mirror of
https://github.com/floor-licker/polyfill-rs.git
synced 2026-08-13 12:38:05 +00:00
fix: resolve book clock merge conflicts
This commit is contained in:
@@ -4,15 +4,16 @@
|
||||
[](https://docs.rs/polyfill-rs)
|
||||
[](LICENSE)
|
||||
|
||||
A high-performance Polymarket Rust client with latency-optimized data structures and zero-allocation hot paths. The `0.4.x` line is V2-native and intentionally breaking for authenticated trading flows.
|
||||
A high-performance Polymarket Rust client with latency-optimized data structures and allocator-conscious hot paths. The `0.4.x` line is V2-native and intentionally breaking for authenticated trading flows.
|
||||
|
||||
At the time that this project was started, `polymarket-rs-client` was a Polymarket Rust Client with a few GitHub stars, but which seemed to be unmaintained. I took on the task of creating a Rust client which could beat the benchmarks quoted in the README.md of that project, with the added constraint of also maintaining zero alloc hot paths.
|
||||
|
||||
I also want to take a moment to clarify what zero-alloc means because I've now recieved double digit messages about this on twitter/x and telegram. In general, zero alloc means either zero alloc in hot paths (which can be a bit more arbitrary) or atlernatively it can mean zero alloc after init/warm-up, which is the objective of this repository. Succinctly that means that **the per-message handling loop never touches the heap**.
|
||||
I also want to take a moment to clarify what zero-alloc means because I've now recieved double digit messages about this on twitter/x and telegram. In this repository the strict claim is limited to tested, warmed hot paths: existing-level book updates and selected read-side calculations are covered by no-heap-traffic tests that count allocations, reallocations, and deallocations. Snapshot churn, first-seen books, and new price levels can still touch the allocator by design.
|
||||
|
||||
Notably order book paths that introduce new allocations by design:
|
||||
Notably order book paths that can touch the allocator by design:
|
||||
- First time seeing a token/book (HashMap insert + key clone): `src/book.rs`
|
||||
- New price levels (sorted Vec insert/growth): `src/book.rs`
|
||||
- New price levels when a sorted side needs to grow: `src/book.rs`
|
||||
- Book removal/drop paths that release owned buffers: `src/book.rs`
|
||||
|
||||
|
||||
## Quick Start
|
||||
@@ -68,7 +69,7 @@ Real-world Polymarket API latency broken down by request phase:
|
||||
|
||||
| Operation | Performance | Notes |
|
||||
|-----------|-------------|-------|
|
||||
| **Order Book Updates (1000 ops)** | 69.6 µs | ~14.4M updates/sec, zero-allocation for warmed existing levels |
|
||||
| **Order Book Updates (1000 ops)** | 69.6 µs | ~14.4M updates/sec, no allocator traffic for warmed existing-level paths |
|
||||
| **Spread/Mid Calculations** | 26.6 ns | best bid/ask + spread + mid over sorted-vector book sides |
|
||||
| **JSON Parsing (480KB)** | ~0.5 ms | SIMD-backed parsing for large REST market responses and benchmarked polyfill typed parse path |
|
||||
| **WS `book` hot path (decode + apply)** | ~0.24 µs / 1.56 µs / 5.92 µs | 1 / 16 / 64 levels-per-side, strict fixed-point tape parser with generation-marked snapshot retention (see `benches/ws_hot_path.rs`) |
|
||||
@@ -83,7 +84,7 @@ The 21.4% performance improvement comes from HTTP/2 tuning with 512KB stream win
|
||||
|
||||
### Memory Architecture
|
||||
|
||||
Configurable book depth limiting prevents memory bloat. Hot data structures group frequently-accessed fields for cache line efficiency. Allocation-sensitive hot paths are covered by targeted no-allocation tests where the implementation is currently allocation-free.
|
||||
Configurable book depth limiting prevents memory bloat. Hot data structures group frequently-accessed fields for cache line efficiency. Allocation-sensitive hot paths are covered by targeted no-heap-traffic tests where the implementation currently avoids allocation, reallocation, and deallocation.
|
||||
|
||||
### Architectural Principles
|
||||
|
||||
|
||||
+170
-67
@@ -16,6 +16,13 @@ struct StoredLevel {
|
||||
generation: u64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub(crate) struct ParsedBookLevel {
|
||||
pub side: Side,
|
||||
pub price_ticks: Price,
|
||||
pub size_units: Qty,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
enum BookSideKind {
|
||||
Bid,
|
||||
@@ -235,7 +242,7 @@ impl OrderBook {
|
||||
Self {
|
||||
token_id,
|
||||
token_id_hash,
|
||||
sequence: 0, // Legacy alias for last_delta_sequence
|
||||
sequence: 0, // Compatibility alias for last_delta_sequence
|
||||
last_delta_sequence: 0,
|
||||
last_snapshot_timestamp_ms: 0,
|
||||
timestamp: Utc::now(),
|
||||
@@ -535,14 +542,12 @@ impl OrderBook {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Begin applying a WebSocket `book` update (hot-path oriented).
|
||||
///
|
||||
/// This is intended for in-place WS processing where we *stream* levels out of a decoded
|
||||
/// message, without constructing intermediate `BookUpdate` structs.
|
||||
///
|
||||
/// Returns `Ok(true)` if the update should be applied, or `Ok(false)` if the update is stale
|
||||
/// and should be skipped.
|
||||
pub(crate) fn begin_ws_book_update(&mut self, asset_id: &str, timestamp: u64) -> Result<bool> {
|
||||
/// Return whether a WebSocket `book` snapshot should be applied.
|
||||
pub(crate) fn should_apply_ws_book_update(
|
||||
&self,
|
||||
asset_id: &str,
|
||||
timestamp: u64,
|
||||
) -> Result<bool> {
|
||||
if asset_id != self.token_id {
|
||||
return Err(PolyfillError::validation("Token ID mismatch"));
|
||||
}
|
||||
@@ -551,39 +556,27 @@ impl OrderBook {
|
||||
return Ok(false);
|
||||
}
|
||||
|
||||
self.last_snapshot_timestamp_ms = timestamp;
|
||||
self.timestamp = chrono::DateTime::<Utc>::from_timestamp_millis(timestamp as i64)
|
||||
.unwrap_or_else(Utc::now);
|
||||
self.begin_snapshot();
|
||||
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
/// Apply a single WS `book` level (already converted to internal fixed-point).
|
||||
/// Atomically apply a WebSocket `book` snapshot.
|
||||
///
|
||||
/// Note: Insertions of new price levels may allocate (Vec growth/shifting). In a strict
|
||||
/// zero-alloc hot path, all expected levels must be warmed up ahead of time.
|
||||
pub(crate) fn apply_ws_book_level_fast(
|
||||
/// The caller should parse levels into `ParsedBookLevel`s before calling this method. This
|
||||
/// method validates tick alignment before mutating sequence/generation or book levels, so a
|
||||
/// malformed snapshot leaves the existing book unchanged.
|
||||
pub(crate) fn apply_ws_book_snapshot_fast(
|
||||
&mut self,
|
||||
side: Side,
|
||||
price_ticks: Price,
|
||||
size_units: Qty,
|
||||
) -> Result<()> {
|
||||
if let Some(tick_size_ticks) = self.tick_size_ticks {
|
||||
if tick_size_ticks > 0 && !price_ticks.is_multiple_of(tick_size_ticks) {
|
||||
return Err(PolyfillError::validation("Price not aligned to tick size"));
|
||||
}
|
||||
asset_id: &str,
|
||||
timestamp: u64,
|
||||
levels: &[ParsedBookLevel],
|
||||
) -> Result<bool> {
|
||||
if !self.should_apply_ws_book_update(asset_id, timestamp)? {
|
||||
return Ok(false);
|
||||
}
|
||||
|
||||
self.apply_snapshot_level(side, price_ticks, size_units);
|
||||
self.apply_validated_snapshot(timestamp, levels)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Finish applying a WS `book` snapshot.
|
||||
pub(crate) fn finish_ws_book_update(&mut self) {
|
||||
self.finish_snapshot();
|
||||
self.trim_depth();
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
/// Apply a WebSocket `book` update for this token.
|
||||
@@ -601,46 +594,44 @@ impl OrderBook {
|
||||
return Err(PolyfillError::validation("Token ID mismatch"));
|
||||
}
|
||||
|
||||
// Use the exchange-provided timestamp as the monotonic snapshot marker.
|
||||
// Snapshot timestamps are separate from legacy/incremental delta sequences.
|
||||
// Use the exchange-provided timestamp as the monotonic marker for snapshots.
|
||||
// This is intentionally separate from legacy/incremental delta sequence numbers.
|
||||
if update.timestamp <= self.last_snapshot_timestamp_ms {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// Validate the whole snapshot before mutating this book. A malformed level must not leave
|
||||
// behind a partial generation or an advanced sequence number.
|
||||
for level in &update.bids {
|
||||
let parsed = self.parse_snapshot_summary(Side::BUY, level)?;
|
||||
self.validate_snapshot_level(parsed)?;
|
||||
}
|
||||
|
||||
for level in &update.asks {
|
||||
let parsed = self.parse_snapshot_summary(Side::SELL, level)?;
|
||||
self.validate_snapshot_level(parsed)?;
|
||||
}
|
||||
|
||||
self.last_snapshot_timestamp_ms = update.timestamp;
|
||||
self.timestamp = chrono::DateTime::<Utc>::from_timestamp_millis(update.timestamp as i64)
|
||||
.unwrap_or_else(Utc::now);
|
||||
self.begin_snapshot();
|
||||
|
||||
// Apply bids (BUY) and asks (SELL) as level upserts.
|
||||
// Re-parse after validation to preserve the existing no-allocation behavior for
|
||||
// `BookUpdate` snapshots. Decimal conversion is deterministic, so these conversions cannot
|
||||
// fail after the validation pass above.
|
||||
for level in &update.bids {
|
||||
let price_ticks = decimal_to_price(level.price)
|
||||
.map_err(|_| PolyfillError::validation("Invalid price"))?;
|
||||
let size_units = decimal_to_qty(level.size)
|
||||
.map_err(|_| PolyfillError::validation("Invalid size"))?;
|
||||
|
||||
if let Some(tick_size_ticks) = self.tick_size_ticks {
|
||||
if tick_size_ticks > 0 && !price_ticks.is_multiple_of(tick_size_ticks) {
|
||||
return Err(PolyfillError::validation("Price not aligned to tick size"));
|
||||
}
|
||||
}
|
||||
|
||||
self.apply_snapshot_level(Side::BUY, price_ticks, size_units);
|
||||
let parsed = self
|
||||
.parse_snapshot_summary(Side::BUY, level)
|
||||
.expect("book update bid level was validated before mutation");
|
||||
self.apply_snapshot_level(parsed.side, parsed.price_ticks, parsed.size_units);
|
||||
}
|
||||
|
||||
for level in &update.asks {
|
||||
let price_ticks = decimal_to_price(level.price)
|
||||
.map_err(|_| PolyfillError::validation("Invalid price"))?;
|
||||
let size_units = decimal_to_qty(level.size)
|
||||
.map_err(|_| PolyfillError::validation("Invalid size"))?;
|
||||
|
||||
if let Some(tick_size_ticks) = self.tick_size_ticks {
|
||||
if tick_size_ticks > 0 && !price_ticks.is_multiple_of(tick_size_ticks) {
|
||||
return Err(PolyfillError::validation("Price not aligned to tick size"));
|
||||
}
|
||||
}
|
||||
|
||||
self.apply_snapshot_level(Side::SELL, price_ticks, size_units);
|
||||
let parsed = self
|
||||
.parse_snapshot_summary(Side::SELL, level)
|
||||
.expect("book update ask level was validated before mutation");
|
||||
self.apply_snapshot_level(parsed.side, parsed.price_ticks, parsed.size_units);
|
||||
}
|
||||
|
||||
self.finish_snapshot();
|
||||
@@ -722,6 +713,55 @@ impl OrderBook {
|
||||
self.snapshot_generation = self.snapshot_generation.wrapping_add(1);
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn parse_snapshot_summary(&self, side: Side, level: &OrderSummary) -> Result<ParsedBookLevel> {
|
||||
let price_ticks = decimal_to_price(level.price)
|
||||
.map_err(|_| PolyfillError::validation("Invalid price"))?;
|
||||
let size_units =
|
||||
decimal_to_qty(level.size).map_err(|_| PolyfillError::validation("Invalid size"))?;
|
||||
|
||||
Ok(ParsedBookLevel {
|
||||
side,
|
||||
price_ticks,
|
||||
size_units,
|
||||
})
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn validate_snapshot_level(&self, level: ParsedBookLevel) -> Result<()> {
|
||||
if let Some(tick_size_ticks) = self.tick_size_ticks {
|
||||
if tick_size_ticks > 0 && !level.price_ticks.is_multiple_of(tick_size_ticks) {
|
||||
return Err(PolyfillError::validation("Price not aligned to tick size"));
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn apply_validated_snapshot(
|
||||
&mut self,
|
||||
timestamp: u64,
|
||||
levels: &[ParsedBookLevel],
|
||||
) -> Result<()> {
|
||||
for &level in levels {
|
||||
self.validate_snapshot_level(level)?;
|
||||
}
|
||||
|
||||
self.last_snapshot_timestamp_ms = timestamp;
|
||||
self.timestamp = chrono::DateTime::<Utc>::from_timestamp_millis(timestamp as i64)
|
||||
.unwrap_or_else(Utc::now);
|
||||
self.begin_snapshot();
|
||||
|
||||
for &level in levels {
|
||||
self.apply_snapshot_level(level.side, level.price_ticks, level.size_units);
|
||||
}
|
||||
|
||||
self.finish_snapshot();
|
||||
self.trim_depth();
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn apply_snapshot_level(&mut self, side: Side, price_ticks: Price, size_units: Qty) {
|
||||
let generation = self.snapshot_generation;
|
||||
@@ -1367,15 +1407,22 @@ mod tests {
|
||||
fn test_ws_snapshot_timestamp_does_not_block_delta_sequence() {
|
||||
let mut book = OrderBook::new("test_token".to_string(), 10);
|
||||
let snapshot_timestamp_ms = 1_757_908_892_351;
|
||||
let levels = [
|
||||
ParsedBookLevel {
|
||||
side: Side::BUY,
|
||||
price_ticks: 5_000,
|
||||
size_units: 100_000,
|
||||
},
|
||||
ParsedBookLevel {
|
||||
side: Side::SELL,
|
||||
price_ticks: 6_000,
|
||||
size_units: 100_000,
|
||||
},
|
||||
];
|
||||
|
||||
assert!(book
|
||||
.begin_ws_book_update("test_token", snapshot_timestamp_ms)
|
||||
.apply_ws_book_snapshot_fast("test_token", snapshot_timestamp_ms, &levels)
|
||||
.unwrap());
|
||||
book.apply_ws_book_level_fast(Side::BUY, 5_000, 100_000)
|
||||
.unwrap();
|
||||
book.apply_ws_book_level_fast(Side::SELL, 6_000, 100_000)
|
||||
.unwrap();
|
||||
book.finish_ws_book_update();
|
||||
|
||||
assert_eq!(book.sequence, 0);
|
||||
assert_eq!(book.last_delta_sequence, 0);
|
||||
@@ -1434,6 +1481,62 @@ mod tests {
|
||||
assert_eq!(book.best_ask().unwrap().price, dec!(0.60));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_book_update_error_keeps_existing_snapshot() {
|
||||
let mut book = OrderBook::new("test_token".to_string(), 10);
|
||||
book.set_tick_size_ticks(100);
|
||||
|
||||
book.apply_book_update(&BookUpdate {
|
||||
asset_id: "test_token".to_string(),
|
||||
market: "0xabc".to_string(),
|
||||
timestamp: 100,
|
||||
bids: vec![OrderSummary {
|
||||
price: dec!(0.50),
|
||||
size: dec!(10),
|
||||
}],
|
||||
asks: vec![OrderSummary {
|
||||
price: dec!(0.60),
|
||||
size: dec!(20),
|
||||
}],
|
||||
hash: None,
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
let err = book
|
||||
.apply_book_update(&BookUpdate {
|
||||
asset_id: "test_token".to_string(),
|
||||
market: "0xabc".to_string(),
|
||||
timestamp: 101,
|
||||
bids: vec![
|
||||
OrderSummary {
|
||||
price: dec!(0.51),
|
||||
size: dec!(11),
|
||||
},
|
||||
OrderSummary {
|
||||
price: dec!(0.515),
|
||||
size: dec!(12),
|
||||
},
|
||||
],
|
||||
asks: vec![OrderSummary {
|
||||
price: dec!(0.61),
|
||||
size: dec!(21),
|
||||
}],
|
||||
hash: None,
|
||||
})
|
||||
.unwrap_err();
|
||||
|
||||
assert!(err.to_string().contains("Price not aligned"));
|
||||
assert_eq!(book.sequence, 0);
|
||||
assert_eq!(book.last_delta_sequence, 0);
|
||||
assert_eq!(book.last_snapshot_timestamp_ms, 100);
|
||||
assert_eq!(book.bids(None).len(), 1);
|
||||
assert_eq!(book.asks(None).len(), 1);
|
||||
assert_eq!(book.best_bid().unwrap().price, dec!(0.50));
|
||||
assert_eq!(book.best_bid().unwrap().size, dec!(10));
|
||||
assert_eq!(book.best_ask().unwrap().price, dec!(0.60));
|
||||
assert_eq!(book.best_ask().unwrap().size, dec!(20));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_book_update_depth_keeps_best_prices_independent_of_payload_order() {
|
||||
let mut book = OrderBook::new("test_token".to_string(), 2);
|
||||
|
||||
@@ -705,4 +705,17 @@ mod tests {
|
||||
let results: Vec<serde_json::Value> = decoder.parse_json_stream(data).unwrap();
|
||||
assert_eq!(results.len(), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stream_book_message_requires_bids_and_asks() {
|
||||
let missing_asks = br#"{"event_type":"book","asset_id":"test_asset_id","market":"0xabc","timestamp":1000,"bids":[]}"#;
|
||||
assert!(parse_stream_messages_bytes(missing_asks).is_err());
|
||||
|
||||
let missing_bids = br#"{"event_type":"book","asset_id":"test_asset_id","market":"0xabc","timestamp":1000,"asks":[]}"#;
|
||||
assert!(parse_stream_messages_bytes(missing_bids).is_err());
|
||||
|
||||
let empty_sides = br#"{"event_type":"book","asset_id":"test_asset_id","market":"0xabc","timestamp":1000,"bids":[],"asks":[]}"#;
|
||||
let messages = parse_stream_messages_bytes(empty_sides).unwrap();
|
||||
assert_eq!(messages.len(), 1);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -429,12 +429,28 @@ impl<'a> WebSocketBookApplier<'a> {
|
||||
}
|
||||
|
||||
/// Apply a single WS text payload (useful for custom transports and for testing).
|
||||
///
|
||||
/// This convenience method consumes an owned `String`, so it may release that
|
||||
/// buffer after processing. Use [`Self::apply_bytes_message`] for the
|
||||
/// allocation-sensitive path when the caller owns a reusable mutable buffer.
|
||||
pub fn apply_text_message(&mut self, text: String) -> Result<WsBookApplyStats> {
|
||||
let stats = self.processor.process_text(text, self.books)?;
|
||||
self.stream.stats.messages_received += 1;
|
||||
self.stream.stats.last_message_time = Some(Utc::now());
|
||||
Ok(stats)
|
||||
}
|
||||
|
||||
/// Apply a single WS payload from a caller-owned mutable byte buffer.
|
||||
///
|
||||
/// The buffer is mutated by `simd-json`. After processor warmup, this is the
|
||||
/// allocation-sensitive book-applier entry point because no owned message buffer
|
||||
/// is created or dropped by this method.
|
||||
pub fn apply_bytes_message(&mut self, bytes: &mut [u8]) -> Result<WsBookApplyStats> {
|
||||
let stats = self.processor.process_bytes(bytes, self.books)?;
|
||||
self.stream.stats.messages_received += 1;
|
||||
self.stream.stats.last_message_time = Some(Utc::now());
|
||||
Ok(stats)
|
||||
}
|
||||
}
|
||||
|
||||
impl<'a> Stream for WebSocketBookApplier<'a> {
|
||||
|
||||
+2
-8
@@ -909,15 +909,9 @@ pub struct BookUpdate {
|
||||
pub market: String,
|
||||
#[serde(deserialize_with = "crate::decode::deserializers::number_from_string")]
|
||||
pub timestamp: u64,
|
||||
#[serde(
|
||||
default,
|
||||
deserialize_with = "crate::decode::deserializers::vec_from_null"
|
||||
)]
|
||||
#[serde(deserialize_with = "crate::decode::deserializers::vec_from_null")]
|
||||
pub bids: Vec<OrderSummary>,
|
||||
#[serde(
|
||||
default,
|
||||
deserialize_with = "crate::decode::deserializers::vec_from_null"
|
||||
)]
|
||||
#[serde(deserialize_with = "crate::decode::deserializers::vec_from_null")]
|
||||
pub asks: Vec<OrderSummary>,
|
||||
#[serde(default)]
|
||||
pub hash: Option<String>,
|
||||
|
||||
+118
-22
@@ -1,13 +1,14 @@
|
||||
//! Zero-allocation-ish WebSocket hot-path processing.
|
||||
//!
|
||||
//! This module is focused on the "decode + apply" path for WS `book` events:
|
||||
//! after warmup, processing a message should not perform heap allocations.
|
||||
//! after warmup, existing-level happy-path processing should not perform heap
|
||||
//! allocation, reallocation, or deallocation.
|
||||
//!
|
||||
//! Important: using the current tokio-tungstenite transport, the *network layer*
|
||||
//! may still allocate when producing `Message::Text(String)`. This module aims to
|
||||
//! make the *processing* layer allocation-free so we can enforce it with tests.
|
||||
|
||||
use crate::book::OrderBookManager;
|
||||
use crate::book::{OrderBookManager, ParsedBookLevel};
|
||||
use crate::errors::{PolyfillError, Result};
|
||||
use crate::types::{Price, Qty, Side, MAX_PRICE_TICKS, MAX_QTY, MIN_PRICE_TICKS, SCALE_FACTOR};
|
||||
use simd_json::prelude::*;
|
||||
@@ -26,6 +27,7 @@ pub struct WsBookApplyStats {
|
||||
pub struct WsBookUpdateProcessor {
|
||||
buffers: simd_json::Buffers,
|
||||
tape: Option<simd_json::Tape<'static>>,
|
||||
parsed_levels: Vec<ParsedBookLevel>,
|
||||
}
|
||||
|
||||
impl WsBookUpdateProcessor {
|
||||
@@ -37,6 +39,7 @@ impl WsBookUpdateProcessor {
|
||||
buffers: simd_json::Buffers::new(input_len_hint),
|
||||
// Store an empty tape with a `'static` lifetime so we can reuse its allocation.
|
||||
tape: Some(simd_json::Tape::null().reset()),
|
||||
parsed_levels: Vec::with_capacity((input_len_hint / 32).max(8)),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -55,7 +58,7 @@ impl WsBookUpdateProcessor {
|
||||
let result = match simd_json::fill_tape(bytes, &mut self.buffers, &mut tape) {
|
||||
Ok(()) => {
|
||||
let root = tape.as_value();
|
||||
process_root_value(root, books)
|
||||
process_root_value(root, books, &mut self.parsed_levels)
|
||||
},
|
||||
Err(e) => Err(PolyfillError::parse(
|
||||
"Failed to parse WebSocket JSON",
|
||||
@@ -82,9 +85,10 @@ impl WsBookUpdateProcessor {
|
||||
fn process_root_value<'tape, 'input>(
|
||||
value: simd_json::tape::Value<'tape, 'input>,
|
||||
books: &OrderBookManager,
|
||||
parsed_levels: &mut Vec<ParsedBookLevel>,
|
||||
) -> Result<WsBookApplyStats> {
|
||||
if let Some(obj) = value.as_object() {
|
||||
return process_stream_object(obj, books);
|
||||
return process_stream_object(obj, books, parsed_levels);
|
||||
}
|
||||
|
||||
let Some(arr) = value.as_array() else {
|
||||
@@ -96,7 +100,7 @@ fn process_root_value<'tape, 'input>(
|
||||
let Some(obj) = elem.as_object() else {
|
||||
continue;
|
||||
};
|
||||
let stats = process_stream_object(obj, books)?;
|
||||
let stats = process_stream_object(obj, books, parsed_levels)?;
|
||||
total.book_messages += stats.book_messages;
|
||||
total.book_levels_applied += stats.book_levels_applied;
|
||||
}
|
||||
@@ -107,6 +111,7 @@ fn process_root_value<'tape, 'input>(
|
||||
fn process_stream_object<'tape, 'input>(
|
||||
obj: simd_json::tape::Object<'tape, 'input>,
|
||||
books: &OrderBookManager,
|
||||
parsed_levels: &mut Vec<ParsedBookLevel>,
|
||||
) -> Result<WsBookApplyStats> {
|
||||
let Some(event_type) = obj.get("event_type").and_then(|v| v.into_string()) else {
|
||||
return Ok(WsBookApplyStats::default());
|
||||
@@ -127,25 +132,36 @@ fn process_stream_object<'tape, 'input>(
|
||||
let timestamp = parse_u64(timestamp_value)
|
||||
.ok_or_else(|| PolyfillError::parse("Invalid timestamp", None))?;
|
||||
|
||||
let bids = obj.get("bids").and_then(|v| v.as_array());
|
||||
let asks = obj.get("asks").and_then(|v| v.as_array());
|
||||
let bids = obj
|
||||
.get("bids")
|
||||
.ok_or_else(|| PolyfillError::parse("Missing bids", None))?
|
||||
.as_array()
|
||||
.ok_or_else(|| PolyfillError::parse("Invalid bids", None))?;
|
||||
let asks = obj
|
||||
.get("asks")
|
||||
.ok_or_else(|| PolyfillError::parse("Missing asks", None))?
|
||||
.as_array()
|
||||
.ok_or_else(|| PolyfillError::parse("Invalid asks", None))?;
|
||||
|
||||
let levels_applied = books.with_book_mut(asset_id, |book| {
|
||||
if !book.begin_ws_book_update(asset_id, timestamp)? {
|
||||
let result = books.with_book_mut(asset_id, |book| {
|
||||
parsed_levels.clear();
|
||||
|
||||
if !book.should_apply_ws_book_update(asset_id, timestamp)? {
|
||||
return Ok(0);
|
||||
}
|
||||
|
||||
let mut applied = 0usize;
|
||||
if let Some(bids) = bids {
|
||||
applied += apply_levels(book, Side::BUY, bids)?;
|
||||
}
|
||||
if let Some(asks) = asks {
|
||||
applied += apply_levels(book, Side::SELL, asks)?;
|
||||
}
|
||||
collect_levels(Side::BUY, bids, parsed_levels)?;
|
||||
collect_levels(Side::SELL, asks, parsed_levels)?;
|
||||
|
||||
book.finish_ws_book_update();
|
||||
Ok(applied)
|
||||
})?;
|
||||
let parsed_count = parsed_levels.len();
|
||||
if book.apply_ws_book_snapshot_fast(asset_id, timestamp, parsed_levels)? {
|
||||
Ok(parsed_count)
|
||||
} else {
|
||||
Ok(0)
|
||||
}
|
||||
});
|
||||
parsed_levels.clear();
|
||||
let levels_applied = result?;
|
||||
|
||||
Ok(WsBookApplyStats {
|
||||
book_messages: 1,
|
||||
@@ -159,10 +175,10 @@ fn parse_u64<'tape, 'input>(value: simd_json::tape::Value<'tape, 'input>) -> Opt
|
||||
.or_else(|| value.into_string().and_then(|s| s.parse::<u64>().ok()))
|
||||
}
|
||||
|
||||
fn apply_levels<'tape, 'input>(
|
||||
book: &mut crate::book::OrderBook,
|
||||
fn collect_levels<'tape, 'input>(
|
||||
side: Side,
|
||||
levels: simd_json::tape::Array<'tape, 'input>,
|
||||
parsed_levels: &mut Vec<ParsedBookLevel>,
|
||||
) -> Result<usize> {
|
||||
let mut applied = 0usize;
|
||||
for level in levels.iter() {
|
||||
@@ -182,7 +198,11 @@ fn apply_levels<'tape, 'input>(
|
||||
let price_ticks = parse_price_ticks_4dp(price_str)?;
|
||||
let size_units = parse_qty_scaled_4dp(size_str)?;
|
||||
|
||||
book.apply_ws_book_level_fast(side, price_ticks, size_units)?;
|
||||
parsed_levels.push(ParsedBookLevel {
|
||||
side,
|
||||
price_ticks,
|
||||
size_units,
|
||||
});
|
||||
applied += 1;
|
||||
}
|
||||
|
||||
@@ -273,6 +293,8 @@ fn parse_scaled_4_u64(value: &str) -> Result<u64> {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::types::{BookUpdate, OrderSummary};
|
||||
use rust_decimal_macros::dec;
|
||||
|
||||
#[test]
|
||||
fn fixed_point_parser_matches_expected_price_ticks() {
|
||||
@@ -318,4 +340,78 @@ mod tests {
|
||||
assert_eq!(stats.book_messages, 1);
|
||||
assert_eq!(stats.book_levels_applied, 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn processor_error_keeps_existing_snapshot() {
|
||||
let books = OrderBookManager::new(10);
|
||||
books.get_or_create_book("test_asset_id").unwrap();
|
||||
books
|
||||
.apply_book_update(&BookUpdate {
|
||||
asset_id: "test_asset_id".to_string(),
|
||||
market: "0xabc".to_string(),
|
||||
timestamp: 1000,
|
||||
bids: vec![OrderSummary {
|
||||
price: dec!(0.50),
|
||||
size: dec!(10),
|
||||
}],
|
||||
asks: vec![OrderSummary {
|
||||
price: dec!(0.60),
|
||||
size: dec!(20),
|
||||
}],
|
||||
hash: None,
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
let mut processor = WsBookUpdateProcessor::new(1024);
|
||||
let mut invalid_snapshot = br#"{"event_type":"book","asset_id":"test_asset_id","market":"0xabc","timestamp":1001,"bids":[{"price":"0.5100","size":"11.0000"},{"price":"0.51001","size":"12.0000"}],"asks":[{"price":"0.6100","size":"21.0000"}]}"#.to_vec();
|
||||
assert!(processor
|
||||
.process_bytes(invalid_snapshot.as_mut_slice(), &books)
|
||||
.is_err());
|
||||
|
||||
let snapshot = books.get_book("test_asset_id").unwrap();
|
||||
assert_eq!(snapshot.sequence, 0);
|
||||
assert_eq!(snapshot.timestamp.timestamp_millis(), 1000);
|
||||
assert_eq!(snapshot.bids.len(), 1);
|
||||
assert_eq!(snapshot.asks.len(), 1);
|
||||
assert_eq!(snapshot.bids[0].price, dec!(0.50));
|
||||
assert_eq!(snapshot.bids[0].size, dec!(10));
|
||||
assert_eq!(snapshot.asks[0].price, dec!(0.60));
|
||||
assert_eq!(snapshot.asks[0].size, dec!(20));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn processor_missing_side_keeps_existing_snapshot() {
|
||||
let books = OrderBookManager::new(10);
|
||||
books.get_or_create_book("test_asset_id").unwrap();
|
||||
books
|
||||
.apply_book_update(&BookUpdate {
|
||||
asset_id: "test_asset_id".to_string(),
|
||||
market: "0xabc".to_string(),
|
||||
timestamp: 1000,
|
||||
bids: vec![OrderSummary {
|
||||
price: dec!(0.50),
|
||||
size: dec!(10),
|
||||
}],
|
||||
asks: vec![OrderSummary {
|
||||
price: dec!(0.60),
|
||||
size: dec!(20),
|
||||
}],
|
||||
hash: None,
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
let mut processor = WsBookUpdateProcessor::new(1024);
|
||||
let mut missing_asks = br#"{"event_type":"book","asset_id":"test_asset_id","market":"0xabc","timestamp":1001,"bids":[{"price":"0.5100","size":"11.0000"}]}"#.to_vec();
|
||||
assert!(processor
|
||||
.process_bytes(missing_asks.as_mut_slice(), &books)
|
||||
.is_err());
|
||||
|
||||
let snapshot = books.get_book("test_asset_id").unwrap();
|
||||
assert_eq!(snapshot.sequence, 0);
|
||||
assert_eq!(snapshot.timestamp.timestamp_millis(), 1000);
|
||||
assert_eq!(snapshot.bids.len(), 1);
|
||||
assert_eq!(snapshot.asks.len(), 1);
|
||||
assert_eq!(snapshot.bids[0].price, dec!(0.50));
|
||||
assert_eq!(snapshot.asks[0].price, dec!(0.60));
|
||||
}
|
||||
}
|
||||
|
||||
+238
-50
@@ -1,6 +1,7 @@
|
||||
use std::alloc::{GlobalAlloc, Layout, System};
|
||||
use std::cell::Cell;
|
||||
use std::collections::hash_map::DefaultHasher;
|
||||
use std::fmt::Write as _;
|
||||
use std::hash::{Hash, Hasher};
|
||||
use std::str::FromStr;
|
||||
|
||||
@@ -11,28 +12,29 @@ use polyfill_rs::{
|
||||
use rust_decimal::Decimal;
|
||||
|
||||
thread_local! {
|
||||
static ALLOCATIONS: Cell<usize> = const { Cell::new(0) };
|
||||
static HEAP_OPERATIONS: Cell<usize> = const { Cell::new(0) };
|
||||
}
|
||||
|
||||
struct CountingAllocator;
|
||||
|
||||
unsafe impl GlobalAlloc for CountingAllocator {
|
||||
unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
|
||||
ALLOCATIONS.with(|count| count.set(count.get() + 1));
|
||||
HEAP_OPERATIONS.with(|count| count.set(count.get() + 1));
|
||||
System.alloc(layout)
|
||||
}
|
||||
|
||||
unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
|
||||
ALLOCATIONS.with(|count| count.set(count.get() + 1));
|
||||
HEAP_OPERATIONS.with(|count| count.set(count.get() + 1));
|
||||
System.alloc_zeroed(layout)
|
||||
}
|
||||
|
||||
unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
|
||||
ALLOCATIONS.with(|count| count.set(count.get() + 1));
|
||||
HEAP_OPERATIONS.with(|count| count.set(count.get() + 1));
|
||||
System.realloc(ptr, layout, new_size)
|
||||
}
|
||||
|
||||
unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
|
||||
HEAP_OPERATIONS.with(|count| count.set(count.get() + 1));
|
||||
System.dealloc(ptr, layout)
|
||||
}
|
||||
}
|
||||
@@ -40,32 +42,45 @@ unsafe impl GlobalAlloc for CountingAllocator {
|
||||
#[global_allocator]
|
||||
static GLOBAL: CountingAllocator = CountingAllocator;
|
||||
|
||||
fn allocation_count() -> usize {
|
||||
ALLOCATIONS.with(|count| count.get())
|
||||
fn heap_operation_count() -> usize {
|
||||
HEAP_OPERATIONS.with(|count| count.get())
|
||||
}
|
||||
|
||||
struct NoAllocGuard {
|
||||
struct NoHeapTrafficGuard {
|
||||
before: usize,
|
||||
}
|
||||
|
||||
impl NoAllocGuard {
|
||||
impl NoHeapTrafficGuard {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
before: allocation_count(),
|
||||
before: heap_operation_count(),
|
||||
}
|
||||
}
|
||||
|
||||
fn assert_no_allocations(self) {
|
||||
let after = allocation_count();
|
||||
fn assert_no_heap_traffic(self) {
|
||||
let after = heap_operation_count();
|
||||
assert_eq!(
|
||||
after,
|
||||
self.before,
|
||||
"expected no heap allocations, but saw {} allocation(s)",
|
||||
"expected no heap traffic, but saw {} allocator operation(s)",
|
||||
after - self.before
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn allocator_counter_tracks_deallocations() {
|
||||
let vec = Vec::<u8>::with_capacity(1024);
|
||||
let before = heap_operation_count();
|
||||
drop(vec);
|
||||
let after = heap_operation_count();
|
||||
|
||||
assert!(
|
||||
after > before,
|
||||
"expected dropping an allocated Vec to count as heap traffic"
|
||||
);
|
||||
}
|
||||
|
||||
fn token_id_hash(token_id: &str) -> u64 {
|
||||
let mut hasher = DefaultHasher::new();
|
||||
token_id.hash(&mut hasher);
|
||||
@@ -89,6 +104,88 @@ fn mk_delta(
|
||||
}
|
||||
}
|
||||
|
||||
fn seed_book_levels(
|
||||
manager: &OrderBookManager,
|
||||
asset_id: &str,
|
||||
bid_ticks: &[i64],
|
||||
ask_ticks: &[i64],
|
||||
) {
|
||||
let mut sequence = 1u64;
|
||||
for &price_ticks in bid_ticks {
|
||||
manager
|
||||
.apply_delta(polyfill_rs::types::OrderDelta {
|
||||
token_id: asset_id.to_string(),
|
||||
timestamp: chrono::Utc::now(),
|
||||
side: Side::BUY,
|
||||
price: Decimal::new(price_ticks, 4),
|
||||
size: Decimal::from_str("100.0").unwrap(),
|
||||
sequence,
|
||||
})
|
||||
.unwrap();
|
||||
sequence += 1;
|
||||
}
|
||||
|
||||
for &price_ticks in ask_ticks {
|
||||
manager
|
||||
.apply_delta(polyfill_rs::types::OrderDelta {
|
||||
token_id: asset_id.to_string(),
|
||||
timestamp: chrono::Utc::now(),
|
||||
side: Side::SELL,
|
||||
price: Decimal::new(price_ticks, 4),
|
||||
size: Decimal::from_str("100.0").unwrap(),
|
||||
sequence,
|
||||
})
|
||||
.unwrap();
|
||||
sequence += 1;
|
||||
}
|
||||
}
|
||||
|
||||
fn ws_book_message(
|
||||
asset_id: &str,
|
||||
timestamp: u64,
|
||||
bid_ticks: &[i64],
|
||||
ask_ticks: &[i64],
|
||||
) -> Vec<u8> {
|
||||
let mut json = String::with_capacity(160 + (bid_ticks.len() + ask_ticks.len()) * 40);
|
||||
write!(
|
||||
&mut json,
|
||||
"{{\"event_type\":\"book\",\"asset_id\":\"{asset_id}\",\"market\":\"0xabc\",\"timestamp\":{timestamp},\"bids\":["
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
for (idx, price_ticks) in bid_ticks.iter().enumerate() {
|
||||
if idx > 0 {
|
||||
json.push(',');
|
||||
}
|
||||
write!(
|
||||
&mut json,
|
||||
"{{\"price\":\"{}\",\"size\":\"100.0000\"}}",
|
||||
Decimal::new(*price_ticks, 4)
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
json.push_str("],\"asks\":[");
|
||||
for (idx, price_ticks) in ask_ticks.iter().enumerate() {
|
||||
if idx > 0 {
|
||||
json.push(',');
|
||||
}
|
||||
write!(
|
||||
&mut json,
|
||||
"{{\"price\":\"{}\",\"size\":\"100.0000\"}}",
|
||||
Decimal::new(*price_ticks, 4)
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
json.push_str("]}");
|
||||
json.into_bytes()
|
||||
}
|
||||
|
||||
fn contiguous_ticks(start: i64, len: usize, step: i64) -> Vec<i64> {
|
||||
(0..len).map(|idx| start + (idx as i64 * step)).collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_alloc_mid_and_spread_fast() {
|
||||
let token_id = "test_token";
|
||||
@@ -101,15 +198,15 @@ fn no_alloc_mid_and_spread_fast() {
|
||||
book.apply_delta_fast(mk_delta(token_hash, Side::SELL, 7600, 1_000_000, 2))
|
||||
.unwrap();
|
||||
|
||||
// Warm up TLS access before measuring (defensive).
|
||||
let _ = allocation_count();
|
||||
// Warm up allocator-counter TLS access before measuring (defensive).
|
||||
let _ = heap_operation_count();
|
||||
|
||||
let guard = NoAllocGuard::new();
|
||||
let guard = NoHeapTrafficGuard::new();
|
||||
assert!(book.best_bid_fast().is_some());
|
||||
assert!(book.best_ask_fast().is_some());
|
||||
assert!(book.spread_fast().is_some());
|
||||
assert!(book.mid_price_fast().is_some());
|
||||
guard.assert_no_allocations();
|
||||
guard.assert_no_heap_traffic();
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -134,9 +231,9 @@ fn no_alloc_book_analysis_fast_paths() {
|
||||
let expected_buy_liquidity = Decimal::from_str("200.0").unwrap();
|
||||
let expected_sell_liquidity = Decimal::from_str("150.0").unwrap();
|
||||
|
||||
let _ = allocation_count();
|
||||
let _ = heap_operation_count();
|
||||
|
||||
let guard = NoAllocGuard::new();
|
||||
let guard = NoHeapTrafficGuard::new();
|
||||
let impact = book
|
||||
.calculate_market_impact(Side::BUY, impact_size)
|
||||
.unwrap();
|
||||
@@ -150,7 +247,7 @@ fn no_alloc_book_analysis_fast_paths() {
|
||||
expected_sell_liquidity
|
||||
);
|
||||
assert!(book.is_valid());
|
||||
guard.assert_no_allocations();
|
||||
guard.assert_no_heap_traffic();
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -163,14 +260,14 @@ fn no_alloc_apply_delta_fast_existing_level_update() {
|
||||
book.apply_delta_fast(mk_delta(token_hash, Side::BUY, 7500, 1_000_000, 1))
|
||||
.unwrap();
|
||||
|
||||
// Warm up TLS access before measuring (defensive).
|
||||
let _ = allocation_count();
|
||||
// Warm up allocator-counter TLS access before measuring (defensive).
|
||||
let _ = heap_operation_count();
|
||||
|
||||
let guard = NoAllocGuard::new();
|
||||
// Updating an existing level should not require heap allocation.
|
||||
let guard = NoHeapTrafficGuard::new();
|
||||
// Updating an existing level should not touch the heap allocator.
|
||||
book.apply_delta_fast(mk_delta(token_hash, Side::BUY, 7500, 2_000_000, 2))
|
||||
.unwrap();
|
||||
guard.assert_no_allocations();
|
||||
guard.assert_no_heap_traffic();
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -200,12 +297,12 @@ fn no_alloc_apply_book_update_existing_levels() {
|
||||
hash: None,
|
||||
};
|
||||
|
||||
// Warm up TLS access before measuring (defensive).
|
||||
let _ = allocation_count();
|
||||
// Warm up allocator-counter TLS access before measuring (defensive).
|
||||
let _ = heap_operation_count();
|
||||
|
||||
let guard = NoAllocGuard::new();
|
||||
let guard = NoHeapTrafficGuard::new();
|
||||
book.apply_book_update(&update).unwrap();
|
||||
guard.assert_no_allocations();
|
||||
guard.assert_no_heap_traffic();
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -214,7 +311,7 @@ fn no_alloc_book_manager_apply_book_update_existing_levels() {
|
||||
let manager = OrderBookManager::new(100);
|
||||
manager.get_or_create_book(asset_id).unwrap();
|
||||
|
||||
// Warm up the internal book with initial levels (allocations allowed).
|
||||
// Warm up the internal book with initial levels (allocator traffic allowed).
|
||||
manager
|
||||
.apply_delta(polyfill_rs::types::OrderDelta {
|
||||
token_id: asset_id.to_string(),
|
||||
@@ -251,12 +348,12 @@ fn no_alloc_book_manager_apply_book_update_existing_levels() {
|
||||
hash: None,
|
||||
};
|
||||
|
||||
// Warm up TLS access before measuring (defensive).
|
||||
let _ = allocation_count();
|
||||
// Warm up allocator-counter TLS access before measuring (defensive).
|
||||
let _ = heap_operation_count();
|
||||
|
||||
let guard = NoAllocGuard::new();
|
||||
let guard = NoHeapTrafficGuard::new();
|
||||
manager.apply_book_update(&update).unwrap();
|
||||
guard.assert_no_allocations();
|
||||
guard.assert_no_heap_traffic();
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -265,7 +362,7 @@ fn no_alloc_ws_book_update_processor_apply_existing_levels() {
|
||||
let manager = OrderBookManager::new(100);
|
||||
manager.get_or_create_book(asset_id).unwrap();
|
||||
|
||||
// Warm up the internal book with initial levels (allocations allowed).
|
||||
// Warm up the internal book with initial levels (allocator traffic allowed).
|
||||
manager
|
||||
.apply_delta(polyfill_rs::types::OrderDelta {
|
||||
token_id: asset_id.to_string(),
|
||||
@@ -303,23 +400,110 @@ fn no_alloc_ws_book_update_processor_apply_existing_levels() {
|
||||
)
|
||||
.into_bytes();
|
||||
|
||||
// Warm up TLS access before measuring (defensive).
|
||||
let _ = allocation_count();
|
||||
// Warm up allocator-counter TLS access before measuring (defensive).
|
||||
let _ = heap_operation_count();
|
||||
|
||||
let guard = NoAllocGuard::new();
|
||||
let guard = NoHeapTrafficGuard::new();
|
||||
processor
|
||||
.process_bytes(msg.as_mut_slice(), &manager)
|
||||
.unwrap();
|
||||
guard.assert_no_allocations();
|
||||
guard.assert_no_heap_traffic();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_alloc_websocket_book_applier_apply_text_message_existing_levels() {
|
||||
fn no_alloc_ws_book_update_processor_one_new_level_with_reserved_capacity() {
|
||||
let asset_id = "test_asset_id";
|
||||
let manager = OrderBookManager::new(100);
|
||||
manager.get_or_create_book(asset_id).unwrap();
|
||||
seed_book_levels(&manager, asset_id, &[7500], &[7600]);
|
||||
|
||||
let mut processor = WsBookUpdateProcessor::new(4096);
|
||||
let mut warmup_msg = ws_book_message(asset_id, 10, &[7500], &[7600]);
|
||||
processor
|
||||
.process_bytes(warmup_msg.as_mut_slice(), &manager)
|
||||
.unwrap();
|
||||
let mut parser_capacity_warmup = ws_book_message(asset_id, 10, &[7500, 7400], &[7600]);
|
||||
processor
|
||||
.process_bytes(parser_capacity_warmup.as_mut_slice(), &manager)
|
||||
.unwrap();
|
||||
|
||||
let mut msg = ws_book_message(asset_id, 11, &[7500, 7400], &[7600]);
|
||||
|
||||
let _ = heap_operation_count();
|
||||
|
||||
let guard = NoHeapTrafficGuard::new();
|
||||
processor
|
||||
.process_bytes(msg.as_mut_slice(), &manager)
|
||||
.unwrap();
|
||||
guard.assert_no_heap_traffic();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_alloc_ws_book_update_processor_one_removed_level() {
|
||||
let asset_id = "test_asset_id";
|
||||
let manager = OrderBookManager::new(100);
|
||||
manager.get_or_create_book(asset_id).unwrap();
|
||||
seed_book_levels(&manager, asset_id, &[7500, 7400], &[7600]);
|
||||
|
||||
let mut processor = WsBookUpdateProcessor::new(4096);
|
||||
let mut warmup_msg = ws_book_message(asset_id, 10, &[7500, 7400], &[7600]);
|
||||
processor
|
||||
.process_bytes(warmup_msg.as_mut_slice(), &manager)
|
||||
.unwrap();
|
||||
|
||||
let mut msg = ws_book_message(asset_id, 11, &[7500], &[7600]);
|
||||
|
||||
let _ = heap_operation_count();
|
||||
|
||||
let guard = NoHeapTrafficGuard::new();
|
||||
processor
|
||||
.process_bytes(msg.as_mut_slice(), &manager)
|
||||
.unwrap();
|
||||
guard.assert_no_heap_traffic();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ws_book_update_processor_full_churn_64_levels_touches_allocator() {
|
||||
let asset_id = "test_asset_id";
|
||||
let levels_per_side = 64;
|
||||
let manager = OrderBookManager::new(levels_per_side);
|
||||
manager.get_or_create_book(asset_id).unwrap();
|
||||
|
||||
let initial_bids = contiguous_ticks(7500, levels_per_side, -1);
|
||||
let initial_asks = contiguous_ticks(7600, levels_per_side, 1);
|
||||
seed_book_levels(&manager, asset_id, &initial_bids, &initial_asks);
|
||||
|
||||
let mut warmup_msg = ws_book_message(asset_id, 1000, &initial_bids, &initial_asks);
|
||||
let mut processor = WsBookUpdateProcessor::new(warmup_msg.len());
|
||||
processor
|
||||
.process_bytes(warmup_msg.as_mut_slice(), &manager)
|
||||
.unwrap();
|
||||
|
||||
let churn_bids = contiguous_ticks(7300, levels_per_side, -1);
|
||||
let churn_asks = contiguous_ticks(7800, levels_per_side, 1);
|
||||
let mut msg = ws_book_message(asset_id, 1001, &churn_bids, &churn_asks);
|
||||
|
||||
let _ = heap_operation_count();
|
||||
|
||||
let before = heap_operation_count();
|
||||
processor
|
||||
.process_bytes(msg.as_mut_slice(), &manager)
|
||||
.unwrap();
|
||||
let after = heap_operation_count();
|
||||
|
||||
assert!(
|
||||
after > before,
|
||||
"expected 64-level full churn to touch the allocator while old and new levels coexist"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_alloc_websocket_book_applier_apply_bytes_message_existing_levels() {
|
||||
let asset_id = "test_asset_id";
|
||||
let manager = OrderBookManager::new(100);
|
||||
manager.get_or_create_book(asset_id).unwrap();
|
||||
|
||||
// Warm up the internal book with initial levels (allocations allowed).
|
||||
// Warm up the internal book with initial levels (allocator traffic allowed).
|
||||
manager
|
||||
.apply_delta(polyfill_rs::types::OrderDelta {
|
||||
token_id: asset_id.to_string(),
|
||||
@@ -346,19 +530,23 @@ fn no_alloc_websocket_book_applier_apply_text_message_existing_levels() {
|
||||
let mut applier = stream.into_book_applier(&manager, processor);
|
||||
|
||||
// Warm up simd-json buffers/tape outside the guarded section.
|
||||
let warmup_msg = format!(
|
||||
let mut warmup_msg = format!(
|
||||
"{{\"event_type\":\"book\",\"asset_id\":\"{asset_id}\",\"market\":\"0xabc\",\"timestamp\":10,\"bids\":[{{\"price\":\"0.75\",\"size\":\"200.0\"}}],\"asks\":[{{\"price\":\"0.76\",\"size\":\"50.0\"}}]}}"
|
||||
);
|
||||
applier.apply_text_message(warmup_msg).unwrap();
|
||||
)
|
||||
.into_bytes();
|
||||
applier
|
||||
.apply_bytes_message(warmup_msg.as_mut_slice())
|
||||
.unwrap();
|
||||
|
||||
let msg = format!(
|
||||
let mut msg = format!(
|
||||
"{{\"event_type\":\"book\",\"asset_id\":\"{asset_id}\",\"market\":\"0xabc\",\"timestamp\":11,\"bids\":[{{\"price\":\"0.75\",\"size\":\"150.0\"}}],\"asks\":[{{\"price\":\"0.76\",\"size\":\"75.0\"}}]}}"
|
||||
);
|
||||
)
|
||||
.into_bytes();
|
||||
|
||||
// Warm up TLS access before measuring (defensive).
|
||||
let _ = allocation_count();
|
||||
// Warm up allocator-counter TLS access before measuring (defensive).
|
||||
let _ = heap_operation_count();
|
||||
|
||||
let guard = NoAllocGuard::new();
|
||||
applier.apply_text_message(msg).unwrap();
|
||||
guard.assert_no_allocations();
|
||||
let guard = NoHeapTrafficGuard::new();
|
||||
applier.apply_bytes_message(msg.as_mut_slice()).unwrap();
|
||||
guard.assert_no_heap_traffic();
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user