Add MatchResult::try_absorb and split reservations (#219)

CHANGELOG.md

@@ -16,6 +16,32 @@ 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.
+- **`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 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`, 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
 

benches/latency/alloc_measurements.rs

@@ -78,9 +78,191 @@ pub fn run_all(config: &Config) -> Vec<AllocReport> {
         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, 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),
     ]
 }
 
+/// 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` 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 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
+/// 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), each level result created with
+/// `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
+/// 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,
+    level_capacity: usize,
+) -> AllocReport {
+    const TRADES_PER_LEVEL: u64 = 2;
+    /// 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 * LEVEL_QTY;
+    let mut inputs: Vec<Vec<MatchResult>> = (0..reps)
+        .map(|_| {
+            (0..levels as u64)
+                .map(|level| {
+                    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");
+                    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(quantity),
+                                Side::Buy,
+                                TimestampMs::new(EXECUTION_TIMESTAMP_MS),
+                            ))
+                            .expect("alloc measurement: level trade");
+                    }
+                    result
+                        .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()
+        })
+        .collect();
+    let mut outputs: Vec<MatchResult> = 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 (index, result) in level_results.iter_mut().enumerate() {
+            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"),
+                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);
+    }
+    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, 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),
+    }
+}
+
 /// 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

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),
+        ),
+        _ => panic!("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();
+}

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<I, O>(
+pub(super) fn time_chunked<I, O>(
     iters: u64,
     mut setup: impl FnMut() -> I,
     mut op: impl FnMut(&mut I) -> O,

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);
 }