Broadcast close with CoroutineStart.DEFAULT produces an unexpected exception

[pauliusrum]

Kotlin: 1.3.61
Coroutines: 1.3.3

runBlocking {
    val a = produce { 
        while (true) send(1) 
    }
    val b = a.broadcast(start = CoroutineStart.DEFAULT)
    b.close()
}

Problem: send(1) throws an exception after b.close() is called.
Expected behavior: send(1) should suspend instead of throwing an exception.

[elizarov]

Why would you expect the sender to suspend instead of throwing an exception? See, conceptually broadcast() is designed as a consuming operation, just like all the other channel operators. The idea is that once you start to broadcast the original channel is not used for any other purpose, so now calling b.close() should not only terminate the broadcast but close the original channel, too.

[pauliusrum]

I see what you are saying but if we have the same example just with CoroutineStart.LAZY

runBlocking {
    val a = produce {
        while (true) send(1)
    }
    val b = a.broadcast()
    b.close()
}

In this case send(1) suspends instead of throwing an exception right? Why is it in one case it suspends in the other it throws an exception?

[elizarov]

That looks like a bug. Should be the same with lazy -- closing the broadcast channel should be closing the original channel.

[antanas-arvasevicius]

to clarify a bit:
Reason that this throws exception is here: "onCompletion = consumes()" in broadcast()
This is intended to cancel upstream and gets called only when BroadcastCoroutine completes.
Problem is that then we are calling "close()" it closes all broadcast subscriptions and itself first and only after that BroadcastCoroutine completes and consumes() get called.
I think it must be in this order:

when close() is called:

1. do cancel upstream (source channel)
2. then upstream is cancelled - close all subscription and broadcast itself.
3. finish BroadcastCoroutine

Currently it is:

1. close broadcast channel and all its subscriptions
2. complete coroutine
3. cancel upstream (source channel)

fun <E> ReceiveChannel<E>.broadcast(
    capacity: Int = 1,
    start: CoroutineStart = CoroutineStart.LAZY
): BroadcastChannel<E> =
    GlobalScope.broadcast(Dispatchers.Unconfined, 
capacity = capacity, 
start = start, 
onCompletion = consumes()) {
        for (e in this@broadcast) {
            send(e)
        }
    }

[elizarov]

Some additional explanaition here.

When you close a broadcast channel it makes the concurrent send operation in your broadcast { .. } to fail. If you want to stop broadcasting coroutine you should use cancel instead. I'm adding more documentation with respect to this in PR #1717 and also fix a problem with the fact that closing a lazily broadcast coroutine does nothing. The fix is to actually start it so that it has its chance to fail.