async.py

Go to the documentation of this file.

#!/usr/bin/env python
'''
Provide tools for running code asynchronously.
'''

def async_caller(obj, inbox, outbox):
    '''

    Accept messages, convert to method calls on given object, pass
    return value as message.

    '''
    while True:
        msg = inbox.get(block=True)
        methname, methnumber, args, kwds = msg
        if methname == 'async_caller_exit':
            break
        meth = eval("obj.%s"%methname)
        ret = meth(*args, **kwds)
        outbox.put((methnumber, ret))
        continue
    return

class AsyncMethod(object):
    '''

    A callable that stands in for a method.

    '''

    def __init__(self, name, number, outbox):
        self.name = name
        self.number = number
        self.outbox = outbox
        self.callback = None
        #print 'AsyncMethod "%s" stacked'%name
        return

    def __call__(self, callback, *args, **kwds):
        if self.callback:
            msg = 'This instance of the %s method has already been caled'
            raise AttributeError, msg % self.name
        self.callback = callback
        self.outbox.put((self.name, self.number, args, kwds))
        return


class AsyncInterface(object):
    '''

    Asynchronous interface to some other object (called "real" below).

    Any method calls on this object, other than the ones explicitly
    defined, must have as their first argument a callable object
    ("callback").  The method name and remaining arguments will be
    queued to another object running in a different process where it
    will be interpted and result in calling the method on the real
    object.  The result will be queued back to this interface object
    where the callback will be called with the result as its argument.

    '''

    def __init__(self, obj, proctype=None):
        if not obj:
            raise ValueError,'Must give a valid object'
        self.__dict__['_obj'] = obj
        self.__dict__['_count'] = 0
        self.__dict__['_callbacks'] = {}

        if proctype:
            meth = eval ("self.do_%s"%proctype)
            meth()
            return
        else:
            try:
                import multiprocessing
            except ImportError:
                self.do_threaded()
            else:
                self.do_multiprocessing()
                pass
            pass
        return

    def do_multiprocessing(self):
        from multiprocessing import Queue, Process
        self.__dict__['_proctype'] = 'multiprocessing'
        self.__dict__['_outbox'] = outbox = Queue()
        self.__dict__['_inbox']  = inbox  = Queue()
        self.__dict__['_proc'] = proc = Process(target=async_caller, 
                                                args=(self._obj, outbox, inbox))
        proc.start()
        return

    def do_threaded(self):
        from Queue import Queue
        from threading import Thread
        self.__dict__['_proctype'] = 'threading'
        self.__dict__['_outbox'] = outbox = Queue()
        self.__dict__['_inbox']  = inbox  = Queue()
        self.__dict__['_proc'] = proc = Thread(target=async_caller, 
                                               args=(self._obj, outbox, inbox))
        proc.start()
        return

    def __getattr__(self, name):
        if not self._obj:
            raise AttributeError,'No object'
        if not hasattr(self._obj, name):
            msg = 'No such attribute: "%s" in "%s"'
            raise AttributeError, msg % (name, self._obj)
        self._count += 1
        meth = AsyncMethod(name, self._count, self._outbox)
        self._callbacks[self._count] = meth
        return meth

    def __del__(self):
        self.shutdown()
        return

    def shutdown(self):
        'Gently shutdown the real object and drain the queue'
        self._outbox.put(('async_caller_exit', None, None, None))
        self.drain_queue()
        self._proc.join()
        return

    def abort(self):
        'Immediately shutdown'
        if self._proctype == 'multiproces':
            self._proc.terminate()
        import sys
        sys.exit(0)
        return

    def drain_queue(self, depth = None):
        'Drain queue of at most depth pending results.  All if depth=None'
        while depth is None or depth > 0:
            if depth is not None: depth -= 1
            if self._inbox.empty(): 
                #print 'queue empty'
                return
            cbnum, res = self._inbox.get()
            meth = self._callbacks[cbnum]
            #print 'calling',meth.callback,'number',cbnum,'with',res
            meth.callback(res)
            del self._callbacks[cbnum]
            continue
        return