why don't we already have a Semaphore implementati...

# coroutines

[v0ldem0rt]

why don't we already have a Semaphore implementation in coroutines?

[Martin Devillers]

Couldn’t you essentially get a “suspending semaphore” behavior with a channel? Receiving on the channel will suspend until an item is offered, i.e. a “permit” is available. If the channel has unlimited capacity, then it can accumulate “permits” indefinitely.

[v0ldem0rt]

True that

[elizarov]

^^ That is not the only reason.

[v0ldem0rt]

👍

[elizarov]

The true reason in why we don’t have a semaphore you’ll get if you answer the question on why you need a semaphore in the first place.

[v0ldem0rt]

Absolutely agree

[v0ldem0rt]

Do channels have any additional overheads?

[elizarov]

Usually it is to to limit concurrency to some resource. But with coroutines there is a much better pattern, called “worker pool”. You just create N coroutines reading tasks from the same channel and performing them. Thus you’ve limited concurrent to N concurrent coroutines without having to use a semaphore.

[elizarov]

There is an additional overhead. But when you have concurrency and the corresponding scheduling you already have so much overhead that any incremental overhead from channels is negligible.

[elizarov]

If you ditch shared mutable stable with semaphores and mutexes alike and switch to sharing immutable data via channels, you get so much benefits in clarity and tractability of your architecture, that will outweigh any performance you might loose.

[v0ldem0rt]

So for fun I made an implementation anyway since it's few lines of code

import kotlinx.coroutines.experimental.channels.Channel
import kotlinx.coroutines.experimental.channels.count
import kotlinx.coroutines.experimental.runBlocking
import kotlinx.coroutines.experimental.timeunit.TimeUnit
import kotlinx.coroutines.experimental.withTimeoutOrNull

class SuspendSemaphore(initialCount: Int) {
    private val channel = Channel<Unit>()

    init {
        runBlocking {
            repeat(initialCount) { channel.send(Unit) }
        }
    }

    val onAcquire get() = channel.onReceive
    suspend fun getCurrentCounter() = channel.count()
    suspend fun acquire() = channel.receive()
    suspend fun release() = channel.send(Unit)
    suspend fun tryAcquire(timeout: Long = 0, timeUnit: TimeUnit = TimeUnit.MILLISECONDS): Boolean {
        return withTimeoutOrNull(timeout, timeUnit) {
            channel.receive()
        } == Unit
    }
}

Let me know if anything sounds suspecious

[elizarov]

LGTM

[louiscad]

@v0ldem0rt Shouldn't you pass initialCount into Channel<Unit>(…) call, and use offer(Unit) without runBlocking { … }? Your code seems deadlocking to me.

[v0ldem0rt]

Yep already ran into it

[v0ldem0rt]

and switched to offer