bases.py 3.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108
  1. # SPDX-FileCopyrightText: 2024 geisserml <geisserml@gmail.com>
  2. # SPDX-License-Identifier: Apache-2.0 OR BSD-3-Clause
  3. __all__ = ("AutoCastable", "AutoCloseable", "DEBUG_AUTOCLOSE", "LIBRARY_AVAILABLE")
  4. import os
  5. import sys
  6. import ctypes
  7. import weakref
  8. import logging
  9. import uuid
  10. logger = logging.getLogger(__name__)
  11. # mutable bools
  12. DEBUG_AUTOCLOSE = ctypes.c_bool(False)
  13. LIBRARY_AVAILABLE = ctypes.c_bool(False) # set to true on library init
  14. STATE_INVALID = -1
  15. STATE_AUTO = 0
  16. STATE_EXPLICIT = 1
  17. STATE_BYPARENT = 2
  18. class AutoCastable:
  19. @property
  20. def _as_parameter_(self):
  21. return self.raw
  22. def _close_template(close_func, raw, obj_repr, state, parent, *args, **kwargs):
  23. if DEBUG_AUTOCLOSE:
  24. desc = {STATE_AUTO: "auto", STATE_EXPLICIT: "explicit", STATE_BYPARENT: "by parent"}[state.value]
  25. # use os.write() rather than print() to avoid "reentrant call" exceptions on shutdown (see https://stackoverflow.com/q/75367828/15547292)
  26. os.write(sys.stderr.fileno(), f"Close ({desc}) {obj_repr}\n".encode())
  27. if not LIBRARY_AVAILABLE:
  28. os.write(sys.stderr.fileno(), f"-> Cannot close object, library is destroyed. This may cause a memory leak!\n".encode())
  29. return
  30. assert (parent is None) or not parent._tree_closed()
  31. close_func(raw, *args, **kwargs)
  32. class AutoCloseable (AutoCastable):
  33. def __init__(self, close_func, *args, obj=None, needs_free=True, **kwargs):
  34. # NOTE proactively prevent accidental double initialization
  35. assert not hasattr(self, "_finalizer")
  36. self._close_func = close_func
  37. self._obj = self if obj is None else obj
  38. self._uuid = uuid.uuid4()
  39. self._ex_args = args
  40. self._ex_kwargs = kwargs
  41. self._autoclose_state = ctypes.c_int8(STATE_AUTO) # mutable int
  42. self._finalizer = None
  43. self._kids = []
  44. if needs_free:
  45. self._attach_finalizer()
  46. def __repr__(self):
  47. return f"<{type(self).__name__} uuid:{str(self._uuid)[:8]}>"
  48. def _attach_finalizer(self):
  49. # NOTE this function captures the value of the `parent` property at finalizer installation time - if it changes, detach the old finalizer and create a new one
  50. assert self._finalizer is None
  51. self._finalizer = weakref.finalize(self._obj, _close_template, self._close_func, self.raw, repr(self), self._autoclose_state, self.parent, *self._ex_args, **self._ex_kwargs)
  52. def _detach_finalizer(self):
  53. self._finalizer.detach()
  54. self._finalizer = None
  55. def _tree_closed(self):
  56. if self.raw is None:
  57. return True
  58. if (self.parent is not None) and self.parent._tree_closed():
  59. return True
  60. return False
  61. def _add_kid(self, k):
  62. self._kids.append( weakref.ref(k) )
  63. def close(self, _by_parent=False):
  64. if not self.raw or not self._finalizer:
  65. return False
  66. for k_ref in self._kids:
  67. k = k_ref()
  68. if k and k.raw:
  69. k.close(_by_parent=True)
  70. self._autoclose_state.value = STATE_BYPARENT if _by_parent else STATE_EXPLICIT
  71. self._finalizer()
  72. self._autoclose_state.value = STATE_INVALID
  73. self.raw = None
  74. self._finalizer = None
  75. self._kids.clear()
  76. return True