Async Channels

Several crates have support for asynchronous channels. For instance tokio:

  1. use tokio::sync::mpsc;
  3. async fn ping_handler(mut input: mpsc::Receiver<()>) {
  4.     let mut count: usize = 0;
  6.     while let Some(_) = input.recv().await {
  7.         count += 1;
  8.         println!("Received {count} pings so far.");
  9.     }
  11.     println!("ping_handler complete");
  12. }
  14. #[tokio::main]
  15. async fn main() {
  16.     let (sender, receiver) = mpsc::channel(32);
  17.     let ping_handler_task = tokio::spawn(ping_handler(receiver));
  18.     for i in 0..10 {
  19.         sender.send(()).await.expect("Failed to send ping.");
  20.         println!("Sent {} pings so far.", i + 1);
  21.     }
  23.     drop(sender);
  24.     ping_handler_task.await.expect("Something went wrong in ping handler task.");
  25. }

This slide should take about 8 minutes.

  • Change the channel size to 3 and see how it affects the execution.

  • Overall, the interface is similar to the sync channels as seen in the morning class.

  • Try removing the std::mem::drop call. What happens? Why?

  • The Flume crate has channels that implement both sync and async send and recv. This can be convenient for complex applications with both IO and heavy CPU processing tasks.

  • What makes working with async channels preferable is the ability to combine them with other futures to combine them and create complex control flow.