(self)
| 19 | @threading_helper.requires_working_threading() |
| 20 | class MmapTests(unittest.TestCase): |
| 21 | def test_read_and_read_byte(self): |
| 22 | ascii_uppercase = string.ascii_uppercase.encode() |
| 23 | # Choose a total mmap size that evenly divides across threads and the |
| 24 | # read pattern (3 bytes per loop). |
| 25 | mmap_size = 3 * NTHREADS * len(ascii_uppercase) |
| 26 | num_bytes_to_read_per_thread = mmap_size // NTHREADS |
| 27 | bytes_read_from_mmap = [] |
| 28 | |
| 29 | def read(mm_obj): |
| 30 | nread = 0 |
| 31 | while nread < num_bytes_to_read_per_thread: |
| 32 | b = mm_obj.read_byte() |
| 33 | bytes_read_from_mmap.append(b) |
| 34 | b = mm_obj.read(2) |
| 35 | bytes_read_from_mmap.extend(b) |
| 36 | nread += 3 |
| 37 | |
| 38 | with mmap.mmap(ANONYMOUS_MEM, mmap_size) as mm_obj: |
| 39 | for i in range(mmap_size // len(ascii_uppercase)): |
| 40 | mm_obj.write(ascii_uppercase) |
| 41 | |
| 42 | mm_obj.seek(0) |
| 43 | run_concurrently( |
| 44 | worker_func=read, |
| 45 | args=(mm_obj,), |
| 46 | nthreads=NTHREADS, |
| 47 | ) |
| 48 | |
| 49 | self.assertEqual(len(bytes_read_from_mmap), mmap_size) |
| 50 | # Count each letter/byte to verify read correctness |
| 51 | counter = Counter(bytes_read_from_mmap) |
| 52 | self.assertEqual(len(counter), len(ascii_uppercase)) |
| 53 | # Each letter/byte should be read (3 * NTHREADS) times |
| 54 | for letter in ascii_uppercase: |
| 55 | self.assertEqual(counter[letter], 3 * NTHREADS) |
| 56 | |
| 57 | def test_readline(self): |
| 58 | num_lines = 1000 |
nothing calls this directly
no test coverage detected