ScopedValue structured forking/forwarding API

[Nikita Bobko]

Glossary.

*suspend* functions in Kotlin are code-colored functions. `suspend` is
keyword that users put on their functions and functional types. All
suspend functions and lambdas undergo a CPS-transformation performed by
the Kotlin compiler. In other languages, such functions are commonly
referred as *async* functions. I may interchangeably call them either
async or suspend functions.

Andrew Haley wrote:
> Is it your intention that a set of ScopedValue bindings be
> associated with a coroutine, and that ScopedValue bindings be captured
> at the point at which the coroutine is suspended, and re-bound later
> when the coroutine is resumed?

Correct.

I'd like to take a step back and mention that there are actually two
possible integrations between ScopedValues and kotlinx.coroutines.

*Integration 1.* kotlinx.coroutines should provide an utility function
that would be an analog of ScopedValue.Carrier.run, with the slight
difference that the function must be suspend, and the lambda that it
accepts must be also suspend. That's the integration for which we don't
need anything else on top of what ScopedValues already provides.

*Integration 2.* Similar to how child threads in StructuredTaskScope
have access to all ScopedValues of their parent thread, we want to
forward all ScopedValues from the blocked parent thread (yes, parent
thread, not parent coroutine, please read further) to children
coroutines. For that kind of integration, we need something along the
lines of what I've described at the beginning of the thread.

In kotlinx.coroutines, we want to provide both integrations.

Andrew Haley wrote:
> However, it is straightforward to create a scoped value class of your
> own that has any properties you wish, including different inheritance
> rules. Given a ScopedValue.Carrier that binds some values, start your
> thread with aCarrier.run(task). Would this work for you?

Andrew Haley wrote:
> I'm baffled. Your proposal involves invoking bindings.forward() when
> (re)starting a task:
>
>                          pool.submit(() -> {
>                              bindings.forward(() -> {
>
> This has to be explicit, and has to be done when there are no bound
> scoped values. Why can't your implementation do it?

I think the misunderstanding may be comming from the fact that we were
thinking about different kinds of integrations.

For integration 1, we indeed are going to do something along the lines
of what you describe with ScopedValue.Carrier object.

To better understand integration 2, I think it is easier to see what I
mean by looking at Kotlin code (I understand that you may not be
familiar with Kotlin, I tried to make it as Java as possible):

    // The lambda below is obviously non-suspend,
    // we just call a Java API from Kotlin
    ScopedValue.where(scopedValue, "Duke").run({

        // Below is a non-suspend lambda with a single CoroutineScope
        // parameter. I extracted the lambda to a "spawner" variable
        // because of some Kotlin-unrelated-details
        val spawner: (CoroutineScope) -> Unit = { scope: CoroutineScope ->

            // Spawn a child coroutine and schedule to run it on UI main
            // thread.
            scope.launch(Dispatchers.Main, block = { // suspend lambda

                // Right now, it fails with:
                //     NoSuchElementException: ScopedValue not bound
                // My goal is make it return "Duke"
                println(scopedValue.get())

                // Non blocking suspension point.
                // delay is a suspend function and can be called only
                // from other suspend functions
                delay(1.second)

                // Right now, it fails with:
                //     NoSuchElementException: ScopedValue not bound
                // My goal is make it return "Duke"
                println(scopedValue.get())

            })
        }

        // runBlocking runs the lambda and keeps the current thread
        // blocked until all the spawned child coroutines are done.
        runBlocking(block = spawner)
    }

- Kotlin's runBlocking function is identical to Java's
  StructuredTaskScope.open() function
- Kotlin's CoroutineScope.launch function is identical to Java's
  StructuredTaskScope.fork function
- Kotlin's CoroutineScope is identical to Java's StructuredTaskScope
- In java, one has to manually do StructuredTaskScope.join(). In Kotlin,
  we do it implicitly at the end of runBlocking function.

Coroutines are not magical. Effectively, CPS-transformation slices
suspend functions by their suspension points. And under layers of
abstractions, kotlinx.coroutine still submits those slices to whatever
thread by doing old and boring threadPool.submit().

Suppose that the API that I'm suggesting at the beginning of the thread
existed, then at the beginning of runBlocking implementation, I'd
capture all the current bindings, and stash them inside a CoroutineScope
instance. To keep everything structural and sound, I obviously call
ScopedValue.Snapshot.join() and ScopedValue.Snapshot.close() on the
stashed bindings at the end of the runBlocking function.

And once we pass all further layers of abstractions, and get to the
point when we actually need to submit a sliece of a coroutine on a
threadPool. Instead of calling threadPool.submit(coroutineSlice), we
would just wrap coroutineSlice into ScopedValue.Snapshot.forward call:

    threadPool.submit(() -> {
        coroutineScope.stashedParentScopedValueBindings.forward(() -> {
            coroutineSlice.run();
        });
    });

so that a child coroutine has access to ScopedValues of its blocked
parent spawner thread, yay! Task accomplished, Java is once again a
great platform for integration, Nikita is happy. Mic drop.

Basically, the difference between these two integrations is that the
first integration bridges suspend world with the very same suspend world
with ScopedValue bindings in the middle. But the second integration
bridges Java's non-suspend world where some of ScopedValues bindings
already existed with Kotlin's suspend world.