The “multiprocessing” module has a class Pool that is quite convenient if we want to do parallel processing. The API is simple and rather straightforward. I wanted to see if I can craft an example out of the official docs and here’s the code:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 |
from multiprocessing import Pool from time import sleep from random import randint import os class AsyncFactory: def __init__(self, func, cb_func): self.func = func self.cb_func = cb_func self.pool = Pool() def call(self,*args, **kwargs): self.pool.apply_async(self.func, args, kwargs, self.cb_func) def wait(self): self.pool.close() self.pool.join() def square(x): sleep_duration = randint(1,5) print "PID: %d \t Value: %d \t Sleep: %d" % (os.getpid(), x ,sleep_duration) sleep(sleep_duration) return x*x def cb_func(x): print x async_square = AsyncFactory(square, cb_func) async_square.call(1) async_square.call(2) async_square.call(3) async_square.call(4) async_square.call(5) async_square.wait() |
Let’s see whta we’re doing here – we have two functions, the main function “square” and the callback function “cb_func” which just prints out the result. We want to run the main function asynchronously and as soon as the result is available, we want the callback to execute, printing out the result.
What we’re doing inside the AsyncFactory is storing the function and callback function inside an instance. Then we create a worker pool. The worker pool by default uses the available CPUs. We can also pass values to the “processes” argument to determine the number of worker processes in the pool.
Then we repeatedly call the apply_async on the Pool object to pass the function with the arguments. Finally, we wait for the pool to close it’s workers and rest in peace. As soon as each worker returns a value, the callback would print it out. We have also added a print statement to check the process ID, passed value and sleep duration.
Isn’t it fun?
6 replies on “Async execution in Python using multiprocessing Pool”
Very neat. Now I only have to make my c++ programs into Python accesible modules and profit!
[…] ASYNC EXECUTION IN PYTHON USING MULTIPROCESSING POOL […]
Beautiful and very useful module. Thanks.
It would be an improvement to make AsyncObject inherit from object:
class AsyncFactory(object):
and change the call method to __call__(self,…)
then you don’t need to explicitly use call method:
async_square(1)
Hi Paul,
Sorry for ignorance for the __call__ semantics. I am new to Python. Can you elaborate the following:
and change the call method to __call__(self,…)
Did you mean replace the 5 async_square.call(x) with a single __call__(self,…) method? And what should be the arguments to __call__?
Thanks
Zaffy
Nevermind Paul. I figured out what you meant. Thanks to Masnun and you for this neat snippet.