@@ -23,65 +23,78 @@ case class PipePaxos[A](
2323):
2424
2525 // private helper functions
26- private lazy val openRounds = log.collect { case (k, ClosingCons .Open (paxos)) => (k, paxos) }
27- private lazy val maxPaxos : (round : Time , paxos : Paxos [A ]) = openRounds.maxBy(_._1)
28- lazy val nextDecisionRound : Long = openRounds.minBy(_._1)._1
26+ lazy val openRounds = log.collect { case (k, ClosingCons .Open (paxos)) => (k, paxos) }
27+ private lazy val maxPaxos : (round : Time , paxos : Paxos [A ]) = openRounds.maxByOption(_._1).getOrElse((- 1 , Paxos [A ]()))
28+ lazy val nextDecisionRound : Long = openRounds.minByOption(_._1).map(_._1).getOrElse(- 1 )
29+ lazy val maxRound = log.keys.maxOption.getOrElse(- 1L )
30+ def nextIdleRound (using Participants ): Option [(round : Time , paxos : Paxos [A ])] =
31+ openRounds.find((_, p) => p.phase == MultipaxosPhase .Idle )
32+
33+ lazy val closedRounds : Map [Long , A ] = log.collect { case (k, ClosingCons .Done (value)) => (k, value) }
34+
35+ def phase (using Participants ) = maxPaxos.paxos.phase
2936
3037 // public API
3138 def leader (using Participants ): Option [Uid ] = maxPaxos.paxos.currentRound match
3239 case Some (PaxosRound (leaderElection, _)) => leaderElection.result
3340 case None => None
3441
35- def phase (using Participants ): MultipaxosPhase =
36- maxPaxos.paxos.currentRound match
37- case None => MultipaxosPhase .LeaderElection
38- case Some (PaxosRound (leaderElection, _)) if leaderElection.result.isEmpty => MultipaxosPhase .LeaderElection
39- case Some (PaxosRound (leaderElection, proposals))
40- if leaderElection.result.nonEmpty && proposals.votes.nonEmpty => MultipaxosPhase .Voting
41- case Some (PaxosRound (leaderElection, proposals))
42- if leaderElection.result.nonEmpty && proposals.votes.isEmpty => MultipaxosPhase .Idle
43- case _ => throw new Error (" Inconsistent Paxos State"
44-
4542 def readDecisionsSince (time : Time ): Iterable [A ] =
46- NumericRange (time, nextDecisionRound, 1L ).view.flatMap(log.get).collect{ case Done (value) => value}
43+ NumericRange (time, nextDecisionRound, 1L ).view.flatMap(log.get).collect { case Done (value) => value }
4744
4845 def startLeaderElection (using LocalUid ): PipePaxos [A ] =
4946 PipePaxos (openRounds.view.mapValues(v => ClosingCons .Open (v.phase1a)).toMap)
5047
5148 def proposeIfLeader (value : A )(using LocalUid , Participants ): PipePaxos [A ] =
52- PipePaxos (
53- Map (maxPaxos.round -> Open (maxPaxos.paxos.phase2a(value)))
54- ) // phase 2a already checks if I am the leader
55-
56- def upkeep (using LocalUid , Participants ): PipePaxos [A ] = PipePaxos {
57- openRounds.map { (roundNumber, paxos) =>
58- // perform upkeep in Paxos
59- val deltaPaxos = paxos.upkeep()
60- val newPaxos = paxos.merge(deltaPaxos)
61-
62- (newPaxos.result, newPaxos.newestBallotWithLeader) match
63- case (Some (decision), Some ((ballotNum, PaxosRound (leaderElection, _)))) =>
64- // we are voting on proposals and there is a decision
65-
66- val newDecision = Map (roundNumber -> Done (decision))
67- if roundNumber == maxPaxos.round then
68- // create new Paxos where leader is already elected
69- val newPaxos = Paxos (rounds =
70- Map (ballotNum -> PaxosRound (
71- leaderElection = leaderElection,
72- proposals = Voting [A ]()
73- ))
74- )
75- newDecision.updated(roundNumber + 1 , Open (newPaxos))
76- else newDecision
77- case _ =>
78- // nothing to do, return upkeep result
79- Map (roundNumber -> Open (deltaPaxos))
80- }.foldLeft(Map .empty[Long , ClosingCons [A ]])(Lattice .merge[Map [Long , ClosingCons [A ]]])
49+ nextIdleRound match {
50+ case Some (round) =>
51+ PipePaxos (
52+ Map (round.round -> Open (round.paxos.phase2a(value)))
53+ ) // phase 2a already checks if I am the leader
54+ case None =>
55+ val rounded = addRound()
56+ val proposed = rounded.maxPaxos.paxos.phase2a(value)
57+ rounded `merge` PipePaxos (Map (rounded.maxRound -> Open (proposed)))
58+
59+ }
60+
61+ def addRound ()(using Participants ): PipePaxos [A ] = {
62+ val maybeshortcut = maxPaxos.paxos.newestBallotWithLeader match {
63+ case Some ((ballotNum, PaxosRound (leaderElection, _))) =>
64+ Paxos (rounds =
65+ Map (ballotNum -> PaxosRound (
66+ leaderElection = leaderElection,
67+ proposals = Voting [A ]()
68+ ))
69+ )
70+ case _ => Paxos [A ]()
71+ }
72+ PipePaxos (Map ((maxPaxos.round + 1 ) -> Open (maybeshortcut)))
8173 }
8274
75+ def upkeep (using LocalUid , Participants ): PipePaxos [A ] =
76+ val res = PipePaxos {
77+ openRounds.map { (roundNumber, paxos) =>
78+ // perform upkeep in Paxos
79+ val deltaPaxos = paxos.upkeep()
80+ val newPaxos = paxos.merge(deltaPaxos)
81+
82+ newPaxos.result match
83+ case Some (decision) =>
84+ // we are voting on proposals and there is a decision
85+ Map (roundNumber -> Done (decision))
86+ case _ =>
87+ // nothing to do, return upkeep result
88+ Map (roundNumber -> Open (deltaPaxos))
89+ }.foldLeft(Map .empty[Long , ClosingCons [A ]])(Lattice .merge[Map [Long , ClosingCons [A ]]])
90+ }
91+ // keep an open round to remember leaderelection
92+ if res.openRounds.nonEmpty
93+ then res
94+ else
95+ res `merge` addRound()
96+
8397object PipePaxos :
84- def empty [A ]: PipePaxos [A ] = PipePaxos [A ](Map ( 0L -> Open ( Paxos ())) )
98+ def empty [A ]: PipePaxos [A ] = PipePaxos [A ](Map .empty )
8599
86100 given [A ] => Lattice [PipePaxos [A ]] = Lattice .derived
87-
0 commit comments