Lab 17: The Multiplexer Server (Milestone 11)

Background

Everything so far has been one process. This lab splits it: a server that daemonizes, owns every PTY master and every Terminal, and listens on a Unix socket; and a client that owns nothing durable and can be killed at any moment with no consequence.

That split is the multiplexer. Get it right and detach/attach in Lab 18 is almost free.

Why This Lab Matters

The kill -9 client test is the architectural acceptance criterion for the entire section.
Daemonization done wrong reintroduces exactly the SIGHUP bug the multiplexer exists to prevent.

Prerequisites

Lab 16 complete.
Why a Server Process and The Protocol read.

Predict First

Your server forgets setsid(). What happens when you close the launching terminal?
A client vanishes mid-write. What signal does the server get, and what is its default disposition?
Two servers race to bind the same socket path. What does the second one see?
A client stops reading but stays connected. What happens to the server's memory?

Step 1: Daemonize

#![allow(unused)]
fn main() {
// crates/terminal-mux/src/daemon.rs

pub fn daemonize() -> io::Result<()> {
    // 1. Fork; the parent exits. The child is reparented to init and, crucially,
    //    is NOT a process group leader — a precondition for setsid().
    match unsafe { libc::fork() } {
        -1 => return Err(io::Error::last_os_error()),
        0 => {}
        _ => unsafe { libc::_exit(0) },
    }

    // 2. New session. DROPS the inherited controlling terminal. Skip this and
    //    closing the launching terminal SIGHUPs the server — the very bug the
    //    multiplexer exists to prevent.
    if unsafe { libc::setsid() } < 0 { return Err(io::Error::last_os_error()); }

    // 3. Fork AGAIN. A session leader can acquire a controlling terminal by
    //    opening a tty; a non-leader cannot. This second fork makes that
    //    accident impossible.
    match unsafe { libc::fork() } {
        -1 => return Err(io::Error::last_os_error()),
        0 => {}
        _ => unsafe { libc::_exit(0) },
    }

    unsafe { libc::chdir(c"/".as_ptr()) };
    unsafe { libc::umask(0o077) };            // sockets and logs are owner-only

    // 4. Redirect 0/1/2 to /dev/null. A stray println! would otherwise scribble
    //    on whatever terminal launched us, and a write to a CLOSED terminal
    //    yields EIO or SIGPIPE.
    let null = unsafe { libc::open(c"/dev/null".as_ptr(), libc::O_RDWR) };
    for fd in 0..=2 { unsafe { libc::dup2(null, fd) }; }
    if null > 2 { unsafe { libc::close(null) }; }

    unsafe { libc::signal(libc::SIGHUP, libc::SIG_IGN) };
    // MANDATORY: SIGPIPE's default disposition TERMINATES the process. A client
    // that disappears mid-write would kill the server and every session with it.
    unsafe { libc::signal(libc::SIGPIPE, libc::SIG_IGN) };
    Ok(())
}
}

Step 2: The Socket

#![allow(unused)]
fn main() {
pub fn socket_dir() -> io::Result<PathBuf> {
    let uid = unsafe { libc::getuid() };
    let dir = std::env::var_os("XDG_RUNTIME_DIR")
        .map(PathBuf::from)
        .unwrap_or_else(|| PathBuf::from(format!("/tmp/mini-mux-{uid}")));
    std::fs::create_dir_all(&dir)?;
    // 0700, and VERIFY it — an attacker may have pre-created the directory with
    // looser permissions, and a world-writable socket is a shell on your account.
    std::fs::set_permissions(&dir, Permissions::from_mode(0o700))?;
    let meta = std::fs::metadata(&dir)?;
    if meta.uid() != uid {
        return Err(io::Error::new(io::ErrorKind::PermissionDenied,
                                  "socket directory owned by another user"));
    }
    if meta.permissions().mode() & 0o077 != 0 {
        return Err(io::Error::new(io::ErrorKind::PermissionDenied,
                                  "socket directory is group/world accessible"));
    }
    Ok(dir)
}

pub fn bind(path: &Path) -> io::Result<UnixListener> {
    match UnixListener::bind(path) {
        Ok(l) => Ok(l),
        Err(e) if e.kind() == io::ErrorKind::AddrInUse => {
            // The file exists. Alive server, or a corpse from a crash?
            match UnixStream::connect(path) {
                Ok(_) => Err(io::Error::new(io::ErrorKind::AddrInUse,
                                            "a server is already running")),
                Err(_) => { std::fs::remove_file(path)?; UnixListener::bind(path) }
            }
        }
        Err(e) => Err(e),
    }
}
}

Optionally verify the peer:

#![allow(unused)]
fn main() {
/// Defence in depth: even with correct permissions, verify the connecting UID.
#[cfg(target_os = "linux")]
fn peer_uid(stream: &UnixStream) -> io::Result<u32> {
    let mut cred: libc::ucred = unsafe { std::mem::zeroed() };
    let mut len = std::mem::size_of::<libc::ucred>() as libc::socklen_t;
    let r = unsafe {
        libc::getsockopt(stream.as_raw_fd(), libc::SOL_SOCKET, libc::SO_PEERCRED,
                         &mut cred as *mut _ as *mut _, &mut len)
    };
    if r != 0 { return Err(io::Error::last_os_error()); }
    Ok(cred.uid)
}
// macOS: LOCAL_PEERCRED / getpeereid(2).
}

Step 3: The Server Event Loop

#![allow(unused)]
fn main() {
pub struct Server {
    listener: UnixListener,
    clients: HashMap<ClientId, ClientConn>,
    sessions: SessionManager,      // owns every Pty and every Terminal
    sig: SignalPipe,
}

impl Server {
    pub fn run(&mut self) -> io::Result<()> {
        loop {
            let mut fds = vec![
                pollfd(self.listener.as_raw_fd(), libc::POLLIN),
                pollfd(self.sig.read_fd, libc::POLLIN),
            ];
            // Every client socket. POLLOUT only when there is queued output —
            // registering it permanently is the classic 100%-CPU bug.
            let client_ids: Vec<ClientId> = self.clients.keys().copied().collect();
            for id in &client_ids {
                let c = &self.clients[id];
                let mut ev = libc::POLLIN;
                if !c.out.is_empty() { ev |= libc::POLLOUT; }
                fds.push(pollfd(c.stream.as_raw_fd(), ev));
            }
            // Every pane master. ALWAYS read, even panes no client is watching.
            let pane_ids: Vec<PaneId> = self.sessions.all_pane_ids();
            for id in &pane_ids {
                fds.push(pollfd(self.sessions.pane(*id).pty.master_fd(), libc::POLLIN));
            }

            let n = unsafe { libc::poll(fds.as_mut_ptr(), fds.len() as _, 16) };
            if n < 0 && last_errno() != libc::EINTR { return Err(io::Error::last_os_error()); }

            self.accept_new_clients(&fds[0]);
            self.handle_signals(&fds[1]);
            self.handle_client_io(&client_ids, &fds[2..]);
            self.handle_pane_io(&pane_ids, &fds[2 + client_ids.len()..]);

            // Rate-limited broadcast: coalesce many parse events into one update
            // per client per frame. Sending an update per read() would flood the
            // socket under `yes`.
            self.broadcast_updates_if_due();
            self.reap_exited_panes();
        }
    }

    fn handle_pane_io(&mut self, ids: &[PaneId], fds: &[libc::pollfd]) {
        for (id, pfd) in ids.iter().zip(fds) {
            if pfd.revents & (libc::POLLIN | libc::POLLHUP) == 0 { continue; }
            let pane = self.sessions.pane_mut(*id);
            loop {
                match pane.pty.read(&mut self.buf) {
                    Ok(0) => { pane.exited = Some(0); break; }
                    Ok(n) => {
                        // Parse into the pane's OWN headless Terminal, ALWAYS —
                        // whether or not any client is watching. This is what
                        // makes detach work and keeps memory constant.
                        pane.terminal.advance(&self.buf[..n]);
                        let replies = pane.terminal.take_replies();
                        if !replies.is_empty() { pane.pty.queue_write(&replies); }
                        // Some state changes must reach clients out of band.
                        if pane.terminal.take_title_changed() {
                            self.pending_events.push(Event::PaneTitle { /* ... */ });
                        }
                        if let Some(cb) = pane.terminal.take_clipboard_write() {
                            // The server has NO display. Only a client can set
                            // the system clipboard.
                            self.pending_events.push(Event::SetClipboard { /* ... */ });
                        }
                    }
                    Err(e) if e.kind() == ErrorKind::WouldBlock => break,
                    Err(e) if e.kind() == ErrorKind::Interrupted => continue,
                    Err(e) if e.raw_os_error() == Some(libc::EIO) => {
                        pane.exited = Some(0); break;
                    }
                    Err(_) => { pane.exited = Some(-1); break; }
                }
            }
        }
    }
}
}

Step 4: The Client

#![allow(unused)]
fn main() {
pub struct Client {
    stream: UnixStream,
    decoder: FrameDecoder,
    /// The user's own terminal, in raw mode. Restored on EVERY exit path.
    _raw: RawMode,
    state: InputState,          // the prefix state machine
    pane_modes: ModeFlags,      // mirrored from the server, for input encoding
    screen: ScreenSnapshot,
}

impl Client {
    pub fn attach(socket: &Path, session: Option<&str>) -> io::Result<Self> {
        let stream = UnixStream::connect(socket)?;
        let raw = RawMode::enable(0)?;
        let mut c = Client { /* ... */ };

        // Version first. A mismatch must be diagnosed HERE, not three messages
        // later as mysterious behavior.
        c.send(Request::Hello {
            protocol_version: PROTOCOL_VERSION,
            client_name: "mini-mux".into(),
            term: std::env::var("TERM").unwrap_or_default(),
            size: current_size(0)?,
            capabilities: detect_capabilities(),
        })?;
        match c.recv()? {
            Event::Welcome { .. } => {}
            Event::ProtocolMismatch { server_version, client_version } => {
                return Err(io::Error::other(
                    format!("protocol mismatch: server {server_version}, client {client_version}")));
            }
            other => return Err(io::Error::other(format!("unexpected: {other:?}"))),
        }
        c.send(Request::Attach { session: session.map(String::from) })?;

        // Enter the alternate screen so the user's shell scrollback survives,
        // and enable the modes the client itself needs.
        c.write_terminal(b"\x1b[?1049h\x1b[?1000h\x1b[?1006h\x1b[?2004h")?;
        Ok(c)
    }

    pub fn run(&mut self) -> io::Result<()> {
        let sig = SignalPipe::install(&[libc::SIGWINCH, libc::SIGTERM, libc::SIGINT])?;
        loop {
            let mut fds = [
                pollfd(0, libc::POLLIN),
                pollfd(self.stream.as_raw_fd(),
                       libc::POLLIN | if self.has_pending() { libc::POLLOUT } else { 0 }),
                pollfd(sig.read_fd, libc::POLLIN),
            ];
            unsafe { libc::poll(fds.as_mut_ptr(), 3, -1) };

            if fds[0].revents & libc::POLLIN != 0 {
                let bytes = read_stdin()?;
                // The PREFIX is parsed here, client-side: zero latency, and it
                // works even when the server is busy.
                for chunk in self.split_on_prefix(&bytes) {
                    match chunk {
                        Chunk::Command(cmd) => self.run_command(cmd)?,
                        Chunk::Forward(b) => self.send(Request::Input { bytes: b.into() })?,
                    }
                }
            }
            if fds[1].revents & libc::POLLIN != 0 { self.handle_server_events()?; }
            if fds[1].revents & libc::POLLOUT != 0 { self.flush_pending()?; }
            if fds[2].revents & libc::POLLIN != 0 {
                for s in sig.drain() {
                    match s {
                        libc::SIGWINCH => {
                            let size = current_size(0)?;
                            self.send(Request::ClientResize { size })?;
                        }
                        libc::SIGTERM | libc::SIGINT => return self.detach(),
                        _ => {}
                    }
                }
            }
        }
    }
}

impl Drop for Client {
    fn drop(&mut self) {
        // Leave the alternate screen and reset every mode we set. Leaving mouse
        // reporting on makes the user's shell emit garbage on every click.
        let _ = self.write_terminal(
            b"\x1b[?1000l\x1b[?1002l\x1b[?1003l\x1b[?1006l\x1b[?2004l\x1b[?1049l\x1b[0m\x1b[?25h");
        // RawMode's own Drop restores termios.
    }
}
}

Expected Output

$ mini-mux server
$ mini-mux ls
work: 1 window (created Wed Jan  1 14:30:00 2025)

$ mini-mux attach
# ...a full-screen session appears; split, run things...
# <prefix> d  to detach

$ mini-mux ls
work: 2 windows (created Wed Jan  1 14:30:00 2025)

# THE test:
$ mini-mux attach &
$ sleep 2
$ pkill -9 -f 'mini-mux attach'
$ ps -o pid,ppid,sid,tty,comm | grep -E 'vim|top|sleep'
  8123  8100  8100 pts/9  vim         ← STILL RUNNING
$ mini-mux attach
# ...vim is exactly where you left it

Debugging Steps

The server dies when you close the launching terminal

setsid() missing or its return value ignored. Verify with ps -o sid,tty -p <server>: the TTY must be ?.

The server dies when a client disappears

SIGPIPE not ignored. Its default disposition terminates the process.

"Address already in use" after a crash

Stale socket file. Implement the connect-then-unlink probe.

The client hangs on connect

The server is not accepting, or the Hello/Welcome handshake deadlocked because both sides are waiting to read.

The server's memory grows without bound

A slow client's output buffer is unbounded, or you are buffering raw pane bytes instead of parsing.

100% CPU when idle

POLLOUT registered permanently on client sockets, or a zero poll timeout with nothing to do.

The client leaves the terminal broken after `kill -9`

Nothing can help — SIGKILL runs no handlers. That is exactly why the server must be robust to it, and why mini-mux attach should offer a reset subcommand.

Panes stop updating when no client is attached

You made pane reading conditional on having clients. It must not be.

Experiment

CLAIM. The client owns nothing durable: kill -9 on it has no effect beyond a dropped connection.

METHOD.

mini-mux server
mini-mux attach &
CLIENT=$!
sleep 1
# Start something identifiable in a pane, then find the pieces:
SERVER=$(pgrep -f 'mini-mux server')

echo "=== who holds what ==="
lsof -p $SERVER 2>/dev/null | grep -E 'ptmx|pts|sock'
#   The SERVER holds ptmx (masters) AND the listening socket.
lsof -p $CLIENT 2>/dev/null | grep -E 'ptmx|pts|sock'
#   The CLIENT holds a socket and its OWN tty. NO ptmx.

echo "=== sessions ==="
ps -o pid,ppid,pgid,sid,tty,comm -p $SERVER,$CLIENT
#   The server's TTY must be "?" — no controlling terminal.

echo "=== kill the client HARD ==="
kill -9 $CLIENT
sleep 1
pgrep -f 'mini-mux server' && echo "server alive"
ps -o pid,comm -p $(pgrep -n sleep) && echo "pane child alive"

PREDICTION. Before the kill -9: what does the server observe on that socket? Does any signal reach any pane child? What is the pane child's PPID?

Test

#![allow(unused)]
fn main() {
#[test]
fn killing_the_client_leaves_everything_running() {
    // THE architectural acceptance test.
    let server = TestServer::spawn().unwrap();
    let mut client = TestClient::attach(&server.socket()).unwrap();
    let pane = client.new_pane("sleep 300").unwrap();
    let child = server.pane_child_pid(pane);
    client.kill_hard();
    std::thread::sleep(Duration::from_millis(500));
    assert!(server.is_running());
    assert!(process_exists(child));
    assert_eq!(server.client_count(), 0);
}

#[test]
fn server_has_no_controlling_terminal_and_is_session_leader() {
    let server = TestServer::spawn().unwrap();
    assert_eq!(controlling_terminal_of(server.pid()), None);
    assert_eq!(session_id_of(server.pid()), server.pid());
}

#[test]
fn server_survives_sigpipe() {
    let server = TestServer::spawn().unwrap();
    let client = TestClient::attach(&server.socket()).unwrap();
    server.new_pane("yes");                    // flood
    client.close_socket_abruptly();
    std::thread::sleep(Duration::from_millis(500));
    assert!(server.is_running());
}

#[test]
fn stale_socket_is_reclaimed() {
    let dir = tempdir();
    let path = dir.path().join("sock");
    { let _l = UnixListener::bind(&path).unwrap(); }   // bound and dropped
    assert!(path.exists());
    let listener = bind(&path);
    assert!(listener.is_ok(), "a corpse socket must be reclaimed");
}

#[test]
fn a_second_server_refuses_to_start() {
    let s1 = TestServer::spawn().unwrap();
    assert!(TestServer::spawn_at(&s1.socket()).is_err());
}

#[test]
fn socket_directory_is_owner_only() {
    let server = TestServer::spawn().unwrap();
    let mode = std::fs::metadata(server.socket().parent().unwrap())
        .unwrap().permissions().mode() & 0o777;
    assert_eq!(mode, 0o700);
}

#[test]
fn panes_keep_updating_with_zero_clients() {
    let server = TestServer::spawn().unwrap();
    let pane = server.new_pane_direct("bash -c 'for i in $(seq 1 3000); do echo $i; done; sleep 30'");
    assert_eq!(server.client_count(), 0);
    std::thread::sleep(Duration::from_secs(2));
    assert!(server.pane_snapshot(pane).contains("3000"));
}

#[test]
fn slow_client_is_bounded_not_unbounded() {
    let server = TestServer::spawn().unwrap();
    let client = TestClient::attach_and_stop_reading(&server.socket()).unwrap();
    server.new_pane("yes");
    std::thread::sleep(Duration::from_secs(3));
    assert!(server.client_buffer_bytes(client.id()) <= MAX_CLIENT_BUFFER);
    assert!(server.is_running());
}
}

Challenge Extensions

mini-mux ls, new-session -d, kill-session as standalone client commands that connect, issue one request, print, and exit — proving the protocol is usable non-interactively.
SO_PEERCRED verification with a test that a different UID is refused.
Server logging to a file (it has no terminal), with rotation and a --log-protocol mode.
Graceful shutdown: SIGTERM to the server sends SIGHUP to every pane, waits, then SIGKILLs stragglers.
Server-side session persistence: serialize the layout tree on change so a crashed server can restore its structure (though not its processes) on restart.
A benchmark: 20 panes each running yes, one client attached. Report server CPU, socket bytes/sec, and client CPU.

Deliverables

A correctly daemonized server: double fork, setsid, redirected fds, SIGPIPE ignored.
A socket in a per-user directory, mode 0700, with ownership verified and stale sockets reclaimed.
A framed protocol with the Hello/Welcome version handshake.
A client that puts its terminal in raw mode, uses the alternate screen, and restores everything on exit.
Prefix parsing client-side, with the literal escape.
Bounded per-client output buffers with a drop-and-repaint policy.
Panes read and parsed with zero clients attached.
All eight tests passing, especially killing_the_client_leaves_everything_running.
The lsof experiment output showing master and socket ownership.

Validation / Self-check

Explain each of the six daemonization steps and what each prevents.
Why the double fork specifically?
Why must SIGPIPE be ignored, and what is its default disposition?
How do you distinguish a live server from a stale socket file?
What are the socket directory's required permissions and checks? What is the attack?
Why does the version handshake come first?
Why is the prefix parsed in the client?
What must be bounded on the server, and what is the recovery when the bound is hit?
Why must panes be read with zero clients attached?
Run the lsof experiment: which process holds ptmx, which holds pts, and what would it mean if one process held both?

Next: Lab 18 — Detach and Attach.

Open-Source Engineer & Contributor