Add thread pool settings (#308)

* Add thread pool settings

* Scalafmt

Author: ponkotuy

doc/20-configuration.md

@@ -227,6 +227,18 @@ usually does not apply. However, it is expected the cache should be fast respond
 and thus **by default the timeout is set to 500 millis** to avoid unnecessary delays.
 This timeout is optional and can be disabled.
 
+## ThreadPool
+
+These are ResourceClient settings passed to Lettuce, a Java Redis client library. For more information see
+
+https://redis.github.io/lettuce/advanced-usage/
+
+```hocon
+play.cache.redis {
+  io-thread-pool-size:          8 // default 8, min 3
+  computation-thread-pool-size: 8 // default 8, min 3
+}
+```
 ## Recovery policy
 
 The intention of cache is usually to optimize the application behavior, not to provide any business logic.

src/main/scala/play/api/cache/redis/configuration/RedisSettings.scala

@@ -26,6 +26,9 @@ trait RedisSettings {
 
   /** instance prefix */
   def prefix: Option[String]
+
+  /** threadpool */
+  def threadPool: RedisThreadPools
   // $COVERAGE-OFF$
   /** trait-specific equals */
   override def equals(obj: scala.Any): Boolean = equalsAsSettings(obj)
@@ -49,6 +52,7 @@ object RedisSettings extends ConfigLoader[RedisSettings] {
     timeout = loadTimeouts(config, path)(RedisTimeouts.requiredDefault),
     source = loadSource(config, path).get,
     prefix = loadPrefix(config, path),
+    threadPool = loadThreadPool(config, path)(RedisThreadPools.requiredDefault),
   )
 
   def withFallback(fallback: RedisSettings): ConfigLoader[RedisSettings] =
@@ -60,19 +64,21 @@ object RedisSettings extends ConfigLoader[RedisSettings] {
         timeout = loadTimeouts(config, path)(fallback.timeout),
         source = loadSource(config, path) getOrElse fallback.source,
         prefix = loadPrefix(config, path) orElse fallback.prefix,
+        threadPool = loadThreadPool(config, path)(fallback.threadPool),
       )
 
-  def apply(dispatcher: String, invocationPolicy: String, timeout: RedisTimeouts, recovery: String, source: String, prefix: Option[String] = None): RedisSettings =
-    create(dispatcher, invocationPolicy, prefix, timeout, recovery, source)
+  def apply(dispatcher: String, invocationPolicy: String, timeout: RedisTimeouts, recovery: String, source: String, prefix: Option[String] = None, threadPool: RedisThreadPools): RedisSettings =
+    create(dispatcher, invocationPolicy, prefix, timeout, recovery, source, threadPool)
 
   @inline
-  private def create(_dispatcher: String, _invocation: String, _prefix: Option[String], _timeout: RedisTimeouts, _recovery: String, _source: String) = new RedisSettings {
+  private def create(_dispatcher: String, _invocation: String, _prefix: Option[String], _timeout: RedisTimeouts, _recovery: String, _source: String, _threadpool: RedisThreadPools) = new RedisSettings {
     override val invocationContext: String = _dispatcher
     override val invocationPolicy: String = _invocation
     override val prefix: Option[String] = _prefix
     override val recovery: String = _recovery
     override val timeout: RedisTimeouts = _timeout
     override val source: String = _source
+    override val threadPool: RedisThreadPools = _threadpool
   }
 
   private def loadInvocationContext(config: Config, path: String): Option[String] =
@@ -93,6 +99,9 @@ object RedisSettings extends ConfigLoader[RedisSettings] {
   private def loadTimeouts(config: Config, path: String)(defaults: RedisTimeouts): RedisTimeouts =
     RedisTimeouts.load(config, path)(defaults)
 
+  private def loadThreadPool(config: Config, path: String)(defaults: RedisThreadPools): RedisThreadPools =
+    RedisThreadPools.load(config, path)(defaults)
+
 }
 
 /** A helper trait delegating properties into the inner settings object */
@@ -104,4 +113,5 @@ trait RedisDelegatingSettings extends RedisSettings {
   override def recovery: String = settings.recovery
   override def invocationContext: String = settings.invocationContext
   override def invocationPolicy: String = settings.invocationPolicy
+  override def threadPool: RedisThreadPools = settings.threadPool
 }

src/main/scala/play/api/cache/redis/configuration/RedisThreadPools.scala

@@ -0,0 +1,48 @@
+package play.api.cache.redis.configuration
+
+import com.typesafe.config.Config
+import play.api.cache.redis._
+
+trait RedisThreadPools {
+  def ioSize: Int
+  def computationSize: Int
+}
+
+final case class RedisThreadPoolsImpl(
+  ioSize: Int,
+  computationSize: Int,
+) extends RedisThreadPools {
+
+  // $COVERAGE-OFF$
+  override def equals(obj: scala.Any): Boolean = obj match {
+    case that: RedisThreadPools => this.ioSize === that.ioSize && this.computationSize === that.computationSize
+    case _                      => false
+  }
+  // $COVERAGE-ON$
+
+}
+
+object RedisThreadPools {
+  import RedisConfigLoader._
+
+  def requiredDefault: RedisThreadPools = new RedisThreadPools {
+    override def ioSize: Int = 8
+    override def computationSize: Int = 8
+  }
+
+  @inline
+  def apply(ioSize: Int, computationSize: Int): RedisThreadPools =
+    RedisThreadPoolsImpl(ioSize, computationSize)
+
+  def load(config: Config, path: String)(default: RedisThreadPools): RedisThreadPools = RedisThreadPools(
+    ioSize = loadIoSize(config, path) getOrElse default.ioSize,
+    computationSize = loadComputationSize(config, path) getOrElse default.computationSize,
+  )
+
+  private def loadIoSize(config: Config, path: String): Option[Int] =
+    config.getOption(path / "io-thread-pool-size", _.getInt)
+
+  private def loadComputationSize(config: Config, path: String): Option[Int] =
+    config.getOption(path / "computation-thread-pool-size", _.getInt)
+
+}

src/main/scala/play/api/cache/redis/connector/RedisCommands.scala

@@ -50,7 +50,12 @@ abstract private[connector] class AbstractRedisCommands(
 
   protected def connectionString: String
 
-  protected lazy val resources: ClientResources = RedisClientFactory.newClientResources()
+  protected def configuration: RedisSettings
+
+  protected lazy val resources: ClientResources = RedisClientFactory.newClientResources(
+    ioThreadPoolSize = configuration.threadPool.ioSize,
+    computationThreadPoolSize = configuration.threadPool.computationSize,
+  )
 
   protected def client: AbstractRedisClient
 
@@ -102,7 +107,7 @@ abstract private[connector] class AbstractRedisCommands(
   *   configures clusters
   */
 private[connector] class RedisCommandsStandalone(
-  configuration: RedisStandalone,
+  override val configuration: RedisStandalone,
 )(implicit
   executionContext: ExecutionContext,
   lifecycle: ApplicationLifecycle,
@@ -147,7 +152,7 @@ private[connector] class RedisCommandsStandalone(
   *   configures clusters
   */
 private[connector] class RedisCommandsCluster(
-  configuration: RedisCluster,
+  override val configuration: RedisCluster,
 )(implicit
   lifecycle: ApplicationLifecycle,
   executionContext: ExecutionContext,
@@ -199,7 +204,7 @@ private[connector] class RedisCommandsCluster(
   *   configures sentinels
   */
 private[connector] class RedisCommandsSentinel(
-  configuration: RedisSentinel,
+  override val configuration: RedisSentinel,
 )(implicit
   lifecycle: ApplicationLifecycle,
   executionContext: ExecutionContext,
@@ -248,7 +253,7 @@ private[connector] class RedisCommandsSentinel(
   */
 //noinspection DuplicatedCode
 private[connector] class RedisCommandsMasterSlaves(
-  configuration: RedisMasterSlaves,
+  override val configuration: RedisMasterSlaves,
 )(implicit
   lifecycle: ApplicationLifecycle,
   executionContext: ExecutionContext,

src/test/scala/play/api/cache/redis/RedisCacheModuleSpec.scala

@@ -1,6 +1,6 @@
 package play.api.cache.redis
 
-import play.api.cache.redis.configuration.{RedisHost, RedisSettings, RedisStandalone, RedisTimeouts}
+import play.api.cache.redis.configuration.{RedisHost, RedisSettings, RedisStandalone, RedisThreadPools, RedisTimeouts}
 import play.api.cache.redis.test._
 import play.api.inject._
 import play.api.inject.guice.GuiceApplicationBuilder
@@ -138,6 +138,7 @@ class RedisCacheModuleSpec extends IntegrationSpec with RedisStandaloneContainer
         recovery = "log-and-default",
         source = "my-instance",
         prefix = None,
+        threadPool = RedisThreadPools(ioSize = 1, computationSize = 1),
       ),
     )
 

src/test/scala/play/api/cache/redis/configuration/RedisInstanceManagerSpec.scala

@@ -58,6 +58,7 @@ class RedisInstanceManagerSpec extends UnitSpec with ImplicitOptionMaterializati
       recovery = "log-and-fail",
       source = "standalone",
       prefix = "redis.",
+      threadPool = RedisThreadPools(1, 1),
     )
 
     manager mustEqual RedisInstanceManagerTest(defaultCacheName)(
@@ -107,6 +108,7 @@ class RedisInstanceManagerSpec extends UnitSpec with ImplicitOptionMaterializati
       recovery = "log-and-fail",
       source = "standalone",
       prefix = "redis.",
+      threadPool = RedisThreadPools(1, 1),
     )
 
     private val defaultCache: RedisInstanceProvider = RedisStandalone(defaultCacheName, RedisHost(localhost, defaultPort, database = 1), otherSettings)

src/test/scala/play/api/cache/redis/configuration/RedisThreadPoolsSpec.scala

@@ -0,0 +1,36 @@
+package play.api.cache.redis.configuration
+
+import play.api.cache.redis.test.{Helpers, ImplicitOptionMaterialization, UnitSpec}
+
+class RedisThreadPoolsSpec extends UnitSpec with ImplicitOptionMaterialization {
+
+  private def orDefault = RedisThreadPools(8, 8)
+
+  "load defined thread pools" in {
+    val configuration = Helpers.configuration.fromHocon {
+      """
+        |play.cache.redis {
+        |
+        |  io-thread-pool-size:          5
+        |  computation-thread-pool-size: 7
+        |}
+      """.stripMargin
+    }
+    val expected = RedisThreadPools(5, 7)
+    val actual = RedisThreadPools.load(configuration.underlying, "play.cache.redis")(RedisThreadPools.requiredDefault)
+    actual mustEqual expected
+  }
+
+  "load with default thread pools" in {
+    val configuration = Helpers.configuration.fromHocon {
+      """
+        |play.cache.redis {
+        |}
+      """.stripMargin
+    }
+    val expected = RedisThreadPools(8, 8)
+    val actual = RedisThreadPools.load(configuration.underlying, "play.cache.redis")(orDefault)
+    actual mustEqual expected
+  }
+
+}

src/test/scala/play/api/cache/redis/test/BaseSpec.scala

@@ -28,6 +28,7 @@ trait DefaultValues {
       timeout = RedisTimeouts(1.second, None, Some(500.millis)),
       recovery = "log-and-default",
       source = "standalone",
+      threadPool = RedisThreadPools(1, 1),
     )
 
   final protected val cacheKey: String = "cache-key"

src/test/scala/play/api/cache/redis/test/RedisSettingsTest.scala

@@ -9,4 +9,5 @@ final case class RedisSettingsTest(
   recovery: String,
   source: String,
   prefix: Option[String] = None,
+  threadPool: RedisThreadPools,
 ) extends RedisSettings