View source for ProcessPoolExecutor: A fast way to implement multiprocessing


A fast way (measured in source code length) for multi-processing a function. Is it the best way? No, but it is easily accessable.

Questions to [mailto:davrot@uni-bremen.de David Rotermund]

== An example ==
We want to run 4 processes (number_of_cpu_processes = 4) at the same time.

The first step is to get rid of most of the arguments of the function function_a. We use [https://docs.python.org/3/library/functools.html#functools.partial functools.partial] to fix all the non relevant (for the multiprocessing) arguments. We will leave only one argument which is akin to a job id.

Then we create a [https://docs.python.org/3/library/concurrent.futures.html#processpoolexecutor concurrent.futures.ProcessPoolExecutor] session with 4 processes: 

The [https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ProcessPoolExecutor ProcessPoolExecutor] class is an [https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.Executor Executor] subclass that uses a pool of processes to execute calls asynchronously. [https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ProcessPoolExecutor ProcessPoolExecutor] uses the multiprocessing module, which allows it to side-step the [https://docs.python.org/3/glossary.html#term-global-interpreter-lock Global Interpreter Lock] but also means that only picklable objects can be executed and returned.

Using the ProcessPoolExecutor instance executor, we now can call the [https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.Executor.map executor.map] method.

[https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.Executor.map map(func, *iterables, timeout=None, chunksize=1)]: &#x3E; When using [https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ProcessPoolExecutor ProcessPoolExecutor], this method chops iterables into a number of chunks which it &#x3E; submits to the pool as separate tasks. The (approximate) size of these chunks can be specified by setting chunksize to a positive integer. &#x3E; For very long iterables, using a large value for chunksize can significantly improve performance compared to the default size of 1.

[https://docs.python.org/3/library/functions.html#map map(function, iterable, …)]: &#x3E; Return an iterator that applies function to every item of iterable, yielding the results. If additional iterable arguments are passed, function must &#x3E; take that many arguments and is applied to the items from all iterables in parallel. With multiple iterables, the iterator stops when the shortest iterable is exhausted.

As you can see in the output, all random numbers are the same. This is the results of the “but also means that only picklable objects can be executed and returned.”. The rng object is pickled and thus frozen in its state. Every time map calls the function, rng is un-pickled – with the same state –.<syntaxhighlight lang="python">import numpy as np
import concurrent.futures
import functools


def function_a(parameter_a: float, rng) -> tuple[float, float]:
    return rng.random() + parameter_a, parameter_a


number_of_cpu_processes: int = 4
rng = np.random.default_rng()
number_of_calls: int = 10

function_a_partial = functools.partial(
    function_a,
    rng=rng,
)

with concurrent.futures.ProcessPoolExecutor(number_of_cpu_processes) as executor:
    for id, (return_value_a, return_value_b) in enumerate(
        executor.map(function_a_partial, range(0, number_of_calls))
    ):
        print(f"{id} {return_value_a:.3f} {return_value_b}")</syntaxhighlight>Output:<syntaxhighlight lang="python">0 0.951 0
1 1.951 1
2 2.951 2
3 3.951 3
4 4.951 4
5 5.951 5
6 6.951 6
7 7.951 7
8 8.951 8
9 9.951 9</syntaxhighlight>Another example but with a Numpy array as result array:<syntaxhighlight lang="python">import numpy as np
import concurrent.futures
import functools


def function_a(parameter_a: float, size: int, rng) -> np.ndarray:
    return rng.random((size,)) + parameter_a


number_of_cpu_processes: int = 4
rng = np.random.default_rng()
number_of_calls: int = 10
size = 12
function_a_partial = functools.partial(
    function_a,
    size=size,
    rng=rng,
)

results: np.ndarray = np.zeros((number_of_calls, size))

with concurrent.futures.ProcessPoolExecutor(number_of_cpu_processes) as executor:
    for id, return_value_a in enumerate(
        executor.map(function_a_partial, range(0, number_of_calls))
    ):
        results[id, :] = return_value_a

print(results)</syntaxhighlight>