15 files changed, 382 insertions, 330 deletions

diff --git a/Makefile b/Makefile
index 8f4dcf6..61eb872 100644
--- a/Makefile
+++ b/Makefile
@@ -69,8 +69,8 @@ CFLAGS_libowfat=-I$(LIBOWFAT_HEADERS)
 LDFLAGS_libowfat=-L$(LIBOWFAT_LIBRARY) -lowfat
 
 BINARY =opentracker
-HEADERS=trackerlogic.h ot_urlencoded.h ot_batch.h ot_buf.h ot_event.h ot_fdstate.h ot_fmt.h ot_hex.h ot_ip.h ot_config.h ot_scan.h ot_socket.h ot_tcp.h ot_mutex.h ot_random.h ot_stats.h ot_vector.h ot_clean.h ot_udp.h ot_iovec.h ot_fullscrape.h ot_accesslist.h ot_http.h ot_livesync.h ot_rijndael.h
-SOURCES=opentracker.c trackerlogic.c ot_urlencoded.c ot_batch.c ot_buf.c ot_event.c ot_fdstate.c ot_fmt.c ot_hex.c ot_ip.c ot_config.c ot_scan.c ot_socket.c ot_tcp.c ot_mutex.c ot_random.c ot_stats.c ot_vector.c ot_clean.c ot_udp.c ot_iovec.c ot_fullscrape.c ot_accesslist.c ot_http.c ot_livesync.c ot_rijndael.c
+HEADERS=trackerlogic.h ot_urlencoded.h ot_batch.h ot_buf.h ot_event.h ot_fdstate.h ot_fmt.h ot_hex.h ot_ip.h ot_config.h ot_scan.h ot_socket.h ot_tcp.h ot_workqueue.h ot_mutex.h ot_random.h ot_stats.h ot_vector.h ot_clean.h ot_udp.h ot_iovec.h ot_fullscrape.h ot_accesslist.h ot_http.h ot_livesync.h ot_rijndael.h
+SOURCES=opentracker.c trackerlogic.c ot_urlencoded.c ot_batch.c ot_buf.c ot_event.c ot_fdstate.c ot_fmt.c ot_hex.c ot_ip.c ot_config.c ot_scan.c ot_socket.c ot_tcp.c ot_workqueue.c ot_mutex.c ot_random.c ot_stats.c ot_vector.c ot_clean.c ot_udp.c ot_iovec.c ot_fullscrape.c ot_accesslist.c ot_http.c ot_livesync.c ot_rijndael.c
 SOURCES_proxy=proxy.c ot_batch.c ot_fdstate.c ot_vector.c ot_mutex.c ot_random.c ot_scan.c ot_ip.c ot_socket.c
 
 all: $(BINARY) $(BINARY).debug
@@ -96,8 +96,8 @@ proxy.debug: $(SOURCES_proxy) $(HEADERS)
 # has to. tests/stubs.c supplies the few symbols those modules reference but
 # the tests never reach.
 TESTBIN=tests/runner
-TEST_SOURCES=tests/runner.c tests/test.c tests/stubs.c tests/t_urlencoded.c tests/t_iovec.c tests/t_vector.c tests/t_random.c tests/t_hex.c tests/t_accesslist.c tests/t_scan.c tests/t_buf.c tests/t_ip.c tests/t_socket.c tests/t_mutex.c tests/t_batch.c tests/t_fdstate.c tests/t_event.c tests/t_fmt.c
-TEST_UNITS=ot_urlencoded.c ot_fmt.c ot_event.c ot_batch.c ot_fdstate.c ot_buf.c ot_ip.c ot_socket.c ot_iovec.c ot_vector.c ot_random.c ot_hex.c ot_scan.c ot_accesslist.c ot_mutex.c
+TEST_SOURCES=tests/runner.c tests/test.c tests/stubs.c tests/t_urlencoded.c tests/t_iovec.c tests/t_vector.c tests/t_random.c tests/t_hex.c tests/t_accesslist.c tests/t_scan.c tests/t_buf.c tests/t_ip.c tests/t_socket.c tests/t_workqueue.c tests/t_batch.c tests/t_fdstate.c tests/t_event.c tests/t_fmt.c
+TEST_UNITS=ot_urlencoded.c ot_fmt.c ot_event.c ot_batch.c ot_fdstate.c ot_buf.c ot_ip.c ot_socket.c ot_iovec.c ot_vector.c ot_random.c ot_hex.c ot_scan.c ot_accesslist.c ot_mutex.c ot_workqueue.c
 TEST_HEADERS=tests/test.h $(HEADERS)
 # Tests reach code that sits behind feature flags, so the units under test are
 # built with those features enabled. A test binary is not a production binary.
diff --git a/opentracker.c b/opentracker.c
index 6e073e9..2f2746d 100644
--- a/opentracker.c
+++ b/opentracker.c
@@ -29,6 +29,7 @@
 #include "ot_ip.h"
 #include "ot_livesync.h"
 #include "ot_mutex.h"
+#include "ot_workqueue.h"
 #include "ot_random.h"
 #include "ot_scan.h"
 #include "ot_event.h"
@@ -148,7 +149,7 @@ static void *server_mainloop(void *args) {
       }
     }
 
-    while ((sock = mutex_workqueue_popresult(&iovec_entries, &iovector, &is_partial)) != -1)
+    while ((sock = workqueue_popresult(&iovec_entries, &iovector, &is_partial)) != -1)
       http_sendiovecdata(sock, &ws, iovec_entries, iovector, is_partial);
 
     if (time_now_seconds() > next_timeout_check) {
diff --git a/ot_fullscrape.c b/ot_fullscrape.c
index a86d6d7..ae29308 100644
--- a/ot_fullscrape.c
+++ b/ot_fullscrape.c
@@ -23,6 +23,7 @@
 #include "ot_fullscrape.h"
 #include "ot_iovec.h"
 #include "ot_mutex.h"
+#include "ot_workqueue.h"
 #include "ot_urlencoded.h"
 #include "trackerlogic.h"
 
@@ -54,7 +55,7 @@ static void *fullscrape_worker(void *args) {
 
   while (g_opentracker_running) {
     ot_tasktype tasktype = TASK_FULLSCRAPE;
-    ot_taskid   taskid   = mutex_workqueue_poptask(&tasktype);
+    ot_taskid   taskid   = workqueue_poptask(&tasktype);
     if (!taskid)
       break;
 #ifdef WANT_COMPRESSION_ZSTD
@@ -68,7 +69,7 @@ static void *fullscrape_worker(void *args) {
     else
 #endif
       fullscrape_make(taskid, tasktype);
-    mutex_workqueue_pushchunked(taskid, NULL);
+    workqueue_pushchunked(taskid, NULL);
   }
   return NULL;
 }
@@ -79,12 +80,12 @@ void fullscrape_init(void) {
 }
 
 void fullscrape_deinit(void) {
-  mutex_workqueue_wakeall();
+  workqueue_wakeall();
   pthread_join( thread_id, NULL );
 }
 
 void fullscrape_deliver( int64_t sock, ot_tasktype tasktype ) {
-  mutex_workqueue_pushtask( sock, TASK_FULLSCRAPE | tasktype );
+  workqueue_pushtask( sock, TASK_FULLSCRAPE | tasktype );
 }
 
 static void fullscrape_write_one( ot_tasktype mode, char **cursor, ot_torrent *torrent, ot_hash *hash ) {
@@ -170,7 +171,7 @@ static void fullscrape_make(int taskid, ot_tasktype mode) {
       if (r > re) {
         iovector.iov_len = r - (char *)iovector.iov_base;
 
-        if (mutex_workqueue_pushchunked(taskid, &iovector)) {
+        if (workqueue_pushchunked(taskid, &iovector)) {
           free(iovector.iov_base);
           mutex_bucket_unlock(bucket, 0);
           return;
@@ -200,7 +201,7 @@ static void fullscrape_make(int taskid, ot_tasktype mode) {
 
   /* Send rest of data */
   iovector.iov_len = r - (char *)iovector.iov_base;
-  if (mutex_workqueue_pushchunked(taskid, &iovector))
+  if (workqueue_pushchunked(taskid, &iovector))
     free(iovector.iov_base);
 }
 
@@ -251,7 +252,7 @@ static void fullscrape_make_gzip(int taskid, ot_tasktype mode) {
       while (!strm.avail_out) {
         iovector.iov_len = (char *)strm.next_out - (char *)iovector.iov_base;
 
-        if (mutex_workqueue_pushchunked(taskid, &iovector)) {
+        if (workqueue_pushchunked(taskid, &iovector)) {
           free(iovector.iov_base);
           mutex_bucket_unlock(bucket, 0);
           return;
@@ -291,7 +292,7 @@ static void fullscrape_make_gzip(int taskid, ot_tasktype mode) {
     fprintf(stderr, "deflate() failed while in fullscrape_make()'s endgame.\n");
 
   iovector.iov_len = (char *)strm.next_out - (char *)iovector.iov_base;
-  if (mutex_workqueue_pushchunked(taskid, &iovector)) {
+  if (workqueue_pushchunked(taskid, &iovector)) {
     free(iovector.iov_base);
     deflateEnd(&strm);
     return;
@@ -314,7 +315,7 @@ static void fullscrape_make_gzip(int taskid, ot_tasktype mode) {
 
       /* Only pass the new buffer if there actually was some data left in the buffer */
       iovector.iov_len = (char *)strm.next_out - (char *)iovector.iov_base;
-      if (!iovector.iov_len || mutex_workqueue_pushchunked(taskid, &iovector))
+      if (!iovector.iov_len || workqueue_pushchunked(taskid, &iovector))
         free(iovector.iov_base);
   }
 
@@ -380,7 +381,7 @@ static void fullscrape_make_zstd(int taskid, ot_tasktype mode) {
       while (outbuf.pos + OT_SCRAPE_MAXENTRYLEN > outbuf.size) {
         iovector.iov_len = outbuf.size;
 
-        if (mutex_workqueue_pushchunked(taskid, &iovector)) {
+        if (workqueue_pushchunked(taskid, &iovector)) {
           free(iovector.iov_base);
           ZSTD_freeCCtx(zstream);
           return mutex_bucket_unlock(bucket, 0);
@@ -418,7 +419,7 @@ static void fullscrape_make_zstd(int taskid, ot_tasktype mode) {
   more_bytes = ZSTD_compressStream2(zstream, &outbuf, &inbuf, ZSTD_e_end);
 
   iovector.iov_len = outbuf.pos;
-  if (mutex_workqueue_pushchunked(taskid, &iovector)) {
+  if (workqueue_pushchunked(taskid, &iovector)) {
     free(iovector.iov_base);
     ZSTD_freeCCtx(zstream);
     return;
@@ -443,7 +444,7 @@ static void fullscrape_make_zstd(int taskid, ot_tasktype mode) {
 
       /* Only pass the new buffer if there actually was some data left in the buffer */
       iovector.iov_len = outbuf.pos;
-      if (!iovector.iov_len || mutex_workqueue_pushchunked(taskid, &iovector))
+      if (!iovector.iov_len || workqueue_pushchunked(taskid, &iovector))
         free(iovector.iov_base);
   }
 
diff --git a/ot_fullscrape.h b/ot_fullscrape.h
index 0e7690c..b759a27 100644
--- a/ot_fullscrape.h
+++ b/ot_fullscrape.h
@@ -9,6 +9,7 @@
 #ifdef WANT_FULLSCRAPE
 
 #include "ot_mutex.h"
+#include "ot_workqueue.h"
 
 void fullscrape_init(void);
 void fullscrape_deinit(void);
diff --git a/ot_http.c b/ot_http.c
index 602ff94..b0e1de3 100644
--- a/ot_http.c
+++ b/ot_http.c
@@ -22,6 +22,7 @@
 #include "ot_iovec.h"
 #include "ot_ip.h"
 #include "ot_mutex.h"
+#include "ot_workqueue.h"
 #include "ot_scan.h"
 #include "ot_event.h"
 #include "ot_fdstate.h"
@@ -77,7 +78,7 @@ void http_close(const int64_t sock) {
     free(cookie->batch);
     buf_reset(&cookie->request);
     if (cookie->flag & (STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER))
-      mutex_workqueue_canceltask(sock);
+      workqueue_canceltask(sock);
     if (!(cookie->flag & STRUCT_HTTP_FLAG_ONSTACK))
       free(cookie);
   }
diff --git a/ot_mutex.c b/ot_mutex.c
index 5392264..d856d5f 100644
--- a/ot_mutex.c
+++ b/ot_mutex.c
@@ -31,7 +31,6 @@ static pthread_mutex_t bucket_mutex[OT_BUCKET_COUNT];
 static _Atomic size_t  g_torrent_count;
 
 /* Self pipe from opentracker.c */
-extern int             g_self_pipe[2];
 
 ot_vector *mutex_bucket_lock(int bucket) {
   pthread_mutex_lock(bucket_mutex + bucket);
@@ -60,237 +59,17 @@ void mutex_bucket_unlock_by_hash(ot_hash const hash, int delta_torrentcount) {
 
 size_t mutex_get_torrent_count(void) { return atomic_load_explicit(&g_torrent_count, memory_order_relaxed); }
 
-/* TaskQueue Magic */
-
-struct ot_task {
-  ot_taskid       taskid;
-  ot_tasktype     tasktype;
-  int64_t         sock;
-  size_t          iovec_entries;
-  struct iovec   *iovec;
-  struct ot_task *next;
-};
-
-static ot_taskid       next_free_taskid = 1;
-static struct ot_task *tasklist;
-static pthread_mutex_t tasklist_mutex;
-static pthread_cond_t  tasklist_being_filled;
-
-int mutex_workqueue_pushtask(int64_t sock, ot_tasktype tasktype) {
-  struct ot_task **tmptask, *task;
-
-  task = malloc(sizeof(struct ot_task));
-  if (!task)
-    return -1;
-
-  task->taskid        = 0;
-  task->tasktype      = tasktype;
-  task->sock          = sock;
-  task->iovec_entries = 0;
-  task->iovec         = NULL;
-  task->next          = 0;
-
-  /* Want exclusive access to tasklist */
-  pthread_mutex_lock(&tasklist_mutex);
-
-  /* Skip to end of list */
-  tmptask = &tasklist;
-  while (*tmptask)
-    tmptask = &(*tmptask)->next;
-  *tmptask = task;
-
-  /* Inform waiting workers and release lock */
-  pthread_cond_broadcast(&tasklist_being_filled);
-  pthread_mutex_unlock(&tasklist_mutex);
-  return 0;
-}
-
-void mutex_workqueue_canceltask(int64_t sock) {
-  struct ot_task **task;
-
-  /* Want exclusive access to tasklist */
-  pthread_mutex_lock(&tasklist_mutex);
-
-  for (task = &tasklist; *task; task = &((*task)->next))
-    if ((*task)->sock == sock) {
-      struct iovec   *iovec = (*task)->iovec;
-      struct ot_task *ptask = *task;
-      size_t                   i;
-
-      /* Free task's iovec, the buffers it points at and the vector itself */
-      for (i = 0; i < (*task)->iovec_entries; ++i)
-        free(iovec[i].iov_base);
-      free(iovec);
-
-      *task = (*task)->next;
-      free(ptask);
-      break;
-    }
-
-  /* Release lock */
-  pthread_mutex_unlock(&tasklist_mutex);
-}
-
-ot_taskid mutex_workqueue_poptask(ot_tasktype *tasktype) {
-  struct ot_task *task;
-  ot_taskid       taskid = 0;
-
-  /* Want exclusive access to tasklist */
-  pthread_mutex_lock(&tasklist_mutex);
-
-  while (!taskid) {
-    /* Skip to the first unassigned task this worker wants to do */
-    for (task = tasklist; task; task = task->next)
-      if (!task->taskid && (TASK_CLASS_MASK & task->tasktype) == *tasktype) {
-        /* If we found an outstanding task, assign a taskid to it
-           and leave the loop */
-        task->taskid = taskid = ++next_free_taskid;
-        *tasktype             = task->tasktype;
-        break;
-      }
-
-    /* Wait until the next task is being fed, unless the tracker is on its
-       way out, in which case the worker is meant to leave rather than wait */
-    if (!taskid) {
-      if (!g_opentracker_running)
-        break;
-      pthread_cond_wait(&tasklist_being_filled, &tasklist_mutex);
-    }
-  }
-
-  /* Release lock */
-  pthread_mutex_unlock(&tasklist_mutex);
-
-  return taskid;
-}
-
-void mutex_workqueue_pushsuccess(ot_taskid taskid) {
-  struct ot_task **task;
-
-  /* Want exclusive access to tasklist */
-  pthread_mutex_lock(&tasklist_mutex);
-
-  for (task = &tasklist; *task; task = &((*task)->next))
-    if ((*task)->taskid == taskid) {
-      struct ot_task *ptask = *task;
-      *task = (*task)->next;
-      free(ptask);
-      break;
-    }
-
-  /* Release lock */
-  pthread_mutex_unlock(&tasklist_mutex);
-}
-
-int mutex_workqueue_pushresult(ot_taskid taskid, size_t iovec_entries, struct iovec *iovec) {
-  struct ot_task *task;
-  const char      byte = 'o';
-
-  /* Want exclusive access to tasklist */
-  pthread_mutex_lock(&tasklist_mutex);
-
-  for (task = tasklist; task; task = task->next)
-    if (task->taskid == taskid) {
-      task->iovec_entries = iovec_entries;
-      task->iovec         = iovec;
-      task->tasktype      = TASK_DONE;
-      break;
-    }
-
-  /* Release lock */
-  pthread_mutex_unlock(&tasklist_mutex);
-
-  /* A full pipe has already done this nudge's job, so a failure is fine */
-  (void)!write(g_self_pipe[1], &byte, 1);
-
-  /* Indicate whether the worker has to throw away results */
-  return task ? 0 : -1;
-}
-
-int mutex_workqueue_pushchunked(ot_taskid taskid, struct iovec *iovec) {
-  struct ot_task *task;
-  const char      byte = 'o';
-
-  /* Want exclusive access to tasklist */
-  pthread_mutex_lock(&tasklist_mutex);
-
-  for (task = tasklist; task; task = task->next)
-    if (task->taskid == taskid) {
-      if (iovec) {
-        if (iovec_append(&task->iovec_entries, &task->iovec, iovec))
-          task->tasktype = TASK_DONE_PARTIAL;
-        else
-          task = NULL;
-      } else
-        task->tasktype = TASK_DONE;
-      break;
-    }
-
-  /* Release lock */
-  pthread_mutex_unlock(&tasklist_mutex);
-
-  /* A full pipe has already done this nudge's job, so a failure is fine */
-  (void)!write(g_self_pipe[1], &byte, 1);
-
-  /* Indicate whether the worker has to throw away results */
-  return task ? 0 : -1;
-}
-
-int64_t mutex_workqueue_popresult(size_t *iovec_entries, struct iovec **iovec, int *is_partial) {
-  struct ot_task **task;
-  int64_t sock = -1;
-
-  *is_partial = 0;
-
-  /* Want exclusive access to tasklist */
-  pthread_mutex_lock(&tasklist_mutex);
-
-  for (task = &tasklist; *task; task = &((*task)->next))
-    if (((*task)->tasktype & TASK_CLASS_MASK) == TASK_DONE) {
-      struct ot_task *ptask = *task;
-      *iovec_entries        = ptask->iovec_entries;
-      *iovec                = ptask->iovec;
-      sock                  = ptask->sock;
-
-      if ((*task)->tasktype == TASK_DONE) {
-        *task = ptask->next;
-        free(ptask);
-      } else {
-        ptask->iovec_entries = 0;
-        ptask->iovec         = NULL;
-        *is_partial          = 1;
-        /* Prevent task from showing up immediately again unless new data was added */
-        (*task)->tasktype    = TASK_FULLSCRAPE;
-      }
-      break;
-    }
-
-  /* Release lock */
-  pthread_mutex_unlock(&tasklist_mutex);
-  return sock;
-}
-
 void mutex_init(void) {
   size_t i;
-  pthread_mutex_init(&tasklist_mutex, NULL);
-  pthread_cond_init(&tasklist_being_filled, NULL);
   for (i = 0; i < OT_BUCKET_COUNT; ++i)
     pthread_mutex_init(bucket_mutex + i, NULL);
   memset(all_torrents, 0, sizeof(all_torrents));
 }
 
 /* Let every worker parked on the queue see that the tracker is stopping */
-void mutex_workqueue_wakeall(void) {
-  pthread_mutex_lock(&tasklist_mutex);
-  pthread_cond_broadcast(&tasklist_being_filled);
-  pthread_mutex_unlock(&tasklist_mutex);
-}
-
 void mutex_deinit(void) {
   size_t i;
   for (i = 0; i < OT_BUCKET_COUNT; ++i)
     pthread_mutex_destroy(bucket_mutex + i);
-  pthread_mutex_destroy(&tasklist_mutex);
-  pthread_cond_destroy(&tasklist_being_filled);
   memset(all_torrents, 0, sizeof(all_torrents));
 }
diff --git a/ot_mutex.h b/ot_mutex.h
index c048fdb..c1218d7 100644
--- a/ot_mutex.h
+++ b/ot_mutex.h
@@ -10,7 +10,6 @@
 #include "trackerlogic.h"
 
 void       mutex_init(void);
-void       mutex_workqueue_wakeall(void);
 void       mutex_deinit(void);
 
 ot_vector *mutex_bucket_lock(int bucket);
@@ -21,60 +20,4 @@ void       mutex_bucket_unlock_by_hash(ot_hash const hash, int delta_torrentcoun
 
 size_t     mutex_get_torrent_count(void);
 
-typedef enum {
-  TASK_STATS_CONNS               = 0x0001,
-  TASK_STATS_TCP                 = 0x0002,
-  TASK_STATS_UDP                 = 0x0003,
-  TASK_STATS_SCRAPE              = 0x0004,
-  TASK_STATS_FULLSCRAPE          = 0x0005,
-  TASK_STATS_TPB                 = 0x0006,
-  TASK_STATS_HTTPERRORS          = 0x0007,
-  TASK_STATS_VERSION             = 0x0008,
-  TASK_STATS_RENEW               = 0x000a,
-  TASK_STATS_SYNCS               = 0x000b,
-  TASK_STATS_COMPLETED           = 0x000c,
-  TASK_STATS_NUMWANTS            = 0x000d,
-
-  TASK_STATS                     = 0x0100, /* Mask */
-  TASK_STATS_TORRENTS            = 0x0101,
-  TASK_STATS_PEERS               = 0x0102,
-  TASK_STATS_SLASH24S            = 0x0103,
-  TASK_STATS_TOP10               = 0x0104,
-  TASK_STATS_TOP100              = 0x0105,
-  TASK_STATS_EVERYTHING          = 0x0106,
-  TASK_STATS_FULLLOG             = 0x0107,
-  TASK_STATS_WOODPECKERS         = 0x0108,
-
-  TASK_FULLSCRAPE                = 0x0200, /* Default mode */
-  TASK_FULLSCRAPE_TPB_BINARY     = 0x0201,
-  TASK_FULLSCRAPE_TPB_ASCII      = 0x0202,
-  TASK_FULLSCRAPE_TPB_ASCII_PLUS = 0x0203,
-  TASK_FULLSCRAPE_TPB_URLENCODED = 0x0204,
-  TASK_FULLSCRAPE_TRACKERSTATE   = 0x0205,
-
-  TASK_DMEM                      = 0x0300,
-
-  TASK_DONE                      = 0x0f00,
-  TASK_DONE_PARTIAL              = 0x0f01,
-
-  TASK_FLAG_GZIP                 = 0x1000,
-  TASK_FLAG_BZIP2                = 0x2000,
-  TASK_FLAG_ZSTD                 = 0x4000,
-  TASK_FLAG_CHUNKED              = 0x8000,
-
-  TASK_TASK_MASK                 = 0x0fff,
-  TASK_CLASS_MASK                = 0x0f00,
-  TASK_FLAGS_MASK                = 0xf000
-} ot_tasktype;
-
-typedef unsigned long ot_taskid;
-
-int        mutex_workqueue_pushtask(int64_t sock, ot_tasktype tasktype);
-void       mutex_workqueue_canceltask(int64_t sock);
-void       mutex_workqueue_pushsuccess(ot_taskid taskid);
-ot_taskid  mutex_workqueue_poptask(ot_tasktype *tasktype);
-int        mutex_workqueue_pushresult(ot_taskid taskid, size_t iovec_entries, struct iovec *iovector);
-int        mutex_workqueue_pushchunked(ot_taskid taskid, struct iovec *iovec);
-int64_t    mutex_workqueue_popresult(size_t *iovec_entries, struct iovec **iovector, int *is_partial);
-
 #endif
diff --git a/ot_stats.c b/ot_stats.c
index 5f97a2f..d38355a 100644
--- a/ot_stats.c
+++ b/ot_stats.c
@@ -23,6 +23,7 @@
 #include "ot_iovec.h"
 #include "ot_ip.h"
 #include "ot_mutex.h"
+#include "ot_workqueue.h"
 #include "ot_fmt.h"
 #include "ot_stats.h"
 #include "trackerlogic.h"
@@ -803,17 +804,17 @@ static void *stats_worker(void *args) {
 
   while (g_opentracker_running) {
     ot_tasktype tasktype = TASK_STATS;
-    ot_taskid   taskid   = mutex_workqueue_poptask(&tasktype);
+    ot_taskid   taskid   = workqueue_poptask(&tasktype);
     if (!taskid)
       break;
     stats_make(&iovec_entries, &iovector, tasktype);
-    if (mutex_workqueue_pushresult(taskid, iovec_entries, iovector))
+    if (workqueue_pushresult(taskid, iovec_entries, iovector))
       iovec_free(&iovec_entries, &iovector);
   }
   return NULL;
 }
 
-void stats_deliver(int64_t sock, int tasktype) { mutex_workqueue_pushtask(sock, tasktype); }
+void stats_deliver(int64_t sock, int tasktype) { workqueue_pushtask(sock, tasktype); }
 
 static pthread_t thread_id;
 void stats_init(void) {
@@ -822,6 +823,6 @@ void stats_init(void) {
 }
 
 void stats_deinit(void) {
-  mutex_workqueue_wakeall();
+  workqueue_wakeall();
   pthread_join(thread_id, NULL);
 }
diff --git a/ot_sync.c b/ot_sync.c
index 3668aa1..c19cf36 100644
--- a/ot_sync.c
+++ b/ot_sync.c
@@ -14,6 +14,7 @@
 #include "ot_fmt.h"
 #include "ot_iovec.h"
 #include "ot_mutex.h"
+#include "ot_workqueue.h"
 #include "ot_scan.h"
 #include "ot_stats.h"
 #include "ot_sync.h"
@@ -139,10 +140,10 @@ static void *sync_worker(void *args) {
 
   while (1) {
     ot_tasktype tasktype = TASK_SYNC_OUT;
-    ot_taskid   taskid   = mutex_workqueue_poptask(&tasktype);
+    ot_taskid   taskid   = workqueue_poptask(&tasktype);
     sync_make(&iovec_entries, &iovector);
     stats_issue_event(ws, EVENT_SYNC_OUT, FLAG_TCP, iovec_length(&iovec_entries, &iovector));
-    if (mutex_workqueue_pushresult(taskid, iovec_entries, iovector))
+    if (workqueue_pushresult(taskid, iovec_entries, iovector))
       iovec_free(&iovec_entries, &iovector);
   }
   return NULL;
@@ -158,7 +159,7 @@ void sync_deinit( ) {
 }
 
 void sync_deliver( int64_t socket ) {
-  mutex_workqueue_pushtask( socket, TASK_SYNC_OUT );
+  workqueue_pushtask( socket, TASK_SYNC_OUT );
 }
 
 #endif
diff --git a/ot_workqueue.c b/ot_workqueue.c
new file mode 100644
index 0000000..aac366c
--- /dev/null
+++ b/ot_workqueue.c
@@ -0,0 +1,243 @@
+/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
+   It is considered beerware. Prost. Skol. Cheers or whatever.
+
+   $id$ */
+
+/* System */
+#include <pthread.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/uio.h>
+#include <unistd.h>
+
+/* Opentracker */
+#include "ot_iovec.h"
+#include "ot_workqueue.h"
+#include "trackerlogic.h"
+
+/* Pushing a result wakes the thread that will send it */
+extern int g_self_pipe[2];
+
+struct ot_task {
+  ot_taskid       taskid;
+  ot_tasktype     tasktype;
+  int64_t         sock;
+  size_t          iovec_entries;
+  struct iovec   *iovec;
+  struct ot_task *next;
+};
+
+static ot_taskid       next_free_taskid = 1;
+static struct ot_task *tasklist;
+static pthread_mutex_t tasklist_mutex;
+static pthread_cond_t  tasklist_being_filled;
+
+int workqueue_pushtask(int64_t sock, ot_tasktype tasktype) {
+  struct ot_task **tmptask, *task;
+
+  task = malloc(sizeof(struct ot_task));
+  if (!task)
+    return -1;
+
+  task->taskid        = 0;
+  task->tasktype      = tasktype;
+  task->sock          = sock;
+  task->iovec_entries = 0;
+  task->iovec         = NULL;
+  task->next          = 0;
+
+  /* Want exclusive access to tasklist */
+  pthread_mutex_lock(&tasklist_mutex);
+
+  /* Skip to end of list */
+  tmptask = &tasklist;
+  while (*tmptask)
+    tmptask = &(*tmptask)->next;
+  *tmptask = task;
+
+  /* Inform waiting workers and release lock */
+  pthread_cond_broadcast(&tasklist_being_filled);
+  pthread_mutex_unlock(&tasklist_mutex);
+  return 0;
+}
+
+void workqueue_canceltask(int64_t sock) {
+  struct ot_task **task;
+
+  /* Want exclusive access to tasklist */
+  pthread_mutex_lock(&tasklist_mutex);
+
+  for (task = &tasklist; *task; task = &((*task)->next))
+    if ((*task)->sock == sock) {
+      struct iovec   *iovec = (*task)->iovec;
+      struct ot_task *ptask = *task;
+      size_t                   i;
+
+      /* Free task's iovec, the buffers it points at and the vector itself */
+      for (i = 0; i < (*task)->iovec_entries; ++i)
+        free(iovec[i].iov_base);
+      free(iovec);
+
+      *task = (*task)->next;
+      free(ptask);
+      break;
+    }
+
+  /* Release lock */
+  pthread_mutex_unlock(&tasklist_mutex);
+}
+
+ot_taskid workqueue_poptask(ot_tasktype *tasktype) {
+  struct ot_task *task;
+  ot_taskid       taskid = 0;
+
+  /* Want exclusive access to tasklist */
+  pthread_mutex_lock(&tasklist_mutex);
+
+  while (!taskid) {
+    /* Skip to the first unassigned task this worker wants to do */
+    for (task = tasklist; task; task = task->next)
+      if (!task->taskid && (TASK_CLASS_MASK & task->tasktype) == *tasktype) {
+        /* If we found an outstanding task, assign a taskid to it
+           and leave the loop */
+        task->taskid = taskid = ++next_free_taskid;
+        *tasktype             = task->tasktype;
+        break;
+      }
+
+    /* Wait until the next task is being fed, unless the tracker is on its
+       way out, in which case the worker is meant to leave rather than wait */
+    if (!taskid) {
+      if (!g_opentracker_running)
+        break;
+      pthread_cond_wait(&tasklist_being_filled, &tasklist_mutex);
+    }
+  }
+
+  /* Release lock */
+  pthread_mutex_unlock(&tasklist_mutex);
+
+  return taskid;
+}
+
+void workqueue_pushsuccess(ot_taskid taskid) {
+  struct ot_task **task;
+
+  /* Want exclusive access to tasklist */
+  pthread_mutex_lock(&tasklist_mutex);
+
+  for (task = &tasklist; *task; task = &((*task)->next))
+    if ((*task)->taskid == taskid) {
+      struct ot_task *ptask = *task;
+      *task = (*task)->next;
+      free(ptask);
+      break;
+    }
+
+  /* Release lock */
+  pthread_mutex_unlock(&tasklist_mutex);
+}
+
+int workqueue_pushresult(ot_taskid taskid, size_t iovec_entries, struct iovec *iovec) {
+  struct ot_task *task;
+  const char      byte = 'o';
+
+  /* Want exclusive access to tasklist */
+  pthread_mutex_lock(&tasklist_mutex);
+
+  for (task = tasklist; task; task = task->next)
+    if (task->taskid == taskid) {
+      task->iovec_entries = iovec_entries;
+      task->iovec         = iovec;
+      task->tasktype      = TASK_DONE;
+      break;
+    }
+
+  /* Release lock */
+  pthread_mutex_unlock(&tasklist_mutex);
+
+  /* A full pipe has already done this nudge's job, so a failure is fine */
+  (void)!write(g_self_pipe[1], &byte, 1);
+
+  /* Indicate whether the worker has to throw away results */
+  return task ? 0 : -1;
+}
+
+int workqueue_pushchunked(ot_taskid taskid, struct iovec *iovec) {
+  struct ot_task *task;
+  const char      byte = 'o';
+
+  /* Want exclusive access to tasklist */
+  pthread_mutex_lock(&tasklist_mutex);
+
+  for (task = tasklist; task; task = task->next)
+    if (task->taskid == taskid) {
+      if (iovec) {
+        if (iovec_append(&task->iovec_entries, &task->iovec, iovec))
+          task->tasktype = TASK_DONE_PARTIAL;
+        else
+          task = NULL;
+      } else
+        task->tasktype = TASK_DONE;
+      break;
+    }
+
+  /* Release lock */
+  pthread_mutex_unlock(&tasklist_mutex);
+
+  /* A full pipe has already done this nudge's job, so a failure is fine */
+  (void)!write(g_self_pipe[1], &byte, 1);
+
+  /* Indicate whether the worker has to throw away results */
+  return task ? 0 : -1;
+}
+
+int64_t workqueue_popresult(size_t *iovec_entries, struct iovec **iovec, int *is_partial) {
+  struct ot_task **task;
+  int64_t sock = -1;
+
+  *is_partial = 0;
+
+  /* Want exclusive access to tasklist */
+  pthread_mutex_lock(&tasklist_mutex);
+
+  for (task = &tasklist; *task; task = &((*task)->next))
+    if (((*task)->tasktype & TASK_CLASS_MASK) == TASK_DONE) {
+      struct ot_task *ptask = *task;
+      *iovec_entries        = ptask->iovec_entries;
+      *iovec                = ptask->iovec;
+      sock                  = ptask->sock;
+
+      if ((*task)->tasktype == TASK_DONE) {
+        *task = ptask->next;
+        free(ptask);
+      } else {
+        ptask->iovec_entries = 0;
+        ptask->iovec         = NULL;
+        *is_partial          = 1;
+        /* Prevent task from showing up immediately again unless new data was added */
+        (*task)->tasktype    = TASK_FULLSCRAPE;
+      }
+      break;
+    }
+
+  /* Release lock */
+  pthread_mutex_unlock(&tasklist_mutex);
+  return sock;
+}
+
+void workqueue_wakeall(void) {
+  pthread_mutex_lock(&tasklist_mutex);
+  pthread_cond_broadcast(&tasklist_being_filled);
+  pthread_mutex_unlock(&tasklist_mutex);
+}
+
+void workqueue_init(void) {
+  pthread_mutex_init(&tasklist_mutex, NULL);
+  pthread_cond_init(&tasklist_being_filled, NULL);
+}
+
+void workqueue_deinit(void) {
+  pthread_mutex_destroy(&tasklist_mutex);
+  pthread_cond_destroy(&tasklist_being_filled);
+}
diff --git a/ot_workqueue.h b/ot_workqueue.h
new file mode 100644
index 0000000..f99a7b5
--- /dev/null
+++ b/ot_workqueue.h
@@ -0,0 +1,78 @@
+/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
+   It is considered beerware. Prost. Skol. Cheers or whatever.
+
+   $id$ */
+
+#ifndef OT_WORKQUEUE_H__
+#define OT_WORKQUEUE_H__
+
+#include <stdint.h>
+#include <sys/uio.h>
+
+#include "trackerlogic.h"
+
+/* Work the tracker hands to a thread of its own: a full scrape or an
+   expensive stats page, which take too long to build while a client waits.
+   The task goes in here, the reply comes back out. */
+
+typedef enum {
+  TASK_STATS_CONNS               = 0x0001,
+  TASK_STATS_TCP                 = 0x0002,
+  TASK_STATS_UDP                 = 0x0003,
+  TASK_STATS_SCRAPE              = 0x0004,
+  TASK_STATS_FULLSCRAPE          = 0x0005,
+  TASK_STATS_TPB                 = 0x0006,
+  TASK_STATS_HTTPERRORS          = 0x0007,
+  TASK_STATS_VERSION             = 0x0008,
+  TASK_STATS_RENEW               = 0x000a,
+  TASK_STATS_SYNCS               = 0x000b,
+  TASK_STATS_COMPLETED           = 0x000c,
+  TASK_STATS_NUMWANTS            = 0x000d,
+
+  TASK_STATS                     = 0x0100, /* Mask */
+  TASK_STATS_TORRENTS            = 0x0101,
+  TASK_STATS_PEERS               = 0x0102,
+  TASK_STATS_SLASH24S            = 0x0103,
+  TASK_STATS_TOP10               = 0x0104,
+  TASK_STATS_TOP100              = 0x0105,
+  TASK_STATS_EVERYTHING          = 0x0106,
+  TASK_STATS_FULLLOG             = 0x0107,
+  TASK_STATS_WOODPECKERS         = 0x0108,
+
+  TASK_FULLSCRAPE                = 0x0200, /* Default mode */
+  TASK_FULLSCRAPE_TPB_BINARY     = 0x0201,
+  TASK_FULLSCRAPE_TPB_ASCII      = 0x0202,
+  TASK_FULLSCRAPE_TPB_ASCII_PLUS = 0x0203,
+  TASK_FULLSCRAPE_TPB_URLENCODED = 0x0204,
+  TASK_FULLSCRAPE_TRACKERSTATE   = 0x0205,
+
+  TASK_DMEM                      = 0x0300,
+
+  TASK_DONE                      = 0x0f00,
+  TASK_DONE_PARTIAL              = 0x0f01,
+
+  TASK_FLAG_GZIP                 = 0x1000,
+  TASK_FLAG_BZIP2                = 0x2000,
+  TASK_FLAG_ZSTD                 = 0x4000,
+  TASK_FLAG_CHUNKED              = 0x8000,
+
+  TASK_TASK_MASK                 = 0x0fff,
+  TASK_CLASS_MASK                = 0x0f00,
+  TASK_FLAGS_MASK                = 0xf000
+} ot_tasktype;
+
+typedef unsigned long ot_taskid;
+
+int        workqueue_pushtask(int64_t sock, ot_tasktype tasktype);
+void       workqueue_canceltask(int64_t sock);
+void       workqueue_pushsuccess(ot_taskid taskid);
+ot_taskid  workqueue_poptask(ot_tasktype *tasktype);
+int        workqueue_pushresult(ot_taskid taskid, size_t iovec_entries, struct iovec *iovector);
+int        workqueue_pushchunked(ot_taskid taskid, struct iovec *iovec);
+int64_t    workqueue_popresult(size_t *iovec_entries, struct iovec **iovector, int *is_partial);
+
+void       workqueue_init(void);
+void       workqueue_deinit(void);
+void       workqueue_wakeall(void);
+
+#endif
diff --git a/tests/runner.c b/tests/runner.c
index 71a5e2a..38bfb86 100644
--- a/tests/runner.c
+++ b/tests/runner.c
@@ -27,7 +27,7 @@ static const struct module modules[] = {{"ot_urlencoded", test_urlencoded},
                                         {"ot_buf", test_buf},
                                         {"ot_ip", test_ip},
                                         {"ot_socket", test_socket},
-                                        {"ot_mutex", test_mutex},
+                                        {"ot_workqueue", test_workqueue},
                                         {"ot_batch", test_batch},
                                         {"ot_fdstate", test_fdstate},
                                         {"ot_event", test_event},
diff --git a/tests/t_mutex.c b/tests/t_workqueue.c
index 3307d70..13e0b2e 100644
--- a/tests/t_mutex.c
+++ b/tests/t_workqueue.c
@@ -8,7 +8,7 @@
 #include <string.h>
 
 /* Opentracker */
-#include "ot_mutex.h"
+#include "ot_workqueue.h"
 #include "test.h"
 
 /* A worker asks for a class of task and blocks until one turns up, so a task
@@ -20,25 +20,25 @@ static void case_task_goes_to_its_own_class(void) {
   ot_taskid   taskid;
 
   test_case("a worker is handed the task of the class it asked for");
-  CHECK_INT(mutex_workqueue_pushtask(41, TASK_FULLSCRAPE_TPB_ASCII), 0);
+  CHECK_INT(workqueue_pushtask(41, TASK_FULLSCRAPE_TPB_ASCII), 0);
   asked  = TASK_FULLSCRAPE;
-  taskid = mutex_workqueue_poptask(&asked);
+  taskid = workqueue_poptask(&asked);
   CHECK(taskid != 0);
 
   test_case("and the full task type, not just its class");
   CHECK_INT(asked, TASK_FULLSCRAPE_TPB_ASCII);
-  mutex_workqueue_canceltask(41);
+  workqueue_canceltask(41);
 }
 
 static void case_flags_survive_the_queue(void) {
   ot_tasktype asked;
 
   test_case("flags outside the class mask are carried through");
-  CHECK_INT(mutex_workqueue_pushtask(42, TASK_FULLSCRAPE_TPB_URLENCODED | TASK_FLAG_GZIP), 0);
+  CHECK_INT(workqueue_pushtask(42, TASK_FULLSCRAPE_TPB_URLENCODED | TASK_FLAG_GZIP), 0);
   asked = TASK_FULLSCRAPE;
-  CHECK(mutex_workqueue_poptask(&asked) != 0);
+  CHECK(workqueue_poptask(&asked) != 0);
   CHECK_INT(asked, TASK_FULLSCRAPE_TPB_URLENCODED | TASK_FLAG_GZIP);
-  mutex_workqueue_canceltask(42);
+  workqueue_canceltask(42);
 }
 
 static void case_classes_do_not_cross(void) {
@@ -46,22 +46,22 @@ static void case_classes_do_not_cross(void) {
   ot_taskid   taskid;
 
   test_case("a stats worker passes over a fullscrape task");
-  CHECK_INT(mutex_workqueue_pushtask(43, TASK_FULLSCRAPE_TPB_BINARY), 0);
-  CHECK_INT(mutex_workqueue_pushtask(44, TASK_STATS_PEERS), 0);
+  CHECK_INT(workqueue_pushtask(43, TASK_FULLSCRAPE_TPB_BINARY), 0);
+  CHECK_INT(workqueue_pushtask(44, TASK_STATS_PEERS), 0);
 
   asked  = TASK_STATS;
-  taskid = mutex_workqueue_poptask(&asked);
+  taskid = workqueue_poptask(&asked);
   CHECK(taskid != 0);
   CHECK_INT(asked, TASK_STATS_PEERS);
 
   test_case("and leaves it for the one that wants it");
   asked  = TASK_FULLSCRAPE;
-  taskid = mutex_workqueue_poptask(&asked);
+  taskid = workqueue_poptask(&asked);
   CHECK(taskid != 0);
   CHECK_INT(asked, TASK_FULLSCRAPE_TPB_BINARY);
 
-  mutex_workqueue_canceltask(43);
-  mutex_workqueue_canceltask(44);
+  workqueue_canceltask(43);
+  workqueue_canceltask(44);
 }
 
 static void case_every_task_gets_its_own_id(void) {
@@ -69,36 +69,36 @@ static void case_every_task_gets_its_own_id(void) {
   ot_taskid   first, second;
 
   test_case("two queued tasks are handed out under different ids");
-  CHECK_INT(mutex_workqueue_pushtask(45, TASK_FULLSCRAPE), 0);
-  CHECK_INT(mutex_workqueue_pushtask(46, TASK_FULLSCRAPE), 0);
+  CHECK_INT(workqueue_pushtask(45, TASK_FULLSCRAPE), 0);
+  CHECK_INT(workqueue_pushtask(46, TASK_FULLSCRAPE), 0);
 
   asked  = TASK_FULLSCRAPE;
-  first  = mutex_workqueue_poptask(&asked);
+  first  = workqueue_poptask(&asked);
   asked  = TASK_FULLSCRAPE;
-  second = mutex_workqueue_poptask(&asked);
+  second = workqueue_poptask(&asked);
   CHECK(first != 0);
   CHECK(second != 0);
   CHECK(first != second);
 
-  mutex_workqueue_canceltask(45);
-  mutex_workqueue_canceltask(46);
+  workqueue_canceltask(45);
+  workqueue_canceltask(46);
 }
 
 static void case_cancelled_task_is_not_handed_out(void) {
   ot_tasktype asked;
 
   test_case("a cancelled task is gone before any worker sees it");
-  CHECK_INT(mutex_workqueue_pushtask(47, TASK_FULLSCRAPE_TPB_ASCII_PLUS), 0);
-  CHECK_INT(mutex_workqueue_pushtask(48, TASK_FULLSCRAPE), 0);
-  mutex_workqueue_canceltask(47);
+  CHECK_INT(workqueue_pushtask(47, TASK_FULLSCRAPE_TPB_ASCII_PLUS), 0);
+  CHECK_INT(workqueue_pushtask(48, TASK_FULLSCRAPE), 0);
+  workqueue_canceltask(47);
 
   asked = TASK_FULLSCRAPE;
-  CHECK(mutex_workqueue_poptask(&asked) != 0);
+  CHECK(workqueue_poptask(&asked) != 0);
   CHECK_INT(asked, TASK_FULLSCRAPE);
-  mutex_workqueue_canceltask(48);
+  workqueue_canceltask(48);
 }
 
-void test_mutex(void) {
+void test_workqueue(void) {
   case_task_goes_to_its_own_class();
   case_flags_survive_the_queue();
   case_classes_do_not_cross();
diff --git a/tests/test.h b/tests/test.h
index 0eddc82..d58742c 100644
--- a/tests/test.h
+++ b/tests/test.h
@@ -86,7 +86,7 @@ void test_scan(void);
 void test_buf(void);
 void test_ip(void);
 void test_socket(void);
-void test_mutex(void);
+void test_workqueue(void);
 void test_batch(void);
 void test_fdstate(void);
 void test_event(void);
diff --git a/trackerlogic.c b/trackerlogic.c
index fc0fa56..b0b7f4e 100644
--- a/trackerlogic.c
+++ b/trackerlogic.c
@@ -20,6 +20,7 @@
 #include "ot_ip.h"
 #include "ot_livesync.h"
 #include "ot_mutex.h"
+#include "ot_workqueue.h"
 #include "ot_random.h"
 #include "ot_stats.h"
 #include "ot_udp.h"
@@ -560,6 +561,7 @@ void trackerlogic_init(void) {
 
   /* Initialise background worker threads */
   mutex_init();
+  workqueue_init();
   clean_init();
   fullscrape_init();
   accesslist_init();
@@ -623,5 +625,6 @@ void trackerlogic_deinit(void) {
   }
 
   /* Release mutexes */
+  workqueue_deinit();
   mutex_deinit();
 }