Improved actor routing

Author: almacken

src/main/resources/application.conf

@@ -14,4 +14,15 @@ akka {
       remote-address-header = on
     }
   }
+}
+
+# Use this dispatcher for actors that make blocking calls to the Elasticsearch database
+elasticsearch-handling-dispatcher {
+  type = Dispatcher
+  executor = "thread-pool-executor"
+  thread-pool-executor {
+    fixed-pool-size = 4
+    # This thread pool is intended for development purposes, and should be increased for production
+  }
+  throughput = 1
 }

src/main/scala/de/upb/cs/swt/delphi/webapi/Configuration.scala

@@ -1,13 +1,20 @@
 package de.upb.cs.swt.delphi.webapi
 
-import com.sksamuel.elastic4s.ElasticsearchClientUri
+import com.sksamuel.elastic4s.{ElasticsearchClientUri, IndexAndType}
+import com.sksamuel.elastic4s.http.ElasticDsl._
 
 /**
   * @author Ben Hermann
   */
-class Configuration(val bindHost: String = "0.0.0.0",
+class Configuration(  //Server and Elasticsearch configuration
+                    val bindHost: String = "0.0.0.0",
                     val bindPort: Int = 8080,
                     val elasticsearchClientUri: ElasticsearchClientUri = ElasticsearchClientUri(
-                      sys.env.getOrElse("DELPHI_ELASTIC_URI", "elasticsearch://localhost:9200"))) {
+                      sys.env.getOrElse("DELPHI_ELASTIC_URI", "elasticsearch://localhost:9200")),
+                    val esProjectIndex: IndexAndType = "delphi" / "project",
+
+                      //Actor system configuration
+                    val elasticActorPoolSize: Int = 8
+                   ) {
 
 }

src/main/scala/de/upb/cs/swt/delphi/webapi/ElasticActor.scala

@@ -1,15 +1,17 @@
 package de.upb.cs.swt.delphi.webapi
 
-import akka.actor.{Actor, ActorLogging, Props, ReceiveTimeout}
+import akka.actor.{Actor, ActorLogging, Props}
 import com.sksamuel.elastic4s.IndexAndType
 import com.sksamuel.elastic4s.http.ElasticDsl._
 import com.sksamuel.elastic4s.http.HttpClient
 import de.upb.cs.swt.delphi.webapi.ElasticActor.GetSource
 
+import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 
 class ElasticActor(configuration: Configuration) extends Actor with ActorLogging{
 
+  implicit val executionContext: ExecutionContext = context.system.dispatchers.lookup("elasticsearch-handling-dispatcher")
   val client = HttpClient(configuration.elasticsearchClientUri)
 
   override def preStart(): Unit = log.info("Search actor started")
@@ -26,9 +28,7 @@ class ElasticActor(configuration: Configuration) extends Actor with ActorLogging
           case Left(_) => Option.empty
         }
       sender().tell(source, context.self)
-      context.stop(self)
     }
-    case ReceiveTimeout => context.stop(self)
   }
 }
 

src/main/scala/de/upb/cs/swt/delphi/webapi/ElasticActorManager.scala

@@ -1,33 +1,48 @@
 package de.upb.cs.swt.delphi.webapi
 
-import akka.actor.{Actor, ActorLogging, Props}
-import com.sksamuel.elastic4s.http.ElasticDsl._
+import akka.actor.{Actor, ActorLogging, Props, Terminated}
+import akka.routing.{ActorRefRoutee, RoundRobinRoutingLogic, Router}
 import de.upb.cs.swt.delphi.webapi.ElasticActorManager.{Enqueue, Retrieve}
 import de.upb.cs.swt.delphi.webapi.ElasticActor.GetSource
 
 class ElasticActorManager(configuration: Configuration) extends Actor with ActorLogging{
 
-  private val index = "delphi" / "project"
+  private val index = configuration.esProjectIndex
+  private var elasticRouter = {
+    val routees = Vector.fill(configuration.elasticActorPoolSize) {
+      val r = context.actorOf(ElasticActor.props(configuration))
+      context watch r
+      ActorRefRoutee(r)
+    }
+    Router(RoundRobinRoutingLogic(), routees)
+  }
 
   override def preStart(): Unit = log.info("Actor manager started")
   override def postStop(): Unit = log.info("Actor manager shut down")
 
   override def receive = {
     case Retrieve(id) => getSource(id)
     case Enqueue(id) =>  getSource(id)
+    case Terminated(id) => {
+      elasticRouter.removeRoutee(id)
+      val r = context.actorOf(ElasticActor.props(configuration))
+      context watch r
+      elasticRouter = elasticRouter.addRoutee(r)
+    }
   }
 
   private def getSource(id: String) = {
-    log.info("Creating actor to search for entry {}", id)
-    val retrieveActor = context.actorOf(ElasticActor.props(configuration))
-    retrieveActor forward GetSource(id, index)
+    log.info("Forwarding search for entry {}", id)
+    elasticRouter.route(GetSource(id, index), sender())
   }
 }
 
 object ElasticActorManager{
   def props(configuration: Configuration) : Props = Props(new ElasticActorManager(configuration))
     .withMailbox("es-priority-mailbox")
 
+  sealed trait ElasticMessage
+
   final case class Retrieve(id: String) extends ElasticMessage
   final case class Enqueue(id: String) extends ElasticMessage
 }

src/main/scala/de/upb/cs/swt/delphi/webapi/ElasticMessage.scala

@@ -5,6 +5,7 @@ import akka.actor.{Actor, ActorLogging, ActorRef, Props}
 import akka.actor.Timers
 import akka.http.scaladsl.model.RemoteAddress
 import de.upb.cs.swt.delphi.webapi.ElasticRequestLimiter._
+import de.upb.cs.swt.delphi.webapi.ElasticActorManager.ElasticMessage
 
 import scala.concurrent.duration._
 import scala.collection.mutable