Is there a coroutine-flavored equivalent of a j.u.c.semaphore? I have a known amount of work to send over the network, and it can only be handled of batches of at most 10, so I’d like to simply spawn a zillion coroutines, each of which will handle 10 items, but to avoid overwhelming the other end I’d like to limit the concurrency ala a semaphore.

I guess you can use Channels and/or Actors for that

That’s unfortunate.

Fanning in to a channel with a set amount of workers reading from it, is a better abstraction than a semaphore imho.

Fan-in via a channel makes it harder to keep track of the result of each unit of work, though.

Semaphores are hard imho or at least for most of us 😛

😥

You can still use java semaphore. That’s the beauty of Kotlin

I can, but that will tie up the whole thread, no?

yes

That defeats the purpose…

probably you can imitate semaphore behavior by having a block locked with Mutex https://kotlinlang.org/docs/reference/coroutines/shared-mutable-state-and-concurrency.html#mutual-exclusion

OK but you see that they’re not equivalent, right…

Immediately inside of your batch worker, you could use an object pool of Mutex and try to obtain the a suspending lock which will create a suspending queue of workers waiting their turn

Answering: "Fan-in via a channel makes it harder to keep track of the result of each unit of work, though." You could send a response channel with the request. The caller could then just suspend by calling

receive

on response channel.

If you will only have a single response that you need to know of, a

CompletableDeferred

is cheaper than a channel is.

Yeah, right, that is better 👍

I have thousands of responses to keep track of

It’s going to be complicated if you’re trying to do things in a difficult way. IMO the better thing would be to only launch the number of coroutines that you want to happen concurrently and then fan out the work using a channel. https://kotlinlang.org/docs/reference/coroutines/channels.html#fan-out

data class WorkItem(val value: String, val response: CompletableDeferred<Response> = CompletableDeferred())

val workValues: ReceiveChannel<WorkValue> = getValues()
coroutineScope {
    repeat(workerCount) {
        launch {
            doSomething(workValues.receiveNextTen())
        }
    }
}

“a difficult way” — the way that semaphores make really simple 😕

How is a semaphore simpler than a MutexPool? Also, semaphores are also blocking 🤷. Semaphore tracks access by the thread. MutexPool tracks access by suspension.

See the start of this thread. I’m looking for a suspending equivalent to a semaphore

Right, which is why I suggested a MutexPool (and even gave an implementation) 🙂

Merely lamenting the lack of a simpler choice.

val semaphore = Semaphore(10)

repeat(20) {
    thread {
        // blocking
        semaphore.aquire()
        try {
            doSomething()
        } finally {
            semaphore.release()
        }
    }
}

vs

val mutexPool = MutexPool(10)

repeat(20) {
    launch {
        // suspending
        mutexPool.withLock {
            doSomething()
        }
    }
}

I’m just trying to understand what “simpler” means

Well, one of them requires having a roll-your-own concurrency building block, and one does not.

Using fan-in on a channel is a common pattern and really straight forward to implement. I really don't get what you think is complicated with it.

I didn’t say fan-in was complicated. I said that fan-in plus enough other machinery to do what semaphores do is complicated. If you disagree, okay. 🤷‍♂️

So the issue is because Semaphore is part of the JDK whereas a MutexPool is not part of coroutines core? (event though it’s a 20 line wrapper, including whitespace, around

Mutex

and

select

from the core library). I would still argue that it sounds like the design causing the need for a semaphore has an alternative that’s simpler and cleaner in coroutines, but I’m not that close to the original problem to say more. The nice part is coroutines being open source makes it nice to open (or implement) feature requests. (Most of which either get implemented fairly quick or get rejected with a real reason)