Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

gh-116738: Make mmap module thread-safe#139237

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to ourterms of service andprivacy statement. We’ll occasionally send you account related emails.

Already on GitHub?Sign in to your account

Merged
kumaraditya303 merged 7 commits intopython:mainfromyoney:ft_mmap
Oct 9, 2025
Merged
Show file tree
Hide file tree
Changes fromall commits
Commits
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
315 changes: 315 additions & 0 deletionsLib/test/test_free_threading/test_mmap.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,315 @@
import unittest

from test.support import import_helper, threading_helper
from test.support.threading_helper import run_concurrently

import os
import string
import tempfile
import threading

from collections import Counter

mmap = import_helper.import_module("mmap")

NTHREADS = 10
ANONYMOUS_MEM = -1


@threading_helper.requires_working_threading()
class MmapTests(unittest.TestCase):
def test_read_and_read_byte(self):
ascii_uppercase = string.ascii_uppercase.encode()
# Choose a total mmap size that evenly divides across threads and the
# read pattern (3 bytes per loop).
mmap_size = 3 * NTHREADS * len(ascii_uppercase)
num_bytes_to_read_per_thread = mmap_size // NTHREADS
bytes_read_from_mmap = []

def read(mm_obj):
nread = 0
while nread < num_bytes_to_read_per_thread:
b = mm_obj.read_byte()
bytes_read_from_mmap.append(b)
b = mm_obj.read(2)
bytes_read_from_mmap.extend(b)
nread += 3

with mmap.mmap(ANONYMOUS_MEM, mmap_size) as mm_obj:
for i in range(mmap_size // len(ascii_uppercase)):
mm_obj.write(ascii_uppercase)

mm_obj.seek(0)
run_concurrently(
worker_func=read,
args=(mm_obj,),
nthreads=NTHREADS,
)

self.assertEqual(len(bytes_read_from_mmap), mmap_size)
# Count each letter/byte to verify read correctness
counter = Counter(bytes_read_from_mmap)
self.assertEqual(len(counter), len(ascii_uppercase))
# Each letter/byte should be read (3 * NTHREADS) times
for letter in ascii_uppercase:
self.assertEqual(counter[letter], 3 * NTHREADS)

def test_readline(self):
num_lines = 1000
lines_read_from_mmap = []
expected_lines = []

def readline(mm_obj):
for i in range(num_lines // NTHREADS):
line = mm_obj.readline()
lines_read_from_mmap.append(line)

# Allocate mmap enough for num_lines (max line 5 bytes including NL)
with mmap.mmap(ANONYMOUS_MEM, num_lines * 5) as mm_obj:
for i in range(num_lines):
line = b"%d\n" % i
mm_obj.write(line)
expected_lines.append(line)

mm_obj.seek(0)
run_concurrently(
worker_func=readline,
args=(mm_obj,),
nthreads=NTHREADS,
)

self.assertEqual(len(lines_read_from_mmap), num_lines)
# Every line should be read once by threads; order is non-deterministic
# Sort numerically by integer value
lines_read_from_mmap.sort(key=lambda x: int(x))
self.assertEqual(lines_read_from_mmap, expected_lines)

def test_write_and_write_byte(self):
thread_letters = list(string.ascii_uppercase)
self.assertLessEqual(NTHREADS, len(thread_letters))
per_thread_write_loop = 100

def write(mm_obj):
# Each thread picks a unique letter to write
thread_letter = thread_letters.pop(0)
thread_bytes = (thread_letter * 2).encode()
for _ in range(per_thread_write_loop):
mm_obj.write_byte(thread_bytes[0])
mm_obj.write(thread_bytes)

with mmap.mmap(
ANONYMOUS_MEM, per_thread_write_loop * 3 * NTHREADS
) as mm_obj:
run_concurrently(
worker_func=write,
args=(mm_obj,),
nthreads=NTHREADS,
)
mm_obj.seek(0)
data = mm_obj.read()
self.assertEqual(len(data), NTHREADS * per_thread_write_loop * 3)
counter = Counter(data)
self.assertEqual(len(counter), NTHREADS)
# Each thread letter should be written `per_thread_write_loop` * 3
for letter in counter:
self.assertEqual(counter[letter], per_thread_write_loop * 3)

def test_move(self):
ascii_uppercase = string.ascii_uppercase.encode()
num_letters = len(ascii_uppercase)

def move(mm_obj):
for i in range(num_letters):
# Move 1 byte from the first half to the second half
mm_obj.move(0 + i, num_letters + i, 1)

with mmap.mmap(ANONYMOUS_MEM, 2 * num_letters) as mm_obj:
mm_obj.write(ascii_uppercase)
run_concurrently(
worker_func=move,
args=(mm_obj,),
nthreads=NTHREADS,
)

def test_seek_and_tell(self):
seek_per_thread = 10

def seek(mm_obj):
self.assertTrue(mm_obj.seekable())
for _ in range(seek_per_thread):
before_seek = mm_obj.tell()
mm_obj.seek(1, os.SEEK_CUR)
self.assertLess(before_seek, mm_obj.tell())

with mmap.mmap(ANONYMOUS_MEM, 1024) as mm_obj:
run_concurrently(
worker_func=seek,
args=(mm_obj,),
nthreads=NTHREADS,
)
# Each thread seeks from current position, the end position should
# be the sum of all seeks from all threads.
self.assertEqual(mm_obj.tell(), NTHREADS * seek_per_thread)

def test_slice_update_and_slice_read(self):
thread_letters = list(string.ascii_uppercase)
self.assertLessEqual(NTHREADS, len(thread_letters))

def slice_update_and_slice_read(mm_obj):
# Each thread picks a unique letter to write
thread_letter = thread_letters.pop(0)
thread_bytes = (thread_letter * 1024).encode()
for _ in range(100):
mm_obj[:] = thread_bytes
read_bytes = mm_obj[:]
# Read bytes should be all the same letter, showing no
# interleaving
self.assertTrue(all_same(read_bytes))

with mmap.mmap(ANONYMOUS_MEM, 1024) as mm_obj:
run_concurrently(
worker_func=slice_update_and_slice_read,
args=(mm_obj,),
nthreads=NTHREADS,
)

def test_item_update_and_item_read(self):
thread_indexes = [i for i in range(NTHREADS)]

def item_update_and_item_read(mm_obj):
# Each thread picks a unique index to write
thread_index = thread_indexes.pop()
for i in range(100):
mm_obj[thread_index] = i
self.assertEqual(mm_obj[thread_index], i)

# Read values set by other threads, all values
# should be less than '100'
for val in mm_obj:
self.assertLess(int.from_bytes(val), 100)

with mmap.mmap(ANONYMOUS_MEM, NTHREADS + 1) as mm_obj:
run_concurrently(
worker_func=item_update_and_item_read,
args=(mm_obj,),
nthreads=NTHREADS,
)

@unittest.skipUnless(os.name == "posix", "requires Posix")
@unittest.skipUnless(hasattr(mmap.mmap, "resize"), "requires mmap.resize")
def test_resize_and_size(self):
thread_indexes = [i for i in range(NTHREADS)]

def resize_and_item_update(mm_obj):
# Each thread picks a unique index to write
thread_index = thread_indexes.pop()
mm_obj.resize(2048)
self.assertEqual(mm_obj.size(), 2048)
for i in range(100):
mm_obj[thread_index] = i
self.assertEqual(mm_obj[thread_index], i)

with mmap.mmap(ANONYMOUS_MEM, 1024, flags=mmap.MAP_PRIVATE) as mm_obj:
run_concurrently(
worker_func=resize_and_item_update,
args=(mm_obj,),
nthreads=NTHREADS,
)

def test_close_and_closed(self):
def close_mmap(mm_obj):
mm_obj.close()
self.assertTrue(mm_obj.closed)

with mmap.mmap(ANONYMOUS_MEM, 1) as mm_obj:
run_concurrently(
worker_func=close_mmap,
args=(mm_obj,),
nthreads=NTHREADS,
)

def test_find_and_rfind(self):
per_thread_loop = 10

def find_and_rfind(mm_obj):
pattern = b'Thread-Ident:"%d"' % threading.get_ident()
mm_obj.write(pattern)
for _ in range(per_thread_loop):
found_at = mm_obj.find(pattern, 0)
self.assertNotEqual(found_at, -1)
# Should not find it after the `found_at`
self.assertEqual(mm_obj.find(pattern, found_at + 1), -1)
found_at_rev = mm_obj.rfind(pattern, 0)
self.assertEqual(found_at, found_at_rev)
# Should not find it after the `found_at`
self.assertEqual(mm_obj.rfind(pattern, found_at + 1), -1)

with mmap.mmap(ANONYMOUS_MEM, 1024) as mm_obj:
run_concurrently(
worker_func=find_and_rfind,
args=(mm_obj,),
nthreads=NTHREADS,
)

@unittest.skipUnless(os.name == "posix", "requires Posix")
@unittest.skipUnless(hasattr(mmap.mmap, "resize"), "requires mmap.resize")
def test_flush(self):
mmap_filename = "test_mmap_file"
resize_to = 1024

def resize_and_flush(mm_obj):
mm_obj.resize(resize_to)
mm_obj.flush()

with tempfile.TemporaryDirectory() as tmpdirname:
file_path = f"{tmpdirname}/{mmap_filename}"
with open(file_path, "wb+") as file:
file.write(b"CPython")
file.flush()
with mmap.mmap(file.fileno(), 1) as mm_obj:
run_concurrently(
worker_func=resize_and_flush,
args=(mm_obj,),
nthreads=NTHREADS,
)

self.assertEqual(os.path.getsize(file_path), resize_to)

def test_mmap_export_as_memoryview(self):
"""
Each thread creates a memoryview and updates the internal state of the
mmap object.
"""
buffer_size = 42

def create_memoryview_from_mmap(mm_obj):
memoryviews = []
for _ in range(100):
mv = memoryview(mm_obj)
memoryviews.append(mv)
self.assertEqual(len(mv), buffer_size)
self.assertEqual(mv[:7], b"CPython")

# Cannot close the mmap while it is exported as buffers
with self.assertRaisesRegex(
BufferError, "cannot close exported pointers exist"
):
mm_obj.close()

with mmap.mmap(ANONYMOUS_MEM, 42) as mm_obj:
mm_obj.write(b"CPython")
run_concurrently(
worker_func=create_memoryview_from_mmap,
args=(mm_obj,),
nthreads=NTHREADS,
)
# Implicit mm_obj.close() verifies all exports (memoryviews) are
# properly freed.


def all_same(lst):
return all(item == lst[0] for item in lst)


if __name__ == "__main__":
unittest.main()
8 changes: 8 additions & 0 deletionsLib/test/test_mmap.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -871,6 +871,10 @@ def test_madvise(self):
size = 2 * PAGESIZE
m = mmap.mmap(-1, size)

class Number:
def __index__(self):
return 2

with self.assertRaisesRegex(ValueError, "madvise start out of bounds"):
m.madvise(mmap.MADV_NORMAL, size)
with self.assertRaisesRegex(ValueError, "madvise start out of bounds"):
Expand All@@ -879,10 +883,14 @@ def test_madvise(self):
m.madvise(mmap.MADV_NORMAL, 0, -1)
with self.assertRaisesRegex(OverflowError, "madvise length too large"):
m.madvise(mmap.MADV_NORMAL, PAGESIZE, sys.maxsize)
with self.assertRaisesRegex(
TypeError, "'str' object cannot be interpreted as an integer"):
m.madvise(mmap.MADV_NORMAL, PAGESIZE, "Not a Number")
self.assertEqual(m.madvise(mmap.MADV_NORMAL), None)
self.assertEqual(m.madvise(mmap.MADV_NORMAL, PAGESIZE), None)
self.assertEqual(m.madvise(mmap.MADV_NORMAL, PAGESIZE, size), None)
self.assertEqual(m.madvise(mmap.MADV_NORMAL, 0, 2), None)
self.assertEqual(m.madvise(mmap.MADV_NORMAL, 0, Number()), None)
self.assertEqual(m.madvise(mmap.MADV_NORMAL, 0, size), None)

@unittest.skipUnless(hasattr(mmap.mmap, 'resize'), 'requires mmap.resize')
Expand Down
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
Make :mod:`mmap` thread-safe on the :term:`free threaded <free threading>`
build.
Loading
Loading
[8]ページ先頭
©2009-2026 Movatter.jp