Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

Appearance settings

Materialized instance dictionaries are not GC-tracked #133543

Copy link
Copy link
Closed
@emontnemery

Description

@emontnemery
Issue body actions

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 fixesbugs and security fixesinterpreter-core(Objects, Python, Grammar, and Parser dirs)(Objects, Python, Grammar, and Parser dirs)type-bugAn unexpected behavior, bug, or errorAn unexpected behavior, bug, or error

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions

      Morty Proxy This is a proxified and sanitized view of the page, visit original site.