IO thread pool for blocking calls

[elizarov]

We need some kind of IO dispatcher in kotlinx.coroutines that will be optimized for offloading of blocking calls and will be backed by an unbounded thread-pool that is smart enough to create new threads only when needed and to shut them down on timeout. The goal is that you could always wrap blocking IO call in withContext(IO) { ... } (UPDATED: it was formerly named run(IO)) and be sure that you will not run into the problem of having not enough threads in your pool. However, it will be up to the user to control the maximal number of concurrent operations via some other means.

It will be somewhat conceptually similar to newThread() scheduler in Rx, but it will be also designed to be used where io() scheduler in Rx is used.

[fvasco]

Java 6 has Executors.newCachedThreadPool, it is easy to create and easy to customize:

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

We can tune maximumPoolSize at startup to limit "the maximum allowed number of threads" throught a System property.

However we should consider that specific use case should create a fine tuned IO executor without using the global one.

[elizarov]

We need to make sure that there is no confusion in difference between run(IO) { /* blocking code here */ } and runBlocking { ... }.

[fvasco]

Kotlin can provide an explicit annotation (like @DslMarker) to avoid invocation from coroutine.

So we can rename the function to

@Blocking
fun runAndWait( ... )

[fvasco]

Using the code:

val job = ...
run(IO + job) { ... }
job.cancel()

I/O thread should be interrupted?

[elizarov]

We should not do interruption by default. It is dangerous. But we definitely need some easy way to adapt an interruptible blocking API (when we are sure it plays well with interrupts) to coroutines. I don't know yet what is going to be the best solution for this. This is also related to #57

[elizarov]

@fvasco W.r.t to @Blocking annotation it is actually a larger issue that spans both Kotlin and Java. Here is a related issue in YT: https://youtrack.jetbrains.com/issue/KT-15525

[voddan]

IMHO the name is misleading. When I see run(IO){} I think it is a coroutine context that guaranties that it runs on the main thread, even if started somewhere else. Or is it an impossible case?

[elizarov]

Hh... the assumption was that the code inside run(IO) { ... } runs inside special IO threads that are designed for blocking IO. The naming problem is how we highlight that it is designed for blocking IO only and that you should not go there if you have async IO....

[LouisCAD]

@elizarov Why not use blockingIO then, instead of just IO?

[mykola-dev]

because it's not blocking? and rxjava already has Schedulers.io() which does the same thing