Table of contents:
This documentation applies to gipc 0.4.0. It was built on April 09, 2014.
gipc provides convenient child process management in the context of gevent. gipc (pronunciation “gipsy”) is a Python package tested on CPython 2.6 and 2.7 on Linux as well as on Windows.
There is plenty of motivation for using multiple processes in event-driven architectures. The assumption behind gipc is that applying multiple processes that communicate among each other (whereas each process has its own event loop) can be a decent solution for many types of problems. First of all, it helps decoupling system components by making each process responsible for one part of the architecture only. Furthermore, even a generally I/O-intense application might at some point become CPU bound. In these cases, the distribution of tasks among multiple processes is an efficient way to make use of multi-core machines and easily increases the application’s performance.
However, canonical usage of Python’s multiprocessing module within a gevent-powered application may raise various problems and most likely breaks the application in many ways. gipc is developed with the motivation to solve these issues transparently and make using gevent in combination with multiprocessing-based child processes and inter-process communication (IPC) a no-brainer again:
In the following code snippet, a Python object is sent from a greenlet in the main process through a pipe to a child process:
import gevent import gipc def child(reader): assert reader.get() == 0 if __name__ == "__main__": with gipc.pipe() as (reader, writer): writelet = gevent.spawn(lambda w: w.put(0), writer) readchild = gipc.start_process(target=child, args=(reader,)) writelet.join() readchild.join()
Although quite simple, this code would have various negative side-effects if used with the canonical multiprocessing API instead of gipc.start_process() and gipc.pipe(), as outlined in the next paragraph.
Depending on the operating system, child process creation via Python’s multiprocessing in the context of gevent requires special treatment. Most care is needed on POSIX-compliant systems. There, a fork might yield a faulty libev event loop state in the child. Most noticeable, greenlets spawned before forking are cloned and haunt in the child upon context switch. Consider this code running on Unix (tested with Python 2.7 & gevent 1.0):
import gevent import multiprocessing def child(c): gevent.sleep(0) assert c.recv() == 0 assert c.recv() == 0 if __name__ == "__main__": def writelet(c): c.send(0) c1, c2 = multiprocessing.Pipe() writelet = gevent.spawn(writelet, c1) readchild = multiprocessing.Process(target=child, args=(c2,)) readchild.start() writelet.join() readchild.join()
It runs without error. Although the code intends to send only one message to the child through a multiprocessing Pipe, the two assert statements verify that the child actually receives two times the same message. One message is sent – as intended – from the writelet in the parent through the c1 end of the pipe. It is retrieved at the c2 end of the pipe in the child. The other message is sent from the spooky writelet clone in the child. It is also written to the c1 end of the pipe which has implicitly been duplicated during forking. Greenlet clones in the child of course only run when a context switch is triggered; in this case via gevent.sleep(0). As you can imagine, this behavior in general might lead to a wide range of side-effects and tedious debugging sessions.
In addition, the code above contains several non-cooperatively blocking method calls: readchild.join() as well as the send()/recv() calls (of multiprocessing.Connection objects in general) block the calling greenlet non-cooperatively and do not allow for context switches into other greenlets.
gipc overcomes these and other issues for you transparently and in a straight- forward fashion: basically, children start off with a fresh gevent state before entering the user-given target function. More specifically, as one of the first actions, children destroy the inherited gevent hub as well as the inherited libev event loop and create their own fresh versions of these objects. This way, inherited greenlets as well as libev watchers become orphaned – the fresh hub and event loop are not connected to them anymore. Consequently, execution of code related to these inherited greenlets and watchers is efficiently prevented without the need to deactivate or kill them one by one.
Furthermore, on POSIX-compliant systems, gipc entirely avoids multiprocessing’s child monitoring implementation (which is based on wait) and instead uses libev’s wonderful child watcher system (based on SIGCHLD signal transmission), enabling gevent-cooperative waiting for child termination.
For the sake of gevent-cooperative inter-process communication, gipc uses efficient pipe-based data transport channels with non-blocking I/O. gipc takes care of closing dispensable pipe handles (file descriptors) in the parent as well as in the child after forking.
Overall, gipc’s main goal is to allow for the integration of child processes in your gevent-powered application via a simple API – on POSIX-compliant systems as well as on Windows.
As of version 0.3, I am not aware of any severe issues. To my knowledge, gipc has already been deployed in serious projects. Generally, you should be aware of the fact that mixing any of fork, threads, greenlets and an event loop library such as libev bears the potential for various kinds of corner-case disasters. One could argue that fork() in the context of libev without doing a clean exec in the child already is broken design. However, many people would like to do exactly this and gipc’s basic approach has proven to work in such cases. gipc is developed with a strong focus on reliability and with best intentions in mind. Via unit testing it has been validated to work reliably in scenarios of low and medium complexity. Of course, gipc cannot rescue an a priori ill-posed approach. Now it is up to you to evaluate gipc in the context of your project – please let me know how it performs for you.
gipc’s Mercurial repository is hosted at Bitbucket.
The latest gipc release from PyPI can be pulled and installed via pip:
$ pip install gipc
pip can also install the current development version of gipc:
$ pip install hg+https://bitbucket.org/jgehrcke/gipc
Windows I/O Completion Ports (IOCP) could solve both issues in an elegant way. Currently, gevent is built on top of libev which does not support IOCP. In the future, however, gevent might become libuv-backed. libuv supports IOCP and would allow for running the same gevent code on Windows as on POSIX-compliant systems. Furthermore, if gevent went with libuv, the strengths of both, the node.js and the gevent worlds would be merged. Denis Bilenko, the maintainer of gevent, seems to be open to such a transition and the first steps are already done.
Note that these examples are invented with the motivation to demonstrate the API and capabilities of gipc rather than showing interesting use cases.
Very basic gevent and gipc concepts are explained by means of this simple messaging example:
import gevent import gipc def main(): with gipc.pipe() as (r, w): p = gipc.start_process(target=child_process, args=(r, )) wg = gevent.spawn(writegreenlet, w) try: p.join() except KeyboardInterrupt: wg.kill(block=True) p.terminate() p.join() def writegreenlet(writer): while True: writer.put("written to pipe from a greenlet running in the main process") gevent.sleep(1) def child_process(reader): while True: print "Child process got message from pipe:\n\t'%s'" % reader.get() if __name__ == "__main__": main()
The context manager with gipc.pipe() as (r, w) creates a pipe with read handle r and write handle w. On context exit (latest) the pipe ends will be closed properly.
After creating the pipe context, the above code spawns a child process via gipc.start_process(). The child process is instructed to execute the target function named child_process whereas the pipe read handle r is provided as an argument to this target function. Within child_process() an endless loop waits for objects on the read end of the pipe via the cooperatively blocking call to reader.get(). Upon retrieval, it immediately writes their string representation to stdout.
After invocation of the the child process (represented by an object bound to name p), a greenlet wg is spawned within the main process. This greenlet executes the function writegreenlet, whereas the pipe write handle w is provided as an argument. Within this greenlet, one string per second is written to the write end of the pipe.
After spawning wg, p.join() is called immediately in the parent process. p.join() is blocking cooperatively, i.e. it allows for a context switch into the write greenlet (this actually is the magic behind gevent/greenlets). Hence, the write greenlet is ‘running’ while p.join() cooperatively waits for the child process to terminate. The write greenlet spends most of its time in gevent.sleep(), which is also blocking cooperatively, allowing for context switches back to the main greenlet in the parent process, which is executing p.join(). In this state, one message per second is passed between parent and child until a KeyboardInterrupt exception is raised in the parent.
Upon KeyboardInterrupt, the parent first kills the write greenlet and blocks cooperatively until it has stopped. Then it terminates the child process (via SIGTER on Unix) and waits for it to exit via p.join().
For pure API and reliability demonstration purposes, this example implements TCP communication between a server in the parent process and multiple clients in one child process:
import gevent from gevent.server import StreamServer from gevent import socket import gipc import time PORT = 1337 N_CLIENTS = 1000 MSG = "HELLO\n" def serve(sock, addr): f = sock.makefile() f.write(f.readline()) f.flush() f.close() def server(): ss = StreamServer(('localhost', PORT), serve).serve_forever() def clientprocess(): t1 = time.time() clients = [gevent.spawn(client) for _ in xrange(N_CLIENTS)] gevent.joinall(clients) duration = time.time()-t1 print "%s clients served within %.2f s." % (N_CLIENTS, duration) def client(): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect(('localhost', PORT)) f = sock.makefile() f.write(MSG) f.flush() assert f.readline() == MSG f.close() if __name__ == "__main__": s = gevent.spawn(server) c = gipc.start_process(clientprocess) c.join() s.kill() s.join()
Output on my test machine: 1000 clients served within 0.54 s.
Child process creation may take a significant amount of time, especially on Windows. This time is not predictable.
When code in the parent should only proceed in the moment the code in the child has reached a certain state, the proper way to tackle this is a bidirectional synchronization handshake:
This concept can easily be implemented using a bidirectional gipc.pipe():
import gevent import gipc import time def main(): with gipc.pipe(duplex=True) as (cend, pend): # `cend` is the channel end for the child, `pend` for the parent. p = gipc.start_process(writer_process, args=(cend,)) # Synchronize with child process. pend.put("SYN") assert pend.get() == "ACK" # Now in sync with child. ptime = time.time() ctime = pend.get() p.join() print "Time delta: %.8f s." % abs(ptime - ctime) def writer_process(cend): with cend: assert cend.get() == "SYN" cend.put("ACK") # Now in sync with parent. cend.put(time.time()) if __name__ == "__main__": main()
The marked code blocks in parent and child are entered quasi-simultaneously. Example output on my test machine (Linux): Time delta: 0.00005388 s. On Windows, time.time()‘s precision is not sufficient to resolve the time delta (and time.clock() is not applicable for comparing times across processes).
Start child process and execute function target(*args, **kwargs). Any existing instance of gipc._GIPCHandle or gipc._GIPCDuplexHandle can be passed to the child process via args and/or kwargs. If any such instance is passed to the child, the corresponding file descriptor is automatically closed in the parent.
Compared to the canonical multiprocessing.Process() constructor, this function
gipc._GProcess instance (inherits from multiprocessing.Process and re-implements some of its methods in a gevent-cooperative fashion).
start_process() triggers most of the magic in gipc. Process creation is based on multiprocessing.Process(), i.e. fork() on POSIX-compliant systems and CreateProcess() on Windows.
Please note that in order to provide reliable signal handling in the context of libev, the default disposition (action) is restored for all signals in the child before executing the user-given target function. You can (re)install any signal handler within target.
Create a pipe-based message transport channel and return two corresponding handles for reading and writing data.
Allows for gevent-cooperative transmission of data (any picklable Python object by default). Data can be transmitted between greenlets within one process or across processes (created via start_process()).
with pipe() as (r, w): do_something(r, w)
reader, writer = pipe() with reader: do_something(reader) with writer as w: do_something(w)
with pipe(duplex=True) as (h1, h2): h1.put(1) assert h2.get() == 1 h2.put(2) assert h1.get() == 2
The _GIPCHandle class implements common features of read and write handles. _GIPCHandle instances are created via pipe().
A _GIPCWriter instance manages the write end of a pipe. It is created via pipe().
A _GIPCReader instance manages the read end of a pipe. It is created via pipe().
Receive, decode and return data from the pipe. Block gevent-cooperatively until data is available or timeout expires. The default decoder is pickle.loads.
|Parameters:||timeout – None (default) or a gevent.Timeout instance. The timeout must be started to take effect and is canceled when the first byte of a new message arrives (i.e. providing a timeout does not guarantee that the method completes within the timeout interval).|
|Returns:||a Python object.|
Recommended usage for silent timeout control:
with gevent.Timeout(TIME_SECONDS, False) as t: reader.get(timeout=t)
The timeout control is currently not available on Windows, because Windows can’t apply select() to pipe handles. An OSError is expected to be raised in case you set a timeout.
A _GIPCDuplexHandle instance manages one end of a bidirectional pipe-based message transport created via pipe() with duplex=True. It provides put(), get(), and close() methods which are forwarded to the corresponding methods of gipc._GIPCWriter and gipc._GIPCReader.
Compatible with the multiprocessing.Process API.
For cooperativeness with gevent and compatibility with libev, it currently re-implements start(), is_alive(), exitcode on Unix and join() on Windows as well as on Unix.
On Unix, child monitoring is implemented via libev child watchers. To that end, libev installs its own SIGCHLD signal handler. Any call to os.waitpid() would compete with that handler, so it is not recommended to call it in the context of this module. gipc prevents multiprocessing from calling os.waitpid() by monkey-patching multiprocessing.forking.Popen.poll to always return None. Calling gipc._GProcess.join() is not required for cleaning up after zombies (libev does). It just waits until the process has terminated.