fd_pipe/
fd_pipe.rs

1//! The lowest-level I/O a developer reaches for: a raw **anonymous pipe** — a pair
2//! of file descriptors with no socket, no address and no filesystem path.
3//!
4//! No network here. `pipe::anonymous()` returns a reader fd and a writer fd on one
5//! ring; a producer task fills buffers borrowed from the crate's [`Pool`] and
6//! writes them into the writer, a consumer task drains the reader and tallies the
7//! bytes, and closing the writer is the EOF the reader stops on. It shows two
8//! things at once: how plain fd streaming looks on `compio`, and that a [`Pool`]
9//! is useful off the network path — any bounded, reused, ring-bound resource fits,
10//! and the `!Send` buffers never leave this thread.
11//!
12//! ```text
13//! cargo run --release --example fd_pipe -- [MESSAGES] [SIZE]
14//!   MESSAGES   how many records to push   default 100000
15//!   SIZE       bytes per record           default 256
16//! ```
17
18use std::{io, time::Instant};
19
20use compio::{
21    BufResult,
22    io::{AsyncRead, AsyncWriteExt},
23    runtime::Runtime,
24};
25use compio_pool::{ManageConnection, Pool};
26
27/// Hands out reusable record buffers sized to one message.
28struct Buffers {
29    size: usize,
30}
31
32impl ManageConnection for Buffers {
33    type Connection = Vec<u8>;
34    type Error = io::Error;
35
36    async fn connect(&self) -> io::Result<Vec<u8>> {
37        Ok(Vec::with_capacity(self.size))
38    }
39
40    async fn is_valid(&self, _buf: &mut Vec<u8>) -> io::Result<()> {
41        Ok(())
42    }
43
44    fn has_broken(&self, _buf: &mut Vec<u8>) -> bool {
45        false
46    }
47}
48
49fn main() -> io::Result<()> {
50    let messages: usize = std::env::args()
51        .nth(1)
52        .and_then(|s| s.parse().ok())
53        .unwrap_or(100_000);
54    let size: usize = std::env::args()
55        .nth(2)
56        .and_then(|s| s.parse().ok())
57        .unwrap_or(256);
58
59    let pool = Pool::builder().max_size(64).build(Buffers { size });
60
61    // A single runtime: a pipe has no fan-out, and both ends must share the ring.
62    let runtime = Runtime::builder().build().expect("build compio runtime");
63    runtime.block_on(async move {
64        let local = pool.local();
65        let (mut reader, mut writer) = compio::fs::pipe::anonymous().await?;
66
67        // Consumer: drain the reader until the writer is closed (read returns 0).
68        let consumer = compio::runtime::spawn(async move {
69            let mut total = 0u64;
70            let mut buf = Vec::with_capacity(64 * 1024);
71            loop {
72                buf.clear();
73                let BufResult(read, b) = reader.read(buf).await;
74                buf = b;
75                match read {
76                    Ok(0) => break,
77                    Ok(n) => total += n as u64,
78                    Err(_) => break,
79                }
80            }
81            total
82        });
83
84        // Producer: borrow a buffer per record, fill it, write it, hand it back.
85        let started = Instant::now();
86        for i in 0..messages {
87            let mut lease = local.get().await.expect("lease a buffer");
88            let mut buf = std::mem::take(&mut *lease);
89            buf.clear();
90            buf.resize(size, i as u8);
91            let BufResult(w, b) = writer.write_all(buf).await;
92            *lease = b; // return the buffer to the pool for the next record
93            w?;
94        }
95        // Closing the writer is the reader's EOF; drop it before awaiting the tally.
96        drop(writer);
97
98        let total = consumer.await.expect("consumer task panicked");
99        let elapsed = started.elapsed();
100        println!(
101            "piped {messages} records x {size} B = {total} bytes in {:.3}s ({:.0} records/s)",
102            elapsed.as_secs_f64(),
103            messages as f64 / elapsed.as_secs_f64(),
104        );
105        Ok(())
106    })
107}
fd_pipe/fd_pipe.rs

fd_pipe/
fd_pipe.rs
