2 files changed, 215 insertions, 0 deletions

diff --git a/tests/integration/harness.py b/tests/integration/harness.py
index 03a0561..122c36d 100644
--- a/tests/integration/harness.py
+++ b/tests/integration/harness.py
@@ -121,6 +121,7 @@ class Tracker:
         all: LeakSanitizer reports from an exit handler, so a killed tracker
         reports nothing and the suite passes over leaks it never looked for.
         """
+        refused_to_exit = False
         if self.process is not None and self.process.poll() is None:
             self.process.send_signal(signal.SIGINT)
             try:
@@ -128,11 +129,16 @@ class Tracker:
             except subprocess.TimeoutExpired:
                 self.process.kill()
                 self.process.wait()
+                refused_to_exit = True
         complaints = [
             line
             for line in self.output().splitlines()
             if "Sanitizer" in line or "runtime error" in line or line.startswith("ERROR")
         ]
+        if refused_to_exit:
+            # Killing it throws away whatever the sanitizer had to say on the
+            # way out, so a test that ends this way has checked nothing.
+            complaints.append("the tracker did not exit within 5 s of SIGINT and had to be killed")
         self._cleanup()
         return complaints
 
diff --git a/tests/integration/test_concurrency.py b/tests/integration/test_concurrency.py
new file mode 100644
index 0000000..1374852
--- /dev/null
+++ b/tests/integration/test_concurrency.py
@@ -0,0 +1,209 @@
+"""Several clients at once.
+
+The rest of the suite talks to the tracker one request at a time, which
+leaves the worker threads and the bucket locks almost untouched. These reach
+them.
+
+What they assert is weak on purpose. A lost update needs two threads inside
+the same few instructions, which is rare enough that these tests pass
+against code that has the race: measured, eight runs in a row. Their worth
+is that they make the racing code run at all, so that "make check-tsan"
+has something to report. Reaching the code and judging it are two jobs, and
+these only do the first.
+"""
+
+import struct
+import threading
+import unittest
+
+from harness import TrackerTestCase, bdecode
+
+THREADS = 8
+PER_THREAD = 40
+
+
+def run_threads(target, count=THREADS):
+    errors = []
+
+    def guarded(index):
+        try:
+            target(index)
+        except Exception as exception:  # noqa: BLE001 - reported, not swallowed
+            errors.append("thread %d: %r" % (index, exception))
+
+    threads = [threading.Thread(target=guarded, args=(i,)) for i in range(count)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+    return errors
+
+
+class ConcurrentAnnounce(TrackerTestCase):
+    def test_every_peer_of_one_swarm_arrives(self):
+        """One torrent, one bucket, every thread contending for its lock."""
+        info_hash = self.hash_for("concurrent-one")
+
+        def announce(index):
+            for i in range(PER_THREAD):
+                self.tracker.announce(info_hash, port=10000 + index * PER_THREAD + i, left=0)
+
+        self.assertEqual(run_threads(announce), [])
+        answer = self.tracker.scrape(info_hash)
+        self.assertEqual(answer[b"files"][info_hash][b"complete"], THREADS * PER_THREAD)
+
+    def test_inserts_spread_over_buckets(self):
+        """Every thread inserts into a different bucket, so the bucket locks
+        serialise nothing between them, and the counter they all adjust is
+        the one global the locks do not cover. Under check-tsan this is what
+        names mutex_bucket_unlock if that counter stops being atomic."""
+
+        def announce(index):
+            for i in range(PER_THREAD):
+                self.tracker.announce(struct.pack(">HH", index, i) + b"\0" * 16, port=11000 + i, left=0)
+
+        self.assertEqual(run_threads(announce), [])
+        self.assertEqual(self.tracker.torrent_count(), THREADS * PER_THREAD)
+
+    def test_a_swarm_that_fills_and_empties_at_once(self):
+        """Inserts and removals racing in the same buckets."""
+        info_hash = self.hash_for("concurrent-churn")
+        for port in range(12000, 12000 + THREADS * PER_THREAD):
+            self.tracker.announce(info_hash, port=port, left=0)
+
+        def churn(index):
+            base = 12000 + index * PER_THREAD
+            for i in range(PER_THREAD):
+                self.tracker.announce(info_hash, port=base + i, left=0, event="stopped")
+                self.tracker.announce(info_hash, port=base + i, left=100)
+
+        self.assertEqual(run_threads(churn), [])
+        answer = self.tracker.scrape(info_hash)
+        self.assertEqual(answer[b"files"][info_hash][b"incomplete"], THREADS * PER_THREAD)
+        self.assertEqual(answer[b"files"][info_hash][b"complete"], 0)
+
+
+class ConcurrentMixedBearers(TrackerTestCase):
+    def test_tcp_and_udp_reach_the_same_swarm(self):
+        info_hash = self.hash_for("concurrent-mixed")
+
+        def over_tcp(index):
+            for i in range(PER_THREAD):
+                self.tracker.announce(info_hash, port=13000 + index * PER_THREAD + i, left=0)
+
+        def over_udp(index):
+            sock, connection_id = self.tracker.udp_connect()
+            try:
+                for i in range(PER_THREAD):
+                    self.tracker.udp_announce(
+                        sock, connection_id, info_hash, left=0, port=14000 + index * PER_THREAD + i
+                    )
+            finally:
+                sock.close()
+
+        errors = []
+        threads = [threading.Thread(target=lambda i=i: over_tcp(i)) for i in range(THREADS // 2)]
+        threads += [threading.Thread(target=lambda i=i: over_udp(i)) for i in range(THREADS // 2)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+
+        self.assertEqual(errors, [])
+        answer = self.tracker.scrape(info_hash)
+        self.assertEqual(answer[b"files"][info_hash][b"complete"], THREADS * PER_THREAD)
+
+
+class ConcurrentWorkerTasks(TrackerTestCase):
+    def test_full_scrapes_while_the_swarm_moves(self):
+        """The full scrape runs on a worker thread, walking the same buckets
+        the announces are changing."""
+        for i in range(200):
+            self.tracker.announce(struct.pack(">I", i) + b"\0" * 16, port=15000 + i, left=0)
+
+        def announce(index):
+            for i in range(PER_THREAD):
+                self.tracker.announce(struct.pack(">I", i) + b"\0" * 16, port=16000 + index * PER_THREAD + i, left=0)
+
+        def scrape(index):
+            for _ in range(6):
+                answer = bdecode(self.tracker.get("/scrape"))
+                for counts in answer[b"files"].values():
+                    if counts[b"complete"] < 0 or counts[b"incomplete"] < 0:
+                        raise AssertionError("negative count in a full scrape: %r" % counts)
+
+        threads = [threading.Thread(target=announce, args=(i,)) for i in range(THREADS // 2)]
+        threads += [threading.Thread(target=scrape, args=(i,)) for i in range(2)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+
+        self.assertEqual(self.tracker.torrent_count(), 200)
+
+    def test_stats_while_the_swarm_moves(self):
+        def announce(index):
+            for i in range(PER_THREAD):
+                self.tracker.announce(struct.pack(">HH", index, i) + b"\0" * 16, port=17000 + i, left=i % 2)
+
+        def poll(index):
+            for _ in range(10):
+                for mode in ("everything", "top100", "peer", "torr", "s24s"):
+                    self.tracker.stats(mode)
+
+        threads = [threading.Thread(target=announce, args=(i,)) for i in range(THREADS // 2)]
+        threads += [threading.Thread(target=poll, args=(i,)) for i in range(2)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+
+        self.assertEqual(self.tracker.torrent_count(), (THREADS // 2) * PER_THREAD)
+
+
+class CountersAcrossContexts(TrackerTestCase):
+    """With udp workers configured, every one of them counts into a block of
+    its own, and the figure reported is those blocks added up. A registry
+    that lost a block, or a worker that wrote into somebody else's, would
+    show here and nowhere else in the suite: without this directive udp is
+    served by the main loop, which is a single context."""
+
+    config = "listen.udp.workers 4\n"
+
+    def test_the_udp_count_is_the_sum_of_every_worker(self):
+        before = int(self.tracker.stats("udp4").splitlines()[0])
+
+        packets = 0
+        for round_ in range(6):
+            sock, connection_id = self.tracker.udp_connect()  # one packet
+            packets += 1
+            try:
+                for i in range(10):
+                    self.tracker.udp_announce(sock, connection_id, self.hash_for("ctx%d" % i), left=0)
+                    packets += 1
+            finally:
+                sock.close()
+
+        after = int(self.tracker.stats("udp4").splitlines()[0])
+        self.assertEqual(after - before, packets, "every packet counted exactly once")
+
+    def test_announces_from_several_workers_reach_one_swarm(self):
+        info_hash = self.hash_for("ctx-swarm")
+
+        def announce(index):
+            sock, connection_id = self.tracker.udp_connect()
+            try:
+                for i in range(PER_THREAD):
+                    self.tracker.udp_announce(
+                        sock, connection_id, info_hash, left=0, port=18000 + index * PER_THREAD + i
+                    )
+            finally:
+                sock.close()
+
+        self.assertEqual(run_threads(announce, 4), [])
+        answer = self.tracker.scrape(info_hash)
+        self.assertEqual(answer[b"files"][info_hash][b"complete"], 4 * PER_THREAD)
+
+
+if __name__ == "__main__":
+    unittest.main()