From 36c21130fa4fd286e7515d5f6d429df983d253d5 Mon Sep 17 00:00:00 2001 From: Dirk Engling Date: Tue, 27 Oct 2026 21:36:12 +0100 Subject: Add concurrency test --- tests/integration/test_concurrency.py | 209 ++++++++++++++++++++++++++++++++++ 1 file changed, 209 insertions(+) create mode 100644 tests/integration/test_concurrency.py (limited to 'tests/integration') 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() -- cgit v1.3