Using python multiprocessing module is quite simple, even though it has some pitfalls. But remembering how to avoid them it can help to really speed up some tasks.
In the examples I will use a simple program calculating Fibonacci numbers
|
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 |
#!/usr/bin/python import time def fib(n): if n < 0: raise ValueError if n < 2: return n return fib(n-1)+fib(n-2) def fibonacci(n): start = time.time() res = fib(n) end = time.time() return res def main(): start = time.time() for i in range(30, 37): res = fibonacci(i) print "fib(%3d) = %7d (took %0.3fs)" % ( n, res, (end-start)) end = time.time() print "Total time: %0.3fs" % (end - start) if __name__ == "__main__": main() |
main is the main function controlling the program. fib is doing the actual work. fibonacci is a wrapper we will change later to use multiprocessing withouth changing fib. Adjust the numbers in range(30, 37) to match speed of your computer, so the program takes about 10-20 seconds to run. Output should be like this:
|
1 2 3 4 5 6 7 8 |
fib( 30) = 832040 (took 0.406s) fib( 31) = 1346269 (took 0.663s) fib( 32) = 2178309 (took 1.069s) fib( 33) = 3524578 (took 1.728s) fib( 34) = 5702887 (took 2.798s) fib( 35) = 9227465 (took 4.565s) fib( 36) = 14930352 (took 7.373s) Total time: 18.603s |
We need a way to make it possible to run the program starting fibonacci only once. I will use Queue to prepare the data for it. Add:
|
1 |
import multiprocessing |
and change the main and fibonacci functions to look like this:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
def fibonacci(in_queue): while True: n = in_queue.get() if n == None: break start = time.time() res = fib(n) end = time.time() return res def main(): start = time.time() in_queue = multiprocessing.Queue() for i in range(30, 37): in_queue.put(i) in_queue.put(None) res = fibonacci(in_queue) print "fib(%3d) = %7d (took %0.3fs)" % ( n, res, (end-start)) end = time.time() print "Total time: %0.3fs" % (end - start) |
What changed? The fibonacci is now a loop, taking values from queue and processing them one after another. This ends when it receives Null as value (“poison pill” – line 4-5). The data is inserted into the queue in the main function together with the poison pill (lines 16-18).
But wait! Now there is no way to send a reply from fibonacci! So let’s create another queue to receive the results. main and fibonacci change again:
|
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 |
def fibonacci(in_queue, out_queue): while True: n = in_queue.get() if n == None: break start = time.time() res = fib(n) end = time.time() out_queue.put((n, res, end-start)) def main(): start = time.time() in_queue = multiprocessing.Queue() out_queue = multiprocessing.Queue() for i in range(30, 37): in_queue.put(i) in_queue.put(None) fibonacci(in_queue, out_queue) while not out_queue.empty(): n, res, spend_time = out_queue.get() print "fib(%3d) = %7d (took %0.3fs)" % ( n, res, spend_time) end = time.time() print "Total time: %0.3fs" % (end - start) |
What’s going on: main puts requests into the first queue (17-19), calls fibonacci (21), and prints results taken from second queue (23-25). fibonacci gets data from first queue (3) until it encounters Null (4-5), and puts results into second queue (10). Now, as the fibonacci is separated from main using the queues we can put it into separate process.
Change the line that calls fibonacci to:
|
1 2 |
multiprocessing.Process(target=fibonacci, args=(in_queue, out_queue)).start() |
Now the fibonacci starts after the first queue is full, but the program won’t wait for it to finish, it immediately goes to displaying the results. But it doesn’t work. Why?
Well, fibonacci wasn’t fast enough to compute any result, so the while loop (in line 23) started when results queue was still empty and quit too early. We can use second poison pill sent from fibonacci to solve this.
Another thing to keep in mind is that the main program waits for the worker to end. Luckily fibonacci ends (thank to the poison pill), but nobody is waiting any more for the results, so they don’t show up. This will be also solved with the poison pill. At the end of fibonacci add:
|
1 |
out_queue.put((None, None, None)) |
and change the loop in main to:
|
1 2 3 4 5 |
while True: n, res, spend_time = out_queue.get() if n == None: break print "fib(%3d) = %7d (took %0.3fs)" % ( n, res, spend_time) |
It works again! Now that is needed to have several processes work in parallel is to start them. Remember to add the necessary poison pills! Complete final version looks like this:
|
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 40 41 42 43 44 45 46 47 48 49 50 51 |
#!/usr/bin/python import time import multiprocessing def fib(n): if n < 0: raise ValueError if n < 2: return n return fib(n-1)+fib(n-2) def fibonacci(in_queue, out_queue): while True: n = in_queue.get() if n == None: break start = time.time() res = fib(n) end = time.time() out_queue.put((n, res, end-start)) out_queue.put((None, None, None)) def main(): start = time.time() in_queue = multiprocessing.Queue() out_queue = multiprocessing.Queue() process_num = multiprocessing.cpu_count() for _ in range(process_num): multiprocessing.Process(target=fibonacci, args=(in_queue, out_queue)).start() for i in range(30, 37): in_queue.put(i) for _ in range(process_num): in_queue.put(None) while True: n, res, spend_time = out_queue.get() if n == None: process_num -= 1 if process_num == 0: break else: print "fib(%3d) = %7d (took %0.3fs)" % ( n, res, spend_time) end = time.time() print "Total time: %0.3fs" % (end - start) if __name__ == "__main__": main() |
Line 22 adds the poison pill to notify main that a fibonacci process has ended. Lines 30-32 start as many processes as we have cpus (determined in line 29). Lines 36-37 add poison pills for all fibonacci processes. Lines 41-43 keep track of how many processes are still running.
Remember, this is not the only solution. For example the results loop could simply count how many results it received. The optimal processes number depends on problem. If it’s just calculating data (like here) then CPU count is good bet. If there is substantial time spend on waiting for network operations, then more may be better. But if the slowdown is caused by some slow resource (like slow disk), then more processes can actually slow down performance. Usually it’s easiest to just try it out.
One more tip: when using multiple processes it is important to remember that one of them could raise exception when it encounters a problem. Always wrap all risky parts in except, and handle the exceptions. Also think of ways of terminating the program early when necessary. Remember: the program ends only, when all processes end.