diff --git a/CHANGELOG.md b/CHANGELOG.md index 63dcaee..4abb170 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,43 @@ only entries that can cost a reader an afternoon. ### Added +- `venues::BinanceSbeDecoder`: the Binance spot SBE binary streams (schema + `spot_stream` 1:0) — depth diff and depth snapshot — decoded to the same + `DecodedMessage` the JSON decoders produce, with the same route-by-symbol + and refuse-rather-than-round contracts, and SBE blockLengths honoured so a + newer minor schema skips cleanly. Tested against hand-encoded frames + including a byte-by-byte truncation sweep; a live capture awaits an + Ed25519 API key, which the endpoint requires even for public data. +- Busy-poll reads: `set_read_timeout(0)` on any transport (and + `WebSocketClient::set_read_timeout`) now means non-blocking spin, not the + platforms' block-forever. `crossbook_capture --busy-poll` uses it, and on + Linux reports a kernel-to-user delivery histogram from `SO_TIMESTAMPING` + receive stamps — measured live against Kraken, busy-poll cut delivery p50 + from 150.4 us to 82.4 us. +- `Transport::last_rx_time_ns`: the kernel's arrival clock for the newest + received data, exposed through TLS on both backends. On the OpenSSL path + this forced reads through a custom BIO, since a backend that lets + `SSL_read` call `recv()` itself can never see the control message the + timestamp rides in on. +- Loopback tests for the transport layer — the busy-poll contract, the timed + path, and the receive timestamp — the first tests the socket has had. + +- `net::ByteBuffer`: a `std::vector` whose `resize` default-initializes + instead of zeroing. The frame reader and the Schannel backend grow their + receive buffers by a 32 KiB chunk on every socket read and trim back to what + arrived; with a plain vector that was 32 KiB of memset per read, all of it + over bytes the transport was about to overwrite. +- `CROSSBOOK_NATIVE` and the `release-native` preset: opt-in `-march=native` + (`/arch:AVX2` on MSVC) plus LTO, for measuring the ceiling on one's own + hardware. Off by default, and the README's numbers stay on plain release, + because a binary tuned to the build machine dies on the next machine. +- The no-allocation probe now covers the decoder, `Feed::handle` end to end + with the checksum verified, and the frame reader's poll loop. It covered + the book — 0.3% of the frame — while the claim it enforces is about the + whole hot path. +- `LATENCY-ROADMAP.md`: where the gap to professional software trading + systems actually is (environment tail, JSON ceiling, compute already + competitive) and the phase order for closing it, with sources. - `json::for_each_member`: walk an object's members once, in wire order, dispatching on key. A completed walk carries `well_formed`'s full guarantee, which is what lets the decoders below drop their separate validation pass. @@ -40,6 +77,12 @@ only entries that can cost a reader an afternoon. ### Changed +- The Schannel decrypt path copies plaintext once, straight into the caller's + buffer, instead of twice through an intermediate; and the unconsumed tail of + a pipelined TLS record is moved in place rather than through a freshly + allocated vector, which was a heap allocation on the common path — a busy + feed routinely lands the next record behind the current one in the same + segment. - Both venue decoders are single-pass. A Kraken frame was being walked ~9x — a `well_formed` pre-pass plus a `find` restart per field, with `checksum` and `timestamp` spelled after the level arrays on the wire so each of those diff --git a/CMakeLists.txt b/CMakeLists.txt index 1da5e65..55e2a6f 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -38,6 +38,30 @@ option(CROSSBOOK_BUILD_FUZZ "Build the fuzz targets (clang only)" OFF) option(CROSSBOOK_BUILD_EXAMPLES "Build the compilable examples" ${CROSSBOOK_IS_TOP_LEVEL}) option(CROSSBOOK_WERROR "Treat warnings as errors" ${CROSSBOOK_IS_TOP_LEVEL}) +# Off by default and never inherited: a binary tuned for the build machine's +# ISA is a binary that dies with an illegal instruction on the next machine, +# and the README's numbers must come from the configuration a consumer gets +# by default. This exists for measuring the ceiling on one's own hardware. +option(CROSSBOOK_NATIVE + "Tune codegen for this machine (-march=native, /arch:AVX2 on MSVC) and enable LTO" OFF) + +if(CROSSBOOK_NATIVE) + if(MSVC) + # MSVC has no -march=native; AVX2 is the widest ISA this project is + # willing to assume behind an explicit opt-in. + add_compile_options(/arch:AVX2) + else() + add_compile_options(-march=native) + endif() + include(CheckIPOSupported) + check_ipo_supported(RESULT crossbook_ipo_supported OUTPUT crossbook_ipo_message) + if(crossbook_ipo_supported) + set(CMAKE_INTERPROCEDURAL_OPTIMIZATION ON) + else() + message(STATUS "crossbook: LTO unavailable, continuing without it: ${crossbook_ipo_message}") + endif() +endif() + # Installing is a decision for whoever owns the prefix. A parent project that # vendors crossbook and then runs `cmake --install .` must not have crossbook's # headers appear in its prefix as a side effect it never asked for, so the diff --git a/CMakePresets.json b/CMakePresets.json index f14d374..e70d282 100644 --- a/CMakePresets.json +++ b/CMakePresets.json @@ -37,6 +37,15 @@ "CROSSBOOK_BUILD_EXAMPLES": "ON" } }, + { + "name": "release-native", + "inherits": "release", + "displayName": "Release, tuned for this machine", + "description": "Adds -march=native (MSVC: /arch:AVX2) and LTO. Numbers from this preset describe one machine and are not comparable across hosts; the README's figures come from the plain release preset.", + "cacheVariables": { + "CROSSBOOK_NATIVE": "ON" + } + }, { "name": "bench", "inherits": "release", diff --git a/LATENCY-ROADMAP.md b/LATENCY-ROADMAP.md new file mode 100644 index 0000000..5e8785c --- /dev/null +++ b/LATENCY-ROADMAP.md @@ -0,0 +1,151 @@ +# Latency roadmap + +Written 2026-08-02, from a four-track research pass: a file-level inventory of +this repository's hot path, and web research into wire formats, software +tick-to-trade practice, and tail-latency engineering. Claims below that came +from outside sources carry links; figures computed rather than measured are +marked as such. This document records where the gap to professional software +trading systems actually is, and the order in which to close it. + +## Where the gap is + +The reference point for a software (non-FPGA) trading system is roughly +2.5 us wire-to-wire ([Carl Cook, CppCon 2017](https://www.youtube.com/watch?v=NH1Tta7purM)). +Against that, this library's position splits three ways: + +1. **The compute core is already in the right league.** ~8 ns book updates, + ~600 ns checksum, ~2.1 us for a fully verified frame. Not the problem. +2. **The tail is the environment, not the code.** At 6k msg/s the handler sits + at ~1.2% utilization; M/D/1 queueing puts the p99 physics floor around + 9-14 us. The measured 912 us p99 is ~100x that — Windows scheduler quanta, + DPCs, C-state exits. A pinned busy-polling thread on an isolated core of + tuned Linux measures p99 1.4 us / p99.9 4.9 us / max 18 us of induced + jitter ([Rigtorp, hiccups](https://github.com/rigtorp/hiccups)); Windows + has a hard ~100 us p99.9 floor from DPCs that no user-mode setting removes + ([LatencyMon docs](https://resplendence.com/latencymon_using)). +3. **JSON is the structural ceiling on the median.** Decode is ~1.9 us, 75% of + the frame. simdjson-class techniques put a validated ~300-byte frame at + roughly 200-500 ns (computed from published GB/s throughput, not a bench), + and binary venue feeds remove the cost entirely. + +One calibration that bounds the whole effort: measured from AWS Tokyo, +Binance's own websocket delivery is avg 4 ms, p99 13 ms +([Deltix/Ember](https://ember.deltixlab.com/docs/performance/ws-market-data/)). +On crypto venues the exchange side is milliseconds; the competitive variable +is tail determinism, not median parse time. + +## What the venues offer (2026) + +| Venue | Fastest wire | Engine location | +|---|---|---| +| Kraken | FIX L3 (tag=value; no binary feed exists) | Equinix London; Beeks hosted colo | +| Binance spot | SBE WebSocket incl. L2 diff depth (Ed25519 key) | AWS Tokyo ap-northeast-1 | +| Deribit | SBE multicast, plaintext UDP; Starbase SBE L3 + order entry | Equinix LD4; AWS eu-west-2/ap-northeast-1 | +| Coinbase Exchange | FIX 5.0 L3 market data | AWS us-east-1 (use1-az4) | +| OKX / Bybit | JSON WebSocket only (public) | AWS HK / AWS Singapore | + +Sources: [Kraken L3](https://docs.kraken.com/exchange/guides/general/l3-data), +[Binance SBE streams](https://developers.binance.com/docs/binance-spot-api-docs/sbe-market-data-streams), +[Deribit multicast](https://insights.deribit.com/exchange-updates/launch-of-our-new-multicast-service/), +[Deribit Starbase](https://insights.deribit.com/exchange-updates/starbase-a-new-era-of-high-performance-trading-on-deribit/), +[Coinbase FIX MD](https://docs.cdp.coinbase.com/exchange/fix-api/market-data). + +## Phases + +**Phase 0 — repository defects (done in the commit series that added this +document, except where noted).** + +- `FrameReader::writable_tail` value-initialized 32 KiB per socket read + (`vector::resize` zeroing), immediately shrunk back by `commit`. Same + pattern on the Schannel ciphertext buffer. +- The Schannel decrypt path heap-allocated a fresh `std::vector` per + pipelined TLS record and copied plaintext twice (OpenSSL path copies once). +- The no-allocation test enforced the book but not `Feed::handle` or the + decoders — the 75% of the frame the claim was actually about. +- No opt-in `-march`/LTO configuration existed. + +**Phase 1 — prove the tail on tuned Linux.** The repo already builds, tests, +and replays on Linux in CI, and `crossbook_verify` already carries `--pin` +and `--realtime`. On a box tuned per the standard recipe (isolcpus + +nohz_full + rcu_nocbs, IRQ affinity away, performance governor, C-states +capped at C1, SMT off, mlockall — [Rigtorp's guide](https://rigtorp.se/low-latency-guide/)), +qualified first with hwlatdetect and rtla osnoise, the existing `--sweep` +should collapse from 912 us p99 to tens of microseconds with zero code +changes. Publish the tuned-vs-untuned pair; it is the honest-measurement +story this README already tells, completed. + +*Measured so far (2026-08-02), same desktop, WSL2 Ubuntu — a VM, not tuned +metal, and still directional:* the identical sweep binary showed p50 falling +from 20-39 us (Windows) to 2.7-3.2 us and top-rung p99 from 912 us to +144 us — a 6.3x tail improvement from the OS change alone. Two predicted +effects reproduced: SCHED_FIFO inside the VM is catastrophic (the spin +starves its own vCPU; p50 collapsed to milliseconds), and the lowest-rate +rung pays the deep-idle wakeup (8 ms p99 at 5 msg/s). Real metal with +isolation remains the open item. + +**Phase 2 — transport for latency.** *Done (2026-08-02), except the +zero-copy read contract.* `set_read_timeout(0)` now means busy-poll on +every transport; Linux reads carry the kernel's `SO_TIMESTAMPING` arrival +stamp (learned the hard way: `SIOCGSTAMP` and `SO_TIMESTAMPNS` are both +dead ends for TCP), and the OpenSSL backend reads through `TcpSocket` via a +custom BIO so timestamps and busy-poll survive TLS. Verified live against +Kraken: busy-poll cut kernel-to-user delivery p50 from 150.4 us to 82.4 us +(unpinned WSL2; the p99 wants the isolated core Phase 1 provides). Still +open: a zero-copy read contract — copy-in `read(buf, len)` cannot express +an rx ring. Skip kTLS: RX-path p99 regressions +([netdev paper](https://netdevconf.info/1.2/papers/ktls.pdf)) and it blocks +the Onload route. Steady-state TLS crypto is under 1 us/record (computed from +~0.64 cycles/byte AES-GCM) and is not the problem. + +**Phase 3 — a binary venue decoder.** *Decoder done (2026-08-02); live +connection blocked on a credential.* `BinanceSbeDecoder` +(`crossbook/venues/binance_sbe.hpp`) decodes the spot SBE depth diff and +depth snapshot streams (schema `spot_stream` 1:0) into the same +`DecodedMessage` the JSON decoders produce, drives `Feed` end to end, and +honours SBE blockLengths for forward compatibility. Tested against +hand-encoded frames including a full truncation sweep — the endpoint +(`stream-sbe.binance.com:9443`) requires an Ed25519 API key even for public +data, so a live capture and a captured-fixture replay await a key. Deribit +SBE multicast follows if derivatives matter — the only feed anywhere that +removes TLS entirely. Kraken's lever is placement plus FIX L3, not +encoding. + +**Phase 4 — the JSON decode floor, for venues stuck with it.** Levers in +order: key dispatch by length/first byte instead of chained `string_view` +compares; deriving canonical-spelling during the parse instead of +re-formatting and byte-comparing every scalar; optionally a SIMD structural +stage. Separately, the checksum's ~600 ns is dominated by re-serializing 20 +levels per message — maintain the top-10 payload incrementally as levels +change instead. + +One lever is measured dead and removed from the list: replacing the +per-digit checked multiply/add in `parse_fixed` with an unchecked fast path +below 19 digits. Interleaved same-state A/B (MSVC /O2, two rounds, median +of 7) showed the change within noise or marginally slower — the compiler +already handles the checked arithmetic well, and the digit loop is not +where decode time lives. The change was reverted; measure before believing +any remaining lever. + +**Phase 5 — placement and bypass.** In-region metal (c7i/c8g/m8azn) in a +shared cluster placement group measures ~20 us p50 / ~23 us p99.9 +instance-to-instance ([AWS tick-to-trade series](https://aws.amazon.com/blogs/web3/optimize-tick-to-trade-latency-for-digital-assets-exchanges-and-trading-platforms-on-aws-part-2/)); +exchanges pull market makers into their placement groups. Onload is the +drop-in kernel bypass for a TCP+TLS websocket client (sockets-compatible, +~6 us plus most network jitter); ef_vi/TCPDirect is a rewrite that buys the +last few hundred nanoseconds and comes last. + +## Expected position + +| Stage | p50/frame | p99 under load | +|---|---|---| +| Untuned Windows desktop (today) | ~2.1 us compute | 912 us | +| Phase 1: tuned Linux, same code | same | ~10-30 us | +| Phases 2-3: busy-poll + SBE venue | ~0.5-1 us | ~5-15 us | +| Phase 5: in-region metal + bypass | sub-us compute | ~20-25 us incl. cloud network | + +The last row is competitive with the crypto-native trading tier. The +remaining distance to traditional-HFT numbers is the venues themselves, +which deliver data in milliseconds. Two standing caveats: several +per-technique figures above are computed or single-source, and nothing here +measures tick-to-trade until an order path exists — this repository has no +egress, so end-to-end latency is unmeasurable by construction. diff --git a/include/crossbook/net/byte_buffer.hpp b/include/crossbook/net/byte_buffer.hpp new file mode 100644 index 0000000..b18dc4c --- /dev/null +++ b/include/crossbook/net/byte_buffer.hpp @@ -0,0 +1,60 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Josh Dardashti +// +// A byte buffer whose resize does not zero. +// +// std::vector::resize value-initializes every new element, which for a +// receive buffer means a memset over bytes the transport is about to +// overwrite anyway. The reassembly path grows its buffer by 32 KiB on every +// socket read and then shrinks it back to what was actually received; with a +// plain vector that is 32 KiB of zeroing per read, all of it wasted. +// +// The standard fix: an allocator whose construct() default-initializes +// instead of value-initializing. For trivially default-constructible types, +// default-initialization is a no-op, so resize becomes pure bookkeeping. +// Everything else about std::vector — growth policy, iterator semantics, +// exception guarantees — is unchanged. + +#pragma once + +#include +#include +#include + +namespace crossbook::net::detail { + +template > +class DefaultInitAllocator : public Base { +public: + template + struct rebind { + using other = + DefaultInitAllocator::template rebind_alloc>; + }; + + using Base::Base; + + /// The point of the class: `new (p) U` default-initializes, so for byte + /// buffers no memory is written until the caller writes it. + template + void construct(U* p) noexcept(std::is_nothrow_default_constructible_v) { + ::new (static_cast(p)) U; + } + + /// Constructions with arguments (insert, push_back, range copies) keep + /// their ordinary value semantics. + template + void construct(U* p, Args&&... args) { + std::allocator_traits::construct(static_cast(*this), p, + std::forward(args)...); + } +}; + +} // namespace crossbook::net::detail + +namespace crossbook::net { + +/// Receive-path byte storage: a std::vector whose resize is free. +using ByteBuffer = std::vector>; + +} // namespace crossbook::net diff --git a/include/crossbook/net/transport.hpp b/include/crossbook/net/transport.hpp index 80cdeb1..14e4612 100644 --- a/include/crossbook/net/transport.hpp +++ b/include/crossbook/net/transport.hpp @@ -95,8 +95,22 @@ class Transport { /// loop stays responsive to shutdown and to its own deadlines. Using one /// value for both means choosing between a client that gives up on a slow /// connect and a loop that hangs for ten seconds on every quiet market. + /// + /// ZERO MEANS BUSY-POLL: the socket goes non-blocking and reads return + /// kTimeout immediately when nothing is buffered, so a caller on a + /// dedicated core can spin instead of taking a scheduler wakeup per + /// message. It does not mean "no timeout" — that spelling of zero is the + /// platforms', and it is never what a latency-sensitive reader wants. virtual void set_read_timeout(int timeout_ms) = 0; + /// Kernel arrival time of the most recently received data, CLOCK_REALTIME + /// nanoseconds, or 0 where the platform offers none for TCP (Windows, + /// macOS) or nothing has arrived. On Linux this is per socket and queried + /// on demand, so it works identically under TLS. One read draining + /// several coalesced segments reports the newest — callers measuring + /// kernel-to-user delivery own that approximation. + [[nodiscard]] virtual std::int64_t last_rx_time_ns() const noexcept { return 0; } + virtual void close() = 0; [[nodiscard]] virtual bool connected() const noexcept = 0; diff --git a/include/crossbook/net/websocket.hpp b/include/crossbook/net/websocket.hpp index 746a175..152d6ad 100644 --- a/include/crossbook/net/websocket.hpp +++ b/include/crossbook/net/websocket.hpp @@ -84,6 +84,15 @@ class WebSocketClient { /// Send a close frame and drop the connection. void close(CloseCode code = CloseCode::kNormal, std::string_view reason = {}); + /// Change the read timeout after connecting. Zero selects busy-poll: poll + /// returns kNeedMore immediately instead of sleeping in the kernel, for a + /// caller that owns a core and spins. See Transport::set_read_timeout. + void set_read_timeout(int timeout_ms); + + /// Kernel arrival time of the newest received data (CLOCK_REALTIME ns), + /// 0 where the platform has none. See Transport::last_rx_time_ns. + [[nodiscard]] std::int64_t last_rx_time_ns() const noexcept; + [[nodiscard]] bool connected() const noexcept; [[nodiscard]] const std::string& last_error() const noexcept { return error_; } [[nodiscard]] const Url& url() const noexcept { return url_; } diff --git a/include/crossbook/net/ws_frame.hpp b/include/crossbook/net/ws_frame.hpp index dd567f0..65c3878 100644 --- a/include/crossbook/net/ws_frame.hpp +++ b/include/crossbook/net/ws_frame.hpp @@ -33,6 +33,8 @@ #include #include +#include "crossbook/net/byte_buffer.hpp" + namespace crossbook::net { /// RFC 6455 §5.2 opcodes. @@ -531,8 +533,12 @@ class FrameReader { static constexpr std::size_t kCompactThreshold = 32 * 1024; std::size_t max_message_bytes_; - std::vector buf_; - std::vector assembled_; + // ByteBuffer, not std::vector: `writable_tail` grows this by a + // 32 KiB read chunk on every transport read and `commit` trims it back, + // so a value-initializing resize would memset 32 KiB per read for bytes + // recv is about to overwrite. + ByteBuffer buf_; + ByteBuffer assembled_; std::array control_{}; std::size_t control_len_{0}; std::size_t read_pos_{0}; diff --git a/include/crossbook/venues/binance_sbe.hpp b/include/crossbook/venues/binance_sbe.hpp new file mode 100644 index 0000000..5328465 --- /dev/null +++ b/include/crossbook/venues/binance_sbe.hpp @@ -0,0 +1,312 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Josh Dardashti +// +// Binance spot SBE stream decoder — the binary wire, schema spot_stream 1:0. +// +// Binance has served its market data streams in Simple Binary Encoding since +// March 2025 (stream-sbe.binance.com, an Ed25519 API key required even for +// public data), and it is the only major spot venue with a binary L2 diff +// feed. The point of consuming it is blunt: the JSON decode that costs ~2 us +// and 75% of a frame becomes a handful of bounds-checked little-endian loads. +// +// This is a hand-rolled reader for one schema, not an SBE framework. The +// schema (sbe/schemas/stream_1_0.xml in binance-spot-api-docs) is four +// messages over fixed-width fields, two repeating groups, and one trailing +// varString8. What is honoured from the SBE spec: header blockLengths are +// trusted for forward compatibility, so a minor-version schema that appends +// root or entry fields skips cleanly rather than desyncing. +// +// Prices and quantities arrive as (int64 mantissa, int8 exponent) pairs. They +// are rescaled to the instrument's configured scales with the same refusal +// semantics as the text path: a value that does not land exactly on the +// instrument's grid is kPrecisionLoss, never rounded. + +#pragma once + +#include +#include +#include +#include +#include +#include + +#include "crossbook/fixed.hpp" +#include "crossbook/sequence.hpp" +#include "crossbook/venue.hpp" + +namespace crossbook::venues { + +class BinanceSbeDecoder { + // The wire is little-endian and the loads below are memcpy-then-use. + // Every platform this project builds for is little-endian; if that ever + // changes, this line is the compile error that says where to add swaps. + static_assert(std::endian::native == std::endian::little, + "BinanceSbeDecoder assumes a little-endian host"); + +public: + explicit BinanceSbeDecoder(InstrumentSpec spec) : spec_(std::move(spec)) { + message_.levels.reserve(256); + } + + [[nodiscard]] const InstrumentSpec& spec() const noexcept { return spec_; } + + /// SBE streams exist for spot only; the spot sequencing rules apply. + [[nodiscard]] SequencePolicy policy() const noexcept { return SequencePolicy::kBinanceSpot; } + + [[nodiscard]] const DecodedMessage& decode(std::string_view frame) { + message_.reset(); + cursor_ = 0; + frame_ = frame; + + std::uint16_t block_length = 0; + std::uint16_t template_id = 0; + std::uint16_t schema_id = 0; + std::uint16_t version = 0; + if (!read_u16(block_length) || !read_u16(template_id) || !read_u16(schema_id) || + !read_u16(version)) { + return fail(DecodeError::kMalformed); + } + // A different schema id is a different wire format, and pretending to + // read it would produce plausible garbage. Loudly malformed, so the + // operator learns the venue rolled the schema. + if (schema_id != kSchemaId) { + return fail(DecodeError::kMalformed); + } + + switch (template_id) { + case kDepthDiffTemplate: + return decode_depth(block_length, /*diff=*/true); + case kDepthSnapshotTemplate: + return decode_depth(block_length, /*diff=*/false); + default: + // Trades and bestBidAsk share the socket legitimately; a book + // decoder ignoring them is routing, not failure. + message_.kind = MessageKind::kIgnored; + return message_; + } + } + +private: + static constexpr std::uint16_t kSchemaId = 1; + static constexpr std::uint16_t kDepthSnapshotTemplate = 10002; + static constexpr std::uint16_t kDepthDiffTemplate = 10003; + /// int64 nanoseconds run out in April 2262; a microsecond event time past + /// this cannot be scaled without overflow and is treated as absent, the + /// same bound (in the venue's unit) the JSON decoder applies. + static constexpr std::int64_t kMaxEventUs = 9'223'372'036'854'775LL; + + [[nodiscard]] const DecodedMessage& decode_depth(std::uint16_t block_length, bool diff) { + const std::size_t root_start = cursor_; + const std::size_t min_block = diff ? 26U : 18U; + if (block_length < min_block || frame_.size() - cursor_ < block_length) { + return fail(DecodeError::kMalformed); + } + + std::int64_t event_us = 0; + std::int64_t first_id = 0; + std::int64_t last_id = 0; + std::int8_t price_exp = 0; + std::int8_t qty_exp = 0; + (void)read_i64(event_us); + if (diff) { + (void)read_i64(first_id); + (void)read_i64(last_id); + } else { + (void)read_i64(last_id); + first_id = last_id; + } + (void)read_i8(price_exp); + (void)read_i8(qty_exp); + cursor_ = root_start + block_length; // Skip fields a newer minor added. + + if (first_id < 0 || last_id < 0) { + return fail(DecodeError::kBadSequence); + } + + // The symbol trails the groups on the wire, but routing must happen + // before any level is trusted. Group byte counts are arithmetic, so + // peek ahead without decoding an entry. + std::size_t peek = cursor_; + std::string_view symbol; + std::uint32_t bid_count = 0; + std::uint32_t ask_count = 0; + std::uint16_t bid_stride = 0; + std::uint16_t ask_stride = 0; + if (!peek_group(peek, bid_stride, bid_count) || !peek_group(peek, ask_stride, ask_count) || + !peek_symbol(peek, symbol)) { + return fail(DecodeError::kMalformed); + } + message_.symbol = symbol; + if (!symbol.empty() && !equals_ignore_case(symbol, spec_.symbol)) { + message_.kind = MessageKind::kIgnored; + message_.symbol = {}; + return message_; + } + + const std::uint64_t total = + static_cast(bid_count) + static_cast(ask_count); + if (total > kMaxLevelsPerMessage) { + return fail(DecodeError::kTooManyLevels); + } + + message_.kind = diff ? MessageKind::kUpdate : MessageKind::kSnapshot; + if (event_us > 0 && event_us <= kMaxEventUs) { + message_.ts = static_cast(event_us) * 1'000; + } + message_.ids.first_id = static_cast(first_id); + message_.ids.final_id = static_cast(last_id); + message_.has_ids = true; + + if (!decode_group(bid_stride, bid_count, price_exp, qty_exp, Side::kBid) || + !decode_group(ask_stride, ask_count, price_exp, qty_exp, Side::kAsk)) { + return message_; // decode_group set the error. + } + return message_; + } + + /// Group dimensions from `at`, advancing it past the whole group. + [[nodiscard]] bool peek_group(std::size_t& at, std::uint16_t& stride, + std::uint32_t& count) const noexcept { + std::uint16_t raw_count = 0; + if (frame_.size() - at < 4) { + return false; + } + std::memcpy(&stride, frame_.data() + at, 2); + std::memcpy(&raw_count, frame_.data() + at + 2, 2); + at += 4; + count = raw_count; + if (stride < 16) { + return false; // An entry is at least two int64s in every version. + } + const std::uint64_t bytes = static_cast(stride) * count; + if (frame_.size() - at < bytes) { + return false; + } + at += static_cast(bytes); + return true; + } + + [[nodiscard]] bool peek_symbol(std::size_t at, std::string_view& out) const noexcept { + if (at >= frame_.size()) { + return false; + } + const auto len = static_cast(frame_[at]); + if (frame_.size() - at - 1 < len) { + return false; + } + out = frame_.substr(at + 1, len); + return true; + } + + [[nodiscard]] bool decode_group(std::uint16_t stride, std::uint32_t count, + std::int8_t price_exp, std::int8_t qty_exp, Side side) { + cursor_ += 4; // The dimensions peek_group already validated. + for (std::uint32_t i = 0; i < count; ++i) { + std::int64_t price_mantissa = 0; + std::int64_t qty_mantissa = 0; + std::memcpy(&price_mantissa, frame_.data() + cursor_, 8); + std::memcpy(&qty_mantissa, frame_.data() + cursor_ + 8, 8); + cursor_ += stride; + + LevelUpdate level{}; + level.side = side; + if (price_mantissa < 0 || qty_mantissa < 0) { + (void)fail(DecodeError::kMalformed); + return false; + } + if (!rescale(price_mantissa, price_exp, spec_.price_scale, level.price.ticks) || + !rescale(qty_mantissa, qty_exp, spec_.qty_scale, level.qty.units)) { + (void)fail(DecodeError::kPrecisionLoss); + return false; + } + message_.levels.push_back(level); + } + return true; + } + + /// mantissa x 10^exponent, re-expressed as an integer at `scale`. + /// Exact or refused: the same contract parse_fixed enforces for text. + [[nodiscard]] static bool rescale(std::int64_t mantissa, std::int8_t exponent, Scale scale, + std::int64_t& out) noexcept { + const int shift = static_cast(scale) + static_cast(exponent); + if (shift == 0) { + out = mantissa; + return true; + } + if (shift > 0) { + if (shift > static_cast(kMaxScale)) { + return false; + } + return detail::checked_mul(mantissa, kPow10[static_cast(shift)], out); + } + if (-shift > static_cast(kMaxScale)) { + if (mantissa == 0) { + out = 0; + return true; + } + return false; + } + const std::int64_t divisor = kPow10[static_cast(-shift)]; + if (mantissa % divisor != 0) { + return false; + } + out = mantissa / divisor; + return true; + } + + [[nodiscard]] const DecodedMessage& fail(DecodeError error) { + message_.kind = MessageKind::kIgnored; + message_.error = error; + message_.levels.clear(); + return message_; + } + + [[nodiscard]] bool read_u16(std::uint16_t& out) noexcept { + if (frame_.size() - cursor_ < 2) { + return false; + } + std::memcpy(&out, frame_.data() + cursor_, 2); + cursor_ += 2; + return true; + } + + [[nodiscard]] bool read_i64(std::int64_t& out) noexcept { + if (frame_.size() - cursor_ < 8) { + return false; + } + std::memcpy(&out, frame_.data() + cursor_, 8); + cursor_ += 8; + return true; + } + + [[nodiscard]] bool read_i8(std::int8_t& out) noexcept { + if (frame_.size() - cursor_ < 1) { + return false; + } + std::memcpy(&out, frame_.data() + cursor_, 1); + cursor_ += 1; + return true; + } + + [[nodiscard]] static bool equals_ignore_case(std::string_view a, std::string_view b) noexcept { + if (a.size() != b.size()) { + return false; + } + for (std::size_t i = 0; i < a.size(); ++i) { + const auto lower = [](char c) { + return (c >= 'A' && c <= 'Z') ? static_cast(c - 'A' + 'a') : c; + }; + if (lower(a[i]) != lower(b[i])) { + return false; + } + } + return true; + } + + DecodedMessage message_; + InstrumentSpec spec_; + std::string_view frame_; + std::size_t cursor_{0}; +}; + +} // namespace crossbook::venues diff --git a/src/net/tcp_socket.cpp b/src/net/tcp_socket.cpp index 35292ad..f0f80d2 100644 --- a/src/net/tcp_socket.cpp +++ b/src/net/tcp_socket.cpp @@ -14,6 +14,7 @@ #include // FormatMessageA / LocalFree; must follow winsock2.h. // clang-format on #else +#include #include #include #include @@ -21,6 +22,11 @@ #include #include #include +#ifdef __linux__ +#include // SOF_TIMESTAMPING_*: the API that works for TCP. + +#include // memcpy out of CMSG_DATA, which is not suitably aligned. +#endif // Linux suppresses SIGPIPE per-send with MSG_NOSIGNAL; macOS and the BSDs do it // per-socket with SO_NOSIGPIPE and do not define the flag at all. Without one of @@ -171,6 +177,18 @@ bool TcpSocket::connect(const std::string& host, std::uint16_t port, int timeout (void)::setsockopt(fd_, SOL_SOCKET, SO_NOSIGPIPE, &one, static_cast(sizeof(one))); #endif +#ifdef __linux__ + // Software receive timestamps, via the one API that covers TCP. The + // venerable SO_TIMESTAMPNS is silently a no-op for SOCK_STREAM, and + // SIOCGSTAMP is explicitly unsupported there — both were tried. Note the + // kernel arms packet stamping through a deferred static key, so the very + // first arrivals after connect can legitimately carry no stamp; consumers + // treat 0 as "not yet", not as an error. + const int stamping = SOF_TIMESTAMPING_RX_SOFTWARE | SOF_TIMESTAMPING_SOFTWARE; + (void)::setsockopt(fd_, SOL_SOCKET, SO_TIMESTAMPING, &stamping, + static_cast(sizeof(stamping))); +#endif + set_read_timeout(timeout_ms); error.clear(); @@ -181,7 +199,18 @@ void TcpSocket::set_read_timeout(int timeout_ms) noexcept { if (fd_ == kInvalidSocket) { return; } + + // Zero is the busy-poll contract (see the header): non-blocking mode, so a + // read with nothing buffered comes back kTimeout immediately and the + // caller's spin owns the waiting. The platforms' native meaning of a zero + // SO_RCVTIMEO — block forever — is never what this client wants. + const bool busy_poll = (timeout_ms == 0); #ifdef _WIN32 + u_long nonblocking = busy_poll ? 1 : 0; + (void)::ioctlsocket(fd_, FIONBIO, &nonblocking); + if (busy_poll) { + return; + } // Windows takes the timeout as a DWORD of milliseconds. auto ms = static_cast(timeout_ms); (void)::setsockopt(fd_, SOL_SOCKET, SO_RCVTIMEO, reinterpret_cast(&ms), @@ -189,6 +218,13 @@ void TcpSocket::set_read_timeout(int timeout_ms) noexcept { (void)::setsockopt(fd_, SOL_SOCKET, SO_SNDTIMEO, reinterpret_cast(&ms), static_cast(sizeof(ms))); #else + const int flags = ::fcntl(fd_, F_GETFL, 0); + if (flags >= 0) { + (void)::fcntl(fd_, F_SETFL, busy_poll ? (flags | O_NONBLOCK) : (flags & ~O_NONBLOCK)); + } + if (busy_poll) { + return; + } ::timeval tv{}; tv.tv_sec = timeout_ms / 1000; tv.tv_usec = static_cast((timeout_ms % 1000) * 1000); @@ -197,6 +233,14 @@ void TcpSocket::set_read_timeout(int timeout_ms) noexcept { #endif } +std::int64_t TcpSocket::last_rx_time_ns() const noexcept { + // Populated by the Linux recvmsg path; permanently 0 on Windows and macOS, + // where no per-segment TCP receive clock exists worth pretending about — + // the measurement starts in user space there, and saying so beats + // fabricating a number. + return last_rx_ns_; +} + IoStatus TcpSocket::read(char* buf, std::size_t len, std::size_t& n_read, std::string& error) { if (fd_ == kInvalidSocket) { error = "read: socket not connected"; @@ -205,6 +249,38 @@ IoStatus TcpSocket::read(char* buf, std::size_t len, std::size_t& n_read, std::s #ifdef _WIN32 const int got = ::recv(fd_, buf, static_cast(len), 0); +#elif defined(__linux__) + // recvmsg rather than recv, because the kernel's arrival timestamp for a + // TCP socket exists ONLY as a control message riding the read that + // delivers the data. The SIOCGSTAMP query-after-the-fact ioctl is not + // supported for SOCK_STREAM, a fact this file learned empirically. + ::iovec iov{}; + iov.iov_base = buf; + iov.iov_len = len; + union { + ::cmsghdr align; + char raw[CMSG_SPACE(3 * sizeof(::timespec))]; // SCM_TIMESTAMPING is timespec[3]. + } ctrl{}; + ::msghdr msg{}; + msg.msg_iov = &iov; + msg.msg_iovlen = 1; + msg.msg_control = ctrl.raw; + msg.msg_controllen = sizeof(ctrl.raw); + const auto got = ::recvmsg(fd_, &msg, 0); + if (got > 0) { + for (::cmsghdr* c = CMSG_FIRSTHDR(&msg); c != nullptr; c = CMSG_NXTHDR(&msg, c)) { + if (c->cmsg_level == SOL_SOCKET && c->cmsg_type == SCM_TIMESTAMPING) { + // timespec[3]: software, deprecated, hardware. Software is the + // one a plain NIC fills. + ::timespec ts[3]{}; + std::memcpy(&ts, CMSG_DATA(c), sizeof(ts)); + if (ts[0].tv_sec != 0 || ts[0].tv_nsec != 0) { + last_rx_ns_ = static_cast(ts[0].tv_sec) * 1'000'000'000 + + static_cast(ts[0].tv_nsec); + } + } + } + } #else const auto got = ::recv(fd_, buf, len, 0); #endif diff --git a/src/net/tcp_socket.hpp b/src/net/tcp_socket.hpp index 20718ed..5bece30 100644 --- a/src/net/tcp_socket.hpp +++ b/src/net/tcp_socket.hpp @@ -64,8 +64,26 @@ class TcpSocket { std::string& error); /// Change the receive timeout on a connected socket. + /// + /// Zero means BUSY-POLL, not block-forever: the socket goes non-blocking + /// and `read` returns kTimeout immediately when nothing is buffered. Both + /// platforms natively treat a zero SO_RCVTIMEO as "no timeout", which is + /// the opposite of what a latency-sensitive caller ever wants; the caller + /// who really wants to block forever can pass a day in milliseconds. void set_read_timeout(int timeout_ms) noexcept; + /// Kernel arrival time of the most recently delivered data, in + /// CLOCK_REALTIME nanoseconds. 0 where the platform has no such clock for + /// TCP (Windows, macOS) or before anything has arrived. + /// + /// Captured from the control message riding each recvmsg — for TCP the + /// kernel offers the stamp no other way (SIOCGSTAMP is unsupported for + /// SOCK_STREAM). A TLS backend therefore only gets timestamps if its + /// reads come through this class rather than the raw fd. One read that + /// drains several coalesced segments carries one stamp — an approximation + /// the measurement consumer must own. + [[nodiscard]] std::int64_t last_rx_time_ns() const noexcept; + /// Write all of `len`, looping over partial sends. [[nodiscard]] IoStatus write(const char* buf, std::size_t len, std::string& error); @@ -76,6 +94,7 @@ class TcpSocket { private: SocketHandle fd_{kInvalidSocket}; + std::int64_t last_rx_ns_{0}; }; } // namespace crossbook::net::detail diff --git a/src/net/tls_openssl.cpp b/src/net/tls_openssl.cpp index 8c8ee8a..645f544 100644 --- a/src/net/tls_openssl.cpp +++ b/src/net/tls_openssl.cpp @@ -3,9 +3,19 @@ // // TLS via OpenSSL — the POSIX backend. // -// Shorter than its Schannel counterpart for one reason: OpenSSL will own the -// file descriptor and do its own reading, so there is no ciphertext buffer to -// manage and no SECBUFFER_EXTRA to get wrong. +// Shorter than its Schannel counterpart for one reason: OpenSSL does its own +// record buffering, so there is no ciphertext buffer to manage and no +// SECBUFFER_EXTRA to get wrong. +// +// OPENSSL DOES NOT OWN THE FD. Reads and writes go through TcpSocket via a +// custom BIO, for a concrete reason: the kernel's arrival timestamp for a TCP +// socket exists only as a control message on the recvmsg that delivers the +// data (SIOCGSTAMP is unsupported for SOCK_STREAM), so a backend that lets +// OpenSSL call recv() itself can never report when a message actually hit the +// machine. The BIO also keeps the busy-poll contract in one place: TcpSocket +// maps EWOULDBLOCK to kTimeout, the BIO maps kTimeout to a retry flag, and +// SSL_read surfaces it as WANT_READ exactly as it did with a timed-out +// blocking read. // // HOSTNAME VERIFICATION IS EXPLICIT. `SSL_CTX_set_verify` alone validates the // chain but not the name, so a certificate legitimately issued for any host at @@ -44,7 +54,36 @@ namespace { return out; } +class OpenSslTransport; + +/// The BIO glue. `BIO_get_data` carries the transport; the callbacks translate +/// between IoStatus and the retry-flag protocol OpenSSL expects. +struct BioGlue { + static int read(::BIO* bio, char* buf, int len); + static int write(::BIO* bio, const char* buf, int len); + static long ctrl(::BIO*, int cmd, long, void*) { + // FLUSH must claim success or SSL_write loops; everything else is for + // file-ish BIOs and does not apply to a socket. + return cmd == BIO_CTRL_FLUSH ? 1 : 0; + } + + /// One method table for the process, built on first use. + [[nodiscard]] static ::BIO_METHOD* method() { + static ::BIO_METHOD* m = [] { + ::BIO_METHOD* built = + ::BIO_meth_new(BIO_TYPE_SOURCE_SINK | ::BIO_get_new_index(), "crossbook-tcp"); + (void)::BIO_meth_set_read(built, &BioGlue::read); + (void)::BIO_meth_set_write(built, &BioGlue::write); + (void)::BIO_meth_set_ctrl(built, &BioGlue::ctrl); + return built; + }(); + return m; + } +}; + class OpenSslTransport final : public Transport { + friend struct BioGlue; + public: ~OpenSslTransport() override { close(); } @@ -100,11 +139,16 @@ class OpenSslTransport final : public Transport { return false; } - if (::SSL_set_fd(ssl_, static_cast(socket_.handle())) != 1) { - error_ = openssl_error("SSL_set_fd"); + ::BIO* bio = ::BIO_new(BioGlue::method()); + if (bio == nullptr) { + error_ = openssl_error("BIO_new"); close(); return false; } + ::BIO_set_data(bio, this); + ::BIO_set_init(bio, 1); + // One BIO for both directions; SSL_set_bio takes ownership. + ::SSL_set_bio(ssl_, bio, bio); const int rc = ::SSL_connect(ssl_); if (rc != 1) { @@ -168,6 +212,10 @@ class OpenSslTransport final : public Transport { void set_read_timeout(int timeout_ms) override { socket_.set_read_timeout(timeout_ms); } + [[nodiscard]] std::int64_t last_rx_time_ns() const noexcept override { + return socket_.last_rx_time_ns(); + } + void close() override { if (ssl_ != nullptr) { // Best-effort close_notify. A venue that has already gone away makes @@ -208,7 +256,11 @@ class OpenSslTransport final : public Transport { return IoStatus::kClosed; // Clean EOF without close_notify. } connected_ = false; - error_ = socket_error_string("SSL_ERROR_SYSCALL"); + if (error_.empty()) { + // The BIO usually recorded the real failure already; the + // errno-based string is the fallback, not an overwrite. + error_ = socket_error_string("SSL_ERROR_SYSCALL"); + } return IoStatus::kError; default: connected_ = false; @@ -224,6 +276,46 @@ class OpenSslTransport final : public Transport { bool connected_{false}; }; +int BioGlue::read(::BIO* bio, char* buf, int len) { + auto* self = static_cast(::BIO_get_data(bio)); + ::BIO_clear_retry_flags(bio); + if (self == nullptr || len <= 0) { + return -1; + } + std::size_t got = 0; + switch (self->socket_.read(buf, static_cast(len), got, self->error_)) { + case IoStatus::kOk: + return static_cast(got); + case IoStatus::kTimeout: + ::BIO_set_retry_read(bio); + return -1; + case IoStatus::kClosed: + return 0; // EOF; SSL_read reports ZERO_RETURN or SYSCALL rc==0. + case IoStatus::kError: + return -1; + } + return -1; +} + +int BioGlue::write(::BIO* bio, const char* buf, int len) { + auto* self = static_cast(::BIO_get_data(bio)); + ::BIO_clear_retry_flags(bio); + if (self == nullptr || len <= 0) { + return -1; + } + switch (self->socket_.write(buf, static_cast(len), self->error_)) { + case IoStatus::kOk: + return len; // TcpSocket::write is all-or-fail. + case IoStatus::kTimeout: + ::BIO_set_retry_write(bio); + return -1; + case IoStatus::kClosed: + case IoStatus::kError: + return -1; + } + return -1; +} + } // namespace std::unique_ptr make_tls_transport() { return std::make_unique(); } diff --git a/src/net/tls_schannel.cpp b/src/net/tls_schannel.cpp index e33e8b7..ebf00d5 100644 --- a/src/net/tls_schannel.cpp +++ b/src/net/tls_schannel.cpp @@ -51,6 +51,7 @@ #include // clang-format on +#include "crossbook/net/byte_buffer.hpp" #include "crossbook/net/transport.hpp" #include "tcp_socket.hpp" #include "tls_backend.hpp" @@ -176,12 +177,8 @@ class SchannelTransport final : public Transport { // the tail of the handshake often arrives with application data // behind it in the same segment. if (!enc_.empty()) { - const IoStatus status = decrypt_buffered(); + const IoStatus status = decrypt_buffered(buf, len, n_read); if (status == IoStatus::kOk) { - const std::size_t take = (std::min)(len, plain_.size()); - std::memcpy(buf, plain_.data(), take); - plain_pos_ = take; - n_read = take; return IoStatus::kOk; } if (status != IoStatus::kTimeout) { @@ -265,6 +262,10 @@ class SchannelTransport final : public Transport { void set_read_timeout(int timeout_ms) override { socket_.set_read_timeout(timeout_ms); } + [[nodiscard]] std::int64_t last_rx_time_ns() const noexcept override { + return socket_.last_rx_time_ns(); + } + void close() override { if (have_ctx_) { (void)::DeleteSecurityContext(&ctx_); @@ -440,11 +441,17 @@ class SchannelTransport final : public Transport { enc_.resize(count); } - /// Decrypt one record out of `enc_` into `plain_`. + /// Decrypt one record out of `enc_`, straight into the caller's buffer. + /// + /// DecryptMessage already decrypts in place inside `enc_`, so the only + /// copy this path needs is the one into `dst`. Plaintext beyond `len` + /// goes to `plain_` for the next call to serve; with a 16 KiB TLS record + /// ceiling and the frame reader asking for 32 KiB chunks, that overflow + /// path is reachable only for callers reading less than a record. /// /// Returns kTimeout to mean "incomplete record, read more" — the caller's /// loop treats it as such, and no other status can express it. - [[nodiscard]] IoStatus decrypt_buffered() { + [[nodiscard]] IoStatus decrypt_buffered(char* dst, std::size_t len, std::size_t& n_read) { SecBuffer buffers[4]{}; buffers[0].BufferType = SECBUFFER_DATA; buffers[0].cbBuffer = static_cast(enc_.size()); @@ -490,23 +497,38 @@ class SchannelTransport final : public Transport { return IoStatus::kError; } + // Copy plaintext out BEFORE touching the ciphertext buffer: both the + // data span and the extra span point into `enc_` itself, and + // `take_extra`'s memmove overwrites the front of it. + std::size_t produced = 0; const SecBuffer* data = find_buffer(buffers, SECBUFFER_DATA); if (data != nullptr && data->cbBuffer != 0) { const char* p = static_cast(data->pvBuffer); - plain_.assign(p, p + data->cbBuffer); + const std::size_t cb = data->cbBuffer; + produced = (std::min)(len, cb); + std::memcpy(dst, p, produced); + if (cb > produced) { + plain_.assign(p + produced, p + cb); + plain_pos_ = 0; + } } + // take_extra memmoves the unconsumed tail to the front of `enc_` in + // place — a fresh vector per pipelined record was an allocation on + // the common path, since a busy feed routinely lands the next record + // behind the current one in the same segment. const SecBuffer* extra = find_buffer(buffers, SECBUFFER_EXTRA); - if (extra != nullptr && extra->cbBuffer != 0) { - // Copy before shrinking: the extra span points into `enc_` itself. - std::vector rest(static_cast(extra->pvBuffer), - static_cast(extra->pvBuffer) + extra->cbBuffer); - enc_.swap(rest); + if (extra != nullptr) { + take_extra(*extra); } else { enc_.clear(); } - return plain_.empty() ? IoStatus::kTimeout : IoStatus::kOk; + if (produced == 0) { + return IoStatus::kTimeout; // A record with no app data; loop. + } + n_read = produced; + return IoStatus::kOk; } /// First buffer of a given type, skipping the one we handed in as input. @@ -525,7 +547,10 @@ class SchannelTransport final : public Transport { CtxtHandle ctx_{}; SecPkgContext_StreamSizes sizes_{}; std::string target_name_; - std::vector enc_; + // ByteBuffer, not std::vector: the read loops grow this by a 32 KiB + // chunk per socket read and trim it back to what arrived, and a + // value-initializing resize would memset the chunk every time. + ByteBuffer enc_; std::vector plain_; std::vector send_buf_; std::string error_; diff --git a/src/net/transport.cpp b/src/net/transport.cpp index f4f28a4..bb8b2d2 100644 --- a/src/net/transport.cpp +++ b/src/net/transport.cpp @@ -49,6 +49,10 @@ class PlainTransport final : public Transport { void set_read_timeout(int timeout_ms) override { socket_.set_read_timeout(timeout_ms); } + [[nodiscard]] std::int64_t last_rx_time_ns() const noexcept override { + return socket_.last_rx_time_ns(); + } + void close() override { socket_.close(); connected_ = false; diff --git a/src/net/websocket.cpp b/src/net/websocket.cpp index 077fd0f..755bfc3 100644 --- a/src/net/websocket.cpp +++ b/src/net/websocket.cpp @@ -49,6 +49,16 @@ bool WebSocketClient::connected() const noexcept { return open_ && transport_ != nullptr && transport_->connected(); } +void WebSocketClient::set_read_timeout(int timeout_ms) { + if (transport_ != nullptr) { + transport_->set_read_timeout(timeout_ms); + } +} + +std::int64_t WebSocketClient::last_rx_time_ns() const noexcept { + return transport_ != nullptr ? transport_->last_rx_time_ns() : 0; +} + std::uint32_t WebSocketClient::next_mask_key() { // mt19937 yields 32 bits per call, which is exactly a masking key. return static_cast(rng_()); diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 026c84f..36da635 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -23,6 +23,7 @@ add_executable(crossbook_tests test_histogram.cpp test_json.cpp test_venues.cpp + test_binance_sbe.cpp test_feed.cpp test_replay.cpp test_execution.cpp @@ -48,6 +49,20 @@ target_link_libraries(crossbook_tests PRIVATE Catch2::Catch2WithMain ) +# The transport is only built when a TLS backend exists, so its loopback tests +# come and go with it. They test TcpSocket directly — an internal header, which +# is why the include path reaches into src/net rather than the public tree. +# Deferred to the end of the top-level configure because this directory is +# processed before src/net, so `TARGET crossbook_net` is not knowable yet here. +function(crossbook_add_transport_tests) + if(TARGET crossbook_net) + target_sources(crossbook_tests PRIVATE ${PROJECT_SOURCE_DIR}/tests/test_transport.cpp) + target_link_libraries(crossbook_tests PRIVATE crossbook::net) + target_include_directories(crossbook_tests PRIVATE ${PROJECT_SOURCE_DIR}/src/net) + endif() +endfunction() +cmake_language(DEFER DIRECTORY ${PROJECT_SOURCE_DIR} CALL crossbook_add_transport_tests) + # Somewhere writable for the capture round-trip test. The build tree, so a test # run leaves nothing behind in the source tree or in a shared temp directory. target_compile_definitions(crossbook_tests PRIVATE diff --git a/tests/test_binance_sbe.cpp b/tests/test_binance_sbe.cpp new file mode 100644 index 0000000..a9b4958 --- /dev/null +++ b/tests/test_binance_sbe.cpp @@ -0,0 +1,274 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Josh Dardashti +// +// The Binance SBE depth decoder, against frames built byte-by-byte from the +// schema. No captured fixtures exist yet — the endpoint requires an Ed25519 +// API key even for public data — so the fixtures here are hand-encoded from +// stream_1_0.xml, which has the useful property that every offset in the +// decoder was reached by an independent reading of the same document. + +#include + +#include +#include +#include +#include + +#include "crossbook/feed.hpp" +#include "crossbook/venues/binance_sbe.hpp" + +using namespace crossbook; +using namespace crossbook::venues; + +namespace { + +/// Little-endian byte appender: the schema's byteOrder, spelled explicitly so +/// the fixture cannot silently inherit the host's. +struct Wire { + std::string bytes; + + void u8(std::uint8_t v) { bytes.push_back(static_cast(v)); } + void u16(std::uint16_t v) { + u8(static_cast(v & 0xFF)); + u8(static_cast(v >> 8)); + } + void i8(std::int8_t v) { u8(static_cast(v)); } + void i64(std::int64_t v) { + auto u = static_cast(v); + for (int i = 0; i < 8; ++i) { + u8(static_cast(u & 0xFF)); + u >>= 8; + } + } + void str8(std::string_view s) { + u8(static_cast(s.size())); + bytes.append(s); + } +}; + +struct Level { + std::int64_t price; + std::int64_t qty; +}; + +/// A DepthDiffStreamEvent (template 10003), schema 1:0. +std::string diff_frame(std::int64_t event_us, std::int64_t first, std::int64_t last, + std::int8_t price_exp, std::int8_t qty_exp, const std::vector& bids, + const std::vector& asks, std::string_view symbol = "BTCUSDT") { + Wire w; + w.u16(26); // blockLength + w.u16(10003); // templateId + w.u16(1); // schemaId + w.u16(0); // version + w.i64(event_us); + w.i64(first); + w.i64(last); + w.i8(price_exp); + w.i8(qty_exp); + for (const auto& side : {bids, asks}) { + w.u16(16); // group blockLength + w.u16(static_cast(side.size())); + for (const Level& level : side) { + w.i64(level.price); + w.i64(level.qty); + } + } + w.str8(symbol); + return w.bytes; +} + +/// A DepthSnapshotStreamEvent (template 10002), schema 1:0. +std::string snapshot_frame(std::int64_t event_us, std::int64_t book_id, std::int8_t price_exp, + std::int8_t qty_exp, const std::vector& bids, + const std::vector& asks, std::string_view symbol = "BTCUSDT") { + Wire w; + w.u16(18); + w.u16(10002); + w.u16(1); + w.u16(0); + w.i64(event_us); + w.i64(book_id); + w.i8(price_exp); + w.i8(qty_exp); + for (const auto& side : {bids, asks}) { + w.u16(16); + w.u16(static_cast(side.size())); + for (const Level& level : side) { + w.i64(level.price); + w.i64(level.qty); + } + } + w.str8(symbol); + return w.bytes; +} + +BinanceSbeDecoder btcusdt() { + return BinanceSbeDecoder(InstrumentSpec{"BTCUSDT", 2, 8}); +} + +} // namespace + +TEST_CASE("an SBE depth diff decodes to the instrument's grid", "[venues][binance][sbe]") { + auto decoder = btcusdt(); + // price 45283.50 spelled as 4528350e-2; qty 0.5 as 50000000e-8. + const std::string frame = diff_frame(1'700'000'000'000'000, 100, 105, -2, -8, + {{4'528'350, 50'000'000}}, {{4'528'360, 30'000'000}}); + + const DecodedMessage& msg = decoder.decode(frame); + REQUIRE(msg.ok()); + CHECK(msg.kind == MessageKind::kUpdate); + CHECK(msg.symbol == "BTCUSDT"); + CHECK(msg.ts == 1'700'000'000'000'000'000); // us in, ns out. + REQUIRE(msg.has_ids); + CHECK(msg.ids.first_id == 100); + CHECK(msg.ids.final_id == 105); + REQUIRE(msg.levels.size() == 2); + CHECK(msg.levels[0].side == Side::kBid); + CHECK(msg.levels[0].price.ticks == 4'528'350); + CHECK(msg.levels[0].qty.units == 50'000'000); + CHECK(msg.levels[1].side == Side::kAsk); + CHECK(msg.levels[1].price.ticks == 4'528'360); +} + +TEST_CASE("a coarser exponent rescales exactly", "[venues][binance][sbe]") { + // The venue is free to send 45283.5 as 452835e-1; at price scale 2 that is + // ticks 4528350 — an exact multiply, not a spelling problem, because the + // binary wire has no spelling. + auto decoder = btcusdt(); + const std::string frame = diff_frame(1, 100, 105, -1, -8, {{452'835, 50'000'000}}, {}); + const DecodedMessage& msg = decoder.decode(frame); + REQUIRE(msg.ok()); + REQUIRE(msg.levels.size() == 1); + CHECK(msg.levels[0].price.ticks == 4'528'350); +} + +TEST_CASE("a value off the instrument's grid is refused, not rounded", "[venues][binance][sbe]") { + // 45283.5001 at price scale 2 would need rounding. Same contract as the + // text path: kPrecisionLoss, and no levels survive. + auto decoder = btcusdt(); + const std::string frame = diff_frame(1, 100, 105, -4, -8, {{452'835'001, 50'000'000}}, {}); + const DecodedMessage& msg = decoder.decode(frame); + CHECK_FALSE(msg.ok()); + CHECK(msg.error == DecodeError::kPrecisionLoss); + CHECK(msg.levels.empty()); +} + +TEST_CASE("a foreign symbol is ignored before any level is trusted", "[venues][binance][sbe]") { + auto decoder = btcusdt(); + const std::string frame = + diff_frame(1, 100, 105, -2, -8, {{4'528'350, 50'000'000}}, {}, "ETHUSDT"); + const DecodedMessage& msg = decoder.decode(frame); + REQUIRE(msg.ok()); + CHECK(msg.kind == MessageKind::kIgnored); + CHECK(msg.levels.empty()); +} + +TEST_CASE("an SBE depth snapshot carries its book id as both sequence ends", + "[venues][binance][sbe]") { + auto decoder = btcusdt(); + const std::string frame = + snapshot_frame(1, 5000, -2, -8, {{4'528'350, 50'000'000}}, {{4'528'360, 30'000'000}}); + const DecodedMessage& msg = decoder.decode(frame); + REQUIRE(msg.ok()); + CHECK(msg.kind == MessageKind::kSnapshot); + CHECK(msg.ids.first_id == 5000); + CHECK(msg.ids.final_id == 5000); + REQUIRE(msg.levels.size() == 2); +} + +TEST_CASE("other templates on the socket are routing, not failure", "[venues][binance][sbe]") { + auto decoder = btcusdt(); + Wire w; + w.u16(34); // BestBidAskStreamEvent root: 8+8+1+1+8+8 exceeds; size irrelevant + w.u16(10001); // bestBidAsk template + w.u16(1); + w.u16(0); + const DecodedMessage& msg = decoder.decode(w.bytes); + REQUIRE(msg.ok()); + CHECK(msg.kind == MessageKind::kIgnored); +} + +TEST_CASE("a foreign schema id is loudly malformed", "[venues][binance][sbe]") { + auto decoder = btcusdt(); + Wire w; + w.u16(26); + w.u16(10003); + w.u16(2); // Not the schema this decoder reads. + w.u16(0); + const DecodedMessage& msg = decoder.decode(w.bytes); + CHECK_FALSE(msg.ok()); + CHECK(msg.error == DecodeError::kMalformed); +} + +TEST_CASE("truncation anywhere is malformed, never a partial decode", "[venues][binance][sbe]") { + auto decoder = btcusdt(); + const std::string whole = + diff_frame(1, 100, 105, -2, -8, {{4'528'350, 50'000'000}}, {{4'528'360, 30'000'000}}); + // Every prefix must refuse. The last byte index is the full frame, which + // must decode — pinning that the loop bound itself is right. + for (std::size_t cut = 0; cut < whole.size(); ++cut) { + const DecodedMessage& msg = decoder.decode(std::string_view(whole).substr(0, cut)); + INFO("cut=" << cut); + REQUIRE((msg.error == DecodeError::kMalformed || msg.kind == MessageKind::kIgnored)); + REQUIRE(msg.levels.empty()); + } + REQUIRE(decoder.decode(whole).ok()); +} + +TEST_CASE("a newer minor version's extra bytes are skipped, not misread", + "[venues][binance][sbe]") { + // Forward compatibility is the point of SBE blockLengths: a schema 1:1 + // that appends a root field and an entry field must still decode as 1:0. + Wire w; + w.u16(30); // Root grew by 4 bytes. + w.u16(10003); + w.u16(1); + w.u16(1); // version 1 + w.i64(1); // eventTime + w.i64(100); // first + w.i64(105); // last + w.i8(-2); + w.i8(-8); + w.u16(0xBEEF); // The appended root field this decoder has never heard of. + w.u16(0xDEAD); + w.u16(20); // Entries grew by 4 bytes too. + w.u16(1); + w.i64(4'528'350); + w.i64(50'000'000); + w.u16(0xAAAA); // Appended entry field. + w.u16(0xBBBB); + w.u16(20); // Empty ask group, same stride. + w.u16(0); + w.str8("BTCUSDT"); + + auto decoder = btcusdt(); + const DecodedMessage& msg = decoder.decode(w.bytes); + REQUIRE(msg.ok()); + CHECK(msg.kind == MessageKind::kUpdate); + REQUIRE(msg.levels.size() == 1); + CHECK(msg.levels[0].price.ticks == 4'528'350); + CHECK(msg.levels[0].qty.units == 50'000'000); +} + +TEST_CASE("the SBE decoder drives a feed end to end", "[venues][binance][sbe][feed]") { + // The decoder satisfies the same contract as its JSON sibling, so the + // whole recovery state machine comes for free — proven, not assumed. + Feed feed("binance-sbe", btcusdt(), SequencePolicy::kBinanceSpot); + + const std::string snap = + snapshot_frame(1, 100, -2, -8, {{4'528'350, 100'000'000}}, {{4'528'360, 100'000'000}}); + REQUIRE(feed.handle(snap) == FeedStatus::kApplied); + REQUIRE(feed.synced()); + + // Contiguous diff applies. + REQUIRE(feed.handle(diff_frame(2, 98, 105, -2, -8, {{4'528'340, 25'000'000}}, {})) == + FeedStatus::kApplied); + CHECK(feed.book().bids().size() == 2); + + // A gap stops the feed and keeps it stopped. + CHECK(feed.handle(diff_frame(3, 110, 112, -2, -8, {{4'528'330, 10'000'000}}, {})) == + FeedStatus::kNeedsSnapshot); + CHECK_FALSE(feed.synced()); + CHECK(feed.handle(diff_frame(4, 113, 114, -2, -8, {{4'528'320, 10'000'000}}, {})) == + FeedStatus::kNeedsSnapshot); +} diff --git a/tests/test_no_alloc.cpp b/tests/test_no_alloc.cpp index 44bf33f..b9f7652 100644 --- a/tests/test_no_alloc.cpp +++ b/tests/test_no_alloc.cpp @@ -20,9 +20,15 @@ #include #include +#include +#include + #include "crossbook/book.hpp" #include "crossbook/checksum.hpp" +#include "crossbook/feed.hpp" +#include "crossbook/net/ws_frame.hpp" #include "crossbook/sequence.hpp" +#include "crossbook/venues/kraken.hpp" // --------------------------------------------------------------------------- // Allocation probe @@ -312,6 +318,140 @@ TEST_CASE("book queries do not allocate", "[alloc][book]") { CHECK(sink != 0); } +TEST_CASE("Kraken decode does not allocate in steady state", "[alloc][decode]") { + // Decode is ~75% of the frame cost, and until this test existed it was + // exactly the part of "no allocation on the hot path" that nothing + // enforced. The book tests above guard 0.3% of the frame. + venues::KrakenBookDecoder decoder(InstrumentSpec{"BTC/USD", 1, 8}); + const std::string frame = + R"({"channel":"book","type":"update","data":[{"symbol":"BTC/USD",)" + R"("bids":[{"price":45283.5,"qty":0.50000000},{"price":45283.4,"qty":1.25000000}],)" + R"("asks":[{"price":45283.6,"qty":0.30000000},{"price":45283.7,"qty":2.00000000}],)" + R"("checksum":1234567890,"timestamp":"2026-07-31T12:00:00.000000Z"}]})"; + + // Warm-up outside the guard: the decoder's level vector is reserved at + // construction, but steady state is what the claim is about. + for (int i = 0; i < 100; ++i) { + REQUIRE(decoder.decode(frame).ok()); + } + + std::uint64_t observed = 0; + std::uint64_t sink = 0; + { + alloc_probe::Guard guard; + for (int i = 0; i < 10'000; ++i) { + const DecodedMessage& msg = decoder.decode(frame); + sink += msg.levels.size(); + } + observed = alloc_probe::Guard::count(); + } + CHECK(observed == 0); + CHECK(sink == 40'000); // Every decode saw all four levels. +} + +TEST_CASE("Feed::handle does not allocate on the applied path", "[alloc][feed]") { + // The full frame: decode, guards, sequence, book update, checksum verify. + // The rejection paths allocate deliberately (Divergence carries strings); + // the applied path must not. + using KrakenFeed = Feed; + KrakenFeed feed("kraken", venues::KrakenBookDecoder(InstrumentSpec{"BTC/USD", 1, 8}), + SequencePolicy::kStrictIncrement); + + // The checksum Kraken would publish for this two-level book. + ArrayBook reference(InstrumentSpec{"BTC/USD", 1, 8}); + reference.apply(Side::kAsk, Price{452836}, Qty{30'000'000}); + reference.apply(Side::kBid, Price{452835}, Qty{50'000'000}); + const std::uint32_t crc = kraken_checksum(reference); + + const std::string snapshot = + std::string(R"({"channel":"book","type":"snapshot","data":[{"symbol":"BTC/USD",)") + + R"("asks":[{"price":45283.6,"qty":0.30000000}],)" + + R"("bids":[{"price":45283.5,"qty":0.50000000}],)" + R"("checksum":)" + std::to_string(crc) + + R"(,"timestamp":"2026-07-31T12:00:00.000000Z")" + "}]}"; + REQUIRE(feed.handle(snapshot) == FeedStatus::kApplied); + + // An update that re-states the same levels leaves the book unchanged, so + // the same checksum stays correct on every iteration and the verify path + // runs armed. + const std::string update = + std::string(R"({"channel":"book","type":"update","data":[{"symbol":"BTC/USD",)") + + R"("bids":[{"price":45283.5,"qty":0.50000000}],)" + + R"("asks":[{"price":45283.6,"qty":0.30000000}],)" + R"("checksum":)" + std::to_string(crc) + + R"(,"timestamp":"2026-07-31T12:00:01.000000Z")" + "}]}"; + + for (int i = 0; i < 100; ++i) { + REQUIRE(feed.handle(update) == FeedStatus::kApplied); + } + const std::uint64_t verified_before = feed.stats().checksums_verified; + + std::uint64_t observed = 0; + { + alloc_probe::Guard guard; + for (int i = 0; i < 10'000; ++i) { + (void)feed.handle(update); + } + observed = alloc_probe::Guard::count(); + } + CHECK(observed == 0); + + // The probe only means something if the guarded region did the work it is + // aimed at: applied frames, with the checksum actually verified. + CHECK(feed.stats().checksums_verified == verified_before + 10'000); + CHECK(feed.stats().checksum_mismatches == 0); + CHECK(feed.synced()); +} + +TEST_CASE("the frame reader poll loop does not allocate in steady state", "[alloc][net]") { + // The transport hands bytes to writable_tail/commit and pulls messages out + // of next(); that loop runs once per socket read, ahead of everything the + // tests above enforce. + net::FrameReader reader; + + // A server-to-client (unmasked) text frame: FIN|text, 7-bit length. + const std::string payload = + R"({"channel":"book","type":"update","data":[{"symbol":"BTC/USD"}]})"; + REQUIRE(payload.size() < 126); + std::string wire; + wire.push_back(static_cast(0x81)); + wire.push_back(static_cast(payload.size())); + wire += payload; + + // No Catch2 macros inside the armed region: an assertion handler is + // allowed to allocate, and a probe that counts the harness is a probe + // that cries wolf. Count successes, assert after disarming. + auto pump_one = [&]() -> bool { + char* dst = reader.writable_tail(wire.size()); + std::memcpy(dst, wire.data(), wire.size()); + reader.commit(wire.size(), wire.size()); + net::Event event; + if (reader.next(event) != net::ReadStatus::kMessage) { + return false; + } + if (event.payload != payload) { + return false; + } + return reader.next(event) == net::ReadStatus::kNeedMore; + }; + + for (int i = 0; i < 100; ++i) { + REQUIRE(pump_one()); + } + + std::uint64_t observed = 0; + std::uint64_t pumped = 0; + { + alloc_probe::Guard guard; + for (int i = 0; i < 10'000; ++i) { + if (pump_one()) { + ++pumped; + } + } + observed = alloc_probe::Guard::count(); + } + CHECK(observed == 0); + CHECK(pumped == 10'000); +} + TEST_CASE("sequence tracking does not allocate", "[alloc][sequence]") { // The gap detector runs on every message, so it belongs to the hot path // just as much as the book does. diff --git a/tests/test_transport.cpp b/tests/test_transport.cpp new file mode 100644 index 0000000..982fdc2 --- /dev/null +++ b/tests/test_transport.cpp @@ -0,0 +1,198 @@ +// SPDX-License-Identifier: MIT +// Copyright (c) 2026 Josh Dardashti +// +// The socket, over loopback. +// +// The transport was the one layer with no tests at all: everything above it is +// a function of bytes, and the socket was quarantined precisely so the +// interesting code would not inherit its untestability. But the busy-poll +// contract — a zero read timeout means "return immediately", not "block +// forever" — is a behavioural cliff worth pinning: both platforms' native +// meaning of a zero SO_RCVTIMEO is the exact opposite. + +#include + +#include +#include +#include +#include + +#include "tcp_socket.hpp" + +#ifdef _WIN32 +// clang-format off +#include +#include +// clang-format on +#else +#include +#include +#include +#include +#endif + +using namespace crossbook::net; + +namespace { + +/// A one-connection loopback listener, raw enough to not depend on the code +/// under test. +class LoopbackListener { +public: + LoopbackListener() { +#ifdef _WIN32 + WSADATA wsa{}; + (void)::WSAStartup(MAKEWORD(2, 2), &wsa); +#endif + listen_fd_ = ::socket(AF_INET, SOCK_STREAM, 0); + ::sockaddr_in addr{}; + addr.sin_family = AF_INET; + // No `::` on htonl/ntohs: macOS defines them as macros, and a scope + // qualifier in front of a macro is a parse error. + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + addr.sin_port = 0; // Any free port. + (void)::bind(listen_fd_, reinterpret_cast<::sockaddr*>(&addr), + static_cast(sizeof(addr))); + (void)::listen(listen_fd_, 1); + detail::SockLen len = static_cast(sizeof(addr)); + (void)::getsockname(listen_fd_, reinterpret_cast<::sockaddr*>(&addr), &len); + port_ = ntohs(addr.sin_port); + } + + ~LoopbackListener() { + close_peer(); + if (listen_fd_ != detail::kInvalidSocket) { +#ifdef _WIN32 + ::closesocket(listen_fd_); +#else + ::close(listen_fd_); +#endif + } + } + + [[nodiscard]] std::uint16_t port() const noexcept { return port_; } + + /// Accept the pending connection. Call after the client's connect. + void accept_peer() { peer_fd_ = ::accept(listen_fd_, nullptr, nullptr); } + + void send_bytes(const char* data, std::size_t len) { +#ifdef _WIN32 + (void)::send(peer_fd_, data, static_cast(len), 0); +#else + (void)::send(peer_fd_, data, len, 0); +#endif + } + + void close_peer() { + if (peer_fd_ != detail::kInvalidSocket) { +#ifdef _WIN32 + ::closesocket(peer_fd_); +#else + ::close(peer_fd_); +#endif + peer_fd_ = detail::kInvalidSocket; + } + } + +private: + detail::SocketHandle listen_fd_{detail::kInvalidSocket}; + detail::SocketHandle peer_fd_{detail::kInvalidSocket}; + std::uint16_t port_{0}; +}; + +} // namespace + +TEST_CASE("a zero read timeout busy-polls instead of blocking", "[transport]") { + LoopbackListener listener; + REQUIRE(listener.port() != 0); + + detail::TcpSocket sock; + std::string error; + REQUIRE(sock.connect("127.0.0.1", listener.port(), 2'000, error)); + listener.accept_peer(); + + sock.set_read_timeout(0); + + // Nothing has been sent: the read must come back kTimeout immediately. + // Under the platforms' native meaning of a zero timeout this call would + // block forever and this test would hang rather than fail, which is why + // the elapsed-time bound is generous but real. + char buf[16]; + std::size_t got = 0; + const auto before = std::chrono::steady_clock::now(); + const IoStatus empty = sock.read(buf, sizeof(buf), got, error); + const auto elapsed = std::chrono::steady_clock::now() - before; + CHECK(empty == IoStatus::kTimeout); + CHECK(elapsed < std::chrono::milliseconds(200)); + + // Data pushed from the far side is picked up by the spin. + listener.send_bytes("hello", 5); + IoStatus status = IoStatus::kTimeout; + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2); + while (status == IoStatus::kTimeout && std::chrono::steady_clock::now() < deadline) { + status = sock.read(buf, sizeof(buf), got, error); + } + REQUIRE(status == IoStatus::kOk); + CHECK(got == 5); + CHECK(std::memcmp(buf, "hello", 5) == 0); +} + +TEST_CASE("a nonzero timeout still blocks and still delivers", "[transport]") { + // The busy-poll contract must not have broken the ordinary path. + LoopbackListener listener; + detail::TcpSocket sock; + std::string error; + REQUIRE(sock.connect("127.0.0.1", listener.port(), 2'000, error)); + listener.accept_peer(); + + sock.set_read_timeout(50); + char buf[16]; + std::size_t got = 0; + CHECK(sock.read(buf, sizeof(buf), got, error) == IoStatus::kTimeout); + + listener.send_bytes("book", 4); + sock.set_read_timeout(2'000); + REQUIRE(sock.read(buf, sizeof(buf), got, error) == IoStatus::kOk); + CHECK(got == 4); +} + +TEST_CASE("the kernel receive timestamp is exposed where the platform has one", "[transport]") { + LoopbackListener listener; + detail::TcpSocket sock; + std::string error; + REQUIRE(sock.connect("127.0.0.1", listener.port(), 2'000, error)); + listener.accept_peer(); + + // Before any data: no timestamp to report. + CHECK(sock.last_rx_time_ns() == 0); + +#ifdef __linux__ + // The kernel arms packet stamping through a deferred static key, so a + // segment sent immediately after connect can legitimately arrive without a + // stamp. Real captures run for seconds and never notice; a test that + // sends within microseconds of connect must wait the key out. + std::this_thread::sleep_for(std::chrono::milliseconds(300)); +#endif + listener.send_bytes("stamp", 5); + sock.set_read_timeout(2'000); + char buf[16]; + std::size_t got = 0; + REQUIRE(sock.read(buf, sizeof(buf), got, error) == IoStatus::kOk); + +#ifdef __linux__ + // Linux reports the kernel's arrival time for the last delivered segment. + // It is a CLOCK_REALTIME value; sanity-bound it against the same clock. + const std::int64_t rx = sock.last_rx_time_ns(); + REQUIRE(rx != 0); + const std::int64_t now = std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()) + .count(); + CHECK(now - rx < 5'000'000'000); // Within 5 s of now. + CHECK(now - rx > -1'000'000'000); // And not from the future. +#else + // Windows and macOS have no per-segment receive timestamp for TCP that is + // worth pretending about; the accessor reports 0 and the measurement + // starts in user space, as documented. + CHECK(sock.last_rx_time_ns() == 0); +#endif +} diff --git a/tools/crossbook_capture.cpp b/tools/crossbook_capture.cpp index 5522853..237130c 100644 --- a/tools/crossbook_capture.cpp +++ b/tools/crossbook_capture.cpp @@ -110,6 +110,8 @@ void print_usage() { " --url dial this URL instead of a venue preset\n" " --subscribe send this after connecting; use with --url\n" " --quiet suppress the periodic progress line\n" + " --busy-poll spin on the socket instead of sleeping in the kernel;\n" + " burns a core, removes the scheduler wakeup per message\n" " --help this text\n" "\n" "Examples:\n" @@ -143,6 +145,7 @@ int main(int argc, char** argv) { int depth = 10; int seconds = 30; bool quiet = false; + bool busy_poll = false; for (int i = 1; i < argc; ++i) { const std::string_view arg(argv[i]); @@ -173,6 +176,8 @@ int main(int argc, char** argv) { seconds = std::atoi(value("--seconds")); } else if (arg == "--quiet") { quiet = true; + } else if (arg == "--busy-poll") { + busy_poll = true; } else { std::fprintf(stderr, "error: unknown option %.*s\n", static_cast(arg.size()), arg.data()); @@ -222,6 +227,13 @@ int main(int argc, char** argv) { std::printf("recording to %s\n", out_path.c_str()); } + if (busy_poll) { + // After the subscribe: the handshake wants its generous timeout, the + // steady state wants no kernel sleep at all. + client.set_read_timeout(0); + std::printf("busy-poll: spinning on the socket, one core is now spoken for\n"); + } + const std::int64_t start = steady_ns(); const std::int64_t deadline = seconds > 0 ? start + static_cast(seconds) * 1'000'000'000LL : 0; @@ -229,6 +241,14 @@ int main(int argc, char** argv) { std::vector gaps; gaps.reserve(1 << 16); + // Kernel-to-user delivery: the gap between the kernel stamping the segment + // and this loop holding the decoded message. This is the number busy-poll + // exists to shrink, and it is only measurable where the transport exposes + // an arrival clock (Linux). + std::vector delivery; + delivery.reserve(1 << 16); + std::int64_t last_rx_seen = 0; + std::uint64_t frames = 0; std::uint64_t bytes = 0; std::int64_t previous = 0; @@ -256,6 +276,23 @@ int main(int argc, char** argv) { } previous = now; + // Only score a message against a socket read it plausibly came + // from: several messages can ride one read, and re-counting the + // same arrival stamp would flatter the tail. First message per + // fresh stamp only. + const std::int64_t rx = client.last_rx_time_ns(); + if (rx != 0 && rx != last_rx_seen) { + last_rx_seen = rx; + const std::int64_t user_now = + std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()) + .count(); + const std::int64_t delta = user_now - rx; + if (delta >= 0) { + delivery.push_back(delta); + } + } + if (writer.is_open() && !writer.write(now, event.payload)) { std::fprintf(stderr, "error: capture write failed\n"); exit_code = 1; @@ -310,6 +347,19 @@ int main(int argc, char** argv) { static_cast(percentile(gaps, 99)) / 1e6, static_cast(gaps.back()) / 1e6); } + if (!delivery.empty()) { + std::sort(delivery.begin(), delivery.end()); + // Kernel arrival stamp to message-in-hand, one sample per socket read. + // Includes TLS decryption and frame reassembly; the mode is what + // busy-poll changes. + std::printf("kernel-to-user p50 %.1f us p99 %.1f us max %.1f us (n=%zu%s)\n", + static_cast(percentile(delivery, 50)) / 1e3, + static_cast(percentile(delivery, 99)) / 1e3, + static_cast(delivery.back()) / 1e3, delivery.size(), + busy_poll ? ", busy-poll" : ""); + } else if (busy_poll) { + std::printf("kernel-to-user unavailable on this platform (no TCP arrival clock)\n"); + } if (writer.frames() > 0) { std::printf("capture %s (%llu frames, %.2f MiB)\n", out_path.c_str(), static_cast(writer.frames()),