typelevel / typelevel/cats-effect
Contention on Map when using IO.unsafeToFuture()
Nobody has claimed this yet.
- Dominant language
- Scala
- Stars
- 2.2k
- Forks
- 576
- Avg merge
- 2d 11h
- Merged PRs (30d)
- 18
Description
From https://discord.com/channels/632277896739946517/632278585700384799/921743315169325087
In a project based on Future, we introduce IO step by step. We use IO.unsafeToFuture() for interoperability.
I can observe the following locking:
The whole application is running with one main ExecutionContext using a ForkJoinPool building very similarly to scala.concurrent.ExecutionContext.opportunistic.
We build our own IORuntime to re-use the main ExecutionContext like this:
import cats.effect.unsafe.{IORuntimeConfig, Scheduler}
import java.util.concurrent.ScheduledThreadPoolExecutor
import scala.concurrent.{blocking, ExecutionContext}
object IORuntimeFactory {
/**
* To be able to run some [[cats.effect.IO]], for example into a [[scala.concurrent.Future]],
* we need a [[cats.effect.unsafe.IORuntime]]. This one is build from existing execution context.
* It does not follow the best practices from cats effects as we are using the same
* execution context for blocking and non blocking operations.
*
* This should be ok as long as we don't use it a lot.
*
* In general, it's better to avoid using it, and to consider using a [[cats.effect.std.Dispatcher]] instead.
*/
def from(ec: ExecutionContext): cats.effect.unsafe.IORuntime = {
val compute = ec
val blockingEC = new ExecutionContext {
override def execute(runnable: Runnable): Unit = ec.execute(() => blocking(runnable.run()))
override def reportFailure(cause: Throwable): Unit = ec.reportFailure(cause)
}
cats.effect.unsafe.IORuntime(compute, blockingEC, scheduler, () => (), IORuntimeConfig())
}
/** same scheduler as in [[cats.effect.unsafe.IORuntime.global]] */
val scheduler: Scheduler = {
val threadPrefix: String = "io-scheduler"
val scheduler = new ScheduledThreadPoolExecutor(
1,
{ r =>
val t = new Thread(r)
t.setName(threadPrefix)
t.setDaemon(true)
t.setPriority(Thread.MAX_PRIORITY)
t
})
scheduler.setRemoveOnCancelPolicy(true)
Scheduler.fromScheduledExecutor(scheduler)
}
}
This runtime can be instantiated many times.
For the observed contention, it is instantiated once and re-used.
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start by examining the reported IO.unsafeToFuture() interaction with the custom IORuntimeFactory and its shared ForkJoinPool-based ExecutionContext. Compare the compute and blocking execution contexts and the single-threaded scheduler against the locking shown in the report; done means identifying and resolving the observed contention, with an appropriate regression test if the relevant runtime code is located.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- scala
- Domain
- backend, performance
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 35/100