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