from core.cache import cached_property, cached_result
from entities.constants import WORLD_ENTITY_INDEX
from entities.entity import BaseEntity, Entity
from entities.helpers import wrap_entity_mem_func
from memory import alloc, Pointer
from gc import collect, is_tracked
from random import randint
from sys import getrefcount
from weakref import ref
world = Entity(WORLD_ENTITY_INDEX)
server_class = world.server_class
assert server_class is world.server_class
Entity.server_class.delete_cached_value(world)
assert server_class is not world.server_class
BaseEntity.server_class.set_cached_value(world, server_class)
assert server_class is world.server_class
del world, server_class
class World(Entity):
caching = True
@wrap_entity_mem_func
def set_transmit(self, *args):
return args
set_transmit = World(WORLD_ENTITY_INDEX).set_transmit
assert set_transmit is World(WORLD_ENTITY_INDEX).set_transmit
del World.cache[WORLD_ENTITY_INDEX]
assert set_transmit is not World(WORLD_ENTITY_INDEX).set_transmit
del set_transmit
# World.cache[WORLD_ENTITY_INDEX], World.cache[WORLD_ENTITY_INDEX].set_transmit.wrapped_self
assert getrefcount(World(WORLD_ENTITY_INDEX)) == 3
r = ref(World(WORLD_ENTITY_INDEX))
assert r()
del World.cache[WORLD_ENTITY_INDEX]
collect()
assert not r()
assert getrefcount(World(WORLD_ENTITY_INDEX, False).set_transmit.wrapped_self) == 2
class Test:
__slots__ = [
'__dict__',
'__weakref__',
]
@cached_property
def test(self):
return self
@cached_property
@property
def test_wrapped(self):
return self
@cached_result
def test_result(self):
return self
test = Test()
# Everything should be the same value
assert test is test.test is test.test_wrapped is test.test_result()
# test, test.test, test.test_wrapped, test.test_result, test.test_result.__self__
assert getrefcount(test) == 6
# Should now be 4, because test.test and test.test_wrapped were removed from the cache
del test.test
Test.test_wrapped.delete_cached_value(test)
assert getrefcount(test) == 4
r = ref(test)
assert r()
del test
# We should still have 4 circular references
assert r()
collect()
# Everything should be collected at that point
assert not r()
class Test:
@cached_property(kwargs=dict(range=(0, 1000)))
def test(self, range):
return randint(*range)
@test.setter
def set_test(self, value, range):
return int(value / 2)
test = Test()
# Compute and cache the value for the first time
i = test.test
# The first computed value was cached, so it should always be the same
assert i is test.test
assert i is test.test
assert i is test.test
assert i is test.test
# Deleting the property is invalidating the cache
del test.test
assert i is not test.test
# The cache will be updated to 5, because our setter computes value / 2
test.test = 10
assert test.test is 5
# The new value should be 1, because we updated our userdata
Test.test['range'] = (1, 1)
del test.test
assert test.test is 1
class Ptr(Pointer):
...
ptr = Ptr(alloc(1024, False).address, True)
# ptr should not be tracked, because it has no __dict__
assert not is_tracked(ptr)
ptr.self = ptr
# Now that ptr has a __dict__, it should be tracked
assert is_tracked(ptr)
r = ref(ptr)
# Our ref should be alive because ptr is still referenced
assert r()
del ptr
# Our ref should still be held hostage
assert r()
collect()
# Now that ptr has been deleted and collected, our ref should be dead
assert not r()
# Let's push the thresholds and get the garbage collector busy
for _ in range(1000000):
ptr = alloc(1024000, True)
ptr.self = ptr
del ptr
collect()
Dict traversal was implemented in accordance with Supporting Cyclic Garbage Collection.
Tests: