polish active-active replication

Author: rkuhn

chapter12/src/main/scala/com/reactivedesignpatterns/chapter12/ActiveActive.scala

@@ -7,6 +7,7 @@ import akka.actor._
 import play.api.libs.json.JsValue
 import scala.collection.immutable.TreeMap
 import scala.concurrent.duration._
+import scala.annotation.tailrec
 
 object ActiveActive {
   import ReplicationProtocol._
@@ -50,7 +51,6 @@ object ActiveActive {
     def deadline: Deadline
     def missing: Set[ActorRef]
     def add(res: SeqResult): ReplyState
-    def isKnown: Boolean
     def isFinished: Boolean = missing.isEmpty
   }
   private case class Unknown(deadline: Deadline, replies: Set[SeqResult], missing: Set[ActorRef]) extends ReplyState {
@@ -66,88 +66,94 @@ object ActiveActive {
         } else Unknown(deadline, nextReplies, missing - res.replica)
       } else Unknown(deadline, nextReplies, missing - res.replica)
     }
-    override def isKnown = false
   }
   private case class Known(deadline: Deadline, reply: SeqResult, wrong: Set[ActorRef], missing: Set[ActorRef]) extends ReplyState {
     override def add(res: SeqResult): ReplyState = {
       val nextWrong = if (res.res == reply.res) wrong else wrong + res.replica
       Known(deadline, reply, nextWrong, missing - res.replica)
     }
-    override def isKnown = true
   }
 
   class Coordinator(N: Int) extends Actor {
-    private var replicas = (1 to N).map(_ => context.actorOf(Replica.props)).toSet
+    private var replicas = (1 to N).map(_ => newReplica()).toSet
     private val seqNr = Iterator from 0
     private var replies = TreeMap.empty[Int, ReplyState]
     private var nextReply = 0
 
+    override def supervisorStrategy = SupervisorStrategy.stoppingStrategy
+    
+    private def newReplica(): ActorRef =
+      context.watch(context.actorOf(Replica.props))
+
+    // schedule timeout messages for quiescent periods
     context.setReceiveTimeout(1.second)
 
-    def receive = {
+    def receive = ({
       case cmd: Command =>
         val c = SeqCommand(seqNr.next, cmd, self)
         replicas foreach (_ ! c)
         replies += c.seq -> Unknown(5 seconds fromNow, Set.empty, replicas)
-        doTimeouts()
       case res: SeqResult if replies.contains(res.seq) && replicas.contains(res.replica) =>
         val prevState = replies(res.seq)
         val nextState = prevState.add(res)
-        // potentially send reply if quorum of replies has been received now
-        nextState match {
-          case Known(seq, reply, _, _) if !prevState.isKnown && seq == nextReply =>
-            reply.replyTo ! reply.res
-            nextReply += 1
-          case _ =>
-        }
-        // clean up state
-        if (nextState.isFinished) {
-          dispose(nextState)
-          replies -= res.seq
-        } else {
-          replies += res.seq -> nextState
-          doTimeouts()
-        }
-      case ReceiveTimeout => doTimeouts()
+        replies += res.seq -> nextState
+      case Terminated(ref) =>
+        replaceReplica(ref, terminate = false)
+      case ReceiveTimeout =>
+    }: Receive) andThen { _ =>
+      doTimeouts()
+      sendReplies()
+      evictFinished()
     }
 
     private def doTimeouts(): Unit = {
       val now = Deadline.now
       val expired = replies.iterator.takeWhile(_._2.deadline <= now)
-      expired.map(_._2).foreach(dispose)
+      for ((seq, state) <- expired) {
+        state match {
+          case Unknown(deadline, received, missing) =>
+            // did not reach consensus on this one, pick simple majority
+            val counts = received.groupBy(_.res)
+            val biggest = counts.iterator.map(_._2.size).max
+            val winners = counts.collectFirst {
+              case (res, win) if win.size == biggest => win
+            }.get
+            val losers = (received -- winners).map(_.replica)
+            // don’t wait for further responses
+            replies += seq -> Known(deadline, winners.head, losers, Set.empty)
+          case Known(deadline, reply, wrong, missing) =>
+            // don’t wait for further responses
+            replies += seq -> Known(deadline, reply, wrong, Set.empty)
+        }
+      }
     }
 
-    /**
-     * The given reply state has been removed from the replies map and is now
-     * being disposed of. This means that we need to act upon wrong replies
-     * from replicas.
-     *
-     * If there are replicas for which no reply has been recorded yet, we
-     * ignore them. If they reply incorrectly later they will be replaced then.
-     * GC pauses are tolerated: do not kick out replicas just for being slow.
-     */
-    private def dispose(state: ReplyState): Unit =
-      state match {
-        case Unknown(_, replies, _) =>
-          // did not reach consensus on this one, pick simple majority
-          val counts = replies.toList.groupBy(_.res)
-          val biggest = counts.iterator.map(_._2.size).max
-          val winners = counts.collectFirst {
-            case (res, win) if win.size == biggest => win
-          }.get
-          val losers = replicas -- winners.iterator.map(_.replica).toSet
-          losers foreach replaceReplica
-        case Known(_, _, wrong, _) =>
-          wrong foreach replaceReplica
+    @tailrec private def sendReplies(): Unit =
+      replies.get(nextReply) match {
+        case Some(k @ Known(_, reply, _, _)) =>
+          reply.replyTo ! reply.res
+          nextReply += 1
+          sendReplies()
+        case _ =>
       }
 
-    private def replaceReplica(r: ActorRef): Unit = {
-      replicas -= r
-      r ! PoisonPill
-      val newReplica = context.actorOf(Replica.props)
-      replicas.head ! SendInitialData(newReplica)
-      replicas += newReplica
-    }
+    @tailrec private def evictFinished(): Unit =
+      replies.headOption match {
+        case Some((seq, k @ Known(_, _, wrong, _))) if k.isFinished =>
+          wrong foreach (replaceReplica(_, terminate = true))
+          replies -= seq
+          evictFinished()
+        case _ =>
+      }
+
+    private def replaceReplica(r: ActorRef, terminate: Boolean): Unit =
+      if (replicas contains r) {
+        replicas -= r
+        if (terminate) r ! PoisonPill
+        val replica = newReplica()
+        replicas.head ! SendInitialData(replica)
+        replicas += replica
+      }
   }
 
 }

chapter12/src/main/scala/com/reactivedesignpatterns/chapter12/MultiMasterCRDT.scala

@@ -92,13 +92,15 @@ object MultiMasterCRDT {
     def receive = {
       case Execute(job) =>
         log.info("executing job {}", job)
-        replicator ! Replicator.Update(StorageComponent, ORMap.empty[Status], Replicator.WriteMajority(5.seconds), Some(s"submit $job")) { map =>
-          if (map.get(job) == Some(New)) map + (job -> Executing)
-          else map
+        replicator ! Replicator.Update(StorageComponent, ORMap.empty[Status], Replicator.WriteMajority(5.seconds), Some(job)) { map =>
+          require(map.get(job) == Some(New))
+          map + (job -> Executing)
         }
       case Finish(job) =>
         log.info("job {} finished", job)
-        replicator ! Replicator.Update(StorageComponent, ORMap.empty[Status], Replicator.WriteMajority(5.seconds), Some(s"cancel $job"))(_ + (job -> Finished))
+        replicator ! Replicator.Update(StorageComponent, ORMap.empty[Status], Replicator.WriteMajority(5.seconds))(_ + (job -> Finished))
+      case Replicator.UpdateSuccess(StorageComponent, Some(job)) =>
+        log.info("starting job {}", job)
       case r: Replicator.UpdateResponse[_] =>
         log.info("received update result: {}", r)
       case ch: Replicator.Changed[_] =>
@@ -159,6 +161,7 @@ object MultiMasterCRDT {
     clientInterface ! Cancel("alpha")
     clientInterface ! Cancel("beta")
     executor ! Execute("beta")
+    executor ! Execute("delta")
     1 second sleep
     clientInterface ! Cancel("gamma")
     1 second sleep