Tokio中File::poll_write()的行为及重复调用疑问

核心疑问 #

针对Tokio的AsyncWrite trait中poll_write方法的设计，有两个关键困惑：

1. 首次调用poll_write传入的缓冲区buf，在Future被唤醒再次执行poll时，之前的buf会被丢弃吗？
2. 示例代码中，当首次poll_write返回Pending后被唤醒，后续poll时的重复逻辑执行、poll_write的具体行为是怎样的？

示例代码（修正编译错误后） #

use tokio::io::{AsyncWrite, File};
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

struct MyWriter {
   file: File,
   my_val: i32
}

impl MyWriter {
   fn new(file: File) -> Self {
      Self {
         file,
         my_val: 0
      }
   }

   fn get_val(&mut self) -> i32 {
      let old = self.my_val;
      self.my_val += 1;
      old
   }

   fn my_logic(&mut self) {}
}

impl Future for MyWriter {
    type Output = i32;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();
        // 执行额外逻辑
        this.my_logic();
        let mut bytes = vec![];
        // 修正：将i32转为u8存入缓冲区
        bytes.push(this.get_val() as u8);

        match Pin::new(&mut this.file).poll_write(cx, &bytes) {
           Poll::Ready(_) => Poll::Ready(0),
           _ => Poll::Pending
        }
    }
}

// 调用示例
// let writer = MyWriter::new(file);
// let ans = writer.await;

问题解答 #

1. 缓冲区buf的生命周期 #

poll_write每次调用时传入的buf都是临时的，首次调用的buf在方法返回后就会被完全丢弃。Tokio的异步IO设计遵循规则：poll_write仅负责尝试写入当前传入的buf内容，它不会保存buf的引用或副本。一旦返回Pending，下次调用poll_write必须重新传入要写入的内容——框架不会帮你留存之前的写入请求。

2. 示例代码的具体行为 #

• 重复执行逻辑：当首次poll返回Pending，任务被Waker唤醒后再次执行poll时，self.my_logic()会被完整执行第二次。因为poll方法的设计就是：每次任务被唤醒，都会从头执行poll内的所有代码，直到返回Ready或者再次进入Pending状态。
• poll_write的写入行为：
  • 首次poll时，传入的是0对应的字节，若此时文件IO未就绪，poll_write返回Pending，并记录当前任务的Waker，等待IO资源可用。
  • 当Waker触发、再次调用poll时，你传入的是1对应的字节，此时poll_write会尝试写入新的字节1，完全不会处理之前的字节0。因为第一次的buf已经被销毁，poll_write没有留存任何之前的写入请求——每次调用都是独立的写入尝试。

正确的异步写入姿势 #

如果要保证写入请求不丢失，你需要自己在结构体中保存待写入的数据缓冲区，确保未完成的写入请求能在下次poll时继续处理：

use tokio::io::{AsyncWrite, File, Error as IoError};
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

struct MyWriter {
    file: File,
    my_val: i32,
    pending_buf: Vec<u8>, // 保存未完成的写入数据
}

impl MyWriter {
    fn new(file: File) -> Self {
        Self {
            file,
            my_val: 0,
            pending_buf: Vec::new(),
        }
    }

    fn get_val(&mut self) -> i32 {
        let old = self.my_val;
        self.my_val += 1;
        old
    }

    fn my_logic(&mut self) {}
}

impl Future for MyWriter {
    type Output = Result<i32, IoError>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();

        // 仅当没有待写入数据时，生成新数据并执行一次逻辑
        if this.pending_buf.is_empty() {
            let val = this.get_val();
            this.pending_buf.push(val as u8);
            this.my_logic();
        }

        let file_pin = Pin::new(&mut this.file);
        match file_pin.poll_write(cx, &this.pending_buf) {
            Poll::Ready(Ok(n)) => {
                // 移除已经成功写入的字节
                this.pending_buf.drain(0..n);
                // 全部写入完成则返回结果
                if this.pending_buf.is_empty() {
                    Poll::Ready(Ok(0))
                } else {
                    // 还有剩余数据，继续等待IO就绪
                    Poll::Pending
                }
            }
            Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
            Poll::Pending => Poll::Pending,
        }
    }
}

内容的提问来源于stack exchange，提问作者Kiwi breeder