Read-write Mutex

[fvasco]

Is it in the roadmap?

[elizarov]

It is on my to-do list, but not for the short-term. It is really up for grabs. This is a great feature for someone, who wants to figure how to write low-level lock-free implementations for kotlinx.coroutines primitives, to go ahead, study Mutex implementation, and figure out how to abstract (reuse) parts of that to implement ReadWriteMutex. It would be a pity if I write it myself and deprive someone else from this great learning opportunity.

[bobvawter]

Here's a basic implementation of the 80% case for ReadWriteMutex built on top of Mutex where you just want to execute coroutines as readers or writers.

https://gist.github.com/bobvawter/4ff642d5996dfccb228425909f303306

[fvasco]

The proposed implementation misses of support for "select" clause.

[elizarov]

Adding "select" in not a critical problem. I have a different concern here -- stateLock approach does not seem efficient to me for a core synchronization primitive. The act of acquiring read lock should be fully lock-free for scalability purposes, so that it scales to multi-core machines where a lot of threads can be acquiring and releasing read lock concurrently.

[bobvawter]

@elizarov If you agree with the general design of that ReadWriteMutex API, I can work on making the implementation more like Mutex (using an atomic queue of operations), adding additional state-management methods (e.g. tryReadLock()), and then put together a PR in the next couple of weeks.

Two questions before I start:

• Should state-change functions like readLock() accept an owner argument?
• If a user were to pass the same object twice to readLock(owner:Any?), should that call succeed?

I can see arguments for owner representing a specific actor (possibly reentrant or concurrent), or for owner representing a specific read operation. The former would look like some internal structure similar to ConcurrentMap<Any, Int> and the latter just a ConcurrentSet<Any>.

[jcornaz]

@elizarov, I don't understand your statement:

The act of acquiring read lock should be fully lock-free

How is it possible to guarantee that there will never be a write-lock and a read-lock open at the same time without locking something when we acquire a read-lock ?

I am not aware of other algorithms than theses two: https://en.wikipedia.org/wiki/Readers%E2%80%93writer_lock#Implementation

And neither of them alow to acquire a readlock in a lock-free manner.

Here my current read-write mutex implementation: https://gist.github.com/jcornaz/e94ee6a3a139ddd33c20554b0380d30f

[jcornaz]

After a second thought I think I now understand how it can be lock free to acquire a read-lock (if there is no write-lock open).

I drafted a first implementation here : https://github.com/jcornaz/kotlinx.coroutines/blob/c2f6831c3d8aa81c3b6c9e9da753873153e2b8a9/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/sync/ReadWriteMutex.kt
(EDIT: This implementation suffer a critical issue allowing to acquire a read lock, when a write lock is active)

This implementation still suffer some problems that I'll address before creating a PR.

But, could you tell me what interface you prefer?

Choice 1:

interface ReadWriteMutex {
  val isReadLocked: Boolean
  val isWriteLocked: Boolean

  fun tryLockRead(): Boolean
  fun tryLockWrite(): Boolean

  suspend fun lockRead()
  fun unlockRead()

  suspend fun lockWrite()
  fun unlockWrite()
}

Choice 2:

interface Lock {
   fun unlock()
}

interface ReadWriteMutex {

  val isReadLocked: Boolean
  val isWriteLocked: Boolean

  fun tryLockRead(): Lock?
  fun tryLockWrite(): Lock?

  suspend fun lockRead(): Lock
  suspend fun lockWrite(): Lock
}

The first interface looks more like the Mutex interface. But it's less safe, because one could call unlockRead() without calling lockRead() before. The second interface make possible to address this issue, but would be less consistent with the Mutex interface.

Note 1: I know there is no select clause yet. I will try to add them.
Note 2: This is not possible to specify owner yet. Should I address it too ?

[jcornaz]

Here would be my implementation using the second interface:
https://github.com/jcornaz/kotlinx.coroutines/blob/feature/readwrite-mutex/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/sync/ReadWriteMutex.kt

If you agree with the concept, I'll try to implement select clauses, add comments and prepare a pull request.

[fvasco]

Regardless of final choice I consider better to unify interfaces.

I like solution 2 but I suspect it requires some extra allocation.
Moreover using a Mutex force me to store locally both Mutex and Lock instances.

I suggest a "Choice 3" opposite to the first.

Choice 3:

interface ReadWriteMutex {
    val read: Mutex
    val write: Mutex
}

comparison:

rwMutex.lockRead()
rwMutex.read.lock()

Plus: choice 3 is compatible with regular mutex.

[jcornaz]

Thanks @fvasco, this is a very good proposition and would be my preferred choice too.

I'll try do it that way

[elizarov]

I like @fvasco API proposal, too. Please, send PR when you are done.

As a side note, we are considering to move "stack overflow prevention" logic that is currently complicating implementation of a regular (exlusive) Mutex into Unconfined dispatcher. It should make Mutex (and ReadWriteMutex) implementation simpler and faster for a typical case of a default dispatcher, while still being free from stack overflow problem under Unconfined dispatcher. See #80 for details.

[Dmitry-Borodin]

@fvasco please clarify whether you are working on it. Since this issue is still with "up for grabs" label, and it is not clear, whether it make sense to work on it.

[fvasco]

Hi @Dmitry-Borodin
sorry for misunderstanding,
I'm considering this job assigned to @jcornaz

I'll try do it that way

Any news?

[jcornaz]

Well hmm... I'd say don't wait for me. The implementation is harder than I thought.

Even if I am actually still trying to do it (at least for my own practice and learning of the subject), I think it's better if you don't wait for my implementation.

@Dmitry-Borodin, feel free to work on it, if you're interested.

[Dmitry-Borodin]

Thank you.
I'll take a look what I can do, I will write here if I will deliver it.

[elizarov]

Let me clarify for people googling up this thread. We are not focused on implementation of RW mutex right-now. We are more focused on CSP/actor-based programming paradigms. However, if there is PR with a high-quality/high-performance RW-mutex implementation, then we'll accept it into the library.

[kobe2000]

We are more focused on CSP/actor-based programming paradigms

Hi, did you mean actor programming can eliminate RW mutex?

[elizarov]

Actor-style programming eliminates the need for all kinds of synchronization primitives, including (but not limited to) locks and RW locks. The whole idea of actor-style programming is that mutable state is never shared, but is encapsulated into an actor.

[kobe2000]

Could you please give me some hint how to use actor-style programming avoiding RW lock in such situation:
I use a large collection in memory(GBs), and it needs to support CRUD operations in multithread, I can not figure it out how to code avoiding RW lock to improve performance.
Thank you!

[btwilk]

Here are efficient read-write lock and semaphore implementations that I would like to contribute.

https://github.com/btwilk/kotlinx.coroutines/tree/new-sync/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/sync

I've been using them for a while without any problems but I would appreciate review for correctness.

Let me know what needs to be fixed/cleaned up/tested for a successful PR.

[fvasco]

Hi @btwilk
are you considered to expose the proposed public interface?

interface ReadWriteMutex {
    val read: Mutex
    val write: Mutex
}

in such case the implementation can be private.

[btwilk]

I think tracking multiple owners is too much overhead for a low-level primitive, which is the main reason I didn't consider it.

Also, the Mutex interface is pretty heavy. I haven't thought about:

• onLock: support for select
• lock: the cancellation behavior documented in Mutex

I hope there is a way forward that allows for incremental progress.

[btwilk]

I just updated my branch with the proposed interface, except using a lighter-weight Lock interface instead of Mutex.

[aleksclark]

Any input on the new PR for this?

[elizarov]

Sorry. We're currently concentrated on flows, so the mutex stuff is on a back-burner. However, we also in the process of introducing an efficient implementation of semaphore (see #1101, paper on the algorithm is to be published, too) and plan to reimplement a regular mutex on top of this efficient semaphore. The architecture of this semaphore actually scales to other synchronization primitives like R/W mutexes.

[alllex]

The "up for grabs" tag was removed, but there seems to be no progress on the issue. Does someone actively work on this?

[elizarov]

It is not an "easy" issue that you can just go and fix, but locks are not that much useful with coroutines as with threads, so no short-terms plans to fix it.

[alllex]

Do I understand correctly, that the new efficient Semaphore has already been implemented and merged? Is it all that is supposedly required for an efficient implementation of the R/W mutex or there are more infrastructural changes on the way?

[elizarov]

Do I understand correctly, that the new efficient Semaphore has already been implemented and merged? Is it all that is supposedly required for an efficient implementation of the R/W mutex or there are more infrastructural changes on the way?

Yes. Semaphore is implemented and merged, yet R/W mutex is quote a non-trivial extension of that algorithm.

[LouisCAD]

Quoting @elizarov from over 4 years ago:

It would be a pity if I write it myself and deprive someone else from this great learning opportunity.

Looks like there are not a lot of folks that feel good enough to take up that learning opportunity 😅

I personally wanted to make one a few months ago as I finally had the use case, and even with all the interesting discussion here and several links, I didn't succeed nor make encouraging progress in the one day time box I set myself, so I ended up giving up and using a catch-all Mutex instead.

[qwwdfsad]

Writing your own RW-mutex or, even more, extending our Semaphore algorithm may be quite a hard task.
But when you already have an efficient semaphore realization, you don't have to!

Semaphore is quite versatile primitive and the rest of the synchronization primitives can be built on top of it.

• Little book of semaphores has a step-by-step implementation of RW-lock on top of a few semaphores with a detailed explanation of why things work (or don't). The target chapter is 4.2.

• The more efficient implementation (lightswitch is replaced with CAS-loop), but in C++, can be found here. This particular C+ code is pretty straightforward to be ported on Kotlin.

• The most straightforward way, though not supported in our API is to have a Semaphore(Int.MAX_VALUE) where issuing read-lock is acquire(1) and issuing write-lock is acquire(Long.MAX_VALUE)

NB: these implementations are good enough for any project-specific uses, but not for kotlinx.coroutines themselves. We have plenty of open questions regarding how RW public API must look like, and, in addition, we would like to extend the internal Semaphore algorithm in the most efficient way

[youcef-debbah]

Without any redirections below is my full implementation of a read/write mutex based on RWMutex implementation (go lang)
check out RWMutex source HERE
THIS article will help you understand the implementation
disclaimer: I wrote this in a hurry in a single workday, more tests and proper documentation are certainly needed but for now this is what I have, I hope this would save someone some time and effort in the future!

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.selects.SelectClause2
import kotlinx.coroutines.sync.Mutex
import java.util.concurrent.atomic.AtomicInteger
import javax.naming.OperationNotSupportedException

interface ReadWriteMutex {
    val read: Mutex
    val write: Mutex
    val state: LockState

    enum class LockState {

        // at least one read lock is acquired (returned from lock fun)
        READ_LOCKED,

        // write lock is acquired, in this state new readers/writers will -obviously-
        // suspended -inside lock()- until the current writer release his lock,
        // in the same time keep in mind that the writer itself may have been
        // already returned from lock fun OR it may be still suspended -inside lock()-
        // waiting for old readers to release their locks
        WRITE_LOCKED,

        // no lock is acquired
        UNLOCKED,
    }

    fun ensure(targetState: LockState)
    fun ensureWriteLocked()
    fun ensureReadLocked()
    fun ensureUnlocked()
}

fun ReadWriteMutex(): ReadWriteMutex {
    val mutex = Mutex()
    val writePermissions = Channel<Unit>()
    val readPermissions = Channel<Unit>()
    val pendingCount = AtomicInteger()
    val readersDeparting = AtomicInteger()
    return SimpleReadWriteMutex(
        pendingCount,
        ReaderMutex(mutex, writePermissions, readPermissions, pendingCount, readersDeparting),
        WriterMutex(mutex, writePermissions, readPermissions, pendingCount, readersDeparting),
    )
}

// simple and efficient non-reentrant read write mutex without timeout shenanigans
// ReaderMutex does not track any locks owners, WriterMutex track only the current writer owner
internal class SimpleReadWriteMutex(
    private val pendingCount: AtomicInteger,
    override val read: ReaderMutex,
    override val write: WriterMutex,
) : ReadWriteMutex {
    override val state: ReadWriteMutex.LockState
        get() {
            val state = pendingCount.get()
            return if (state > 0)
                ReadWriteMutex.LockState.READ_LOCKED
            else if (state == 0)
                ReadWriteMutex.LockState.UNLOCKED
            else
                ReadWriteMutex.LockState.WRITE_LOCKED
        }

    override fun ensure(targetState: ReadWriteMutex.LockState) {
        val currentState = state
        if (currentState != targetState)
            throw IllegalStateException("the ReadWriteMutex was expected to be $targetState but was $currentState")
    }

    override fun ensureWriteLocked() {
        ensure(ReadWriteMutex.LockState.WRITE_LOCKED)
    }

    override fun ensureReadLocked() {
        ensure(ReadWriteMutex.LockState.READ_LOCKED)
    }

    override fun ensureUnlocked() {
        ensure(ReadWriteMutex.LockState.UNLOCKED)
    }

    override fun toString(): String = "SimpleReadWriteMutex($state)"
}

internal sealed class AbstractReadOrWriteMutex(
    protected val mutex: Mutex,
    protected val writePermissions: Channel<Unit>,
    protected val readPermissions: Channel<Unit>,
    protected val pendingCount: AtomicInteger,
    protected val readersDeparting: AtomicInteger,
) : Mutex {

    companion object {
        const val MAX_READERS = 1 shl 30
    }

    @Deprecated(
        message = "Mutex.onLock deprecated without replacement. For additional details please refer to #2794",
        level = DeprecationLevel.WARNING,
        replaceWith = ReplaceWith(""),
    )
    override val onLock: SelectClause2<Any?, Mutex>
        get() = throw OperationNotSupportedException("")

    override fun toString(): String {
        val lockState = if (isLocked) "locked" else "unlocked"
        val className = this::class.simpleName ?: "Mutex"
        return "$className($lockState)"
    }

    protected fun overUnlockedException() =
        IllegalStateException("cannot unlock because the mutex is already unlocked!")

    fun debug() = toString() + "{readersDeparting=$readersDeparting,pendingCount=$pendingCount}"
}

internal class ReaderMutex(
    mutex: Mutex,
    writePermissions: Channel<Unit>,
    readPermissions: Channel<Unit>,
    pendingCount: AtomicInteger,
    readersDeparting: AtomicInteger,
) : AbstractReadOrWriteMutex(mutex, writePermissions, readPermissions, pendingCount, readersDeparting) {

    override val isLocked: Boolean
        get() = pendingCount.get() > 0

    override suspend fun lock(owner: Any?) {
        if (owner != null) throw UnsupportedOperationException()
        if (pendingCount.incrementAndGet() < 0) {
            readPermissions.receive()
        }
    }

    override fun unlock(owner: Any?) {
        if (owner != null) throw UnsupportedOperationException()
        val readersCount = pendingCount.decrementAndGet()
        if (readersCount < 0) {
            if (readersCount == -1 || readersCount == -MAX_READERS - 1) {
                throw overUnlockedException()
            }
            if (readersDeparting.decrementAndGet() == 0) {
                writePermissions.trySend(Unit)
            }
        }
    }

    override fun tryLock(owner: Any?): Boolean {
        if (owner != null) throw UnsupportedOperationException()
        while (true) {
            val readersCount = pendingCount.get()
            if (readersCount < 0) {
                return false
            }
            if (pendingCount.compareAndSet(readersCount, readersCount + 1)) {
                return true
            }
        }
    }

    override fun holdsLock(owner: Any): Boolean {
        throw UnsupportedOperationException()
    }
}

internal class WriterMutex(
    mutex: Mutex,
    writePermissions: Channel<Unit>,
    readPermissions: Channel<Unit>,
    pendingCount: AtomicInteger,
    readersDeparting: AtomicInteger,
) : AbstractReadOrWriteMutex(mutex, writePermissions, readPermissions, pendingCount, readersDeparting) {

    override val isLocked: Boolean
        get() = pendingCount.get() < 0

    override suspend fun lock(owner: Any?) {
        mutex.lock(owner)
        val oldReadersCount = pendingCount.getAndAdd(-MAX_READERS)
        if (oldReadersCount != 0 && readersDeparting.addAndGet(oldReadersCount) != 0) {
            writePermissions.receive()
        }
    }

    override fun unlock(owner: Any?) {
        val newReadersCount = pendingCount.addAndGet(MAX_READERS)
        if (newReadersCount >= MAX_READERS) {
            throw overUnlockedException()
        }
        repeat(newReadersCount) {
            readPermissions.trySend(Unit)
        }
        mutex.unlock(owner)
    }

    override fun tryLock(owner: Any?): Boolean {
        if (!mutex.tryLock(owner)) {
            return false
        }
        if (!pendingCount.compareAndSet(0, -MAX_READERS)) {
            mutex.unlock(owner)
            return false
        }
        return true
    }

    override fun holdsLock(owner: Any): Boolean = mutex.holdsLock(owner)
}

// not used, but may need it again to test some old school concurrent algorithms later
internal class Condition(val mutex: Mutex = Mutex()) {
    private val signals = Channel<Unit>()

    fun signal(): Boolean = signals.trySend(Unit).isSuccess

    suspend fun waitSignalUnlocked(owner: Any? = null) {
        mutex.unlock(owner)
        signals.receive()
        mutex.lock(owner)
    }

    suspend fun waitSignal() {
        signals.receive()
    }
}

[UnderMybrella]

I've been working on a semaphore-backed implementation of this, it's not super clean at the moment but I think I've got a basic iteration working over here. I'm gonna do some polishing before submitting a PR though, make sure it's working as intended.

[qwwdfsad]

We still haven't decided whether we are going to provide our own RW-mutex (though we have a working prototype here: #2045), and we are not ready to accept a PR with that either. Writing your own might be a tricky task, so for your own use I suggest using a wrapper over Semaphore using any technique described above

[kevincianfarini]

It's been a few years. Any chance we can get this re-looked at? In just the past few months several people have been redirected to this Github issue, including myself.

1. https://kotlinlang.slack.com/archives/C1CFAFJSK/p1717606632683559
2. https://kotlinlang.slack.com/archives/C1CFAFJSK/p1710684177345599
3. https://kotlinlang.slack.com/archives/C1CFAFJSK/p1625331416161000
4. https://kotlinlang.slack.com/archives/C1CFAFJSK/p1700152822282589
5. https://kotlinlang.slack.com/archives/C0B8MA7FA/p1640832952043000?thread_ts=1640810328.037800&cid=C0B8MA7FA
6. https://kotlinlang.slack.com/archives/C1H43FDRB/p1582641977015100?thread_ts=1581601415.043900&cid=C1H43FDRB

[glcanvas]

task has been created in 2017,
now 2024
maybe jet brains allocate a budget for this task?

[qwwdfsad]

@kevincianfarini thanks for the comprehensive list! It indeed helps

[MSDarwish2000]

Thanks to @youcef-debbah, I have written a simple ReadWriteMutex based on the Go's implementation (MayakaApps/KMP-RWMutex). It is different in that it uses Semaphore/Mutex instead of Channels, is tested, and documented, and have different API. The source code is here.

If you prefer @youcef-debbah's implementation, a small fix is needed. The readPermissions and writePermissions should be buffered to fix race situation when the permission is sent before it is received. So that it should be:

val writePermissions = Channel<Unit>(capacity = 1)
val readPermissions = Channel<Unit>(capacity = AbstractReadOrWriteMutex.MAX_READERS)

[youcef-debbah]

@MSDarwish2000 thanks for taking the time to write a documented solution, I think I will do the same once I have some free time (the KMP community is in dire need of all kinds of quality contributions)

concerning your "fix", using a channel in Kotlin with such a huge capacity defies the very purpose of a ReadWriteMutex (maximizing performance) because this will cause an array with MAX_READERS size to be created each time you create a new Mutex (you should never create such array EVER!)

now about the bug in my code, to be honest, I have no idea right now (I have completely forgotten about this code), I need to re-read my code, understand it first, write some real tests, and then fix any potential bugs, I will do this the next time a have some free time (probably in a few weeks) but until that time can't your fix be implemented using a linked list channel like this:

val writePermissions = Channel<Unit>(capacity = 1)
val readPermissions = Channel<Unit>(capacity = Channel.UNLIMITED)

it's the same except that no huge array is created for each channel (only a simple linked list), what do you think @MSDarwish2000?

in any case, I recommend to anyone interested in this to use @MSDarwish2000 implementation for now, once I have a well-tested solution that has better performance than his I will let you know :D it's not like his implementation particularly bad, I'm just into this kind of challenge :3

[MSDarwish2000]

@youcef-debbah Thank you for your response. Actually, I don't think that such an array is pre-allocated. Through a quick lock on the current implementation and tests, the Channel's buffer is expanded dynamically. As far as my tests are concerned, there is no considerable difference (if any) in memory usage between the two capacities.

Yet, I'd prefer your suggestion of usimg Channel.UNLIMITED for the potential of further optimizations. BTW, the memory usage concern, at least theoretically, is what made me prefer Semaphore to Channel. But for CPU time, the two implementations have no significant difference (again by my tests).

As long as it is a friendly challenge, I'm more than happy with it ❤️. Feel free to use the tests from the linked repo as they are actually translated from Go's tests. Also, have a look at the documentation of my code, your suggestions or contributions are welcome.

To make the discussion more fruitful, I suspect that BenchmarkRWMutexUncontended in rwmutex_test.go results in deadlocks as even Go themselves clearly mention that recursive read-locking is prohibited, yet there is a nested read-lock. I know almost nothing about Go, so I'm not sure if it is a negligible bug in a benchmark function, or it would undergo some compilation optimizations that would make the code work fine. Anyway, I didn't have the time to translate the benchmarks.

I hope that I'm not abusing this issue by going slightly off-topic.

[LouisCAD]

@youcef-debbah No huge array is created anymore with Channel (unless it's actually filled up to a huge level I think).

Version 1.7.0, released more than a year ago, brought a new implementation that leverages a dynamic buffer.

[code-hunger]

Here is an attempt without magic constants that tries to generalize the particularities of reads and writes. My goal is to give the end programmer full control over how locks are granted. Here's what I came up with (full code gist). I have not done extensive testing, I'm open to suggestions about that.

Define a general MultiLock<S, T> over a lock state type S and a waiter tag type W - for the most common scenario,

1. S is the Free/Read(int)/Write class hierarchy (instantiate it with immutable types)
2. W is a simple Read/Write enum (to be used in the waiting list later or determine priorities)

class MultiLock<S, W>(
    private var state: S,
    private val waiters: Queue<Waiter<W>>
) {
    class Waiter<W>(
        val waiterType: W,
        val awake: () -> Unit
    )

    interface LockStrategy<S, W> {
        fun acquire(state: S): AcquireDecision<S, W>
        fun release(state: S, waiters: Queue<Waiter<W>>): S
    }

    sealed interface AcquireDecision<S, W> {
        data class Allow<S, W>(val newState: S) : AcquireDecision<S, W>
        data class Block<S, W>(val waiterType: W) : AcquireDecision<S, W>
    }

    private val mutex = Mutex()

    suspend fun <R> withLock(
        lockStrategy: LockStrategy<S, W>,
        f: suspend () -> R
    ): R { /* see impl in gist */ }
    private suspend fun LockStrategy<S, W>.lock()  { /* see impl in gist */ }
    private suspend fun LockStrategy<S, W>.lock()  { /* see impl in gist */ }
}

The multilock:

• is initialized with an initial state + a container for waiters
• allows to run functions with a locking strategry (that abstraacts over Reads and Writes)
• the decision of a lock acquire attempt
  • either succeeds (Allow) with a new state,
  • or fails (Block) with a waiter type (to be stored in the waiters list)
• handles the whole suspend/resume functionality in the lock() functions and exposes only the Waitert::awake to the user, who decides how and when to awake blocked waiters.

This makes writing a read-write lock into mere manipulation of queues and datatype logistics, here's how the writer lock strategy looks like (full code in gist):

class ReadWriteLock {
    private val multiLock = MultiLock<State, WaiterType>(State.Free, LinkedList())

    sealed interface State {
        data class Read(var size: Int) : State
        data object Write : State
        data object Free : State
    }

    enum class WaiterType { Reader, Writer }

    suspend fun <R> write(f: suspend () -> R): R = multiLock.withLock(WriteLockStrategy, f)

    private object WriteLockStrategy : LockStrategy<State, WaiterType> {
        override fun acquire(state: State): MultiLock.AcquireDecision<State, WaiterType> =
            when (state) {
                State.Free -> Allow(State.Write)
                else -> Block(WaiterType.Writer)
            }

        override fun release(state: State, waiters: Queue<Waiter<WaiterType>>): State {
            if (waiters.peek().waiterType == WaiterType.Writer) {
                waiters
                    .poll()
                    .awake()
                return State.Write
            }

            var readCount = 0
            while (waiters.peek().waiterType == WaiterType.Reader) {
                waiters
                    .poll()
                    .awake()
                readCount++
            }
            return if (readCount > 0)
                State.Read(readCount)
            else
                State.Free
        }
    }

  // read is analogous

Notice how the WriteLockStrategy functions are not even suspend. They are called by MultiLock under protection of the mutex so it is safe to wake up whomever we want.

We can pass whatever instance of Queue we want (best if constructed in-place to prevent ownership leaks) and get control over priorities - be it "first come first serve" or "all reads first".

We can also add other waiter enum values (to distinguish e.g. FastRead vs SlowRead (Write)) or even use a full data class.

The critical part is in the handling of AcquireDecision in MultiLock::lock:

            when (val acquireDecision = acquire(state)) {
                is Allow ->
                    state = acquireDecision.newState

                is Block -> suspendCoroutine { continuation ->
                    waiters.add(Waiter(acquireDecision.waiterType, continuation))
                    // we suspend to let others finish
                    // so we have to release the lock now
                    mutex.unlock(owner)

On Block we suspend until resumed by the awake() method on the newly added waiter, called by some lock strategy's release.