Merge pull request #18 from ionite34/dev

docs/source/conf.py

@@ -15,7 +15,7 @@
 project = "einspect"
 copyright = "2023, Ionite"
 author = "Ionite"
-release = "v0.4.3"
+release = "v0.4.4"
 
 
 # -- General configuration ---------------------------------------------------

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "einspect"
-version = "0.4.3"
+version = "0.4.4"
 description = "Extended Inspect - view and modify memory structs of runtime objects."
 authors = ["ionite34 <dev@ionite.io>"]
 license = "MIT"

src/einspect/__init__.py

@@ -9,4 +9,6 @@
 
 __all__ = ("view", "unsafe", "impl", "orig")
 
+__version__ = "0.4.4"
+
 unsafe: ContextManager[None] = global_unsafe

src/einspect/api.py

@@ -8,14 +8,17 @@
 from einspect.compat import Version, python_req
 from einspect.protocols.delayed_bind import bind_api
 
-__all__ = ("Py", "Py_ssize_t", "Py_hash_t", "PyObj_FromPtr")
+__all__ = ("Py", "Py_ssize_t", "Py_hash_t", "uintptr_t", "PyObj_FromPtr")
 
 Py_ssize_t = ctypes.c_ssize_t
 """Constant for type Py_ssize_t."""
 
 Py_hash_t = ctypes.c_uint64
 """Constant for type Py_hash_t."""
 
+uintptr_t = ctypes.c_uint64
+"""Constant for type uintptr_t."""
+
 ObjectOrRef = Union[py_object, object]
 IntSize = Union[int, Py_ssize_t]
 PyObjectPtr = POINTER(py_object)

src/einspect/structs/py_gc.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from ctypes import Structure, c_void_p, cast
+from enum import IntEnum
+
+from typing_extensions import Annotated
+
+from einspect.api import uintptr_t
+from einspect.structs.deco import struct
+from einspect.types import AsRef, ptr
+
+
+class PyGC(IntEnum):
+    """Bit flags for _gc_prev"""
+
+    # Bit 0 is set when tp_finalize is called
+    PREV_MASK_FINALIZED = 1
+    # Bit 1 is set when the object is in generation which is GCed currently.
+    PREV_MASK_COLLECTING = 2
+    # The (N-2) most significant bits contain the real address.
+    PREV_SHIFT = 2
+    PREV_MASK = uintptr_t(-1).value << PREV_SHIFT
+
+
+@struct
+class PyGC_Head(Structure, AsRef):
+    """
+    Defines the PyGC_Head Structure.
+
+    GC information is stored BEFORE the object structure.
+    https://github.com/python/cpython/blob/3.11/Include/internal/pycore_gc.h#L11-L20
+    """
+
+    # Pointer to next object in the list.
+    # 0 means the object is not tracked
+    _gc_next: Annotated[int, uintptr_t]
+    # Pointer to previous object in the list.
+    # Lowest two bits are used for flags.
+    _gc_prev: Annotated[int, uintptr_t]
+
+    def Next(self) -> ptr[PyGC_Head]:
+        return cast(self._gc_next, ptr[PyGC_Head])
+
+    def Set_Next(self, item: ptr[PyGC_Head]) -> None:
+        addr = cast(item, c_void_p).value
+        self._gc_next = addr
+
+    def Prev(self) -> ptr[PyGC_Head]:
+        return cast(self._gc_prev & PyGC.PREV_MASK, ptr[PyGC_Head])
+
+    def Set_Prev(self, item: ptr[PyGC_Head]) -> None:
+        item = cast(item, c_void_p).value
+        if item & ~PyGC.PREV_MASK != 0:
+            raise ValueError("item is not valid")
+        self._gc_prev = (self._gc_prev & ~PyGC.PREV_MASK) | item
+
+    def Finalized(self) -> bool:
+        return (self._gc_prev & PyGC.PREV_MASK_FINALIZED) != 0
+
+    def Set_Finalized(self) -> None:
+        self._gc_prev |= PyGC.PREV_MASK_FINALIZED

src/einspect/structs/py_object.py

@@ -10,7 +10,8 @@
 from einspect.compat import Version, python_req
 from einspect.protocols.delayed_bind import bind_api
 from einspect.structs.deco import struct
-from einspect.types import ptr
+from einspect.structs.py_gc import PyGC_Head
+from einspect.types import AsRef, ptr
 
 if TYPE_CHECKING:
     from einspect.structs import PyTypeObject
@@ -23,7 +24,7 @@
 
 
 @struct
-class PyObject(Structure, Generic[_T, _KT, _VT]):
+class PyObject(Structure, AsRef, Generic[_T, _KT, _VT]):
     """Defines a base PyObject Structure."""
 
     ob_refcnt: int
@@ -93,9 +94,39 @@ def into_object(self) -> _T:
         py_obj = ctypes.cast(self.as_ref(), ctypes.py_object)
         return py_obj.value
 
-    def as_ref(self) -> ptr[Self]:
-        """Return a pointer to the PyObject."""
-        return ctypes.pointer(self)
+    def is_gc(self) -> bool:
+        """
+        Returns True if the object implements the GC protocol.
+
+        The object cannot be tracked by the garbage collector if False.
+
+        https://github.com/python/cpython/blob/3.11/Include/internal/pycore_object.h#L209-L216
+        """
+        return (type_obj := self.ob_type.contents).is_gc() and (
+            not type_obj.tp_is_gc or type_obj.tp_is_gc(self.into_object())
+        )
+
+    def as_gc(self) -> PyGC_Head:
+        """Return the PyGC_Head struct of this object."""
+        addr = self.address - ctypes.sizeof(PyGC_Head)
+        return PyGC_Head.from_address(addr)  # type: ignore
+
+    def gc_is_tracked(self) -> bool:
+        """Return True if the PyObject is currently tracked by the GC."""
+        return self.as_gc()._gc_next != 0
+
+    def gc_may_be_tracked(self) -> bool:
+        """
+        Return True if the PyObject may be tracked by
+        the GC in the future, or already is.
+
+        https://github.com/python/cpython/blob/3.11/Include/internal/pycore_gc.h#L28-L32
+        """
+        if not self.is_gc():
+            return False
+        if self.ob_type.contents.into_object() is tuple:
+            return self.gc_is_tracked()
+        return True
 
     @bind_api(python_req(Version.PY_3_10) or pythonapi["Py_NewRef"])
     def NewRef(self) -> object:

src/einspect/structs/py_type.py

@@ -143,6 +143,15 @@ def setattr_safe(self, name: str, value: Any) -> None:
     def Modified(self) -> None:
         """Mark the type as modified."""
 
+    def is_gc(self) -> bool:
+        """
+        Return True if the type has GC support.
+
+        https://docs.python.org/3/c-api/type.html#c.PyType_IS_GC
+        https://github.com/python/cpython/blob/3.11/Include/objimpl.h#L160-L161
+        """
+        return bool(self.tp_flags & TpFlags.HAVE_GC)
+
 
 # Mapping of CField name to type
 # noinspection PyProtectedMember

src/einspect/type_orig.py

@@ -1,26 +1,27 @@
 """Proxy for retrieving original methods and slot wrappers of types."""
 from __future__ import annotations
 
-from types import WrapperDescriptorType
-from typing import Type, TypeVar
+from typing import Any, Type, TypeVar
 
 _T = TypeVar("_T")
 MISSING = object()
 
+obj_getattr = object.__getattribute__
 type_hash = type.__hash__
+str_eq = str.__eq__
 dict_setdefault = dict.setdefault
 dict_contains = dict.__contains__
 dict_get = dict.get
 dict_getitem = dict.__getitem__
 
-_slots_cache: dict[type, dict[str, WrapperDescriptorType]] = {}
+_slots_cache: dict[type, dict[str, Any]] = {}
 
 
-def add_cache(type_: type, method_name: str, method: WrapperDescriptorType):
+def add_cache(type_: type, name: str, method: Any):
     """Add a method to the cache."""
     type_methods = dict_setdefault(_slots_cache, type_, {})
     # Only allow adding once, ignore if already added
-    dict_setdefault(type_methods, method_name, method)
+    dict_setdefault(type_methods, name, method)
 
 
 def in_cache(type_: type, name: str) -> bool:
@@ -29,7 +30,7 @@ def in_cache(type_: type, name: str) -> bool:
     return dict_contains(type_methods, name)
 
 
-def get_cache(type_: type, name: str) -> WrapperDescriptorType:
+def get_cache(type_: type, name: str) -> Any:
     """Get the method from the type in cache."""
     type_methods = dict_setdefault(_slots_cache, type_, {})
     return dict_getitem(type_methods, name)
@@ -58,8 +59,8 @@ def __call__(self, *args, **kwargs):
     def __getattribute__(self, name: str):
         """Get an attribute from the original type."""
         # Overrides
-        _type = object.__getattribute__(self, "_orig__type")
-        if name in ("_orig__type",):
+        _type = obj_getattr(self, "_orig__type")
+        if str_eq(name, "_orig__type"):
             return _type
 
         # Check if the attribute is cached

src/einspect/types.py

@@ -7,7 +7,7 @@
 
 from typing_extensions import Self
 
-__all__ = ("ptr", "Array", "_SelfPtr")
+__all__ = ("ptr", "Array", "AsRef", "_SelfPtr")
 
 _T = TypeVar("_T")
 
@@ -81,3 +81,9 @@ def __getitem__(self, item):
             "Array": ctypes.Array,
         }
     )
+
+
+class AsRef:
+    def as_ref(self) -> ptr[Self]:
+        """Return a pointer to the Structure."""
+        return ctypes.pointer(self)  # type: ignore

src/einspect/views/view_base.py

@@ -172,6 +172,25 @@ def mem_size(self) -> int:
         """Memory size of the object in bytes."""
         return sizeof(self._pyobject)
 
+    def is_gc(self) -> bool:
+        """
+        Returns True if the object implements the Garbage Collector protocol.
+
+        If True, a PyGC_HEAD struct will precede the object struct in memory.
+        """
+        return self._pyobject.is_gc()
+
+    def gc_may_be_tracked(self) -> bool:
+        """
+        Return True if the object may be tracked by
+        the GC in the future, or already is.
+        """
+        return self._pyobject.gc_may_be_tracked()
+
+    def gc_is_tracked(self) -> bool:
+        """Returns True if the object is tracked by the GC."""
+        return self._pyobject.is_gc() and self._pyobject.gc_is_tracked()
+
     def drop(self) -> None:
         """
         Drop all references to the base object.

src/einspect/views/view_tuple.py

@@ -5,7 +5,6 @@
 from typing import Sequence, TypeVar, overload
 
 from einspect.api import Py_ssize_t
-from einspect.compat import abc
 from einspect.errors import UnsafeIndexError
 from einspect.structs import PyObject, PyTupleObject
 from einspect.utils import new_ref
@@ -17,7 +16,7 @@
 _VT = TypeVar("_VT")
 
 
-class TupleView(VarView[tuple, None, _VT], abc.Sequence):
+class TupleView(VarView[tuple, None, _VT]):
     _pyobject: PyTupleObject[_VT]
 
     @overload

src/einspect/views/view_type.py

@@ -2,7 +2,7 @@
 
 from collections.abc import Generator
 from contextlib import contextmanager
-from typing import Any, Type, TypeVar
+from typing import Any, Callable, Type, TypeVar
 
 from typing_extensions import Self
 
@@ -19,6 +19,7 @@
 MISSING = object()
 
 _T = TypeVar("_T")
+_Fn = TypeVar("_Fn", bound=Callable)
 
 
 class TypeView(VarView[_T, None, None]):
@@ -56,8 +57,10 @@ def as_mutable(self) -> Generator[Self, None, None]:
             yield self
             return
         self.immutable = False
+        self._pyobject.Modified()
         yield self
         self.immutable = True
+        self._pyobject.Modified()
 
     def _try_alloc(self, slot: Slot):
         # Check if there is a ptr class
@@ -87,12 +90,8 @@ def __setitem__(self, key: str, value):
             self._try_alloc(slot)
 
         with self.as_mutable():
-            self._pyobject.Modified()
             self._pyobject.setattr_safe(key, value)
 
-        # Invalidate type lookup cache
-        self._pyobject.Modified()
-
     def __getattr__(self, item: str):
         # Forward `tp_` attributes from PyTypeObject
         if item.startswith("tp_"):
@@ -111,14 +110,14 @@ def __setattr__(self, key: str, value: Any) -> None:
         super().__setattr__(key, value)
 
 
-def impl(cls: Type[_T]):
+def impl(cls: Type[_T]) -> Callable[[_Fn], _Fn]:
     """Decorator for implementing methods on built-in types."""
     if not isinstance(cls, type):
         raise TypeError("cls must be a type")
 
     t_view = TypeView(cls)
 
-    def wrapper(func):
+    def wrapper(func: _Fn) -> _Fn:
         if isinstance(func, property):
             name = func.fget.__name__
         else: