Python Multiprocessing Techniques

Python的多进程编程技巧 苏剑林 · 2017-02-19 · Machine Learning programming Python

Process

In Python, if you want to do multiprocess computation, this is generally implemented via multiprocessing, and the most commonly used tool is the process pool in multiprocessing, for example:

from multiprocessing import Pool
import time

def f(x):
    time.sleep(1)
    print x+1
    return x+1

a = range(10)
pool = Pool(4)
b = pool.map(f, a)
pool.close()
pool.join()

print b

This is concise and clear, and indeed convenient. What's interesting is that just by replacing multiprocessing with multiprocessing.dummy, you can turn the program from multiprocessing into multithreading. more

Object

Python is an object-oriented programming language, and quite often we wrap some of our code into a class. But inside a class, the method above stops working. For example:

from multiprocessing import Pool
import time

class test:
    def __init__(self):
        self.a = range(10)
    def run(self):
        def f(x):
            time.sleep(1)
            print x+1
            return x+1
        pool = Pool(4)
        self.b = pool.map(f, self.a)
        pool.close()
        pool.join()

t = test()
t.run()
print t.b

This code looks perfectly natural, yet running it throws an error:

cPickle.PicklingError: Can't pickle
: attribute lookup __builtin__.function failed

But if you replace multiprocessing with multiprocessing.dummy, no error occurs. Put simply, this is still because variables can't be shared across processes, whereas multiple threads live within the same process, so naturally there's no such problem there.

Imitation

In order to figure out multiprocess programming inside objects, I made quite a few attempts. Later it occurred to me that many modules in gensim support parallelism, so I could try imitating them. Sure enough, I found ldamulticore.py, and after repeatedly comparing it with various resources online, I worked out a fairly concise, convenient, and generally applicable way of writing this.

As with most multiprocess programming, in order to communicate between processes you need to set up Queue objects. The difference is that most tutorials online start multiple processes via multiprocessing's Process function combined with a loop, and using Pool fails to work (unless you use multiprocessing.Manager.Queue, see this article). But gensim uses a trick with Pool that still lets you launch multiple processes directly through Pool — the work of a real expert really is different. Reference code as follows:

from multiprocessing import Pool,Queue
import time

class test:
    def __init__(self):
        self.a = range(10)
    def run(self):
        in_queue, out_queue = Queue(), Queue()
        for i in self.a:
            in_queue.put(i)
        def f(in_queue, out_queue):
            while not in_queue.empty():
                time.sleep(1)
                out_queue.put(in_queue.get()+1)
        pool = Pool(4, f, (in_queue, out_queue))
        self.b = []
        while len(self.b) < len(self.a):
            if not out_queue.empty():
                t = out_queue.get()
                print t
                self.b.append(t)
        pool.terminate()

t = test()
t.run()
print t.b

In short, the approach is to set up two Queues, one responsible for queuing tasks and the other for retrieving results. What's rather surprising is that Pool actually has a second and third argument! For the details, see the official documentation — this is Pool's initialization function, and it too runs automatically in parallel.

Note that after running the line pool = Pool(4, f, (in_queue, out_queue)), the multiple processes start up, but the program doesn't wait for them to finish; it immediately moves on to run the following statements. At this point, as before, you could use pool.close() and pool.join() to make subsequent statements wait until the processes are done. But the approach used here is to directly execute the statement that retrieves results, and use this process itself to determine whether the processes have finished; once they're done, the process pool is shut down via pool.terminate(). This way of writing it is essentially general-purpose.