0
@@ -8,20 +8,24 @@
0
 
0
 static void backend_accept_cb(EV_P_ struct ev_io *watcher, int revents)
0
 {
0
-  backend_t    *server = get_ev_data(backend, watcher, accept);
0
+  backend_t         *backend = get_ev_data(backend, watcher, accept);
0
   struct sockaddr_in remote_addr;
0
   socklen_t          sin_size = sizeof(remote_addr);
0
   int                fd, flags;
0
   
0
-  fd = accept(server->fd, (struct sockaddr *)&remote_addr, &sin_size);
0
-  if (fd == -1)
0
-    rb_sys_fail("accept");
0
+  fd = accept(backend->fd, (struct sockaddr *)&remote_addr, &sin_size);
0
+  if (fd == -1) {
0
+    log_errno(backend);
0
+    return;
0
+  }
0
   
0
   flags = fcntl(fd, F_GETFL, 0);
0
-  if ((fcntl(fd, F_SETFL, flags | O_NONBLOCK)) < 0)
0
-    rb_sys_fail("fcntl");
0
+  if ((fcntl(fd, F_SETFL, flags | O_NONBLOCK)) < 0) {
0
+    log_errno(backend);
0
+    return;
0
+  }
0
   
0
-  connection_start(server, fd, remote_addr);
0
+  connection_start(backend, fd, remote_addr);
0
 }
0
 
0
 

0
@@ -4,59 +4,58 @@ static VALUE sInternedCall;
0
 static VALUE sInternedKeys;
0
 static VALUE sRackInput;
0
 
0
-#define connection_error(connection, msg) \
0
-  rb_funcall(connection->backend->obj, rb_intern("log_error"), 1, \
0
-             rb_exc_new(rb_eRuntimeError, msg, strlen(msg))); \
0
-  connection_close(connection)
0
+#define connection_error(c, msg) \
0
+  log_error(c->backend, msg); \
0
+  connection_close(c)
0
 
0
-#define connection_errorno(connection) \
0
-  connection_error(connection, strerror(errno))
0
+#define connection_errno(c) \
0
+  connection_error(c, strerror(errno))
0
 
0
 /* event callbacks */
0
 
0
 static void connection_closable_cb(EV_P_ struct ev_io *watcher, int revents)
0
 {
0
-  connection_t *connection = get_ev_data(connection, watcher, write);
0
+  connection_t *c = get_ev_data(connection, watcher, write);
0
   
0
-  connection_close(connection);
0
+  connection_close(c);
0
 }
0
 
0
 static void connection_writable_cb(EV_P_ struct ev_io *watcher, int revents)
0
 {
0
-  connection_t *connection = get_ev_data(connection, watcher, write);
0
-  int                sent;
0
+  connection_t *c = get_ev_data(connection, watcher, write);
0
+  int           sent;
0
   
0
-  sent = send(connection->fd,
0
-              (char *) connection->write_buffer.ptr + connection->write_buffer.current,
0
-              connection->write_buffer.len - connection->write_buffer.current,
0
+  sent = send(c->fd,
0
+              (char *) c->write_buffer.ptr + c->write_buffer.current,
0
+              c->write_buffer.len - c->write_buffer.current,
0
               0);
0
   
0
   if (sent > 0) {
0
-    connection->write_buffer.current += sent;
0
+    c->write_buffer.current += sent;
0
   } else {
0
-    connection_errorno(connection);
0
+    connection_errno(c);
0
     return;    
0
   }
0
   
0
-  if (connection->write_buffer.current == connection->write_buffer.len) {
0
+  if (c->write_buffer.current == c->write_buffer.len) {
0
     watcher->cb = connection_closable_cb;
0
   }
0
 }
0
 
0
 static void connection_readable_cb(EV_P_ struct ev_io *watcher, int revents)
0
 {
0
-  connection_t *connection = get_ev_data(connection, watcher, read);
0
-  size_t             n;
0
-  char               buf[BUFFER_SIZE];
0
+  connection_t *c = get_ev_data(connection, watcher, read);
0
+  size_t        n;
0
+  char          buf[BUFFER_SIZE];
0
 
0
-  n = recv(connection->fd, buf, BUFFER_SIZE, 0);
0
+  n = recv(c->fd, buf, BUFFER_SIZE, 0);
0
   
0
   if (n == -1) {
0
-    connection_errorno(connection);
0
+    connection_errno(c);
0
     return;
0
   }
0
   
0
-  connection_parse(connection, buf, n);
0
+  connection_parse(c, buf, n);
0
 }
0
 
0
 
0
@@ -64,142 +63,138 @@ static void connection_readable_cb(EV_P_ struct ev_io *watcher, int revents)
0
 
0
 void connection_start(backend_t *backend, int fd, struct sockaddr_in remote_addr)
0
 {
0
-  connection_t *connection = NULL;
0
-  connection_t *connections = backend->connections->items;
0
-  int                i = 0;
0
+  connection_t *c  = NULL;
0
+  connection_t *cs = backend->connections->items;
0
+  int           i  = 0;
0
   
0
   /* select the first closed connection */
0
   for (i = 0; i < backend->connections->nitems; i++) {
0
-    if (!connections[i].open) {
0
-      connection = &connections[i];
0
+    if (!cs[i].open) {
0
+      c = &cs[i];
0
       break;
0
     }
0
   }
0
   
0
   /* no free connection found, add more */
0
-  if (connection == NULL) {
0
+  if (c == NULL) {
0
     connections_create(backend->connections, CONNECTIONS_SIZE);
0
-    connections = backend->connections->items;
0
+    cs = backend->connections->items;
0
     /* FIXME: bug here on high concurrency, causes segfault on line 88 */
0
-    connection = &connections[++i];
0
+    c = &cs[++i];
0
   }
0
   
0
-  assert(connection != NULL);
0
-  assert(!connection->open);
0
+  assert(c != NULL);
0
+  assert(!c->open);
0
   
0
   /* init connection */
0
-  connection->open = 1;
0
-  connection->loop = backend->loop;
0
-  connection->buffer_pool = backend->buffer_pool;
0
-  connection->backend = backend;
0
-  connection->content_length = 0;
0
-  connection->fd = fd;
0
-  connection->remote_addr = remote_addr;
0
+  c->open           = 1;
0
+  c->loop           = backend->loop;
0
+  c->buffer_pool    = backend->buffer_pool;
0
+  c->backend        = backend;
0
+  c->content_length = 0;
0
+  c->fd             = fd;
0
+  c->remote_addr    = remote_addr;
0
   
0
   /* mark as used to Ruby GC */
0
-  connection->env = rb_hash_new();
0
-  rb_gc_register_address(&connection->env);
0
+  c->env = rb_hash_new();
0
+  rb_gc_register_address(&c->env);
0
   
0
   /* alloc read buffer from pool */
0
-  connection->read_buffer.ptr = palloc(connection->buffer_pool, 1);
0
-  if (connection->read_buffer.ptr == NULL)
0
-    rb_sys_fail("palloc");
0
-  connection->read_buffer.nalloc = 1;
0
-  connection->read_buffer.salloc = connection->buffer_pool->size;
0
-  connection->read_buffer.len = 0;
0
-  connection->read_buffer.current = 0;
0
+  c->read_buffer.ptr     = palloc(c->buffer_pool, 1);
0
+  assert(c->read_buffer.ptr);
0
+  c->read_buffer.nalloc  = 1;
0
+  c->read_buffer.salloc  = c->buffer_pool->size;
0
+  c->read_buffer.len     = 0;
0
+  c->read_buffer.current = 0;
0
   
0
   /* assign env[rack.input] */
0
-  connection->input = input_new(&connection->read_buffer);
0
-  rb_gc_register_address(&connection->input);
0
-  rb_hash_aset(connection->env, sRackInput, connection->input);
0
-  
0
-  connection->write_buffer.ptr = palloc(connection->buffer_pool, 1);
0
-  if (connection->write_buffer.ptr == NULL)
0
-    rb_sys_fail("palloc");
0
-  connection->write_buffer.nalloc = 1;
0
-  connection->write_buffer.salloc = connection->buffer_pool->size;
0
-  connection->write_buffer.len = 0;
0
-  connection->write_buffer.current = 0;
0
+  c->input = input_new(&c->read_buffer);
0
+  rb_gc_register_address(&c->input);
0
+  rb_hash_aset(c->env, sRackInput, c->input);
0
+  
0
+  /* alloc write buffer from pool */
0
+  c->write_buffer.ptr     = palloc(c->buffer_pool, 1);
0
+  assert(c->write_buffer.ptr);
0
+  c->write_buffer.nalloc  = 1;
0
+  c->write_buffer.salloc  = c->buffer_pool->size;
0
+  c->write_buffer.len     = 0;
0
+  c->write_buffer.current = 0;
0
   
0
   /* reinit parser */
0
-  http_parser_init(&connection->parser);
0
-  connection->parser.data = connection;
0
+  http_parser_init(&c->parser);
0
+  c->parser.data = c;
0
   
0
   /* init libev stuff */
0
-  watch(connection, connection_readable_cb, read, EV_READ);
0
+  watch(c, connection_readable_cb, read, EV_READ);
0
   
0
   /* TODO add timeout watcher */
0
 }
0
 
0
-void connection_parse(connection_t *connection, char *buf, int len)
0
+void connection_parse(connection_t *c, char *buf, int len)
0
 {
0
-  if (!http_parser_is_finished(&connection->parser)
0
-      && connection->read_buffer.len + len > MAX_HEADER) {
0
-    connection_error(connection, "Header too big");
0
+  if (!http_parser_is_finished(&c->parser) && c->read_buffer.len + len > MAX_HEADER) {
0
+    connection_error(c, "Header too big");
0
     return;
0
   }
0
   
0
   /* alloc more mem when buffer full */
0
   /* TODO extract this into buffer.c and optimize */
0
   /* TODO store big body in tempfile */
0
-  if (connection->read_buffer.len >= connection->read_buffer.salloc) {
0
+  if (c->read_buffer.len >= c->read_buffer.salloc) {
0
     char *new, *old;
0
     
0
     /* TODO if last alloc, just alloc next block */
0
-    old = connection->read_buffer.ptr;
0
-    new = (char *) palloc(connection->buffer_pool,
0
-                          connection->read_buffer.nalloc + 1);
0
-    if (new == NULL)
0
-      rb_sys_fail("palloc");
0
-      
0
-    memcpy(new, old, connection->read_buffer.len);
0
+    old = c->read_buffer.ptr;
0
+    new = (char *) palloc(c->buffer_pool, c->read_buffer.nalloc + 1);
0
+    assert(new);
0
+    
0
+    memcpy(new, old, c->read_buffer.len);
0
 
0
-    connection->read_buffer.ptr = new;
0
-    connection->read_buffer.nalloc ++;
0
-    connection->read_buffer.salloc += connection->buffer_pool->size;
0
-    pfree(connection->buffer_pool, old);
0
+    c->read_buffer.ptr     = new;
0
+    c->read_buffer.nalloc ++;
0
+    c->read_buffer.salloc += c->buffer_pool->size;
0
+    pfree(c->buffer_pool, old);
0
   }
0
   
0
-  memcpy(connection->read_buffer.ptr + connection->read_buffer.len, buf, len);
0
-  connection->read_buffer.len += len;
0
+  memcpy(c->read_buffer.ptr + c->read_buffer.len, buf, len);
0
+  c->read_buffer.len += len;
0
   
0
-  if (!http_parser_is_finished(&connection->parser)) {
0
+  if (!http_parser_is_finished(&c->parser)) {
0
     /* header not all received, we continue parsing ... */
0
     
0
     /* terminate string with null (required by ragel v5) */
0
-    memset(connection->read_buffer.ptr + connection->read_buffer.len, '\0', 1);
0
+    memset(c->read_buffer.ptr + c->read_buffer.len, '\0', 1);
0
     
0
     /* parse the request into connection->env */
0
-    connection->parser.nread = http_parser_execute(&connection->parser,
0
-                                                    connection->read_buffer.ptr,
0
-                                                    connection->read_buffer.len,
0
-                                                    connection->parser.nread);
0
+    c->parser.nread = http_parser_execute(&c->parser,
0
+                                           c->read_buffer.ptr,
0
+                                           c->read_buffer.len,
0
+                                           c->parser.nread);
0
   
0
     /* parser error */
0
-    if (http_parser_has_error(&connection->parser)) {
0
-      connection_error(connection, "Invalid request");
0
+    if (http_parser_has_error(&c->parser)) {
0
+      connection_error(c, "Invalid request");
0
       return;
0
     }
0
   }
0
   
0
   /* request fully received */
0
-  if (http_parser_is_finished(&connection->parser) && connection->read_buffer.len >= connection->content_length) {
0
-    unwatch(connection, read);
0
-    connection_process(connection);
0
+  if (http_parser_is_finished(&c->parser) && c->read_buffer.len >= c->content_length) {
0
+    unwatch(c, read);
0
+    connection_process(c);
0
   }
0
 }
0
 
0
-void connection_send_status(connection_t *connection, const int status)
0
+void connection_send_status(connection_t *c, const int status)
0
 {
0
   size_t n;
0
   
0
-  n = sprintf(connection->write_buffer.ptr, "HTTP/1.1 %s" CRLF, get_status_line(status));
0
+  n = sprintf(c->write_buffer.ptr, "HTTP/1.1 %s" CRLF, get_status_line(status));
0
   
0
-  connection->write_buffer.len = n;
0
+  c->write_buffer.len = n;
0
 }
0
 
0
-void connection_send_headers(connection_t *connection, VALUE headers)
0
+void connection_send_headers(connection_t *c, VALUE headers)
0
 {
0
   VALUE   hash, keys, key, value;
0
   size_t  i, n;
0
@@ -210,55 +205,54 @@ void connection_send_headers(connection_t *connection, VALUE headers)
0
   for (i = 0; i < RARRAY_LEN(keys); ++i) {
0
     key = RARRAY_PTR(keys)[i];
0
     value = rb_hash_aref(headers, key);
0
-    n += sprintf((char *) connection->write_buffer.ptr + connection->write_buffer.len +  n,
0
+    n += sprintf((char *) c->write_buffer.ptr + c->write_buffer.len +  n,
0
                  "%s: %s" CRLF,
0
                  RSTRING_PTR(key),
0
                  RSTRING_PTR(value));
0
   }
0
-
0
-  connection->write_buffer.len += n;
0
+  c->write_buffer.len += n;
0
   
0
-  memcpy(connection->write_buffer.ptr + connection->write_buffer.len, CRLF, 2);
0
-  connection->write_buffer.len += 2;
0
+  memcpy(c->write_buffer.ptr + c->write_buffer.len, CRLF, 2);
0
+  c->write_buffer.len += 2;
0
 }
0
 
0
 static VALUE iter_body(VALUE chunk, VALUE *val_conn)
0
 {
0
-  connection_t *conn = (connection_t *) val_conn;
0
-  size_t             len  = RSTRING_LEN(chunk);
0
+  connection_t *c = (connection_t *) val_conn;
0
+  size_t        len  = RSTRING_LEN(chunk);
0
   
0
-  memcpy(conn->write_buffer.ptr + conn->write_buffer.len, RSTRING_PTR(chunk), len);
0
-  conn->write_buffer.len += len;
0
+  memcpy(c->write_buffer.ptr + c->write_buffer.len, RSTRING_PTR(chunk), len);
0
+  c->write_buffer.len += len;
0
   
0
   return Qnil;
0
 }
0
 
0
-int connection_send_body(connection_t *conn, VALUE body)
0
+int connection_send_body(connection_t *c, VALUE body)
0
 {
0
   if (TYPE(body) == T_STRING) {
0
     /* Calling String#each creates several other strings which is slower and use more mem,
0
      * also Ruby 1.9 doesn't define that method anymore, so it's better to send one big string. */
0
     size_t len = RSTRING_LEN(body);
0
     
0
-    memcpy(conn->write_buffer.ptr + conn->write_buffer.len, RSTRING_PTR(body), len);
0
-    conn->write_buffer.len += len;
0
+    memcpy(c->write_buffer.ptr + c->write_buffer.len, RSTRING_PTR(body), len);
0
+    c->write_buffer.len += len;
0
     
0
   } else {
0
     /* Iterate over body#each and send each yielded chunk */
0
-    rb_iterate(rb_each, body, iter_body, (VALUE) conn);
0
+    rb_iterate(rb_each, body, iter_body, (VALUE) c);
0
     
0
   }
0
 }
0
 
0
-void connection_process(connection_t *connection)
0
+void connection_process(connection_t *c)
0
 {
0
   /* Call the app to process the request */
0
-  VALUE response = rb_funcall_rescue(connection->backend->app, sInternedCall, 1, connection->env);
0
+  VALUE response = rb_funcall_rescue(c->backend->app, sInternedCall, 1, c->env);
0
 
0
   if (response == Qundef) {
0
     /* log any error */
0
-    rb_funcall(connection->backend->obj, rb_intern("log_error"), 0);
0
-    connection_close(connection);
0
+    rb_funcall(c->backend->obj, rb_intern("log_error"), 0);
0
+    connection_close(c);
0
   } else {
0
     /* store response info and prepare for writing */
0
     int   status  = FIX2INT(rb_ary_entry(response, 0));
0
@@ -266,36 +260,36 @@ void connection_process(connection_t *connection)
0
     VALUE body    = rb_ary_entry(response, 2);
0
     
0
     /* TODO grow buffer if too small */
0
-    connection_send_status(connection, status);
0
-    connection_send_headers(connection, headers);
0
-    connection_send_body(connection, body);
0
+    connection_send_status(c, status);
0
+    connection_send_headers(c, headers);
0
+    connection_send_body(c, body);
0
   
0
-    watch(connection, connection_writable_cb, write, EV_WRITE);
0
+    watch(c, connection_writable_cb, write, EV_WRITE);
0
   }
0
 }
0
 
0
-void connection_close(connection_t *connection)
0
+void connection_close(connection_t *c)
0
 {
0
-  unwatch(connection, read);
0
-  unwatch(connection, write);
0
+  unwatch(c, read);
0
+  unwatch(c, write);
0
 
0
-  close(connection->fd);
0
+  close(c->fd);
0
   
0
-  if (connection->read_buffer.ptr != NULL)
0
-    pfree(connection->buffer_pool, connection->read_buffer.ptr);
0
-  connection->read_buffer.salloc = 0;
0
-  connection->read_buffer.nalloc = 0;
0
+  if (c->read_buffer.ptr != NULL)
0
+    pfree(c->buffer_pool, c->read_buffer.ptr);
0
+  c->read_buffer.salloc = 0;
0
+  c->read_buffer.nalloc = 0;
0
   
0
-  if (connection->write_buffer.ptr != NULL)
0
-    pfree(connection->buffer_pool, connection->write_buffer.ptr);
0
-  connection->write_buffer.salloc = 0;
0
-  connection->write_buffer.nalloc = 0;
0
+  if (c->write_buffer.ptr != NULL)
0
+    pfree(c->buffer_pool, c->write_buffer.ptr);
0
+  c->write_buffer.salloc = 0;
0
+  c->write_buffer.nalloc = 0;
0
   
0
   /* tell Ruby GC vars are not used anymore */
0
-  rb_gc_unregister_address(&connection->env);
0
-  rb_gc_unregister_address(&connection->input);
0
+  rb_gc_unregister_address(&c->env);
0
+  rb_gc_unregister_address(&c->input);
0
   
0
-  connection->open = 0;
0
+  c->open = 0;
0
 }
0
 
0
 

0
@@ -112,6 +112,12 @@ struct backend_s {
0
   (type##_t *) w->data; \
0
   assert(&((type##_t *)w->data)->event##_watcher == w);
0
 
0
+#define log_error(b, msg) \
0
+  rb_funcall(b->obj, rb_intern("log_error"), 1, \
0
+             rb_exc_new(rb_eRuntimeError, msg, strlen(msg)))
0
+
0
+#define log_errno(b) log_error(b, strerror(errno))
0
+
0
 void backend_define(void);
0
 
0
 void input_define(void);