blob: dd6016489c88ec7b638fe6e8f395163057be950b (
plain) (
blame)
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
|
import unittest
from threading import Barrier, Thread
from test.support import threading_helper
threading_helper.requires_working_threading(module=True)
class TestReversed(unittest.TestCase):
@threading_helper.reap_threads
def test_reversed(self):
# Iterating over the iterator with multiple threads should not
# emit TSAN warnings
number_of_iterations = 10
number_of_threads = 10
size = 1_000
barrier = Barrier(number_of_threads)
def work(r):
barrier.wait()
while True:
try:
l = r.__length_hint__()
next(r)
except StopIteration:
break
assert 0 <= l <= size
x = tuple(range(size))
for _ in range(number_of_iterations):
r = reversed(x)
worker_threads = []
for _ in range(number_of_threads):
worker_threads.append(Thread(target=work, args=[r]))
with threading_helper.start_threads(worker_threads):
pass
barrier.reset()
if __name__ == "__main__":
unittest.main()
|
