Work with ThreadLocal-sensitive Components #119

ian4hu · 2017-09-13T23:47:22Z

Many application trace framework use ThreadLocal to store the call context of a application, like the SLF4j MDC, CAT , and other distributed trace system(which use thread local for trace the context in some async situation).

But the coroutine will be dispatched to a unspecified thread to execute, how can i make the tracing work?

The text was updated successfully, but these errors were encountered:

elizarov · 2017-09-14T07:32:47Z

Coroutine's analog to ThreadLocal is CoroutineContext.

To interoperate with ThradLocal-using libraries you need to implement a custom ContinuationInterceptor that supports framework-specific thread-locals. Here is an example. Assume that use some framework that relies on a specific ThreadLocal to store some application-specific data (MyData in this example):

val myThreadLocal = ThreadLocal<MyData>()

To use it with coroutines, you'll need to implement a context that keeps the current value of MyData and puts it the corresponding ThreadLocal every time the coroutine is resumed on a thread. The code should look like this:

class MyContext(
    private var myData: MyData,
    private val dispatcher: CoroutineDispatcher
) : AbstractCoroutineContextElement(ContinuationInterceptor), ContinuationInterceptor {
    override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> =
        dispatcher.interceptContinuation(Wrapper(continuation))

    inner class Wrapper<T>(private val continuation: Continuation<T>): Continuation<T> {
        private inline fun wrap(block: () -> Unit) {
            try {
                myThreadLocal.set(myData)
                block()
            } finally {
                myData = myThreadLocal.get()
            }
        }

        override val context: CoroutineContext get() = continuation.context
        override fun resume(value: T) = wrap { continuation.resume(value) }
        override fun resumeWithException(exception: Throwable) = wrap { continuation.resumeWithException(exception) }
    }
}

To use it in your coroutines, you wrap the dispatcher that you want to use with MyContext and give it the initial value of your data. This value will be put into the thread-local on the thread where the coroutine is resumed.

launch(MyContext(MyData(), CommonPool)) {
    // do something...
}

The implementation above would also track any changes to the thread-local that was done and store it in this context, so this way multiple invocation can share "thread-local" data via context.

I do think we need a ready-to-use implementation for SLF4j in the integration directory of kotlinx.coroutines project (kotlinx-coroutines-slf4j). You are welcome to contribute it.

elizarov · 2017-09-14T20:09:15Z

I've reposted this Q&A to StackOverflow: https://stackoverflow.com/questions/46227462/how-to-use-code-that-relies-on-threadlocal-with-kotlin-coroutines/46227463
Closing this issue.

elizarov · 2018-04-27T12:56:51Z

We should really have it out-of-the box in a separate integration module.

userquin · 2018-04-28T18:22:12Z

@elizarov can you take a look at link below?

It is related to Spring Security Context (thread local), but there is an interceptor that cache the coroutine context so I cannot use the approach you mention. Due that the continuation is created in the thread in witch Spring Security Context "exists", I can capture it in the continuation and then apply your wrap solution.

konrad-kaminski/spring-kotlin-coroutine#19

thanks in advance and sorry to link to another repository.

oshai · 2018-06-11T19:36:15Z

@elizarov - I will try to contribute it.

elizarov · 2018-07-25T08:01:57Z

The problem with "wrapping dispatcher" approach that I've provided in SO Q&A is that this "wrapped dispatcher" would get lost if dispatcher gets replaced. Consider, the proposed SLF4J MDC context from #403, for example. If you do

launch (MDCContext(DefaultDispatcher)) { 
    foo()
}

But then in foo() you'll need to switch to UI context:

suspend fun foo() {
    withContext(UI) { 
        // MDCContext is lost here.
    }
}

I'm implementing an alternative approach, where you can implement and register a special extension point called CoroutineContextThreadLocal like this:

// declare custom coroutine context element
class MyElement : AbstractCoroutineContextElement(Key) {
    companion object Key : CoroutineContext.Key<MyElement>
    // some state is kept here
}

// declare thread local variable
private val myThreadLocal = ThreadLocal<MyElement?>()

// declare extension point implementation
class MyCoroutineContextThreadLocal : CoroutineContextThreadLocal<MyElement?> {
    // this is invoked before coroutine is resumed on current thread
    override fun updateThreadContext(context: CoroutineContext): MyElement? {
        val oldValue = myThreadLocal.get()
        myThreadLocal.set(context[MyElement])
        return oldValue
    }

    // this is invoked after coroutine has suspended on current thread
    override fun restoreThreadContext(context: CoroutineContext, oldValue: MyElement?) {
        myThreadLocal.set(oldValue)
    }
}

Now, MyCoroutineContextThreadLocal fully qualified class named shall be registered via META-INF/services/kotlinx.coroutines.experimental.CoroutineContextThreadLocal file.

…cal sensitive code Fixes #119