diff --git a/sdk/std/sys/fs.rs b/sdk/std/sys/fs.rs index ad451b34b2..45a2e2a285 100644 --- a/sdk/std/sys/fs.rs +++ b/sdk/std/sys/fs.rs @@ -39,18 +39,12 @@ enum Inner { Served(Arc), } -/// A file open on a file server. +/// A file open on a file server, which keeps its offset. struct Served { dir: Arc, rel: String, - append: bool, - state: Mutex, -} - -struct Position { fid: u64, generation: u64, - offset: u64, } #[path = "turns.rs"] @@ -378,6 +372,7 @@ fn served_error(e: toyos::fs::Error) -> io::Error { Error::InvalidRequest => io::const_error!(Kind::InvalidInput, "the file server takes no such request"), Error::Unsupported => io::const_error!(Kind::Unsupported, "the volume's format has no such thing"), Error::OutOfResources => io::const_error!(Kind::OutOfMemory, "the kernel had no room for the request"), + Error::NoDirectory => io::const_error!(Kind::NotADirectory, "the connection was given files and no directory"), Error::Gone => io::const_error!(Kind::StaleNetworkFileHandle, "the file server is gone"), Error::NotHeld => io::const_error!(Kind::NotFound, "this process holds no such directory"), Error::Malformed => io::const_error!(Kind::InvalidData, "the file server answered outside its protocol"), @@ -480,19 +475,16 @@ pub fn is_served(path: &Path) -> bool { impl Served { fn with( &self, - op: impl FnOnce(&mut toyos::fs::Dir, &mut Position) -> Result, + op: impl FnOnce(&mut toyos::fs::Dir, u64, u64) -> Result, ) -> io::Result { let mut dir = self.dir.lock(); - let mut pos = self.state.lock().unwrap(); - op(&mut dir, &mut pos).map_err(served_error) + op(&mut dir, self.fid, self.generation).map_err(served_error) } } impl Drop for Served { fn drop(&mut self) { - let mut dir = self.dir.lock(); - let pos = self.state.lock().unwrap(); - dir.close(pos.fid, pos.generation); + self.dir.lock().close(self.fid, self.generation); } } @@ -515,16 +507,13 @@ impl File { } /// A kernel handle a child may write this file through: the file's own, or - /// a pipe its server appends to the file from this file's offset. The - /// pipe lives as long as the answer and the child's copy of it. + /// a pipe its server writes to the file at this file's offset. The pipe + /// lives as long as the answer and the child's copy of it. pub fn as_child_stdio(&self) -> io::Result<(RawHandle, Option)> { match &self.0 { Inner::Kernel(h) => Ok((*h, None)), Inner::Served(s) => { - let pipe = s.with(|dir, pos| { - let at = if s.append { dir.fstat(pos.fid, pos.generation)?.size } else { pos.offset }; - dir.stream(pos.fid, pos.generation, at) - })?; + let pipe = s.with(|dir, fid, generation| dir.stream(fid, generation))?; Ok((toyos::AsHandle::as_handle(&pipe), Some(pipe))) } } @@ -543,12 +532,8 @@ impl File { Ok(File(Inner::Served(Arc::new(Served { dir: Arc::clone(cap), rel: String::from(rel), - append: opts.append, - state: Mutex::new(Position { - fid: opened.fid, - generation: opened.generation, - offset: 0, - }), + fid: opened.fid, + generation: opened.generation, })))) }, ) @@ -557,14 +542,14 @@ impl File { pub fn file_attr(&self) -> io::Result { match &self.0 { Inner::Kernel(h) => syscall::fstat(*h).map(FileAttr::of_kernel).map_err(to_io_error), - Inner::Served(s) => s.with(|dir, pos| dir.fstat(pos.fid, pos.generation)).map(FileAttr::of), + Inner::Served(s) => s.with(|dir, fid, generation| dir.fstat(fid, generation)).map(FileAttr::of), } } pub fn fsync(&self) -> io::Result<()> { match &self.0 { Inner::Kernel(h) => syscall::fsync(*h).map_err(to_io_error), - Inner::Served(s) => s.with(|dir, pos| dir.fsync(pos.fid, pos.generation)), + Inner::Served(s) => s.with(|dir, fid, generation| dir.fsync(fid, generation)), } } @@ -591,18 +576,14 @@ impl File { pub fn truncate(&self, size: u64) -> io::Result<()> { match &self.0 { Inner::Kernel(h) => syscall::ftruncate(*h, size).map_err(to_io_error), - Inner::Served(s) => s.with(|dir, pos| dir.truncate(pos.fid, pos.generation, size)), + Inner::Served(s) => s.with(|dir, fid, generation| dir.truncate(fid, generation, size)), } } pub fn read(&self, buf: &mut [u8]) -> io::Result { match &self.0 { Inner::Kernel(h) => syscall::read(*h, buf).map_err(to_io_error), - Inner::Served(s) => s.with(|dir, pos| { - let n = dir.read(pos.fid, pos.generation, pos.offset, buf)?; - pos.offset += n as u64; - Ok(n) - }), + Inner::Served(s) => s.with(|dir, fid, generation| dir.read(fid, generation, buf)), } } @@ -637,11 +618,7 @@ impl File { pub fn write(&self, buf: &[u8]) -> io::Result { match &self.0 { Inner::Kernel(h) => syscall::write(*h, buf).map_err(to_io_error), - Inner::Served(s) => s.with(|dir, pos| { - let written = dir.write(pos.fid, pos.generation, pos.offset, buf)?; - pos.offset = written.offset; - Ok(written.len) - }), + Inner::Served(s) => s.with(|dir, fid, generation| dir.write(fid, generation, buf).map(|w| w.len)), } } @@ -681,20 +658,18 @@ impl File { }; syscall::seek(*h, abi_pos).map_err(to_io_error) } - Inner::Served(s) => s.with(|dir, p| { - let (base, delta) = match pos { - SeekFrom::Start(n) => (n, 0), - SeekFrom::Current(d) => (p.offset, d), - SeekFrom::End(d) => (dir.fstat(p.fid, p.generation)?.size, d), + Inner::Served(s) => { + let (whence, delta) = match pos { + // Past the largest offset a file has is the request's fault, as the kernel's route says. + SeekFrom::Start(n) => { + let n = i64::try_from(n).map_err(|_| io::Error::from(io::ErrorKind::InvalidInput))?; + (toyos::fs::SEEK_START, n) + } + SeekFrom::Current(d) => (toyos::fs::SEEK_CURRENT, d), + SeekFrom::End(d) => (toyos::fs::SEEK_END, d), }; - // Past either end is the request's fault, as the kernel's route says. - let at = base - .checked_add_signed(delta) - .filter(|&at| at <= toyos::fs::MAX_FILE_BYTES) - .ok_or(toyos::fs::Error::InvalidRequest)?; - p.offset = at; - Ok(at) - }), + s.with(|dir, fid, generation| dir.seek(fid, generation, whence, delta)) + } } } diff --git a/sdk/toyos/src/fs.rs b/sdk/toyos/src/fs.rs index 6a85cf00b7..86c7f3f0a0 100644 --- a/sdk/toyos/src/fs.rs +++ b/sdk/toyos/src/fs.rs @@ -18,6 +18,12 @@ //! target and the rest of the path in the window, and resolved again in the //! client's own table, so it reaches nothing the client does not hold. //! +//! **An open file is a description the server keeps**: the file, the access +//! the open fixed and the offset. A connection names descriptions by id, and +//! [`DUP`] and [`EXPORT`] make second ids that share one description's offset, +//! on the same connection or on a connection made for another process. An +//! exported connection was given files and no directory: it names no path. +//! //! **One connection per directory per process, a request and then its //! reply.** The client lends the server a [`WINDOW_BYTES`] region at connect //! ([`HELLO`]); paths, data and listings travel through it and the frames stay @@ -75,6 +81,21 @@ pub const SYMLINK: u32 = 16; pub const READLINK: u32 = 17; pub const STREAM: u32 = 18; pub const SYNC: u32 = 19; +/// A second id on the description `fid` names: `value`. +pub const DUP: u32 = 20; +/// Move the description's offset: `flags` is [`SEEK_START`], [`SEEK_CURRENT`] +/// or [`SEEK_END`], `offset` the signed distance from it; `value` is where it +/// landed. +pub const SEEK: u32 = 21; +/// A connection to this server holding the descriptions of the `len` ids in +/// the window, little-endian `u64`s, as ids 1.. in that order: the reply +/// carries its other end. It spends a served slot of this connection's share +/// from here, and names no path. +pub const EXPORT: u32 = 22; + +pub const SEEK_START: u64 = 0; +pub const SEEK_CURRENT: u64 = 1; +pub const SEEK_END: u64 = 2; // Replies, by frame type. pub const REPLY: u32 = 0x100; @@ -220,6 +241,8 @@ errors! { /// The kernel had no room for what the request needs: a handle, memory, /// a place in a queue. OutOfResources = 28, + /// A path on a connection that was given files and no directory. + NoDirectory = 32, } local { /// The server ended, and the connection with it. @@ -448,11 +471,17 @@ impl From for Refused { } } -/// One directory capability, connected. +/// Where a [`Dir`]'s connection comes from. +enum Source { + /// A directory capability, connected again through the same connector. + Named { names: &'static Namespace, name: [u8; MAX_SERVICE_NAME], len: usize }, + /// A connection another process exported: one connection, never another. + Exported, +} + +/// One directory capability, or one exported connection, connected. pub struct Dir { - names: &'static Namespace, - name: [u8; MAX_SERVICE_NAME], - name_len: usize, + source: Source, conn: Option, window: SharedMemory, generation: u64, @@ -470,16 +499,23 @@ impl Dir { } let mut buf = [0u8; MAX_SERVICE_NAME]; buf[..name.len()].copy_from_slice(name.as_bytes()); - let window = SharedMemory::create(WINDOW_BYTES).map_err(|e| Error::kernel("a window", e))?; - let mut dir = - Self { names, name: buf, name_len: name.len(), conn: None, window, generation: 0, writable: false, volume: 0 }; + let mut dir = Self::unconnected(Source::Named { names, name: buf, len: name.len() })?; dir.reconnect()?; Ok(dir) } - /// The capability's namespace name. - pub fn name(&self) -> &str { - core::str::from_utf8(&self.name[..self.name_len]).expect("copied from a str") + /// Serve the connection an [`EXPORT`] answered, whose descriptions are ids + /// 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)?; + 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, volume: 0 }) } /// Counts this directory's connections; a file id is good only on the one @@ -500,14 +536,22 @@ impl Dir { } /// 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. + /// 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; - let conn = match self.names.open(self.name()) { + let Source::Named { names, name, len } = &self.source else { return Err(Error::Gone) }; + let name = core::str::from_utf8(&name[..*len]).expect("copied from a str"); + let conn = match names.open(name) { Ok(conn) => conn, Err(SyscallError::NotFound) => return Err(Error::NotHeld), Err(e) => return Err(Error::kernel("a connection", e)), }; + 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.volume = reply.value2 | SERVED_VOLUME; @@ -602,24 +646,65 @@ impl Dir { let _ = self.fid_call(CLOSE, generation, &Request { fid, ..Request::new() }); } - /// Read at most `out.len()` bytes, and at most the window, at `offset`. - pub fn read(&mut self, fid: u64, generation: u64, offset: u64, out: &mut [u8]) -> Result { + /// 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: u64, generation: u64, out: &mut [u8]) -> Result { let len = out.len().min(WINDOW_BYTES); - let reply = self.fid_call(READ, generation, &Request { fid, offset, len: len as u64, ..Request::new() })?; + let reply = self.fid_call(READ, generation, &Request { fid, len: len as u64, ..Request::new() })?; let n = counted(reply.value, len)?; window_take(self.window(), 0, &mut out[..n]); Ok(n) } - /// Write at most the window of `data` at `offset`, or at the end for a file - /// opened to append. Answers how much, and the offset after it. - pub fn write(&mut self, fid: u64, generation: u64, offset: u64, data: &[u8]) -> Result { + /// Write at most the window of `data` at the description's offset, or at + /// the end for a file opened to append; the offset moves past it. Answers + /// how much, and the offset after it. + pub fn write(&mut self, fid: u64, generation: u64, data: &[u8]) -> Result { let len = data.len().min(WINDOW_BYTES); window_put(self.window(), 0, &data[..len]); - let reply = self.fid_call(WRITE, generation, &Request { fid, offset, len: len as u64, ..Request::new() })?; + let reply = self.fid_call(WRITE, generation, &Request { fid, len: len as u64, ..Request::new() })?; Ok(Written { len: counted(reply.value, len)?, offset: reply.value2 }) } + /// Move the description's offset `delta` from `whence` ([`SEEK_START`], + /// [`SEEK_CURRENT`] or [`SEEK_END`]): where it landed. + pub fn seek(&mut self, fid: u64, generation: u64, whence: u64, delta: i64) -> Result { + let request = Request { fid, offset: delta as u64, flags: whence, ..Request::new() }; + self.fid_call(SEEK, generation, &request).map(|r| r.value) + } + + /// A second id on `fid`'s description, sharing its offset. + pub fn dup(&mut self, fid: u64, generation: u64) -> Result { + self.fid_call(DUP, generation, &Request { fid, ..Request::new() }).map(|r| r.value) + } + + /// 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, generation: u64, fids: &[u64]) -> Result { + if generation != self.generation || self.conn.is_none() { + return Err(Error::Stale); + } + if fids.is_empty() || fids.len() * 8 > WINDOW_BYTES { + return Err(Error::InvalidRequest); + } + for (i, fid) in fids.iter().enumerate() { + window_put(self.window(), i * 8, &fid.to_le_bytes()); + } + let mut batch = [toyos_abi::HANDLE_INVALID; MAX_TRANSFER_HANDLES]; + let request = Request { len: fids.len() as u64, ..Request::new() }; + let (msg_type, reply, carried) = self.call_with(EXPORT, &request, &mut batch)?; + match (answer(msg_type, reply), carried) { + (Ok(_), 1) => Ok(Connection::received(batch[0])), + (Err(e), 0) => Err(e), + (_, n) => { + close_all(&batch[..n]); + self.conn = None; + Err(Error::Malformed) + } + } + } + pub fn fstat(&mut self, fid: u64, generation: u64) -> Result { let volume = self.volume; self.fid_call(FSTAT, generation, &Request { fid, ..Request::new() }).map(|r| stat_of(&r, volume)) @@ -643,15 +728,14 @@ impl Dir { }) } - /// A pipe whose bytes the server appends to this file from `offset` on, - /// for a program's standard output: the write end. - pub fn stream(&mut self, fid: u64, generation: u64, offset: u64) -> Result { + /// 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: u64, generation: u64) -> Result { if generation != self.generation || self.conn.is_none() { return Err(Error::Stale); } let mut batch = [toyos_abi::HANDLE_INVALID; MAX_TRANSFER_HANDLES]; - let (msg_type, reply, carried) = - self.call_with(STREAM, &Request { fid, offset, ..Request::new() }, &mut batch)?; + let (msg_type, reply, carried) = self.call_with(STREAM, &Request { fid, ..Request::new() }, &mut batch)?; match (answer(msg_type, reply), carried) { (Ok(_), 1) => Ok(Pipe(crate::OwnedHandle(batch[0]))), (Err(e), 0) => Err(e), @@ -957,7 +1041,7 @@ mod tests { /// refused by name, never read as another answer. #[test] fn a_reply_outside_the_protocol_is_malformed() { - let past = Error::WIRE.len() as u64 + 1; + let past = Error::WIRE.iter().filter_map(|e| e.code()).max().expect("wire words") + 1; // 29 to 31 are what the client's own words once were, and 0x80 on what // their discriminants are. for status in [past, 29, 30, 31, 0x80, 0x81, 0x82, u64::MAX, SyscallError::NotFound.to_u64()] { diff --git a/system/fileserver/src/absent.rs b/system/fileserver/src/absent.rs index fb9074e6ed..6b26b43345 100644 --- a/system/fileserver/src/absent.rs +++ b/system/fileserver/src/absent.rs @@ -53,8 +53,6 @@ impl Volume for Absent { Ok(()) } - fn hold(&mut self, _node: Node) {} - fn node_meta(&mut self, _node: Node) -> Result { Err(Error::NotFound) } diff --git a/system/fileserver/src/bounds.rs b/system/fileserver/src/bounds.rs index 24c63fccb9..4a7dba93f9 100644 --- a/system/fileserver/src/bounds.rs +++ b/system/fileserver/src/bounds.rs @@ -9,6 +9,11 @@ //! connection taken past its share's part of the handshakes is answered //! `TooManyConnections` and let go. //! +//! **An export takes its served slot when it is made**, of the share of the +//! connection that asked, and is refused there: a connection made for a +//! child is served from its export, waits on no handshake bound or timeout, +//! and its hello is not counted again. +//! //! **Every client number is bounded where it is kept**: an offset a request //! names, and an offset a stream grows by what it drains, stay at or below the //! largest a file has ([`MAX_FILE_BYTES`]). @@ -20,6 +25,8 @@ //! held to UTF-8 alone. use toyos::fs::{canonical, Error, MAX_FILE_BYTES, MAX_PATH, WINDOW_BYTES}; + +use crate::descriptions::MAX_IDS; use toyos::shm::Lent; /// Clients served at once, machine-wide: one connection per directory per @@ -59,15 +66,53 @@ pub fn handshake(share: u64, waiting: usize) -> Result<(), String> { } } -/// A hello, while `served` clients are served and `of_share` of them are its -/// share's. `window` is the region it lent as the kernel describes it, `None` -/// for a hello that lent none, and is asked only once the bounds have passed. -pub fn hello(served: usize, of_share: usize, window: impl FnOnce() -> Option>) -> Hello { +/// Whether a connection's served slot is still to be taken at its hello, or +/// was taken when it was exported. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Slot { + AtHello, + Taken, +} + +/// The slot a new served connection takes, while `served` clients are served +/// and `of_share` of them are its share's: `Err` is why it is refused. +fn served_slot(served: usize, of_share: usize) -> Result<(), String> { if served >= MAX_SERVED { - return Hello::Refused(format!("it is refused, since {MAX_SERVED} clients are served already")); + return Err(format!("it is refused, since {MAX_SERVED} clients are served already")); } if of_share >= SERVED_SHARE { - return Hello::Refused(format!("it is refused, since its share is served {SERVED_SHARE} clients already")); + return Err(format!("it is refused, since its share is served {SERVED_SHARE} clients already")); + } + Ok(()) +} + +/// An export, while `served` clients are served and `of_share` of them are +/// the asking connection's share's: it takes its slot here or is refused +/// `TooManyConnections`. +pub fn export(served: usize, of_share: usize) -> Result<(), Error> { + served_slot(served, of_share).map_err(|_| Error::TooManyConnections) +} + +/// How many ids an export names, read before its window is: 1 to +/// [`MAX_IDS`], what one connection holds, or `InvalidRequest`. +pub fn export_ids(count: u64) -> Result { + usize::try_from(count).ok().filter(|n| (1..=MAX_IDS).contains(n)).ok_or(Error::InvalidRequest) +} + +/// A hello, while `served` clients are served and `of_share` of them are its +/// share's; `slot` says whether it takes a slot here. `window` is the region +/// it lent as the kernel describes it, `None` for a hello that lent none, and +/// is asked only once the bounds have passed. +pub fn hello( + slot: Slot, + served: usize, + of_share: usize, + window: impl FnOnce() -> Option>, +) -> Hello { + if slot == Slot::AtHello { + if let Err(why) = served_slot(served, of_share) { + return Hello::Refused(why); + } } match window() { None => Hello::Dropped("its hello carried no window"), @@ -84,10 +129,14 @@ pub fn in_a_file(at: u64, len: u64) -> Result { at.checked_add(len).filter(|&end| end <= MAX_FILE_BYTES).ok_or(Error::FileTooLarge) } -/// A stream asked to start at `at`, while `streams` are served and `of_share` -/// of them are its share's. -pub fn stream(at: u64, streams: usize, of_share: usize) -> Result<(), Error> { - in_a_file(at, 0)?; +/// Where a seek of `delta` from `base` lands: at or below the largest offset a +/// file has and not before its start, or `InvalidRequest`. +pub fn seek(base: u64, delta: i64) -> Result { + base.checked_add_signed(delta).filter(|&at| at <= MAX_FILE_BYTES).ok_or(Error::InvalidRequest) +} + +/// A stream, while `streams` are served and `of_share` of them are its share's. +pub fn stream(streams: usize, of_share: usize) -> Result<(), Error> { match streams >= MAX_STREAMS || of_share >= STREAM_SHARE { true => Err(Error::TooManyStreams), false => Ok(()), @@ -140,9 +189,40 @@ mod tests { #[test] fn a_hello_past_the_server_s_bound_or_its_share_s_part_is_refused_before_its_window_is_asked() { - assert!(matches!(hello(MAX_SERVED, 0, unasked), Hello::Refused(why) if why.contains("128 clients are served"))); - assert!(matches!(hello(SERVED_SHARE, SERVED_SHARE, unasked), Hello::Refused(why) if why.contains("its share is served 32"))); - assert_eq!(hello(MAX_SERVED - 1, SERVED_SHARE - 1, || Some(Ok(()))), Hello::Served(())); + let at = Slot::AtHello; + assert!(matches!(hello(at, MAX_SERVED, 0, unasked), Hello::Refused(why) if why.contains("128 clients are served"))); + assert!(matches!(hello(at, SERVED_SHARE, SERVED_SHARE, unasked), Hello::Refused(why) if why.contains("its share is served 32"))); + assert_eq!(hello(at, MAX_SERVED - 1, SERVED_SHARE - 1, || Some(Ok(()))), Hello::Served(())); + } + + /// An export is refused at the bounds a hello is, before any connection is + /// made for it. + #[test] + fn an_export_past_the_server_s_bound_or_its_share_s_part_is_refused() { + assert_eq!(export(MAX_SERVED, 0), Err(Error::TooManyConnections)); + assert_eq!(export(SERVED_SHARE, SERVED_SHARE), Err(Error::TooManyConnections)); + assert_eq!(export(MAX_SERVED - 1, SERVED_SHARE - 1), Ok(())); + } + + /// An export's count is the client's word: none, more than a connection + /// holds, or one whose window span would overflow is refused before + /// anything is read. + #[test] + fn an_export_naming_no_ids_or_more_than_a_connection_holds_is_refused() { + for count in [0, MAX_IDS as u64 + 1, 1 << 61, u64::MAX] { + assert_eq!(export_ids(count), Err(Error::InvalidRequest), "{count:#x}"); + } + assert_eq!(export_ids(1), Ok(1)); + assert_eq!(export_ids(MAX_IDS as u64), Ok(MAX_IDS)); + } + + /// The exported connection's slot was taken at its export, so its hello is + /// served at a full server and share, which counted it already; its window + /// is still held to what a hello's is. + #[test] + fn an_exported_connection_s_hello_is_not_counted_again() { + assert_eq!(hello(Slot::Taken, MAX_SERVED, SERVED_SHARE, || Some(Ok(()))), Hello::Served(())); + assert_eq!(hello::<()>(Slot::Taken, 0, 0, || None), Hello::Dropped("its hello carried no window")); } /// The kernel's answer for a lent region that is not one region of @@ -157,19 +237,20 @@ mod tests { (None, "its hello carried no window"), ]; for (window, why) in refused { - assert_eq!(hello(0, 0, || window), Hello::Dropped(why)); + assert_eq!(hello(Slot::AtHello, 0, 0, || window), Hello::Dropped(why)); } } - /// A stream asked to start past the largest offset a file has is refused, - /// since its offset would grow by what it drains; and the same bound holds - /// a write's end and a truncation. + /// A seek lands at or below the largest offset a file has, and never + /// before its start; a write's end and a truncation are held to the same. #[test] fn an_offset_past_the_largest_a_file_has_is_refused() { - for at in [MAX_FILE_BYTES + 1, u64::MAX - 1] { - assert_eq!(stream(at, 0, 0), Err(Error::FileTooLarge), "a stream at {at:#x}"); - } - assert_eq!(stream(MAX_FILE_BYTES, 0, 0), Ok(())); + assert_eq!(seek(0, (MAX_FILE_BYTES + 1) as i64), Err(Error::InvalidRequest)); + assert_eq!(seek(MAX_FILE_BYTES, 1), Err(Error::InvalidRequest)); + assert_eq!(seek(3, -4), Err(Error::InvalidRequest), "before the start"); + assert_eq!(seek(u64::MAX, 1), Err(Error::InvalidRequest), "an overflow"); + assert_eq!(seek(10, -4), Ok(6)); + assert_eq!(seek(0, MAX_FILE_BYTES as i64), Ok(MAX_FILE_BYTES)); assert_eq!(in_a_file(MAX_FILE_BYTES - 2, 4), Err(Error::FileTooLarge), "a write straddling the ceiling"); assert_eq!(in_a_file(u64::MAX, 1), Err(Error::FileTooLarge), "a write whose end overflows"); assert_eq!(in_a_file(MAX_FILE_BYTES - 4, 4), Ok(MAX_FILE_BYTES)); @@ -177,9 +258,9 @@ mod tests { #[test] fn a_stream_past_the_server_s_bound_or_its_share_s_part_is_refused() { - assert_eq!(stream(0, MAX_STREAMS, 0), Err(Error::TooManyStreams)); - assert_eq!(stream(0, STREAM_SHARE, STREAM_SHARE), Err(Error::TooManyStreams)); - assert_eq!(stream(0, MAX_STREAMS - 1, STREAM_SHARE - 1), Ok(())); + assert_eq!(stream(MAX_STREAMS, 0), Err(Error::TooManyStreams)); + assert_eq!(stream(STREAM_SHARE, STREAM_SHARE), Err(Error::TooManyStreams)); + assert_eq!(stream(MAX_STREAMS - 1, STREAM_SHARE - 1), Ok(())); } /// A path put on the wire past std is held to what std would have sent diff --git a/system/fileserver/src/data.rs b/system/fileserver/src/data.rs index 49f530c6c0..daeefb4189 100644 --- a/system/fileserver/src/data.rs +++ b/system/fileserver/src/data.rs @@ -474,12 +474,6 @@ impl Volume for DataVolume { Ok(()) } - fn hold(&mut self, node: Node) { - if let Some(open) = self.open.get_mut(&node) { - open.holders += 1; - } - } - fn node_meta(&mut self, node: Node) -> Result { let open = self.entry(node)?; Ok(Meta { kind: Kind::File, size: open.size, mtime: open.mtime }) diff --git a/system/fileserver/src/descriptions.rs b/system/fileserver/src/descriptions.rs new file mode 100644 index 0000000000..34dcc56338 --- /dev/null +++ b/system/fileserver/src/descriptions.rs @@ -0,0 +1,158 @@ +//! Open files as POSIX has them: a **description** is the file, the access an +//! open fixed and the offset; a connection names descriptions by **id**, and +//! two ids, on one connection or on two, may name one description and share +//! its offset. +//! +//! A description lives while an id or a stream holds it, and its node is +//! closed when the last goes: every removal answers the node that is then +//! no longer held, for the server to close on the volume. + +use std::cell::RefCell; +use std::collections::BTreeMap; +use std::rc::Rc; + +use toyos::fs::Error; + +use crate::volume::Node; + +/// Ids one connection holds at once. +pub const MAX_IDS: usize = 1024; + +/// One open of a file, shared by every id and stream that names it. +#[derive(Debug, PartialEq, Eq)] +pub struct Description { + pub node: Node, + /// The file's `Volume::identity` when it was opened, which its renames keep. + pub identity: u64, + pub write: bool, + pub append: bool, + /// Where the next read or write that names no offset begins. + pub offset: u64, +} + +/// A description, shared. +pub type Shared = Rc>; + +/// The node `shared` held, when it was its last holder. +pub fn release(shared: Shared) -> Option { + Rc::try_unwrap(shared).ok().map(|d| d.into_inner().node) +} + +/// One connection's ids. +#[derive(Default)] +pub struct Ids { + ids: BTreeMap, + next: u64, +} + +impl Ids { + pub fn len(&self) -> usize { + self.ids.len() + } + + pub fn is_empty(&self) -> bool { + self.ids.is_empty() + } + + /// A new id for `shared`, or `TooManyOpen` past [`MAX_IDS`]. + pub fn insert(&mut self, shared: Shared) -> Result { + if self.ids.len() >= MAX_IDS { + return Err(Error::TooManyOpen); + } + self.next += 1; + self.ids.insert(self.next, shared); + Ok(self.next) + } + + pub fn get(&self, id: u64) -> Result<&Shared, Error> { + self.ids.get(&id).ok_or(Error::UnknownFile) + } + + /// A second id on `id`'s description. + pub fn dup(&mut self, id: u64) -> Result { + let shared = Rc::clone(self.get(id)?); + self.insert(shared) + } + + /// Forget `id`: the node, when it was its description's last holder. + pub fn close(&mut self, id: u64) -> Result, Error> { + self.ids.remove(&id).map(release).ok_or(Error::UnknownFile) + } + + /// The descriptions `ids` name, in order, for another connection: every + /// one known, and at most [`MAX_IDS`] of them. + pub fn export(&self, ids: &[u64]) -> Result { + if ids.is_empty() { + return Err(Error::InvalidRequest); + } + let mut out = Ids::default(); + for &id in ids { + out.insert(Rc::clone(self.get(id)?))?; + } + Ok(out) + } + + /// Every id gone with its connection: the nodes no longer held. + pub fn drain(&mut self) -> Vec { + core::mem::take(&mut self.ids).into_values().filter_map(release).collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn opened(node: Node) -> Shared { + Rc::new(RefCell::new(Description { node, identity: node, write: true, append: false, offset: 0 })) + } + + #[test] + fn ids_on_one_description_share_its_offset_and_its_node_closes_with_the_last() { + let mut ids = Ids::default(); + let first = ids.insert(opened(7)).unwrap(); + let second = ids.dup(first).unwrap(); + ids.get(first).unwrap().borrow_mut().offset = 42; + assert_eq!(ids.get(second).unwrap().borrow().offset, 42, "a dup has an offset of its own"); + assert_eq!(ids.close(first), Ok(None), "the node closed while a dup still held it"); + assert_eq!(ids.close(second), Ok(Some(7))); + assert_eq!(ids.close(second), Err(Error::UnknownFile)); + } + + /// An export installs the same descriptions on another connection, in the + /// order named, so the receiver's ids are 1.. and the offset is shared. + #[test] + fn an_export_shares_descriptions_across_connections_in_the_order_named() { + let mut parent = Ids::default(); + let a = parent.insert(opened(1)).unwrap(); + let b = parent.insert(opened(2)).unwrap(); + let mut child = parent.export(&[b, a]).unwrap(); + assert_eq!(child.get(1).unwrap().borrow().node, 2); + assert_eq!(child.get(2).unwrap().borrow().node, 1); + child.get(1).unwrap().borrow_mut().offset = 9; + assert_eq!(parent.get(b).unwrap().borrow().offset, 9, "the child moved an offset of its own"); + assert_eq!(parent.drain(), Vec::::new(), "the parent's end closed nodes the child holds"); + let mut closed = child.drain(); + closed.sort_unstable(); + assert_eq!(closed, [1, 2]); + } + + #[test] + fn an_export_naming_nothing_or_an_unknown_id_installs_nothing() { + let mut parent = Ids::default(); + let a = parent.insert(opened(1)).unwrap(); + assert!(matches!(parent.export(&[]), Err(Error::InvalidRequest))); + assert!(matches!(parent.export(&[a, a + 1]), Err(Error::UnknownFile))); + assert_eq!(parent.close(a), Ok(Some(1)), "a refused export kept a holder"); + } + + #[test] + fn a_connection_holds_at_most_max_ids() { + let mut ids = Ids::default(); + let first = ids.insert(opened(1)).unwrap(); + for _ in 1..MAX_IDS { + ids.dup(first).unwrap(); + } + assert_eq!(ids.dup(first), Err(Error::TooManyOpen)); + assert_eq!(ids.insert(opened(2)), Err(Error::TooManyOpen)); + } +} diff --git a/system/fileserver/src/fat.rs b/system/fileserver/src/fat.rs index 89d60e606f..3c5a97d6b7 100644 --- a/system/fileserver/src/fat.rs +++ b/system/fileserver/src/fat.rs @@ -323,12 +323,6 @@ impl Volume for FatVolume { Ok(()) } - fn hold(&mut self, node: Node) { - if let Some(open) = self.open.get_mut(&node) { - open.holders += 1; - } - } - fn node_meta(&mut self, node: Node) -> Result { let open = self.entry(node)?; let (path, size) = (open.path.clone(), open.file.len()); diff --git a/system/fileserver/src/lib.rs b/system/fileserver/src/lib.rs index 659d3e3465..6876c9fa14 100644 --- a/system/fileserver/src/lib.rs +++ b/system/fileserver/src/lib.rs @@ -4,8 +4,9 @@ //! can serve ([`data`] for the bcachefs DATA role, [`fat`] for FAT32's LOG and //! BOOT, [`absent`] for a role with no volume this boot), and the resolver that //! keeps every path inside the directory a connection was given ([`resolve`]), -//! which requests a read-only one is refused ([`rights`]), and what a client -//! may take of the server and name in a request ([`bounds`]). +//! which requests a read-only one is refused ([`rights`]), what a client +//! may take of the server and name in a request ([`bounds`]), and the open +//! files its connections share ([`descriptions`]). //! //! **This is the page cache.** A block of the volume — a btree node, a FAT //! sector, a file's data — is read into [`cache::Cache`] once and served from @@ -19,6 +20,7 @@ pub mod absent; pub mod bounds; pub mod cache; pub mod data; +pub mod descriptions; pub mod fat; pub mod ram; pub mod resolve; diff --git a/system/fileserver/src/main.rs b/system/fileserver/src/main.rs index 3b3fc5a544..250416cf61 100644 --- a/system/fileserver/src/main.rs +++ b/system/fileserver/src/main.rs @@ -25,6 +25,12 @@ //! (`toyos_manifest::launch`). Four at their parts take the server //! (`issues/four-shares-take-a-file-server.md`). //! +//! **An export is a connection this server makes to itself** through a private +//! port only it holds a connector to, for its life: connect, then accept, which +//! answers exactly that connection since this loop is single-threaded. The end +//! handed out cannot be duplicated, so it has one holder. A restart takes the +//! port with it, so nothing pending reaches a new instance. +//! //! **A server never blocks on a client.** Accept and the first frame are two //! events; a request is buffered until whole; every reply is one `try_send`, //! and a client whose pipe will not take one is dropped by name. The window a @@ -36,12 +42,15 @@ //! sooner. A kill loses what no sync covered, which is POSIX's promise and no //! more. +use std::cell::RefCell; use std::collections::BTreeMap; +use std::rc::Rc; use std::time::{Duration, Instant}; use fileserver::absent::Absent; -use fileserver::bounds::{self, Hello, MAX_HANDSHAKES, MAX_SERVED, MAX_STREAMS}; +use fileserver::bounds::{self, Hello, Slot, MAX_HANDSHAKES, MAX_SERVED, MAX_STREAMS}; use fileserver::data::{DataVolume, Located, Probed}; +use fileserver::descriptions::{self, Description, Ids, Shared}; use diskserver::disk::{Claimed, Disk, Served}; use fileserver::ram::Ram; use fileserver::fat::FatVolume; @@ -52,19 +61,31 @@ use fileserver::writeback::WriteBack; use toyos::endow::{self, Endowments}; use toyos::fs::*; use toyos::ipc::{self, Connection, RxStep}; +use toyos::namespace::Namespace; use toyos::poller::{Poller, READABLE}; -use toyos::port::Acceptor; +use toyos::port::{self, Acceptor}; use toyos::shm::SharedMemory; use toyos::volatile::Window; use toyos::Pipe; -use toyos_abi::syscall::{SyscallError, DEV_PREFIX, MAX_BADGE, SERVE_PREFIX}; - -/// Files one client holds open at once. -const MAX_FIDS: usize = 1024; +use toyos_abi::handle::Rights; +use toyos_abi::syscall::{self, SyscallError, DEV_PREFIX, MAX_BADGE, SERVE_PREFIX}; /// What one turn of a stream appends at most. const STREAM_READ: usize = 64 * 1024; +/// The turns that drain what one stream holds before a request on its +/// description: the kernel's whole pipe ring (`PIPE_SIZE`, 2 MiB), so a +/// writer that keeps writing cannot hold the loop. +const STREAM_PRESENT: usize = (2 << 20) / STREAM_READ; + +/// The name the private export port has in this server's own namespace. +const EXPORT_PORT: &str = "export"; + +/// What the end an export hands out carries: a connection's rights without +/// `DUP`, so whoever holds it is its one holder, and may pass it on. +const EXPORTED: Rights = + Rights::TRANSFER.union(Rights::WAIT).union(Rights::READ).union(Rights::WRITE).union(Rights::MAP); + // One wait watches the acceptor, every connection and every stream at once. const _: () = assert!(1 + MAX_SERVED + MAX_HANDSHAKES + MAX_STREAMS <= Poller::MAX_HANDLES as usize); @@ -110,39 +131,77 @@ impl Role { } } -/// One file a client holds open. -struct Fid { - node: Node, - /// The file's [`Volume::identity`] when it was opened, which its renames keep. - identity: u64, - write: bool, - append: bool, -} - struct Client { conn: Connection, rx: ipc::FrameRx<{ core::mem::size_of::() }>, - /// Its grant's root, beneath which every path it names is resolved. - root: String, - /// Its grant's share, which it spends. + /// Its grant's root, beneath which every path it names is resolved: + /// `None` for an exported connection, which names no path. + root: Option, + /// Its grant's share, which it spends; an exported connection's is the + /// share of the connection that exported it. share: u64, /// Its grant is read-write and the volume is: what it may change. writes: bool, window: Option, - fids: BTreeMap, - next_fid: u64, + fids: Ids, since: Instant, } -/// A pipe a program writes and this server appends to a file. +impl Client { + /// Whether it was exported: its served slot was taken then, so it waits on + /// no handshake. + fn exported(&self) -> bool { + self.root.is_none() + } + + /// Whether it holds a served slot. + fn served(&self) -> bool { + self.window.is_some() || self.exported() + } + + /// Whether it is a handshake still waiting on its hello. + fn waiting(&self) -> bool { + !self.served() + } +} + +/// A pipe a program writes and this server writes to a file at its +/// description's offset. struct Stream { pipe: Pipe, - node: Node, - offset: u64, + file: Shared, /// The share of the client that asked for it, which it spends. share: u64, } +/// The private port exports are made on: its acceptor, and the namespace +/// holding this server's one connector to it. +struct Exports { + acceptor: Acceptor, + names: Namespace, +} + +impl Exports { + fn new() -> Self { + let (acceptor, connector) = port::create().unwrap_or_else(|e| panic!("fileserver: no export port ({e:?})")); + let names = toyos::namespace::build() + .add(EXPORT_PORT, &connector) + .finish() + .unwrap_or_else(|e| panic!("fileserver: no namespace for the export port ({e:?})")); + Self { acceptor, names } + } + + /// A connection to this server: the end to hand away, narrowed to + /// [`EXPORTED`], and this server's. + fn connect(&self) -> Result<(Connection, Connection), Error> { + let theirs = self.names.open(EXPORT_PORT).map_err(|e| Error::kernel("an export's connection", e))?; + let ours = self.acceptor.accept().unwrap_or_else(|e| panic!("fileserver: the export port refused an accept ({e:?})")); + let narrowed = syscall::dup_narrowed(toyos::AsHandle::as_handle(&theirs), EXPORTED) + .map_err(|e| Error::kernel("an export's narrowed end", e))?; + Ok((Connection::received(narrowed), ours)) + } +} + fn main() { let args: Vec = std::env::args().collect(); let role = args.get(1).and_then(|r| Role::parse(r)).unwrap_or_else(|| { @@ -176,6 +235,7 @@ fn main() { next_client: 0, streams: BTreeMap::new(), next_stream: 0, + exports: Exports::new(), writeback: WriteBack::default(), scratch: Vec::new(), } @@ -310,6 +370,7 @@ struct Server { next_client: u64, streams: BTreeMap, next_stream: u64, + exports: Exports, /// When the sync nobody asked for is due. writeback: WriteBack, /// A write's bytes, copied out of the client's window or a stream's pipe @@ -364,7 +425,7 @@ impl Server { let poller = Poller::new(Poller::MAX_HANDLES); let mut ready = Vec::new(); loop { - if self.clients.values().filter(|c| c.window.is_none()).count() < MAX_HANDSHAKES { + if self.clients.values().filter(|c| c.waiting()).count() < MAX_HANDSHAKES { poller.watch(&self.acceptor, READABLE, TOKEN_ACCEPTOR); } for (id, c) in &self.clients { @@ -377,7 +438,7 @@ impl Server { let timeout = self .clients .values() - .filter(|c| c.window.is_none()) + .filter(|c| c.waiting()) .map(|c| HANDSHAKE_TIMEOUT.saturating_sub(now.duration_since(c.since))) .chain(self.writeback.left(now)) .min() @@ -398,7 +459,7 @@ impl Server { let late: Vec = self .clients .iter() - .filter(|(_, c)| c.window.is_none() && now.duration_since(c.since) >= HANDSHAKE_TIMEOUT) + .filter(|(_, c)| c.waiting() && now.duration_since(c.since) >= HANDSHAKE_TIMEOUT) .map(|(id, _)| *id) .collect(); for id in late { @@ -429,7 +490,7 @@ impl Server { }, Err(why) => return println!("fileserver: letting a connection go: it carries no grant ({why:?})"), }; - let mine = self.clients.values().filter(|c| c.share == grant.share && c.window.is_none()).count(); + let mine = self.clients.values().filter(|c| c.share == grant.share && c.waiting()).count(); if let Err(why) = bounds::handshake(grant.share, mine) { // Let go whether or not the refusal went: either way it is said. let _ = conn.try_send(REPLY, &Reply::refused(Error::TooManyConnections)); @@ -440,26 +501,25 @@ impl Server { let client = Client { conn, rx: ipc::FrameRx::new(), - root: grant.root.to_string(), + root: Some(grant.root.to_string()), share: grant.share, writes: grant.access == Access::ReadWrite && self.volume.writable(), window: None, - fids: BTreeMap::new(), - next_fid: 1, + fids: Ids::default(), since: Instant::now(), }; self.clients.insert(id, client); } fn drop_client(&mut self, id: u64, why: &str) { - let Some(client) = self.clients.remove(&id) else { return }; + let Some(mut client) = self.clients.remove(&id) else { return }; if !why.is_empty() { println!("fileserver: dropping client {id} of {:?}, share {}: {why}", client.root, client.share); } - for fid in client.fids.values() { + for node in client.fids.drain() { // Nobody is left to answer: a refused close is said, and its node // kept for the next sync. - let _ = self.volume.close(fid.node); + let _ = self.volume.close(node); } } @@ -504,8 +564,12 @@ impl Server { } } - /// A path from the window: `len` bytes at `at`, UTF-8. + /// A path from the window: `len` bytes at `at`, UTF-8. A connection that + /// was given files and no directory names none. fn path(&self, id: u64, at: u64, len: u64) -> Result { + if self.clients[&id].root.is_none() { + return Err(Error::NoDirectory); + } let window = self.clients[&id].window.as_ref().expect("a request comes after the window"); let (at, len) = bounds::path_span(at, len)?; let mut bytes = vec![0u8; len]; @@ -516,7 +580,7 @@ impl Server { /// `rel`, from the client's directory, as a volume path. fn resolve(&mut self, id: u64, rel: &str, follow_last: bool) -> Result { bounds::relative(rel)?; - let root = self.clients[&id].root.clone(); + let root = self.clients[&id].root.clone().ok_or(Error::NoDirectory)?; let volume = &mut self.volume; let mut lookup = |p: &str| match volume.lstat(p) { Ok(meta) if meta.kind == Kind::Symlink => volume.read_link(p).map(Found::Link).map_err(drop), @@ -539,16 +603,15 @@ impl Server { fn answer(&mut self, id: u64, op: u32, r: Request) -> Answer { if op == HELLO { - let share = self.clients[&id].share; - let served = self.clients.values().filter(|c| c.window.is_some()); - let (all, mine) = served.fold((0, 0), |(all, mine), c| (all + 1, mine + usize::from(c.share == share))); + let (all, mine) = self.served(self.clients[&id].share); let client = self.clients.get_mut(&id).expect("pumped"); if client.window.is_some() { return Answer::Drop("it lent a second window"); } + let slot = if client.exported() { Slot::Taken } else { Slot::AtHello }; // The window travels in the HELLO frame: its reader took it with the frame. let lent = || client.rx.take_handles::<1>().map(|[lent]| SharedMemory::lent(lent, WINDOW_BYTES)); - match bounds::hello(all, mine, lent) { + match bounds::hello(slot, all, mine, lent) { Hello::Served(window) => client.window = Some(window), Hello::Refused(why) => return Answer::Refuse(Error::TooManyConnections, why), Hello::Dropped(why) => return Answer::Drop(why), @@ -568,8 +631,32 @@ impl Server { } } - fn fid(&self, id: u64, fid: u64) -> Result<&Fid, Error> { - self.clients[&id].fids.get(&fid).ok_or(Error::UnknownFile) + /// How many clients hold a served slot, and how many of them are `share`'s. + fn served(&self, share: u64) -> (usize, usize) { + let served = self.clients.values().filter(|c| c.served()); + served.fold((0, 0), |(all, mine), c| (all + 1, mine + usize::from(c.share == share))) + } + + /// The description `fid` names on client `id`. + fn fid(&self, id: u64, fid: u64) -> Result { + self.clients[&id].fids.get(fid).cloned() + } + + /// The description `fid` names, with what its streams hold already + /// written at its offset, so a request that reads or moves the offset + /// comes after every byte a stream was given before it. + fn settled(&mut self, id: u64, fid: u64) -> Result { + let file = self.fid(id, fid)?; + let streams: Vec = + self.streams.iter().filter(|(_, s)| Rc::ptr_eq(&s.file, &file)).map(|(sid, _)| *sid).collect(); + for sid in streams { + for _ in 0..STREAM_PRESENT { + if !self.drain(sid) { + break; + } + } + } + Ok(file) } /// Mark the volume written: the write-back sync is due from here. @@ -602,7 +689,7 @@ impl Server { Resolved::Absolute(p) => return Ok(self.link(id, &p)), Resolved::Path(p) => p, }; - if self.clients[&id].fids.len() >= MAX_FIDS { + if self.clients[&id].fids.len() >= descriptions::MAX_IDS { return Err(Error::TooManyOpen); } let how = OpenHow { @@ -623,34 +710,40 @@ impl Server { if how.create || how.truncate || how.create_new { self.dirtied(); } - let client = self.clients.get_mut(&id).expect("pumped"); - let fid = client.next_fid; - client.next_fid += 1; let write = r.flags & (O_WRITE | O_APPEND) != 0; let append = r.flags & O_APPEND != 0; + let file = Description { node, identity, write, append, offset: 0 }; let client = self.clients.get_mut(&id).expect("pumped"); - client.fids.insert(fid, Fid { node, identity, write, append }); + let fid = client.fids.insert(Rc::new(RefCell::new(file)))?; Ok(Answer::Reply(Reply { value: fid, ..stat_reply(meta, identity) })) } CLOSE => { let client = self.clients.get_mut(&id).expect("pumped"); - let fid = client.fids.remove(&r.fid).ok_or(Error::UnknownFile)?; - self.volume.close(fid.node)?; + if let Some(node) = client.fids.close(r.fid)? { + self.volume.close(node)?; + } Ok(Answer::Reply(Reply::ok())) } + DUP => { + let client = self.clients.get_mut(&id).expect("pumped"); + let fid = client.fids.dup(r.fid)?; + Ok(Answer::Reply(Reply { value: fid, ..Reply::ok() })) + } READ => { - let node = self.fid(id, r.fid)?.node; + let file = self.settled(id, r.fid)?; let len = (r.len as usize).min(WINDOW_BYTES); let window = window_of(self.clients[&id].window.as_ref().expect("lent")).sub(0, len); - let n = self.volume.read(node, r.offset, &mut WindowOut(window))?; + let (node, at) = { + let f = file.borrow(); + (f.node, f.offset) + }; + let n = self.volume.read(node, at, &mut WindowOut(window))?; + file.borrow_mut().offset = at + n as u64; Ok(Answer::Reply(Reply { value: n as u64, ..Reply::ok() })) } WRITE => { - let (node, write, append) = { - let f = self.fid(id, r.fid)?; - (f.node, f.write, f.append) - }; - if !write { + let file = self.settled(id, r.fid)?; + if !file.borrow().write { return Err(Error::NotOpenForWriting); } let len = r.len as usize; @@ -662,47 +755,62 @@ impl Server { } let window = window_of(self.clients[&id].window.as_ref().expect("lent")); window_take(window, 0, &mut self.scratch[..len]); - let at = if append { self.volume.node_meta(node)?.size } else { r.offset }; - bounds::in_a_file(at, len as u64)?; - self.volume.write(node, at, &self.scratch[..len])?; - self.dirtied(); - Ok(Answer::Reply(Reply { value: len as u64, value2: at + len as u64, ..Reply::ok() })) + let end = self.write_at(&file, len)?; + Ok(Answer::Reply(Reply { value: len as u64, value2: end, ..Reply::ok() })) + } + SEEK => { + let file = self.settled(id, r.fid)?; + let base = match r.flags { + SEEK_START => 0, + SEEK_CURRENT => file.borrow().offset, + SEEK_END => self.volume.node_meta(file.borrow().node)?.size, + _ => return Err(Error::InvalidRequest), + }; + let at = bounds::seek(base, r.offset as i64)?; + file.borrow_mut().offset = at; + Ok(Answer::Reply(Reply { value: at, ..Reply::ok() })) } + EXPORT => self.export(id, r.len), FSTAT => { - let (node, identity) = self.fid(id, r.fid).map(|f| (f.node, f.identity))?; + let (node, identity) = { + let f = self.settled(id, r.fid)?; + let f = f.borrow(); + (f.node, f.identity) + }; let meta = self.volume.node_meta(node)?; Ok(Answer::Reply(stat_reply(meta, identity))) } TRUNCATE => { - let f = self.fid(id, r.fid)?; - if !f.write { + let file = self.fid(id, r.fid)?; + let (node, write) = { + let f = file.borrow(); + (f.node, f.write) + }; + if !write { return Err(Error::NotOpenForWriting); } - let node = f.node; bounds::in_a_file(r.offset, 0)?; self.volume.truncate(node, r.offset)?; self.dirtied(); Ok(Answer::Reply(Reply::ok())) } FSYNC => { - let node = self.fid(id, r.fid)?.node; + let node = self.settled(id, r.fid)?.borrow().node; self.sync(Some(node)) } SYNC => self.sync(None), STREAM => { - let f = self.fid(id, r.fid)?; - if !f.write { + let file = self.fid(id, r.fid)?; + if !file.borrow().write { return Err(Error::NotOpenForWriting); } - let node = f.node; let share = self.clients[&id].share; let mine = self.streams.values().filter(|s| s.share == share).count(); - bounds::stream(r.offset, self.streams.len(), mine)?; + bounds::stream(self.streams.len(), mine)?; let (read, write) = toyos::pipe_pair().map_err(|e| Error::kernel("a stream's pipe", e))?; - self.volume.hold(node); let sid = self.next_stream; self.next_stream += 1; - self.streams.insert(sid, Stream { pipe: read, node, offset: r.offset, share }); + self.streams.insert(sid, Stream { pipe: read, file, share }); Ok(Answer::WithHandle(Reply::ok(), write.into())) } STAT | LSTAT => { @@ -821,32 +929,71 @@ impl Server { Answer::Link(path.len()) } - /// One read of what a stream's writer has put in the pipe, appended to - /// its file, and no more: a pipe with more waiting fires at the next - /// wait, after every other client that was ready — [`Self::pump`]'s rule. - fn drain(&mut self, sid: u64) { + /// Write the scratch's first `len` bytes to `file` at its offset, or at + /// its end when it appends, and move the offset past them: the offset. + fn write_at(&mut self, file: &Shared, len: usize) -> Result { + let (node, append, offset) = { + let f = file.borrow(); + (f.node, f.append, f.offset) + }; + let at = if append { self.volume.node_meta(node)?.size } else { offset }; + let end = bounds::in_a_file(at, len as u64)?; + self.volume.write(node, at, &self.scratch[..len])?; + self.dirtied(); + file.borrow_mut().offset = end; + Ok(end) + } + + /// A connection to this server for another process, holding the + /// descriptions of the `count` ids in client `id`'s window. Its served + /// slot is taken here, of `id`'s share, or the export is refused. + fn export(&mut self, id: u64, count: u64) -> Result { + let count = bounds::export_ids(count)?; + let mut words = vec![0u8; count * 8]; + window_take(window_of(self.clients[&id].window.as_ref().expect("lent")), 0, &mut words); + let fids: Vec = + words.chunks_exact(8).map(|w| u64::from_le_bytes(w.try_into().expect("eight bytes"))).collect(); + let parent = &self.clients[&id]; + let (share, writes) = (parent.share, parent.writes); + let fids = parent.fids.export(&fids)?; + let (all, mine) = self.served(share); + bounds::export(all, mine)?; + let (theirs, ours) = self.exports.connect()?; + let client = Client { + conn: ours, + rx: ipc::FrameRx::new(), + root: None, + share, + writes, + window: None, + fids, + since: Instant::now(), + }; + let child = self.next_client; + self.next_client += 1; + self.clients.insert(child, client); + Ok(Answer::WithHandle(Reply { value: count as u64, ..Reply::ok() }, theirs.into_owned())) + } + + /// One read of what a stream's writer has put in the pipe, written to its + /// file at its description's offset, and no more: a pipe with more waiting + /// fires at the next wait, after every other client that was ready — + /// [`Self::pump`]'s rule. Whether it wrote anything. + fn drain(&mut self, sid: u64) -> bool { if self.scratch.len() < STREAM_READ { self.scratch.resize(STREAM_READ, 0); } - let Some(stream) = self.streams.get_mut(&sid) else { return }; + let Some(stream) = self.streams.get(&sid) else { return false }; let ended = match stream.pipe.read_nonblock(&mut self.scratch[..STREAM_READ]) { - Err(SyscallError::WouldBlock) => return, + Err(SyscallError::WouldBlock) => return false, Ok(0) | Err(SyscallError::Gone) => None, Err(e) => Some(format!("its pipe would not read ({e:?})")), Ok(n) => { - let (node, at) = (stream.node, stream.offset); - match bounds::in_a_file(at, n as u64).ok() { - None => Some("it reached the largest offset a file has".to_string()), - Some(end) => { - stream.offset = end; - match self.volume.write(node, at, &self.scratch[..n]) { - Ok(()) => { - self.dirtied(); - return; - } - Err(e) => Some(format!("its write failed ({e:?})")), - } - } + let file = Rc::clone(&stream.file); + match self.write_at(&file, n) { + Ok(_) => return true, + Err(Error::FileTooLarge) => Some("it reached the largest offset a file has".to_string()), + Err(e) => Some(format!("its write failed ({e:?})")), } } }; @@ -854,9 +1001,12 @@ impl Server { println!("fileserver: a stream is ended: {why}"); } if let Some(stream) = self.streams.remove(&sid) { - // Nobody is left to answer: a refused close is said, and its node - // kept for the next sync. - let _ = self.volume.close(stream.node); + if let Some(node) = descriptions::release(stream.file) { + // Nobody is left to answer: a refused close is said, and its + // node kept for the next sync. + let _ = self.volume.close(node); + } } + false } } diff --git a/system/fileserver/src/rights.rs b/system/fileserver/src/rights.rs index 12a6bdb2d7..eba3a0afa0 100644 --- a/system/fileserver/src/rights.rs +++ b/system/fileserver/src/rights.rs @@ -13,7 +13,9 @@ use toyos::fs::*; pub fn changes(op: u32, flags: u64) -> Option { match op { OPEN => Some(flags & (O_WRITE | O_APPEND | O_CREATE | O_TRUNCATE | O_CREATE_NEW) != 0), - HELLO | CLOSE | READ | STAT | LSTAT | FSTAT | FSYNC | SYNC | READDIR | READLINK => Some(false), + HELLO | CLOSE | READ | STAT | LSTAT | FSTAT | FSYNC | SYNC | READDIR | READLINK | DUP | SEEK | EXPORT => { + Some(false) + } WRITE | TRUNCATE | MKDIR | RMDIR | UNLINK | RENAME | SYMLINK | STREAM => Some(true), _ => None, } @@ -25,7 +27,7 @@ mod tests { /// Every request the wire has, by what it does to the directory: the /// independent spelling `changes` is held to. - const READS: [u32; 10] = [HELLO, CLOSE, READ, STAT, LSTAT, FSTAT, FSYNC, SYNC, READDIR, READLINK]; + const READS: [u32; 13] = [HELLO, CLOSE, READ, STAT, LSTAT, FSTAT, FSYNC, SYNC, READDIR, READLINK, DUP, SEEK, EXPORT]; const WRITES: [u32; 8] = [WRITE, TRUNCATE, MKDIR, RMDIR, UNLINK, RENAME, SYMLINK, STREAM]; #[test] @@ -39,7 +41,7 @@ mod tests { // Every request number the wire has is one of the two, or `OPEN`. let mut named: Vec = READS.iter().chain(&WRITES).copied().chain([OPEN]).collect(); named.sort_unstable(); - assert_eq!(named, (HELLO..=SYNC).collect::>()); + assert_eq!(named, (HELLO..=EXPORT).collect::>()); } /// An open changes the directory by any one flag that writes, creates or @@ -58,7 +60,7 @@ mod tests { /// client is let go whatever its grant. #[test] fn a_request_the_wire_does_not_have_is_neither() { - for op in [0, SYNC + 1, REPLY, LINK, u32::MAX] { + for op in [0, EXPORT + 1, REPLY, LINK, u32::MAX] { assert_eq!(changes(op, 0), None, "request {op}"); } } diff --git a/system/fileserver/src/volume.rs b/system/fileserver/src/volume.rs index 16ad41fc8a..2fb7c53b4d 100644 --- a/system/fileserver/src/volume.rs +++ b/system/fileserver/src/volume.rs @@ -98,9 +98,6 @@ pub trait Volume { /// next sync to write, and the refusal is the last holder's answer. fn close(&mut self, node: Node) -> Result<(), Error>; - /// Hold the node once more, for a second file id or a stream. - fn hold(&mut self, node: Node); - fn node_meta(&mut self, node: Node) -> Result; /// The number of the file `path` names, without following it 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 new file mode 100644 index 0000000000..acfd552975 --- /dev/null +++ b/tests/guest/rust-tests/src/bin/an_export_shares_its_descriptions.rs @@ -0,0 +1,76 @@ +//! A file server's export: a connection it makes to itself for another +//! process, holding descriptions an open made, which names no path. +//! +//! - The exported connection says hello and is served, and its ids are the +//! exported descriptions, 1.. in the order named. +//! - A read or a write through either connection moves the one offset both +//! share, and a seek through one is where the other reads next. +//! - A path on the exported connection is refused `NoDirectory`. +//! - Closing the exporter's id leaves the description to the exported one. +//! - What a stream was given before a write through an id is written first: +//! the write lands after every byte of it, and the file ends there. + +use toyos::fs::{Dir, Error, Refused, O_CREATE, O_READ, O_TRUNCATE, O_WRITE, SEEK_CURRENT, SEEK_END, SEEK_START}; + +const DIR: &str = "fs:/home"; +const FILE: &str = "an-export-shares-its-descriptions"; +const STREAMED: &str = "an-export-shares-its-descriptions-streamed"; + +/// What the stream is given: half the pipe's ring, so it is all in the pipe +/// before the write through the id is asked, and the server has drained at +/// most a turn or two of it. +const STREAM_BYTES: usize = 1 << 20; + +fn main() { + let names = toyos::endow::namespace().expect("a test is endowed a namespace"); + let mut parent = Dir::connect(names, DIR).expect("the home directory"); + let opened = parent.open(FILE, O_READ | O_WRITE | O_CREATE | O_TRUNCATE).expect("open"); + let (fid, generation) = (opened.fid, opened.generation); + let written = parent.write(fid, generation, b"abcdef").expect("write"); + assert_eq!((written.len, written.offset), (6, 6)); + + let end = parent.export(generation, &[fid]).expect("export"); + let mut child = Dir::adopt(end).expect("the exported connection says hello"); + let theirs = child.generation(); + assert_eq!(child.seek(1, theirs, SEEK_CURRENT, 0), Ok(6), "the export's id 1 has an offset of its own"); + assert_eq!(child.seek(1, theirs, SEEK_START, 2), Ok(2)); + let mut buf = [0u8; 2]; + assert_eq!(parent.read(fid, generation, &mut buf), Ok(2)); + assert_eq!(&buf, b"cd", "the exporter read away from where the export moved the offset"); + assert_eq!(child.read(1, theirs, &mut buf), Ok(2)); + assert_eq!(&buf, b"ef", "the export read away from where the exporter left the offset"); + println!("an_export_shares_its_descriptions: one offset, two connections"); + + assert_eq!(child.stat(FILE, true), Err(Refused::Error(Error::NoDirectory))); + assert!(matches!(child.open(FILE, O_READ), Err(Refused::Error(Error::NoDirectory)))); + println!("an_export_shares_its_descriptions: the export names no path"); + + parent.close(fid, generation); + assert_eq!(child.seek(1, theirs, SEEK_START, 0), Ok(0), "the description went with the exporter's id"); + assert_eq!(child.read(1, theirs, &mut buf), Ok(2)); + assert_eq!(&buf, b"ab"); + child.close(1, theirs); + parent.unlink(FILE).expect("unlink"); + println!("an_export_shares_its_descriptions: the description outlived its exporter's id"); + + let opened = parent.open(STREAMED, O_READ | O_WRITE | O_CREATE | O_TRUNCATE).expect("open the streamed file"); + let (fid, generation) = (opened.fid, opened.generation); + let stream = parent.stream(fid, generation).expect("a stream on the description"); + let pattern: Vec = (0..STREAM_BYTES).map(|i| (i % 251) as u8).collect(); + let mut sent = 0; + while sent < pattern.len() { + sent += stream.write(&pattern[sent..]).expect("the stream took its bytes"); + } + let written = parent.write(fid, generation, b"tail").expect("a write through the id"); + assert_eq!(written.offset, STREAM_BYTES as u64 + 4, "the write did not land after what the stream was given"); + assert_eq!(parent.seek(fid, generation, SEEK_END, 0), Ok(STREAM_BYTES as u64 + 4), "the file does not end at the tail"); + assert_eq!(parent.seek(fid, generation, SEEK_START, STREAM_BYTES as i64 - 2), Ok(STREAM_BYTES as u64 - 2)); + let mut tail = [0u8; 6]; + assert_eq!(parent.read(fid, generation, &mut tail), Ok(6)); + assert_eq!(&tail[..2], &pattern[STREAM_BYTES - 2..], "the stream's last bytes are not before the tail"); + assert_eq!(&tail[2..], b"tail"); + drop(stream); + parent.close(fid, generation); + parent.unlink(STREAMED).expect("unlink the streamed file"); + println!("an_export_shares_its_descriptions: ok"); +}