diff --git a/issues/anonymous-memory-swaps-to-an-encrypted-partition.md b/issues/anonymous-memory-swaps-to-an-encrypted-partition.md index 747d4424227..7d7a5325355 100644 --- a/issues/anonymous-memory-swaps-to-an-encrypted-partition.md +++ b/issues/anonymous-memory-swaps-to-an-encrypted-partition.md @@ -134,18 +134,29 @@ The symbols: atomically, and is refused whole if any is not free. A referenced block is never written. - **Free after durable.** `clean_end` comes only after a flush answers durable - for every write of the batch. The block client keeps acknowledged writes and - reissues them after a loss (`lib/blockring/src/client.rs`), and durable rests - on that. + for every write of the batch. The block client keeps an acknowledged + write's blocks, which the service hands the session shared, until a flush + covers it, and reissues it from them after a loss + (`lib/blockring/src/client.rs`); durable rests on that. - **Install before freeing.** Swap-in installs the frame and frees the block in one hold of the space lock. - **Re-dirtying.** A store, a kernel write copy, or a `mem_write` into a page being cleaned keeps it dirty, and keeps its frame. - **Arena reserve.** A swap session's arena has 511 blocks and its ring has 64 - entries. - - Writes hold at most 447 blocks and 48 entries, leaving 64 blocks and 16 - entries for swap-in reads. - - The service flushes once 223 acknowledged writes are held. + entries, and both carry only requests on the wire: an acknowledged write + holds neither. + - Writes on the wire hold at most 447 blocks and 48 entries, leaving 64 + blocks and 16 entries for swap-in reads. + - What the session keeps shared is bounded by the batch: `clean_end` waits + for a flush covering it. Past `diskserver::session::KEPT_BLOCKS` the + session flushes for itself before the next write. The service's batch size + is its own choice and needs no arena arithmetic. + - **A swap-in read can wait behind a write.** Requests go out in the order + asked, so a read asked after a write waits while that write has no + contiguous arena room, or waits at the session's kept limit for its flush. + The 64-block reserve bounds how many reads fit, not that wait. If swap-in + latency matters, reads go out ahead of waiting writes, which the client's + one outbox does not do today. - **`CLEANING` on private leaves.** - A store faults and clears it. So does a kernel write copy, which takes its pin under the space lock. A read copy leaves it. diff --git a/issues/the-file-servers-clean-blocks-are-given-back-to-no-other-job.md b/issues/the-file-servers-clean-blocks-are-given-back-to-no-other-job.md index b0f7b290b08..c07fa931631 100644 --- a/issues/the-file-servers-clean-blocks-are-given-back-to-no-other-job.md +++ b/issues/the-file-servers-clean-blocks-are-given-back-to-no-other-job.md @@ -23,13 +23,15 @@ interim, not a sizing. A clean block the allocator refuses, or whose place in the eviction order it refuses, is not kept: a read is still answered, and a flushed block is already -on the disk. Four allocations still have no refusal path, so a server whose job is out of memory can still -abort at them: +on the disk. These allocations still have no refusal path, so a server whose +job is out of memory can still abort at them: - the block map's node, a small one every few inserts of a clean or a dirty block; +- a block's share count, a small one per block kept; - a dirty block; - a fetch's staging buffer; -- a flush's list of the dirty blocks and its buffer for one run. +- a flush's list of runs, the writes it keeps in flight, and each run's list + of shares. **Exit**: the file servers drop clean blocks when the memory service reports pressure, and `CLEAN_SHARE` is deleted. diff --git a/lib/blockdev/src/lib.rs b/lib/blockdev/src/lib.rs index 92d476ec2a4..6375848ec58 100644 --- a/lib/blockdev/src/lib.rs +++ b/lib/blockdev/src/lib.rs @@ -33,6 +33,13 @@ pub const BLOCK: usize = 4096; /// One block. pub type Block = [u8; BLOCK]; +/// A block's bytes shared between whoever wrote them and a device that may +/// write them again: a served disk keeps a write's until a flush covers it. +/// The block is boxed on its own so the file server's cache can allocate a +/// block it reads in fallibly (`system/fileserver/src/cache.rs`, `boxed`), +/// which `Rc::new` of the block itself cannot. +pub type SharedBlock = Rc>; + /// Why a device did not do what it was asked. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum Error { diff --git a/lib/blockring/src/client.rs b/lib/blockring/src/client.rs index 8d65fec4894..bf8c5012c79 100644 --- a/lib/blockring/src/client.rs +++ b/lib/blockring/src/client.rs @@ -7,16 +7,29 @@ //! that is not on the wire is the server breaking the session //! ([`Violation::Tag`]), not a second answer. //! -//! **An acknowledged write is kept until a flush covers it.** Its arena blocks -//! stay pinned ([`Client::take_released`] is when they come back) because a -//! loss — a flush answered [`Status::Lost`] or [`Status::Device`], or a session -//! that ended — means the device may no longer hold it, and the only copy left -//! is the client's. After a loss every such write is issued again, in the order -//! it was first acknowledged and never beside an earlier one it overlaps, and -//! nothing else goes on the wire until they are all acknowledged again; the -//! flush that was waiting is then asked again. A flush is answered -//! [`Outcome::Durable`] only by a device flush that succeeded after every loss -//! it could have been told about. +//! **The arena holds only what is on the wire.** A read or a write is given +//! its arena blocks as it goes out ([`Client::next_request`]'s `place`) and +//! gives them back when it is answered ([`Client::take_released`]); nothing +//! waiting to go out, and nothing acknowledged, holds any. +//! +//! **A write's payload is kept until a flush covers it.** The client holds +//! what the caller gave with the write ([`Ask::Write`]) from submit until the +//! write is answered anything but [`Outcome::Done`], or a covering flush or a +//! give-up ends it, and hands it to `place` each time the write goes out. A +//! loss (a flush answered [`Status::Lost`] or [`Status::Device`], or a session +//! that ended) means the device may no longer hold an acknowledged write, and +//! the payload is the only copy left. After a loss every such write is issued +//! again, in the order it was first acknowledged and never beside an earlier +//! one it overlaps, and nothing else goes on the wire until they are all +//! acknowledged again; the flush that was waiting is then asked again. A flush +//! is answered [`Outcome::Durable`] only by a device flush that succeeded +//! after every loss it could have been told about. +//! +//! **What it keeps is bounded.** Once the acknowledged writes it keeps reach +//! the limit it was made with, the caller's next write waits while the client +//! asks a flush of its own. If that flush fails past [`MAX_ATTEMPTS`], writes +//! that meet the limit are answered [`Outcome::Device`] unsent, until a +//! caller's flush succeeds. //! //! **A write given up is a loss no later flush can answer for.** One the device //! refused on more than [`MAX_ATTEMPTS`] reissues is dropped, and its caller @@ -31,7 +44,8 @@ //! //! **Requests in flight at once are unordered.** A caller must not have two in //! flight whose ranges overlap when either is a write, nor reuse arena blocks -//! before they are released. +//! before they are released. Requests go out in the order asked, so one that +//! waits (for arena room, or behind the limit) holds back those after it. use alloc::collections::{BTreeMap, VecDeque}; use alloc::vec::Vec; @@ -41,6 +55,24 @@ use toyos::transport::{Inflight, Run, Violation}; use crate::entry::{Completion, Op, Request, Status}; use crate::layout::DEPTH; +/// What a caller asks for. A transfer names its blocks, not the arena blocks +/// it will move through: those are given as it goes on the wire. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub enum Ask

{ + Read { lba: u64, blocks: u32 }, + Write { lba: u64, blocks: u32, payload: P }, + Flush, +} + +/// A transfer going on the wire, for `place` to give arena blocks. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub enum Placing<'a, P> { + /// To be filled by the device, for the caller's `ticket`. + Read { ticket: Ticket, blocks: u32 }, + /// To be filled with `payload` before it goes. + Write { payload: &'a P, blocks: u32 }, +} + /// The caller's name for one thing it asked for. pub type Ticket = u64; @@ -72,9 +104,10 @@ pub enum Outcome { /// A write acknowledged and not yet covered by a flush. #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] -struct Acked { +struct Acked

{ lba: u64, - run: Run, + blocks: u32, + payload: P, /// When it was first acknowledged, which is the order it is issued again /// in. first: u64, @@ -83,48 +116,86 @@ struct Acked { attempts: u32, } -impl Acked { - fn overlaps_range(&self, lba: u64, blocks: u32) -> bool { - self.lba < lba + u64::from(blocks) && lba < self.lba + u64::from(self.run.count()) - } -} - /// Why a request is being sent. #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] enum Kind { /// The caller asked for it. User(Ticket), /// An acknowledged write, issued again after a loss. - Reissue(Acked), + Reissue { first: u64, attempts: u32 }, /// One attempt at the caller's flush `Ticket`. Flush(Ticket), + /// One attempt at the client's own flush, for the writes it keeps. + Pushback, } /// A request not yet on the wire. #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] -struct Queued { - op: Op, +struct Queued

{ + ask: Ask

, kind: Kind, } -/// A request on the wire, and for a flush the acknowledgement it covers below -/// and the losses it was sent after. +impl

Queued

{ + fn reissue(acked: Acked

) -> Self { + let Acked { lba, blocks, payload, first, attempts, .. } = acked; + Self { ask: Ask::Write { lba, blocks, payload }, kind: Kind::Reissue { first, attempts } } + } + + /// The write this reissue is, acknowledged again at `seq`. + fn acked(self, seq: u64) -> Acked

{ + match (self.ask, self.kind) { + (Ask::Write { lba, blocks, payload }, Kind::Reissue { first, attempts }) => { + Acked { lba, blocks, payload, first, seq, attempts } + } + _ => unreachable!("blockring: only a reissue is acknowledged again"), + } + } + + fn is_flush(&self) -> bool { + matches!(self.kind, Kind::Flush(_) | Kind::Pushback) + } +} + +/// A request on the wire, the op it went as, and for a flush the +/// acknowledgement it covers below and the losses it was sent after. #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] -struct Sent { - queued: Queued, +struct Sent

{ + queued: Queued

, + op: Op, covers: u64, losses: u64, } -/// `D` is the most requests it has on the wire at once: a session's [`DEPTH`]. -#[derive(Clone, Debug, PartialEq, Eq, Hash, Default)] -pub struct Client { +impl

Sent

{ + fn run(&self) -> Option { + match self.op { + Op::Read { run, .. } | Op::Write { run, .. } => Some(run), + Op::Flush => None, + } + } +} + +/// The client's own flush. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +enum Pushback { + Idle, + /// Asked, and failed this many times so far. + Asked(u32), + /// Failed past [`MAX_ATTEMPTS`]: a write that meets the limit is refused. + Failed, +} + +/// `P` is what a write carries for `place`; `D` is the most requests it has +/// on the wire at once: a session's [`DEPTH`]. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct Client { up: bool, /// Not yet on the wire, in order. Writes being issued again are always at /// the front, then any flush attempt waiting on them. - outbox: VecDeque, - wire: Inflight, - acked: VecDeque, + outbox: VecDeque>, + wire: Inflight, D>, + acked: VecDeque>, next_seq: u64, /// Losses taken so far: a flush sent before one says nothing about what it /// lost. @@ -132,101 +203,153 @@ pub struct Client { attempts: BTreeMap, answers: VecDeque<(Ticket, Outcome)>, released: VecDeque, + /// Blocks of the acknowledged writes kept, wherever they are. + kept: u64, + /// The most blocks kept before the caller's writes wait for a flush. + limit: u64, + pushback: Pushback, /// Writes put on the wire again after a loss, over the client's life. reissued: u64, /// An acknowledged write was given up: no flush is durable again. gave_up: bool, } -impl Client { +impl Client { /// A client with no session yet: requests wait until - /// [`Self::session_started`]. - pub fn new() -> Self { - Self::default() + /// [`Self::session_started`]. It keeps at most `limit` blocks of + /// acknowledged writes, and what was on the wire when it reached them, + /// before the caller's writes wait for a flush. + pub fn new(limit: u64) -> Self { + Self { + up: false, + outbox: VecDeque::new(), + wire: Inflight::new(), + acked: VecDeque::new(), + next_seq: 0, + losses: 0, + attempts: BTreeMap::new(), + answers: VecDeque::new(), + released: VecDeque::new(), + kept: 0, + limit, + pushback: Pushback::Idle, + reissued: 0, + gave_up: false, + } } /// Ask for a read, a write or a flush. - pub fn submit(&mut self, ticket: Ticket, op: Op) { - let kind = match op { - Op::Flush => Kind::Flush(ticket), - Op::Read { .. } | Op::Write { .. } => Kind::User(ticket), + pub fn submit(&mut self, ticket: Ticket, ask: Ask

) { + let kind = match ask { + Ask::Flush => Kind::Flush(ticket), + Ask::Read { .. } | Ask::Write { .. } => Kind::User(ticket), }; - self.outbox.push_back(Queued { op, kind }); + self.outbox.push_back(Queued { ask, kind }); } fn reissuing(&self) -> bool { - self.wire.values().any(|sent| matches!(sent.queued.kind, Kind::Reissue(_))) - || self.outbox.front().is_some_and(|q| matches!(q.kind, Kind::Reissue(_))) + self.wire.values().any(|sent| matches!(sent.queued.kind, Kind::Reissue { .. })) + || self.outbox.front().is_some_and(|q| matches!(q.kind, Kind::Reissue { .. })) } - /// The next request to put on the wire, tagged; `None` while there is no - /// session, nothing to send, `D` on the wire, or what is next must - /// wait for writes being issued again. - pub fn next_request(&mut self) -> Option { - if !self.up { + /// The next request to put on the wire, tagged, its transfer placed in the + /// arena by `place`; `None` while there is no session, nothing to send, + /// `D` on the wire, what is next must wait for writes being issued again + /// or for the client's own flush, or `place` has no room for it, which a + /// request on the wire answered makes. + pub fn next_request(&mut self, mut place: impl FnMut(Placing<'_, P>) -> Option) -> Option { + if !self.up || self.wire.values().count() == D { return None; } - let front = *self.outbox.front()?; - match front.kind { - // Never beside a write on the wire it overlaps: the device may - // apply two writes in flight at once in either order, and the one - // on the wire is either an earlier one being issued again or a - // later one of the caller's. - Kind::Reissue(acked) => { - let blocked = self.wire.values().any(|sent| { - matches!(sent.queued.op, Op::Write { run, lba } if acked.overlaps_range(lba, run.count())) - }); - if blocked { - return None; - } - } - Kind::User(_) | Kind::Flush(_) => { - if self.reissuing() { - return None; + loop { + let front = self.outbox.front()?; + match (front.kind, &front.ask) { + // Never beside a write on the wire it overlaps: the device may + // apply two writes in flight at once in either order, and the + // one on the wire is either an earlier one being issued again + // or a later one of the caller's. + (Kind::Reissue { .. }, &Ask::Write { lba, blocks, .. }) => { + let blocked = self.wire.values().any(|sent| { + matches!(sent.op, Op::Write { run, lba: at } if at < lba + u64::from(blocks) && lba < at + u64::from(run.count())) + }); + if blocked { + return None; + } } + (_, _) if self.reissuing() => return None, + (Kind::User(ticket), Ask::Write { .. }) if self.kept >= self.limit => match self.pushback { + Pushback::Failed => { + self.outbox.pop_front(); + self.answers.push_back((ticket, Outcome::Device)); + continue; + } + Pushback::Asked(_) => return None, + // A caller's flush on the wire covers what it keeps. + Pushback::Idle if self.wire.values().any(|sent| matches!(sent.queued.kind, Kind::Flush(_))) => { + return None + } + Pushback::Idle => { + self.pushback = Pushback::Asked(0); + self.queue_flush(Kind::Pushback); + continue; + } + }, + _ => {} } + break; } - let tag = self.wire.insert(Sent { queued: front, covers: self.next_seq, losses: self.losses }).ok()?; - self.outbox.pop_front(); - if matches!(front.kind, Kind::Reissue(_)) { + let front = self.outbox.front()?; + let op = match &front.ask { + Ask::Flush => Op::Flush, + &Ask::Read { lba, blocks } => { + let Kind::User(ticket) = front.kind else { unreachable!("blockring: a read is the caller's") }; + Op::Read { run: place(Placing::Read { ticket, blocks })?, lba } + } + Ask::Write { lba, blocks, payload } => Op::Write { run: place(Placing::Write { payload, blocks: *blocks })?, lba: *lba }, + }; + let queued = self.outbox.pop_front().expect("just seen"); + if matches!(queued.kind, Kind::Reissue { .. }) { self.reissued += 1; } - Some(Request { op: front.op, tag }) + let sent = Sent { queued, op, covers: self.next_seq, losses: self.losses }; + let tag = self.wire.insert(sent).ok().expect("blockring: a slot counted free was taken"); + Some(Request { op, tag }) } /// What the server answered. pub fn complete(&mut self, completion: Completion) -> Result<(), Violation> { let sent = self.wire.answer(completion.tag)?; - let q = sent.queued; - match (q.kind, completion.status) { + // Answered, whatever the word: its arena blocks are free once a read's + // data has been taken, which the caller does before it releases them. + self.released.extend(sent.run()); + let Sent { queued, op, covers, losses } = sent; + match (queued.kind, completion.status) { (Kind::User(ticket), Status::Ok) => { - match q.op { - Op::Write { run, lba } => { + match (queued.ask, op) { + (Ask::Write { lba, blocks, payload }, Op::Write { .. }) => { let seq = self.bump(); - let acked = Acked { lba, run, first: seq, seq, attempts: 0 }; + self.kept += u64::from(blocks); + let acked = Acked { lba, blocks, payload, first: seq, seq, attempts: 0 }; // Sent before a loss it did not see: the device may // have taken it and lost it, and an earlier write it // overlaps is being issued again — so it is issued // again too, after that one. - if sent.losses != self.losses { + if losses != self.losses { self.requeue_reissue(acked); } else { self.acked.push_back(acked); } } - Op::Read { run, .. } => self.released.push_back(run), - Op::Flush => return Err(Violation::Entry), + (Ask::Read { .. }, Op::Read { .. }) => {} + _ => return Err(Violation::Entry), } self.answers.push_back((ticket, Outcome::Done)); } (Kind::User(ticket), status) => { - if let Op::Read { run, .. } | Op::Write { run, .. } = q.op { - self.released.push_back(run); - } let outcome = match status { Status::Invalid => Outcome::Invalid, Status::Device => Outcome::Device, - Status::ReadOnly if matches!(q.op, Op::Write { .. }) => Outcome::ReadOnly, + Status::ReadOnly if matches!(op, Op::Write { .. }) => Outcome::ReadOnly, // A read or a write answered `Lost`, or a read answered // `ReadOnly`, is a server that does not know which op it // was. @@ -237,23 +360,25 @@ impl Client { // A write the session took once, refused by the same grant on its // way again, or a flush refused as a write: a server that does // not know what it granted. - (Kind::Reissue(_) | Kind::Flush(_), Status::ReadOnly) => return Err(Violation::Entry), - (Kind::Reissue(mut acked), Status::Ok) => { - acked.seq = self.bump(); + (Kind::Reissue { .. } | Kind::Flush(_) | Kind::Pushback, Status::ReadOnly) => return Err(Violation::Entry), + (Kind::Reissue { .. }, Status::Ok) => { + let seq = self.bump(); + let acked = queued.acked(seq); // A loss since it went out: an earlier write it overlaps may // be going out again, so it goes again, after that one. - if sent.losses != self.losses { + if losses != self.losses { self.requeue_reissue(acked); } else { self.insert_acked(acked); } } - (Kind::Reissue(mut acked), _) => { + (Kind::Reissue { .. }, _) => { + let mut acked = queued.acked(0); acked.attempts += 1; if acked.attempts > MAX_ATTEMPTS { // The device will not take the only copy there is: every // flush waiting is told so, and the write is gone. - self.released.push_back(acked.run); + self.kept -= u64::from(acked.blocks); self.gave_up = true; self.fail_waiting_flushes(); } else { @@ -262,17 +387,18 @@ impl Client { } // A flush that succeeded after a loss it was sent before covers // writes that are being issued again: it is asked again, after them. - (Kind::Flush(ticket), Status::Ok) if sent.losses != self.losses => { - self.queue_flush_attempt(ticket); - } - (Kind::Flush(ticket), Status::Ok) => { - while self.acked.front().is_some_and(|a| a.seq < sent.covers) { + (kind @ (Kind::Flush(_) | Kind::Pushback), Status::Ok) if losses != self.losses => self.queue_flush(kind), + (kind @ (Kind::Flush(_) | Kind::Pushback), Status::Ok) => { + while self.acked.front().is_some_and(|a| a.seq < covers) { let a = self.acked.pop_front().expect("just seen"); - self.released.push_back(a.run); + self.kept -= u64::from(a.blocks); + } + self.pushback = Pushback::Idle; + if let Kind::Flush(ticket) = kind { + self.attempts.remove(&ticket); + let outcome = if self.gave_up { Outcome::Device } else { Outcome::Durable }; + self.answers.push_back((ticket, outcome)); } - self.attempts.remove(&ticket); - let outcome = if self.gave_up { Outcome::Device } else { Outcome::Durable }; - self.answers.push_back((ticket, outcome)); } (Kind::Flush(ticket), _) => { self.lost(); @@ -282,7 +408,20 @@ impl Client { self.attempts.remove(&ticket); self.answers.push_back((ticket, Outcome::Device)); } else { - self.queue_flush_attempt(ticket); + self.queue_flush(Kind::Flush(ticket)); + } + } + (Kind::Pushback, _) => { + self.lost(); + // A caller's flush that covered the writes since has made it + // moot: nothing waits on it. + if let Pushback::Asked(failed) = self.pushback { + if failed + 1 > MAX_ATTEMPTS { + self.pushback = Pushback::Failed; + } else { + self.pushback = Pushback::Asked(failed + 1); + self.queue_flush(Kind::Pushback); + } } } } @@ -296,26 +435,21 @@ impl Client { self.wire.end(|_, sent| wire.push(sent)); let mut flushes = Vec::new(); for sent in wire { - let q = sent.queued; - match q.kind { + self.released.extend(sent.run()); + match sent.queued.kind { Kind::User(ticket) => { #[cfg(not(feature = "mutate-session-end-forgets"))] - { - if let Op::Read { run, .. } | Op::Write { run, .. } = q.op { - self.released.push_back(run); - } - self.answers.push_back((ticket, Outcome::Refused)); - } + self.answers.push_back((ticket, Outcome::Refused)); #[cfg(feature = "mutate-session-end-forgets")] let _ = ticket; } - Kind::Reissue(acked) => self.requeue_reissue(acked), - Kind::Flush(ticket) => flushes.push(ticket), + Kind::Reissue { .. } => self.requeue_reissue(sent.queued.acked(0)), + kind @ (Kind::Flush(_) | Kind::Pushback) => flushes.push(kind), } } self.lost(); - for ticket in flushes { - self.queue_flush_attempt(ticket); + for kind in flushes { + self.queue_flush(kind); } } @@ -345,6 +479,11 @@ impl Client { self.outbox.is_empty() && self.wire.values().next().is_none() && self.acked.is_empty() } + /// Blocks of acknowledged writes kept for a flush to cover. + pub fn kept(&self) -> u64 { + self.kept + } + /// The tags on the wire. pub fn on_the_wire(&self) -> impl Iterator + '_ { self.wire.tags() @@ -362,33 +501,34 @@ impl Client { } /// Keep `acked` in first-acknowledged order. - fn insert_acked(&mut self, acked: Acked) { + fn insert_acked(&mut self, acked: Acked

) { let at = self.acked.iter().position(|a| a.first > acked.first).unwrap_or(self.acked.len()); self.acked.insert(at, acked); } /// Put `acked` back among the writes being issued again, in order. - fn requeue_reissue(&mut self, acked: Acked) { + fn requeue_reissue(&mut self, acked: Acked

) { let at = self .outbox .iter() .position(|q| match q.kind { - Kind::Reissue(other) => other.first > acked.first, - Kind::User(_) | Kind::Flush(_) => true, + Kind::Reissue { first, .. } => first > acked.first, + Kind::User(_) | Kind::Flush(_) | Kind::Pushback => true, }) .unwrap_or(self.outbox.len()); - self.outbox.insert(at, Queued { op: Op::Write { run: acked.run, lba: acked.lba }, kind: Kind::Reissue(acked) }); + self.outbox.insert(at, Queued::reissue(acked)); } - /// Ask the flush `ticket` again, once every write being issued again is - /// back: after them and before anything the caller asked for since. - fn queue_flush_attempt(&mut self, ticket: Ticket) { + /// Ask a flush, the caller's or the client's own, once every write being + /// issued again is back: after them and before anything the caller asked + /// for since. + fn queue_flush(&mut self, kind: Kind) { let at = self .outbox .iter() - .position(|q| !matches!(q.kind, Kind::Reissue(_) | Kind::Flush(_))) + .position(|q| !(matches!(q.kind, Kind::Reissue { .. }) || q.is_flush())) .unwrap_or(self.outbox.len()); - self.outbox.insert(at, Queued { op: Op::Flush, kind: Kind::Flush(ticket) }); + self.outbox.insert(at, Queued { ask: Ask::Flush, kind }); } /// The device may no longer hold any write acknowledged and not covered: @@ -404,8 +544,8 @@ impl Client { } } - /// Every flush waiting to be asked again is answered: a write it covers - /// is gone. + /// Every caller's flush waiting to be asked again is answered: a write it + /// covers is gone. fn fail_waiting_flushes(&mut self) { let mut kept = VecDeque::with_capacity(self.outbox.len()); for q in self.outbox.drain(..) { @@ -414,7 +554,7 @@ impl Client { self.attempts.remove(&ticket); self.answers.push_back((ticket, Outcome::Device)); } - Kind::User(_) | Kind::Reissue(_) => kept.push_back(q), + Kind::User(_) | Kind::Reissue { .. } | Kind::Pushback => kept.push_back(q), } } self.outbox = kept; @@ -426,56 +566,122 @@ mod tests { use super::*; use crate::layout::ARENA; - fn run(first: u32, count: u32) -> Run { - ARENA.run(first, count).unwrap() + /// A write's payload here is its ticket. + type C = Client; + + /// A limit no test here reaches unless it means to. + const ROOMY: u64 = 1 << 20; + + /// The arena blocks every test gives a transfer: four from its ticket's + /// fourth, so each ticket's are its own and a test can name them. + fn run_of(ticket: Ticket, blocks: u32) -> Run { + ARENA.run(ticket as u32 * 4, blocks).unwrap() + } + + fn place(p: Placing<'_, Ticket>) -> Option { + Some(match p { + Placing::Read { ticket, blocks } => run_of(ticket, blocks), + Placing::Write { payload, blocks } => run_of(*payload, blocks), + }) + } + + fn next(client: &mut C) -> Option { + client.next_request(place) } - fn answer(client: &mut Client, request: Request, status: Status) { + fn answer(client: &mut C, request: Request, status: Status) { client.complete(Completion { tag: request.tag, status }).unwrap(); } + fn write(ticket: Ticket, lba: u64, blocks: u32) -> Ask { + Ask::Write { lba, blocks, payload: ticket } + } + + fn up(limit: u64) -> C { + let mut client = C::new(limit); + client.session_started(); + client + } + #[test] fn nothing_goes_out_before_a_session() { - let mut client: Client = Client::new(); - client.submit(1, Op::Read { run: run(0, 1), lba: 0 }); - assert_eq!(client.next_request(), None); + let mut client = C::new(ROOMY); + client.submit(1, Ask::Read { lba: 0, blocks: 1 }); + assert_eq!(next(&mut client), None); client.session_started(); - assert!(client.next_request().is_some()); + assert!(next(&mut client).is_some()); } #[test] fn a_second_answer_for_one_tag_is_a_violation() { - let mut client: Client = Client::new(); - client.session_started(); - client.submit(1, Op::Read { run: run(0, 1), lba: 0 }); - let r = client.next_request().unwrap(); + let mut client = up(ROOMY); + client.submit(1, Ask::Read { lba: 0, blocks: 1 }); + let r = next(&mut client).unwrap(); answer(&mut client, r, Status::Ok); assert_eq!(client.complete(Completion { tag: r.tag, status: Status::Ok }), Err(Violation::Tag)); } + /// An acknowledged write gives its arena blocks back at once, and is kept + /// only until the flush that covers it. + #[test] + fn a_write_holds_arena_blocks_only_on_the_wire() { + let mut client = up(ROOMY); + client.submit(1, write(1, 5, 2)); + let w = next(&mut client).unwrap(); + assert_eq!(w.op, Op::Write { run: run_of(1, 2), lba: 5 }); + answer(&mut client, w, Status::Ok); + assert_eq!(client.take_released().collect::>(), [run_of(1, 2)]); + assert_eq!(client.kept(), 2, "acknowledged, not yet covered"); + client.submit(2, Ask::Flush); + let f = next(&mut client).unwrap(); + answer(&mut client, f, Status::Ok); + assert_eq!(client.kept(), 0); + assert!(client.quiet()); + } + + /// A request `place` has no room for waits, unsent and unanswered, and + /// goes once there is room. + #[test] + fn a_request_without_arena_room_waits_for_it() { + let mut client = up(ROOMY); + client.submit(1, write(1, 0, 1)); + assert_eq!(client.next_request(|_| None), None); + assert_eq!(client.take_answers().count(), 0); + let w = next(&mut client).unwrap(); + assert_eq!(w.op, Op::Write { run: run_of(1, 1), lba: 0 }); + } + /// A write acknowledged, then a flush answered `Lost`: the write goes out - /// again before the flush is asked again, and the flush's caller hears - /// `Durable` only from the second. + /// again with its own payload before the flush is asked again, and the + /// flush's caller hears `Durable` only from the second. #[test] fn a_lost_flush_reissues_then_asks_again() { - let mut client: Client = Client::new(); - client.session_started(); - client.submit(1, Op::Write { run: run(3, 2), lba: 5 }); - let w = client.next_request().unwrap(); + let mut client = up(ROOMY); + client.submit(1, write(1, 5, 2)); + let w = next(&mut client).unwrap(); answer(&mut client, w, Status::Ok); - client.submit(2, Op::Flush); - let f = client.next_request().unwrap(); + client.submit(2, Ask::Flush); + let f = next(&mut client).unwrap(); answer(&mut client, f, Status::Lost); - let again = client.next_request().unwrap(); - assert_eq!(again.op, Op::Write { run: run(3, 2), lba: 5 }); - assert_eq!(client.next_request(), None, "the flush waits for the write"); + let mut placed = Vec::new(); + let again = client + .next_request(|p| { + if let Placing::Write { payload, blocks } = p { + placed.push((*payload, blocks)); + } + place(p) + }) + .unwrap(); + assert_eq!(placed, [(1, 2)]); + assert_eq!(again.op, Op::Write { run: run_of(1, 2), lba: 5 }); + assert_eq!(next(&mut client), None, "the flush waits for the write"); answer(&mut client, again, Status::Ok); - let f2 = client.next_request().unwrap(); + let f2 = next(&mut client).unwrap(); assert_eq!(f2.op, Op::Flush); answer(&mut client, f2, Status::Ok); let answers: Vec<_> = client.take_answers().collect(); assert_eq!(answers, [(1, Outcome::Done), (2, Outcome::Durable)]); - assert_eq!(client.take_released().collect::>(), [run(3, 2)]); + assert_eq!(client.take_released().collect::>(), [run_of(1, 2), run_of(1, 2)]); assert!(client.quiet()); } @@ -483,48 +689,47 @@ mod tests { /// other, in the order they were first acknowledged. #[test] fn overlapping_writes_go_out_again_in_order() { - let mut client: Client = Client::new(); - client.session_started(); - client.submit(1, Op::Write { run: run(0, 2), lba: 0 }); - let a = client.next_request().unwrap(); + let mut client = up(ROOMY); + client.submit(1, write(1, 0, 2)); + let a = next(&mut client).unwrap(); answer(&mut client, a, Status::Ok); - client.submit(2, Op::Write { run: run(2, 1), lba: 1 }); - let b = client.next_request().unwrap(); + client.submit(2, write(2, 1, 1)); + let b = next(&mut client).unwrap(); answer(&mut client, b, Status::Ok); client.session_ended(); client.session_started(); - let first = client.next_request().unwrap(); - assert_eq!(first.op, Op::Write { run: run(0, 2), lba: 0 }); - assert_eq!(client.next_request(), None, "the overlapping one waits"); + let first = next(&mut client).unwrap(); + assert_eq!(first.op, Op::Write { run: run_of(1, 2), lba: 0 }); + assert_eq!(next(&mut client), None, "the overlapping one waits"); answer(&mut client, first, Status::Ok); - let second = client.next_request().unwrap(); - assert_eq!(second.op, Op::Write { run: run(2, 1), lba: 1 }); + let second = next(&mut client).unwrap(); + assert_eq!(second.op, Op::Write { run: run_of(2, 1), lba: 1 }); } /// A write the device refused on every reissue is gone, and its caller was - /// told it was done: a flush asked for after that is answered the loss, - /// never durable. + /// told it was done: it is kept no longer, and a flush asked for after + /// that is answered the loss, never durable. #[test] fn a_write_given_up_poisons_every_later_flush() { - let mut client: Client = Client::new(); - client.session_started(); - client.submit(1, Op::Write { run: run(0, 1), lba: 0 }); - let w = client.next_request().unwrap(); + let mut client = up(ROOMY); + client.submit(1, write(1, 0, 1)); + let w = next(&mut client).unwrap(); answer(&mut client, w, Status::Ok); client.session_ended(); for _ in 0..=MAX_ATTEMPTS { client.session_started(); - let again = client.next_request().unwrap(); - assert_eq!(again.op, Op::Write { run: run(0, 1), lba: 0 }); + let again = next(&mut client).unwrap(); + assert_eq!(again.op, Op::Write { run: run_of(1, 1), lba: 0 }); answer(&mut client, again, Status::Device); } - client.submit(2, Op::Flush); - let f = client.next_request().unwrap(); + assert_eq!(client.kept(), 0); + client.submit(2, Ask::Flush); + let f = next(&mut client).unwrap(); assert_eq!(f.op, Op::Flush); answer(&mut client, f, Status::Ok); assert_eq!(client.take_answers().collect::>(), [(1, Outcome::Done), (2, Outcome::Device)]); - client.submit(3, Op::Flush); - let f = client.next_request().unwrap(); + client.submit(3, Ask::Flush); + let f = next(&mut client).unwrap(); answer(&mut client, f, Status::Ok); assert_eq!(client.take_answers().collect::>(), [(3, Outcome::Device)]); } @@ -534,32 +739,101 @@ mod tests { /// not know which op it answered. #[test] fn read_only_answers_a_write_alone() { - let mut client: Client = Client::new(); - client.session_started(); - client.submit(1, Op::Write { run: run(0, 1), lba: 0 }); - let w = client.next_request().unwrap(); + let mut client = up(ROOMY); + client.submit(1, write(1, 0, 1)); + let w = next(&mut client).unwrap(); answer(&mut client, w, Status::ReadOnly); assert_eq!(client.take_answers().collect::>(), [(1, Outcome::ReadOnly)]); - assert_eq!(client.take_released().collect::>(), [run(0, 1)]); - for op in [Op::Read { run: run(1, 1), lba: 0 }, Op::Flush] { - client.submit(2, op); - let r = client.next_request().unwrap(); + assert_eq!(client.take_released().collect::>(), [run_of(1, 1)]); + assert_eq!(client.kept(), 0); + for ask in [Ask::Read { lba: 0, blocks: 1 }, Ask::Flush] { + client.submit(2, ask); + let r = next(&mut client).unwrap(); assert_eq!(client.complete(Completion { tag: r.tag, status: Status::ReadOnly }), Err(Violation::Entry)); } } #[test] fn what_was_on_the_wire_at_the_end_is_refused_and_what_was_not_waits() { - let mut client: Client = Client::new(); - client.session_started(); - client.submit(1, Op::Write { run: run(0, 1), lba: 0 }); - client.submit(2, Op::Read { run: run(1, 1), lba: 4 }); - let _ = client.next_request().unwrap(); + let mut client = up(ROOMY); + client.submit(1, write(1, 0, 1)); + client.submit(2, Ask::Read { lba: 4, blocks: 1 }); + let _ = client.next_request(place).unwrap(); client.session_ended(); assert_eq!(client.take_answers().collect::>(), [(1, Outcome::Refused)]); - assert_eq!(client.take_released().collect::>(), [run(0, 1)]); + assert_eq!(client.take_released().collect::>(), [run_of(1, 1)]); client.session_started(); - let r = client.next_request().unwrap(); - assert_eq!(r.op, Op::Read { run: run(1, 1), lba: 4 }); + let r = next(&mut client).unwrap(); + assert_eq!(r.op, Op::Read { run: run_of(2, 1), lba: 4 }); + } + + /// A long run of writes with no flush of the caller's: once the writes + /// kept reach the limit, the next waits while the client flushes for them, + /// and what it keeps never passes the limit by more than was on the wire. + #[test] + fn writes_past_the_limit_wait_for_the_clients_own_flush() { + let mut client = up(4); + let mut most = 0; + for t in 0..32 { + client.submit(t, write(t, t * 2, 2)); + loop { + let r = next(&mut client).expect("a write or the client's flush goes out"); + answer(&mut client, r, Status::Ok); + most = most.max(client.kept()); + if r.op != Op::Flush { + break; + } + } + assert!(client.kept() <= 4, "kept {} blocks after write {t}", client.kept()); + } + assert_eq!(most, 4); + assert!(client.take_answers().all(|(_, o)| o == Outcome::Done), "the caller hears only its own writes"); + } + + /// A write at the limit with a caller's flush on the wire waits for that + /// flush rather than asking one of the client's own. + #[test] + fn a_write_at_the_limit_waits_for_a_callers_flush_on_the_wire() { + let mut client = up(1); + client.submit(1, write(1, 0, 1)); + let w = next(&mut client).unwrap(); + answer(&mut client, w, Status::Ok); + client.submit(2, Ask::Flush); + let f = next(&mut client).unwrap(); + client.submit(3, write(3, 4, 1)); + assert_eq!(next(&mut client), None, "the write goes out, or the client asks its own flush"); + answer(&mut client, f, Status::Ok); + let w = next(&mut client).unwrap(); + assert_eq!(w.op, Op::Write { run: run_of(3, 1), lba: 4 }); + } + + /// The client's own flush, refused past `MAX_ATTEMPTS`, refuses a write + /// that meets the limit unsent, as the device's word; a caller's flush + /// that succeeds lets writes go again. + #[test] + fn a_failed_own_flush_refuses_writes_until_a_flush_succeeds() { + let mut client = up(2); + client.submit(1, write(1, 0, 2)); + let w = next(&mut client).unwrap(); + answer(&mut client, w, Status::Ok); + client.submit(2, write(2, 8, 1)); + for _ in 0..=MAX_ATTEMPTS { + let f = next(&mut client).expect("the client's flush"); + assert_eq!(f.op, Op::Flush); + answer(&mut client, f, Status::Device); + if let Some(again) = next(&mut client) { + assert_eq!(again.op, Op::Write { run: run_of(1, 2), lba: 0 }, "the kept write goes again"); + answer(&mut client, again, Status::Ok); + } + } + assert_eq!(next(&mut client), None); + assert_eq!(client.take_answers().collect::>(), [(1, Outcome::Done), (2, Outcome::Device)]); + + client.submit(3, Ask::Flush); + let f = next(&mut client).unwrap(); + answer(&mut client, f, Status::Ok); + client.submit(4, write(4, 8, 1)); + let w = next(&mut client).unwrap(); + assert_eq!(w.op, Op::Write { run: run_of(4, 1), lba: 8 }); } } diff --git a/lib/blockring/src/model.rs b/lib/blockring/src/model.rs index 18337cdfd32..0709f14f4e6 100644 --- a/lib/blockring/src/model.rs +++ b/lib/blockring/src/model.rs @@ -26,7 +26,7 @@ //! than [`MAX_ATTEMPTS`], which is the fewest that can make the client give //! a write or a flush up ([`Law::Durable`]). -use alloc::collections::{BTreeMap, VecDeque}; +use alloc::collections::{BTreeMap, BTreeSet, VecDeque}; use alloc::format; use alloc::string::String; use alloc::vec::Vec; @@ -37,13 +37,17 @@ use std::collections::HashSet; use toyos_blockhold::Holds; use toyos::transport::{Consumer, Place, Producer, Violation, Word}; -use crate::client::{Client, Outcome, Ticket, MAX_ATTEMPTS}; +use crate::client::{Ask, Client, Outcome, Placing, Ticket, MAX_ATTEMPTS}; use crate::entry::{Completion, Op, Request}; use crate::layout::{ARENA, CQE_WORDS, SQE_WORDS}; use crate::server::{ServerSession, Taken}; const BLOCKS: usize = 2; +/// The blocks the client keeps before its caller's writes wait for its own +/// flush: one, so every script reaches it. +const LIMIT: u64 = 1; + /// How many of each failure one run may meet. #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] pub struct Failures { @@ -84,6 +88,8 @@ pub enum Step { pub enum Law { Answers, Durable, + /// A caller's write first goes out while the client keeps its limit. + Kept, } /// One word of the session page. The model runs on one thread, so every order @@ -209,7 +215,11 @@ struct Server { #[derive(Clone)] struct World { - client: Client<{ DEPTH as usize }>, + client: Client, + /// The writes that have gone out at least once. + sent: BTreeSet, + /// The blocks the client keeps before its caller's writes wait. + limit: u64, next: usize, /// Every answer each ticket has had. answers: BTreeMap>, @@ -257,14 +267,16 @@ pub fn explore(script: &[Step], failures: Failures) -> Explored { may_give_up: failures.total() > MAX_ATTEMPTS, path: Vec::new(), }; - dfs(&mut run, start(failures)); + dfs(&mut run, start(failures, LIMIT)); Explored { broken: run.broken, ends: run.ends, given_up: run.given_up, filled: run.filled } } /// The client connected to a fresh server, with nothing asked yet. -fn start(failures: Failures) -> World { +fn start(failures: Failures, limit: u64) -> World { let mut world = World { - client: Client::new(), + client: Client::new(limit), + limit, + sent: BTreeSet::new(), next: 0, answers: BTreeMap::new(), acked_before: BTreeMap::new(), @@ -278,23 +290,38 @@ fn start(failures: Failures) -> World { media: [0; BLOCKS], left: failures, }; - connect(&mut world); + connect(&mut world).expect("nothing is asked yet"); world } -fn connect(world: &mut World) { +fn connect(world: &mut World) -> Result<(), (Law, String)> { let mut holds = Holds::new(); holds.hold(0, BLOCKS as u64, 1).expect("a fresh server holds nothing"); world.server = Some(Server { session: ServerSession::new(0, BLOCKS as u64, true), holds }); world.alive = true; world.losses = 0; world.client.session_started(); - pump(world); + pump(world) } /// Every request the client will send goes onto the ring. -fn pump(world: &mut World) { - while let Some(request) = world.client.next_request() { +fn pump(world: &mut World) -> Result<(), (Law, String)> { + // A transfer's arena block is its ticket, a write's from its payload, which + // is its ticket, so the device knows a write's value by the block it + // carries, and a write issued again with another's payload is found. + let place = |p: Placing<'_, Ticket>| match p { + Placing::Read { ticket, blocks } => ARENA.run(ticket as u32, blocks), + Placing::Write { payload, blocks } => ARENA.run(*payload as u32, blocks), + }; + loop { + let kept = world.client.kept(); + let Some(request) = world.client.next_request(place) else { return Ok(()) }; + if let Op::Write { run, .. } = request.op { + let ticket = Ticket::from(run.first()); + if world.sent.insert(ticket) && kept >= world.limit { + return Err((Law::Kept, format!("write {ticket} went out with {kept} blocks kept"))); + } + } world.queues.send(request); } } @@ -381,8 +408,7 @@ fn settle(script: &[Step], world: &mut World) -> Result<(), (Law, String)> { durable(script, world, ticket).map_err(|why| (Law::Durable, why))?; } } - pump(world); - Ok(()) + pump(world) } /// What the flush `ticket`, just answered durable, promised is on the medium. @@ -460,8 +486,9 @@ fn key(world: &World) -> String { }); let posted: Vec = world.posted.iter().map(|&t| names.of(t)).collect(); format!( - "{:?} {wire:?} {sq:?} {device:?} {cq:?} {server:?} {posted:?} {} {:?} {:?} {} {} {:?} {:?} {:?}", + "{:?} {:?} {wire:?} {sq:?} {device:?} {cq:?} {server:?} {posted:?} {} {:?} {:?} {} {} {:?} {:?} {:?}", world.client, + world.sent, world.next, world.answers, world.acked_before, @@ -518,8 +545,7 @@ fn next(script: &[Step], world: &World) -> Vec<(String, After)> { let mut w = world.clone(); match script[world.next] { Step::Write { block, .. } => { - let run = ARENA.run(ticket as u32, 1).expect("a script's ticket is an arena block"); - w.client.submit(ticket, Op::Write { run, lba: block }); + w.client.submit(ticket, Ask::Write { lba: block, blocks: 1, payload: ticket }); } Step::Flush => { let acked = w @@ -529,7 +555,7 @@ fn next(script: &[Step], world: &World) -> Vec<(String, After)> { .map(|(t, _)| *t) .collect(); w.acked_before.insert(ticket, acked); - w.client.submit(ticket, Op::Flush); + w.client.submit(ticket, Ask::Flush); } } w.next += 1; @@ -652,8 +678,8 @@ fn next(script: &[Step], world: &World) -> Vec<(String, After)> { // It reconnects to the server started in its place. if !world.alive && !world.client.up() { let mut w = world.clone(); - connect(&mut w); - after("reconnect".into(), Ok(w)); + let connected = connect(&mut w).map(|()| w); + after("reconnect".into(), connected); } next @@ -747,6 +773,17 @@ mod tests { } } + /// No caller's write goes out while the client keeps its limit: it waits + /// for the client's own flush, under every failure the bounds allow. + #[test] + fn no_write_goes_out_while_the_client_keeps_its_limit() { + for (script, failures) in every_bound() { + if let Some((Law::Kept, why)) = verdict(script, failures).broken { + panic!("{failures:?}: {why}"); + } + } + } + /// The model is not vacuous: with no failure at all, a run ends with both /// flushes durable and the medium holding the script's last values; each /// ring holds its depth, and wraps, because one session carries more than @@ -788,7 +825,7 @@ mod tests { /// refuses as the server breaking the protocol. #[test] fn a_used_page_reads_fresh_once_a_servers_ends_are_made_over_it() { - let world = walk(start(at_most(0, 1, 0)), &["ask", "take", "done", "read", "crash"]); + let world = walk(start(at_most(0, 1, 0), u64::MAX), &["ask", "take", "done", "read", "crash"]); let page = &world.queues.page; let mut client = world.client.clone(); client.session_ended(); @@ -810,7 +847,7 @@ mod tests { /// alike: with W3 taken and the device reset, they are still named alike. #[test] fn states_named_alike_act_alike() { - let fresh = start(at_most(1, 0, 0)); + let fresh = start(at_most(1, 0, 0), u64::MAX); let answered_first = walk(fresh.clone(), &["ask", "take", "done", "read", "ask", "take", "done", "read", "ask", "take"]); let together = walk(fresh, &["ask", "ask", "take", "done", "read", "take", "done", "read", "ask", "take"]); diff --git a/system/diskserver/src/disk.rs b/system/diskserver/src/disk.rs index 85a2f34063f..5350b1fba1a 100644 --- a/system/diskserver/src/disk.rs +++ b/system/diskserver/src/disk.rs @@ -21,18 +21,19 @@ use std::collections::{BTreeMap, BTreeSet}; use std::future::poll_fn; use std::task::{Poll, Waker}; -use toyos_blockdev::{within, Block, BlockDev, Error, BLOCK}; +use toyos_blockdev::{within, Block, BlockDev, Error, SharedBlock, BLOCK}; +use toyos_blockring::layout::{ARENA, DEPTH}; use toyos_blockring::MAX_REQUEST_BLOCKS; use crate::session::{Answer, Error as SessionError, Session, Unsent, Waited}; use crate::{Outcome, Ticket}; /// What [`Served`] drives: a session's requests and their answers, as -/// [`Session`] keeps them, whose contract this is. A flush takes no arena. +/// [`Session`] keeps them, whose contract this is. pub trait Link { fn blocks(&self) -> u64; fn submit_read(&mut self, lba: u64, blocks: u32) -> Result; - fn submit_write(&mut self, lba: u64, data: &[u8]) -> Result; + fn submit_write(&mut self, lba: u64, blocks: &[SharedBlock]) -> Result; fn submit_flush(&mut self) -> Result; fn wait(&mut self, block: bool) -> Waited; fn reconnect(&mut self) -> Result<(), SessionError>; @@ -45,8 +46,8 @@ impl Link for Session { fn submit_read(&mut self, lba: u64, blocks: u32) -> Result { Session::submit_read(self, lba, blocks) } - fn submit_write(&mut self, lba: u64, data: &[u8]) -> Result { - Session::submit_write(self, lba, data) + fn submit_write(&mut self, lba: u64, blocks: &[SharedBlock]) -> Result { + Session::submit_write(self, lba, blocks) } fn submit_flush(&mut self) -> Result { Session::submit_flush(self) @@ -81,13 +82,8 @@ struct State { gone: bool, } -/// Why a request was not answered. -enum NotAnswered { - /// The session's arena has no room: a flush releases what acknowledged - /// writes hold. - Full, - Gone, -} +/// Why a request was not answered: the disk is gone. +struct Gone; impl Served { pub fn new(link: L) -> Self { @@ -132,21 +128,23 @@ impl Served { self.progress(true); } + /// How many requests of [`MAX_REQUEST_BLOCKS`] the session carries at + /// once: the most a caller gains by having in flight. + pub fn depth(&self) -> usize { + (DEPTH as usize).min(ARENA.slots() as usize / MAX_REQUEST_BLOCKS as usize) + } + /// One request, asked again after each session end it was on the wire /// for. - async fn ask(&self, mut submit: impl FnMut(&mut L) -> Result) -> Result { + async fn ask(&self, mut submit: impl FnMut(&mut L) -> Result) -> Result { let mut refusals = 0; loop { if self.state.borrow().gone { - return Err(NotAnswered::Gone); + return Err(Gone); } - let ticket = match submit(&mut self.link.borrow_mut()) { - Ok(ticket) => ticket, - Err(Unsent::ArenaFull) => return Err(NotAnswered::Full), - // `progress` reconnects as it finds an end, so a session - // still ended is one that would not reconnect. - Err(Unsent::Ended) => return Err(NotAnswered::Gone), - }; + // `progress` reconnects as it finds an end, so a session still + // ended is one that would not reconnect. + let Ok(ticket) = submit(&mut self.link.borrow_mut()) else { return Err(Gone) }; self.state.borrow_mut().asked.insert(ticket); let answer = self.answer(ticket).await?; if answer.outcome != Outcome::Refused { @@ -155,12 +153,12 @@ impl Served { refusals += 1; if refusals > MAX_RECONNECTS { println!("a block request was refused across {refusals} session ends; the disk is gone"); - return Err(NotAnswered::Gone); + return Err(Gone); } } } - async fn answer(&self, ticket: Ticket) -> Result { + async fn answer(&self, ticket: Ticket) -> Result { poll_fn(|cx| { let mut state = self.state.borrow_mut(); if let Some(answer) = state.answered.remove(&ticket) { @@ -169,7 +167,7 @@ impl Served { } if state.gone { state.asked.remove(&ticket); - return Poll::Ready(Err(NotAnswered::Gone)); + return Poll::Ready(Err(Gone)); } state.waiting.insert(ticket, cx.waker().clone()); Poll::Pending @@ -177,15 +175,20 @@ impl Served { .await } - /// [`Served::ask`], flushing first wherever the arena has no room. - async fn ask_room(&self, mut submit: impl FnMut(&mut L) -> Result) -> Result { - loop { - match self.ask(&mut submit).await { - Ok(answer) => return Ok(answer), - Err(NotAnswered::Full) => self.flush().await?, - Err(NotAnswered::Gone) => return Err(Error::Gone), + /// `blocks` written at `lba`, shared: the session keeps them until a + /// flush covers the write, rather than a copy of them. + pub async fn write_shared(&self, lba: u64, blocks: &[SharedBlock]) -> Result<(), Error> { + within(lba, blocks.len(), self.blocks)?; + let per = MAX_REQUEST_BLOCKS as usize; + for (i, chunk) in blocks.chunks(per).enumerate() { + let at = lba + (i * per) as u64; + match self.ask(|link| link.submit_write(at, chunk)).await.map_err(|Gone| Error::Gone)?.outcome { + Outcome::Done => {} + Outcome::ReadOnly => return Err(Error::ReadOnly), + _ => return Err(Error::Device), } } + Ok(()) } } @@ -199,7 +202,7 @@ impl BlockDev for Served { let per = MAX_REQUEST_BLOCKS as usize; for (i, chunk) in out.chunks_mut(per).enumerate() { let (at, n) = (lba + (i * per) as u64, chunk.len() as u32); - let answer = self.ask_room(|link| link.submit_read(at, n)).await?; + let answer = self.ask(|link| link.submit_read(at, n)).await.map_err(|Gone| Error::Gone)?; let chunk = chunk.as_flattened_mut(); match (answer.outcome, answer.data) { (Outcome::Done, Some(data)) if data.len() == chunk.len() => chunk.copy_from_slice(&data), @@ -209,26 +212,17 @@ impl BlockDev for Served { Ok(()) } + /// A copy of `data`, which the session keeps until a flush covers it. async fn write(&self, lba: u64, data: &[Block]) -> Result<(), Error> { - within(lba, data.len(), self.blocks)?; - let per = MAX_REQUEST_BLOCKS as usize; - for (i, chunk) in data.chunks(per).enumerate() { - let at = lba + (i * per) as u64; - match self.ask_room(|link| link.submit_write(at, chunk.as_flattened())).await?.outcome { - Outcome::Done => {} - Outcome::ReadOnly => return Err(Error::ReadOnly), - _ => return Err(Error::Device), - } - } - Ok(()) + let shared: Vec = data.iter().map(|block| std::rc::Rc::new(Box::new(*block))).collect(); + self.write_shared(lba, &shared).await } async fn flush(&self) -> Result<(), Error> { match self.ask(Link::submit_flush).await { Ok(Answer { outcome: Outcome::Durable, .. }) => Ok(()), Ok(_) => Err(Error::Device), - Err(NotAnswered::Gone) => Err(Error::Gone), - Err(NotAnswered::Full) => unreachable!("diskserver: a flush took arena room"), + Err(Gone) => Err(Error::Gone), } } } diff --git a/system/diskserver/src/disk/tests.rs b/system/diskserver/src/disk/tests.rs index fa70af1b8ab..4da6d8d514c 100644 --- a/system/diskserver/src/disk/tests.rs +++ b/system/diskserver/src/disk/tests.rs @@ -6,7 +6,7 @@ use std::cell::Cell; use std::collections::VecDeque; use std::rc::Rc; -use toyos_blockdev::{block_on, Block, BlockDev, Error, Executor, BLOCK}; +use toyos_blockdev::{block_on, Block, BlockDev, Error, Executor, SharedBlock, BLOCK}; use super::{Link, Served, MAX_RECONNECTS}; use crate::session::{Answer, Error as SessionError, Unsent, Waited}; @@ -36,8 +36,6 @@ struct Script { /// Each reconnect's answer, in turn; `Ok` once they run out. reconnects: VecDeque>, reconnected: u32, - /// Writes the next this many write submissions refuse for want of room. - full: u32, read_only: bool, } @@ -53,7 +51,6 @@ impl Script { up: true, reconnects: VecDeque::new(), reconnected: 0, - full: 0, read_only: false, } } @@ -96,12 +93,8 @@ impl Link for Script { self.ask(Asked::Read { lba, blocks }) } - fn submit_write(&mut self, lba: u64, data: &[u8]) -> Result { - if self.full > 0 && self.up { - self.full -= 1; - return Err(Unsent::ArenaFull); - } - self.ask(Asked::Write { lba, data: data.to_vec() }) + fn submit_write(&mut self, lba: u64, blocks: &[SharedBlock]) -> Result { + self.ask(Asked::Write { lba, data: blocks.iter().flat_map(|b| b.iter().copied()).collect() }) } fn submit_flush(&mut self) -> Result { @@ -197,17 +190,6 @@ fn a_refused_reconnect_makes_the_disk_gone_for_every_request() { assert_eq!((script.log.len(), script.reconnected), (1, 1)); } -/// A write the arena has no room for is asked again after a flush, which is -/// what releases the room acknowledged writes hold. -#[test] -fn a_write_without_arena_room_is_asked_again_after_a_flush() { - let mut script = Script::new(4); - script.full = 1; - let disk = Served::new(script); - now(&disk, disk.write(2, &blocks(7, 1))).unwrap(); - assert_eq!(disk.link.borrow().log, [Asked::Flush, Asked::Write { lba: 2, data: vec![7; BLOCK] }]); -} - /// A transfer longer than one request carries is asked in requests of at /// most `MAX_REQUEST_BLOCKS`; one past the partition's end is refused before /// anything is asked; a write the grant does not make is `ReadOnly`. diff --git a/system/diskserver/src/session.rs b/system/diskserver/src/session.rs index e4eb518ce54..f2e55a2a7fb 100644 --- a/system/diskserver/src/session.rs +++ b/system/diskserver/src/session.rs @@ -3,9 +3,11 @@ //! //! **What a request means is `toyos_blockring::client::Client`'s to decide**; //! this is the mapping, the arena and the doorbells around it. -//! A write's data is copied into the arena and stays there until a flush -//! covers it, because after a loss it is the only copy there is; a read's is -//! copied out when it is answered. +//! A request takes arena blocks only while it is on the wire. A write's blocks +//! are the caller's, shared: the client keeps them with the write until a +//! flush covers it, at most [`KEPT_BLOCKS`] before it flushes for them, and +//! the session copies them into the arena each time the write goes out. A +//! read's data is copied out when it is answered. //! //! **A service that ends is survivable and not invisible.** [`Session::wait`] //! reads every completion the server posted before it hung up, answers what @@ -22,14 +24,21 @@ use toyos::namespace::Namespace; use toyos::poller::READABLE; use toyos::wake::{Bell, Waker}; use toyos_abi::syscall::SyscallError; -use toyos_blockring::client::{Client, Outcome, Ticket}; -use toyos_blockring::entry::{Completion, Op}; +use toyos_blockdev::SharedBlock; +use toyos_blockring::client::{Ask, Client, Outcome, Placing, Ticket}; +use toyos_blockring::entry::Completion; use toyos_blockring::layout::{self, ClientRings, ARENA, MAX_REQUEST_BLOCKS}; use toyos_blockring::wire::{self, Opened, Refusal}; use toyos_blockring::{Run, BLOCK_BYTES}; use toyos_blockserve::region::Region; +/// The most blocks of acknowledged writes a session keeps before the caller's +/// next write waits for a flush of its own: 32 MiB, no less than a file +/// server cache's dirty bound, so a cache that flushes by that bound is never +/// made to flush by this one. +pub const KEPT_BLOCKS: u64 = 8 * 1024; + /// Why a session could not be opened or go on. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum Error { @@ -70,8 +79,6 @@ pub struct Waited { /// caller that gave up on it would find it done under writes it made since. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum Unsent { - /// The arena has no room: a flush releases what acknowledged writes hold. - ArenaFull, /// The session has ended: [`Session::reconnect`] first. Ended, } @@ -110,7 +117,7 @@ impl Arena { } enum Pending { - Read { run: Run }, + Read, Write, Flush, } @@ -123,11 +130,13 @@ pub struct Session { /// The open session's doorbells, `None` once it ended. bells: Option, rings: ClientRings, - client: Client, + client: Client>, opened: Opened, arena: Arena, next_ticket: Ticket, pending: BTreeMap, + /// Each read on the wire's arena blocks. + reads: BTreeMap, /// The most requests this session has had on the wire at once. peak: usize, } @@ -139,7 +148,7 @@ impl Session { let region = Region::create().map_err(Error::Kernel)?; let rings = layout::client(region.words()); let (bells, opened) = handshake(&names, service, guid, ®ion)?; - let mut client = Client::new(); + let mut client = Client::new(KEPT_BLOCKS); client.session_started(); Ok(Self { names, @@ -153,6 +162,7 @@ impl Session { arena: Arena::new(), next_ticket: 0, pending: BTreeMap::new(), + reads: BTreeMap::new(), peak: 0, }) } @@ -178,30 +188,29 @@ impl Session { t } - /// Ask for `data` (whole blocks) written at `lba`. - pub fn submit_write(&mut self, lba: u64, data: &[u8]) -> Result { - let blocks = whole_blocks(data.len()); + /// Ask for `blocks` written at `lba`, which the session keeps a share of + /// until a flush covers the write. With [`KEPT_BLOCKS`] kept, it goes out + /// after a flush of the session's own. + pub fn submit_write(&mut self, lba: u64, blocks: &[SharedBlock]) -> Result { + let count = transfer_blocks(blocks.len()); if self.bells.is_none() { return Err(Unsent::Ended); } - let run = self.arena.alloc(blocks).ok_or(Unsent::ArenaFull)?; - self.region.arena(&run).copy_in(0, data); let ticket = self.ticket(); - self.client.submit(ticket, Op::Write { run, lba }); + self.client.submit(ticket, Ask::Write { lba, blocks: count, payload: blocks.to_vec() }); self.pending.insert(ticket, Pending::Write); Ok(ticket) } /// Ask for `blocks` read from `lba`; the answer carries the data. pub fn submit_read(&mut self, lba: u64, blocks: u32) -> Result { - assert!((1..=MAX_REQUEST_BLOCKS).contains(&blocks), "a read of {blocks} blocks"); + transfer_blocks(blocks as usize); if self.bells.is_none() { return Err(Unsent::Ended); } - let run = self.arena.alloc(blocks).ok_or(Unsent::ArenaFull)?; let ticket = self.ticket(); - self.client.submit(ticket, Op::Read { run, lba }); - self.pending.insert(ticket, Pending::Read { run }); + self.client.submit(ticket, Ask::Read { lba, blocks }); + self.pending.insert(ticket, Pending::Read); Ok(ticket) } @@ -211,7 +220,7 @@ impl Session { return Err(Unsent::Ended); } let ticket = self.ticket(); - self.client.submit(ticket, Op::Flush); + self.client.submit(ticket, Ask::Flush); self.pending.insert(ticket, Pending::Flush); Ok(ticket) } @@ -219,25 +228,42 @@ impl Session { /// Put what the client will send on the ring, and ring the doorbell once /// for all of it. fn pump(&mut self) { - let Some(bells) = &self.bells else { return }; - let page = self.region.words(); + let Self { bells, region, rings, client, arena, reads, .. } = self; + let Some(bells) = bells else { return }; + let page = region.words(); + let mut place = |p: Placing<'_, Vec>| match p { + Placing::Read { ticket, blocks } => { + let run = arena.alloc(blocks)?; + reads.insert(ticket, run); + Some(run) + } + Placing::Write { payload, blocks } => { + let run = arena.alloc(blocks)?; + let window = region.arena(&run); + for (k, block) in payload.iter().enumerate() { + window.copy_in(k * BLOCK_BYTES, &***block); + } + Some(run) + } + }; loop { - match self.rings.0.space(page) { + match rings.0.space(page) { Ok(0) => break, Ok(_) => {} // The server wrote an index no server writes: the session is // over, and the next drain is where that is acted on. Err(_) => break, } - let Some(request) = self.client.next_request() else { break }; - let pushed = self.rings.0.push(page, request.encode()); + let Some(request) = client.next_request(&mut place) else { break }; + let pushed = rings.0.push(page, request.encode()); assert_eq!(pushed, Ok(true), "diskserver: a request past the room just counted"); } - self.peak = self.peak.max(self.client.on_the_wire().count()); - if self.rings.0.publish(page) { + let on_the_wire = client.on_the_wire().count(); + if rings.0.publish(page) { // A server that has ended is found by the wait. bells.ring.wake(); } + self.peak = self.peak.max(on_the_wire); } /// Every answer there is; with `block`, wait until there is at least one @@ -307,12 +333,18 @@ impl Session { for (ticket, outcome) in decided { let pending = self.pending.remove(&ticket).expect("diskserver: an answer for no ticket"); let data = match (pending, outcome) { - (Pending::Read { run }, Outcome::Done) => { - let mut data = vec![0u8; run.span().len]; - self.region.arena(&run).copy_out(0, &mut data); - Some(data) + (Pending::Read, outcome) => { + let run = self.reads.remove(&ticket); + match (run, outcome) { + (Some(run), Outcome::Done) => { + let mut data = vec![0u8; run.span().len]; + self.region.arena(&run).copy_out(0, &mut data); + Some(data) + } + _ => None, + } } - _ => None, + (Pending::Write | Pending::Flush, _) => None, }; answers.push(Answer { ticket, outcome, data }); } @@ -351,11 +383,9 @@ impl Session { } } -fn whole_blocks(len: usize) -> u32 { - assert!(len > 0 && len % BLOCK_BYTES == 0, "diskserver: a transfer of {len} bytes is no whole blocks"); - let blocks = (len / BLOCK_BYTES) as u32; - assert!(blocks <= MAX_REQUEST_BLOCKS, "diskserver: a transfer of {blocks} blocks"); - blocks +fn transfer_blocks(blocks: usize) -> u32 { + assert!((1..=MAX_REQUEST_BLOCKS as usize).contains(&blocks), "diskserver: a transfer of {blocks} blocks"); + blocks as u32 } pub(crate) fn connect(names: &Namespace, service: &str) -> Result { diff --git a/system/fileserver/src/cache.rs b/system/fileserver/src/cache.rs index 88a6d8eb78c..81f0b1f096d 100644 --- a/system/fileserver/src/cache.rs +++ b/system/fileserver/src/cache.rs @@ -2,11 +2,13 @@ //! metadata and its files' data alike. It is a [`BlockDev`] over the disk. //! //! **A write lands here and reaches the disk at a flush.** A flush writes -//! every dirty block, lowest first and in runs, then asks the disk to make -//! them durable; a volume's sync is its own metadata written into the cache -//! and then this. A write that would take the cache past [`DIRTY_LIMIT`] dirty -//! blocks flushes first, and is refused whole when that flush is, so memory -//! bounds the write-back and not the other way round. +//! every dirty block in runs, as many at once as the disk takes +//! ([`Disk::depth`]), handing the disk the cache's own blocks to share +//! ([`Disk::write_shared`]), then asks the disk to make them durable; a +//! volume's sync is its own metadata written into the cache and then this. A +//! write that would take the cache past [`DIRTY_LIMIT`] dirty blocks flushes +//! first, and is refused whole when that flush is, so memory bounds the +//! write-back and not the other way round. //! //! **Clean blocks are kept up to the limit the cache is made with** and the //! oldest-read goes first; a dirty block is never dropped. A block read in @@ -14,7 +16,8 @@ //! room: refused, the read is answered from the disk's bytes and nothing is //! cached. A block a flush cleans whose place is refused is dropped, since the //! disk holds it. The block map's node, a small allocation every few inserts, -//! is the one part of keeping a block read in that still cannot be refused. +//! and the block's share count are the parts of keeping a block read in that +//! still cannot be refused. //! //! **Several tasks may use it at once**, and the disk never has two requests //! for one block in flight when either is a write, which the block service's @@ -29,34 +32,63 @@ use std::cell::RefCell; use std::collections::{BTreeMap, VecDeque}; use std::future::{poll_fn, Future}; +use std::pin::Pin; +use std::rc::Rc; use std::task::{Poll, Waker}; -use toyos_blockdev::{within, Block, BlockDev, Error, BLOCK}; +use toyos_blockdev::{within, Block, BlockDev, Error, SharedBlock, BLOCK}; /// A disk this server owns, and what to do when nothing can go on until it /// answers. +// Every future runs on the server's one thread, as `BlockDev`'s do. +#[allow(async_fn_in_trait)] pub trait Disk: BlockDev { /// Returns once a request asked of this disk may have been answered. A /// disk that answers every request as it is asked is never waited on. fn wait(&self) { panic!("fileserver: a disk that answers as it is asked was waited on"); } + + /// How many writes a flush keeps in flight at once: at least one. + fn depth(&self) -> usize { + 1 + } + + /// `blocks` written at `lba`: shared by a disk that keeps a write's bytes + /// until a flush covers it, copied by one that does not. + async fn write_shared(&self, lba: u64, blocks: &[SharedBlock]) -> Result<(), Error> { + let data: Vec = blocks.iter().map(|block| ***block).collect(); + self.write(lba, &data).await + } } impl Disk for diskserver::disk::Served { fn wait(&self) { diskserver::disk::Served::wait(self) } + + fn depth(&self) -> usize { + diskserver::disk::Served::depth(self) + } + + async fn write_shared(&self, lba: u64, blocks: &[SharedBlock]) -> Result<(), Error> { + diskserver::disk::Served::write_shared(self, lba, blocks).await + } } /// Dirty blocks held before the cache flushes by itself: 32 MiB. pub const DIRTY_LIMIT: usize = 8 * 1024; +// A cache that flushes by its own dirty bound never meets a session's. +const _: () = assert!(DIRTY_LIMIT as u64 <= diskserver::session::KEPT_BLOCKS); + /// The longest run one disk request carries, in blocks. const RUN: usize = 32; struct Slot { - data: Box, + /// Shared with the disk while it keeps a write of it: a write here then + /// takes a block of its own. + data: SharedBlock, dirty: bool, /// Which write put `data` here, so a flush that carried an older copy /// leaves the block dirty. @@ -146,7 +178,7 @@ impl Cache { let mut inner = self.inner.borrow_mut(); inner.reads += 1; if let Some(slot) = inner.slots.get(&block) { - put(i, &slot.data); + put(i, &**slot.data); inner.hits += 1; i += 1; continue; @@ -173,12 +205,12 @@ impl Cache { // disk answered, and stays held until a flush that waited for // this fetch has written it. match inner.slots.get(&at) { - Some(slot) => put(i + k, &slot.data), + Some(slot) => put(i + k, &**slot.data), None => { put(i + k, data); if inner.order.try_reserve(1).is_ok() { if let Some(data) = boxed(data) { - inner.slots.insert(at, Slot { data, dirty: false, version: 0 }); + inner.slots.insert(at, Slot { data: Rc::new(data), dirty: false, version: 0 }); inner.order.push_back(at); } } @@ -258,12 +290,15 @@ impl BlockDev for Cache { let version = inner.versions; let newly = match inner.slots.get_mut(&block) { Some(slot) => { - *slot.data = *chunk; + match Rc::get_mut(&mut slot.data) { + Some(data) => **data = *chunk, + None => slot.data = Rc::new(Box::new(*chunk)), + } slot.version = version; !std::mem::replace(&mut slot.dirty, true) } None => { - inner.slots.insert(block, Slot { data: Box::new(*chunk), dirty: true, version }); + inner.slots.insert(block, Slot { data: Rc::new(Box::new(*chunk)), dirty: true, version }); true } }; @@ -274,58 +309,40 @@ impl BlockDev for Cache { Ok(()) } - /// Every dirty block to the disk, lowest first and in runs, then the disk - /// made durable. A block the disk refused stays dirty. A disk written + /// Every dirty block to the disk in runs, up to [`Disk::depth`] of them in + /// flight, then the disk made durable. A block the disk refused stays + /// dirty, and once one run is refused no other is begun. A disk written /// nothing since the last flush it answered is asked nothing: every write /// it took is already durable. async fn flush(&self) -> Result<(), Error> { self.until(|inner| !inner.flushing).await; self.inner.borrow_mut().flushing = true; let _flushing = Flushing(self); - let dirty: Vec = self.inner.borrow().slots.iter().filter(|(_, s)| s.dirty).map(|(b, _)| *b).collect(); - let mut buf: Vec = vec![[0; BLOCK]; RUN]; - let mut i = 0; - while i < dirty.len() { - let first = dirty[i]; - let mut run = 1; - while i + run < dirty.len() && run < RUN && dirty[i + run] == first + run as u64 { - run += 1; + let mut runs = self.dirty_runs().into_iter(); + let mut flight: Vec> + '_>>> = Vec::new(); + let mut refused = Ok(()); + poll_fn(|cx| loop { + while refused.is_ok() && flight.len() < self.disk.depth() { + let Some((first, count)) = runs.next() else { break }; + flight.push(Box::pin(self.write_run(first, count))); } - let end = first + run as u64; - self.until(|inner| !inner.fetches.values().any(|f| f.first < end && first < f.first + f.count as u64)) - .await; - let versions: Vec = { - let inner = self.inner.borrow(); - (0..run) - .map(|k| { - // Only a flush cleans a block, and this is the one. - let slot = inner.slots.get(&(first + k as u64)).expect("fileserver: a dirty block left the cache during its flush"); - buf[k] = *slot.data; - slot.version - }) - .collect() - }; - self.inner.borrow_mut().unflushed = true; - self.disk.write(first, &buf[..run]).await?; - let mut inner = self.inner.borrow_mut(); - for (k, version) in versions.into_iter().enumerate() { - let block = first + k as u64; - // A block written again meanwhile stays dirty. - let slot = inner.slots.get_mut(&block).expect("fileserver: a dirty block left the cache during its flush"); - if slot.version == version { - slot.dirty = false; - inner.dirty -= 1; - // The disk holds it now, so a block whose place in the - // eviction order is refused is dropped rather than kept. - if inner.order.try_reserve(1).is_ok() { - inner.order.push_back(block); - } else { - inner.slots.remove(&block); - } + let before = flight.len(); + flight.retain_mut(|run| match run.as_mut().poll(cx) { + Poll::Ready(done) => { + refused = std::mem::replace(&mut refused, Ok(())).and(done); + false } + Poll::Pending => true, + }); + if flight.is_empty() && (refused.is_err() || runs.len() == 0) { + return Poll::Ready(()); } - i += run; - } + if flight.len() == before { + return Poll::Pending; + } + }) + .await; + refused?; if self.inner.borrow().unflushed { self.disk.flush().await?; self.inner.borrow_mut().unflushed = false; @@ -335,6 +352,56 @@ impl BlockDev for Cache { } } +impl Cache { + /// The dirty blocks as runs of consecutive blocks, each at most [`RUN`]. + fn dirty_runs(&self) -> Vec<(u64, usize)> { + let mut runs: Vec<(u64, usize)> = Vec::new(); + for (&block, _) in self.inner.borrow().slots.iter().filter(|(_, s)| s.dirty) { + match runs.last_mut() { + Some((first, count)) if *first + *count as u64 == block && *count < RUN => *count += 1, + _ => runs.push((block, 1)), + } + } + runs + } + + /// One run of a flush: written once no fetch is reading it, and each block + /// clean afterwards unless written again meanwhile. + async fn write_run(&self, first: u64, count: usize) -> Result<(), Error> { + let end = first + count as u64; + self.until(|inner| !inner.fetches.values().any(|f| f.first < end && first < f.first + f.count as u64)).await; + let (blocks, versions): (Vec, Vec) = { + let inner = self.inner.borrow(); + (first..end) + .map(|block| { + // Only a flush cleans a block, and this is the one. + let slot = inner.slots.get(&block).expect("fileserver: a dirty block left the cache during its flush"); + (Rc::clone(&slot.data), slot.version) + }) + .unzip() + }; + self.inner.borrow_mut().unflushed = true; + self.disk.write_shared(first, &blocks).await?; + let mut inner = self.inner.borrow_mut(); + for (block, version) in (first..).zip(versions) { + // A block written again meanwhile stays dirty. + let slot = inner.slots.get_mut(&block).expect("fileserver: a dirty block left the cache during its flush"); + if slot.version == version { + slot.dirty = false; + inner.dirty -= 1; + // The disk holds it now, so a block whose place in the + // eviction order is refused is dropped rather than kept. + if inner.order.try_reserve(1).is_ok() { + inner.order.push_back(block); + } else { + inner.slots.remove(&block); + } + } + } + Ok(()) + } +} + impl Inner { /// Drop the oldest clean blocks past the limit. fn evict(&mut self) { @@ -627,6 +694,15 @@ mod tests { /// `(first, count, write)` of each request in flight. flight: RefCell>, overlaps: Cell, + /// What [`Disk::depth`] answers. + depth: Cell, + /// Every write is refused while this is set. + refusing: Cell, + /// Requests the closed gate still lets through, one each. + passes: Cell, + writes: Cell, + /// Every block handed to [`Disk::write_shared`], as handed. + kept: RefCell>, } impl Held { @@ -637,20 +713,35 @@ mod tests { waiting: RefCell::default(), flight: RefCell::default(), overlaps: Cell::new(0), + depth: Cell::new(1), + refusing: Cell::new(false), + passes: Cell::new(0), + writes: Cell::new(0), + kept: RefCell::default(), } } async fn gate(&self) { - poll_fn(|cx| match self.open.get() { - true => Poll::Ready(()), - false => { - self.waiting.borrow_mut().push(cx.waker().clone()); - Poll::Pending + poll_fn(|cx| { + if self.open.get() { + return Poll::Ready(()); + } + if self.passes.get() > 0 { + self.passes.set(self.passes.get() - 1); + return Poll::Ready(()); } + self.waiting.borrow_mut().push(cx.waker().clone()); + Poll::Pending }) .await } + /// Let one request waiting at the gate through, the first polled. + fn release_one(&self) { + self.passes.set(self.passes.get() + 1); + self.waiting.take().into_iter().for_each(Waker::wake); + } + fn release(&self) { self.open.set(true); self.waiting.take().into_iter().for_each(Waker::wake); @@ -690,9 +781,13 @@ mod tests { Ok(()) } async fn write(&self, lba: u64, data: &[Block]) -> Result<(), Error> { + self.writes.set(self.writes.get() + 1); self.send(lba, data.len(), true); self.gate().await; self.landed(lba, true); + if self.refusing.get() { + return Err(Error::Device); + } self.ram.write(lba, data).await } async fn flush(&self) -> Result<(), Error> { @@ -701,7 +796,81 @@ mod tests { } } - impl Disk for Held {} + impl Disk for Held { + fn depth(&self) -> usize { + self.depth.get() + } + + async fn write_shared(&self, lba: u64, blocks: &[SharedBlock]) -> Result<(), Error> { + self.kept.borrow_mut().extend((lba..).zip(blocks.iter().cloned())); + let data: Vec = blocks.iter().map(|block| ***block).collect(); + self.write(lba, &data).await + } + } + + /// Blocks 1, 3, 5, … written, one run each: `runs` of them. + fn runs_apart(c: &Cache, runs: u64) { + for k in 0..runs { + block_on_ready(c.write(2 * k + 1, &[[k as u8 + 1; BLOCK]])).unwrap(); + } + } + + /// A flush keeps as many runs in flight as the disk takes, and no more. + #[test] + fn a_flush_keeps_as_many_runs_in_flight_as_the_disk_takes() { + let c = Cache::new(Held::new(16), CLEAN_LIMIT); + c.disk.depth.set(3); + runs_apart(&c, 5); + c.disk.open.set(false); + let exec = Executor::new(); + exec.spawn(async { c.flush().await.unwrap() }); + exec.run_until_stalled(); + assert_eq!(c.disk.writes_in_flight(), 3); + c.disk.release(); + assert_eq!(exec.run_until_stalled(), 0); + assert_eq!(c.disk.overlaps.get(), 0); + assert_eq!(c.counts().dirty, 0); + for k in 0..5 { + assert_eq!(on_disk(&c.disk, 2 * k + 1), [k as u8 + 1; BLOCK]); + } + } + + /// The disk is handed the cache's own blocks, not copies of them. + #[test] + fn a_flush_shares_the_cache_s_blocks_with_the_disk() { + let c = Cache::new(Held::new(16), CLEAN_LIMIT); + block_on_ready(c.write(3, &[[1; BLOCK]])).unwrap(); + block_on_ready(c.flush()).unwrap(); + let kept = c.disk.kept.borrow(); + let inner = c.inner.borrow(); + assert_eq!(kept.len(), 1); + assert!(Rc::ptr_eq(&kept[0].1, &inner.slots[&3].data), "the disk was handed a copy"); + } + + /// Once a run is refused no other is begun, though another is still in + /// flight, and every block not written stays dirty for the next flush. + #[test] + fn a_refused_run_begins_no_other_and_leaves_the_rest_dirty() { + let c = Cache::new(Held::new(16), CLEAN_LIMIT); + c.disk.depth.set(2); + runs_apart(&c, 4); + c.disk.refusing.set(true); + c.disk.open.set(false); + let exec = Executor::new(); + exec.spawn(async { assert_eq!(c.flush().await, Err(Error::Device)) }); + exec.run_until_stalled(); + assert_eq!(c.disk.writes_in_flight(), 2); + c.disk.release_one(); + exec.run_until_stalled(); + assert_eq!(c.disk.writes_in_flight(), 1, "the refused run's place was taken"); + c.disk.release(); + assert_eq!(exec.run_until_stalled(), 0); + assert_eq!(c.disk.writes.get(), 2, "a run was begun after one was refused"); + assert_eq!(c.counts().dirty, 4); + c.disk.refusing.set(false); + block_on_ready(c.flush()).unwrap(); + assert_eq!(c.counts().dirty, 0); + } /// What block `block` holds on the disk. fn on_disk(disk: &Held, block: u64) -> Block {