"""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()