Added BlockingExecutor + separate executor for callbacks

Alexandre Curreli · Alexandre Curreli · commit fb1968dfd2d6 · 2014-05-05T13:49:15.000-04:00
diff --git a/build.sbt b/build.sbt
@@ -6,7 +6,7 @@ organization := "com.livestream"
 
 name := "scredis"
 
-version := "1.1.1"
+version := "1.1.2"
 
 scalaVersion := "2.9.3"
 
diff --git a/src/main/resources/reference.conf b/src/main/resources/reference.conf
@@ -128,41 +128,79 @@ scredis {
     # This parameter has no effect unless auto-pipeline is enabled.
     timer-thread-naming-pattern = "scredis-$p-pipeliner"
     
-    # The default execution context is a FixedThreadPool. The following parameters have no
-    # effect if a custom execution context is provided with the Redis.withExecutionContext()
-    # factory method.
+    # The default executors are CachedThreadPools with a maximum number of concurrent tasks.
     executors {
-      # Sets the number of threads in the fixed thread pool.
-      threads = 10
       
-      # Defines the maximum number of asynchronous tasks that can reside in the queue before
-      # the executor starts blocking. This prevents OutOfMemory exceptions to occur under super
-      # high load scenarios where commands are queued faster than processed.
-      queue-capacity = 10000
+      # The internal thread pool is the one scredis uses to run asynchronous commands.
+      # The following parameters have no effect if a custom execution context is provided with
+      # the Redis.withExecutionContext() factory method.
+      internal {
+        # Defines the maximum number of asynchronous tasks (single command or pipeline) that can run
+        # simultaneously before the executor starts blocking. This prevents OutOfMemory exceptions
+        # to occur under super high load scenarios where commands are queued faster than processed.
+        # This parameter basically determines the maximum number of threads that can be created.
+        max-concurrent = 8
+        
+        # Sets the name of the threads in the thread pool.
+        # The "$p" (pool number) and "$t" (thread number) variable can be used in the pattern and
+        # will be replaced at runtime.
+        threads-naming-pattern = "scredis-$p-worker-$t"
+        
+        # Determines whether all internal threads should be created as daemon threads.
+        #
+        # Daemon threads cannot prevent an application from terminating. When all non-daemon
+        # threads (such as the main thread) terminate, all daemon threads are automatically
+        # terminated and the application itself terminates. As a consequence, if Redis.quit()
+        # is not invoked before the application terminates, all queued or processing commands will
+        # be dropped.
+        #
+        # In contrast, non-daemon threads will prevent the application from terminating unless there
+        # is an explicit invocation of the Redis.quit() method (which will terminate all internal
+        # threads) or if the application is shutdown using System.exit().
+        daemon-threads = true
+        
+        # Sets the priority of all threads in the thread pool.
+        # "min" - lowest priority
+        # "normal" - normal priority
+        # "max" - highest priority
+        threads-priority = normal
+      }
       
-      # Sets the name of the threads in the thread pool.
-      # The "$p" (pool number) and "$t" (thread number) variable can be used in the pattern and
-      # will be replaced at runtime.
-      threads-naming-pattern = "scredis-$p-worker-$t"
-      
-      # Determines whether all internal threads should be created as daemon threads.
-      #
-      # Daemon threads cannot prevent an application from terminating. When all non-daemon
-      # threads (such as the main thread) terminate, all daemon threads are automatically
-      # terminated and the application itself terminates. As a consequence, if Redis.quit()
-      # is not invoked before the application terminates, all queued or processing commands will
-      # be dropped.
-      #
-      # In contrast, non-daemon threads will prevent the application from terminating unless there
-      # is an explicit invocation of the Redis.quit() method (which will terminate all internal
-      # threads) or if the application is shutdown using System.exit().
-      daemon-threads = true
-      
-      # Sets the priority of all threads in the thread pool.
-      # "min" - lowest priority
-      # "normal" - normal priority
-      # "max" - highest priority
-      threads-priority = normal
+      # The callback thread pool is used when registering callbacks on asynchronous commands.
+      # It is an implicit public member of the Redis instance, i.e. Redis.ec and can be imported
+      # as follows "import redis.ec". Note that the callback executor is lazily defined, meaning
+      # that it will only be initialized (and shutdown) if it is explicitly imported.
+      callback {
+        # Defines the maximum number of asynchronous tasks (single command or pipeline) that can run
+        # simultaneously before the executor starts blocking. This prevents OutOfMemory exceptions
+        # to occur under super high load scenarios where commands are queued faster than processed.
+        # This parameter basically determines the maximum number of threads that can be created.
+        max-concurrent = 16
+        
+        # Sets the name of the threads in the thread pool.
+        # The "$p" (pool number) and "$t" (thread number) variable can be used in the pattern and
+        # will be replaced at runtime.
+        threads-naming-pattern = "scredis-$p-callback-$t"
+        
+        # Determines whether all internal threads should be created as daemon threads.
+        #
+        # Daemon threads cannot prevent an application from terminating. When all non-daemon
+        # threads (such as the main thread) terminate, all daemon threads are automatically
+        # terminated and the application itself terminates. As a consequence, if Redis.quit()
+        # is not invoked before the application terminates, all queued or processing commands will
+        # be dropped.
+        #
+        # In contrast, non-daemon threads will prevent the application from terminating unless there
+        # is an explicit invocation of the Redis.quit() method (which will terminate all internal
+        # threads) or if the application is shutdown using System.exit().
+        daemon-threads = true
+        
+        # Sets the priority of all threads in the thread pool.
+        # "min" - lowest priority
+        # "normal" - normal priority
+        # "max" - highest priority
+        threads-priority = normal
+      }
     }
   }
 
diff --git a/src/main/scala/scredis/Redis.scala b/src/main/scala/scredis/Redis.scala
@@ -26,6 +26,7 @@ import akka.util.duration._
 
 import scredis.commands.async._
 import scredis.exceptions._
+import scredis.util.BlockingExecutor
 
 import scala.collection.mutable.ListBuffer
 
@@ -139,19 +140,36 @@ final class Redis private[scredis] (
   private val timerTask = if(IsAutomaticPipeliningEnabled) Some((
     new Timer(
       config.Async.TimerThreadNamingPattern.replace("$p", PoolNumber),
-      config.Async.Executors.DaemonThreads
+      true
     ),
     new TimerTask() {
       def run(): Unit = executePipelineIfNeeded()
     }
   )) else None
   
-  private val executorOpt = ecOpt match {
+  private val internalExecutorOpt = ecOpt match {
     case Some(_) => None
-    case None => Some(newThreadPool())
+    case None => Some(
+      newThreadPool(
+        config.Async.Executors.Internal.ThreadsNamingPattern,
+        config.Async.Executors.Internal.DaemonThreads,
+        config.Async.Executors.Internal.ThreadsPriority
+      )
+    )
   }
+  private val interalExecutionContext = ecOpt.getOrElse(
+    BlockingExecutor(internalExecutorOpt.get, config.Async.Executors.Internal.MaxConcurrent)
+  )
   
-  implicit val ec = ecOpt.getOrElse(ExecutionContext.fromExecutorService(executorOpt.get))
+  private var shouldShutdownCallbackExecutor = false
+  private lazy val callbackExecutor: ExecutorService = {
+    shouldShutdownCallbackExecutor = true
+    newThreadPool(
+      config.Async.Executors.Callback.ThreadsNamingPattern,
+      config.Async.Executors.Callback.DaemonThreads,
+      config.Async.Executors.Callback.ThreadsPriority
+    )
+  }
   
   private val pool = ClientPool(config)
 
@@ -170,23 +188,26 @@ final class Redis private[scredis] (
     config.Client.Sleep
   )
   
-  private def newThreadPool(): ExecutorService = {
+  lazy implicit val ec = ecOpt.getOrElse(
+    BlockingExecutor(callbackExecutor, config.Async.Executors.Callback.MaxConcurrent)
+  )
+  
+  private def execute[A](f: => A): Future[A] = Future {
+    f
+  }(interalExecutionContext)
+  
+  private def newThreadPool(
+    namingPattern: String, daemon: Boolean, priority: Int
+  ): ExecutorService = {
     val threadFactory = new BasicThreadFactory.Builder()
       .namingPattern(
-        config.Async.Executors.ThreadsNamingPattern.replace("$p", PoolNumber).replace("$t", "%d")
+        namingPattern.replace("$p", PoolNumber).replace("$t", "%d")
       )
-      .daemon(config.Async.Executors.DaemonThreads)
-      .priority(config.Async.Executors.ThreadsPriority)
+      .daemon(daemon)
+      .priority(priority)
       .build()
     
-    new ThreadPoolExecutor(
-      config.Async.Executors.Threads,
-      config.Async.Executors.Threads,
-      0L,
-      TimeUnit.MILLISECONDS,
-      new LinkedBlockingQueue[Runnable](config.Async.Executors.QueueCapacity),
-      threadFactory
-    )
+    Executors.newCachedThreadPool(threadFactory)
   }
   
   private def nextIndex(): Int = synchronized {
@@ -273,7 +294,7 @@ final class Redis private[scredis] (
           -1
         }
       }
-      if(executeIndex >= 0) Future { executePipeline(executeIndex) }
+      if(executeIndex >= 0) execute { executePipeline(executeIndex) }
     } catch {
       case e: Throwable => logger.error("An unexpected error occurred", e)
     } finally {
@@ -293,7 +314,11 @@ final class Redis private[scredis] (
       val (future, executeIndex) = synchronized {
         val commands = this.commands.get(index)
         val future = try {
-          val promise = Promise[Any]()
+          /*
+           * Akka promises need explicit execution context.
+           * Cannot re-use internal execution context otherwise it can create deadlocks.
+           */
+          val promise = Promise[Any]()(ec)
           commands += ((body, opts, promise))
           promise.future.asInstanceOf[Future[A]]
         } catch {
@@ -311,7 +336,7 @@ final class Redis private[scredis] (
         }
         (future, executeIndex)
       }
-      if(executeIndex >= 0) Future { executePipeline(executeIndex) }
+      if(executeIndex >= 0) execute { executePipeline(executeIndex) }
       future
     } else {
       withClientAsync(body)
@@ -335,7 +360,7 @@ final class Redis private[scredis] (
   def borrowClient(): Client = pool.borrowClient()
   def returnClient(client: Client): Unit = pool.returnClient(client)
   def withClient[A](body: Client => A): A = pool.withClient(body)
-  def withClientAsync[A](body: Client => A): Future[A] = Future {
+  def withClientAsync[A](body: Client => A): Future[A] = execute {
     withClient(body)
   }
   
@@ -357,7 +382,7 @@ final class Redis private[scredis] (
    */
   override def select(
     db: Int
-  )(implicit opts: CommandOptions = DefaultCommandOptions): Future[Unit] = Future {
+  )(implicit opts: CommandOptions = DefaultCommandOptions): Future[Unit] = execute {
     selectInternal(db)
   }
   
@@ -375,7 +400,7 @@ final class Redis private[scredis] (
    */
   override def auth(password: String)(
     implicit opts: CommandOptions = DefaultCommandOptions
-  ): Future[Unit] = Future {
+  ): Future[Unit] = execute {
     authInternal(if(password.isEmpty) None else Some(password))
   }
   
@@ -420,7 +445,7 @@ final class Redis private[scredis] (
     }
     
     try {
-      executorOpt.foreach { executor =>
+      internalExecutorOpt.foreach { executor =>
         executor.shutdown()
         awaitTerminationOpt.foreach { timeout =>
           if (timeout.isFinite) {
@@ -432,9 +457,18 @@ final class Redis private[scredis] (
       }
     } catch {
       case e: Throwable => logger.error(
-        "An error occurred while shutting down default executor", e
+        "An error occurred while shutting down internal executor", e
       )
     }
+    if (shouldShutdownCallbackExecutor) {
+      try {
+        callbackExecutor.shutdown()
+      } catch {
+        case e: Throwable => logger.error(
+          "An error occurred while shutting down callback executor", e
+        )
+      }
+    }
     try {
       pool.close()
     } catch {
@@ -444,7 +478,9 @@ final class Redis private[scredis] (
     }
   }
   
-  if(IsAutomaticPipeliningEnabled) nextIndex()
+  if (IsAutomaticPipeliningEnabled) {
+    nextIndex()
+  }
   
   timerTask.foreach {
     case (timer, task) => timer.scheduleAtFixedRate(task, Interval.toMillis, Interval.toMillis)
diff --git a/src/main/scala/scredis/RedisConfig.scala b/src/main/scala/scredis/RedisConfig.scala
@@ -177,17 +177,33 @@ class RedisConfig(config: Config = ConfigFactory.load().getConfig("scredis")) {
     val TimerThreadNamingPattern = async.getString("timer-thread-naming-pattern")
     
     object Executors {
-      val Threads = asyncExecutors.getInt("threads")
-      val QueueCapacity = asyncExecutors.getInt("queue-capacity")
-      val ThreadsNamingPattern = asyncExecutors.getString("threads-naming-pattern")
-      val DaemonThreads = asyncExecutors.getBoolean("daemon-threads")
-      val ThreadsPriority = asyncExecutors.getString("threads-priority").toLowerCase match {
-        case "min"    => Thread.MIN_PRIORITY
-        case "normal" => Thread.NORM_PRIORITY
-        case "max"    => Thread.MAX_PRIORITY
-        case _        => throw new IllegalArgumentException(
-          "threads-priority can only be min, normal or max"
-        )
+      object Internal {
+        private val config = asyncExecutors.getConfig("internal")
+        val MaxConcurrent = config.getInt("max-concurrent")
+        val ThreadsNamingPattern = config.getString("threads-naming-pattern")
+        val DaemonThreads = config.getBoolean("daemon-threads")
+        val ThreadsPriority = config.getString("threads-priority").toLowerCase match {
+          case "min"    => Thread.MIN_PRIORITY
+          case "normal" => Thread.NORM_PRIORITY
+          case "max"    => Thread.MAX_PRIORITY
+          case _        => throw new IllegalArgumentException(
+            "threads-priority can only be min, normal or max"
+          )
+        }
+      }
+      object Callback {
+        private val config = asyncExecutors.getConfig("callback")
+        val MaxConcurrent = config.getInt("max-concurrent")
+        val ThreadsNamingPattern = config.getString("threads-naming-pattern")
+        val DaemonThreads = config.getBoolean("daemon-threads")
+        val ThreadsPriority = config.getString("threads-priority").toLowerCase match {
+          case "min"    => Thread.MIN_PRIORITY
+          case "normal" => Thread.NORM_PRIORITY
+          case "max"    => Thread.MAX_PRIORITY
+          case _        => throw new IllegalArgumentException(
+            "threads-priority can only be min, normal or max"
+          )
+        }
       }
     }
   }
diff --git a/src/main/scala/scredis/commands/KeysCommands.scala b/src/main/scala/scredis/commands/KeysCommands.scala
@@ -21,7 +21,6 @@ import scredis.CommandOptions
 import scredis.protocol.Protocol
 import scredis.parsing._
 import scredis.parsing.Implicits._
-import scredis.util.ScanLikeIterator
 
 import scala.collection.mutable.ListBuffer
 
diff --git a/src/main/scala/scredis/commands/async/TransactionalCommands.scala b/src/main/scala/scredis/commands/async/TransactionalCommands.scala
@@ -32,8 +32,6 @@ import scala.collection.generic.CanBuildFrom
  * @define m [[scredis.TransactionalClient]]
  */
 trait TransactionalCommands extends Async {
-
-  protected implicit val ec: ExecutionContext
   
   protected def force(opts: CommandOptions): CommandOptions = CommandOptions(
     opts.timeout,
@@ -91,7 +89,8 @@ trait TransactionalCommands extends Async {
    */
   def pipelinedN[A, B[_] <: Traversable[_]](body: PipelineClient => B[Future[A]])(
     implicit opts: CommandOptions = DefaultCommandOptions,
-    cbf: CanBuildFrom[B[Future[A]], A, B[A]]
+    cbf: CanBuildFrom[B[Future[A]], A, B[A]],
+    ec: ExecutionContext
   ): Future[B[A]] = async(_.pipelinedN(body))(force(opts))
   
   /**
@@ -164,7 +163,8 @@ trait TransactionalCommands extends Async {
    */
   def transactionalN[A, B[_] <: Traversable[_]](body: TransactionalClient => B[Future[A]])(
     implicit opts: CommandOptions = DefaultCommandOptions,
-    cbf: CanBuildFrom[B[Future[A]], A, B[A]]
+    cbf: CanBuildFrom[B[Future[A]], A, B[A]],
+    ec: ExecutionContext
   ): Future[B[A]] = async(_.transactionalN(body))(force(opts))
   
   /**
diff --git a/src/main/scala/scredis/util/BlockingExecutor.scala b/src/main/scala/scredis/util/BlockingExecutor.scala
diff --git a/src/main/scala/scredis/util/ScanLikeIterator.scala b/src/main/scala/scredis/util/ScanLikeIterator.scala
