refactor into more, smaller modules. Write Small Things.

Author: djmitche

cluster/acceptor.py

@@ -0,0 +1,29 @@
+from collections import defaultdict
+from protocol import Ballot
+from member import Component
+
+
+class Acceptor(Component):
+
+    def __init__(self, member):
+        super(Acceptor, self).__init__(member)
+        self.ballot_num = Ballot(-1, -1)
+        self.accepted = defaultdict()  # { (b,s) : p }
+
+    def do_PREPARE(self, scout_id, ballot_num):  # p1a
+        if ballot_num > self.ballot_num:
+            self.ballot_num = ballot_num
+        self.send([scout_id.address], 'PROMISE',  # p1b
+                         scout_id=scout_id,
+                         acceptor=self.address,
+                         ballot_num=self.ballot_num,
+                         accepted=self.accepted)
+
+    def do_ACCEPT(self, commander_id, ballot_num, slot, proposal):  # p2a
+        if ballot_num >= self.ballot_num:
+            self.ballot_num = ballot_num
+            self.accepted[(ballot_num, slot)] = proposal
+        self.send([commander_id.address], 'ACCEPTED',  # p2b
+                         commander_id=commander_id,
+                         acceptor=self.address,
+                         ballot_num=self.ballot_num)

cluster/client.py

@@ -1,70 +1,50 @@
 import sys
-import logging
-from deterministic_network import Node
+from member import Member
+from member import Component
 
 
-class Client(Node):
+# TODO: eliminate - this library doesn't have distinct client nodes
+class Client(Member):
 
-    def __init__(self, members):
-        super(Client, self).__init__()
-        self.cid = 1000000
+    def __init__(self, node):
+        super(Client, self).__init__(node)
         self.current_request = None
 
     def start(self):
         def re_invoke(n):
             self.invoke(n, lambda output: re_invoke(n+1))
-        self.set_timer(1, lambda: re_invoke(1))
+        self.node.set_timer(1, lambda: re_invoke(1))
 
-    def invoke(self, input, callback):
+    def invoke(self, n, callback):
         assert self.current_request is None
+        def done(output):
+            self.current_request = None
+            callback(output)
+        self.current_request = Request(self, n, done)
+        self.current_request.start()
+
+
+class Request(Component):
+
+    client_ids = xrange(1000000, sys.maxint).__iter__()
+    RETRANSMIT_TIME = 0.1
+
+    def __init__(self, member, n, callback):
+        super(Request, self).__init__(member)
+        self.cid = self.client_ids.next()
+        self.n = n
         self.output = None
-        self.current_request = (self.cid, input, callback)
-        self.cid += 1
-        self.send_invoke()
-        return self.output
+        self.callback = callback
 
-    def send_invoke(self):
-        cid, input, callback = self.current_request
-        nodes = [k for k in self.core.nodes.keys() if k.startswith('Node-')]
-        self.send([self.core.rnd.choice(nodes)], 'INVOKE',
-                  caller=self.address, cid=cid, input=input)
-        self.invoke_timer = self.set_timer(3, self.send_invoke)
+    def start(self):
+        self.send([self.member.node.network.rnd.choice(self.member.node.network.nodes.keys())], 'INVOKE',
+                  caller=self.address, cid=self.cid, input=self.n)
+        self.invoke_timer = self.set_timer(self.RETRANSMIT_TIME, self.start)
 
     def do_INVOKED(self, cid, output):
-        if not self.current_request or cid != self.current_request[0]:
+        if cid != self.cid:
             return
         self.logger.debug("received output %r" % (output,))
-        callback = self.current_request[2]
-        self.current_request = None
         self.cancel_timer(self.invoke_timer)
-        callback(output)
-
-if __name__ == "__main__":
-    logging.basicConfig(
-        format="%(asctime)s %(name)s %(message)s", level=logging.DEBUG)
-    client = Client(sys.argv[1])
-    print client.invoke(4)
-    print client.invoke(1)
-
-# tests
-
-import unittest
-import threading
-
-
-class FakeMember(Node):
-
-    def do_INVOKE(self, caller, cid, input):
-        self.send([caller], 'INVOKED', cid=cid, output=input * 10)
+        self.callback(output)
         self.stop()
-
-
-class ClientTests(unittest.TestCase):
-
-    def test_invoke(self):
-        member = FakeMember()
-        client = Client(member.address)
-        memberthd = threading.Thread(target=member.run)
-        memberthd.start()
-        self.assertEqual(client.invoke(5), 50)
-        memberthd.join()

cluster/commander.py

@@ -0,0 +1,41 @@
+from protocol import CommanderId
+from member import Component
+
+
+class Commander(Component):
+
+    def __init__(self, member, leader, ballot_num, slot, proposal):
+        super(Commander, self).__init__(member)
+        self.leader = leader
+        self.ballot_num = ballot_num
+        self.slot = slot
+        self.proposal = proposal
+        self.commander_id = CommanderId(self.address, slot, proposal)
+        self.accepted = set([])
+        self.peers = member.peers  # TODO: pass this in
+        self.quorum = len(self.peers) / 2 + 1
+
+    def start(self):
+        self.send(self.peers, 'ACCEPT',  # p2a
+                         commander_id=self.commander_id,
+                         ballot_num=self.ballot_num,
+                         slot=self.slot,
+                         proposal=self.proposal)
+
+    def finished(self, ballot_num, preempted):
+        self.leader.commander_finished(self.commander_id, ballot_num, preempted)
+        self.stop()
+
+    def do_ACCEPTED(self, commander_id, acceptor, ballot_num):  # p2b
+        if commander_id != self.commander_id:
+            return
+        if ballot_num == self.ballot_num:
+            self.accepted.add(acceptor)
+            if len(self.accepted) < self.quorum:
+                return
+            self.send(self.peers, 'DECISION',
+                                slot=self.slot,
+                                proposal=self.proposal)
+            self.finished(ballot_num, False)
+        else:
+            self.finished(ballot_num, True)

cluster/deterministic_network.py

@@ -1,4 +1,3 @@
-import uuid
 import time
 import logging
 import heapq
@@ -8,38 +7,48 @@ class Node(object):
 
     unique_ids = xrange(1000).__iter__()
 
-    def __init__(self):
+    def __init__(self, network):
+        self.network = network
         self.unique_id = self.unique_ids.next()
-
-    def set_up_node(self, address, core):
-        self.core = core
-        self.address = address
-        self.core.nodes[self.address] = self
-        self.logger = logging.getLogger('node.%s' % (self.address,))
+        self.address = 'N%d' % self.unique_id
+        self.components = []
+        self.logger = logging.getLogger(self.address)
         self.logger.info('starting')
 
-    def stop(self):
-        self.logger.error('STOPPING')
-        if self.address in self.core.nodes:
-            del self.core.nodes[self.address]
-
-    def start(self):
-        pass
+    def kill(self):
+        self.logger.error('node dying')
+        if self.address in self.network.nodes:
+            del self.network.nodes[self.address]
 
     def set_timer(self, seconds, callable):
-        # TODO: refactor so this won't call a stopped node
-        return self.core.set_timer(seconds, self.address, callable)
+        return self.network.set_timer(seconds, self.address, callable)
 
     def cancel_timer(self, timer):
-        self.core.cancel_timer(timer)
+        self.network.cancel_timer(timer)
 
     def send(self, destinations, action, **kwargs):
         self.logger.debug("sending %s with args %s to %s" %
                           (action, kwargs, destinations))
-        self.core.send(destinations, action, **kwargs)
+        self.network.send(destinations, action, **kwargs)
+
+    def register(self, component):
+        self.components.append(component)
+
+    def unregister(self, component):
+        self.components.remove(component)
 
+    def receive(self, action, kwargs):
+        import sys
+        for comp in self.components[:]:
+            try:
+                fn = getattr(comp, 'do_%s' % action)
+            except AttributeError:
+                continue
+            comp.logger.debug("received %r with args %r" % (action, kwargs))
+            fn(**kwargs)
 
-class Core(object):
+
+class Network(object):
 
     PROP_DELAY = 0.03
     PROP_JITTER = 0.02
@@ -50,11 +59,14 @@ def __init__(self, seed, pause=False):
         self.pause = pause
         self.timers = []
         self.now = 1000.0
-        self.logger = logging.getLogger('core')
+        self.logger = logging.getLogger('network')
+
+    def new_node(self):
+        node = Node(self)
+        self.nodes[node.address] = node
+        return node
 
     def run(self):
-        for node in sorted(self.nodes.values()):
-            node.start()
         while self.timers:
             next_timer = self.timers[0][0]
             if next_timer > self.now:
@@ -71,6 +83,7 @@ def stop(self):
         self.timers = []
 
     def set_timer(self, seconds, address, callable):
+        # TODO: return an obj with 'cancel'
         timer = [self.now + seconds, True, address, callable]
         heapq.heappush(self.timers, timer)
         return timer
@@ -83,13 +96,7 @@ def _receive(self, address, action, kwargs):
             node = self.nodes[address]
         except KeyError:
             return
-        try:
-            fn = getattr(node, 'do_%s' % action)
-        except AttributeError:
-            return
-
-        node.logger.debug("received %r with args %r" % (action, kwargs))
-        fn(**kwargs)
+        node.receive(action, kwargs)
 
     def send(self, destinations, action, **kwargs):
         for dest in destinations:

cluster/leader.py

@@ -0,0 +1,99 @@
+from util import defaultlist, view_primary
+from protocol import Ballot
+from member import Component
+from scout import Scout
+from commander import Commander
+
+class Leader(Component):
+
+    HEARTBEAT_INTERVAL = 1
+
+    def __init__(self, member, unique_id, commander_cls=Commander, scout_cls=Scout):
+        super(Leader, self).__init__(member)
+        self.ballot_num = Ballot(0, unique_id)
+        self.active = False
+        self.proposals = defaultlist()
+        self.commander_cls = commander_cls
+        self.commanders = {}
+        self.scout_cls = scout_cls
+        self.scout = None
+
+    def view_change(self, viewchange):
+        is_primary = view_primary(viewchange.viewid, viewchange.peers) == self.address
+        self.is_primary = is_primary
+        if is_primary:
+            if not self.scout and not self.active:
+                self.spawn_scout()
+        else:
+            if self.scout:
+                self.scout.finished(False, None)
+                # .. which eventually calls self.preempted
+            elif self.active:
+                self.preempted(None)
+
+    def spawn_scout(self):
+        assert not self.scout
+        sct = self.scout = self.scout_cls(self.member, self, self.ballot_num)
+        sct.start()
+
+    def scout_finished(self, adopted, ballot_num, pvals):
+        self.scout = None
+        if adopted:
+            # pvals is a defaultlist of (slot, proposal) by ballot num; we need the
+            # highest ballot number for each slot.  TODO: this is super
+            # inefficient!
+            last_by_slot = defaultlist()
+            for b, s in reversed(sorted(pvals.keys())):
+                p = pvals[b, s]
+                if last_by_slot[s] is None:
+                    last_by_slot[s] = p
+            for s, p in enumerate(last_by_slot):
+                if p is not None:
+                    self.proposals[s] = p
+            for s, p in enumerate(self.proposals):
+                if p is not None:
+                    self.spawn_commander(ballot_num, s, p)
+            self.logger.info("leader becoming active")
+            self.active = True
+        else:
+            self.preempted(ballot_num)
+
+    def preempted(self, ballot_num):
+        # ballot_num is None when we are preempted by a view change
+        if ballot_num:
+            self.logger.info("leader preempted by %s" % (ballot_num.leader,))
+        else:
+            self.logger.info("leader preempted by view change")
+        self.active = False
+        self.ballot_num = Ballot(
+            (ballot_num if ballot_num else self.ballot_num).n + 1, self.unique_id)
+        # if we're the primary for this view, re-scout immediately
+        if not self.scout and self.is_primary:
+            self.logger.info("re-scouting as the primary for this view")
+            self.spawn_scout()
+
+    def spawn_commander(self, ballot_num, slot, proposal):
+        cmd = self.commander_cls(self.member, self, ballot_num, slot, proposal)
+        if cmd.commander_id in self.commanders:
+            return
+        print "set", cmd.commander_id
+        self.commanders[cmd.commander_id] = cmd
+        cmd.start()
+
+    def commander_finished(self, commander_id, ballot_num, preempted):
+        print "del", commander_id
+        del self.commanders[commander_id]
+        if preempted:
+            self.preempted(ballot_num)
+
+    def do_PROPOSE(self, slot, proposal):
+        if self.proposals[slot] is None:
+            if self.active:
+                self.proposals[slot] = proposal
+                self.spawn_commander(self.ballot_num, slot, proposal)
+            else:
+                if not self.scout:
+                    self.logger.warning(
+                        "got PROPOSE when not active - scouting")
+                    self.spawn_scout()
+