rewrite to use MultiPaxos

* add deterministic networking
* add run.py to run three cluster members and a client

This is still incredibly inefficient:
* leaders battle it out for leadership on every invocation
* message sizes are unbounded
* the size of local state is unbounded

Author: djmitche

cluster/README.txt

@@ -7,6 +7,8 @@ that can also feed clients information on cluster membership.
 
 * statemachine.py -- illustrates the state machine client API
 * network.py -- handles network communication
+* deterministic_network.py -- similar, but deterministic
 * client.py -- a simple client
 * member_single.py -- a non-clustered member server, talking to the client
 * member_replicated.py -- a replicated state machine with consensus and a fixed membership
+  This is based on "Paxos Made Moderately Complex" (van Renesse, 2011)

cluster/client.py

@@ -1,25 +1,46 @@
+import time
 import sys
 import logging
-from network import Node
+from deterministic_network import Node
 
 
 class Client(Node):
 
     def __init__(self, member_address):
         super(Client, self).__init__()
         self.member_address = member_address
+        self.cid = 1000000
+        self.current_request = None
 
-    def invoke(self, input):
+    def start(self):
+        def re_invoke(n):
+            if n < 10:
+                self.invoke(n, lambda output: re_invoke(n+1))
+        self.set_timer(1, lambda: re_invoke(1))
+
+    def invoke(self, input, callback):
+        assert self.current_request is None
         self.output = None
-        self.send([self.member_address], 'INVOKE',
-                  input=input, caller=self.address)
-        self.run()
+        self.current_request = (self.cid, input, callback)
+        self.cid += 1
+        self.send_invoke()
         return self.output
 
-    def do_INVOKED(self, output):
+    def send_invoke(self):
+        cid, input, callback = self.current_request
+        self.send([self.member_address], 'INVOKE',
+                  caller=self.address, cid=cid, input=input)
+        self.invoke_timer = self.set_timer(3, self.send_invoke)
+
+    def do_INVOKED(self, cid, output):
+        if not self.current_request or cid != self.current_request[0]:
+            return
+        print "GOT", cid, output
         self.logger.debug("received output %r" % (output,))
-        self.output = output
-        self.stop()
+        callback = self.current_request[2]
+        self.current_request = None
+        self.cancel_timer(self.invoke_timer)
+        callback(output)
 
 if __name__ == "__main__":
     logging.basicConfig(
@@ -36,8 +57,8 @@ def do_INVOKED(self, output):
 
 class FakeMember(Node):
 
-    def do_INVOKE(self, caller, input):
-        self.send([caller], 'INVOKED', output=input * 10)
+    def do_INVOKE(self, caller, cid, input):
+        self.send([caller], 'INVOKED', cid=cid, output=input * 10)
         self.stop()
 
 

cluster/deterministic_network.py

@@ -0,0 +1,77 @@
+import uuid
+import logging
+import heapq
+import random
+
+class Node(object):
+
+    unique_ids = xrange(1000).__iter__()
+
+    def __init__(self):
+        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,))
+
+    def start(self):
+        pass
+
+    def set_timer(self, seconds, callable):
+        return self.core.set_timer(seconds, callable)
+
+    def cancel_timer(self, timer):
+        self.core.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)
+
+    def receive(self, action, kwargs):
+        self.logger.debug("received %r with args %r" % (action, kwargs))
+        getattr(self, 'do_%s' % action)(**kwargs)
+
+
+class Core(object):
+
+    PROP_DELAY = 0.03
+    PROP_JITTER = 0.02
+
+    def __init__(self, seed):
+        self.nodes = {}
+        self.rnd = random.Random(seed)
+        self.timers = []
+        self.now = 0.0
+        self.logger = logging.getLogger('core')
+
+    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:
+                self.now = next_timer
+            when, do, callable = heapq.heappop(self.timers)
+            if do:
+                callable()
+
+    def stop(self):
+        self.timers = []
+
+    def set_timer(self, seconds, callable):
+        timer = [self.now + seconds, True, callable]
+        heapq.heappush(self.timers, timer)
+        return timer
+
+    def cancel_timer(self, timer):
+        timer[1] = False
+
+    def send(self, destinations, action, **kwargs):
+        for dest in destinations:
+            node = self.nodes[dest]
+            delay = self.PROP_DELAY + self.rnd.uniform(-self.PROP_JITTER, self.PROP_JITTER)
+            self.set_timer(delay, lambda node=node: node.receive(action, kwargs))
+