fix(history): stop reprocessing the slice that was just appended to
`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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DnsaJg74eNSva3PJjK2eej
This commit is contained in:
+38
-1
@@ -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<usize> {
|
||||
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.
|
||||
|
||||
@@ -646,6 +646,8 @@ impl<T: Time, D: Drift<T>, O: Observer<T>, K: Eq + Hash + Clone> History<T, D, O
|
||||
agent.last_time = Some(t);
|
||||
agent.message = time_slice.forward_prior_out(&agent_idx);
|
||||
}
|
||||
|
||||
k += 1;
|
||||
} else {
|
||||
let mut time_slice = TimeSlice::new(t, self.p_draw, self.convergence);
|
||||
time_slice.add_events(composition, results, weights, kinds_chunk, &self.agents);
|
||||
|
||||
+24
-15
@@ -12,59 +12,68 @@ use crate::Index;
|
||||
/// crate. Power users can promote `&K` to `Index` via `get_or_create` and
|
||||
/// skip the lookup on subsequent hot-path calls.
|
||||
#[derive(Debug)]
|
||||
pub struct KeyTable<K>(HashMap<K, Index>);
|
||||
pub struct KeyTable<K> {
|
||||
forward: HashMap<K, Index>,
|
||||
/// 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<K>,
|
||||
}
|
||||
|
||||
impl<K> KeyTable<K>
|
||||
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<Q: ?Sized + Hash + Eq>(&self, k: &Q) -> Option<Index>
|
||||
where
|
||||
K: Borrow<Q>,
|
||||
{
|
||||
self.0.get(k).cloned()
|
||||
self.forward.get(k).cloned()
|
||||
}
|
||||
|
||||
pub fn get_or_create<Q: ?Sized + Hash + Eq + ToOwned<Owned = K>>(&mut self, k: &Q) -> Index
|
||||
where
|
||||
K: Borrow<Q>,
|
||||
{
|
||||
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<Item = &K> {
|
||||
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<K> Default for KeyTable<K>
|
||||
where
|
||||
K: Eq + Hash,
|
||||
K: Eq + Hash + Clone,
|
||||
{
|
||||
fn default() -> Self {
|
||||
KeyTable::new()
|
||||
|
||||
+22
-1
@@ -174,6 +174,14 @@ pub struct TimeSlice<T: Time = i64> {
|
||||
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<T: Time> TimeSlice<T> {
|
||||
@@ -186,6 +194,7 @@ impl<T: Time> TimeSlice<T> {
|
||||
convergence,
|
||||
arena: ScratchArena::new(),
|
||||
color_groups: ColorGroups::new(),
|
||||
color_groups_dirty: false,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -198,6 +207,7 @@ impl<T: Time> TimeSlice<T> {
|
||||
let n = self.events.len();
|
||||
if n == 0 {
|
||||
self.color_groups = ColorGroups::new();
|
||||
self.color_groups_dirty = false;
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -221,6 +231,12 @@ impl<T: Time> TimeSlice<T> {
|
||||
|
||||
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<D: Drift<T>>(
|
||||
@@ -306,8 +322,9 @@ impl<T: Time> TimeSlice<T> {
|
||||
|
||||
self.events.extend(events);
|
||||
|
||||
self.color_groups_dirty = true;
|
||||
|
||||
self.iteration(from, agents);
|
||||
self.recompute_color_groups();
|
||||
}
|
||||
|
||||
pub(crate) fn posteriors(&self) -> HashMap<Index, Gaussian> {
|
||||
@@ -318,6 +335,10 @@ impl<T: Time> TimeSlice<T> {
|
||||
}
|
||||
|
||||
pub fn iteration<D: Drift<T>>(&mut self, from: usize, agents: &CompetitorStore<T, D>) {
|
||||
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) {
|
||||
|
||||
@@ -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<i64, String> {
|
||||
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<Event<i64, String>>, batched: bool) -> Vec<(String, Gaussian)> {
|
||||
let mut h: History<i64, _, _, String> =
|
||||
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");
|
||||
}
|
||||
Reference in New Issue
Block a user