Shared memory for communicating Numpy arrays

1_worker_producer_shared_memory.py

def worker_producer_shared_memory(np_arr_shape, shared_memory, n_frames):
    """A frame producer function that writes to shared memory"""
    mp_array, np_array = shared_memory
    for _ in range(n_frames):
        mp_array.acquire()
        np_array[:] = prepare_random_frame(np_arr_shape)  # produce a fresh array

2_worker_consumer.py

def consumer_shared_memory(n_frames, shared_memory):
    """A frame consumer function, which draws frames from the worker process via shared memory."""
    mp_array, np_array = shared_memory
    for _ in range(n_frames):
        _ = np_array.astype("uint8").copy() * 2  # example of some processing done on the array
        while True:
            try:
                mp_array.release()
                break
            # it already unlocked, wait until its locked again which means a new frame is ready
            except ValueError:
                time.sleep(0.001)

3_minimal_example_shared_memory_application.py

mp_array = mp.Array("I", int(np.prod(np_arr_shape)), lock=mp.Lock())
# create a numpy view of the array
np_array = np.frombuffer(mp_array.get_obj(), dtype="I").reshape(np_arr_shape)
shared_memory = (mp_array, np_array)
proc = mp.Process(target=worker_producer_shared_memory,
                  args=(np_arr_shape, shared_memory, n_frames))
proc.start()
# will consume n_frames from the producer
consumer_shared_memory(n_frames, shared_memory)