Introduce SingleFlight (function de-duplication)

[recheej]

This is to resolve #1097. I have a local branch right now testing this out, but wanted to submit this issue to talk about the design. Essentially, we want to have the same functionality as SingleFlight from Go. The idea is that if you have a function A with a certain key in flight and another function B with the same key comes, function B will get function A's result.

Here's how I picture the public API for coroutines:

/**
 * Represents work that gives you duplicate suppression.
 *
 * Given a suspend function and key [K], if another function with the same key [K] comes while the original function is in flight/not yet completed,
 * the duplicate function will wait for original result.
 *
 */
public interface SingleFlightGroup<T, K> {
    /**
     * Get the result of function [block]. Only one execution of [block] will be "in-flight" at a time for a given key [K]
     *
     * @param key Unique identifier for this call.
     * @param block Function for which you want to get result. This will NOT be called if there's an original execution in-flight with same [key]. Will use [CoroutineContext] from [singleFlightGroup].
     */
    public suspend fun getResult(key: K, block: suspend CoroutineScope.() -> T): FlightResult<T>

    /**
     * Forget about this given [key]. Future calls will call into block of [getResult].
     */
    public suspend fun forget(key: K)
}

/**
 * Creates a new [SingleFlightGroup].
 */
public fun <T> CoroutineScope.singleFlightGroup(): SingleFlightGroup<T, Any> 

/**
 * Result of a flight.
 *
 * [shared] indicates whether this was shared with other callers.
 */
public data class FlightResult<T> (
    val result: T,
    val shared: Boolean
)

And here's how it would be used:

fun main() {
     GlobalScope.launch {
            val key = "test"
            val expectedResultValue = "22"
            val group = singleFlightGroup<String>()

            val deferredOne: Deferred<FlightResult<String>> = async {
                group.getResult(key) {
                    // this is the "original" function. If any other call comes in before it's completed with the same key, they will await its result
                    delay(100)
                    expectedResultValue
                }
            }

            val deferredTwo: Deferred<FlightResult<String>> = async {
                group.getResult(key) {
                    // this block will never be called since we wait for result from first function call
                    expectedResultValue
                }
            }

            val resultOne: FightResult<String> = deferredOne.await()
            val resultTwo: FightResult<String> = deferredTwo.await()
            println(resultOne.result == expectedResultValue)
            // prints out: true
        
     }
}

Hopefully that all makes sense. One question I've been dating is how to handle exceptions. If there's an exception within the block function... how should we handle it? My first thought is to just let it bubble up. The original Go API provides exception with the result if I'm not mistaken. We could go that route also.

[recheej]

cc @elizarov @qwwdfsad

[fvasco]

Memoization is not strictly related to coroutine, so why you should put it here instead of in a third party library?

[recheej]

Memoization is not strictly related to coroutine, so why you should put it here instead of in a third party library?

Hmm. I'll need to look into Memorization more but i believe the difference between single flight and that is that single flight only applies to things that are in progress. So I don't know it's really considered a cache.

[elizarov]

On the surface, this seems to be an interesting coroutine-related primitive to have in the library (it is not a cache and is indeed very coroutine-specific), but it will definitely require a non-trivial design effort to get a composable and minimal API here.

However, I do have a feeling that having just an "in-flight deduplication" supported will be clearly not enough. People will immediately want "to keep this key for a few more seconds after the answer", "to limit the number of such kept entries", etc, and it will end up with a full-blown cache implementation with tons of configuration options. This definitely deserves a separate library for now. We will not have the resources to engage in such an extensive design effort in the near future as we have quite a number of core concerns to cater for.

[recheej]

• @elizarov fair enough!