Rust中如何检测写入stdout的数据是否已被其他进程读取？

我正在用Rust编写一个持续向stdout写入数据的工具，program#2会以可变频率从该stdout读取数据。

为避免program#2挂起，我原本按最大频率写入数据，当前实现示例代码如下：

use std::io::Write;
use std::{io, time, thread};

fn main() -> io::Result<()> {

    let mut data = [0u8; 256];
    let mut counter = 0u32;
    // program#2 will read from its `stdin` every 20ms ~ 30ms
    let min_period = time::Duration::from_millis(20);

    loop {
        let start_time = time::Instant::now();

        // complex data processing
        data[..4].copy_from_slice(&(counter.to_le_bytes()));
        // write to stdout
        io::stdout().write_all(&data)?;

        let elapsed = start_time.elapsed();
        if elapsed < min_period {
            thread::sleep(min_period - elapsed);
        }
        counter += 1;
    }

    Ok(())
}

但我不希望stdout缓冲区中留存过多数据，因为若数据处理逻辑变更会引入延迟。因此我希望在向stdout写入新数据前，检查之前的数据是否已被program#2读取，示例代码如下：

use std::io::Write;
use std::{io, time, thread};

fn main() -> io::Result<()> {

    let mut data = [0u8; 256];
    let mut counter = 0u32;
    let one_ms = time::Duration::from_millis(1);

    loop {
        if (/* data in stdout has been read yet?*/) {
            // complex data processing
            data[..4].copy_from_slice(&(counter.to_le_bytes()));
            // write to stdout
            io::stdout().write_all(&data)?;
            counter += 1;
            
        } else {
            thread::sleep(one_ms);
        }
    }

    Ok(())
}

请问在Rust中是否可以实现？或者有什么更好的方法？

解决方案 #

在Rust中完全可以实现这个需求，以下是几种实用方案：

方案1：利用管道的天然阻塞特性（最简单） #

当两个程序通过管道连接（如./program1 | ./program2）时，操作系统管道的缓冲区满时，write操作会自动阻塞，直到program#2读取数据腾出空间。你可以将stdout设置为无缓冲，确保写入的数据直接进入管道，而非停留在用户态缓冲区：

use std::io::{self, Write};
use std::time;
use std::thread;

fn main() -> io::Result<()> {
    let mut stdout = io::stdout();
    // 关闭stdout的用户态缓冲区
    stdout.set_buf_writer(None);

    let mut data = [0u8; 256];
    let mut counter = 0u32;

    loop {
        // 复杂数据处理
        data[..4].copy_from_slice(&counter.to_le_bytes());
        // 写入会自动阻塞，直到管道有空间（即program#2读取了之前的数据）
        stdout.write_all(&data)?;
        // 强制刷新，确保数据进入管道（无缓冲下可省略，但更稳妥）
        stdout.flush()?;
        counter += 1;
    }

    Ok(())
}

这种方案无需额外依赖，完全利用操作系统特性，既避免了缓冲区堆积，也省去了手动轮询的开销，适合绝大多数场景。

方案2：手动轮询管道可写状态 #

如果需要更精细的控制（比如不想阻塞主线程），可以使用nix crate来检查管道的可写状态。首先在Cargo.toml添加依赖：

[dependencies]
nix = "0.27"

然后实现轮询逻辑：

use std::io::{self, Write};
use std::time;
use std::thread;
use nix::poll::{PollFd, PollFlags};
use nix::unistd::Stdio;

fn main() -> io::Result<()> {
    let mut stdout = io::stdout();
    stdout.set_buf_writer(None);
    let stdout_fd = Stdio::stdout().as_raw_fd();

    let mut data = [0u8; 256];
    let mut counter = 0u32;
    let one_ms = time::Duration::from_millis(1);

    loop {
        // 检查管道是否有可写空间
        let mut fds = [PollFd::new(stdout_fd, PollFlags::POLLOUT)];
        match nix::poll::poll(&mut fds, 0) {
            Ok(_) => {
                if fds[0].revents().contains(PollFlags::POLLOUT) {
                    // 数据处理与写入
                    data[..4].copy_from_slice(&counter.to_le_bytes());
                    stdout.write_all(&data)?;
                    stdout.flush()?;
                    counter += 1;
                } else {
                    // 管道暂时不可写，短暂休眠后重试
                    thread::sleep(one_ms);
                }
            }
            Err(_) => {
                thread::sleep(one_ms);
            }
        }
    }

    Ok(())
}

该方案适合需要自定义轮询间隔、避免阻塞的场景，但仅适用于类Unix系统。

方案3：异步IO实现（适合异步架构） #

如果你的程序采用异步架构，可以用tokio等异步框架，通过异步写入自动等待管道可用：

首先添加依赖：

[dependencies]
tokio = { version = "1.0", features = ["full"] }

实现代码：

use tokio::io::{self, AsyncWriteExt};

#[tokio::main]
async fn main() -> io::Result<()> {
    let mut stdout = io::stdout();
    let mut data = [0u8; 256];
    let mut counter = 0u32;

    loop {
        data[..4].copy_from_slice(&counter.to_le_bytes());
        // 异步写入，自动等待管道可用
        stdout.write_all(&data).await?;
        stdout.flush().await?;
        counter += 1;
    }

    Ok(())
}

异步IO会自动处理等待逻辑，无需手动轮询或阻塞，适合大型异步程序。

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