bug修复

Administrator

2025-05-23 c285883d71ef8a362b012983dadc7ce4256b40f6

 test/test_shared_memery.py

@@ -1,23 +1,33 @@
import multiprocessing.shared_memory
import multiprocessing
import time
from multiprocessing import Manager, Process


def producer(shr):
    with shr.txn() as shm:
        for i in range(10):
            shm[i] = i
def producer(d):
    l = []
    for i in range(10):
        d.append(i)
        if i == 4:
            d[:] = []
        time.sleep(0.1)


def consumer(shr):
    with shr.txn() as shm:
        for i in range(10):
            print(shm[i])
def consumer(d):
    for i in range(10000):
        start_time = time.time()
        print(len(d))
        end_time = time.time()
        print("耗时：", end_time - start_time)
        time.sleep(1)


if __name__ == "__main__":
    with multiprocessing.shared_memory.SharedMemory(create=True, size=80) as shr:
        p1 = multiprocessing.Process(target=producer, args=(shr,))
        p2 = multiprocessing.Process(target=consumer, args=(shr,))
    with Manager() as manager:
        d = manager.list()
        p1 = multiprocessing.Process(target=producer, args=(d,))
        for i in range(100):
            p2 = multiprocessing.Process(target=consumer, args=(d,))
            p2.start()
        p1.start()
        p2.start()
        p1.join()
        p2.join()
        input()