消除

丢弃 Future 意味着无法再对其进行轮询。这称为 取消，在任何 await 点都可能发生。请务必小心谨慎，确保即使 Future 任务被取消，系统也能正常运行。例如，系统不应死锁或丢失数据。

  1. use std::io::{self, ErrorKind};
  2. use std::time::Duration;
  3. use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream};
  5. struct LinesReader {
  6.     stream: DuplexStream,
  7. }
  9. impl LinesReader {
  10.     fn new(stream: DuplexStream) -> Self {
  11.         Self { stream }
  12.     }
  14.     async fn next(&mut self) -> io::Result<Option<String>> {
  15.         let mut bytes = Vec::new();
  16.         let mut buf = [0];
  17.         while self.stream.read(&mut buf[..]).await? != 0 {
  18.             bytes.push(buf[0]);
  19.             if buf[0] == b'\n' {
  20.                 break;
  21.             }
  22.         }
  23.         if bytes.is_empty() {
  24.             return Ok(None);
  25.         }
  26.         let s = String::from_utf8(bytes)
  27.             .map_err(|_| io::Error::new(ErrorKind::InvalidData, "not UTF-8"))?;
  28.         Ok(Some(s))
  29.     }
  30. }
  32. async fn slow_copy(source: String, mut dest: DuplexStream) -> std::io::Result<()> {
  33.     for b in source.bytes() {
  34.         dest.write_u8(b).await?;
  35.         tokio::time::sleep(Duration::from_millis(10)).await
  36.     }
  37.     Ok(())
  38. }
  40. #[tokio::main]
  41. async fn main() -> std::io::Result<()> {
  42.     let (client, server) = tokio::io::duplex(5);
  43.     let handle = tokio::spawn(slow_copy("hi\nthere\n".to_owned(), client));
  45.     let mut lines = LinesReader::new(server);
  46.     let mut interval = tokio::time::interval(Duration::from_millis(60));
  47.     loop {
  48.         tokio::select! {
  49.             _ = interval.tick() => println!("tick!"),
  50.             line = lines.next() => if let Some(l) = line? {
  51.                 print!("{}", l)
  52.             } else {
  53.                 break
  54.             },
  55.         }
  56.     }
  57.     handle.await.unwrap()?;
  58.     Ok(())
  59. }

  • 编译器无法确保取消操作的安全性。您需要阅读 API 文档，并考虑 async fn 所持状态。

  • 与 panic 和 ? 不同，取消属于正常控制流的一部分（而非错误处理）。

  • 该示例丢失了字符串的某些部分。

    • 每当 tick() 分支先完成操作时，next() 及其 buf` 均会被丢弃。

    • 通过将 buf 整合到结构体中，`LinesReader 可以确保取消操作的安全性：

      1. #![allow(unused)]
      2. fn main() {
      3. struct LinesReader {
      4.     stream: DuplexStream,
      5.     bytes: Vec<u8>,
      6.     buf: [u8; 1],
      7. }
      9. impl LinesReader {
      10.     fn new(stream: DuplexStream) -> Self {
      11.         Self { stream, bytes: Vec::new(), buf: [0] }
      12.     }
      13.     async fn next(&mut self) -> io::Result<Option<String>> {
      14.         // prefix buf and bytes with self.
      15.         // ...
      16.         let raw = std::mem::take(&mut self.bytes);
      17.         let s = String::from_utf8(raw)
      18.         // ...
      19.     }
      20. }
      21. }

  • Interval::tick is cancellation-safe because it keeps track of whether a tick has been ‘delivered’.

  • AsyncReadExt::read is cancellation-safe because it either returns or doesn’t read data.

  • AsyncBufReadExt::read_line is similar to the example and isn’t cancellation-safe. See its documentation for details and alternatives.