more work

Author: jrydberg

setup.py

@@ -1,6 +1,6 @@
 from setuptools import setup, find_packages
 setup(name='txgossip',
-      version='0.0',
+      version='0.1',
       description='simple cluster communication protocol',
       author='Johan Rydberg',
       author_email='[email protected]',

txgossip/gossip.py

@@ -144,7 +144,8 @@ def _gossip(self):
             self._gossip_with_peer(random.choice(dead_peers))
 
         for state in self._states.values():
-            state.check_suspected()
+            if state.name != self.name:
+                state.check_suspected()
 
     def _gossip_with_peer(self, peer):
         """Send a gossip message to C{peer}."""
@@ -212,9 +213,12 @@ def get(self):
     def __getitem__(self, key):
         return self.state[key]
 
-    def __setitem__(self, key, value):
+    def set(self, key, value):
         self.state[key] = value
 
+    def __setitem__(self, key, value):
+        self.set(key, value)
+
     def __contains__(self, key):
         return key in self.state
 

txgossip/recipies.py

@@ -1,5 +1,4 @@
 # Copyright (C) 2011 Johan Rydberg
-# Copyright (C) 2010 Bob Potter
 #
 # Permission is hereby granted, free of charge, to any person
 # obtaining a copy of this software and associated documentation files
@@ -46,31 +45,31 @@ class LeaderElectionMixin:
     VOTE_KEY = 'leader:vote'
     LEADER_KEY = 'leader:leader'
 
-    def __init__(self, clock):
+    def __init__(self, clock, vote_delay=5):
         self.clock = clock
         self._election_timeout = None
         self._gossiper = None
+        self.is_leader = None
+        self.vote_delay = vote_delay
 
     def make_connection(self, gossiper):
         self._gossiper = gossiper
+        self.start_election()
 
     def _check_consensus(self, key):
         """Check if all peers have the same value for C{key}.
 
         Return the value if they all have the same value, otherwise
         return C{None}.
         """
-        try:
-            correct = self._gossiper[key]
-            for peer in self._gossiper.live_peers:
-                value = peer[key]
-                if value != correct:
-                    return None
-        except KeyError:
-            # One peer did not even have the key.
-            return None
-        else:
-            return correct
+        correct = self._gossiper.get(key)
+        for peer in self._gossiper.live_peers:
+            if not key in peer.keys():
+                return None
+            value = peer.get(key)
+            if value != correct:
+                return None
+        return correct
 
     def value_changed(self, peer, key, value):
         """Inform about a change of a key-value pair.
@@ -85,7 +84,7 @@ def value_changed(self, peer, key, value):
         if key == self.VOTE_KEY:
             leader = self._check_consensus(self.VOTE_KEY)
             if leader:
-                self._gossiper[self.LEADER_KEY] = leader
+                self._gossiper.set(self.LEADER_KEY, leader)
         elif key == self.LEADER_KEY:
             leader = self._check_consensus(self.LEADER_KEY)
             if leader:
@@ -102,19 +101,13 @@ def _vote(self):
 
         # Check if we're one of the peers that would like to become
         # master.
-        vote, currp = None, None
-        try:
-            curr = self._gossiper[self.PRIO_KEY]
-            if curr is not None:
-                vote = self._gossiper
-        except KeyError:
-            pass
+        vote = None
+        curr = self._gossiper.get(self.PRIO_KEY)
+        if curr is not None:
+            vote = self._gossiper
 
         for peer in self._gossiper.live_peers:
-            try:
-                prio = peer[self.PRIO_KEY]
-            except KeyError:
-                continue
+            prio = peer.get(self.PRIO_KEY)
             if prio is None:
                 # This peer did not want to become leader.
                 continue
@@ -126,15 +119,9 @@ def _vote(self):
                 vote = peer
             elif prio == curr:
                 # We need to break the tie.
-                if hash(peer) > hash(vote):
+                if hash(peer.name) > hash(vote.name):
                     vote = peer
-
-        # See if there's a need to change our vote, or if we'll stand
-        # by our last vote.
-        if self.VOTE_KEY in self._gossiper:
-            if self._gossiper[self.VOTE_KEY] == vote.name:
-                return
-        self._gossiper[self.VOTE_KEY] = vote.name
+        self._gossiper.set(self.VOTE_KEY, vote.name)
 
     def start_election(self):
         """Start an election.
@@ -155,6 +142,7 @@ def leader_elected(self, is_leader, leader):
         @param is_leader: C{True} if this peer is the leader.
         @param leader: The address of the peer that is the leader.
         """
+        self.is_leader = is_leader
 
     def peer_dead(self, peer):
         """A peer is dead."""
@@ -186,26 +174,39 @@ def __init__(self, clock, storage, ignore_keys=[]):
     def make_connection(self, gossiper):
         self._gossiper = gossiper
 
-    def value_changed(self, peer, key, timestamp_value):
-        """A peer has changed its value."""
-        if key == '__heartbeat__':
-            # We do not care about updates to the heartbeat value.
-            return
-        timestamp, value = timestamp_value
+    def persist_key_value(self, key, timestamped_value):
+        self._storage[key] = timestamped_value
+        if hasattr(self._storage, 'sync'):
+            self._storage.sync()
+
+    def replicate_key_value(self, peer, key, timestamped_value):
+        timestamp, value = timestamped_value
         if key in self._storage:
             current_timestamp, current_value = self._storage[key]
             if timestamp <= current_timestamp:
                 return
-        self._storage[key] = (timestamp, value)
+        self._gossiper.set(key, timestamped_value)
+
+    def value_changed(self, peer, key, timestamp_value):
+        """A peer has changed its value."""
+        if key == '__heartbeat__' or key in self._ignore_keys:
+            return
+        if peer.name == self._gossiper.name:
+            self.persist_key_value(key, timestamp_value)
+        else:
+            self.replicate_key_value(peer, key, timestamp_value)
+
+    def set(self, key, value):
+        self._gossiper.set(key, [self.clock.seconds(), value])
 
     def __setitem__(self, key, value):
-        self._gossiper[key] = (self.clock.seconds(), value)
+        self.set(key, value)
 
     def __getitem__(self, key):
-        return self._storage[key][1]
+        return self._gossiper.get(key)[1]
 
     def get(self, key, default=None):
-        if key in self._storage:
+        if key in self.keys():
             return self[key]
         return default
 
@@ -218,19 +219,16 @@ def keys(self, pattern=None):
             return [key for key in keys
                     if fnmatch.fnmatch(key, pattern)]
 
+    def load_from(self, storage):
+        for key in storage:
+            if key not in self._ignore_keys:
+                self._gossiper.set(key, storage[key])
+
     def __contains__(self, key):
         return key in self.keys()
 
-    def timestamp_for_key(self, key):
-        return self._storage[key][0]
-
-    def synchronize_keys_with_peer(self, peer):
-        """Synchronize keys with C{peer}.
+    def peer_dead(self, peer):
+        """A peer is dead."""
 
-        Will iterate through all the keys that C{peer} has and see if
-        there's some values that are newer than ours.
-        """
-        for key, value in peer.items():
-            if key in self._ignore_keys:
-                continue
-            self.value_changed(peer, key, value)
+    def peer_alive(self, peer):
+        """A peer is alive."""

txgossip/state.py

@@ -67,9 +67,17 @@ def __contains__(self, key):
     def __setitem__(self, key, value):
         self.update_local(key, value)
 
+    def set(self, key, value):
+        self.update_local(key, value)
+
     def __getitem__(self, key):
         return self.attrs[key][0]
 
+    def get(self, key, default=None):
+        if key in self.attrs:
+            return self.attrs[key][0]
+        return default
+
     def has_key(self, key):
         return key in self.attrs
 
@@ -82,7 +90,7 @@ def items(self):
 
     def set_key(self, k, v, n):
         self.attrs[k] = (v, n)
-        self.participant.value_changed(self.name, str(k), v)
+        self.participant.value_changed(self, str(k), v)
 
     def beat_that_heart(self):
         self.heart_beat_version += 1
@@ -111,9 +119,9 @@ def check_suspected(self):
     def mark_alive(self):
         alive, self.alive = self.alive, True
         if not alive:
-            self.participant.peer_alive(self.name)
+            self.participant.peer_alive(self)
 
     def mark_dead(self):
         if self.alive:
             self.alive = False
-            self.participant.peer_dead(self.name)
+            self.participant.peer_dead(self)

txgossip/test/__init__.py

@@ -0,0 +1,76 @@
+# Copyright (C) 2011 Johan Rydberg
+#
+# Permission is hereby granted, free of charge, to any person
+# obtaining a copy of this software and associated documentation files
+# (the "Software"), to deal in the Software without restriction,
+# including without limitation the rights to use, copy, modify, merge,
+# publish, distribute, sublicense, and/or sell copies of the Software,
+# and to permit persons to whom the Software is furnished to do so,
+# subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be
+# included in all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
+# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
+# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+from mockito import mock, when, verify
+
+from twisted.trial import unittest
+from twisted.internet import task
+
+from txgossip.recipies import KeyStoreMixin
+
+
+class KeyStoreTestCase(unittest.TestCase):
+    """Test cases for the key-value store mixin."""
+
+    def setUp(self):
+        self.clock = task.Clock()
+        self.storage = {}
+        self.keystore = KeyStoreMixin(self.clock, self.storage)
+        self.gossiper = mock()
+        self.gossiper.name = 'self'
+        self.keystore.make_connection(self.gossiper)
+
+    def test_replicate_when_remote_peer_changed_value(self):
+        self.keystore.value_changed('a', 'k', (0, 'value'))
+        verify(self.gossiper).set('k', (0, 'value'))
+        self.assertNotIn('k', self.storage)
+
+    def test_ignore_replication_when_remote_peer_has_old_value(self):
+        self.storage['k'] = (1, 'value')
+        self.keystore.value_changed('a', 'k', (0, 'value'))
+        verify(self.gossiper, times=0).set('k', (0, 'value'))
+
+    def test_persist_value_when_set_on_local_peer(self):
+        self.keystore.value_changed('self', 'k', (0, 'value'))
+        self.assertIn('k', self.storage)
+        self.assertEquals(self.storage['k'], (0, 'value'))
+
+    def test_keys_returns_all_keys_in_gossiper(self):
+        when(self.gossiper).keys().thenReturn(['a', 'b'])
+        self.assertEquals(self.keystore.keys(), ['a', 'b'])
+
+    def test_keys_can_be_filtered(self):
+        when(self.gossiper).keys().thenReturn(['ab', 'ba'])
+        self.assertEquals(self.keystore.keys('b*'), ['ba'])
+
+    def test_contains_use_keys_from_gossiper(self):
+        when(self.gossiper).keys().thenReturn(['a', 'b'])
+        self.assertIn('a', self.keystore)
+
+    def test_get_results_default_value_if_value_not_present(self):
+        when(self.gossiper).keys().thenReturn([])
+        self.assertEquals(self.keystore.get('a', '!'), '!')
+
+    def test_get_returns_value_from_gossip_state(self):
+        when(self.gossiper).keys().thenReturn(['a'])
+        when(self.gossiper).get('a').thenReturn((0, '!'))
+        self.assertEquals(self.keystore.get('a'), '!')