What would the idiomatic way to set a

CompletableDeferred

as a mirror of another one? As in, I have two or more `CompletableDeferred`that I know are waiting for the same value, so I'd like to ‘merge' them so that completing the ‘master' automatically completes the ‘slaves'. So far I've come up with:

fun <T> CoroutineScope.merge(
  master: CompletableDeferred<T>,
  slave: CompletableDeferred<T>
) = launch {
  master.await().also { result ->
    slave.complete(result)
  }
}

However this creates a new coroutine, and I'm wondering if there's a better way to achieve a similar goal.

Looks a bit like XY problem. But I think what you want is

master.invokeOnCompletion { slave.complete(it.getCompleted()) }

. Wouldn't recommend doing this though.

Yeah, that looks better. Why wouldn't you recommend it?

What's the original use case to have multiple completable deferreds that are working to obtain the same value? Why can't you reuse the same single completable deferred everywhere you expect that same value?

@Erik I'm working on a cache implementation with structured concurrency, where clients request an object with an ID and by giving a

CompletableDeferred

so the cache can give them back their answer. If two requests for the same object are done before the first one has resulted, the cache merges the Deferred so the completion of the first one also completes the second one.

Yes, do you have code you don't mind sharing, so it can be edited?

Yep, give me a minute ^^

Yeah, this can be simplified. One sec.

I had something similar in my head, but I saw you were answering already, so I awaited this.

Ha! Just like the

get

method.

I haven't looked at the original piece(s) of code, but I'd like to add that a cache could not only have a cache being a

Map<I, Deferred<T>>

, but also a cache of the actual values. E.g. and in-memory cache could be a private

MutableMap<I, T>

.

That is simplified, yes. Correct me if I'm wrong, but but in this case if a first call to

get

is not in the cache (and therefore the

worker

is called), I think another call to

get

until the

worker

is done will have to wait for the first one to terminate?

Yes?

Basically something like: Time 1: Request to get 1, sent to the worker Time 2: Request to get 2, it's in the cache, replies immediately Time 3: The worker has replied, the new value is added to the cache and the first request is done

Yes, the code I posted does that.

I'm not sure I understand how it works... Does that mean that the

async

in the

semaphore.withPermit

unlocks the semaphore while the worker works, and locks it again when it's done?

Oh the

async

starts the request concurrently. So the

withPermit

exits while the request is running.

So what happens is; Time 1: Request to get 1, the worker starts to work on it and the semaphore is unlocked Time 2: Request to get 2, it's in the cache, the caller gets an already completed Deferred, so the await() doesn't suspend Time 3: Request to get 1,

getOrPut

gives the on-going Deferred created in time 1 Time 4: The worker terminates, both requests that were awaiting for 1 complete Is that how it works?

Exactly!

Thanks! That's exactly what I wanted 🙂

I repeat: I haven't looked at the original code precisely, but this is something that could work:

class Cache<I, T> {
    private val cache = mutableMapOf<I, T>()
    private val deferreds = mutableMapOf<I, Deferred<T>>()

    suspend fun getById(id: I) = cache.getOrPut(id) {
        val value = deferreds.getOrPut(id) {
            coroutineScope {
                async { TODO("Asynchronously obtain the value by ID") }
            }
        }.await()
        deferreds.remove(id)
        value
    }
}

I think this would work too for multiple request coming in: they just get an existing cached value, or if not, at most 1

Deferred<T>

is started and all requests await that single deferred

Time 1: get(1), misses cache Time 2: time 1 starts request Time 3: get(1), misses cache Time 4: time 1 finishes and removes request Time 5: time 3, starts request (no cache hit) I think the lock is required.

Yeah, the

deferreds.remove(id)

is a bit buggy here, but it's not required. It's only necessary to not keep completed deferreds in memory. So that line can be removed entirely

Do you recommend that the Cache use the main

CoroutineScope

, or have its own child of it?

Then your time 5 example would get the value either from the cache, or if it just misses the cache, then it gets it from the deferred in memory

It should create a child of it. A supervised one too.

So something like

Cache(...): CoroutineScope by CoroutineScope(SupervisorJob(myMainJob))

?

CoroutineScope

s should be kept private, but otherwise, that ctor looks fine.