-
Notifications
You must be signed in to change notification settings - Fork 29
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
simplify multipaxos quite a lot and remove some inconsistencies
- Loading branch information
Showing
4 changed files
with
90 additions
and
102 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
130 changes: 38 additions & 92 deletions
130
Modules/RDTs/src/main/scala/rdts/datatypes/experiments/protocols/MultiPaxos.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,139 +1,85 @@ | ||
| package rdts.datatypes.experiments.protocols | ||
|
|
||
| import rdts.base.{Bottom, Lattice, LocalUid, Uid} | ||
| import rdts.datatypes.Epoch | ||
| import rdts.datatypes.experiments.protocols.Paxos.given | ||
| import rdts.datatypes.{Epoch, LastWriterWins} | ||
|
|
||
| enum MultipaxosPhase: | ||
| case LeaderElection | ||
| case Voting | ||
| case Idle | ||
|
|
||
| case class MultiPaxosRound[A]( | ||
| leader: LastWriterWins[Option[Uid]] = LastWriterWins.empty, | ||
| phase: LastWriterWins[MultipaxosPhase] = LastWriterWins.fallback(MultipaxosPhase.LeaderElection), | ||
| paxos: Paxos[A] = Paxos[A]() | ||
| ) | ||
|
|
||
| case class MultiPaxos[A]( | ||
| rounds: Epoch[MultiPaxosRound[A]] = Epoch.empty[MultiPaxosRound[A]], | ||
| rounds: Epoch[Paxos[A]] = Epoch.empty[Paxos[A]], | ||
| log: Map[Long, A] = Map.empty | ||
| ): | ||
|
|
||
| // private helper functions | ||
| private def currentPaxos = rounds.value.paxos | ||
| private def currentPaxos = rounds.value | ||
|
|
||
| // public API | ||
| def leader: Option[Uid] = rounds.value.leader.value | ||
| def paxosLeader(using Participants): Option[Uid] = currentPaxos.currentRound match | ||
| def leader(using Participants): Option[Uid] = currentPaxos.currentRound match | ||
| case Some((_, PaxosRound(leaderElection, _))) => leaderElection.result | ||
| case None => None | ||
|
|
||
| def phase: MultipaxosPhase = rounds.value.phase.value | ||
| def read: List[A] = log.toList.sortBy(_._1).map(_._2) | ||
| def phase(using Participants): MultipaxosPhase = | ||
| currentPaxos.currentRound match | ||
| case None => MultipaxosPhase.LeaderElection | ||
| case Some((_, PaxosRound(leaderElection, _))) if leaderElection.result.isEmpty => MultipaxosPhase.LeaderElection | ||
| case Some((_, PaxosRound(leaderElection, proposals))) | ||
| if leaderElection.result.nonEmpty && proposals.votes.nonEmpty => MultipaxosPhase.Voting | ||
| case Some((_, PaxosRound(leaderElection, proposals))) | ||
| if leaderElection.result.nonEmpty && proposals.votes.isEmpty => MultipaxosPhase.Idle | ||
| case _ => throw new Error("Inconsistent Paxos State") | ||
|
|
||
| def read: List[A] = log.toList.sortBy(_._1).map(_._2) | ||
|
|
||
| def startLeaderElection(using LocalUid, Participants): MultiPaxos[A] = | ||
| MultiPaxos( | ||
| rounds = | ||
| // transition to new epoch | ||
| rounds.epocheWrite(MultiPaxosRound( | ||
| leader = rounds.value.leader.write(None), // set leader to none | ||
| paxos = Paxos().phase1a, // start new Paxos round with self proposed as leader | ||
| phase = rounds.value.phase.write(MultipaxosPhase.LeaderElection) | ||
| )) | ||
| ) | ||
| MultiPaxos(rounds.write(currentPaxos.phase1a)) // start new Paxos round with self proposed as leader | ||
|
|
||
| def proposeIfLeader(value: A)(using LocalUid, Participants): MultiPaxos[A] = | ||
| val afterProposal = currentPaxos.phase2a(value) | ||
| // check if proposing does anything, otherwise return empty delta | ||
| if currentPaxos.subsumes(afterProposal) then | ||
| MultiPaxos() | ||
| else | ||
| MultiPaxos( | ||
| rounds = rounds.write(MultiPaxosRound( | ||
| paxos = afterProposal, // phase 2a already checks if I am the leader | ||
| phase = rounds.value.phase.write(MultipaxosPhase.Voting) | ||
| )) | ||
| ) | ||
| MultiPaxos(rounds = rounds.write(afterProposal)) // phase 2a already checks if I am the leader | ||
|
|
||
| def upkeep(using LocalUid, Participants): MultiPaxos[A] = { | ||
| // perform upkeep in Paxos | ||
| val deltaPaxos = currentPaxos.upkeep() | ||
| val newPaxos = currentPaxos.merge(deltaPaxos) | ||
|
|
||
| rounds.value.phase.value match | ||
| case MultipaxosPhase.LeaderElection => | ||
| // we are in a leader election | ||
| // check if this process was elected as a leader | ||
| newPaxos.currentRound match | ||
| case Some((_, PaxosRound(leaderElection, _))) | ||
| if leaderElection.result.nonEmpty // if leaderElection.result == Some(replicaId) | ||
| => | ||
| // there is a result | ||
| MultiPaxos(rounds = | ||
| rounds.write(MultiPaxosRound( | ||
| leader = rounds.value.leader.write(leaderElection.result), | ||
| paxos = deltaPaxos, | ||
| phase = rounds.value.phase.write(MultipaxosPhase.Idle) | ||
| )) | ||
| ) | ||
| case _ => | ||
| // there is no result, return upkeep result | ||
| MultiPaxos(rounds = | ||
| rounds.write(MultiPaxosRound( | ||
| paxos = deltaPaxos | ||
| )) | ||
| ) | ||
| case MultipaxosPhase.Voting => | ||
| // we are voting on proposals | ||
| // check if there is a decision | ||
| (newPaxos.decision, newPaxos.decidedLeaderElection) match | ||
| case (Some(decision), Some((ballotNum, leaderElection))) => | ||
| // append log | ||
| val newLog = Map(rounds.counter -> decision) | ||
| // create new Paxos where leader is already elected | ||
| val newPaxos = Paxos(rounds = | ||
| Map(ballotNum -> PaxosRound( | ||
| leaderElection = leaderElection, | ||
| proposals = Voting[A]() | ||
| )) | ||
| ) | ||
| // return new Multipaxos with: appended log, set leader, phase as idle, | ||
| MultiPaxos( | ||
| rounds = rounds.epocheWrite(MultiPaxosRound( | ||
| leader = rounds.value.leader.write(paxosLeader), | ||
| phase = rounds.value.phase.write(MultipaxosPhase.Idle), | ||
| paxos = newPaxos | ||
| )), | ||
| log = newLog | ||
| ) | ||
| case _ => | ||
| // no decision yet, just send upkeep result | ||
| MultiPaxos(rounds = | ||
| rounds.write(MultiPaxosRound( | ||
| paxos = deltaPaxos | ||
| )) | ||
| ) | ||
| case MultipaxosPhase.Idle => | ||
| // nothing to do | ||
| MultiPaxos() | ||
| (newPaxos.decision, newPaxos.newestBallotWithLeader) match | ||
| case (Some(decision), Some((ballotNum, PaxosRound(leaderElection, _)))) => | ||
| // we are voting on proposals and there is a decision | ||
|
|
||
| val newLog = Map(rounds.counter -> decision) // append log | ||
| // create new Paxos where leader is already elected | ||
| val newPaxos = Paxos(rounds = | ||
| Map(ballotNum -> PaxosRound( | ||
| leaderElection = leaderElection, | ||
| proposals = Voting[A]() | ||
| )) | ||
| ) | ||
| // return new Multipaxos with: appended log | ||
| MultiPaxos( | ||
| rounds = rounds.epocheWrite(newPaxos), | ||
| log = newLog | ||
| ) | ||
| case _ => | ||
| // nothing to do, return upkeep result | ||
| MultiPaxos(rounds = rounds.write(deltaPaxos)) | ||
| } | ||
|
|
||
| override def toString: String = | ||
| lazy val s = s"MultiPaxos(epoch: ${rounds.counter}, phase: $phase, log: $read)" | ||
| lazy val s = s"MultiPaxos(epoch: ${rounds.counter}, log: $read)" | ||
| s | ||
|
|
||
| object MultiPaxos: | ||
| given [A]: Bottom[MultiPaxosRound[A]] = Bottom.provide(MultiPaxosRound()) | ||
| given [A]: Lattice[MultiPaxosRound[A]] = Lattice.derived | ||
|
|
||
| // for the log | ||
| given [A]: Lattice[Map[Long, A]] = | ||
| given Lattice[A] = Lattice.assertEquals | ||
| Lattice.mapLattice | ||
|
|
||
| given [A]: Lattice[MultiPaxos[A]] = Lattice.derived | ||
|
|
||
| // given consensus: Consensus[MultiPaxos] = | ||
| // extension [A](c: MultiPaxos[A]) | ||
| // override def propose |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters