Skip to content

Materialized instance dictionaries are not GC-tracked #133543

@emontnemery

Description

@emontnemery

Bug report

Bug description:

Objects can't be garbage collected after accessing __dict__. The issue was noticed on a class with many cached_property which store the cache entries by directly accessing the instance __dict__.

If cached_property is modified by removing this line, the issue goes away:
https://github.com/python/cpython/blob/3.13/Lib/functools.py#L1028

The attached reproduction creates instances of two classes and checks if they are freed or not.

Reproduction of the issue with the attached pytest test cases:

mkdir cached_property_issue
python3 -m venv .venv
source .venv/bin/activate
pip3 install pytest
pytest test_cached_property_issue.py
from __future__ import annotations

from collections.abc import Generator
from functools import cached_property
import gc
from typing import Any, Self
import weakref

import pytest

hass_instances: list[weakref.ref[HackHomeAssistant]] = []
sensor_instances: list[weakref.ReferenceType[HackEntity]] = []


@pytest.fixture(autouse=True, scope="module")
def garbage_collection() -> Generator[None]:
    """Run garbage collection and check instances."""
    yield
    gc.collect()
    assert [bool(obj()) for obj in hass_instances] == [False, True]
    assert [bool(obj()) for obj in sensor_instances] == [False, True]


class HackEntity:
    """A class with many properties."""

    entity_id: str | None = None
    platform = None
    registry_entry = None
    _unsub_device_updates = None

    @cached_property
    def should_poll(self) -> bool:
        return False

    @cached_property
    def unique_id(self) -> None:
        return None

    @cached_property
    def has_entity_name(self) -> bool:
        return False

    @cached_property
    def _device_class_name(self) -> None:
        return None

    @cached_property
    def _unit_of_measurement_translation_key(self) -> None:
        return None

    @property
    def name(self) -> None:
        return None

    @cached_property
    def capability_attributes(self) -> None:
        return None

    @cached_property
    def state_attributes(self) -> None:
        return None

    @cached_property
    def extra_state_attributes(self) -> None:
        return None

    @cached_property
    def device_class(self) -> None:
        return None

    @cached_property
    def unit_of_measurement(self) -> None:
        return None

    @cached_property
    def icon(self) -> None:
        return None

    @cached_property
    def entity_picture(self) -> None:
        return None

    @cached_property
    def available(self) -> bool:
        return True

    @cached_property
    def assumed_state(self) -> bool:
        return False

    @cached_property
    def force_update(self) -> bool:
        return False

    @cached_property
    def supported_features(self) -> None:
        return None

    @cached_property
    def attribution(self) -> None:
        return None

    @cached_property
    def translation_key(self) -> None:
        return None

    @cached_property
    def options(self) -> None:
        return None

    @cached_property
    def state_class(self) -> None:
        return None

    @cached_property
    def native_unit_of_measurement(self) -> None:
        return None

    @cached_property
    def suggested_unit_of_measurement(self) -> None:
        return None

    @property
    def state(self) -> Any:
        self._unit_of_measurement_translation_key
        self.native_unit_of_measurement
        self.options
        self.state_class
        self.suggested_unit_of_measurement
        self.translation_key
        return None

    def async_write_ha_state(self) -> None:
        self._verified_state_writable = True
        self.capability_attributes

        self.available
        self.state
        self.extra_state_attributes
        self.state_attributes

        self.unit_of_measurement
        self.assumed_state
        self.attribution
        self.device_class
        self.entity_picture
        self.icon
        self._device_class_name
        self.has_entity_name
        self.supported_features
        self.force_update

    def async_on_remove(self) -> None:
        self._on_remove = []

    def add_to_platform_start(
        self, hass: HackHomeAssistant, platform: HackEntityPlatform
    ) -> None:
        self.hass = hass
        self.platform = platform


class CompensationSensor(HackEntity):
    """This concrete class won't be garbage collected."""

    def __init__(self) -> None:
        """Initialize the Compensation sensor."""
        self.__dict__
        sensor_instances.append(weakref.ref(self))
        self.parallel_updates = None
        self._platform_state = None

        self._state_info = {}
        self.async_write_ha_state()


class CompensationSensor2(HackEntity):
    """This concrete class will be garbage collected."""

    def __init__(self) -> None:
        """Initialize the Compensation sensor."""
        sensor_instances.append(weakref.ref(self))
        self.parallel_updates = None
        self._platform_state = None

        self._state_info = {}
        self.async_write_ha_state()


class HackHomeAssistant:
    """Root object of the Home Assistant home automation."""

    def __new__(cls) -> Self:
        """Set the _hass thread local data."""
        hass = super().__new__(cls)
        hass_instances.append(weakref.ref(hass))
        return hass


class HackEntityPlatform:
    """Manage the entities for a single platform."""

    def __init__(
        self,
        hass: HackHomeAssistant,
    ) -> None:
        """Initialize the entity platform."""
        self.hass = hass
        self.entities: dict[str, HackEntity] = {}

    def async_add_entity(
        self,
        new_entity: HackEntity,
    ) -> None:
        """Add entities for a single platform async."""

        entity = new_entity
        entity.add_to_platform_start(self.hass, self)

        entity.unique_id

        entity.entity_id = "sensor.test"
        entity_id = entity.entity_id
        self.entities[entity_id] = entity

        entity.async_on_remove()

        entity.should_poll


def test_limits1() -> None:
    """Create an object which will be garbage collected."""
    hass = HackHomeAssistant()
    sens = CompensationSensor2()
    entity_platform = HackEntityPlatform(hass)
    entity_platform.async_add_entity(sens)


def test_limits2() -> None:
    """Create an object which won't be garbage collected."""
    hass = HackHomeAssistant()
    sens = CompensationSensor()
    entity_platform = HackEntityPlatform(hass)
    entity_platform.async_add_entity(sens)


def test_limits3() -> None:
    """Last test, garbage collection check runs after this test."""

CPython versions tested on:

3.13

Operating systems tested on:

Linux

Linked PRs

Metadata

Metadata

Assignees

No one assigned

    Labels

    3.13bugs and security fixesinterpreter-core(Objects, Python, Grammar, and Parser dirs)type-bugAn unexpected behavior, bug, or error

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions