Customize blocking thread pool

[wb14123]

Is there anyway to override blocking thread pool in IOApp without override IORuntime? I want the blocking thread pool to have a limited size and larger caching time.

[armanbilge]

It's not, but overriding the runtime in IOApp is easy. Something like this I believe:

override def runtime = IORuntime.builder().setBlocking(...).build()

[wb14123]

Thanks! This is much simpler than what I copied from IOApp. I saw in the comment of IOApp:

In other words, this method is made available to users, but its use is strongly discouraged in favor of other, more precise solutions to specific use-cases.

So I just want to make sure there is no other better way to do it.

[wb14123]

@armanbilge I tried to override runtime like this with thread name io-blocking-custom-*:

object FetchServer extends IOApp {

  override def runtime: IORuntime = {
    val (blocking, blockDown) = createCustomBlockingExecutionContext()
    IORuntime.builder().setBlocking(blocking, blockDown).build()
  }

  private def createCustomBlockingExecutionContext(): (ExecutionContext, () => Unit) = {
    val threadCount = new AtomicInteger(0)
    val executor = Executors.newFixedThreadPool(16,  { (r: Runnable) =>
      val t = new Thread(r)
      t.setName(s"io-blocking-custom-${threadCount.getAndIncrement()}")
      t.setDaemon(true)
      t
    })
    (ExecutionContext.fromExecutor(executor), { () => executor.shutdown() })
  }

  override def run(args: List[String]): IO[ExitCode] = {
    BaseServer().evalMap { baseServer =>
      MetricServer.start() >> baseServer.fetcher.run()
    }.useForever
  }

}

But I still see blocking threads named io-compute-blocker-* being created. Am I doing anything wrong?

[armanbilge]

But I still see blocking threads named io-compute-blocker-* being created. Am I doing anything wrong?

Ah, no you are not doing anything wrong. Because the WorkStealingThreadPool supports blocking itself, it actually doesn't touch the (customized) blocking pool at all, it uses its own internal mechanisms. Hmm, I didn't really think through this, sorry.

cats-effect/core/shared/src/main/scala/cats/effect/IOFiber.scala

Lines 978 to 1001 in a11d81a

	if (cur.hint eq IOFiber.TypeBlocking) {
	val ec = currentCtx
	if (ec.isInstanceOf[WorkStealingThreadPool[_]]) {
	val wstp = ec.asInstanceOf[WorkStealingThreadPool[_]]
	if (wstp.canExecuteBlockingCode()) {
	var error: Throwable = null
	val r =
	try {
	scala.concurrent.blocking(cur.thunk())
	} catch {
	case t if NonFatal(t) =>
	error = t
	case t: Throwable =>
	onFatalFailure(t)
	}
	val next = if (error eq null) succeeded(r, 0) else failed(error, 0)
	runLoop(next, nextCancelation, nextAutoCede)
	} else {
	blockingFallback(cur)
	}
	} else {
	blockingFallback(cur)
	}

A workaround for this would be to replace the IORuntime's compute e.g.

IORuntime.builder().transformCompute { ec =>
  new ExecutionContext {
    def execute(r: Runnable) = ec.execute(r)
  }
}

This will cause the ec.isInstanceOf[WorkStealingThreadPool] check to fail and force the use of the fallback, but this will cause other de-optimizations.

---

I want the blocking thread pool to have a limited size and larger caching time.

We can expose a configuration for caching time of the WTSP's blocking mechanism. We currently have no way to limit the size, however.

We could also add a configuration to disable the WSTP's internal blocking mechanism and always use the fallback blocking pool, but this is generally a de-optimization.

[armanbilge]

We can expose a configuration for caching time of the WTSP's blocking mechanism.

I forgot that we've already exposed this. So it's already possible to configure this.

cats-effect/core/jvm/src/main/scala/cats/effect/unsafe/IORuntimeCompanionPlatform.scala

Lines 53 to 59 in 22cc77b

	def createWorkStealingComputeThreadPool(
	threads: Int = Math.max(2, Runtime.getRuntime().availableProcessors()),
	threadPrefix: String = "io-compute",
	blockerThreadPrefix: String = DefaultBlockerPrefix,
	runtimeBlockingExpiration: Duration = 60.seconds,
	reportFailure: Throwable => Unit = _.printStackTrace(),
	blockedThreadDetectionEnabled: Boolean = false): (WorkStealingThreadPool, () => Unit) = {

[wb14123]

@armanbilge Thanks for the info! If that's the case, I think maybe I will just use unlimited blocking threads for now. But I think it's a useful use case to limit the thread size, since if some code (unintentionally) generates lots of slow blocking operations, it may blow up the threads and make the whole system unresponsive. It's better to limit the thread size so that developer has a working system to debug what is going on.

I see there is a blockedWorkerThreadCounter in WorkStealingThreadPool. Can we add a configuration to limit it? And if it's full, put the pending job into a queue like the implementation of ThreadPoolExecutor. Will there be any draw back with this approach?

Additionally, maybe we should update the doc at https://typelevel.org/cats-effect/docs/thread-model#blocking? Since there is no mentioning of such a behavior for WTSP. It only says "CE3 has a builtin blocking which will shift execution to an internal blocking threadpool and shift it back afterwards using Async."

[armanbilge]

Can we add a configuration to limit it? And if it's full, put the pending job into a queue like the implementation of ThreadPoolExecutor. Will there be any draw back with this approach?

Have you read the excellent Tuning Guide? It explains how to address this in userland without limiting the number of blocking threads.

Realistically, you can't do that many concurrent blocking operations before it starts to become a performance problem. For that reason, it is worth using techniques similar to the ones we examined in the Restrict Work section: a global Semaphore and/or a concurrent work Queue.

You can use this technique today and it doesn't require any changes to the WSTP in Cats Effect.

---

Additionally, maybe we should update the doc at https://typelevel.org/cats-effect/docs/thread-model#blocking?

Hmm, yes, that entire page is very out-of-date 😅 As of CE 3.6 we have significantly revamped the entire thread model.

[wb14123]

Yes I'm aware of that guide and it's well written. However those approaches only applies when we know where the blocking operations come from. That requires a good understanding of the blocking operations including the ones in the libraries (for example in quill-doobie, I don't know why it generates blocking operations even I'm already using Hikari transactor, which has a thread pool itself). I think it's safer to have a global thread limit so it doesn't affect the whole OS and let user have an opportunity to debug it when something goes wrong.

Regarding the doc, is there any info about the new thread model? I'm very interested to read some internals of Cats Effect since it's such a critical component. If you can point some directions, I can contribute to the doc and have someone review it.

[armanbilge]

I think it's safer to have a global thread limit so it doesn't affect the whole OS and let user have an opportunity to debug it when something goes wrong.

Yes, it's a good question. I asked Daniel about it and he will add some comments on this discussion later. If you are on Discord you can read the thread here.

---

is there any info about the new thread model

Yes! See:

• I/O Integrated Runtime Concept #3070
• https://youtu.be/PLApcas04V0

[wb14123]

Hi @armanbilge , I've watched the video and read the discussion. I don't see the doc at https://typelevel.org/cats-effect/docs/thread-model#work-stealing-pool is outdated other than the blocking section we've discussed. It's more like lack of details but I guess it makes it still consistent with the new implementation? Unless I miss something? Actually because of the lack of details in the doc, I don't really know what's the old implementation. Did it have a scheduler thread(s) that polls instead of polling from each worker thread?

[armanbilge]

@wb14123 thanks for getting caught up!

Here are three major changes to the WSTP that are not covered in that section:

• Blocking mechanism with cached threads #2687
• Integrated timers #3219
• Polling system #3332

Altogether, they mean that the higher-level goals discussed earlier on that page are also outdated.
https://typelevel.org/cats-effect/docs/thread-model#high-level-goals