switch to inter-component events for better isolation

Author: djmitche

cluster/commander.py

@@ -4,16 +4,16 @@
 
 class Commander(Component):
 
-    def __init__(self, member, leader, ballot_num, slot, proposal):
+    def __init__(self, member, leader, ballot_num, slot, proposal, peers):
         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
+        self.peers = peers
+        self.quorum = len(peers) / 2 + 1
 
     def start(self):
         self.send(self.peers, 'ACCEPT',  # p2a

cluster/leader.py

@@ -17,8 +17,10 @@ def __init__(self, member, unique_id, commander_cls=Commander, scout_cls=Scout):
         self.commanders = {}
         self.scout_cls = scout_cls
         self.scout = None
+        self.peers = None
 
-    def view_change(self, viewchange):
+    def on_view_change_event(self, viewchange):
+        self.peers = viewchange.peers
         is_primary = view_primary(viewchange.viewid, viewchange.peers) == self.address
         self.is_primary = is_primary
         if is_primary:
@@ -33,7 +35,7 @@ def view_change(self, viewchange):
 
     def spawn_scout(self):
         assert not self.scout
-        sct = self.scout = self.scout_cls(self.member, self, self.ballot_num)
+        sct = self.scout = self.scout_cls(self.member, self, self.ballot_num, self.peers)
         sct.start()
 
     def scout_finished(self, adopted, ballot_num, pvals):
@@ -73,7 +75,7 @@ def preempted(self, ballot_num):
             self.spawn_scout()
 
     def spawn_commander(self, ballot_num, slot, proposal):
-        cmd = self.commander_cls(self.member, self, ballot_num, slot, proposal)
+        cmd = self.commander_cls(self.member, self, ballot_num, slot, proposal, self.peers)
         if cmd.commander_id in self.commanders:
             return
         print "set", cmd.commander_id

cluster/member.py

@@ -6,16 +6,34 @@ class Member(object):  # TODO: rename
     def __init__(self, node):
         self.node = node
         self.address = self.node.address
+        self.components = []
+
+    def register(self, component):
+        self.components.append(component)
+        self.node.register(component)
+
+    def unregister(self, component):
+        self.components.remove(component)
+        self.node.unregister(component)
+
+    def event(self, message, **kwargs):
+        method = 'on_' + message + '_event'
+        for comp in self.components:
+            if hasattr(comp, method):
+                getattr(comp, method)(**kwargs)
 
 
 class Component(object):  # TODO: rename
 
     def __init__(self, member):
         self.member = member
-        self.member.node.register(self)
+        self.member.register(self)
         self.address = member.address
         self.logger = logging.getLogger("%s.%s" % (self.address, self.__class__.__name__))
 
+    def event(self, message, **kwargs):
+        self.member.event(message, **kwargs)
+
     def send(self, destinations, action, **kwargs):
         self.member.node.send(destinations, action, **kwargs)
 
@@ -27,4 +45,4 @@ def cancel_timer(self, timer):
         self.member.node.cancel_timer(timer)
 
     def stop(self):
-        self.member.node.unregister(self)
+        self.member.unregister(self)

cluster/member_replicated.py

@@ -21,10 +21,6 @@ def __init__(self, node, execute_fn, peers,
     def start(self):
         self.replica.start()
 
-    def view_change(self, viewchange):
-        self.peers = viewchange.peers
-        self.leader.view_change(viewchange)
-
 
 class ClusterSeed(Member):
 

cluster/replica.py

@@ -22,14 +22,15 @@ def __init__(self, member, execute_fn, peers):
         self.proposals = defaultlist()
         self.decisions = defaultlist()
         self.viewid = 0
+        self.peers = peers
         self.last_heard_from = {}
         self.lost_peer_proposal = None
 
     def start(self):
         "Try to join the cluster"
         if not self.ready:
             # TODO: do something more deterministic
-            self.send([self.member.node.network.rnd.choice(self.member.peers)], 'JOIN',
+            self.send([self.member.node.network.rnd.choice(self.peers)], 'JOIN',
                              requester=self.address)
             self.set_timer(self.JOIN_RETRANSMIT, self.start)
 
@@ -49,7 +50,7 @@ def invoke(self, proposal):
                 self.viewid = viewchange.viewid
                 self.send(
                     list(set(viewchange.peers) -
-                         set(self.member.peers)), 'WELCOME',
+                         set(self.peers)), 'WELCOME',
                     state=self.state,
                     slot_num=self.slot_num,
                     decisions=self.decisions,
@@ -58,7 +59,7 @@ def invoke(self, proposal):
                 if self.address not in viewchange.peers:
                     self.stop()
                     return
-                self.member.view_change(viewchange)
+                self.event('view_change', viewchange=viewchange)
             else:
                 self.logger.info(
                     "ignored out-of-sequence view change operation")
@@ -69,8 +70,8 @@ def propose(self, proposal, slot=None):
             self.next_slot += 1
         self.proposals[slot] = proposal
         # find a leader we think is working, deterministically
-        leaders = [view_primary(self.viewid, self.member.peers)] + \
-            list(self.member.peers)
+        leaders = [view_primary(self.viewid, self.peers)] + \
+            list(self.peers)
         for leader in leaders:
             # TODO: better way to get current time
             if self.last_heard_from.get(leader, 0) > self.member.node.network.now - 2 * self.HEARTBEAT_INTERVAL:
@@ -91,8 +92,8 @@ def repropose(self):
 
     def heartbeat(self):
         "send and monitor heartbeats"
-        self.send(self.member.peers, 'HEARTBEAT', sender=self.address)
-        for peer in self.member.peers:
+        self.send(self.peers, 'HEARTBEAT', sender=self.address)
+        for peer in self.peers:
             if peer == self.address or peer not in self.last_heard_from:
                 continue
             if self.last_heard_from[peer] < self.member.node.network.now - 2 * self.HEARTBEAT_INTERVAL:
@@ -108,7 +109,7 @@ def lost_peer(self, peer):
         self.lost_peer_proposal = Proposal(None, None,
                                            ViewChange(
                                                self.viewid + 1,
-                                               tuple(sorted(set(self.member.peers) - set([peer])))))
+                                               tuple(sorted(set(self.peers) - set([peer])))))
         self.propose(self.lost_peer_proposal)
 
     def do_HEARTBEAT(self, sender):
@@ -157,10 +158,10 @@ def do_DECISION(self, slot, proposal):
                                  cid=decided_proposal.cid, output=output)
 
     def do_JOIN(self, requester):
-        if self.ready and requester not in self.member.peers:
+        if self.ready and requester not in self.peers:
             self.last_heard_from[requester] = self.member.node.network.now
             viewchange = ViewChange(self.viewid + 1,
-                                    tuple(sorted(set(self.member.peers) | set([requester]))))
+                                    tuple(sorted(set(self.peers) | set([requester]))))
             self.propose(Proposal(None, None, viewchange))
 
     def do_WELCOME(self, state, slot_num, decisions, viewid, peers):
@@ -172,6 +173,9 @@ def do_WELCOME(self, state, slot_num, decisions, viewid, peers):
             self.decisions = defaultlist(decisions)
             self.viewid = viewid
             self.last_heard_from = {}
-            self.member.view_change(ViewChange(viewid, peers))  # TODO: WELCOME should include a ViewChange
+            self.event('view_change', viewchange=ViewChange(viewid, peers))  # TODO: WELCOME should include a ViewChange
             self.heartbeat()
             self.repropose()
+
+    def on_view_change_event(self, viewchange):
+        self.peers = viewchange.peers

cluster/scout.py

@@ -6,22 +6,23 @@ class Scout(Component):
 
     PREPARE_RETRANSMIT = 1
 
-    def __init__(self, member, leader, ballot_num):
+    def __init__(self, member, leader, ballot_num, peers):
         super(Scout, self).__init__(member)
         self.leader = leader
         self.scout_id = ScoutId(self.address, ballot_num)
         self.ballot_num = ballot_num
         self.pvals = defaultdict()
         self.accepted = set([])
-        self.quorum = len(member.peers) / 2 + 1
+        self.peers = peers
+        self.quorum = len(peers) / 2 + 1
         self.retransmit_timer = None
 
     def start(self):
         self.logger.info("scout starting")
         self.send_prepare()
 
     def send_prepare(self):
-        self.send(self.member.peers, 'PREPARE',  # p1a
+        self.send(self.peers, 'PREPARE',  # p1a
                        scout_id=self.scout_id,
                        ballot_num=self.ballot_num)
         self.retransmit_timer = self.set_timer(