4 files changed, 94 insertions, 27 deletions

diff --git a/tests/integration/test_tcpworkers.py b/tests/integration/test_tcpworkers.py
new file mode 100644
index 0000000..e43612a
--- /dev/null
+++ b/tests/integration/test_tcpworkers.py
@@ -0,0 +1,67 @@
+"""The tracker with its tcp side handed to worker threads.
+
+With workers configured, a connection is accepted and answered on a worker's
+own stack. Anything it cannot finish there is copied to the heap and handed
+to the event loop, which is the one place where a worker and the main thread
+touch the same structures, so these drive all three outcomes: answered
+inline, held open, and parked waiting for another thread entirely.
+"""
+
+import threading
+import unittest
+import urllib.parse
+
+from harness import TrackerTestCase, bdecode
+
+WORKERS = 4
+
+
+class TcpWorkers(TrackerTestCase):
+    config = "listen.tcp.workers %d\n" % WORKERS
+
+    def announce_path(self, name, port=6881):
+        return "/announce?info_hash=%s&peer_id=%s&port=%d&left=0" % (
+            urllib.parse.quote_from_bytes(self.hash_for(name), safe=""),
+            "x" * 20,
+            port,
+        )
+
+    def test_many_at_once_are_all_answered(self):
+        """Every worker gets something to do, and the swarm adds up."""
+        info_hash = self.hash_for("workers-swarm")
+        errors = []
+
+        def announce(index):
+            try:
+                self.tracker.announce(info_hash, peer_id=b"-OT0001-%012d" % index, port=7000 + index, left=0)
+            except Exception as problem:  # noqa: BLE001 - reported, not swallowed
+                errors.append("%d: %r" % (index, problem))
+
+        threads = [threading.Thread(target=announce, args=(i,)) for i in range(32)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+
+        self.assertEqual(errors, [], "\n".join(errors))
+        self.assertEqual(self.tracker.scrape(info_hash)[b"files"][info_hash][b"complete"], 32)
+
+    def test_a_kept_connection_outlives_the_worker_that_took_it(self):
+        """Keep-alive is the plain case of a connection a worker hands on."""
+        connection = self.tracker.connect()
+        try:
+            for i in range(4):
+                connection.send(self.announce_path("workers-kept"))
+                status, _, body = connection.reply()
+                self.assertIn(b"200", status)
+                self.assertIn(b"complete", bdecode(body))
+        finally:
+            connection.close()
+
+    def test_a_full_scrape_asked_for_by_a_worker_still_arrives(self):
+        """The reply is built by another thread, so the worker parks the
+           connection waiting for nothing at all before handing it over."""
+        for i in range(20):
+            self.tracker.announce(bytes([i]) + b"\0" * 19, port=7100 + i, left=0)
+        body = self.tracker.get("/scrape")
+        self.assertEqual(len(bdecode(body)[b"files"]), 20)
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);