apppublic/appPublic/psm.py
2025-04-10 06:07:26 +00:00

57 lines
1.6 KiB
Python

import json
from time import sleep
from multiprocessing import Manager
from multiprocessing.shared_memory import SharedMemory
from multiprocessing.resource_tracker import unregister
class PSharedMemory:
tailstring=b'#:@#'
def __init__(self, name, datalen, data=None):
self.sm = None
self.name = name
self.datalen = datalen
self.lock = Manager().Lock()
if data:
self.sm = SharedMemory(name=self.name, create=True, size=self.datalen)
self.creator = True
self.set(data)
else:
self.sm = SharedMemory(name=self.name)
unregister(self.sm._name, 'shared_memory')
self.creator = False
def get(self):
b = self.sm.buf.tobytes().split(self.tailstring)[0]
d = b.decode('utf-8')
return json.loads(d)
def set(self, data):
with self.lock:
d = json.dumps(data)
b = d.encode('utf-8') + self.tailstring
if self.datalen < len(b):
raise Exception(f'SharedMemory allocated size is {self.datalen} set size is {len(b)}')
self.sm.buf[:len(b)] = b
def __del__(self):
if self.sm is None:
return
self.sm.close()
if self.creator:
self.sm.unlink()
if __name__ == '__main__':
import sys
data = {
"aaa":"134902t34gf",
"bbb":36
}
if len(sys.argv) > 1:
sm = PSharedMemory('rtgerigreth', datalen=200, data=data)
sleep(10000)
else:
sm = PSharedMemory('rtgerigreth', datalen=200 )
x = sm.get()
print(f'data in shared memory: {x}')