Run several tasks and wait for first

Support
#1

Hi,

I’m trying to to the following function that can get list of suspend lamdas run all of them in parallel and wait for the first one that will finish?, in case that I had the blocks not in list I could just use select<>{} but for my problem i get the suspend blocks in list,

Thanks,

Ronen

#2

I don’t sure to fully understand your problem,
to await for a suspend block in a select clause you can wrap it in a Job using launch or async.

#3

Code always will be better:
typealias SUSPEND_METHOD = suspend () → Unit
fun waitForFirst(tasks: List<SUSPEND_METHOD>) {
// need the help here, how to implement it
}

Thanks a lot for the help!!!

#4

Thanks to @fvasco I found a way to do it if there is a better way I’ll be more than happy to know:

typealias SUSPEND_METHOD = suspend () → Unit
fun waitForFirst(tasks: List<SUSPEND_METHOD>) {
select {
tasks.forEach{task → launch { task() }.onJoin{}}
}
}

Now the tricky question is how to wait for more than one, for example to the first 2 tasks

#5

I usually do that for timed select, ie:

select {
    ....
    launch{ delay(1234) }.onJoin { ...}
}

Cycle onJoin over uncompleted tasks until n tasks are completed, otherwise consider to decorate:

https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CyclicBarrier.html

#6

what about the discussion in CountDownLatch implementation · Issue #59 · Kotlin/kotlinx.coroutines · GitHub about CountDownLatch which is similar to CyclicBarrier?

#7

Hi @sabagronen,
issue 59 is really use-case sensitive, I don’t know how this apply to your use case.
However Job.onAwait in a select allow you to wait only for ther first Job, so you should iterate to wait for n jobs.

Finally in linked issue 58 “alek-sys” propose an interesting implementation using a Channel: “Use counter + done channel”, in such case it is easy to get/await only n results.

PS: totalJobs = AtomicInteger should be a regular Int, in your use case must be “2”.

#8

Actually, my use case is to write a class that knows, among other things, to run x tasks in parallel and waits for y of them to finish, where y<=x. For waiting to all I’ve put all the Jobs in a list and apply join() on each one of them, for one thanks to you I figured it up how to do it.
If the way to wait for 2 is working, I get wait for any y. I wonder what is the preferred way for waitForFirst since we can iterate until the first

#9

Hey, How did you solve this problem ? Can you share the snippet ? Thanks.

#10

In case anyone else stumbles onto this, a contemporary solution for waiting N jobs to complete is to use flows.

fun <T> flowOfAsync(vararg blocks: suspend () -> T) = channelFlow {
    blocks.forEach { block ->
        launch {
            send(block())
        }
    }
    awaitClose()
}

flowOfAsync(...).drop(N-1).first()