objectsdb.py

#!/usr/bin/env python
#coding=utf8

'''
The current living objects database.
'''

from __future__ import absolute_import

from datetime import datetime, timedelta
import logging
from threading import RLock

from ccserver.tql import TqlObject
from ccserver.orderedset import OrderedSet
from ccserver.exceptions import AlreadyRegistered

DEFAULT_TTL = 0
TTL_SERVER_DELTA = 1 # Delta to apply for all tags

class ObjectsDB(object):
    '''
    The object database.

    This class abstract the access to the current living objects database. The
    database store a list of :class:`TqlObject` instances, each defining an
    object with its tags.

    Objects are registered by :meth:`register` method, and removed by
    :meth:`unregister` method. You can update the tags of registered objects
    with :meth:`update` method. A call to :meth:`all` method give you the
    complete list of objects with an automatic update before to return it.

    .. note:
       All calls to the :class:`ObjectDB` are thread-safe.
    '''

    def __init__(self, server):

        # The server to use to complete database:
        self._server = server

        # The object database:
        self._objects = {} # Dict of oid -> TqlObject instances

        # The TTL database:
        self._ttls = {} # (obj_id, tag) -> time of death

        # The access lock of the object database:
        self._lock = RLock()

        # The list of received message while update:
        self._msgs = []

        # Total amount of query updates:
        self._nb_queries = 0

    def update(self, ids=None, tags=None, tags_novalue=None):
        '''
        Update the database according to the TTL values.

        :param ids: list of ids to update (or None for all)
        :param tags: list of tags to update (or None for all)
        :param tags_noval: list of tags to update (only the name, not the
            attached value, None for all)
        '''

        with self._lock:
            self._nb_queries += 1
            self._update(ids, tags, tags_novalue)
            self._process_responses()
            self._update_user_defined(ids)

    def _update(self, ids=None, tags=None, tags_novalue=None):
        '''
        First step of the database update process.

        This methods make essentially two things:

            * Calculate the final list of tags to get and to check
            * Send the request to each client involved in the update

        .. warning:
            The call to this method is not threadsafe and must be done only
            by :meth:`update` method.
        '''

        self._msgs = []
        
        # Cast input to the right type:
        if tags is not None:
            tags = set(tags)

        if tags_novalue is not None:
            tags_novalue = set(tags_novalue)
            if tags is not None:
                tags_novalue -= tags
        
        if ids is not None:
            ids = set(ids)
        else:
            ids = set(self._objects)

        # The main-loop, we will pop each object id from the set and
        # process it:
        while ids:
            oid = ids.pop()
            obj = self._objects.get(oid)

            assert obj is not None, 'object not found'

            if '__parent' in obj:
                # Current object is proxified by a parent:
                parent = obj['__parent']
                self._update_object(parent, tags, tags_novalue)

                # Parent is removed from the list if it present (because
                # it has been updated):
                if parent['id'] in ids:
                    ids.remove(parent['id'])

                # Search for sibling objects:
                sg = set((o for o in ids if o.startswith('%s.' % parent['id'])))
                sg &= ids
                ids -= sg
                sg.add(oid)

                for sibling in sg:
                    sibling = self.get_by_id(sibling)
                    self._update_object(sibling, tags, tags_novalue)
            else:
                # Current object is a standard node:
                self._update_object(obj, tags, tags_novalue)

    def _update_user_defined(self, ids):
        '''
        Update the user defined tags (previously fetched
        by :meth:`_update_object`).
        '''

        if ids is not None:
            ids = set(ids)
        else:
            ids = set(self._objects)

        for objid in ids:
            obj = self.get_by_id(objid)

            if '__parent' in obj:
                user_tags = obj['__parent']['__user_defined_tags']
            else:
                user_tags = obj['__user_defined_tags']

            obj.update(user_tags)
    
    def _process_responses(self):
        '''
        Process the responses received from clients.
        '''
        
        msgs = self._server.manager.wait(frozenset(self._msgs), timeout=4)
        now = datetime.now()

        for msg in msgs:
            # Update tags if received from the peer:
            if msg.get('error') is not None:
                logging.error('Error from client while getting tags')
            else:
                returned = msg['return']
                obj = msg['data']

                assert isinstance(returned, dict), 'returned tags is not a dict'
                assert isinstance(obj, TqlObject), 'data obj is not TqlObject'

                for name, attrs in returned.iteritems():
                    obj[name] = attrs.get('value')

                    # Update the ttl
                    ttl = attrs.get('ttl', DEFAULT_TTL)
                    if ttl == -1:
                        tod = datetime.max
                    else:
                        ttl += TTL_SERVER_DELTA
                        dt = timedelta(seconds=ttl)
                        tod = now + dt

                    self._ttls[(obj['id'], name)] = tod

    def _update_object(self, obj, tags, tags_novalue):
        '''
        Calculate the list of tags to request for a client, and send the
        request to the client.
        '''

        parent = obj.get('__parent')

        if parent is None:
            user_tags = self._server.conf.show(obj['id'])['tags']
            obj['__user_defined_tags'] = user_tags
            client_id = obj['id']
        else:
            user_tags = parent['__user_defined_tags']
            client_id = parent['id']

        if tags is not None:
            tags = set(tags)
            tags -= set(user_tags)

        if tags_novalue is not None:
            tags_novalue = set(tags_novalue)
            tags_novalue -= set(user_tags)

        try:
            client = self._server.get_connection(client_id)
        except KeyError:
            # The client is not connected, we need to keep only unttlised tags:
            for name in obj.keys():
                if (obj['id'], name) in self._ttls:
                    del obj[name]
        else:
            # The client is connected:
            if parent is None:
                obj.update(client.get_tags())

            # First, check if each tag is not already cached:
            now = datetime.now()
            if tags is not None:
                for tag in tags.copy():
                    if now < self._ttls.get((obj['id'], tag), now):
                        tags.remove(tag)
                # Cast as jsonizable type:
                tags = tuple(tags)

            if tags_novalue is not None:
                for tag in tags_novalue.copy():
                    if now < self._ttls.get((obj['id'], tag), now):
                        tags_novalue.remove(tag)
                # Cast as jsonizable type:
                tags_novalue = tuple(tags_novalue)

            # Ask tags to the node:
            if (tags is None or tags) or (tags_novalue is None or tags_novalue):
                self._request_tags(obj, client, tags, tags_novalue)

    def _request_tags(self, obj, client, tags, tags_novalue):
        '''
        Send the request to the client.
        '''
        
        if '__parent' in obj:
            subid = obj['id'].partition('.')[2]
            msg_id = client.connection.async_call('sub_tags', subid, tags,
                                                  tags_novalue, _data=obj)
        else:
            msg_id = client.connection.async_call('get_tags', tags,
                                                  tags_novalue, _data=obj)

        self._msgs.append(msg_id)

    def get_by_id(self, oid):
        '''
        Get an registered object by its id.

        :param oid: the complete oid of the object to get
        :return: a :class:`TqlObject` instance
        '''
        return self._objects[oid]

    def register(self, obj, cleanable=()):
        '''
        Populate the database with the specified object. All the tags specified
        in the passed object are not declared in cache and will never be
        cleaned.

        :param obj: the object to register
        :param clenable: the optionnal list of tags to declare as cleanable when
            the cache is cleaned
        '''

        with self._lock:
            if 'id' not in obj:
                raise ValueError('id tag must exists on the object')
            elif obj['id'] in self._objects:
                raise AlreadyRegistered('The object is already registered')
            else:
                self._objects[obj['id']] = obj
                for tag in cleanable:
                    self._ttls[(obj['id'], tag)] = datetime.max

    def unregister(self, obj_id):
        '''
        Remove an object from the manager by its id.
        '''

        with self._lock:
            if obj_id in self._objects:
                self.unregister_children(obj_id)
                del self._objects[obj_id]
                self.clean_ttls(obj_id)
            else:
                raise ValueError('The object is not registered')

    def unregister_children(self, obj_id):
        '''
        Remove all children of an object.
        '''

        with self._lock:
            if obj_id in self._objects:
                parent = self._objects[obj_id]

                for key, obj in self._objects.items():
                    if obj.get('__parent') is parent:
                        del self._objects[key]
                        self.clean_ttls(obj['id'])
            else:
                raise ValueError('The object %r is not registered' % obj_id)

    def clean_ttls(self, obj_id):
        '''
        Remove TTLs for all tags for the specified object.
        '''

        with self._lock:
            for oid, tag in self._ttls.keys():
                if oid == obj_id:
                    del self._ttls[(oid, tag)]

    def get_ids(self):
        '''
        Get the set of all the first level ids.
        '''

        ids = set()
        for oid in self._objects:
            parid = oid.partition('.')[0]
            ids.add(parid)

        return ids

    def all(self, tags, to_check):
        '''
        Get all the objects registered on the server with specified tags.
        '''

        self.update(tags=tags, tags_novalue=to_check)
        for obj in self._objects.values():
            yield obj

    def some(self, ids, tags=set(), to_check=set()):
        '''
        Get specified objects registed on the server with specified tags.
        '''

        self.update(ids, tags=tags, tags_novalue=to_check)
        for obj_id in ids:
            yield self._objects[obj_id]

    def stats(self):
        '''
        Get stats about the current state of the object database.
        '''

        stats = {}

        stats['count'] = len(self._objects)
        cached = dead = 0
        now = datetime.now()
        for date in self._ttls.values():
            if date > now:
                dead += 1
            else:
                cached += 1
        stats['cached'] = cached
        stats['dead'] = dead
        stats['total'] = cached + dead
        stats['queries'] = self._nb_queries

        return stats