1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
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()
|
