def execute_operation(state, operation):
  if operation.name == 'deposit':
if not verify_signature(operation.deposit_signature):
return state, False
state.accounts[operation.destination_account] += operation.amount
return state, True
  elif operation.name == 'transfer':
if state.accounts[operation.source_account] < operation.amount:
 return state, False
 state.accounts[operation.source_account] -= operation.amount
state.accounts[operation.destination_account] += operation.amount
return state, True
  elif operation.name == 'get-balance':
  return state, state.accounts[operation.account]

Accepted = namedtuple('Accepted', ['slot', 'ballot_num'])
 Accept = namedtuple('Accept', ['slot', 'ballot_num', 'proposal'])
 Decision = namedtuple('Decision', ['slot', 'proposal'])
 Invoked = namedtuple('Invoked', ['client_id', 'output'])
 Invoke = namedtuple('Invoke', ['caller', 'client_id', 'input_value'])
 Join = namedtuple('Join', [])
 Active = namedtuple('Active', [])
 Prepare = namedtuple('Prepare', ['ballot_num'])
 Promise = namedtuple('Promise', ['ballot_num', 'accepted_proposals'])
 Propose = namedtuple('Propose', ['slot', 'proposal'])
 Welcome = namedtuple('Welcome', ['state', 'slot', 'decisions'])
 Decided = namedtuple('Decided', ['slot'])
 Preempted = namedtuple('Preempted', ['slot', 'preempted_by'])
 Adopted = namedtuple('Adopted', ['ballot_num', 'accepted_proposals'])
 Accepting = namedtuple('Accepting', ['leader'])

msg = Accepted(slot=10, ballot_num=30)

got_ballot_num = msg.ballot_num

JOIN_RETRANSMIT = 0.7
 CATCHUP_INTERVAL = 0.6
 ACCEPT_RETRANSMIT = 1.0
 PREPARE_RETRANSMIT = 1.0
 INVOKE_RETRANSMIT = 0.5
 LEADER_TIMEOUT = 1.0
 NULL_BALLOT = Ballot(-1, -1)  # sorts before all real ballots
 NOOP_PROPOSAL = Proposal(None, None, None)  # no-op to fill otherwise empty slots

Proposal = namedtuple('Proposal', ['caller', 'client_id', 'input'])
 Ballot = namedtuple('Ballot', ['n', 'leader'])

class Role(object):
 def __init__(self, node):
  self.node = node
  self.node.register(self)
  self.running = True
  self.logger = node.logger.getChild(type(self).__name__)
 def set_timer(self, seconds, callback):
  return self.node.network.set_timer(self.node.address, seconds,
lambda: self.running and callback())
 def stop(self):
  self.running = False
  self.node.unregister(self)

class Node(object):
 unique_ids = itertools.count()
 def __init__(self, network, address):
  self.network = network
  self.address = address or 'N%d' % self.unique_ids.next()
  self.logger = SimTimeLogger(
logging.getLogger(self.address), {'network': self.network})
  self.logger.info('starting')
  self.roles = []
  self.send = functools.partial(self.network.send, self)
 def register(self, roles):
  self.roles.append(roles)
 def unregister(self, roles):
  self.roles.remove(roles)
 def receive(self, sender, message):
  handler_name = 'do_%s' % type(message).__name__
  for comp in self.roles[:]:
if not hasattr(comp, handler_name):
 continue
comp.logger.debug("received %s from %s", message, sender)
fn = getattr(comp, handler_name)
fn(sender=sender, **message._asdict())

class Member(object):
 def __init__(self, state_machine, network, peers, seed=None,
  seed_cls=Seed, bootstrap_cls=Bootstrap):
  self.network = network
  self.node = network.new_node()
  if seed is not None:
self.startup_role = seed_cls(self.node, initial_state=seed, peers=peers,
 execute_fn=state_machine)
  else:
self.startup_role = bootstrap_cls(self.node,
 execute_fn=state_machine, peers=peers)
  self.requester = None
 def start(self):
  self.startup_role.start()
  self.thread = threading.Thread(target=self.network.run)
  self.thread.start()
 def invoke(self, input_value, request_cls=Requester):
  assert self.requester is None
  q = Queue.Queue()
  self.requester = request_cls(self.node, input_value, q.put)
  self.requester.start()
  output = q.get()
  self.requester = None
  return output

class Acceptor(Role):
 def __init__(self, node):
  super(Acceptor, self).__init__(node)
  self.ballot_num = NULL_BALLOT
  self.accepted_proposals = {}  # {slot: (ballot_num, proposal)}
 def do_Prepare(self, sender, ballot_num):
  if ballot_num > self.ballot_num:
self.ballot_num = ballot_num
# we've heard from a scout, so it might be the next leader
self.node.send([self.node.address], Accepting(leader=sender))
  self.node.send([sender], Promise(
ballot_num=self.ballot_num, 
accepted_proposals=self.accepted_proposals
  ))
 def do_Accept(self, sender, ballot_num, slot, proposal):
  if ballot_num >= self.ballot_num:
self.ballot_num = ballot_num
acc = self.accepted_proposals
if slot not in acc or acc[slot][0] < ballot_num:
 acc[slot] = (ballot_num, proposal)
  self.node.send([sender], Accepted(
slot=slot, ballot_num=self.ballot_num))

BootStrap  Replica  Replica Replica
  X---------->| | | Join
  |<----------X X | Welcome
  X------------------------>| | Join
  |<------------------------X | Welcome
  X-------------------------------------->| Join
  |<--------------------------------------X Welcome
class Replica(Role):
 def __init__(self, node, execute_fn, state, slot, decisions, peers):
  super(Replica, self).__init__(node)
  self.execute_fn = execute_fn
  self.state = state
  self.slot = slot
  self.decisions = decisions
  self.peers = peers
  self.proposals = {}
  # next slot num for a proposal (may lead slot)
  self.next_slot = slot
  self.latest_leader = None
  self.latest_leader_timeout = None
 # making proposals
 def do_Invoke(self, sender, caller, client_id, input_value):
  proposal = Proposal(caller, client_id, input_value)
  slot = next((s for s, p in self.proposals.iteritems() if p == proposal), None)
  # propose, or re-propose if this proposal already has a slot
  self.propose(proposal, slot)
 def propose(self, proposal, slot=None):
  """Send (or resend, if slot is specified) a proposal to the leader"""
  if not slot:
slot, self.next_slot = self.next_slot, self.next_slot + 1
  self.proposals[slot] = proposal
  # find a leader we think is working - either the latest we know of, or
  # ourselves (which may trigger a scout to make us the leader)
  leader = self.latest_leader or self.node.address
  self.logger.info(
"proposing %s at slot %d to leader %s" % (proposal, slot, leader))
  self.node.send([leader], Propose(slot=slot, proposal=proposal))
 # handling decided proposals
 def do_Decision(self, sender, slot, proposal):
  assert not self.decisions.get(self.slot, None), \
 "next slot to commit is already decided"
  if slot in self.decisions:
assert self.decisions[slot] == proposal, \
 "slot %d already decided with %r!" % (slot, self.decisions[slot])
return
  self.decisions[slot] = proposal
  self.next_slot = max(self.next_slot, slot + 1)
  # re-propose our proposal in a new slot if it lost its slot and wasn't a no-op
  our_proposal = self.proposals.get(slot)
  if (our_proposal is not None and 
our_proposal != proposal and our_proposal.caller):
self.propose(our_proposal)
  # execute any pending, decided proposals
  while True:
commit_proposal = self.decisions.get(self.slot)
if not commit_proposal:
 break  # not decided yet
commit_slot, self.slot = self.slot, self.slot + 1
self.commit(commit_slot, commit_proposal)
 def commit(self, slot, proposal):
  """Actually commit a proposal that is decided and in sequence"""
  decided_proposals = [p for s, p in self.decisions.iteritems() if s < slot]
  if proposal in decided_proposals:
self.logger.info(
 "not committing duplicate proposal %r, slot %d", proposal, slot)
return  # duplicate
  self.logger.info("committing %r at slot %d" % (proposal, slot))
  if proposal.caller is not None:
# perform a client operation
self.state, output = self.execute_fn(self.state, proposal.input)
self.node.send([proposal.caller], 
 Invoked(client_id=proposal.client_id, output=output))
 # tracking the leader
 def do_Adopted(self, sender, ballot_num, accepted_proposals):
  self.latest_leader = self.node.address
  self.leader_alive()
 def do_Accepting(self, sender, leader):
  self.latest_leader = leader
  self.leader_alive()
 def do_Active(self, sender):
  if sender != self.latest_leader:
return
  self.leader_alive()
 def leader_alive(self):
  if self.latest_leader_timeout:
self.latest_leader_timeout.cancel()
  def reset_leader():
idx = self.peers.index(self.latest_leader)
self.latest_leader = self.peers[(idx + 1) % len(self.peers)]
self.logger.debug("leader timed out; tring the next one, %s", 
 self.latest_leader)
  self.latest_leader_timeout = self.set_timer(LEADER_TIMEOUT, reset_leader)
 # adding new cluster members
 def do_Join(self, sender):
  if sender in self.peers:
self.node.send([sender], Welcome(
 state=self.state, slot=self.slot, decisions=self.decisions))

class Leader(Role):
 def __init__(self, node, peers, commander_cls=Commander, scout_cls=Scout):
  super(Leader, self).__init__(node)
  self.ballot_num = Ballot(0, node.address)
  self.active = False
  self.proposals = {}
  self.commander_cls = commander_cls
  self.scout_cls = scout_cls
  self.scouting = False
  self.peers = peers
 def start(self):
  # reminder others we're active before LEADER_TIMEOUT expires
  def active():
if self.active:
 self.node.send(self.peers, Active())
self.set_timer(LEADER_TIMEOUT / 2.0, active)
  active()
 def spawn_scout(self):
  assert not self.scouting
  self.scouting = True
  self.scout_cls(self.node, self.ballot_num, self.peers).start()
 def do_Adopted(self, sender, ballot_num, accepted_proposals):
  self.scouting = False
  self.proposals.update(accepted_proposals)
  # note that we don't re-spawn commanders here; if there are undecided
  # proposals, the replicas will re-propose
  self.logger.info("leader becoming active")
  self.active = True
 def spawn_commander(self, ballot_num, slot):
  proposal = self.proposals[slot]
  self.commander_cls(self.node, ballot_num, slot, proposal, self.peers).start()
 def do_Preempted(self, sender, slot, preempted_by):
  if not slot:  # from the scout
self.scouting = False
  self.logger.info("leader preempted by %s", preempted_by.leader)
  self.active = False
  self.ballot_num = Ballot((preempted_by or self.ballot_num).n + 1, 
  self.ballot_num.leader)
 def do_Propose(self, sender, slot, proposal):
  if slot not in self.proposals:
if self.active:
 self.proposals[slot] = proposal
 self.logger.info("spawning commander for slot %d" % (slot,))
 self.spawn_commander(self.ballot_num, slot)
else:
 if not self.scouting:
  self.logger.info("got PROPOSE when not active - scouting")
  self.spawn_scout()
 else:
  self.logger.info("got PROPOSE while scouting; ignored")
  else:
self.logger.info("got PROPOSE for a slot already being proposed")

def __init__(self, node, ballot_num, peers):
  super(Scout, self).__init__(node)
  self.ballot_num = ballot_num
  self.accepted_proposals = {}
  self.acceptors = set([])
  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.node.send(self.peers, Prepare(ballot_num=self.ballot_num))
  self.retransmit_timer = self.set_timer(PREPARE_RETRANSMIT, self.send_prepare)
 def update_accepted(self, accepted_proposals):
  acc = self.accepted_proposals
  for slot, (ballot_num, proposal) in accepted_proposals.iteritems():
if slot not in acc or acc[slot][0] < ballot_num:
 acc[slot] = (ballot_num, proposal)
 def do_Promise(self, sender, ballot_num, accepted_proposals):
  if ballot_num == self.ballot_num:
self.logger.info("got matching promise; need %d" % self.quorum)
self.update_accepted(accepted_proposals)
self.acceptors.add(sender)
if len(self.acceptors) >= self.quorum:
 # strip the ballot numbers from self.accepted_proposals, now that it
 # represents a majority
 accepted_proposals = \ 
  dict((s, p) for s, (b, p) in self.accepted_proposals.iteritems())
 # We're adopted; note that this does *not* mean that no other
 # leader is active.  # Any such conflicts will be handled by the
 # commanders.
 self.node.send([self.node.address],
  Adopted(ballot_num=ballot_num, 
accepted_proposals=accepted_proposals))
 self.stop()
  else:
# this acceptor has promised another leader a higher ballot number,
# so we've lost
self.node.send([self.node.address], 
 Preempted(slot=None, preempted_by=ballot_num))
self.stop()

def __init__(self, node, ballot_num, slot, proposal, peers):
  super(Commander, self).__init__(node)
  self.ballot_num = ballot_num
  self.slot = slot
  self.proposal = proposal
  self.acceptors = set([])
  self.peers = peers
  self.quorum = len(peers) / 2 + 1
 def start(self):
  self.node.send(set(self.peers) - self.acceptors, Accept(
slot=self.slot, ballot_num=self.ballot_num, proposal=self.proposal))
  self.set_timer(ACCEPT_RETRANSMIT, self.start)
 def finished(self, ballot_num, preempted):
  if preempted:
self.node.send([self.node.address], Preempted(slot=self.slot, preempted_by=ballot_num))
  else:
self.node.send([self.node.address], Decided(slot=self.slot))
  self.stop()
 def do_Accepted(self, sender, slot, ballot_num):
  if slot != self.slot:
return
  if ballot_num == self.ballot_num:
self.acceptors.add(sender)
if len(self.acceptors) < self.quorum:
 return
self.node.send(self.peers, Decision(slot=self.slot, proposal=self.proposal))
self.finished(ballot_num, False)
  else:
self.finished(ballot_num, True)

class Bootstrap(Role):
 def __init__(self, node, peers, execute_fn,
  replica_cls=Replica, acceptor_cls=Acceptor, leader_cls=Leader,
  commander_cls=Commander, scout_cls=Scout):
  super(Bootstrap, self).__init__(node)
  self.execute_fn = execute_fn
  self.peers = peers
  self.peers_cycle = itertools.cycle(peers)
  self.replica_cls = replica_cls
  self.acceptor_cls = acceptor_cls
  self.leader_cls = leader_cls
  self.commander_cls = commander_cls
  self.scout_cls = scout_cls
 def start(self):
  self.join()
 def join(self):
  self.node.send([next(self.peers_cycle)], Join())
  self.set_timer(JOIN_RETRANSMIT, self.join)
 def do_Welcome(self, sender, state, slot, decisions):
  self.acceptor_cls(self.node)
  self.replica_cls(self.node, execute_fn=self.execute_fn, peers=self.peers, state=state, slot=slot, decisions=decisions)
  self.leader_cls(self.node, peers=self.peers, commander_cls=self.commander_cls,scout_cls=self.scout_cls).start()
  self.stop()