22 files changed, 789 insertions, 135 deletions

diff --git a/Makefile b/Makefile
index 55be5ba..c7105d4 100644
--- a/Makefile
+++ b/Makefile
@@ -64,13 +64,13 @@ GIT_VERSION=`sh -c 'command -v git >/dev/null && test -d .git && git rev-parse H
 OPTS_debug=-D_DEBUG -g -ggdb # -pg -fprofile-arcs -ftest-coverage
 OPTS_production=-O3
 
-CFLAGS+=-I$(LIBOWFAT_HEADERS) -DGIT_VERSION=$(GIT_VERSION) -Wall -pipe -pthread -Wextra #-ansi -pedantic
+CFLAGS+=-I$(LIBOWFAT_HEADERS) -DGIT_VERSION=$(GIT_VERSION) -Wall -pipe -pthread -Wextra -Wwrite-strings #-ansi -pedantic
 LDFLAGS+=-L$(LIBOWFAT_LIBRARY) -lowfat -pthread
 
 BINARY =opentracker
-HEADERS=trackerlogic.h scan_urlencoded_query.h ot_buf.h ot_hex.h ot_ip.h ot_config.h ot_scan.h ot_socket.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 scan_urlencoded_query.c ot_buf.c ot_hex.c ot_ip.c ot_config.c ot_scan.c ot_socket.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_vector.c ot_mutex.c ot_random.c ot_scan.c ot_ip.c ot_socket.c
+HEADERS=trackerlogic.h scan_urlencoded_query.h ot_batch.h ot_buf.h ot_hex.h ot_ip.h ot_config.h ot_scan.h ot_socket.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 scan_urlencoded_query.c ot_batch.c ot_buf.c ot_hex.c ot_ip.c ot_config.c ot_scan.c ot_socket.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_vector.c ot_mutex.c ot_random.c ot_scan.c ot_ip.c ot_socket.c
 
 OBJECTS = $(SOURCES:%.c=%.o)
 OBJECTS_debug = $(SOURCES:%.c=%.debug.o)
@@ -119,8 +119,8 @@ $(OBJECTS_proxy_debug): $(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_scan_urlencoded_query.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
-TEST_UNITS=scan_urlencoded_query.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
+TEST_SOURCES=tests/runner.c tests/test.c tests/stubs.c tests/t_scan_urlencoded_query.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
+TEST_UNITS=scan_urlencoded_query.c ot_batch.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_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 0af0d17..60bdceb 100644
--- a/opentracker.c
+++ b/opentracker.c
@@ -23,7 +23,6 @@
 
 /* Libowfat */
 #include "io.h"
-#include "iob.h"
 
 /* Opentracker */
 #include "ot_accesslist.h"
@@ -36,6 +35,7 @@
 #include "ot_random.h"
 #include "ot_scan.h"
 #include "ot_socket.h"
+#include "ot_timeout.h"
 #include "ot_stats.h"
 #include "ot_udp.h"
 #include "trackerlogic.h"
@@ -120,9 +120,8 @@ static ssize_t header_complete(char *request, ssize_t byte_count) {
 static void handle_dead(const int64_t sock) {
   struct http_data *cookie = io_getcookie(sock);
   if (cookie) {
-    size_t i;
-    for (i = 0; i < cookie->batches; ++i)
-      iob_reset(cookie->batch + i);
+    if (cookie->batch)
+      batch_reset(cookie->batch);
     free(cookie->batch);
     buf_reset(&cookie->request);
     if (cookie->flag & (STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER))
@@ -178,33 +177,23 @@ static void handle_read(const int64_t sock, struct ot_workstruct *ws) {
 
 static void handle_write(const int64_t sock) {
   struct http_data *cookie = io_getcookie(sock);
-  size_t            i;
   int               chunked = 0;
 
-  /* Look for the first io_batch still containing bytes to write */
   if (cookie) {
     if (cookie->flag & STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER)
       chunked = 1;
 
-    for (i = 0; i < cookie->batches; ++i) {
-      if (cookie->batch[i].bytesleft) {
-        int64_t res = iob_send(sock, cookie->batch + i);
-
-        if (res == -3) {
-          handle_dead(sock);
-          return;
-        }
-
-        if (!cookie->batch[i].bytesleft)
-          continue;
-
-        if (res == -1 || res > 0 || i < cookie->batches - 1)
-          return;
+    if (cookie->batch) {
+      if (batch_send(sock, cookie->batch) == BATCH_FAILED) {
+        handle_dead(sock);
+        return;
       }
+      if (batch_pending(cookie->batch))
+        return;
     }
   }
 
-  /* In a chunked transfer after all batches accumulated have been sent, wait for the next one */
+  /* In a chunked transfer, wait for the next piece once everything queued has gone out */
   if (chunked)
     io_dontwantwrite(sock);
   else
@@ -216,7 +205,6 @@ static void handle_accept(const int64_t serversocket) {
   int64_t           sock;
   ot_ip6            ip;
   uint16_t          port;
-  tai6464           t;
 
   while ((sock = socket_accept(serversocket, ip, &port)) != -1) {
 
@@ -235,11 +223,7 @@ static void handle_accept(const int64_t serversocket) {
 
     stats_issue_event(EVENT_ACCEPT, FLAG_TCP, (uintptr_t)ip);
 
-    /* That breaks taia encapsulation. But there is no way to take system
-       time this often in FreeBSD and libowfat does not allow to set unix time */
-    taia_uint(&t, 0); /* Clear t */
-    tai_unix(&(t.sec), (g_now_seconds + OT_CLIENT_TIMEOUT));
-    io_timeout(sock, t);
+    timeout_in(sock, OT_CLIENT_TIMEOUT);
   }
   io_eagain(serversocket);
 }
@@ -305,7 +289,7 @@ static void listen_bind_one(ot_listen *entry) {
 
 #ifdef _DEBUG
   {
-    char *protos[] = {"TCP", "UDP", "UDP mcast"};
+    const char *protos[] = {"TCP", "UDP", "UDP mcast"};
     char  _debug[512];
     int   off  = snprintf(_debug, sizeof(_debug), "Binding socket type %s to address [", protos[entry->proto]);
     off       += ip_format(_debug + off, entry->ip);
diff --git a/ot_accesslist.c b/ot_accesslist.c
index 2465ada..d7e88ee 100644
--- a/ot_accesslist.c
+++ b/ot_accesslist.c
@@ -32,10 +32,10 @@
 
 /* GLOBAL VARIABLES */
 #ifdef WANT_ACCESSLIST
-char *g_accesslist_filename = NULL;
+const char *g_accesslist_filename = NULL;
 #ifdef WANT_DYNAMIC_ACCESSLIST
-char *g_accesslist_pipe_add    = NULL;
-char *g_accesslist_pipe_delete = NULL;
+const char *g_accesslist_pipe_add    = NULL;
+const char *g_accesslist_pipe_delete = NULL;
 #endif
 static pthread_mutex_t g_accesslist_mutex;
 
@@ -272,7 +272,7 @@ static void *accesslist_worker(void *args) {
 
 #ifdef WANT_DYNAMIC_ACCESSLIST
 static pthread_t thread_adder_id, thread_deleter_id;
-static void     *accesslist_adddel_worker(char *fifoname, ot_accesslist *_Atomic *adding_to, ot_accesslist *_Atomic *removing_from) {
+static void     *accesslist_adddel_worker(const char *fifoname, ot_accesslist *_Atomic *adding_to, ot_accesslist *_Atomic *removing_from) {
   struct stat st;
 
   if (!stat(fifoname, &st)) {
diff --git a/ot_accesslist.h b/ot_accesslist.h
index 30aaebd..9dfdd95 100644
--- a/ot_accesslist.h
+++ b/ot_accesslist.h
@@ -23,10 +23,10 @@ void         accesslist_cleanup(void);
    line, and return how many were written to dest */
 size_t       accesslist_parse_hashes(ot_hash *dest, size_t capacity, const char *buf, size_t len);
 
-extern char *g_accesslist_filename;
+extern const char *g_accesslist_filename;
 #ifdef WANT_DYNAMIC_ACCESSLIST
-extern char *g_accesslist_pipe_add;
-extern char *g_accesslist_pipe_delete;
+extern const char *g_accesslist_pipe_add;
+extern const char *g_accesslist_pipe_delete;
 #endif
 
 #else
diff --git a/ot_batch.c b/ot_batch.c
new file mode 100644
index 0000000..84eb4a8
--- /dev/null
+++ b/ot_batch.c
@@ -0,0 +1,158 @@
+/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
+   It is considered beerware. Prost. Skol. Cheers or whatever.
+
+   $id$ */
+
+/* System */
+#include <errno.h>
+#include <limits.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sys/uio.h>
+#include <unistd.h>
+
+/* Opentracker */
+#include "ot_batch.h"
+
+#ifndef IOV_MAX
+#define IOV_MAX 1024
+#endif
+
+#define OT_BATCH_MIN_ENTRIES 16
+
+static int batch_room_for(ot_batch *batch, size_t wanted) {
+  size_t        space = batch->space ? batch->space : OT_BATCH_MIN_ENTRIES;
+  struct iovec *grown;
+
+  if (batch->count + wanted <= batch->space)
+    return 0;
+
+  /* During a chunked transfer entries are appended while earlier ones are
+     still going out, so the front of the array frees up as we go. Slide the
+     rest down before asking for more. */
+  if (batch->first && batch->count - batch->first + wanted <= batch->space) {
+    memmove(batch->entries, batch->entries + batch->first, (batch->count - batch->first) * sizeof(struct iovec));
+    batch->count -= batch->first;
+    batch->first  = 0;
+    return 0;
+  }
+
+  while (space < batch->count + wanted)
+    space *= 2;
+
+  if (!(grown = realloc(batch->entries, space * sizeof(struct iovec))))
+    return -1;
+
+  batch->entries = grown;
+  batch->space   = space;
+  return 0;
+}
+
+int batch_add(ot_batch *batch, void *buf, size_t len) {
+  if (batch_room_for(batch, 1)) {
+    free(buf);
+    return -1;
+  }
+
+  batch->entries[batch->count].iov_base = buf;
+  batch->entries[batch->count].iov_len  = len;
+  ++batch->count;
+  return 0;
+}
+
+int batch_add_vector(ot_batch *batch, const struct iovec *entries, size_t count) {
+  if (batch_room_for(batch, count)) {
+    while (count--)
+      free(entries[count].iov_base);
+    return -1;
+  }
+
+  memcpy(batch->entries + batch->count, entries, count * sizeof(struct iovec));
+  batch->count += count;
+  return 0;
+}
+
+/* Drop everything the kernel accepted, freeing each buffer as it goes */
+static void batch_consume(ot_batch *batch, size_t written) {
+  while (written) {
+    struct iovec *entry = batch->entries + batch->first;
+    size_t        left  = entry->iov_len - batch->offset;
+
+    if (written < left) {
+      batch->offset += written;
+      return;
+    }
+
+    free(entry->iov_base);
+    written      -= left;
+    batch->offset = 0;
+    ++batch->first;
+  }
+
+  /* Nothing is outstanding, so the array can start over from the front */
+  if (batch->first == batch->count)
+    batch->first = batch->count = 0;
+}
+
+/* A peer that has gone away must come back as an error rather than as a
+   signal that kills the tracker, so ask the kernel not to raise one where it
+   can be asked. macOS cannot, and relies on SIGPIPE being blocked. */
+static ssize_t batch_writev(int s, struct iovec *entries, int count) {
+#ifdef MSG_NOSIGNAL
+  struct msghdr msg;
+
+  memset(&msg, 0, sizeof(msg));
+  msg.msg_iov    = entries;
+  msg.msg_iovlen = count;
+  return sendmsg(s, &msg, MSG_NOSIGNAL);
+#else
+  return writev(s, entries, count);
+#endif
+}
+
+ssize_t batch_send(int s, ot_batch *batch) {
+  size_t  entries = batch->count - batch->first;
+  ssize_t written;
+
+  if (!batch_pending(batch))
+    return BATCH_DONE;
+
+  if (entries > IOV_MAX)
+    entries = IOV_MAX;
+
+  /* The first entry may be half way out already */
+  batch->entries[batch->first].iov_base = (char *)batch->entries[batch->first].iov_base + batch->offset;
+  batch->entries[batch->first].iov_len -= batch->offset;
+  written                               = batch_writev(s, batch->entries + batch->first, (int)entries);
+  batch->entries[batch->first].iov_base = (char *)batch->entries[batch->first].iov_base - batch->offset;
+  batch->entries[batch->first].iov_len += batch->offset;
+
+  if (written < 0)
+    return (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) ? BATCH_AGAIN : BATCH_FAILED;
+  if (!written)
+    return BATCH_FAILED;
+
+  batch_consume(batch, (size_t)written);
+  return written;
+}
+
+void batch_reset(ot_batch *batch) {
+  while (batch->first < batch->count)
+    free(batch->entries[batch->first++].iov_base);
+
+  free(batch->entries);
+  batch->entries   = NULL;
+  batch->count = batch->first = batch->offset = batch->space = 0;
+}
+
+/* For the few constants the chunked encoding needs, which the batch frees
+   along with everything else */
+int batch_add_copy(ot_batch *batch, const void *buf, size_t len) {
+  void *copy = malloc(len);
+
+  if (!copy)
+    return -1;
+  memcpy(copy, buf, len);
+  return batch_add(batch, copy, len);
+}
diff --git a/ot_batch.h b/ot_batch.h
new file mode 100644
index 0000000..c855671
--- /dev/null
+++ b/ot_batch.h
@@ -0,0 +1,48 @@
+/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
+   It is considered beerware. Prost. Skol. Cheers or whatever.
+
+   $id$ */
+
+#ifndef OT_BATCH_H__
+#define OT_BATCH_H__
+
+#include <stddef.h>
+#include <sys/types.h>
+#include <sys/uio.h>
+
+/* A queue of buffers on their way to one socket. The batch owns every buffer
+   handed to it and frees each one as soon as it has left, so a client that
+   reads slowly costs memory only for what it has not received yet.
+
+   A zeroed ot_batch is ready to use. */
+typedef struct {
+  struct iovec *entries;
+  size_t        space;
+  size_t        count;
+  size_t        first;  /* entries before this one are sent and freed */
+  size_t        offset; /* bytes of entries[first] already sent */
+} ot_batch;
+
+static inline int batch_pending(const ot_batch *batch) { return batch->first != batch->count; }
+
+/* What batch_send reports instead of a byte count */
+enum { BATCH_DONE = 0, BATCH_AGAIN = -1, BATCH_FAILED = -2 };
+
+/* Queue buf and take ownership of it. Returns 0, or -1 having freed buf. */
+int     batch_add(ot_batch *batch, void *buf, size_t len);
+
+/* Write as much as the socket takes. Returns the number of bytes written, or
+   BATCH_DONE, BATCH_AGAIN or BATCH_FAILED. */
+ssize_t batch_send(int s, ot_batch *batch);
+
+/* Queue every entry of a vector, taking ownership of each buffer but not of
+   the vector itself. Returns 0, or -1 having freed all of them. */
+int     batch_add_vector(ot_batch *batch, const struct iovec *entries, size_t count);
+
+/* Queue a copy of buf, which the batch then owns. Returns 0, or -1. */
+int     batch_add_copy(ot_batch *batch, const void *buf, size_t len);
+
+/* Free whatever has not left yet, along with the batch's own array */
+void    batch_reset(ot_batch *batch);
+
+#endif
diff --git a/ot_config.c b/ot_config.c
index 43c07c5..e9666f9 100644
--- a/ot_config.c
+++ b/ot_config.c
@@ -22,10 +22,10 @@
 #include "ot_stats.h"
 #include "trackerlogic.h"
 
-char         *g_redirecturl;
-char         *g_serverdir;
-char         *g_serveruser;
-char         *g_statefile;
+const char   *g_redirecturl;
+const char   *g_serverdir;
+const char   *g_serveruser;
+const char   *g_statefile;
 
 #define OT_LISTEN_INHERIT (-1)
 
@@ -122,13 +122,13 @@ static int listen_scan_options(char *p, int *shared, int *workers) {
 }
 
 
-char *set_config_option(char **option, char *value) {
+static const char *set_config_option(const char **option, char *value) {
 #ifdef _DEBUG
   fprintf(stderr, "Setting config option: %s\n", value);
 #endif
   while (isspace(*value))
     ++value;
-  free(*option);
+  free((char *)*option);
   return *option = strdup(value);
 }
 
diff --git a/ot_config.h b/ot_config.h
index ab25e2c..059573a 100644
--- a/ot_config.h
+++ b/ot_config.h
@@ -22,10 +22,10 @@ typedef struct {
   int64_t    sock;
 } ot_listen;
 
-extern char *g_redirecturl;
-extern char *g_serverdir;
-extern char *g_serveruser;
-extern char *g_statefile;
+extern const char *g_redirecturl;
+extern const char *g_serverdir;
+extern const char *g_serveruser;
+extern const char *g_statefile;
 
 /* Reads the command line and every config file it names. Exits on error. */
 void       config_parse(int argc, char **argv);
diff --git a/ot_fullscrape.c b/ot_fullscrape.c
index 9dcbe68..eade79d 100644
--- a/ot_fullscrape.c
+++ b/ot_fullscrape.c
@@ -83,7 +83,7 @@ void fullscrape_deinit(void) {
 }
 
 void fullscrape_deliver( int64_t sock, ot_tasktype tasktype ) {
-  mutex_workqueue_pushtask( sock, tasktype );
+  mutex_workqueue_pushtask( sock, TASK_FULLSCRAPE | tasktype );
 }
 
 static char * fullscrape_write_one( ot_tasktype mode, char *r, ot_torrent *torrent, ot_hash *hash ) {
diff --git a/ot_http.c b/ot_http.c
index 46ca5a9..331a457 100644
--- a/ot_http.c
+++ b/ot_http.c
@@ -15,7 +15,7 @@
 #include <unistd.h>
 
 /* Libowfat */
-#include "iob.h"
+#include "io.h"
 
 /* Opentracker */
 #include "ot_accesslist.h"
@@ -26,20 +26,14 @@
 #include "ot_mutex.h"
 #include "ot_scan.h"
 #include "ot_stats.h"
+#include "ot_timeout.h"
 #include "scan_urlencoded_query.h"
 #include "trackerlogic.h"
 
-#ifdef WANT_NO_AUTO_FREE
-#define OT_IOB_INIT(B) (bzero(B, sizeof(io_batch)), 0)
-#else
-#define OT_IOB_INIT(B) iob_init_autofree(B, 0)
-#endif
-
 #define OT_MAXMULTISCRAPE_COUNT 64
-#define OT_BATCH_LIMIT          (1024 * 1024 * 16)
-extern char *g_redirecturl;
+extern const char *g_redirecturl;
 
-char        *g_stats_path;
+const char  *g_stats_path;
 ssize_t      g_stats_path_len;
 
 enum { SUCCESS_HTTP_HEADER_LENGTH = 80, SUCCESS_HTTP_SIZE_OFF = 17 };
@@ -79,7 +73,6 @@ static void http_senddata(const int64_t sock, struct ot_workstruct *ws) {
 
   if (written_size < ws->reply_size) {
     char   *outbuf;
-    tai6464 t;
 
     if (!(outbuf = malloc(ws->reply_size - written_size))) {
       buf_reset(&cookie->request);
@@ -90,25 +83,21 @@ static void http_senddata(const int64_t sock, struct ot_workstruct *ws) {
 
     memcpy(outbuf, ws->reply + written_size, ws->reply_size - written_size);
     if (!cookie->batch) {
-      cookie->batch = malloc(sizeof(io_batch));
-      if (!cookie->batch || OT_IOB_INIT(cookie->batch) == -1) {
-        free(cookie->batch);
+      cookie->batch = calloc(1, sizeof(ot_batch));
+      if (!cookie->batch) {
         free(outbuf);
         buf_reset(&cookie->request);
         free(cookie);
         io_close(sock);
         return;
       }
-
-      cookie->batches = 1;
     }
 
-    iob_addbuf_free(cookie->batch, outbuf, ws->reply_size - written_size);
+    batch_add(cookie->batch, outbuf, ws->reply_size - written_size);
 
     /* writeable short data sockets just have a tcp timeout */
     if (!ws->keep_alive) {
-      taia_uint(&t, 0);
-      io_timeout(sock, t);
+      timeout_never(sock);
       io_dontwantread(sock);
     }
     io_wantwrite(sock);
@@ -125,9 +114,9 @@ static void http_senddata(const int64_t sock, struct ot_workstruct *ws) {
 #define HTTPERROR_404            return http_issue_error(sock, ws, CODE_HTTPERROR_404)
 #define HTTPERROR_500            return http_issue_error(sock, ws, CODE_HTTPERROR_500)
 ssize_t http_issue_error(const int64_t sock, struct ot_workstruct *ws, int code) {
-  char *error_code[] = {"302 Found",      "400 Invalid Request", "400 Invalid Request", "400 Invalid Request",      "402 Payment Required",
+  const char *error_code[] = {"302 Found",      "400 Invalid Request", "400 Invalid Request", "400 Invalid Request",      "402 Payment Required",
                         "403 Not Modest", "403 Access Denied",   "404 Not Found",       "500 Internal Server Error"};
-  char *title        = error_code[code];
+  const char *title  = error_code[code];
 
   ws->reply          = ws->outbuf;
   if (code == CODE_HTTPERROR_302)
@@ -146,11 +135,9 @@ ssize_t http_issue_error(const int64_t sock, struct ot_workstruct *ws, int code)
 
 ssize_t http_sendiovecdata(const int64_t sock, struct ot_workstruct *ws, size_t iovec_entries, struct iovec *iovector, int is_partial) {
   struct http_data *cookie = io_getcookie(sock);
-  io_batch         *current;
   char             *header;
   const char       *encoding = "";
-  size_t            i, header_size, size = iovec_length(&iovec_entries, (const struct iovec **)&iovector);
-  tai6464           t;
+  size_t            header_size, size = iovec_length(&iovec_entries, (const struct iovec **)&iovector);
 
   /* No cookie? Bad socket. Leave. */
   if (!cookie) {
@@ -199,46 +186,28 @@ ssize_t http_sendiovecdata(const int64_t sock, struct ot_workstruct *ws, size_t
     }
 
     if (!cookie->batch) {
-      cookie->batch = malloc(sizeof(io_batch));
-      if (!cookie->batch || OT_IOB_INIT(cookie->batch) == -1) {
-        free(cookie->batch);
+      cookie->batch = calloc(1, sizeof(ot_batch));
+      if (!cookie->batch) {
         free(header);
         iovec_free(&iovec_entries, &iovector);
         HTTPERROR_500;
       }
-      cookie->batches = 1;
     }
-    current = cookie->batch + cookie->batches - 1;
-    iob_addbuf_free(current, header, header_size);
 
-    /* Split huge iovectors into separate io_batches */
-    for (i = 0; i < (size_t)iovec_entries; ++i) {
-      /* If the current batch's limit is reached, try to reallocate a new batch to work on */
-      if (current->bytesleft > OT_BATCH_LIMIT) {
-        io_batch *new_batch = realloc(cookie->batch, (cookie->batches + 1) * sizeof(io_batch));
-        if (new_batch) {
-          cookie->batch = new_batch;
-          if (OT_IOB_INIT(current) != -1)
-            current = cookie->batch + cookie->batches++;
-        }
-      }
-      iob_addbuf_free(current, iovector[i].iov_base, iovector[i].iov_len);
-    }
+    batch_add(cookie->batch, header, header_size);
+    batch_add_vector(cookie->batch, iovector, iovec_entries);
     free(iovector);
+
     if (cookie->flag & STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER)
-      iob_addbuf(current, "\r\n", 2);
+      batch_add_copy(cookie->batch, "\r\n", 2);
   }
 
   if ((cookie->flag & STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER) && cookie->batch && !is_partial) {
-    current = cookie->batch + cookie->batches - 1;
-    iob_addbuf(current, "0\r\n\r\n", 5);
+    batch_add_copy(cookie->batch, "0\r\n\r\n", 5);
     cookie->flag &= ~STRUCT_HTTP_FLAG_CHUNKED_IN_TRANSFER;
   }
 
-  /* writeable sockets timeout after 10 minutes */
-  taia_now(&t);
-  taia_addsec(&t, &t, OT_CLIENT_TIMEOUT_SEND);
-  io_timeout(sock, t);
+  timeout_in(sock, OT_CLIENT_TIMEOUT_SEND);
   io_wantwrite(sock);
   return 0;
 }
@@ -317,7 +286,6 @@ static ssize_t http_handle_stats(const int64_t sock, struct ot_workstruct *ws, c
   }
 
   if (mode == TASK_STATS_TPB) {
-    tai6464 t;
 #ifdef WANT_COMPRESSION_GZIP
     ws->request[ws->request_size - 1] = 0;
 #ifndef WANT_COMPRESSION_GZIP_ALWAYS
@@ -333,8 +301,7 @@ static ssize_t http_handle_stats(const int64_t sock, struct ot_workstruct *ws, c
     cookie->flag |= STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED;
 
     /* Clients waiting for us should not easily timeout */
-    taia_uint(&t, 0);
-    io_timeout(sock, t);
+    timeout_never(sock);
     fullscrape_deliver(sock, format);
     io_dontwantread(sock);
     return ws->reply_size = -2;
@@ -343,13 +310,11 @@ static ssize_t http_handle_stats(const int64_t sock, struct ot_workstruct *ws, c
 
   /* default format for now */
   if ((mode & TASK_CLASS_MASK) == TASK_STATS) {
-    tai6464 t;
 
     cookie->flag |= STRUCT_HTTP_FLAG_WAITINGFORTASK;
 
     /* Complex stats also include expensive memory debugging tools */
-    taia_uint(&t, 0);
-    io_timeout(sock, t);
+    timeout_never(sock);
     stats_deliver(sock, mode);
     return ws->reply_size = -2;
   }
@@ -371,7 +336,6 @@ typedef struct {
 static ssize_t http_handle_fullscrape(const int64_t sock, struct ot_workstruct *ws) {
   struct http_data *cookie = io_getcookie(sock);
   int               format = 0;
-  tai6464           t;
 
 #ifdef WANT_MODEST_FULLSCRAPES
   {
@@ -430,9 +394,8 @@ static ssize_t http_handle_fullscrape(const int64_t sock, struct ot_workstruct *
   /* Pass this task to the worker thread */
   cookie->flag |= STRUCT_HTTP_FLAG_WAITINGFORTASK | STRUCT_HTTP_FLAG_CHUNKED;
   /* Clients waiting for us should not easily timeout */
-  taia_uint(&t, 0);
-  io_timeout(sock, t);
-  fullscrape_deliver(sock, TASK_FULLSCRAPE | format);
+  timeout_never(sock);
+  fullscrape_deliver(sock, format);
   io_dontwantread(sock);
   return ws->reply_size = -2;
 }
@@ -491,7 +454,7 @@ unsigned long long numwants[201];
 #endif
 
 #if defined(WANT_KEEPALIVE) || defined(WANT_IP_FROM_PROXY)
-static char *http_header(char *data, size_t byte_count, char *header) {
+static char *http_header(char *data, size_t byte_count, const char *header) {
   size_t i;
   long   sl = strlen(header);
   for (i = 0; i + sl + 2 < byte_count; ++i) {
diff --git a/ot_http.h b/ot_http.h
index 965820f..003b545 100644
--- a/ot_http.h
+++ b/ot_http.h
@@ -10,9 +10,7 @@
 #include <sys/types.h>
 #include <sys/uio.h>
 
-/* Libowfat */
-#include "iob.h"
-
+#include "ot_batch.h"
 #include "ot_buf.h"
 #include "trackerlogic.h"
 
@@ -27,8 +25,7 @@ typedef enum {
 
 struct http_data {
   ot_buf           request;
-  io_batch        *batch;
-  size_t           batches;
+  ot_batch        *batch;
   ot_ip6           ip;
   STRUCT_HTTP_FLAG flag;
 };
@@ -37,7 +34,7 @@ ssize_t        http_handle_request(const int64_t s, struct ot_workstruct *ws);
 ssize_t        http_sendiovecdata(const int64_t s, struct ot_workstruct *ws, size_t iovec_entries, struct iovec *iovector, int is_partial);
 ssize_t        http_issue_error(const int64_t s, struct ot_workstruct *ws, int code);
 
-extern char   *g_stats_path;
+extern const char *g_stats_path;
 extern ssize_t g_stats_path_len;
 
 #endif
diff --git a/ot_stats.c b/ot_stats.c
index f938e91..6589aef 100644
--- a/ot_stats.c
+++ b/ot_stats.c
@@ -53,7 +53,7 @@ static unsigned long long ot_full_scrape_count;
 static unsigned long long ot_full_scrape_request_count;
 static unsigned long long ot_full_scrape_size;
 static unsigned long long ot_failed_request_counts[CODE_HTTPERROR_COUNT];
-static char              *ot_failed_request_names[] = {
+static const char        *ot_failed_request_names[] = {
     "302 Redirect", "400 Parse Error", "400 Invalid Parameter", "400 Invalid Parameter (compact=0)", "400 Not Modest",
     "402 Payment Required", "403 Access Denied", "404 Not found", "500 Internal Server Error"};
 static unsigned long long ot_renewed[OT_PEER_TIMEOUT];
diff --git a/ot_timeout.h b/ot_timeout.h
new file mode 100644
index 0000000..1027c2e
--- /dev/null
+++ b/ot_timeout.h
@@ -0,0 +1,39 @@
+/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
+   It is considered beerware. Prost. Skol. Cheers or whatever.
+
+   $id$ */
+
+#ifndef OT_TIMEOUT_H__
+#define OT_TIMEOUT_H__
+
+#include <stdint.h>
+#include <string.h>
+
+/* Libowfat */
+#include "io.h"
+
+#include "trackerlogic.h"
+
+/* io_timeout wants an absolute TAI64 deadline and io_timeouted skips any
+   socket whose deadline is zero, so that is how a socket is told to wait
+   indefinitely.
+
+   The deadline is taken from the cached clock rather than from
+   gettimeofday, which is too expensive to call at this rate on FreeBSD. */
+
+static inline void timeout_never(int64_t sock) {
+  tai6464 t;
+
+  memset(&t, 0, sizeof(t));
+  io_timeout(sock, t);
+}
+
+static inline void timeout_in(int64_t sock, unsigned int seconds) {
+  tai6464 t;
+
+  memset(&t, 0, sizeof(t));
+  tai_unix(&t.sec, g_now_seconds + seconds);
+  io_timeout(sock, t);
+}
+
+#endif
diff --git a/proxy.c b/proxy.c
index a870c63..63b4c4f 100644
--- a/proxy.c
+++ b/proxy.c
@@ -16,9 +16,9 @@
 
 /* Libowfat */
 #include "io.h"
-#include "iob.h"
 
 /* Opentracker */
+#include "ot_batch.h"
 #include "ot_ip.h"
 #include "ot_mutex.h"
 #include "ot_random.h"
@@ -72,7 +72,7 @@ static void *livesync_worker(void *args);
 static void *streamsync_worker(void *args);
 static void livesync_proxytell(uint8_t prefix, uint8_t *info_hash, uint8_t *peer);
 
-void exerr(char *message) {
+void exerr(const char *message) {
   fprintf(stderr, "%s\n", message);
   exit(111);
 }
@@ -228,7 +228,7 @@ typedef struct {
   size_t   indata_length;     /* Length of unprocessed data */
   uint32_t tracker_id;        /* How the other end greeted */
   int64_t  fd;                /* A file handle, if connected, <= 0 is disconnected (0 initially, -1 else) */
-  io_batch outdata;           /* The iobatch containing our sync data */
+  ot_batch outdata;           /* The batch containing our sync data */
 
   size_t   packet_tcount;     /* Number of unprocessed torrents in packet we currently receive */
   uint8_t  packet_tprefix;    /* Prefix byte for all torrents in current packet */
@@ -243,7 +243,7 @@ void reset_info_block(proxy_peer *peer) {
   peer->tracker_id    = 0;
   peer->fd            = -1;
   peer->packet_tcount = 0;
-  iob_reset(&peer->outdata);
+  batch_reset(&peer->outdata);
   PROXYPEER_SETDISCONNECTED(peer->state);
 }
 
@@ -424,11 +424,11 @@ static void handle_write(int64_t peersocket) {
     }
     break;
   case FLAG_CONNECTED:
-    switch (iob_send(peersocket, &peer->outdata)) {
-    case 0: /* all data sent */
+    switch (batch_send(peersocket, &peer->outdata)) {
+    case BATCH_DONE:
       io_dontwantwrite(peersocket);
       break;
-    case -3: /* an error occured */
+    case BATCH_FAILED:
       io_close(peersocket);
       reset_info_block(peer);
       break;
@@ -728,7 +728,7 @@ static void *streamsync_worker(void *args) {
             void *tmp = malloc(mem);
             if (tmp) {
               memcpy(tmp, ptr, mem);
-              iob_addbuf_free(&g_connections[i].outdata, tmp, mem);
+              batch_add(&g_connections[i].outdata, tmp, mem);
               io_wantwrite(g_connections[i].fd);
             }
           }
diff --git a/scan_urlencoded_query.h b/scan_urlencoded_query.h
index d0929cf..f71752d 100644
--- a/scan_urlencoded_query.h
+++ b/scan_urlencoded_query.h
@@ -11,8 +11,8 @@
 #include <sys/types.h>
 
 typedef struct {
-  char *key;
-  int   value;
+  const char *key;
+  int         value;
 } ot_keywords;
 
 typedef enum {
diff --git a/tests/runner.c b/tests/runner.c
index db62246..148b928 100644
--- a/tests/runner.c
+++ b/tests/runner.c
@@ -23,6 +23,8 @@ static const struct module modules[] = {{"scan_urlencoded_query", test_scan_urle
                                         {"ot_buf", test_buf},
                                         {"ot_ip", test_ip},
                                         {"ot_socket", test_socket},
+                                        {"ot_mutex", test_mutex},
+                                        {"ot_batch", test_batch},
                                         {"ot_accesslist", test_accesslist},
                                         {NULL, NULL}};
 
diff --git a/tests/stubs.c b/tests/stubs.c
index ce93071..09a7312 100644
--- a/tests/stubs.c
+++ b/tests/stubs.c
@@ -18,3 +18,12 @@ void free_peerlist(ot_peerlist *peer_list) { (void)peer_list; }
    neither of which any test drives. */
 time_t       g_now_seconds;
 volatile int g_opentracker_running;
+
+/* ot_mutex.c wakes the main loop through the self pipe when a result is
+   ready. The queue tests never get that far. */
+int  g_self_pipe[2] = {-1, -1};
+long io_trywrite(long d, const char *buf, long len) {
+  (void)d;
+  (void)buf;
+  return len;
+}
diff --git a/tests/t_batch.c b/tests/t_batch.c
new file mode 100644
index 0000000..02743f1
--- /dev/null
+++ b/tests/t_batch.c
@@ -0,0 +1,348 @@
+/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
+   It is considered beerware. Prost. Skol. Cheers or whatever.
+
+   $id$ */
+
+/* System */
+#include <fcntl.h>
+#include <poll.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+/* Opentracker */
+#include "ot_batch.h"
+#include "test.h"
+
+/* The batch writes to a real socket pair, because what is worth pinning down
+   is what it does when the kernel takes only part of what it was offered. */
+
+static char *dup_of(const char *text) {
+  size_t len  = strlen(text);
+  char  *copy = malloc(len);
+
+  memcpy(copy, text, len);
+  return copy;
+}
+
+static void drain(int s, char *into, size_t len) {
+  size_t got = 0;
+
+  while (got < len) {
+    ssize_t r = read(s, into + got, len - got);
+
+    if (r <= 0)
+      break;
+    got += (size_t)r;
+  }
+  into[got] = 0;
+}
+
+static void case_queue_accounting(void) {
+  ot_batch batch;
+
+  memset(&batch, 0, sizeof(batch));
+
+  test_case("a zeroed batch is empty and sends nothing");
+  CHECK_INT(batch_pending(&batch), 0);
+  CHECK_INT(batch_send(-1, &batch), BATCH_DONE);
+
+  test_case("anything queued makes it pending");
+  CHECK_INT(batch_add(&batch, dup_of("hello "), 6), 0);
+  CHECK_INT(batch_add_copy(&batch, "world", 5), 0);
+  CHECK_INT(batch_pending(&batch), 1);
+  CHECK_INT(batch.count, 2);
+
+  batch_reset(&batch);
+  test_case("and a reset empties it");
+  CHECK_INT(batch_pending(&batch), 0);
+  CHECK_PTR(batch.entries, NULL);
+}
+
+static void case_send_in_one_go(void) {
+  ot_batch batch;
+  int      sv[2];
+  char     got[32];
+
+  memset(&batch, 0, sizeof(batch));
+  CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0);
+
+  test_case("several buffers arrive as one stream, in order");
+  CHECK_INT(batch_add(&batch, dup_of("opentracker"), 11), 0);
+  CHECK_INT(batch_add_copy(&batch, " says ", 6), 0);
+  CHECK_INT(batch_add(&batch, dup_of("hello"), 5), 0);
+  CHECK_INT(batch_send(sv[0], &batch), 22);
+  CHECK_INT(batch_pending(&batch), 0);
+
+  drain(sv[1], got, 22);
+  CHECK_STR(got, "opentracker says hello");
+
+  test_case("and an empty batch reports it has nothing left");
+  CHECK_INT(batch_send(sv[0], &batch), BATCH_DONE);
+
+  batch_reset(&batch);
+  close(sv[0]);
+  close(sv[1]);
+}
+
+/* Send everything the batch holds, reading the other end as we go so the
+   socket never stays full, and return what arrived. */
+static size_t pump(int writer, int reader, ot_batch *batch, char *into, size_t room) {
+  size_t got = 0;
+
+  while (batch_pending(batch) || got < room) {
+    ssize_t r;
+
+    if (batch_pending(batch)) {
+      ssize_t res = batch_send(writer, batch);
+
+      if (res == BATCH_FAILED)
+        break;
+      if (res > 0)
+        continue;
+    }
+
+    r = read(reader, into + got, room - got);
+    if (r <= 0)
+      break;
+    got += (size_t)r;
+  }
+  return got;
+}
+
+static void case_partial_write_resumes(void) {
+  const size_t size = 200000;
+  ot_batch     batch;
+  int          sv[2], sndbuf = 4096;
+  char        *sent, *got;
+  size_t       i, arrived;
+
+  memset(&batch, 0, sizeof(batch));
+  CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0);
+  setsockopt(sv[0], SOL_SOCKET, SO_SNDBUF, &sndbuf, sizeof(sndbuf));
+  fcntl(sv[0], F_SETFL, fcntl(sv[0], F_GETFL, 0) | O_NONBLOCK);
+
+  sent = malloc(size);
+  got  = malloc(size);
+  for (i = 0; i < size; ++i)
+    sent[i] = (char)(i % 251);
+
+  test_case("a buffer the kernel takes in pieces arrives whole and in order");
+  CHECK_INT(batch_add(&batch, memcpy(malloc(size), sent, size), size), 0);
+  arrived = pump(sv[0], sv[1], &batch, got, size);
+  CHECK_INT(arrived, size);
+  CHECK_INT(batch_pending(&batch), 0);
+  CHECK_MEM(got, sent, size);
+
+  free(sent);
+  free(got);
+  batch_reset(&batch);
+  close(sv[0]);
+  close(sv[1]);
+}
+
+/* writev refuses more than IOV_MAX entries at once, so a long batch has to be
+   sent across several calls. */
+static void case_more_entries_than_one_writev(void) {
+  const size_t entries = 3000, each = 4;
+  ot_batch     batch;
+  int          sv[2], sndbuf = 4096;
+  char        *sent, *got;
+  size_t       i, arrived, refused = 0;
+
+  memset(&batch, 0, sizeof(batch));
+  CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0);
+  setsockopt(sv[0], SOL_SOCKET, SO_SNDBUF, &sndbuf, sizeof(sndbuf));
+  fcntl(sv[0], F_SETFL, fcntl(sv[0], F_GETFL, 0) | O_NONBLOCK);
+
+  sent = malloc(entries * each);
+  got  = malloc(entries * each);
+
+  test_case("a batch of more entries than writev accepts still arrives whole");
+  for (i = 0; i < entries; ++i) {
+    char piece[4];
+
+    piece[0] = (char)(i), piece[1] = (char)(i >> 8), piece[2] = 'x', piece[3] = 'y';
+    memcpy(sent + i * each, piece, each);
+    if (batch_add(&batch, memcpy(malloc(each), piece, each), each))
+      ++refused;
+  }
+  CHECK_INT(refused, 0);
+  CHECK_INT(batch.count, entries);
+
+  arrived = pump(sv[0], sv[1], &batch, got, entries * each);
+  CHECK_INT(arrived, entries * each);
+  CHECK_MEM(got, sent, entries * each);
+
+  free(sent);
+  free(got);
+  batch_reset(&batch);
+  close(sv[0]);
+  close(sv[1]);
+}
+
+static void case_reset_frees_the_unsent(void) {
+  ot_batch batch;
+  int      sv[2];
+
+  memset(&batch, 0, sizeof(batch));
+  CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0);
+
+  test_case("a batch torn down mid flight keeps no bytes outstanding");
+  CHECK_INT(batch_add(&batch, dup_of("first"), 5), 0);
+  CHECK_INT(batch_add(&batch, dup_of("second"), 6), 0);
+  CHECK_INT(batch_send(sv[0], &batch), 11);
+  CHECK_INT(batch_add(&batch, dup_of("never sent"), 10), 0);
+  CHECK_INT(batch_pending(&batch), 1);
+
+  /* Whether the unsent buffer is freed is for the sanitizer to say */
+  batch_reset(&batch);
+  CHECK_INT(batch_pending(&batch), 0);
+  CHECK_INT(batch.count, 0);
+
+  close(sv[0]);
+  close(sv[1]);
+}
+
+static void case_closed_socket_is_fatal(void) {
+  ot_batch batch;
+  int      sv[2];
+
+  memset(&batch, 0, sizeof(batch));
+  CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0);
+  close(sv[1]);
+
+  test_case("writing to a socket whose peer is gone reports a dead connection");
+  CHECK_INT(batch_add(&batch, dup_of("nobody is listening"), 19), 0);
+  CHECK_INT(batch_send(sv[0], &batch), BATCH_FAILED);
+
+  batch_reset(&batch);
+  close(sv[0]);
+}
+
+static void case_vector_is_taken_whole(void) {
+  ot_batch      batch;
+  struct iovec *vector = malloc(3 * sizeof(struct iovec));
+  int           sv[2];
+  char          got[32];
+
+  memset(&batch, 0, sizeof(batch));
+  CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0);
+
+  test_case("a whole vector is queued in one go, in order");
+  vector[0].iov_base = dup_of("one "), vector[0].iov_len = 4;
+  vector[1].iov_base = dup_of("two "), vector[1].iov_len = 4;
+  vector[2].iov_base = dup_of("three"), vector[2].iov_len = 5;
+  CHECK_INT(batch_add_vector(&batch, vector, 3), 0);
+  free(vector);
+  CHECK_INT(batch.count, 3);
+  CHECK_INT(batch_send(sv[0], &batch), 13);
+
+  drain(sv[1], got, 13);
+  CHECK_STR(got, "one two three");
+
+  batch_reset(&batch);
+  close(sv[0]);
+  close(sv[1]);
+}
+
+/* The entry array starts small and doubles, so crossing that boundary one
+   buffer at a time is worth its own case. */
+static void case_growing_one_entry_at_a_time(void) {
+  const size_t entries = 100;
+  ot_batch     batch;
+  int          sv[2];
+  char         sent[400], got[401];
+  size_t       i, refused = 0;
+
+  memset(&batch, 0, sizeof(batch));
+  CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0);
+
+  test_case("a batch grown one buffer at a time keeps them all, in order");
+  for (i = 0; i < entries; ++i) {
+    char piece[4];
+
+    piece[0] = (char)('a' + i % 26), piece[1] = (char)('0' + i % 10);
+    piece[2] = (char)('A' + i % 26), piece[3] = '.';
+    memcpy(sent + i * 4, piece, 4);
+    if (batch_add(&batch, memcpy(malloc(4), piece, 4), 4))
+      ++refused;
+  }
+  CHECK_INT(refused, 0);
+  CHECK_INT(batch.count, entries);
+  CHECK_INT(batch_send(sv[0], &batch), (ssize_t)(entries * 4));
+
+  drain(sv[1], got, entries * 4);
+  CHECK_MEM(got, sent, entries * 4);
+
+  batch_reset(&batch);
+  close(sv[0]);
+  close(sv[1]);
+}
+
+/* Appending while earlier entries are still on their way is what a chunked
+   transfer does. Rather than depend on when a kernel decides to block, this
+   takes entries off the front the way a completed send would and checks that
+   the array reuses the space instead of growing. */
+static void case_append_while_sending(void) {
+  const size_t per_round = 12, each = 8;
+  ot_batch     batch;
+  int          sv[2];
+  char         expected[16], got[17];
+  size_t       i, round, refused = 0, space_after_first = 0;
+
+  memset(&batch, 0, sizeof(batch));
+  CHECK_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, sv), 0);
+
+  test_case("an array whose front has gone out is reused, not grown");
+  for (round = 0; round < 8; ++round) {
+    for (i = 0; i < per_round; ++i) {
+      /* Distinct per entry, so a move from the wrong index shows up */
+      char *piece = malloc(each);
+
+      memset(piece, (int)('a' + (round * per_round + i) % 26), each);
+      piece[0] = (char)('0' + i);
+      piece[1] = (char)('0' + round);
+      if (batch_add(&batch, piece, each))
+        ++refused;
+    }
+
+    /* All but the last entry have left and been freed */
+    while (batch.count - batch.first > 1)
+      free(batch.entries[batch.first++].iov_base);
+
+    if (!round)
+      space_after_first = batch.space;
+  }
+  CHECK_INT(refused, 0);
+  CHECK_INT(batch.space, space_after_first);
+
+  test_case("and what is still queued survives the move, in order");
+  CHECK_INT(batch.count - batch.first, 1);
+  CHECK_INT(batch_add(&batch, memset(malloc(each), 'Z', each), each), 0);
+  CHECK_INT(batch_send(sv[0], &batch), (ssize_t)(2 * each));
+
+  memset(expected, (int)('a' + (7 * per_round + per_round - 1) % 26), each);
+  expected[0] = (char)('0' + per_round - 1);
+  expected[1] = '7';
+  memset(expected + each, 'Z', each);
+  drain(sv[1], got, 2 * each);
+  CHECK_MEM(got, expected, 2 * each);
+
+  batch_reset(&batch);
+  close(sv[0]);
+  close(sv[1]);
+}
+
+void test_batch(void) {
+  case_queue_accounting();
+  case_send_in_one_go();
+  case_vector_is_taken_whole();
+  case_growing_one_entry_at_a_time();
+  case_append_while_sending();
+  case_partial_write_resumes();
+  case_more_entries_than_one_writev();
+  case_reset_frees_the_unsent();
+  case_closed_socket_is_fatal();
+}
diff --git a/tests/t_mutex.c b/tests/t_mutex.c
new file mode 100644
index 0000000..3307d70
--- /dev/null
+++ b/tests/t_mutex.c
@@ -0,0 +1,107 @@
+/* This software was written by Dirk Engling <erdgeist@erdgeist.org>
+   It is considered beerware. Prost. Skol. Cheers or whatever.
+
+   $id$ */
+
+/* System */
+#include <stdlib.h>
+#include <string.h>
+
+/* Opentracker */
+#include "ot_mutex.h"
+#include "test.h"
+
+/* A worker asks for a class of task and blocks until one turns up, so a task
+   whose class matches no worker is never claimed and whoever is waiting for
+   its answer waits forever. These cases pin the matching. */
+
+static void case_task_goes_to_its_own_class(void) {
+  ot_tasktype asked;
+  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);
+  asked  = TASK_FULLSCRAPE;
+  taskid = mutex_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);
+}
+
+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);
+  asked = TASK_FULLSCRAPE;
+  CHECK(mutex_workqueue_poptask(&asked) != 0);
+  CHECK_INT(asked, TASK_FULLSCRAPE_TPB_URLENCODED | TASK_FLAG_GZIP);
+  mutex_workqueue_canceltask(42);
+}
+
+static void case_classes_do_not_cross(void) {
+  ot_tasktype asked;
+  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);
+
+  asked  = TASK_STATS;
+  taskid = mutex_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);
+  CHECK(taskid != 0);
+  CHECK_INT(asked, TASK_FULLSCRAPE_TPB_BINARY);
+
+  mutex_workqueue_canceltask(43);
+  mutex_workqueue_canceltask(44);
+}
+
+static void case_every_task_gets_its_own_id(void) {
+  ot_tasktype asked;
+  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);
+
+  asked  = TASK_FULLSCRAPE;
+  first  = mutex_workqueue_poptask(&asked);
+  asked  = TASK_FULLSCRAPE;
+  second = mutex_workqueue_poptask(&asked);
+  CHECK(first != 0);
+  CHECK(second != 0);
+  CHECK(first != second);
+
+  mutex_workqueue_canceltask(45);
+  mutex_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);
+
+  asked = TASK_FULLSCRAPE;
+  CHECK(mutex_workqueue_poptask(&asked) != 0);
+  CHECK_INT(asked, TASK_FULLSCRAPE);
+  mutex_workqueue_canceltask(48);
+}
+
+void test_mutex(void) {
+  case_task_goes_to_its_own_class();
+  case_flags_survive_the_queue();
+  case_classes_do_not_cross();
+  case_every_task_gets_its_own_id();
+  case_cancelled_task_is_not_handed_out();
+}
diff --git a/tests/test.h b/tests/test.h
index d79c4d2..fd574c2 100644
--- a/tests/test.h
+++ b/tests/test.h
@@ -86,5 +86,7 @@ void test_scan(void);
 void test_buf(void);
 void test_ip(void);
 void test_socket(void);
+void test_mutex(void);
+void test_batch(void);
 
 #endif
diff --git a/trackerlogic.c b/trackerlogic.c
index 52c8c18..74fea6d 100644
--- a/trackerlogic.c
+++ b/trackerlogic.c
@@ -10,9 +10,6 @@
 #include <stdlib.h>
 #include <string.h>
 
-/* Libowfat */
-#include "io.h"
-#include "iob.h"
 
 /* Opentracker */
 #include "ot_accesslist.h"
@@ -550,7 +547,7 @@ void trackerlogic_add_random_torrents(size_t amount) {
 }
 #endif
 
-void exerr(char *message) {
+void exerr(const char *message) {
   fprintf(stderr, "%s\n", message);
   exit(111);
 }
diff --git a/trackerlogic.h b/trackerlogic.h
index aa10a54..42018c3 100644
--- a/trackerlogic.h
+++ b/trackerlogic.h
@@ -181,7 +181,7 @@ struct ot_workstruct {
 
 void   trackerlogic_init(void);
 void   trackerlogic_deinit(void);
-void   exerr(char *message);
+void   exerr(const char *message);
 
 /* add_peer_to_torrent does only release the torrent bucket if from_sync is set,
    otherwise it is released in return_peers_for_torrent */