hey do I understand correctly that launch { x = async {...};y=async{}; x.await(); y.await()} will try to take 3 threads from ForkJoinPool and in case of 1core computer 2nd async simply will not start until first finished even without "await"?

No, all those coroutines run on CommonPool dispatcher which has

numCpu - 1

threads

my confusion with coroutines (still) is their relationship with physical hardware threads... because I have case where I have code something like launch { }

Think about coroutines as about callbacks

launch { in = inputchannel.receive outputWorkers.forEach { async{doWork(in)}.await()} } And I somewhat worried that if my doWork(in) of the first element takes long time, then other doWorks will wait in wain

wat is

doWork

?

doWork in general is call into ZeroMQ send

And this call is blocking or not?

with greedy pirate algorithm, so it can take a while while it returns

If this call to 0mq using some callbacks + coroutine adapter it’s non-blocking, otherwise it’s blocking and you shouldn’t wrap it to coroutines with CommonPool

if other end is down, then particular call my take like 12 seconds to finish

see, “12 seconds to finish” doesn’t mean that something is blocking or not, it’s more about callbacks vs thread blocking

it is blocking... of course... zeromq is pretty strightforward...

Yes, if 0mq lib has non-blocking API is the best choice in case of coroutines

currently I am using it with coroutines... only parts which are blocking - well they are blocking

Something like this:

val ZeroMQ = newFixedThreadPoolContext(ZERO_MQ_THREAD_POOL_SIZE, "ZeroMQ")

launch {
in = inputchannel.receive
outputWorkers.forEach { async(ZeroMQ) {doWork(in)}.await()}
}

but async will dispatch into CommonPool, in case if I have several threadss... wouldn't it dispatch each blocking call onto it's own thread?

async will dispatch into CommonPool

No, see my example

so if I stick up my dispatcher and I have like don't know 4-5 threads reserved, then I am good?

Yes

thanks... sounds about nice exercise 🙂

See, CommonPool is default pool that used by default and intended to use only with non-blocking calls, even 1 thread usually is enough to dispatch all non-blocking calls (tho, there is an issue about this to allow configure minimal amount of threads in CommonPool)

but when we have callback based invocations of parallel tasks (think Firebase on Android) then this "dispatching" problem is problem of FB implementation, and I only write my Coroutine wrapper as per examples which wraps the callback, and I even don't worry about dispatching as it is hidden under the hood of FB

But if you want to use some blocking API (blocking file io, blocking network, DB etc) it would be better to wrap such calls to own custom thread pool

yea that is what is this confusing part of Coroutines documentation, as all the examples start with fun doSomeBigWork() And then show how sticking this into async or launch servse the problem... But in reality, if it's blocking, it IS blocking... and coroutines won't save if you have long-running file-crunching anyway

Coroutine cannot magically convert your blocking code to non-blocking

well stick this one sentence as first on coroutine documetnation 🙂 people get confused 🙂

BTW, You don’t need asunc/await here

outputWorkers.forEach { async(ZeroMQ) {doWork(in)}.await()}

withContext would be enough:

outputWorkers.forEach {
    withContext(ZeroMQ) { doWork(in) }
}

I think this is more efficient:

withContext(ZeroMQ) {
    outputWorkers.forEach {
        doWork(it); yield()
    }
}

Yes, you right, better to consume on the same thread

Neither version consumes on a single thread since ZeroMQ dispatcher could be thread pool based, and both invoke a suspending function, which means it then resumes in the coroutineContext, on any of its threads. But my version avoids having two suspension points (withContext can suspend before and after its block) when you can have only one suspension point instead (yield makes it cooperative with other coroutines running on the same dispatcher)

Good point about two suspension points, sorry for tautology 😅

I don't quite understand how it works, but looks like it's working

What works? yield?

no worker dividing... I printed thread on which actual zeromq send happens, and they are different > I gave 5 threads to that newContext as you mentioned above

It’s just a pool of tasks that should be dispatched, so this pull processed by coroutine dispatcher

pool?

haha, yes, end of working day %)

@gildor I am trying to understand yours and Louis code difference... What I want (and I can't quite test it on my case) to have like 5 workers (potentially) work in parallel (litearry) so that all of them start at the same time, but all finish when all are finished... So I am using coroutines not for starting in parallel, but be sure that when I exit the code block, all of them has finished... I don't care of order of execution, I care that noone blocks another for startup

There is no real difference, version of Louis little bit more efficient, because have one less suspension point (point where coroutine dispatch callback and do some other work)

then I either don't undersatnd coroutines yet, or I am too embedded in Java thinking... If the worker.dowork is blocking code, then the execution of like (execution pointer) stops at the line no matter if it's in coroutine or outside? Isn't it

This is not about blocking or not blocking

yea looking at internals of withContext, it finally runs the executor, so the block passed to it is put on thread... your case, will 5 times put the code block on one thread, but Luis case puts all the workers sequentially to run on one dispatched thread... so if first worker takes long time, the next ones even won't start until first one finished

is about what will be generated on coroutine side: how many states in state machine

ok

again, it depends on what you want to achive, withContext is operations that waits for result

I want to achieve, that 5 workers START in parallel, but execution order is independent and they can each one take as much as they need, and I need to wait until all 5 finish eventually, and only then take next portion of data to give again for those 5 workers to work on

Yes, you need worker pool pattern with channel

yes, only how do I arrange it effectively with coroutine framework... I was thinking Ineed like 5 asyncs and then 5 awaits after those 1st ones started

You need async only if you want to get result

now I think I need something like launch { in = inputChannel.receive() launch { withContext(5ThreadPoolContext) { workers.foreach{doWork(in))} } }

take a look here: https://github.com/Kotlin/kotlinx.coroutines/issues/172

yea... but my thinking (where is it wrong) -> First launch in general runs in infinite loop which reads channel second launch is to "consolidate" inner coroutiens when they finish, so the 3rd one - probably can be launch too -> is the one which is run for each worker so somehow 3 launches one inside another to get that pooled effect?

Check samples from this issue. There is example of

map

for ReceiveChannel where you can pass your IO context and do some operation (even blocking) and map some values with required parallelism

ahhh thanks... when you say so, I actually myself did paralell map some time ago, which was working... thanks will look at it

Don’t forget to pass dispatcher suitable for blocking operations to context

Here is my blog post some time ago, where I achieved something similar to what is proposed in the issue... it was more like experiment, at how big data pool parallel map is effective

@Olekss I think this is what you're looking for:

outputWorkers.map {
    launch(ZeroMQ) {
        doWork(it)
    }
}.forEach { it.join() }

Yes, exactly, if you don't need results of work