dpp-coordinator.py

class Coordinator:
    def __init__(self, nodes):
        self.nodes = nodes
        self.q = 1+len(nodes)/2
    def add_node(self, node, timeout):
        due = now() + timeout
        while node not in self.nodes and due > now():
            try:
                if len(self.nodes)%2==1:
                    # set_q has the same effect as filter from the DPP article
                    resps = net.send(self.nodes, lambda x: x.set_q(self.q+1), due-now())
                    resps.where(lambda x: isinstance(x.msg, OK)).wait(len(self.nodes))
                    # increase replication (if the value was accepted)
                    net.send(self.nodes, lambda x: x.get(due-now()), due-now())
                    resps.where(lambda x: isinstance(x.msg, OK)).wait(len(self.nodes))
                    # add a node
                    net.send(self.nodes, lambda x: x.add_node(node), due-now())
                    resps.where(lambda x: isinstance(x.msg, OK)).wait(len(self.nodes))
                    self.nodes.append(node)
                    self.q+=1
                else:
                    # add a node
                    net.send(self.nodes, lambda x: x.add_node(node), due-now())
                    resps.where(lambda x: isinstance(x.msg, OK)).wait(len(self.nodes))
                    self.nodes.append(node)
            except CantResolveWait:
                continue
        if node in self.nodes: return OK()
        return Error()
    def rm_node(self, node, timeout):
        due = now() + timeout
        while node in self.nodes and due > now():
            try:
                if len(self.nodes)%2==1:
                    resps = net.send(self.nodes, lambda x: x.rm_node(node), due-now())
                    resps.where(lambda x: isinstance(x.msg, OK)).wait(len(self.nodes))
                    self.nodes.rm(node)
                else:
                    resps = net.send(self.nodes, lambda x: x.rm_node(node), due-now())
                    resps.where(lambda x: isinstance(x.msg, OK)).wait(len(self.nodes))
                    resps = net.send(self.nodes, lambda x: x.set_q(self.q-1), due-now())
                    resps.where(lambda x: isinstance(x.msg, OK)).wait(len(self.nodes))
                    self.nodes.rm(node)
                    self.q-=1
            except CantResolveWait:
                continue
        if node not in self.nodes: return OK()
        return Error()