Merge branch 'isolate_libevent_2_squashed'

6 files changed, 224 insertions, 42 deletions

diff --git a/src/common/compat_libevent.c b/src/common/compat_libevent.c
index 735385557c..cb311ea462 100644
--- a/src/common/compat_libevent.c
+++ b/src/common/compat_libevent.c
@@ -221,6 +221,121 @@ periodic_timer_free_(periodic_timer_t *timer)
   tor_free(timer);
 }
 
+/**
+ * Type used to represent events that run directly from the main loop,
+ * either because they are activated from elsewhere in the code, or
+ * because they have a simple timeout.
+ *
+ * We use this type to avoid exposing Libevent's API throughout the rest
+ * of the codebase.
+ *
+ * This type can't be used for all events: it doesn't handle events that
+ * are triggered by signals or by sockets.
+ */
+struct mainloop_event_t {
+  struct event *ev;
+  void (*cb)(mainloop_event_t *, void *);
+  void *userdata;
+};
+
+/**
+ * Internal: Implements mainloop event using a libevent event.
+ */
+static void
+mainloop_event_cb(evutil_socket_t fd, short what, void *arg)
+{
+  (void)fd;
+  (void)what;
+  mainloop_event_t *mev = arg;
+  mev->cb(mev, mev->userdata);
+}
+
+/**
+ * Create and return a new mainloop_event_t to run the function <b>cb</b>.
+ *
+ * When run, the callback function will be passed the mainloop_event_t
+ * and <b>userdata</b> as its arguments.  The <b>userdata</b> pointer
+ * must remain valid for as long as the mainloop_event_t event exists:
+ * it is your responsibility to free it.
+ *
+ * The event is not scheduled by default: Use mainloop_event_activate()
+ * or mainloop_event_schedule() to make it run.
+ */
+mainloop_event_t *
+mainloop_event_new(void (*cb)(mainloop_event_t *, void *),
+                   void *userdata)
+{
+  tor_assert(cb);
+
+  struct event_base *base = tor_libevent_get_base();
+  mainloop_event_t *mev = tor_malloc_zero(sizeof(mainloop_event_t));
+  mev->ev = tor_event_new(base, -1, 0, mainloop_event_cb, mev);
+  tor_assert(mev->ev);
+  mev->cb = cb;
+  mev->userdata = userdata;
+  return mev;
+}
+
+/**
+ * Schedule <b>event</b> to run in the main loop, immediately.  If it is
+ * not scheduled, it will run anyway. If it is already scheduled to run
+ * later, it will run now instead.  This function will have no effect if
+ * the event is already scheduled to run.
+ *
+ * This function may only be called from the main thread.
+ */
+void
+mainloop_event_activate(mainloop_event_t *event)
+{
+  tor_assert(event);
+  event_active(event->ev, EV_READ, 1);
+}
+
+/** Schedule <b>event</b> to run in the main loop, after a delay of <b>tv</b>.
+ *
+ * If the event is scheduled for a different time, cancel it and run
+ * after this delay instead.  If the event is currently pending to run
+ * <em>now</b>, has no effect.
+ *
+ * Do not call this function with <b>tv</b> == NULL -- use
+ * mainloop_event_activate() instead.
+ *
+ * This function may only be called from the main thread.
+ */
+int
+mainloop_event_schedule(mainloop_event_t *event, const struct timeval *tv)
+{
+  tor_assert(event);
+  if (BUG(tv == NULL)) {
+    // LCOV_EXCL_START
+    mainloop_event_activate(event);
+    return 0;
+    // LCOV_EXCL_STOP
+  }
+  return event_add(event->ev, tv);
+}
+
+/** Cancel <b>event</b> if it is currently active or pending. (Do nothing if
+ * the event is not currently active or pending.) */
+void
+mainloop_event_cancel(mainloop_event_t *event)
+{
+  if (!event)
+    return;
+  event_del(event->ev);
+}
+
+/** Cancel <b>event</b> and release all storage associated with it. */
+void
+mainloop_event_free_(mainloop_event_t *event)
+{
+  if (!event)
+    return;
+  tor_event_free(event->ev);
+  memset(event, 0xb8, sizeof(*event));
+  tor_free(event);
+}
+
 int
 tor_init_libevent_rng(void)
 {
@@ -248,6 +363,38 @@ tor_libevent_free_all(void)
   the_event_base = NULL;
 }
 
+/**
+ * Run the event loop for the provided event_base, handling events until
+ * something stops it.  If <b>once</b> is set, then just poll-and-run
+ * once, then exit.  Return 0 on success, -1 if an error occurred, or 1
+ * if we exited because no events were pending or active.
+ *
+ * This isn't reentrant or multithreaded.
+ */
+int
+tor_libevent_run_event_loop(struct event_base *base, int once)
+{
+  const int flags = once ? EVLOOP_ONCE : 0;
+  return event_base_loop(base, flags);
+}
+
+/** Tell the event loop to exit after <b>delay</b>.  If <b>delay</b> is NULL,
+ * instead exit after we're done running the currently active events. */
+void
+tor_libevent_exit_loop_after_delay(struct event_base *base,
+                                   const struct timeval *delay)
+{
+  event_base_loopexit(base, delay);
+}
+
+/** Tell the event loop to exit after running whichever callback is currently
+ * active. */
+void
+tor_libevent_exit_loop_after_callback(struct event_base *base)
+{
+  event_base_loopbreak(base);
+}
+
 #if defined(LIBEVENT_VERSION_NUMBER) &&         \
   LIBEVENT_VERSION_NUMBER >= V(2,1,1) &&        \
   !defined(TOR_UNIT_TESTS)
diff --git a/src/common/compat_libevent.h b/src/common/compat_libevent.h
index 1853e50917..c8b0371564 100644
--- a/src/common/compat_libevent.h
+++ b/src/common/compat_libevent.h
@@ -7,8 +7,6 @@
 #include "orconfig.h"
 #include "testsupport.h"
 
-#include <event2/event.h>
-
 void configure_libevent_logging(void);
 void suppress_libevent_log_msg(const char *msg);
 
@@ -19,6 +17,9 @@ void suppress_libevent_log_msg(const char *msg);
   evdns_add_server_port_with_base(tor_libevent_get_base(), \
   (sock),(tcp),(cb),(data));
 
+struct event;
+struct event_base;
+
 void tor_event_free_(struct event *ev);
 #define tor_event_free(ev) \
   FREE_AND_NULL(struct event, tor_event_free_, (ev))
@@ -33,8 +34,16 @@ void periodic_timer_free_(periodic_timer_t *);
 #define periodic_timer_free(t) \
   FREE_AND_NULL(periodic_timer_t, periodic_timer_free_, (t))
 
-#define tor_event_base_loopexit event_base_loopexit
-#define tor_event_base_loopbreak event_base_loopbreak
+typedef struct mainloop_event_t mainloop_event_t;
+mainloop_event_t *mainloop_event_new(void (*cb)(mainloop_event_t *, void *),
+                                     void *userdata);
+void mainloop_event_activate(mainloop_event_t *event);
+int mainloop_event_schedule(mainloop_event_t *event,
+                            const struct timeval *delay);
+void mainloop_event_cancel(mainloop_event_t *event);
+void mainloop_event_free_(mainloop_event_t *event);
+#define mainloop_event_free(event) \
+  FREE_AND_NULL(mainloop_event_t, mainloop_event_free_, (event))
 
 /** Defines a configuration for using libevent with Tor: passed as an argument
  * to tor_libevent_initialize() to describe how we want to set up. */
@@ -63,6 +72,11 @@ void tor_gettimeofday_cache_set(const struct timeval *tv);
 void tor_libevent_postfork(void);
 #endif
 
+int tor_libevent_run_event_loop(struct event_base *base, int once);
+void tor_libevent_exit_loop_after_delay(struct event_base *base,
+                                        const struct timeval *delay);
+void tor_libevent_exit_loop_after_callback(struct event_base *base);
+
 #ifdef COMPAT_LIBEVENT_PRIVATE
 
 /** Macro: returns the number of a Libevent version as a 4-byte number,
diff --git a/src/common/procmon.c b/src/common/procmon.c
index abcbbeaa21..73c14cd584 100644
--- a/src/common/procmon.c
+++ b/src/common/procmon.c
@@ -10,8 +10,6 @@
 
 #include "util.h"
 
-#include <event2/event.h>
-
 #ifdef HAVE_SIGNAL_H
 #include <signal.h>
 #endif
@@ -44,7 +42,7 @@ typedef int pid_t;
 /* Currently we need to poll in some way on all systems. */
 
 #ifdef PROCMON_POLLS
-static void tor_process_monitor_poll_cb(evutil_socket_t unused1, short unused2,
+static void tor_process_monitor_poll_cb(periodic_timer_t *ev,
                                         void *procmon_);
 #endif
 
@@ -136,7 +134,7 @@ struct tor_process_monitor_t {
 
   /** A Libevent event structure, to either poll for the process's
    * existence or receive a notification when the process ends. */
-  struct event *e;
+  periodic_timer_t *e;
 
   /** A callback to be called when the process ends. */
   tor_procmon_callback_t cb;
@@ -159,9 +157,6 @@ tor_validate_process_specifier(const char *process_spec,
   return parse_process_specifier(process_spec, &ppspec, msg);
 }
 
-/* XXXX we should use periodic_timer_new() for this stuff */
-#define PERIODIC_TIMER_FLAGS EV_PERSIST
-
 /* DOCDOC poll_interval_tv */
 static const struct timeval poll_interval_tv = {15, 0};
 
@@ -225,13 +220,9 @@ tor_process_monitor_new(struct event_base *base,
   procmon->cb_arg = cb_arg;
 
 #ifdef PROCMON_POLLS
-  procmon->e = tor_event_new(base, -1 /* no FD */, PERIODIC_TIMER_FLAGS,
-                             tor_process_monitor_poll_cb, procmon);
-  /* Note: If you port this file to plain Libevent 2, check that
-   * procmon->e is non-NULL.  We don't need to here because
-   * tor_evtimer_new never returns NULL. */
-
-  evtimer_add(procmon->e, &poll_interval_tv);
+  procmon->e = periodic_timer_new(base,
+                                  &poll_interval_tv,
+                                  tor_process_monitor_poll_cb, procmon);
 #else /* !(defined(PROCMON_POLLS)) */
 #error OOPS?
 #endif /* defined(PROCMON_POLLS) */
@@ -246,14 +237,12 @@ tor_process_monitor_new(struct event_base *base,
 /** Libevent callback to poll for the existence of the process
  * monitored by <b>procmon_</b>. */
 static void
-tor_process_monitor_poll_cb(evutil_socket_t unused1, short unused2,
-                            void *procmon_)
+tor_process_monitor_poll_cb(periodic_timer_t *event, void *procmon_)
 {
+  (void)event;
   tor_process_monitor_t *procmon = (tor_process_monitor_t *)(procmon_);
   int its_dead_jim;
 
-  (void)unused1; (void)unused2;
-
   tor_assert(procmon != NULL);
 
 #ifdef _WIN32
@@ -336,7 +325,7 @@ tor_process_monitor_free_(tor_process_monitor_t *procmon)
 #endif
 
   if (procmon->e != NULL)
-    tor_event_free(procmon->e);
+    periodic_timer_free(procmon->e);
 
   tor_free(procmon);
 }
diff --git a/src/common/timers.c b/src/common/timers.c
index 552080b11e..a90817da1c 100644
--- a/src/common/timers.c
+++ b/src/common/timers.c
@@ -37,8 +37,6 @@
 #include "torlog.h"
 #include "util.h"
 
-#include <event2/event.h>
-
 struct timeout_cb {
   timer_cb_fn_t cb;
   void *arg;
@@ -69,7 +67,7 @@ struct timeout_cb {
 #include "src/ext/timeouts/timeout.c"
 
 static struct timeouts *global_timeouts = NULL;
-static struct event *global_timer_event = NULL;
+static struct mainloop_event_t *global_timer_event = NULL;
 
 static monotime_t start_of_time;
 
@@ -147,7 +145,7 @@ libevent_timer_reschedule(void)
   if (delay > MIN_CHECK_TICKS)
     delay = MIN_CHECK_TICKS;
   timeout_to_tv(delay, &d);
-  event_add(global_timer_event, &d);
+  mainloop_event_schedule(global_timer_event, &d);
 }
 
 /** Run the callback of every timer that has expired, based on the current
@@ -170,10 +168,9 @@ timers_run_pending(void)
  * have fired, activate their callbacks, and reschedule the libevent timer.
  */
 static void
-libevent_timer_callback(evutil_socket_t fd, short what, void *arg)
+libevent_timer_callback(mainloop_event_t *ev, void *arg)
 {
-  (void)fd;
-  (void)what;
+  (void)ev;
   (void)arg;
 
   timers_run_pending();
@@ -203,9 +200,8 @@ timers_initialize(void)
   monotime_init();
   monotime_get(&start_of_time);
 
-  struct event *timer_event;
-  timer_event = tor_event_new(tor_libevent_get_base(),
-                              -1, 0, libevent_timer_callback, NULL);
+  mainloop_event_t *timer_event;
+  timer_event = mainloop_event_new(libevent_timer_callback, NULL);
   tor_assert(timer_event);
   global_timer_event = timer_event;
 
@@ -219,7 +215,7 @@ void
 timers_shutdown(void)
 {
   if (global_timer_event) {
-    tor_event_free(global_timer_event);
+    mainloop_event_free(global_timer_event);
     global_timer_event = NULL;
   }
   if (global_timeouts) {
diff --git a/src/common/workqueue.c b/src/common/workqueue.c
index ec96959b7d..12e31414e7 100644
--- a/src/common/workqueue.c
+++ b/src/common/workqueue.c
@@ -1,3 +1,4 @@
+
 /* copyright (c) 2013-2015, The Tor Project, Inc. */
 /* See LICENSE for licensing information */
 
@@ -24,6 +25,7 @@
 
 #include "orconfig.h"
 #include "compat.h"
+#include "compat_libevent.h"
 #include "compat_threads.h"
 #include "crypto.h"
 #include "util.h"
@@ -31,6 +33,8 @@
 #include "tor_queue.h"
 #include "torlog.h"
 
+#include <event2/event.h>
+
 #define WORKQUEUE_PRIORITY_FIRST WQ_PRI_HIGH
 #define WORKQUEUE_PRIORITY_LAST WQ_PRI_LOW
 #define WORKQUEUE_N_PRIORITIES (((int) WORKQUEUE_PRIORITY_LAST)+1)
@@ -63,6 +67,9 @@ struct threadpool_s {
   void (*free_update_arg_fn)(void *);
   /** Array of n_threads update arguments. */
   void **update_args;
+  /** Event to notice when another thread has sent a reply. */
+  struct event *reply_event;
+  void (*reply_cb)(threadpool_t *);
 
   /** Number of elements in threads. */
   int n_threads;
@@ -597,15 +604,41 @@ replyqueue_new(uint32_t alertsocks_flags)
   return rq;
 }
 
-/**
- * Return the "read socket" for a given reply queue.  The main thread should
- * listen for read events on this socket, and call replyqueue_process() every
- * time it triggers.
+/** Internal: Run from the libevent mainloop when there is work to handle in
+ * the reply queue handler. */
+static void
+reply_event_cb(evutil_socket_t sock, short events, void *arg)
+{
+  threadpool_t *tp = arg;
+  (void) sock;
+  (void) events;
+  replyqueue_process(tp->reply_queue);
+  if (tp->reply_cb)
+    tp->reply_cb(tp);
+}
+
+/** Register the threadpool <b>tp</b>'s reply queue with the libevent
+ * mainloop of <b>base</b>. If <b>tp</b> is provided, it is run after
+ * each time there is work to process from the reply queue. Return 0 on
+ * success, -1 on failure.
  */
-tor_socket_t
-replyqueue_get_socket(replyqueue_t *rq)
+int
+threadpool_register_reply_event(threadpool_t *tp,
+                                void (*cb)(threadpool_t *tp))
 {
-  return rq->alert.read_fd;
+  struct event_base *base = tor_libevent_get_base();
+
+  if (tp->reply_event) {
+    tor_event_free(tp->reply_event);
+  }
+  tp->reply_event = tor_event_new(base,
+                                  tp->reply_queue->alert.read_fd,
+                                  EV_READ|EV_PERSIST,
+                                  reply_event_cb,
+                                  tp);
+  tor_assert(tp->reply_event);
+  tp->reply_cb = cb;
+  return event_add(tp->reply_event, NULL);
 }
 
 /**
diff --git a/src/common/workqueue.h b/src/common/workqueue.h
index eb885e680d..e1fe612e2b 100644
--- a/src/common/workqueue.h
+++ b/src/common/workqueue.h
@@ -56,8 +56,11 @@ threadpool_t *threadpool_new(int n_threads,
 replyqueue_t *threadpool_get_replyqueue(threadpool_t *tp);
 
 replyqueue_t *replyqueue_new(uint32_t alertsocks_flags);
-tor_socket_t replyqueue_get_socket(replyqueue_t *rq);
 void replyqueue_process(replyqueue_t *queue);
 
+struct event_base;
+int threadpool_register_reply_event(threadpool_t *tp,
+                                    void (*cb)(threadpool_t *tp));
+
 #endif /* !defined(TOR_WORKQUEUE_H) */