BlockingExecutionContext.scala

/**
    * Create an ExecutionContext for blocking operations that grows with demand
    *
    * @param threadNamePrefix the prefix to put at the beginning of the name of threads for this pool
    * @param unhandledExceptionLogger a function to log unhandled exceptions
    * @param minPoolSize the minimum number of threads to allocate for this pool
    * @param maxPoolSize the maximum allowed number of threads to allocate for this pool (requests beyond
    *                    this maximum will block the thread of the requester)
    * @param keepAliveTime the amount of time to keep an unused thread alive before destorying it
    * @return an ExecutionContext for blocking operations
    */
def mkBlockingExecutionContext(
     threadNamePrefix: String,
     unhandledExceptionLogger: Throwable => Unit,
     minPoolSize: Int = 0,
     maxPoolSize: Int = 46034, // 75% of 61379 (max ubuntu threads) `cat /proc/sys/kernel/threads-max`
     keepAliveTime: Duration = 60.seconds
  ) : ExecutionContext = {
    val threadFactory =
      new ThreadFactoryBuilder()
        .setNameFormat(s"$threadNamePrefix-blocking-%d")
        .build()
    val threadPoolExecutor =
      new ThreadPoolExecutor(
        /* corePoolSize */ minPoolSize,
        /* maximumPoolSize */ maxPoolSize,
        /* keepAliveTime */ keepAliveTime.toMillis, TimeUnit.MILLISECONDS,
        /* workQueue */ new SynchronousQueue[Runnable](),
        threadFactory
      )
    ExecutionContext.fromExecutor(threadPoolExecutor,unhandledExceptionLogger)
  }