I have some (readonly) data (around 1 or 2 GB in the actual use case) and I want to send it to a pool of processes (one for each available processor minus one, for a total of 7 on my late 2011 macbook pro) for some computations using Python 3.9.
The simplest approach is to use the tools provided by the multiprocessing module of the standard library.
I implemented this solution in the test_standard_ipc function below.
As far as I know this is what happens when that function is called: the data dictionary is serialised using pickle, then a single os pipe is in charge to stream the serialised data to the first process of the pool; only when this stream is completed the same os pipe is used to stream the serialised data to the second process in the pool and so on.
This means that each process in the pool needs to wait for its turn to receive the serialised data (then deserialise it and start working). When the dimension of the data is around 1 or 2 GB this means that many processes in the pool have to wait a large amount of time before starting doing things.
To overcome this problem I came up with the following idea: since my data is only made of built-in data types I will use the marshal module (which is way faster than pickle) to serialise data, I will put the resulting bytes in a shared array and I will pass to each process in the pool the address of such array. In this way I should be able to immediately start all processes in the pool that will concurrently deserialise the data and start working.
HOWEVER, it seems that reading the shared array of bytes from each process in the pool is extremely slow (even when I use a pool with just one process).
My guess is that the byte array with the serialised data is still accessed from the subprocesses through a single os pipe instead of directly accessing it.
Is there a way to speed up this approach?
Here is the test code:
# test_ipc.py
import marshal
import os
from multiprocessing import Manager, Pool
from time import time
from timeit import timeit
from pympler.asizeof import asizeof
def marshal_worker(data_array):
pid = os.getpid()
print(f" -> [pid {pid}] Marshal worker ready at {time()}")
# Building a bytearray is a waste of time but I did not found
# found a way to feed `data_array` directly to marshal.loads()
t = time()
ba = bytearray(data_array)
print(f" -> [pid {pid}] Building bytearray took {time() - t} s")
t = time()
data = marshal.loads(ba)
print(f" -> [pid {pid}] Marshal loads() took {time() - t} s")
return len(data)
def test_marshal_ipc(data):
print("Running test_marshal_ipc():")
n_processes = os.cpu_count() - 1 or 1
with Manager() as manager:
with Pool(processes=n_processes) as pool:
data_bytes = marshal.dumps(data)
data_array = manager.Array('B', data_bytes, lock=False)
async_results = [pool.apply_async(marshal_worker, (data_array,)) for _ in range(n_processes)]
subprocess_results = [res.get() for res in async_results]
return subprocess_results
def standard_worker(data):
print(f" -> [pid {os.getpid()}] Standard worker ready at {time()}")
return len(data)
def test_standard_ipc(data):
print("Running test_standard_ipc():")
n_processes = os.cpu_count() - 1 or 1
with Pool(processes=n_processes) as pool:
async_results = [pool.apply_async(standard_worker, (data,)) for _ in range(n_processes)]
subprocess_results = [res.get() for res in async_results]
return subprocess_results
if __name__ == '__main__':
REPETITIONS = 1
DATA_SIZE = 10_000
data = {
'foo': list(range(DATA_SIZE)),
'bar': dict(zip(range(DATA_SIZE), range(DATA_SIZE)))
}
print(f"Data size: {asizeof(data)} bytes")
marsall_time = timeit(
stmt="test_marshal_ipc(data)",
setup="from __main__ import test_marshal_ipc, data",
number=REPETITIONS
)
print(f"marshal ipc took: {marsall_time} s")
standard_time = timeit(
stmt="test_standard_ipc(data)",
setup="from __main__ import test_standard_ipc, data",
number=REPETITIONS
)
print(f"standard ipc took: {standard_time} s")
and the output:
$ python test_ipc.py
Data size: 1318944 bytes
Running test_marshal_ipc():
-> [pid 17950] Marshal worker ready at 1633625344.844704
-> [pid 17953] Marshal worker ready at 1633625344.8449469
-> [pid 17951] Marshal worker ready at 1633625344.8453
-> [pid 17955] Marshal worker ready at 1633625344.860242
-> [pid 17954] Marshal worker ready at 1633625344.864512
-> [pid 17952] Marshal worker ready at 1633625344.871718
-> [pid 17956] Marshal worker ready at 1633625344.876148
-> [pid 17950] Building bytearray took 58.384530782699585 s
-> [pid 17950] Marshal loads() took 0.0020139217376708984 s
-> [pid 17952] Building bytearray took 58.448140144348145 s
-> [pid 17952] Marshal loads() took 0.0024509429931640625 s
-> [pid 17956] Building bytearray took 58.71299409866333 s
-> [pid 17956] Marshal loads() took 0.002827167510986328 s
-> [pid 17954] Building bytearray took 58.93824005126953 s
-> [pid 17954] Marshal loads() took 0.0023200511932373047 s
-> [pid 17955] Building bytearray took 59.62452507019043 s
-> [pid 17955] Marshal loads() took 0.001924276351928711 s
-> [pid 17951] Building bytearray took 59.66379499435425 s
-> [pid 17951] Marshal loads() took 0.002319812774658203 s
-> [pid 17953] Building bytearray took 59.7155179977417 s
-> [pid 17953] Marshal loads() took 0.0018548965454101562 s
marshal ipc took: 60.396030886999995 s
Running test_standard_ipc():
-> [pid 17974] Standard worker ready at 1633625405.037303
-> [pid 17975] Standard worker ready at 1633625405.0419872
-> [pid 17974] Standard worker ready at 1633625405.043684
-> [pid 17975] Standard worker ready at 1633625405.045311
-> [pid 17974] Standard worker ready at 1633625405.047421
-> [pid 17974] Standard worker ready at 1633625405.05076
-> [pid 17975] Standard worker ready at 1633625405.05163
standard ipc took: 0.4552726120000017 s
UPDATE:
Switching from manager.Array to multiprocessing.Array throws the following error:
$ python test_ipc.py
Data size: 1318944 bytes
Running test_marshal_ipc():
Traceback (most recent call last):
File "test_ipc.py", line 67, in <module>
marsall_time = timeit(
File "***OMISSIS***/python3.9/timeit.py", line 233, in timeit
return Timer(stmt, setup, timer, globals).timeit(number)
File "***OMISSIS***/python3.9/timeit.py", line 177, in timeit
timing = self.inner(it, self.timer)
File "<timeit-src>", line 6, in inner
File "test_ipc.py", line 36, in test_marshal_ipc
subprocess_results = [res.get() for res in async_results]
File "test_ipc.py", line 36, in <listcomp>
subprocess_results = [res.get() for res in async_results]
File "***OMISSIS***/python3.9/multiprocessing/pool.py", line 771, in get
raise self._value
File "***OMISSIS***/python3.9/multiprocessing/pool.py", line 537, in _handle_tasks
put(task)
File "***OMISSIS***/python3.9/multiprocessing/connection.py", line 211, in send
self._send_bytes(_ForkingPickler.dumps(obj))
File "***OMISSIS***/python3.9/multiprocessing/reduction.py", line 51, in dumps
cls(buf, protocol).dump(obj)
File "***OMISSIS***/python3.9/multiprocessing/sharedctypes.py", line 129, in reduce_ctype
assert_spawning(obj)
File "***OMISSIS***/python3.9/multiprocessing/context.py", line 359, in assert_spawning
raise RuntimeError(
RuntimeError: c_ubyte_Array_150019 objects should only be shared between processes through inheritance