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