-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
62 lines (48 loc) · 1.93 KB
/
Copy pathmain.py
File metadata and controls
62 lines (48 loc) · 1.93 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
from multiprocessing import shared_memory
import pickle
# the class does not use lock, user is responsible in using locks
# todo cannot add methods that does not start with '_'
class ProcessSafeClass:
def __init__(self):
# Create shared memory for attribute dictionary
self._shm = shared_memory.SharedMemory(create=True, size=4096)
self._initialize_dict()
def _initialize_dict(self):
# Initialize an empty dictionary in shared memory
initial_dict = {}
encoded_dict = pickle.dumps(initial_dict)
buffer = self._shm.buf
buffer[:len(encoded_dict)] = encoded_dict
def _get_dict(self):
# Retrieve the dictionary from shared memory
buffer = self._shm.buf
encoded_dict = bytes(buffer[:self._shm.size])
return pickle.loads(encoded_dict)
def _set_dict(self, dictionary):
# Store the dictionary in shared memory
encoded_dict = pickle.dumps(dictionary)
if len(encoded_dict) > self._shm.size:
raise ValueError("Shared memory size is too small to hold the dictionary")
# todo raise the shm size and continue
buffer = self._shm.buf
buffer[:len(encoded_dict)] = encoded_dict
def __getattribute__(self, item):
if item.startswith('_'):
return super().__getattribute__(item)
shm_dict = self._get_dict()
if item in shm_dict:
return shm_dict[item]
else:
raise AttributeError(f"'ProcessSafeClass' object has no attribute '{item}'")
def __setattr__(self, attr_name, attr):
# Custom attribute setting
if attr_name.startswith('_'):
super().__setattr__(attr_name, attr)
else:
shm_dict = self._get_dict()
shm_dict[attr_name] = attr
self._set_dict(shm_dict)
def _cleanup(self):
# Clean up shared memory
self._shm.close()
self._shm.unlink()