path: root/sharedstate.git/sharedstate/sharedobject.py

# sharedobject.py, classes to aid activities in sharing a state
# Reinier Heeres, reinier@heeres.eu
# Miguel Alvarez, miguel@laptop.org
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
#
# Change log:
#   2007-07-14: rwh, the big merge. Old function are kept with _nc
#               (non-contiguous) appended
#   2007-07-07: miguel, conflict resolution added
#   2007-06-21: rwh, first version

import copy
import pickle
import base64
import difflib
import time

import logging
_logger = logging.getLogger('sharinghelper')

class DiffRec:
    def __init__(self, versionid, sender, incremental, objstr):
        self.version_id = versionid
        self.sender = sender
        self.incremental = incremental
        self.obj = objstr

    def is_newer(self, v2):
        if self.version_id > v2.version_id or \
            (self.version_id == v2.version_id and self.sender > v2.sender):
            return True
        else:
            return False

    def __str__(self):
        return "[%d]%s" % (self.version_id, str(self.obj))

    def __repr__(self):
        return self.__str__()

class SharedObject:
    """Base class for shared objects, able to share python objects the
    dumb way"""

    def __init__(self, name, helper, opt = {}):
        self._name = name
        self._options = opt
        self._helper = helper
        self._value = None
        self._version_id = 0
        self._received_diffs = []
        self._inverse_diffs = []
        self._pending_diffs = {}
        self._cached_versions = 8
        self._locked = False
        self._locked_by = None
        self._locked_time = 0

    def encode(self, obj):
        return base64.b64encode(pickle.dumps(obj))

    def decode(self, obj):
        return pickle.loads(base64.b64decode(obj))

    def _should_encode_incremental(self, incremental):
        """Decide whether to encode an object incrementally. If nothing is
        specified (incremental=None) determine from self._options, else
        return what is specified."""

        if incremental is None:
            if 'incremental' in self._options and self._options['incremental'] is True:
                return True
            else:
                return False
        return incremental

    def changed_nc(self, diffobj, incremental):
        """This function should be called when the object has changed."""

        self._version_id += 1
        try:
            diff = DiffRec(self._version_id, self._helper._own_bus_name, incremental, diffobj)
            self.insert_diff(diff)
            enc = self.encode(diff.obj)
            if self._helper.tube_connected():
                self._helper.SendObject(self._name, self._version_id, incremental, enc)
        except Exception, inst:
            _logger.error('changed(): %s', inst)

        self.do_changed_callback()

    def changed(self, diffobj, incremental=False):
        """This function should be called when the object has change
        If diffobj is None the whole object will be sent
        If defined, only this difference will be sent
        A diff entry is also added to our received_diffs so that
        we can properly sequence and undo/redo this change.
        """

        try:
            self.insert_diff(DiffRec(self._version_id+1,
            self._helper._own_bus_name, incremental, diffobj))
            if not incremental:
                enc = self.encode(self._value)
            else:
                enc = self.encode(diffobj)
            self._helper.SendObject(self._name, self._version_id, incremental, enc)
            _logger.debug("Modification sent")
        except Exception, inst:
            _logger.error('changed(): %s, currval: %s', inst, self._value)
        self.do_changed_callback()

    def set_value_nc(self, v, incremental=None):
        """Function to set value of this object. Specifying incremental
        (a boolean) allows forcing of either incremental/full encoding.
        If not specified, the default will be used from self._options"""

        incremental = self._should_encode_incremental(incremental)
        _logger.debug('Setting value of %s to %s, incremental=%s', self._name, v, incremental)
        if incremental:
            old = copy.deepcopy(self._value)
            self._value = v
            d = self.diff(v, old)
            _logger.debug('set_value(): generated diff:\n%r', d)
            del old
            if d is not None:
                self.changed(d, True)
        else:
            self._value = v
            self.changed_nc(v, False)

    def set_value(self, v, incremental=False):
        incremental = self._should_encode_incremental(incremental)
        _logger.debug('Setting value of %s to %s, incremental=%s', self._name, v, incremental)
        old = copy.deepcopy(self._value)
        d = self.diff(v, old)
        _logger.debug("Diff= %s", d)
        self.changed(d, incremental)

    def get_value(self):
        return self._value

    def do_changed_callback(self):
        if 'changed' in self._options:
            self._options['changed'](self._value)

    def output_diff_stack(self):
        _logger.debug('Diff stack:')
        for versionid, incremental, objstr, sender in self._received_diffs:
            _logger.debug('\tv:%d, inc:%d, sender:%d', versionid,
                incremental, sender)

    def get_version(self, versionid):
        if versionid == self._version_id:
            return self._value
        if versionid > self._version_id or versionid < 0:
            return None
        target_index = self._get_diff_index(versionid + 1) #versionid of a diff== the version it leads _up_ to
        if 0 > target_index or index > len(self._received_diffs):
            return None
        object = self._value
        for i in range(len(self._received_diffs)-1, target_index -1, -1):
            d = self._received_diffs[i]
            object = self._apply_diff_to(object, self.inverse_diff(diff))
        return object

    def get_version(self, versionid):
        if versionid == self._version_id:
            return self._value
        if versionid > self._version_id or versionid < 0:
            return None
        if versionid < self._received_diffs[0].version_id:
            return None

        i = len(self._received_diffs) - 1
        while self._version_id > versionid:
            obj = self._apply_diff_to(obj, self.inverse_diff(diff))
            i -= 1

        return obj

    def inverse_diff(self, diff):
        """Invert a diff object, so that if was the result of an o -> n comparison, the
        result is associated to a n -> o comparison."""
        index = self._get_diff_index(diff)
        if 0 < index < len(self._inverse_diffs): #found (?)
            return self._inverse_diffs[index]
        return None

    def _get_diff_index(self, vi):
        for i in xrange(len(self._received_diffs)-1, -1 , -1):
            if self._received_diffs[i].version_id == vi:
                return i
        return -1

    def insert_diff_nc(self, d):
        if len(self._received_diffs) > 0 and self._received_diffs[0].is_newer(d):
            return -1

        if len(self._received_diffs) >= self._cached_versions:
            del(self._received_diffs[0])

        i = len(self._received_diffs) - 1
        while i > -1:
            if d.is_newer(self._received_diffs[i]):
                break
            i -= 1

        self._received_diffs.insert(i + 1, d)
        self._inverse_diffs.insert(i + 1, DiffRec(d.version_id, d.sender, d.incremental, None))

        return i + 1

    def insert_diff(self, recv_diff, old=None):
        """ Places a new and compatible change of incremental type in its correct place, and
        actualizes thevalue of the shared object accordingly."""
        _logger.debug("insert_diff(): Current versionid:%s, recv_diff:%s", self._version_id, recv_diff)
        if not recv_diff.incremental:
            _logger.debug("insert_diff(): Applying non-incremental diff")
        if recv_diff in self._received_diffs:
            _logger.debug("Dupe")
            return True
        recvi = recv_diff.version_id
        index = self._get_diff_index(recvi)
        if old == None:
            old = self.get_version(recvi - 1)
            #if old == None: #can't find the father, doesn't use
            #    self._discarded_diff = recv_diff
            #    #TODO: signal that we have a discarded diff --> return False
            #    return False

        _logger.debug("insert_diff(): old=%s", old)
        if index < 0:
            diffs = []
        else:
                diffs = self._received_diffs[index:]
        i = 0
        while len(diffs) > 0 and recv_diff.is_newer(diffs[0]):
            diffs = diffs[1:]
            index += 1
            i += 1
        recv_diff.version_id += i
        for d in diffs:
            d.version_id += 1
        res = [recv_diff] + diffs

        ret = old
        invdiffs = []
        for d in res:
            _logger.debug("[insert_diff] Applying %s to %s", d.obj, ret)
            ret, id = self._apply_diff_to(ret, d.obj)
            invdiffs.append(id)
        self._value = ret
        if index >= 0:
            self._received_diffs = self._received_diffs[:index] + res
            self._inverse_diffs = self._inverse_diffs[:index] + invdiffs
            _logger.info("insert_diff(): recv_diff inserted at %d", index)
        else:
            self._received_diffs = self._received_diffs + res
            self._inverse_diffs = self._inverse_diffs + invdiffs
            _logger.debug("insert_diff(): recv_diff inserted at %d", len(self._received_diffs))
        self._version_id += 1
        if not self._version_id == self._received_diffs[-1].version_id:
            _logger.debug("Version inconsistency: expected %s, got %s", self._received_diffs[-1], self._version_id)
            self._version_id = self._received_diffs[-1].version_id
        _logger.debug("new version_id: %s, actualized received_diffs: %s", self._version_id, self._received_diffs)
        return True

    def process_update_nc(self, versionid, incremental, objstr, sender, force=False):
        """Process an update:
            -Undo all newer diffs
            -Apply the just added one
            -Redo all newer diffs
        """

        obj = self.decode(objstr)
        d = DiffRec(versionid, sender, incremental, obj)

        i = self.insert_diff(d)
        _logger.debug('Inserted diff at position %d', i)
        if i == -1:
            return False

        j = len(self._received_diffs) - 1        # Don't include ourselve
        while j > i:
            if not self._received_diffs[j].incremental:
                break        # not necessary to continue beyond replacement object
            self.apply_diff(self._inverse_diffs[j].obj)
            j -= 1

        while j < len(self._received_diffs):
            if not self._received_diffs[j].incremental:
                self._value = self._received_diffs[j].obj
            else:
                self._inverse_diffs[j].obj = self.apply_diff(self._received_diffs[j].obj)
            self._version_id = self._received_diffs[j].version_id
            j += 1

        self.do_changed_callback()

        return True

    def process_update(self, versionid, incremental, objstr, sender, force=False):
        """Process an update:
            -Undo all newer diffs to get to the common version
            - Check for compatability between the received update and the 'combined' newer diffs
            - If compatible, insert the received one
            - If not, apply the 'winning' one, and pass the other to the 'rejected' folder
            We return 'True' if there is no conflict, and 'False' if there was one
        """
        if  (versionid == self._version_id):
            _logger.debug("Maybe dupe? our versionid:%d, received:%d", self._version_id, versionid)
            a = len(self._received_diffs) > 0
            b = (sender == self._received_diffs[-1].sender)
            _logger.debug("Two tests: %s, %s", a, b)
            if a and b:
                return True #Dupe
        obj = self.decode(objstr)
        _logger.debug( "process_update: version: %s, obj: %s" % (versionid, obj))
        if versionid > self._version_id + 1 and incremental and not force:
            #Disordered diffs. Store for later use
            self._pending_diffs[versionid] = DiffRec(versionid, sender, incremental, obj)
            for i in range(self._version_id+1, versionid):
                #TODO : Call for missing diffs:
                pass
            _logger.debug("Disordered diffs. returning")
            return True
        old = self.get_version(versionid - 1) #supposed common ancestor
        if not incremental: #we get the incremental version maually
            obj2 = self.diff(obj, old)
        else:
            obj2 = obj
        db = DiffRec(versionid, sender, True, obj2)
        if versionid > self._version_id or force: #expected version number
            if not incremental:
                _logger.debug("Update in expected range, non incremental")
                self._value = obj
                self._received_diffs.append(db)
                self._version_id = versionid
            else:
                _logger.debug("Update in expected range, incremental")
                self.insert_diff(db, old)
                while self._version_id + 1 in self._pending_diffs:
                    #We get the pending diff and apply it
                    pd = self._pending_diffs[self._version_id + 1]
                    if not pd.incremental:
                        self._value = pd.obj
                        pd.obj = self.diff(pd.obj, self._value)
                        pd.incremental = True
                    else:
                        self.insert_diff(pd, self._value)
                    self._received_diffs.append(pd)
                    self._version_id += 1
                    del self._pending_diffs[self._version_id +1]
            _logger.debug("Updated value: %s", self._value)
            self.do_changed_callback()
            return True
        else:
            _logger.debug("Conflicting versionids: mine:%d, his:%d" %(self._version_id, versionid))
            obj_da = self.diff(self._value, old)
            if len(self._received_diffs)>0:
                sender_a = self._received_diffs[-1].sender
            else:
                sender_a = None
            da = DiffRec(self._version_id, sender_a, True, obj_da)
            _logger.debug("Verifying compatibility")
            if self._compatible_diffs(da.obj, db.obj):
                _logger.debug("compatible diff")
                if self.insert_diff(db, old):
                    #insert does take care of version actualization and updates self._version_id
                    _logger.debug("changed without error")
                    self.do_changed_callback()
                    return True
                else:
                    _logger.debug("Error with insertion")
                    return False

    def _compatible_diffs(self, da, db):
        return True

    def diff(self, new, old):
        return None

    def apply_diff_to(self, obj, diffobj):
        return (obj, diffobj)

    def apply_diff(self, diffobj):
        (newobj, idiff) = self.apply_diff_to(self._value, diffobj)
        self._value = newobj
        return idiff

    def is_locked(self):
        return self._locked

    def lock(self):
        if not self._locked:
            self._locked = True
            self._locked_by = "me"
            self._locked_time = time.time()
            self._helper.LockObject(self._name, self._locked_time)

    def receive_lock(self, sender, when):
        if not self._locked or \
            (self._locked and self._locked_time > when):
            if self._locked and 'locklost' in self._options:
                self._options['locklost']()
            self._locked = True
            self._locked_by = sender
            self._locked_time = when
            if 'locked' in self._options:
                self._options['locked'](sender)

    def unlock(self):
        if self._locked and self._locked_by is 'me':
            self._helper.UnlockObject(self._name)

    def receive_unlock(self, sender):
        if self._locked:
            self._locked = False
            self._locked_by = None
            self._locked_time = 0
            if 'unlocked' in self._options:
                self._options['unlocked'](sender)