From 82611004582629c6de778c9996f7625fafba4419 Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 29 Sep 2026 11:29:48 +0200 Subject: [PATCH 1/7] feat(execution): split MatchResult trade / filled-id reservations (#219) Add MatchResult::try_reserve_trades and try_reserve_filled_order_ids so a caller can size each vector on its own; a taker that only partially fills its makers no longer has to allocate a filled-id buffer. try_reserve and try_with_capacity keep their shared-capacity behaviour (#148). Also repairs the #217 CHANGELOG entry for is_poisoned (missing word, long line). --- CHANGELOG.md | 6 +++ src/execution/match_result.rs | 64 ++++++++++++++++++++++++----- src/execution/tests/capacity.rs | 71 +++++++++++++++++++++++++++++++++ 3 files changed, 130 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 82045ef..0b17b0f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 (recovering the guard does not clear the lock's poison). The answer can become `true` concurrently, so the result of the operation itself stays the authoritative report. +- **Split `MatchResult` reservations (#219).** + `MatchResult::try_reserve_trades(n)` and + `MatchResult::try_reserve_filled_order_ids(n)` size one vector each, so a + caller that expects trades without fully consumed makers does not also + allocate a filled-id buffer. `try_reserve(n)` and `try_with_capacity` keep + sizing both vectors from one count. ### Fixed diff --git a/src/execution/match_result.rs b/src/execution/match_result.rs index 6bc6b8d..3d066c0 100644 --- a/src/execution/match_result.rs +++ b/src/execution/match_result.rs @@ -293,7 +293,10 @@ impl MatchResult { /// Never allocates when the spare capacity already suffices. The matching /// engine calls this before committing each maker step, so the step's /// [`Self::add_trade`] / [`Self::add_filled_order_id`] cannot then fail on - /// growth. + /// growth. A caller that knows the two counts separately (for example a + /// taker that only partially fills its makers produces trades but no + /// filled ids) can size each vector on its own with + /// [`Self::try_reserve_trades`] and [`Self::try_reserve_filled_order_ids`]. /// /// # Errors /// @@ -304,7 +307,47 @@ impl MatchResult { #[inline] pub fn try_reserve(&mut self, additional: usize) -> Result<(), PriceLevelError> { self.trades.try_reserve(additional)?; - self.try_reserve_filled(additional) + self.try_reserve_filled_order_ids(additional) + } + + /// Reserves room for at least `additional` more trades (amortized + /// growth), leaving the filled-id vector alone (issue #219). + /// + /// Never allocates when the spare capacity already suffices; after `Ok`, + /// `additional` further [`Self::add_trade`] calls cannot fail on growth. + /// + /// # Errors + /// + /// Returns [`PriceLevelError::CapacityExceeded`] with resource + /// [`CapacityResource::Trades`] if the trade vector cannot grow (an + /// unrepresentable size or an allocator refusal). Every observable field + /// is left unchanged. + #[inline] + pub fn try_reserve_trades(&mut self, additional: usize) -> Result<(), PriceLevelError> { + self.trades.try_reserve(additional) + } + + /// Reserves room for at least `additional` more filled order ids + /// (amortized growth), leaving the trade vector alone (issue #219). + /// + /// Never allocates when the spare capacity already suffices; after `Ok`, + /// `additional` further [`Self::add_filled_order_id`] calls cannot fail on + /// growth. + /// + /// # Errors + /// + /// Returns [`PriceLevelError::CapacityExceeded`] with resource + /// [`CapacityResource::FilledOrderIds`] if the id vector cannot grow (an + /// unrepresentable size or an allocator refusal). Every observable field + /// is left unchanged. + #[inline] + pub fn try_reserve_filled_order_ids( + &mut self, + additional: usize, + ) -> Result<(), PriceLevelError> { + self.filled_order_ids.try_reserve(additional).map_err(|_| { + PriceLevelError::capacity_exceeded(CapacityResource::FilledOrderIds, additional) + }) } /// Exact-growth variant of [`Self::try_reserve`], used by the fill-or-kill @@ -318,6 +361,13 @@ impl MatchResult { }) } + /// Capacity of the filled-id vector (issue #219 test seam). + #[cfg(test)] + #[must_use] + pub(crate) fn test_filled_order_ids_capacity(&self) -> usize { + self.filled_order_ids.capacity() + } + /// `true` when one more trade AND one more filled id fit without growing. /// The matching engine's per-step fast-path check. #[inline] @@ -333,14 +383,6 @@ impl MatchResult { && self.filled_order_ids.len() < self.filled_order_ids.capacity() } - /// Reserves `additional` filled-order-id slots (amortized growth). - #[inline] - fn try_reserve_filled(&mut self, additional: usize) -> Result<(), PriceLevelError> { - self.filled_order_ids.try_reserve(additional).map_err(|_| { - PriceLevelError::capacity_exceeded(CapacityResource::FilledOrderIds, additional) - }) - } - /// Add a trade to this match result. /// /// All validation and the storage reservation happen BEFORE any field is @@ -406,7 +448,7 @@ impl MatchResult { /// [`CapacityResource::FilledOrderIds`] if the id vector cannot grow; the /// result is left unchanged. pub fn add_filled_order_id(&mut self, order_id: Id) -> Result<(), PriceLevelError> { - self.try_reserve_filled(1)?; + self.try_reserve_filled_order_ids(1)?; // Capacity reserved: this push cannot reallocate. self.filled_order_ids.push(order_id); Ok(()) diff --git a/src/execution/tests/capacity.rs b/src/execution/tests/capacity.rs index 37a15e0..f0d4011 100644 --- a/src/execution/tests/capacity.rs +++ b/src/execution/tests/capacity.rs @@ -104,6 +104,77 @@ mod tests { assert!(result.error().is_none()); } + /// Issue #219: each split reservation sizes only its own vector. + #[test] + fn split_reservations_size_one_vector_each() { + let mut result = MatchResult::new(Id::from_u64(10), Quantity::new(20)); + result.try_reserve_trades(4).expect("reserve trades"); + assert!(result.trades().capacity() >= 4); + assert_eq!(result.test_filled_order_ids_capacity(), 0); + + let mut result = MatchResult::new(Id::from_u64(10), Quantity::new(20)); + result + .try_reserve_filled_order_ids(3) + .expect("reserve filled ids"); + assert!(result.test_filled_order_ids_capacity() >= 3); + assert_eq!(result.trades().capacity(), 0); + + // Zero never allocates. + let mut result = MatchResult::new(Id::from_u64(10), Quantity::new(20)); + result.try_reserve_trades(0).expect("zero trades"); + result.try_reserve_filled_order_ids(0).expect("zero ids"); + assert_eq!(result.trades().capacity(), 0); + assert_eq!(result.test_filled_order_ids_capacity(), 0); + + // The combined convenience still sizes both from one count. + let mut result = MatchResult::new(Id::from_u64(10), Quantity::new(20)); + result.try_reserve(2).expect("reserve both"); + assert!(result.trades().capacity() >= 2); + assert!(result.test_filled_order_ids_capacity() >= 2); + } + + /// Issue #219: a refused split reservation reports its own resource and + /// changes no observable field; spare capacity makes it allocation-free. + #[test] + fn split_reservation_failures_change_nothing() { + let mut result = MatchResult::new(Id::from_u64(10), Quantity::new(20)); + result.add_trade(trade(1, 5)).expect("add_trade"); + result + .add_filled_order_id(Id::from_u64(1)) + .expect("filled id"); + + let err = result + .try_reserve_trades(usize::MAX) + .expect_err("must overflow"); + assert_capacity_error(&err, CapacityResource::Trades, usize::MAX); + let err = result + .try_reserve_filled_order_ids(usize::MAX) + .expect_err("must overflow"); + assert_capacity_error(&err, CapacityResource::FilledOrderIds, usize::MAX); + let n = byte_overflow_len::(); + let err = result + .try_reserve_filled_order_ids(n) + .expect_err("byte size overflows"); + assert_capacity_error(&err, CapacityResource::FilledOrderIds, n); + + let _limit = trade_list_seam::limit_trades(1); + let err = result.try_reserve_trades(1).expect_err("limited"); + assert_capacity_error(&err, CapacityResource::Trades, 1); + + assert_eq!(result.trades().len(), 1); + assert_eq!(result.filled_order_ids(), &[Id::from_u64(1)]); + assert_eq!(result.remaining_quantity().as_u64(), 15); + assert_eq!(result.outcome(), MatchOutcome::PartiallyFilled); + assert!(result.error().is_none()); + + // Spare capacity: the reservation reuses the buffer. + let mut result = MatchResult::new(Id::from_u64(10), Quantity::new(20)); + result.try_reserve_filled_order_ids(4).expect("reserve"); + let before = result.filled_order_ids().as_ptr(); + result.try_reserve_filled_order_ids(4).expect("fits"); + assert_eq!(result.filled_order_ids().as_ptr(), before); + } + /// A failed growth in `add_trade` must leave every observable field as it /// was: the reservation happens before remaining / completion / outcome /// are committed. From 89163d837b6e18ac7a6d17e5ecb9dc0b94312b62 Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 29 Sep 2026 11:34:07 +0200 Subject: [PATCH 2/7] feat(execution): MatchResult::try_absorb for multi-level aggregation (#219) Fold a later level's MatchResult for the same taker into an aggregate. Validation (taker id, the level matched with the aggregate's remaining quantity, aggregate not terminal) and reservation happen before anything moves, so on any Err the aggregate is unchanged and the level result still holds its committed trades. Per vector, an empty aggregate buffer that is too small is swapped with the level's (no allocation); otherwise entries are appended into the reserved buffer (TradeList::append_reserved). Outcome and completion are recomputed with the finalize rules, a level's Killed / Rejected is adopted only when nothing traded, and the level's error becomes the aggregate's. The drained level is left as a consistent empty result. Allocation pass (PL_LATENCY_ONLY=alloc cargo bench --bench latency), two trades + one filled id per level: 1 level: copy 2.00 allocs/op (704 B), absorb 0.00 3 levels: copy 3.00 allocs/op (1,856 B), absorb 3.00 (1,856 B) --- CHANGELOG.md | 15 + benches/latency/alloc_measurements.rs | 119 +++++++ src/execution/list.rs | 10 + src/execution/match_result.rs | 168 ++++++++++ src/execution/tests/absorb.rs | 446 ++++++++++++++++++++++++++ src/execution/tests/mod.rs | 1 + 6 files changed, 759 insertions(+) create mode 100644 src/execution/tests/absorb.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 0b17b0f..a66825a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 caller that expects trades without fully consumed makers does not also allocate a filled-id buffer. `try_reserve(n)` and `try_with_capacity` keep sizing both vectors from one count. +- **`MatchResult::try_absorb` (#219).** Folds one price level's result into + a multi-level aggregate for the same taker. Trades and filled ids are + appended in order; remaining quantity, completion and outcome are + recomputed (a level's `Killed` / `Rejected` is adopted only when nothing + traded); the level's error becomes the aggregate's. The level must have + been matched with the aggregate's remaining quantity, and a terminal + aggregate (error, killed, rejected) refuses further levels. An empty, + too-small aggregate vector takes the level's buffer by swap, so absorbing + the first level allocates nothing; otherwise the entries are appended into + the existing reservation. On any refusal (`InvalidOperation`, + `CapacityExceeded`) the aggregate is unchanged and the level result still + holds its committed trades. Allocation pass (`PL_LATENCY_ONLY=alloc make + bench-latency`): one level of two trades and one filled id, 2.00 allocs/op + (704 B) with reserve-and-copy against 0 with `try_absorb`; three such + levels, 3.00 allocs/op (1,856 B) for both. ### Fixed diff --git a/benches/latency/alloc_measurements.rs b/benches/latency/alloc_measurements.rs index 3c49e7f..4efd8ae 100644 --- a/benches/latency/alloc_measurements.rs +++ b/benches/latency/alloc_measurements.rs @@ -78,9 +78,128 @@ pub fn run_all(config: &Config) -> Vec { measure_snapshot_capture(config), measure_checksum_validate(config), measure_restore(config), + // Issue #219: folding per-level results into a multi-level aggregate. + measure_aggregate(config, 1, Aggregation::Copy), + measure_aggregate(config, 1, Aggregation::Absorb), + measure_aggregate(config, 3, Aggregation::Copy), + measure_aggregate(config, 3, Aggregation::Absorb), ] } +/// How [`measure_aggregate`] folds each level's result into the aggregate. +#[derive(Debug, Clone, Copy)] +enum Aggregation { + /// The pre-#219 pattern: reserve the aggregate for the level's entries, + /// then copy them in with `add_trade` / `add_filled_order_id`. + Copy, + /// `MatchResult::try_absorb`: move the level's buffers when the + /// aggregate's are empty, append otherwise. + Absorb, +} + +/// Measures folding `levels` per-level results (two trades and one filled id +/// each, sized like `match_order` sizes its own result) into a fresh +/// aggregate (issue #219). The level results are built before counting +/// starts; the aggregates are kept and dropped after it stops, so only the +/// folding itself is counted. `reps` is the number of aggregations. The +/// `Copy` pattern reserves exactly each level's entry counts, the cheapest +/// form of that pattern (a caller reserving each level's worst case before +/// matching allocates at least as often). +fn measure_aggregate(config: &Config, levels: usize, aggregation: Aggregation) -> AllocReport { + const TRADES_PER_LEVEL: u64 = 2; + const FILL: u64 = 5; + let reps = config.alloc_reps; + let taker = Id::from_u64(TAKER_ID_BASE); + let total = levels as u64 * TRADES_PER_LEVEL * FILL; + let mut inputs: Vec> = (0..reps) + .map(|_| { + (0..levels as u64) + .map(|level| { + let incoming = total - level * TRADES_PER_LEVEL * FILL; + let mut result = MatchResult::try_with_capacity( + taker, + Quantity::new(incoming), + TRADES_PER_LEVEL as usize, + ) + .expect("alloc measurement: level result capacity"); + for step in 0..TRADES_PER_LEVEL { + let maker = level * TRADES_PER_LEVEL + step + 1; + result + .add_trade(Trade::with_timestamp( + Id::from_u64(TAKER_ID_BASE + 1_000 + maker), + taker, + Id::from_u64(maker), + Price::new(LEVEL_PRICE), + Quantity::new(FILL), + Side::Buy, + TimestampMs::new(EXECUTION_TIMESTAMP_MS), + )) + .expect("alloc measurement: level trade"); + } + result + .add_filled_order_id(Id::from_u64(level * TRADES_PER_LEVEL + 1)) + .expect("alloc measurement: level filled id"); + result + }) + .collect() + }) + .collect(); + let mut outputs: Vec = Vec::with_capacity(reps); + + alloc::reset(); + alloc::enable(); + let before = AllocStats::read(); + for level_results in &mut inputs { + let mut aggregate = MatchResult::new(taker, Quantity::new(total)); + for result in level_results.iter_mut() { + match aggregation { + Aggregation::Copy => { + aggregate + .try_reserve_trades(result.trades().len()) + .expect("alloc measurement: reserve trades"); + aggregate + .try_reserve_filled_order_ids(result.filled_order_ids().len()) + .expect("alloc measurement: reserve filled ids"); + for &trade in result.trades().as_vec() { + aggregate + .add_trade(trade) + .expect("alloc measurement: add_trade"); + } + for &id in result.filled_order_ids() { + aggregate + .add_filled_order_id(id) + .expect("alloc measurement: add_filled_order_id"); + } + } + Aggregation::Absorb => aggregate + .try_absorb(result) + .expect("alloc measurement: try_absorb"), + } + } + outputs.push(aggregate); + } + let after = AllocStats::read(); + alloc::disable(); + + assert!( + outputs.iter().all(MatchResult::is_complete), + "alloc measurement (aggregate): every aggregate must fill the taker" + ); + drop(outputs); + drop(inputs); + + AllocReport { + name: match (levels, aggregation) { + (1, Aggregation::Copy) => "aggregate_1_copy", + (1, Aggregation::Absorb) => "aggregate_1_absorb", + (_, Aggregation::Copy) => "aggregate_3_copy", + (_, Aggregation::Absorb) => "aggregate_3_absorb", + }, + reps, + totals: after.since(before), + } +} + /// Measures `add_order` of a fresh order into a level with fixed resting /// depth. Every harness-owned buffer (`orders`, `handles`) is allocated /// BEFORE counting starts and dropped AFTER counting stops, so only diff --git a/src/execution/list.rs b/src/execution/list.rs index 8d3a6c5..4c37748 100644 --- a/src/execution/list.rs +++ b/src/execution/list.rs @@ -112,6 +112,16 @@ impl TradeList { self.trades.push(trade); } + /// Moves every trade of `other` to the end of this list, leaving `other` + /// empty (its capacity is kept). The caller must already have reserved + /// room for `other.len()` more trades (via [`Self::try_reserve`]): with + /// that spare capacity `Vec::append` never reallocates, so this cannot + /// fail or panic (issue #219). + #[inline] + pub(crate) fn append_reserved(&mut self, other: &mut TradeList) { + self.trades.append(&mut other.trades); + } + /// Clones the list without an infallible allocation. /// /// `Clone` is still derived (it aborts / panics like any `Vec` clone on diff --git a/src/execution/match_result.rs b/src/execution/match_result.rs index 3d066c0..d458f9e 100644 --- a/src/execution/match_result.rs +++ b/src/execution/match_result.rs @@ -361,6 +361,168 @@ impl MatchResult { }) } + /// Folds a later price level's result for the same taker into this + /// aggregate (issue #219), moving the level's buffers instead of copying + /// them when that is lossless. + /// + /// A caller that sweeps several levels matches each one with this + /// aggregate's current [`Self::remaining_quantity`] and absorbs the + /// level's result before moving on. The merge is: + /// + /// - **Trades and filled ids** are appended in order, after this + /// aggregate's own. Filled ids are assumed unique across levels (an + /// order rests at one price) and are not re-checked. + /// - **Remaining quantity** becomes the level's remaining quantity, so + /// the aggregate's executed quantity is the sum of both. + /// - **Outcome and completion** are recomputed from the combined fields: + /// `Filled` when nothing remains, `NotFilled` without trades, otherwise + /// `PartiallyFilled`. When the combined result has no trades and the + /// level was killed or rejected, that outcome is adopted. + /// - **Error:** the level's error, if any, becomes the aggregate's. The + /// aggregate can hold no error of its own here (see below), so the + /// first failure always wins. + /// + /// # Allocation + /// + /// Per vector: when this aggregate's vector is empty and too small for + /// the level's contents, the two vectors are swapped, so the aggregate + /// takes the level's buffer and the level gets the aggregate's (empty) + /// reservation back; nothing is allocated or copied. Otherwise the + /// entries are appended into this aggregate's buffer, which allocates + /// only if its spare capacity is too small (amortized growth). Existing + /// entries and a sufficient reservation are always kept. Absorbing the + /// first level into a fresh [`Self::new`] aggregate therefore never + /// allocates. + /// + /// # Failure ownership + /// + /// Every check and reservation happens before anything moves. On `Err` + /// no observable field of `self` has changed (only a vector's spare + /// capacity may have grown) and `level` is untouched, so the committed + /// trades it records are still in the caller's hands. On `Ok` `level` is + /// left as an empty result for its remaining quantity: no trades, no + /// filled ids, no error, outcome `Filled` or `NotFilled`. + /// + /// # Errors + /// + /// Returns [`PriceLevelError::InvalidOperation`] if `level` belongs to a + /// different taker order id; if this aggregate is already terminal (it + /// carries an error, or it was killed or rejected), since no later level + /// may be matched after either; if the level's executed quantity overflows + /// `u64`; or if the level's executed plus remaining quantity differs from + /// this aggregate's remaining quantity (the level was not matched with + /// what the aggregate had left). Returns + /// [`PriceLevelError::CapacityExceeded`] (resource + /// [`CapacityResource::Trades`] or [`CapacityResource::FilledOrderIds`]) if + /// a vector cannot grow. + /// + /// # Examples + /// + /// ```rust + /// use pricelevel::{MatchResult, PriceLevelError, Quantity}; + /// # use pricelevel::{Id, PriceLevel, TakerKind, TimeInForce, TimestampMs, UuidGenerator}; + /// + /// fn sweep( + /// levels: &[PriceLevel], + /// taker: Id, + /// quantity: u64, + /// ids: &UuidGenerator, + /// ) -> Result { + /// let mut aggregate = MatchResult::new(taker, Quantity::new(quantity)); + /// for level in levels { + /// let remaining = aggregate.remaining_quantity().as_u64(); + /// if remaining == 0 { + /// break; + /// } + /// let mut result = level.match_order( + /// remaining, + /// taker, + /// TimeInForce::Ioc, + /// TakerKind::Standard, + /// TimestampMs::new(1_700_000_000_000), + /// ids, + /// ); + /// // On `Err`, `result` still holds the level's committed trades. + /// aggregate.try_absorb(&mut result)?; + /// if aggregate.is_failed() { + /// break; + /// } + /// } + /// Ok(aggregate) + /// } + /// # let ids = UuidGenerator::new(uuid::Uuid::nil()); + /// # let aggregate = sweep(&[PriceLevel::new(100)], Id::from_u64(1), 5, &ids).unwrap(); + /// # assert_eq!(aggregate.remaining_quantity().as_u64(), 5); + /// ``` + pub fn try_absorb(&mut self, level: &mut MatchResult) -> Result<(), PriceLevelError> { + // 1. Validate: nothing changes on a refusal. + if level.order_id != self.order_id { + return Err(absorb_refused(format!( + "absorbed result's taker order id {} does not match the aggregate's {}", + level.order_id, self.order_id + ))); + } + if self.error.is_some() || self.outcome.was_killed() || self.outcome.was_rejected() { + return Err(absorb_refused(format!( + "the aggregate is terminal ({:?}, error recorded: {}); no later level may be absorbed", + self.outcome, + self.error.is_some() + ))); + } + let level_executed = level.executed_quantity()?.as_u64(); + if level_executed.checked_add(level.remaining_quantity) != Some(self.remaining_quantity) { + return Err(absorb_refused(format!( + "absorbed result executed {level_executed} with {} remaining, but the aggregate has {} remaining", + level.remaining_quantity, self.remaining_quantity + ))); + } + + // 2. Plan and reserve: still nothing observable changes. + let swap_trades = self.trades.is_empty() && self.trades.capacity() < level.trades.len(); + let swap_filled = self.filled_order_ids.is_empty() + && self.filled_order_ids.capacity() < level.filled_order_ids.len(); + if !swap_trades { + self.trades.try_reserve(level.trades.len())?; + } + if !swap_filled { + self.try_reserve_filled_order_ids(level.filled_order_ids.len())?; + } + + // 3. Commit: nothing below can fail. + if swap_trades { + std::mem::swap(&mut self.trades, &mut level.trades); + } else { + self.trades.append_reserved(&mut level.trades); + } + if swap_filled { + std::mem::swap(&mut self.filled_order_ids, &mut level.filled_order_ids); + } else { + // Reserved above: `append` cannot reallocate. + self.filled_order_ids.append(&mut level.filled_order_ids); + } + self.remaining_quantity = level.remaining_quantity; + self.is_complete = self.remaining_quantity == 0; + self.outcome = if self.is_complete { + MatchOutcome::Filled + } else if self.trades.is_empty() { + // Nothing traded anywhere: a level's kill / rejection is the + // whole story (remaining > 0, no trades, no filled ids). + if level.outcome.was_killed() || level.outcome.was_rejected() { + level.outcome + } else { + MatchOutcome::NotFilled + } + } else { + MatchOutcome::PartiallyFilled + }; + if let Some(error) = level.error.take() { + self.set_error(error); + } + // The level is drained: leave it as a consistent empty result. + level.finalize(Quantity::new(level.remaining_quantity)); + Ok(()) + } + /// Capacity of the filled-id vector (issue #219 test seam). #[cfg(test)] #[must_use] @@ -931,6 +1093,12 @@ fn executed_quantity_overflow() -> PriceLevelError { } } +/// The typed refusal of [`MatchResult::try_absorb`] (issue #219). +#[cold] +fn absorb_refused(message: String) -> PriceLevelError { + PriceLevelError::InvalidOperation { message } +} + impl fmt::Display for MatchResult { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!( diff --git a/src/execution/tests/absorb.rs b/src/execution/tests/absorb.rs new file mode 100644 index 0000000..5aa6061 --- /dev/null +++ b/src/execution/tests/absorb.rs @@ -0,0 +1,446 @@ +//! Issue #219: `MatchResult::try_absorb`, folding one price level's result +//! into a multi-level aggregate. Covers the validation refusals (aggregate +//! unchanged, source intact), the lossless buffer move into an empty +//! aggregate, the reserve-and-append path, first-error-wins, kill / reject +//! adoption, and a property over sequences of level results. + +#[cfg(test)] +// Scoped to this co-located test module only (issue #173): raw arithmetic is +// permitted inside `mod tests` per the Testing section of +// `rules/global_rules.md`. Production code outside this module keeps the +// full deny list. +#[allow(clippy::arithmetic_side_effects)] +mod tests { + use crate::errors::{CapacityResource, ExhaustedCounter, PriceLevelError}; + use crate::execution::match_result::{MatchOutcome, MatchResult}; + use crate::execution::trade::Trade; + use crate::execution::trade_list_seam; + use crate::orders::{Id, Side}; + use crate::utils::{Price, Quantity, TimestampMs}; + use proptest::prelude::*; + use std::str::FromStr; + + const TAKER: u64 = 10; + + fn trade(taker: u64, maker: u64, quantity: u64) -> Trade { + Trade::with_timestamp( + Id::from_u64(1_000_000 + maker), + Id::from_u64(taker), + Id::from_u64(maker), + Price::new(1_000), + Quantity::new(quantity), + Side::Buy, + TimestampMs::new(1_616_823_000_000), + ) + } + + /// A level result for `incoming`: one trade per `(maker, quantity, + /// consumed)`, a filled id for each consumed maker, finalized like the + /// engine does. + fn level(incoming: u64, fills: &[(u64, u64, bool)]) -> MatchResult { + let mut result = MatchResult::new(Id::from_u64(TAKER), Quantity::new(incoming)); + for &(maker, quantity, consumed) in fills { + result + .add_trade(trade(TAKER, maker, quantity)) + .expect("add_trade"); + if consumed { + result + .add_filled_order_id(Id::from_u64(maker)) + .expect("filled id"); + } + } + let remaining = result.remaining_quantity(); + result.finalize(remaining); + result + } + + fn json(result: &MatchResult) -> String { + serde_json::to_string(result).expect("serialize") + } + + /// The aggregate decodes through the validator (serde and text forms). + fn assert_round_trips(result: &MatchResult) { + let decoded: MatchResult = serde_json::from_str(&json(result)).expect("validated decode"); + assert_eq!(json(&decoded), json(result)); + MatchResult::from_str(&result.to_string()).expect("validated text decode"); + } + + fn maker_ids(result: &MatchResult) -> Vec { + result + .trades() + .as_vec() + .iter() + .map(Trade::maker_order_id) + .collect() + } + + fn assert_refused_unchanged( + aggregate: &mut MatchResult, + source: &mut MatchResult, + ) -> PriceLevelError { + let aggregate_before = json(aggregate); + let source_before = json(source); + let err = aggregate.try_absorb(source).expect_err("must refuse"); + assert_eq!(json(aggregate), aggregate_before, "aggregate unchanged"); + assert_eq!(json(source), source_before, "source intact"); + err + } + + #[test] + fn taker_mismatch_is_refused() { + let mut aggregate = MatchResult::new(Id::from_u64(TAKER), Quantity::new(10)); + let mut other = MatchResult::new(Id::from_u64(TAKER + 1), Quantity::new(10)); + other.add_trade(trade(TAKER + 1, 1, 4)).expect("add_trade"); + let err = assert_refused_unchanged(&mut aggregate, &mut other); + assert!(matches!(err, PriceLevelError::InvalidOperation { .. })); + } + + #[test] + fn quantity_mismatch_is_refused() { + // The level was matched with more, and with less, than the aggregate + // has left. + for incoming in [9, 11] { + let mut aggregate = level(20, &[(1, 10, true)]); + let mut source = level(incoming, &[(2, 3, false)]); + let err = assert_refused_unchanged(&mut aggregate, &mut source); + assert!(matches!(err, PriceLevelError::InvalidOperation { .. })); + } + } + + #[test] + fn terminal_aggregate_is_refused() { + let mut failed = level(20, &[(1, 5, true)]); + failed.set_error(PriceLevelError::counter_exhausted( + ExhaustedCounter::QueueSequence, + )); + let mut killed = MatchResult::new(Id::from_u64(TAKER), Quantity::new(20)); + killed.mark_killed(20); + let mut rejected = MatchResult::new(Id::from_u64(TAKER), Quantity::new(20)); + rejected.mark_rejected(20); + + for mut aggregate in [failed, killed, rejected] { + let remaining = aggregate.remaining_quantity().as_u64(); + let mut source = level(remaining, &[(2, 1, false)]); + let err = assert_refused_unchanged(&mut aggregate, &mut source); + assert!(matches!(err, PriceLevelError::InvalidOperation { .. })); + } + } + + #[test] + fn empty_aggregate_takes_the_level_buffers_by_move() { + let mut aggregate = MatchResult::new(Id::from_u64(TAKER), Quantity::new(20)); + let mut source = + MatchResult::try_with_capacity(Id::from_u64(TAKER), Quantity::new(20), 4).expect("cap"); + source.add_trade(trade(TAKER, 1, 6)).expect("t1"); + source.add_filled_order_id(Id::from_u64(1)).expect("f1"); + source.add_trade(trade(TAKER, 2, 4)).expect("t2"); + let remaining = source.remaining_quantity(); + source.finalize(remaining); + let trades_ptr = source.trades().as_vec().as_ptr(); + let trades_capacity = source.trades().capacity(); + let filled_ptr = source.filled_order_ids().as_ptr(); + let filled_capacity = source.test_filled_order_ids_capacity(); + + aggregate.try_absorb(&mut source).expect("absorb"); + + // The level's buffers moved: same allocation, same capacity, so the + // absorb itself allocated nothing. + assert_eq!(aggregate.trades().as_vec().as_ptr(), trades_ptr); + assert_eq!(aggregate.trades().capacity(), trades_capacity); + assert_eq!(aggregate.filled_order_ids().as_ptr(), filled_ptr); + assert_eq!(aggregate.test_filled_order_ids_capacity(), filled_capacity); + assert_eq!( + maker_ids(&aggregate), + vec![Id::from_u64(1), Id::from_u64(2)] + ); + assert_eq!(aggregate.filled_order_ids(), &[Id::from_u64(1)]); + assert_eq!(aggregate.remaining_quantity().as_u64(), 10); + assert_eq!(aggregate.outcome(), MatchOutcome::PartiallyFilled); + assert_round_trips(&aggregate); + + // The source is drained into a consistent empty result. + assert!(source.trades().is_empty()); + assert!(source.filled_order_ids().is_empty()); + assert!(source.error().is_none()); + assert_eq!(source.remaining_quantity().as_u64(), 10); + assert_eq!(source.outcome(), MatchOutcome::NotFilled); + assert_round_trips(&source); + } + + #[test] + fn sufficient_reservation_is_kept_and_appended_into() { + let mut aggregate = MatchResult::new(Id::from_u64(TAKER), Quantity::new(20)); + aggregate.try_reserve(8).expect("reserve"); + let trades_ptr = aggregate.trades().as_vec().as_ptr(); + let filled_ptr = aggregate.filled_order_ids().as_ptr(); + let mut source = level(20, &[(1, 6, true), (2, 4, true)]); + + aggregate.try_absorb(&mut source).expect("absorb"); + + // Appended into the caller's reservation; nothing moved or grew. + assert_eq!(aggregate.trades().as_vec().as_ptr(), trades_ptr); + assert_eq!(aggregate.filled_order_ids().as_ptr(), filled_ptr); + assert!(aggregate.trades().capacity() >= 8); + assert_eq!( + maker_ids(&aggregate), + vec![Id::from_u64(1), Id::from_u64(2)] + ); + assert_eq!( + aggregate.filled_order_ids(), + &[Id::from_u64(1), Id::from_u64(2)] + ); + assert_round_trips(&aggregate); + } + + #[test] + fn existing_filled_ids_are_not_lost() { + // The aggregate holds a filled id but no trades: its trade vector is + // moved in, its id vector is appended to. + let mut aggregate = MatchResult::new(Id::from_u64(TAKER), Quantity::new(20)); + aggregate + .add_filled_order_id(Id::from_u64(7)) + .expect("filled id"); + let mut source = level(20, &[(1, 5, true)]); + + aggregate.try_absorb(&mut source).expect("absorb"); + + assert_eq!(maker_ids(&aggregate), vec![Id::from_u64(1)]); + assert_eq!( + aggregate.filled_order_ids(), + &[Id::from_u64(7), Id::from_u64(1)] + ); + assert_eq!(aggregate.remaining_quantity().as_u64(), 15); + } + + #[test] + fn append_path_keeps_order_and_refuses_growth_atomically() { + let mut aggregate = level(20, &[(1, 5, true)]); + let mut source = level(15, &[(2, 3, true), (3, 2, false)]); + + // A capped trade list refuses the append: nothing changes. + { + let _limit = trade_list_seam::limit_trades(2); + let err = assert_refused_unchanged(&mut aggregate, &mut source); + assert!(matches!( + err, + PriceLevelError::CapacityExceeded { + resource: CapacityResource::Trades, + .. + } + )); + } + + aggregate.try_absorb(&mut source).expect("absorb"); + assert_eq!( + maker_ids(&aggregate), + vec![Id::from_u64(1), Id::from_u64(2), Id::from_u64(3)] + ); + assert_eq!( + aggregate.filled_order_ids(), + &[Id::from_u64(1), Id::from_u64(2)] + ); + assert_eq!(aggregate.remaining_quantity().as_u64(), 10); + assert_eq!(aggregate.executed_quantity().expect("sum").as_u64(), 10); + assert_eq!(aggregate.outcome(), MatchOutcome::PartiallyFilled); + assert!(!aggregate.is_complete()); + assert_round_trips(&aggregate); + } + + #[test] + fn filling_level_completes_the_aggregate() { + let mut aggregate = level(10, &[(1, 4, true)]); + let mut source = level(6, &[(2, 6, true)]); + aggregate.try_absorb(&mut source).expect("absorb"); + assert!(aggregate.is_complete()); + assert_eq!(aggregate.outcome(), MatchOutcome::Filled); + assert_eq!(source.outcome(), MatchOutcome::Filled); + assert_round_trips(&aggregate); + assert_round_trips(&source); + } + + #[test] + fn first_error_wins() { + let first = PriceLevelError::counter_exhausted(ExhaustedCounter::QueueSequence); + let mut aggregate = MatchResult::new(Id::from_u64(TAKER), Quantity::new(20)); + let mut source = level(20, &[(1, 5, true)]); + source.set_error(first.clone()); + + aggregate.try_absorb(&mut source).expect("absorb"); + assert_eq!(aggregate.error(), Some(&first)); + assert!(source.error().is_none(), "the error was taken"); + assert_round_trips(&aggregate); + + // The failed aggregate is terminal: a later level cannot replace it. + let mut later = level(15, &[(2, 1, false)]); + later.set_error(PriceLevelError::capacity_exceeded( + CapacityResource::Trades, + 1, + )); + let _ = assert_refused_unchanged(&mut aggregate, &mut later); + assert_eq!(aggregate.error(), Some(&first)); + } + + #[test] + fn kill_and_reject_adopted_only_without_trades() { + for reject in [false, true] { + let mark = |result: &mut MatchResult, quantity: u64| { + if reject { + result.mark_rejected(quantity); + } else { + result.mark_killed(quantity); + } + }; + let expected = if reject { + MatchOutcome::Rejected + } else { + MatchOutcome::Killed + }; + + // Nothing traded anywhere: the level's verdict is adopted. + let mut aggregate = MatchResult::new(Id::from_u64(TAKER), Quantity::new(20)); + let mut source = MatchResult::new(Id::from_u64(TAKER), Quantity::new(20)); + mark(&mut source, 20); + aggregate.try_absorb(&mut source).expect("absorb"); + assert_eq!(aggregate.outcome(), expected); + assert_eq!(aggregate.remaining_quantity().as_u64(), 20); + assert!(!aggregate.is_complete()); + assert_round_trips(&aggregate); + + // Earlier trades: the aggregate stays partially filled. + let mut aggregate = level(20, &[(1, 5, true)]); + let mut source = MatchResult::new(Id::from_u64(TAKER), Quantity::new(15)); + mark(&mut source, 15); + aggregate.try_absorb(&mut source).expect("absorb"); + assert_eq!(aggregate.outcome(), MatchOutcome::PartiallyFilled); + assert_round_trips(&aggregate); + } + } + + #[test] + fn zero_quantity_aggregate_absorbs_a_vacuous_level() { + let mut aggregate = MatchResult::new(Id::from_u64(TAKER), Quantity::new(0)); + let mut source = MatchResult::new(Id::from_u64(TAKER), Quantity::new(0)); + aggregate.try_absorb(&mut source).expect("absorb"); + assert!(aggregate.is_complete()); + assert_eq!(aggregate.outcome(), MatchOutcome::Filled); + assert_round_trips(&aggregate); + } + + /// One generated level: fills `(quantity, consumed)`, then an optional + /// kill (only when nothing traded) and an optional error. + #[derive(Debug, Clone)] + struct LevelSpec { + fills: Vec<(u64, bool)>, + killed: bool, + error: bool, + } + + fn level_spec() -> impl Strategy { + ( + prop::collection::vec((1u64..=8, any::()), 0..4), + any::(), + prop::bool::weighted(0.15), + ) + .prop_map(|(fills, killed, error)| LevelSpec { + fills, + killed, + error, + }) + } + + proptest! { + #![proptest_config(ProptestConfig { cases: 512, ..ProptestConfig::default() })] + + #[test] + fn absorb_keeps_the_aggregate_invariants( + initial in 0u64..=40, + specs in prop::collection::vec(level_spec(), 0..6), + ) { + let mut aggregate = MatchResult::new(Id::from_u64(TAKER), Quantity::new(initial)); + let mut next_maker = 1u64; + let mut expected_makers: Vec = Vec::new(); + let mut expected_filled: Vec = Vec::new(); + let mut first_error: Option = None; + + for spec in specs { + let remaining = aggregate.remaining_quantity().as_u64(); + let mut source = + MatchResult::new(Id::from_u64(TAKER), Quantity::new(remaining)); + let mut level_makers = Vec::new(); + let mut level_filled = Vec::new(); + let mut left = remaining; + for (quantity, consumed) in spec.fills { + let quantity = quantity.min(left); + if quantity == 0 { + break; + } + let maker = next_maker; + next_maker += 1; + source.add_trade(trade(TAKER, maker, quantity)).expect("add_trade"); + left -= quantity; + level_makers.push(Id::from_u64(maker)); + if consumed { + source.add_filled_order_id(Id::from_u64(maker)).expect("filled id"); + level_filled.push(Id::from_u64(maker)); + } + } + let source_remaining = source.remaining_quantity(); + source.finalize(source_remaining); + if spec.killed && level_makers.is_empty() && remaining > 0 { + source.mark_killed(remaining); + } + let level_error = spec.error.then(|| { + PriceLevelError::capacity_exceeded(CapacityResource::Trades, expected_makers.len() + 1) + }); + if let Some(error) = &level_error { + source.set_error(error.clone()); + } + + let terminal = aggregate.is_failed() + || aggregate.was_killed() + || aggregate.was_rejected(); + let before = json(&aggregate); + let source_before = json(&source); + let absorbed = aggregate.try_absorb(&mut source); + if terminal { + prop_assert!(absorbed.is_err()); + prop_assert_eq!(json(&aggregate), before); + prop_assert_eq!(json(&source), source_before); + continue; + } + prop_assert!(absorbed.is_ok()); + expected_makers.extend(level_makers); + expected_filled.extend(level_filled); + if first_error.is_none() { + first_error = level_error; + } + + let remaining = aggregate.remaining_quantity().as_u64(); + let executed = aggregate.executed_quantity().expect("sum").as_u64(); + prop_assert_eq!(aggregate.is_complete(), remaining == 0); + prop_assert_eq!(executed + remaining, initial); + prop_assert_eq!(maker_ids(&aggregate), expected_makers.clone()); + prop_assert_eq!(aggregate.filled_order_ids(), expected_filled.as_slice()); + prop_assert_eq!(aggregate.error(), first_error.as_ref()); + let consistent = match aggregate.outcome() { + MatchOutcome::Filled => remaining == 0, + MatchOutcome::PartiallyFilled => { + remaining > 0 && !aggregate.trades().is_empty() + } + MatchOutcome::NotFilled | MatchOutcome::Killed | MatchOutcome::Rejected => { + remaining > 0 && aggregate.trades().is_empty() + } + }; + prop_assert!(consistent, "{:?}", aggregate.outcome()); + let decoded: MatchResult = + serde_json::from_str(&json(&aggregate)).expect("validated decode"); + prop_assert_eq!(json(&decoded), json(&aggregate)); + // The drained source is a consistent empty result too. + prop_assert!(source.trades().is_empty()); + prop_assert!(source.filled_order_ids().is_empty()); + let _: MatchResult = + serde_json::from_str(&json(&source)).expect("drained source decodes"); + } + } + } +} diff --git a/src/execution/tests/mod.rs b/src/execution/tests/mod.rs index 70f22e9..4407796 100644 --- a/src/execution/tests/mod.rs +++ b/src/execution/tests/mod.rs @@ -1,3 +1,4 @@ +mod absorb; mod analytics; mod capacity; mod filled_ids; From 02765643db85921778acc12887513076eefac08a Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 29 Sep 2026 11:40:08 +0200 Subject: [PATCH 3/7] fix(execution): try_absorb review fixes (#219) - Adopt a level's Killed / Rejected only when the aggregate then holds no trades AND no filled ids; otherwise NotFilled. - Per vector merge plan (utils::alloc::plan_merge / merge_planned): append into the aggregate's spare capacity, else adopt the level's buffer when it holds both sets of entries (aggregate entries rotated to the front in place, no allocation), else grow. All fallible work stays before the infallible commit. - Refusal messages are built inside #[cold] constructors. - Doc example keeps an unabsorbed level result on Err and uses ?; document the caller reservation pattern and that a killed / rejected level after earlier trades leaves the aggregate PartiallyFilled and open. - Tests renamed to test___; new tests for the adopt path with a non-empty aggregate, a kill into ids-without-trades and a kill after trades; proptest covers caller-added ids, reservations and level capacities. Allocation pass, allocs/op (2 trades + 1 filled id per level): 1 level: copy 2.00, absorb 0, absorb with spare level buffers 0 3 levels: copy 3.00, absorb 3.00 (exact level buffers), absorb 0 (spare level buffers), absorb+reserve 3.00 --- CHANGELOG.md | 25 ++-- benches/latency/alloc_measurements.rs | 67 +++++++-- src/execution/list.rs | 27 +++- src/execution/match_result.rs | 206 +++++++++++++++++--------- src/execution/tests/absorb.rs | 189 ++++++++++++++++++++--- src/execution/tests/capacity.rs | 4 +- src/utils/alloc.rs | 62 ++++++++ 7 files changed, 453 insertions(+), 127 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a66825a..180dd66 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,18 +25,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **`MatchResult::try_absorb` (#219).** Folds one price level's result into a multi-level aggregate for the same taker. Trades and filled ids are appended in order; remaining quantity, completion and outcome are - recomputed (a level's `Killed` / `Rejected` is adopted only when nothing - traded); the level's error becomes the aggregate's. The level must have - been matched with the aggregate's remaining quantity, and a terminal - aggregate (error, killed, rejected) refuses further levels. An empty, - too-small aggregate vector takes the level's buffer by swap, so absorbing - the first level allocates nothing; otherwise the entries are appended into - the existing reservation. On any refusal (`InvalidOperation`, + recomputed (a level's `Killed` / `Rejected` is adopted only when the + aggregate then holds no trades and no filled ids); the level's error + becomes the aggregate's. The level must have been matched with the + aggregate's remaining quantity, and a failed, killed or rejected aggregate + refuses further levels (a killed level after earlier trades leaves the + aggregate `PartiallyFilled`; stopping the sweep is the caller's decision). + Per vector, the entries go into the aggregate's spare capacity, else into + the level's buffer when it has room for both (adopted without + allocating), else the aggregate grows. On any refusal (`InvalidOperation`, `CapacityExceeded`) the aggregate is unchanged and the level result still holds its committed trades. Allocation pass (`PL_LATENCY_ONLY=alloc make - bench-latency`): one level of two trades and one filled id, 2.00 allocs/op - (704 B) with reserve-and-copy against 0 with `try_absorb`; three such - levels, 3.00 allocs/op (1,856 B) for both. + bench-latency`, two trades and one filled id per level): one level, 2.00 + allocs/op (704 B) with reserve-and-copy against 0 with `try_absorb`; + three levels, 3.00 allocs/op (1,856 B) for reserve-and-copy and for + `try_absorb` over exactly sized level buffers, and 0 when the level + buffers have the spare capacity `match_order` gives them (ten resting + orders). ### Fixed diff --git a/benches/latency/alloc_measurements.rs b/benches/latency/alloc_measurements.rs index 4efd8ae..7d37a5c 100644 --- a/benches/latency/alloc_measurements.rs +++ b/benches/latency/alloc_measurements.rs @@ -79,10 +79,13 @@ pub fn run_all(config: &Config) -> Vec { measure_checksum_validate(config), measure_restore(config), // Issue #219: folding per-level results into a multi-level aggregate. - measure_aggregate(config, 1, Aggregation::Copy), - measure_aggregate(config, 1, Aggregation::Absorb), - measure_aggregate(config, 3, Aggregation::Copy), - measure_aggregate(config, 3, Aggregation::Absorb), + measure_aggregate(config, 1, Aggregation::Copy, EXACT), + measure_aggregate(config, 1, Aggregation::Absorb, EXACT), + measure_aggregate(config, 1, Aggregation::Absorb, SPARE), + measure_aggregate(config, 3, Aggregation::Copy, EXACT), + measure_aggregate(config, 3, Aggregation::Absorb, EXACT), + measure_aggregate(config, 3, Aggregation::Absorb, SPARE), + measure_aggregate(config, 3, Aggregation::AbsorbReserved, EXACT), ] } @@ -92,20 +95,38 @@ enum Aggregation { /// The pre-#219 pattern: reserve the aggregate for the level's entries, /// then copy them in with `add_trade` / `add_filled_order_id`. Copy, - /// `MatchResult::try_absorb`: move the level's buffers when the - /// aggregate's are empty, append otherwise. + /// `MatchResult::try_absorb` alone: append into spare capacity, adopt a + /// roomy level buffer, or grow. Absorb, + /// The documented caller pattern: absorb the first level with no + /// reservation, and reserve each later level's entry counts with + /// `try_reserve_trades` / `try_reserve_filled_order_ids` before + /// absorbing it. + AbsorbReserved, } +/// Level result buffers sized exactly for their entries. +const EXACT: usize = 2; +/// Level result buffers with spare capacity, as `match_order` sizes its own +/// result (`min(incoming quantity, resting order count)`, here a level of +/// ten resting orders). +const SPARE: usize = 10; + /// Measures folding `levels` per-level results (two trades and one filled id /// each, sized like `match_order` sizes its own result) into a fresh -/// aggregate (issue #219). The level results are built before counting +/// aggregate (issue #219), each level result created with +/// `level_capacity`. The level results are built before counting /// starts; the aggregates are kept and dropped after it stops, so only the /// folding itself is counted. `reps` is the number of aggregations. The /// `Copy` pattern reserves exactly each level's entry counts, the cheapest /// form of that pattern (a caller reserving each level's worst case before /// matching allocates at least as often). -fn measure_aggregate(config: &Config, levels: usize, aggregation: Aggregation) -> AllocReport { +fn measure_aggregate( + config: &Config, + levels: usize, + aggregation: Aggregation, + level_capacity: usize, +) -> AllocReport { const TRADES_PER_LEVEL: u64 = 2; const FILL: u64 = 5; let reps = config.alloc_reps; @@ -119,7 +140,7 @@ fn measure_aggregate(config: &Config, levels: usize, aggregation: Aggregation) - let mut result = MatchResult::try_with_capacity( taker, Quantity::new(incoming), - TRADES_PER_LEVEL as usize, + level_capacity, ) .expect("alloc measurement: level result capacity"); for step in 0..TRADES_PER_LEVEL { @@ -151,7 +172,7 @@ fn measure_aggregate(config: &Config, levels: usize, aggregation: Aggregation) - let before = AllocStats::read(); for level_results in &mut inputs { let mut aggregate = MatchResult::new(taker, Quantity::new(total)); - for result in level_results.iter_mut() { + for (index, result) in level_results.iter_mut().enumerate() { match aggregation { Aggregation::Copy => { aggregate @@ -174,6 +195,19 @@ fn measure_aggregate(config: &Config, levels: usize, aggregation: Aggregation) - Aggregation::Absorb => aggregate .try_absorb(result) .expect("alloc measurement: try_absorb"), + Aggregation::AbsorbReserved => { + if index > 0 { + aggregate + .try_reserve_trades(result.trades().len()) + .expect("alloc measurement: reserve trades"); + aggregate + .try_reserve_filled_order_ids(result.filled_order_ids().len()) + .expect("alloc measurement: reserve filled ids"); + } + aggregate + .try_absorb(result) + .expect("alloc measurement: try_absorb"); + } } } outputs.push(aggregate); @@ -189,11 +223,14 @@ fn measure_aggregate(config: &Config, levels: usize, aggregation: Aggregation) - drop(inputs); AllocReport { - name: match (levels, aggregation) { - (1, Aggregation::Copy) => "aggregate_1_copy", - (1, Aggregation::Absorb) => "aggregate_1_absorb", - (_, Aggregation::Copy) => "aggregate_3_copy", - (_, Aggregation::Absorb) => "aggregate_3_absorb", + name: match (levels, aggregation, level_capacity) { + (1, Aggregation::Copy, _) => "aggregate_1_copy", + (1, Aggregation::Absorb, EXACT) => "aggregate_1_absorb", + (1, _, _) => "aggregate_1_absorb_spare", + (_, Aggregation::Copy, _) => "aggregate_3_copy", + (_, Aggregation::Absorb, EXACT) => "aggregate_3_absorb", + (_, Aggregation::Absorb, _) => "aggregate_3_absorb_spare", + (_, Aggregation::AbsorbReserved, _) => "aggregate_3_absorb_rsv", }, reps, totals: after.since(before), diff --git a/src/execution/list.rs b/src/execution/list.rs index 4c37748..f8a6f6b 100644 --- a/src/execution/list.rs +++ b/src/execution/list.rs @@ -1,5 +1,6 @@ use crate::errors::{CapacityResource, PriceLevelError}; use crate::execution::trade::Trade; +use crate::utils::alloc::{MergePlan, merge_planned, plan_merge}; use crate::utils::text::{ MAX_TEXT_NESTING_DEPTH, MAX_TEXT_NESTING_DEPTH_INSIDE_LIST, NestingError, TopLevelSplit, try_push, @@ -112,14 +113,26 @@ impl TradeList { self.trades.push(trade); } - /// Moves every trade of `other` to the end of this list, leaving `other` - /// empty (its capacity is kept). The caller must already have reserved - /// room for `other.len()` more trades (via [`Self::try_reserve`]): with - /// that spare capacity `Vec::append` never reallocates, so this cannot - /// fail or panic (issue #219). + /// The allocation-free way to put `other`'s trades behind this list's, + /// if any (issue #219; see [`plan_merge`]). #[inline] - pub(crate) fn append_reserved(&mut self, other: &mut TradeList) { - self.trades.append(&mut other.trades); + #[must_use] + pub(crate) fn plan_merge(&self, other: &TradeList) -> Option { + plan_merge( + self.trades.len(), + self.trades.capacity(), + other.trades.len(), + other.trades.capacity(), + ) + } + + /// Moves every trade of `other` behind this list's, leaving `other` + /// empty, per a `plan` from [`Self::plan_merge`] (or + /// [`MergePlan::Append`] after [`Self::try_reserve`] of `other.len()`). + /// Never allocates, so it cannot fail or panic (issue #219). + #[inline] + pub(crate) fn merge_planned(&mut self, other: &mut TradeList, plan: MergePlan) { + merge_planned(&mut self.trades, &mut other.trades, plan); } /// Clones the list without an infallible allocation. diff --git a/src/execution/match_result.rs b/src/execution/match_result.rs index d458f9e..da704bf 100644 --- a/src/execution/match_result.rs +++ b/src/execution/match_result.rs @@ -3,6 +3,7 @@ use crate::execution::list::TradeList; use crate::execution::trade::Trade; use crate::orders::Id; use crate::utils::Quantity; +use crate::utils::alloc::{MergePlan, merge_planned, plan_merge}; use crate::utils::dedup::first_repeat_position; use crate::utils::text::{MAX_TEXT_NESTING_DEPTH, NestingError, matching_close, try_push}; use serde::{Deserialize, Serialize}; @@ -362,8 +363,8 @@ impl MatchResult { } /// Folds a later price level's result for the same taker into this - /// aggregate (issue #219), moving the level's buffers instead of copying - /// them when that is lossless. + /// aggregate (issue #219), moving buffers instead of allocating whenever + /// that is lossless. /// /// A caller that sweeps several levels matches each one with this /// aggregate's current [`Self::remaining_quantity`] and absorbs the @@ -376,42 +377,56 @@ impl MatchResult { /// the aggregate's executed quantity is the sum of both. /// - **Outcome and completion** are recomputed from the combined fields: /// `Filled` when nothing remains, `NotFilled` without trades, otherwise - /// `PartiallyFilled`. When the combined result has no trades and the - /// level was killed or rejected, that outcome is adopted. + /// `PartiallyFilled`. A level's `Killed` / `Rejected` is adopted only + /// when the combined result has no trades and no filled ids. /// - **Error:** the level's error, if any, becomes the aggregate's. The /// aggregate can hold no error of its own here (see below), so the /// first failure always wins. /// + /// A failed, killed or rejected aggregate is terminal and refuses every + /// later level. A killed or rejected *level* makes the aggregate terminal + /// only when nothing traded before it; after earlier trades the aggregate + /// is `PartiallyFilled` and still accepts further levels. Whether to stop + /// sweeping after a killed or rejected level is the caller's decision + /// (check the level result's [`Self::outcome`] before absorbing it); + /// `try_absorb` does not enforce it. + /// /// # Allocation /// - /// Per vector: when this aggregate's vector is empty and too small for - /// the level's contents, the two vectors are swapped, so the aggregate - /// takes the level's buffer and the level gets the aggregate's (empty) - /// reservation back; nothing is allocated or copied. Otherwise the - /// entries are appended into this aggregate's buffer, which allocates - /// only if its spare capacity is too small (amortized growth). Existing - /// entries and a sufficient reservation are always kept. Absorbing the - /// first level into a fresh [`Self::new`] aggregate therefore never - /// allocates. + /// Per vector, the first applicable of: append into this aggregate's + /// spare capacity (a caller's reservation is used, never given up); + /// adopt the level's buffer when its capacity holds both sets of entries + /// (this aggregate's entries are moved to its front in place, and the + /// level gets this aggregate's emptied buffer); otherwise grow this + /// aggregate's buffer and append (amortized growth, the only case that + /// allocates). Absorbing the first level into a fresh [`Self::new`] + /// aggregate therefore never allocates. + /// + /// Recommended pattern: absorb the first level without reserving + /// anything (the zero-allocation adopt). Only if another level must be + /// matched, reserve this aggregate for that level's worst case with + /// [`Self::try_reserve_trades`] / [`Self::try_reserve_filled_order_ids`] + /// *before* matching it, so absorbing its committed trades cannot then + /// fail on growth. /// /// # Failure ownership /// /// Every check and reservation happens before anything moves. On `Err` /// no observable field of `self` has changed (only a vector's spare /// capacity may have grown) and `level` is untouched, so the committed - /// trades it records are still in the caller's hands. On `Ok` `level` is - /// left as an empty result for its remaining quantity: no trades, no - /// filled ids, no error, outcome `Filled` or `NotFilled`. + /// trades it records are still in the caller's hands: keep it (see the + /// example). On `Ok` `level` is left as an empty result for its + /// remaining quantity: no trades, no filled ids, no error, outcome + /// `Filled` or `NotFilled`. /// /// # Errors /// /// Returns [`PriceLevelError::InvalidOperation`] if `level` belongs to a - /// different taker order id; if this aggregate is already terminal (it - /// carries an error, or it was killed or rejected), since no later level - /// may be matched after either; if the level's executed quantity overflows - /// `u64`; or if the level's executed plus remaining quantity differs from - /// this aggregate's remaining quantity (the level was not matched with - /// what the aggregate had left). Returns + /// different taker order id; if this aggregate is terminal (it carries an + /// error, or it was killed or rejected); if the level's executed quantity + /// overflows `u64`; or if the level's executed plus remaining quantity + /// differs from this aggregate's remaining quantity (the level was not + /// matched with what the aggregate had left). Returns /// [`PriceLevelError::CapacityExceeded`] (resource /// [`CapacityResource::Trades`] or [`CapacityResource::FilledOrderIds`]) if /// a vector cannot grow. @@ -419,21 +434,35 @@ impl MatchResult { /// # Examples /// /// ```rust - /// use pricelevel::{MatchResult, PriceLevelError, Quantity}; - /// # use pricelevel::{Id, PriceLevel, TakerKind, TimeInForce, TimestampMs, UuidGenerator}; + /// use pricelevel::{ + /// Id, MatchResult, PriceLevel, PriceLevelError, Quantity, TakerKind, TimeInForce, + /// TimestampMs, UuidGenerator, + /// }; /// + /// /// Sweeps `levels` in price order. A level result that could not be + /// /// absorbed is pushed to `unabsorbed`, so its committed trades are + /// /// never dropped. /// fn sweep( /// levels: &[PriceLevel], /// taker: Id, /// quantity: u64, /// ids: &UuidGenerator, + /// unabsorbed: &mut Vec, /// ) -> Result { /// let mut aggregate = MatchResult::new(taker, Quantity::new(quantity)); - /// for level in levels { + /// for (index, level) in levels.iter().enumerate() { /// let remaining = aggregate.remaining_quantity().as_u64(); /// if remaining == 0 { /// break; /// } + /// if index > 0 { + /// // Later levels: reserve the worst case before matching. + /// let bound = level + /// .order_count() + /// .min(usize::try_from(remaining).unwrap_or(usize::MAX)); + /// aggregate.try_reserve_trades(bound)?; + /// aggregate.try_reserve_filled_order_ids(bound)?; + /// } /// let mut result = level.match_order( /// remaining, /// taker, @@ -442,72 +471,80 @@ impl MatchResult { /// TimestampMs::new(1_700_000_000_000), /// ids, /// ); - /// // On `Err`, `result` still holds the level's committed trades. - /// aggregate.try_absorb(&mut result)?; + /// if let Err(error) = aggregate.try_absorb(&mut result) { + /// unabsorbed.push(result); + /// return Err(error); + /// } /// if aggregate.is_failed() { /// break; /// } /// } /// Ok(aggregate) /// } - /// # let ids = UuidGenerator::new(uuid::Uuid::nil()); - /// # let aggregate = sweep(&[PriceLevel::new(100)], Id::from_u64(1), 5, &ids).unwrap(); - /// # assert_eq!(aggregate.remaining_quantity().as_u64(), 5); + /// + /// let ids = UuidGenerator::new(uuid::Uuid::nil()); + /// let mut unabsorbed = Vec::new(); + /// let aggregate = sweep(&[PriceLevel::new(100)], Id::from_u64(1), 5, &ids, &mut unabsorbed)?; + /// assert_eq!(aggregate.remaining_quantity().as_u64(), 5); + /// assert!(unabsorbed.is_empty()); + /// # Ok::<(), PriceLevelError>(()) /// ``` pub fn try_absorb(&mut self, level: &mut MatchResult) -> Result<(), PriceLevelError> { // 1. Validate: nothing changes on a refusal. if level.order_id != self.order_id { - return Err(absorb_refused(format!( - "absorbed result's taker order id {} does not match the aggregate's {}", - level.order_id, self.order_id - ))); + return Err(absorb_taker_mismatch(level.order_id, self.order_id)); } if self.error.is_some() || self.outcome.was_killed() || self.outcome.was_rejected() { - return Err(absorb_refused(format!( - "the aggregate is terminal ({:?}, error recorded: {}); no later level may be absorbed", - self.outcome, - self.error.is_some() - ))); + return Err(absorb_terminal(self.outcome, self.error.is_some())); } let level_executed = level.executed_quantity()?.as_u64(); if level_executed.checked_add(level.remaining_quantity) != Some(self.remaining_quantity) { - return Err(absorb_refused(format!( - "absorbed result executed {level_executed} with {} remaining, but the aggregate has {} remaining", - level.remaining_quantity, self.remaining_quantity - ))); + return Err(absorb_quantity_mismatch( + level_executed, + level.remaining_quantity, + self.remaining_quantity, + )); } // 2. Plan and reserve: still nothing observable changes. - let swap_trades = self.trades.is_empty() && self.trades.capacity() < level.trades.len(); - let swap_filled = self.filled_order_ids.is_empty() - && self.filled_order_ids.capacity() < level.filled_order_ids.len(); - if !swap_trades { - self.trades.try_reserve(level.trades.len())?; - } - if !swap_filled { - self.try_reserve_filled_order_ids(level.filled_order_ids.len())?; - } + let trades_plan = match self.trades.plan_merge(&level.trades) { + Some(plan) => plan, + None => { + self.trades.try_reserve(level.trades.len())?; + MergePlan::Append + } + }; + let filled_plan = match plan_merge( + self.filled_order_ids.len(), + self.filled_order_ids.capacity(), + level.filled_order_ids.len(), + level.filled_order_ids.capacity(), + ) { + Some(plan) => plan, + None => { + self.try_reserve_filled_order_ids(level.filled_order_ids.len())?; + MergePlan::Append + } + }; - // 3. Commit: nothing below can fail. - if swap_trades { - std::mem::swap(&mut self.trades, &mut level.trades); - } else { - self.trades.append_reserved(&mut level.trades); - } - if swap_filled { - std::mem::swap(&mut self.filled_order_ids, &mut level.filled_order_ids); - } else { - // Reserved above: `append` cannot reallocate. - self.filled_order_ids.append(&mut level.filled_order_ids); - } + // 3. Commit: nothing below can fail or allocate. + self.trades.merge_planned(&mut level.trades, trades_plan); + merge_planned( + &mut self.filled_order_ids, + &mut level.filled_order_ids, + filled_plan, + ); self.remaining_quantity = level.remaining_quantity; self.is_complete = self.remaining_quantity == 0; self.outcome = if self.is_complete { MatchOutcome::Filled } else if self.trades.is_empty() { // Nothing traded anywhere: a level's kill / rejection is the - // whole story (remaining > 0, no trades, no filled ids). - if level.outcome.was_killed() || level.outcome.was_rejected() { + // whole story, unless the aggregate carries filled ids a caller + // added without trades (a killed / rejected result has none). + if self.filled_order_ids.is_empty() + && (level.outcome.was_killed() || level.outcome.was_rejected()) + { level.outcome } else { MatchOutcome::NotFilled @@ -1093,10 +1130,41 @@ fn executed_quantity_overflow() -> PriceLevelError { } } -/// The typed refusal of [`MatchResult::try_absorb`] (issue #219). +/// [`MatchResult::try_absorb`] refusal: the level belongs to another taker +/// (issue #219). +#[cold] +fn absorb_taker_mismatch(level: Id, aggregate: Id) -> PriceLevelError { + PriceLevelError::InvalidOperation { + message: format!( + "absorbed result's taker order id {level} does not match the aggregate's {aggregate}" + ), + } +} + +/// [`MatchResult::try_absorb`] refusal: the aggregate is terminal (issue +/// #219). #[cold] -fn absorb_refused(message: String) -> PriceLevelError { - PriceLevelError::InvalidOperation { message } +fn absorb_terminal(outcome: MatchOutcome, has_error: bool) -> PriceLevelError { + PriceLevelError::InvalidOperation { + message: format!( + "the aggregate is terminal ({outcome:?}, error recorded: {has_error}); no later level may be absorbed" + ), + } +} + +/// [`MatchResult::try_absorb`] refusal: the level was not matched with the +/// aggregate's remaining quantity (issue #219). +#[cold] +fn absorb_quantity_mismatch( + level_executed: u64, + level_remaining: u64, + aggregate_remaining: u64, +) -> PriceLevelError { + PriceLevelError::InvalidOperation { + message: format!( + "absorbed result executed {level_executed} with {level_remaining} remaining, but the aggregate has {aggregate_remaining} remaining" + ), + } } impl fmt::Display for MatchResult { diff --git a/src/execution/tests/absorb.rs b/src/execution/tests/absorb.rs index 5aa6061..6d86e36 100644 --- a/src/execution/tests/absorb.rs +++ b/src/execution/tests/absorb.rs @@ -54,6 +54,31 @@ mod tests { result } + /// As [`level`], with the trade buffer sized exactly for its fills (no + /// spare capacity). + fn exact_level(incoming: u64, fills: &[(u64, u64, bool)]) -> MatchResult { + let mut result = MatchResult::try_with_capacity( + Id::from_u64(TAKER), + Quantity::new(incoming), + fills.len(), + ) + .expect("capacity"); + for &(maker, quantity, consumed) in fills { + result + .add_trade(trade(TAKER, maker, quantity)) + .expect("add_trade"); + if consumed { + result + .add_filled_order_id(Id::from_u64(maker)) + .expect("filled id"); + } + } + assert_eq!(result.trades().capacity(), fills.len()); + let remaining = result.remaining_quantity(); + result.finalize(remaining); + result + } + fn json(result: &MatchResult) -> String { serde_json::to_string(result).expect("serialize") } @@ -87,7 +112,7 @@ mod tests { } #[test] - fn taker_mismatch_is_refused() { + fn test_try_absorb_taker_mismatch_refused_unchanged() { let mut aggregate = MatchResult::new(Id::from_u64(TAKER), Quantity::new(10)); let mut other = MatchResult::new(Id::from_u64(TAKER + 1), Quantity::new(10)); other.add_trade(trade(TAKER + 1, 1, 4)).expect("add_trade"); @@ -96,7 +121,7 @@ mod tests { } #[test] - fn quantity_mismatch_is_refused() { + fn test_try_absorb_quantity_mismatch_refused_unchanged() { // The level was matched with more, and with less, than the aggregate // has left. for incoming in [9, 11] { @@ -108,7 +133,7 @@ mod tests { } #[test] - fn terminal_aggregate_is_refused() { + fn test_try_absorb_terminal_aggregate_refused_unchanged() { let mut failed = level(20, &[(1, 5, true)]); failed.set_error(PriceLevelError::counter_exhausted( ExhaustedCounter::QueueSequence, @@ -127,7 +152,7 @@ mod tests { } #[test] - fn empty_aggregate_takes_the_level_buffers_by_move() { + fn test_try_absorb_empty_aggregate_adopts_level_buffers() { let mut aggregate = MatchResult::new(Id::from_u64(TAKER), Quantity::new(20)); let mut source = MatchResult::try_with_capacity(Id::from_u64(TAKER), Quantity::new(20), 4).expect("cap"); @@ -168,7 +193,7 @@ mod tests { } #[test] - fn sufficient_reservation_is_kept_and_appended_into() { + fn test_try_absorb_sufficient_reservation_appended_into() { let mut aggregate = MatchResult::new(Id::from_u64(TAKER), Quantity::new(20)); aggregate.try_reserve(8).expect("reserve"); let trades_ptr = aggregate.trades().as_vec().as_ptr(); @@ -193,7 +218,7 @@ mod tests { } #[test] - fn existing_filled_ids_are_not_lost() { + fn test_try_absorb_existing_filled_ids_kept() { // The aggregate holds a filled id but no trades: its trade vector is // moved in, its id vector is appended to. let mut aggregate = MatchResult::new(Id::from_u64(TAKER), Quantity::new(20)); @@ -213,11 +238,13 @@ mod tests { } #[test] - fn append_path_keeps_order_and_refuses_growth_atomically() { - let mut aggregate = level(20, &[(1, 5, true)]); - let mut source = level(15, &[(2, 3, true), (3, 2, false)]); + fn test_try_absorb_growth_refusal_unchanged_then_appends_in_order() { + // Both buffers are exactly full, so neither the append nor the adopt + // plan fits and the aggregate must grow. + let mut aggregate = exact_level(20, &[(1, 5, true)]); + let mut source = exact_level(15, &[(2, 3, true), (3, 2, false)]); - // A capped trade list refuses the append: nothing changes. + // A capped trade list refuses the growth: nothing changes. { let _limit = trade_list_seam::limit_trades(2); let err = assert_refused_unchanged(&mut aggregate, &mut source); @@ -247,7 +274,7 @@ mod tests { } #[test] - fn filling_level_completes_the_aggregate() { + fn test_try_absorb_filling_level_completes_aggregate() { let mut aggregate = level(10, &[(1, 4, true)]); let mut source = level(6, &[(2, 6, true)]); aggregate.try_absorb(&mut source).expect("absorb"); @@ -259,7 +286,7 @@ mod tests { } #[test] - fn first_error_wins() { + fn test_try_absorb_level_error_first_error_wins() { let first = PriceLevelError::counter_exhausted(ExhaustedCounter::QueueSequence); let mut aggregate = MatchResult::new(Id::from_u64(TAKER), Quantity::new(20)); let mut source = level(20, &[(1, 5, true)]); @@ -281,7 +308,7 @@ mod tests { } #[test] - fn kill_and_reject_adopted_only_without_trades() { + fn test_try_absorb_kill_or_reject_adopted_only_without_trades() { for reject in [false, true] { let mark = |result: &mut MatchResult, quantity: u64| { if reject { @@ -317,7 +344,96 @@ mod tests { } #[test] - fn zero_quantity_aggregate_absorbs_a_vacuous_level() { + fn test_try_absorb_nonempty_aggregate_adopts_roomy_level_buffers() { + // The aggregate's buffers are full; the level's have room for both + // sets of entries, so the aggregate's entries move to the front of + // the level's buffers and nothing is allocated. + let mut aggregate = + MatchResult::try_with_capacity(Id::from_u64(TAKER), Quantity::new(20), 1).expect("cap"); + aggregate.add_trade(trade(TAKER, 1, 5)).expect("t1"); + aggregate.add_filled_order_id(Id::from_u64(1)).expect("f1"); + let mut source = + MatchResult::try_with_capacity(Id::from_u64(TAKER), Quantity::new(15), 8).expect("cap"); + source.add_trade(trade(TAKER, 2, 3)).expect("t2"); + source.add_filled_order_id(Id::from_u64(2)).expect("f2"); + source.add_trade(trade(TAKER, 3, 2)).expect("t3"); + let trades_ptr = source.trades().as_vec().as_ptr(); + let filled_ptr = source.filled_order_ids().as_ptr(); + let aggregate_trades_ptr = aggregate.trades().as_vec().as_ptr(); + + aggregate.try_absorb(&mut source).expect("absorb"); + + assert_eq!(aggregate.trades().as_vec().as_ptr(), trades_ptr); + assert_eq!(aggregate.filled_order_ids().as_ptr(), filled_ptr); + assert_eq!(aggregate.trades().capacity(), 8); + assert_eq!( + maker_ids(&aggregate), + vec![Id::from_u64(1), Id::from_u64(2), Id::from_u64(3)] + ); + assert_eq!( + aggregate.filled_order_ids(), + &[Id::from_u64(1), Id::from_u64(2)] + ); + assert_eq!(aggregate.remaining_quantity().as_u64(), 10); + // The level got the aggregate's emptied buffer back. + assert_eq!(source.trades().as_vec().as_ptr(), aggregate_trades_ptr); + assert!(source.trades().is_empty()); + assert_round_trips(&aggregate); + assert_round_trips(&source); + } + + #[test] + fn test_try_absorb_killed_level_into_ids_without_trades_not_filled() { + // `add_filled_order_id` can leave ids without trades; a kill must + // not be adopted then (Killed carries no filled ids). + let mut aggregate = MatchResult::new(Id::from_u64(TAKER), Quantity::new(20)); + aggregate + .add_filled_order_id(Id::from_u64(7)) + .expect("filled id"); + let mut source = MatchResult::new(Id::from_u64(TAKER), Quantity::new(20)); + source.mark_killed(20); + + aggregate.try_absorb(&mut source).expect("absorb"); + + assert_eq!(aggregate.outcome(), MatchOutcome::NotFilled); + assert!(!aggregate.was_killed()); + assert_eq!(aggregate.filled_order_ids(), &[Id::from_u64(7)]); + // The outcome agrees with the fields; the decoder still rejects the + // payload, but only because the caller-added id is backed by no + // trade (the same rejection the aggregate drew before absorbing). + let err = serde_json::from_str::(&json(&aggregate)) + .expect_err("an id without a trade is not a decodable result"); + let message = err.to_string(); + assert!(message.contains("not an in-order maker"), "{message}"); + assert!(!message.contains("outcome contradicts"), "{message}"); + } + + #[test] + fn test_try_absorb_killed_level_after_trades_partially_filled_and_open() { + // A kill after earlier trades does not make the aggregate terminal: + // it stays PartiallyFilled and accepts a further level. Stopping the + // sweep is the caller's decision. + let mut aggregate = level(20, &[(1, 5, true)]); + let mut killed = MatchResult::new(Id::from_u64(TAKER), Quantity::new(15)); + killed.mark_killed(15); + + aggregate + .try_absorb(&mut killed) + .expect("absorb killed level"); + assert_eq!(aggregate.outcome(), MatchOutcome::PartiallyFilled); + assert!(!aggregate.was_killed()); + + let mut next = level(15, &[(2, 15, true)]); + aggregate + .try_absorb(&mut next) + .expect("a later level is accepted"); + assert!(aggregate.is_complete()); + assert_eq!(aggregate.outcome(), MatchOutcome::Filled); + assert_round_trips(&aggregate); + } + + #[test] + fn test_try_absorb_zero_quantity_vacuously_filled() { let mut aggregate = MatchResult::new(Id::from_u64(TAKER), Quantity::new(0)); let mut source = MatchResult::new(Id::from_u64(TAKER), Quantity::new(0)); aggregate.try_absorb(&mut source).expect("absorb"); @@ -333,6 +449,9 @@ mod tests { fills: Vec<(u64, bool)>, killed: bool, error: bool, + /// Capacity the level result is created with (exercises the append, + /// adopt and grow plans). + capacity: usize, } fn level_spec() -> impl Strategy { @@ -340,11 +459,13 @@ mod tests { prop::collection::vec((1u64..=8, any::()), 0..4), any::(), prop::bool::weighted(0.15), + 0usize..10, ) - .prop_map(|(fills, killed, error)| LevelSpec { + .prop_map(|(fills, killed, error, capacity)| LevelSpec { fills, killed, error, + capacity, }) } @@ -352,20 +473,33 @@ mod tests { #![proptest_config(ProptestConfig { cases: 512, ..ProptestConfig::default() })] #[test] - fn absorb_keeps_the_aggregate_invariants( + fn test_try_absorb_level_sequences_keep_invariants( initial in 0u64..=40, specs in prop::collection::vec(level_spec(), 0..6), + // Filled ids the caller added to the destination without trades. + prefix_ids in 0u64..3, + reservation in 0usize..6, ) { let mut aggregate = MatchResult::new(Id::from_u64(TAKER), Quantity::new(initial)); + aggregate.try_reserve_trades(reservation).expect("reserve"); + let mut expected_filled: Vec = Vec::new(); + for id in 0..prefix_ids { + let id = Id::from_u64(900 + id); + aggregate.add_filled_order_id(id).expect("prefix id"); + expected_filled.push(id); + } let mut next_maker = 1u64; let mut expected_makers: Vec = Vec::new(); - let mut expected_filled: Vec = Vec::new(); let mut first_error: Option = None; for spec in specs { let remaining = aggregate.remaining_quantity().as_u64(); - let mut source = - MatchResult::new(Id::from_u64(TAKER), Quantity::new(remaining)); + let mut source = MatchResult::try_with_capacity( + Id::from_u64(TAKER), + Quantity::new(remaining), + spec.capacity, + ) + .expect("level capacity"); let mut level_makers = Vec::new(); let mut level_filled = Vec::new(); let mut left = remaining; @@ -427,14 +561,21 @@ mod tests { MatchOutcome::PartiallyFilled => { remaining > 0 && !aggregate.trades().is_empty() } - MatchOutcome::NotFilled | MatchOutcome::Killed | MatchOutcome::Rejected => { - remaining > 0 && aggregate.trades().is_empty() + MatchOutcome::NotFilled => remaining > 0 && aggregate.trades().is_empty(), + MatchOutcome::Killed | MatchOutcome::Rejected => { + remaining > 0 + && aggregate.trades().is_empty() + && aggregate.filled_order_ids().is_empty() } }; prop_assert!(consistent, "{:?}", aggregate.outcome()); - let decoded: MatchResult = - serde_json::from_str(&json(&aggregate)).expect("validated decode"); - prop_assert_eq!(json(&decoded), json(&aggregate)); + // Caller-added ids without trades are outside the decodable + // domain from the start; otherwise the aggregate round-trips. + if prefix_ids == 0 { + let decoded: MatchResult = + serde_json::from_str(&json(&aggregate)).expect("validated decode"); + prop_assert_eq!(json(&decoded), json(&aggregate)); + } // The drained source is a consistent empty result too. prop_assert!(source.trades().is_empty()); prop_assert!(source.filled_order_ids().is_empty()); diff --git a/src/execution/tests/capacity.rs b/src/execution/tests/capacity.rs index f0d4011..fcd256c 100644 --- a/src/execution/tests/capacity.rs +++ b/src/execution/tests/capacity.rs @@ -106,7 +106,7 @@ mod tests { /// Issue #219: each split reservation sizes only its own vector. #[test] - fn split_reservations_size_one_vector_each() { + fn test_match_result_split_reservations_size_one_vector_each() { let mut result = MatchResult::new(Id::from_u64(10), Quantity::new(20)); result.try_reserve_trades(4).expect("reserve trades"); assert!(result.trades().capacity() >= 4); @@ -136,7 +136,7 @@ mod tests { /// Issue #219: a refused split reservation reports its own resource and /// changes no observable field; spare capacity makes it allocation-free. #[test] - fn split_reservation_failures_change_nothing() { + fn test_match_result_split_reservation_failure_changes_nothing() { let mut result = MatchResult::new(Id::from_u64(10), Quantity::new(20)); result.add_trade(trade(1, 5)).expect("add_trade"); result diff --git a/src/utils/alloc.rs b/src/utils/alloc.rs index ceee1ca..96cfb7d 100644 --- a/src/utils/alloc.rs +++ b/src/utils/alloc.rs @@ -167,6 +167,68 @@ pub(crate) fn try_copy_str( Ok(out) } +/// How [`merge_planned`] moves a source vector's elements behind a +/// destination's without allocating (issue #219). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum MergePlan { + /// Append into the destination's spare capacity; the destination's + /// buffer is kept. + Append, + /// Adopt the source's buffer: its spare capacity holds the destination's + /// elements, which are moved to its front. The source gets the + /// destination's emptied buffer. + Adopt, +} + +/// The allocation-free way to put `src_len` elements behind a destination +/// of `dest_len` elements, if there is one: [`MergePlan::Append`] when the +/// destination's spare capacity fits them (a caller's reservation is used +/// first), else [`MergePlan::Adopt`] when the source's capacity fits both. +/// `None` means the destination must grow first. +#[inline] +#[must_use] +pub(crate) fn plan_merge( + dest_len: usize, + dest_capacity: usize, + src_len: usize, + src_capacity: usize, +) -> Option { + if dest_capacity + .checked_sub(dest_len) + .is_some_and(|spare| spare >= src_len) + { + Some(MergePlan::Append) + } else if dest_len + .checked_add(src_len) + .is_some_and(|total| total <= src_capacity) + { + Some(MergePlan::Adopt) + } else { + None + } +} + +/// Moves every element of `src` behind those of `dest`, per a `plan` that +/// [`plan_merge`] returned for these two vectors (or [`MergePlan::Append`] +/// after reserving `src.len()` on `dest`). Never allocates: `Append` fits +/// the destination's spare capacity, and `Adopt` swaps the buffers, appends +/// the destination's elements into the source buffer's spare capacity and +/// rotates them to the front in place. `src` is left empty. +#[inline] +pub(crate) fn merge_planned(dest: &mut Vec, src: &mut Vec, plan: MergePlan) { + match plan { + MergePlan::Append => dest.append(src), + MergePlan::Adopt => { + let moved = dest.len(); + std::mem::swap(dest, src); + dest.append(src); + // `moved <= dest.len()`: the appended elements are exactly the + // `moved` former destination elements. + dest.rotate_right(moved); + } + } +} + /// An [`std::io::Write`] sink over a `Vec` whose every growth goes /// through [`try_reserve_vec`] (resource /// [`CapacityResource::SerializationBuffer`]). From 81b3e529ef0a056feb53d2e1685e90a4d09190ac Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 29 Sep 2026 12:11:16 +0200 Subject: [PATCH 4/7] address review: build genuinely full buffers for absorb growth fixtures try_with_capacity only guarantees a lower bound, so the growth and adopt tests and the EXACT allocation scenario now fill buffers to their observed capacity instead of asserting an exact size. --- benches/latency/alloc_measurements.rs | 44 ++++++++--- src/execution/tests/absorb.rs | 104 +++++++++++++++++--------- 2 files changed, 105 insertions(+), 43 deletions(-) diff --git a/benches/latency/alloc_measurements.rs b/benches/latency/alloc_measurements.rs index 7d37a5c..eab783d 100644 --- a/benches/latency/alloc_measurements.rs +++ b/benches/latency/alloc_measurements.rs @@ -105,7 +105,10 @@ enum Aggregation { AbsorbReserved, } -/// Level result buffers sized exactly for their entries. +/// Level result buffers requested for exactly their two trades. The +/// allocator may hand back more, so [`measure_aggregate`] then fills the +/// trade buffer to its observed capacity: it is always genuinely full and +/// absorbing it deterministically takes the growth path. const EXACT: usize = 2; /// Level result buffers with spare capacity, as `match_order` sizes its own /// result (`min(incoming quantity, resting order count)`, here a level of @@ -115,7 +118,9 @@ const SPARE: usize = 10; /// Measures folding `levels` per-level results (two trades and one filled id /// each, sized like `match_order` sizes its own result) into a fresh /// aggregate (issue #219), each level result created with -/// `level_capacity`. The level results are built before counting +/// `level_capacity`. An [`EXACT`] level has as many trades as its trade +/// buffer's observed capacity (two unless the allocator over-provides), +/// splitting the level's quantity between them. The level results are built before counting /// starts; the aggregates are kept and dropped after it stops, so only the /// folding itself is counted. `reps` is the number of aggregations. The /// `Copy` pattern reserves exactly each level's entry counts, the cheapest @@ -128,38 +133,59 @@ fn measure_aggregate( level_capacity: usize, ) -> AllocReport { const TRADES_PER_LEVEL: u64 = 2; - const FILL: u64 = 5; + /// Quantity each level executes, split across its trades. + const LEVEL_QTY: u64 = 1_000; let reps = config.alloc_reps; let taker = Id::from_u64(TAKER_ID_BASE); - let total = levels as u64 * TRADES_PER_LEVEL * FILL; + let total = levels as u64 * LEVEL_QTY; let mut inputs: Vec> = (0..reps) .map(|_| { (0..levels as u64) .map(|level| { - let incoming = total - level * TRADES_PER_LEVEL * FILL; + let incoming = total - level * LEVEL_QTY; let mut result = MatchResult::try_with_capacity( taker, Quantity::new(incoming), level_capacity, ) .expect("alloc measurement: level result capacity"); - for step in 0..TRADES_PER_LEVEL { - let maker = level * TRADES_PER_LEVEL + step + 1; + let trades = if level_capacity == EXACT { + result.trades().capacity() as u64 + } else { + TRADES_PER_LEVEL + }; + assert!( + (1..=LEVEL_QTY).contains(&trades), + "alloc measurement: level trade count" + ); + let fill = LEVEL_QTY / trades; + for step in 0..trades { + let maker = level * LEVEL_QTY + step + 1; + let quantity = if step + 1 == trades { + LEVEL_QTY - fill * (trades - 1) + } else { + fill + }; result .add_trade(Trade::with_timestamp( Id::from_u64(TAKER_ID_BASE + 1_000 + maker), taker, Id::from_u64(maker), Price::new(LEVEL_PRICE), - Quantity::new(FILL), + Quantity::new(quantity), Side::Buy, TimestampMs::new(EXECUTION_TIMESTAMP_MS), )) .expect("alloc measurement: level trade"); } result - .add_filled_order_id(Id::from_u64(level * TRADES_PER_LEVEL + 1)) + .add_filled_order_id(Id::from_u64(level * LEVEL_QTY + 1)) .expect("alloc measurement: level filled id"); + assert!( + level_capacity != EXACT + || result.trades().len() == result.trades().capacity(), + "alloc measurement: an EXACT level's trade buffer must be full" + ); result }) .collect() diff --git a/src/execution/tests/absorb.rs b/src/execution/tests/absorb.rs index 6d86e36..be673cb 100644 --- a/src/execution/tests/absorb.rs +++ b/src/execution/tests/absorb.rs @@ -54,12 +54,22 @@ mod tests { result } - /// As [`level`], with the trade buffer sized exactly for its fills (no - /// spare capacity). - fn exact_level(incoming: u64, fills: &[(u64, u64, bool)]) -> MatchResult { + /// A level result whose trade AND filled-id buffers are genuinely full + /// (length equal to the observed capacity), so absorbing into or from it + /// can neither append into spare room nor adopt it. + /// + /// `try_with_capacity` only guarantees *at least* the requested + /// capacity, so after the `fills` the buffers are padded with filler + /// trades of quantity 1 (makers `filler_base`, `filler_base + 1`, ...): + /// consumed while the id buffer has room, unconsumed after. A filler + /// that meets a full trade buffer grows it; the id buffer never grows, + /// so the padding terminates with both full. The result is finalized + /// with `remaining` left, so its incoming quantity is its executed + /// quantity plus `remaining`. + fn full_level(remaining: u64, fills: &[(u64, u64, bool)], filler_base: u64) -> MatchResult { let mut result = MatchResult::try_with_capacity( Id::from_u64(TAKER), - Quantity::new(incoming), + Quantity::new(u64::MAX), fills.len(), ) .expect("capacity"); @@ -73,12 +83,33 @@ mod tests { .expect("filled id"); } } - assert_eq!(result.trades().capacity(), fills.len()); - let remaining = result.remaining_quantity(); - result.finalize(remaining); + let mut maker = filler_base; + loop { + let ids_room = + result.filled_order_ids().len() < result.test_filled_order_ids_capacity(); + if !ids_room && result.trades().len() == result.trades().capacity() { + break; + } + result + .add_trade(trade(TAKER, maker, 1)) + .expect("filler trade"); + if ids_room { + result + .add_filled_order_id(Id::from_u64(maker)) + .expect("filler id"); + } + maker += 1; + } + result.finalize(Quantity::new(remaining)); result } + /// What `level` needs from the aggregate: its executed plus remaining + /// quantity. + fn incoming(level: &MatchResult) -> u64 { + level.executed_quantity().expect("sum").as_u64() + level.remaining_quantity().as_u64() + } + fn json(result: &MatchResult) -> String { serde_json::to_string(result).expect("serialize") } @@ -239,14 +270,21 @@ mod tests { #[test] fn test_try_absorb_growth_refusal_unchanged_then_appends_in_order() { - // Both buffers are exactly full, so neither the append nor the adopt - // plan fits and the aggregate must grow. - let mut aggregate = exact_level(20, &[(1, 5, true)]); - let mut source = exact_level(15, &[(2, 3, true), (3, 2, false)]); + // Every buffer is full, so neither the append nor the adopt plan + // fits and the aggregate must grow. + let mut source = full_level(10, &[(2, 3, true), (3, 2, false)], 200); + let mut aggregate = full_level(incoming(&source), &[(1, 5, true)], 100); + let total = incoming(&aggregate); + let trades_capacity = aggregate.trades().capacity(); + let filled_capacity = aggregate.test_filled_order_ids_capacity(); + let mut expected_makers = maker_ids(&aggregate); + expected_makers.extend(maker_ids(&source)); + let mut expected_filled = aggregate.filled_order_ids().to_vec(); + expected_filled.extend_from_slice(source.filled_order_ids()); // A capped trade list refuses the growth: nothing changes. { - let _limit = trade_list_seam::limit_trades(2); + let _limit = trade_list_seam::limit_trades(aggregate.trades().len()); let err = assert_refused_unchanged(&mut aggregate, &mut source); assert!(matches!( err, @@ -258,16 +296,16 @@ mod tests { } aggregate.try_absorb(&mut source).expect("absorb"); + // Both vectors grew: the growth path, not append or adopt, ran. + assert!(aggregate.trades().capacity() > trades_capacity); + assert!(aggregate.test_filled_order_ids_capacity() > filled_capacity); + assert_eq!(maker_ids(&aggregate), expected_makers); + assert_eq!(aggregate.filled_order_ids(), expected_filled.as_slice()); + assert_eq!(aggregate.remaining_quantity().as_u64(), 10); assert_eq!( - maker_ids(&aggregate), - vec![Id::from_u64(1), Id::from_u64(2), Id::from_u64(3)] - ); - assert_eq!( - aggregate.filled_order_ids(), - &[Id::from_u64(1), Id::from_u64(2)] + aggregate.executed_quantity().expect("sum").as_u64(), + total - 10 ); - assert_eq!(aggregate.remaining_quantity().as_u64(), 10); - assert_eq!(aggregate.executed_quantity().expect("sum").as_u64(), 10); assert_eq!(aggregate.outcome(), MatchOutcome::PartiallyFilled); assert!(!aggregate.is_complete()); assert_round_trips(&aggregate); @@ -348,32 +386,30 @@ mod tests { // The aggregate's buffers are full; the level's have room for both // sets of entries, so the aggregate's entries move to the front of // the level's buffers and nothing is allocated. - let mut aggregate = - MatchResult::try_with_capacity(Id::from_u64(TAKER), Quantity::new(20), 1).expect("cap"); - aggregate.add_trade(trade(TAKER, 1, 5)).expect("t1"); - aggregate.add_filled_order_id(Id::from_u64(1)).expect("f1"); + let mut aggregate = full_level(15, &[(1, 5, true)], 100); + let room = aggregate.trades().len() + 8; let mut source = - MatchResult::try_with_capacity(Id::from_u64(TAKER), Quantity::new(15), 8).expect("cap"); + MatchResult::try_with_capacity(Id::from_u64(TAKER), Quantity::new(15), room) + .expect("cap"); source.add_trade(trade(TAKER, 2, 3)).expect("t2"); source.add_filled_order_id(Id::from_u64(2)).expect("f2"); source.add_trade(trade(TAKER, 3, 2)).expect("t3"); let trades_ptr = source.trades().as_vec().as_ptr(); + let trades_capacity = source.trades().capacity(); let filled_ptr = source.filled_order_ids().as_ptr(); let aggregate_trades_ptr = aggregate.trades().as_vec().as_ptr(); + let mut expected_makers = maker_ids(&aggregate); + expected_makers.extend([Id::from_u64(2), Id::from_u64(3)]); + let mut expected_filled = aggregate.filled_order_ids().to_vec(); + expected_filled.push(Id::from_u64(2)); aggregate.try_absorb(&mut source).expect("absorb"); assert_eq!(aggregate.trades().as_vec().as_ptr(), trades_ptr); assert_eq!(aggregate.filled_order_ids().as_ptr(), filled_ptr); - assert_eq!(aggregate.trades().capacity(), 8); - assert_eq!( - maker_ids(&aggregate), - vec![Id::from_u64(1), Id::from_u64(2), Id::from_u64(3)] - ); - assert_eq!( - aggregate.filled_order_ids(), - &[Id::from_u64(1), Id::from_u64(2)] - ); + assert_eq!(aggregate.trades().capacity(), trades_capacity); + assert_eq!(maker_ids(&aggregate), expected_makers); + assert_eq!(aggregate.filled_order_ids(), expected_filled.as_slice()); assert_eq!(aggregate.remaining_quantity().as_u64(), 10); // The level got the aggregate's emptied buffer back. assert_eq!(source.trades().as_vec().as_ptr(), aggregate_trades_ptr); From 20d35a7808084233a098b2053e0a4a10e7d4ef08 Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 29 Sep 2026 12:11:16 +0200 Subject: [PATCH 5/7] address review: try_absorb example keeps aggregate and unabsorbed result on error Reserve for the remaining quantity (order_count is advisory only: iceberg and reserve makers replenish and trade again) and return an error struct carrying the aggregate and the unabsorbed level result. --- src/execution/match_result.rs | 62 ++++++++++++++++++++++++----------- 1 file changed, 43 insertions(+), 19 deletions(-) diff --git a/src/execution/match_result.rs b/src/execution/match_result.rs index da704bf..b98edbd 100644 --- a/src/execution/match_result.rs +++ b/src/execution/match_result.rs @@ -407,7 +407,12 @@ impl MatchResult { /// matched, reserve this aggregate for that level's worst case with /// [`Self::try_reserve_trades`] / [`Self::try_reserve_filled_order_ids`] /// *before* matching it, so absorbing its committed trades cannot then - /// fail on growth. + /// fail on growth. The general worst case is the quantity still to + /// fill: every trade executes at least one unit. The level's resting + /// order count is NOT a bound (an iceberg or reserve maker replenishes + /// and can trade several times in one match), so a reservation sized + /// from [`crate::PriceLevel::order_count`] is advisory only and + /// `try_absorb` may still have to grow. /// /// # Failure ownership /// @@ -439,16 +444,22 @@ impl MatchResult { /// TimestampMs, UuidGenerator, /// }; /// - /// /// Sweeps `levels` in price order. A level result that could not be - /// /// absorbed is pushed to `unabsorbed`, so its committed trades are - /// /// never dropped. + /// /// A sweep that stopped early. Nothing committed is lost: `aggregate` + /// /// holds every level absorbed so far and `unabsorbed` the level result + /// /// (if any) that could not be folded in. + /// struct SweepError { + /// error: PriceLevelError, + /// aggregate: MatchResult, + /// unabsorbed: Option, + /// } + /// + /// /// Sweeps `levels` in price order. /// fn sweep( /// levels: &[PriceLevel], /// taker: Id, /// quantity: u64, /// ids: &UuidGenerator, - /// unabsorbed: &mut Vec, - /// ) -> Result { + /// ) -> Result> { /// let mut aggregate = MatchResult::new(taker, Quantity::new(quantity)); /// for (index, level) in levels.iter().enumerate() { /// let remaining = aggregate.remaining_quantity().as_u64(); @@ -456,12 +467,16 @@ impl MatchResult { /// break; /// } /// if index > 0 { - /// // Later levels: reserve the worst case before matching. - /// let bound = level - /// .order_count() - /// .min(usize::try_from(remaining).unwrap_or(usize::MAX)); - /// aggregate.try_reserve_trades(bound)?; - /// aggregate.try_reserve_filled_order_ids(bound)?; + /// // Later levels: reserve the worst case (one trade per unit + /// // still to fill) before matching. On failure nothing was + /// // matched yet, so only the aggregate is handed back. + /// let bound = usize::try_from(remaining).unwrap_or(usize::MAX); + /// let reserved = aggregate + /// .try_reserve_trades(bound) + /// .and_then(|()| aggregate.try_reserve_filled_order_ids(bound)); + /// if let Err(error) = reserved { + /// return Err(Box::new(SweepError { error, aggregate, unabsorbed: None })); + /// } /// } /// let mut result = level.match_order( /// remaining, @@ -472,8 +487,12 @@ impl MatchResult { /// ids, /// ); /// if let Err(error) = aggregate.try_absorb(&mut result) { - /// unabsorbed.push(result); - /// return Err(error); + /// // Both the aggregate and the level's committed trades survive. + /// return Err(Box::new(SweepError { + /// error, + /// aggregate, + /// unabsorbed: Some(result), + /// })); /// } /// if aggregate.is_failed() { /// break; @@ -483,11 +502,16 @@ impl MatchResult { /// } /// /// let ids = UuidGenerator::new(uuid::Uuid::nil()); - /// let mut unabsorbed = Vec::new(); - /// let aggregate = sweep(&[PriceLevel::new(100)], Id::from_u64(1), 5, &ids, &mut unabsorbed)?; - /// assert_eq!(aggregate.remaining_quantity().as_u64(), 5); - /// assert!(unabsorbed.is_empty()); - /// # Ok::<(), PriceLevelError>(()) + /// let levels = [PriceLevel::new(100), PriceLevel::new(101)]; + /// match sweep(&levels, Id::from_u64(1), 5, &ids) { + /// Ok(aggregate) => assert_eq!(aggregate.remaining_quantity().as_u64(), 5), + /// Err(stopped) => panic!( + /// "sweep stopped: {} ({} trades kept, unabsorbed: {})", + /// stopped.error, + /// stopped.aggregate.trades().len(), + /// stopped.unabsorbed.is_some() + /// ), + /// } /// ``` pub fn try_absorb(&mut self, level: &mut MatchResult) -> Result<(), PriceLevelError> { // 1. Validate: nothing changes on a refusal. From 6b600a0829ab7fc789db29614a26ed6914d1fde6 Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 29 Sep 2026 12:11:16 +0200 Subject: [PATCH 6/7] address review: Criterion group comparing try_absorb with reserve-and-copy Covers empty destination (swap), existing reservation (append), adoption into a non-empty destination (rotate) and growth, inputs built outside the timed region. --- benches/price_level/absorb.rs | 153 ++++++++++++++++++++++++++++ benches/price_level/isolated_ops.rs | 2 +- benches/price_level/mod.rs | 2 + 3 files changed, 156 insertions(+), 1 deletion(-) create mode 100644 benches/price_level/absorb.rs diff --git a/benches/price_level/absorb.rs b/benches/price_level/absorb.rs new file mode 100644 index 0000000..d5f7c14 --- /dev/null +++ b/benches/price_level/absorb.rs @@ -0,0 +1,153 @@ +//! Folding one level's `MatchResult` into a multi-level aggregate +//! (issue #219): `MatchResult::try_absorb` against the pre-#219 +//! reserve-and-copy pattern (`try_reserve_trades` / +//! `try_reserve_filled_order_ids`, then `add_trade` / `add_filled_order_id` +//! per entry). +//! +//! One fold per iteration, through [`time_chunked`]: the aggregate and the +//! level result are built before the clock starts and dropped after it +//! stops, so only the fold is timed. Each level result holds [`TRADES`] +//! trades, every maker consumed (one filled id per trade). The cases pick +//! the buffer shapes that select each `try_absorb` plan: +//! +//! * `empty`: a fresh aggregate (`try_absorb` swaps the buffers in); +//! * `reserved`: an aggregate holding one level with room reserved for the +//! next (`try_absorb` appends into the reservation); +//! * `rotate`: a full aggregate and a level whose buffer has room for both +//! (`try_absorb` adopts the level's buffer and rotates its own entries to +//! the front); +//! * `growth`: a full aggregate and a full level (both patterns grow). +//! +//! "Full" means the trade buffer's length equals its observed capacity: +//! `MatchResult::try_with_capacity` guarantees only a lower bound, so the +//! fixtures add trades until the buffer is full instead of trusting the +//! request. The filled-id buffer's capacity is not public; it is requested +//! at the same size and gets one id per trade. + +use super::isolated_ops::time_chunked; +use criterion::Criterion; +use pricelevel::{Id, MatchResult, Price, Quantity, Side, TimestampMs, Trade}; + +/// Trades (and filled ids) per level result. +const TRADES: u64 = 8; +/// Quantity each level executes, split across its trades. +const LEVEL_QTY: u64 = 8_000; +const TAKER: u64 = 1_000_000_000; +const PRICE: u128 = 10_000; +const EXEC_TS: u64 = 1_800_000_000_000; + +/// A result for `incoming` whose first level (makers from `first_maker`) +/// executes [`LEVEL_QTY`]. With `full`, the trade count is the trade +/// buffer's observed capacity (requested [`TRADES`]); otherwise it is +/// [`TRADES`] in a buffer requested at `capacity`. +fn level(incoming: u64, first_maker: u64, capacity: usize, full: bool) -> MatchResult { + let taker = Id::from_u64(TAKER); + let mut result = MatchResult::try_with_capacity(taker, Quantity::new(incoming), capacity) + .expect("level capacity"); + let trades = if full { + result.trades().capacity() as u64 + } else { + TRADES + }; + assert!((1..=LEVEL_QTY).contains(&trades), "level trade count"); + let fill = LEVEL_QTY / trades; + for step in 0..trades { + let maker = first_maker + step; + let quantity = if step + 1 == trades { + LEVEL_QTY - fill * (trades - 1) + } else { + fill + }; + result + .add_trade(Trade::with_timestamp( + Id::from_u64(TAKER + 1 + maker), + taker, + Id::from_u64(maker), + Price::new(PRICE), + Quantity::new(quantity), + Side::Buy, + TimestampMs::new(EXEC_TS), + )) + .expect("level trade"); + result + .add_filled_order_id(Id::from_u64(maker)) + .expect("level filled id"); + } + assert!( + !full || result.trades().len() == result.trades().capacity(), + "a full level's trade buffer must be full" + ); + result +} + +/// The pre-#219 fold: reserve exactly the level's entry counts, then copy. +fn copy(aggregate: &mut MatchResult, level: &MatchResult) { + aggregate + .try_reserve_trades(level.trades().len()) + .expect("reserve trades"); + aggregate + .try_reserve_filled_order_ids(level.filled_order_ids().len()) + .expect("reserve filled ids"); + for &trade in level.trades().as_vec() { + aggregate.add_trade(trade).expect("add_trade"); + } + for &id in level.filled_order_ids() { + aggregate + .add_filled_order_id(id) + .expect("add_filled_order_id"); + } +} + +/// The fixture for `case`: an aggregate and the next level's result. +fn fixture(case: &str) -> (MatchResult, MatchResult) { + let per_level = TRADES as usize; + // The next level's result, matched with LEVEL_QTY left. + let next = |capacity, full| level(LEVEL_QTY, 1_000, capacity, full); + match case { + "empty" => ( + MatchResult::new(Id::from_u64(TAKER), Quantity::new(LEVEL_QTY)), + next(per_level, false), + ), + "reserved" => { + let mut aggregate = level(2 * LEVEL_QTY, 0, per_level, false); + aggregate.try_reserve(per_level).expect("reservation"); + (aggregate, next(per_level, false)) + } + "rotate" => { + let aggregate = level(2 * LEVEL_QTY, 0, per_level, true); + let room = 2 * aggregate.trades().len(); + (aggregate, next(room, false)) + } + "growth" => ( + level(2 * LEVEL_QTY, 0, per_level, true), + next(per_level, true), + ), + _ => unreachable!("unknown absorb case {case}"), + } +} + +/// Register the aggregation benchmarks. +pub fn register_benchmarks(c: &mut Criterion) { + let mut group = c.benchmark_group("PriceLevel - Absorb"); + for case in ["empty", "reserved", "rotate", "growth"] { + group.bench_function(format!("{case}/copy"), |b| { + b.iter_custom(|iters| { + time_chunked( + iters, + || fixture(case), + |(aggregate, level)| copy(aggregate, level), + ) + }); + }); + group.bench_function(format!("{case}/absorb"), |b| { + b.iter_custom(|iters| { + time_chunked( + iters, + || fixture(case), + |(aggregate, level)| aggregate.try_absorb(level).expect("try_absorb"), + ) + }); + }); + } + group.finish(); +} diff --git a/benches/price_level/isolated_ops.rs b/benches/price_level/isolated_ops.rs index 5e9ac94..e7f3b60 100644 --- a/benches/price_level/isolated_ops.rs +++ b/benches/price_level/isolated_ops.rs @@ -83,7 +83,7 @@ const CHUNK: usize = 64; /// are built before, and inputs and outputs dropped after, the timed region /// of each chunk; only the calls (and moving each output into a pre-sized /// buffer) are timed. -fn time_chunked( +pub(super) fn time_chunked( iters: u64, mut setup: impl FnMut() -> I, mut op: impl FnMut(&mut I) -> O, diff --git a/benches/price_level/mod.rs b/benches/price_level/mod.rs index 7083959..36dcfc4 100644 --- a/benches/price_level/mod.rs +++ b/benches/price_level/mod.rs @@ -1,4 +1,5 @@ // benches/price_level/mod.rs +pub mod absorb; pub mod add_orders; pub mod checked_arithmetic; pub mod fok_depth; @@ -36,4 +37,5 @@ pub fn register_benchmarks(c: &mut criterion::Criterion) { result_capacity::register_benchmarks(c); residual_reuse::register_benchmarks(c); isolated_ops::register_benchmarks(c); + absorb::register_benchmarks(c); } From 5877ec7d0e6fff648eeb832eab36bfb263b8479f Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 29 Sep 2026 12:17:50 +0200 Subject: [PATCH 7/7] fix(bench): use panic! instead of the denied unreachable! in the absorb bench --- benches/price_level/absorb.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/benches/price_level/absorb.rs b/benches/price_level/absorb.rs index d5f7c14..d9e814e 100644 --- a/benches/price_level/absorb.rs +++ b/benches/price_level/absorb.rs @@ -122,7 +122,7 @@ fn fixture(case: &str) -> (MatchResult, MatchResult) { level(2 * LEVEL_QTY, 0, per_level, true), next(per_level, true), ), - _ => unreachable!("unknown absorb case {case}"), + _ => panic!("unknown absorb case {case}"), } }