diff --git a/docs/src/architecture/files.md b/docs/src/architecture/files.md index 2e4abe37ebf..51ddf1b6ea2 100644 --- a/docs/src/architecture/files.md +++ b/docs/src/architecture/files.md @@ -18,6 +18,12 @@ prefix. An empty view names nothing. directory moved elsewhere never reach outside it. An absolute link is resolved again in the client's own view, so it reaches nothing the client does not hold. +- **A process's threads share its connection.** Each directory capability is + one connection per process, on which its threads' requests are in flight at + once, each in its own share of the window the client lends; none waits for + another's round trip but one that needs the whole window, or one beyond the + sixteen a connection carries. The server answers connections in turn, so a + process's share of the server does not grow with its threads. - **Sharing is granted, never reached.** A file outside a program's view reaches it only through the file picker, which hands over that one file. - **No special cases.** The boot partition is in the updater's view and nobody @@ -79,6 +85,7 @@ directory, which the child resolves in its own view. - Ordinary I/O failure returns an error. A file server's crash invalidates that server's open files, which then answer that they are gone; a fresh open reaches the restarted server. Unsaved changes may be lost; data already - confirmed durable stays durable. + confirmed durable stays durable. A server that stays silent past its bound + while it owes a reply is let go as if it had crashed, as a block server is. - Per-session temporary storage is an ordinary file server holding its contents in RAM. diff --git a/issues/a-call-beside-another-thread-of-its-process-is-unmeasured-on-metal.md b/issues/a-call-beside-another-thread-of-its-process-is-unmeasured-on-metal.md new file mode 100644 index 00000000000..70a6ff3f818 --- /dev/null +++ b/issues/a-call-beside-another-thread-of-its-process-is-unmeasured-on-metal.md @@ -0,0 +1,29 @@ +--- +status: open +kind: tooling +opened: 2026-11-11 +--- + +# A call beside another thread of its process is unmeasured on metal + +A process's threads have their requests on one directory capability in flight at once: up to the wire's 16, each in +its own share of the window, every thread waiting for its own tag's reply (`sdk/toyos/src/fs.rs`, its in-flight core +`fs/flight.rs`, modelled under loom in `kernel/loom/tests/fs_flight.rs`). One connection per process and capability +stays, so the server's per-connection fairness is unchanged; a connection per thread was the alternative, rejected +because it multiplies windows and server slots by threads and gives a many-threaded process a larger share of the +server. + +Before, the process took one turn on the capability for each whole round trip. On the T14 (osbench `fileops`, +scratch stage timers, 2026-11-11, evidence of PR 1225), `fstat` on /tmp beside a thread of the same process asking as +fast as it is answered took 21.4 to 21.6 µs p50 against 7.6 alone (2.8x), and beside another process 8.4 to 8.5 µs +(1.1x). What a call beside another thread costs now is unmeasured: the benches are down. + +- **Owed on metal:** `osbench_disk` on the T14, main from before the tagged replies and after, two boots a side in + A, B, B, A order. On both volumes, main after must show `stat.beside_a_thread` and + `open_close.beside_a_thread` within 1.2x of `stat` and `open_close` in the same boot; main before is the + comparison. + +**Exit**: that measurement, the two rows within 1.2x on both volumes, or each remaining difference measured to its +parts. + +**Owner**: agents/fileops (the file server's client). diff --git a/issues/a-file-server-reply-has-no-derived-worst-case.md b/issues/a-file-server-reply-has-no-derived-worst-case.md new file mode 100644 index 00000000000..a4e751c3527 --- /dev/null +++ b/issues/a-file-server-reply-has-no-derived-worst-case.md @@ -0,0 +1,24 @@ +--- +status: open +kind: defect +opened: 2026-11-12 +severity: [hang] +--- + +# A file server's reply has no derived worst case + +A file server's client lets its server go once it has waited `REPLY_BOUND_NS` for a reply it is owed: the block +protocol's silence bound, 365 s, and 60 s of room (`sdk/toyos/src/fs.rs`). Every request in flight on that connection +then fails `Unanswered`, and every file it opened answers `Stale`. + +Nothing derives what a reply can take. One request may wait on several block requests in turn, each allowed the +block protocol's silence: an fsync writes back every page its file owes inside the server's one loop +(`system/fileserver/src/pages.rs`), and a write that needs room writes back first. The server answers its other +connections one request each in between, and nothing bounds how many there are. So a server that keeps its block +contract can stay silent past the bound and be let go, and the bound itself rests on no figure of the server's. + +**Exit**: a derived worst case for a file server's reply, from the block requests one request may wait on and the +turns it may wait behind, with the client's bound set from it; or a server that tells a client it owes a reply is +still being worked on, so that silence past a bound means a hang. + +**Owner**: agents/fileops (the file server and its client). diff --git a/issues/a-process-s-threads-take-turns-on-a-directory-capability-for-whole-round-trips.md b/issues/a-process-s-threads-take-turns-on-a-directory-capability-for-whole-round-trips.md deleted file mode 100644 index 00be89ce337..00000000000 --- a/issues/a-process-s-threads-take-turns-on-a-directory-capability-for-whole-round-trips.md +++ /dev/null @@ -1,25 +0,0 @@ ---- -status: assigned -kind: defect -opened: 2026-11-11 ---- - -# A process's threads take turns on a directory capability for whole round trips - -`sdk/toyos/src/files/owner.rs` holds one connection per directory capability behind a `Turnstile`, and every -request through it (`fs.rs` `Dir::call_with`) takes the turn for its whole round trip: one request in flight per -capability per process, though the wire carries up to 16 tagged requests. Two threads of one program calling into -one capability serialise, and unfairly. - -Measured on the T14 (osbench `fileops`, scratch stage timers, 2026-11-11, evidence of PR 1225): `fstat` on /tmp -beside a thread of the same process asking as fast as it is answered takes 21.4 to 21.6 µs p50 against 7.6 alone -(2.8x), and the other thread is answered 7.4 to 7.9 times for each answer to this one. Beside another process the -same call takes 8.4 to 8.5 µs (1.1x): the server interleaves two clients fine, so the cost is the client's turn. -Every multi-threaded program pays it on every path call: a parallel compiler, an async runtime. - -Owner: agents/fileops, after storage step 3a (holder-side `fstat` and writes, which leave the turnstile already). -Exit: requests from different threads are in flight on one capability at once, each thread waiting for its own -tag's reply, the window divided into per-request slots; one connection per process and capability stays, so the -server's per-client fairness is unchanged (a connection per thread was the alternative, rejected because it -multiplies windows and server slots by threads and gives a many-threaded process a larger share of the server). -Done when the `fstat`-beside-a-thread row reads within 1.2x of alone for `stat`, open and a path call. diff --git a/kernel/loom/src/lib.rs b/kernel/loom/src/lib.rs index 74955aab90f..967c9ff431b 100644 --- a/kernel/loom/src/lib.rs +++ b/kernel/loom/src/lib.rs @@ -266,3 +266,69 @@ pub mod sched; #[cfg(feature = "loom")] pub mod sched_model; + +/// What a file server connection's requests in flight are built on, as loom's: +/// its atomics and lock, and the futex. A wait sleeps only while its word +/// still says what the waiter saw, checked under the lock a wake takes after +/// its store, and returns only once that word is woken, as the kernel's futex +/// wakes only a waiter of the word posted to: a wake the protocol leaves out +/// is loom's deadlock. +#[cfg(feature = "loom")] +pub mod flight_prim { + use std::collections::HashMap; + + pub use loom::sync::atomic::{AtomicU32, Ordering}; + use loom::sync::{Condvar, Mutex}; + + pub struct Lock(Mutex); + + impl Lock { + pub fn new(value: T) -> Self { + Self(Mutex::new(value)) + } + + pub fn lock(&self) -> loom::sync::MutexGuard<'_, T> { + self.0.lock().unwrap() + } + } + + /// Wakes counted by the word's address. + struct Futex { + wakes: Mutex>, + woken: Condvar, + } + + loom::lazy_static! { + static ref FUTEX: Futex = Futex { wakes: Mutex::new(HashMap::new()), woken: Condvar::new() }; + } + + fn address(word: &AtomicU32) -> usize { + word as *const AtomicU32 as usize + } + + pub fn wait(word: &AtomicU32, expected: u32) { + let mut wakes = FUTEX.wakes.lock().unwrap(); + if word.load(Ordering::SeqCst) != expected { + return; + } + let seen = wakes.get(&address(word)).copied().unwrap_or(0); + while wakes.get(&address(word)).copied().unwrap_or(0) == seen { + wakes = FUTEX.woken.wait(wakes).unwrap(); + } + } + + pub fn wake_one(word: &AtomicU32) { + wake_all(word); + } + + pub fn wake_all(word: &AtomicU32) { + *FUTEX.wakes.lock().unwrap().entry(address(word)).or_insert(0) += 1; + FUTEX.woken.notify_all(); + } +} + +/// One file server connection's requests in flight, the SDK's, driven by +/// `tests/fs_flight.rs` over [`flight_prim`]. +#[cfg(feature = "loom")] +#[path = "../../../sdk/toyos/src/fs/flight.rs"] +pub mod fs_flight; diff --git a/kernel/loom/tests/fs_flight.rs b/kernel/loom/tests/fs_flight.rs new file mode 100644 index 00000000000..1c12b64bada --- /dev/null +++ b/kernel/loom/tests/fs_flight.rs @@ -0,0 +1,318 @@ +//! Loom: one file server connection's requests in flight, the SDK's +//! (`sdk/toyos/src/fs/flight.rs`). +//! +//! Callers share connections whose replies whoever waits receives: a reply +//! filed in another caller's slot, the receive turn handed on, a connection +//! that ends under requests in flight, a send that fails beside its own +//! connection's callers and beside a fresh connection, two connections each +//! receiving for itself and handing its turn on within itself, a slot used +//! again, and a request too large for a slot holding every slot. The futex +//! the shim gives wakes only the word woken, and every wire here blocks, so a +//! caller left waiting with nobody receiving or nobody waking it is loom's +//! deadlock. + +#![cfg(feature = "loom")] + +use std::collections::VecDeque; + +use kernel_loom::fs_flight::{Flight, Taken, SLOTS}; +use loom::sync::atomic::{AtomicU32, Ordering}; +use loom::sync::{Arc, Condvar, Mutex}; +use loom::thread; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Why { + Gone, + Malformed, +} + +/// One connection: requests to the server, replies back, and its end, each +/// wait on it blocking. +#[derive(Default)] +struct Wire { + state: Mutex, + moved: Condvar, +} + +#[derive(Default)] +struct Queues { + asked: VecDeque, + answered: VecDeque, + ended: bool, +} + +impl Wire { + fn ask(&self, tag: u64) { + self.state.lock().unwrap().asked.push_back(tag); + self.moved.notify_all(); + } + + /// The next reply, its tag answered as its value, or the end. + fn receive(&self) -> Result<(u64, u64), Why> { + let mut state = self.state.lock().unwrap(); + loop { + if let Some(tag) = state.answered.pop_front() { + return Ok((tag, tag)); + } + if state.ended { + return Err(Why::Gone); + } + state = self.moved.wait(state).unwrap(); + } + } + + /// The next `n` requests, as the server takes them. + fn take(&self, n: usize) -> Vec { + let mut state = self.state.lock().unwrap(); + while state.asked.len() < n { + state = self.moved.wait(state).unwrap(); + } + state.asked.drain(..n).collect() + } + + fn answer(&self, tag: u64) { + self.state.lock().unwrap().answered.push_back(tag); + self.moved.notify_all(); + } + + fn end(&self) { + self.state.lock().unwrap().ended = true; + self.moved.notify_all(); + } +} + +/// The requests in flight, two connections by generation (1 and 2), and the +/// slots callers hold. +struct World { + flight: Flight, + wires: [Wire; 2], + holding: AtomicU32, +} + +impl World { + fn new() -> Arc { + Arc::new(Self { flight: Flight::new(), wires: Default::default(), holding: AtomicU32::new(0) }) + } + + fn wire(&self, link: u64) -> &Wire { + &self.wires[link as usize - 1] + } + + /// Slots for one request, as `Dir`'s ticket takes them: held alone. + fn take(&self, whole: bool) -> Taken { + let taken = self.flight.take(whole); + let mine = if whole { u32::MAX } else { 1 << taken.first() }; + let before = self.holding.fetch_or(mine, Ordering::AcqRel); + assert_eq!(before & mine, 0, "two callers held one slot"); + assert!(!whole || before == 0, "every slot was taken beside a slot held"); + assert!(before != u32::MAX, "a slot was taken beside every slot held"); + taken + } + + fn give(&self, taken: Taken) { + let mine = if taken.count() == SLOTS { u32::MAX } else { 1 << taken.first() }; + self.holding.fetch_and(!mine, Ordering::AcqRel); + self.flight.give(taken); + } + + /// One request on connection `link` and its reply, as `Dir` makes it. + fn ask(&self, link: u64, whole: bool) -> Result { + let taken = self.take(whole); + let tag = self.flight.sent(taken, link); + self.wire(link).ask(tag); + let answered = self.flight.answer(taken, || self.wire(link).receive(), Why::Malformed); + self.give(taken); + if let Ok(reply) = answered { + assert_eq!(reply, tag, "a caller was answered another caller's reply"); + } + answered + } + + /// A request on connection `link` whose send fails: its slot goes back + /// unanswered. + fn unsent(&self, link: u64) { + let taken = self.take(false); + self.flight.sent(taken, link); + self.give(taken); + } +} + +/// Every interleaving of `f` with at most `preemptions` preemptions: 3 for a +/// model of three threads, 2 for one whose third thread asks twice or whose +/// fourth makes it run past minutes. +fn model(preemptions: usize, f: impl Fn() + Sync + Send + 'static) { + let mut builder = loom::model::Builder::new(); + builder.preemption_bound = Some(preemptions); + builder.check(f); +} + +fn spawn(world: &Arc, f: impl FnOnce(&World) -> Result + Send + 'static) -> thread::JoinHandle> { + let world = Arc::clone(world); + thread::spawn(move || f(&world)) +} + +/// Two callers, the server answering the later request first: each is +/// answered its own reply, whichever of them receives. +#[test] +fn each_caller_is_answered_its_own_reply_whoever_receives() { + model(3, || { + let world = World::new(); + let callers: Vec<_> = (0..2).map(|_| spawn(&world, |w| w.ask(1, false))).collect(); + for tag in world.wire(1).take(2).into_iter().rev() { + world.wire(1).answer(tag); + } + for caller in callers { + assert!(caller.join().unwrap().is_ok()); + } + }); +} + +/// One caller asks twice, its second request in the slot it gave back, +/// beside a caller asking once: each request is answered its own reply. +#[test] +fn a_slot_used_again_is_answered_its_own_reply() { + model(2, || { + let world = World::new(); + let twice = spawn(&world, |w| w.ask(1, false).and_then(|_| w.ask(1, false))); + let once = spawn(&world, |w| w.ask(1, false)); + for tag in world.wire(1).take(2).into_iter().rev() { + world.wire(1).answer(tag); + } + let last = world.wire(1).take(1)[0]; + world.wire(1).answer(last); + assert!(twice.join().unwrap().is_ok()); + assert!(once.join().unwrap().is_ok()); + }); +} + +/// The connection ends with a request in flight and another on its way: both +/// fail, and neither waits on. +#[test] +fn a_connection_that_ends_fails_every_request_on_it() { + model(3, || { + let world = World::new(); + let callers: Vec<_> = (0..2).map(|_| spawn(&world, |w| w.ask(1, false))).collect(); + world.wire(1).take(1); + world.wire(1).end(); + for caller in callers { + assert_eq!(caller.join().unwrap(), Err(Why::Gone)); + } + }); +} + +/// A caller on a connection that ends, a caller whose send on it fails, and a +/// caller on a fresh connection, which may take the slot the failed send gave +/// back: the failure of the first connection is filed in no request of the +/// second, which is answered. +#[test] +fn a_connection_that_ends_fails_no_request_on_a_fresh_one() { + model(2, || { + let world = World::new(); + let ended = spawn(&world, |w| w.ask(1, false)); + let unsent = spawn(&world, |w| { + w.unsent(1); + Ok(0) + }); + let fresh = spawn(&world, |w| w.ask(2, false)); + world.wire(1).take(1); + world.wire(1).end(); + let tag = world.wire(2).take(1)[0]; + world.wire(2).answer(tag); + assert_eq!(ended.join().unwrap(), Err(Why::Gone)); + unsent.join().unwrap().unwrap(); + assert!(fresh.join().unwrap().is_ok(), "a fresh connection's request failed with the old one"); + }); +} + +/// Two callers and a send that fails, all on one connection: a turn handed to +/// the failed send goes on to the caller still waiting, which is answered. +#[test] +fn a_failed_send_hands_its_turn_on() { + model(2, || { + let world = World::new(); + let unsent = spawn(&world, |w| { + w.unsent(1); + Ok(0) + }); + let callers: Vec<_> = (0..2).map(|_| spawn(&world, |w| w.ask(1, false))).collect(); + for _ in 0..2 { + let tag = world.wire(1).take(1)[0]; + world.wire(1).answer(tag); + } + unsent.join().unwrap().unwrap(); + for caller in callers { + assert!(caller.join().unwrap().is_ok()); + } + }); +} + +/// Two callers on one connection beside a caller leading another, whose +/// server answers only once the first two have returned: the turn the first +/// hands on goes to the caller waiting on its own connection, never to the +/// other connection's leader. +#[test] +fn a_turn_handed_on_stays_on_its_connection() { + model(2, || { + let world = World::new(); + let other = spawn(&world, |w| w.ask(2, false)); + let callers: Vec<_> = (0..2).map(|_| spawn(&world, |w| w.ask(1, false))).collect(); + for tag in world.wire(1).take(2) { + world.wire(1).answer(tag); + } + for caller in callers { + assert!(caller.join().unwrap().is_ok()); + } + let tag = world.wire(2).take(1)[0]; + world.wire(2).answer(tag); + assert!(other.join().unwrap().is_ok()); + }); +} + +/// A caller on one connection is answered while a caller on another holds +/// that connection's turn, whose server answers only once the first caller +/// has returned: each connection receives for itself, so neither waits on +/// the other. +#[test] +fn each_connection_receives_for_itself() { + model(3, || { + let world = World::new(); + let second = spawn(&world, |w| w.ask(2, false)); + let first = spawn(&world, |w| w.ask(1, false)); + let tag = world.wire(1).take(1)[0]; + world.wire(1).answer(tag); + assert!(first.join().unwrap().is_ok()); + let tag = world.wire(2).take(1)[0]; + world.wire(2).answer(tag); + assert!(second.join().unwrap().is_ok()); + }); +} + +/// A reply whose tag names no request in flight fails the connection's +/// requests rather than answering one. +#[test] +fn a_reply_naming_no_request_fails_the_connection() { + model(3, || { + let world = World::new(); + let caller = spawn(&world, |w| w.ask(1, false)); + let tag = world.wire(1).take(1)[0]; + world.wire(1).answer(tag + SLOTS as u64); + assert_eq!(caller.join().unwrap(), Err(Why::Malformed)); + }); +} + +/// A request holding every slot and one holding a slot, each answered as it +/// comes: neither is held beside the other, and each is answered its own. +#[test] +fn every_slot_is_held_by_one_caller_alone() { + model(3, || { + let world = World::new(); + let whole = spawn(&world, |w| w.ask(1, true)); + let one = spawn(&world, |w| w.ask(1, false)); + for _ in 0..2 { + let tag = world.wire(1).take(1)[0]; + world.wire(1).answer(tag); + } + assert!(whole.join().unwrap().is_ok()); + assert!(one.join().unwrap().is_ok()); + }); +} diff --git a/sdk/toyos/src/failure.rs b/sdk/toyos/src/failure.rs index 708657a3e08..2e111fb3a9a 100644 --- a/sdk/toyos/src/failure.rs +++ b/sdk/toyos/src/failure.rs @@ -154,6 +154,7 @@ impl From for Failure { Error::Unsupported => Self::Unsupported, Error::OutOfResources => Self::OutOfMemory, Error::Malformed => Self::Protocol, + Error::Unanswered => Self::TimedOut, } } } diff --git a/sdk/toyos/src/files/owner.rs b/sdk/toyos/src/files/owner.rs index eab0607fc5e..de7428c4b1c 100644 --- a/sdk/toyos/src/files/owner.rs +++ b/sdk/toyos/src/files/owner.rs @@ -12,24 +12,19 @@ use super::routing::Routed; use super::{encode, At, Error, Open, Stat, MAX_LINKS, SEEK_CURRENT, SEEK_END, SEEK_START}; use crate::fd::{self, Fd, Object, SERVED_FILE}; use crate::fs::{self, Base, Change, Dir, Fid, Refused}; -use crate::sync::{Lock, Turnstile}; +use crate::sync::Lock; -/// One directory capability this process holds, connected: threads sharing -/// it are served in turn. -struct Capability { - dir: Turnstile, -} - -/// Every capability of the view this process has connected: bounded by the -/// view, as only a held directory is asked for. -static CAPABILITIES: Lock>> = Lock::new(BTreeMap::new()); +/// Every directory capability of the view this process has connected, which +/// its threads share: bounded by the view, as only a held directory is asked +/// for. +static CAPABILITIES: Lock>> = Lock::new(BTreeMap::new()); /// A path's capability, starting directory and relative name. -enum Place { Served(Arc, Base, String), Directory(String) } +enum Place { Served(Arc, Base, String), Directory(String) } /// A file or directory held open by its server, with a paged holder where supplied. struct Served { - cap: Arc, + cap: Arc, fid: Fid, position: Position, holder: Option, @@ -42,7 +37,7 @@ enum Held { impl Drop for Served { fn drop(&mut self) { - self.cap.dir.lock().close(self.fid); + self.cap.close(self.fid); } } @@ -93,7 +88,7 @@ impl Position { Self { append, at: Lock::new(at) } } - fn read(&self, dir: &mut Dir, fid: Fid, out: &mut [u8]) -> Result { + fn read(&self, dir: &Dir, fid: Fid, out: &mut [u8]) -> Result { let mut at = self.at.lock(); match *at { Offset::Here(offset) => { @@ -105,9 +100,9 @@ impl Position { } } - /// A read at the local offset through the file's paged object, which - /// takes no turn on the capability: `None` where the offset is the - /// server's or the object would not answer, and the server is asked. + /// A read at the local offset through the file's paged object: `None` + /// where the offset is the server's or the object would not answer, and + /// the server is asked. fn read_held(&self, holder: &crate::paged::Paged, out: &mut [u8]) -> Option> { let mut at = self.at.lock(); let Offset::Here(offset) = *at else { return None }; @@ -137,7 +132,7 @@ impl Position { Some(n) } - fn write(&self, dir: &mut Dir, fid: Fid, data: &[u8]) -> Result { + fn write(&self, dir: &Dir, fid: Fid, data: &[u8]) -> Result { let mut at = self.at.lock(); match *at { Offset::Here(offset) if !self.append => { @@ -157,7 +152,7 @@ impl Position { /// Where a seek lands, by the rule the server keeps for its own offsets: /// at or below [`fs::MAX_FILE_BYTES`] and not before the start. From the /// end, the server is asked: it reads the end as of the writes it orders. - fn seek(&self, dir: &mut Dir, fid: Fid, whence: u64, delta: i64) -> Result { + fn seek(&self, dir: &Dir, fid: Fid, whence: u64, delta: i64) -> Result { let mut at = self.at.lock(); let Offset::Here(offset) = *at else { return dir.seek(fid, whence, delta) }; let landed = match whence { @@ -172,7 +167,7 @@ impl Position { /// Share this description with another process. Once handed over, every /// read, write and seek uses the server's one offset. - fn export(&self, dir: &mut Dir, fid: Fid) -> Result { + fn export(&self, dir: &Dir, fid: Fid) -> Result { let mut at = self.at.lock(); if let Offset::Here(offset) = *at { dir.seek(fid, fs::SEEK_START, offset as i64)?; @@ -184,7 +179,7 @@ impl Position { /// A stream the server writes at the description's offset for another /// process ([`Dir::stream`]): the offset is handed to the server first, /// and is its from then on. - fn stream(&self, dir: &mut Dir, fid: Fid) -> Result { + fn stream(&self, dir: &Dir, fid: Fid) -> Result { let mut at = self.at.lock(); if let Offset::Here(offset) = *at { dir.seek(fid, fs::SEEK_START, offset as i64)?; @@ -240,14 +235,13 @@ fn utf8(path: &[u8]) -> Result<&str, Error> { /// refused connect is the caller's error. The connect waits for the server's /// hello, so it is made outside the lock: of two threads that connect at once /// the first connection kept is the one used. -fn capability(prefix: &str) -> Result, Error> { +fn capability(prefix: &str) -> Result, Error> { if let Some(found) = CAPABILITIES.lock().get(prefix) { return Ok(Arc::clone(found)); } let name = alloc::format!("{}{prefix}", fs::CAPABILITY_PREFIX); let dir = Dir::named(&name).map_err(Error::Served)?; - let found = Arc::new(Capability { dir: Turnstile::new(dir) }); - Ok(Arc::clone(CAPABILITIES.lock().entry(String::from(prefix)).or_insert(found))) + Ok(Arc::clone(CAPABILITIES.lock().entry(String::from(prefix)).or_insert(Arc::new(dir)))) } /// Which server the absolute, lexically clean `abs` is on, and the path there. @@ -287,14 +281,12 @@ fn from_dir(fd: Fd, path: &str) -> Result { } /// Ask whoever serves `path` from `at`, following an absolute link a server -/// meets to wherever it lands in this process's own table. `served` runs on -/// the capability's turn, so nothing it does may reach the descriptor table, -/// whose release of an open file takes that turn again. +/// meets to wherever it lands in this process's own table. fn on_path( at: At, path: &[u8], mut synthetic: impl FnMut(&str) -> Result, - mut served: impl FnMut(&Arc, &mut Dir, Base, &str) -> Result, + mut served: impl FnMut(&Arc, Base, &str) -> Result, ) -> Result { let mut place = place(at, path)?; for _ in 0..MAX_LINKS { @@ -302,17 +294,10 @@ fn on_path( Place::Served(cap, base, rel) => (cap, base, rel), Place::Directory(path) => return synthetic(&path), }; - let mut dir = cap.dir.lock(); - match served(&cap, &mut dir, base, &rel) { + match served(&cap, base, &rel) { Ok(answer) => return Ok(answer), Err(Refused::Error(e)) => return Err(Error::Served(e)), - Err(Refused::Follow(len)) => { - let mut buf = [0u8; fs::MAX_PATH]; - let target = dir.follow_target(len, &mut buf); - let target = super::absolute("/", utf8(target)?); - drop(dir); - place = route(&target)?; - } + Err(Refused::Follow(target)) => place = route(&super::absolute("/", utf8(&target)?))?, } } Err(Error::Loop) @@ -324,7 +309,7 @@ fn on_fd( fd: Fd, kernel: impl FnOnce(crate::BorrowedHandle<'_>) -> Result, synthetic: impl FnOnce(&str) -> Result, - served: impl FnOnce(&mut Dir, Fid, &Position) -> Result, + served: impl FnOnce(&Dir, Fid, &Position) -> Result, ) -> Result { on_fd_held(fd, kernel, synthetic, |_, _| None, served) } @@ -341,15 +326,13 @@ fn held_write(written: Result) -> Option( fd: Fd, kernel: impl FnOnce(crate::BorrowedHandle<'_>) -> Result, synthetic: impl FnOnce(&str) -> Result, holding: impl FnOnce(&crate::paged::Paged, &Position) -> Option>, - served: impl FnOnce(&mut Dir, Fid, &Position) -> Result, + served: impl FnOnce(&Dir, Fid, &Position) -> Result, ) -> Result { // A kernel call is made while the descriptor's object is held; an object // of this module's is taken out of it, counted, and asked after. @@ -366,14 +349,13 @@ fn on_fd_held( Held::Directory(path) => synthetic(path), Held::Served(held) => match held.holder.as_ref().and_then(|h| holding(h, &held.position)) { Some(answer) => answer, - None => served(&mut held.cap.dir.lock(), held.fid, &held.position).map_err(Error::Served), + None => served(&held.cap, held.fid, &held.position).map_err(Error::Served), }, }, } } -/// A descriptor for `held`; on a full table it is let go again, which takes -/// its capability's turn, so no turn may be held here. +/// A descriptor for `held`; on a full table it is let go again. fn install(held: Held, cloexec: bool) -> Result { fd::install(object(held), !cloexec, 0) } @@ -399,11 +381,11 @@ fn object(held: Held) -> Object { /// Take up the one description exported for an endowed file descriptor. /// This constructs an object without entering the descriptor table again. pub(super) fn adopt_exported(connection: crate::ipc::Connection) -> Result { - let mut dir = Dir::adopt(connection).map_err(Error::Served)?; + let dir = Dir::adopt(connection).map_err(Error::Served)?; let fid = Fid { id: 1, generation: dir.generation() }; dir.fstat(fid).map_err(Error::Served)?; Ok(object(Held::Served(Served { - cap: Arc::new(Capability { dir: Turnstile::new(dir) }), + cap: Arc::new(dir), fid, position: Position { append: false, at: Lock::new(Offset::Server) }, holder: None, @@ -414,7 +396,7 @@ pub(super) fn adopt_exported(connection: crate::ipc::Connection) -> Result i64 { let held = held(word).expect("a held descriptor names this file object"); let exported = match &*held { - Held::Served(file) => file.position.export(&mut file.cap.dir.lock(), file.fid).map_err(Error::Served), + Held::Served(file) => file.position.export(&file.cap, file.fid).map_err(Error::Served), Held::Directory(_) => Err(Error::Served(fs::Error::Unsupported)), }; answer(exported.map(|conn| u64::from(conn.into_raw().0))) @@ -469,7 +451,7 @@ unsafe extern "C" fn toyos_fs_open(at: Fd, path: *const u8, len: usize, how: Ope } Ok(Held::Directory(String::from(path))) }, - |cap, dir, base, rel| { + |cap, base, rel| { let mut flags = 0; if how.read { flags |= fs::O_READ; } if how.write { flags |= fs::O_WRITE; } @@ -479,8 +461,8 @@ unsafe extern "C" fn toyos_fs_open(at: Fd, path: *const u8, len: usize, how: Ope if how.truncate { flags |= fs::O_TRUNCATE; } if how.directory { flags |= fs::O_DIRECTORY; } let opened = match how.unnamed { - true => dir.create_unnamed(base, rel, flags)?, - false => dir.open(base, rel, flags)?, + true => cap.create_unnamed(base, rel, flags)?, + false => cap.open(base, rel, flags)?, }; let position = Position::new(how.append, opened.stat.kind == fs::KIND_DIR); // What a reply carries is a holder only as the kernel answers for @@ -489,14 +471,14 @@ unsafe extern "C" fn toyos_fs_open(at: Fd, path: *const u8, len: usize, how: Ope None | Some(Err(crate::paged::NotAHolder::Refused | crate::paged::NotAHolder::Ended)) => None, Some(Ok(paged)) => Some(paged), Some(Err(crate::paged::NotAHolder::Other)) => { - dir.close(opened.fid); + cap.close(opened.fid); return Err(Refused::Error(fs::Error::Malformed)); } }; Ok(Held::Served(Served { cap: Arc::clone(cap), fid: opened.fid, position, holder })) }, ); - // Install outside the capability turn: refusal closes the opened fid. + // A refused install closes the opened fid. let installed = opened.and_then(|held| install(held, how.cloexec).map_err(Error::Fd)); answer(installed.map(|fd| fd as u64)) } @@ -518,7 +500,7 @@ unsafe extern "C" fn toyos_fs_stat(at: Fd, path: *const u8, len: usize, follow: at, path, directory_stat, - |_, dir, base, rel| dir.stat(base, rel, follow).map(stat_of), + |dir, base, rel| dir.stat(base, rel, follow).map(stat_of), ), }; answer(stat.map(|stat| { @@ -645,7 +627,7 @@ extern "C" fn toyos_fs_image(fd: Fd) -> i64 { unsafe extern "C" fn toyos_fs_mkdir(at: Fd, path: *const u8, len: usize) -> i64 { // SAFETY: the caller's. let path = unsafe { bytes(path, len) }; - let made = on_path(At::of(at), path, read_only, |_, dir, base, rel| dir.mkdir(base, rel)); + let made = on_path(At::of(at), path, read_only, |dir, base, rel| dir.mkdir(base, rel)); answer(made.map(|()| 0)) } @@ -657,7 +639,7 @@ unsafe extern "C" fn toyos_fs_unlink(at: Fd, path: *const u8, len: usize, is_dir At::of(at), path, read_only, - |_, dir, base, rel| dir.unlink(base, rel, is_dir), + |dir, base, rel| dir.unlink(base, rel, is_dir), ); answer(removed.map(|()| 0)) } @@ -671,7 +653,7 @@ unsafe extern "C" fn toyos_fs_chdir(path: *const u8, len: usize) -> i64 { if abs.len() > fs::MAX_PATH { return Err(Error::Served(fs::Error::NameTooLong)); } - let stat = on_path(At::Cwd, abs.as_bytes(), directory_stat, |_, dir, base, rel| dir.stat(base, rel, true).map(stat_of))?; + let stat = on_path(At::Cwd, abs.as_bytes(), directory_stat, |dir, base, rel| dir.stat(base, rel, true).map(stat_of))?; if stat.file_type != FileType::Directory { return Err(Error::NotADirectory); } @@ -682,7 +664,7 @@ unsafe extern "C" fn toyos_fs_chdir(path: *const u8, len: usize) -> i64 { } /// Two paths on one capability, with their starting directories. -struct Pair(Arc, (Base, String), (Base, String)); +struct Pair(Arc, (Base, String), (Base, String)); fn pair(from: (At, &[u8]), to: (At, &[u8])) -> Result { pair_of(place(from.0, from.1)?, place(to.0, to.1)?) @@ -709,7 +691,7 @@ unsafe extern "C" fn toyos_fs_rename(from_at: Fd, from: *const u8, from_len: usi // SAFETY: the caller's. let (from, to) = unsafe { (bytes(from, from_len), bytes(to, to_len)) }; let renamed = pair((At::of(from_at), from), (At::of(to_at), to)).and_then(|pair| match pair { - Pair(cap, (ba, a), (bb, b)) => cap.dir.lock().rename((ba, &a), (bb, &b)).map_err(crossing), + Pair(cap, (ba, a), (bb, b)) => cap.rename((ba, &a), (bb, &b)).map_err(crossing), }); answer(renamed.map(|()| 0)) } @@ -732,7 +714,7 @@ unsafe extern "C" fn toyos_fs_link( (at, _) => place(at, from), }; let linked = from_place.and_then(|from| pair_of(from, place(At::of(to_at), to)?)).and_then(|pair| match pair { - Pair(cap, (ba, a), (bb, b)) => cap.dir.lock().link((ba, &a), follow, (bb, &b), None).map_err(crossing), + Pair(cap, (ba, a), (bb, b)) => cap.link((ba, &a), follow, (bb, &b), None).map_err(crossing), }); answer(linked.map(|()| 0)) } @@ -742,7 +724,7 @@ unsafe extern "C" fn toyos_fs_symlink(target: *const u8, target_len: usize, at: // SAFETY: the caller's. let (target, link) = unsafe { (bytes(target, target_len), bytes(link, link_len)) }; let made = utf8(target).and_then(|target| { - on_path(At::of(at), link, read_only, |_, dir, base, rel| { + on_path(At::of(at), link, read_only, |dir, base, rel| { dir.symlink(target, (base, rel)) }) }); @@ -756,7 +738,7 @@ unsafe extern "C" fn toyos_fs_readlink(at: Fd, path: *const u8, len: usize, out: // SAFETY: the caller lends `cap` writable bytes for the call, and each // closure makes the slice only for its own run. let target = || unsafe { core::slice::from_raw_parts_mut(out, cap) }; - let n = on_path(At::of(at), path, |_| Err(Error::Served(fs::Error::NotALink)), |_, dir, base, rel| { + let n = on_path(At::of(at), path, |_| Err(Error::Served(fs::Error::NotALink)), |dir, base, rel| { let mut buf = [0u8; fs::MAX_PATH]; let n = dir.read_link(base, rel, &mut buf)?; let out = target(); @@ -777,10 +759,10 @@ unsafe extern "C" fn toyos_fs_set_times(at: Fd, path: *const u8, len: usize, fol }; let set = match (At::of(at), path.is_empty()) { (At::Fd(fd), true) => from_dir(fd, "").and_then(|place| match place { - Place::Served(cap, base, rel) => cap.dir.lock().set_attr(base, &rel, follow, &change).map_err(crossing), + Place::Served(cap, base, rel) => cap.set_attr(base, &rel, follow, &change).map_err(crossing), Place::Directory(_) => Err(Error::Served(fs::Error::ReadOnly)), }), - (at, _) => on_path(at, path, read_only, |_, dir, base, rel| dir.set_attr(base, rel, follow, &change)), + (at, _) => on_path(at, path, read_only, |dir, base, rel| dir.set_attr(base, rel, follow, &change)), }; answer(set.map(|()| 0)) } diff --git a/sdk/toyos/src/fs.rs b/sdk/toyos/src/fs.rs index f1d851c5216..4f1503bfeb5 100644 --- a/sdk/toyos/src/fs.rs +++ b/sdk/toyos/src/fs.rs @@ -42,21 +42,38 @@ //! the copy. //! //! **A server that ends is survivable and not invisible.** [`Dir`] connects -//! again through the same connector — the supervisor keeps a file server's ports open -//! across its restart — and counts its connections in a generation. A file id -//! from an earlier generation names nothing on the new connection. Nothing -//! here keeps a write the server acknowledged and never made durable: that is -//! what `Fsync` is for. - +//! again through the same connector — the supervisor keeps a file server's +//! ports open across its restart — lends the new connection a window of its +//! own, and counts its connections in a generation. A file id from an earlier +//! generation names nothing on the new connection. A server that answers +//! nothing for [`REPLY_BOUND_NS`] while a reply is owed is let go as one that +//! ended. Nothing here keeps a write the server acknowledged and never made +//! durable: that is what `Fsync` is for. + +use alloc::sync::Arc; use alloc::vec::Vec; +use core::sync::atomic::{AtomicU64, Ordering}; use toyos_abi::syscall::{SyscallError, MAX_BADGE}; use crate::ipc::{Connection, FrameRx, IpcError}; use crate::ipc_payload; use crate::shm::{Bytes, SharedMemory}; +use crate::sync::Lock; use crate::Pipe; +mod flight; + +use flight::{Flight, Taken}; + +/// What [`flight`] is built on: the SDK's futex and lock, which `kernel-loom` +/// swaps for loom's. +mod flight_prim { + pub use core::sync::atomic::{AtomicU32, Ordering}; + + pub(crate) use crate::sync::{wait, wake_all, wake_one, Lock}; +} + /// What a directory capability's service name begins with. pub const CAPABILITY_PREFIX: &str = "fs:"; @@ -362,6 +379,9 @@ errors! { NotHeld = 0x81, /// The server answered something this protocol does not say. Malformed = 0x82, + /// The server answered nothing within [`REPLY_BOUND_NS`] while a reply + /// was owed: the connection is let go, as one that ended. + Unanswered = 0x83, } } @@ -729,12 +749,12 @@ impl Change { } /// Why a request on a path was not answered with its result. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq)] pub enum Refused { Error(Error), - /// The path met an absolute symlink; [`Dir::follow_target`] holds the - /// absolute path, `.0` bytes long, to resolve in the client's own table. - Follow(usize), + /// The path met an absolute symlink: the absolute path to resolve in the + /// client's own table, as the server wrote it. + Follow(Vec), } impl From for Refused { @@ -751,31 +771,72 @@ enum Source { Exported, } -/// One directory capability, or one exported connection, connected. It keeps -/// one request outstanding at a time, its span the window's start. +/// The window's share one request takes, unless it needs more, when it takes +/// the whole window. +const SLOT_BYTES: usize = WINDOW_BYTES / MAX_OUTSTANDING; + +const _: () = assert!(flight::SLOTS == MAX_OUTSTANDING); + +/// A reply as it came: its frame type, its words and the handles it carried. +type Answered = (u32, Reply, crate::ipc::Batch); + +/// One directory capability, or one exported connection, connected. Its +/// threads' requests are in flight on one connection at once, each in its +/// own share of the window ([`flight`]). pub struct Dir { source: Source, - conn: Option, + /// The connection requests go out on: `None` once it ended, until the + /// next request connects again. + link: Lock>>, + /// Held while a connection is made, so two callers make one. + connecting: Lock<()>, + /// Connections made: the last one's generation. + generation: AtomicU64, + flight: Flight, +} + +/// One connection and the window it was lent, which requests in flight on it +/// keep after a newer one replaced it: a server still writing a request it was +/// let go with writes this window, never a newer connection's. +struct Link { + conn: Connection, window: SharedMemory, generation: u64, + /// Whether its grant lets this process change what it names. writable: bool, - /// The tag the next request carries. - tag: u64, } -/// A path request's two paths: the second is a path, or a symlink's target, -/// which is stored as written and so held to no rule but its length. +/// A request's share of the window, on one connection: given back when it +/// drops. +struct Ticket<'a> { + dir: &'a Dir, + link: Arc, + taken: Taken, +} + +/// What a path request names: two paths from `bases`, the second a path or a +/// symlink's target, which is stored as written and so held to no rule but +/// its length; and the open file a replacing link names. struct Paths<'a> { + bases: [Base; 2], first: &'a str, second: &'a [u8], + expected: Option, +} + +impl<'a> Paths<'a> { + /// One path from `base`. + fn one(base: Base, first: &'a str) -> Self { + Self { bases: [base, Base::Root], first, second: &[], expected: None } + } } impl Dir { /// Connect to the directory capability `connector`, which this process /// holds for its life: [`Error::NotHeld`] for one it may only pass on. pub fn connect(connector: crate::BorrowedHandle<'static>) -> Result { - let mut dir = Self::unconnected(Source::Held(connector))?; - dir.reconnect()?; + let dir = Self::unconnected(Source::Held(connector)); + dir.connection()?; Ok(dir) } @@ -789,36 +850,46 @@ impl Dir { /// 1.. in the order exported. It names no path, and is not connected again /// once it has ended. pub fn adopt(conn: Connection) -> Result { - let mut dir = Self::unconnected(Source::Exported)?; - dir.attach(conn)?; + let dir = Self::unconnected(Source::Exported); + let link = dir.attach(conn)?; + *dir.link.lock() = Some(Arc::new(link)); Ok(dir) } - fn unconnected(source: Source) -> Result { - let window = SharedMemory::create(WINDOW_BYTES).map_err(|e| Error::kernel("a window", e))?; - Ok(Self { source, conn: None, window, generation: 0, writable: false, tag: 0 }) + fn unconnected(source: Source) -> Self { + Self { source, link: Lock::new(None), connecting: Lock::new(()), generation: AtomicU64::new(0), flight: Flight::new() } } /// Counts this directory's connections; a file id is good only on the one /// it was answered on. pub fn generation(&self) -> u64 { - self.generation + self.generation.load(Ordering::Relaxed) } - /// Whether the capability lets this process change what it names. + /// Whether the capability lets this process change what it names, as its + /// connection was answered: `false` while none is open. pub fn writable(&self) -> bool { - self.writable + self.link.lock().as_ref().is_some_and(|link| link.writable) } - fn window(&self) -> Bytes<'_> { - self.window.bytes() + /// The connection to send on, made where the last one ended. + fn connection(&self) -> Result, Error> { + if let Some(link) = &*self.link.lock() { + return Ok(Arc::clone(link)); + } + let _connecting = self.connecting.lock(); + if let Some(link) = &*self.link.lock() { + return Ok(Arc::clone(link)); + } + let link = Arc::new(self.reconnect()?); + *self.link.lock() = Some(Arc::clone(&link)); + Ok(link) } - /// Open a fresh connection through the same connector, and lend it the - /// window again. The old one's file ids name nothing from here on; an - /// exported connection has no connector, and stays gone. - fn reconnect(&mut self) -> Result<(), Error> { - self.conn = None; + /// A fresh connection through the same connector, lent a window of its + /// own. The old one's file ids name nothing on it; an exported connection + /// has no connector, and stays gone. + fn reconnect(&self) -> Result { let Source::Held(connector) = self.source else { return Err(Error::Gone) }; let conn = match crate::port::connect(connector) { Ok(conn) => conn, @@ -828,110 +899,67 @@ impl Dir { self.attach(conn) } - /// Lend `conn` the window and serve on it. - fn attach(&mut self, conn: Connection) -> Result<(), Error> { - let reply = hello(&conn, &self.window)?; - self.writable = reply.value & RIGHT_WRITE != 0; - self.conn = Some(conn); - self.generation += 1; - Ok(()) - } - - /// The one handle a reply that answers with one carries; an error carries - /// none, and any other count ends the connection. - fn one(&mut self, msg_type: u32, reply: Reply, carried: crate::ipc::Batch) -> Result { - let n = carried.len(); - match (answer(msg_type, reply), carried.exact::<1>()) { - (Ok(_), Some([handle])) => Ok(handle), - (Err(e), None) if n == 0 => Err(e), - _ => { - self.conn = None; - Err(Error::Malformed) - } - } - } - - /// One request and its reply on the current connection. A connection that - /// has ended answers `Gone` and is dropped, so the next call reconnects. - fn call(&mut self, op: u32, request: Request) -> Result<(u32, Reply), Error> { - let (msg_type, reply, carried) = self.call_with(op, request)?; - if !carried.is_empty() { - // A reply that brings handles nobody asked for speaks another protocol. - self.conn = None; - return Err(Error::Malformed); + /// Lend `conn` a window of its own: the connection to serve on. + fn attach(&self, conn: Connection) -> Result { + let window = SharedMemory::create(WINDOW_BYTES).map_err(|e| Error::kernel("a window", e))?; + let reply = hello(&conn, &window)?; + let generation = self.generation.fetch_add(1, Ordering::Relaxed) + 1; + Ok(Link { conn, window, generation, writable: reply.value & RIGHT_WRITE != 0 }) + } + + /// No request goes out on `link` again: one that ended, or whose server + /// speaks another protocol. Requests in flight on it are answered or fail + /// with it, and the next request connects again. + fn retire(&self, link: &Arc) { + let mut current = self.link.lock(); + if current.as_ref().is_some_and(|now| Arc::ptr_eq(now, link)) { + *current = None; } - Ok((msg_type, reply)) } - /// [`Self::call`], and the handles the reply frame carries into `batch`. - /// A reply to another tag than the request's is [`Error::Malformed`]. - fn call_with( - &mut self, - op: u32, - mut request: Request, - ) -> Result<(u32, Reply, crate::ipc::Batch), Error> { - if self.conn.is_none() { - self.reconnect()?; - } - self.tag = self.tag.wrapping_add(1); - request.tag = self.tag; - let conn = self.conn.as_ref().expect("connected above"); - let mut rx = FrameRx::<{ core::mem::size_of::() }>::new(); - let answered = conn.send(op, &request).map_err(transport).and_then(|()| { - let frame = rx.recv(conn).map_err(transport)?; - let reply: Reply = crate::ipc::decode_payload(rx.payload(frame.payload_len)).map_err(transport)?; - if reply.tag != request.tag { - return Err(Error::Malformed); + /// A share of the window for a request of `span` bytes, on the current + /// connection: one slot, or the whole window for more than a slot holds. + /// The slot comes first: a request that failed retired its connection + /// before it gave its slot back, so the slot's next request goes out on a + /// newer connection and window than the one a let-go server may still + /// write. + fn ticket(&self, span: usize) -> Result, Error> { + let taken = self.flight.take(span > SLOT_BYTES); + match self.connection() { + Ok(link) => Ok(Ticket { dir: self, link, taken }), + Err(e) => { + self.flight.give(taken); + Err(e) } - Ok((frame.msg_type, reply, rx.take_batch())) - }); - if answered.is_err() { - self.conn = None; } - answered } - /// The file id `base` names on the current connection, and whether a - /// request from it means the same on a fresh one: [`Error::Stale`] for an - /// id of an earlier connection. - fn base(&self, base: Base) -> Result<(u64, bool), Error> { - match base { - Base::Root => Ok((ROOT, true)), - Base::Fid(fid) => self.held(fid).map(|id| (id, false)), - } - } - - /// `fid`'s id on the current connection, or [`Error::Stale`]. - fn held(&self, fid: Fid) -> Result { - match fid.generation == self.generation && self.conn.is_some() { - true => Ok(fid.id), - false => Err(Error::Stale), - } + /// A call on a file id, with `span` bytes of its share of the window. + fn fid_call(&self, op: u32, fid: Fid, request: Request, span: usize) -> Result { + self.ticket(span)?.fid_call(op, fid, request, span) } /// A request naming paths, from `bases`: put them in the span first, which /// is `room` bytes at least for what the answer puts there. A server that /// has gone is asked again once on a fresh connection when every base is /// the root, since a path from it means the same thing on both. - fn path_call(&mut self, op: u32, request: Request, bases: [Base; 2], paths: Paths<'_>, room: usize) -> Result { - let (reply, carried) = self.path_call_with(op, request, bases, paths, room)?; - if !carried.is_empty() { - // A reply that brings handles nobody asked for speaks another protocol. - self.conn = None; - return Err(Refused::Error(Error::Malformed)); - } - Ok(reply) + fn path_call(&self, op: u32, request: Request, paths: Paths<'_>, room: usize) -> Result { + self.path_call_then(op, request, paths, room, |ticket, reply, carried| { + ticket.none_carried(carried).map_err(Refused::Error)?; + Ok(reply) + }) } - /// [`Self::path_call`], and the handles the reply carries. - fn path_call_with( - &mut self, + /// [`Self::path_call`], and `then` on its reply and the handles it + /// carried, while the request still holds its share of the window. + fn path_call_then( + &self, op: u32, mut request: Request, - bases: [Base; 2], paths: Paths<'_>, room: usize, - ) -> Result<(Reply, crate::ipc::Batch), Refused> { + then: impl FnOnce(&Ticket<'_>, Reply, crate::ipc::Batch) -> Result, + ) -> Result { if !relative(paths.first) { return Err(Refused::Error(Error::InvalidPath)); } @@ -939,93 +967,85 @@ impl Dir { let span = (paths.first.len() + paths.second.len()).max(room).clamp(MAX_PATH, WINDOW_BYTES); let mut attempt = 0; loop { - let (fid, again) = self.base(bases[0])?; - let (fid2, again2) = self.base(bases[1])?; - self.window().write(0, paths.first.as_bytes()); - self.window().write(paths.first.len(), paths.second); + let ticket = self.ticket(span)?; + let (fid, again) = ticket.base(paths.bases[0])?; + let (fid2, again2) = ticket.base(paths.bases[1])?; + if let Some(expected) = paths.expected { + request.expected = ticket.held(expected)?; + } + ticket.window().write(0, paths.first.as_bytes()); + ticket.window().write(paths.first.len(), paths.second); request.fid = fid; request.fid2 = fid2; - request.at = 0; - request.span = span as u64; request.len = paths.first.len() as u64; request.len2 = paths.second.len() as u64; - match self.call_with(op, request) { - Ok((msg_type, reply, carried)) => return path_answer(msg_type, reply, span).map(|reply| (reply, carried)), - Err(Error::Gone) if attempt == 0 && again && again2 => { - attempt += 1; - self.reconnect()?; - } + match ticket.call_with(op, request, span) { + Ok((msg_type, reply, carried)) => match path_answer(msg_type, reply, span)? { + PathAnswer::Reply(reply) => return then(&ticket, reply, carried), + PathAnswer::Follow(len) => { + ticket.none_carried(carried)?; + let mut target = alloc::vec![0; len]; + ticket.window().read(0, &mut target); + return Err(Refused::Follow(target)); + } + }, + // The ticket drops, and the next takes a fresh connection. + Err(Error::Gone) if attempt == 0 && again && again2 => attempt += 1, Err(e) => return Err(Refused::Error(e)), } } } - /// The absolute path a [`Refused::Follow`] of `len` bytes left in the - /// window. - pub fn follow_target<'b>(&self, len: usize, buf: &'b mut [u8; MAX_PATH]) -> &'b [u8] { - self.window().read(0, &mut buf[..len]); - &buf[..len] - } - - /// A call on a file id, with `span` bytes of the window from its start. - fn fid_call(&mut self, op: u32, fid: Fid, mut request: Request, span: usize) -> Result { - request.fid = self.held(fid)?; - request.at = 0; - request.span = span as u64; - let (msg_type, reply) = self.call(op, request)?; - answer(msg_type, reply) - } - - pub fn open(&mut self, base: Base, path: &str, flags: u64) -> Result { + pub fn open(&self, base: Base, path: &str, flags: u64) -> Result { self.opened(OPEN, base, path, flags) } /// A file with no name in the directory `path` names, opened by `flags` /// as [`Self::open`] opens: [`Self::link`] gives it one. - pub fn create_unnamed(&mut self, base: Base, path: &str, flags: u64) -> Result { + pub fn create_unnamed(&self, base: Base, path: &str, flags: u64) -> Result { self.opened(CREATE_UNNAMED, base, path, flags) } - fn opened(&mut self, op: u32, base: Base, path: &str, flags: u64) -> Result { - let paths = Paths { first: path, second: &[] }; - let (reply, mut carried) = self.path_call_with(op, Request { flags, ..Request::new() }, [base, Base::Root], paths, 0)?; - let fid = Fid { id: reply.value, generation: self.generation }; - let holder = carried.next().map(crate::OwnedHandle::from); - if !carried.is_empty() { - self.close(fid); - self.conn = None; - return Err(Refused::Error(Error::Malformed)); - } - Ok(Opened { fid, stat: stat_of(&reply), holder }) + fn opened(&self, op: u32, base: Base, path: &str, flags: u64) -> Result { + self.path_call_then(op, Request { flags, ..Request::new() }, Paths::one(base, path), 0, |ticket, reply, mut carried| { + let fid = Fid { id: reply.value, generation: ticket.link.generation }; + let holder = carried.next().map(crate::OwnedHandle::from); + if !carried.is_empty() { + self.close(fid); + self.retire(&ticket.link); + return Err(Refused::Error(Error::Malformed)); + } + Ok(Opened { fid, stat: stat_of(&reply), holder }) + }) } /// Let `fid` go, without waiting for the server: [`CLOSE`] is answered - /// with nothing. - pub fn close(&mut self, fid: Fid) { - let Ok(id) = self.held(fid) else { return }; - let conn = self.conn.as_ref().expect("an id is held on a connection"); - if conn.send(CLOSE, &Request { fid: id, ..Request::new() }).is_err() { - self.conn = None; + /// with nothing, and so holds no share of the window. + pub fn close(&self, fid: Fid) { + let Some(link) = self.link.lock().clone() else { return }; + if fid.generation == link.generation && link.conn.send(CLOSE, &Request { fid: fid.id, ..Request::new() }).is_err() { + self.retire(&link); } } /// Read at most `out.len()` bytes, and at most the window, at the /// description's offset, which moves past them. - pub fn read(&mut self, fid: Fid, out: &mut [u8]) -> Result { + pub fn read(&self, fid: Fid, out: &mut [u8]) -> Result { self.read_with(READ, fid, 0, out) } /// Read at most `out.len()` bytes, and at most the window, at `offset`, /// leaving the description's offset where it is. - pub fn pread(&mut self, fid: Fid, offset: u64, out: &mut [u8]) -> Result { + pub fn pread(&self, fid: Fid, offset: u64, out: &mut [u8]) -> Result { self.read_with(PREAD, fid, offset, out) } - fn read_with(&mut self, op: u32, fid: Fid, offset: u64, out: &mut [u8]) -> Result { + fn read_with(&self, op: u32, fid: Fid, offset: u64, out: &mut [u8]) -> Result { let len = out.len().min(WINDOW_BYTES); - let reply = self.fid_call(op, fid, Request { offset, ..Request::new() }, len)?; + let ticket = self.ticket(len)?; + let reply = ticket.fid_call(op, fid, Request { offset, ..Request::new() }, len)?; let n = counted(reply.value, len)?; - self.window().read(0, &mut out[..n]); + ticket.window().read(0, &mut out[..n]); Ok(n) } @@ -1033,32 +1053,34 @@ impl Dir { /// the end for a file opened to append; the offset moves past it. Answers /// how much, and the offset after it, which is never past /// [`MAX_FILE_BYTES`]. - pub fn write(&mut self, fid: Fid, data: &[u8]) -> Result { + pub fn write(&self, fid: Fid, data: &[u8]) -> Result { let len = data.len().min(WINDOW_BYTES); - self.window().write(0, &data[..len]); - let reply = self.fid_call(WRITE, fid, Request::new(), len)?; + let ticket = self.ticket(len)?; + ticket.window().write(0, &data[..len]); + let reply = ticket.fid_call(WRITE, fid, Request::new(), len)?; let offset = Some(reply.value2).filter(|&at| at <= MAX_FILE_BYTES).ok_or(Error::Malformed)?; Ok(Written { len: counted(reply.value, len)?, offset }) } /// Write at most the window of `data` at `offset`, leaving the /// description's offset where it is: how much. - pub fn pwrite(&mut self, fid: Fid, offset: u64, data: &[u8]) -> Result { + pub fn pwrite(&self, fid: Fid, offset: u64, data: &[u8]) -> Result { let len = data.len().min(WINDOW_BYTES); - self.window().write(0, &data[..len]); - let reply = self.fid_call(PWRITE, fid, Request { offset, ..Request::new() }, len)?; + let ticket = self.ticket(len)?; + ticket.window().write(0, &data[..len]); + let reply = ticket.fid_call(PWRITE, fid, Request { offset, ..Request::new() }, len)?; counted(reply.value, len) } /// Move the description's offset `delta` from `whence` ([`SEEK_START`], /// [`SEEK_CURRENT`] or [`SEEK_END`]): where it landed. - pub fn seek(&mut self, fid: Fid, whence: u64, delta: i64) -> Result { + pub fn seek(&self, fid: Fid, whence: u64, delta: i64) -> Result { let request = Request { offset: delta as u64, flags: whence, ..Request::new() }; self.fid_call(SEEK, fid, request, 0).map(|r| r.value) } /// A second id on `fid`'s description, sharing its offset. - pub fn dup(&mut self, fid: Fid) -> Result { + pub fn dup(&self, fid: Fid) -> Result { let reply = self.fid_call(DUP, fid, Request::new(), 0)?; Ok(Fid { id: reply.value, generation: fid.generation }) } @@ -1066,66 +1088,61 @@ impl Dir { /// A connection for another process holding the descriptions `fids` name, /// as ids 1.. in this order ([`Dir::adopt`]). It spends a served slot of /// this connection's share for as long as anybody holds it. - pub fn export(&mut self, fids: &[Fid]) -> Result { + pub fn export(&self, fids: &[Fid]) -> Result { if fids.is_empty() || fids.len() * 8 > WINDOW_BYTES { return Err(Error::InvalidRequest); } + let span = fids.len() * 8; + let ticket = self.ticket(span)?; for (i, fid) in fids.iter().enumerate() { - self.window().write(i * 8, &self.held(*fid)?.to_le_bytes()); + ticket.window().write(i * 8, &ticket.held(*fid)?.to_le_bytes()); } - let request = Request { len: fids.len() as u64, span: (fids.len() * 8) as u64, ..Request::new() }; - let (msg_type, reply, carried) = self.call_with(EXPORT, request)?; - self.one(msg_type, reply, carried)?.typed().map_err(|_| Error::Malformed) + let request = Request { len: fids.len() as u64, ..Request::new() }; + let (msg_type, reply, carried) = ticket.call_with(EXPORT, request, span)?; + ticket.one(msg_type, reply, carried)?.typed().map_err(|_| Error::Malformed) } /// What `path` names from `base`: the link itself where `follow` is /// false. - pub fn stat(&mut self, base: Base, path: &str, follow: bool) -> Result { + pub fn stat(&self, base: Base, path: &str, follow: bool) -> Result { let flags = if follow { FOLLOW_LAST } else { 0 }; - let reply = self.path_call(STAT, Request { flags, ..Request::new() }, [base, Base::Root], Paths { first: path, second: &[] }, 0)?; + let reply = self.path_call(STAT, Request { flags, ..Request::new() }, Paths::one(base, path), 0)?; Ok(stat_of(&reply)) } /// What the file or directory `fid` names is. - pub fn fstat(&mut self, fid: Fid) -> Result { - self.stat(Base::Fid(fid), "", false).map_err(|refused| match refused { - Refused::Error(e) => e, - // An id names one node, which nothing follows from. - Refused::Follow(_) => Error::Malformed, - }) + pub fn fstat(&self, fid: Fid) -> Result { + self.stat(Base::Fid(fid), "", false).map_err(unfollowed) } /// Make `change` to what `path` names from `base`, following a link in /// its last component where `follow`. A size is changed only through an /// id opened to write, with the empty path. - pub fn set_attr(&mut self, base: Base, path: &str, follow: bool, change: &Change) -> Result<(), Refused> { + pub fn set_attr(&self, base: Base, path: &str, follow: bool, change: &Change) -> Result<(), Refused> { let mut request = Request { flags: if follow { FOLLOW_LAST } else { 0 }, ..Request::new() }; change.encode(&mut request); - self.path_call(SETATTR, request, [base, Base::Root], Paths { first: path, second: &[] }, 0).map(drop) + self.path_call(SETATTR, request, Paths::one(base, path), 0).map(drop) } /// Set the size of the file `fid` opened to write. - pub fn truncate(&mut self, fid: Fid, size: u64) -> Result<(), Error> { + pub fn truncate(&self, fid: Fid, size: u64) -> Result<(), Error> { let change = Change { size: Some(size), ..Change::default() }; - self.set_attr(Base::Fid(fid), "", false, &change).map_err(|refused| match refused { - Refused::Error(e) => e, - Refused::Follow(_) => Error::Malformed, - }) + self.set_attr(Base::Fid(fid), "", false, &change).map_err(unfollowed) } /// Make what this file was written durable on the volume. - pub fn fsync(&mut self, fid: Fid) -> Result<(), Error> { + pub fn fsync(&self, fid: Fid) -> Result<(), Error> { self.fid_call(FSYNC, fid, Request::new(), 0).map(drop) } /// Make the whole volume durable. It names no path, and so is asked again /// once on a fresh connection. - pub fn sync(&mut self) -> Result<(), Error> { + pub fn sync(&self) -> Result<(), Error> { self.unpathed(SYNC).map(drop) } /// The volume's [`Room`]. Like [`Self::sync`], it names no path. - pub fn room(&mut self) -> Result { + pub fn room(&self) -> Result { let reply = self.unpathed(ROOM)?; match reply.value2 { 0 => Err(Error::Malformed), @@ -1133,28 +1150,27 @@ impl Dir { } } - fn unpathed(&mut self, op: u32) -> Result { - self.path_call(op, Request::new(), [Base::Root; 2], Paths { first: "", second: &[] }, 0).map_err(|refused| match refused { - Refused::Error(e) => e, - Refused::Follow(_) => Error::Malformed, - }) + fn unpathed(&self, op: u32) -> Result { + self.path_call(op, Request::new(), Paths::one(Base::Root, ""), 0).map_err(unfollowed) } /// A pipe whose bytes the server writes to this file at the description's /// offset, which they move, for a program's standard output: the write end. - pub fn stream(&mut self, fid: Fid) -> Result { - let mut request = Request::new(); - request.fid = self.held(fid)?; - let (msg_type, reply, carried) = self.call_with(STREAM, request)?; - self.one(msg_type, reply, carried)?.typed().map_err(|_| Error::Malformed) + pub fn stream(&self, fid: Fid) -> Result { + self.handle_call(STREAM, fid)?.typed().map_err(|_| Error::Malformed) } /// The bytes of the file `fid` opened, as [`IMAGE`] answers them. - pub fn image(&mut self, fid: Fid) -> Result { - let mut request = Request::new(); - request.fid = self.held(fid)?; - let (msg_type, reply, carried) = self.call_with(IMAGE, request)?; - self.one(msg_type, reply, carried).map(crate::OwnedHandle::from) + pub fn image(&self, fid: Fid) -> Result { + self.handle_call(IMAGE, fid).map(crate::OwnedHandle::from) + } + + /// A request on `fid` answered with one handle. + fn handle_call(&self, op: u32, fid: Fid) -> Result { + let ticket = self.ticket(0)?; + let request = Request { fid: ticket.held(fid)?, ..Request::new() }; + let (msg_type, reply, carried) = ticket.call_with(op, request, 0)?; + ticket.one(msg_type, reply, carried) } /// The entries of the directory `dir` opened, from `cursor` (0 at its @@ -1162,78 +1178,193 @@ impl Dir { /// most the window of them, into `out`: the cursor to ask from next, /// `None` after the last. A first entry larger than `room` is refused /// [`Error::LimitExceeded`]. - pub fn read_dir(&mut self, dir: Fid, cursor: u64, room: usize, out: &mut Vec) -> Result, Error> { + pub fn read_dir(&self, dir: Fid, cursor: u64, room: usize, out: &mut Vec) -> Result, Error> { let room = room.min(WINDOW_BYTES); - let reply = self.fid_call(READDIR, dir, Request { offset: cursor, ..Request::new() }, room)?; + let ticket = self.ticket(room)?; + let reply = ticket.fid_call(READDIR, dir, Request { offset: cursor, ..Request::new() }, room)?; let n = counted(reply.value, room)?; let next = next_batch(n, reply.value2)?; out.clear(); out.resize(n, 0); - self.window().read(0, out); + ticket.window().read(0, out); Ok(next) } - pub fn mkdir(&mut self, base: Base, path: &str) -> Result<(), Refused> { - self.path_call(MKDIR, Request::new(), [base, Base::Root], Paths { first: path, second: &[] }, 0).map(drop) + pub fn mkdir(&self, base: Base, path: &str) -> Result<(), Refused> { + self.path_call(MKDIR, Request::new(), Paths::one(base, path), 0).map(drop) } /// Remove the name `path`: a directory's only where `dir`, which removes /// nothing else. - pub fn unlink(&mut self, base: Base, path: &str, dir: bool) -> Result<(), Refused> { + pub fn unlink(&self, base: Base, path: &str, dir: bool) -> Result<(), Refused> { let flags = if dir { UNLINK_DIR } else { 0 }; - self.path_call(UNLINK, Request { flags, ..Request::new() }, [base, Base::Root], Paths { first: path, second: &[] }, 0).map(drop) + self.path_call(UNLINK, Request { flags, ..Request::new() }, Paths::one(base, path), 0).map(drop) } - pub fn rename(&mut self, from: (Base, &str), to: (Base, &str)) -> Result<(), Refused> { + pub fn rename(&self, from: (Base, &str), to: (Base, &str)) -> Result<(), Refused> { if !relative(to.1) { return Err(Refused::Error(Error::InvalidPath)); } - self.path_call(RENAME, Request::new(), [from.0, to.0], Paths { first: from.1, second: to.1.as_bytes() }, 0).map(drop) + let paths = Paths { bases: [from.0, to.0], first: from.1, second: to.1.as_bytes(), expected: None }; + self.path_call(RENAME, Request::new(), paths, 0).map(drop) } /// Make `link` a symlink holding `target`, which is stored as written and /// resolved by whoever follows it. - pub fn symlink(&mut self, target: &str, link: (Base, &str)) -> Result<(), Refused> { + pub fn symlink(&self, target: &str, link: (Base, &str)) -> Result<(), Refused> { if target.is_empty() { return Err(Refused::Error(Error::InvalidRequest)); } if target.len() > MAX_PATH { return Err(Refused::Error(Error::NameTooLong)); } - self.path_call(SYMLINK, Request::new(), [link.0, Base::Root], Paths { first: link.1, second: target.as_bytes() }, 0).map(drop) + let paths = Paths { second: target.as_bytes(), ..Paths::one(link.0, link.1) }; + self.path_call(SYMLINK, Request::new(), paths, 0).map(drop) } /// What the symlink at `path` holds, into `out`. - pub fn read_link(&mut self, base: Base, path: &str, out: &mut [u8; MAX_PATH]) -> Result { - let reply = self.path_call(READLINK, Request::new(), [base, Base::Root], Paths { first: path, second: &[] }, MAX_PATH)?; - let n = counted(reply.value, MAX_PATH)?; - self.window().read(0, &mut out[..n]); - Ok(n) + pub fn read_link(&self, base: Base, path: &str, out: &mut [u8; MAX_PATH]) -> Result { + self.path_call_then(READLINK, Request::new(), Paths::one(base, path), MAX_PATH, |ticket, reply, carried| { + ticket.none_carried(carried)?; + let n = counted(reply.value, MAX_PATH)?; + ticket.window().read(0, &mut out[..n]); + Ok(n) + }) } /// Give what `from` names, following a link in its last component where /// `follow`, the name `to` too. With `replace`, `to` is taken from the /// node of that open file, and only if it still names it: [`Error::Stale`] /// when it no longer does. - pub fn link(&mut self, from: (Base, &str), follow: bool, to: (Base, &str), replace: Option) -> Result<(), Refused> { + pub fn link(&self, from: (Base, &str), follow: bool, to: (Base, &str), replace: Option) -> Result<(), Refused> { if !relative(to.1) { return Err(Refused::Error(Error::InvalidPath)); } let mut request = Request { flags: if follow { FOLLOW_LAST } else { 0 }, ..Request::new() }; - if let Some(expected) = replace { + if replace.is_some() { request.flags |= LINK_REPLACE; - request.expected = self.held(expected)?; } - self.path_call(LINK, request, [from.0, to.0], Paths { first: from.1, second: to.1.as_bytes() }, 0).map(drop) + let paths = Paths { bases: [from.0, to.0], first: from.1, second: to.1.as_bytes(), expected: replace }; + self.path_call_then(LINK, request, paths, 0, |ticket, _, carried| { + ticket.none_carried(carried).map_err(Refused::Error) + }) + } +} + +impl Ticket<'_> { + /// This request's share of the window. + fn window(&self) -> Bytes<'_> { + let at = self.taken.first() * SLOT_BYTES; + self.link.window.bytes().sub(at, self.taken.count() * SLOT_BYTES) + } + + /// `fid`'s id on this request's connection, or [`Error::Stale`] for one + /// answered on another. + fn held(&self, fid: Fid) -> Result { + match fid.generation == self.link.generation { + true => Ok(fid.id), + false => Err(Error::Stale), + } + } + + /// The file id `base` names on this connection, and whether a request + /// from it means the same on a fresh one. + fn base(&self, base: Base) -> Result<(u64, bool), Error> { + match base { + Base::Root => Ok((ROOT, true)), + Base::Fid(fid) => self.held(fid).map(|id| (id, false)), + } + } + + /// The request, its span the first `span` bytes of its share, and its + /// reply with the handles it carried. A connection that ended answers + /// `Gone`, and the next request connects again. + fn call_with(&self, op: u32, mut request: Request, span: usize) -> Result { + request.at = (self.taken.first() * SLOT_BYTES) as u64; + request.span = span as u64; + request.tag = self.dir.flight.sent(self.taken, self.link.generation); + let conn = &self.link.conn; + let answered = conn.send(op, &request).map_err(transport).and_then(|()| { + let receive = || { + let mut rx = FrameRx::<{ core::mem::size_of::() }>::new(); + let (msg_type, reply) = next_reply(conn, &mut rx)?; + Ok((reply.tag, (msg_type, reply, rx.take_batch()))) + }; + self.dir.flight.answer(self.taken, receive, Error::Malformed) + }); + if answered.is_err() { + self.dir.retire(&self.link); + } + answered + } + + /// [`Self::call_with`] for a reply that carries no handle. + fn call(&self, op: u32, request: Request, span: usize) -> Result { + let (msg_type, reply, carried) = self.call_with(op, request, span)?; + self.none_carried(carried)?; + answer(msg_type, reply) + } + + /// A call on a file id, with `span` bytes of this share. + fn fid_call(&self, op: u32, fid: Fid, mut request: Request, span: usize) -> Result { + request.fid = self.held(fid)?; + self.call(op, request, span) + } + + /// A reply that brings handles nobody asked for speaks another protocol: + /// its connection takes no request again. + fn none_carried(&self, carried: crate::ipc::Batch) -> Result<(), Error> { + if carried.is_empty() { + return Ok(()); + } + self.dir.retire(&self.link); + Err(Error::Malformed) + } + + /// The one handle a reply that answers with one carries; an error carries + /// none, and any other count retires the connection. + fn one(&self, msg_type: u32, reply: Reply, carried: crate::ipc::Batch) -> Result { + let n = carried.len(); + match (answer(msg_type, reply), carried.exact::<1>()) { + (Ok(_), Some([handle])) => Ok(handle), + (Err(e), None) if n == 0 => Err(e), + _ => { + self.dir.retire(&self.link); + Err(Error::Malformed) + } + } } } -/// A path request's reply frame: its reply, an absolute link to resolve again -/// from the span's first `span` bytes, or the refusal it names. -fn path_answer(msg_type: u32, reply: Reply, span: usize) -> Result { +impl Drop for Ticket<'_> { + fn drop(&mut self) { + self.dir.flight.give(self.taken); + } +} + +/// What a path request's reply frame answers, refusals aside. +enum PathAnswer { + Reply(Reply), + /// An absolute link to resolve again, this many bytes from the span's + /// start. + Follow(usize), +} + +/// A path request's reply frame, whose span is `span` bytes: its reply, a +/// link to follow, or the refusal it names. +fn path_answer(msg_type: u32, reply: Reply, span: usize) -> Result { match msg_type { - FOLLOW => Err(Refused::Follow(counted(reply.value, MAX_PATH.min(span))?)), - _ => answer(msg_type, reply).map_err(Refused::Error), + FOLLOW => Ok(PathAnswer::Follow(counted(reply.value, MAX_PATH.min(span))?)), + _ => answer(msg_type, reply).map(PathAnswer::Reply), + } +} + +/// A refusal of a request on a file id, which no link follows from: an id +/// names one node. +fn unfollowed(refused: Refused) -> Error { + match refused { + Refused::Error(e) => e, + Refused::Follow(_) => Error::Malformed, } } @@ -1338,9 +1469,35 @@ fn hello(conn: &Connection, window: &SharedMemory) -> Result { Err(e) => return Err(e), } let mut rx = FrameRx::<{ core::mem::size_of::() }>::new(); + let (msg_type, reply) = next_reply(conn, &mut rx)?; + answer(msg_type, reply) +} + +/// How long a client waits for the next reply it is owed before it lets its +/// server go as hung: a hang's bound, never an answer's expected time. +/// +/// A reply waits on the server's block requests and on its turns behind its +/// other connections. A block server may stay silent for the block protocol's +/// bound ([`BLOCK_SILENCE_NS`]); this one is that and 60 s of room for the +/// server's own work and turns. A file server's reply has no derived worst +/// case: one request may wait on several block requests in turn, and nothing +/// sums them or the turns +/// (`issues/a-file-server-reply-has-no-derived-worst-case.md`). +pub const REPLY_BOUND_NS: u64 = BLOCK_SILENCE_NS + 60_000_000_000; + +/// The block protocol's bound on a block server's silence: the block crate's, +/// which depends on this one, and which the file server holds this equal to. +pub const BLOCK_SILENCE_NS: u64 = 365_000_000_000; + +/// The next reply frame on `conn`, into `rx`: its frame type and words, +/// [`Error::Unanswered`] for none within [`REPLY_BOUND_NS`]. +fn next_reply(conn: &Connection, rx: &mut FrameRx) -> Result<(u32, Reply), Error> { + if !crate::poller::wait_one(conn, crate::poller::READABLE, REPLY_BOUND_NS).map_err(|e| Error::kernel("a reply's wait", e))? { + return Err(Error::Unanswered); + } let frame = rx.recv(conn).map_err(transport)?; let reply = crate::ipc::decode_payload(rx.payload(frame.payload_len)).map_err(transport)?; - answer(frame.msg_type, reply) + Ok((frame.msg_type, reply)) } /// A connection's transport refused, as the protocol names it. @@ -1543,14 +1700,14 @@ mod tests { } for msg_type in [0, HELLO, OPEN, FOLLOW + 1, u32::MAX] { assert_eq!(answer(msg_type, Reply::ok()).err(), Some(Error::Malformed), "frame {msg_type:#x}"); - assert_eq!(path_answer(msg_type, Reply::ok(), MAX_PATH).err(), Some(Refused::Error(Error::Malformed)), "frame {msg_type:#x}"); + assert_eq!(path_answer(msg_type, Reply::ok(), MAX_PATH).err(), Some(Error::Malformed), "frame {msg_type:#x}"); } assert_eq!(answer(FOLLOW, Reply::ok()).err(), Some(Error::Malformed), "a link to follow answers only a path"); let follow = |value| Reply { value, ..Reply::ok() }; - assert_eq!(path_answer(FOLLOW, follow(MAX_PATH as u64), MAX_PATH).err(), Some(Refused::Follow(MAX_PATH))); - assert_eq!(path_answer(FOLLOW, follow(MAX_PATH as u64 + 1), 2 * MAX_PATH).err(), Some(Refused::Error(Error::Malformed))); - assert_eq!(path_answer(FOLLOW, follow(9), 8).err(), Some(Refused::Error(Error::Malformed)), "a link past the span"); - assert_eq!(path_answer(REPLY, Reply::refused(Error::Loop), MAX_PATH).err(), Some(Refused::Error(Error::Loop))); + assert!(matches!(path_answer(FOLLOW, follow(MAX_PATH as u64), MAX_PATH), Ok(PathAnswer::Follow(MAX_PATH)))); + assert_eq!(path_answer(FOLLOW, follow(MAX_PATH as u64 + 1), 2 * MAX_PATH).err(), Some(Error::Malformed)); + assert_eq!(path_answer(FOLLOW, follow(9), 8).err(), Some(Error::Malformed), "a link past the span"); + assert_eq!(path_answer(REPLY, Reply::refused(Error::Loop), MAX_PATH).err(), Some(Error::Loop)); assert_eq!(next_batch(0, READDIR_END), Ok(None)); assert_eq!(next_batch(19, READDIR_END), Ok(None)); assert_eq!(next_batch(19, 5), Ok(Some(5))); diff --git a/sdk/toyos/src/fs/flight.rs b/sdk/toyos/src/fs/flight.rs new file mode 100644 index 00000000000..20f608c0803 --- /dev/null +++ b/sdk/toyos/src/fs/flight.rs @@ -0,0 +1,273 @@ +//! One connection's requests in flight, up to [`SLOTS`] at once. A request +//! holds a slot, which is its span of the window and where its reply is +//! filed, and a tag naming the slot; one too large for a slot holds them all. +//! +//! **Whoever waits receives.** Each connection has its own receive turn. A +//! caller waiting for its reply takes its connection's turn if nobody holds +//! it, and takes replies off that connection, filing each other caller's in +//! that caller's slot, until its own comes; then it wakes a caller still +//! waiting on the same connection, which takes the turn if it is still free. +//! A receive that fails, or a reply naming no request in flight, fails every +//! request in flight on that connection. +//! +//! **One lock.** Every slot's request and the turn are read and changed only +//! under the table's lock, which is held for nothing else; a caller sleeps on +//! its slot's own word, which whoever changes its request moves after. +//! +//! Built on `flight_prim`, which `kernel-loom` swaps for loom's lock and a +//! futex that wakes only the word woken, so its models run this file. + +use super::flight_prim::{wait, wake_all, wake_one, AtomicU32, Lock, Ordering}; + +/// The requests one connection has in flight at most. +pub const SLOTS: usize = 16; + +/// Every slot free. +const ALL: u32 = (1 << SLOTS) - 1; + +/// Requests in flight, answered with `R` or failed with `E`. +pub struct Flight { + table: Lock>, + /// Moved whenever slots are given back; a caller waiting for one sleeps + /// on it. + freed: AtomicU32, + /// Moved whenever a slot's request is answered or failed, or its caller + /// is asked to take the turn; that caller sleeps on it. + woken: [AtomicU32; SLOTS], +} + +struct Table { + /// One bit for each slot nobody holds. + free: u32, + /// Callers waiting for every slot, ahead of whom no slot is taken. + wholes: u32, + /// The slots whose callers hold their connection's receive turn. + receiving: u32, + /// Requests sent, which a tag counts above its slot. + sent: u64, + requests: [Request; SLOTS], +} + +struct Request { + /// The connection it went out on, by generation. + link: u64, + tag: u64, + state: State, +} + +enum State { + /// Not in flight. + Idle, + /// Sent, or about to be, its reply not yet filed. + Waiting, + Filed(Result), +} + +/// The slots one caller holds, from `first`: one, or all of them. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct Taken { + first: usize, + count: usize, +} + +impl Taken { + pub fn first(self) -> usize { + self.first + } + + pub fn count(self) -> usize { + self.count + } +} + +impl Default for Flight { + fn default() -> Self { + Self::new() + } +} + +impl Flight { + pub fn new() -> Self { + Self { + table: Lock::new(Table { + free: ALL, + wholes: 0, + receiving: 0, + sent: 0, + requests: core::array::from_fn(|_| Request { link: 0, tag: 0, state: State::Idle }), + }), + freed: AtomicU32::new(0), + woken: core::array::from_fn(|_| AtomicU32::new(0)), + } + } + + /// A slot, or every slot where `whole`, once they are free. A caller + /// waiting for every slot goes before any that asks for one after it. + pub fn take(&self, whole: bool) -> Taken { + let mut counted = false; + loop { + let seen = self.freed.load(Ordering::Acquire); + { + let mut table = self.table.lock(); + if whole { + if table.free == ALL { + table.free = 0; + table.wholes -= u32::from(counted); + return Taken { first: 0, count: SLOTS }; + } + table.wholes += u32::from(!counted); + counted = true; + } else if table.wholes == 0 && table.free != 0 { + let first = table.free.trailing_zeros() as usize; + table.free &= !(1 << first); + return Taken { first, count: 1 }; + } + } + wait(&self.freed, seen); + } + } + + /// The request on `taken` goes out on the connection of generation + /// `link`: the tag it carries. Called before the send, so its reply always + /// finds it waiting. + pub fn sent(&self, taken: Taken, link: u64) -> u64 { + let mut table = self.table.lock(); + let tag = (table.sent << SLOTS.trailing_zeros()) | taken.first as u64; + table.sent += 1; + table.requests[taken.first] = Request { link, tag, state: State::Waiting }; + tag + } + + /// The reply to the request on `taken`, or the failure of its + /// connection. While nobody else does, this caller receives, with + /// `receive`, which waits for the next reply on that connection and + /// answers its tag; a tag naming no request in flight there fails them + /// all with `malformed`. + pub fn answer(&self, taken: Taken, mut receive: impl FnMut() -> Result<(u64, R), E>, malformed: E) -> Result + where + E: Copy, + { + let mine = taken.first; + loop { + let seen = self.woken[mine].load(Ordering::Acquire); + { + let mut table = self.table.lock(); + if let Some(filed) = table.take_filed(mine) { + return filed; + } + let link = table.requests[mine].link; + if !table.receives(link) { + table.receiving |= 1 << mine; + drop(table); + return self.lead(mine, link, &mut receive, malformed); + } + } + wait(&self.woken[mine], seen); + } + } + + /// Receive on connection `link`, filing every reply, until slot `mine`'s + /// is filed; then let the turn go to a caller still waiting. + fn lead(&self, mine: usize, link: u64, receive: &mut impl FnMut() -> Result<(u64, R), E>, malformed: E) -> Result + where + E: Copy, + { + loop { + let got = receive(); + let mut table = self.table.lock(); + let woken = match got { + Ok((tag, reply)) => table.file(link, tag, reply).unwrap_or_else(|_| table.fail(link, malformed)), + Err(why) => table.fail(link, why), + }; + let answered = table.take_filed(mine); + let next = answered.is_some().then(|| { + table.receiving &= !(1 << mine); + table.waiting(link) + }); + drop(table); + self.wake(woken & !(1 << mine)); + if let Some(answered) = answered { + self.wake(next.flatten().map_or(0, |i| 1 << i)); + return answered; + } + } + } + + /// Give `taken` back, answered or never sent. A request never sent, whose + /// caller may have been woken to take its connection's turn, wakes + /// another caller waiting on that connection while nobody receives there. + pub fn give(&self, taken: Taken) { + let next = { + let mut table = self.table.lock(); + let request = &mut table.requests[taken.first]; + let unsent = matches!(request.state, State::Waiting); + request.state = State::Idle; + let link = request.link; + table.free |= if taken.count == SLOTS { ALL } else { 1 << taken.first }; + if unsent && !table.receives(link) { table.waiting(link) } else { None } + }; + self.freed.fetch_add(1, Ordering::Release); + wake_all(&self.freed); + self.wake(next.map_or(0, |i| 1 << i)); + } + + /// Wake the callers of the slots `slots` names. + fn wake(&self, slots: u32) { + for i in (0..SLOTS).filter(|i| slots & (1 << i) != 0) { + self.woken[i].fetch_add(1, Ordering::Release); + wake_one(&self.woken[i]); + } + } +} + +impl Table { + /// File `reply` where it waits on connection `link`: the slot to wake, + /// or `Err` for a tag naming no request waiting there. + fn file(&mut self, link: u64, tag: u64, reply: R) -> Result { + let i = (tag % SLOTS as u64) as usize; + let request = &mut self.requests[i]; + if !matches!(request.state, State::Waiting) || request.tag != tag || request.link != link { + return Err(()); + } + request.state = State::Filed(Ok(reply)); + Ok(1 << i) + } + + /// Fail every request waiting on connection `link` with `why`: the slots + /// to wake. + fn fail(&mut self, link: u64, why: E) -> u32 + where + E: Copy, + { + let mut failed = 0; + for (i, request) in self.requests.iter_mut().enumerate() { + if matches!(request.state, State::Waiting) && request.link == link { + request.state = State::Filed(Err(why)); + failed |= 1 << i; + } + } + failed + } + + /// What was filed in slot `i`, taken by its caller. + fn take_filed(&mut self, i: usize) -> Option> { + match core::mem::replace(&mut self.requests[i].state, State::Idle) { + State::Filed(filed) => Some(filed), + other => { + self.requests[i].state = other; + None + } + } + } + + /// Whether a caller holds connection `link`'s receive turn. + fn receives(&self, link: u64) -> bool { + (0..SLOTS).any(|i| self.receiving & (1 << i) != 0 && self.requests[i].link == link) + } + + /// A slot whose caller still waits on connection `link`: the caller to + /// take its turn next. + fn waiting(&self, link: u64) -> Option { + (0..SLOTS).find(|&i| matches!(self.requests[i].state, State::Waiting) && self.requests[i].link == link) + } +} diff --git a/sdk/toyos/src/sync.rs b/sdk/toyos/src/sync.rs index 9c0f533888a..f3a602572bd 100644 --- a/sdk/toyos/src/sync.rs +++ b/sdk/toyos/src/sync.rs @@ -4,8 +4,7 @@ //! [`Lock`] is one `u32`, 0 free, 1 held, 2 held with waiters, as Drepper's //! "Futexes Are Tricky" (2011) builds its mutex in §"Mutex, Take 2". A holder //! that finds 2 at its release wakes one waiter, and a waiter sleeps only while -//! the word still says 2. [`Turnstile`] serves its holders in the order they -//! asked. +//! the word still says 2. use core::cell::UnsafeCell; use core::sync::atomic::{AtomicU32, Ordering}; @@ -60,101 +59,36 @@ impl Drop for Held<'_, T> { /// Sleep while `word` still says `expected`; a spurious return is allowed. #[cfg(not(test))] -fn wait(word: &AtomicU32, expected: u32) { +pub(crate) fn wait(word: &AtomicU32, expected: u32) { // SAFETY: `word` is a live, aligned u32. unsafe { syscall::futex_wait(word.as_ptr(), expected, None) }; } /// Wake every sleeper on `word`. #[cfg(not(test))] -fn wake_all(word: &AtomicU32) { +pub(crate) fn wake_all(word: &AtomicU32) { // SAFETY: `word` is a live, aligned u32. unsafe { syscall::futex_wake(word.as_ptr(), u32::MAX) }; } /// Wake one sleeper on `word`. #[cfg(not(test))] -fn wake_one(word: &AtomicU32) { +pub(crate) fn wake_one(word: &AtomicU32) { // SAFETY: `word` is a live, aligned u32. unsafe { syscall::futex_wake(word.as_ptr(), 1) }; } // A host test has no futex: a waiter yields and spins. #[cfg(test)] -fn wait(_: &AtomicU32, _: u32) { +pub(crate) fn wait(_: &AtomicU32, _: u32) { std::thread::yield_now(); } #[cfg(test)] -fn wake_all(_: &AtomicU32) {} +pub(crate) fn wake_all(_: &AtomicU32) {} #[cfg(test)] -fn wake_one(_: &AtomicU32) {} - -/// `T` held in turn: each [`Turnstile::lock`] takes a ticket, and the value -/// goes to the tickets in the order they were taken. A plain lock lets the -/// thread that let go take it straight back ahead of the one it woke, so -/// threads sharing one connection would not be served in turn. -pub struct Turnstile { - /// The next ticket handed out. - next: AtomicU32, - /// The ticket whose turn it is, which waiters sleep on. - serving: AtomicU32, - value: UnsafeCell, -} - -// SAFETY: the value is reached only through a `Turn`, which only the ticket -// being served holds. -unsafe impl Sync for Turnstile {} - -/// The value, held for one turn. -pub struct Turn<'a, T>(&'a Turnstile); - -impl Turnstile { - pub const fn new(value: T) -> Self { - Self { - next: AtomicU32::new(0), - serving: AtomicU32::new(0), - value: UnsafeCell::new(value), - } - } - - /// The value, once every ticket taken before this one has had its turn. - pub fn lock(&self) -> Turn<'_, T> { - let ticket = self.next.fetch_add(1, Ordering::Relaxed); - loop { - let serving = self.serving.load(Ordering::Acquire); - if serving == ticket { - return Turn(self); - } - wait(&self.serving, serving); - } - } -} - -impl core::ops::Deref for Turn<'_, T> { - type Target = T; - fn deref(&self) -> &T { - // SAFETY: this ticket is the one being served. - unsafe { &*self.0.value.get() } - } -} - -impl core::ops::DerefMut for Turn<'_, T> { - fn deref_mut(&mut self) -> &mut T { - // SAFETY: this ticket is the one being served. - unsafe { &mut *self.0.value.get() } - } -} - -impl Drop for Turn<'_, T> { - fn drop(&mut self) { - self.0.serving.fetch_add(1, Ordering::Release); - // Every waiter, since only the next ticket's may go on and only it - // knows it is the next. - wake_all(&self.0.serving); - } -} +pub(crate) fn wake_one(_: &AtomicU32) {} /// How many times a contender looks again before it sleeps. A futex wait and /// the wake that ends it cost a kernel round trip each way, microseconds, @@ -195,13 +129,10 @@ pub fn unlock_word(state: &AtomicU32) { #[cfg(test)] mod tests { use std::sync::Arc; - use std::time::{Duration, Instant}; use std::vec::Vec; use super::*; - /// A thread that lets the value go and asks for it again waits behind - /// the threads that asked first, in the order they asked. /// Threads contending for one lock, each holding it only briefly, never /// hold it together: every increment made under it lands. #[test] @@ -228,32 +159,4 @@ mod tests { } assert_eq!(*lock.lock(), THREADS * EACH); } - - #[test] - fn the_value_goes_to_the_tickets_in_the_order_they_were_taken() { - let stile: Arc>> = Arc::new(Turnstile::new(Vec::new())); - let mut held = stile.lock(); - held.push("first"); - let queued: Vec<_> = ["second", "third"] - .into_iter() - .enumerate() - .map(|(i, name)| { - let theirs = Arc::clone(&stile); - let thread = std::thread::spawn(move || theirs.lock().push(name)); - // The next waits for this one's ticket, so the two are taken in order. - let asked = Instant::now(); - while stile.next.load(Ordering::Relaxed) < i as u32 + 2 { - assert!(asked.elapsed() < Duration::from_secs(10), "{name} never took a ticket"); - std::thread::yield_now(); - } - thread - }) - .collect(); - drop(held); - stile.lock().push("first again"); - for thread in queued { - thread.join().unwrap(); - } - assert_eq!(*stile.lock(), ["first", "second", "third", "first again"]); - } } diff --git a/system/fileserver/src/bounds.rs b/system/fileserver/src/bounds.rs index bd4baf01793..593abc88760 100644 --- a/system/fileserver/src/bounds.rs +++ b/system/fileserver/src/bounds.rs @@ -34,6 +34,11 @@ use toyos::shm::Lent; /// process. pub const MAX_SERVED: usize = 128; +// A client lets this server go once a reply it owes has waited past a bound +// the client derives from its copy of the block protocol's silence bound, +// which is the bound a block request this server waits on keeps. +const _: () = assert!(toyos::fs::BLOCK_SILENCE_NS == toyos_blockring::SILENCE_BOUND_NS); + /// Connections taken and not yet answered at their hello, a quarter of them /// one share's. While the table is full, the next waits in the port's queue. pub const HANDSHAKES: Policy = Policy::quartered(32); diff --git a/system/pkg/src/main.rs b/system/pkg/src/main.rs index 28918a71411..bbe407b041d 100644 --- a/system/pkg/src/main.rs +++ b/system/pkg/src/main.rs @@ -72,7 +72,7 @@ fn install(file: &Path, assume_yes: bool) -> Result<(), String> { .map_err(|e| format!("pkg: cannot read {}: {e}", beside.display()))?; let want = sums::digest_for(&sums_text, &name)?; - let mut apps = writable_apps()?; + let apps = writable_apps()?; let room = apps.room().map_err(|e| format!("pkg: {} answers no room: {e:?}", package::DIR))?; let (planner, got) = survey(file, Planner::new(&name, room.free, room.unit))?; if got != want { diff --git a/system/supervisor/src/main.rs b/system/supervisor/src/main.rs index 658c4460143..2f49a3a4115 100644 --- a/system/supervisor/src/main.rs +++ b/system/supervisor/src/main.rs @@ -2068,7 +2068,7 @@ impl<'a> Supervisor<'a> { let (done_tx, done_rx) = std::sync::mpsc::channel(); let Some((_, held)) = self.files.iter().find(|(held, _)| *held == name) else { continue }; let spawned = std::thread::Builder::new().name(format!("sync-{role}")).spawn(move || { - let answer = toyos::fs::Dir::connect(held.as_handle()).and_then(|mut dir| dir.sync()); + let answer = toyos::fs::Dir::connect(held.as_handle()).and_then(|dir| dir.sync()); let _ = done_tx.send(answer); }); match spawned { diff --git a/tests/guest/osbench/src/files.rs b/tests/guest/osbench/src/files.rs index a0e6e1e4505..5f865a0bab2 100644 --- a/tests/guest/osbench/src/files.rs +++ b/tests/guest/osbench/src/files.rs @@ -117,6 +117,20 @@ fn parts( each(&name("pread_8"), 100, 2000, fields, |i| io("pread", f.read_at(&mut buf[..8], page_at(i, pages))).map(drop)); } +/// `measure`, while another thread asks `metadata` of `path` until it ends. +fn beside_a_thread(path: &Path, measure: impl FnOnce()) { + let done = std::sync::atomic::AtomicBool::new(false); + std::thread::scope(|scope| { + scope.spawn(|| { + while !done.load(std::sync::atomic::Ordering::Relaxed) { + let _ = fs::metadata(path); + } + }); + measure(); + done.store(true, std::sync::atomic::Ordering::Relaxed); + }); +} + /// Every benchmark under `target`, each line naming what served it and its /// medium. Each one's comment is its warm-up and repetitions. pub fn suite(target: &Target) { @@ -154,6 +168,14 @@ pub fn suite(target: &Target) { Err(e) => refuse(&name("fstat"), e), } + // The same `metadata` and open, while another thread of this process asks + // `metadata` of the same file as fast as it is answered: what a second + // thread's calls on one directory cost this one's. 2,000 after 100 each. + beside_a_thread(&small, || { + each(&name("stat.beside_a_thread"), 100, 2000, &with(&[]), |_| io("stat", fs::metadata(&small)).map(drop)); + each(&name("open_close.beside_a_thread"), 100, 2000, &with(&[]), |_| io("open", File::open(&small)).map(drop)); + }); + // A new file, 100 bytes written to it, closed and removed: 1,000 after 50. let fresh = scratch.at("fresh"); each(&name("create_write_unlink"), 50, 1000, &with(&[("bytes", "100".into())]), |_| { @@ -188,11 +210,11 @@ pub fn suite(target: &Target) { parts(&name, &fields, &mut f, &mut buf, pages); #[cfg(target_os = "toyos")] match crate::sys::direct::connect(&target.dir) { - Ok(Some(mut dir)) => { + Ok(Some(dir)) => { let under = |p: &Path| p.strip_prefix(&target.dir).ok().and_then(Path::to_str).map(str::to_string); match (under(&small), under(&big)) { (Some(s), Some(b)) => { - crate::sys::direct::rows(&mut dir, &name, &with(&[]), &fields, &s, &b, |i| page_at(i, pages)); + crate::sys::direct::rows(&dir, &name, &with(&[]), &fields, &s, &b, |i| page_at(i, pages)); } _ => refuse(&name("direct"), "the scratch files are not under the served directory"), } diff --git a/tests/guest/osbench/src/sys/toyos.rs b/tests/guest/osbench/src/sys/toyos.rs index b562f789d00..826259a126f 100644 --- a/tests/guest/osbench/src/sys/toyos.rs +++ b/tests/guest/osbench/src/sys/toyos.rs @@ -653,7 +653,7 @@ pub mod direct { /// the target's `fields`, and `big`, with `big_fields`; each a path /// under the directory `dir` serves. pub fn rows( - dir: &mut Dir, + dir: &Dir, name: &dyn Fn(&str) -> String, fields: &[(&'static str, String)], big_fields: &[(&'static str, String)], diff --git a/tests/guest/rust-tests/src/bin/a_served_file_reads_through_its_paged_object.rs b/tests/guest/rust-tests/src/bin/a_served_file_reads_through_its_paged_object.rs index 9cd2d32cd05..b6e6f3f1d20 100644 --- a/tests/guest/rust-tests/src/bin/a_served_file_reads_through_its_paged_object.rs +++ b/tests/guest/rust-tests/src/bin/a_served_file_reads_through_its_paged_object.rs @@ -33,7 +33,7 @@ fn read(holder: &Paged, offset: u64, len: usize) -> Vec { } fn main() { - let mut dir = Dir::named(DIR).expect("the home directory"); + let dir = Dir::named(DIR).expect("the home directory"); let data = pattern(5 * 4096 + 123); let mut writer = dir.open(Base::Root, NAME, O_WRITE | O_CREATE | O_TRUNCATE).expect("open to write"); diff --git a/tests/guest/rust-tests/src/bin/an_export_shares_its_descriptions.rs b/tests/guest/rust-tests/src/bin/an_export_shares_its_descriptions.rs index ab9b5e63ab8..1d04e8dc661 100644 --- a/tests/guest/rust-tests/src/bin/an_export_shares_its_descriptions.rs +++ b/tests/guest/rust-tests/src/bin/an_export_shares_its_descriptions.rs @@ -22,13 +22,13 @@ const STREAMED: &str = "an-export-shares-its-descriptions-streamed"; const STREAM_BYTES: usize = 1 << 20; fn main() { - let mut parent = Dir::named(DIR).expect("the home directory"); + let parent = Dir::named(DIR).expect("the home directory"); let fid = parent.open(Base::Root, FILE, O_READ | O_WRITE | O_CREATE | O_TRUNCATE).expect("open").fid; let written = parent.write(fid, b"abcdef").expect("write"); assert_eq!((written.len, written.offset), (6, 6)); let end = parent.export(&[fid]).expect("export"); - let mut child = Dir::adopt(end).expect("the exported connection says hello"); + let child = Dir::adopt(end).expect("the exported connection says hello"); let theirs = Fid { id: 1, generation: child.generation() }; assert_eq!(child.seek(theirs, SEEK_CURRENT, 0), Ok(6), "the export's id 1 has an offset of its own"); assert_eq!(child.seek(theirs, SEEK_START, 2), Ok(2)); diff --git a/tests/guest/rust-tests/src/bin/app_view.rs b/tests/guest/rust-tests/src/bin/app_view.rs index 5e9addcba71..12ff5809bdb 100644 --- a/tests/guest/rust-tests/src/bin/app_view.rs +++ b/tests/guest/rust-tests/src/bin/app_view.rs @@ -200,7 +200,7 @@ fn app() { match Dir::named(&format!("fs:{PACKAGE}")) { Err(e) => red.push(format!("fs:{PACKAGE} would not connect: {e:?}")), - Ok(mut dir) => { + Ok(dir) => { if dir.writable() { red.push(format!("fs:{PACKAGE} says it is writable")); } diff --git a/tests/guest/rust-tests/src/bin/served_file_offset.rs b/tests/guest/rust-tests/src/bin/served_file_offset.rs index 34da9e6d83f..d2335504b9b 100644 --- a/tests/guest/rust-tests/src/bin/served_file_offset.rs +++ b/tests/guest/rust-tests/src/bin/served_file_offset.rs @@ -129,7 +129,7 @@ fn refused_spawn_keeps_export(fd: i32) { // The kernel refuses after the SDK duplicated the descriptors and laid the program out. let refused = helper.spawn(Some(&image)).err(); assert_eq!(refused, Some(toyos::spawn::Refused::Syscall(toyos_abi::syscall::SyscallError::InvalidArgument))); - let mut dir = toyos::fs::Dir::adopt(exported).expect("refusal kept the original export"); + let dir = toyos::fs::Dir::adopt(exported).expect("refusal kept the original export"); let fid = toyos::fs::Fid { id: 1, generation: dir.generation() }; assert_eq!(dir.seek(fid, toyos::fs::SEEK_CURRENT, 0).unwrap(), 4, "refusal kept the shared offset"); dir.close(fid); diff --git a/tests/guest/rust-tests/src/bin/stat_identity.rs b/tests/guest/rust-tests/src/bin/stat_identity.rs index fca0c07e0db..e9ff37d163e 100644 --- a/tests/guest/rust-tests/src/bin/stat_identity.rs +++ b/tests/guest/rust-tests/src/bin/stat_identity.rs @@ -94,7 +94,7 @@ fn directory_replacement(root: &str) { fs::create_dir(&source).unwrap(); fs::create_dir(&target).unwrap(); write(&format!("{source}/file")); - let mut directory = Dir::named(&format!("fs:{root}")).unwrap(); + let directory = Dir::named(&format!("fs:{root}")).unwrap(); let original = directory.open(Base::Root, "stat-identity-source", O_DIRECTORY | O_READ).unwrap().fid; let alias = directory.dup(original).unwrap(); let replaced = directory.open(Base::Root, "stat-identity-target", O_DIRECTORY | O_READ).unwrap().fid;