Closed
Description
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
Labels
bugs and security fixesbugs and security fixes(Objects, Python, Grammar, and Parser dirs)(Objects, Python, Grammar, and Parser dirs)An unexpected behavior, bug, or errorAn unexpected behavior, bug, or error