Pfthb

I am trying to understand how futures::sync::mpsc::Receiver works. In the below example, the receiver thread sleeps for two seconds and the sender sends every second.

I expect that the sender will need to be blocked because of the wait and then send when the buffer is released.

What I see instead is that it is deadlocked after a time. Increasing the buffer of the channel only extends the time until it is blocked.

What should I do to make the sender send data when the buffer is available and put some backpressure to the sender in such cases? futures::sync::mpsc::channel has its own documentation, but I do not understand how to use it properly.

extern crate futures;
extern crate tokio_core;

use std::thread, time;

use futures::sync::mpsc;
use futures::Future, Sink, Stream;

use tokio_core::reactor::Core;

#[derive(Debug)]
struct Stats 
 pub success: usize,
 pub failure: usize,


fn main() 
 println!("Received");
 let delay = time::Duration::from_secs(2);
 thread::sleep(delay);

 match res 
 Ok(_) => stats.success += 1,
 Err(_) => stats.failure += 1,
 
 println!("stats = :?", stats);

 Ok(())
 );

 core.run(f2).expect("Core failed to run");

1. Never call wait inside of a future. That's blocking, and blocking should never be done inside a future.




2. Never call sleep inside of a future. That's blocking, and blocking should never be done inside a future.




3. Channel backpressure is implemented by the fact that send consumes the Sender and returns a future. The future yields the Sender back to you when there is room in the queue.

extern crate futures; // 0.1.25
extern crate tokio; // 0.1.11

use futures::future, sync::mpsc, Future, Sink, Stream;
use std::time::Duration;
use tokio::timer::Interval;

#[derive(Debug)]
struct Stats 
 pub success: usize,
 pub failure: usize,


fn main() 
 let (tx, rx) = mpsc::channel::<Result<(), ()>>(1);

 tokio::spawn( 
 tx.send(Ok(())).map_err()
 .map(drop) // discard the tx
 );

 let mut stats = Stats 
 success: 0,
 failure: 0,
 ;

 let i = Interval::new_interval(Duration::from_millis(20))
 .map_err());