Make tcp worker multithreaded, too

This shamelessly steals the model established in the UDP workers, i.e.
if the kernel is able to bind multiple listening sockets to the same
endpoint, we will share them across threads. If it can't, all threads
will accept on the same socket.

The fast path handles most of the connections, only the rare case of
no data yet on the socket is offloaded to the slow path (which used to
be the mainloop) for completion. This scales nicely, as the higher the
load is, the higher chances are that an accept happens only after the
network roundtrip to the peer completes, leading to more connections
that already have data waiting for us.

We also consolidate the workstruct init and deinit code, preparation
for later unifying all thread state structs for all protocols (mca,
udc and tcp).

The config file gains a per tcp connection worker count and some docs
explaining what this knob buys and costs, nudging the user to either
keep it 0 or dial it up to at least 2.

1 files changed, 67 insertions, 0 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)