Generic array of objects using multiprocessing in python
I am trying to create an array of shared objects (class instances) between different processes in a python program. Each of these objects will be modified in the program. For this purpose, I use multiprocessing as follows:
import multiprocessing import numpy as np sh = multiprocessing.RawArray (ctypes.py_object, 10) f = np.frombuffer(sh, dtype=object)
and I get the following message:
Traceback (most recent call last):
File "<pyshell#14>", line 1, in <module>
f = np.frombuffer(sh, dtype=object)
ValueError: cannot create an OBJECT array from memory buffer
Now my question is, first of all, that this is generally the correct way to solve this problem, and second, what is my mistake in the above code? Thank you in advance.
source to share
ctypes.py_object
represents a PyObject *
in C. It is a pointer to struct
that represents a Python object, resides in the private memory of the process, and contains more pointers. Another process cannot access it; trying to share a pointer between processes is not helpful.
See also this answer from Alex Martelli .
source to share
You might want to use multiprocessing.Queue
one where you can just dump objects without worrying about types. It is also safe and process-safe.
Here's a simple example of a queue used to ease the Producer-Consumer problem ( original source , Pizzas were a bit premium by me).
from multiprocessing import Process, Queue
class Pizza(object):
def __init__(self, pizza_num):
self.pizza_num = pizza_num
self.num_slices = 8
sentinel = "NO PIZZA"
def producer(initial_num_pizzas, total_num_pizzas, q):
"""Cooks Pizzas to be consumed and waits for the consumer to finish eating."""
print("Producer: I am cooking %s Pizzas and putting them on the Queue!"%(total_num_pizzas-initial_num_pizzas))
for i in range(q.qsize(), total_num_pizzas):
print("Producer: Behold, for I have cooked Pizza no. %s"%i)
q.put(Pizza(i))
q.put(sentinel)
def consumer(q):
"""Consumes some Pizza. In this case, all it does is set the number of slices to 0."""
while True:
pizza = q.get()
pizza.num_slices = 0
if pizza == sentinel:
break
print("Comsumer: Pizza no. %s was found! It has %s slices, yum!"%(pizza.pizza_num, pizza.num_slices))
if __name__ == '__main__':
q = Queue()
total_num_pizzas = 10
initial_num_pizzas = 4
## Let add some Pizzas beforehand:
for i in range(0, initial_num_pizzas):
q.put(Pizza(i))
print("Main: I have precooked %s Pizzas."%q.qsize())
producer_proc = Process(target=producer, args=(initial_num_pizzas, total_num_pizzas, q))
consumer_proc = Process(target=consumer, args=(q,))
producer_proc.start()
consumer_proc.start()
q.close() ## Shop is closed, no more Pizzas will be added to Queue!
q.join_thread()
producer_proc.join()
consumer_proc.join()
Below is an example of the output. If you run it, Producer and Consumer print claims may alternate differently due to the non-deterministic execution of concurrent processes.
Main: I have precooked 4 Pizzas.
Producer: I am cooking 6 Pizzas and putting them on the Queue!
Producer: Behold, for I have cooked Pizza no. 4
Producer: Behold, for I have cooked Pizza no. 5
Producer: Behold, for I have cooked Pizza no. 6
Producer: Behold, for I have cooked Pizza no. 7
Comsumer: Pizza no. 0 was found! It has 8 slices, yum!
Comsumer: Pizza no. 1 was found! It has 8 slices, yum!
Producer: Behold, for I have cooked Pizza no. 8
Comsumer: Pizza no. 2 was found! It has 8 slices, yum!
Producer: Behold, for I have cooked Pizza no. 9
Comsumer: Pizza no. 3 was found! It has 8 slices, yum!
Comsumer: Pizza no. 4 was found! It has 8 slices, yum!
Comsumer: Pizza no. 5 was found! It has 8 slices, yum!
Comsumer: Pizza no. 6 was found! It has 8 slices, yum!
Comsumer: Pizza no. 7 was found! It has 8 slices, yum!
Comsumer: Pizza no. 8 was found! It has 8 slices, yum!
Comsumer: Pizza no. 9 was found! It has 8 slices, yum!
Note that you must use Sentinels to mark the end of your queue . I used "NO PIZZA" here, but they can be anything at all.
source to share