nilmdb/nilmdb/utils/serializer.py

import queue
import threading
import sys
import decorator
import inspect
import types
import functools

# This file provides a class that will wrap an object and serialize
# all calls to its methods.  All calls to that object will be queued
# and executed from a single thread, regardless of which thread makes
# the call.

# Based partially on http://stackoverflow.com/questions/2642515/

class SerializerThread(threading.Thread):
    """Thread that retrieves call information from the queue, makes the
    call, and returns the results."""
    def __init__(self, classname, call_queue):
        threading.Thread.__init__(self)
        self.name = "Serializer-" + classname + "-" + self.name
        self.call_queue = call_queue

    def run(self):
        while True:
            result_queue, func, args, kwargs = self.call_queue.get()
            # Terminate if result_queue is None
            if result_queue is None:
                return
            exception = None
            result = None
            try:
                result = func(*args, **kwargs) # wrapped
            except:
                exception = sys.exc_info()
            # Ensure we delete these before returning a result, so
            # we don't unncessarily hold onto a reference while
            # we're waiting for the next call.
            del func, args, kwargs
            result_queue.put((exception, result))
            del exception, result

def serializer_proxy(obj_or_type):
    """Wrap the given object or type in a SerializerObjectProxy.

    Returns a SerializerObjectProxy object that proxies all method
    calls to the object, as well as attribute retrievals.

    The proxied requests, including instantiation, are performed in a
    single thread and serialized between caller threads.
    """
    class SerializerCallProxy(object):
        def __init__(self, call_queue, func, objectproxy):
            self.call_queue = call_queue
            self.func = func
            # Need to hold a reference to object proxy so it doesn't
            # go away (and kill the thread) until after get called.
            self.objectproxy = objectproxy
        def __call__(self, *args, **kwargs):
            result_queue = queue.Queue()
            self.call_queue.put((result_queue, self.func, args, kwargs))
            ( exc_info, result ) = result_queue.get()
            if exc_info is None:
                return result
            else:
                raise exc_info[1].with_traceback(exc_info[2])

    class SerializerObjectProxy(object):
        def __init__(self, obj_or_type, *args, **kwargs):
            self.__object = obj_or_type
            try:
                if type(obj_or_type) == type:
                    classname = obj_or_type.__name__
                else:
                    classname = obj_or_type.__class__.__name__
            except AttributeError: # pragma: no cover
                classname = "???"
            self.__call_queue = queue.Queue()
            self.__thread = SerializerThread(classname, self.__call_queue)
            self.__thread.daemon = True
            self.__thread.start()
            self._thread_safe = True

        def __getattr__(self, key):
            attr = getattr(self.__object, key)
            if not callable(attr):
                getter = SerializerCallProxy(self.__call_queue, getattr, self)
                return getter(self.__object, key)
            r = SerializerCallProxy(self.__call_queue, attr, self)
            return r

        # For an interable object, on __iter__(), save the object's
        # iterator and return this proxy.  On next(), call the object's
        # iterator through this proxy.
        def __iter__(self):
            attr = getattr(self.__object, "__iter__")
            self.__iter = SerializerCallProxy(self.__call_queue, attr, self)()
            return self
        def __next__(self):
            return SerializerCallProxy(self.__call_queue,
                                       self.__iter.__next__, self)()

        def __getitem__(self, key):
            return self.__getattr__("__getitem__")(key)

        def __call__(self, *args, **kwargs):
            """Call this to instantiate the type, if a type was passed
            to serializer_proxy.  Otherwise, pass the call through."""
            ret = SerializerCallProxy(self.__call_queue,
                                      self.__object, self)(*args, **kwargs)
            if type(self.__object) == type:
                # Instantiation
                self.__object = ret
                return self
            return ret

        def __del__(self):
            try:
                # Signal thread to exit, but don't wait for it.
                self.__call_queue.put((None, None, None, None))
            except: # pragma: no cover
                pass

    return SerializerObjectProxy(obj_or_type)