Re: [Libevent-users] workers for evhttp (was stable, now not!)

[Michael Herf <herf@xxxxxxxxx>]

Nick Mathewson wrote:

1. I added BEV_OPT_THREADSAFE in evhttp_connection_base_new (evcon->bufev =
    

We won't want to do this unconditionally: not everybody will want the
overhead of locking in an evhttp that isn't going to be used in
multiple threads.
  

Maybe evthread_use_pthreads() could enable this behavior?
There could be a more complicated way to change it, but I think keeping these things roughly in sync makes sense.

First thing to do is, in the future, please use the "unified diff"
format that "diff -u" generates?  It's lots easier to read.
  

Oops. below --

As for the patch itself: there's going to be more work needed to make
evhttp threadsafe.  It looks like most of what you're doing so far is
to make evhttp's use of bufferevents threadsafe.  That's a necessary
component, but we'll also need to make sure that the evhttp structures
themselves are safe to use from multiple threads at a time.  That
means at a minimum getting them locks to prevent concurrent
modifications.  We'll also need to figure out the correct behavior wrt
preventing deadlocks between http locking and bufferevent locking --
possibly by the expedient of just having the bufferevent and its
evhttp objects share a lock.
  

I'm parsing the request in the network thread, so I think this is all working because I assume that what's in evhttp_connection & evhttp is *read-only* by the time I'm in EVCON_WRITING.

If this is true, everything but the bufferevent could be lock-free if it were formalized a bit more (i.e. a read-only/const portion and a writable portion of the evhttp structure?), so this might actually avoid unnecessary locking. There certainly may be writes to these structures in other code paths that I didn't check for, so it might not be so easy. Certainly my "sloppy" approach to evhttp doesn't generalize to dealing with the request earlier (e.g., a multi-CPU header parser), but that would require a new event base per thread as well.

mike

--- http.c    2011-04-27 14:07:21.000000000 -0700
+++ /home/mike/libevent/http.c    2011-05-21 23:23:14.813264896 -0700
@@ -369,8 +369,6 @@
     evcon->cb = cb;
     evcon->cb_arg = arg;
 
-    bufferevent_enable(evcon->bufev, EV_WRITE);
-
     /* Disable the read callback: we don't actually care about data;
      * we only care about close detection.  (We don't disable reading,
      * since we *do* want to learn about any close events.) */
@@ -379,6 +377,9 @@
         evhttp_write_cb,
         evhttp_error_cb,
         evcon);
+
+    // mbh: moved after setcb, so we don't race with the network thread:
+    bufferevent_enable(evcon->bufev, EV_WRITE);
 }
 
 static void
@@ -391,7 +392,6 @@
 evhttp_send_continue(struct evhttp_connection *evcon,
             struct evhttp_request *req)
 {
-    bufferevent_enable(evcon->bufev, EV_WRITE);
     evbuffer_add_printf(bufferevent_get_output(evcon->bufev),
             "HTTP/%d.%d 100 Continue\r\n\r\n",
             req->major, req->minor);
@@ -402,6 +402,9 @@
         evhttp_write_cb,
         evhttp_error_cb,
         evcon);
+
+    // mbh: moved after, to avoid races:
+    bufferevent_enable(evcon->bufev, EV_WRITE);
 }
 
 /** Helper: returns true iff evconn is in any connected state. */
@@ -995,6 +998,7 @@
 static void
 evhttp_read_cb(struct bufferevent *bufev, void *arg)
 {
+
     struct evhttp_connection *evcon = arg;
     struct evhttp_request *req = TAILQ_FIRST(&evcon->requests);
 
@@ -2037,6 +2041,7 @@
         goto error;
     }
 
+#if 0
     if ((evcon->bufev = bufferevent_new(-1,
             evhttp_read_cb,
             evhttp_write_cb,
@@ -2044,6 +2049,19 @@
         event_warn("%s: bufferevent_new failed", __func__);
         goto error;
     }
+#else
+    // threadsafe, using newer API
+    evcon->bufev = bufferevent_socket_new(NULL, -1, BEV_OPT_THREADSAFE);
+
+    if (evcon->bufev == NULL) {
+        printf("bufferevent_socket_net failed\n");
+        event_warn("%s: bufferevent_new failed", __func__);
+        goto error;
+    }
+
+    bufferevent_setcb(evcon->bufev, evhttp_read_cb, evhttp_write_cb, evhttp_error_cb, evcon);
+
+#endif
 
     evcon->state = EVCON_DISCONNECTED;
     TAILQ_INIT(&evcon->requests);
@@ -2142,8 +2160,8 @@
         evcon);
     bufferevent_settimeout(evcon->bufev, 0,
         evcon->timeout != -1 ? evcon->timeout : HTTP_CONNECT_TIMEOUT);
-    /* make sure that we get a write callback */
-    bufferevent_enable(evcon->bufev, EV_WRITE);
+
+    // mbh: bufferevent_enable was here
 
     if (bufferevent_socket_connect_hostname(evcon->bufev, evcon->dns_base,
         AF_UNSPEC, evcon->address, evcon->port) < 0) {
@@ -2159,6 +2177,10 @@
 
     evcon->state = EVCON_CONNECTING;
 
+    // mbh: moved to avoid races
+    /* make sure that we get a write callback */
+    bufferevent_enable(evcon->bufev, EV_WRITE);
+
     return (0);
 }
 
@@ -2367,7 +2389,15 @@
 {
     evhttp_response_code(req, code, reason);
 
+    if (req->evcon && req->evcon->bufev) {
+        struct bufferevent *bev = req->evcon->bufev;
+
+        bufferevent_incref(bev);
     evhttp_send(req, databuf);
+
+        // note: req may be freed, but our incref is still around:
+        bufferevent_decref(bev);
+    }
 }