The pprocess module defines a simple parallel processing API
for Python, inspired somewhat by the thread module, slightly less
by pypar, and slightly
less still by pypvm.
This document complements the tutorial by providing an overview of the different styles of parallel programming supported by the module. For an introduction and in order to get a clearer idea of the most suitable styles for your own programs, consult the tutorial.
To create new processes to run a function or any callable object, specify the "callable" and any arguments as follows:
channel = pprocess.start(fn, arg1, arg2, named1=value1, named2=value2)
This returns a channel which can then be used to communicate with the created process. Meanwhile, in the created process, the given callable will be invoked with another channel as its first argument followed by the specified arguments:
def fn(channel, arg1, arg2, named1, named2):
# Read from and write to the channel.
# Return value is ignored.
...
To create new processes in a similar way to that employed when using os.fork
(ie. the fork system call on various operating systems), use the following
method:
channel = pprocess.create()
if channel.pid == 0:
# This code is run by the created process.
# Read from and write to the channel to communicate with the
# creating/calling process.
# An explicit exit of the process may be desirable to prevent the process
# from running code which is intended for the creating/calling process.
...
pprocess.exit(channel)
else:
# This code is run by the creating/calling process.
# Read from and write to the channel to communicate with the created
# process.
...
When creating many processes, each providing results for the consumption of the main process, the collection of those results in an efficient fashion can be problematic: if some processes take longer than others, and if we decide to read from those processes when they are not ready instead of other processes which are ready, the whole activity will take much longer than necessary.
One solution to the problem of knowing when to read from channels is to create
an Exchange object, optionally initialising it with a list of channels
through which data is expected to arrive:
exchange = pprocess.Exchange() # populate the exchange later exchange = pprocess.Exchange(channels) # populate the exchange with channels
We can add channels to the exchange using the add method:
exchange.add(channel)
To test whether an exchange is active - that is, whether it is actually
monitoring any channels - we can use the active method which
returns all channels being monitored by the exchange:
channels = exchange.active()
We may then check the exchange to see whether any data is ready to be received; for example:
for channel in exchange.ready():
# Read from and write to the channel.
...
If we do not wish to wait indefinitely for a list of channels, we can set a
timeout value as an argument to the ready method (as a floating
point number specifying the timeout in seconds, where 0 means a
non-blocking poll as stated in the select module's select
function documentation).
A convenient form of message exchanges can be adopted by defining a subclass of
the Exchange class and defining a particular method:
class MyExchange(pprocess.Exchange):
def store_data(self, channel):
data = channel.receive()
# Do something with data here.
The exact operations performed on the received data might be as simple as storing it on an instance attribute. To make use of the exchange, we would instantiate it as usual:
exchange = MyExchange() # populate the exchange later exchange = MyExchange(limit=10) # set a limit for later population
The exchange can now be used in a simpler fashion than that shown above. We can
add channels as before using the add method, or we can choose to only
add channels if the specified limit of channels is not exceeded:
exchange.add(channel) # add a channel as normal
exchange.add_wait(channel) # add a channel, waiting if the limit would be
# exceeded
Or we can request that the exchange create a channel on our behalf:
channel = exchange.create()
We can even start processes and monitor channels without ever handling the channel ourselves:
exchange.start(fn, arg1, arg2, named1=value1, named2=value2)
We can explicitly wait for "free space" for channels by calling the
wait method, although the start and add_wait
methods make this less interesting:
exchange.wait()
Finally, when finishing the computation, we can choose to merely call the
finish method and have the remaining data processed automatically:
exchange.finish()
Clearly, this approach is less flexible but more convenient than the raw message exchange API as described above. However, it permits much simpler and clearer code.
Instead of having to subclass the pprocess.Exchange class and
to define the store_data method, it might be more desirable to let
the exchange manage the communications between created and creating processes
and to let the creating process just consume received data as it arrives,
without particular regard for the order of the received data - perhaps the
creating process has its own way of managing such issues.
For such situations, the Queue class may be instantiated and
channels added to the queue using the various methods provided:
queue = pprocess.Queue(limit=10)
channel = queue.create()
if channel:
# Do some computation.
pprocess.exit(channel)
The results can then be consumed by treating the queue like an iterator:
for result in queue:
# Capture each result.
This approach does not, of course, require the direct handling of channels.
One could instead use the start method on the queue to create
processes and to initiate computations (since a queue is merely an enhanced
exchange with a specific implementation of the store_data
method).
Where the above Queue class appears like an attractive solution
for the management of the results of computations, but where the order of their
consumption by the creating process remains important, the Map
class may offer a suitable way of collecting and accessing results:
results = pprocess.Map(limit=10)
for value in inputs:
results.start(fn, args)
The results can then be consumed in an order corresponding to the order of the computations which produced them:
for result in results:
# Process each result.
Internally, the Map object records a particular ordering of
channels, ensuring that the received results can be mapped to this ordering,
and that the results can be made available with this ordering preserved.
A further simplification of the above convenient use of message exchanges
involves the creation of callables (eg. functions) which are automatically
monitored by an exchange. We create such a callable by calling the
manage method on an exchange:
myfn = exchange.manage(fn)
This callable can then be invoked instead of using the exchange's
start method:
myfn(arg1, arg2, named1=value1, named2=value2)
The exchange's finish method can be used as usual to process
incoming data.
In making a program parallel, existing functions which only return results can
be manually modified to accept and use channels to communicate results back to
the main process. However, a simple alternative is to use the MakeParallel
class to provide a wrapper around unmodified functions which will return the results
from those functions in the channels provided. For example:
fn = pprocess.MakeParallel(originalfn)
In situations where a callable would normally be used in conjunction with the
Python built-in map function, an alternative solution can be adopted by using
the pmap function:
pprocess.pmap(fn, sequence)
Here, the sequence would have to contain elements that each contain the
required parameters of the specified callable, fn. Note that the
callable does not need to be a parallel-aware function which has a
channel argument: the pmap function automatically
wraps the given callable internally.
So far, all parallel computations have been done with newly-created
processes. However, this can seem somewhat inefficient, especially if processes
are being continually created and destroyed (although if this happens too
often, the amount of work done by each process may be too little, anyway). One
solution is to retain processes after they have done their work and request
that they perform more work for each new parallel task or invocation. To enable
the reuse of processes in this way, a special keyword argument may be specified
when creating Exchange instances (and instances of subclasses such
as Map and Queue). For example:
exchange = MyExchange(limit=10, reuse=1) # reuse up to 10 processes
Code invoked through such exchanges must be aware of channels and be
constructed in such a way that it does not terminate after sending a result
back to the creating process. Instead, it should repeatedly wait for subsequent
sets of parameters (compatible with those either in the signature of a callable
or with the original values read from the channel). Reusable code is terminated
when the special value of None is sent from the creating process
to the created process, indicating that no more parameters will be sent; this
should cause the code to terminate.
An easier way of making reusable code sections for parallel use is to employ the
MakeReusable class to wrap an existing callable:
fn = pprocess.MakeReusable(originalfn)
This wraps the callable in a similar fashion to MakeParallel, but
provides the necessary mechanisms described above for reusable code.
When created/child processes terminate, one would typically want to be informed of such conditions using a signal handler. Unfortunately, Python seems to have issues with restartable reads from file descriptors when interrupted by signals:
The exact combination of conditions indicating closed pipes remains relatively obscure. Here is a message/thread describing them (in the context of another topic):
It would seem, from using sockets and from studying the asyncore
module, that sockets are more predictable than pipes.
Notes about poll implementations can be found here: