foundry
Foundry OS: the operating system built on the Rustbelt kernel
| author | Tyler Fowler <tyler@localhost> | 2026-10-30 10:37:18 -0700 |
|---|---|---|
| commit | 97b0cc346c0cd9dda26e765fdd20cf91ecf13b28 | |
| parent | 5c8fa817a8d6e7db7ede3ec8406defb7bb97444e | |
| tree | browse files at this commit | |
WIP stage 15: libc plumbing (poll/epoll, nonblock, loopback, futex/threads, ptys)
| A | echoserver.rs | +206 -0 |
| M | getty.rs | +41 -19 |
| A | nbtest.rs | +117 -0 |
| A | ptytest.rs | +106 -0 |
| M | sh.rs | +4 -1 |
| A | threads.rs | +135 -0 |
| M | ulib.rs | +387 -13 |
7 files changed, 996 insertions, 33 deletions
@@ -0,0 +1,206 @@+//! echoserver -- an event-loop echo server over loopback (127.0.0.1).+//!+//! One process serves several clients at once without threads: it never+//! blocks on any single socket. Every socket is nonblocking, and the loop+//! asks the kernel "which of my sockets are ready?" (epoll_wait, or plain+//! poll), then only accepts/reads/writes those:+//!+//! loop {+//! ready = epoll_wait(...) // sleep until something happens+//! for each ready fd:+//! listener -> accept new clients, watch them too+//! client -> read; echo it back; 0 bytes = client left+//! }+//!+//! The self-test forks 3 client processes that talk to it over loopback,+//! once with epoll and once with poll, then checks UDP over loopback.+//! With an argument (`echoserver PORT`) it just serves forever.+#![no_std]+#![no_main]++#[macro_use]+#[path = "ulib.rs"]+mod ulib;++extern crate alloc;++use alloc::format;+use alloc::vec::Vec;+use ulib::*;++const LOOPBACK: Ip = Ip([127, 0, 0, 1]);+const CLIENTS: usize = 3;+const ROUNDS: usize = 3;++/// A client process: connect, send ROUNDS messages, check each echo.+fn client(port: u16, id: usize) -> i64 {+ let fd = socket(SOCK_STREAM, 0);+ if connect(fd, LOOPBACK, port) < 0 {+ return 10;+ }+ for round in 0..ROUNDS {+ let msg = format!("hello {} from client {}", round, id);+ write_all(fd, msg.as_bytes());+ let mut buf = [0u8; 64];+ let mut got = 0;+ while got < msg.len() {+ let n = read(fd, &mut buf[got..]);+ if n <= 0 {+ return 11;+ }+ got += n as usize;+ }+ if &buf[..got] != msg.as_bytes() {+ return 12;+ }+ }+ close(fd); // the server sees end-of-file+ 0+}++fn listener(port: u16) -> i64 {+ let fd = socket(SOCK_STREAM | O_NONBLOCK, 0); // SOCK_NONBLOCK+ if bind(fd, port) < 0 || listen(fd, 4) < 0 {+ return -1;+ }+ fd+}++/// Accept everyone waiting (nonblocking: stop at EAGAIN).+fn accept_all(lfd: i64, mut add: impl FnMut(i64)) {+ loop {+ let (fd, _) = accept(lfd);+ if fd < 0 {+ break; // -EAGAIN: nobody else waiting+ }+ set_nonblocking(fd, true);+ add(fd);+ }+}++/// Read what's there and echo it back. Returns false when the client left.+fn echo(fd: i64, served: &mut usize) -> bool {+ let mut buf = [0u8; 512];+ loop {+ let n = read(fd, &mut buf);+ if n == -EAGAIN {+ return true; // drained for now+ }+ if n <= 0 {+ return false; // 0 = end of file (or an error)+ }+ write_all(fd, &buf[..n as usize]);+ *served += 1;+ }+}++/// The epoll flavour: register fds once, then wait for just the ready ones.+/// Returns the number of clients served to the end (or -1 on a stall).+fn serve_epoll(lfd: i64, want: usize) -> i64 {+ let ep = epoll_create1(0);+ epoll_ctl(ep, EPOLL_CTL_ADD, lfd, EPOLLIN, lfd as u64);+ let (mut finished, mut served) = (0, 0);+ let mut events = [EpollEvent::default(); 8];+ while finished < want {+ let n = epoll_wait(ep, &mut events, 5000);+ if n <= 0 {+ close(ep);+ return -1; // 5 s with nothing happening: something's wrong+ }+ for ev in &events[..n as usize] {+ let fd = ev.data as i64;+ if fd == lfd {+ accept_all(lfd, |c| {+ epoll_ctl(ep, EPOLL_CTL_ADD, c, EPOLLIN, c as u64);+ });+ } else if !echo(fd, &mut served) {+ epoll_ctl(ep, EPOLL_CTL_DEL, fd, 0, 0);+ close(fd);+ finished += 1;+ }+ }+ }+ close(ep);+ println!("echoserver: epoll: {} clients, {} echoes", finished, served);+ finished as i64+}++/// The poll flavour: pass the whole fd list on every call.+fn serve_poll(lfd: i64, want: usize) -> i64 {+ let mut fds = alloc::vec![PollFd { fd: lfd as i32, events: POLLIN, revents: 0 }];+ let (mut finished, mut served) = (0, 0);+ while finished < want {+ if poll(&mut fds, 5000) <= 0 {+ return -1;+ }+ let mut new = Vec::new();+ let mut gone = Vec::new();+ for p in fds.iter() {+ if p.revents == 0 {+ continue;+ }+ if p.fd as i64 == lfd {+ accept_all(lfd, |c| new.push(c));+ } else if !echo(p.fd as i64, &mut served) {+ close(p.fd as i64);+ gone.push(p.fd);+ finished += 1;+ }+ }+ fds.retain(|p| !gone.contains(&p.fd));+ fds.extend(new.into_iter().map(|c| PollFd { fd: c as i32, events: POLLIN, revents: 0 }));+ }+ println!("echoserver: poll: {} clients, {} echoes", finished, served);+ finished as i64+}++/// Start the clients, run the server, collect the clients' results.+fn round(c: &mut Checks, port: u16, use_epoll: bool) {+ let lfd = listener(port);+ c.check(lfd >= 0, "listen on a loopback port");+ let mut pids = Vec::new();+ for id in 0..CLIENTS {+ let pid = fork();+ if pid == 0 {+ close(lfd);+ exit(client(port, id));+ }+ pids.push(pid);+ }+ let finished = if use_epoll { serve_epoll(lfd, CLIENTS) } else { serve_poll(lfd, CLIENTS) };+ close(lfd);+ let ok = pids.iter().all(|&pid| wait(pid) == 0);+ let what = if use_epoll { "epoll" } else { "poll" };+ c.check(finished == CLIENTS as i64, &format!("{}: served {} clients to the end", what, CLIENTS));+ c.check(ok, &format!("{}: every client got its echoes back", what));+}++fn main() -> i64 {+ if let Some(port) = arg(1).and_then(|a| a.parse().ok()) {+ let lfd = listener(port);+ if lfd < 0 {+ eprintln!("echoserver: can't listen on port {}", port);+ return 1;+ }+ println!("echoserver: listening on port {} (epoll)", port);+ loop {+ serve_epoll(lfd, usize::MAX);+ }+ }+ let mut c = Checks::new("echoserver");+ round(&mut c, 7070, true);+ round(&mut c, 7071, false);++ // UDP over loopback: a datagram to ourselves.+ let a = socket(SOCK_DGRAM, 0);+ let b = socket(SOCK_DGRAM, 0);+ bind(b, 7072);+ c.check(sendto(a, b"ping", LOOPBACK, 7072) == 4, "UDP sendto 127.0.0.1");+ set_recv_timeout(b, 2000);+ let mut buf = [0u8; 16];+ let (n, from) = recvfrom(b, &mut buf);+ c.check(n == 4 && &buf[..4] == b"ping" && from.ip() == LOOPBACK, "UDP datagram arrives, from 127.0.0.1");+ close(a);+ close(b);+ c.done()+}
@@ -1,14 +1,22 @@ //! getty [-1] DEVICE -- run a shell on another terminal, e.g. /dev/hvc0. //! //! Real Unix systems run one "getty" per terminal line: it opens the line-//! and starts a login shell on it. Ours skips the login: it opens DEVICE-//! and spawns `sh` with stdin, stdout and stderr all on it. When the shell-//! exits (`exit`) a new one starts, unless `-1` was given.+//! and starts a login shell on it. Ours skips the login. Since stage 15 it+//! does it the classic Unix way, with fork + exec:+//!+//! 1. fork: the child will become the shell.+//! 2. setsid (in the child): a new *session* with no terminal yet.+//! 3. open DEVICE: a session leader without a terminal that opens one+//! gets it as its *controlling terminal*. That's what lets ^C on+//! that terminal send SIGINT to the right programs.+//! 4. dup2 it onto fds 0, 1 and 2 (stdin, stdout, stderr).+//! 5. execve the shell: it keeps the fds and the session.+//!+//! getty waits for the shell; when it exits (`exit`) a new one starts,+//! unless `-1` was given. //! //! Try it: `getty /dev/hvc0 &` here, then on the host-//! `socat - UNIX-CONNECT:target/hvc0.sock` and type commands. That shell-//! is not "interactive" (no line editing): it reads one line at a time and-//! prints each line after the prompt, like a script.+//! `socat - UNIX-CONNECT:target/hvc0.sock` and type commands. #![no_std] #![no_main] @@ -20,6 +28,27 @@ use ulib::*; const SHELL: &str = "/mnt/disk0/bin/sh"; +/// The child's side (steps 2-5). Only returns on failure.+fn become_shell(dev: &str) -> i64 {+ setsid();+ let fd = open(dev, O_RDWR); // becomes our controlling terminal+ if fd < 0 {+ eprintln!("getty: {}: {}", dev, strerror(fd));+ return 126;+ }+ for target in 0..3 {+ dup2(fd, target);+ }+ if fd > 2 {+ close(fd);+ }+ let greeting = "\nRustbelt getty: shell on this terminal (type `exit` to end it)\n";+ write(1, greeting.as_bytes());+ let err = execve(SHELL, &["sh"], &environ());+ eprintln!("getty: {}: {}", SHELL, strerror(err));+ 127+}+ fn main() -> i64 { let (once, dev) = match (arg(1), arg(2)) { (Some("-1"), Some(dev)) => (true, dev),@@ -30,24 +59,17 @@ fn main() -> i64 { } }; loop {- let fd = open(dev, 0);- if fd < 0 {- eprintln!("getty: {}: {}", dev, strerror(fd));- return 1;+ let pid = fork();+ if pid == 0 {+ exit(become_shell(dev)); }- let greeting = "\nRustbelt getty: shell on this terminal (type `exit` to end it)\n";- write(fd, greeting.as_bytes());- let opts = SpawnOptions { stdin: fd, stdout: fd, stderr: fd, flags: 0 };- let pid = spawn(SHELL, "sh", &opts); if pid < 0 {- eprintln!("getty: {}: {}", SHELL, strerror(pid));- close(fd);+ eprintln!("getty: fork: {}", strerror(pid)); return 1; } let status = wait(pid);- close(fd);- if once {- return status;+ if once || status == 126 || status == 127 {+ return status; // done, or the terminal/shell can't be opened } sleep_ms(500); // don't spin if the shell keeps dying at once }
@@ -0,0 +1,117 @@+//! nbtest -- checks O_NONBLOCK: "would block" becomes EAGAIN instead.+//!+//! A blocking read on an empty pipe sleeps until data comes. With+//! O_NONBLOCK (set with fcntl(F_SETFL), ioctl(FIONBIO), pipe2 or+//! SOCK_NONBLOCK) it returns -EAGAIN at once ("try again later"), which+//! is what an event loop wants: it only reads when poll said "ready", and+//! must never get stuck on one fd.+#![no_std]+#![no_main]++#[macro_use]+#[path = "ulib.rs"]+mod ulib;++extern crate alloc;++use ulib::*;++const FIONBIO: u64 = 0x5421;++fn main() -> i64 {+ let mut c = Checks::new("nbtest");+ let mut buf = [0u8; 4096];++ // Pipes.+ let (r, w) = pipe2(O_NONBLOCK).unwrap();+ c.check(read(r, &mut buf) == -EAGAIN, "pipe: read with nothing there: EAGAIN");+ let mut p = [PollFd { fd: r as i32, events: POLLIN, revents: 0 }];+ c.check(poll(&mut p, 0) == 0, "pipe: poll says not readable");+ let mut total = 0;+ loop {+ let n = write(w, &buf);+ if n < 0 {+ c.check(n == -EAGAIN, "pipe: write to a full pipe: EAGAIN");+ break;+ }+ total += n;+ }+ println!("nbtest: the pipe took {} bytes before it was full", total);+ c.check(total == 65536, "pipe: holds 64 KiB, like Linux");+ let mut p = [PollFd { fd: w as i32, events: POLLOUT, revents: 0 }];+ c.check(poll(&mut p, 0) == 0, "pipe: poll says a full pipe isn't writable");+ c.check(read(r, &mut buf) == 4096, "pipe: then reading works");+ c.check(poll(&mut p, 0) == 1 && p[0].revents & POLLOUT != 0, "pipe: room again -> POLLOUT");+ close(w);+ while read(r, &mut buf) > 0 {}+ c.check(read(r, &mut buf) == 0, "pipe: writer gone -> end of file, not EAGAIN");+ close(r);++ // fcntl and FIONBIO switch it on and off.+ let (r, w) = pipe().unwrap();+ c.check(fcntl(r, F_GETFL, 0) as u64 & O_NONBLOCK == 0, "fcntl F_GETFL: blocking by default");+ set_nonblocking(r, true);+ c.check(fcntl(r, F_GETFL, 0) as u64 & O_NONBLOCK != 0, "fcntl F_SETFL O_NONBLOCK sticks");+ c.check(read(r, &mut buf) == -EAGAIN, "fcntl-made nonblocking pipe: EAGAIN");+ let off = 0i32;+ ioctl(r, FIONBIO, &off as *const i32 as u64);+ c.check(fcntl(r, F_GETFL, 0) as u64 & O_NONBLOCK == 0, "ioctl FIONBIO 0 turns it off");+ close(r);+ close(w);++ // Sockets (over loopback).+ let l = socket(SOCK_STREAM, 0);+ bind(l, 7080);+ listen(l, 2);+ set_nonblocking(l, true);+ c.check(accept(l).0 == -EAGAIN, "socket: accept with nobody calling: EAGAIN");+ let s = socket(SOCK_STREAM | O_NONBLOCK, 0);+ let r = connect(s, Ip([127, 0, 0, 1]), 7080);+ c.check(r == 0 || r == -EINPROGRESS, "socket: nonblocking connect: EINPROGRESS (or done at once)");+ let mut p = [PollFd { fd: s as i32, events: POLLOUT, revents: 0 }];+ c.check(poll(&mut p, 2000) == 1 && p[0].revents & POLLOUT != 0, "socket: poll POLLOUT = connected");+ let mut p = [PollFd { fd: l as i32, events: POLLIN, revents: 0 }];+ c.check(poll(&mut p, 2000) == 1, "socket: listener POLLIN = a connection to accept");+ let (a, _) = accept(l);+ c.check(a >= 0, "socket: accept it");+ c.check(read(s, &mut buf) == -EAGAIN, "socket: recv with no data: EAGAIN");+ write(a, b"hi");+ let mut p = [PollFd { fd: s as i32, events: POLLIN, revents: 0 }];+ poll(&mut p, 2000);+ c.check(read(s, &mut buf) == 2, "socket: data after POLLIN");+ close(a);+ close(s);+ close(l);++ // A terminal: the slave side of a pty.+ if let Ok((m, n)) = openpty() {+ let slave = open(&alloc::format!("/dev/pts/{}", n), O_RDWR | O_NOCTTY | O_NONBLOCK);+ c.check(read(slave, &mut buf) == -EAGAIN, "tty: read with no line typed: EAGAIN");+ close(slave);+ close(m);+ } else {+ c.check(false, "openpty");+ }++ // Small extras libc needs: pread/pwrite, writev, clocks, nanosleep.+ let fd = open("/tmp/nbtest.txt", O_CREAT | O_TRUNC | O_RDWR);+ writev(fd, &[b"hello ", b"vectored ", b"world"]);+ pwrite(fd, b"HELLO", 0);+ let n = pread(fd, &mut buf, 6);+ c.check(&buf[..n.max(0) as usize] == b"vectored world", "writev + pread at an offset");+ c.check(lseek(fd, 0, SEEK_CUR) == 20, "pread/pwrite leave the file position alone");+ lseek(fd, 0, SEEK_SET);+ let n = read(fd, &mut buf);+ c.check(&buf[..n.max(0) as usize] == b"HELLO vectored world", "pwrite overwrote the start");+ close(fd);+ unlink("/tmp/nbtest.txt");+ let (secs, _) = clock_gettime(0);+ c.check(secs > 1_600_000_000, "CLOCK_REALTIME is a real date (after 2020)");+ let t0 = now_us();+ nanosleep_us(30_000);+ c.check(now_us() - t0 >= 25_000, "nanosleep 30 ms");+ let mut link = [0u8; 64];+ let n = readlink("/proc/self/exe", &mut link);+ c.check(n > 0 && link[..n as usize].ends_with(b"nbtest"), "readlink /proc/self/exe = this program");+ c.done()+}
@@ -0,0 +1,106 @@+//! ptytest -- checks pseudo-terminals (ptys) and the line discipline.+//!+//! A pty is a pair: the *master* side is held by a terminal emulator (or+//! ssh, tmux, a compositor's terminal window), the *slave* side+//! (/dev/pts/N) looks exactly like a real terminal to the program running+//! on it. Between them sits the line discipline: in *canonical* mode it+//! collects a whole line (with backspace editing) before the program can+//! read it, echoes what's typed, turns ^C into SIGINT; in *raw* mode bytes+//! pass straight through (what editors and games use).+#![no_std]+#![no_main]++#[macro_use]+#[path = "ulib.rs"]+mod ulib;++extern crate alloc;++use alloc::format;+use ulib::*;++/// Read whatever the other side has (waits up to 1 s for the first byte).+fn read_some(fd: i64, buf: &mut [u8]) -> usize {+ let mut p = [PollFd { fd: fd as i32, events: POLLIN, revents: 0 }];+ if poll(&mut p, 1000) <= 0 {+ return 0;+ }+ read(fd, buf).max(0) as usize+}++fn main() -> i64 {+ let mut c = Checks::new("ptytest");+ let (master, n) = match openpty() {+ Ok(x) => x,+ Err(e) => {+ println!("ptytest: openpty: {}", strerror(e));+ return 1;+ }+ };+ let path = format!("/dev/pts/{}", n);+ let slave = open(&path, O_RDWR | O_NOCTTY);+ c.check(slave >= 0, &format!("open {} (the slave side)", path));+ let mut buf = [0u8; 64];++ // Canonical mode: a line is readable only once it's complete.+ write(master, b"hel");+ set_nonblocking(slave, true);+ c.check(read(slave, &mut buf) == -EAGAIN, "canonical: half a line can't be read yet (EAGAIN)");+ write(master, b"lo\x7f\x7fp me\r"); // two backspaces, then Enter ('\r')+ let got = read_some(slave, &mut buf);+ c.check(&buf[..got] == b"help me\n", "canonical: backspace edits the line, Enter ('\\r') ends it as '\\n'");+ let got = read_some(master, &mut buf);+ c.check(got > 0 && buf[..got].ends_with(b"\r\n"), "echo: what's typed comes back to the master");++ // Output processing: the program's '\n' reaches the terminal as "\r\n".+ write(slave, b"out\n");+ let got = read_some(master, &mut buf);+ c.check(&buf[..got] == b"out\r\n", "output: '\\n' becomes \"\\r\\n\" (ONLCR)");++ // Raw mode: every byte at once, no echo, no conversions.+ let saved = tcgetattr(slave).unwrap_or_default();+ let mut raw = saved;+ cfmakeraw(&mut raw);+ c.check(tcsetattr(slave, &raw) == 0, "TCSETS: switch to raw mode");+ write(master, b"x\r\x03");+ let got = read_some(slave, &mut buf);+ c.check(&buf[..got] == b"x\r\x03", "raw: bytes arrive at once, unchanged (even ^C)");+ set_nonblocking(master, true);+ c.check(read(master, &mut buf) == -EAGAIN, "raw: no echo");+ set_nonblocking(master, false);+ tcsetattr(slave, &saved);+ c.check(tcgetattr(slave).map(|t| t.lflag & ICANON != 0).unwrap_or(false), "TCGETS: canonical mode back on");+ // (We keep `slave` open: with no slave open at all, the master reports+ // a hang-up (POLLHUP) instead of waiting for the child to open it.)++ // ^C: a child makes the pty its controlling terminal (setsid + open)+ // and waits; typing ^C on the master must kill it with SIGINT.+ let pid = fork();+ if pid == 0 {+ setsid();+ let fd = open(&path, O_RDWR); // session leader: becomes our ctty+ if fd < 0 {+ exit(3);+ }+ write(fd, b"ready\n");+ loop {+ pause();+ }+ }+ let got = read_some(master, &mut buf); // "ready\r\n"+ c.check(&buf[..got] == b"ready\r\n", "child opened the pty as its controlling terminal");+ write(master, b"\x03");+ let status = match wait4(pid, 0) {+ Ok((_, s)) => s,+ Err(_) => 0,+ };+ c.check(wifsignaled(status) && wtermsig(status) == SIGINT, "^C on the master: SIGINT for the foreground job");+ let got = read_some(master, &mut buf);+ c.check(got >= 2 && &buf[..2] == b"^C", "^C is echoed as \"^C\"");++ // Hang-up: closing the master gives the slave end-of-file.+ close(master);+ c.check(read(slave, &mut buf) == 0, "master closed: the slave reads end-of-file");+ close(slave);+ c.done()+}
@@ -212,7 +212,10 @@ impl Shell { let mut line: Vec<u8> = Vec::new(); let mut hist = self.history.len(); // position while browsing history let mut esc = 0; // 0: normal, 1: after ESC, 2: after ESC [- let mut buf = [0u8; 16];+ // One byte per read: if several lines were typed (or pasted) ahead,+ // the rest stays in the tty for the next prompt instead of being+ // lost when we stop at the first Enter.+ let mut buf = [0u8; 1]; 'read: loop { let n = read(0, &mut buf); if n <= 0 {
@@ -0,0 +1,135 @@+//! threads [N] -- several threads in one process, pthread-style.+//!+//! Threads share *everything* a process owns: memory, open files, signal+//! handlers. Each has only its own registers, stack and TLS block (and a+//! thread id). ulib's `spawn_thread` = pthread_create: clone with+//! CLONE_VM | CLONE_FILES | CLONE_SIGHAND | CLONE_THREAD | ..., a fresh+//! stack, CLONE_SETTLS for the TLS block, and CLONE_CHILD_CLEARTID so+//! `join` can sleep on a futex until the kernel says "thread gone".+//!+//! The demo: a parallel sum. 1..=COUNT is split into N slices (default 4,+//! one per CPU); each thread adds up its slice, then adds its result to a+//! shared total under a futex mutex. Many small increments of a shared+//! counter (also under the mutex) check the mutex really excludes; a+//! counter bumped *without* it shows why that matters.+#![no_std]+#![no_main]++#[macro_use]+#[path = "ulib.rs"]+mod ulib;++extern crate alloc;++use alloc::sync::Arc;+use alloc::vec::Vec;+use core::cell::UnsafeCell;+use core::sync::atomic::{AtomicU32, AtomicU64, Ordering};+use ulib::*;++const COUNT: u64 = 40_000_000;+const BUMPS: u64 = 20_000;++/// Data guarded by a FutexMutex (a tiny `Mutex<T>`).+struct Shared {+ lock: FutexMutex,+ total: UnsafeCell<u64>,+ bumps: UnsafeCell<u64>,+ /// Bumped with a plain read + write, no lock: races lose updates.+ racy: UnsafeCell<u64>,+}+unsafe impl Sync for Shared {}+unsafe impl Send for Shared {}++fn main() -> i64 {+ let n: u64 = arg(1).and_then(|a| a.parse().ok()).unwrap_or(4).clamp(1, 16);+ let mut c = Checks::new("threads");+ let shared = Arc::new(Shared {+ lock: FutexMutex::new(),+ total: UnsafeCell::new(0),+ bumps: UnsafeCell::new(0),+ racy: UnsafeCell::new(0),+ });+ let cpus_seen = Arc::new(AtomicU64::new(0)); // bit i = some thread ran on CPU i+ let tids = Arc::new(AtomicU64::new(0)); // sum of the threads' tids+ let tls_ok = Arc::new(AtomicU32::new(0));+ let start = now_us();+ let mut threads = Vec::new();+ for i in 0..n {+ let (shared, cpus_seen, tids, tls_ok) = (shared.clone(), cpus_seen.clone(), tids.clone(), tls_ok.clone());+ let from = COUNT * i / n + 1;+ let to = COUNT * (i + 1) / n;+ // Each thread gets `i` as its TLS value.+ let t = spawn_thread(i, move || {+ let mut sum = 0u64;+ for x in from..=to {+ sum = sum.wrapping_add(core::hint::black_box(x));+ }+ cpus_seen.fetch_or(1 << sysinfo().this_cpu, Ordering::Relaxed);+ tids.fetch_add(gettid() as u64, Ordering::Relaxed);+ if tls_value() == i {+ tls_ok.fetch_add(1, Ordering::Relaxed);+ }+ for _ in 0..BUMPS {+ shared.lock.lock();+ unsafe { *shared.bumps.get() += 1 };+ shared.lock.unlock();+ unsafe {+ let v = core::ptr::read_volatile(shared.racy.get());+ core::ptr::write_volatile(shared.racy.get(), v + 1);+ }+ }+ shared.lock.lock();+ unsafe { *shared.total.get() += sum };+ shared.lock.unlock();+ });+ match t {+ Ok(t) => threads.push(t),+ Err(e) => {+ println!("threads: spawn_thread: {}", strerror(e));+ return 1;+ }+ }+ }+ let spawned_tids: u64 = threads.iter().map(|t| t.tid as u64).sum();+ for t in threads {+ t.join();+ }+ let ms = (now_us() - start) / 1000;+ let total = unsafe { *shared.total.get() };+ let bumps = unsafe { *shared.bumps.get() };+ let racy = unsafe { *shared.racy.get() };+ let cpus = cpus_seen.load(Ordering::Relaxed).count_ones();+ println!("threads: {} threads summed 1..={} = {} in {} ms on {} CPU(s)", n, COUNT, total, ms, cpus);+ println!("threads: locked counter {} (want {}), unlocked counter {} (races lose updates)", bumps, n * BUMPS, racy);+ c.check(total == COUNT * (COUNT + 1) / 2, "parallel sum is right");+ c.check(bumps == n * BUMPS, "the futex mutex never let two threads in at once");+ c.check(tls_ok.load(Ordering::Relaxed) == n as u32, "every thread saw its own TLS value (fs:[8])");+ c.check(tids.load(Ordering::Relaxed) == spawned_tids, "gettid in each thread = the tid clone returned");+ c.check(gettid() == getpid(), "the main thread's tid is the pid");++ // A futex wait whose value doesn't match returns at once (EAGAIN); one+ // that matches sleeps until its timeout.+ let word = AtomicU32::new(7);+ c.check(futex_wait(&word, 8, None) == -EAGAIN, "futex_wait with a stale value: EAGAIN");+ let t0 = now_us();+ let r = futex_wait(&word, 7, Some(50_000));+ c.check(r == -ETIMEDOUT && now_us() - t0 >= 40_000, "futex_wait times out after 50 ms");+ c.check(futex_wake(&word, 1) == 0, "futex_wake with nobody waiting wakes 0");++ // exit vs exit_group: a thread that calls exit (60) ends only itself.+ let alive = Arc::new(AtomicU32::new(0));+ let a2 = alive.clone();+ let t = spawn_thread(0, move || {+ a2.store(1, Ordering::SeqCst);+ // falls off the end: ulib calls exit(0) = just this thread+ })+ .unwrap();+ t.join();+ c.check(alive.load(Ordering::SeqCst) == 1, "a thread's exit ends only that thread");++ // tgkill: a signal for one particular thread (here: ourselves).+ c.check(tgkill(getpid(), gettid(), 0) == 0, "tgkill(pid, tid, 0) finds the thread");+ c.check(tgkill(getpid(), 99999, 0) == -ESRCH, "tgkill to a missing thread: ESRCH");+ c.done()+}
@@ -29,7 +29,7 @@ pub const SYS_LSEEK: u64 = 8; pub const SYS_IOCTL: u64 = 16; pub const SYS_PIPE: u64 = 22; pub const SYS_SCHED_YIELD: u64 = 24;-pub const SYS_SLEEP_MS: u64 = 35;+pub const SYS_NANOSLEEP: u64 = 35; pub const SYS_GETPID: u64 = 39; pub const SYS_EXIT: u64 = 60; pub const SYS_WAIT4: u64 = 61;@@ -227,7 +227,12 @@ pub fn sched_yield() { syscall3(SYS_SCHED_YIELD, 0, 0, 0); } pub fn sleep_ms(ms: u64) {- syscall3(SYS_SLEEP_MS, ms, 0, 0);+ nanosleep_us(ms * 1000);+}+/// nanosleep(*timespec, *remaining): 0, or -EINTR if a signal woke us.+pub fn nanosleep_us(us: u64) -> i64 {+ let ts = [us / 1_000_000, (us % 1_000_000) * 1000];+ syscall3(SYS_NANOSLEEP, ts.as_ptr() as u64, 0, 0) } pub fn getpid() -> i64 { syscall3(SYS_GETPID, 0, 0, 0)@@ -590,9 +595,11 @@ pub fn sysinfo() -> SysInfo { syscall3(SYS_SYSINFO, &mut info as *mut SysInfo as u64, 0, 0); info }-/// Current directory into `buf`; returns its length (or -errno).+/// Current directory into `buf`; returns its length (or -errno). The+/// kernel, like Linux, also writes a NUL and counts it; we don't. pub fn getcwd(buf: &mut [u8]) -> i64 {- syscall3(SYS_GETCWD, buf.as_mut_ptr() as u64, buf.len() as u64, 0)+ let n = syscall3(SYS_GETCWD, buf.as_mut_ptr() as u64, buf.len() as u64, 0);+ if n > 0 { n - 1 } else { n } } pub fn chdir(path: &str) -> i64 { syscall3(SYS_CHDIR, path.as_ptr() as u64, path.len() as u64, 0)@@ -1068,8 +1075,9 @@ fn panic(info: &core::panic::PanicInfo) -> ! { // up gives us more (zeroed, demand-paged) memory. // Same idea as the kernel's fixed-size-block allocator: sizes are rounded up // to a power of two (16 B .. 1 MiB) and freed blocks go on one free list-// per size, to be reused by the next allocation of that size. Programs are-// single-threaded, so no lock.+// per size, to be reused by the next allocation of that size. Programs may+// have several threads (stage 15), so a tiny spinlock guards the heap; a+// thread that finds it taken yields its CPU instead of spinning long. const CLASSES: usize = 17; // 16 << 16 = 1 MiB @@ -1078,9 +1086,21 @@ struct Heap { used: UnsafeCell<u64>, // bump pointer: end of what's handed out end: UnsafeCell<u64>, // the current break free: UnsafeCell<[usize; CLASSES]>, // free list heads (addresses, 0 = empty)+ lock: AtomicBool, } -unsafe impl Sync for Heap {} // single-threaded programs+unsafe impl Sync for Heap {} // the fields are only touched under `lock`++impl Heap {+ fn lock(&self) {+ while self.lock.swap(true, Ordering::Acquire) {+ sched_yield();+ }+ }+ fn unlock(&self) {+ self.lock.store(false, Ordering::Release);+ }+} #[global_allocator] static HEAP: Heap = Heap {@@ -1088,6 +1108,7 @@ static HEAP: Heap = Heap { used: UnsafeCell::new(0), end: UnsafeCell::new(0), free: UnsafeCell::new([0; CLASSES]),+ lock: AtomicBool::new(false), }; /// Size class for a layout: 0 = 16 bytes, 1 = 32, ...@@ -1098,6 +1119,26 @@ fn class_of(layout: Layout) -> usize { unsafe impl GlobalAlloc for Heap { unsafe fn alloc(&self, layout: Layout) -> *mut u8 {+ self.lock();+ let p = unsafe { self.alloc_locked(layout) };+ self.unlock();+ p+ }++ unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {+ let class = class_of(layout);+ self.lock();+ unsafe {+ let free = &mut *self.free.get();+ *(ptr as *mut usize) = free[class];+ free[class] = ptr as usize;+ }+ self.unlock();+ }+}++impl Heap {+ unsafe fn alloc_locked(&self, layout: Layout) -> *mut u8 { let class = class_of(layout); if class >= CLASSES { return core::ptr::null_mut(); // too big: "out of memory"@@ -1133,14 +1174,347 @@ unsafe impl GlobalAlloc for Heap { start as *mut u8 } }+} - unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {- let class = class_of(layout);- unsafe {- let free = &mut *self.free.get();- *(ptr as *mut usize) = free[class];- free[class] = ptr as usize;+// --- Stage 15: what a libc's plumbing needs ----------------------------------++pub const SYS_POLL: u64 = 7;+pub const SYS_PREAD64: u64 = 17;+pub const SYS_PWRITE64: u64 = 18;+pub const SYS_READV: u64 = 19;+pub const SYS_WRITEV: u64 = 20;+pub const SYS_CLONE: u64 = 56;+pub const SYS_EXIT_THREAD: u64 = 60; // Linux's "exit": just this thread+pub const SYS_SETSID: u64 = 112;+pub const SYS_ARCH_PRCTL: u64 = 158;+pub const SYS_GETTID: u64 = 186;+pub const SYS_FUTEX: u64 = 202;+pub const SYS_EPOLL_WAIT: u64 = 232;+pub const SYS_EPOLL_CTL: u64 = 233;+pub const SYS_TGKILL: u64 = 234;+pub const SYS_EPOLL_CREATE1: u64 = 291;++pub const O_NOCTTY: u64 = 0o400;+pub const O_NONBLOCK: u64 = 0o4000;+pub const O_RDWR: u64 = 2;+pub const F_GETFL: u64 = 3;+pub const F_SETFL: u64 = 4;+pub const EINPROGRESS: i64 = 115;++/// Turn O_NONBLOCK on or off for `fd` (fcntl F_GETFL + F_SETFL).+pub fn set_nonblocking(fd: i64, on: bool) -> i64 {+ let flags = fcntl(fd, F_GETFL, 0);+ if flags < 0 {+ return flags;+ }+ let flags = if on { flags as u64 | O_NONBLOCK } else { flags as u64 & !O_NONBLOCK };+ fcntl(fd, F_SETFL, flags)+}++pub fn pread(fd: i64, buf: &mut [u8], offset: u64) -> i64 {+ syscall6(SYS_PREAD64, fd as u64, buf.as_mut_ptr() as u64, buf.len() as u64, offset, 0, 0)+}+pub fn pwrite(fd: i64, buf: &[u8], offset: u64) -> i64 {+ syscall6(SYS_PWRITE64, fd as u64, buf.as_ptr() as u64, buf.len() as u64, offset, 0, 0)+}+/// writev: several buffers in one call (`struct iovec {base, len}`).+pub fn writev(fd: i64, bufs: &[&[u8]]) -> i64 {+ let iov: alloc::vec::Vec<[u64; 2]> = bufs.iter().map(|b| [b.as_ptr() as u64, b.len() as u64]).collect();+ syscall3(SYS_WRITEV, fd as u64, iov.as_ptr() as u64, iov.len() as u64)+}++/// Linux 228: 0 = CLOCK_REALTIME (wall clock), 1 = CLOCK_MONOTONIC.+pub fn clock_gettime(clock: u64) -> (u64, u64) {+ let mut ts = [0u64; 2];+ syscall3(SYS_CLOCK_GETTIME, clock, ts.as_mut_ptr() as u64, 0);+ (ts[0], ts[1])+}++// poll / epoll (see the kernel's poll.rs).+pub const POLLIN: u16 = 0x1;+pub const POLLOUT: u16 = 0x4;+pub const POLLERR: u16 = 0x8;+pub const POLLHUP: u16 = 0x10;+pub const POLLNVAL: u16 = 0x20;++/// `struct pollfd`.+#[repr(C)]+#[derive(Clone, Copy, Default)]+pub struct PollFd {+ pub fd: i32,+ pub events: u16,+ pub revents: u16,+}+/// Wait until one of `fds` is ready (timeout_ms -1 = forever); returns+/// how many are, 0 on timeout.+pub fn poll(fds: &mut [PollFd], timeout_ms: i64) -> i64 {+ syscall3(SYS_POLL, fds.as_mut_ptr() as u64, fds.len() as u64, timeout_ms as u64)+}++pub const EPOLL_CTL_ADD: u64 = 1;+pub const EPOLL_CTL_DEL: u64 = 2;+pub const EPOLL_CTL_MOD: u64 = 3;+pub const EPOLLIN: u32 = 0x1;+pub const EPOLLOUT: u32 = 0x4;++/// `struct epoll_event` is *packed* on x86_64: 4 + 8 = 12 bytes.+#[repr(C, packed)]+#[derive(Clone, Copy, Default)]+pub struct EpollEvent {+ pub events: u32,+ pub data: u64,+}+pub fn epoll_create1(flags: u64) -> i64 {+ syscall3(SYS_EPOLL_CREATE1, flags, 0, 0)+}+pub fn epoll_ctl(epfd: i64, op: u64, fd: i64, events: u32, data: u64) -> i64 {+ let ev = EpollEvent { events, data };+ syscall5(SYS_EPOLL_CTL, epfd as u64, op, fd as u64, &ev as *const EpollEvent as u64, 0)+}+pub fn epoll_wait(epfd: i64, out: &mut [EpollEvent], timeout_ms: i64) -> i64 {+ syscall5(SYS_EPOLL_WAIT, epfd as u64, out.as_mut_ptr() as u64, out.len() as u64, timeout_ms as u64, 0)+}++// Sessions and terminals.+pub fn setsid() -> i64 {+ syscall3(SYS_SETSID, 0, 0, 0)+}+pub const TCGETS: u64 = 0x5401;+pub const TCSETS: u64 = 0x5402;+pub const TIOCSCTTY: u64 = 0x540E;+pub const TIOCGPTN: u64 = 0x8004_5430;+pub const TIOCSPTLCK: u64 = 0x4004_5431;+pub const ICANON: u32 = 0x2;+pub const ECHO: u32 = 0x8;+pub const ISIG: u32 = 0x1;+pub const VMIN: usize = 6;+pub const VTIME: usize = 5;++/// Linux's `struct termios` (36 bytes): the terminal's settings.+#[repr(C)]+#[derive(Clone, Copy, Default)]+pub struct Termios {+ pub iflag: u32,+ pub oflag: u32,+ pub cflag: u32,+ pub lflag: u32,+ pub line: u8,+ pub cc: [u8; 19],+}+pub fn tcgetattr(fd: i64) -> Result<Termios, i64> {+ let mut t = Termios::default();+ match ioctl(fd, TCGETS, &mut t as *mut Termios as u64) {+ 0 => Ok(t),+ e => Err(e),+ }+}+pub fn tcsetattr(fd: i64, t: &Termios) -> i64 {+ ioctl(fd, TCSETS, t as *const Termios as u64)+}+/// Raw mode, like libc's cfmakeraw: no line editing, echo or signals,+/// no character conversions; read returns as soon as 1 byte is there.+pub fn cfmakeraw(t: &mut Termios) {+ t.iflag &= !(0x1 | 0x2 | 0x8 | 0x20 | 0x40 | 0x80 | 0x100 | 0x400); // IGNBRK BRKINT PARMRK ISTRIP INLCR IGNCR ICRNL IXON+ t.oflag &= !0x1; // OPOST+ t.lflag &= !(ECHO | 0x40 | ICANON | ISIG | 0x8000); // ECHO ECHONL ICANON ISIG IEXTEN+ t.cflag = (t.cflag & !0x130) | 0x30; // CS8, no parity+ t.cc[VMIN] = 1;+ t.cc[VTIME] = 0;+}++/// Open a new pseudo-terminal: returns (master fd, slave number N); the+/// slave is /dev/pts/N. Like posix_openpt + grantpt + unlockpt.+pub fn openpty() -> Result<(i64, u32), i64> {+ let master = open("/dev/ptmx", O_RDWR | O_NOCTTY);+ if master < 0 {+ return Err(master);+ }+ let mut n = 0u32;+ let unlock = 0u32;+ let r = ioctl(master, TIOCGPTN, &mut n as *mut u32 as u64);+ let r = if r == 0 { ioctl(master, TIOCSPTLCK, &unlock as *const u32 as u64) } else { r };+ if r != 0 {+ close(master);+ return Err(r);+ }+ Ok((master, n))+}++// Threads: clone + futex + TLS.+pub const FUTEX_WAIT: u64 = 0;+pub const FUTEX_WAKE: u64 = 1;+pub const FUTEX_PRIVATE_FLAG: u64 = 128;+pub const CLONE_VM: u64 = 0x100;+pub const CLONE_FS: u64 = 0x200;+pub const CLONE_FILES: u64 = 0x400;+pub const CLONE_SIGHAND: u64 = 0x800;+pub const CLONE_THREAD: u64 = 0x10000;+pub const CLONE_SYSVSEM: u64 = 0x40000;+pub const CLONE_SETTLS: u64 = 0x80000;+pub const CLONE_PARENT_SETTID: u64 = 0x100000;+pub const CLONE_CHILD_CLEARTID: u64 = 0x200000;+pub const ARCH_SET_FS: u64 = 0x1002;+pub const ARCH_GET_FS: u64 = 0x1003;++use core::sync::atomic::{AtomicBool, AtomicU32, Ordering};++pub fn gettid() -> i64 {+ syscall3(SYS_GETTID, 0, 0, 0)+}+pub fn tgkill(tgid: i64, tid: i64, sig: i32) -> i64 {+ syscall3(SYS_TGKILL, tgid as u64, tid as u64, sig as u64)+}+/// Sleep while `*word == expected` (until futex_wake, or `timeout_us`).+pub fn futex_wait(word: &AtomicU32, expected: u32, timeout_us: Option<u64>) -> i64 {+ let ts = timeout_us.map(|us| [us / 1_000_000, (us % 1_000_000) * 1000]);+ let ts_ptr = ts.as_ref().map_or(0, |t| t.as_ptr() as u64);+ syscall6(SYS_FUTEX, word.as_ptr() as u64, FUTEX_WAIT | FUTEX_PRIVATE_FLAG, expected as u64, ts_ptr, 0, 0)+}+/// Wake up to `n` threads sleeping on `word`; returns how many woke.+pub fn futex_wake(word: &AtomicU32, n: u32) -> i64 {+ syscall6(SYS_FUTEX, word.as_ptr() as u64, FUTEX_WAKE | FUTEX_PRIVATE_FLAG, n as u64, 0, 0, 0)+}++/// A mutex built on a futex (Ulrich Drepper's "Futexes Are Tricky",+/// mutex #2). The word is 0 = unlocked, 1 = locked, 2 = locked and+/// someone may be sleeping. Taking a free lock is one atomic instruction,+/// no syscall; only contention goes to the kernel.+pub struct FutexMutex {+ state: AtomicU32,+}+impl FutexMutex {+ pub const fn new() -> FutexMutex {+ FutexMutex { state: AtomicU32::new(0) }+ }+ pub fn lock(&self) {+ if self.state.compare_exchange(0, 1, Ordering::Acquire, Ordering::Relaxed).is_ok() {+ return; // fast path: it was free+ }+ // Slow path: mark "contended" (2) and sleep until it's free. The+ // kernel only puts us to sleep if the word is *still* 2, so an+ // unlock between our swap and the futex call can't be missed.+ while self.state.swap(2, Ordering::Acquire) != 0 {+ futex_wait(&self.state, 2, None);+ }+ }+ pub fn unlock(&self) {+ // 1 -> 0: nobody waits. 2 -> 0: wake one sleeper.+ if self.state.swap(0, Ordering::Release) == 2 {+ futex_wake(&self.state, 1);+ }+ }+}++/// Thread-local storage, the x86_64 way: each thread's FS register base+/// points at its own block, whose first word points at itself (so code can+/// find the block with `mov rax, fs:[0]`). We keep one u64 per thread in+/// the second word.+#[repr(C)]+pub struct TlsBlock {+ pub self_ptr: u64,+ pub value: u64,+}+pub fn arch_prctl(code: u64, addr: u64) -> i64 {+ syscall3(SYS_ARCH_PRCTL, code, addr, 0)+}+/// This thread's TLS value (fs:[8]).+pub fn tls_value() -> u64 {+ let v: u64;+ unsafe { asm!("mov {}, fs:[8]", out(reg) v, options(nostack, readonly)) };+ v+}++/// A running thread (see `spawn_thread`).+pub struct Thread {+ pub tid: i64,+ /// The kernel writes our tid here (CLONE_PARENT_SETTID) and zeroes it ++ /// futex-wakes it when the thread exits (CLONE_CHILD_CLEARTID): that's+ /// how `join` knows it's done, exactly like pthread_join.+ tid_word: alloc::boxed::Box<AtomicU32>,+ stack: u64,+ tls: alloc::boxed::Box<TlsBlock>,+}++const THREAD_STACK: u64 = 64 * 1024;+type ThreadFn = alloc::boxed::Box<dyn FnOnce()>;++/// The new thread starts here, on its own stack, with the boxed closure.+extern "C" fn thread_entry(arg: u64) -> ! {+ let f = unsafe { alloc::boxed::Box::from_raw(arg as *mut ThreadFn) };+ f();+ // exit (60), not exit_group: only this thread ends.+ syscall3(SYS_EXIT_THREAD, 0, 0, 0);+ loop {}+}++/// Start `f` in a new thread of this process (same memory, same fds),+/// with `tls_value` in its TLS block. Like pthread_create.+pub fn spawn_thread(tls_value: u64, f: impl FnOnce() + Send + 'static) -> Result<Thread, i64> {+ let stack = mmap(0, THREAD_STACK, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);+ if stack < 0 {+ return Err(stack);+ }+ let stack = stack as u64;+ let boxed: alloc::boxed::Box<ThreadFn> = alloc::boxed::Box::new(alloc::boxed::Box::new(f));+ let arg = alloc::boxed::Box::into_raw(boxed) as u64;+ let mut tls = alloc::boxed::Box::new(TlsBlock { self_ptr: 0, value: tls_value });+ tls.self_ptr = &*tls as *const TlsBlock as u64;+ let tid_word = alloc::boxed::Box::new(AtomicU32::new(0));+ // The child's stack starts with [entry, arg] for the asm below.+ let sp = stack + THREAD_STACK - 16;+ unsafe {+ *(sp as *mut u64) = thread_entry as usize as u64;+ *((sp + 8) as *mut u64) = arg;+ }+ let flags = CLONE_VM | CLONE_FS | CLONE_FILES | CLONE_SIGHAND | CLONE_THREAD | CLONE_SYSVSEM+ | CLONE_SETTLS | CLONE_PARENT_SETTID | CLONE_CHILD_CLEARTID;+ let tid_ptr = tid_word.as_ptr() as u64;+ let ret: i64;+ // clone returns twice: the parent gets the tid, the child 0 -- on its+ // new stack, so it must not touch the parent's stack frame. It pops+ // [entry, arg] and calls entry(arg) straight from assembly (after the+ // two pops rsp is 16-byte aligned, as `call` wants).+ unsafe {+ asm!(+ "syscall",+ "test rax, rax",+ "jnz 2f",+ "pop rax",+ "pop rdi",+ "call rax",+ "ud2",+ "2:",+ inlateout("rax") SYS_CLONE as i64 => ret,+ in("rdi") flags,+ in("rsi") sp,+ in("rdx") tid_ptr,+ in("r10") tid_ptr,+ in("r8") tls.self_ptr,+ lateout("rcx") _,+ lateout("r11") _,+ options(nostack)+ );+ }+ if ret < 0 {+ unsafe { drop(alloc::boxed::Box::from_raw(arg as *mut ThreadFn)) };+ munmap(stack, THREAD_STACK);+ return Err(ret);+ }+ Ok(Thread { tid: ret, tid_word, stack, tls })+}++impl Thread {+ /// Wait for the thread to finish, then free its stack.+ pub fn join(self) {+ loop {+ let tid = self.tid_word.load(Ordering::Acquire);+ if tid == 0 {+ break; // the kernel cleared it: the thread is gone+ }+ futex_wait(&self.tid_word, tid, None); }+ munmap(self.stack, THREAD_STACK);+ drop(self.tls); } }