Flange, a penguin in a hard hat, feeds the bytes of the word hello into a pipe. They come out of the other end in the same order, where a gopher is waiting to catch them.

Pipes

A pipe is a one-way stream of bytes between processes, buffered in a fixed amount of kernel memory. The shell creates one for each | on a command line. In seq 1 100000 | grep 7 | wc -l, the first pipe carries the output of seq to the input of grep, and the second carries the output of grep to the input of wc.

In this chapter, we will specify what a pipe has to do, measure how the Linux pipe behaves, and then implement the same design ourselves. We will write pipefacts, which measures a kernel pipe, handpipe, which implements a pipe in user space with the kernel’s data structure, and pipeline, which connects real commands with kernel pipes the way a shell does. Each program is written in Go and in Rust.

This chapter covers

the requirements of a byte stream between processes, and the mechanism that meets each
the capacity of a Linux pipe and the unit in which a read frees space
the pipe buffer: a ring of pages indexed by the counters head and tail
blocking: how a full buffer suspends the writer and an empty buffer suspends the reader
end of stream and SIGPIPE: how each side learns that the other has closed
the descriptor rule a program must follow when it builds a pipeline
the kernel’s struct pipe_inode_info and struct pipe_buffer
resizing a pipe, named pipes, and how to inspect pipes on a running system

Why a pipe

Consider the job of counting how many of the numbers from 1 to 100000 contain the digit 7. seq 1 100000 prints the numbers, 588895 bytes of text, and grep 7 | wc -l counts the matching lines. The output of the first program has to reach the second somehow.

The obvious way is a temporary file: run seq and store its output, then run grep on the file. This works, and it has 3 costs. The whole output is stored, although each number is needed only once. The second program cannot start until the first has finished. And if the output were endless, as it is for a program that follows a log, the method would not work at all.

A pipe removes all 3. It is a small buffer in memory with a writer on one side and a reader on the other. The writer adds bytes, the reader removes them, and both programs run at the same time. The buffer holds 65536 bytes at most, whether 600 kilobytes pass through it or 600 gigabytes.

2 ways to pass the output of seq to wc. Without a pipe, seq writes all 588895 bytes to a temporary file and wc reads it afterwards: seq must finish before wc starts, and the disk holds everything. With a pipe, seq writes and wc reads at the same time through a buffer of at most 65536 bytes, and memory stays small and fixed. — **Figure 1.1** A temporary file stores everything and forces an order. A pipe passes the bytes along while both programs run.

A small buffer between 2 running programs raises questions that a file does not. What happens when the buffer is full? What happens when it is empty? How does the reader learn that the writer has finished? The answers are the behaviour of a pipe, and the next section lists them as requirements.

Requirements

Linux runs each program as a process, which has its own memory. A process cannot read or write the memory of another. The kernel is the part of the operating system that manages memory and devices for all processes, and a process asks it for a service with a system call: a function call that enters the kernel and returns a result. read and write are system calls. Each takes a file descriptor, a small integer that names an open file or another source of bytes.

Since processes share no memory, a stream of bytes between them has to pass through the kernel. The table lists what such a stream must provide and the mechanism a pipe uses to provide it.

Requirement	Mechanism in a pipe
Bytes arrive in the order they were sent	a first-in, first-out buffer
Memory use is independent of the length of the stream	the buffer has a fixed capacity, 16 pages by default
A fast writer cannot overrun a slow reader	a write to a full buffer blocks until a read frees space
The reader can detect the end of the stream	a read from an empty buffer returns 0 once every write end is closed
The writer can detect that the reader is gone	a write with no read end open raises `SIGPIPE` and fails with `EPIPE`

A call blocks when the kernel suspends the calling thread until the call can proceed. A suspended thread consumes no processor time.

The interface

The pipe2 system call creates a pipe and returns a pair of file descriptors: the read end, then the write end. Bytes passed to write on the write end are appended to the pipe’s buffer, and read on the read end removes them from the front.

A writer process calls write on the write end of a pipe. Inside the kernel the pipe is a ring of 16 pages of 4096 bytes; 6 of them hold bytes, from the oldest to the newest. A reader process calls read on the read end, and bytes leave in the order they entered. — **Figure 1.2** A pipe: a write end, a read end, and a buffer in kernel memory between them.

A pipe has no name in the file system and stores nothing on disk. It exists for as long as some process holds a descriptor for either end. Linux lists the descriptors of a process under /proc/<pid>/fd, where <pid> is the process number, so we can observe a pipe from outside. For sleep 47 | sleep 48, with the processes numbered 4121903 and 4121904, descriptor 1 (standard output) of the first and descriptor 0 (standard input) of the second refer to the same pipe object:

$ ls -l /proc/4121903/fd/1 /proc/4121904/fd/0
/proc/4121903/fd/1 -> pipe:[33835818]
/proc/4121904/fd/0 -> pipe:[33835818]

The listing is read as follows.

The 2 lines of the listing explained. The first line is process 4121903, descriptor 1, standard output, which is the write end of pipe number 33835818. The second line is process 4121904, descriptor 0, standard input, which is the read end of the same pipe, because the pipe numbers match. — **Figure 1.3** How to read the 2 lines: the number after `pipe:` identifies the pipe, and equal numbers mean the same pipe.

The same can be seen from inside a pipeline. Here ls lists its own descriptors while its output goes to cat. The lines for descriptors 1 and 3 are shown: descriptor 1 of ls is a pipe, and descriptor 3 is the directory it is reading.

$ ls -l /proc/self/fd | cat
1 -> pipe:[35453230]
3 -> /proc/559492/fd

A pipe carries bytes in 1 direction. A program that needs to send and receive creates 2 pipes, or uses a socket (chapter 10). The descriptors, the table they live in and what a child process inherits are the subject of chapter 2. For this chapter it is enough that a descriptor is a number that names 1 end of the pipe.

One write, step by step

Before looking at the data structures, we follow a single call. A process holds the write end of a pipe as descriptor 4 and calls write(4, "hello", 5).

The kernel takes entry 4 of the process’s descriptor table and finds a struct file, its record of an open file. The record contains a table of functions, f_op, and for a pipe the write function in that table is pipe_write.
pipe_write locks the pipe. A mutex is a lock that 1 thread can hold at a time, and a thread that asks for a held mutex sleeps until it is released. All further steps happen with the lock held, so a reader running at the same moment sees the pipe either before the write or after it.
It checks that a read end is open. If none is, it sends the process the signal SIGPIPE and returns the error EPIPE.
It looks for room. The bytes can go into the unused end of the page written last, or into a new page if a slot of the buffer is free. If there is no room, the process is put to sleep until a reader makes some, and the step is repeated.
It copies the 5 bytes from the memory of the process into the page.
It wakes any reader that is asleep waiting for data, releases the lock, and returns 5.

The path of write(4, hello, 5) on a pipe. Descriptor 4 is the write end. Its struct file has the pipe's function table in f_op, which leads to pipe_write. pipe_write locks the pipe's mutex, checks that a reader is left and otherwise sends SIGPIPE and fails with EPIPE, finds room in the newest page or a free slot of the ring, copies the 5 bytes from the process into the page, and wakes the readers asleep on rd_wait. — **Figure 1.4** A `write` to a pipe, from the system call to the wake-up of the reader.

Each step corresponds to a field of the structure that represents a pipe in the kernel.

The kernel’s structures

A pipe is a struct pipe_inode_info, and each slot of its buffer is a struct pipe_buffer. These are the definitions in include/linux/pipe_fs_i.h of the kernel that runs on our machine, Linux 5.15, with the fields this chapter does not use left out.

struct pipe_buffer {
        struct page *page;             /* the page that holds the bytes */
        unsigned int offset, len;      /* where the unread bytes start, and how many */
        unsigned int flags;            /* PIPE_BUF_FLAG_CAN_MERGE: may be appended to */
        /* ... */
};

struct pipe_inode_info {
        struct mutex mutex;                    /* the lock of step 2 */
        wait_queue_head_t rd_wait, wr_wait;    /* sleeping readers, sleeping writers */
        unsigned int head;                     /* slots ever filled */
        unsigned int tail;                     /* slots ever emptied */
        unsigned int max_usage;                /* slots that may be in use */
        unsigned int ring_size;                /* slots in bufs, a power of 2 */
        unsigned int readers;                  /* open read ends */
        unsigned int writers;                  /* open write ends */
        struct pipe_buffer *bufs;              /* the array of slots */
        /* ... */
};

#define PIPE_DEF_BUFFERS        16

Field	Meaning
`bufs`, `ring_size`	The buffer: an array of `ring_size` slots. A new pipe has `PIPE_DEF_BUFFERS`, 16.
`page`	A page is the unit in which the kernel allocates memory, 4096 bytes on this machine. Each slot in use refers to 1 page.
`offset`, `len`	The unread bytes of a page are `len` bytes starting at `offset`. A read advances `offset` and reduces `len`.
`head`, `tail`	Counters that only increase. `head - tail` slots are in use.
`mutex`	Held during every read and write.
`rd_wait`, `wr_wait`	Wait queues: lists of sleeping threads. A reader that finds the pipe empty adds itself to `rd_wait` and sleeps, and a writer wakes the threads on it. `wr_wait` is the same for writers and a full pipe.
`readers`, `writers`	How many read ends and write ends are open. Step 3 tests `readers`.

The kernel structures of a pipe. The read end and the write end are 2 struct file objects that refer to the same inode. The inode refers to a struct pipe_inode_info, which has the mutex, the wait queues rd_wait and wr_wait, the counters head and tail, ring_size, the counts readers and writers, and the array bufs. bufs holds 16 struct pipe_buffer, each with a page, an offset, a length and flags. — **Figure 1.5** The structures of a pipe and the pointers between them.

The capacity follows from these numbers: 16 slots of 4096 bytes are 65536 bytes. The first measurement below checks it.

Setting up the project

The Go programs share a module, and the Rust programs share a Cargo workspace. The Rust crates use the 2024 edition and Rust 1.99, the current stable release. A rust-toolchain.toml file in the rust directory selects it.

pipefacts calls the kernel through golang.org/x/sys/unix in Go and through the nix crate in Rust. handpipe uses the parking_lot crate in Rust for its lock and condition variables. pipeline needs only the standard library in both languages.

To create the Go module with its dependency, we run:

$ mkdir go && cd go
$ go mod init democratizing-linux
$ go get golang.org/x/sys/unix

This produces the following go.mod. Each program is a directory with a main.go, such as pipefacts/main.go.

module democratizing-linux

go 1.26.0

require golang.org/x/sys v0.49.0

For Rust, we create a directory rust with this workspace manifest in Cargo.toml:

[workspace]
members = []
resolver = "3"

and this rust-toolchain.toml beside it:

[toolchain]
channel = "1.99.0"
components = ["rustfmt", "clippy"]

Then we create the crates and add their dependencies. cargo new adds each crate to members.

$ cargo new pipefacts
$ cargo add -p pipefacts nix --features fs
$ cargo new handpipe
$ cargo add -p handpipe parking_lot
$ cargo new pipeline

The manifest of pipefacts is now:

[package]
name = "pipefacts"
version = "0.1.0"
edition = "2024"
rust-version = "1.99"
publish = false

[dependencies]
nix = { version = "0.31.3", features = ["fs"] }

and the manifest of handpipe:

[package]
name = "handpipe"
version = "0.1.0"
edition = "2024"
rust-version = "1.99"
publish = false

[dependencies]
parking_lot = "0.12.5"

We run a program with go run ./pipefacts from the go directory and with cargo run -p pipefacts from the rust directory.

Measuring a kernel pipe

pipefacts measures the behaviours in the requirements table on a kernel pipe. It creates each pipe with the O_NONBLOCK flag. In this mode, a call that would block returns the error EAGAIN immediately, which lets a thread detect a full or an empty buffer without being suspended. We will come back to O_CLOEXEC when we start other programs.

Listing 1.1 The imports of pipefacts, in Go. code/go/pipefacts/main.go

import (
	"errors"
	"fmt"
	"log"

	"golang.org/x/sys/unix"
)

Listing 1.2 The imports of pipefacts, in Rust. code/rust/pipefacts/src/main.rs

use std::os::fd::OwnedFd;

use nix::errno::Errno;
use nix::fcntl::{FcntlArg, OFlag, fcntl};
use nix::unistd::{pipe2, read, write};

Listing 1.3 Creating a non-blocking pipe, in Go. code/go/pipefacts/main.go

// block is the size of every write the program attempts: one page.
const block = 4096

// newPipe creates a pipe in non-blocking mode: a read or a write that cannot
// proceed returns EAGAIN where it would otherwise suspend the caller.
func newPipe() (reader, writer int, err error) {
	var fds [2]int

	if err := unix.Pipe2(fds[:], unix.O_NONBLOCK|unix.O_CLOEXEC); err != nil {
		return 0, 0, fmt.Errorf("pipe2: %w", err)
	}

	return fds[0], fds[1], nil
}

Listing 1.4 Creating a non-blocking pipe, in Rust. code/rust/pipefacts/src/main.rs

/// The size of every write the program attempts: one page.
const BLOCK: usize = 4096;

/// Creates a pipe in non-blocking mode: a read or a write that cannot proceed
/// returns `EAGAIN` where it would otherwise suspend the caller. The read end
/// is first.
fn new_pipe() -> nix::Result<(OwnedFd, OwnedFd)> {
    pipe2(OFlag::O_NONBLOCK | OFlag::O_CLOEXEC)
}

fill measures free space. It writes 4096-byte blocks until a write returns EAGAIN and returns the total number of bytes accepted. drain removes bytes.

Listing 1.5 Measuring free space by writing until the kernel refuses, in Go. code/go/pipefacts/main.go

// fill writes whole blocks until the kernel refuses one and returns the
// number of bytes accepted.
func fill(writer int) (int, error) {
	data := make([]byte, block)
	total := 0

	for {
		n, err := unix.Write(writer, data)

		if errors.Is(err, unix.EAGAIN) {
			return total, nil
		}

		if err != nil {
			return total, fmt.Errorf("write: %w", err)
		}

		total += n
	}
}

// drain reads and discards up to n bytes.
func drain(reader, n int) error {
	_, err := unix.Read(reader, make([]byte, n))

	return err
}

Listing 1.6 Measuring free space by writing until the kernel refuses, in Rust. code/rust/pipefacts/src/main.rs

/// Writes whole blocks until the kernel refuses one and returns the number of
/// bytes accepted.
fn fill(writer: &OwnedFd) -> nix::Result<usize> {
    let data = [0u8; BLOCK];
    let mut total = 0;

    loop {
        match write(writer, &data) {
            Ok(n) => total += n,
            Err(Errno::EAGAIN) => return Ok(total),
            Err(error) => return Err(error),
        }
    }
}

/// Reads and discards up to `n` bytes.
fn drain(reader: &OwnedFd, n: usize) -> nix::Result<usize> {
    read(reader, &mut vec![0u8; n])
}

capacity calls fill on a new pipe, again after 1 byte has been read from the full pipe, and again after 4096 bytes have been read in total.

Listing 1.7 Capacity, and the space a read frees, in Go. code/go/pipefacts/main.go

// capacity fills a pipe when it is new, after 1 byte has been read, and
// after 4096 bytes have been read. Each fill reports the free space.
func capacity() error {
	reader, writer, err := newPipe()

	if err != nil {
		return err
	}

	defer unix.Close(reader)
	defer unix.Close(writer)

	steps := []struct {
		read   int
		report string
	}{
		{0, "new pipe: %d bytes accepted\n"},
		{1, "after reading 1 byte: %d bytes accepted\n"},
		{block - 1, "after reading 4096 bytes in total: %d bytes accepted\n"},
	}

	for _, step := range steps {
		if err := drain(reader, step.read); err != nil {
			return err
		}

		accepted, err := fill(writer)

		if err != nil {
			return err
		}

		fmt.Printf(step.report, accepted)
	}

	return nil
}

Listing 1.8 Capacity, and the space a read frees, in Rust. code/rust/pipefacts/src/main.rs

/// Fills a pipe when it is new, after 1 byte has been read, and after 4096
/// bytes have been read. Each fill reports the free space.
fn capacity() -> nix::Result<()> {
    let (reader, writer) = new_pipe()?;

    let steps = [
        (0, "new pipe"),
        (1, "after reading 1 byte"),
        (BLOCK - 1, "after reading 4096 bytes in total"),
    ];

    for (count, label) in steps {
        drain(&reader, count)?;

        let accepted = fill(&writer)?;
        println!("{label}: {accepted} bytes accepted");
    }

    Ok(())
}

resize changes the number of slots. fcntl with the command F_SETPIPE_SZ asks for a capacity in bytes and returns the capacity granted. The program asks for 16384 bytes, for 5000 bytes, and for 1048576 bytes, and measures the pipe after each request.

Listing 1.9 Changing the capacity of a pipe, in Go. code/go/pipefacts/main.go

// resize changes the capacity of a pipe with fcntl and measures it after
// each change.
func resize() error {
	reader, writer, err := newPipe()

	if err != nil {
		return err
	}

	defer unix.Close(reader)
	defer unix.Close(writer)

	for _, request := range []int{4 * block, 5000, 1 << 20} {
		granted, err := unix.FcntlInt(uintptr(writer), unix.F_SETPIPE_SZ, request)

		if err != nil {
			return fmt.Errorf("F_SETPIPE_SZ: %w", err)
		}

		accepted, err := fill(writer)

		if err != nil {
			return err
		}

		fmt.Printf("capacity of %d bytes requested: %d granted, %d bytes accepted\n", request, granted, accepted)

		if err := drain(reader, accepted); err != nil {
			return err
		}
	}

	return nil
}

Listing 1.10 Changing the capacity of a pipe, in Rust. code/rust/pipefacts/src/main.rs

/// Changes the capacity of a pipe with fcntl and measures it after each
/// change.
fn resize() -> nix::Result<()> {
    let (reader, writer) = new_pipe()?;

    for request in [4 * BLOCK, 5000, 1 << 20] {
        let granted = fcntl(&writer, FcntlArg::F_SETPIPE_SZ(request as i32))?;
        let accepted = fill(&writer)?;

        println!(
            "capacity of {request} bytes requested: {granted} granted, {accepted} bytes accepted"
        );

        drain(&reader, accepted)?;
    }

    Ok(())
}

endOfStream reads an empty pipe with the write end open, and again with it closed.

Listing 1.11 Reading an empty pipe, in Go. code/go/pipefacts/main.go

// endOfStream reads an empty pipe while a write end is open, and again after
// the write end has been closed.
func endOfStream() error {
	reader, writer, err := newPipe()

	if err != nil {
		return err
	}

	defer unix.Close(reader)

	buf := make([]byte, block)

	_, err = unix.Read(reader, buf)
	fmt.Printf("empty pipe, write end open: read fails with %s\n", name(err))

	unix.Close(writer)

	n, err := unix.Read(reader, buf)

	if err != nil {
		return fmt.Errorf("read: %w", err)
	}

	fmt.Printf("empty pipe, write end closed: read returns %d\n", n)

	return nil
}

Listing 1.12 Reading an empty pipe, in Rust. code/rust/pipefacts/src/main.rs

/// Reads an empty pipe while a write end is open, and again after the write
/// end has been closed.
fn end_of_stream() -> nix::Result<()> {
    let (reader, writer) = new_pipe()?;
    let mut buf = [0u8; BLOCK];

    if let Err(error) = read(&reader, &mut buf) {
        println!("empty pipe, write end open: read fails with {error:?}");
    }

    drop(writer);

    let n = read(&reader, &mut buf)?;
    println!("empty pipe, write end closed: read returns {n}");

    Ok(())
}

brokenPipe closes the read end and then writes.

Listing 1.13 Writing with no read end open, in Go. code/go/pipefacts/main.go

// brokenPipe writes to a pipe whose read end has been closed.
func brokenPipe() error {
	reader, writer, err := newPipe()

	if err != nil {
		return err
	}

	defer unix.Close(writer)

	unix.Close(reader)

	_, err = unix.Write(writer, []byte("hello"))
	fmt.Printf("read end closed: write fails with %s\n", name(err))

	return nil
}

// name returns the symbolic name of a system call error, such as EAGAIN.
func name(err error) string {
	var errno unix.Errno

	if errors.As(err, &errno) {
		return unix.ErrnoName(errno)
	}

	return fmt.Sprint(err)
}

Listing 1.14 Writing with no read end open, in Rust. code/rust/pipefacts/src/main.rs

/// Writes to a pipe whose read end has been closed.
fn broken_pipe() -> nix::Result<()> {
    let (reader, writer) = new_pipe()?;

    drop(reader);

    if let Err(error) = write(&writer, b"hello") {
        println!("read end closed: write fails with {error:?}");
    }

    Ok(())
}

The Go and Rust programs print the following on a Jetson Orin NX running Linux 5.15:

new pipe: 65536 bytes accepted
after reading 1 byte: 0 bytes accepted
after reading 4096 bytes in total: 4096 bytes accepted
capacity of 16384 bytes requested: 16384 granted, 16384 bytes accepted
capacity of 5000 bytes requested: 8192 granted, 8192 bytes accepted
capacity of 1048576 bytes requested: 1048576 granted, 1048576 bytes accepted
empty pipe, write end open: read fails with EAGAIN
empty pipe, write end closed: read returns 0
read end closed: write fails with EPIPE

Output	Conclusion
a new pipe accepts 65536 bytes	The capacity is 16 pages. A page is the unit in which the kernel allocates memory: 4096 bytes on this machine.
after 1 byte is read, 0 bytes are accepted	Free space is counted in whole pages. The pipe held 65535 bytes at that point and still refused a write.
after 4096 bytes are read, 4096 are accepted	Reading the last byte of a page releases that page.
16384 requested, 16384 granted	The ring was replaced by a ring of 4 slots.
5000 requested, 8192 granted	The kernel rounds the request up to whole pages and to a power of 2 slots, here 2.
1048576 requested and granted	256 slots. This is the largest size an ordinary user may ask for, set in `/proc/sys/fs/pipe-max-size`.
`EAGAIN` with the write end open	The buffer is empty and data may still arrive. A blocking read would be suspended here.
0 with the write end closed	The buffer is empty and no data can arrive. A return value of 0 from `read` is the end of the stream.
`EPIPE` with the read end closed	No process can consume the data, so the write is refused.

In the last case the kernel also sends the writing process SIGPIPE. A signal is an asynchronous notification from the kernel to a process, and the default action for SIGPIPE is to terminate the process. The programs continue and print EPIPE because their runtimes change that default: the Go runtime handles SIGPIPE for descriptors other than 1 and 2, and a Rust program starts with SIGPIPE ignored. The pipeline program shows the default action.

Rows 2 and 3 confirm that the buffer is made of whole pages and rule out a flat byte array, since a flat array with 1 byte removed would accept 1 byte.

The buffer: a ring of pages

The kernel allocates physical memory in pages, and the pipe buffer is built from that unit. A pipe has an array of slots, 16 by default, and each slot refers to a page. Each slot also records which part of its page holds unread data: an offset to the first unread byte and a length.

Since the buffer is built from page references, an empty pipe needs no data pages and an idle pipe costs little memory. A page that already contains the data can also be placed in a slot directly: the splice system call moves page references between a file and a pipe and copies no bytes.

The slots are used as a ring, indexed by the counters head and tail:

head is the number of pages added to the ring since the pipe was created.
tail is the number of pages removed.

head and tail only increase. The ring maintains these invariants:

head - tail is the number of pages in use, and it is between 0 and the number of slots.
Counter value c refers to slot c mod slots. The oldest page is at tail and the newest at head - 1.
New bytes are appended to the newest page if it has unused space after its data. The remainder goes into new pages at head.
Bytes are removed from the oldest page. When its length reaches 0, the page is released and tail advances by 1.

The kernel computes the slot with a mask in place of a division: head & (ring_size - 1). The 2 are equal when ring_size is a power of 2, which is the reason a request for 5000 bytes was rounded up to 2 slots. The counters are unsigned 32-bit numbers and wrap around to 0 after about 4 billion pages. head - tail remains correct across the wrap, because unsigned subtraction wraps in the same way.

Invariants 3 and 4 explain the measurement. Removing bytes from the oldest page frees space at the front of that page, and invariant 3 places no new bytes there. The space becomes usable only when the whole page is released.

The table below traces a ring of 4 slots with 8-byte pages. The last column lists the length of each slot, with a dot for a free slot.

Call	Result	`head`	`tail`	Slots 0 to 3	Explanation
`put` 12	12	2	0	`[8 4 . .]`	2 new pages: 8 bytes and 4 bytes.
`take` 3	3	2	0	`[5 4 . .]`	Slot 0 has `offset` 3 and `length` 5. It still holds data, so `tail` is unchanged.
`put` 30	20	4	0	`[5 8 8 8]`	4 bytes complete slot 1, and slots 2 and 3 take 8 each. All slots are in use. The 3 free bytes in slot 0 are unusable.
`take` 5	5	4	1	`[. 8 8 8]`	Slot 0 is empty and is released.
`put` 30	8	5	1	`[8 8 8 8]`	Counter 4 maps to slot 0. The newest page now precedes the oldest in the array: the ring has wrapped.

A ring of 4 slots, each a page of 8 bytes, after each call of the trace. Put 12 fills slot 0 and half of slot 1. Take 3 leaves 5 bytes in slot 0. Put 30 copies 20 bytes and fills slots 1, 2 and 3. Take 5 empties slot 0 and tail becomes 1. Put 30 copies 8 bytes into slot 0, and head becomes 5. — **Figure 1.6** The trace as a picture. The shaded part of a slot is its unread data.

Low-level design: handpipe

Type	Fields	Invariant
`page`	`data`, `offset`, `length`	unread bytes are `data[offset .. offset+length]`
`ring`	`pages`, `head`, `tail`	`0 <= head - tail <= len(pages)`; counter `c` maps to slot `c mod len(pages)`
`pipe`	`ring`, a lock, `readable`, `writable`, `readers`, `writers`	the ring and the counts are accessed only with the lock held

Operation	Behaviour	Blocks
`put(data)`	appends as many bytes as fit; returns the count	no
`take(buf)`	removes up to `len(buf)` of the oldest bytes; returns the count	no
`write(data)`	appends all of `data`; fails with `EPIPE` when `readers` is 0	while the ring is full
`read(buf)`	removes at least 1 byte; returns 0 when the ring is empty and `writers` is 0	while the ring is empty and `writers` is above 0
`closeWriter`, `closeReader`	decrements a count and wakes the other side	no

Implementing the ring

page and ring follow the description directly. slot applies invariant 2. Both versions allocate every page when the ring is created, which keeps the code short; the kernel allocates a page when a slot first needs one.

Listing 1.15 The imports of handpipe, in Go. code/go/handpipe/main.go

import (
	"errors"
	"fmt"
	"strings"
	"sync"
)

Listing 1.16 The imports of handpipe, in Rust. code/rust/handpipe/src/main.rs

use std::fmt;
use std::thread;

use parking_lot::{Condvar, Mutex};

Listing 1.17 The page, the ring, and its counters, in Go. code/go/handpipe/main.go

// A page is one slot of the ring: a block of memory, and the position and
// length of the bytes in it that have not been read yet.
type page struct {
	data   []byte
	offset int
	length int
}

// A ring is the storage of a pipe. head counts the pages ever put into use
// and tail counts the pages ever emptied, so head-tail pages are in use, and
// a counter's remainder after division by the number of slots is its slot.
type ring struct {
	pages []page
	head  uint
	tail  uint
}

func newRing(slots, pageSize int) *ring {
	pages := make([]page, slots)

	for i := range pages {
		pages[i].data = make([]byte, pageSize)
	}

	return &ring{pages: pages}
}

func (r *ring) empty() bool {
	return r.head == r.tail
}

func (r *ring) full() bool {
	return r.head-r.tail == uint(len(r.pages))
}

// slot returns the page that a counter refers to.
func (r *ring) slot(counter uint) *page {
	return &r.pages[counter%uint(len(r.pages))]
}

Listing 1.18 The page, the ring, and its counters, in Rust. code/rust/handpipe/src/main.rs

/// One slot of the ring: a block of memory, and the position and length of
/// the bytes in it that have not been read yet.
struct Page {
    data: Vec<u8>,
    offset: usize,
    length: usize,
}

/// The storage of a pipe. `head` counts the pages ever put into use and
/// `tail` counts the pages ever emptied, so `head - tail` pages are in use,
/// and a counter's remainder after division by the number of slots is its
/// slot.
struct Ring {
    pages: Vec<Page>,
    head: usize,
    tail: usize,
}

impl Ring {
    fn new(slots: usize, page_size: usize) -> Ring {
        let pages = (0..slots)
            .map(|_| Page {
                data: vec![0; page_size],
                offset: 0,
                length: 0,
            })
            .collect();

        Ring {
            pages,
            head: 0,
            tail: 0,
        }
    }

    fn empty(&self) -> bool {
        self.head == self.tail
    }

    fn full(&self) -> bool {
        self.head - self.tail == self.pages.len()
    }

    /// The page that a counter refers to.
    fn slot(&mut self, counter: usize) -> &mut Page {
        let slots = self.pages.len();

        &mut self.pages[counter % slots]
    }
}

put implements invariant 3: first the unused space of the newest page, then free slots. It returns the number of bytes copied, which is smaller than the request when the ring fills.

Listing 1.19 Appending bytes to the ring, in Go. code/go/handpipe/main.go

// put copies as much of data as the ring has room for and returns the number
// of bytes it copied. The bytes go first into the unused end of the newest
// page, then into free slots, one page each.
func (r *ring) put(data []byte) int {
	copied := 0

	if !r.empty() {
		newest := r.slot(r.head - 1)
		end := newest.offset + newest.length

		n := copy(newest.data[end:], data)
		newest.length += n
		copied += n
	}

	for copied < len(data) && !r.full() {
		next := r.slot(r.head)
		next.offset = 0
		next.length = copy(next.data, data[copied:])

		copied += next.length
		r.head++
	}

	return copied
}

Listing 1.20 Appending bytes to the ring, in Rust. code/rust/handpipe/src/main.rs

/// Copies as many bytes as both slices allow and returns the count.
fn copy(to: &mut [u8], from: &[u8]) -> usize {
    let n = to.len().min(from.len());
    to[..n].copy_from_slice(&from[..n]);

    n
}

impl Ring {
    /// Copies as much of `data` as the ring has room for and returns the
    /// number of bytes it copied. The bytes go first into the unused end of
    /// the newest page, then into free slots, one page each.
    fn put(&mut self, data: &[u8]) -> usize {
        let mut copied = 0;

        if !self.empty() {
            let newest = self.slot(self.head - 1);
            let end = newest.offset + newest.length;

            let n = copy(&mut newest.data[end..], data);
            newest.length += n;
            copied += n;
        }

        while copied < data.len() && !self.full() {
            let next = self.slot(self.head);
            next.offset = 0;
            next.length = copy(&mut next.data, &data[copied..]);

            copied += next.length;
            self.head += 1;
        }

        copied
    }
}

take implements invariant 4. It advances offset in the oldest page and increments tail when the page is exhausted.

Listing 1.21 Removing the oldest bytes from the ring, in Go. code/go/handpipe/main.go

// take moves up to len(buf) of the oldest bytes into buf and returns the
// number it moved. A slot becomes free when the last byte of its page has
// been taken.
func (r *ring) take(buf []byte) int {
	moved := 0

	for moved < len(buf) && !r.empty() {
		oldest := r.slot(r.tail)

		n := copy(buf[moved:], oldest.data[oldest.offset:oldest.offset+oldest.length])
		oldest.offset += n
		oldest.length -= n
		moved += n

		if oldest.length == 0 {
			r.tail++
		}
	}

	return moved
}

Listing 1.22 Removing the oldest bytes from the ring, in Rust. code/rust/handpipe/src/main.rs

impl Ring {
    /// Moves up to `buf.len()` of the oldest bytes into `buf` and returns the
    /// number it moved. A slot becomes free when the last byte of its page
    /// has been taken.
    fn take(&mut self, buf: &mut [u8]) -> usize {
        let mut moved = 0;

        while moved < buf.len() && !self.empty() {
            let oldest = self.slot(self.tail);
            let unread = oldest.offset..oldest.offset + oldest.length;

            let n = copy(&mut buf[moved..], &oldest.data[unread]);
            oldest.offset += n;
            oldest.length -= n;
            moved += n;

            if oldest.length == 0 {
                self.tail += 1;
            }
        }

        moved
    }
}

smallRing applies the calls of the trace and prints the ring after each.

Listing 1.23 Running the trace, in Go. code/go/handpipe/main.go

// picture renders the ring as one cell per slot: the number of unread bytes
// in its page, or a dot when the slot is free.
func (r *ring) picture() string {
	cells := make([]string, len(r.pages))

	for i := range cells {
		cells[i] = "."
	}

	for counter := r.tail; counter != r.head; counter++ {
		cells[counter%uint(len(r.pages))] = fmt.Sprint(r.slot(counter).length)
	}

	return "[" + strings.Join(cells, " ") + "]"
}

// smallRing applies the calls of the trace to a ring of 4 pages of 8 bytes
// and prints the ring after each.
func smallRing() {
	r := newRing(4, 8)
	buf := make([]byte, 32)

	calls := []struct {
		op string
		n  int
	}{{"put", 12}, {"take", 3}, {"put", 30}, {"take", 5}, {"put", 30}}

	fmt.Println("ring of 4 pages of 8 bytes")

	for _, call := range calls {
		var done int

		switch call.op {
		case "put":
			done = r.put(buf[:call.n])
		case "take":
			done = r.take(buf[:call.n])
		}

		fmt.Printf("%-4s %2d: %2d bytes   head %d  tail %d   %s\n", call.op, call.n, done, r.head, r.tail, r.picture())
	}
}

Listing 1.24 Running the trace, in Rust. code/rust/handpipe/src/main.rs

impl Ring {
    /// Renders the ring as one cell per slot: the unread bytes in its page,
    /// or a dot for a free slot.
    fn picture(&self) -> String {
        let slots = self.pages.len();
        let mut cells = vec![".".to_string(); slots];

        for counter in self.tail..self.head {
            cells[counter % slots] = self.pages[counter % slots].length.to_string();
        }

        format!("[{}]", cells.join(" "))
    }
}

/// Applies the calls of the trace to a ring of 4 pages of 8 bytes and prints
/// the ring after each.
fn small_ring() {
    let mut ring = Ring::new(4, 8);
    let mut buf = [0u8; 32];

    let calls = [
        ("put", 12),
        ("take", 3),
        ("put", 30),
        ("take", 5),
        ("put", 30),
    ];

    println!("ring of 4 pages of 8 bytes");

    for (op, n) in calls {
        let done = match op {
            "put" => ring.put(&buf[..n]),
            _ => ring.take(&mut buf[..n]),
        };

        println!(
            "{op:<4} {n:2}: {done:2} bytes   head {}  tail {}   {}",
            ring.head,
            ring.tail,
            ring.picture()
        );
    }
}

ring of 4 pages of 8 bytes
put  12: 12 bytes   head 2  tail 0   [8 4 . .]
take  3:  3 bytes   head 2  tail 0   [5 4 . .]
put  30: 20 bytes   head 4  tail 0   [5 8 8 8]
take  5:  5 bytes   head 4  tail 1   [. 8 8 8]
put  30:  8 bytes   head 5  tail 1   [8 8 8 8]

The output agrees with the table.

Checking the model against the kernel

kernelSize repeats the capacity measurement of pipefacts on a ring with the kernel’s dimensions, 16 pages of 4096 bytes.

Listing 1.25 The capacity measurement on a ring of 16 pages, in Go. code/go/handpipe/main.go

// fill puts whole blocks into the ring until it accepts none and returns the
// number of bytes accepted.
func fill(r *ring, block []byte) int {
	total := 0

	for {
		n := r.put(block)

		if n == 0 {
			return total
		}

		total += n
	}
}

// kernelSize repeats the capacity measurement of pipefacts on a ring with
// the kernel's dimensions: 16 pages of 4096 bytes.
func kernelSize() {
	const pageSize = 4096

	r := newRing(16, pageSize)
	block := make([]byte, pageSize)

	steps := []struct {
		read   int
		report string
	}{
		{0, "new pipe: %d bytes accepted\n"},
		{1, "after reading 1 byte: %d bytes accepted\n"},
		{pageSize - 1, "after reading 4096 bytes in total: %d bytes accepted\n"},
	}

	for _, step := range steps {
		r.take(block[:step.read])
		fmt.Printf(step.report, fill(r, block))
	}
}

Listing 1.26 The capacity measurement on a ring of 16 pages, in Rust. code/rust/handpipe/src/main.rs

/// Puts whole blocks into the ring until it accepts none and returns the
/// number of bytes accepted.
fn fill(ring: &mut Ring, block: &[u8]) -> usize {
    let mut total = 0;

    loop {
        let n = ring.put(block);

        if n == 0 {
            return total;
        }

        total += n;
    }
}

/// Repeats the capacity measurement of pipefacts on a ring with the kernel's
/// dimensions: 16 pages of 4096 bytes.
fn kernel_size() {
    const PAGE_SIZE: usize = 4096;

    let mut ring = Ring::new(16, PAGE_SIZE);
    let mut block = [0u8; PAGE_SIZE];

    let steps = [
        (0, "new pipe"),
        (1, "after reading 1 byte"),
        (PAGE_SIZE - 1, "after reading 4096 bytes in total"),
    ];

    for (count, label) in steps {
        ring.take(&mut block[..count]);

        let accepted = fill(&mut ring, &block);
        println!("{label}: {accepted} bytes accepted");
    }
}

new pipe: 65536 bytes accepted
after reading 1 byte: 0 bytes accepted
after reading 4096 bytes in total: 4096 bytes accepted

These lines are identical to lines 1 to 3 of the pipefacts output. The ring of pages reproduces the kernel’s capacity and its page-sized release of space.

Where the model simplifies. handpipe uses the kernel’s names and its ring. It differs in 3 details. It allocates every page when the ring is created, where the kernel allocates a page when a slot first needs one. It appends to the newest page whenever there is space, where the kernel does so only when the final partial page of a write fits there completely, which keeps the pages of a large write aligned. And its ring has a fixed size, where the kernel’s can be resized with F_SETPIPE_SZ, as pipefacts showed.

Blocking

The ring is a single-threaded data structure. A pipe adds safe access from a reader and a writer running concurrently, and a defined behaviour when a call cannot make progress. Figure 1.7 lists the cases of each call.

What a read does: when the ring holds bytes it returns them; when the ring is empty and a write end is open the caller sleeps; when the ring is empty and no write end is open it returns 0. What a write does: when the ring has room it copies the bytes; when the ring is full and a read end is open the caller sleeps; when no read end is open the kernel sends SIGPIPE and the call fails with EPIPE. — **Figure 1.7** The cases of `read` and of `write`.

4 crabs show what a write to a pipe can meet. A happy crab: the ring has room, and the write returns. A sleeping crab: the ring is full, and the writer sleeps. An angry crab: no reader is left, and the write fails with SIGPIPE and EPIPE. A cute crab: the writer closes its end, and read returns 0.

The blocking cases provide flow control. If seq produces output faster than grep consumes it, the ring fills and seq is suspended inside write. When grep releases a page, seq resumes. The writer’s rate is therefore limited to the reader’s rate, and the memory between them stays at 16 pages for a stream of any length. Neither program contains code for this; it follows from the semantics of write.

In addition to the ring, the implementation needs:

A lock, which gives a thread exclusive access to the ring while it holds the lock.
The condition variables readable and writable. A condition variable is a queue on which a thread sleeps until another thread wakes it. Waiting releases the lock and suspends the thread atomically, and reacquires the lock before returning. Readers wait on readable and writers on writable. The kernel’s equivalents are the wait queues rd_wait and wr_wait.
The counts readers and writers of the open ends.

Listing 1.27 The pipe: a ring, a lock, wait queues, and the open ends, in Go. code/go/handpipe/main.go

// errBrokenPipe is what a write returns when no reader is left.
var errBrokenPipe = errors.New("EPIPE")

// A pipe is a ring shared by a reader and a writer goroutine. readable is where a reader
// sleeps while the ring is empty, and writable is where a writer sleeps while
// the ring is full. readers and writers count the open ends.
type pipe struct {
	mu       sync.Mutex
	readable *sync.Cond
	writable *sync.Cond
	ring     *ring
	readers  int
	writers  int
}

func newPipe(slots, pageSize int) *pipe {
	p := &pipe{ring: newRing(slots, pageSize), readers: 1, writers: 1}
	p.readable = sync.NewCond(&p.mu)
	p.writable = sync.NewCond(&p.mu)

	return p
}

Listing 1.28 The pipe: a ring, a lock, wait queues, and the open ends, in Rust. code/rust/handpipe/src/main.rs

/// What a write returns when no reader is left.
#[derive(Debug)]
struct BrokenPipe;

impl fmt::Display for BrokenPipe {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str("EPIPE")
    }
}

/// What the lock guards: the ring and the count of the open ends.
struct State {
    ring: Ring,
    readers: usize,
    writers: usize,
}

/// A ring shared by a reader and a writer thread. `readable` is where a reader sleeps while
/// the ring is empty, and `writable` is where a writer sleeps while the ring
/// is full.
struct Pipe {
    state: Mutex<State>,
    readable: Condvar,
    writable: Condvar,
}

impl Pipe {
    fn new(slots: usize, page_size: usize) -> Pipe {
        let state = State {
            ring: Ring::new(slots, page_size),
            readers: 1,
            writers: 1,
        };

        Pipe {
            state: Mutex::new(state),
            readable: Condvar::new(),
            writable: Condvar::new(),
        }
    }
}

write loops until all of data is in the ring. On each pass it checks for a reader, appends what fits, and either wakes the readers or waits on writable.

A woken thread must test its condition again before proceeding. A wake-up indicates that the state changed, and the change may be unrelated to the condition the thread waits for: in row 2 of the trace, a take of 3 bytes freed no slot. For this reason the wait is inside the loop, and the writer calls put again after every wake-up.

Listing 1.29 Blocking write, in Go. code/go/handpipe/main.go

// write returns when every byte of data is in the ring, or when no reader is
// left. While the ring is full it sleeps on writable.
func (p *pipe) write(data []byte) (int, error) {
	p.mu.Lock()
	defer p.mu.Unlock()

	written := 0

	for written < len(data) {
		if p.readers == 0 {
			return written, errBrokenPipe
		}

		n := p.ring.put(data[written:])
		written += n

		if n == 0 {
			p.writable.Wait()
			continue
		}

		p.readable.Broadcast()
	}

	return written, nil
}

Listing 1.30 Blocking write, in Rust. code/rust/handpipe/src/main.rs

impl Pipe {
    /// Returns when every byte of `data` is in the ring, or when no reader is
    /// left. While the ring is full it sleeps on `writable`.
    fn write(&self, data: &[u8]) -> Result<usize, BrokenPipe> {
        let mut state = self.state.lock();
        let mut written = 0;

        while written < data.len() {
            if state.readers == 0 {
                return Err(BrokenPipe);
            }

            let n = state.ring.put(&data[written..]);
            written += n;

            if n == 0 {
                self.writable.wait(&mut state);
                continue;
            }

            self.readable.notify_all();
        }

        Ok(written)
    }
}

read waits while the ring is empty and a writer exists. If it wakes with the ring still empty, writers is 0, take returns 0, and so does read. A read returns as soon as at least 1 byte is available, so the caller may receive fewer bytes than it asked for and must be prepared to call read again.

Listing 1.31 Blocking read, in Go. code/go/handpipe/main.go

// read returns as soon as it has moved at least one byte into buf. While the
// ring is empty and a writer exists it sleeps on readable. It returns 0 when
// the ring is empty and no writer is left: the end of the stream.
func (p *pipe) read(buf []byte) int {
	p.mu.Lock()
	defer p.mu.Unlock()

	for p.ring.empty() && p.writers > 0 {
		p.readable.Wait()
	}

	n := p.ring.take(buf)

	if n > 0 {
		p.writable.Broadcast()
	}

	return n
}

Listing 1.32 Blocking read, in Rust. code/rust/handpipe/src/main.rs

impl Pipe {
    /// Returns as soon as it has moved at least one byte into `buf`. While
    /// the ring is empty and a writer exists it sleeps on `readable`. It
    /// returns 0 when the ring is empty and no writer is left: the end of the
    /// stream.
    fn read(&self, buf: &mut [u8]) -> usize {
        let mut state = self.state.lock();

        self.readable
            .wait_while(&mut state, |state| state.ring.empty() && state.writers > 0);

        let n = state.ring.take(buf);

        if n > 0 {
            self.writable.notify_all();
        }

        n
    }
}

Closing an end decrements its count and wakes the opposite side. Without the wake-up, a reader already waiting on an empty ring would remain suspended after writers reached 0.

Listing 1.33 Closing an end, in Go. code/go/handpipe/main.go

// closeWriter gives up a write end. A reader asleep on an empty ring wakes
// and finds that no writer is left.
func (p *pipe) closeWriter() {
	p.mu.Lock()
	defer p.mu.Unlock()

	p.writers--
	p.readable.Broadcast()
}

// closeReader gives up a read end. A writer asleep on a full ring wakes and
// finds that no reader is left.
func (p *pipe) closeReader() {
	p.mu.Lock()
	defer p.mu.Unlock()

	p.readers--
	p.writable.Broadcast()
}

Listing 1.34 Closing an end, in Rust. code/rust/handpipe/src/main.rs

impl Pipe {
    /// Gives up a write end. A reader asleep on an empty ring wakes and finds
    /// that no writer is left.
    fn close_writer(&self) {
        self.state.lock().writers -= 1;
        self.readable.notify_all();
    }

    /// Gives up a read end. A writer asleep on a full ring wakes and finds
    /// that no reader is left.
    fn close_reader(&self) {
        self.state.lock().readers -= 1;
        self.writable.notify_all();
    }
}

Animation 1.1 runs a writer and a reader on the 4-slot ring. The writer’s second call writes 30 bytes. It blocks after 20 bytes and again after 28, and returns when all 30 bytes are in the ring.

A writer and a reader share a ring of 4 pages of 8 bytes. The writer puts 12 bytes, and the reader takes 3. The writer starts a write of 30 bytes; 20 fit, every slot is in use, and the writer sleeps. The reader takes 5 more bytes, which frees slot 0; the writer wakes, puts 8 bytes into slot 0, and sleeps again with 2 bytes left. The reader takes 8, the writer puts its last 2 bytes and its write returns 30. The writer closes its end, the reader takes the 26 bytes that remain, and its next read returns 0. — **Animation 1.1** A write of 30 bytes into a ring that holds 32. The writer blocks while no slot is free.

transfer tests the complete pipe. A writer thread sends 1048576 bytes in writes of 100000 bytes. Each write is larger than the ring’s capacity, so each one blocks several times. The reader verifies the order of the bytes, then reads again after the write end is closed. The read end is then closed and another write is attempted.

Listing 1.35 One megabyte through a 16-page ring, in Go. code/go/handpipe/main.go

// transfer sends one megabyte through a pipe of the kernel's size. Byte i of
// the stream has the value i mod 256, which lets the reader verify the order.
func transfer() {
	const total = 1 << 20

	p := newPipe(16, 4096)

	go func() {
		defer p.closeWriter()

		chunk := make([]byte, 100_000)

		for sent := 0; sent < total; sent += len(chunk) {
			chunk = chunk[:min(len(chunk), total-sent)]

			for i := range chunk {
				chunk[i] = byte(sent + i)
			}

			if _, err := p.write(chunk); err != nil {
				return
			}
		}
	}()

	buf := make([]byte, 30_000)
	received, inOrder := 0, true

	for n := p.read(buf); n > 0; n = p.read(buf) {
		for i, b := range buf[:n] {
			inOrder = inOrder && b == byte(received+i)
		}

		received += n
	}

	fmt.Printf("1048576 bytes written: %d bytes read, in order: %v\n", received, inOrder)
	fmt.Printf("write end closed: read returns %d\n", p.read(buf))

	p.closeReader()

	_, err := p.write([]byte("hello"))
	fmt.Printf("read end closed: write fails with %v\n", err)
}

Listing 1.36 One megabyte through a 16-page ring, in Rust. code/rust/handpipe/src/main.rs

/// Sends one megabyte through a pipe of the kernel's size. Byte `i` of the
/// stream has the value `i` mod 256, which lets the reader verify the order.
fn transfer() {
    const TOTAL: usize = 1 << 20;

    let pipe = Pipe::new(16, 4096);
    let mut buf = vec![0u8; 30_000];

    thread::scope(|scope| {
        scope.spawn(|| {
            let mut chunk = vec![0u8; 100_000];

            for sent in (0..TOTAL).step_by(chunk.len()) {
                chunk.truncate(TOTAL - sent);

                for (i, byte) in chunk.iter_mut().enumerate() {
                    *byte = (sent + i) as u8;
                }

                if pipe.write(&chunk).is_err() {
                    break;
                }
            }

            pipe.close_writer();
        });

        let (mut received, mut in_order) = (0, true);

        loop {
            let n = pipe.read(&mut buf);

            if n == 0 {
                break;
            }

            for (i, byte) in buf[..n].iter().enumerate() {
                in_order = in_order && *byte == (received + i) as u8;
            }

            received += n;
        }

        println!("1048576 bytes written: {received} bytes read, in order: {in_order}");
    });

    println!("write end closed: read returns {}", pipe.read(&mut buf));

    pipe.close_reader();

    if let Err(error) = pipe.write(b"hello") {
        println!("read end closed: write fails with {error}");
    }
}

1048576 bytes written: 1048576 bytes read, in order: true
write end closed: read returns 0
read end closed: write fails with EPIPE

All bytes arrived in order through 65536 bytes of buffer. The read that returns 0 and the write that fails with EPIPE match the pipefacts output.

Atomic writes. When several processes write to the same pipe, the kernel guarantees that a write of at most PIPE_BUF bytes (4096 on Linux) is kept contiguous in the stream. Larger writes can be interleaved with the writes of other processes. handpipe omits this guarantee: its write can block after copying part of a small request.

The end of a stream

readers and writers in struct pipe_inode_info count open ends, and both rules for the end of a stream are tests of these counts.

Call	Condition	Result
`read`	the ring is empty and `writers` is 0	returns 0
`write`	`readers` is 0	`SIGPIPE`, and the error `EPIPE`

The counts change when an end is opened or closed, and “closed” has a precise meaning. An end of a pipe is a struct file, and several descriptors can refer to the same struct file: dup creates another one, and fork gives the child a copy of every descriptor of the parent. The kernel releases the struct file, and lowers readers or writers, when the last of those descriptors is closed. A single forgotten descriptor in any process keeps the end open.

Figure 1.8 follows the counts through the life of the pipeline a | b.

The number of open read ends and write ends of the pipe in the pipeline a | b. After pipe the shell holds 1 of each. After 2 forks the shell, a and b hold 3 of each. After each process closes what it does not use, a holds the write end and b the read end: 1 and 1. After a exits the write count is 0. When b has read the last byte, read returns 0. — **Figure 1.8** The open ends of a pipe, counted over all processes.

The figure counts descriptors, as a programmer has to. Row 3 is the step that programs get wrong. After the forks, 3 processes hold a descriptor for the write end. The reader b holds one itself, and if it does not close it, b waits for the end of a stream whose last writer is b.

Building a pipeline

pipeline runs seq 1 100000 | grep 7 | wc -l with kernel pipes. The kernel’s rules for descriptors determine how it must manage them.

Descriptors are inherited. When a process (the parent) starts another (the child), the child receives copies of the parent’s descriptors. A copy refers to the same pipe end as the original.
An end is closed when its last descriptor is closed. The kernel counts the descriptors that refer to each end. read returns 0 only when the count for the write end is 0.

Suppose the parent creates a pipe, starts seq with the write end and grep with the read end, and keeps its own descriptors open. When seq exits, the write end still has an open descriptor, the parent’s. grep then blocks in read indefinitely and the pipeline does not finish. The rule for the parent follows: after starting a child, close the parent’s copy of every pipe end that was passed to it.

O_CLOEXEC enforces the same rule for descriptors that were not passed deliberately. The kernel closes a descriptor carrying this flag when the process executes a new program. Go’s os.Pipe and Rust’s io::pipe both set it, so each command receives its own ends and no descriptors for the other pipes.

A pipeline of 3 commands and 3 pipes, top to bottom. The standard output of seq 1 100000 is the write end of pipe 1, and the standard input of grep 7 is its read end; 588895 bytes pass through. grep writes to pipe 2, which wc -l reads; 241976 bytes pass through. wc writes 6 bytes to pipe 3, which the pipeline program reads: 40951. — **Figure 1.9** The pipeline that `pipeline` builds. The parent reads pipe 3 to collect the result.

start creates a pipe for each command. Each command writes to its own pipe and reads from the previous pipe, and the read end of the newest pipe is carried to the next iteration. The parent’s copies are closed in each iteration: explicitly in Go, and in Rust by ownership, since the ends are moved into the Command, which is dropped at the end of the statement.

Listing 1.37 The imports of pipeline, in Go. code/go/pipeline/main.go

import (
	"fmt"
	"io"
	"log"
	"os"
	"os/exec"
	"strings"
	"syscall"
)

Listing 1.38 The imports of pipeline, in Rust. code/rust/pipeline/src/main.rs

use std::io::{self, PipeReader, Read};
use std::os::unix::process::ExitStatusExt;
use std::process::{Child, Command, Stdio};

Listing 1.39 Starting the commands, each with its own pipe, in Go. code/go/pipeline/main.go

// start runs the commands of a pipeline. Command i writes to pipe i and reads
// from pipe i-1; the first command reads /dev/null. It returns the commands
// and the read end of the last pipe.
func start(stages [][]string) ([]*exec.Cmd, *os.File, error) {
	var input *os.File
	commands := make([]*exec.Cmd, 0, len(stages))

	for _, stage := range stages {
		reader, writer, err := os.Pipe()

		if err != nil {
			return nil, nil, err
		}

		command := exec.Command(stage[0], stage[1:]...)
		command.Stdout = writer

		if input != nil {
			command.Stdin = input
		}

		err = command.Start()

		// The child holds its own copies of both ends. Closing the parent's
		// copies leaves the child's as the only ones.
		writer.Close()

		if input != nil {
			input.Close()
		}

		if err != nil {
			reader.Close()

			return nil, nil, fmt.Errorf("%s: %w", stage[0], err)
		}

		commands, input = append(commands, command), reader
	}

	return commands, input, nil
}

Listing 1.40 Starting the commands, each with its own pipe, in Rust. code/rust/pipeline/src/main.rs

/// Runs the commands of a pipeline. Command `i` writes to pipe `i` and reads
/// from pipe `i - 1`; the first command reads /dev/null. Returns the children
/// and the read end of the last pipe.
fn start(stages: &[Vec<&str>]) -> io::Result<(Vec<Child>, PipeReader)> {
    let mut input = None;
    let mut children = Vec::with_capacity(stages.len());

    for stage in stages {
        let (reader, writer) = io::pipe()?;
        let stdin = input.take().map_or_else(Stdio::null, Stdio::from);

        // The `Command` owns the parent's copies of both ends. It is dropped
        // when this statement ends, which closes them and leaves the child's
        // copies as the only ones.
        let child = Command::new(stage[0])
            .args(&stage[1..])
            .stdin(stdin)
            .stdout(writer)
            .spawn()?;

        children.push(child);
        input = Some(reader);
    }

    let output = input.ok_or(io::ErrorKind::InvalidInput)?;

    Ok((children, output))
}

The kernel records how each process terminated: with an exit status chosen by the process, where 0 means success by convention, or by a signal. The parent retrieves the record with the wait system call.

Listing 1.41 Reporting how a command terminated, in Go. code/go/pipeline/main.go

// ending waits for a command and reports how it terminated: with an exit
// status, or by a signal.
func ending(command *exec.Cmd) string {
	command.Wait()

	state := command.ProcessState

	if status, ok := state.Sys().(syscall.WaitStatus); ok && status.Signaled() {
		return fmt.Sprintf("killed by signal %d", status.Signal())
	}

	return fmt.Sprintf("exited with status %d", state.ExitCode())
}

Listing 1.42 Reporting how a command terminated, in Rust. code/rust/pipeline/src/main.rs

/// Waits for a command and reports how it terminated: with an exit status,
/// or by a signal.
fn ending(child: &mut Child) -> io::Result<String> {
    let status = child.wait()?;

    let text = match (status.signal(), status.code()) {
        (Some(signal), _) => format!("killed by signal {signal}"),
        (None, Some(code)) => format!("exited with status {code}"),
        (None, None) => status.to_string(),
    };

    Ok(text)
}

run reads the last pipe until read returns 0. That occurs when the last command has exited and no descriptor for the write end remains, which depends on start having closed the parent’s copy.

Listing 1.43 Running a pipeline, in Go. code/go/pipeline/main.go

// run starts a pipeline, reads the last command's output to the end of the
// stream, and prints the output and each command's termination.
func run(line string) error {
	var stages [][]string

	for _, stage := range strings.Split(line, " | ") {
		stages = append(stages, strings.Fields(stage))
	}

	commands, output, err := start(stages)

	if err != nil {
		return err
	}

	defer output.Close()

	text, err := io.ReadAll(output)

	if err != nil {
		return err
	}

	fmt.Println(line)
	fmt.Printf("  output: %s\n", strings.Join(strings.Fields(string(text)), " "))

	for i, command := range commands {
		fmt.Printf("  %s: %s\n", stages[i][0], ending(command))
	}

	return nil
}

Listing 1.44 Running a pipeline, in Rust. code/rust/pipeline/src/main.rs

/// Starts a pipeline, reads the last command's output to the end of the
/// stream, and prints the output and each command's termination.
fn run(line: &str) -> io::Result<()> {
    let stages: Vec<Vec<&str>> = line
        .split(" | ")
        .map(|stage| stage.split_whitespace().collect())
        .collect();

    let (mut children, mut output) = start(&stages)?;

    let mut text = String::new();
    output.read_to_string(&mut text)?;

    let words: Vec<&str> = text.split_whitespace().collect();

    println!("{line}");
    println!("  output: {}", words.join(" "));

    for (stage, child) in stages.iter().zip(&mut children) {
        println!("  {}: {}", stage[0], ending(child)?);
    }

    Ok(())
}

seq 1 100000 | grep 7 | wc -l
  output: 40951
  seq: exited with status 0
  grep: exited with status 0
  wc: exited with status 0
yes | head -n 3
  output: y y y
  yes: killed by signal 13
  head: exited with status 0

In the first pipeline, each command reached the end of its input when the previous command exited, and all of them exited with status 0. The result can be verified independently. Of the 100000 5-digit strings from 00000 to 99999, 9⁵ = 59049 contain no 7. The range 1 to 100000 replaces 00000 with 100000, and neither contains a 7, so 100000 - 59049 = 40951 numbers in the range contain a 7.

The second pipeline terminates a writer that has no end of its own. yes writes the line y in a loop. head -n 3 exits after 3 lines, which closes the only read end of the pipe. The next write by yes raises SIGPIPE, signal 13, and the default action terminates the process. In a longer pipeline this propagates from right to left: each command that exits closes a read end, and the command before it is terminated on its next write.

Named pipes

The pipes so far have no name, so only processes that inherited the descriptors can use them. A named pipe, or FIFO, is the same kernel object with an entry in the file system, created by mkfifo. Unrelated processes open it by path, one for writing and one for reading, and the bytes pass through a struct pipe_inode_info exactly as before. Nothing is written to the disk. The p at the start of the permissions marks the type.

$ mkfifo /tmp/x/queue
$ ls -l /tmp/x/queue
prw-rw-r-- /tmp/x/queue
$ echo hello > /tmp/x/queue &
$ cat /tmp/x/queue
hello

An open of a FIFO for writing blocks until some process opens it for reading, and the reverse, which is why the echo above is started in the background.

Inspecting pipes on a live system

Command	Shows
`ls -l /proc/<pid>/fd`	which descriptors of a process are pipes, and the number of each pipe; equal numbers are the 2 ends of the same pipe
`cat /proc/sys/fs/pipe-max-size`	the largest capacity an ordinary user can set with `F_SETPIPE_SZ`: 1048576 here
`cat /proc/sys/fs/pipe-user-pages-soft`	the number of pages all pipes of a user may hold before new pipes are created with a single slot: 16384 here
`fcntl(fd, F_GETPIPE_SZ)`	the capacity of a pipe, from inside a program

A pipeline that appears to hang is usually diagnosed with the first command: find the process that is blocked, read its descriptors, and look for another process that still holds the write end of the same pipe number.

The complete programs

Listing 1.45 The complete pipefacts, in Go. code/go/pipefacts/main.go

// Command pipefacts measures a kernel pipe: its capacity,
// the unit in which a read frees space, how a reader observes the end of the
// stream, and the result of a write when no reader exists.
//
// Run it from code/go with:
//
//	go run ./pipefacts
package main

import (
	"errors"
	"fmt"
	"log"

	"golang.org/x/sys/unix"
)


// block is the size of every write the program attempts: one page.
const block = 4096

// newPipe creates a pipe in non-blocking mode: a read or a write that cannot
// proceed returns EAGAIN where it would otherwise suspend the caller.
func newPipe() (reader, writer int, err error) {
	var fds [2]int

	if err := unix.Pipe2(fds[:], unix.O_NONBLOCK|unix.O_CLOEXEC); err != nil {
		return 0, 0, fmt.Errorf("pipe2: %w", err)
	}

	return fds[0], fds[1], nil
}


// fill writes whole blocks until the kernel refuses one and returns the
// number of bytes accepted.
func fill(writer int) (int, error) {
	data := make([]byte, block)
	total := 0

	for {
		n, err := unix.Write(writer, data)

		if errors.Is(err, unix.EAGAIN) {
			return total, nil
		}

		if err != nil {
			return total, fmt.Errorf("write: %w", err)
		}

		total += n
	}
}

// drain reads and discards up to n bytes.
func drain(reader, n int) error {
	_, err := unix.Read(reader, make([]byte, n))

	return err
}


// capacity fills a pipe when it is new, after 1 byte has been read, and
// after 4096 bytes have been read. Each fill reports the free space.
func capacity() error {
	reader, writer, err := newPipe()

	if err != nil {
		return err
	}

	defer unix.Close(reader)
	defer unix.Close(writer)

	steps := []struct {
		read   int
		report string
	}{
		{0, "new pipe: %d bytes accepted\n"},
		{1, "after reading 1 byte: %d bytes accepted\n"},
		{block - 1, "after reading 4096 bytes in total: %d bytes accepted\n"},
	}

	for _, step := range steps {
		if err := drain(reader, step.read); err != nil {
			return err
		}

		accepted, err := fill(writer)

		if err != nil {
			return err
		}

		fmt.Printf(step.report, accepted)
	}

	return nil
}


// resize changes the capacity of a pipe with fcntl and measures it after
// each change.
func resize() error {
	reader, writer, err := newPipe()

	if err != nil {
		return err
	}

	defer unix.Close(reader)
	defer unix.Close(writer)

	for _, request := range []int{4 * block, 5000, 1 << 20} {
		granted, err := unix.FcntlInt(uintptr(writer), unix.F_SETPIPE_SZ, request)

		if err != nil {
			return fmt.Errorf("F_SETPIPE_SZ: %w", err)
		}

		accepted, err := fill(writer)

		if err != nil {
			return err
		}

		fmt.Printf("capacity of %d bytes requested: %d granted, %d bytes accepted\n", request, granted, accepted)

		if err := drain(reader, accepted); err != nil {
			return err
		}
	}

	return nil
}


// endOfStream reads an empty pipe while a write end is open, and again after
// the write end has been closed.
func endOfStream() error {
	reader, writer, err := newPipe()

	if err != nil {
		return err
	}

	defer unix.Close(reader)

	buf := make([]byte, block)

	_, err = unix.Read(reader, buf)
	fmt.Printf("empty pipe, write end open: read fails with %s\n", name(err))

	unix.Close(writer)

	n, err := unix.Read(reader, buf)

	if err != nil {
		return fmt.Errorf("read: %w", err)
	}

	fmt.Printf("empty pipe, write end closed: read returns %d\n", n)

	return nil
}


// brokenPipe writes to a pipe whose read end has been closed.
func brokenPipe() error {
	reader, writer, err := newPipe()

	if err != nil {
		return err
	}

	defer unix.Close(writer)

	unix.Close(reader)

	_, err = unix.Write(writer, []byte("hello"))
	fmt.Printf("read end closed: write fails with %s\n", name(err))

	return nil
}

// name returns the symbolic name of a system call error, such as EAGAIN.
func name(err error) string {
	var errno unix.Errno

	if errors.As(err, &errno) {
		return unix.ErrnoName(errno)
	}

	return fmt.Sprint(err)
}


func main() {
	for _, experiment := range []func() error{capacity, resize, endOfStream, brokenPipe} {
		if err := experiment(); err != nil {
			log.Fatal(err)
		}
	}
}

Listing 1.46 The complete pipefacts, in Rust. code/rust/pipefacts/src/main.rs

//! Measures a kernel pipe: its capacity, the unit in which
//! a read frees space, how a reader observes the end of the stream, and the
//! result of a write when no reader exists.
//!
//! Run it from code/rust with:
//!
//! ```text
//! cargo run -p pipefacts
//! ```

use std::os::fd::OwnedFd;

use nix::errno::Errno;
use nix::fcntl::{FcntlArg, OFlag, fcntl};
use nix::unistd::{pipe2, read, write};

/// The size of every write the program attempts: one page.
const BLOCK: usize = 4096;

/// Creates a pipe in non-blocking mode: a read or a write that cannot proceed
/// returns `EAGAIN` where it would otherwise suspend the caller. The read end
/// is first.
fn new_pipe() -> nix::Result<(OwnedFd, OwnedFd)> {
    pipe2(OFlag::O_NONBLOCK | OFlag::O_CLOEXEC)
}

/// Writes whole blocks until the kernel refuses one and returns the number of
/// bytes accepted.
fn fill(writer: &OwnedFd) -> nix::Result<usize> {
    let data = [0u8; BLOCK];
    let mut total = 0;

    loop {
        match write(writer, &data) {
            Ok(n) => total += n,
            Err(Errno::EAGAIN) => return Ok(total),
            Err(error) => return Err(error),
        }
    }
}

/// Reads and discards up to `n` bytes.
fn drain(reader: &OwnedFd, n: usize) -> nix::Result<usize> {
    read(reader, &mut vec![0u8; n])
}

/// Fills a pipe when it is new, after 1 byte has been read, and after 4096
/// bytes have been read. Each fill reports the free space.
fn capacity() -> nix::Result<()> {
    let (reader, writer) = new_pipe()?;

    let steps = [
        (0, "new pipe"),
        (1, "after reading 1 byte"),
        (BLOCK - 1, "after reading 4096 bytes in total"),
    ];

    for (count, label) in steps {
        drain(&reader, count)?;

        let accepted = fill(&writer)?;
        println!("{label}: {accepted} bytes accepted");
    }

    Ok(())
}

/// Changes the capacity of a pipe with fcntl and measures it after each
/// change.
fn resize() -> nix::Result<()> {
    let (reader, writer) = new_pipe()?;

    for request in [4 * BLOCK, 5000, 1 << 20] {
        let granted = fcntl(&writer, FcntlArg::F_SETPIPE_SZ(request as i32))?;
        let accepted = fill(&writer)?;

        println!(
            "capacity of {request} bytes requested: {granted} granted, {accepted} bytes accepted"
        );

        drain(&reader, accepted)?;
    }

    Ok(())
}

/// Reads an empty pipe while a write end is open, and again after the write
/// end has been closed.
fn end_of_stream() -> nix::Result<()> {
    let (reader, writer) = new_pipe()?;
    let mut buf = [0u8; BLOCK];

    if let Err(error) = read(&reader, &mut buf) {
        println!("empty pipe, write end open: read fails with {error:?}");
    }

    drop(writer);

    let n = read(&reader, &mut buf)?;
    println!("empty pipe, write end closed: read returns {n}");

    Ok(())
}

/// Writes to a pipe whose read end has been closed.
fn broken_pipe() -> nix::Result<()> {
    let (reader, writer) = new_pipe()?;

    drop(reader);

    if let Err(error) = write(&writer, b"hello") {
        println!("read end closed: write fails with {error:?}");
    }

    Ok(())
}

fn main() -> nix::Result<()> {
    capacity()?;
    resize()?;
    end_of_stream()?;
    broken_pipe()
}

Listing 1.47 The complete handpipe, in Go. code/go/handpipe/main.go

// Command handpipe builds a pipe by hand, with the same parts the kernel
// uses: a ring of pages for the bytes, a lock, a place for a reader to sleep
// and a place for a writer to sleep, and a count of the open ends.
//
// It traces a small ring call by call, repeats the capacity measurement of
// pipefacts on a ring of the kernel's size, and sends 1048576 bytes from a
// writer goroutine to a reader.
//
// Run it from code/go with:
//
//	go run ./handpipe
package main

import (
	"errors"
	"fmt"
	"strings"
	"sync"
)


// A page is one slot of the ring: a block of memory, and the position and
// length of the bytes in it that have not been read yet.
type page struct {
	data   []byte
	offset int
	length int
}

// A ring is the storage of a pipe. head counts the pages ever put into use
// and tail counts the pages ever emptied, so head-tail pages are in use, and
// a counter's remainder after division by the number of slots is its slot.
type ring struct {
	pages []page
	head  uint
	tail  uint
}

func newRing(slots, pageSize int) *ring {
	pages := make([]page, slots)

	for i := range pages {
		pages[i].data = make([]byte, pageSize)
	}

	return &ring{pages: pages}
}

func (r *ring) empty() bool {
	return r.head == r.tail
}

func (r *ring) full() bool {
	return r.head-r.tail == uint(len(r.pages))
}

// slot returns the page that a counter refers to.
func (r *ring) slot(counter uint) *page {
	return &r.pages[counter%uint(len(r.pages))]
}


// put copies as much of data as the ring has room for and returns the number
// of bytes it copied. The bytes go first into the unused end of the newest
// page, then into free slots, one page each.
func (r *ring) put(data []byte) int {
	copied := 0

	if !r.empty() {
		newest := r.slot(r.head - 1)
		end := newest.offset + newest.length

		n := copy(newest.data[end:], data)
		newest.length += n
		copied += n
	}

	for copied < len(data) && !r.full() {
		next := r.slot(r.head)
		next.offset = 0
		next.length = copy(next.data, data[copied:])

		copied += next.length
		r.head++
	}

	return copied
}


// take moves up to len(buf) of the oldest bytes into buf and returns the
// number it moved. A slot becomes free when the last byte of its page has
// been taken.
func (r *ring) take(buf []byte) int {
	moved := 0

	for moved < len(buf) && !r.empty() {
		oldest := r.slot(r.tail)

		n := copy(buf[moved:], oldest.data[oldest.offset:oldest.offset+oldest.length])
		oldest.offset += n
		oldest.length -= n
		moved += n

		if oldest.length == 0 {
			r.tail++
		}
	}

	return moved
}


// errBrokenPipe is what a write returns when no reader is left.
var errBrokenPipe = errors.New("EPIPE")

// A pipe is a ring shared by a reader and a writer goroutine. readable is where a reader
// sleeps while the ring is empty, and writable is where a writer sleeps while
// the ring is full. readers and writers count the open ends.
type pipe struct {
	mu       sync.Mutex
	readable *sync.Cond
	writable *sync.Cond
	ring     *ring
	readers  int
	writers  int
}

func newPipe(slots, pageSize int) *pipe {
	p := &pipe{ring: newRing(slots, pageSize), readers: 1, writers: 1}
	p.readable = sync.NewCond(&p.mu)
	p.writable = sync.NewCond(&p.mu)

	return p
}


// write returns when every byte of data is in the ring, or when no reader is
// left. While the ring is full it sleeps on writable.
func (p *pipe) write(data []byte) (int, error) {
	p.mu.Lock()
	defer p.mu.Unlock()

	written := 0

	for written < len(data) {
		if p.readers == 0 {
			return written, errBrokenPipe
		}

		n := p.ring.put(data[written:])
		written += n

		if n == 0 {
			p.writable.Wait()
			continue
		}

		p.readable.Broadcast()
	}

	return written, nil
}


// read returns as soon as it has moved at least one byte into buf. While the
// ring is empty and a writer exists it sleeps on readable. It returns 0 when
// the ring is empty and no writer is left: the end of the stream.
func (p *pipe) read(buf []byte) int {
	p.mu.Lock()
	defer p.mu.Unlock()

	for p.ring.empty() && p.writers > 0 {
		p.readable.Wait()
	}

	n := p.ring.take(buf)

	if n > 0 {
		p.writable.Broadcast()
	}

	return n
}


// closeWriter gives up a write end. A reader asleep on an empty ring wakes
// and finds that no writer is left.
func (p *pipe) closeWriter() {
	p.mu.Lock()
	defer p.mu.Unlock()

	p.writers--
	p.readable.Broadcast()
}

// closeReader gives up a read end. A writer asleep on a full ring wakes and
// finds that no reader is left.
func (p *pipe) closeReader() {
	p.mu.Lock()
	defer p.mu.Unlock()

	p.readers--
	p.writable.Broadcast()
}


// picture renders the ring as one cell per slot: the number of unread bytes
// in its page, or a dot when the slot is free.
func (r *ring) picture() string {
	cells := make([]string, len(r.pages))

	for i := range cells {
		cells[i] = "."
	}

	for counter := r.tail; counter != r.head; counter++ {
		cells[counter%uint(len(r.pages))] = fmt.Sprint(r.slot(counter).length)
	}

	return "[" + strings.Join(cells, " ") + "]"
}

// smallRing applies the calls of the trace to a ring of 4 pages of 8 bytes
// and prints the ring after each.
func smallRing() {
	r := newRing(4, 8)
	buf := make([]byte, 32)

	calls := []struct {
		op string
		n  int
	}{{"put", 12}, {"take", 3}, {"put", 30}, {"take", 5}, {"put", 30}}

	fmt.Println("ring of 4 pages of 8 bytes")

	for _, call := range calls {
		var done int

		switch call.op {
		case "put":
			done = r.put(buf[:call.n])
		case "take":
			done = r.take(buf[:call.n])
		}

		fmt.Printf("%-4s %2d: %2d bytes   head %d  tail %d   %s\n", call.op, call.n, done, r.head, r.tail, r.picture())
	}
}


// fill puts whole blocks into the ring until it accepts none and returns the
// number of bytes accepted.
func fill(r *ring, block []byte) int {
	total := 0

	for {
		n := r.put(block)

		if n == 0 {
			return total
		}

		total += n
	}
}

// kernelSize repeats the capacity measurement of pipefacts on a ring with
// the kernel's dimensions: 16 pages of 4096 bytes.
func kernelSize() {
	const pageSize = 4096

	r := newRing(16, pageSize)
	block := make([]byte, pageSize)

	steps := []struct {
		read   int
		report string
	}{
		{0, "new pipe: %d bytes accepted\n"},
		{1, "after reading 1 byte: %d bytes accepted\n"},
		{pageSize - 1, "after reading 4096 bytes in total: %d bytes accepted\n"},
	}

	for _, step := range steps {
		r.take(block[:step.read])
		fmt.Printf(step.report, fill(r, block))
	}
}


// transfer sends one megabyte through a pipe of the kernel's size. Byte i of
// the stream has the value i mod 256, which lets the reader verify the order.
func transfer() {
	const total = 1 << 20

	p := newPipe(16, 4096)

	go func() {
		defer p.closeWriter()

		chunk := make([]byte, 100_000)

		for sent := 0; sent < total; sent += len(chunk) {
			chunk = chunk[:min(len(chunk), total-sent)]

			for i := range chunk {
				chunk[i] = byte(sent + i)
			}

			if _, err := p.write(chunk); err != nil {
				return
			}
		}
	}()

	buf := make([]byte, 30_000)
	received, inOrder := 0, true

	for n := p.read(buf); n > 0; n = p.read(buf) {
		for i, b := range buf[:n] {
			inOrder = inOrder && b == byte(received+i)
		}

		received += n
	}

	fmt.Printf("1048576 bytes written: %d bytes read, in order: %v\n", received, inOrder)
	fmt.Printf("write end closed: read returns %d\n", p.read(buf))

	p.closeReader()

	_, err := p.write([]byte("hello"))
	fmt.Printf("read end closed: write fails with %v\n", err)
}


func main() {
	smallRing()
	fmt.Println()
	kernelSize()
	fmt.Println()
	transfer()
}

Listing 1.48 The complete handpipe, in Rust. code/rust/handpipe/src/main.rs

//! Builds a pipe by hand, with the same parts the kernel uses: a ring of
//! pages for the bytes, a lock, a place for a reader to sleep and a place for
//! a writer to sleep, and a count of the open ends.
//!
//! It traces a small ring call by call, repeats the capacity measurement of
//! pipefacts on a ring of the kernel's size, and sends 1048576 bytes from a
//! writer thread to a reader.
//!
//! Run it from code/rust with:
//!
//! ```text
//! cargo run -p handpipe
//! ```

use std::fmt;
use std::thread;

use parking_lot::{Condvar, Mutex};

/// One slot of the ring: a block of memory, and the position and length of
/// the bytes in it that have not been read yet.
struct Page {
    data: Vec<u8>,
    offset: usize,
    length: usize,
}

/// The storage of a pipe. `head` counts the pages ever put into use and
/// `tail` counts the pages ever emptied, so `head - tail` pages are in use,
/// and a counter's remainder after division by the number of slots is its
/// slot.
struct Ring {
    pages: Vec<Page>,
    head: usize,
    tail: usize,
}

impl Ring {
    fn new(slots: usize, page_size: usize) -> Ring {
        let pages = (0..slots)
            .map(|_| Page {
                data: vec![0; page_size],
                offset: 0,
                length: 0,
            })
            .collect();

        Ring {
            pages,
            head: 0,
            tail: 0,
        }
    }

    fn empty(&self) -> bool {
        self.head == self.tail
    }

    fn full(&self) -> bool {
        self.head - self.tail == self.pages.len()
    }

    /// The page that a counter refers to.
    fn slot(&mut self, counter: usize) -> &mut Page {
        let slots = self.pages.len();

        &mut self.pages[counter % slots]
    }
}

/// Copies as many bytes as both slices allow and returns the count.
fn copy(to: &mut [u8], from: &[u8]) -> usize {
    let n = to.len().min(from.len());
    to[..n].copy_from_slice(&from[..n]);

    n
}

impl Ring {
    /// Copies as much of `data` as the ring has room for and returns the
    /// number of bytes it copied. The bytes go first into the unused end of
    /// the newest page, then into free slots, one page each.
    fn put(&mut self, data: &[u8]) -> usize {
        let mut copied = 0;

        if !self.empty() {
            let newest = self.slot(self.head - 1);
            let end = newest.offset + newest.length;

            let n = copy(&mut newest.data[end..], data);
            newest.length += n;
            copied += n;
        }

        while copied < data.len() && !self.full() {
            let next = self.slot(self.head);
            next.offset = 0;
            next.length = copy(&mut next.data, &data[copied..]);

            copied += next.length;
            self.head += 1;
        }

        copied
    }
}

impl Ring {
    /// Moves up to `buf.len()` of the oldest bytes into `buf` and returns the
    /// number it moved. A slot becomes free when the last byte of its page
    /// has been taken.
    fn take(&mut self, buf: &mut [u8]) -> usize {
        let mut moved = 0;

        while moved < buf.len() && !self.empty() {
            let oldest = self.slot(self.tail);
            let unread = oldest.offset..oldest.offset + oldest.length;

            let n = copy(&mut buf[moved..], &oldest.data[unread]);
            oldest.offset += n;
            oldest.length -= n;
            moved += n;

            if oldest.length == 0 {
                self.tail += 1;
            }
        }

        moved
    }
}

/// What a write returns when no reader is left.
#[derive(Debug)]
struct BrokenPipe;

impl fmt::Display for BrokenPipe {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str("EPIPE")
    }
}

/// What the lock guards: the ring and the count of the open ends.
struct State {
    ring: Ring,
    readers: usize,
    writers: usize,
}

/// A ring shared by a reader and a writer thread. `readable` is where a reader sleeps while
/// the ring is empty, and `writable` is where a writer sleeps while the ring
/// is full.
struct Pipe {
    state: Mutex<State>,
    readable: Condvar,
    writable: Condvar,
}

impl Pipe {
    fn new(slots: usize, page_size: usize) -> Pipe {
        let state = State {
            ring: Ring::new(slots, page_size),
            readers: 1,
            writers: 1,
        };

        Pipe {
            state: Mutex::new(state),
            readable: Condvar::new(),
            writable: Condvar::new(),
        }
    }
}

impl Pipe {
    /// Returns when every byte of `data` is in the ring, or when no reader is
    /// left. While the ring is full it sleeps on `writable`.
    fn write(&self, data: &[u8]) -> Result<usize, BrokenPipe> {
        let mut state = self.state.lock();
        let mut written = 0;

        while written < data.len() {
            if state.readers == 0 {
                return Err(BrokenPipe);
            }

            let n = state.ring.put(&data[written..]);
            written += n;

            if n == 0 {
                self.writable.wait(&mut state);
                continue;
            }

            self.readable.notify_all();
        }

        Ok(written)
    }
}

impl Pipe {
    /// Returns as soon as it has moved at least one byte into `buf`. While
    /// the ring is empty and a writer exists it sleeps on `readable`. It
    /// returns 0 when the ring is empty and no writer is left: the end of the
    /// stream.
    fn read(&self, buf: &mut [u8]) -> usize {
        let mut state = self.state.lock();

        self.readable
            .wait_while(&mut state, |state| state.ring.empty() && state.writers > 0);

        let n = state.ring.take(buf);

        if n > 0 {
            self.writable.notify_all();
        }

        n
    }
}

impl Pipe {
    /// Gives up a write end. A reader asleep on an empty ring wakes and finds
    /// that no writer is left.
    fn close_writer(&self) {
        self.state.lock().writers -= 1;
        self.readable.notify_all();
    }

    /// Gives up a read end. A writer asleep on a full ring wakes and finds
    /// that no reader is left.
    fn close_reader(&self) {
        self.state.lock().readers -= 1;
        self.writable.notify_all();
    }
}

impl Ring {
    /// Renders the ring as one cell per slot: the unread bytes in its page,
    /// or a dot for a free slot.
    fn picture(&self) -> String {
        let slots = self.pages.len();
        let mut cells = vec![".".to_string(); slots];

        for counter in self.tail..self.head {
            cells[counter % slots] = self.pages[counter % slots].length.to_string();
        }

        format!("[{}]", cells.join(" "))
    }
}

/// Applies the calls of the trace to a ring of 4 pages of 8 bytes and prints
/// the ring after each.
fn small_ring() {
    let mut ring = Ring::new(4, 8);
    let mut buf = [0u8; 32];

    let calls = [
        ("put", 12),
        ("take", 3),
        ("put", 30),
        ("take", 5),
        ("put", 30),
    ];

    println!("ring of 4 pages of 8 bytes");

    for (op, n) in calls {
        let done = match op {
            "put" => ring.put(&buf[..n]),
            _ => ring.take(&mut buf[..n]),
        };

        println!(
            "{op:<4} {n:2}: {done:2} bytes   head {}  tail {}   {}",
            ring.head,
            ring.tail,
            ring.picture()
        );
    }
}

/// Puts whole blocks into the ring until it accepts none and returns the
/// number of bytes accepted.
fn fill(ring: &mut Ring, block: &[u8]) -> usize {
    let mut total = 0;

    loop {
        let n = ring.put(block);

        if n == 0 {
            return total;
        }

        total += n;
    }
}

/// Repeats the capacity measurement of pipefacts on a ring with the kernel's
/// dimensions: 16 pages of 4096 bytes.
fn kernel_size() {
    const PAGE_SIZE: usize = 4096;

    let mut ring = Ring::new(16, PAGE_SIZE);
    let mut block = [0u8; PAGE_SIZE];

    let steps = [
        (0, "new pipe"),
        (1, "after reading 1 byte"),
        (PAGE_SIZE - 1, "after reading 4096 bytes in total"),
    ];

    for (count, label) in steps {
        ring.take(&mut block[..count]);

        let accepted = fill(&mut ring, &block);
        println!("{label}: {accepted} bytes accepted");
    }
}

/// Sends one megabyte through a pipe of the kernel's size. Byte `i` of the
/// stream has the value `i` mod 256, which lets the reader verify the order.
fn transfer() {
    const TOTAL: usize = 1 << 20;

    let pipe = Pipe::new(16, 4096);
    let mut buf = vec![0u8; 30_000];

    thread::scope(|scope| {
        scope.spawn(|| {
            let mut chunk = vec![0u8; 100_000];

            for sent in (0..TOTAL).step_by(chunk.len()) {
                chunk.truncate(TOTAL - sent);

                for (i, byte) in chunk.iter_mut().enumerate() {
                    *byte = (sent + i) as u8;
                }

                if pipe.write(&chunk).is_err() {
                    break;
                }
            }

            pipe.close_writer();
        });

        let (mut received, mut in_order) = (0, true);

        loop {
            let n = pipe.read(&mut buf);

            if n == 0 {
                break;
            }

            for (i, byte) in buf[..n].iter().enumerate() {
                in_order = in_order && *byte == (received + i) as u8;
            }

            received += n;
        }

        println!("1048576 bytes written: {received} bytes read, in order: {in_order}");
    });

    println!("write end closed: read returns {}", pipe.read(&mut buf));

    pipe.close_reader();

    if let Err(error) = pipe.write(b"hello") {
        println!("read end closed: write fails with {error}");
    }
}

fn main() {
    small_ring();
    println!();
    kernel_size();
    println!();
    transfer();
}

Listing 1.49 The complete pipeline, in Go. code/go/pipeline/main.go

// Command pipeline runs `a | b | c` the way a shell does: each command's
// standard output is the write end of a pipe, and the next command's standard
// input is the read end of the same pipe.
//
// It runs the pipelines listed in main and prints the last command's output and how each
// command terminated.
//
// Run it from code/go with:
//
//	go run ./pipeline
package main

import (
	"fmt"
	"io"
	"log"
	"os"
	"os/exec"
	"strings"
	"syscall"
)


// start runs the commands of a pipeline. Command i writes to pipe i and reads
// from pipe i-1; the first command reads /dev/null. It returns the commands
// and the read end of the last pipe.
func start(stages [][]string) ([]*exec.Cmd, *os.File, error) {
	var input *os.File
	commands := make([]*exec.Cmd, 0, len(stages))

	for _, stage := range stages {
		reader, writer, err := os.Pipe()

		if err != nil {
			return nil, nil, err
		}

		command := exec.Command(stage[0], stage[1:]...)
		command.Stdout = writer

		if input != nil {
			command.Stdin = input
		}

		err = command.Start()

		// The child holds its own copies of both ends. Closing the parent's
		// copies leaves the child's as the only ones.
		writer.Close()

		if input != nil {
			input.Close()
		}

		if err != nil {
			reader.Close()

			return nil, nil, fmt.Errorf("%s: %w", stage[0], err)
		}

		commands, input = append(commands, command), reader
	}

	return commands, input, nil
}


// ending waits for a command and reports how it terminated: with an exit
// status, or by a signal.
func ending(command *exec.Cmd) string {
	command.Wait()

	state := command.ProcessState

	if status, ok := state.Sys().(syscall.WaitStatus); ok && status.Signaled() {
		return fmt.Sprintf("killed by signal %d", status.Signal())
	}

	return fmt.Sprintf("exited with status %d", state.ExitCode())
}


// run starts a pipeline, reads the last command's output to the end of the
// stream, and prints the output and each command's termination.
func run(line string) error {
	var stages [][]string

	for _, stage := range strings.Split(line, " | ") {
		stages = append(stages, strings.Fields(stage))
	}

	commands, output, err := start(stages)

	if err != nil {
		return err
	}

	defer output.Close()

	text, err := io.ReadAll(output)

	if err != nil {
		return err
	}

	fmt.Println(line)
	fmt.Printf("  output: %s\n", strings.Join(strings.Fields(string(text)), " "))

	for i, command := range commands {
		fmt.Printf("  %s: %s\n", stages[i][0], ending(command))
	}

	return nil
}


func main() {
	for _, line := range []string{"seq 1 100000 | grep 7 | wc -l", "yes | head -n 3"} {
		if err := run(line); err != nil {
			log.Fatal(err)
		}
	}
}

Listing 1.50 The complete pipeline, in Rust. code/rust/pipeline/src/main.rs

//! Runs `a | b | c` the way a shell does: each command's standard output is
//! the write end of a pipe, and the next command's standard input is the read
//! end of the same pipe.
//!
//! It runs the pipelines listed in main and prints the last command's output and how each
//! command terminated.
//!
//! Run it from code/rust with:
//!
//! ```text
//! cargo run -p pipeline
//! ```

use std::io::{self, PipeReader, Read};
use std::os::unix::process::ExitStatusExt;
use std::process::{Child, Command, Stdio};

/// Runs the commands of a pipeline. Command `i` writes to pipe `i` and reads
/// from pipe `i - 1`; the first command reads /dev/null. Returns the children
/// and the read end of the last pipe.
fn start(stages: &[Vec<&str>]) -> io::Result<(Vec<Child>, PipeReader)> {
    let mut input = None;
    let mut children = Vec::with_capacity(stages.len());

    for stage in stages {
        let (reader, writer) = io::pipe()?;
        let stdin = input.take().map_or_else(Stdio::null, Stdio::from);

        // The `Command` owns the parent's copies of both ends. It is dropped
        // when this statement ends, which closes them and leaves the child's
        // copies as the only ones.
        let child = Command::new(stage[0])
            .args(&stage[1..])
            .stdin(stdin)
            .stdout(writer)
            .spawn()?;

        children.push(child);
        input = Some(reader);
    }

    let output = input.ok_or(io::ErrorKind::InvalidInput)?;

    Ok((children, output))
}

/// Waits for a command and reports how it terminated: with an exit status,
/// or by a signal.
fn ending(child: &mut Child) -> io::Result<String> {
    let status = child.wait()?;

    let text = match (status.signal(), status.code()) {
        (Some(signal), _) => format!("killed by signal {signal}"),
        (None, Some(code)) => format!("exited with status {code}"),
        (None, None) => status.to_string(),
    };

    Ok(text)
}

/// Starts a pipeline, reads the last command's output to the end of the
/// stream, and prints the output and each command's termination.
fn run(line: &str) -> io::Result<()> {
    let stages: Vec<Vec<&str>> = line
        .split(" | ")
        .map(|stage| stage.split_whitespace().collect())
        .collect();

    let (mut children, mut output) = start(&stages)?;

    let mut text = String::new();
    output.read_to_string(&mut text)?;

    let words: Vec<&str> = text.split_whitespace().collect();

    println!("{line}");
    println!("  output: {}", words.join(" "));

    for (stage, child) in stages.iter().zip(&mut children) {
        println!("  {}: {}", stage[0], ending(child)?);
    }

    Ok(())
}

fn main() -> io::Result<()> {
    run("seq 1 100000 | grep 7 | wc -l")?;
    run("yes | head -n 3")
}

Summary

A pipe is a one-way byte stream between processes, buffered in kernel memory and reached through a read end and a write end.
The buffer is a ring of 16 slots, each referring to one page: 65536 bytes with 4096-byte pages.
head counts pages added and tail counts pages removed. Space is released a whole page at a time, when the last byte of the oldest page has been read.
A write to a full pipe blocks until a read releases a page. This limits a writer to the rate of its reader and bounds the memory between them.
A thread woken from a wait tests its condition again, because a wake-up reports that the state changed and says nothing about which change occurred.
read returns 0 when the pipe is empty and no descriptor for the write end remains.
write raises SIGPIPE and fails with EPIPE when no descriptor for the read end remains.
A program that builds a pipeline closes its own copy of each pipe end after passing it to a child. A copy left open prevents the end of the stream from being reported.

References

pipe(7) and pipe(2), Linux manual pages: https://man7.org/linux/man-pages/man7/pipe.7.html
fs/pipe.c and include/linux/pipe_fs_i.h, Linux 5.15: https://elixir.bootlin.com/linux/v5.15/source/fs/pipe.c
M. Kerrisk, The Linux Programming Interface, No Starch Press, 2010, chapter 44.
D. M. Ritchie, “The Evolution of the Unix Time-sharing System”, AT&T Bell Laboratories Technical Journal, 63(8), 1984.
Go, package os/signal, the section on SIGPIPE: https://pkg.go.dev/os/signal

Keyboard shortcuts

Democratizing Linux: A Pipe Dream