Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- process Proposer{ val MaximumAcceptorsFailures val MaximumProposersFailures val MaximumClientFailures var lastDecidedSequentialSlot = -1 var nextSlot = 0 var slotProposals[slot => values[{value}]] = {0, {}} var slotAcceptorsProposals[slot => values[{value}]] = {0, {}} var slotProposal[slot => value] = {0, null} var slotVersion[slot => num] = {0, increment(self, 0)} var slotAccepted[slot => num] = {0, 0} var slotStatus[slot => phase[NONE, PHASE1, PHASE2, DECIDED]] = {0, NONE} var cumulativeDecidedAck[client => slot] = {null, -1} var lastAcceptorDroppedSlot = -1 while(acceptorsCurrentSize/2 < MaximumAcceptorsFailures) wait() and stores received proposals IF CRASHED does not send membership or paxos messages, requests a state snapshot from a quorumProposers() of Proposers trigger(self, Phase1(0)) while(true){ receive { nonLeaderReceive } } nonLeaderReceive{ case e: LeaderElectedEvent(leaderId) => { if (leaderId = self){ for(slot := lastDecidedSequentialSlot; slot <= nextSlot; slot++){ if(slotStatus(slot).phase = DECIDED) send(ALL, Decided(slot, slotProposal(slot).value)) if(slotStatus(slot).phase = PHASE1) send(Acceptors, Prepare(slot, slotStatus(slot).num)) if(slotStatus(slot).phase = PHASE2) send(Acceptors, Accept(slot, slotStatus(slot).num, slotProposal(slot).value)) } become(leaderReceive) } } case m: Proposal(slot, value) => { if(slot > lastDecidedSequentialSlot && slotStatus(slot).phase != PHASE2 && slotStatus(slot).phase != DECIDED){ slotProposal(slot).values := BusinessLogic(slotProposal(slot).values, value) if(slotStatus(slot).phase = NONE && BusinessLogic(slotProposal(slot).values)) trigger(self, Phase1(slot)) } } case m: PrepareNack(slot, num) => { if(slot > lastDecidedSequentialSlot && slotStatus(slot).phase = PHASE1 && num >= slotVersion(slot).num){ slotVersion(slot).num := increment(self, num) } } case m: PrepareAck(slot, num, value) => { if(slot > lastDecidedSequentialSlot && slotStatus(slot).phase = PHASE1 && num = slotVersion(slot).num){ slotAcceptorsProposals(slot).values += value if(|slotAcceptorsProposals(slot).values| > quorum()){ slotStatus(slot).phase := PHASE2 slotProposal(slot).value := BusinessLogic(slotAcceptorsProposals(slot).values, slotProposals(slot).values) } } } case m: Accepted(slot, num, value) => { if(slot > lastDecidedSequentialSlot && slotStatus(slot).phase = PHASE2 && num = slotStatus(slot).num){ slotAccepted(slot).num++ if(slotAccepted(slot).num > quorum()){ slotStatus(slot).phase := DECIDED while(slotStatus(lastDecidedSequentialSlot + 1).phase = DECIDED) lastDecidedSequentialSlot++ DROP ALL DATA ON SLOTS < lastDecidedSequentialSlot while(slotStatus(nextSlot).phase != NONE) nextSlot++ if(nextSlot - lastDecidedSequentialSlot < threshold) trigger(self, Phase1(nextSlot)) } } } } leaderReceive { case e: LeaderElectedEvent(leaderId) => { if (leaderId != self){ become(nonLeaderReceive) } } case m: Proposal(slot, value) => { if(slot > lastDecidedSequentialSlot && slotStatus(slot).phase != PHASE2 && slotStatus(slot).phase != DECIDED){ slotProposal(slot).values := BusinessLogic(slotProposal(slot).values, value) if(slotStatus(slot).phase = NONE && BusinessLogic(slotProposal(slot).values)) trigger(self, Phase1(slot)) } } case e: Phase1(slot) => { if(slot > lastDecidedSequentialSlot && slotStatus(slot).phase = NONE){ slotStatus(slot).phase := PHASE1 send(Acceptors, Prepare(slot, slotStatus(slot).num)) } } case m: PrepareNack(slot, num) => { if(slot > lastDecidedSequentialSlot && slotStatus(slot).phase = PHASE1 && num >= slotVersion(slot).num){ slotVersion(slot).num := increment(self, num) send(Acceptors, Prepare(slot, slotVersion(slot).num)) } } case m: PrepareAck(slot, num, value) => { if(slot > lastDecidedSequentialSlot && slotStatus(slot).phase = PHASE1 && num = slotVersion(slot).num){ slotAcceptorsProposals(slot).values += value if(|slotAcceptorsProposals(slot).values| > quorum()){ slotStatus(slot).phase := PHASE2 slotProposal(slot).value := BusinessLogic(slotAcceptorsProposals(slot).values, slotProposals(slot).values) send(Acceptors, Accept(slot, slotStatus(slot).num, slotProposal(slot).value)) } } } case m: Accepted(slot, num, value) => { if(slot > lastDecidedSequentialSlot && slotStatus(slot).phase = PHASE2 && num = slotStatus(slot).num){ slotAccepted(slot).num++ if(slotAccepted(slot).num > quorum()){ slotStatus(slot).phase := DECIDED while(slotStatus(lastDecidedSequentialSlot + 1).phase = DECIDED) lastDecidedSequentialSlot++ send(ALL, Decided(slot, slotProposal(slot).value)) DROP ALL DATA ON SLOTS < lastDecidedSequentialSlot while(slotStatus(nextSlot).phase != NONE) nextSlot++ if(nextSlot - lastDecidedSequentialSlot < threshold) trigger(self, Phase1(nextSlot)) } } } case m: CumulativeDecidedAck(client, slot) => { if(slot > lastAcceptorDroppedSlot && slot <= lastDecidedSequentialSlot){ if(cumulativeDecidedAck(client).slot < slot) cumulativeDecidedAck(client).slot := slot dropSlot = FIND the slot number such that at least a quorumClient() has sent a greater or equal CumulativeDecidedAck in cumulativeDecidedAck send (Acceptors, CumulativeDropSlots(dropSlot)) } } } } } def increment(self, num): Int = { num.leastSignificantBits = self num.mostSignificantBits ++ return num } def quorum(): Int = { if(acceptorsCurrentSize / 2 < MaximumAcceptorsFailures){ MaximumAcceptorsFailures }else{ acceptorsCurrentSize / 2 } } def quorumProposers(): Int = { if(proposersCurrentSize / 2 < MaximumProposersFailures){ MaximumProposersFailures }else{ proposersCurrentSize / 2 } } def quorumClient(): Int = { if(clientsCurrentSize / 2 < MaximumClientFailures){ MaximumClientFailures }else{ clientsCurrentSize / 2 } }
Advertisement
Add Comment
Please, Sign In to add comment
