Python Basics: Wait for Multiple Local Streams with selectors

When several streams can become readable independently, a selector lets one loop wait for them together. This fixture registers two local socket pairs and collects each stream until EOF, without assuming notification order or receive boundaries.

Prepare a scratch directory

Tested on Linux with CPython 3.12.14. You need Bash, Python, a text editor, and the standard mktemp, rm, and rmdir utilities. Run without -O so the assertions remain enabled. No external packages are required.

Keep the same terminal open. Stop if setup fails, before saving the source:

start_dir="$PWD"
work_dir="$(mktemp -d)" || exit 1
cd -- "$work_dir" || exit 1

Run the example

Save this as demonstrate-selectors.py:

from contextlib import ExitStack
import selectors
import socket
import time

results = {}
with ExitStack() as stack:
    selector = stack.enter_context(selectors.DefaultSelector())
    for name, payload in [("alpha", b"one"), ("beta", b"two")]:
        writer, reader = socket.socketpair()
        stack.enter_context(writer)
        stack.enter_context(reader)
        writer.settimeout(2)
        writer.sendall(payload)
        writer.shutdown(socket.SHUT_WR)
        reader.setblocking(False)
        results[name] = bytearray()
        selector.register(reader, selectors.EVENT_READ, data=name)

    deadline = time.monotonic() + 2
    while selector.get_map():
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise TimeoutError("local streams did not finish")
        for key, mask in selector.select(remaining):
            try:
                chunk = key.fileobj.recv(2)
            except BlockingIOError:
                continue
            if chunk:
                results[key.data].extend(chunk)
            else:
                selector.unregister(key.fileobj)

assert results == {"alpha": b"one", "beta": b"two"}
for name in sorted(results):
    print(f"{name}: {results[name].decode('ascii')}")
print("All streams reached EOF")

Run it:

python3 -B demonstrate-selectors.py

Observed output:

alpha: one
beta: two
All streams reached EOF

Read the result

The registration data associates each socket with its buffer. Readability can mean data is available or that the peer has reached EOF. Unregistering EOF streams prevents a loop that repeatedly wakes for a finished socket. ExitStack then closes the sockets and selector.

Sorting the names controls presentation only. The program does not depend on the order in which the operating system returns ready descriptors. The monotonic deadline limits the read loop even if a stream never finishes.

Apply it carefully

A readiness notification is not a full application message. Keep sockets nonblocking and tolerate BlockingIOError. This is a tiny bounded fixture with already queued data, not a production server: real applications also need connection limits, bounded buffers, write queues, and message framing. File-object support differs by platform; this example was run with local sockets on Linux.

The official selectors reference describes the relevant APIs and their limits.

Clean up

Remove the saved source, return to the starting directory, and remove the empty practice directory:

rm -- demonstrate-selectors.py
cd -- "$start_dir" || exit 1
rmdir -- "$work_dir"
unset start_dir work_dir

If the directory is not empty, inspect its remaining contents before removing anything else.

Acknowledgements

This post was created with assistance from AI.