Test joining a string of characters being replaced with ephemeral strings by another thread.
(self)
| 35 | reader.join() |
| 36 | |
| 37 | def test_racing_join_replace(self): |
| 38 | ''' |
| 39 | Test joining a string of characters being replaced with ephemeral |
| 40 | strings by another thread. |
| 41 | ''' |
| 42 | l = [*'abcdefg'] |
| 43 | MAX_ORDINAL = 1_000 |
| 44 | READERS = 10 |
| 45 | done_event = Event() |
| 46 | |
| 47 | def writer_func(): |
| 48 | for i, c in zip(cycle(range(len(l))), |
| 49 | map(chr, range(128, MAX_ORDINAL))): |
| 50 | l[i] = c |
| 51 | done_event.set() |
| 52 | |
| 53 | def reader_func(): |
| 54 | while not done_event.is_set(): |
| 55 | ''.join(l) |
| 56 | ''.join(l) |
| 57 | ''.join(l) |
| 58 | ''.join(l) |
| 59 | |
| 60 | writer = Thread(target=writer_func) |
| 61 | readers = [] |
| 62 | for x in range(READERS): |
| 63 | reader = Thread(target=reader_func) |
| 64 | readers.append(reader) |
| 65 | reader.start() |
| 66 | |
| 67 | writer.start() |
| 68 | writer.join() |
| 69 | for reader in readers: |
| 70 | reader.join() |
| 71 | |
| 72 | def test_maketrans_dict_concurrent_modification(self): |
| 73 | for _ in range(5): |