Kotlin Coroutine suspend function Integration uses single thread dispatcher

[Legion2]

Describe the bug

When working with kotlin coroutines and suspend functions in quarkus, the code is executed sequentially by a single thread by default. One has to explicitly change the kotlin coroutine dispatcher to get the expected behavior of the Dispatchers.Default which uses a Thread pool.

Expected behavior

By default when using suspend function with quarkus the Dispatchers.Default or Vertx.vertx().dispatcher() should be used as the dispatchers. Or it should be possible to configure the used Dispatcher.

Actual behavior

Some single threaded Dispatcher is used (maybe runBlocking is used) to run the suspend functions.

How to Reproduce?

Add the following code to an RestEasy Resource and call the endpoint

    @GET
    @Path("test")
    suspend fun test() {
        val start = System.currentTimeMillis()
        coroutineScope {
            repeat(10) {
                launch {
                    val timestamp = System.currentTimeMillis()
                    while (System.currentTimeMillis() < timestamp + 1000) {
                        // Busy wait
                    }
                    println("Hello, world!")
                }
            }
        }
        println("Total time: ${System.currentTimeMillis() - start}")
    }

Output: total time 10 seconds

    @GET
    @Path("test")
    suspend fun test() {
        val start = System.currentTimeMillis()
        coroutineScope {
            repeat(10) {
                launch(Vertx.vertx().dispatcher()) {
                    val timestamp = System.currentTimeMillis()
                    while (System.currentTimeMillis() < timestamp + 1000) {
                        // Busy wait
                    }
                    println("Hello, world!")
                }
            }
        }
        println("Total time: ${System.currentTimeMillis() - start}")
    }

Output: total time 1 second

    @GET
    @Path("test")
    suspend fun test() {
        val start = System.currentTimeMillis()
        coroutineScope {
            repeat(10) {
                launch(Dispatchers.Default) { //Dispatchers.IO
                    val timestamp = System.currentTimeMillis()
                    while (System.currentTimeMillis() < timestamp + 1000) {
                        // Busy wait
                    }
                    println("Hello, world!")
                }
            }
        }
        println("Total time: ${System.currentTimeMillis() - start}")
    }

Output: total time 1 second

Output of uname -a or ver

Linux leon-pc 6.5.0-35-generic #35~22.04.1-Ubuntu SMP PREEMPT_DYNAMIC Tue May 7 09:00:52 UTC 2 x86_64 x86_64 x86_64 GNU/Linux

Output of java -version

openjdk version "17.0.11" 2024-04-16 OpenJDK Runtime Environment (build 17.0.11+9-Ubuntu-122.04.1) OpenJDK 64-Bit Server VM (build 17.0.11+9-Ubuntu-122.04.1, mixed mode, sharing)

Quarkus version or git rev

3.7.3

Build tool (ie. output of mvnw --version or gradlew --version)

Gradle 8.5

Additional information

No response

[quarkus-bot]

/cc @geoand (kotlin)

[Legion2]

Can someone point me to the source code of the coroutine dispatcher implementation, then I can take a look and try to debug the problem.

[Legion2]

I found this dispatcher implementation which is used by the resteasy reactive client.

quarkus/extensions/resteasy-reactive/rest-kotlin/runtime/src/main/kotlin/org/jboss/resteasy/reactive/server/runtime/kotlin/ApplicationCoroutineScope.kt

Lines 34 to 51 in 32d0c2d

	class VertxDispatcher(
	private val vertxContext: Context,
	private val requestScope: ThreadSetupAction.ThreadState,
	private val rrContext: ResteasyReactiveRequestContext
	) : CoroutineDispatcher() {
	override fun dispatch(context: CoroutineContext, block: Runnable) {
	// context propagation for suspending functions is not enabled yet, will be handled later
	vertxContext.runOnContext {
	requestScope.activate()
	CurrentRequestManager.set(rrContext)
	try {
	block.run()
	} finally {
	requestScope.deactivate()
	}
	}
	}
	}

There is also a similar implementation in the vertx extension https://github.com/quarkusio/quarkus/blob/ea2c6a4090d1eccb4b39250198601f7da9454833/extensions/vertx/kotlin/runtime/src/main/kotlin/io/quarkus/vertx/kotlin/runtime/VertxDispatcher.kt

Both implementations do not use the existing vertx Dispatcher and implement a very basic execution of coroutine on the vertx event loop.

[mschorsch]

I am also interested in the outcome of the analysis of this issue.

[geoand]

Thanks for opening this!

Just so I understand this, you are saying that when are manually handling the launch of coroutines, the dispatcher is single threader?

Also, some comments on things mentioned above:

I found this dispatcher implementation which is used by the resteasy reactive client.

The code you have linked to is for the Quarkus REST server part, not the client.

Both implementations do not use the existing vertx Dispatcher

Which one are you refering to as the existing vertx Dispatcher?

[mschorsch]

Which one are you refering to as the existing vertx Dispatcher?

The vertx.dispatcher() from vertx-lang-kotlin-coroutines (https://vertx.io/docs/vertx-lang-kotlin-coroutines/kotlin/#_running_a_coroutine_from_a_vert_x_context)

[geoand]

Oh, I was not aware of that one at all...

I wonder if we should start using it elsewhere too Never mind, that would not work in Quarkus

[Legion2]

@geoand why can the Dispatcher from vertx-lang-kotlin-coroutines not be used in quarkus. If it can not reused, what changes are needed in quarkus to port the Dispatcher from vertx-lang-kotlin-coroutines.

[geoand]

See the code in RESTEasy Reactive that uses its own dispatcher