diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index 0d05c564..c95c8241 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -255,6 +255,11 @@ class ManagedResource: The native pointer is freed automatically via `_free_native_ptr`. """ + _inflight = 0 + _mut_inflight = 0 + _pending_teardown: Optional[bool] = None + _released = False + def _init_attrs(self): """Set this class's own attributes to their defaults. @@ -333,7 +338,7 @@ def _ensure_not_borrowed(self): Raises: C2paError: If a native call is in flight on this resource. """ - if getattr(self, '_inflight', 0) > 0: + if self._inflight > 0: name = type(self).__name__ raise C2paError( f"{name} is in use by another operation and " @@ -345,7 +350,7 @@ def _ensure_no_mutating_call(self): Raises: C2paError: when a mutating native call is in progress. """ - if getattr(self, '_mut_inflight', 0) > 0: + if self._mut_inflight > 0: raise C2paError( f"{type(self).__name__} is running a mutating operation") @@ -371,45 +376,63 @@ def _guarded_op(self, *, refuse_mut=True): finally: self._maybe_flush_pending() - @contextlib.contextmanager - def _native_call(self): - """Hold the handle valid across a native call that runs - caller-supplied stream callbacks, so _live_op_lock() can't be held. - Count the call as in-flight/in-progress. - A free intent (teardown) is registered and the last caller frees. - A free intent marks the resource as closed, preventing further use. + def _begin_reservation(self, *, mutating, refuse): + """Count a native call as in flight, or raise. + + Only place counters go up. + `refuse` runs under the lock, before counting, + and raises to reject the call. """ - with self._live_op_lock(): - self._ensure_valid_state() - self._ensure_no_mutating_call() - self._inflight = getattr(self, '_inflight', 0) + 1 try: - with _native_section(): - yield - finally: with self._live_op_lock(): - self._inflight -= 1 + self._ensure_valid_state() + refuse() + if mutating: + self._mut_inflight += 1 + self._inflight += 1 + except BaseException: self._maybe_flush_pending() + raise + + def _end_reservation(self, *, mutating): + """Uncount a _begin_reservation() call and run any queued teardown. + The only place the counters go down.""" + with self._live_op_lock(): + if mutating: + self._mut_inflight -= 1 + self._inflight -= 1 + self._maybe_flush_pending() @contextlib.contextmanager - def _exclusive_native_call(self): - """Exclusively marks this handle as being mutated. - A free intent (teardown) is registered and the last caller frees. - A free intent marks the resource as closed, preventing further use. + def _reserve(self, *, mutating, refuse): + """Hold a reservation across a native call. + The lock is not held across the call itself: + the call may run caller-supplied reentrant callbacks. + A teardown that arrives meanwhile queues until + the last call out. """ - with self._live_op_lock(): - self._ensure_valid_state() - self._ensure_no_mutating_call() - self._mut_inflight = getattr(self, '_mut_inflight', 0) + 1 - self._inflight = getattr(self, '_inflight', 0) + 1 + self._begin_reservation(mutating=mutating, refuse=refuse) try: with _native_section(): yield finally: - with self._live_op_lock(): - self._mut_inflight -= 1 - self._inflight -= 1 - self._maybe_flush_pending() + self._end_reservation(mutating=mutating) + + @contextlib.contextmanager + def _native_call(self): + """Reserve the handle for a shared call: + other shared calls may run alongside it, no mutating call may.""" + with self._reserve(mutating=False, + refuse=self._ensure_no_mutating_call): + yield + + @contextlib.contextmanager + def _exclusive_native_call(self): + """Reserve the handle for a mutating call: + no other mutating call may run alongside it.""" + with self._reserve(mutating=True, + refuse=self._ensure_no_mutating_call): + yield @staticmethod def _free_native_ptr(ptr): @@ -480,11 +503,10 @@ def _teardown(self, free_handle: bool): than report an error, so it cannot rely on acquiring. """ if is_foreign_process(self): - self._handle = None - self._lifecycle_state = LifecycleState.CLOSED + self._detach_in_child() return - if getattr(self, '_released', False): + if self._released: return self._record_pending_intent(free_handle) @@ -495,10 +517,10 @@ def _teardown(self, free_handle: bool): return try: - if getattr(self, '_released', False): + if self._released: # Checks released as it recorded possible free intents. return - if getattr(self, '_inflight', 0) > 0 or _in_native_section(): + if self._inflight > 0 or _in_native_section(): # Closes the resource so it can't be used anymore. # Records also pending actual frees. self._close_lifecycle() @@ -507,7 +529,7 @@ def _teardown(self, free_handle: bool): return with self._live_teardown_lock(): - pending = getattr(self, '_pending_teardown', None) + pending = self._pending_teardown if pending is not None: free_handle = pending and free_handle self._pending_teardown = None @@ -523,7 +545,7 @@ def _record_pending_intent(self, free_handle: bool): free only leaks. """ with self._live_teardown_lock(): - pending = getattr(self, '_pending_teardown', None) + pending = self._pending_teardown if pending is None: self._pending_teardown = free_handle else: @@ -542,16 +564,25 @@ def _record_pending_teardown(self, free_handle: bool): self._record_pending_intent(free_handle) self._close_lifecycle() + def _detach_in_child(self): + """In a forked child: + null this copy's handle and mark it closed, without freeing. + The parent still owns the native pointer. + """ + if hasattr(self, '_handle'): + self._handle = None + if hasattr(self, '_lifecycle_state'): + self._lifecycle_state = LifecycleState.CLOSED + def _finish_teardown(self, free_handle: bool): """Once teardown can run, runs the actual release. Steps: release, null the handle, free if requested. """ if is_foreign_process(self): - self._handle = None - self._lifecycle_state = LifecycleState.CLOSED + self._detach_in_child() return - if getattr(self, '_released', False): + if self._released: # Already done by another caller (concurrent caller). return @@ -569,16 +600,16 @@ def _finish_teardown(self, free_handle: bool): def _has_pending_teardown(self) -> bool: """Check if a teardown request is waiting for the resource.""" - return getattr(self, '_pending_teardown', None) is not None + return self._pending_teardown is not None def _flush_pending_pass(self): """Attempt to run pending teardowns. """ with self._live_op_lock(): - if getattr(self, '_pending_teardown', None) is None: + if self._pending_teardown is None: return - if getattr(self, '_inflight', 0) > 0: + if self._inflight > 0: return if _in_native_section(): _register_for_section_flush(self) @@ -596,8 +627,7 @@ def _maybe_flush_pending(self): return self._flush_pending_pass() - if self._has_pending_teardown() and not getattr( - self, '_released', False): + if self._has_pending_teardown() and not self._released: self._flush_pending_pass() def _release_handle(self): @@ -608,8 +638,7 @@ def _release_handle(self): to free. """ with self._live_op_lock(): - owned_elsewhere = getattr( - self, '_pending_teardown', None) is not None + owned_elsewhere = self._pending_teardown is not None if not owned_elsewhere and ( self._lifecycle_state != LifecycleState.ACTIVE): self._handle = None @@ -787,49 +816,36 @@ def _begin_consume(self): Raises: C2paError: Unusable resource or native call in progress. """ - with self._live_op_lock(): - # A consumed or closed resource has no handle left to hand over; - # without this the call would pass a null pointer to native. - self._ensure_valid_state() - self._ensure_not_borrowed() - self._mut_inflight = getattr(self, '_mut_inflight', 0) + 1 - self._inflight = getattr(self, '_inflight', 0) + 1 + self._begin_reservation(mutating=True, + refuse=self._ensure_not_borrowed) def _end_consume(self): """Release a _begin_consume() reservation, then run any teardown that arrived while it was held.""" - with self._live_op_lock(): - self._mut_inflight -= 1 - self._inflight -= 1 - self._maybe_flush_pending() + self._end_reservation(mutating=True) def _consume_and_swap(self, ffi_call, error_message): """Run an FFI call consuming the handle, reserving it. A replacement handle will be swapping in on success (a returned null value is a failure). """ + def swap(new_ptr): + with self._live_op_lock(): + self._handle = new_ptr - self._begin_consume() - try: - with _native_section(): - new_ptr = self._invoke_consume( - ffi_call, error_message, reserved=True) - if new_ptr: - with self._live_op_lock(): - self._handle = new_ptr - return - self._raise_consume_failure(error_message, reserved=True) - finally: - self._end_consume() + self._consume_reserved(ffi_call, error_message, + succeeded=bool, on_success=swap) - def _consume_reserved(self, ffi_call, error_message, *, succeeded): - """Run a reserved consuming call and mark the handle consumed on - success. + def _consume_reserved(self, ffi_call, error_message, *, succeeded, + on_success=None): + """Run a reserved consuming call, then act on success. Args: succeeded: Reads the call's raw result and returns whether it succeeded. Each entry point has its own convention: a status code, or a replacement pointer. + on_success: Called with the raw result on success. Default: + mark the handle consumed, closed, not freed. Returns: The call's raw result, for callers that hand it on. @@ -840,7 +856,10 @@ def _consume_reserved(self, ffi_call, error_message, *, succeeded): result = self._invoke_consume( ffi_call, error_message, reserved=True) if succeeded(result): - self._teardown(free_handle=False) + if on_success is None: + self._teardown(free_handle=False) + else: + on_success(result) return result self._raise_consume_failure(error_message, reserved=True) finally: @@ -896,15 +915,7 @@ def _cleanup_resources(self): """Release native resources idempotently.""" try: if is_foreign_process(self): - # A forked child holds a separate copy of this object and the - # parent still owns the real handle and frees it. Mark this - # copy closed and null its handle so the child cannot mistake - # it for usable or free it, but do not free here. - # Mutating this copy does not touch the parent's. - if hasattr(self, '_handle'): - self._handle = None - if hasattr(self, '_lifecycle_state'): - self._lifecycle_state = LifecycleState.CLOSED + self._detach_in_child() return if hasattr(self, '_lifecycle_state'): # Closes here must defer to the teardown checks. @@ -921,7 +932,7 @@ def is_valid(self) -> bool: return ( self._lifecycle_state == LifecycleState.ACTIVE and self._handle is not None - and getattr(self, '_mut_inflight', 0) == 0 + and self._mut_inflight == 0 ) def close(self) -> None: diff --git a/tests/test_unit_tests_threaded.py b/tests/test_unit_tests_threaded.py index eb0d6bb3..e4ca8134 100644 --- a/tests/test_unit_tests_threaded.py +++ b/tests/test_unit_tests_threaded.py @@ -4293,7 +4293,19 @@ def test_close_during_sign_does_not_deadlock(self): "claim_generator_info": [ {"name": "python_test", "version": "0.0.1"}], "format": "image/jpeg", - "assertions": [], + "assertions": [ + { + "label": "c2pa.actions", + "data": { + "actions": [ + { + "action": "c2pa.created", + "digitalSourceType": "http://cv.iptc.org/newscodes/digitalsourcetype/digitalCreation" + } + ] + } + } + ], } errors = [] @@ -5007,7 +5019,19 @@ def test_sign_with_internal_close_frees_once(self): "claim_generator_info": [ {"name": "python_test", "version": "0.0.1"}], "format": "image/jpeg", - "assertions": [], + "assertions": [ + { + "label": "c2pa.actions", + "data": { + "actions": [ + { + "action": "c2pa.created", + "digitalSourceType": "http://cv.iptc.org/newscodes/digitalsourcetype/digitalCreation" + } + ] + } + } + ], } signer = Signer.from_info(signer_info) builder = Builder(manifest)