How to tell when a channel is disconnected?

[dcarosone]

I would like an is_connected() / is_disconnected() method on channels.

In a simple app, with some workers cloning the rx of a channel, and the main thread putting work into the channel, the channel is disconnected once the main thread drops the send side. The workers continue draining the queue until they get None when the queue is empty and disconnected. All works as intended.

However, I have another thread as well, and this makes the termination condition more complex. That thread is printing progress statistics. I want it to exit when the queue is empty and disconnected; the app can then wait on just this thread, rather than join on all the workers as well. But at present it can only determine when the queue is empty.

Further discussion and example code here: https://users.rust-lang.org/t/test-for-when-crossbeam-channel-is-disconnected/23554

It looks like this idea has come up before; it's mentioned in #236

[BurntSushi]

The standard and simplest way to do this is to simply use another channel. You can send a message after your work is done, thereby alerting your stats thread that it can quit. (You don't even need to send a message, you can just drop this signaling channel since receives on a disconnected channel never block.)

[ghost]

Exactly what @BurntSushi said, but I'll just elaborate a little bit...

Create a channel whose disconnection signals that worker threads have completed their work:

let (done, quit) = bounded(0);

Give each worker thread a clone of done and drop it when the main work channel gets exhausted.

Then, in the stats thread, you can do something like this:

let ticker = tick(Duration::from_secs(60));
loop {
    select! {
        recv(ticker) -> _ => STAT.print(),
        recv(quit) -> _ => break,
    }
}

When all clones of done get dropped, this special channel gets disconnected, so the recv(quit) branch in select! will get activated.

[dcarosone]

Yeah, I was generally aware of this pattern, among a collection of other options. Thanks for the loop { select! {}} example, @stjepang - that's not quite as cumbersome as I expected, I was trying to think how to fit it into the existing if - but it still feels a little like a work-around.

[ghost]

This is really the idiomatic approach :)

Disconnection is only used to inform send/recv methods that the other side of the channel went away. It's not really a flag intended to be polled independently of blocking methods.

To make an analogy with iterators, this is kind of similar to how iterators get exhausted once .next() returns None, but there's no method named .is_exhausted() that would tell if there are more items without consuming them.

(There exist peekable iterators, but they're more like plumbing a channel to a peekable bounded(1) channel that simply buffers the next element).

[ghost]

Closing the issue. If you have any other questions, I'd be happy to help!

[Timmmm]

I think this is a bit unfortunate. I have code that receives some data from a channel, and sends it on another one. Ideally I could do something like this:

loop {
  select! {
     recv(input) -> msg => output.send(process(msg))?,
     closed(output) -> return Err(...),
  }
}

Adding another channel just to send a "closed" signal seems inelegant and a fragile duplication of information - I now have to do extra work to make sure that the "closed" signal is sent iff output is closed. The above code is a lot nicer as a user.

Perhaps it is simply too difficult to implement?