Event Loops and epoll

An event loop is a program structure in which a single thread waits for activity on many file descriptors at once and handles each descriptor when it is ready. The kernel mechanism behind it on Linux is epoll: the program registers descriptors with an epoll instance and then asks which of them are ready. Network servers, the Go runtime and Rust’s async runtimes are all built on this mechanism.

In this chapter, we will learn how epoll finds the ready descriptors without examining all of them. We will first measure its behaviour on pipes with epollfacts. Then we will write eventloop, a model of the interest list and the ready list, check it against the measurements, and run an event loop on it.

Blocking and readiness

By default a read on a pipe or a socket blocks: if no data is available, the kernel puts the calling thread to sleep until some arrives (chapter 1). A thread that is asleep in a read on 1 descriptor cannot notice that another descriptor has data.

A descriptor can be switched to non-blocking mode with the flag O_NONBLOCK. A read that would have to wait then returns the error EAGAIN immediately. A program could call read on each of its descriptors in a loop and skip those that return EAGAIN, and it would consume a whole CPU doing nothing. What is needed is a single call that sleeps until any of a set of descriptors is ready, meaning that a read or a write on it would not block. That call is the subject of this chapter.

The problem

A server with 10000 connections has to respond to whichever connection sends data next. A blocking read waits on 1 descriptor. The alternatives each have a cost.

With select and poll, the program passes all its descriptors on each call, and the kernel checks each one. The work per call is proportional to the number of descriptors registered. On a server most connections are idle at any moment, so most of that work finds nothing.

epoll separates registration from waiting. The program registers a descriptor once, with epoll_ctl. From then on the kernel itself records when the descriptor becomes ready. epoll_wait returns the recorded descriptors, and its work is proportional to the number that are ready.

The ready list is filled at the moment of the event. In chapter 1 a write to a pipe woke the readers sleeping on the pipe’s wait queue. Registering a descriptor with epoll adds an entry to that same wait queue, and the entry’s wake-up function puts the registration on the ready list. The writer does this work as part of its write.

The interface

epoll consists of 3 system calls.

A registration and an event are both a struct epoll_event. This is the definition in include/uapi/linux/eventpoll.h.

struct epoll_event {
        __poll_t events;       /* which conditions: EPOLLIN, EPOLLOUT, ... */
        __u64 data;            /* chosen by the program, returned with each event */
};

data is how the program recognises an event. The kernel does not interpret it. A program stores the descriptor number or a pointer to its own record of the connection, and gets it back in epoll_wait.

A server on a single thread uses the calls in the following pattern.

The descriptors are in non-blocking mode. epoll reports that a descriptor is ready, and the non-blocking mode guarantees that the handler does not sleep if the situation has changed by the time it reads.

From arriving data to epoll_wait

The claim that epoll_wait does work proportional to the number of ready descriptors rests on 1 mechanism, which we follow for a registered descriptor 7.

1. At registration, epoll_ctl creates an item for descriptor 7 and adds an entry to the wait queue of the file behind it. A wait queue is the list of sleepers that the file wakes when its state changes (chapter 1). The entry that epoll adds is a function in place of a sleeping thread.
2. Later, data arrives for descriptor 7. The code that delivers the data, a write by another process or the network stack, wakes the wait queue of the file.
3. Waking the queue calls the function, ep_poll_callback. It puts the item for descriptor 7 on the ready list of the epoll instance.
4. It then wakes any thread that is asleep in epoll_wait on this instance.
5. epoll_wait takes the items from the ready list, asks each file once more whether it is really ready, and copies the events to the program.

No step searches the registered descriptors. The descriptor announces itself at the moment it becomes ready.

The kernel’s structures

An epoll instance is a struct eventpoll, and each registration is a struct epitem. Both are defined in fs/eventpoll.c of Linux 5.15. These are the fields we use.

struct eventpoll {
        struct mutex mtx;                      /* held by epoll_ctl and epoll_wait */
        wait_queue_head_t wq;                  /* threads asleep in epoll_wait */
        struct list_head rdllist;              /* the ready list */
        struct rb_root_cached rbr;             /* the interest list: a tree of epitems */
        /* ... */
};

struct epitem {
        struct rb_node rbn;                    /* its node in rbr */
        struct list_head rdllink;              /* its link in rdllist */
        struct epoll_filefd ffd;               /* the file and descriptor it watches */
        struct eppoll_entry *pwqlist;          /* its entry on the file's wait queue */
        struct eventpoll *ep;                  /* the instance it belongs to */
        struct epoll_event event;              /* the events asked for, and data */
        /* ... */
};

An epoll instance is itself a file, and the kernel shows its registrations in /proc. This is the session manager of our user account, pid 1203, whose descriptor 4 is an epoll instance:

$ ls -l /proc/1203/fd | grep eventpoll
4 -> anon_inode:[eventpoll]
$ cat /proc/1203/fdinfo/4
pos:	0
flags:	02000002
tfd:       29 events:       19 data:     aaaaf18e8be0  pos:0 ino:b06c sdev:8
tfd:       35 events:       19 data:     aaaaf1953a90  pos:0 ino:b06e sdev:8
tfd:       15 events:       19 data:     aaaaf173f750  pos:0 ino:b049 sdev:8

Each tfd line is an item of the interest list: the descriptor it watches, the events in hexadecimal, and data. 19 is EPOLLIN (1), EPOLLERR (8) and EPOLLHUP (0x10) together. The data values are addresses in the memory of the program, which stored a pointer to its own record of each descriptor.

Measuring epoll

epollfacts uses golang.org/x/sys/unix in Go and the nix crate with the event and fs features in Rust.

$ cargo new epollfacts
$ cargo add -p epollfacts nix --features event,fs

watch creates an epoll instance and a number of pipes, and registers the read end of each pipe for the event EPOLLIN, readability. A registration carries a value chosen by the program, here the number of the pipe, which the kernel returns with each event. ready calls epoll_wait with a timeout of 0, so that it reports the current state without blocking.

Listing 9.1 The imports of epollfacts, in Go. code/go/epollfacts/main.go

import (
	"fmt"
	"log"

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

Listing 9.2 The imports of epollfacts, in Rust. code/rust/epollfacts/src/main.rs

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

use nix::fcntl::OFlag;
use nix::sys::epoll::{Epoll, EpollCreateFlags, EpollEvent, EpollFlags, EpollTimeout};
use nix::unistd::{pipe2, read, write};

Listing 9.3 Creating an epoll instance and registering pipes, in Go. code/go/epollfacts/main.go

// A pipe is the pair of descriptors of one pipe.
type pipe struct {
	reader, writer int
}

// watch creates an epoll instance and n pipes, and registers the read end of
// every pipe for readability with the given extra flags. The number of the
// pipe is stored in the registration and comes back with each event.
func watch(n int, flags uint32) (int, []pipe, error) {
	epoll, err := unix.EpollCreate1(unix.EPOLL_CLOEXEC)

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

	pipes := make([]pipe, n)

	for i := range pipes {
		var fds [2]int

		if err := unix.Pipe2(fds[:], unix.O_CLOEXEC); err != nil {
			return 0, nil, err
		}

		pipes[i] = pipe{reader: fds[0], writer: fds[1]}
		event := unix.EpollEvent{Events: unix.EPOLLIN | flags, Fd: int32(i)}

		if err := unix.EpollCtl(epoll, unix.EPOLL_CTL_ADD, fds[0], &event); err != nil {
			return 0, nil, err
		}
	}

	return epoll, pipes, nil
}

// ready returns the events that are ready now. A timeout of 0 makes the
// wait return at once.
func ready(epoll int) ([]unix.EpollEvent, error) {
	events := make([]unix.EpollEvent, 16)
	n, err := unix.EpollWait(epoll, events, 0)

	if err != nil {
		return nil, err
	}

	return events[:n], nil
}

// release closes an epoll instance and its pipes.
func release(epoll int, pipes []pipe) {
	unix.Close(epoll)

	for _, p := range pipes {
		unix.Close(p.reader)
		unix.Close(p.writer)
	}
}

Listing 9.4 Creating an epoll instance and registering pipes, in Rust. code/rust/epollfacts/src/main.rs

/// The pair of descriptors of one pipe. The write end is an `Option` so that
/// an experiment can close it.
struct Pipe {
    reader: OwnedFd,
    writer: Option<OwnedFd>,
}

/// Creates an epoll instance and `n` pipes, and registers the read end of
/// every pipe for readability with the given extra flags. The number of the
/// pipe is stored in the registration and comes back with each event.
fn watch(n: usize, flags: EpollFlags) -> nix::Result<(Epoll, Vec<Pipe>)> {
    let epoll = Epoll::new(EpollCreateFlags::EPOLL_CLOEXEC)?;
    let mut pipes = Vec::with_capacity(n);

    for i in 0..n {
        let (reader, writer) = pipe2(OFlag::O_CLOEXEC)?;

        epoll.add(
            &reader,
            EpollEvent::new(EpollFlags::EPOLLIN | flags, i as u64),
        )?;

        pipes.push(Pipe {
            reader,
            writer: Some(writer),
        });
    }

    Ok((epoll, pipes))
}

/// Returns the events that are ready now. A timeout of 0 makes the wait
/// return at once.
fn ready(epoll: &Epoll) -> nix::Result<Vec<EpollEvent>> {
    let mut events = [EpollEvent::empty(); 16];
    let n = epoll.wait(&mut events, EpollTimeout::ZERO)?;

    Ok(events[..n].to_vec())
}

/// Writes "hello" to a pipe whose write end is open.
fn send(pipe: &Pipe) -> nix::Result<()> {
    if let Some(writer) = &pipe.writer {
        write(writer, b"hello")?;
    }

    Ok(())
}

level uses the default kind of registration, called level-triggered. It waits before any write, after a write to pipe 1, a second time without reading the data, after reading it, and after closing the write end.

Listing 9.5 Level-triggered events, in Go. code/go/epollfacts/main.go

// level registers 3 pipes level-triggered and waits before a write, after a
// write, again without reading, and after reading.
func level() error {
	epoll, pipes, err := watch(3, 0)

	if err != nil {
		return err
	}

	defer release(epoll, pipes)

	events, err := ready(epoll)

	if err != nil {
		return err
	}

	fmt.Printf("3 pipes registered, none written: %d event(s)\n", len(events))

	if _, err := unix.Write(pipes[1].writer, []byte("hello")); err != nil {
		return err
	}

	events, err = ready(epoll)

	if err != nil || len(events) != 1 {
		return fmt.Errorf("wait after a write: %d events, %v", len(events), err)
	}

	fmt.Printf("write to pipe 1: 1 event, for pipe %d\n", events[0].Fd)

	events, err = ready(epoll)

	if err != nil {
		return err
	}

	fmt.Printf("level-triggered, data not read, wait again: %d event(s)\n", len(events))

	if _, err := unix.Read(pipes[1].reader, make([]byte, 16)); err != nil {
		return err
	}

	events, err = ready(epoll)

	if err != nil {
		return err
	}

	fmt.Printf("after reading the data: %d event(s)\n", len(events))

	unix.Close(pipes[1].writer)
	events, err = ready(epoll)

	if err != nil || len(events) != 1 {
		return fmt.Errorf("wait after a close: %d events, %v", len(events), err)
	}

	fmt.Printf("write end of pipe 1 closed: event has EPOLLHUP %v\n", events[0].Events&unix.EPOLLHUP != 0)

	return nil
}

Listing 9.6 Level-triggered events, in Rust. code/rust/epollfacts/src/main.rs

/// Registers 3 pipes level-triggered and waits before a write, after a write,
/// again without reading, and after reading.
fn level() -> nix::Result<()> {
    let (epoll, mut pipes) = watch(3, EpollFlags::empty())?;

    println!(
        "3 pipes registered, none written: {} event(s)",
        ready(&epoll)?.len()
    );

    send(&pipes[1])?;

    if let [event] = ready(&epoll)?.as_slice() {
        println!("write to pipe 1: 1 event, for pipe {}", event.data());
    }

    println!(
        "level-triggered, data not read, wait again: {} event(s)",
        ready(&epoll)?.len()
    );

    read(&pipes[1].reader, &mut [0u8; 16])?;
    println!("after reading the data: {} event(s)", ready(&epoll)?.len());

    pipes[1].writer = None;

    if let [event] = ready(&epoll)?.as_slice() {
        println!(
            "write end of pipe 1 closed: event has EPOLLHUP {}",
            event.events().contains(EpollFlags::EPOLLHUP)
        );
    }

    Ok(())
}

edge registers a pipe with the flag EPOLLET, edge-triggered, and waits 2 times after a write and again after a further write.

Listing 9.7 Edge-triggered events, in Go. code/go/epollfacts/main.go

// edge registers a pipe edge-triggered and waits 2 times after one write,
// and once more after a second write.
func edge() error {
	epoll, pipes, err := watch(1, unix.EPOLLET)

	if err != nil {
		return err
	}

	defer release(epoll, pipes)

	var counts [3]int

	for i := range counts {
		if i != 1 {
			if _, err := unix.Write(pipes[0].writer, []byte("hello")); err != nil {
				return err
			}
		}

		events, err := ready(epoll)

		if err != nil {
			return err
		}

		counts[i] = len(events)
	}

	fmt.Printf("edge-triggered, data not read: wait %d event(s), wait again %d event(s)\n", counts[0], counts[1])
	fmt.Printf("edge-triggered, after another write: %d event(s)\n", counts[2])

	return nil
}

Listing 9.8 Edge-triggered events, in Rust. code/rust/epollfacts/src/main.rs

/// Registers a pipe edge-triggered and waits 2 times after one write, and
/// once more after a second write.
fn edge() -> nix::Result<()> {
    let (epoll, pipes) = watch(1, EpollFlags::EPOLLET)?;
    let mut counts = [0; 3];

    for (i, count) in counts.iter_mut().enumerate() {
        if i != 1 {
            send(&pipes[0])?;
        }

        *count = ready(&epoll)?.len();
    }

    println!(
        "edge-triggered, data not read: wait {} event(s), wait again {} event(s)",
        counts[0], counts[1]
    );
    println!(
        "edge-triggered, after another write: {} event(s)",
        counts[2]
    );

    Ok(())
}

many registers 1000 pipes and writes to pipe 700.

Listing 9.9 1000 registrations and 1 ready descriptor, in Go. code/go/epollfacts/main.go

// many registers 1000 pipes and writes to 1 of them.
func many() error {
	epoll, pipes, err := watch(1000, 0)

	if err != nil {
		return err
	}

	defer release(epoll, pipes)

	if _, err := unix.Write(pipes[700].writer, []byte("hello")); err != nil {
		return err
	}

	events, err := ready(epoll)

	if err != nil || len(events) != 1 {
		return fmt.Errorf("wait with 1000 pipes: %d events, %v", len(events), err)
	}

	fmt.Printf("1000 pipes registered, pipe 700 written: 1 event, for pipe %d\n", events[0].Fd)

	return nil
}

Listing 9.10 1000 registrations and 1 ready descriptor, in Rust. code/rust/epollfacts/src/main.rs

/// Registers 1000 pipes and writes to 1 of them.
fn many() -> nix::Result<()> {
    let (epoll, pipes) = watch(1000, EpollFlags::empty())?;

    send(&pipes[700])?;

    if let [event] = ready(&epoll)?.as_slice() {
        println!(
            "1000 pipes registered, pipe 700 written: 1 event, for pipe {}",
            event.data()
        );
    }

    Ok(())
}

3 pipes registered, none written: 0 event(s)
write to pipe 1: 1 event, for pipe 1
level-triggered, data not read, wait again: 1 event(s)
after reading the data: 0 event(s)
write end of pipe 1 closed: event has EPOLLHUP true
edge-triggered, data not read: wait 1 event(s), wait again 0 event(s)
edge-triggered, after another write: 1 event(s)
1000 pipes registered, pipe 700 written: 1 event, for pipe 700

The choice has a consequence for the handler. With a level-triggered registration, a handler may read part of the data and return; the descriptor is reported again. With an edge-triggered registration, a handler has to read until read returns EAGAIN, because no further event will announce the data it leaves behind. Edge-triggered registrations save repeated reports and are used by runtimes that follow this rule strictly.

Modelling epoll

A pipe is reduced to a count of buffered bytes. The queued flag keeps a pipe from appearing on the ready list more than once.

Listing 9.11 The imports of eventloop, in Go. code/go/eventloop/main.go

import "fmt"

Listing 9.12 Pipes, the interest list and the ready list, in Go. code/go/eventloop/main.go

// A pipe is the read end of a pipe as epoll sees it: the bytes waiting to be
// read, and its registration. queued is set while the pipe is on the ready
// list.
type pipe struct {
	buffered   int
	registered bool
	edge       bool
	queued     bool
}

// A kernel holds the pipes and one epoll instance. The registered pipes are
// the interest list. ready is the ready list, in the order in which the
// pipes became ready. examined counts the registrations that waits and scans
// have looked at.
type kernel struct {
	pipes    []pipe
	ready    []int
	examined int
}

func newKernel(pipes int) *kernel {
	return &kernel{pipes: make([]pipe, pipes)}
}

Listing 9.13 Pipes, the interest list and the ready list, in Rust. code/rust/eventloop/src/main.rs

/// The read end of a pipe as epoll sees it: the bytes waiting to be read, and
/// its registration. `queued` is set while the pipe is on the ready list.
#[derive(Clone, Default)]
struct Pipe {
    buffered: usize,
    registered: bool,
    edge: bool,
    queued: bool,
}

/// Holds the pipes and one epoll instance. The registered pipes are the
/// interest list. `ready` is the ready list, in the order in which the pipes
/// became ready. `examined` counts the registrations that waits and scans
/// have looked at.
struct Kernel {
    pipes: Vec<Pipe>,
    ready: Vec<usize>,
    examined: usize,
}

impl Kernel {
    fn new(pipes: usize) -> Kernel {
        Kernel {
            pipes: vec![Pipe::default(); pipes],
            ready: Vec::new(),
            examined: 0,
        }
    }
}

write is where a pipe becomes ready, and it is write that queues the pipe. No part of the model searches for ready pipes.

Listing 9.14 Registering, and queuing on a write, in Go. code/go/eventloop/main.go

// register adds a pipe to the interest list. A pipe that is already
// readable goes on the ready list at once.
func (k *kernel) register(index int, edge bool) {
	p := &k.pipes[index]
	p.registered, p.edge = true, edge

	if p.buffered > 0 {
		k.enqueue(index)
	}
}

// enqueue puts a pipe on the ready list unless it is already there.
func (k *kernel) enqueue(index int) {
	if p := &k.pipes[index]; !p.queued {
		p.queued = true
		k.ready = append(k.ready, index)
	}
}

// write adds bytes to a pipe. This is the moment the pipe becomes ready, and
// the writer puts it on the ready list.
func (k *kernel) write(index, n int) {
	k.pipes[index].buffered += n

	if k.pipes[index].registered {
		k.enqueue(index)
	}
}

// read removes up to n bytes from a pipe and returns the number removed.
func (k *kernel) read(index, n int) int {
	n = min(n, k.pipes[index].buffered)
	k.pipes[index].buffered -= n

	return n
}

Listing 9.15 Registering, and queuing on a write, in Rust. code/rust/eventloop/src/main.rs

impl Kernel {
    /// Adds a pipe to the interest list. A pipe that is already readable goes
    /// on the ready list at once.
    fn register(&mut self, index: usize, edge: bool) {
        let pipe = &mut self.pipes[index];

        pipe.registered = true;
        pipe.edge = edge;

        if pipe.buffered > 0 {
            self.enqueue(index);
        }
    }

    /// Puts a pipe on the ready list unless it is already there.
    fn enqueue(&mut self, index: usize) {
        if !self.pipes[index].queued {
            self.pipes[index].queued = true;
            self.ready.push(index);
        }
    }

    /// Adds bytes to a pipe. This is the moment the pipe becomes ready, and
    /// the writer puts it on the ready list.
    fn write(&mut self, index: usize, n: usize) {
        self.pipes[index].buffered += n;

        if self.pipes[index].registered {
            self.enqueue(index);
        }
    }

    /// Removes up to `n` bytes from a pipe and returns the number removed.
    fn read(&mut self, index: usize, n: usize) -> usize {
        let n = n.min(self.pipes[index].buffered);
        self.pipes[index].buffered -= n;

        n
    }
}

wait takes the ready list and checks each entry again, since the data may have been read in the meantime. The difference between the kinds of registration is a single statement: a level-triggered pipe that is reported is queued again, and an edge-triggered pipe is left off the list until the next write queues it.

Listing 9.16 Waiting, and scanning for comparison, in Go. code/go/eventloop/main.go

// wait empties the ready list and returns the pipes on it that are still
// readable. A level-triggered pipe that is reported goes back on the list,
// so the next wait checks it again. An edge-triggered pipe does not, and is
// reported again only after another write.
func (k *kernel) wait() []int {
	queued := k.ready
	k.ready = nil

	var events []int

	for _, index := range queued {
		p := &k.pipes[index]
		p.queued = false
		k.examined++

		if p.buffered == 0 {
			continue
		}

		events = append(events, index)

		if !p.edge {
			k.enqueue(index)
		}
	}

	return events
}

// scan is the method of select and poll: it examines every registered pipe
// on every call.
func (k *kernel) scan() []int {
	var events []int

	for index, p := range k.pipes {
		if !p.registered {
			continue
		}

		k.examined++

		if p.buffered > 0 {
			events = append(events, index)
		}
	}

	return events
}

Listing 9.17 Waiting, and scanning for comparison, in Rust. code/rust/eventloop/src/main.rs

impl Kernel {
    /// Empties the ready list and returns the pipes on it that are still
    /// readable. A level-triggered pipe that is reported goes back on the
    /// list, so the next wait checks it again. An edge-triggered pipe does
    /// not, and is reported again only after another write.
    fn wait(&mut self) -> Vec<usize> {
        let queued = std::mem::take(&mut self.ready);
        let mut events = Vec::new();

        for index in queued {
            self.pipes[index].queued = false;
            self.examined += 1;

            if self.pipes[index].buffered == 0 {
                continue;
            }

            events.push(index);

            if !self.pipes[index].edge {
                self.enqueue(index);
            }
        }

        events
    }

    /// The method of select and poll: it examines every registered pipe on
    /// every call.
    fn scan(&mut self) -> Vec<usize> {
        let mut events = Vec::new();

        for (index, pipe) in self.pipes.iter().enumerate() {
            if !pipe.registered {
                continue;
            }

            self.examined += 1;

            if pipe.buffered > 0 {
                events.push(index);
            }
        }

        events
    }
}

The model repeats the measurements. many also counts how many registrations a wait and a scan examine.

Listing 9.18 The measurements on the model, in Go. code/go/eventloop/main.go

// level repeats the level-triggered measurement of epollfacts.
func level() {
	k := newKernel(3)

	for index := range k.pipes {
		k.register(index, false)
	}

	fmt.Printf("3 pipes registered, none written: %d event(s)\n", len(k.wait()))

	k.write(1, 5)

	if events := k.wait(); len(events) == 1 {
		fmt.Printf("write to pipe 1: 1 event, for pipe %d\n", events[0])
	}

	fmt.Printf("level-triggered, data not read, wait again: %d event(s)\n", len(k.wait()))

	k.read(1, 16)
	fmt.Printf("after reading the data: %d event(s)\n", len(k.wait()))
}

// edge repeats the edge-triggered measurement.
func edge() {
	k := newKernel(1)
	k.register(0, true)

	k.write(0, 5)
	first, second := len(k.wait()), len(k.wait())
	fmt.Printf("edge-triggered, data not read: wait %d event(s), wait again %d event(s)\n", first, second)

	k.write(0, 5)
	fmt.Printf("edge-triggered, after another write: %d event(s)\n", len(k.wait()))
}

// many compares the work of a wait with the work of a scan when 1000 pipes
// are registered and 1 is ready.
func many() {
	k := newKernel(1000)

	for index := range k.pipes {
		k.register(index, false)
	}

	k.write(700, 5)

	if events := k.wait(); len(events) == 1 {
		fmt.Printf("1000 pipes registered, pipe 700 written: 1 event, for pipe %d\n", events[0])
	}

	byWait := k.examined
	k.scan()
	fmt.Printf("registrations examined: %d by the wait, %d by a scan\n", byWait, k.examined-byWait)
}

Listing 9.19 The measurements on the model, in Rust. code/rust/eventloop/src/main.rs

/// Repeats the level-triggered measurement of epollfacts.
fn level() {
    let mut kernel = Kernel::new(3);

    for index in 0..3 {
        kernel.register(index, false);
    }

    println!(
        "3 pipes registered, none written: {} event(s)",
        kernel.wait().len()
    );

    kernel.write(1, 5);

    if let [index] = kernel.wait().as_slice() {
        println!("write to pipe 1: 1 event, for pipe {index}");
    }

    println!(
        "level-triggered, data not read, wait again: {} event(s)",
        kernel.wait().len()
    );

    kernel.read(1, 16);
    println!("after reading the data: {} event(s)", kernel.wait().len());
}

/// Repeats the edge-triggered measurement.
fn edge() {
    let mut kernel = Kernel::new(1);
    kernel.register(0, true);

    kernel.write(0, 5);
    let (first, second) = (kernel.wait().len(), kernel.wait().len());
    println!("edge-triggered, data not read: wait {first} event(s), wait again {second} event(s)");

    kernel.write(0, 5);
    println!(
        "edge-triggered, after another write: {} event(s)",
        kernel.wait().len()
    );
}

/// Compares the work of a wait with the work of a scan when 1000 pipes are
/// registered and 1 is ready.
fn many() {
    let mut kernel = Kernel::new(1000);

    for index in 0..1000 {
        kernel.register(index, false);
    }

    kernel.write(700, 5);

    if let [index] = kernel.wait().as_slice() {
        println!("1000 pipes registered, pipe 700 written: 1 event, for pipe {index}");
    }

    let by_wait = kernel.examined;
    kernel.scan();
    println!(
        "registrations examined: {by_wait} by the wait, {} by a scan",
        kernel.examined - by_wait
    );
}

The loop

An event loop repeats 2 steps: wait for events, then run the handler of each ready descriptor. A handler must return quickly and must not block, because every other descriptor waits while it runs.

Listing 9.20 An event loop on the model, in Go. code/go/eventloop/main.go

// turn is one iteration of an event loop: wait for ready pipes, then run the
// handler for each. It returns the pipes it handled.
func (k *kernel) turn(handle func(k *kernel, index int)) []int {
	events := k.wait()

	for _, index := range events {
		handle(k, index)
	}

	return events
}

// loop registers 4 pipes, writes to 2 of them, and runs an event loop whose
// handler reads at most 4 bytes for each event.
func loop() {
	k := newKernel(4)

	for index := range k.pipes {
		k.register(index, false)
	}

	k.write(2, 5)
	k.write(0, 3)

	handle := func(k *kernel, index int) { k.read(index, 4) }

	fmt.Printf("event loop: turn 1 handles %v, turn 2 handles %v, turn 3 handles %v\n",
		k.turn(handle), k.turn(handle), k.turn(handle))
}

Listing 9.21 An event loop on the model, in Rust. code/rust/eventloop/src/main.rs

impl Kernel {
    /// One iteration of an event loop: wait for ready pipes, then run the
    /// handler for each. Returns the pipes it handled.
    fn turn(&mut self, handle: fn(&mut Kernel, usize)) -> Vec<usize> {
        let events = self.wait();

        for &index in &events {
            handle(self, index);
        }

        events
    }
}

/// Formats pipe numbers the way Go prints a slice.
fn list(pipes: &[usize]) -> String {
    let words: Vec<String> = pipes.iter().map(usize::to_string).collect();

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

/// Registers 4 pipes, writes to 2 of them, and runs an event loop whose
/// handler reads at most 4 bytes for each event.
fn event_loop() {
    let mut kernel = Kernel::new(4);

    for index in 0..4 {
        kernel.register(index, false);
    }

    kernel.write(2, 5);
    kernel.write(0, 3);

    let handle = |kernel: &mut Kernel, index: usize| {
        kernel.read(index, 4);
    };

    let turns = [
        kernel.turn(handle),
        kernel.turn(handle),
        kernel.turn(handle),
    ];

    println!(
        "event loop: turn 1 handles {}, turn 2 handles {}, turn 3 handles {}",
        list(&turns[0]),
        list(&turns[1]),
        list(&turns[2])
    );
}

3 pipes registered, none written: 0 event(s)
write to pipe 1: 1 event, for pipe 1
level-triggered, data not read, wait again: 1 event(s)
after reading the data: 0 event(s)
edge-triggered, data not read: wait 1 event(s), wait again 0 event(s)
edge-triggered, after another write: 1 event(s)
1000 pipes registered, pipe 700 written: 1 event, for pipe 700
registrations examined: 1 by the wait, 1000 by a scan
event loop: turn 1 handles [2 0], turn 2 handles [2], turn 3 handles []

Epoll inside language runtimes

Go and Rust programs rarely call epoll directly, and nearly all of their network I/O goes through it.

The Go runtime has a component called the netpoller. Every socket a Go program opens is put in non-blocking mode and registered, edge-triggered, with a single epoll instance. When a goroutine calls Read and the kernel answers EAGAIN, the runtime parks the goroutine, which costs no thread, and runs another. A wake-up from epoll makes the parked goroutine runnable again. The code reads as if it blocked, and underneath it is the event loop of this chapter.

In Rust, an async runtime such as Tokio does the same for futures. A future that cannot make progress registers its descriptor and returns Pending. The runtime’s reactor calls epoll_wait, and wakes the task that owns a descriptor when its event arrives.

Inspecting epoll on a live system

The complete programs

Listing 9.22 The complete epollfacts, in Go. code/go/epollfacts/main.go

// Command epollfacts measures epoll with pipes: which descriptors a wait
// reports, how level-triggered and edge-triggered registrations differ, how
// a closed peer is reported, and how many events a wait returns when 1000
// descriptors are registered and 1 is ready.
//
// Run it from code/go with:
//
//	go run ./epollfacts
package main

import (
	"fmt"
	"log"

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


// A pipe is the pair of descriptors of one pipe.
type pipe struct {
	reader, writer int
}

// watch creates an epoll instance and n pipes, and registers the read end of
// every pipe for readability with the given extra flags. The number of the
// pipe is stored in the registration and comes back with each event.
func watch(n int, flags uint32) (int, []pipe, error) {
	epoll, err := unix.EpollCreate1(unix.EPOLL_CLOEXEC)

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

	pipes := make([]pipe, n)

	for i := range pipes {
		var fds [2]int

		if err := unix.Pipe2(fds[:], unix.O_CLOEXEC); err != nil {
			return 0, nil, err
		}

		pipes[i] = pipe{reader: fds[0], writer: fds[1]}
		event := unix.EpollEvent{Events: unix.EPOLLIN | flags, Fd: int32(i)}

		if err := unix.EpollCtl(epoll, unix.EPOLL_CTL_ADD, fds[0], &event); err != nil {
			return 0, nil, err
		}
	}

	return epoll, pipes, nil
}

// ready returns the events that are ready now. A timeout of 0 makes the
// wait return at once.
func ready(epoll int) ([]unix.EpollEvent, error) {
	events := make([]unix.EpollEvent, 16)
	n, err := unix.EpollWait(epoll, events, 0)

	if err != nil {
		return nil, err
	}

	return events[:n], nil
}

// release closes an epoll instance and its pipes.
func release(epoll int, pipes []pipe) {
	unix.Close(epoll)

	for _, p := range pipes {
		unix.Close(p.reader)
		unix.Close(p.writer)
	}
}


// level registers 3 pipes level-triggered and waits before a write, after a
// write, again without reading, and after reading.
func level() error {
	epoll, pipes, err := watch(3, 0)

	if err != nil {
		return err
	}

	defer release(epoll, pipes)

	events, err := ready(epoll)

	if err != nil {
		return err
	}

	fmt.Printf("3 pipes registered, none written: %d event(s)\n", len(events))

	if _, err := unix.Write(pipes[1].writer, []byte("hello")); err != nil {
		return err
	}

	events, err = ready(epoll)

	if err != nil || len(events) != 1 {
		return fmt.Errorf("wait after a write: %d events, %v", len(events), err)
	}

	fmt.Printf("write to pipe 1: 1 event, for pipe %d\n", events[0].Fd)

	events, err = ready(epoll)

	if err != nil {
		return err
	}

	fmt.Printf("level-triggered, data not read, wait again: %d event(s)\n", len(events))

	if _, err := unix.Read(pipes[1].reader, make([]byte, 16)); err != nil {
		return err
	}

	events, err = ready(epoll)

	if err != nil {
		return err
	}

	fmt.Printf("after reading the data: %d event(s)\n", len(events))

	unix.Close(pipes[1].writer)
	events, err = ready(epoll)

	if err != nil || len(events) != 1 {
		return fmt.Errorf("wait after a close: %d events, %v", len(events), err)
	}

	fmt.Printf("write end of pipe 1 closed: event has EPOLLHUP %v\n", events[0].Events&unix.EPOLLHUP != 0)

	return nil
}


// edge registers a pipe edge-triggered and waits 2 times after one write,
// and once more after a second write.
func edge() error {
	epoll, pipes, err := watch(1, unix.EPOLLET)

	if err != nil {
		return err
	}

	defer release(epoll, pipes)

	var counts [3]int

	for i := range counts {
		if i != 1 {
			if _, err := unix.Write(pipes[0].writer, []byte("hello")); err != nil {
				return err
			}
		}

		events, err := ready(epoll)

		if err != nil {
			return err
		}

		counts[i] = len(events)
	}

	fmt.Printf("edge-triggered, data not read: wait %d event(s), wait again %d event(s)\n", counts[0], counts[1])
	fmt.Printf("edge-triggered, after another write: %d event(s)\n", counts[2])

	return nil
}


// many registers 1000 pipes and writes to 1 of them.
func many() error {
	epoll, pipes, err := watch(1000, 0)

	if err != nil {
		return err
	}

	defer release(epoll, pipes)

	if _, err := unix.Write(pipes[700].writer, []byte("hello")); err != nil {
		return err
	}

	events, err := ready(epoll)

	if err != nil || len(events) != 1 {
		return fmt.Errorf("wait with 1000 pipes: %d events, %v", len(events), err)
	}

	fmt.Printf("1000 pipes registered, pipe 700 written: 1 event, for pipe %d\n", events[0].Fd)

	return nil
}


func main() {
	for _, experiment := range []func() error{level, edge, many} {
		if err := experiment(); err != nil {
			log.Fatal(err)
		}
	}
}

Listing 9.23 The complete epollfacts, in Rust. code/rust/epollfacts/src/main.rs

//! Measures epoll with pipes: which descriptors a wait reports, how
//! level-triggered and edge-triggered registrations differ, how a closed peer
//! is reported, and how many events a wait returns when 1000 descriptors are
//! registered and 1 is ready.
//!
//! Run it from code/rust with:
//!
//! ```text
//! cargo run -p epollfacts
//! ```

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

use nix::fcntl::OFlag;
use nix::sys::epoll::{Epoll, EpollCreateFlags, EpollEvent, EpollFlags, EpollTimeout};
use nix::unistd::{pipe2, read, write};

/// The pair of descriptors of one pipe. The write end is an `Option` so that
/// an experiment can close it.
struct Pipe {
    reader: OwnedFd,
    writer: Option<OwnedFd>,
}

/// Creates an epoll instance and `n` pipes, and registers the read end of
/// every pipe for readability with the given extra flags. The number of the
/// pipe is stored in the registration and comes back with each event.
fn watch(n: usize, flags: EpollFlags) -> nix::Result<(Epoll, Vec<Pipe>)> {
    let epoll = Epoll::new(EpollCreateFlags::EPOLL_CLOEXEC)?;
    let mut pipes = Vec::with_capacity(n);

    for i in 0..n {
        let (reader, writer) = pipe2(OFlag::O_CLOEXEC)?;

        epoll.add(
            &reader,
            EpollEvent::new(EpollFlags::EPOLLIN | flags, i as u64),
        )?;

        pipes.push(Pipe {
            reader,
            writer: Some(writer),
        });
    }

    Ok((epoll, pipes))
}

/// Returns the events that are ready now. A timeout of 0 makes the wait
/// return at once.
fn ready(epoll: &Epoll) -> nix::Result<Vec<EpollEvent>> {
    let mut events = [EpollEvent::empty(); 16];
    let n = epoll.wait(&mut events, EpollTimeout::ZERO)?;

    Ok(events[..n].to_vec())
}

/// Writes "hello" to a pipe whose write end is open.
fn send(pipe: &Pipe) -> nix::Result<()> {
    if let Some(writer) = &pipe.writer {
        write(writer, b"hello")?;
    }

    Ok(())
}

/// Registers 3 pipes level-triggered and waits before a write, after a write,
/// again without reading, and after reading.
fn level() -> nix::Result<()> {
    let (epoll, mut pipes) = watch(3, EpollFlags::empty())?;

    println!(
        "3 pipes registered, none written: {} event(s)",
        ready(&epoll)?.len()
    );

    send(&pipes[1])?;

    if let [event] = ready(&epoll)?.as_slice() {
        println!("write to pipe 1: 1 event, for pipe {}", event.data());
    }

    println!(
        "level-triggered, data not read, wait again: {} event(s)",
        ready(&epoll)?.len()
    );

    read(&pipes[1].reader, &mut [0u8; 16])?;
    println!("after reading the data: {} event(s)", ready(&epoll)?.len());

    pipes[1].writer = None;

    if let [event] = ready(&epoll)?.as_slice() {
        println!(
            "write end of pipe 1 closed: event has EPOLLHUP {}",
            event.events().contains(EpollFlags::EPOLLHUP)
        );
    }

    Ok(())
}

/// Registers a pipe edge-triggered and waits 2 times after one write, and
/// once more after a second write.
fn edge() -> nix::Result<()> {
    let (epoll, pipes) = watch(1, EpollFlags::EPOLLET)?;
    let mut counts = [0; 3];

    for (i, count) in counts.iter_mut().enumerate() {
        if i != 1 {
            send(&pipes[0])?;
        }

        *count = ready(&epoll)?.len();
    }

    println!(
        "edge-triggered, data not read: wait {} event(s), wait again {} event(s)",
        counts[0], counts[1]
    );
    println!(
        "edge-triggered, after another write: {} event(s)",
        counts[2]
    );

    Ok(())
}

/// Registers 1000 pipes and writes to 1 of them.
fn many() -> nix::Result<()> {
    let (epoll, pipes) = watch(1000, EpollFlags::empty())?;

    send(&pipes[700])?;

    if let [event] = ready(&epoll)?.as_slice() {
        println!(
            "1000 pipes registered, pipe 700 written: 1 event, for pipe {}",
            event.data()
        );
    }

    Ok(())
}

fn main() -> nix::Result<()> {
    level()?;
    edge()?;
    many()
}

Listing 9.24 The complete eventloop, in Go. code/go/eventloop/main.go

// Command eventloop models epoll: an interest list of registered pipes, a
// ready list that a write fills, and a wait that examines only the ready
// list. It repeats the measurements of epollfacts on the model and then runs
// a small event loop on it.
//
// Run it from code/go with:
//
//	go run ./eventloop
package main

import "fmt"


// A pipe is the read end of a pipe as epoll sees it: the bytes waiting to be
// read, and its registration. queued is set while the pipe is on the ready
// list.
type pipe struct {
	buffered   int
	registered bool
	edge       bool
	queued     bool
}

// A kernel holds the pipes and one epoll instance. The registered pipes are
// the interest list. ready is the ready list, in the order in which the
// pipes became ready. examined counts the registrations that waits and scans
// have looked at.
type kernel struct {
	pipes    []pipe
	ready    []int
	examined int
}

func newKernel(pipes int) *kernel {
	return &kernel{pipes: make([]pipe, pipes)}
}


// register adds a pipe to the interest list. A pipe that is already
// readable goes on the ready list at once.
func (k *kernel) register(index int, edge bool) {
	p := &k.pipes[index]
	p.registered, p.edge = true, edge

	if p.buffered > 0 {
		k.enqueue(index)
	}
}

// enqueue puts a pipe on the ready list unless it is already there.
func (k *kernel) enqueue(index int) {
	if p := &k.pipes[index]; !p.queued {
		p.queued = true
		k.ready = append(k.ready, index)
	}
}

// write adds bytes to a pipe. This is the moment the pipe becomes ready, and
// the writer puts it on the ready list.
func (k *kernel) write(index, n int) {
	k.pipes[index].buffered += n

	if k.pipes[index].registered {
		k.enqueue(index)
	}
}

// read removes up to n bytes from a pipe and returns the number removed.
func (k *kernel) read(index, n int) int {
	n = min(n, k.pipes[index].buffered)
	k.pipes[index].buffered -= n

	return n
}


// wait empties the ready list and returns the pipes on it that are still
// readable. A level-triggered pipe that is reported goes back on the list,
// so the next wait checks it again. An edge-triggered pipe does not, and is
// reported again only after another write.
func (k *kernel) wait() []int {
	queued := k.ready
	k.ready = nil

	var events []int

	for _, index := range queued {
		p := &k.pipes[index]
		p.queued = false
		k.examined++

		if p.buffered == 0 {
			continue
		}

		events = append(events, index)

		if !p.edge {
			k.enqueue(index)
		}
	}

	return events
}

// scan is the method of select and poll: it examines every registered pipe
// on every call.
func (k *kernel) scan() []int {
	var events []int

	for index, p := range k.pipes {
		if !p.registered {
			continue
		}

		k.examined++

		if p.buffered > 0 {
			events = append(events, index)
		}
	}

	return events
}


// level repeats the level-triggered measurement of epollfacts.
func level() {
	k := newKernel(3)

	for index := range k.pipes {
		k.register(index, false)
	}

	fmt.Printf("3 pipes registered, none written: %d event(s)\n", len(k.wait()))

	k.write(1, 5)

	if events := k.wait(); len(events) == 1 {
		fmt.Printf("write to pipe 1: 1 event, for pipe %d\n", events[0])
	}

	fmt.Printf("level-triggered, data not read, wait again: %d event(s)\n", len(k.wait()))

	k.read(1, 16)
	fmt.Printf("after reading the data: %d event(s)\n", len(k.wait()))
}

// edge repeats the edge-triggered measurement.
func edge() {
	k := newKernel(1)
	k.register(0, true)

	k.write(0, 5)
	first, second := len(k.wait()), len(k.wait())
	fmt.Printf("edge-triggered, data not read: wait %d event(s), wait again %d event(s)\n", first, second)

	k.write(0, 5)
	fmt.Printf("edge-triggered, after another write: %d event(s)\n", len(k.wait()))
}

// many compares the work of a wait with the work of a scan when 1000 pipes
// are registered and 1 is ready.
func many() {
	k := newKernel(1000)

	for index := range k.pipes {
		k.register(index, false)
	}

	k.write(700, 5)

	if events := k.wait(); len(events) == 1 {
		fmt.Printf("1000 pipes registered, pipe 700 written: 1 event, for pipe %d\n", events[0])
	}

	byWait := k.examined
	k.scan()
	fmt.Printf("registrations examined: %d by the wait, %d by a scan\n", byWait, k.examined-byWait)
}


// turn is one iteration of an event loop: wait for ready pipes, then run the
// handler for each. It returns the pipes it handled.
func (k *kernel) turn(handle func(k *kernel, index int)) []int {
	events := k.wait()

	for _, index := range events {
		handle(k, index)
	}

	return events
}

// loop registers 4 pipes, writes to 2 of them, and runs an event loop whose
// handler reads at most 4 bytes for each event.
func loop() {
	k := newKernel(4)

	for index := range k.pipes {
		k.register(index, false)
	}

	k.write(2, 5)
	k.write(0, 3)

	handle := func(k *kernel, index int) { k.read(index, 4) }

	fmt.Printf("event loop: turn 1 handles %v, turn 2 handles %v, turn 3 handles %v\n",
		k.turn(handle), k.turn(handle), k.turn(handle))
}


func main() {
	level()
	edge()
	many()
	loop()
}

Listing 9.25 The complete eventloop, in Rust. code/rust/eventloop/src/main.rs

//! Models epoll: an interest list of registered pipes, a ready list that a
//! write fills, and a wait that examines only the ready list. It repeats the
//! measurements of epollfacts on the model and then runs a small event loop
//! on it.
//!
//! Run it from code/rust with:
//!
//! ```text
//! cargo run -p eventloop
//! ```

/// The read end of a pipe as epoll sees it: the bytes waiting to be read, and
/// its registration. `queued` is set while the pipe is on the ready list.
#[derive(Clone, Default)]
struct Pipe {
    buffered: usize,
    registered: bool,
    edge: bool,
    queued: bool,
}

/// Holds the pipes and one epoll instance. The registered pipes are the
/// interest list. `ready` is the ready list, in the order in which the pipes
/// became ready. `examined` counts the registrations that waits and scans
/// have looked at.
struct Kernel {
    pipes: Vec<Pipe>,
    ready: Vec<usize>,
    examined: usize,
}

impl Kernel {
    fn new(pipes: usize) -> Kernel {
        Kernel {
            pipes: vec![Pipe::default(); pipes],
            ready: Vec::new(),
            examined: 0,
        }
    }
}

impl Kernel {
    /// Adds a pipe to the interest list. A pipe that is already readable goes
    /// on the ready list at once.
    fn register(&mut self, index: usize, edge: bool) {
        let pipe = &mut self.pipes[index];

        pipe.registered = true;
        pipe.edge = edge;

        if pipe.buffered > 0 {
            self.enqueue(index);
        }
    }

    /// Puts a pipe on the ready list unless it is already there.
    fn enqueue(&mut self, index: usize) {
        if !self.pipes[index].queued {
            self.pipes[index].queued = true;
            self.ready.push(index);
        }
    }

    /// Adds bytes to a pipe. This is the moment the pipe becomes ready, and
    /// the writer puts it on the ready list.
    fn write(&mut self, index: usize, n: usize) {
        self.pipes[index].buffered += n;

        if self.pipes[index].registered {
            self.enqueue(index);
        }
    }

    /// Removes up to `n` bytes from a pipe and returns the number removed.
    fn read(&mut self, index: usize, n: usize) -> usize {
        let n = n.min(self.pipes[index].buffered);
        self.pipes[index].buffered -= n;

        n
    }
}

impl Kernel {
    /// Empties the ready list and returns the pipes on it that are still
    /// readable. A level-triggered pipe that is reported goes back on the
    /// list, so the next wait checks it again. An edge-triggered pipe does
    /// not, and is reported again only after another write.
    fn wait(&mut self) -> Vec<usize> {
        let queued = std::mem::take(&mut self.ready);
        let mut events = Vec::new();

        for index in queued {
            self.pipes[index].queued = false;
            self.examined += 1;

            if self.pipes[index].buffered == 0 {
                continue;
            }

            events.push(index);

            if !self.pipes[index].edge {
                self.enqueue(index);
            }
        }

        events
    }

    /// The method of select and poll: it examines every registered pipe on
    /// every call.
    fn scan(&mut self) -> Vec<usize> {
        let mut events = Vec::new();

        for (index, pipe) in self.pipes.iter().enumerate() {
            if !pipe.registered {
                continue;
            }

            self.examined += 1;

            if pipe.buffered > 0 {
                events.push(index);
            }
        }

        events
    }
}

/// Repeats the level-triggered measurement of epollfacts.
fn level() {
    let mut kernel = Kernel::new(3);

    for index in 0..3 {
        kernel.register(index, false);
    }

    println!(
        "3 pipes registered, none written: {} event(s)",
        kernel.wait().len()
    );

    kernel.write(1, 5);

    if let [index] = kernel.wait().as_slice() {
        println!("write to pipe 1: 1 event, for pipe {index}");
    }

    println!(
        "level-triggered, data not read, wait again: {} event(s)",
        kernel.wait().len()
    );

    kernel.read(1, 16);
    println!("after reading the data: {} event(s)", kernel.wait().len());
}

/// Repeats the edge-triggered measurement.
fn edge() {
    let mut kernel = Kernel::new(1);
    kernel.register(0, true);

    kernel.write(0, 5);
    let (first, second) = (kernel.wait().len(), kernel.wait().len());
    println!("edge-triggered, data not read: wait {first} event(s), wait again {second} event(s)");

    kernel.write(0, 5);
    println!(
        "edge-triggered, after another write: {} event(s)",
        kernel.wait().len()
    );
}

/// Compares the work of a wait with the work of a scan when 1000 pipes are
/// registered and 1 is ready.
fn many() {
    let mut kernel = Kernel::new(1000);

    for index in 0..1000 {
        kernel.register(index, false);
    }

    kernel.write(700, 5);

    if let [index] = kernel.wait().as_slice() {
        println!("1000 pipes registered, pipe 700 written: 1 event, for pipe {index}");
    }

    let by_wait = kernel.examined;
    kernel.scan();
    println!(
        "registrations examined: {by_wait} by the wait, {} by a scan",
        kernel.examined - by_wait
    );
}

impl Kernel {
    /// One iteration of an event loop: wait for ready pipes, then run the
    /// handler for each. Returns the pipes it handled.
    fn turn(&mut self, handle: fn(&mut Kernel, usize)) -> Vec<usize> {
        let events = self.wait();

        for &index in &events {
            handle(self, index);
        }

        events
    }
}

/// Formats pipe numbers the way Go prints a slice.
fn list(pipes: &[usize]) -> String {
    let words: Vec<String> = pipes.iter().map(usize::to_string).collect();

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

/// Registers 4 pipes, writes to 2 of them, and runs an event loop whose
/// handler reads at most 4 bytes for each event.
fn event_loop() {
    let mut kernel = Kernel::new(4);

    for index in 0..4 {
        kernel.register(index, false);
    }

    kernel.write(2, 5);
    kernel.write(0, 3);

    let handle = |kernel: &mut Kernel, index: usize| {
        kernel.read(index, 4);
    };

    let turns = [
        kernel.turn(handle),
        kernel.turn(handle),
        kernel.turn(handle),
    ];

    println!(
        "event loop: turn 1 handles {}, turn 2 handles {}, turn 3 handles {}",
        list(&turns[0]),
        list(&turns[1]),
        list(&turns[2])
    );
}

fn main() {
    level();
    edge();
    many();
    event_loop();
}

References

• epoll(7), epoll_ctl(2) and epoll_wait(2), Linux manual pages: https://man7.org/linux/man-pages/man7/epoll.7.html
• M. Kerrisk, The Linux Programming Interface, No Starch Press, 2010, chapter 63.
• fs/eventpoll.c, Linux 5.15: https://elixir.bootlin.com/linux/v5.15/source/fs/eventpoll.c