diff --git a/src/app/collection.rs b/src/app/collection.rs index 702725a..bf42b98 100644 --- a/src/app/collection.rs +++ b/src/app/collection.rs @@ -22,7 +22,7 @@ use crate::collect::run::{FailureKind, RunFailure}; use crate::collect::tracker::{OpenFailure, Trackers}; use crate::collect::worktree; use crate::config::{Config, Project, Scope}; -use crate::model::join::{self, BeadKey, Listed, ProjectRows}; +use crate::model::join::{self, Listed, ProjectRows}; use crate::model::snapshot::{ self, AgentProvider, Collected, FailedProject, Filter, Node, ProviderState, Said, Session, SessionState, Snapshot, TrackerFailure, TrackerState, Tree, @@ -174,12 +174,41 @@ pub struct Collection { /// in the journal it has read to, and what the journal said since the /// answers were last handed over. journals: BTreeMap, + /// What the collection in hand found of each project it read. + found: BTreeMap, + /// For each project a watcher was last told of, the readiness its trees + /// gave each bead where that differs from bd's own. bd's own moves only + /// with a read of the project, so a project not read since says nothing + /// new to a watcher unless these move. + answered: BTreeMap>, /// How many reads `read` has taken in, so trees assembled from the /// standing set can tell when one of them has been replaced. reads_taken: u64, reached: Reached, } +/// Whether a read found its project's tracker had moved since the read +/// standing. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Found { + /// Read again, or found unreachable. + Moved, + /// Nothing had moved, so the read standing still speaks for it. + Unmoved, +} + +/// What one collection says to a watcher, and what the source reading for +/// it arms the next read from. +pub struct Watched { + /// Every project the collection reads, as a snapshot names them. + pub projects: Vec, + /// When each project's read stops speaking for its tracker with nothing + /// written, where it does. + pub speaks_until: BTreeMap>, + /// What each project with something new to say says. + pub answers: Vec, +} + /// The trees that reach into another project's answer, kept from one /// collection to the next until what they were assembled from changes. #[derive(Default)] @@ -269,19 +298,84 @@ impl Collection { // with a producer is never polled — so a refresh naming it is the // only chance the agent join gets. let (panes, provider, out_of_reach) = self.every_pane(agents); + self.read_wanted(cfg, trackers, wanted, &panes, now); + self.drawn(cfg, trackers, &panes, &out_of_reach, &provider, filter, now) + } + + /// Read what `wanted` names, and say to a watcher what each project + /// standing has new to say. Where no read found anything moved, nothing + /// is drawn and only the projects read are spoken of. + pub fn watched( + &mut self, + cfg: &Config, + agents: &dyn Agents, + trackers: &dyn Trackers, + wanted: &Wanted, + now: DateTime, + ) -> Watched { + let (panes, provider, out_of_reach) = self.every_pane(agents); + self.read_wanted(cfg, trackers, wanted, &panes, now); + if !self.found.values().any(|found| *found == Found::Moved) { + let cfg = self.widened(cfg); + return Watched { + projects: cfg.read().map(|project| project.name.clone()).collect(), + speaks_until: self.speaks_until(&cfg), + answers: self.vouched(), + }; + } + let snapshot = self.drawn( + cfg, + trackers, + &panes, + &out_of_reach, + &provider, + Filter::All, + now, + ); + let answers = self.answers(&snapshot); + Watched { + projects: snapshot.projects, + speaks_until: snapshot.speaks_until, + answers, + } + } + fn read_wanted( + &mut self, + cfg: &Config, + trackers: &dyn Trackers, + wanted: &Wanted, + panes: &[Pane], + now: DateTime, + ) { + self.found.clear(); let reading = self.widened(cfg); self.read( &reading, trackers, &|project| wanted.names(project), &|project| cfg.reads(project), - &panes, + panes, now, ); + } + + /// Everything standing drawn, reading as well any project the scope left + /// out that holds a drawn bead's blocker. + #[allow(clippy::too_many_arguments)] + fn drawn( + &mut self, + cfg: &Config, + trackers: &dyn Trackers, + panes: &[Pane], + out_of_reach: &BTreeSet, + provider: &AgentProvider, + filter: Filter, + now: DateTime, + ) -> Snapshot { loop { self.reach_across(cfg); - match self.draw(cfg, &panes, &out_of_reach, &provider, filter, now) { + match self.draw(cfg, panes, out_of_reach, provider, filter, now) { Ok(snapshot) => return snapshot, Err(needed) => { self.read_on_demand.extend(needed.iter().cloned()); @@ -291,7 +385,7 @@ impl Collection { trackers, &|project| needed.contains(project), &|project| cfg.reads(project), - &panes, + panes, now, ); } @@ -299,62 +393,70 @@ impl Collection { } } - /// What every project standing says to a watcher, each bead with the - /// readiness `snapshot`, the collection just drawn from these reads, gives - /// it. + /// What each project with something new to say says to a watcher, each + /// bead with the readiness `snapshot`, the collection just drawn from + /// these reads, gives it. /// - /// Every project rather than only those just read, because a blocker - /// closing in one project frees a bead in another. A bead no tree reaches - /// takes bd's own readiness, which is all there is to say of it. - pub fn answers(&mut self, snapshot: &Snapshot) -> Vec { - let mut drawn: BTreeMap = BTreeMap::new(); + /// A project read again says so. So does one whose trees moved a bead's + /// readiness, because a blocker closing in one project frees a bead in + /// another. A project read and found unmoved is vouched for, and any + /// other is not spoken of. A bead no tree reaches takes bd's own + /// readiness, which is all there is to say of it. + fn answers(&mut self, snapshot: &Snapshot) -> Vec { + let mut drawn: BTreeMap<&str, BTreeMap<&str, &Node>> = BTreeMap::new(); for node in snapshot.collected.iter().flat_map(|tree| &tree.beads) { - drawn.entry(node.key()).or_insert(node); + drawn + .entry(node.project.as_str()) + .or_default() + .entry(node.id.as_str()) + .or_insert(node); + } + let mut answers = Vec::new(); + for (project, read) in &self.read { + let found = self.found.get(project).copied(); + let said = match &read.work { + Ok(work) => { + let differing = differing_readiness(work, drawn.get(project.as_str())); + let moved = self.answered.get(project) != Some(&differing); + let said = (found == Some(Found::Moved) || moved) + .then(|| held_by_watcher(work, &differing, read.at)); + self.answered.insert(project.clone(), differing); + said + } + Err(failure) => (found == Some(Found::Moved)) + .then(|| watcher::Said::Unreachable(failure.clone())), + }; + let said = said.or_else(|| { + (found == Some(Found::Unmoved)).then_some(watcher::Said::Vouched { at: read.at }) + }); + if let Some(said) = said { + answers.push(Answer { + project: project.clone(), + said, + journal: self + .journals + .get_mut(project) + .and_then(|journal| journal.said.take()), + }); + } } - self.read + answers + } + + /// What a watcher is told where no read found anything moved: that each + /// project read is as current as the read. + fn vouched(&self) -> Vec { + self.found .iter() - .map(|(project, read)| Answer { - project: project.clone(), - said: match &read.work { - Ok(work) => watcher::Said::Read { - at: read.at, - beads: work - .beads - .iter() - .map(|bead| { - let key = BeadKey { - project: project.clone(), - id: bead.id.clone(), - }; - let bd = BeadReadiness { - ready: work.readiness.ready.contains(&bead.id), - blocked_by: work - .readiness - .blocked_by - .get(&bead.id) - .cloned() - .unwrap_or_default(), - }; - let (ready, blocked_by) = match drawn.get(&key) { - Some(node) => (node.ready, node.blocked_by.clone()), - None => (bd.ready, bd.blocked_by.clone()), - }; - let held = Held { - bead: Arc::clone(bead), - ready, - blocked_by, - bd, - }; - (key.id, held) - }) - .collect(), + .filter(|(_, found)| **found == Found::Unmoved) + .filter_map(|(project, _)| { + Some(Answer { + project: project.clone(), + said: watcher::Said::Vouched { + at: self.read.get(project)?.at, }, - Err(failure) => watcher::Said::Unreachable(failure.clone()), - }, - journal: self - .journals - .get_mut(project) - .and_then(|journal| journal.said.take()), + journal: None, + }) }) .collect() } @@ -387,6 +489,9 @@ impl Collection { { match answer { Ok(Refresh::Unchanged { as_of }) => { + self.found + .entry(project.name.clone()) + .or_insert(Found::Unmoved); // A skipped read is a successful read: `bdi` knows the // tracker has not moved, so the project is as fresh as if // the cascade had run and the foot must not draw it as @@ -401,6 +506,7 @@ impl Collection { as_of, journal, }) => { + self.found.insert(project.name.clone(), Found::Moved); if let Some(journal) = journal { self.journals .entry(project.name.clone()) @@ -418,6 +524,7 @@ impl Collection { ); } Err(failure) => { + self.found.insert(project.name.clone(), Found::Moved); // Storing what the probe just read would make this // failure sticky: the next probe would match it, the // cascade that would have recovered is skipped, and the @@ -621,10 +728,7 @@ impl Collection { .map(|(project, read)| (project.to_string(), read.at)) .collect(); - let speaks_until = self - .that_answered(cfg) - .filter_map(|(project, work)| Some((project.to_string(), work.speaks_until?))) - .collect(); + let speaks_until = self.speaks_until(cfg); let read_for_reach = cfg .read() @@ -657,6 +761,14 @@ impl Collection { .filter_map(|p| Some((p.name.as_str(), self.read.get(&p.name)?))) } + /// When each project whose tracker answered stops speaking for it with + /// nothing written, where it does. + fn speaks_until(&self, cfg: &Config) -> BTreeMap> { + self.that_answered(cfg) + .filter_map(|(project, work)| Some((project.to_string(), work.speaks_until?))) + .collect() + } + /// The projects whose trackers answered, in the same order. fn that_answered<'a>( &'a self, @@ -864,6 +976,66 @@ fn held_by_unread(drawn: &[Drawn<'_>], scope: &Scope) -> BTreeSet { .collect() } +/// bd's own readiness for bead `id` of `work`. +fn bd_readiness(work: &ProjectWork, id: &str) -> BeadReadiness { + BeadReadiness { + ready: work.readiness.ready.contains(id), + blocked_by: work + .readiness + .blocked_by + .get(id) + .cloned() + .unwrap_or_default(), + } +} + +/// The readiness the trees gave each of `work`'s beads they drew, where it +/// differs from bd's own. +fn differing_readiness( + work: &ProjectWork, + drawn: Option<&BTreeMap<&str, &Node>>, +) -> BTreeMap { + drawn + .into_iter() + .flatten() + .filter_map(|(id, node)| { + let bd = bd_readiness(work, id); + (node.ready != bd.ready || node.blocked_by != bd.blocked_by).then(|| { + let drawn = BeadReadiness { + ready: node.ready, + blocked_by: node.blocked_by.clone(), + }; + (id.to_string(), drawn) + }) + }) + .collect() +} + +/// Every bead of `work` as the watcher holds it, read `at`, each taking the +/// readiness `differing` gives it and bd's own otherwise. +fn held_by_watcher( + work: &ProjectWork, + differing: &BTreeMap, + at: DateTime, +) -> watcher::Said { + let beads = work + .beads + .iter() + .map(|bead| { + let bd = bd_readiness(work, &bead.id); + let given = differing.get(&bead.id).unwrap_or(&bd); + let held = Held { + bead: Arc::clone(bead), + ready: given.ready, + blocked_by: given.blocked_by.clone(), + bd, + }; + (bead.id.clone(), held) + }) + .collect(); + watcher::Said::Read { at, beads } +} + /// A tree's beads, gathered under the project whose answer holds each. fn by_project<'a>(project: &'a str, assembled: &'a Assembled) -> Vec<(&'a str, Vec>)> { let mut rows: BTreeMap<&str, Vec>> = BTreeMap::new(); @@ -3372,6 +3544,177 @@ prefix = "kad" )); } + // ---- what a watcher is told after the first collection -------------- + + /// Ferry's tracker holding its bead waiting on dunwich's epic, which bd + /// cannot see and so calls the bead ready. + fn ferry_waiting() -> Fake { + Fake::holding(beads(WAITING_ON_DUNWICH)).ready(["fer-2"]) + } + + /// Both projects read once, as a watcher starts. + fn watching_ferry_wait() -> Collection { + let trackers = Fakes::default() + .with("dunwich", dunwich_tracker()) + .with("ferry", ferry_waiting()); + let mut collection = Collection::default(); + watched(&mut collection, &trackers, &Wanted::Everything); + collection + } + + fn watched(collection: &mut Collection, trackers: &Fakes, wanted: &Wanted) -> Watched { + collection.watched(&two_projects(), &no_panes(), trackers, wanted, now()) + } + + /// What each answer said, by project: read, vouched for or unreachable. + fn said_of(watched: &Watched) -> Vec<(&str, &'static str)> { + watched + .answers + .iter() + .map(|answer| { + let said = match answer.said { + watcher::Said::Read { .. } => "read", + watcher::Said::Vouched { .. } => "vouched", + watcher::Said::Unreachable(_) => "unreachable", + }; + (answer.project.as_str(), said) + }) + .collect() + } + + #[test] + fn a_watcher_starts_with_every_project_read() { + let trackers = Fakes::default() + .with("dunwich", dunwich_tracker()) + .with("ferry", ferry_waiting()); + + let first = watched(&mut Collection::default(), &trackers, &Wanted::Everything); + + assert_eq!(said_of(&first), [("dunwich", "read"), ("ferry", "read")]); + assert_eq!(first.projects, ["dunwich", "ferry"]); + } + + /// The probe found nothing, so there is nothing to draw and nothing new + /// to say of any bead: the project read is vouched for, and no other is + /// spoken of. + #[test] + fn a_watcher_is_told_only_that_a_project_nothing_moved_in_is_current() { + let mut collection = watching_ferry_wait(); + let trackers = Fakes::default() + .with("dunwich", dunwich_tracker()) + .with("ferry", ferry_waiting()); + let nested = crate::model::tree::nestings_on_this_thread(); + + let again = watched(&mut collection, &trackers, &dunwich_alone()); + + assert_eq!(said_of(&again), [("dunwich", "vouched")]); + assert_eq!( + crate::model::tree::nestings_on_this_thread() - nested, + 0, + "nothing moved, so nothing was drawn" + ); + assert_eq!(again.projects, ["dunwich", "ferry"]); + } + + /// Dunwich moved without changing anything ferry's bead waits on, so + /// ferry has nothing new to say. + #[test] + fn a_watcher_is_told_of_a_project_that_moved_and_not_of_one_it_left_alone() { + let mut collection = watching_ferry_wait(); + let trackers = Fakes::default() + .with("dunwich", dunwich_tracker().moved()) + .with("ferry", ferry_waiting()); + + let again = watched(&mut collection, &trackers, &dunwich_alone()); + + assert_eq!(said_of(&again), [("dunwich", "read")]); + } + + /// Closing dunwich's epic frees ferry's bead, which ferry's own tracker + /// cannot see, so ferry is answered again though it was not read. + #[test] + fn a_watcher_is_told_of_a_bead_a_read_of_another_project_freed() { + let mut collection = watching_ferry_wait(); + let closed = DUNWICH_TREE.replacen(r#""status":"in_progress""#, r#""status":"closed""#, 1); + let trackers = Fakes::default() + .with("dunwich", dunwich_holding(&closed).moved()) + .with("ferry", ferry_waiting()); + + let again = watched(&mut collection, &trackers, &dunwich_alone()); + + assert_eq!(said_of(&again), [("dunwich", "read"), ("ferry", "read")]); + let freed = held(&again.answers, "ferry", "fer-2"); + assert_eq!((freed.ready, freed.blocked_by.len()), (true, 0)); + } + + /// A project the collection was not asked to read, and whose beads it + /// left as they were, is not spoken of. + #[test] + fn a_watcher_is_not_told_of_a_project_the_collection_did_not_read() { + let mut collection = watching_ferry_wait(); + let trackers = Fakes::default() + .with("dunwich", dunwich_tracker()) + .with("ferry", ferry_waiting().moved()); + + let again = watched(&mut collection, &trackers, &ferry()); + + assert_eq!(said_of(&again), [("ferry", "read")]); + } + + /// A tracker the watcher has been told it cannot reach is not said to be + /// unreachable again by a collection that never asked it. + #[test] + fn a_watcher_is_not_told_again_of_a_tracker_the_collection_did_not_ask() { + let unreachable = |dunwich| { + Fakes::default().with("dunwich", dunwich).with( + "ferry", + ferry_waiting().failing(Asked::All, failing(FailureKind::Auth)), + ) + }; + let mut collection = Collection::default(); + let first = watched( + &mut collection, + &unreachable(dunwich_tracker()), + &Wanted::Everything, + ); + + let again = watched( + &mut collection, + &unreachable(dunwich_tracker().moved()), + &dunwich_alone(), + ); + + assert_eq!( + said_of(&first), + [("dunwich", "read"), ("ferry", "unreachable")] + ); + assert_eq!(said_of(&again), [("dunwich", "read")]); + } + + /// The instant a project nothing moved in is current as of travels with + /// it, so a consumer's freshness moves on. + #[test] + fn a_project_vouched_for_is_current_as_of_the_collection() { + let mut collection = watching_ferry_wait(); + let trackers = Fakes::default() + .with("dunwich", dunwich_tracker()) + .with("ferry", ferry_waiting()); + let later = now() + TimeDelta::seconds(30); + + let again = collection.watched( + &two_projects(), + &no_panes(), + &trackers, + &dunwich_alone(), + later, + ); + + assert!(matches!( + again.answers[0].said, + watcher::Said::Vouched { at } if at == later + )); + } + // ---- bd's events journal ---------------------------------------------- fn record(seq: u64, op: &str, id: &str) -> Value { diff --git a/src/app/mod.rs b/src/app/mod.rs index 0aa7866..a948716 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -22,12 +22,12 @@ mod watcher; mod watching; pub use armed::{armed_unread, Armed, Arming}; -pub use collection::{run, Asked, Awaited, Collection, Wanted}; +pub use collection::{run, Asked, Awaited, Collection, Wanted, Watched}; pub use due::due_after; pub use outstanding::Outstanding; pub use reading::Reading; pub use reading_trackers::{ReadingTrackers, Reads}; -pub use watcher::{hold, serve, Hold}; +pub use watcher::{allocate_from_one_arena, hold, serve, Hold}; /// The fake trackers and panes both halves read in their tests. /// diff --git a/src/app/reading_trackers.rs b/src/app/reading_trackers.rs index f312ecb..55d685a 100644 --- a/src/app/reading_trackers.rs +++ b/src/app/reading_trackers.rs @@ -10,15 +10,13 @@ use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender}; use chrono::{DateTime, Utc}; use crate::collect::changes::Heard; -use crate::model::snapshot::Snapshot; use super::watcher::{Answer, ChangeSource, Said}; -use super::{Asked, Outstanding, Reading, Wanted}; +use super::{Asked, Outstanding, Reading, Wanted, Watched}; -/// One read of what `Wanted` names, as of the instant handed in: the -/// collection drawn from it, and what each project standing says to the -/// watcher. -pub type Reads = Box) -> (Snapshot, Vec) + Send>; +/// One read of what `Wanted` names, as of the instant handed in, and what +/// each project with something new to say says to the watcher. +pub type Reads = Box) -> Watched + Send>; /// A change source that reads every configured project's tracker, once at /// the start and then whenever a project's poll comes round or a producer @@ -64,14 +62,14 @@ impl ReadingTrackers { let Ok(Asked::Read(wanted)) = self.asking.1.try_recv() else { return false; }; - let (snapshot, answers) = (self.reads)(&wanted, now); + let watched = (self.reads)(&wanted, now); self.reading.came_back( self.outstanding.came_back(), - &snapshot.projects, - &snapshot.speaks_until, + &watched.projects, + &watched.speaks_until, Utc::now(), ); - self.told.extend(answers); + self.told.extend(watched.answers); true } @@ -140,38 +138,36 @@ mod tests { use crate::collect::agents::testing::Fake as Provider; use crate::collect::changes::Reported; use crate::collect::tracker::testing::Fakes; - use crate::model::snapshot::Filter; /// Every project `trackers` holds, read through a collection as /// `bdi watch` reads them, with no window and nothing polling. fn reading(trackers: Arc) -> (ReadingTrackers, Sender) { - polling_every(None, trackers) + dunwich_polling_every(None, trackers) } - /// As [`reading`], with every project polling `every` after each read. - fn polling_every( + /// As [`reading`], with dunwich polling `every` after each read and ferry + /// never polling, so whatever is said of ferry after its first read was + /// said for a producer. + fn dunwich_polling_every( every: Option, trackers: Arc, ) -> (ReadingTrackers, Sender) { let cfg = two_projects(); let mut collection = Collection::default(); let reads: Reads = Box::new(move |wanted, now| { - let snapshot = collection.collect( + collection.watched( &cfg, &Provider::holding(Vec::new()), trackers.as_ref(), wanted, - Filter::All, now, - ); - let answers = collection.answers(&snapshot); - (snapshot, answers) + ) }); let (tell, heard) = mpsc::channel(); - let armed = ["dunwich", "ferry"] - .into_iter() - .map(|project| Armed::polling(project.to_string(), every)) - .collect(); + let armed = vec![ + Armed::polling("dunwich".to_string(), every), + Armed::polling("ferry".to_string(), None), + ]; let source = ReadingTrackers::new( reads, heard, @@ -215,13 +211,13 @@ mod tests { let read_once = trackers.tracker("dunwich").asked().len(); tell.send(Heard::Changed("dunwich".to_string())).unwrap(); - let answers: Vec = (0..2).map(|_| source.next().expect("an answer")).collect(); + let answer = source.next().expect("an answer"); assert!( trackers.tracker("dunwich").asked().len() > read_once, "dunwich's tracker was asked again" ); - assert!(answers.iter().all(|a| matches!(a.said, Said::Read { .. }))); + assert_eq!(answer.project, "dunwich"); } #[test] @@ -243,7 +239,7 @@ mod tests { #[test] fn a_producers_line_is_taken_while_polls_keep_coming_due() { - let (mut source, tell) = polling_every(Some(std::time::Duration::ZERO), trackers()); + let (mut source, tell) = dunwich_polling_every(Some(std::time::Duration::ZERO), trackers()); for _ in 0..2 { source.next(); } @@ -251,7 +247,10 @@ mod tests { tell.send(Heard::Covered("ferry".to_string())).unwrap(); assert!( - (0..10).any(|_| matches!(source.next().expect("an answer").said, Said::Vouched { .. })), + (0..10).any(|_| { + let answer = source.next().expect("an answer"); + answer.project == "ferry" && matches!(answer.said, Said::Vouched { .. }) + }), "ferry's cover was never taken while polls kept coming due" ); } diff --git a/src/app/watcher.rs b/src/app/watcher.rs index ff96f80..815c706 100644 --- a/src/app/watcher.rs +++ b/src/app/watcher.rs @@ -117,9 +117,13 @@ impl Standing { /// with how current that is. None until there is something to say: a /// project neither read nor found unreachable has a watch wait rather /// than be told it is empty. - fn told(&self, project: &str, interest: &mut Interest) -> Option> { + /// + /// Every interest is caught up whenever the beads are replaced, so one + /// whose beads have not been replaced since has nothing to catch up on. + fn told(&self, project: &str, interest: &mut Interest, replaced: bool) -> Option> { let mut lines = match &self.beads { - Some(beads) => interest.catch_up(project, beads), + Some(beads) if replaced => interest.catch_up(project, beads), + Some(_) => Vec::new(), None if self.unreachable.is_some() => Vec::new(), None => return None, }; @@ -190,6 +194,7 @@ impl Hold { /// Take what a source said of one project, and tell everyone watching it. pub fn take(&mut self, answer: Answer) { let standing = self.projects.entry(answer.project.clone()).or_default(); + let replaced = matches!(answer.said, Said::Read { .. }); match answer.said { Said::Read { at, beads } => { standing.beads = Some(beads); @@ -225,10 +230,12 @@ impl Hold { }) .map(|record| watching::event_line(&project, record)) .collect(); - standing.told(&project, interest).is_none_or(|told| { - lines.extend(told); - consumer.tells(lines) - }) + standing + .told(&project, interest, replaced) + .is_none_or(|told| { + lines.extend(told); + consumer.tells(lines) + }) }); } @@ -268,7 +275,7 @@ impl Hold { interest.widen(watch); lines.extend( self.projects[project] - .told(project, interest) + .told(project, interest, true) .into_iter() .flatten(), ); @@ -320,6 +327,24 @@ fn give_back_freed_memory() { #[cfg(not(all(target_os = "linux", target_env = "gnu")))] fn give_back_freed_memory() {} +/// Have every thread of the watcher allocate from one arena. +/// +/// glibc gives each thread that allocates an arena of its own, and what a +/// thread frees stays in its arena for that thread to reuse. A read and a +/// catch-up each free tens of megabytes on threads that then sit idle, so +/// every arena keeps its own share resident and the watcher holds about +/// twice what it uses. An arena once made is kept, so this comes before the +/// watcher starts any thread. +#[cfg(all(target_os = "linux", target_env = "gnu"))] +pub fn allocate_from_one_arena() { + // SAFETY: `mallopt` takes no pointer, and `M_ARENA_MAX` only limits the + // arenas made from here on. + unsafe { libc::mallopt(libc::M_ARENA_MAX, 1) }; +} + +#[cfg(not(all(target_os = "linux", target_env = "gnu")))] +pub fn allocate_from_one_arena() {} + /// How many answers a connection may fall behind by before it is hung up /// on. A consumer that reconnects is sent the beads as they then stand, so /// hanging up loses it nothing, where queueing for it without end would cost diff --git a/src/cli.rs b/src/cli.rs index 52528e2..19ce378 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -12,7 +12,8 @@ use chrono::{DateTime, Utc}; use clap::{Parser, Subcommand}; use crate::app::{ - armed_unread, hold, serve, Armed, Arming, Hold, Outstanding, ReadingTrackers, Reads, + allocate_from_one_arena, armed_unread, hold, serve, Armed, Arming, Hold, Outstanding, + ReadingTrackers, Reads, }; use crate::app::{Asked, Wanted}; use crate::collect::agents::{Agents, Unasked}; @@ -479,6 +480,7 @@ fn arming(polling: Polling) -> Arming { /// /// The config is read once. Replacing what the watcher reads is a restart. fn watch(socket: Option, config: &Path) -> anyhow::Result { + allocate_from_one_arena(); let cwd = std::env::current_dir().context("finding the current directory")?; let cfg = read_config( &RealRunner, @@ -529,11 +531,8 @@ fn watch(socket: Option, config: &Path) -> anyhow::Result { let trackers = bd::Cli::new(&RealRunner) .reading_journals() .caching_environments(EnvironmentCache::here()); - let reads: Reads = Box::new(move |wanted, now| { - let snapshot = collection.collect(&cfg, &Unasked, &trackers, wanted, Filter::All, now); - let answers = collection.answers(&snapshot); - (snapshot, answers) - }); + let reads: Reads = + Box::new(move |wanted, now| collection.watched(&cfg, &Unasked, &trackers, wanted, now)); let mut source = ReadingTrackers::new(reads, heard, outstanding, reading); std::thread::spawn(move || hold(&mut source, &held)); diff --git a/src/model/tree.rs b/src/model/tree.rs index 6a5874a..66203ff 100644 --- a/src/model/tree.rs +++ b/src/model/tree.rs @@ -15,37 +15,53 @@ use crate::model::types::{Bead, Dependency, Edge}; /// Everything else — letters, punctuation, digit runs of equal value — /// compares as written. pub(crate) fn numeric_id_order(a: &str, b: &str) -> std::cmp::Ordering { - let (mut a, mut b) = (a.chars().peekable(), b.chars().peekable()); + // Bytes rather than chars: UTF-8 orders as the chars it encodes, and no + // byte of a multi-byte char is an ASCII digit. + let (mut a, mut b) = (a.as_bytes(), b.as_bytes()); loop { - return match (a.peek(), b.peek()) { - (None, None) => std::cmp::Ordering::Equal, - (None, Some(_)) => std::cmp::Ordering::Less, - (Some(_), None) => std::cmp::Ordering::Greater, + let order = match (a.first(), b.first()) { + (None, None) => return std::cmp::Ordering::Equal, + (None, Some(_)) => return std::cmp::Ordering::Less, + (Some(_), None) => return std::cmp::Ordering::Greater, (Some(x), Some(y)) if x.is_ascii_digit() && y.is_ascii_digit() => { - let digits = |it: &mut std::iter::Peekable| -> String { - std::iter::from_fn(|| it.next_if(char::is_ascii_digit)).collect() - }; - let (a_digits, b_digits) = (digits(&mut a), digits(&mut b)); - let (a_value, b_value): (u128, u128) = ( - a_digits.parse().unwrap_or(u128::MAX), - b_digits.parse().unwrap_or(u128::MAX), - ); - match a_value.cmp(&b_value) { - std::cmp::Ordering::Equal => match a_digits.cmp(&b_digits) { - std::cmp::Ordering::Equal => continue, - other => other, - }, - other => other, - } + let (a_digits, a_rest) = digit_run(a); + let (b_digits, b_rest) = digit_run(b); + (a, b) = (a_rest, b_rest); + digits_value(a_digits) + .cmp(&digits_value(b_digits)) + .then_with(|| a_digits.cmp(b_digits)) + } + (Some(x), Some(y)) => { + (a, b) = (&a[1..], &b[1..]); + x.cmp(y) } - (Some(_), Some(_)) => match a.next().cmp(&b.next()) { - std::cmp::Ordering::Equal => continue, - other => other, - }, }; + if order.is_ne() { + return order; + } } } +/// The run of ASCII digits `id` starts with, and what follows it. +fn digit_run(id: &[u8]) -> (&[u8], &[u8]) { + let end = id + .iter() + .position(|byte| !byte.is_ascii_digit()) + .unwrap_or(id.len()); + id.split_at(end) +} + +/// The value of a run of ASCII digits, or the largest there is where it +/// will not fit. +fn digits_value(digits: &[u8]) -> u128 { + digits + .iter() + .try_fold(0u128, |value, digit| { + value.checked_mul(10)?.checked_add(u128::from(digit - b'0')) + }) + .unwrap_or(u128::MAX) +} + /// One way down from a bead to a bead beneath it. #[derive(Debug, Clone, PartialEq, Eq)] pub struct Link { @@ -872,6 +888,27 @@ mod tests { use crate::collect::bd::parse_shared_beads; use crate::model::types::Edge; + #[test] + fn ids_compare_as_a_reader_reads_them() { + use std::cmp::Ordering::{Equal, Greater, Less}; + let huge = "9".repeat(40); + let (beyond, further) = (format!("t.{huge}0"), format!("t.{huge}1")); + let cases = [ + ("t.2", "t.10", Less), + ("t.10", "t.2", Greater), + ("t.10", "t.10", Equal), + ("t", "t.1", Less), + ("t.1a", "t.1b", Less), + ("t.01", "t.1", Less), + ("é.2", "é.10", Less), + ("a.1", "é.1", Less), + (beyond.as_str(), further.as_str(), Less), + ]; + for (a, b, order) in cases { + assert_eq!(numeric_id_order(a, b), order, "{a} against {b}"); + } + } + /// A slice of this project's own tracker as `bd list --all --json` /// writes it: one open root, one bead in flight, and closed beads whose /// blocks edges cross between them.