I'd like for someone to clarify as how does `Chann...

# coroutines

[ursus]

I'd like for someone to clarify as how does Channel behaves in the limit

val events = Channel<Event>()

scope.launch {
	events.send(SomeEvent(42))
}

differentScope.launch {
	for (event in events) {
		doSomethingBlocking(event)
	}
}

imagine this setup

so a randezevous channel, the semantics are events.send(SomeEvent(42)) won't resume until it's delivered

Is that really the case?

From what I see that loop is sugar for

while (iterator.hasNext()) {
    val element = iterator.next()
    // ... handle the element ...
}

So two suspension points..

and I'm not super clear on the docs of hasNext (kotlinlang.org/api/kotlinx.coroutines/…/has-next.html)

It seems hasNext is the one which dequeues the buffer, so .. is that what signals the events.send to resume? or is it the next? There is this note of "p*rompt cancellation guarantee"* but I can't really parse it.

In other words, can there be a situation where events.send resumes but the other side not even start doSomethingBlocking due to cancellation?

[Youssef Shoaib [MOD]]

If the channel has an element available, returns true and removes it from the channel.

next is not even suspend, so only hasNext can possibly be taking the element out.

[ursus]

facepalm sorry, idk what I though that

[ursus]

to the final question, what's your opinion? can it happen?

[Youssef Shoaib [MOD]]

Because of the prompt cancellation guarantee, some values retrieved from the channel can become lost. See the "Undelivered elements" section in the Channel documentation for details on handling undelivered elements.

The channel docs say:

The Channel() factory function has the optional parameter onUndeliveredElement. When that parameter is set, the corresponding function is called once for each element that was sent to the channel with the call to the send function but failed to be delivered, which can happen in the following cases:

[...]

• When the receiving operations like receive, onReceive, or hasNext throw an exception after retrieving the element from the channel because of being cancelled before the code following them had a chance to resume.

so it seems like a yes!

[ursus]

at which point the sender will appear as if it passed okay?

damn..that sort of breaks my worldview about Channels

pretending it was delivered when it wasn't...not a great default

now I have to .. idk, pass mutablestateflows and let receiver clear the pending state masquarading as event if I want a delivery guarantee?

[Youssef Shoaib [MOD]]

Imagine that every hasNext(), receive, etc actually has a coroutineContext.ensureActive() right after it. That should fix the mental model hopefully
If you really want, you can always force it to not happen:

withContext(NonCancellable) { channel.receive() }

You could pretty easily do the same with channel iterators (untested code):

fun <T> Channel<T>.nonCancellable(): ChannelIterator<T> {
  val iterator = iterator()
  object: ChannelIterator<T> by iterator {
    override suspend fun hasNext() = withContext(NonCancellable) { iterator.hasNext() }
  }
}
// you might need this as well, not sure
operator fun <T> ChannelIterator<T>.iterator(): ChannelIterator<T> = this

[ursus]

well I'd like for the loop be cancellable somehow, say due to app going to the background, at which point the sender should just suspend until receiver comes back

so not sure if NotCancellable is the way to go, what would make the loop uncancellable at all times right?

[ursus]

kotlinlang.org/api/kotlinx.coroutines/…/-channel

All suspending functions with channels provide prompt cancellation guarantee. If the job was cancelled while send or receive function was suspended, it will not resume successfully, even if it already changed the channel's state, but throws a CancellationException. With a single-threaded dispatcher like Dispatchers.Main, this gives a guarantee that the coroutine promptly reacts to the cancellation of its Job and does not resume its execution.

I'm really trying to understand this but somehow escapes me

it will not resume successfully, even if it already changed the channel's state

what is here meant to resume? the other side? or the same side (which got cancelled)

[Youssef Shoaib [MOD]]

I believe it's the receiving side won't resume successfully.
Again, just imagine that hasNext() is actually secretHasNext().also { coroutineContext.ensureActive() }
The iterator I've provided keeps the loop cancellable I'm pretty sure. It only wraps the hasNext in NonCancellable.

[ursus]

but what else would have the loop be cancellable? you said in the beginning only hasNext is suspending

and say there is no more suspension pointa in user code

means it would never cancel, no?

[Dmitry Khalanskiy [JB]]

To put it in very specific terms, prompt cancellation guarantee means that the following may happen:
• The channel is empty.
• hasNext suspends awaiting an element.
• Someone `send`s an element to the channel.
• hasNext grabs that element, and its code is getting scheduled to execute on its CoroutineDispatcher.
• While hasNext is in the dispatcher's queue, the coroutine running hasNext gets cancelled.
• (Here's where the prompt cancellation guarantee comes into play) Instead of running, hasNext finishes with a CancellationException. The element that hasNext grabbed has already left the channel, it's not going to be put back in, so now it's simply dropped (and onUndeliveredElement gets invoked).
Without prompt cancellation guarantee, hasNext would resume, even though its coroutine already got cancelled.

The secretHasNext().also { coroutineContext.ensureActive() } model is good!

[ursus]

at which point does however send resume? whats the trigger?
a) hasNext "grabbing" that element? Or
b) hasNext returning successfully?
c) something else?

[Dmitry Khalanskiy [JB]]

It's a).

[ursus]

I see thank you

is there a way for me to get the guarantee of receiving? i.e. b behavior somehow?

I dont mind value staying in the queue and getting processed later when conditions are met again, just dequeueing and not even starting processing..idk how to reason over such system, because this is not a function of backpressure or channel confkg or anything, all it seems I can do is to log in onUndeliveredElement and ..shrug

Is then there a better primitive?

[ursus]

maybe onUndelivered send it again?

[Dmitry Khalanskiy [JB]]

Certainly, there are different primitives for different needs, as well as ways to disable prompt cancellation for the channel. The question is, what are you trying to achieve? Could you please describe how you intend to use the channel?

maybe onUndelivered send it again?

Nope, won't work, onUndeliveredElement isn't a suspend callback but a normal function.

[ursus]

Well, the concrete situation here is the old "(android) navigate from view model after an async operation completes (by sending one off event)", where the viewmodel (the sending side) has a longer lifecycle than the ui (receiving side)

And randezevous channel seemed a good for the case, as it would seemingly wait for the receiver to come online

And if I made the handling code (body of channel for loop) blocking, I naively expected it to be effectively uncancellable once it starts, at which point I can treat successful send as "for sure handled fully"

[Dmitry Khalanskiy [JB]]

If the UI is the receiving side, that means the element got dropped because the UI got destroyed. What does the element represent? Why doesn't it get invalidated by the UI getting destroyed?

[ursus]

element represent a navigational command object instance, upon receiving which. the ui should mutate the backstack and thus actually causing navigation to happen (or call some other system api which depends on android activity context, say like permission request ui prompt)

Because on android uis can get destroyed temporarily (pausing/resuming when app goes to background) as well or rather, youre not supposed to cause navigation on background, since savedstate mechanism only snapshots onGoneToBackground once, so you have to queue the events somehow and only process once ui resumes

[ursus]

tldr; logically ui is scoped to the viewmodel, but physically it might not be there at all times and I intended the channel buffer to be a way to bridge the intermittent absence

[ursus]

Would a stateflow with pending event "state" i clear after better solution? But that sort of implies just one pending, which is not great

[ursus]

What are the primitives? I'm only aware of Channel, StateFlow, SharedFlow, is there something else?

[Dmitry Khalanskiy [JB]]

I do think that having some state representing the UI state, which state then gets updated according to, is a sound decision. StateFlow representing a queue of commands to execute on the UI looks like a good fit for your scenario, but maybe you could benefit from using Jetpack Compose, which is also built on that idea.

Other than that, it does look like your use case calls exactly for an atomic `Channel`: you send values where each value needs to be processed by exactly one consumer. Please take a look at github.com/Kotlin/kotlinx.coroutines/issues/4414#… for a way to disable prompt cancellation selectively. Here, you'd need to use onReceiveCatching as the select clause.

[ursus]

I've checked it out but it does look scary 😄

class AckedEventQueue<T> {
    fun add(event: T)
    suspend fun next(): Element<T>
    fun ack(id: Long)

    data class Element<T>(val id: Long, val event: T)
}

what would you say to a purpose built thing, something like this? So it explicitly dequeues not in next but after `ack`ing it. But it does mix suspend/blocking semantics so idk, what do you think? Or would the atomic channel extensions be more idiomatic?

(Also the deque backing this technically is also state, so google should be happy also 😄 )

[Dmitry Khalanskiy [JB]]

The API itself is reasonable, though I'm wondering how you're planning to implement it. If it's a wrapper around the atomic consumption from a Channel, sure, makes sense! I personally would go for an extension like atomicReceive().