Also, if doing blocking processing in an `Applicat...

# ktor

[dave08]

Also, if doing blocking processing in an ApplicationCall context, what's the better practice, to use suspending functions, withContext(CommonPool), or async { } (since some processing can be done in parallel..)?

[cy]

the first one is better if you have a single blocking task that you need to wait for

[cy]

with async you can launch multiple tasks

[cy]

also async could be called in non-suspend function

[dave08]

It's safe to use CommonPool for this? Also, does each request have it's own coroutine running in a thread pool, so that if one request is held up by a blocking process, the other one would be handled?

[cy]

Yes, no coroutines reused so using CommonPool for blocking tasks is safe: other non-blocking requests will be handled properly

[dave08]

If I would just use a suspend fun doing a single blocking process in a request, the other requests won't be blocked (or are they all running on the same thread... and I will need to do withContext(CommonPool) for it)?

[cy]

withContext launches a new child coroutine on a specificed context (separate thread pool) while the original one is suspended until the child will complete or crash

[cy]

so child coroutines will concur each other but not request handler coroutines

[dave08]

Right, but if I would just use suspend fun w/o withContext, then no other requests are handled meanwhile? I haven't found anything in the docs about these points (in Vert.x they write it a few time DON'T BLOCK THE EVENT LOOP... same in Ktor?)

[cy]

Yes but there are plans to fix it to allow users to block handler

[cy]

this is why we have it unspecified in the docs

[dave08]

Meanwhile, it's very important to document...! Since people can make such mistakes (I almost did 🤕)

[cy]

^^ @Deactivated User

[Deactivated User]

A couple of questions:

Is this blocking? https://github.com/ktorio/ktor-samples/blob/e9bd44f53dab4b0a45bf7b538d32af022325c2f9/app/youkube/src/Upload.kt#L58

Why is PartData.FileItem using InputStream instead of an Asynchronous channel?

Other than that, going to update the documentation to reflect this. Thanks!

[dave08]

Ooops! I'm also doing that! Thanks for pointing it out @Deactivated User... It's also like that in the upload docs https://ktor.io/servers/uploads.html

[dave08]

How should I do it then?

[cy]

Because for now it is not critical as input stream provided by FileItem is always ByteArrayInputStream or FileInputStream and never from the socket

[dave08]

But a very large file upload won't stop other requests from being processed?

[cy]

yes, it could but there is no way to make reading a file asynchronous on JVM

[cy]

the only we can do is to hide it as it is done in AsynchronousFileChannel

[Deactivated User]

Maybe internally it uses another thread I don’t know, but there is a signature using a completion handler for reading:
https://docs.oracle.com/javase/7/docs/api/java/nio/channels/AsynchronousFileChannel.html#read(java.nio.ByteBuffer,%20long,%20A,%20java.nio.channels.CompletionHandler)

And in the case it doesn’t work, is it possible to hide the usage of the IO pool to the final user by wrapping it in an asynchronous stream using the IO pool under the hood?

That way people won’t have to worry about blocking when handling uploads

[dave08]

What's the simplest way I could do it now in the meantime @Deactivated User?

[cy]

@Deactivated User yes, JDK's implementation does IO on a separate thread pool that is quite slow

[Deactivated User]

I’m not completely sure, if this would work as expected @cy:

async(ioCoroutineDispatcher) {
    part.streamProvider().use { its -> file.outputStream().buffered().use { its.copyTo(it) } }
}.await()

Maybe it can be reworked to multiplex reading instead of reading the whole file in that thread at once.

[Deactivated User]

Maybe increasing the read chunk size or doing buffering would improve performance and reduce the threadpool overhead?

I was hoping that JVM would call asynchronous APIs from the OS to do that. Specially on SSDs or systems with several HDDs, that could benefit from it

[Deactivated User]

Untested and unoptimized and with less throughtput but not blocking dispatchers with big files but blocking them temporarily with small chunk sizes:

//part.streamProvider().use { its -> file.outputStream().buffered().use { its.copyToSuspend(it) } }

suspend fun InputStream.copyToSuspend(out: OutputStream, bufferSize: Int = DEFAULT_BUFFER_SIZE, dispatcher: CoroutineDispatcher = ioCoroutineDispatcher): Long {
    var bytesCopied: Long = 0
    val buffer = ByteArray(bufferSize)
    while (true) {
        val bytes = withContext(dispatcher) { read(buffer) }
        if (bytes < 0) break
        withContext(dispatcher) { out.write(buffer, 0, bytes) }
        bytesCopied += bytes
    }
    return bytesCopied
}

[dave08]

What about using withContext() {} instead of async { }.await(), since you're in a suspend fun? Since anyways you await on each looping...

[cy]

launching a new coroutine for every block would be very slow

[dave08]

Or a produce { } might be better?

[cy]

well, you can try to launch a reading loop on a separate pool and use a channel or a byte channel to transfer bytes to the main handler coroutine

[cy]

val channel = writer(CommonPool) { 
    val buffer = ByteArray(4096)
    val stream = file.inputStream()
    while (true) {
        val rc = stream.read(buffer)
        if (rc == -1) break
        channel.writeFully(buffer, 0, rc)
    }
}

// here we have a channel that is asynchronous

[dave08]

Btw, I think withContext doesn't start a new coroutine, it just switches the coroutine context... so it could be used to surround the whole function, if you don't need to switch back and forth...

[cy]

writer is similar to produce but for byte channel

[Deactivated User]

I’m not sure the best solution here. If it was asynchronous without threadpools in the first place, probably the overhead would be smaller.

What I tried to do (but maybe I did it wrong) is that for example you have a io threadpool of 4 threads. And you have 8 uploads of 4 terabytes (that would take some time). Instead of blocking the threadpool with 4 of those tasks, I tried to process them all in parts.
Maybe to reduce the overhead of switching, I can process several parts and reuse the coroutine of reading/writing.

@cy In your snippet, the stream reading is still synchronous right?

[cy]

Yes but it is running on a separate thread pool so request handler pool is not affected

[cy]

and consuming bytes from a byte channel is safe

[dave08]

So @cy, how would writing the uploaded file look using this (the copyToSuspend part)?

[dave08]

I need to be able to use something that doesn't block in my current project in the meantime.. I won't get to terrabytes, but this is a microservice that MUST be responding to other requests while uploading... maybe I should just put the original code in a withContext(CommonPool) until this is fixed?

[dave08]

Or maybe in ioCoroutineDispatcher?

[dave08]

Also, in the meantime, I think the docs also need to have some kind of temporary solution for others not to have surprises...

[Deactivated User]

I’m interested in it too. I will update youkube sample (and uploads.html) too with the recommended way for doing this

[dave08]

@Deactivated User Also: https://ktor.io/servers/uploads.html, that's where I looked... I wouldn't have gone to Youkube unless there was nothing there...

[dave08]

Please let me know what you did, I need to release this microservice soon... Thanks!

[cy]

One should never block on ioCoroutineDispatcher! Unlike blocking in request handler coroutine, blocking on ioCoroutineDispatcher could cause infinite deadlock

[dave08]

Thanks for the warning @cy! If there are dispatchers that may be used by Ktor end-users it might be nice to have some docs on them too... instead of creating pools that might have been made for the purpose, or mistakingly using pools not for that purpose..

[cy]

You can simply create your own: https://github.com/Kotlin/kotlinx.coroutines/blob/master/coroutines-guide.md#coroutine-context-and-dispatchers

[cy]

other functions you can use to create your pool: newFixedThreadPoolContext and ExecutorService.asCoroutineDispatcher()

[dave08]

Ok, so there aren't any interesting dispatchers to reuse.. I'll do that, I suppose that for IO it's probably better than just using CommonPool...

[Deactivated User]

For the copyTo part:

//part.streamProvider().use { its -> file.outputStream().buffered().use { its.copyToSuspend(it) } }

suspend fun InputStream.copyToSuspend(
    out: OutputStream,
    bufferSize: Int = DEFAULT_BUFFER_SIZE,
    yieldSize: Int = 4 * 1024 * 1024,
    dispatcher: CoroutineDispatcher = ioCoroutineDispatcher
): Long {
    return withContext(dispatcher) {
        val buffer = ByteArray(bufferSize)
        var bytesCopied = 0L
        var bytesAfterYield = 0L
        while (true) {
            val bytes = read(buffer).takeIf { it >= 0 } ?: break
            out.write(buffer, 0, bytes)
            if (bytesAfterYield >= yieldSize) {
                yield()
                bytesAfterYield %= yieldSize
            }
            bytesCopied += bytes
            bytesAfterYield += bytes
        }
        return@withContext bytesCopied
    }
}