ConflatedBroadcastChannel hangs in offer()

[chrisly42]

Hi there,

maybe there is a better way to achieve what I want, but with my current wrapper class around MutableMaps that allows coroutine waiting for key additions will at times get stuck and cause 100% cpu load. I'm using the recently added ConflatedBroadcastChannel class to inform all coroutines waiting for a key to arrive on a put operation.

Actually, I would only need a coroutine compatible wait/notify primitives... hints?

Anyway, when I pause the threads during the lockup, I will get a stacktrace that may look like this one:

<15> ForkJoinPool.commonPool-worker-3"@830,030,709,776 in group "main": RUNNING
getNext:121, LockFreeLinkedListNode {kotlinx.coroutines.experimental.internal}
takeFirstReceiveOrPeekClosed:920, AbstractSendChannel {kotlinx.coroutines.experimental.channels}
offerInternal:54, AbstractSendChannel {kotlinx.coroutines.experimental.channels}
offerInternal:39, ConflatedChannel {kotlinx.coroutines.experimental.channels}
offerInternal:268, ConflatedBroadcastChannel$Subscriber {kotlinx.coroutines.experimental.channels}
offerInternal:239, ConflatedBroadcastChannel {kotlinx.coroutines.experimental.channels}
offer:219, ConflatedBroadcastChannel {kotlinx.coroutines.experimental.channels}
put:11, AwaitableMap {com.example}

Any ideas what's going on here?

package com.example

import kotlinx.coroutines.experimental.channels.ConflatedBroadcastChannel

class AwaitableMap<K : Any, V : Any>(val delegateMap: MutableMap<K, V>) : MutableMap<K, V> by delegateMap {

    val channel = ConflatedBroadcastChannel<K>()

    override fun put(key: K, value: V): V? {
        val putVal = delegateMap.put(key, value)
        channel.offer(key)
        return putVal
    }

    suspend fun await(key: K): V {
        return delegateMap[key] ?: channel.open().use {
            do {
                val fetched = delegateMap[key]
                if (fetched != null) {
                    return@use fetched
                }
                it.receive()
            } while (true)
            return@use null
        }!!
    }
}

[elizarov]

Hm... ConflatedBroadcastChannel is well, conflated. If you have two back-to-back invocations to put with different keys, then the second offer will replace the previous one (it will be conflated) and the waiter will never wake up.

There is no await/notify-like not event-like mechanism in coroutines out-of-the-bix yet. Mutex should support conditions but it does not do so now, so you'll leave this ticket open.

If wonder what is the larger context of the problem that you are trying to solve with AwaitableMap. What are you planning to use it for?

[chrisly42]

Thanks for your quick reply. As far as I understood, there is no problem for me to use the conflated version as I am not using the value transmitted for anything. The offer() should just wake up very listener that currently is "subscribed" by having a channel open. I'm not sure why the waiter would not wake up if any put is broadcasted to all the receivers.

Moreover, I still think there is a bug in the implementation of ConflatedBroadcastChannel somewhere because it works well in most cases but in higher loads, the put-replacement routine gets stuck in an infinite loop in offer which is considered to be a non-blocking call. In no case, offer() should do that.

My use case is the following. I have an object that several child objects that are to be loaded asynchronously via HTTP (slow) but I want to update new items in the UI as soon as they are completely fetched by the coroutine. There is another coroutine that already knows that the will be there as some point in the future, waiting for its arrival (this one is calling the await() method). It will then process the item data before displaying it on the UI.

I wrote a workaround that uses a map of channels, that works for my use-case right now, but is not generic.

I'm fairly new to Kotlin and Android programming, but I think Kotlin and coroutines are just wonderful :) Maybe there are already implementations that I'm not aware of (I haven't checked out RxJava yet)...

[elizarov]

The stuck invocation to offer is definitely a bug. Thanks for the report!

I've just cursory glanced over your code and how I see how you are actually using it. I was misled by the fact that you send a key to the channel, but you don't actually use this key in await. Indeed, you use ConflatedBroadcastChannel as a simple event mechanism to notify waiter that something had changed. It should work this way, though a more efficient event-like mechanism is definitely needed (but is missing now).

[elizarov]

Good news. I can reproduce your problem (offer hangs) with a stress test.

[elizarov]

Fixed in develop branch

[chrisly42]

Thanks a lot for the quick fix. Will stick to the alternate solution until next release though and then test it again. Your changes don't seem trivial, though :) Feel free to close the issue, if you think the stress test covers the problem.

[elizarov]

It turned out to be a whole complex of problems that happen when you concurrently close/send/close channels. It was also independently discovered by @cy6erGn0m