Teach read_event/write_event manipulators about bufferevents.

Add an --enable-bufferevents config switch.

configure.in

@@ -118,6 +118,9 @@ if test "$enable_local_appdata" = "yes"; then
             [Defined if we default to host local appdata paths on Windows])
 fi
 
+AC_ARG_ENABLE(bufferevents,
+     AS_HELP_STRING(--enable-bufferevents, use Libevent's buffered IO.))
+
 AC_PROG_CC
 AC_PROG_CPP
 AC_PROG_MAKE_SET
@@ -302,7 +305,7 @@ AC_CHECK_MEMBERS([struct event.min_heap_idx], , ,
 [#include <event.h>
 ])
 
-AC_CHECK_HEADERS(event2/event.h event2/dns.h)
+AC_CHECK_HEADERS(event2/event.h event2/dns.h event2/bufferevent_ssl.h)
 
 LIBS="$save_LIBS"
 LDFLAGS="$save_LDFLAGS"
@@ -323,6 +326,17 @@ fi
 AC_SUBST(TOR_LIBEVENT_LIBS)
 
 
+dnl This isn't the best test for Libevent 2.0.3-alpha.  Once it's released,
+dnl we can do much better.
+if test "$enable_bufferevents" = "yes" && test "$ac_cv_header_event2_bufferevent_ssl_h" != "yes" ; then
+  AC_MSG_ERROR([You've asked for bufferevent support, but you're using a version of Libevent without SSL support.  This won't work.  We need Libevent 2.0.3-alpha or later.  If it isn't released yet, use Libevent from SVN, and talk to Nick.])
+fi
+
+AM_CONDITIONAL(USE_BUFFEREVENTS, test "$enable_bufferevents" = "yes")
+if test "$enable_bufferevents" = "yes"; then
+   AC_DEFINE(USE_BUFFEREVENTS, 1, [Defined if we're going to use Libevent's buffered IO API])
+fi
+
 dnl ------------------------------------------------------
 dnl Where do you live, openssl?  And how do we call you?
 

src/common/compat_libevent.h

@@ -8,6 +8,9 @@
 
 struct event;
 struct event_base;
+#ifdef USE_BUFFEREVENTS
+struct bufferevent;
+#endif
 
 #ifdef HAVE_EVENT2_EVENT_H
 #include <event2/util.h>

src/or/buffers.c

@@ -541,6 +541,7 @@ buf_free(buf_t *buf)
 {
   if (!buf)
     return;
+
   buf_clear(buf);
   buf->magic = 0xdeadbeef;
   tor_free(buf);

src/or/connection.c

@@ -36,6 +36,10 @@
 #include "router.h"
 #include "routerparse.h"
 
+#ifdef USE_BUFFEREVENTS
+#include <event2/bufferevent.h>
+#endif
+
 static connection_t *connection_create_listener(
                                struct sockaddr *listensockaddr,
                                socklen_t listensocklen, int type,
@@ -424,6 +428,10 @@ _connection_free(connection_t *conn)
 
   tor_free(conn->read_event); /* Probably already freed by connection_free. */
   tor_free(conn->write_event); /* Probably already freed by connection_free. */
+  IF_HAS_BUFFEREVENT(conn, {
+      bufferevent_free(conn->bufev);
+      conn->bufev = NULL;
+  });
 
   if (conn->type == CONN_TYPE_DIR) {
     dir_connection_t *dir_conn = TO_DIR_CONN(conn);

src/or/main.c

@@ -56,6 +56,10 @@
 #include <event.h>
 #endif
 
+#ifdef USE_BUFFEREVENTS
+#include <event2/bufferevent.h>
+#endif
+
 void evdns_shutdown(int);
 
 /********* PROTOTYPES **********/
@@ -168,7 +172,7 @@ connection_add(connection_t *conn)
   conn->conn_array_index = smartlist_len(connection_array);
   smartlist_add(connection_array, conn);
 
-  if (conn->s >= 0 || conn->linked) {
+  if (!HAS_BUFFEREVENT(conn) && (conn->s >= 0 || conn->linked)) {
     conn->read_event = tor_event_new(tor_libevent_get_base(),
          conn->s, EV_READ|EV_PERSIST, conn_read_callback, conn);
     conn->write_event = tor_event_new(tor_libevent_get_base(),
@@ -196,6 +200,12 @@ connection_unregister_events(connection_t *conn)
       log_warn(LD_BUG, "Error removing write event for %d", conn->s);
     tor_free(conn->write_event);
   }
+#ifdef USE_BUFFEREVENTS
+  if (conn->bufev) {
+    bufferevent_free(conn->bufev);
+    conn->bufev = NULL;
+  }
+#endif
   if (conn->dns_server_port) {
     dnsserv_close_listener(conn);
   }
@@ -310,6 +320,15 @@ get_connection_array(void)
 void
 connection_watch_events(connection_t *conn, watchable_events_t events)
 {
+  IF_HAS_BUFFEREVENT(conn, {
+      short ev = ((short)events) & (EV_READ|EV_WRITE);
+      short old_ev = bufferevent_get_enabled(conn->bufev);
+      if ((ev & ~old_ev) != 0)
+        bufferevent_enable(conn->bufev, ev);
+      if ((old_ev & ~ev) != 0)
+        bufferevent_disable(conn->bufev, old_ev & ~ev);
+      return;
+  });
   if (events & READ_EVENT)
     connection_start_reading(conn);
   else
@@ -327,6 +346,9 @@ connection_is_reading(connection_t *conn)
 {
   tor_assert(conn);
 
+  IF_HAS_BUFFEREVENT(conn,
+    return (bufferevent_get_enabled(conn->bufev) & EV_READ) != 0;
+  );
   return conn->reading_from_linked_conn ||
     (conn->read_event && event_pending(conn->read_event, EV_READ, NULL));
 }
@@ -336,6 +358,12 @@ void
 connection_stop_reading(connection_t *conn)
 {
   tor_assert(conn);
+
+  IF_HAS_BUFFEREVENT(conn, {
+      bufferevent_disable(conn->bufev, EV_READ);
+      return;
+  });
+
   tor_assert(conn->read_event);
 
   if (conn->linked) {
@@ -355,6 +383,12 @@ void
 connection_start_reading(connection_t *conn)
 {
   tor_assert(conn);
+
+  IF_HAS_BUFFEREVENT(conn, {
+      bufferevent_enable(conn->bufev, EV_READ);
+      return;
+  });
+
   tor_assert(conn->read_event);
 
   if (conn->linked) {
@@ -376,6 +410,10 @@ connection_is_writing(connection_t *conn)
 {
   tor_assert(conn);
 
+  IF_HAS_BUFFEREVENT(conn,
+    return (bufferevent_get_enabled(conn->bufev) & EV_WRITE) != 0;
+  );
+
   return conn->writing_to_linked_conn ||
     (conn->write_event && event_pending(conn->write_event, EV_WRITE, NULL));
 }
@@ -385,6 +423,12 @@ void
 connection_stop_writing(connection_t *conn)
 {
   tor_assert(conn);
+
+  IF_HAS_BUFFEREVENT(conn, {
+      bufferevent_disable(conn->bufev, EV_WRITE);
+      return;
+  });
+
   tor_assert(conn->write_event);
 
   if (conn->linked) {
@@ -407,6 +451,11 @@ connection_start_writing(connection_t *conn)
   tor_assert(conn);
   tor_assert(conn->write_event);
 
+  IF_HAS_BUFFEREVENT(conn, {
+      bufferevent_enable(conn->bufev, EV_WRITE);
+      return;
+  });
+
   if (conn->linked) {
     conn->writing_to_linked_conn = 1;
     if (conn->linked_conn &&

src/or/main.h

@@ -24,6 +24,7 @@ int connection_is_on_closeable_list(connection_t *conn);
 smartlist_t *get_connection_array(void);
 
 typedef enum watchable_events {
+  /* Yes, it is intentional that these match Libevent's EV_READ and EV_WRITE */
   READ_EVENT=0x02,
   WRITE_EVENT=0x04
 } watchable_events_t;

src/or/or.h

@@ -959,6 +959,7 @@ typedef struct connection_t {
   /** Our socket; -1 if this connection is closed, or has no socket. */
   evutil_socket_t s;
   int conn_array_index; /**< Index into the global connection array. */
+
   struct event *read_event; /**< Libevent event structure. */
   struct event *write_event; /**< Libevent event structure. */
   buf_t *inbuf; /**< Buffer holding data read over this connection. */
@@ -969,6 +970,11 @@ typedef struct connection_t {
                               * read? */
   time_t timestamp_lastwritten; /**< When was the last time libevent said we
                                  * could write? */
+
+#ifdef USE_BUFFEREVENTS
+  struct bufferevent *bufev; /**< A Libevent buffered IO structure. */
+#endif
+
   time_t timestamp_created; /**< When was this connection_t created? */
 
   /* XXXX_IP6 make this IPv6-capable */
@@ -1264,6 +1270,19 @@ static INLINE control_connection_t *TO_CONTROL_CONN(connection_t *c)
   return DOWNCAST(control_connection_t, c);
 }
 
+#ifdef USE_BUFFEREVENTS
+#define HAS_BUFFEREVENT(c) (((c)->bufev) != NULL)
+#define IF_HAS_BUFFEREVENT(c, stmt)               \
+  do {                                               \
+    if ((conn)->bufev) do {                          \
+        stmt ;                                       \
+      } while(0);                                    \
+  } while (0)
+#else
+#define HAS_BUFFEREVENT(c) (0)
+#define IF_HAS_BUFFEREVENT(c, stmt) (void)0
+#endif
+
 /** What action type does an address policy indicate: accept or reject? */
 typedef enum {
   ADDR_POLICY_ACCEPT=1,

src/or/relay.c

@@ -2214,7 +2214,7 @@ set_streams_blocked_on_circ(circuit_t *circ, or_connection_t *orconn,
       edge->edge_blocked_on_circ = block;
     }
 
-    if (!conn->read_event) {
+    if (!conn->read_event && !HAS_BUFFEREVENT(conn)) {
       /* This connection is a placeholder for something; probably a DNS
        * request.  It can't actually stop or start reading.*/
       continue;