Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/src/channels/ArrayBroadcastChannel.kt (Kotlin)

Code before:
    private fun computeMinHead(): Long {
        var minHead = Long.MAX_VALUE
        for (sub in subscribers)
            minHead = minHead.coerceAtMost(sub.subHead) // volatile (atomic) reads of subHead
        return minHead
    }

    @Suppress("UNCHECKED_CAST")
    private fun elementAt(index: Long): E = buffer[(index % capacity).toInt()] as E

    private class Subscriber<E>(
        private val broadcastChannel: ArrayBroadcastChannel<E>
    ) : AbstractChannel<E>(), ReceiveChannel<E> {
        private val subLock = ReentrantLock()

        private val _subHead = atomic(0L)
        var subHead: Long // guarded by subLock
            get() = _subHead.value
            set(value) { _subHead.value = value }

        override val isBufferAlwaysEmpty: Boolean get() = false
        override val isBufferEmpty: Boolean get() = subHead >= broadcastChannel.tail
        override val isBufferAlwaysFull: Boolean get() = error("Should not be used")
        override val isBufferFull: Boolean get() = error("Should not be used")

        override fun onCancelIdempotent(wasClosed: Boolean) {
            if (wasClosed) {
                broadcastChannel.updateHead(removeSub = this)
                subLock.withLock {
                    subHead = broadcastChannel.tail
                }
            }
        }

        // returns true if subHead was updated and broadcast channel's head must be checked
        // this method is lock-free (it never waits on lock)
        @Suppress("UNCHECKED_CAST")
        fun checkOffer(): Boolean {
            var updated = false
            var closed: Closed<*>? = null
        loop@
            while (needsToCheckOfferWithoutLock()) {
                // just use `tryLock` here and break when some other thread is checking under lock
                // it means that `checkOffer` must be retried after every `unlock`
                if (!subLock.tryLock()) break
                val receive: ReceiveOrClosed<E>?
                var result: Any?
                try {
                    result = peekUnderLock()
                    when {
                        result === POLL_FAILED -> continue@loop // must retest `needsToCheckOfferWithoutLock` outside of the lock

Review comment (suggestion):
It looks like `onCancelIdempotent` should be overridden (to do nothing) to avoid an extra bit of work.


Code after:
    private fun computeMinHead(): Long {
        var minHead = Long.MAX_VALUE
        for (sub in subscribers)
            minHead = minHead.coerceAtMost(sub.subHead) // volatile (atomic) reads of subHead
        return minHead
    }

    @Suppress("UNCHECKED_CAST")
    private fun elementAt(index: Long): E = buffer[(index % capacity).toInt()] as E

    private class Subscriber<E>(
        private val broadcastChannel: ArrayBroadcastChannel<E>
    ) : AbstractChannel<E>(), ReceiveChannel<E> {
        private val subLock = ReentrantLock()

        private val _subHead = atomic(0L)
        var subHead: Long // guarded by subLock
            get() = _subHead.value
            set(value) { _subHead.value = value }

        override val isBufferAlwaysEmpty: Boolean get() = false
        override val isBufferEmpty: Boolean get() = subHead >= broadcastChannel.tail
        override val isBufferAlwaysFull: Boolean get() = error("Should not be used")
        override val isBufferFull: Boolean get() = error("Should not be used")

        override fun close(cause: Throwable?): Boolean {
            val wasClosed = super.close(cause)
            if (wasClosed) {
                broadcastChannel.updateHead(removeSub = this)
                subLock.withLock {
                    subHead = broadcastChannel.tail
                }
            }
            return wasClosed
        }

        // returns true if subHead was updated and broadcast channel's head must be checked
        // this method is lock-free (it never waits on lock)
        @Suppress("UNCHECKED_CAST")
        fun checkOffer(): Boolean {
            var updated = false
            var closed: Closed<*>? = null
        loop@
            while (needsToCheckOfferWithoutLock()) {
                // just use `tryLock` here and break when some other thread is checking under lock
                // it means that `checkOffer` must be retried after every `unlock`
                if (!subLock.tryLock()) break
                val receive: ReceiveOrClosed<E>?
                var result: Any?
                try {
                    result = peekUnderLock()