gipc: child processes and IPC for gevent¶

Quick start example¶

The following code snippet uses gipc for spawning a child process and for creating a pipe, and then sends a Python object from a greenlet in the main (parent) process through the pipe to the child process:

import gevent
import gipc

def writelet(w):
    # This function runs as a greenlet in the parent process.
    # Put a Python object into the write end of the transport channel.
    w.put(0)

def readchild(r):
    # This function runs in a child process.
    # Read and validate object from the read end of the transport channel.
    assert r.get() == 0

def main():
    with gipc.pipe() as (readend, writeend):
        # Start 'writer' greenlet. Provide it with the pipe write end.
        g = gevent.spawn(writelet, writeend)
        # Start 'reader' child process. Provide it with the pipe read end.
        p = gipc.start_process(target=readchild, args=(readend,))
        # Wait for both to finish.
        g.join()
        p.join()

# Protect entry point from being executed upon import; crucial on Windows.
if __name__ == "__main__":
    main()

Although quite simple, this code would have various unwanted side-effects if used with the canonical multiprocessing API instead of gipc.start_process() and gipc.pipe(), as outlined in the Challenges paragraph.

What are the challenges and what is gipc’s solution?¶

Challenges:

Depending on the operating system in use, the creation of child processes via Python’s multiprocessing in the context of a gevent application requires special treatment. Most care is required on POSIX-compliant systems: greenlets spawned in the current process before forking obviously become cloned by fork() and haunt in the child, which usually is undesired behavior. The following code snippet clarifies this behavior by implementing the example from above, but this time by directly using multiprocessing instead of gipc (this has been tested on Linux with Python 3.4 & gevent 1.1):

import gevent
import multiprocessing

def writelet(c):
    c.send(0)

def readchild(c):
    gevent.sleep(0)
    assert c.recv() == 0
    assert c.recv() == 0

if __name__ == "__main__":
    c1, c2 = multiprocessing.Pipe()
    g = gevent.spawn(writelet, c1)
    p = multiprocessing.Process(target=readchild, args=(c2,))
    p.start()
    g.join()
    p.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 may lead to a wide range of side-effects including race conditions, and therefore almost guarantees especially tedious debugging sessions.

The second class of serious issues in the code above is that it contains several non-cooperatively blocking function calls: p.join() as well as the send()/recv() calls (of multiprocessing.Connection objects) block the calling greenlet non-cooperatively, i.e. they do not allow for a context switch into other greenlets. While this does not lead to an error in the simple example code above, this behavior is not tolerable in real-world gevent applications.

Solution:

gipc overcomes these and other issues for you transparently and in a straight- forward fashion.

First of all, the most basic design assumption behind gipc is that application developers never actually want to duplicate all running greenlets during fork. This leads to the rational of first destroying the inherited “gevent state” in the child and then creating a fresh gevent context, before invoking the target function.

So, the goal is that children start off with a fresh gevent state before entering the user-given target function. Consequently, as one of the first actions, children created via gipc destroy the inherited gevent hub as well as the inherited libev event loop and create their own fresh versions of these entities. 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 actively, one by one.

Furthermore, on POSIX-compliant systems, gipc entirely avoids multiprocessing’s child monitoring implementation (which is based on the class of wait system calls) and instead uses libev’s wonderful child watcher system (based on SIGCHLD signal transmission), enabling gevent-cooperative waiting for child termination. That’s how p.join() from the example above can be made cooperative.

For implementing gevent-cooperative inter-process communication, gipc uses efficient pipe-based data transport channels with non-blocking I/O system calls. gipc’s transport channel system has been carefully designed (for instance, it takes care of closing dispensable file descriptors in the parent as well as in the child after forking) and also abstracts away the difficulties of passing pipe handles among processes on Windows. gipc also abstracts away the implementation differences of the multiprocessing package between Python 2 and 3.

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, and on Python 2 and 3.

Example 1: gipc.pipe()-based messaging from greenlet in parent to child¶

Very basic gevent and gipc concepts are explained by means of the following 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 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().

Example 2: serving multiple clients (in child) from one server (in parent)¶

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:

1. gevent’s StreamServer is started in a greenlet within the initial (parent) process. For each connecting client, it receives one newline-terminated message and echoes it back.
2. A child process is started using gipc. Its starting point is the function clientprocess. There, N TCP clients are started concurrently from N greenlets.
3. Each client sends one message, validates the echo response and terminates.
4. The child process terminates.
5. After the child process is joined in the parent, the server is killed.
6. The server greenlet is joined.

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.

Example 3: time-synchronization between processes¶

Child process creation may take a significant amount of time, especially on Windows. The exact amount of 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:

• Process A sends a synchronization request to process B and waits for an acknowledgment response. It proceeds upon retrieval.
• Process B sends the acknowledgment in the moment it retrieves the sync request and proceeds.

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).

Spawning child processes¶

gipc.start_process(target, args=(), kwargs={}, daemon=None, name=None)¶

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, gipc automatically closes the corresponding file descriptor(s) in the parent.

Note

Compared to the canonical multiprocessing.Process() constructor, this function

• returns a gipc._GProcess instance which is compatible with the multiprocessing.Process API.
• just as well takes the target, arg=(), and kwargs={} arguments.
• introduces the daemon=None argument.
• does not accept the group argument (being an artifact from multiprocessing‘s compatibility with threading).
• starts the process, i.e. a subsequent call to the start() method of the returned object is not required.

Parameters:	target – Function to be called in the child process. Signature: target(*args, **kwargs). args – Tuple defining the positional arguments provided to target. kwargs – Dictionary defining the keyword arguments provided to target. name – Forwarded to multiprocessing.Process.name. daemon – Forwarded to multiprocessing.Process.daemon.
Returns:	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.

Warning

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. The notable exception is the SIGPIPE signal, whose handler is not reset to its default handler in child processes created by gipc. That is, the SIGPIPE action in children is inherited from the parent. In CPython, the default action for SIGPIPE is SIG_IGN, i.e. the signal is ignored.

Creating a pipe and its handle pair¶

gipc.pipe(duplex=False, encoder='default', decoder='default')¶

Create a pipe-based message transport channel and return two corresponding handles for reading and writing data.

Allows for gevent-cooperative transmission of data between greenlets within one process or across processes (created via start_process()). The default behavior allows for transmission of any picklable Python object.

The transport layer is based on os.pipe() (i.e. CreatePipe() on Windows and pipe() on POSIX-compliant systems).

Parameters:

duplex – If False (default), create a unidirectional pipe-based message transport channel and return the corresponding handle pair, a 2-tuple with the first element of type _GIPCReader and the second element of type _GIPCWriter. If True, create a bidirectional message transport channel (using two pipes internally) and return the corresponding 2-tuple with both elements being of type _GIPCDuplexHandle. encoder – Defines the entity used for object serialization before writing object o to the pipe via put(o). Must be either a callable returning a byte string, None, or 'default'. 'default' translates to pickle.dumps (in this mode, any pickleable Python object can be provided to put() and transmitted through the pipe). When setting this to None, no automatic object serialization is performed. In that case only byte strings are allowed to be provided to put(), and a TypeError is thrown otherwise. A TypeError will also be thrown if the encoder callable does not return a byte string. decoder – Defines the entity used for data deserialization after reading raw binary data from the pipe. Must be a callable retrieving a byte string as first and only argument, None or 'default'. 'default' translates to pickle.loads. When setting this to None, no data decoding is performed, and a raw byte string is returned.

Returns:

duplex=False: (reader, writer) 2-tuple. The first element is of type gipc._GIPCReader, the second of type gipc._GIPCWriter. Both inherit from gipc._GIPCHandle. duplex=True: (handle, handle) 2-tuple. Both elements are of type gipc._GIPCDuplexHandle.

gipc._GIPCHandle and gipc._GIPCDuplexHandle instances are recommended to be used with a context manager as indicated in the following examples:

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

An example for using the encoder/decoder arguments for implementing JSON (de)serialization:

import json
enc = lambda o: json.dumps(o).encode("ascii")
dec = lambda b: json.loads(b.decode("ascii"))
with pipe(encoder=enc, decoder=dec) as (r, w):
    ...

Note that JSON representation is text whereas the encoder/decoder callables must return/accept byte strings, as ensured here by ASCII en/decoding. Also note that in practice JSON serializaton has normally no advantage over pickling, so this is just an educational example.

Handling handles¶

class gipc.gipc._GIPCHandle¶

The _GIPCHandle class implements common features of read and write handles. _GIPCHandle instances are created via pipe().

close()¶

Close underlying file descriptor and de-register handle from further usage. Is called on context exit.

Raises:

• GIPCError
• GIPCClosed
• GIPCLocked

class gipc.gipc._GIPCWriter¶

Bases: gipc.gipc._GIPCHandle

A _GIPCWriter instance manages the write end of a pipe. It is created via pipe().

put(o)¶

Encode object o and write it to the pipe. Block gevent-cooperatively until all data is written. The default encoder is pickle.dumps.

Parameters:	o – a Python object that is encodable with the encoder of choice.

Raises:

• GIPCError
• GIPCClosed
• pickle.PicklingError

class gipc.gipc._GIPCReader¶

Bases: gipc.gipc._GIPCHandle

A _GIPCReader instance manages the read end of a pipe. It is created via pipe().

get(timeout=None)¶

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.

Raises:

• gevent.Timeout (if provided)
• GIPCError
• GIPCClosed
• pickle.UnpicklingError

Recommended usage for silent timeout control:

with gevent.Timeout(TIME_SECONDS, False) as t:
    reader.get(timeout=t)

Warning

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.

class gipc.gipc._GIPCDuplexHandle¶

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.

Controlling child processes¶

class gipc.gipc._GProcess¶

Bases: multiprocessing.context.Process

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.

Note

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’s Popen.poll to be no-op and to always return None. Calling gipc._GProcess.join() is not required for cleaning up after zombies (libev does). It just waits for the process to terminate.

Exception types¶

exception gipc.GIPCError¶

Is raised upon general errors. All other exception types derive from this one.

exception gipc.GIPCLocked¶

Is raised upon attempt to close a handle which is currently locked for I/O.

exception gipc.GIPCClosed¶

Is raised upon operation on closed handle.