Merge branch 'better_workqueue_v3_squashed'

7 files changed, 775 insertions, 182 deletions

diff --git a/src/test/include.am b/src/test/include.am
index 134c2ff56c..3c59a8b3c7 100644
--- a/src/test/include.am
+++ b/src/test/include.am
@@ -2,7 +2,7 @@ TESTS += src/test/test
 
 noinst_PROGRAMS+= src/test/bench
 if UNITTESTS_ENABLED
-noinst_PROGRAMS+= src/test/test src/test/test-child
+noinst_PROGRAMS+= src/test/test src/test/test-child src/test/test_workqueue
 endif
 
 src_test_AM_CPPFLAGS = -DSHARE_DATADIR="\"$(datadir)\"" \
@@ -47,6 +47,7 @@ src_test_test_SOURCES = \
 	src/test/test_routerkeys.c \
 	src/test/test_scheduler.c \
 	src/test/test_socks.c \
+	src/test/test_threads.c \
 	src/test/test_util.c \
 	src/test/test_config.c \
 	src/test/test_hs.c \
@@ -63,6 +64,11 @@ src_test_test_CPPFLAGS= $(src_test_AM_CPPFLAGS)
 src_test_bench_SOURCES = \
 	src/test/bench.c
 
+src_test_test_workqueue_SOURCES = \
+	src/test/test_workqueue.c
+src_test_test_workqueue_CPPFLAGS= $(src_test_AM_CPPFLAGS)
+src_test_test_workqueue_CFLAGS = $(AM_CFLAGS) $(TEST_CFLAGS)
+
 src_test_test_LDFLAGS = @TOR_LDFLAGS_zlib@ @TOR_LDFLAGS_openssl@ \
         @TOR_LDFLAGS_libevent@
 src_test_test_LDADD = src/or/libtor-testing.a src/common/libor-testing.a \
@@ -81,6 +87,15 @@ src_test_bench_LDADD = src/or/libtor.a src/common/libor.a \
 	@TOR_OPENSSL_LIBS@ @TOR_LIB_WS32@ @TOR_LIB_GDI@ @CURVE25519_LIBS@ \
 	@TOR_SYSTEMD_LIBS@
 
+src_test_test_workqueue_LDFLAGS = @TOR_LDFLAGS_zlib@ @TOR_LDFLAGS_openssl@ \
+        @TOR_LDFLAGS_libevent@
+src_test_test_workqueue_LDADD = src/or/libtor-testing.a \
+	src/common/libor-testing.a \
+	src/common/libor-crypto-testing.a $(LIBDONNA) \
+	src/common/libor-event-testing.a \
+	@TOR_ZLIB_LIBS@ @TOR_LIB_MATH@ @TOR_LIBEVENT_LIBS@ \
+	@TOR_OPENSSL_LIBS@ @TOR_LIB_WS32@ @TOR_LIB_GDI@ @CURVE25519_LIBS@
+
 noinst_HEADERS+= \
 	src/test/fakechans.h \
 	src/test/test.h \
diff --git a/src/test/test.c b/src/test/test.c
index 65d15f12bd..fc5290f0b9 100644
--- a/src/test/test.c
+++ b/src/test/test.c
@@ -1258,6 +1258,23 @@ test_stats(void *arg)
   tor_free(s);
 }
 
+static void *
+passthrough_test_setup(const struct testcase_t *testcase)
+{
+  return testcase->setup_data;
+}
+static int
+passthrough_test_cleanup(const struct testcase_t *testcase, void *ptr)
+{
+  (void)testcase;
+  (void)ptr;
+  return 1;
+}
+
+const struct testcase_setup_t passthrough_setup = {
+  passthrough_test_setup, passthrough_test_cleanup
+};
+
 #define ENT(name)                                                       \
   { #name, test_ ## name , 0, NULL, NULL }
 #define FORK(name)                                                      \
@@ -1297,6 +1314,7 @@ extern struct testcase_t cell_queue_tests[];
 extern struct testcase_t options_tests[];
 extern struct testcase_t socks_tests[];
 extern struct testcase_t entrynodes_tests[];
+extern struct testcase_t thread_tests[];
 extern struct testcase_t extorport_tests[];
 extern struct testcase_t controller_event_tests[];
 extern struct testcase_t logging_tests[];
@@ -1324,6 +1342,7 @@ static struct testgroup_t testgroups[] = {
   { "container/", container_tests },
   { "util/", util_tests },
   { "util/logging/", logging_tests },
+  { "util/thread/", thread_tests },
   { "cellfmt/", cell_format_tests },
   { "cellqueue/", cell_queue_tests },
   { "dir/", dir_tests },
diff --git a/src/test/test.h b/src/test/test.h
index 48037a5ba3..b8057c59bf 100644
--- a/src/test/test.h
+++ b/src/test/test.h
@@ -158,5 +158,7 @@ crypto_pk_t *pk_generate(int idx);
 #define NS_MOCK(name) MOCK(name, NS(name))
 #define NS_UNMOCK(name) UNMOCK(name)
 
+extern const struct testcase_setup_t passthrough_setup;
+
 #endif
 
diff --git a/src/test/test_crypto.c b/src/test/test_crypto.c
index 4a5a12c50a..8426c715a4 100644
--- a/src/test/test_crypto.c
+++ b/src/test/test_crypto.c
@@ -1975,30 +1975,14 @@ test_crypto_siphash(void *arg)
   ;
 }
 
-static void *
-pass_data_setup_fn(const struct testcase_t *testcase)
-{
-  return testcase->setup_data;
-}
-static int
-pass_data_cleanup_fn(const struct testcase_t *testcase, void *ptr)
-{
-  (void)ptr;
-  (void)testcase;
-  return 1;
-}
-static const struct testcase_setup_t pass_data = {
-  pass_data_setup_fn, pass_data_cleanup_fn
-};
-
 #define CRYPTO_LEGACY(name)                                            \
   { #name, test_crypto_ ## name , 0, NULL, NULL }
 
 struct testcase_t crypto_tests[] = {
   CRYPTO_LEGACY(formats),
   CRYPTO_LEGACY(rng),
-  { "aes_AES", test_crypto_aes, TT_FORK, &pass_data, (void*)"aes" },
-  { "aes_EVP", test_crypto_aes, TT_FORK, &pass_data, (void*)"evp" },
+  { "aes_AES", test_crypto_aes, TT_FORK, &passthrough_setup, (void*)"aes" },
+  { "aes_EVP", test_crypto_aes, TT_FORK, &passthrough_setup, (void*)"evp" },
   CRYPTO_LEGACY(sha),
   CRYPTO_LEGACY(pk),
   { "pk_fingerprints", test_crypto_pk_fingerprints, TT_FORK, NULL, NULL },
@@ -2006,23 +1990,25 @@ struct testcase_t crypto_tests[] = {
   CRYPTO_LEGACY(dh),
   CRYPTO_LEGACY(s2k_rfc2440),
 #ifdef HAVE_LIBSCRYPT_H
-  { "s2k_scrypt", test_crypto_s2k_general, 0, &pass_data,
+  { "s2k_scrypt", test_crypto_s2k_general, 0, &passthrough_setup,
     (void*)"scrypt" },
-  { "s2k_scrypt_low", test_crypto_s2k_general, 0, &pass_data,
+  { "s2k_scrypt_low", test_crypto_s2k_general, 0, &passthrough_setup,
     (void*)"scrypt-low" },
 #endif
-  { "s2k_pbkdf2", test_crypto_s2k_general, 0, &pass_data,
+  { "s2k_pbkdf2", test_crypto_s2k_general, 0, &passthrough_setup,
     (void*)"pbkdf2" },
-  { "s2k_rfc2440_general", test_crypto_s2k_general, 0, &pass_data,
+  { "s2k_rfc2440_general", test_crypto_s2k_general, 0, &passthrough_setup,
     (void*)"rfc2440" },
-  { "s2k_rfc2440_legacy", test_crypto_s2k_general, 0, &pass_data,
+  { "s2k_rfc2440_legacy", test_crypto_s2k_general, 0, &passthrough_setup,
     (void*)"rfc2440-legacy" },
   { "s2k_errors", test_crypto_s2k_errors, 0, NULL, NULL },
   { "scrypt_vectors", test_crypto_scrypt_vectors, 0, NULL, NULL },
   { "pbkdf2_vectors", test_crypto_pbkdf2_vectors, 0, NULL, NULL },
   { "pwbox", test_crypto_pwbox, 0, NULL, NULL },
-  { "aes_iv_AES", test_crypto_aes_iv, TT_FORK, &pass_data, (void*)"aes" },
-  { "aes_iv_EVP", test_crypto_aes_iv, TT_FORK, &pass_data, (void*)"evp" },
+  { "aes_iv_AES", test_crypto_aes_iv, TT_FORK, &passthrough_setup,
+    (void*)"aes" },
+  { "aes_iv_EVP", test_crypto_aes_iv, TT_FORK, &passthrough_setup,
+    (void*)"evp" },
   CRYPTO_LEGACY(base32_decode),
   { "kdf_TAP", test_crypto_kdf_TAP, 0, NULL, NULL },
   { "hkdf_sha256", test_crypto_hkdf_sha256, 0, NULL, NULL },
diff --git a/src/test/test_threads.c b/src/test/test_threads.c
new file mode 100644
index 0000000000..2ac08d4d28
--- /dev/null
+++ b/src/test/test_threads.c
@@ -0,0 +1,316 @@
+/* Copyright (c) 2001-2004, Roger Dingledine.
+ * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
+ * Copyright (c) 2007-2015, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+#include "orconfig.h"
+#include "or.h"
+#include "compat_threads.h"
+#include "test.h"
+
+/** mutex for thread test to stop the threads hitting data at the same time. */
+static tor_mutex_t *thread_test_mutex_ = NULL;
+/** mutexes for the thread test to make sure that the threads have to
+ * interleave somewhat. */
+static tor_mutex_t *thread_test_start1_ = NULL,
+                   *thread_test_start2_ = NULL;
+/** Shared strmap for the thread test. */
+static strmap_t *thread_test_strmap_ = NULL;
+/** The name of thread1 for the thread test */
+static char *thread1_name_ = NULL;
+/** The name of thread2 for the thread test */
+static char *thread2_name_ = NULL;
+
+static int thread_fns_failed = 0;
+
+static unsigned long thread_fn_tid1, thread_fn_tid2;
+
+static void thread_test_func_(void* _s) ATTR_NORETURN;
+
+/** How many iterations have the threads in the unit test run? */
+static int t1_count = 0, t2_count = 0;
+
+/** Helper function for threading unit tests: This function runs in a
+ * subthread. It grabs its own mutex (start1 or start2) to make sure that it
+ * should start, then it repeatedly alters _test_thread_strmap protected by
+ * thread_test_mutex_. */
+static void
+thread_test_func_(void* _s)
+{
+  char *s = _s;
+  int i, *count;
+  tor_mutex_t *m;
+  char buf[64];
+  char **cp;
+  if (!strcmp(s, "thread 1")) {
+    m = thread_test_start1_;
+    cp = &thread1_name_;
+    count = &t1_count;
+    thread_fn_tid1 = tor_get_thread_id();
+  } else {
+    m = thread_test_start2_;
+    cp = &thread2_name_;
+    count = &t2_count;
+    thread_fn_tid2 = tor_get_thread_id();
+  }
+
+  tor_snprintf(buf, sizeof(buf), "%lu", tor_get_thread_id());
+  *cp = tor_strdup(buf);
+
+  tor_mutex_acquire(m);
+
+  for (i=0; i<10000; ++i) {
+    tor_mutex_acquire(thread_test_mutex_);
+    strmap_set(thread_test_strmap_, "last to run", *cp);
+    ++*count;
+    tor_mutex_release(thread_test_mutex_);
+  }
+  tor_mutex_acquire(thread_test_mutex_);
+  strmap_set(thread_test_strmap_, s, *cp);
+  if (in_main_thread())
+    ++thread_fns_failed;
+  tor_mutex_release(thread_test_mutex_);
+
+  tor_mutex_release(m);
+
+  spawn_exit();
+}
+
+/** Run unit tests for threading logic. */
+static void
+test_threads_basic(void *arg)
+{
+  char *s1 = NULL, *s2 = NULL;
+  int done = 0, timedout = 0;
+  time_t started;
+#ifndef _WIN32
+  struct timeval tv;
+  tv.tv_sec=0;
+  tv.tv_usec=100*1000;
+#endif
+  (void) arg;
+
+  set_main_thread();
+
+  thread_test_mutex_ = tor_mutex_new();
+  thread_test_start1_ = tor_mutex_new();
+  thread_test_start2_ = tor_mutex_new();
+  thread_test_strmap_ = strmap_new();
+  s1 = tor_strdup("thread 1");
+  s2 = tor_strdup("thread 2");
+  tor_mutex_acquire(thread_test_start1_);
+  tor_mutex_acquire(thread_test_start2_);
+  spawn_func(thread_test_func_, s1);
+  spawn_func(thread_test_func_, s2);
+  tor_mutex_release(thread_test_start2_);
+  tor_mutex_release(thread_test_start1_);
+  started = time(NULL);
+  while (!done) {
+    tor_mutex_acquire(thread_test_mutex_);
+    strmap_assert_ok(thread_test_strmap_);
+    if (strmap_get(thread_test_strmap_, "thread 1") &&
+        strmap_get(thread_test_strmap_, "thread 2")) {
+      done = 1;
+    } else if (time(NULL) > started + 150) {
+      timedout = done = 1;
+    }
+    tor_mutex_release(thread_test_mutex_);
+#ifndef _WIN32
+    /* Prevent the main thread from starving the worker threads. */
+    select(0, NULL, NULL, NULL, &tv);
+#endif
+  }
+  tor_mutex_acquire(thread_test_start1_);
+  tor_mutex_release(thread_test_start1_);
+  tor_mutex_acquire(thread_test_start2_);
+  tor_mutex_release(thread_test_start2_);
+
+  tor_mutex_free(thread_test_mutex_);
+
+  if (timedout) {
+    printf("\nTimed out: %d %d", t1_count, t2_count);
+    tt_assert(strmap_get(thread_test_strmap_, "thread 1"));
+    tt_assert(strmap_get(thread_test_strmap_, "thread 2"));
+    tt_assert(!timedout);
+  }
+
+  /* different thread IDs. */
+  tt_assert(strcmp(strmap_get(thread_test_strmap_, "thread 1"),
+                     strmap_get(thread_test_strmap_, "thread 2")));
+  tt_assert(!strcmp(strmap_get(thread_test_strmap_, "thread 1"),
+                      strmap_get(thread_test_strmap_, "last to run")) ||
+              !strcmp(strmap_get(thread_test_strmap_, "thread 2"),
+                      strmap_get(thread_test_strmap_, "last to run")));
+
+  tt_int_op(thread_fns_failed, ==, 0);
+  tt_int_op(thread_fn_tid1, !=, thread_fn_tid2);
+
+ done:
+  tor_free(s1);
+  tor_free(s2);
+  tor_free(thread1_name_);
+  tor_free(thread2_name_);
+  if (thread_test_strmap_)
+    strmap_free(thread_test_strmap_, NULL);
+  if (thread_test_start1_)
+    tor_mutex_free(thread_test_start1_);
+  if (thread_test_start2_)
+    tor_mutex_free(thread_test_start2_);
+}
+
+typedef struct cv_testinfo_s {
+  tor_cond_t *cond;
+  tor_mutex_t *mutex;
+  int value;
+  int addend;
+  int shutdown;
+  int n_shutdown;
+  int n_wakeups;
+  int n_timeouts;
+  int n_threads;
+  const struct timeval *tv;
+} cv_testinfo_t;
+
+static cv_testinfo_t *
+cv_testinfo_new(void)
+{
+  cv_testinfo_t *i = tor_malloc_zero(sizeof(*i));
+  i->cond = tor_cond_new();
+  i->mutex = tor_mutex_new_nonrecursive();
+  return i;
+}
+
+static void
+cv_testinfo_free(cv_testinfo_t *i)
+{
+  if (!i)
+    return;
+  tor_cond_free(i->cond);
+  tor_mutex_free(i->mutex);
+  tor_free(i);
+}
+
+static void cv_test_thr_fn_(void *arg) ATTR_NORETURN;
+
+static void
+cv_test_thr_fn_(void *arg)
+{
+  cv_testinfo_t *i = arg;
+  int tid, r;
+
+  tor_mutex_acquire(i->mutex);
+  tid = i->n_threads++;
+  tor_mutex_release(i->mutex);
+  (void) tid;
+
+  tor_mutex_acquire(i->mutex);
+  while (1) {
+    if (i->addend) {
+      i->value += i->addend;
+      i->addend = 0;
+    }
+
+    if (i->shutdown) {
+      ++i->n_shutdown;
+      i->shutdown = 0;
+      tor_mutex_release(i->mutex);
+      spawn_exit();
+    }
+    r = tor_cond_wait(i->cond, i->mutex, i->tv);
+    ++i->n_wakeups;
+    if (r == 1) {
+      ++i->n_timeouts;
+      tor_mutex_release(i->mutex);
+      spawn_exit();
+    }
+  }
+}
+
+static void
+test_threads_conditionvar(void *arg)
+{
+  cv_testinfo_t *ti=NULL;
+  const struct timeval msec100 = { 0, 100*1000 };
+  const int timeout = !strcmp(arg, "tv");
+
+  ti = cv_testinfo_new();
+  if (timeout) {
+    ti->tv = &msec100;
+  }
+  spawn_func(cv_test_thr_fn_, ti);
+  spawn_func(cv_test_thr_fn_, ti);
+  spawn_func(cv_test_thr_fn_, ti);
+  spawn_func(cv_test_thr_fn_, ti);
+
+  tor_mutex_acquire(ti->mutex);
+  ti->addend = 7;
+  ti->shutdown = 1;
+  tor_cond_signal_one(ti->cond);
+  tor_mutex_release(ti->mutex);
+
+#define SPIN()                                  \
+  while (1) {                                   \
+    tor_mutex_acquire(ti->mutex);               \
+    if (ti->addend == 0) {                      \
+      break;                                    \
+    }                                           \
+    tor_mutex_release(ti->mutex);               \
+  }
+
+  SPIN();
+
+  ti->addend = 30;
+  ti->shutdown = 1;
+  tor_cond_signal_all(ti->cond);
+  tor_mutex_release(ti->mutex);
+  SPIN();
+
+  ti->addend = 1000;
+  if (! timeout) ti->shutdown = 1;
+  tor_cond_signal_one(ti->cond);
+  tor_mutex_release(ti->mutex);
+  SPIN();
+  ti->addend = 300;
+  if (! timeout) ti->shutdown = 1;
+  tor_cond_signal_all(ti->cond);
+  tor_mutex_release(ti->mutex);
+
+  SPIN();
+  tor_mutex_release(ti->mutex);
+
+  tt_int_op(ti->value, ==, 1337);
+  if (!timeout) {
+    tt_int_op(ti->n_shutdown, ==, 4);
+  } else {
+#ifdef _WIN32
+    Sleep(500); /* msec */
+#elif defined(HAVE_USLEEP)
+    usleep(500*1000); /* usec */
+#else
+    {
+      struct tv = { 0, 500*1000 };
+      select(0, NULL, NULL, NULL, &tv);
+    }
+#endif
+    tor_mutex_acquire(ti->mutex);
+    tt_int_op(ti->n_shutdown, ==, 2);
+    tt_int_op(ti->n_timeouts, ==, 2);
+    tor_mutex_release(ti->mutex);
+  }
+
+ done:
+  cv_testinfo_free(ti);
+}
+
+#define THREAD_TEST(name)                                               \
+  { #name, test_threads_##name, TT_FORK, NULL, NULL }
+
+struct testcase_t thread_tests[] = {
+  THREAD_TEST(basic),
+  { "conditionvar", test_threads_conditionvar, TT_FORK,
+    &passthrough_setup, (void*)"no-tv" },
+  { "conditionvar_timeout", test_threads_conditionvar, TT_FORK,
+    &passthrough_setup, (void*)"tv" },
+  END_OF_TESTCASES
+};
+
diff --git a/src/test/test_util.c b/src/test/test_util.c
index 2ee1d6dfc1..b53c8fc7a3 100644
--- a/src/test/test_util.c
+++ b/src/test/test_util.c
@@ -1607,142 +1607,6 @@ test_util_pow2(void *arg)
   ;
 }
 
-/** mutex for thread test to stop the threads hitting data at the same time. */
-static tor_mutex_t *thread_test_mutex_ = NULL;
-/** mutexes for the thread test to make sure that the threads have to
- * interleave somewhat. */
-static tor_mutex_t *thread_test_start1_ = NULL,
-                   *thread_test_start2_ = NULL;
-/** Shared strmap for the thread test. */
-static strmap_t *thread_test_strmap_ = NULL;
-/** The name of thread1 for the thread test */
-static char *thread1_name_ = NULL;
-/** The name of thread2 for the thread test */
-static char *thread2_name_ = NULL;
-
-static void thread_test_func_(void* _s) ATTR_NORETURN;
-
-/** How many iterations have the threads in the unit test run? */
-static int t1_count = 0, t2_count = 0;
-
-/** Helper function for threading unit tests: This function runs in a
- * subthread. It grabs its own mutex (start1 or start2) to make sure that it
- * should start, then it repeatedly alters _test_thread_strmap protected by
- * thread_test_mutex_. */
-static void
-thread_test_func_(void* _s)
-{
-  char *s = _s;
-  int i, *count;
-  tor_mutex_t *m;
-  char buf[64];
-  char **cp;
-  if (!strcmp(s, "thread 1")) {
-    m = thread_test_start1_;
-    cp = &thread1_name_;
-    count = &t1_count;
-  } else {
-    m = thread_test_start2_;
-    cp = &thread2_name_;
-    count = &t2_count;
-  }
-
-  tor_snprintf(buf, sizeof(buf), "%lu", tor_get_thread_id());
-  *cp = tor_strdup(buf);
-
-  tor_mutex_acquire(m);
-
-  for (i=0; i<10000; ++i) {
-    tor_mutex_acquire(thread_test_mutex_);
-    strmap_set(thread_test_strmap_, "last to run", *cp);
-    ++*count;
-    tor_mutex_release(thread_test_mutex_);
-  }
-  tor_mutex_acquire(thread_test_mutex_);
-  strmap_set(thread_test_strmap_, s, *cp);
-  tor_mutex_release(thread_test_mutex_);
-
-  tor_mutex_release(m);
-
-  spawn_exit();
-}
-
-/** Run unit tests for threading logic. */
-static void
-test_util_threads(void *arg)
-{
-  char *s1 = NULL, *s2 = NULL;
-  int done = 0, timedout = 0;
-  time_t started;
-#ifndef _WIN32
-  struct timeval tv;
-  tv.tv_sec=0;
-  tv.tv_usec=100*1000;
-#endif
-  (void)arg;
-  thread_test_mutex_ = tor_mutex_new();
-  thread_test_start1_ = tor_mutex_new();
-  thread_test_start2_ = tor_mutex_new();
-  thread_test_strmap_ = strmap_new();
-  s1 = tor_strdup("thread 1");
-  s2 = tor_strdup("thread 2");
-  tor_mutex_acquire(thread_test_start1_);
-  tor_mutex_acquire(thread_test_start2_);
-  spawn_func(thread_test_func_, s1);
-  spawn_func(thread_test_func_, s2);
-  tor_mutex_release(thread_test_start2_);
-  tor_mutex_release(thread_test_start1_);
-  started = time(NULL);
-  while (!done) {
-    tor_mutex_acquire(thread_test_mutex_);
-    strmap_assert_ok(thread_test_strmap_);
-    if (strmap_get(thread_test_strmap_, "thread 1") &&
-        strmap_get(thread_test_strmap_, "thread 2")) {
-      done = 1;
-    } else if (time(NULL) > started + 150) {
-      timedout = done = 1;
-    }
-    tor_mutex_release(thread_test_mutex_);
-#ifndef _WIN32
-    /* Prevent the main thread from starving the worker threads. */
-    select(0, NULL, NULL, NULL, &tv);
-#endif
-  }
-  tor_mutex_acquire(thread_test_start1_);
-  tor_mutex_release(thread_test_start1_);
-  tor_mutex_acquire(thread_test_start2_);
-  tor_mutex_release(thread_test_start2_);
-
-  tor_mutex_free(thread_test_mutex_);
-
-  if (timedout) {
-    printf("\nTimed out: %d %d", t1_count, t2_count);
-    tt_assert(strmap_get(thread_test_strmap_, "thread 1"));
-    tt_assert(strmap_get(thread_test_strmap_, "thread 2"));
-    tt_assert(!timedout);
-  }
-
-  /* different thread IDs. */
-  tt_assert(strcmp(strmap_get(thread_test_strmap_, "thread 1"),
-                     strmap_get(thread_test_strmap_, "thread 2")));
-  tt_assert(!strcmp(strmap_get(thread_test_strmap_, "thread 1"),
-                      strmap_get(thread_test_strmap_, "last to run")) ||
-              !strcmp(strmap_get(thread_test_strmap_, "thread 2"),
-                      strmap_get(thread_test_strmap_, "last to run")));
-
- done:
-  tor_free(s1);
-  tor_free(s2);
-  tor_free(thread1_name_);
-  tor_free(thread2_name_);
-  if (thread_test_strmap_)
-    strmap_free(thread_test_strmap_, NULL);
-  if (thread_test_start1_)
-    tor_mutex_free(thread_test_start1_);
-  if (thread_test_start2_)
-    tor_mutex_free(thread_test_start2_);
-}
-
 /** Run unit tests for compression functions */
 static void
 test_util_gzip(void *arg)
@@ -4783,23 +4647,6 @@ test_util_socket(void *arg)
     tor_close_socket(fd4);
 }
 
-static void *
-socketpair_test_setup(const struct testcase_t *testcase)
-{
-  return testcase->setup_data;
-}
-static int
-socketpair_test_cleanup(const struct testcase_t *testcase, void *ptr)
-{
-  (void)testcase;
-  (void)ptr;
-  return 1;
-}
-
-static const struct testcase_setup_t socketpair_setup = {
-  socketpair_test_setup, socketpair_test_cleanup
-};
-
 /* Test for socketpair and ersatz_socketpair().  We test them both, since
  * the latter is a tolerably good way to exersize tor_accept_socket(). */
 static void
@@ -4928,7 +4775,6 @@ struct testcase_t util_tests[] = {
   UTIL_LEGACY(memarea),
   UTIL_LEGACY(control_formats),
   UTIL_LEGACY(mmap),
-  UTIL_LEGACY(threads),
   UTIL_LEGACY(sscanf),
   UTIL_LEGACY(format_time_interval),
   UTIL_LEGACY(path_is_relative),
@@ -4975,10 +4821,10 @@ struct testcase_t util_tests[] = {
   UTIL_TEST(mathlog, 0),
   UTIL_TEST(weak_random, 0),
   UTIL_TEST(socket, TT_FORK),
-  { "socketpair", test_util_socketpair, TT_FORK, &socketpair_setup,
+  { "socketpair", test_util_socketpair, TT_FORK, &passthrough_setup,
     (void*)"0" },
   { "socketpair_ersatz", test_util_socketpair, TT_FORK,
-    &socketpair_setup, (void*)"1" },
+    &passthrough_setup, (void*)"1" },
   UTIL_TEST(max_mem, 0),
   UTIL_TEST(hostname_validation, 0),
   UTIL_TEST(ipv4_validation, 0),
diff --git a/src/test/test_workqueue.c b/src/test/test_workqueue.c
new file mode 100644
index 0000000000..aaff5069be
--- /dev/null
+++ b/src/test/test_workqueue.c
@@ -0,0 +1,409 @@
+/* Copyright (c) 2001-2004, Roger Dingledine.
+ * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
+ * Copyright (c) 2007-2015, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+#include "or.h"
+#include "compat_threads.h"
+#include "onion.h"
+#include "workqueue.h"
+#include "crypto.h"
+#include "crypto_curve25519.h"
+#include "compat_libevent.h"
+
+#include <stdio.h>
+#ifdef HAVE_EVENT2_EVENT_H
+#include <event2/event.h>
+#else
+#include <event.h>
+#endif
+
+static int opt_verbose = 0;
+static int opt_n_threads = 8;
+static int opt_n_items = 10000;
+static int opt_n_inflight = 1000;
+static int opt_n_lowwater = 250;
+static int opt_n_cancel = 0;
+static int opt_ratio_rsa = 5;
+
+#ifdef TRACK_RESPONSES
+tor_mutex_t bitmap_mutex;
+int handled_len;
+bitarray_t *handled;
+#endif
+
+typedef struct state_s {
+  int magic;
+  int n_handled;
+  crypto_pk_t *rsa;
+  curve25519_secret_key_t ecdh;
+  int is_shutdown;
+} state_t;
+
+typedef struct rsa_work_s {
+  int serial;
+  uint8_t msg[128];
+  uint8_t msglen;
+} rsa_work_t;
+
+typedef struct ecdh_work_s {
+  int serial;
+  union {
+    curve25519_public_key_t pk;
+    uint8_t msg[32];
+  } u;
+} ecdh_work_t;
+
+static void
+mark_handled(int serial)
+{
+#ifdef TRACK_RESPONSES
+  tor_mutex_acquire(&bitmap_mutex);
+  tor_assert(serial < handled_len);
+  tor_assert(! bitarray_is_set(handled, serial));
+  bitarray_set(handled, serial);
+  tor_mutex_release(&bitmap_mutex);
+#else
+  (void)serial;
+#endif
+}
+
+static int
+workqueue_do_rsa(void *state, void *work)
+{
+  rsa_work_t *rw = work;
+  state_t *st = state;
+  crypto_pk_t *rsa = st->rsa;
+  uint8_t sig[256];
+  int len;
+
+  tor_assert(st->magic == 13371337);
+
+  len = crypto_pk_private_sign(rsa, (char*)sig, 256,
+                               (char*)rw->msg, rw->msglen);
+  if (len < 0) {
+    rw->msglen = 0;
+    return WQ_RPL_ERROR;
+  }
+
+  memset(rw->msg, 0, sizeof(rw->msg));
+  rw->msglen = len;
+  memcpy(rw->msg, sig, len);
+  ++st->n_handled;
+
+  mark_handled(rw->serial);
+
+  return WQ_RPL_REPLY;
+}
+
+static int
+workqueue_do_shutdown(void *state, void *work)
+{
+  (void)state;
+  (void)work;
+  crypto_pk_free(((state_t*)state)->rsa);
+  tor_free(state);
+  return WQ_RPL_SHUTDOWN;
+}
+
+static int
+workqueue_do_ecdh(void *state, void *work)
+{
+  ecdh_work_t *ew = work;
+  uint8_t output[CURVE25519_OUTPUT_LEN];
+  state_t *st = state;
+
+  tor_assert(st->magic == 13371337);
+
+  curve25519_handshake(output, &st->ecdh, &ew->u.pk);
+  memcpy(ew->u.msg, output, CURVE25519_OUTPUT_LEN);
+  ++st->n_handled;
+  mark_handled(ew->serial);
+  return WQ_RPL_REPLY;
+}
+
+static void *
+new_state(void *arg)
+{
+  state_t *st;
+  (void)arg;
+
+  st = tor_malloc(sizeof(*st));
+  /* Every thread gets its own keys. not a problem for benchmarking */
+  st->rsa = crypto_pk_new();
+  if (crypto_pk_generate_key_with_bits(st->rsa, 1024) < 0) {
+    crypto_pk_free(st->rsa);
+    tor_free(st);
+    return NULL;
+  }
+  curve25519_secret_key_generate(&st->ecdh, 0);
+  st->magic = 13371337;
+  return st;
+}
+
+static void
+free_state(void *arg)
+{
+  state_t *st = arg;
+  crypto_pk_free(st->rsa);
+  tor_free(st);
+}
+
+static tor_weak_rng_t weak_rng;
+static int n_sent = 0;
+static int rsa_sent = 0;
+static int ecdh_sent = 0;
+static int n_received = 0;
+
+#ifdef TRACK_RESPONSES
+bitarray_t *received;
+#endif
+
+static void
+handle_reply(void *arg)
+{
+#ifdef TRACK_RESPONSES
+  rsa_work_t *rw = arg; /* Naughty cast, but only looking at serial. */
+  tor_assert(! bitarray_is_set(received, rw->serial));
+  bitarray_set(received,rw->serial);
+#endif
+
+  tor_free(arg);
+  ++n_received;
+}
+
+static workqueue_entry_t *
+add_work(threadpool_t *tp)
+{
+  int add_rsa =
+    opt_ratio_rsa == 0 ||
+    tor_weak_random_range(&weak_rng, opt_ratio_rsa) == 0;
+
+  if (add_rsa) {
+    rsa_work_t *w = tor_malloc_zero(sizeof(*w));
+    w->serial = n_sent++;
+    crypto_rand((char*)w->msg, 20);
+    w->msglen = 20;
+    ++rsa_sent;
+    return threadpool_queue_work(tp, workqueue_do_rsa, handle_reply, w);
+  } else {
+    ecdh_work_t *w = tor_malloc_zero(sizeof(*w));
+    w->serial = n_sent++;
+    /* Not strictly right, but this is just for benchmarks. */
+    crypto_rand((char*)w->u.pk.public_key, 32);
+    ++ecdh_sent;
+    return threadpool_queue_work(tp, workqueue_do_ecdh, handle_reply, w);
+  }
+}
+
+static int n_failed_cancel = 0;
+static int n_successful_cancel = 0;
+
+static int
+add_n_work_items(threadpool_t *tp, int n)
+{
+  int n_queued = 0;
+  int n_try_cancel = 0, i;
+  workqueue_entry_t **to_cancel;
+  workqueue_entry_t *ent;
+
+  to_cancel = tor_malloc(sizeof(workqueue_entry_t*) * opt_n_cancel);
+
+  while (n_queued++ < n) {
+    ent = add_work(tp);
+    if (! ent) {
+      tor_event_base_loopexit(tor_libevent_get_base(), NULL);
+      return -1;
+    }
+    if (n_try_cancel < opt_n_cancel &&
+        tor_weak_random_range(&weak_rng, n) < opt_n_cancel) {
+      to_cancel[n_try_cancel++] = ent;
+    }
+  }
+
+  for (i = 0; i < n_try_cancel; ++i) {
+    void *work = workqueue_entry_cancel(to_cancel[i]);
+    if (! work) {
+      n_failed_cancel++;
+    } else {
+      n_successful_cancel++;
+      tor_free(work);
+    }
+  }
+
+  tor_free(to_cancel);
+  return 0;
+}
+
+static int shutting_down = 0;
+
+static void
+replysock_readable_cb(tor_socket_t sock, short what, void *arg)
+{
+  threadpool_t *tp = arg;
+  replyqueue_t *rq = threadpool_get_replyqueue(tp);
+
+  int old_r = n_received;
+  (void) sock;
+  (void) what;
+
+  replyqueue_process(rq);
+  if (old_r == n_received)
+    return;
+
+  if (opt_verbose) {
+    printf("%d / %d", n_received, n_sent);
+    if (opt_n_cancel)
+      printf(" (%d cancelled, %d uncancellable)",
+             n_successful_cancel, n_failed_cancel);
+    puts("");
+  }
+#ifdef TRACK_RESPONSES
+  tor_mutex_acquire(&bitmap_mutex);
+  for (i = 0; i < opt_n_items; ++i) {
+    if (bitarray_is_set(received, i))
+      putc('o', stdout);
+    else if (bitarray_is_set(handled, i))
+      putc('!', stdout);
+    else
+      putc('.', stdout);
+  }
+  puts("");
+  tor_mutex_release(&bitmap_mutex);
+#endif
+
+  if (n_sent - (n_received+n_successful_cancel) < opt_n_lowwater) {
+    int n_to_send = n_received + opt_n_inflight - n_sent;
+    if (n_to_send > opt_n_items - n_sent)
+      n_to_send = opt_n_items - n_sent;
+    add_n_work_items(tp, n_to_send);
+  }
+
+  if (shutting_down == 0 &&
+      n_received+n_successful_cancel == n_sent &&
+      n_sent >= opt_n_items) {
+    shutting_down = 1;
+    threadpool_queue_update(tp, NULL,
+                             workqueue_do_shutdown, NULL, NULL);
+  }
+}
+
+static void
+help(void)
+{
+  puts(
+     "Options:\n"
+     "    -N <items>    Run this many items of work\n"
+     "    -T <threads>  Use this many threads\n"
+     "    -I <inflight> Have no more than this many requests queued at once\n"
+     "    -L <lowwater> Add items whenever fewer than this many are pending\n"
+     "    -C <cancel>   Try to cancel N items of every batch that we add\n"
+     "    -R <ratio>    Make one out of this many items be a slow (RSA) one\n"
+     "    --no-{eventfd2,eventfd,pipe2,pipe,socketpair}\n"
+     "                  Disable one of the alert_socket backends.");
+}
+
+int
+main(int argc, char **argv)
+{
+  replyqueue_t *rq;
+  threadpool_t *tp;
+  int i;
+  tor_libevent_cfg evcfg;
+  struct event *ev;
+  uint32_t as_flags = 0;
+
+  for (i = 1; i < argc; ++i) {
+    if (!strcmp(argv[i], "-v")) {
+      opt_verbose = 1;
+    } else if (!strcmp(argv[i], "-T") && i+1<argc) {
+      opt_n_threads = atoi(argv[++i]);
+    } else if (!strcmp(argv[i], "-N") && i+1<argc) {
+      opt_n_items = atoi(argv[++i]);
+    } else if (!strcmp(argv[i], "-I") && i+1<argc) {
+      opt_n_inflight = atoi(argv[++i]);
+    } else if (!strcmp(argv[i], "-L") && i+1<argc) {
+      opt_n_lowwater = atoi(argv[++i]);
+    } else if (!strcmp(argv[i], "-R") && i+1<argc) {
+      opt_ratio_rsa = atoi(argv[++i]);
+    } else if (!strcmp(argv[i], "-C") && i+1<argc) {
+      opt_n_cancel = atoi(argv[++i]);
+    } else if (!strcmp(argv[i], "--no-eventfd2")) {
+      as_flags |= ASOCKS_NOEVENTFD2;
+    } else if (!strcmp(argv[i], "--no-eventfd")) {
+      as_flags |= ASOCKS_NOEVENTFD;
+    } else if (!strcmp(argv[i], "--no-pipe2")) {
+      as_flags |= ASOCKS_NOPIPE2;
+    } else if (!strcmp(argv[i], "--no-pipe")) {
+      as_flags |= ASOCKS_NOPIPE;
+    } else if (!strcmp(argv[i], "--no-socketpair")) {
+      as_flags |= ASOCKS_NOSOCKETPAIR;
+    } else if (!strcmp(argv[i], "-h")) {
+      help();
+      return 0;
+    } else {
+      help();
+      return 1;
+    }
+  }
+  if (opt_n_threads < 1 ||
+      opt_n_items < 1 || opt_n_inflight < 1 || opt_n_lowwater < 0 ||
+      opt_n_cancel > opt_n_inflight ||
+      opt_ratio_rsa < 0) {
+    help();
+    return 1;
+  }
+
+  init_logging(1);
+  crypto_global_init(1, NULL, NULL);
+  crypto_seed_rng(1);
+
+  rq = replyqueue_new(as_flags);
+  tor_assert(rq);
+  tp = threadpool_new(opt_n_threads,
+                      rq, new_state, free_state, NULL);
+  tor_assert(tp);
+
+  crypto_seed_weak_rng(&weak_rng);
+
+  memset(&evcfg, 0, sizeof(evcfg));
+  tor_libevent_initialize(&evcfg);
+
+  ev = tor_event_new(tor_libevent_get_base(),
+                     replyqueue_get_socket(rq), EV_READ|EV_PERSIST,
+                     replysock_readable_cb, tp);
+
+  event_add(ev, NULL);
+
+#ifdef TRACK_RESPONSES
+  handled = bitarray_init_zero(opt_n_items);
+  received = bitarray_init_zero(opt_n_items);
+  tor_mutex_init(&bitmap_mutex);
+  handled_len = opt_n_items;
+#endif
+
+  for (i = 0; i < opt_n_inflight; ++i) {
+    if (! add_work(tp)) {
+      puts("Couldn't add work.");
+      return 1;
+    }
+  }
+
+  {
+    struct timeval limit = { 30, 0 };
+    tor_event_base_loopexit(tor_libevent_get_base(), &limit);
+  }
+
+  event_base_loop(tor_libevent_get_base(), 0);
+
+  if (n_sent != opt_n_items || n_received+n_successful_cancel != n_sent) {
+    printf("%d vs %d\n", n_sent, opt_n_items);
+    printf("%d+%d vs %d\n", n_received, n_successful_cancel, n_sent);
+    puts("FAIL");
+    return 1;
+  } else {
+    puts("OK");
+    return 0;
+  }
+}
+