Test joining a string being extended by another thread
(self)
| 9 | @threading_helper.requires_working_threading() |
| 10 | class TestStr(TestCase): |
| 11 | def test_racing_join_extend(self): |
| 12 | '''Test joining a string being extended by another thread''' |
| 13 | l = [] |
| 14 | ITERS = 100 |
| 15 | READERS = 10 |
| 16 | done_event = Event() |
| 17 | def writer_func(): |
| 18 | for i in range(ITERS): |
| 19 | l.extend(map(str, range(i))) |
| 20 | l.clear() |
| 21 | done_event.set() |
| 22 | def reader_func(): |
| 23 | while not done_event.is_set(): |
| 24 | ''.join(l) |
| 25 | writer = Thread(target=writer_func) |
| 26 | readers = [] |
| 27 | for x in range(READERS): |
| 28 | reader = Thread(target=reader_func) |
| 29 | readers.append(reader) |
| 30 | reader.start() |
| 31 | |
| 32 | writer.start() |
| 33 | writer.join() |
| 34 | for reader in readers: |
| 35 | reader.join() |
| 36 | |
| 37 | def test_racing_join_replace(self): |
| 38 | ''' |