From 8e7d18f6e660e97ca21c31dff2705791b402ad32 Mon Sep 17 00:00:00 2001 From: christopher5106 Date: Sat, 24 Oct 2026 16:42:46 +0200 Subject: [PATCH] fix: Read and write regular files natively so streams release the GIL A Stream serves every read, seek and write of the native library through a Python callback, tens of thousands per asset (one or more per PNG chunk or BMFF box), and each callback retakes the GIL. Signing or reading files from several threads is therefore slower than from one, and any busy Python thread stalls a signature for tens of seconds. When the wrapped object is a regular file on a 64-bit POSIX platform, Stream now hands the native library libc read/lseek/write with a private duplicate of the file descriptor as context: the native call never re-enters Python. The duplicate starts at the Python object's position and gives the final position back on close(), and closing the Python file meanwhile cannot hand its number to another file. In-memory streams, pipes, append-mode files and other platforms keep the Python callbacks, as does Stream.native_file_io = False. Co-Authored-By: Claude Opus 5.5 --- src/c2pa/c2pa.py | 116 +++++++++++++++++++++++++++++++++-- tests/test_unit_tests.py | 126 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 237 insertions(+), 5 deletions(-) diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index a9090cc1..f8a055f1 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -21,6 +21,7 @@ import logging import sys import os +import stat import threading import warnings import weakref @@ -1005,6 +1006,31 @@ def __del__(self): ctypes.c_uint8), ctypes.c_ssize_t) FlushCallback = ctypes.CFUNCTYPE(ctypes.c_ssize_t, ctypes.c_void_p) + + +def _load_native_file_callbacks(): + """libc read/lseek/write as stream callbacks over a file descriptor. + + With the descriptor as the stream context, the native library reads and + writes the file without calling back into Python: the GIL stays released + for the whole native call. Their C signatures match the callback types on + 64-bit POSIX ABIs only; elsewhere this returns None and every stream keeps + the Python callbacks. + """ + if os.name != "posix" or sys.maxsize <= 2**32: + return None + try: + libc = ctypes.CDLL(None) + return ( + ctypes.cast(libc.read, ReadCallback), + ctypes.cast(libc.lseek, SeekCallback), + ctypes.cast(libc.write, WriteCallback), + ) + except (AttributeError, OSError): # pragma: no cover + return None + + +_NATIVE_FILE_CALLBACKS = _load_native_file_callbacks() SignerCallback = ctypes.CFUNCTYPE( ctypes.c_ssize_t, ctypes.c_void_p, ctypes.POINTER( ctypes.c_ubyte), ctypes.c_size_t, ctypes.POINTER( @@ -2422,6 +2448,11 @@ class Stream: # Methods an object must expose to be wrapped as Stream. _REQUIRED_STREAM_METHODS = ("read", "write", "seek", "tell", "flush") + # Regular files are read and written by the native library directly + # (see _native_file_descriptor). Set to False to always go through + # Python callbacks. + native_file_io = True + # Class-level error messages to avoid multiple creation _ERROR_MESSAGES = { 'stream_error': "Error cleaning up stream: {}", @@ -2446,6 +2477,7 @@ def __init__(self, file_like_stream): self._closed = False self._initialized = False self._stream = None + self._native_fd = None # Serializes close() and __del__ against a concurrent double-free. self._close_lock = threading.RLock() @@ -2474,6 +2506,7 @@ def __init__(self, file_like_stream): ) self._file_like_stream = file_like_stream + self._native_fd = self._native_file_descriptor(file_like_stream) # Weakref breaks avoids references cycles: # Stream -> ctypes_cb -> closure -> Stream. @@ -2629,20 +2662,33 @@ def flush_callback(ctx): return -1 # Create callbacks that will be kept alive by being instance attributes - self._read_cb = ReadCallback(read_callback) - self._seek_cb = SeekCallback(seek_callback) - self._write_cb = WriteCallback(write_callback) - self._flush_cb = FlushCallback(flush_callback) + if self._native_fd is not None: + # libc on the stream's own descriptor: no Python callback, no GIL. + # Flushing a descriptor is a no-op (the Python buffer was flushed + # when the stream was created and is not used meanwhile). + self._read_cb, self._seek_cb, self._write_cb = ( + _NATIVE_FILE_CALLBACKS) + self._flush_cb = FlushCallback(lambda ctx: 0) + context = ctypes.cast( + ctypes.c_void_p(self._native_fd), + ctypes.POINTER(StreamContext)) + else: + self._read_cb = ReadCallback(read_callback) + self._seek_cb = SeekCallback(seek_callback) + self._write_cb = WriteCallback(write_callback) + self._flush_cb = FlushCallback(flush_callback) + context = None # Create the stream self._stream = _lib.c2pa_create_stream( - None, + context, self._read_cb, self._seek_cb, self._write_cb, self._flush_cb ) if not self._stream: + self._close_native_fd() error = _read_native_error() raise C2paError( "Failed to create stream: {}".format(error or "Unknown error")) @@ -2687,6 +2733,7 @@ def __del__(self): pass finally: self._stream = None + self._close_native_fd() finally: if lock is not None: lock.release() @@ -2732,6 +2779,8 @@ def close(self): str(e))) finally: self._stream = None + self._sync_native_position() + self._close_native_fd() # Clean up callbacks for attr in [ @@ -2752,6 +2801,63 @@ def close(self): self._closed = True self._initialized = False + @classmethod + def _native_file_descriptor(cls, file_like_stream): + """A private descriptor for native I/O on a regular file, or None. + + The descriptor is a duplicate, owned by this Stream: closing the + Python file meanwhile cannot hand its number to another file. It + shares the file's offset, which is first aligned with the Python + object's position (its buffer flushed), and given back on close(). + Streams that are not regular files (in-memory, pipes, sockets), + files opened for appending, and platforms without native callbacks + keep the Python callbacks. + """ + if not cls.native_file_io or _NATIVE_FILE_CALLBACKS is None: + return None + try: + fd = file_like_stream.fileno() + if not stat.S_ISREG(os.fstat(fd).st_mode): + return None + if "a" in str(getattr(file_like_stream, "mode", "")): + return None + file_like_stream.flush() + position = file_like_stream.tell() + native_fd = os.dup(fd) + os.lseek(native_fd, position, os.SEEK_SET) + return native_fd + except (AttributeError, OSError, ValueError, io.UnsupportedOperation): + return None + + def _sync_native_position(self): + """Move the Python file to where the native library left the file.""" + if self._native_fd is None: + return + try: + self._file_like_stream.seek( + os.lseek(self._native_fd, 0, os.SEEK_CUR)) + except (OSError, ValueError) as e: + logger.error( + Stream._ERROR_MESSAGES['stream_error'].format(str(e))) + + def _close_native_fd(self): + """Close the descriptor the native library used, if any.""" + fd, self._native_fd = self._native_fd, None + if fd is not None: + try: + os.close(fd) + except OSError: # pragma: no cover + pass + + @property + def uses_native_file_io(self) -> bool: + """Whether the native library reads and writes the file directly. + + Returns: + bool: True for a regular file served without Python callbacks + """ + return self._native_fd is not None + def write_to_target(self, dest_stream): self._file_like_stream.seek(0) dest_stream.write(self._file_like_stream.getvalue()) diff --git a/tests/test_unit_tests.py b/tests/test_unit_tests.py index 69ba7875..026e9e92 100644 --- a/tests/test_unit_tests.py +++ b/tests/test_unit_tests.py @@ -6208,6 +6208,132 @@ def test_stream_flush_after_close(self): self.assertEqual(flush_cb(None), -1) + +class _CountingFile(io.FileIO): + """A regular file that counts the reads and writes done through Python.""" + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.python_calls = 0 + + def readinto(self, b): + self.python_calls += 1 + return super().readinto(b) + + def read(self, *args): + self.python_calls += 1 + return super().read(*args) + + def write(self, b): + self.python_calls += 1 + return super().write(b) + + +class TestNativeFileStreams(unittest.TestCase): + """Regular files are read and written by the native library, not Python.""" + + def setUp(self): + with open(os.path.join(FIXTURES_DIR, "es256_certs.pem"), "rb") as f: + certs = f.read() + with open(os.path.join(FIXTURES_DIR, "es256_private.key"), "rb") as f: + key = f.read() + # No timestamp authority: the test stays offline. + self.signer = Signer.from_info( + C2paSignerInfo(alg=b"es256", sign_cert=certs, private_key=key, + ta_url=None)) + self.manifest = { + "claim_generator_info": [{"name": "python_test", "version": "0.0.1"}], + "title": "Native streams", + "format": "image/jpeg", + "assertions": [{ + "label": "c2pa.actions", + "data": {"actions": [{ + "action": "c2pa.created", + "digitalSourceType": "http://cv.iptc.org/newscodes/" + "digitalsourcetype/digitalCapture", + }]}, + }], + } + self.temp_dir = tempfile.mkdtemp() + + def tearDown(self): + self.signer.close() + shutil.rmtree(self.temp_dir, ignore_errors=True) + + def test_only_regular_files_use_native_io(self): + with open(DEFAULT_TEST_FILE, "rb") as f, Stream(f) as stream: + self.assertTrue(stream.uses_native_file_io) + with Stream(io.BytesIO(b"data")) as stream: + self.assertFalse(stream.uses_native_file_io) + append_path = os.path.join(self.temp_dir, "append.bin") + with open(append_path, "ab") as f, Stream(f) as stream: + self.assertFalse(stream.uses_native_file_io) + + def test_native_io_can_be_turned_off(self): + Stream.native_file_io = False + try: + with open(DEFAULT_TEST_FILE, "rb") as f, Stream(f) as stream: + self.assertFalse(stream.uses_native_file_io) + finally: + Stream.native_file_io = True + + def test_signing_files_never_calls_back_into_python(self): + dest_path = os.path.join(self.temp_dir, "signed.jpg") + with _CountingFile(DEFAULT_TEST_FILE, "rb") as source, \ + _CountingFile(dest_path, "w+b") as dest: + Builder(self.manifest).sign(self.signer, "image/jpeg", source, dest) + self.assertEqual(source.python_calls, 0) + self.assertEqual(dest.python_calls, 0) + # The Python file continues where the native library stopped. + self.assertEqual(source.tell(), os.path.getsize(DEFAULT_TEST_FILE)) + + with Reader(dest_path) as reader: + manifest = json.loads(reader.json()) + active = manifest["manifests"][manifest["active_manifest"]] + self.assertEqual(active["title"], "Native streams") + + def test_native_and_python_streams_sign_alike(self): + signed = [] + for native in (True, False): + Stream.native_file_io = native + try: + dest_path = os.path.join(self.temp_dir, f"signed-{native}.jpg") + with open(DEFAULT_TEST_FILE, "rb") as source, \ + open(dest_path, "w+b") as dest: + Builder(self.manifest).sign( + self.signer, "image/jpeg", source, dest) + with Reader(dest_path) as reader: + manifest = json.loads(reader.json()) + signed.append((os.path.getsize(dest_path), + manifest.get("validation_state"))) + finally: + Stream.native_file_io = True + self.assertEqual(signed[0], signed[1]) + + def test_starts_at_the_python_position(self): + with open(DEFAULT_TEST_FILE, "rb") as f: + expected = f.read()[100:116] + f.seek(100) + with Stream(f) as stream: + buffer = (ctypes.c_ubyte * 16)() + read = stream._read_cb( + ctypes.c_void_p(stream._native_fd), buffer, 16) + self.assertEqual(bytes(buffer[:read]), expected) + self.assertEqual(f.tell(), 116) + + def test_survives_the_python_file_closing_first(self): + f = open(DEFAULT_TEST_FILE, "rb") + stream = Stream(f) + f.close() + try: + buffer = (ctypes.c_ubyte * 4)() + read = stream._read_cb( + ctypes.c_void_p(stream._native_fd), buffer, 4) + self.assertEqual(read, 4) + finally: + stream._close_native_fd() + stream.close() + class TestLegacyAPI(unittest.TestCase): def setUp(self): # Filter specific deprecation warnings for legacy API tests