From c088214fed1b697cc9734a1bb84aa6536426991b Mon Sep 17 00:00:00 2001 From: Anders Olsson Date: Tue, 4 Aug 2026 21:50:40 +0200 Subject: [PATCH] fix(history): stop reprocessing the slice that was just appended to MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `add_events_with_prior` advanced `k` past the slice it had written when it created a new one, but not when it appended to an existing one. The trailing forward-refresh loop therefore started *on* the slice just modified and ran `new_forward_info` over it again. That is not merely redundant work. The loop immediately above it sets each agent's message to `forward * likelihood` for that slice, and `new_forward_info` then assigns `skill.forward = message.forget(drift)` — folding the slice's own likelihood back into its own forward prior. The skills it produced depended on how events had been batched. Ingesting one event at a time now converges to the same fixed point as ingesting the same events in a single call, which it previously did not: for five events sharing a timestamp, competitor `a` converged to mu=7.44 sigma=3.90 batched versus mu=7.99 sigma=3.10 incrementally. Both runs had converged; the gap was not a convergence residual. The numerical goldens never caught this because they all ingest in one call with a distinct timestamp per event, so the append-to-existing-slice branch is never taken. `tests/ingestion_equivalence.rs` covers it directly, and asserts convergence before comparing so that a residual cannot be mistaken for agreement. Removing the redundant re-inference also removes the dominant cost of incremental ingestion, which was quadratic in the number of events already in the slice: events before after speedup 500 45.8ms 1.1ms 42x 1000 179.5ms 2.8ms 64x 2000 721.8ms 9.9ms 73x 4000 2.9s 35.4ms 82x Ingesting one at a time is now 1.8x a single batched call, down from 148x. Two supporting changes are included: - Color groups are rebuilt lazily rather than on every append. Nothing reads the partition between an append and the next full sweep, so the per-append rebuild was pure waste. - `ColorGroups::groups_are_contiguous` is asserted after each rebuild and in `color_range`. The parallel sweep derives one `&mut` sub-slice per color from those ranges and relies on them being disjoint; that invariant was established by construction but never checked. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01DnsaJg74eNSva3PJjK2eej --- src/color_group.rs | 39 ++++++++- src/history.rs | 2 + src/key_table.rs | 39 +++++---- src/time_slice.rs | 23 +++++- tests/ingestion_equivalence.rs | 147 +++++++++++++++++++++++++++++++++ 5 files changed, 233 insertions(+), 17 deletions(-) create mode 100644 tests/ingestion_equivalence.rs diff --git a/src/color_group.rs b/src/color_group.rs index 6add43c..b33ea41 100644 --- a/src/color_group.rs +++ b/src/color_group.rs @@ -49,16 +49,53 @@ impl ColorGroups { /// Contiguous index range for one color after events have been reordered /// into color-contiguous positions by `TimeSlice::recompute_color_groups`. - #[allow(dead_code)] pub(crate) fn color_range(&self, color_idx: usize) -> std::ops::Range { let group = &self.groups[color_idx]; if group.is_empty() { return 0..0; } + let start = *group.first().unwrap(); let end = *group.last().unwrap() + 1; + + debug_assert_eq!( + end - start, + group.len(), + "color {color_idx} is not contiguous; its range would overlap other colors" + ); + start..end } + + /// Whether every color occupies a contiguous, ascending range of event + /// indices, and no two colors overlap. + /// + /// The parallel sweep derives one `&mut` sub-slice per color from these + /// ranges and relies on them being disjoint. That disjointness is what + /// makes concurrent writes to distinct skills sound, so it is checked + /// rather than assumed. + pub(crate) fn groups_are_contiguous(&self) -> bool { + let mut expected_start = 0; + + for group in &self.groups { + if group.is_empty() { + continue; + } + + let ascending_run = group + .iter() + .enumerate() + .all(|(offset, &idx)| idx == group[0] + offset); + + if !ascending_run || group[0] != expected_start { + return false; + } + + expected_start += group.len(); + } + + true + } } /// Compute color groups greedily. diff --git a/src/history.rs b/src/history.rs index aea04e2..8690e44 100644 --- a/src/history.rs +++ b/src/history.rs @@ -646,6 +646,8 @@ impl, O: Observer, K: Eq + Hash + Clone> History(HashMap); +pub struct KeyTable { + forward: HashMap, + /// Reverse mapping, indexed by `Index.0`. + /// + /// Indices are handed out densely and sequentially, so position *is* the + /// index and `key()` is a lookup rather than a scan over every entry. + reverse: Vec, +} impl KeyTable where - K: Eq + Hash, + K: Eq + Hash + Clone, { pub fn new() -> Self { - Self(HashMap::new()) + Self { + forward: HashMap::new(), + reverse: Vec::new(), + } } pub fn get(&self, k: &Q) -> Option where K: Borrow, { - self.0.get(k).cloned() + self.forward.get(k).cloned() } pub fn get_or_create>(&mut self, k: &Q) -> Index where K: Borrow, { - if let Some(idx) = self.0.get(k) { + if let Some(idx) = self.forward.get(k) { *idx } else { - let idx = Index::from(self.0.len()); - self.0.insert(k.to_owned(), idx); + let idx = Index::from(self.reverse.len()); + let owned = k.to_owned(); + self.reverse.push(owned.clone()); + self.forward.insert(owned, idx); idx } } pub fn key(&self, idx: Index) -> Option<&K> { - self.0 - .iter() - .find(|&(_, value)| *value == idx) - .map(|(key, _)| key) + self.reverse.get(idx.0) } pub fn keys(&self) -> impl Iterator { - self.0.keys() + self.forward.keys() } pub fn len(&self) -> usize { - self.0.len() + self.reverse.len() } pub fn is_empty(&self) -> bool { - self.0.is_empty() + self.reverse.is_empty() } } impl Default for KeyTable where - K: Eq + Hash, + K: Eq + Hash + Clone, { fn default() -> Self { KeyTable::new() diff --git a/src/time_slice.rs b/src/time_slice.rs index 1aed80b..0f001b1 100644 --- a/src/time_slice.rs +++ b/src/time_slice.rs @@ -174,6 +174,14 @@ pub struct TimeSlice { pub(crate) convergence: crate::ConvergenceOptions, arena: ScratchArena, pub(crate) color_groups: ColorGroups, + /// Whether `color_groups` still reflects `events`. + /// + /// Coloring is rebuilt lazily, on the first full sweep after an append, + /// rather than eagerly per append: the partition is thrown away and + /// recomputed wholesale either way, so doing it per append made ingesting + /// n events O(n^2) with no benefit — nothing reads the partition between + /// an append and the next full sweep. + color_groups_dirty: bool, } impl TimeSlice { @@ -186,6 +194,7 @@ impl TimeSlice { convergence, arena: ScratchArena::new(), color_groups: ColorGroups::new(), + color_groups_dirty: false, } } @@ -198,6 +207,7 @@ impl TimeSlice { let n = self.events.len(); if n == 0 { self.color_groups = ColorGroups::new(); + self.color_groups_dirty = false; return; } @@ -221,6 +231,12 @@ impl TimeSlice { self.events = reordered; self.color_groups = ColorGroups { groups: new_groups }; + self.color_groups_dirty = false; + + debug_assert!( + self.color_groups.groups_are_contiguous(), + "color groups must occupy contiguous event ranges" + ); } pub fn add_events>( @@ -306,8 +322,9 @@ impl TimeSlice { self.events.extend(events); + self.color_groups_dirty = true; + self.iteration(from, agents); - self.recompute_color_groups(); } pub(crate) fn posteriors(&self) -> HashMap { @@ -318,6 +335,10 @@ impl TimeSlice { } pub fn iteration>(&mut self, from: usize, agents: &CompetitorStore) { + if from == 0 && self.color_groups_dirty { + self.recompute_color_groups(); + } + if from > 0 || self.color_groups.is_empty() { // Initial pass (add_events) or no color groups yet: simple sequential sweep. for event in self.events.iter_mut().skip(from) { diff --git a/tests/ingestion_equivalence.rs b/tests/ingestion_equivalence.rs new file mode 100644 index 0000000..afd6b31 --- /dev/null +++ b/tests/ingestion_equivalence.rs @@ -0,0 +1,147 @@ +//! Ingesting the same events must give the same answer however they were +//! batched. +//! +//! The numerical goldens all ingest in a single call with one slice per +//! timestamp, so they never exercise the "append to an existing slice" path. +//! These do. + +use smallvec::smallvec; +use trueskill_tt::{ConvergenceOptions, Event, Gaussian, History, Member, Outcome, Team}; + +/// Converge tightly: the default cap of 30 iterations leaves a residual around +/// 1e-6, which would swamp the comparison. Both paths must reach the same +/// fixed point, so drive both well past it. +fn tight() -> ConvergenceOptions { + ConvergenceOptions { + max_iter: 2_000, + epsilon: 1e-12, + ..ConvergenceOptions::default() + } +} + +fn event(a: &str, b: &str, time: i64) -> Event { + Event { + time, + teams: smallvec![ + Team::with_members([Member::new(a.to_string())]), + Team::with_members([Member::new(b.to_string())]), + ], + outcome: Outcome::winner(0, 2), + } +} + +fn converged_skills(events: Vec>, batched: bool) -> Vec<(String, Gaussian)> { + let mut h: History = + History::builder_with_key().convergence(tight()).build(); + + if batched { + h.add_events(events).unwrap(); + } else { + for ev in events { + h.add_events(std::iter::once(ev)).unwrap(); + } + } + + let report = h.converge().unwrap(); + assert!( + report.converged, + "fixture must converge before results can be compared; final step {:?}", + report.final_step + ); + + let mut skills: Vec<(String, Gaussian)> = h + .learning_curves() + .into_iter() + .map(|(key, curve)| (key, curve.last().unwrap().1)) + .collect(); + skills.sort_by(|a, b| a.0.cmp(&b.0)); + skills +} + +fn assert_same(batched: &[(String, Gaussian)], incremental: &[(String, Gaussian)], what: &str) { + assert_eq!( + batched.len(), + incremental.len(), + "{what}: competitor count differs" + ); + + for ((kb, gb), (ki, gi)) in batched.iter().zip(incremental.iter()) { + assert_eq!(kb, ki, "{what}: key order differs"); + assert!( + (gb.mu() - gi.mu()).abs() < 1e-8 && (gb.sigma() - gi.sigma()).abs() < 1e-8, + "{what}: {kb} differs — batched mu={} sigma={}, incremental mu={} sigma={}", + gb.mu(), + gb.sigma(), + gi.mu(), + gi.sigma() + ); + } +} + +/// All events share one timestamp, so incremental ingestion repeatedly appends +/// to an existing slice. +#[test] +fn same_slice_incremental_matches_batched() { + let events = vec![ + event("a", "b", 1), + event("c", "d", 1), + event("e", "f", 1), + event("a", "c", 1), + event("b", "e", 1), + ]; + + let batched = converged_skills(events.clone(), true); + let incremental = converged_skills(events, false); + assert_same(&batched, &incremental, "single shared slice"); +} + +/// Distinct timestamps, so each append lands in a fresh slice appended after +/// the existing ones. +#[test] +fn distinct_slices_incremental_matches_batched() { + let events = vec![ + event("a", "b", 1), + event("b", "c", 2), + event("c", "a", 3), + event("a", "c", 4), + ]; + + let batched = converged_skills(events.clone(), true); + let incremental = converged_skills(events, false); + assert_same(&batched, &incremental, "distinct slices"); +} + +/// Several events per timestamp across several timestamps — appends to +/// existing slices interleaved with new ones. +#[test] +fn mixed_slices_incremental_matches_batched() { + let events = vec![ + event("a", "b", 1), + event("c", "d", 1), + event("a", "c", 2), + event("b", "d", 2), + event("a", "d", 3), + event("b", "c", 3), + ]; + + let batched = converged_skills(events.clone(), true); + let incremental = converged_skills(events, false); + assert_same(&batched, &incremental, "mixed slices"); +} + +/// Appending an event to a slice that is *not* the most recent one exercises +/// the forward refresh of every later slice. +#[test] +fn back_dated_event_matches_batched() { + let events = vec![ + event("a", "b", 1), + event("b", "c", 5), + event("c", "a", 9), + // arrives last, but belongs to the middle slice + event("a", "c", 5), + ]; + + let batched = converged_skills(events.clone(), true); + let incremental = converged_skills(events, false); + assert_same(&batched, &incremental, "back-dated event"); +}