Rust：如何获取含被取消实例的两个copy future的返回值？

要解决这个问题，核心是避免在select!完成一个分支后直接丢弃未完成的future，而是主动等待它结束（即使是因取消/流关闭而终止），从而拿到它已拷贝的字节数。

解决方案代码 #

async fn copy(a: ReadStream, b: WriteStream) -> usize {
    let mut total = 0;
    let mut buf = [0; 4096];
    loop {
        match a.read(&mut buf).await {
            Ok(0) => break, // 读取到流末尾，结束拷贝
            Ok(n) => {
                if b.write_all(&buf[..n]).await.is_ok() {
                    total += n;
                } else {
                    break; // 写入失败，终止拷贝
                }
            }
            Err(_) => break, // 读取失败，终止拷贝
        }
    }
    total
}

async fn bidirectional_copy(a_read: ReadStream, b_write: WriteStream, b_read: ReadStream, a_write: WriteStream) -> (usize, usize) {
    let a_to_b = copy(a_read, b_write).fuse();
    let b_to_a = copy(b_read, a_write).fuse();
    pin_mut!(a_to_b, b_to_a);

    let (bytes_a_to_b, bytes_b_to_a) = select! {
        res_a = a_to_b => {
            // 第一个拷贝任务完成，等待第二个任务结束（此时流可能已关闭，任务会快速终止）
            let res_b = b_to_a.await;
            (res_a, res_b)
        },
        res_b = b_to_a => {
            // 第二个拷贝任务完成，等待第一个任务结束
            let res_a = a_to_b.await;
            (res_a, res_b)
        },
    };

    (bytes_a_to_b, bytes_b_to_a)
}

关键说明 #

1. 避免直接丢弃future：原代码中select!分支执行时，未完成的future会被丢弃，导致无法获取其返回值。修改后，在第一个任务完成后，主动await另一个任务，确保拿到它累计的拷贝字节数。
2. copy函数的鲁棒性：copy函数需要处理流的读取/写入错误或结束情况，在终止时返回已拷贝的总字节数，这样即使任务因取消（比如对应流被关闭）而终止，也能正确返回进度。
3. 使用fuse和pin_mut!：fuse()确保future不会被重复poll，pin_mut!固定future的内存位置，满足异步任务的Pin要求。

内容的提问来源于stack exchange，提问作者Ramon de Klein