Hello there, for my work project I would like to c...

# coroutines

[Alexander Schmidt]

Hello there,
for my work project I would like to create a concurrency module that can be used from Kotlin and Java. Therefore my initial idea was to provide some spring-beans that provide access to a Default and an Io thread pool (just like Dispatchers.Io and Dispatchers.Default) managed by Spring. Every module should then be able to just use the pools via dependency injection for all kinds of tasks. Also I am deriving the coroutine dispatchers directly from the thread pools

@Configuration
class ConcurrencyConfig {
    @Bean
    fun executors(properties: ConcurrencyProperties): CustomExecutors {
        val taskDecorator = MDCTaskDecorator()
        return CustomExecutors(
            Default = createExecutor(properties.default, taskDecorator),
            Io = createExecutor(properties.io, taskDecorator, useVirtual = true)
        )
    }
    
    @Bean
    fun dispatchers(executors: CustomExecutors): CustomDispatchers {
        return CustomDispatchers(
            Default = executors.Default.asCoroutineDispatcher(),
            Io = executors.Io.asCoroutineDispatcher()
        )
    }
}

Sharing pools between Java and Kotlin makes configuring max-threads and so on much simpler, although Java-Threads might block coroutines. That is the reason why the Io-Executor uses virtual threads. This works quite well - so far, but now I face problems when adding some required features our applications need.

1. All threads and coroutines must have an MDC correlationId
2. Executors and dispatchers must shutdown gracefully within the spring context, allowing running tasks to continue (until a timeout is reached) and also create sub-tasks (which the task needs to successfully complete, e.g. the last step of a task is to create 3 subtasks (io) and then aggregate results (default)). New tasks should be rejected while shutdown is in progress.

[Alexandru Caraus]

Afaik, platforms have a shutdown handler, i dont remember the details, but there you can do cleanups on graceful shutdowns. Ps. Now wait for the "in a thread" guys to come.

[Alexander Schmidt]

For the first criteria, I have added a basic TaskDecorator to the Spring-Pool, which seems to work on the Java side. Since coroutines can context switch, I could not find a solution to fully automate adding MDCs to the coroutines. MDCContext from kotlinx-coroutines-slf4j adds a copy of the ContextMap as a ThreadLocalElement, but I have to call it manually every time I use something like launch, async, withContext, runBlocking etc. or sometimes the context will be missing or updates to the MDC values within a coroutine are lost in child coroutines or after a suspension.
My best approach so far was creating a dispatcher-wrapper that adds + MDCContext() to every dispatch-call.

@Test
fun `current behaviour`() {
    val softly = SoftAssertions()
    val oldValue = "oldValue"
    val newValue = "newValue"
    val ref = <http://customDispatchers.Io|customDispatchers.Io>
    putMdcValue(oldValue)
    runBlocking(customDispatchers.Default) { // dispatch of Default is not called, value is initially preserved due (why!?)
        softly.assertMdcValue(1, oldValue)
        putMdcValue(newValue)
        softly.assertMdcValue(2, newValue)
        delay(100)  // Snapshot is restored, update is lost
        softly.assertMdcValue(3, null)
        putMdcValue(newValue)
        softly.assertMdcValue(4, newValue)
        withContext(<http://customDispatchers.Io|customDispatchers.Io>) { // value is initially preserved due (why!?), there thread is changed?
            softly.assertMdcValue(5, newValue)
            delay(100)
            softly.assertMdcValue(6, null)
        }
        withContext(ref) { // dis does not have any snapshot, so no mdc propagation
            softly.assertMdcValue(7, null)
            delay(100)
            softly.assertMdcValue(8, null)
        }
        softly.assertMdcValue(9, null) // after context-switch, Snapshot is not available, so its null
        delay(100)
        softly.assertMdcValue(10, null)
    }
    softly.assertAll()
}

@Test
fun `desired bahaviour`() {
    val softly = SoftAssertions()
    val oldValue = "oldValue"
    val newValue = "newValue"
    val ref = <http://customDispatchers.Io|customDispatchers.Io>
    putMdcValue(oldValue)
    runBlocking(customDispatchers.Default) {
        softly.assertMdcValue(1, oldValue)
        putMdcValue(newValue)
        softly.assertMdcValue(2, newValue)
        delay(100)
        softly.assertMdcValue(3, newValue)
        putMdcValue(newValue)
        softly.assertMdcValue(4, newValue)
        withContext(<http://customDispatchers.Io|customDispatchers.Io>) {
            softly.assertMdcValue(5, newValue)
            delay(100)
            softly.assertMdcValue(6, newValue)
        }
        withContext(ref) {
            softly.assertMdcValue(7, newValue)
            delay(100)
            softly.assertMdcValue(8, newValue)
        }
        softly.assertMdcValue(9, newValue)
        delay(100)
        softly.assertMdcValue(10, newValue)
    }
    softly.assertAll()
}

I am trying a solution that 'just works' and no developer has to bother with MDCs.


For my other problem, I can add a RejectedExecutionHandler what for example offers runnables to the queue if they are submitted from within an executor thread (identified by thread prefix). This seems to work on the java side. The Kotlin dispatchers are derived from the Java executors so I was curious if it worked out of the box in Kotlin. It didn't, because closing the application context also cancels the dispatchers scope and therefore all coroutines will stop working when they reach a suspension point, although they would be accepted by the RejectedExecutionHandler.

@Test
@Timeout(DEFAULT_TEST_TIMEOUT, unit = TimeUnit.SECONDS)
fun `threads should complete and not interrupt on cancel`() {
    val context = AnnotationConfigApplicationContext(TestConcurrencyConfig::class.java)
    val customExecutors = context.getBean(CustomExecutors::class.java)
    val taskCompleted = CountDownLatch(1)
    val spawnTime = measureTimeMillis {
        customExecutors.Io.submit {
            <http://log.info|log.info>("Task starting")
            Thread.sleep(100)
            <http://log.info|log.info>("Task completed")
            taskCompleted.countDown()
        }
    }
    context.close()
    assertThat(spawnTime).isLessThan(100L)
    assertThat(taskCompleted.count).isEqualTo(0)
}

@Test
@Timeout(DEFAULT_TEST_TIMEOUT, unit = TimeUnit.SECONDS)
fun `coroutine should complete and not cancel upon shutdown`() {
    runBlocking {
        val context = AnnotationConfigApplicationContext(TestConcurrencyConfig::class.java)
        val customDispatchers = context.getBean(CustomDispatchers::class.java)
        val taskCompleted = CountDownLatch(1)
        withContext(<http://customDispatchers.Io|customDispatchers.Io>) {
            launch {
                <http://log.info|log.info>("Task starting")
                delay(100)
                <http://log.info|log.info>("Task completed")
                taskCompleted.countDown()
            }
            context.close()
        }
        assertThat(taskCompleted.count).isEqualTo(0) // fails
    }
}

While structured concurrency and cancellation to prevent 'unused' computations is nice in general, in my use case the tasks have to run till the end, because there are no transactions or rollbacks and unfinished tasks can lead to inconsistent data.

I am looking forward to thoughts, ideas or feedback on this :-)