How to handle errors when using UltraDict in multiprocessing and how to asynchronously wait for UltraDict to be written to in the main process.

[xingguang12]

`import time
import random
import threading
from multiprocessing import Process, Manager
from UltraDict import UltraDict

class A():
def init(self, name, dictname):
self.name = name
self.dictname = dictname
self.personnum = -1
self.time = time.time()
self.task = threading.Thread(target=self.a)

def a(self):
    while True:
        self.personnum = random.random()
        self.time = time.time()
        other = UltraDict(name=self.dictname)
        other[self.name] = {"time": self.time, "num": self.personnum}
        time.sleep(1)

def start(self):
    self.task.start()

class ProcessClass():
def init(self, i, name):
self.dict1 = {}
self.dictname = name
self._process = Process(target=self.run, args=(i,))

def run(self, i):
    for j in range(5):
        # 每个 A 对象与共享字典关联
        a = A(str(i) + str(j), self.dictname)
        a.start()
        self.dict1[str(i) + str(j)] = a
    while True:
        time.sleep(1)  # 保持进程运行

def start(self):
    self._process.start()

if name == 'main':
ultra = UltraDict(name='getpic', shared_lock=True)
name = ultra.name
processes = []
for i in range(5):
process_instance = ProcessClass(i, name)
process_instance.start()
processes.append(process_instance)

while True:
    time.sleep(2)  # 每隔2秒打印一次
    print(ultra)
    # print("实时获取的数据:", dict(shared_dict))`
   UltraDict.Exceptions.CannotAttachSharedMemory: Could not get memory 'wnsm_cfd5be17'

[ronny-rentner]

Hey, what kind of errors? Can you reformat your code? I only speak English and German.

To your second question: You don't need to wait for UltraDict. It will make sure that you always have the newest data. You can work with a lock as explained in the docs https://github.com/ronny-rentner/UltraDict?tab=readme-ov-file#locking if your use case requires it.

PS: Are you using Windows? Check out the issue mentioned here: https://github.com/ronny-rentner/UltraDict?tab=readme-ov-file#issues

[xingguang12]

import time
import random
import threading
from multiprocessing import Process, Manager
from UltraDict import UltraDict

class A():
    def __init__(self, name, dictname):
        self.name = name
        self.dictname = dictname
        self.personnum = -1
        self.time = time.time()
        self.task = threading.Thread(target=self.a)

    def a(self):
        while True:
            self.personnum = random.random()
            self.time = time.time()
            other = UltraDict(name=self.dictname)
            other[self.name] = {"time": self.time, "num": self.personnum}
            time.sleep(1)

    def start(self):
        self.task.start()


class ProcessClass():
    def __init__(self, i, name):
        self.dict1 = {}
        self.dictname = name
        self._process = Process(target=self.run, args=(i,))

    def run(self, i):
        for j in range(5):
            a = A(str(i) + str(j), self.dictname)
            a.start()
            self.dict1[str(i) + str(j)] = a
        while True:
            time.sleep(1)

    def start(self):
        self._process.start()


if __name__ == '__main__':
    ultra = UltraDict(name='getpic', shared_lock=True)
    name = ultra.name
    processes = []
    for i in range(5):
        process_instance = ProcessClass(i, name)
        process_instance.start()
        processes.append(process_instance)

    while True:
        time.sleep(2)
        print(ultra)

[xingguang12]

the code raise UltraDict.Exceptions.CannotAttachSharedMemory: Could not get memory 'wnsm_0cbb1722'

[ronny-rentner]

Thanks for providing the code sample. I'll check it out later today and then get back to you.

[ronny-rentner]

@xingguang12

I've tried your code on Linux and Windows with the dev branch. On Linux, it runs just fine. On Windows, I get the same error as you.

When you do other = UltraDict(name=self.dictname) in your function a(), you create a new connection to the shared memory for every single time you write something to the UltraDict. This is quite inefficient and as it seems, on Windows you get issues when you do multiple connections to the same shared memory (UltraDict) in parallel threads.

What has worked for me on Windows is using only one central UltraDict per process which is being used in each thread like this:

class A():
    def __init__(self, name, dictname):
        self.name = name
        self.dictname = dictname
        self.personnum = -1
        self.time = time.time()
        self.other = UltraDict(name=self.dictname)
        self.task = threading.Thread(target=self.a)

    def a(self):
        while True:
            self.personnum = random.random()
            self.time = time.time()
            self.other[self.name] = {"time": self.time, "num": self.personnum}
            time.sleep(1)

    def start(self):
        self.task.start()
                                 

Please let me know if this works for you.

Depending on your real data, it might also be advisable to increase the buffer_size.

[xingguang12]

thanks a lot