7 files changed, 587 insertions, 19 deletions

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/t_socket.c b/tests/t_socket.c
index bbc9e5e..8eef7fd 100644
--- a/tests/t_socket.c
+++ b/tests/t_socket.c
@@ -91,14 +91,14 @@ static void case_socket_options(void) {
   int udp, tcp;
 
   test_case("a bound udp socket is non blocking and serves both families");
-  udp = socket_udp_bind(ip_any, 0);
+  udp = socket_udp_bind(ip_any, 0, SOCKET_EXCLUSIVE);
   CHECK(udp != -1);
   CHECK_INT(is_nonblocking(udp), 1);
   CHECK_INT(sock_flag(udp, IPPROTO_IPV6, IPV6_V6ONLY), 0);
   close(udp);
 
   test_case("a listening tcp socket is non blocking and serves both families");
-  tcp = socket_tcp_listen(ip_any, 0);
+  tcp = socket_tcp_listen(ip_any, 0, SOCKET_EXCLUSIVE);
   CHECK(tcp != -1);
   CHECK_INT(is_nonblocking(tcp), 1);
   CHECK_INT(sock_flag(tcp, IPPROTO_IPV6, IPV6_V6ONLY), 0);
@@ -106,24 +106,20 @@ static void case_socket_options(void) {
 }
 
 #ifdef SO_REUSEPORT
-static void case_reuseport_is_opt_in(void) {
+static void case_sharing_is_opt_in(void) {
   int s;
 
-  test_case("reuseport stays off unless the operator asks for it");
-  g_socket_reuseport = 0;
-  s                  = socket_udp_bind(ip_loopback, 0);
+  test_case("an address is exclusive unless sharing is asked for");
+  s = socket_udp_bind(ip_loopback, 0, SOCKET_EXCLUSIVE);
   CHECK(s != -1);
   CHECK_INT(sock_flag(s, SOL_SOCKET, SO_REUSEPORT), 0);
   close(s);
 
-  test_case("reuseport reaches the socket when it is asked for");
-  g_socket_reuseport = 1;
-  s                  = socket_udp_bind(ip_loopback, 0);
+  test_case("and shared when it is");
+  s = socket_udp_bind(ip_loopback, 0, SOCKET_SHARED);
   CHECK(s != -1);
   CHECK_INT(sock_flag(s, SOL_SOCKET, SO_REUSEPORT), 1);
   close(s);
-
-  g_socket_reuseport = 0;
 }
 #endif
 
@@ -135,7 +131,7 @@ static void case_udp_roundtrip(void) {
   int      s;
 
   test_case("a datagram comes back with the sender's address and port");
-  s = socket_udp_bind(ip_loopback, 0);
+  s = socket_udp_bind(ip_loopback, 0, SOCKET_EXCLUSIVE);
   CHECK(s != -1);
   bound_address(s, own_ip, &own_port);
   CHECK(own_port != 0);
@@ -158,9 +154,9 @@ static void case_udp_v4mapped(void) {
   int      receiver, sender;
 
   test_case("an IPv4 sender is reported in its v4 mapped form");
-  receiver = socket_udp_bind(ip_v4_loopback, 0);
+  receiver = socket_udp_bind(ip_v4_loopback, 0, SOCKET_EXCLUSIVE);
   CHECK(receiver != -1);
-  sender = socket_udp_bind(ip_v4_loopback, 0);
+  sender = socket_udp_bind(ip_v4_loopback, 0, SOCKET_EXCLUSIVE);
   CHECK(sender != -1);
   bound_address(receiver, receiver_ip, &receiver_port);
   bound_address(sender, sender_ip, &sender_port);
@@ -184,7 +180,7 @@ static void case_tcp_roundtrip(void) {
   int      listener, client, accepted;
 
   test_case("a connection is accepted and carries the caller's address");
-  listener = socket_tcp_listen(ip_loopback, 0);
+  listener = socket_tcp_listen(ip_loopback, 0, SOCKET_EXCLUSIVE);
   CHECK(listener != -1);
   bound_address(listener, listener_ip, &listener_port);
   CHECK(listener_port != 0);
@@ -216,7 +212,7 @@ static void case_connect_refused(void) {
 
   test_case("a refused connection is reported by socket_is_connected");
   /* Take a port, then drop it, so nothing is listening on it any more */
-  listener = socket_tcp_listen(ip_loopback, 0);
+  listener = socket_tcp_listen(ip_loopback, 0, SOCKET_EXCLUSIVE);
   CHECK(listener != -1);
   bound_address(listener, listener_ip, &port);
   close(listener);
@@ -247,8 +243,8 @@ static void case_bind_failure(void) {
   test_case("binding an address no interface carries fails");
   before = next_descriptor();
   CHECK(before != -1);
-  CHECK_INT(socket_tcp_listen(ip_unassigned, 0), -1);
-  CHECK_INT(socket_udp_bind(ip_unassigned, 0), -1);
+  CHECK_INT(socket_tcp_listen(ip_unassigned, 0, SOCKET_EXCLUSIVE), -1);
+  CHECK_INT(socket_udp_bind(ip_unassigned, 0, SOCKET_EXCLUSIVE), -1);
 
   test_case("a failed bind leaves no socket behind");
   CHECK_INT(next_descriptor(), before);
@@ -294,7 +290,7 @@ static void case_mcast_sender(void) {
 void test_socket(void) {
   case_socket_options();
 #ifdef SO_REUSEPORT
-  case_reuseport_is_opt_in();
+  case_sharing_is_opt_in();
 #endif
   case_udp_roundtrip();
   case_udp_v4mapped();
diff --git a/tests/t_vector.c b/tests/t_vector.c
index fee336b..a8186b2 100644
--- a/tests/t_vector.c
+++ b/tests/t_vector.c
@@ -4,10 +4,12 @@
    $id$ */
 
 /* System */
+#include <stdint.h>
 #include <stdlib.h>
 #include <string.h>
 
 /* Opentracker */
+#include "ot_vector.h"
 #include "trackerlogic.h"
 #include "test.h"
 
@@ -149,6 +151,105 @@ static void case_insert_compare_prefix(void) {
   free(vector.data);
 }
 
+/* ------------------------------------------------- bucket redistribution */
+
+/* A flat peer list of count distinct v4 peers. The low byte of the index goes
+   into the first address byte, so the peers are distinct but arrive in no
+   particular order: a list generated in ascending order would come back
+   sorted whether or not vector_redistribute_buckets sorted it. */
+static void build_flat_peerlist(ot_peerlist *peer_list, size_t count) {
+  size_t i;
+
+  memset(peer_list, 0, sizeof(*peer_list));
+  peer_list->peers.data  = malloc(count * OT_PEER_SIZE4);
+  peer_list->peers.size  = count;
+  peer_list->peers.space = count;
+  peer_list->peer_count  = count;
+
+  for (i = 0; i < count; ++i) {
+    uint8_t *peer = (uint8_t *)peer_list->peers.data + i * OT_PEER_SIZE4;
+
+    memset(peer, 0, OT_PEER_SIZE4);
+    peer[0] = (uint8_t)i;
+    peer[1] = (uint8_t)(i >> 8);
+    peer[2] = (uint8_t)(i >> 16);
+    peer[3] = 1;
+  }
+}
+
+static void case_redistribute_holds_every_peer(void) {
+  const size_t count = 513;
+  ot_peerlist  peer_list;
+  size_t       bucket, total_size = 0, empty = 0, largest = 0;
+  ot_vector   *buckets;
+
+  test_case("a peer list past the threshold is split into buckets");
+  build_flat_peerlist(&peer_list, count);
+  CHECK(peer_list.peers.data != NULL);
+  vector_redistribute_buckets(&peer_list, OT_PEER_SIZE4);
+  CHECK_INT(OT_PEERLIST_HASBUCKETS(&peer_list), 1);
+  CHECK_INT(peer_list.peers.size, 16);
+
+  buckets = peer_list.peers.data;
+  for (bucket = 0; bucket < peer_list.peers.size; ++bucket) {
+    total_size += buckets[bucket].size;
+    if (!buckets[bucket].size)
+      ++empty;
+    if (buckets[bucket].size > largest)
+      largest = buckets[bucket].size;
+  }
+
+  test_case("no peer is lost on the way");
+  CHECK_INT(total_size, count);
+
+  /* The point of the exercise is that a lookup only searches one bucket, so
+     a hash that piled everything into one would defeat it. The input and the
+     hash are both fixed, so this is not a matter of luck. */
+  test_case("and they are spread over the buckets rather than piled up");
+  CHECK_INT(empty, 0);
+  CHECK(largest < count / 2);
+
+  vector_clean_list(peer_list.peers.data, peer_list.peers.size);
+}
+
+static void case_redistribute_sorts_each_bucket(void) {
+  const size_t count = 1000;
+  ot_peerlist  peer_list;
+  size_t       bucket, i, out_of_order = 0;
+  ot_vector   *buckets;
+
+  test_case("every bucket comes back sorted, so bsearch can be used on it");
+  build_flat_peerlist(&peer_list, count);
+  CHECK(peer_list.peers.data != NULL);
+  vector_redistribute_buckets(&peer_list, OT_PEER_SIZE4);
+  CHECK_INT(OT_PEERLIST_HASBUCKETS(&peer_list), 1);
+
+  buckets = peer_list.peers.data;
+  for (bucket = 0; bucket < peer_list.peers.size; ++bucket)
+    for (i = 1; i < buckets[bucket].size; ++i) {
+      const uint8_t *prev = (uint8_t *)buckets[bucket].data + (i - 1) * OT_PEER_SIZE4;
+      const uint8_t *next = (uint8_t *)buckets[bucket].data + i * OT_PEER_SIZE4;
+
+      if (memcmp(prev, next, OT_PEER_COMPARE_SIZE4) >= 0)
+        ++out_of_order;
+    }
+  CHECK_INT(out_of_order, 0);
+
+  vector_clean_list(peer_list.peers.data, peer_list.peers.size);
+}
+
+static void case_redistribute_below_threshold(void) {
+  ot_peerlist peer_list;
+
+  test_case("a small peer list is left flat");
+  build_flat_peerlist(&peer_list, 100);
+  CHECK(peer_list.peers.data != NULL);
+  vector_redistribute_buckets(&peer_list, OT_PEER_SIZE4);
+  CHECK_INT(OT_PEERLIST_HASBUCKETS(&peer_list), 0);
+  CHECK_INT(peer_list.peers.size, 100);
+  free(peer_list.peers.data);
+}
+
 void test_vector(void) {
   case_search_finds_member();
   case_search_finds_first_and_last();
@@ -158,4 +259,7 @@ void test_vector(void) {
   case_insert_keeps_order();
   case_insert_finds_existing();
   case_insert_compare_prefix();
+  case_redistribute_holds_every_peer();
+  case_redistribute_sorts_each_bucket();
+  case_redistribute_below_threshold();
 }
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