(self)
| 55 | self.assertEqual(counter[letter], 3 * NTHREADS) |
| 56 | |
| 57 | def test_readline(self): |
| 58 | num_lines = 1000 |
| 59 | lines_read_from_mmap = [] |
| 60 | expected_lines = [] |
| 61 | |
| 62 | def readline(mm_obj): |
| 63 | for i in range(num_lines // NTHREADS): |
| 64 | line = mm_obj.readline() |
| 65 | lines_read_from_mmap.append(line) |
| 66 | |
| 67 | # Allocate mmap enough for num_lines (max line 5 bytes including NL) |
| 68 | with mmap.mmap(ANONYMOUS_MEM, num_lines * 5) as mm_obj: |
| 69 | for i in range(num_lines): |
| 70 | line = b"%d\n" % i |
| 71 | mm_obj.write(line) |
| 72 | expected_lines.append(line) |
| 73 | |
| 74 | mm_obj.seek(0) |
| 75 | run_concurrently( |
| 76 | worker_func=readline, |
| 77 | args=(mm_obj,), |
| 78 | nthreads=NTHREADS, |
| 79 | ) |
| 80 | |
| 81 | self.assertEqual(len(lines_read_from_mmap), num_lines) |
| 82 | # Every line should be read once by threads; order is non-deterministic |
| 83 | # Sort numerically by integer value |
| 84 | lines_read_from_mmap.sort(key=lambda x: int(x)) |
| 85 | self.assertEqual(lines_read_from_mmap, expected_lines) |
| 86 | |
| 87 | def test_write_and_write_byte(self): |
| 88 | thread_letters = list(string.ascii_uppercase) |
nothing calls this directly
no test coverage detected