`mpsc::Sender::try_send` does not work!

[andreytkachenko]

Consider this example:

use futures::channel::mpsc;

fn main() {
    let (tx, rx) = mpsc::channel(1);
    
    for i in 0..100 {
        tx.clone().try_send(i).unwrap();
    }
    
    println!("BAM!!!")
}

And you will see the message in the console!

[binier]

It seems channel isn't respecting bound and sends succeeds inspite buffer size limit. All the values are sent successfuly though as shown in here (link to the playground).

fn main() {
    let (tx, mut rx) = mpsc::channel(1);
    
    for i in 0..100 {
        tx.clone().try_send(i).unwrap();
    }
    drop(tx);
    
    for i in 0..100 {
        assert!(matches!(rx.try_next(), Ok(Some(x)) if x == i));
    }
    
    assert!(matches!(rx.try_next(), Ok(None)));
    println!("BAM!!!");
}

[binier]

futures-rs/futures-channel/src/mpsc/mod.rs

Lines 352 to 358 in da71932

	/// Creates a bounded mpsc channel for communicating between asynchronous tasks.
	///
	/// Being bounded, this channel provides backpressure to ensure that the sender
	/// outpaces the receiver by only a limited amount. The channel's capacity is
	/// equal to `buffer + num-senders`. In other words, each sender gets a
	/// guaranteed slot in the channel capacity, and on top of that there are
	/// `buffer` "first come, first serve" slots available to all senders.

Since you are cloning tx when sending value through the channel, as documented, you get +1 slot for each sender, so there are no issues in code you wrote above, it works as designed.

[andreytkachenko]

Yep, I saw this comment. But!
Consider this example:

use futures::SinkExt;
use futures::channel::mpsc;

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(1);
    
    for i in 0..100 {
        tx.clone().send(i).await.unwrap();
    }
    
    println!("BAM!!!")
}

aaand it works as I would expect!

PS: buffer + num-senders the num-senders is a number of currently alive senders or all once created clones (even dropped)?

[binier]

PS: buffer + num-senders the num-senders is a number of currently alive senders or all once created clones (even dropped)?

That +1 slot only belongs to one sender. So no matter how much cloning you do, if channel's buffer that you set(in this case 1) is full, sender can only use it's own +1 slot left - hence send only 1 more item.

[andreytkachenko]

Am I right with the statement:

use futures::channel::mpsc;

fn main() {
    let (tx, rx) = mpsc::channel(1); // channel capacity 1
    
    for i in 0..100 {
        let mut clone = tx.clone();  // channel capacity 2
        clone.try_send(i).unwrap(); 
        drop(clone);  // channel capacity back to 1
    }
    
    println!("BAM!!!")
}

Or channel capacity remains 2, 3, 4 .... ??

[binier]

Channel capacity doesn't change at all. It stays the same even when you clone the senders.

So when you try to send value using tx when capacity is reached, instead of appending value to the channel queue, send task gets parked (appended to the parked_queue) and only one SenderTask can be parked per clone of Sender.

[andreytkachenko]

Is this behavior "as intended" or there is an issue? Should I come up with some workaround and wait when it will be fixed, or I need completely redesign interface of my application, because in that case this channels are completely useless in sync context (senders are require &mut and clone does not solve the problem) and I didn't find any other implementation of async channels with poll_ready/try_send interface.

[binier]

Yes, this is an intended behavior.

What are you trying to achieve though? Why is this behavior an issue?

[andreytkachenko]

Here is my workaround https://github.com/andreytkachenko/messagebus/blob/master/src/receivers/mpsc_futures.rs.
Basically I need channels with poll_ready/try_send interface where try_send and poll_ready will be by &self not &mut self.