#rng #pipe #reader #tail #head #writer #byte #sync #n #asyncwrite

no-std piper

Async pipes, channels, mutexes, and more

6 releases

new 0.2.1 Sep 25, 2023
0.2.0 Apr 24, 2023
0.1.3 May 31, 2020
0.1.1 Apr 26, 2020
0.1.0 Feb 9, 2020

#728 in Algorithms

Download history 455/week @ 2023-06-04 248/week @ 2023-06-11 723/week @ 2023-06-18 528/week @ 2023-06-25 565/week @ 2023-07-02 474/week @ 2023-07-09 503/week @ 2023-07-16 588/week @ 2023-07-23 373/week @ 2023-07-30 467/week @ 2023-08-06 545/week @ 2023-08-13 337/week @ 2023-08-20 481/week @ 2023-08-27 473/week @ 2023-09-03 541/week @ 2023-09-10 444/week @ 2023-09-17

1,975 downloads per month
Used in 2,703 crates (4 directly)


435 lines


A single-consumer single-producer pipe for Rust async programs.


Licensed under either of

at your option.


Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in the work by you, as defined in the Apache-2.0 license, shall be dual licensed as above, without any additional terms or conditions.


A bounded single-producer single-consumer pipe.

This crate provides a ring buffer that can be asynchronously read from and written to. It is created via the pipe function, which returns a pair of Reader and Writer handles. They implement the AsyncRead and AsyncWrite traits, respectively.

The handles are single-producer/single-consumer; to clarify, they cannot be cloned and need &mut access to read or write to them. If multiple-producer/multiple-consumer handles are needed, consider wrapping them in an Arc<Mutex<...>> or similar.

When the sender is dropped, remaining bytes in the pipe can still be read. After that, attempts to read will result in Ok(0), i.e. they will always 'successfully' read 0 bytes.

When the receiver is dropped, the pipe is closed and no more bytes and be written into it. Further writes will result in Ok(0), i.e. they will always 'successfully' write 0 bytes.

Version 0.2.0 Notes

Previously, this crate contained other synchronization primitives, such as bounded channels, locks, and event listeners. These have been split out into their own crates:


Asynchronous Tasks

Communicate between asynchronous tasks, potentially on other threads.

use async_channel::unbounded;
use async_executor::Executor;
use easy_parallel::Parallel;
use futures_lite::{future, prelude::*};
use std::time::Duration;

// Create a pair of handles.
let (mut reader, mut writer) = piper::pipe(1024);

// Create the executor.
let ex = Executor::new();
let (signal, shutdown) = unbounded::<()>();

// Spawn a detached task for random data to the pipe.
let writer = ex.spawn(async move {
    for _ in 0..1_000 {
        // Generate 8 random numnbers.
        let random = fastrand::u64(..).to_le_bytes();

        // Write them to the pipe.

        // Wait a bit.

    // Drop the writer to close the pipe.

// Detach the task so that it runs in the background.

// Spawn a task for reading from the pipe.
let reader = ex.spawn(async move {
    let mut buf = vec![];

    // Read all bytes from the pipe.
    reader.read_to_end(&mut buf).await.unwrap();

    println!("Random data: {:#?}", buf);

    // Run four executor threads.
    .each(0..4, |_| future::block_on(ex.run(shutdown.recv())))
    // Run the main future on the current thread.
    .finish(|| future::block_on(async {
        // Wait for the reader to finish.

        // Signal the executor threads to shut down.

Blocking I/O

File I/O is blocking; therefore, in async code, you must run it on another thread. This example spawns another thread for reading a file and writing it to a pipe.

use futures_lite::{future, prelude::*};
use std::fs::File;
use std::io::prelude::*;
use std::thread;

// Create a pair of handles.
let (mut r, mut w) = piper::pipe(1024);

// Spawn a thread for reading a file.
thread::spawn(move || {
    let mut file = File::open("Cargo.toml").unwrap();

    // Read the file into a buffer.
    let mut buf = [0u8; 16384];
    future::block_on(async move {
        loop {
            // Read a chunk of bytes from the file.
            // Blocking is okay here, since this is a separate thread.
            let n = file.read(&mut buf).unwrap();
            if n == 0 {

            // Write the chunk to the pipe.

        // Close the pipe.

// Read bytes from the pipe.
let mut buf = vec![];
r.read_to_end(&mut buf).await.unwrap();

println!("Read {} bytes", buf.len());

However, the lower-level poll_fill and poll_drain methods take impl Read and impl Write arguments, respectively. This allows you to skip the buffer entirely and read/write directly from the file into the pipe. This approach should be preferred when possible, as it avoids an extra copy.

// In the `future::block_on` call above...
loop {
    let n = future::poll_fn(|cx| w.poll_fill(cx, &mut file)).await.unwrap();
    if n == 0 {

The blocking crate is preferred in this use case, since it uses more efficient strategies for thread management and pipes.