#!/usr/bin/env python  """ A simple parallel processing API for Python, inspired somewhat by the thread module, slightly less by pypar, and slightly less still by pypvm.  Thread-style Processing -----------------------  To create new processes to run a function or any callable object, specify the "callable" and any arguments as follows:  channel = 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.     ...  Fork-style Processing ---------------------  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 = 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.     ... else:     # This code is run by the creating/calling process.     # Read from and write to the channel to communicate with the created     # process.     ...  Message Exchanges -----------------  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, initialising it with a list of channels through which data is expected to arrive:  exchange = Exchange(channels)  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).  Signals and Waiting -------------------  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:  http://mail.python.org/pipermail/python-dev/2002-September/028572.html http://twistedmatrix.com/bugs/issue733  Select and Poll ---------------  The exact combination of conditions indicating closed pipes remains relatively obscure. Here is a message/thread describing them (in the context of another topic):  http://twistedmatrix.com/pipermail/twisted-python/2005-February/009666.html """  import os import sys import select import socket  try:     import cPickle as pickle except ImportError:     import pickle  class Channel:      "A communications channel."      def __init__(self, pid, read_pipe, write_pipe):          """         Initialise the channel with a process identifier 'pid', a 'read_pipe'         from which messages will be received, and a 'write_pipe' into which         messages will be sent.         """          self.pid = pid         self.read_pipe = read_pipe         self.write_pipe = write_pipe         self.closed = 0      def __del__(self):          # Since signals don't work well with I/O, we close pipes and wait for         # created processes upon finalisation.          if not self.closed:             self.close()      def close(self):          "Explicitly close the channel."          self.closed = 1         self.read_pipe.close()         self.write_pipe.close()         self.wait(os.WNOHANG)      def wait(self, options=0):          "Wait for the created process, if any, to exit."          if self.pid != 0:             try:                 os.waitpid(self.pid, options)             except OSError:                 pass      def send(self, obj):          "Send the given object 'obj' through the channel."          pickle.dump(obj, self.write_pipe)         self.write_pipe.flush()      def receive(self):          "Receive an object through the channel, returning the object."          obj = pickle.load(self.read_pipe)         if isinstance(obj, Exception):             raise obj         else:             return obj  class Exchange:      """     A communications exchange that can be used to detect channels which are     ready to communicate.     """      def __init__(self, channels):          "Initialise the exchange with the given 'channels'."          self.readables = {}         self.poller = select.poll()         for channel in channels:             self.readables[channel.read_pipe.fileno()] = channel             self.poller.register(channel.read_pipe.fileno(), select.POLLIN | select.POLLHUP | select.POLLNVAL | select.POLLERR)      def active(self):          "Return the number of active channels."          return len(self.readables.keys())      def ready(self, timeout=None):          """         Wait for a period of time specified by the optional 'timeout' (or until         communication is possible) and return a list of channels which are ready         to be read from.         """          fds = self.poller.poll(timeout)         readables = []         for fd, status in fds:             channel = self.readables[fd]              # Remove ended/error channels.              if status & (select.POLLHUP | select.POLLNVAL | select.POLLERR):                 self.remove(channel)              # Record readable channels.              elif status & select.POLLIN:                 readables.append(channel)          return readables      def remove(self, channel):          """         Remove the given 'channel' from the exchange.         """          del self.readables[channel.read_pipe.fileno()]         self.poller.unregister(channel.read_pipe.fileno())  def create():      """     Create a new process, returning a communications channel to both the     creating process and the created process.     """      parent, child = socket.socketpair()     for s in [parent, child]:         s.setblocking(1)      pid = os.fork()     if pid == 0:         parent.close()         return Channel(pid, child.makefile("r"), child.makefile("w"))     else:         child.close()         return Channel(pid, parent.makefile("r"), parent.makefile("w"))  def start(callable, *args, **kwargs):      """     Create a new process which shall start running in the given 'callable'.     Return a communications channel to the creating process, and supply such a     channel to the created process as the 'channel' parameter in the given     'callable'. Additional arguments to the 'callable' can be given as     additional arguments to this function.     """      channel = create()     if channel.pid == 0:         try:             try:                 callable(channel, *args, **kwargs)             except:                 exc_type, exc_value, exc_traceback = sys.exc_info()                 channel.send(exc_value)         finally:             channel.close()             sys.exit(0)     else:         return channel  def waitall():      "Wait for all created processes to terminate."      try:         while 1:             os.wait()     except OSError:         pass  # vim: tabstop=4 expandtab shiftwidth=4