Using Eight Cores (Incorrectly) With Python

One of my web apps, The Wub Machine, is very computationally expensive. Audio decoding, processing, encoding, and streaming, all in Python. Naturally, my first instinct was to turn to the multiprocessing module to spread the CPU-bound work across multiple processes, thus avoiding Python’s global interpreter lock.

Remixing is hard work.

In theory, it’s simple enough, but I did run into a few very nasty problems when dealing with multiprocessing in Python:

The multiprocessing module, at least on *nixes, forks the current process and communicates with the child with a pipe. This works wonderfully if the data you’re transferring can be easily pickled, and if the child process doesn’t need to modify any global state in the parent. Unfortunately, certain useful constructs in Python can’t be pickled, including functions and lambdas (or pretty much anything callable).

In my app, I had a peculiar use case – I would callback pass lambdas into the constructor of a class, then spawn another process on one of that class’s methods like so:
```
    class MyClass(object):
        def __init__(self, my_callback):
            self.my_callback = my_callback

        def start_work():
            p = multiprocessing.Process(target=self.do_work)
            p.start()
            p.join()

        def do_work():
            # Calculate fibonacci or something, iunno
            self.my_callback("hey look, some data!")
```
This lead to a baffling bug – while the callbacks were being run, their side-effects weren’t persistent. I inserted logging in the callback function to verify, and noted that not only was it running, but at the end of its execution, the global state had been set properly. However, from the perspective of the parent process, nothing had changed.

The reason was simple: the callback had been run in the child process and had modified the global state of the child process, not the parent. A simple fix would be to have eliminated these callbacks, but I instead used some of Armin Ronacher’s bad ideas in Python to create an experimental module that allows pseudo-function-calls between processes. Use (or even just read) at your own risk – it’s a hack.

""" multiprocesscallback.py, by Peter Sobot (psobot.com), May 13, 2012 Handles callback functions in classes that have member functions that are executed in a different process. A crazy experiment in Python magic that breaks a lot of rules. Do not use in production, for any reason. (Although I do.) If your class takes in a callback, like so: class MyClass(object): def __init__(self, callback): you can call MultiprocessCallback.register_all(queue) to auto-create member functions with the names of the callback variables, which, when called, will be safely executed in the parent process. E.g.: class MyClass(object): def __init__(self, my_callback): self._pq = multiprocessing.Queue() MultiprocessCallback.register_all(self._pq) def start_other_process(): target = MultiprocessCallback.target(self.runs_in_another_process) p = multiprocessing.Process(target=target) p.start() MultiprocessCallback.listen() p.join() def runs_in_another_process(): self.my_callback("hey look, some data!") The data in the callback must be picklable, as it will be sent across the multiprocess boundary. The parent process can read from the queue itself and run .execute() on the MultiprocessCallback objects, or it can use the *blocking* MultiprocessCallback.listen(), which provides a basic listener. Known bugs or omissions: - Will straight-up just not work in Windows. (The scenario doesn't exist - you can't use multiprocessing on member functions on Windows.) """ import multiprocessing import traceback import inspect import sys import time __author__ = "psobot" class EndListener(Exception): pass def register_all(queue=None): _locals = inspect.getargvalues(sys._getframe(1))[3] if not queue: queue = multiprocessing.Queue() setattr(_locals['self'], "_mpcq", queue) for n, v in dict([(_n, _v) for (_n, _v) in _locals.iteritems() if _n != "self"]).iteritems(): if hasattr(v, "__call__"): proc = multiprocessing.current_process().name setattr(_locals['self'], n, lambda *args, **kwargs: _safecall(proc, queue, n, v, *args, **kwargs)) def _safecall(proc, queue, n, _c, *args, **kwargs): if multiprocessing.current_process().name != proc: queue.put(MultiprocessCallback(n, *args, **kwargs)) else: return _c(*args, **kwargs) def listen(queue=None): if not queue: _locals = inspect.getargvalues(sys._getframe(1))[3] queue = _locals['self'].__dict__["_mpcq"] data = queue.get() while not isinstance(data, EndListener): if isinstance(data, MultiprocessCallback): data.execute() data = queue.get() def target(_callable, queue=None): def _target(*args, **kwargs): _callable(*args, **kwargs) end(queue) return _target def end(queue=None): if not queue: i = 1 _locals = inspect.getargvalues(sys._getframe(i))[3] while not 'self' in _locals or not '_mpcq' in _locals['self'].__dict__: i += 1 _locals = inspect.getargvalues(sys._getframe(i))[3] queue = _locals['self'].__dict__["_mpcq"] queue.put(EndListener()) class MultiprocessCallback(object): def __init__(self, name, *args, **kwargs): self.name = name self.stackf = traceback.format_stack(sys._getframe(3), 2) self.originator = multiprocessing.current_process().name self.args = args self.kwargs = kwargs def execute(self, search=None): if not search: i = 1 search = inspect.getargvalues(sys._getframe(i))[3] while not self.name in search: i += 1 search = inspect.getargvalues(sys._getframe(i))[3] if not self.name in search and 'self' in search: search = search['self'].__dict__ if self.name in search: if hasattr(search[self.name], '__call__'): try: r = search[self.name](*self.args, **self.kwargs) if r is not None: print "Warning: return value from callback ignored." except Exception, e: e.args = (" ".join(list(e.args) + ["\nOriginally called from %s (most recent call last):\n" % self.originator] + self.stackf), ) raise else: raise ValueError("Function %s not callable." % self.name) else: raise KeyError("Function %s not provided." % self.name) if __name__ == "__main__": class Test(object): def __init__(self, callback = None): register_all() def run_me(self): """ Run self.separate_process in its own process. When the callback is called, it will execute in the parent process. """ p = multiprocessing.Process(target=target(self.separate_process)) p.start() listen() p.join() def separate_process(self): for i in xrange(0, 10): # some intense computation self.callback(i) time.sleep(0.1) count = 0 def my_callback(i): """ Increments a global variable by i in the main process. """ if multiprocessing.current_process().name != "MainProcess": raise multiprocessing.ProcessError("The global is being incremented in the wrong process!") global count count += i print "Counter in main process is now: %s" % count Test(callback=my_callback).run_me()

Logging, the wonderful built-in Python module for meticulously logging everything, is thread-safe. Sadly, it doesn’t seem to be multiprocessing-safe. Logging makes use of its own internal I/O thread, to ensure that all log messages are properly queued and written without clobbering eachother. This thread is locked for every write.

After forking another process, the first call to the logger often hangs while waiting for the logging thread to become unlocked. If the logging thread was in use (i.e.: locked) at the exact instant the process was forked, then the locked thread will be copied to the new process. However, whatever log operation was in progress will then unlock the original thread, not the copied thread, leaving the new process to wait forever on a lock that will never be unlocked.

The solution, in my case, was to replace the logger in use with the one provided by multiprocessing if logging from a new process:
```
def initlog():
    if multiprocessing.current_process().name == "MainProcess":
        _log = logging.getLogger(config.log_name)
    else:
        _log = multiprocessing.get_logger()
    ...
```

To find and fix these bugs took a lot of time, and a good debugging strategy. The most valuable tool turned out, surprisingly, to be GDB. GDB 7 has support for debugging Python runtimes, complete with pseudo-stack traces. Take a look at the following backtrace of a Python process provided by GDB and formatted for clarity:

    [Thread debugging using libthread_db enabled]
    [New Thread 0xb0c2fb70 (LWP 12895)]
    0x006da405 in __kernel_vsyscall ()

    Thread 1 (Thread 0xaf23ab70 (LWP 12894)):
    #0  0x006da405 in __kernel_vsyscall ()
    #1  0x003a27d5 in sem_wait@@GLIBC_2.1 ()
                  from /lib/i386-linux-gnu/libpthread.so.0
    #2  0x080f2139 in PyThread_acquire_lock (...)
                  at ../Python/thread_pthread.h:309
    #3  0x080f2fd8 in lock_PyThread_acquire_lock (...)
                  at ../Modules/threadmodule.c:52
    #4  0x080da7d5 in call_function
                    (f=Frame 0x937b47c,
                      for file /usr/lib/python2.7/threading.py,
                      line 128,
                      in acquire
                      (self=<_RLock(...) at remote 0x9caabec>,
                        blocking=1,
                        me=-1356616848),
                      throwflag=0) at ../Python/ceval.c:4013
    ...
    (goes down 79 frames)

Obviously, this looks much more complicated than a normal Python stack trace, but it’s a huge step up from zero debugability. If I proceed down a couple more frames, I find:

    #7  0x080dac2a in fast_function
                      (f=Frame 0x9ca278c,
                      for file /usr/lib/python2.7/logging/__init__.py,
                      line 693,
                      in acquire (self=<FileHandler(stream=...

…which is the first piece of familiar code. Line 693 of logging/__init__.py is surrounded by a short function, and has a comment that brings the first bit of understanding:

    def acquire(self):
        """
        Acquire the I/O thread lock.
        """
        if self.lock:
            self.lock.acquire()

Well, there you go. After fixing these race conditions and deadlocks, the Wub Machine’s success rate immediately jumped from horrible to 95% under load.

$Look'it dat 95\% success rate.$

All it took was GDB and an understanding of fork() to solve these bugs. My only advice: be very, very, very careful when using multiprocessing.