pre-refactoring (active-passive)

Author: rkuhn

.gitignore

@@ -110,3 +110,5 @@ src_managed/
 project/boot/
 .history
 .cache
+
+theDatabase*

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

@@ -34,7 +34,7 @@ object ActiveActive {
         cmd match {
           case Put(key, value, r) =>
             map += key -> value
-            replyTo ! SeqResult(seq, PutConfirmed(key), self, r)
+            replyTo ! SeqResult(seq, PutConfirmed(key, value), self, r)
           case Get(key, r) =>
             replyTo ! SeqResult(seq, GetResult(key, map get key), self, r)
         }

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

@@ -17,6 +17,7 @@ import scala.annotation.tailrec
 import akka.contrib.pattern.ClusterSingletonManager
 import akka.contrib.pattern.ClusterSingletonProxy
 import com.typesafe.config.ConfigFactory
+import scala.io.StdIn
 
 object ActivePassive {
   import ReplicationProtocol._
@@ -37,6 +38,8 @@ object ActivePassive {
     private val toReplicate = Queue.empty[Replicate]
     private var replicating = TreeMap.empty[Int, (Replicate, Int)]
 
+    private var rejected = 0
+
     val cluster = Cluster(context.system)
 
     import context.dispatcher
@@ -45,9 +48,10 @@ object ActivePassive {
 
     log.info("taking over from local replica")
     localReplica ! TakeOver(self)
-    
+
     def receive = {
       case InitialState(m, s) =>
+        log.info("took over at sequence {}", s)
         theStore = m
         seqNr = Iterator from s
         context.become(running)
@@ -56,11 +60,12 @@ object ActivePassive {
     }
 
     val running: Receive = {
-      case Put(key, value, replyTo) =>
+      case p @ Put(key, value, replyTo) =>
         if (toReplicate.size < MaxOutstanding) {
           toReplicate.enqueue(Replicate(seqNr.next, key, value, replyTo))
           replicate()
         } else {
+          rejected += 1
           replyTo ! PutRejected(key, value)
         }
       case Get(key, replyTo) =>
@@ -69,13 +74,17 @@ object ActivePassive {
         replicating.valuesIterator foreach {
           case (replicate, count) => disseminate(replicate)
         }
+        if (rejected > 0) {
+          log.info("rejected {} PUT requests", rejected)
+          rejected = 0
+        }
       case Replicated(confirm) =>
         replicating.get(confirm) match {
           case None => // already removed
           case Some((rep, 1)) =>
             replicating -= confirm
             theStore += rep.key -> rep.value
-            rep.replyTo ! PutConfirmed(rep.key)
+            rep.replyTo ! PutConfirmed(rep.key, rep.value)
           case Some((rep, n)) =>
             replicating += confirm -> (rep, n - 1)
         }
@@ -106,7 +115,7 @@ object ActivePassive {
     private val applied = Queue.empty[Replicate]
     private var waiting = TreeMap.empty[Int, Replicate]
 
-    val name = self.path.address.toString
+    val name = Cluster(context.system).selfAddress.toString.replaceAll("[:/]", "_")
     val cluster = Cluster(context.system)
     val random = new Random
 
@@ -116,26 +125,30 @@ object ActivePassive {
         expectedSeq = s + 1
     }
     log.info("started at sequence {}", expectedSeq)
-    
+
     override def postStop(): Unit = {
       log.info("stopped at sequence {}", expectedSeq)
     }
 
     def receive = {
       case TakeOver(active) =>
+        log.info("active replica starting at sequence {}", expectedSeq)
         active ! InitialState(theStore, expectedSeq)
       case Replicate(s, _, _, replyTo) if s < expectedSeq =>
         replyTo ! Replicated(s)
       case r: Replicate =>
         waiting += r.seq -> r
         consolidate()
       case GetSingle(s, replyTo) =>
+        log.info("GetSingle from {}", replyTo)
         if (applied.nonEmpty && applied.head.seq <= s && applied.last.seq >= s)
           replyTo ! applied.find(_.seq == s).get
         else if (s < expectedSeq) replyTo ! InitialState(theStore, expectedSeq)
       case GetFull(replyTo) =>
+        log.info("sending full info to {}", replyTo)
         replyTo ! InitialState(theStore, expectedSeq)
       case InitialState(m, s) if s > expectedSeq =>
+        log.info("received newer state at sequence {} (was at {})", s, expectedSeq)
         theStore = m
         expectedSeq = s + 1
         persist(name, s, m)
@@ -163,12 +176,12 @@ object ActivePassive {
         }
         waiting = waiting.drop(prefix)
       }
-      
+
       // cap the size of the applied buffer
       applied.drop(Math.max(0, applied.size - 10))
 
       // check if we fell behind by too much
-      if (waiting.lastKey - expectedSeq > 10) {
+      if (waiting.nonEmpty && (waiting.lastKey - expectedSeq > 10)) {
         val outstanding = (expectedSeq to waiting.lastKey).iterator.filterNot(waiting.contains).toList
         if (outstanding.size <= 3) outstanding foreach askAround
         else askAroundFullState()
@@ -180,17 +193,23 @@ object ActivePassive {
       random.shuffle(cluster.state.members.iterator.map(_.address).toSeq).take(n)
     }
     private def askAround(seq: Int): Unit = {
+      log.info("asking around for sequence number {}", seq)
       getMembers(askAroundCount).foreach(addr => replicaOn(addr) ! GetSingle(seq, self))
     }
     private def askAroundFullState(): Unit = {
+      log.info("asking for full data")
       getMembers(1).foreach(addr => replicaOn(addr) ! GetFull(self))
     }
     private def replicaOn(addr: Address): ActorSelection =
       context.actorSelection(self.path.toStringWithAddress(addr))
   }
-  
+
   val commonConfig = ConfigFactory.parseString("""
     akka.actor.provider = akka.cluster.ClusterActorRefProvider
+    akka.remote.netty.tcp {
+      host = "127.0.0.1"
+      port = 0
+    }
     akka.cluster {
       gossip-interval = 100ms
       failure-detector {
@@ -199,51 +218,126 @@ object ActivePassive {
       }
     }
     """)
-  def roleConfig(name: String) = ConfigFactory.parseString(s"""akka.cluster.roles = ["$name"]""")
+  def roleConfig(name: String, port: Option[Int]) = {
+    val roles = ConfigFactory.parseString(s"""akka.cluster.roles = ["$name"]""")
+    port match {
+      case None => roles
+      case Some(p) =>
+        ConfigFactory.parseString(s"""akka.remote.netty.tcp.port = $p""")
+          .withFallback(roles)
+    }
+  }
 
-  def start(n: Int): ActorSystem = {
-    val system = ActorSystem(s"node$n", roleConfig("backend") withFallback commonConfig)
+  def start(port: Option[Int]): ActorSystem = {
+    val system = ActorSystem("ActivePassive", roleConfig("backend", port) withFallback commonConfig)
     val localReplica = system.actorOf(Props(new Passive(3, 3.seconds)), "passive")
     val managerProps =
-      ClusterSingletonManager.props(Props(new Active(localReplica, 2, 10)), "active", PoisonPill, Some("backend"))
+      ClusterSingletonManager.props(Props(new Active(localReplica, 2, 120)), "active", PoisonPill,
+        role = Some("backend"), retryInterval = 150.millis)
     val manager = system.actorOf(managerProps, "activeManager")
     system
   }
 
   def main(args: Array[String]): Unit = {
-    val systems = Array.tabulate(5)(start(_))
+    val systems = Array.fill(5)(start(None))
     val seedNode = Cluster(systems(0)).selfAddress
     systems foreach (Cluster(_).join(seedNode))
 
-    val sys = ActorSystem("ActivePassive", commonConfig)
+    val sys = ActorSystem("ActivePassive", ConfigFactory.parseString("akka.loglevel=INFO") withFallback commonConfig)
     Cluster(sys).join(seedNode)
+
+    awaitMembers(sys, systems.length + 1)
+
+    val proxy = sys.actorOf(ClusterSingletonProxy.props("/user/activeManager/active", Some("backend")), "proxy")
+
+    val useStorage = sys.actorOf(Props(new UseStorage(proxy)), "useStorage")
+    useStorage ! Run(0)
+
+    sys.actorOf(Props(new Actor {
+      def receive = {
+        case Run =>
+          StdIn.readLine()
+          useStorage ! Stop
+      }
+    })) ! Run
+
+    Thread.sleep(10000)
 
-    while (Cluster(sys).state.members.size < 6) {
+    val rnd = new Random
+    while (!terminate) {
+      Thread.sleep(5000)
+      val sysidx = rnd.nextInt(systems.length)
+      val oldsys = systems(sysidx)
+      val port = Cluster(oldsys).selfAddress.port
+      oldsys.shutdown()
+      oldsys.awaitTermination()
+      val newsys = start(port)
+      val seed = Cluster(if (sysidx == 0) systems(1) else systems(0)).selfAddress
+      Cluster(newsys).join(seed)
+      systems(sysidx) = newsys
+      awaitMembers(sys, systems.length + 1)
+    }
+
+    Thread.sleep(3000)
+
+    sys.shutdown()
+    systems foreach (_.shutdown())
+  }
+  
+  private def awaitMembers(sys: ActorSystem, count: Int): Unit = {
+    while (Cluster(sys).state.members.size < count) {
       Thread.sleep(500)
       print('.')
       Console.flush()
     }
     println("cluster started")
-    
-    val path = sys / "user" / "activeManager" / "active"
-    val proxy = sys.actorOf(ClusterSingletonProxy.props(path.toString, Some("backend")))
-    
-    sys.actorOf(Props(new UseStorage(proxy)), "useStorage") ! Run
-    
-    while (!terminate) {
-      
-    }
   }
-  
-  private case object Run
+
+  private case class Run(round: Int)
   private case object Stop
   @volatile private var terminate = false
-  
-  private class UseStorage(db: ActorRef) extends Actor {
+
+  private class UseStorage(db: ActorRef) extends Actor with ActorLogging {
+    val N = 200
+    var theStore = Map.empty[String, JsValue]
+    val keys = (1 to N).map(i => f"$i%03d")
+    var outstanding = Set.empty[String]
+    val rnd = new Random
+    var lastOutstandingCount = 0
+
     def receive = {
-      case Run =>
-        
-        self ! Run
+      case Run(0) =>
+        db ! Get("initial", self)
+      case GetResult("initial", _) =>
+        self ! Run(1)
+      case Run(round) =>
+        if (round % 100 == 0) log.info("round {}", round)
+        val nowOutstanding = outstanding.size
+        if (nowOutstanding != lastOutstandingCount) {
+          lastOutstandingCount = nowOutstanding
+          log.info("{} outstanding", nowOutstanding)
+        }
+        for (k <- keys) {
+          db ! Get(k, self)
+          if (!outstanding.contains(k) && rnd.nextBoolean()) {
+            db ! Put(k, JsNumber(round), self)
+            outstanding += k
+          }
+        }
+        context.system.scheduler.scheduleOnce(100.millis, self, Run(round + 1))(context.dispatcher)
+      case GetResult(key, value) =>
+        if (outstanding.contains(key)) {
+          outstanding -= key
+          value foreach (theStore += key -> _)
+        } else if (value != theStore.get(key)) {
+          log.warning("returned wrong value for key {}: {} (expected {})", key, value, theStore.get(key))
+          context.stop(self)
+        }
+      case PutConfirmed(key, value) =>
+        outstanding -= key
+        theStore += key -> value
+      case PutRejected(key, value) =>
+        outstanding -= key
       case Stop => context.stop(self)
     }
     override def postStop(): Unit = terminate = true

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

@@ -14,7 +14,7 @@ object ReplicationProtocol {
   sealed trait Result
 
   case class Put(key: String, value: JsValue, replyTo: ActorRef) extends Command
-  case class PutConfirmed(key: String) extends Result
+  case class PutConfirmed(key: String, value: JsValue) extends Result
   case class PutRejected(key: String, value: JsValue) extends Result
   case class Get(key: String, replyTo: ActorRef) extends Command
   case class GetResult(key: String, value: Option[JsValue]) extends Result
@@ -26,8 +26,8 @@ object Persistence {
 
   def persist(name: String, seq: Int, kv: Map[String, JsValue]): Unit = {
     val bytes = Json.stringify(Json.toJson(Database(seq, kv)))
-    val current = new File(s"theDataBase-$name.json")
-    val next = new File(s"theDataBase-$name.json.new")
+    val current = new File(s"./theDataBase-$name.json")
+    val next = new File(s"./theDataBase-$name.json.new")
     IO.write(next, bytes)
     IO.move(next, current) // atomically update the database
   }

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

@@ -2,4 +2,27 @@ package com.reactivedesignpatterns.chapter12
 
 class MultiMasterCRDT {
 
+  import akka.actor._
+  
+  case class Greet(whom: String)
+  
+  class Greeter extends Actor {
+    def receive = {
+      case Greet(whom) =>
+        sender() ! s"Hello $whom!"
+        val delegate = context.actorOf(grumpyProps)
+        context.become(grumpy(delegate))
+    }
+    def grumpy(delegate: ActorRef): Receive = {
+      case g: Greet => delegate forward g
+    }
+  }
+  
+  val grumpyProps = Props(new Actor {
+    def receive = {
+      case Greet(whom) =>
+        sender() ! s"Go away, $whom!"
+    }
+  })
+  
 }