Skip to content

Instantly share code, notes, and snippets.

@VieVie31
Created March 1, 2019 14:09
Show Gist options
  • Star 1 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save VieVie31/9d2ddebce576b50f9892723d3b545013 to your computer and use it in GitHub Desktop.
Save VieVie31/9d2ddebce576b50f9892723d3b545013 to your computer and use it in GitHub Desktop.
just a test with a class with only one attribute... the goal is to serialize the attributes of the class to free memory when too much memory is used... (python 3.7 only)
import gc
import os
import time
import pickle
import psutil
import pathlib
import threading
from dataclasses import dataclass, field
from typing import Any
class CacheManager:
keep = 1 #NB Object allowed to be kept in memory >= 1
period = 2 #NB sec before next clear
memory_threshold = 0 #Do not clear if below this percentage [0, 100]
__objects = []
__started = False
def run():
if not CacheManager.__started:
CacheManager.__started = True
CacheManager.__periodic_check()
return CacheManager
def stop():
CacheManager.__started = False
return CacheManager
def __periodic_check():
if not CacheManager.__started:
return
if psutil.virtual_memory().percent > CacheManager.memory_threshold:
CacheManager.__clear()
threading.Timer(CacheManager.period, CacheManager.__periodic_check).start()
def __clear():
objects = sorted(filter(lambda obj: obj.ts() > 0., CacheManager.__objects), key=lambda obj: obj.ts())
for obj in objects[:-CacheManager.keep]:
obj.free()
gc.collect()
def add(obj):
CacheManager.__objects.append(obj)
def remove(obj):
if obj in CacheManager.__objects:
CacheManager.__objects.remove(obj)
class Empty: #used to not mislead a real `None` value and an empty serialized value
def __repr__(self):
print("SerializedValue")
@dataclass
class MaClasse:
#static
__cm = CacheManager.run()
#instance
_x : Any = field(default_factory=Empty)
_x_ts : float = 0.
def __post_init__(self):
MaClasse.__cm.add(self)
self.__pid = os.getpid()
if not pathlib.Path(f"/tmp/cm_{self.__pid}").exists():
os.makedirs(f"/tmp/cm_{self.__pid}")
def _dumps(value, path):
with open(path, "wb") as f:
f.write(pickle.dumps(value))
def _loads(path):
if not pathlib.Path(path).exists():
return None
with open(path, "rb") as f:
print("loaded from disk") #just to see how well it's working
return pickle.load(f)
@property
def x(self):
self._x_ts = time.time()
if type(self._x) != type(Empty()):
return self._x
self._x = MaClasse._loads(f"/tmp/cm_{self.__pid}/{id(self)}_x.pkl")
return self._x
@x.setter
def x(self, value):
MaClasse._dumps(value, f"/tmp/cm_{self.__pid}/{id(self)}_x.pkl")
self._x = value
self._x_ts = time.time()
@x.deleter
def x(self):
if pathlib.Path(f"/tmp/cm_{self.__pid}/{id(self)}_x.pkl").exists():
os.remove(f"/tmp/cm_{self.__pid}/{id(self)}_x.pkl")
del self._x
self._x_ts = 0.
gc.collect()
def free(self):
if type(self._x) == type(Empty()):
return
MaClasse._dumps(self._x, f"/tmp/cm_{self.__pid}/{id(self)}_x.pkl")
del self._x
gc.collect()
self._x = Empty()
def ts(self):
return self._x_ts
def __del__(self):
#FIXME: it seems like `del var` do not call `var.__del__()`
MaClasse.__cm.remove(self)
del self.x
print(f"Serialized values will be stored on : /tmp/cm_{os.getpid()}/")
a = MaClasse()
a.x = 5
a.free()
assert a.x == 5
lst = [MaClasse() for i in range(10)]
for i, mc in enumerate(lst):
mc.x = i
print(mc.x)
time.sleep(3)
assert a.x == 5
print(lst[0].x)
print(a.x)
print(a.x)
#don't forget to remove the serialized objects when finished...
#rm -fr /tmp/cm_{os.getpid()}
#TODO: make the attributes creation process easier (adding properties on the fly for examples)...
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment