Coverage for src/refcount/interop.py: 96.28%
166 statements
« prev ^ index » next coverage.py v7.6.10, created at 2025-01-14 17:01 +1100
1"""Implementation of reference counting classes for external resources accessed via interoperability software such as cffi."""
3from typing import Any, Callable, Dict, Optional, Union
5from cffi import FFI
6from typing_extensions import TypeAlias
8from refcount.base import NativeHandle
10# This is a Hack. I cannot use FFI.CData in type hints.
11# CffiData: TypeAlias = FFI().CData
12CffiData: TypeAlias = Any
13"""A dummy type to use in type hints for limited documentation purposes
15FFI.CData is a type, but it seems it cannot be used in type hinting.
16"""
19class CffiNativeHandle(NativeHandle):
20 """Reference counting wrapper class for CFFI pointers.
22 This class is originally inspired from a class with a similar purpose in C#. See https://github.com/rdotnet/dynamic-interop-dll
24 Say you have a C API as follows:
26 * `void* create_some_object();`
27 * `dispose_of_some_object(void* obj);`
29 and accessing it using Python and [CFFI](https://cffi.readthedocs.io).
30 Users would use the `calllib` function:
32 ```python
33 from cffi import FFI
35 ffi = FFI()
37 # cdef() expects a single string declaring the C types, functions and
38 # globals needed to use the shared object. It must be in valid C syntax.
39 ffi.cdef('''
40 void* create_some_object();
41 dispose_of_some_object(void* obj);
42 ''')
43 mydll_so = ffi.dlopen("/path/to/mydll.so", ffi.RTLD_LAZY)
44 cffi_void_ptr = mydll_so.create_some_object()
45 ```
47 at some point when done you need to dispose of it to clear native memory:
49 ```python
50 mydll_so.dispose_of_some_object(cffi_void_ptr)
51 ```
53 In practice in real systems one quickly ends up with cases
54 where it is unclear when to dispose of the object.
55 If you call the `dispose_of_some_object` function more
56 than once, or too soon, you quickly crash the program, or possibly worse outcomes with numeric non-sense.
57 `CffiNativeHandle` is designed to alleviate this headache by
58 using native reference counting of `handle` classes to reliably dispose of objects.
60 Attributes:
61 _handle (object): The handle (e.g. cffi pointer) to the native resource.
62 _type_id (Optional[str]): An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation.
63 _finalizing (bool): a flag telling whether this object is in its deletion phase. This has a use in some advanced cases with reverse callback, possibly not relevant in Python.
64 """
66 def __init__(self, handle: "CffiData", type_id: Optional[str] = None, prior_ref_count: int = 0):
67 """Initialize a reference counter for a resource handle, with an initial reference count.
69 Args:
70 handle (object): The handle (e.g. cffi pointer) to the native resource.
71 type_id (Optional[str]): An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation.
72 prior_ref_count (int): the initial reference count. Default 0 if this NativeHandle is sole responsible for the lifecycle of the resource.
73 """
74 super().__init__(handle, prior_ref_count)
75 # TODO checks on handle
76 self._type_id = type_id
77 self._finalizing: bool = False
78 self._handle: CffiData = None
79 if handle is None:
80 return # defer setting the handle to the inheritor.
81 self._set_handle(handle, prior_ref_count)
83 def _is_valid_handle(self, h: "CffiData") -> bool:
84 """Checks if the handle is a CFFI CData pointer, acceptable handle for this wrapper.
86 Args:
87 handle (object): The handle (e.g. cffi pointer) to the native resource.
88 """
89 return isinstance(h, FFI.CData)
91 def __dispose_impl(self, decrement: bool) -> None:
92 """An implementation of the dispose method in a 'Dispose' software pattern. Avoids cyclic method calls.
94 Args:
95 decrement (bool): indicating whether the reference count should be decreased. It should almost always be True except in very unusual use cases (argument is for possible future use).
96 """
97 if self.disposed:
98 return
99 if decrement: 99 ↛ 101line 99 didn't jump to line 101 because the condition on line 99 was always true
100 self._ref_count = self._ref_count - 1
101 if self._ref_count <= 0: # noqa: SIM102
102 if self._release_handle(): 102 ↛ exitline 102 didn't return from function '__dispose_impl' because the condition on line 102 was always true
103 self._handle = None
105 @property
106 def disposed(self) -> bool:
107 """Has the native object and memory already been disposed of.
109 Returns:
110 (bool): The underlying native handle has been disposed of from this wrapper
111 """
112 return self._handle is None
114 @property
115 def is_invalid(self) -> bool:
116 """Is the underlying handle valid? In practice synonym with the disposed attribute.
118 Returns:
119 (bool): True if this handle is valid
120 """
121 return self._handle is None
123 def _release_handle(self) -> bool:
124 """Must of overriden. Method disposing of the object pointed to by the CFFI pointer (handle).
126 Raises:
127 NotImplementedError: thrown if this method is not overriden by inheritors
129 Returns:
130 bool: Overriding implementation should return True if the release of native resources was successful, False otherwise.
131 """
132 # See also https://stackoverflow.com/questions/4714136/how-to-implement-virtual-methods-in-python
133 # May want to make this abstract using ABC - we'll see.
134 raise NotImplementedError("method _release_handle must be overriden by child classes")
136 def get_handle(self) -> Union["CffiData", None]:
137 """Gets the underlying low-level CFFI handle this object wraps.
139 Returns:
140 (Union[CffiData, None]): CFFI handle or None
141 """
142 return self._handle
144 # TODO?
145 # @property.getter
146 # def ptr(self):
147 # """ Return the pointer to a cffi object """
148 # return self._handle
150 @property
151 def type_id(self) -> Optional[str]:
152 """Return an optional type identifier for the underlying native type.
154 This can be in practice useful to be more transparent about the underlying
155 type obtained via a C API with opaque pointers (i.e. void*)
157 Returns:
158 str: optional type identifier
159 """
160 return self._type_id
162 # @property
163 # def obj(self):
164 # """ Return the native object pointed to (cffi object) """
165 # return self._handle
167 def __str__(self) -> str:
168 """String representation."""
169 if self.type_id is None or self.type_id == "":
170 return "CFFI pointer handle to a native pointer " + str(self._handle)
171 return 'CFFI pointer handle to a native pointer of type id "' + self.type_id + '"'
173 @property
174 def ptr(self) -> "CffiData":
175 """Return the pointer (cffi object)."""
176 return self._handle
178 @property
179 def obj(self) -> Any:
180 """Return the object pointed to (cffi object)."""
181 return self._handle[0]
183 def __repr__(self) -> str:
184 """String representation."""
185 return str(self)
187 def __del__(self) -> None:
188 """destructor, triggering the release of the underlying handled resource if the reference count is 0."""
189 if self._handle is not None:
190 # if not self._release_native is None:
191 # Protect against accessing properties
192 # of partially constructed objects (May not be an issue in Python?)
193 self._finalizing = True
194 self.release()
196 def dispose(self) -> None:
197 """Disposing of the object pointed to by the CFFI pointer (handle) if the reference counts allows it."""
198 self.__dispose_impl(True)
200 def release(self) -> None:
201 """Manually decrements the reference counter. Triggers disposal if reference count is down to zero."""
202 self.__dispose_impl(True)
205class DeletableCffiNativeHandle(CffiNativeHandle):
206 """Reference counting wrapper class for CFFI pointers.
208 Attributes:
209 _handle (object): The handle (e.g. cffi pointer) to the native resource.
210 _type_id (Optional[str]): An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation.
211 _finalizing (bool): a flag telling whether this object is in its deletion phase. This has a use in some advanced cases with reverse callback, possibly not relevant in Python.
212 _release_native (Callable[[CffiData],None]): function to call on deleting this wrapper. The function should have one argument accepting the object _handle.
213 """
215 def __init__(
216 self,
217 handle: "CffiData",
218 release_native: Optional[Callable[["CffiData"], None]],
219 type_id: Optional[str] = None,
220 prior_ref_count: int = 0,
221 ):
222 """New reference counter for a CFFI resource handle.
224 Args:
225 handle (CffiData): The handle (expected cffi pointer) to the native resource.
226 release_native (Callable[[CffiData],None]): function to call on deleting this wrapper. The function should have one argument accepting the object handle.
227 type_id (str, optional): [description]. An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation. Defaults to None.
228 prior_ref_count (int, optional): [description]. The initial reference count. Defaults to 0 if this NativeHandle is sole responsible for the lifecycle of the resource.
229 """
230 super().__init__(
231 handle,
232 type_id,
233 prior_ref_count,
234 )
235 self._release_native = release_native
236 self._set_handle(handle, prior_ref_count)
238 def _release_handle(self) -> bool:
239 """Release the handle, dispose of the native resource.
241 Returns:
242 bool: Return True if the release of native resources handle was successful, False otherwise.
243 """
244 if self._handle is None: 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true
245 return False
246 if self._release_native is not None: 246 ↛ 252line 246 didn't jump to line 252 because the condition on line 246 was always true
247 self._release_native(
248 self._handle,
249 ) # TODO are trapped exceptions acceptable here?
250 return True
251 # if self._release_native is None:
252 return False
255class OwningCffiNativeHandle(CffiNativeHandle):
256 """Reference counting wrapper class for CFFI pointers that own and already manage the native memory.
258 Attributes:
259 _handle (object): The handle (e.g. cffi pointer) to the native resource.
260 _type_id (Optional[str]): An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation.
261 _finalizing (bool): a flag telling whether this object is in its deletion phase. This has a use in some advanced cases with reverse callback, possibly not relevant in Python.
262 """
264 # """ a global function that can be called to release an external pointer """
265 # release_native = None
267 def __init__(
268 self,
269 handle: "CffiData",
270 type_id: Optional[str] = None,
271 prior_ref_count: int = 0,
272 ):
273 """Reference counting wrapper class for CFFI pointers that own and already manage the native memory.
275 Args:
276 handle (CffiData): The handle (expected cffi pointer) to the native resource.
277 type_id (str, optional): [description]. An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation. Defaults to None.
278 prior_ref_count (int, optional): [description]. The initial reference count. Defaults to 0 if this NativeHandle is sole responsible for the lifecycle of the resource.
279 """
280 super().__init__(
281 handle,
282 type_id,
283 prior_ref_count,
284 )
285 self._set_handle(handle, prior_ref_count)
287 def _release_handle(self) -> bool:
288 """Does nothing, as the wrapped cffi pointer is already owning and managing the memory.
290 Returns:
291 bool: Always returns True.
292 """
293 return True
296def wrap_cffi_native_handle(
297 obj: Union["CffiData", Any],
298 type_id: str = "",
299 release_native: Optional[Callable[["CffiData"], None]] = None,
300) -> Union[DeletableCffiNativeHandle, Any]:
301 """Create a reference counting wrapper around an object if this object is a CFFI pointer.
303 Args:
304 obj (Union[CffiData,Any]): An object, which will be wrapped if this is a CFFI pointer, i.e. an instance of `CffiData`
305 release_native (Callable[[CffiData],None]): function to call on deleting this wrapper. The function should have one argument accepting the object handle.
306 type_id (Optional[str]): An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation.
307 """
308 if isinstance(obj, FFI.CData):
309 return DeletableCffiNativeHandle(
310 obj,
311 release_native=release_native,
312 type_id=type_id,
313 )
314 return obj
317def is_cffi_native_handle(x: Any, type_id: str = "") -> bool:
318 """Checks whether an object is a ref counting wrapper around a CFFI pointer.
320 Args:
321 x (object): object to test, presumed to be an instance of `CffiNativeHandle`
322 type_id (Optional[str]): Optional identifier for the type of underlying resource being wrapped.
323 """
324 if x is None:
325 return False
326 if not isinstance(x, CffiNativeHandle):
327 return False
328 if type_id is None or type_id == "":
329 return True
330 return x.type_id == type_id
333def unwrap_cffi_native_handle(
334 obj_wrapper: Any,
335 stringent: bool = False,
336) -> Union["CffiData", Any, None]:
337 """Unwrap a reference counting wrapper and returns its CFFI pointer if it is found (wrapped or 'raw').
339 Args:
340 obj_wrapper (Any): An object, which will be unwrapped if this is a CFFI pointer, i.e. an instance of `CffiData`
341 stringent (bool, optional): [description]. if True an error is raised if obj_wrapper is neither None, a CffiNativeHandle nor an CffiData. Defaults to False.
343 Raises:
344 Exception: A CFFI pointer could not be found in the object.
346 Returns:
347 Union[CffiData,Any,None]: A CFFI pointer if it was found. Returns None or unchanged if not found, and stringent is equal to False. Exception otherwise.
348 """
349 # 2016-01-28 allowing null pointers, to unlock behavior of EstimateERRISParameters.
350 # Reassess approach, even if other C API function will still catch the issue of null ptrs.
351 if obj_wrapper is None:
352 return None
353 if isinstance(obj_wrapper, CffiNativeHandle):
354 return obj_wrapper.get_handle()
355 if isinstance(obj_wrapper, FFI.CData):
356 return obj_wrapper
357 if stringent:
358 raise TypeError(
359 "Argument is neither a CffiNativeHandle nor a CFFI external pointer",
360 )
361 return obj_wrapper
364def cffi_arg_error_external_obj_type(x: Any, expected_type_id: str) -> str:
365 """Build an error message that an unexpected object is in lieu of an expected refcount external ref object.
367 Args:
368 x (object): object passed as an argument to a function but with an unexpected type or type id.
369 expected_type_id (Optional[str]): Expected identifier for the type of underlying resource being wrapped.
371 Returns (str): the error message
372 """
373 if x is None:
374 return "Expected a 'CffiNativeHandle' but instead got 'None'"
375 if not is_cffi_native_handle(x):
376 return f"Expected a 'CffiNativeHandle' but instead got object of type '{type(x)!s}'"
377 return f"Expected a 'CffiNativeHandle' with underlying type id '{expected_type_id}' but instead got one with type id '{x.type_id}'"
380# Maybe, pending use cases:
381# def checked_unwrap_cffi_native_handle (obj_wrapper, stringent=False):
382# if not is_cffi_native_handle (obj_wrapper, expected_type_id):
383# raise Exception(cffi_arg_error_external_obj_type(obj_wrapper, expected_type_id)
384# else:
385# return unwrap_cffi_native_handle (obj_wrapper, stringent=True)
388class GenericWrapper:
389 """A pass-through wrapper for python objects that are ready for C interop. "bytes" can be passed as C 'char*'.
391 This is mostly a facility to generate glue code more easily
392 """
394 def __init__(self, handle: "CffiData"):
395 """A pass-through wrapper for python objects that are ready for C interop. "bytes" can be passed as C 'char*'."""
396 self._handle = handle
398 @property
399 def ptr(self) -> "CffiData":
400 """The wrapped python object that is ready for C interop."""
401 return self._handle
404def wrap_as_pointer_handle(
405 obj_wrapper: Any,
406 stringent: bool = False,
407) -> Union[CffiNativeHandle, OwningCffiNativeHandle, GenericWrapper]:
408 """Wrap an object, if need be, so that its C API pointer appears accessible via a 'ptr' property.
410 Args:
411 obj_wrapper (Any): Object to wrap, if necessary
412 stringent (bool, optional): Throws an exception if the input type is unhandled. Defaults to False.
414 Raises:
415 TypeError: neither a CffiNativeHandle nor a CFFI external pointer, nor bytes
417 Returns:
418 Union[CffiNativeHandle, OwningCffiNativeHandle, GenericWrapper, None]: wrapped object or None
419 """
420 # 2016-01-28 allowing null pointers, to unlock behavior of EstimateERRISParameters.
421 # Reassess approach, even if other C API function will still catch the issue of null ptrs.
422 if obj_wrapper is None:
423 return GenericWrapper(None)
424 # return GenericWrapper(FFI.NULL) # Ended with kernel crashes and API call return, but unclear why
425 if isinstance(obj_wrapper, CffiNativeHandle):
426 return obj_wrapper
427 if isinstance(obj_wrapper, FFI.CData):
428 return OwningCffiNativeHandle(obj_wrapper)
429 if isinstance(obj_wrapper, bytes):
430 return GenericWrapper(obj_wrapper)
431 if stringent:
432 raise TypeError(
433 "Argument is neither a CffiNativeHandle nor a CFFI external pointer, nor bytes",
434 )
435 return obj_wrapper
438def type_error_cffi(x: Union[CffiNativeHandle, Any], expected_type: str) -> str:
439 """DEPRECATED.
441 This function is deprecated; may still be in use in 'uchronia'. Use `cffi_arg_error_external_obj_type` instead.
443 Build an error message for situations where a cffi pointer handler is not that, or not of the expected type
445 Args:
446 x (Union[CffiNativeHandle, Any]): actual object that is not of the expected type or underlying type for the external pointer.
447 expected_type (str): underlying type expected for the CFFI pointer handler
449 Returns:
450 str: error message that the caller can use to report the issue
451 """
452 return cffi_arg_error_external_obj_type(x, expected_type)
455class CffiWrapperFactory:
456 """A class that creates custom python wrappers based on the type identifier of the external pointer being wrapped."""
458 def __init__(self, api_type_wrapper: Dict[str, Any], strict_wrapping: bool = False) -> None:
459 """A class that creates custom python wrappers based on the type identifier of the external pointer being wrapped.
461 Args:
462 api_type_wrapper (Dict[str,Any]): dictionary, mapping from type identifiers to callables, class constructors
463 strict_wrapping (bool, optional): If true, type identifiers passed at wrapper creation time `create_wrapper`
464 must be known or exceptions are raised. If False, it falls back on creating generic wrappers. Defaults to False.
465 """
466 self._strict_wrapping = strict_wrapping
467 self._api_type_wrapper = api_type_wrapper
469 def create_wrapper(
470 self,
471 obj: Any,
472 type_id: str,
473 release_native: Optional[Callable[["CffiData"], None]],
474 ) -> "CffiNativeHandle":
475 """Create a CffiNativeHandle wrapper around an object, if this object is a CFFI pointer.
477 Args:
478 obj (Union[CffiData,Any]): An object, which will be wrapped if this is a CFFI pointer, i.e. an instance of `CffiData`
479 type_id (Optional[str]): An optional identifier for the type of underlying resource. This can be used to usefully maintain type information about the pointer/handle across an otherwise opaque C API. See package documentation.
480 release_native (Callable[[CffiData],None]): function to call on deleting this wrapper. The function should have one argument accepting the object handle.
482 Raises:
483 ValueError: Missing type_id
484 ValueError: If this object is in strict mode, and `type_id` is not known in the mapping
485 NotImplementedError: `type_id` is known, but mapping to None (wrapper not yet implemented)
486 TypeError: The function to create the wrapper does not accept any argument.
488 Returns:
489 CffiNativeHandle: cffi wrapper
490 """
491 from inspect import signature
493 # from typing import get_type_hints
494 if type_id is None:
495 raise ValueError("Type ID provided cannot be None")
496 if type_id not in self._api_type_wrapper:
497 if self._strict_wrapping:
498 raise ValueError(f"Type ID {type_id} is unknown")
499 return wrap_cffi_native_handle(obj, type_id, release_native)
500 wrapper_type = self._api_type_wrapper[type_id]
501 if wrapper_type is None:
502 if self._strict_wrapping:
503 raise NotImplementedError(
504 f"Python object wrapper for foreign type ID {wrapper_type} is not yet implemented",
505 )
506 return wrap_cffi_native_handle(obj, type_id, release_native)
507 s = signature(wrapper_type)
508 n = len(s.parameters)
509 parameters = [v for k, v in s.parameters.items()]
510 # [<Parameter "handle: Any">, <Parameter "release_native: Callable[[Any], NoneType]">, <Parameter "type_id: Optional[str] = None">, <Parameter "prior_ref_count: int = 0">]
511 if n == 0:
512 raise TypeError(
513 f"Wrapper class '{wrapper_type.__name__}' has no constructor arguments; at least one is required",
514 )
515 if n == 1:
516 return wrapper_type(obj)
517 if n == 2: # noqa: PLR2004
518 if release_native is None:
519 raise ValueError(
520 f"Wrapper class '{wrapper_type.__name__}' has two constructor arguments; the argument 'release_native' cannot be None",
521 )
522 return wrapper_type(obj, release_native)
523 if n == 3: # noqa: PLR2004
524 if release_native is None:
525 raise ValueError(
526 f"Wrapper class '{type(wrapper_type)}' has three constructor arguments; the argument 'release_native' cannot be None",
527 )
528 return wrapper_type(obj, release_native, type_id)
529 if n == 4: # noqa: PLR2004
530 p = parameters[3]
531 # constructor = wrapper_type.__init__
532 # type_hints = get_type_hints(constructor)
533 param_type = p.annotation
534 if param_type is not int:
535 raise TypeError(
536 f"Wrapper class '{type(wrapper_type)}' has four constructor arguments; the last argument 'prior_ref_count' must be an integer",
537 )
538 if parameters[3].default == parameters[3].empty: 538 ↛ 539line 538 didn't jump to line 539 because the condition on line 538 was never true
539 raise ValueError(
540 f"Wrapper class '{type(wrapper_type)}' has four constructor arguments; the last argument 'prior_ref_count' must have a default value",
541 )
542 return wrapper_type(obj, release_native, type_id)
543 raise NotImplementedError(
544 f"Wrapper class '{wrapper_type.__name__}' has more than 4 arguments; this is not supported",
545 )
548WrapperCreationFunction = Callable[[Any, str, Callable], DeletableCffiNativeHandle]