# @brief An IPC channel allowing to call methods of an object from a # remote process. It is based on python's multiprocessing pipes: # instead of sending data, we send method call requests to an object # and transmit back the results (or exceptions). Works between # processes created by fork() (eg. multiprocessing.Process::start()). # # A CallPipe is bound to a 'target' object, and is made of 2 endpoints # and one multiprocessing.Pipe inbetween: # - a CallPipe_Callee, which is basically a thread listening to method # call requests coming on the pipe, calling the method on the target # object and returning the result onto the pipe to the caller # - a CallPipe_Caller, which is a proxy object transforming the calls # to its __getattr__ into a MethodProxy object which in turns # transforms the calls to its __call__ method into request/response on # the pipe. # # A callpipe is associated with 1 'target' object, and should be used # by 2 different processes: the one that contains the object, and the # one that remotely calls the methods of the object. If there are more # processes that have to call the methods on the target object, more # callpipes have to be created (associated with the same target # object). # # On the 'target' object side (CallPipe_Callee), each callpipe # corresponds to a thread that may call its methods. As a consequence, # the methods that might be called from the callpipe threads have to be # thread-safe. # # On the remote side, the callpipe is multithread-safe. Currently, the # implementation is very crude: only one remote thread is allowed to # call a method from the target object at any given time (mutual # exclusion). # # Note: Once the remote processes have a reference to the # CallPipe_Caller object, they can discard any reference to the real # object that might still exist in their image (due to the fork() # nature of multiprocessing.Process::start()). In any case, if they # call the methods of the (copy they have of the) target object, # instead of the methods of the proxy, then they will act on their # private copy of the object, not on the remote target object. import threading, thread, os, select try: # Tested with python 2.6 b3 import multiprocessing except ImportError: # Tested with standalone processing 0.52 import processing as multiprocessing class CallPipe_MethodProxy: """The remote proxy whose __call__ method sends the request and waits for the answer on the pipe""" def __init__(self, callpipe_caller, attr_name, cb_unregister = None): self.__caller = callpipe_caller self.__attr_name = attr_name self.__unregister = cb_unregister def __call__(self, *args, **kw): # We need to acquire the lock because if 2 threads concurrently # try to send/recv concurrently on the same pipe, the callee # will not be able to tell who is calling what and who it needs to # answer to. self.__caller._lock.acquire() try: self.__caller._endpoint.send((self.__attr_name, args, kw)) recvdata = self.__caller._endpoint.recv() finally: self.__caller._lock.release() if len(recvdata) != 2: self.__caller._unregister(self.__attr_name) raise AttributeError(self.__attr_name) status, r = recvdata if status != "OK": raise r return r class CallPipe_CallerProxy: """The remote proxy whose __getattr__ method simply calls CallPipe_Caller::get_method_proxy()""" def __init__(self, gateway): self.__gateway = gateway def __getattr__(self, attr_name): return self.__gateway._get_method_proxy(attr_name) class CallPipe_Caller: """The callpipe endpoint which binds a CallPipe_CallerProxy to a request/response session over a pipe""" def __init__(self, endpoint): """Endpoint is an endpoint of a bidirectional multiprocessing.Pipe""" self._endpoint = endpoint self._lock = threading.Lock() self.__method_proxies = dict() # Cache of MethodProxy objects self.__proxy = CallPipe_CallerProxy(self) def get_proxy(self): """Returns the associated CallPipe_CallerProxy object""" return self.__proxy def _get_method_proxy(self, method_name): """Called by the proxy object: creates a CallPipe_MethodProxy object for the given method name and store it into the local cache (__method_proxies). Returns the MethodProxy object.""" self._lock.acquire() try: method_proxy = self.__method_proxies[method_name] except KeyError: method_proxy = CallPipe_MethodProxy(self, method_name) self.__method_proxies[method_name] = method_proxy finally: self._lock.release() return method_proxy def start(self): pass def stop(self): """Send a termination request to the CallPipe_Callee""" self.__endpoint.send(("TERMINATE",)) self.__endpoint.close() def _unregister(self, attr_name): """Called by the CallPipe_MethodProxy object when the remote object signals an AttributeError, to remove it from the __method_proxies cache""" del self.__method_proxies[attr_name] class CallPipe_CalleeThread(threading.Thread): """The thread that waits for the remote requests coming from the CallPipe_Caller at the other end of the pipe and performes the requested local method calls on the target object""" def __init__(self, termfd, endpoint, obj): """ \param termfd is a file descriptor on which to select() to wait for termination requests from the main CallPipe_Callee thread \param Endpoint is an endpoint of a bidirectional multiprocessing.Pipe \param obj is the object on which to perform the method calls """ threading.Thread.__init__(self) self.__endpoint = endpoint self.__obj = obj self.__waitset = select.poll() eventmask = select.POLLIN | select.POLLERR \ | select.POLLHUP | select.POLLPRI self.__waitset.register(self.__endpoint.fileno(), eventmask) self.__waitset.register(termfd, eventmask) def run(self): while True: # Check whether we received something from either the # callpipe or the terminating pipe fds = set([fd for fd, evt in self.__waitset.poll()]) if len(fds) > 1: break if self.__endpoint.fileno() not in fds: break request = self.__endpoint.recv() if not isinstance(request, tuple) or len(request) != 3: # We have a problem. stopping ourselves try: if request[0] == "TERMINATE": raise SystemExit("Received a termination request.") except IndexError: pass raise SystemExit("Invalid requests received by Callee.") method_name, args, kw = request try: method = None try: method = getattr(self.__obj, method_name) except AttributeError: self.__endpoint.send(("ATTRIBUTE_ERROR",)) if method: result = method(*args, **kw) self.__endpoint.send(("OK", result)) except Exception, ex: self.__endpoint.send(("EXCEPTION", ex)) except: self.__endpoint.send(("EXCEPTION", RuntimeError("Uncaught exception"))) # Closing the endpoint (never reached) self.__endpoint.close() class CallPipe_Callee: """The object that binds the local target object to a thread listening for the method requests coming from the pipe""" def __init__(self, endpoint, obj): """ \param Endpoint is an endpoint of a bidirectional multiprocessing.Pipe \param obj is the object on which to perform the method calls """ term_r, term_w = os.pipe() self.__obj = obj self.__termpipe = term_w self.__thread = CallPipe_CalleeThread(term_r, endpoint, obj) def get_object(self): return self.__obj def get_request_handler(self): return self.__thread def start(self): self.__thread.start() def stop(self): """Stop the thread performing the requests. Returns when the thread has been stopped""" # Send a terminate request to the terminating pipe. # Has to be called from outside the thread (otherwise: deadlock) os.write(self.__termpipe, "TERMINATE") self.__thread.join() def CallPipe(obj): """ Returns the endpoints of the callpipe: result[0]: a CallPipe_Callee object, the local endpoint referencing the target object (listening for requests) result[1]: a CallPipe_Caller object, the endpoint to remotely call the methods on the target object (sending requests) """ src, dst = multiprocessing.Pipe() return (CallPipe_Callee(dst, obj), CallPipe_Caller(src)) if __name__ == "__main__": import time class MyObject: def __init__(self, name): self.__name = name self.__trace = [] self.__lock = threading.Lock() def add_trace(self, item): self.__lock.acquire() try: self.__trace.append(item) finally: self.__lock.release() def get_traces(self): self.__lock.acquire() try: return self.__trace finally: self.__lock.release() def raise_exception(self): return 1/0 class Process1(multiprocessing.Process): def __init__(self, obj_proxy): multiprocessing.Process.__init__(self) self.__obj = obj_proxy def run(self): for i in xrange(10): print "[%d] Calling add_trace %d..." % (os.getpid(), i) self.__obj.add_trace("I am pid %d in loop %d" % (os.getpid(), i)) time.sleep(1) print "[%d] Now trying to get an exception..." % os.getpid() try: self.__obj.raise_exception() except ZeroDivisionError, ex: self.__obj.add_trace('I am pid %d and got the expected exception "%s".' % (os.getpid(), ex)) finally: print "[%d] Exception test executed." % os.getpid() print "[%d] End of process." % os.getpid() class Process2(multiprocessing.Process): def __init__(self, obj_proxy): multiprocessing.Process.__init__(self) self.__obj = obj_proxy def run(self): for i in xrange(10): print "[%d] Calling add_trace %d..." % (os.getpid(), i) self.__obj.add_trace("I am pid %d in loop %d" % (os.getpid(), i)) print "[%d] Calling get_traces %d..." % (os.getpid(), i) self.__obj.add_trace("I am pid %d and see %d traces before."\ % (os.getpid(), len(self.__obj.get_traces()))) time.sleep(1) print "[%d] Pausing 3 seconds..." % os.getpid() time.sleep(3) print "[%d] Calling final get_traces:" % os.getpid() for t in self.__obj.get_traces(): print "[%d] Trace: '%s'" % (os.getpid(), t) print "[%d] End of process." % os.getpid() # Creating object called by everybody obj = MyObject("The object") obj.add_trace("[%d] I am the parent of everything." % os.getpid()) # Creating the callpipes print "[%d] Creating callpipes and processes..." % os.getpid() callpipe1 = CallPipe(obj) callpipe2 = CallPipe(obj) p1 = Process1(callpipe1[1].get_proxy()) p2 = Process2(callpipe2[1].get_proxy()) # # The next 2 steps can be done in any order # # Starting the children processes print "Starting children processes..." p1.start() p2.start() # Starting the callpipe threads print "Starting callpipes..." callpipe1[0].start() callpipe2[0].start() # # The previous 2 steps could be done in any order # # Waiting for children processes to finish print "Waiting for processes to terminate..." try: for p in p1,p2: p.join() except KeyboardInterrupt: for p in p1,p2: p.terminate() print "Processes terminated. Cleaning up..." obj.add_trace("[%d] I am the parent, and my children are done." \ % os.getpid()) # Stopping the call pipe threads callpipe1[0].stop() callpipe2[0].stop() # Dump the contents of the object print "[%d] Children done. Dumping object:" % os.getpid() for t in obj.get_traces(): print " Trace: '%s'" % t print "Bye."