Rust Tokio开发ISO-TP库：如何正确处理WouldBlock后的轮询？

问题描述 #

我正在开发适用于Linux的Rust Tokio版ISO-TP CAN协议库，该协议支持发送更大的消息。开发中使用Tokio的AsyncFd结构，调用write时会创建Future并进行轮询。问题出在连续执行两次写入操作时：

socket_tx1.write_packet(packet.clone())?.await?;
socket_tx1.write_packet(packet.clone())?.await?;

第一次写入成功，但第二次会返回std::io::ErrorKind::WouldBlock（缓冲区已满，需等待可用空间），但我不知道如何正确处理该行为。

我尝试了两种实现：
第一种实现可以工作，但在遇到WouldBlock时会进入忙等待，这是我希望避免的：

impl Future for IsoTpWriteFuture {
    type Output = io::Result<()>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        loop {
            let guard = ready!(self.socket.0.poll_write_ready(cx))?;
            match self.socket.0.get_ref().0.write(&self.packet) {
                Err(err) if err.kind() == io::ErrorKind::WouldBlock => continue,
                Ok(_) => return Poll::Ready(Ok(())),
                Err(err) => return Poll::Ready(Err(err))
            }
        }
    }
}

第二种实现逻辑更合理，但无法正常工作：

impl Future for IsoTpWriteFuture {
    type Output = io::Result<()>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        loop {
            let guard = ready!(self.socket.0.poll_write_ready(cx))?;
            match guard.try_io(|inner|inner.get_ref().0.write(&self.packet)) {
                Err(err) => continue,
                Ok(_) => return Poll::Ready(Ok(())),
            }
        }
    }
}

原因是当try_io遇到WouldBlock时会清除guard的就绪状态，而轮询是边缘触发的，导致程序在poll_write_ready处挂起无法推进。

请问是否可以在write返回WouldBlock后重新轮询状态变化？或者忙等待是不可避免的？

解决方案：避免忙等待的正确实现 #

忙等待完全可以避免，核心是在try_io返回WouldBlock后重新向AsyncFd注册任务唤醒器，确保缓冲区可用时Future能被再次触发轮询。

基础正确实现 #

impl Future for IsoTpWriteFuture {
    type Output = io::Result<()>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        loop {
            let mut guard = ready!(self.socket.0.poll_write_ready(cx))?;
            
            match guard.try_io(|inner| inner.get_ref().0.write(&self.packet)) {
                // 写入成功，返回结果
                Ok(result) => return Poll::Ready(result.map(|_| ())),
                // 遇到WouldBlock，重新注册唤醒器
                Err(_would_block) => {
                    guard.clear_ready();
                    self.socket.0.writable().register(cx.waker());
                }
            }
        }
    }
}

关键逻辑说明

1. try_io的错误处理：当try_io返回WouldBlock时，显式调用guard.clear_ready()确保就绪状态被清除，避免重复触发无效轮询。
2. 重新注册唤醒：通过self.socket.0.writable().register(cx.waker())将当前任务的唤醒器绑定到AsyncFd的可写事件上，内核会在缓冲区可用时主动唤醒任务，彻底避免忙等待。

进阶优化：处理部分写入场景 #

如果ISO-TP的write存在部分写入（返回字节数小于数据包长度），需要跟踪已写入字节数，下次轮询时从剩余位置继续写入：

// 调整Future结构体，添加已写入字节的跟踪字段
struct IsoTpWriteFuture {
    socket: ...,
    packet: Vec<u8>,
    written: usize,
}

impl Future for IsoTpWriteFuture {
    type Output = io::Result<()>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();
        let remaining = &this.packet[this.written..];
        
        loop {
            let mut guard = ready!(this.socket.0.poll_write_ready(cx))?;
            
            match guard.try_io(|inner| inner.get_ref().0.write(remaining)) {
                Ok(Ok(n)) => {
                    this.written += n;
                    // 全部写入完成，返回成功
                    if this.written == this.packet.len() {
                        return Poll::Ready(Ok(()));
                    }
                    // 还有剩余数据，继续轮询
                }
                Ok(Err(err)) => return Poll::Ready(Err(err)),
                Err(_would_block) => {
                    guard.clear_ready();
                    this.socket.0.writable().register(cx.waker());
                }
            }
        }
    }
}

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