1 files changed, 101 insertions, 70 deletions

diff --git a/tests/t_mutex.c b/tests/t_mutex.c
index 3307d70..7c1463b 100644
--- a/tests/t_mutex.c
+++ b/tests/t_mutex.c
@@ -4,104 +4,135 @@
    $id$ */
 
 /* System */
-#include <stdlib.h>
+#include <pthread.h>
+#include <stdatomic.h>
 #include <string.h>
+#include <unistd.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. */
+/* Every announce passes through here, and what it has to get right is which
+   bucket a hash belongs to and that two threads never hold one bucket at
+   once. Both are checked against real mutexes rather than a stand-in. */
 
-static void case_task_goes_to_its_own_class(void) {
-  ot_tasktype asked;
-  ot_taskid   taskid;
+/* A hash that lands in a chosen bucket: the bucket is the top
+   OT_BUCKET_COUNT_BITS of the first four bytes, read in network order */
+static void hash_for_bucket(ot_hash hash, int bucket) {
+  memset(hash, 0, sizeof(ot_hash));
+  hash[0] = (uint8_t)(bucket >> (OT_BUCKET_COUNT_BITS - 8));
+  hash[1] = (uint8_t)((bucket << (16 - OT_BUCKET_COUNT_BITS)) & 0xff);
+}
+
+static void case_a_hash_names_its_bucket(void) {
+  static const int buckets[] = {0, 1, 2, 17, 511, 512, OT_BUCKET_COUNT - 1};
+  size_t           i;
+
+  test_case("a hash is served by the bucket its leading bits name");
+  for (i = 0; i < sizeof(buckets) / sizeof(buckets[0]); ++i) {
+    ot_hash    hash;
+    ot_vector *by_hash, *by_number;
+
+    hash_for_bucket(hash, buckets[i]);
+    by_hash = mutex_bucket_lock_by_hash(hash);
+    mutex_bucket_unlock_by_hash(hash, 0);
 
-  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);
+    by_number = mutex_bucket_lock(buckets[i]);
+    mutex_bucket_unlock(buckets[i], 0);
 
-  test_case("and the full task type, not just its class");
-  CHECK_INT(asked, TASK_FULLSCRAPE_TPB_ASCII);
-  mutex_workqueue_canceltask(41);
+    CHECK(by_hash == by_number);
+  }
 }
 
-static void case_flags_survive_the_queue(void) {
-  ot_tasktype asked;
+static void case_every_bucket_is_its_own(void) {
+  ot_vector *first, *last, *middle;
 
-  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);
+  test_case("no two buckets are the same one");
+  first  = mutex_bucket_lock(0);
+  mutex_bucket_unlock(0, 0);
+  middle = mutex_bucket_lock(OT_BUCKET_COUNT / 2);
+  mutex_bucket_unlock(OT_BUCKET_COUNT / 2, 0);
+  last   = mutex_bucket_lock(OT_BUCKET_COUNT - 1);
+  mutex_bucket_unlock(OT_BUCKET_COUNT - 1, 0);
+
+  CHECK(first != middle);
+  CHECK(middle != last);
+  CHECK(first != last);
 }
 
-static void case_classes_do_not_cross(void) {
-  ot_tasktype asked;
-  ot_taskid   taskid;
+static void case_the_count_follows_what_unlock_is_told(void) {
+  size_t before = mutex_get_torrent_count();
+
+  test_case("the torrent count moves by what unlocking reports, and not otherwise");
+  mutex_bucket_lock(3);
+  mutex_bucket_unlock(3, 0);
+  CHECK_INT((int)(mutex_get_torrent_count() - before), 0);
 
-  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);
+  mutex_bucket_lock(3);
+  mutex_bucket_unlock(3, 5);
+  CHECK_INT((int)(mutex_get_torrent_count() - before), 5);
 
-  asked  = TASK_STATS;
-  taskid = mutex_workqueue_poptask(&asked);
-  CHECK(taskid != 0);
-  CHECK_INT(asked, TASK_STATS_PEERS);
+  mutex_bucket_lock(3);
+  mutex_bucket_unlock(3, -2);
+  CHECK_INT((int)(mutex_get_torrent_count() - before), 3);
+}
 
-  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);
+static _Atomic int g_other_got_in;
 
-  mutex_workqueue_canceltask(43);
-  mutex_workqueue_canceltask(44);
+static void *take_a_bucket(void *args) {
+  int bucket = (int)(intptr_t)args;
+
+  mutex_bucket_lock(bucket);
+  atomic_store(&g_other_got_in, 1);
+  mutex_bucket_unlock(bucket, 0);
+  return NULL;
+}
+
+static int waited_for_entry(int patience_ms) {
+  int waited;
+
+  for (waited = 0; waited < patience_ms && !atomic_load(&g_other_got_in); ++waited)
+    usleep(1000);
+  return atomic_load(&g_other_got_in);
 }
 
-static void case_every_task_gets_its_own_id(void) {
-  ot_tasktype asked;
-  ot_taskid   first, second;
+static void case_a_held_bucket_keeps_others_out(void) {
+  pthread_t other;
 
-  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);
+  test_case("a second thread waits for the bucket this one holds");
+  atomic_store(&g_other_got_in, 0);
+  mutex_bucket_lock(9);
+  CHECK_INT(pthread_create(&other, NULL, take_a_bucket, (void *)(intptr_t)9), 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);
+  CHECK_INT(waited_for_entry(200), 0);
 
-  mutex_workqueue_canceltask(45);
-  mutex_workqueue_canceltask(46);
+  mutex_bucket_unlock(9, 0);
+  CHECK_INT(waited_for_entry(2000), 1);
+  pthread_join(other, NULL);
 }
 
-static void case_cancelled_task_is_not_handed_out(void) {
-  ot_tasktype asked;
+static void case_another_bucket_is_free_meanwhile(void) {
+  pthread_t other;
 
-  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);
+  test_case("and a different bucket is not held up by it");
+  atomic_store(&g_other_got_in, 0);
+  mutex_bucket_lock(9);
+  CHECK_INT(pthread_create(&other, NULL, take_a_bucket, (void *)(intptr_t)10), 0);
 
-  asked = TASK_FULLSCRAPE;
-  CHECK(mutex_workqueue_poptask(&asked) != 0);
-  CHECK_INT(asked, TASK_FULLSCRAPE);
-  mutex_workqueue_canceltask(48);
+  CHECK_INT(waited_for_entry(2000), 1);
+  pthread_join(other, NULL);
+  mutex_bucket_unlock(9, 0);
 }
 
 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();
+  mutex_init();
+
+  case_a_hash_names_its_bucket();
+  case_every_bucket_is_its_own();
+  case_the_count_follows_what_unlock_is_told();
+  case_a_held_bucket_keeps_others_out();
+  case_another_bucket_is_free_meanwhile();
+
+  mutex_deinit();
 }