Query dispatcher

Author: jmarin

publication/src/test/scala/hmda/publication/reports/ReportGenerators.scala

@@ -1,7 +1,6 @@
 package hmda.publication.reports
 
 import hmda.model.census.Census
-import hmda.model.publication.reports.MinorityStatusEnum.WhiteNonHispanic
 import hmda.model.publication.reports._
 import org.scalacheck.Gen
 

query/src/main/resources/application.conf

@@ -75,3 +75,20 @@ cassandra {
   retries = 60
   retry-interval = 1000
 }
+
+query-dispatcher {
+  type = Dispatcher
+  executor = "fork-join-executor"
+  fork-join-executor {
+    # Min number of threads to cap factor-based parallelism number to
+    parallelism-min = 2
+    # Parallelism (threads) ... ceil(available processors * factor)
+    parallelism-factor = 2.0
+    # Max number of threads to cap factor-based parallelism number to
+    parallelism-max = 10
+  }
+  # Throughput defines the maximum number of messages to be
+  # processed per actor before the thread jumps to the next actor.
+  # Set to 1 for as fair as possible.
+  throughput = 100
+}

query/src/main/scala/hmda/query/HmdaQuerySupervisor.scala

@@ -39,7 +39,7 @@ class HmdaQuerySupervisor extends HmdaSupervisorActor {
 
   override protected def createActor(name: String): ActorRef = name match {
     case id @ InstitutionView.name =>
-      val actor = context.actorOf(InstitutionView.props(), id)
+      val actor = context.actorOf(InstitutionView.props().withDispatcher("query-dispatcher"), id)
       supervise(actor, id)
   }
 
@@ -49,7 +49,7 @@ class HmdaQuerySupervisor extends HmdaSupervisorActor {
 
   private def createHmdaFilingView(period: String)(implicit ec: ExecutionContext): ActorRef = {
     val id = s"${HmdaFilingView.name}-$period"
-    val actor = context.actorOf(HmdaFilingView.props(period), id)
+    val actor = context.actorOf(HmdaFilingView.props(period).withDispatcher("query-dispatcher"), id)
     for {
       p <- (actor ? GetProjectionActorRef).mapTo[ActorRef]
     } yield {

query/src/main/scala/hmda/query/view/filing/HmdaFilingView.scala

@@ -41,7 +41,9 @@ class HmdaFilingView(period: String) extends HmdaPersistentActor {
 
   var counter = 0
 
-  val queryProjector = context.actorOf(HmdaFilingDBProjection.props(period), "queryProjector")
+  val queryProjector = context
+    .actorOf(HmdaFilingDBProjection.props(period)
+      .withDispatcher("query-dispatcher"), "queryProjector")
 
   val conf = ConfigFactory.load()
   val snapshotCounter = conf.getInt("hmda.journal.snapshot.counter")

query/src/main/scala/hmda/query/view/institutions/InstitutionView.scala

@@ -50,7 +50,9 @@ class InstitutionView extends HmdaPersistentActor {
 
   var counter = 0
 
-  val queryProjector = context.actorOf(InstitutionDBProjection.props(), "institution-projection")
+  val queryProjector = context
+    .actorOf(InstitutionDBProjection.props()
+      .withDispatcher("query-dispatcher"), "institution-projection")
 
   val snapshotCounter = configuration.getInt("hmda.journal.snapshot.counter")