root/trunk/storage/__init__.py

"""Storage Managers for Dejavu."""

import datetime
import threading
import thread
import recur
try:
    import cPickle as pickle
except ImportError:
    import pickle


class StorageManager(object):
    """A Manager base class for storing and retrieving Units.
    
    The base StorageManager class doesn't actually store anything;
    it needs to be subclassed.
    """
    
    def __init__(self, name, arena, allOptions={}):
        self.name = name
        self.arena = arena
        self.shutdownOrder = int(allOptions.get('Shutdown Order', '0'))
    
    def recall(self, unitClass, expr=None):
        """Return an iterable object which will populate Units."""
        raise NotImplementedError
    
    def save(self, unit, forceSave=False):
        """Store the object's data{} dictionary."""
        raise NotImplementedError
    
    def destroy(self, unit):
        """Delete the object."""
        raise NotImplementedError
    
    def create_storage(self, unitClass):
        raise NotImplementedError
    
    def reserve(self, unit):
        """Reserve storage space for the Unit."""
        raise AttributeError
    
    def shutdown(self):
        pass


class ProxyStorage(StorageManager):
    """A Storage Manager which passes calls to another Storage Manager."""
    
    nextstore = None
    
    def __init__(self, name, arena, allOptions={}):
        StorageManager.__init__(self, name, arena, allOptions)
        
        nextstore = allOptions.get('Next Store')
        if nextstore:
            self.nextstore = arena.stores[nextstore]
    
    def recall(self, unitClass, expr=None):
        if self.nextstore:
            for unit in self.nextstore.recall(unitClass, expr):
                yield unit
    
    def save(self, unit, forceSave=False):
        """Store the unit."""
        if self.nextstore:
            self.nextstore.save(unit, forceSave)
    
    def destroy(self, unit):
        """Delete the unit."""
        if self.nextstore:
            self.nextstore.destroy(unit)
    
    def create_storage(self, unitClass):
        if self.nextstore:
            self.nextstore.create_storage(unitClass)
    
    def reserve(self, unit):
        """Reserve storage space for the Unit."""
        if self.nextstore:
            self.nextstore.reserve(unit)


class CachingProxy(ProxyStorage):
    """A Proxy Storage Manager which recalls and keeps Units in memory."""
    
    def __init__(self, name, arena, allOptions={}):
        ProxyStorage.__init__(self, name, arena, allOptions)
        
        self._caches = {}       # id: pickled Unit
        self._cache_locks = {}
        self._recallTimes = {}
        
        # Create and motivate a worker to sweep out idle Units.
        lifetime = allOptions.get('Lifetime', '')
        if lifetime:
            self.recurrence = recur.Recurrence(None, lifetime)
            if self.recurrence:
                # Throw away the first occurrence value,
                # which is almost always .now()
                self.recurrence.next()
            self.lastrun = datetime.datetime.now()
            self.active = True
            threading.Timer(self.recurrence.interval(), self._cycle).start()
    
    def _cycle(self):
        """Run the scheduled work."""
        threading.Timer(self.recurrence.interval(), self._cycle).start()
        if self.active:
            self.sweep_all(self.lastrun)
            self.lastrun = datetime.datetime.now()
    
    def cachelen(self, cls):
        return len(self._caches.get(cls, {}))
    
    def cached_units(self, cls):
        return [pickle.loads(data) for data
                in self._caches.get(cls, {}).itervalues()]
    
    def _get_lock(self, unitClass):
        if unitClass not in self._caches:
            self._caches[unitClass] = {}
            self._cache_locks[unitClass] = thread.allocate_lock()
        lock = self._cache_locks[unitClass]
        lock.acquire(True)
        return lock
    
    def recall(self, unitClass, expr=None):
        """Return a Unit iterator."""
        currentTime = datetime.datetime.now()
        lock = self._get_lock(unitClass)
        try:
            cache = self._caches[unitClass]
            seen = {}
            
            # Run through our cache first. Hopefully, this will save us
            # calling expr.evaluate twice for each unit.
            for id, pickledUnit in cache.iteritems():
                unit = pickle.loads(pickledUnit)
                if expr is None or expr.evaluate(unit):
                    seen[id] = unit
            
            if self.nextstore:
                for unit in self.nextstore.recall(unitClass, expr):
                    if unit.ID not in seen:
                        seen[unit.ID] = unit
                    if unit.ID not in cache:
                        # Pickle the Unit to discard extraneous attributes,
                        # and avoid identity issues.
                        cache[unit.ID] = pickle.dumps(unit)
                        self._recallTimes[unit.ID] = currentTime
            
            return iter(seen.values())
        finally:
            lock.release()
    
    def save(self, unit, forceSave=False):
        """Store the unit."""
        # Defer saves to sweep().
        pass
    
    def destroy(self, unit):
        """Delete the unit."""
        unitClass = unit.__class__
        lock = self._get_lock(unitClass)
        try:
            cache = self._caches[unitClass]
            if self.nextstore:
                self.nextstore.destroy(unit)
            if unit.ID in cache:
                del cache[unit.ID]
                del self._recallTimes[unit.ID]
        finally:
            lock.release()
    
    def reserve(self, unit):
        """Reserve storage space for the Unit."""
        unitClass = unit.__class__
        lock = self._get_lock(unitClass)
        try:
            cache = self._caches[unitClass]
            if self.nextstore:
                self.nextstore.reserve(unit)
            # Pickle the Unit to discard extraneous attributes,
            # and avoid identity issues.
            cache[unit.ID] = pickle.dumps(unit)
            self._recallTimes[unit.ID] = datetime.datetime.now()
        finally:
            lock.release()
    
    def sweep(self, unitClass, lastSweepTime=None):
        lock = self._get_lock(unitClass)
        try:
            cache = self._caches[unitClass]
            for id in cache.keys():
                lastRecall = self._recallTimes.setdefault(id, None)
                if (lastRecall is None or lastSweepTime is None
                    or lastRecall < lastSweepTime):
                    unit = pickle.loads(cache[id])
                    if unit.dirty:
                        self.nextstore.save(unit)
                    
                    del cache[id]
                    del self._recallTimes[id]
        finally:
            lock.release()
    
    def sweep_all(self, lastSweepTime=None):
        for cls in self._caches:
            self.sweep(cls, lastSweepTime)
    
    def shutdown(self):
        self.sweep_all()


class BurnedProxy(CachingProxy):
    """A Caching Proxy Storage Manager which recalls and caches ALL Units.
    
    The big performance difference for a burned cache is that, once _any_
    Units have been recalled, further recalls won't hit the next store.
    Notice we didn't say "performance _benefit_" ;) That would depend to
    a great extent on the proxied store.
    """
    
    def _icached_units(self, cls, expr=None):
        cache = self._caches.get(cls, {})
        if expr is None:
            for data in cache.itervalues():
                yield pickle.loads(data)
        else:
            for data in cache.itervalues():
                unit = pickle.loads(data)
                if expr.evaluate(unit):
                    yield unit
    
    def recall(self, unitClass, expr=None):
        """Return a Unit iterator."""
        lock = self._get_lock(unitClass)
        try:
            cache = self._caches[unitClass]
            
            if (not cache) and self.nextstore:
                # Read ALL units from storage.
                now = datetime.datetime.now()
                for unit in self.nextstore.recall(unitClass, None):
                    cache[unit.ID] = pickle.dumps(unit)
                    self._recallTimes[unit.ID] = now
            
            return self._icached_units(unitClass, expr)
        finally:
            lock.release()


class Adapter(object):
    """Transform values according to their type. Must be subclassed.
    
    In order for your subclass to work, you need to provide functions
    named 'coerce_' + type, where 'type' refers to the type you wish to
    support. Replace dots in the type name by underscores. For example,
    to coerce datetime.date objects, you must provide a function in your
    subclass named 'coerce_datetime_date'. For builtins, do not include
    the module name '__builtin__', just use 'coerce_unicode', for example.
    
    If you try to coerce a value for whose type you have not provided a
    coercion function, a TypeError is raised.
    
    When writing Adapters for Storage Managers, you should at least
    provide coerce_* functions for: bool, dict, float, int, list, long,
    NoneType, str, tuple, and unicode. For most applications, you should
    also provide:
        datetime.datetime
                .date
                .time
                .timedelta
    and:
        fixedpoint.FixedPoint or decimal.Decimal (preferably both)
    """
    
    def coerce(self, value, valuetype=None):
        """coerce(value, valuetype=None) -> value, coerced by valuetype."""
        if valuetype is None:
            valuetype = type(value)
        
        mod = valuetype.__module__
        if mod == "__builtin__":
            xform = "coerce_%s" % valuetype.__name__
        else:
            xform = "coerce_%s_%s" % (mod, valuetype.__name__)
        xform = xform.replace(".", "_")
        try:
            xform = getattr(self, xform)
        except AttributeError:
            raise TypeError("'%s' is not handled by %s." %
                            (valuetype, self.__class__))
        return xform(value)