From b08f4e111b858da233f49222d33730a56958f2e2 Mon Sep 17 00:00:00 2001 From: Joshua Dardashti <135767837+jdardash@users.noreply.github.com> Date: Sun, 2 Aug 2026 10:39:02 -0700 Subject: [PATCH 01/11] docs: latency roadmap, from the gap actually measured Four-track research pass: repo hot-path inventory, venue wire formats, software tick-to-trade practice, tail-latency engineering. Records the three-way split of the gap (environment tail, JSON ceiling, compute already competitive) and the phase order for closing it. --- LATENCY-ROADMAP.md | 125 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 125 insertions(+) create mode 100644 LATENCY-ROADMAP.md diff --git a/LATENCY-ROADMAP.md b/LATENCY-ROADMAP.md new file mode 100644 index 0000000..354fe84 --- /dev/null +++ b/LATENCY-ROADMAP.md @@ -0,0 +1,125 @@ +# 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. + +**Phase 2 — transport for latency.** The transport is a blocking `recv` with +a 1 s timeout and takes no timestamps. In order: a busy-poll read mode +(non-blocking socket, spin on an isolated core); `SO_TIMESTAMPING` +kernel/NIC receive timestamps threaded into the event, so measurement starts +at the wire rather than after recv + TLS + reassembly; and a revised +`Transport` read contract — the current copy-in `read(buf, len)` cannot +express zero-copy completion. 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.** Binance spot SBE is the only major +binary L2 diff-depth feed today and turns the ~1.9 us JSON decode into +struct-field reads; it also exercises the venue-decoder seam properly. +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; SWAR digit parsing in `parse_fixed`; 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. + +**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. From a551b1abfe04b8137d4ed2c49c86f56f713e18cb Mon Sep 17 00:00:00 2001 From: Joshua Dardashti <135767837+jdardash@users.noreply.github.com> Date: Sun, 2 Aug 2026 10:44:11 -0700 Subject: [PATCH 02/11] test(alloc): enforce no-allocation over decode, the feed, and the frame reader MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The probe covered the book — 0.3% of the frame. The claim it existed to enforce is about the other 99.7%: the decoder, Feed::handle end to end with the checksum verified, and the transport's poll loop. All three are allocation-free in steady state; now CI says so instead of the README. No Catch2 macros run inside the armed regions — an assertion handler may allocate, and each guarded loop's work is asserted after disarming (levels seen, checksums verified, messages pumped). --- tests/test_no_alloc.cpp | 138 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 138 insertions(+) diff --git a/tests/test_no_alloc.cpp b/tests/test_no_alloc.cpp index 44bf33f..848b623 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,138 @@ 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) { + pumped += pump_one() ? 1 : 0; + } + 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. From 7d3b15e6b357663d1a54471102e0479ea9ad0212 Mon Sep 17 00:00:00 2001 From: Joshua Dardashti <135767837+jdardash@users.noreply.github.com> Date: Sun, 2 Aug 2026 10:51:15 -0700 Subject: [PATCH 03/11] perf(net): stop zeroing receive buffers, and decrypt straight to the caller MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two receive paths grew a std::vector by a 32 KiB chunk on every socket read and trimmed it back to what arrived; vector::resize value-initializes, so that was 32 KiB of memset per read over bytes the transport was about to overwrite. net::ByteBuffer is the same vector with an allocator whose construct() default-initializes — resize becomes bookkeeping, everything else is unchanged. The Schannel decrypt path also copied plaintext twice (DecryptMessage's in-place output into plain_, plain_ into the caller) and built a fresh vector for the unconsumed tail of every pipelined TLS record — a heap allocation on the common path, since a busy feed routinely lands the next record behind the current one in the same segment. Plaintext now goes straight to the caller's buffer, overflow beyond the caller's ask is the only thing plain_ holds, and the tail is moved in place by take_extra, which already existed and already did exactly this. Verified live: 8 s Kraken capture through the new path, then replayed — 250 of 250 checksums match, so the decrypted bytes are byte-exact. --- include/crossbook/net/byte_buffer.hpp | 60 +++++++++++++++++++++++++++ include/crossbook/net/ws_frame.hpp | 10 ++++- src/net/tls_schannel.cpp | 51 ++++++++++++++++------- 3 files changed, 104 insertions(+), 17 deletions(-) create mode 100644 include/crossbook/net/byte_buffer.hpp 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/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/src/net/tls_schannel.cpp b/src/net/tls_schannel.cpp index e33e8b7..fde1b26 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) { @@ -440,11 +437,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 +493,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 +543,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_; From bf597924d7f4a68e4ae75a2524273a610957c264 Mon Sep 17 00:00:00 2001 From: Joshua Dardashti <135767837+jdardash@users.noreply.github.com> Date: Sun, 2 Aug 2026 10:51:22 -0700 Subject: [PATCH 04/11] build: 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 not inherited by any other preset: the README's numbers must come from the configuration a consumer gets, and a binary tuned to the build machine dies with an illegal instruction on the next one. --- CMakeLists.txt | 24 ++++++++++++++++++++++++ CMakePresets.json | 9 +++++++++ 2 files changed, 33 insertions(+) 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", From 914b4f6cf2865e0339cce0632b9a7cf8382da0f1 Mon Sep 17 00:00:00 2001 From: Joshua Dardashti <135767837+jdardash@users.noreply.github.com> Date: Sun, 2 Aug 2026 10:51:22 -0700 Subject: [PATCH 05/11] docs: changelog entries for the receive-path work and the roadmap --- CHANGELOG.md | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 63dcaee..dc1411e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,22 @@ only entries that can cost a reader an afternoon. ### Added +- `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 +56,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 From 9433ea43297d7f2ec1b19af876a86abb4856e4d5 Mon Sep 17 00:00:00 2001 From: Joshua Dardashti <135767837+jdardash@users.noreply.github.com> Date: Sun, 2 Aug 2026 11:03:57 -0700 Subject: [PATCH 06/11] style(test): wrap the frame literals the way clang-format wants --- tests/test_no_alloc.cpp | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_no_alloc.cpp b/tests/test_no_alloc.cpp index 848b623..a82030a 100644 --- a/tests/test_no_alloc.cpp +++ b/tests/test_no_alloc.cpp @@ -366,8 +366,8 @@ TEST_CASE("Feed::handle does not allocate on the applied path", "[alloc][feed]") 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")" + "}]}"; + 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 @@ -376,8 +376,8 @@ TEST_CASE("Feed::handle does not allocate on the applied path", "[alloc][feed]") 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")" + "}]}"; + 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); From e9d130c064ddbe348d81cbc74e0eb98dacb95984 Mon Sep 17 00:00:00 2001 From: Joshua Dardashti <135767837+jdardash@users.noreply.github.com> Date: Sun, 2 Aug 2026 11:16:41 -0700 Subject: [PATCH 07/11] fix(test): count pumped frames without a sign conversion GCC's -Wsign-conversion under -Werror rejects promoting the int from the conditional into a uint64_t accumulator; MSVC and Clang did not flag it, which is what the GCC floor job exists to catch. --- tests/test_no_alloc.cpp | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/test_no_alloc.cpp b/tests/test_no_alloc.cpp index a82030a..b9f7652 100644 --- a/tests/test_no_alloc.cpp +++ b/tests/test_no_alloc.cpp @@ -442,7 +442,9 @@ TEST_CASE("the frame reader poll loop does not allocate in steady state", "[allo { alloc_probe::Guard guard; for (int i = 0; i < 10'000; ++i) { - pumped += pump_one() ? 1 : 0; + if (pump_one()) { + ++pumped; + } } observed = alloc_probe::Guard::count(); } From 0b80c8d76d91c7dd8f7976a6dd4b3af3c76893e4 Mon Sep 17 00:00:00 2001 From: Joshua Dardashti <135767837+jdardash@users.noreply.github.com> Date: Sun, 2 Aug 2026 15:28:26 -0700 Subject: [PATCH 08/11] feat(net): busy-poll reads and kernel receive timestamps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit set_read_timeout(0) now means busy-poll — non-blocking socket, reads return kTimeout immediately — rather than the platforms' block-forever. crossbook_capture grows --busy-poll and, where the platform can say, a kernel-to-user delivery histogram: the gap between the kernel stamping a segment and the loop holding the decoded message, which is the number busy-poll exists to shrink. The timestamps were a lesson in what Linux actually offers TCP: SIOCGSTAMP is unsupported for SOCK_STREAM and SO_TIMESTAMPNS is silently a no-op there — both learned empirically. The one API that works is SO_TIMESTAMPING with RX_SOFTWARE, delivered as SCM_TIMESTAMPING control messages on the recvmsg that carries the data. Which forces an architectural correction: OpenSSL can no longer own the fd, because a backend that lets SSL_read call recv() itself can never see a control message. Reads now flow through TcpSocket via a custom BIO on the OpenSSL path — the same route Schannel always took — with kTimeout mapped to the retry flags SSL_read expects. The transport also gains its first tests: loopback coverage of the busy-poll contract, the ordinary timed path, and the timestamp, wired in via a deferred CMake hook because tests/ is configured before src/net exists. Verified live on Linux against Kraken over TLS: busy-poll cut kernel-to-user p50 from 150.4 us to 82.4 us in back-to-back 15 s captures (WSL2, unpinned; the p99 wants the isolated core the roadmap pairs it with). Windows: 344/344 tests, live capture clean, timestamps honestly reported unavailable. --- include/crossbook/net/transport.hpp | 14 ++ include/crossbook/net/websocket.hpp | 9 ++ src/net/tcp_socket.cpp | 76 +++++++++++ src/net/tcp_socket.hpp | 19 +++ src/net/tls_openssl.cpp | 104 ++++++++++++++- src/net/tls_schannel.cpp | 4 + src/net/transport.cpp | 4 + src/net/websocket.cpp | 10 ++ tests/CMakeLists.txt | 14 ++ tests/test_transport.cpp | 197 ++++++++++++++++++++++++++++ tools/crossbook_capture.cpp | 50 +++++++ 11 files changed, 495 insertions(+), 6 deletions(-) create mode 100644 tests/test_transport.cpp 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/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 fde1b26..ebf00d5 100644 --- a/src/net/tls_schannel.cpp +++ b/src/net/tls_schannel.cpp @@ -262,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_); 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..9e7a039 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -48,6 +48,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_transport.cpp b/tests/test_transport.cpp new file mode 100644 index 0000000..f8e63d9 --- /dev/null +++ b/tests/test_transport.cpp @@ -0,0 +1,197 @@ +// 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; + 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()), From 5114ca349e32adc0afc562f5027f593dc58858df Mon Sep 17 00:00:00 2001 From: Joshua Dardashti <135767837+jdardash@users.noreply.github.com> Date: Sun, 2 Aug 2026 15:50:44 -0700 Subject: [PATCH 09/11] feat(venues): Binance spot SBE decoder, the binary wire the roadmap pointed at MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hand-rolled reader for schema spot_stream 1:0 — depth diff and depth snapshot to the same DecodedMessage the JSON decoders produce, same route-before-trust and refuse-rather-than-round contracts, and SBE blockLengths honoured so a newer minor schema skips cleanly instead of desyncing. Ten fixture tests hand-encoded from the schema XML, including a byte-by-byte truncation sweep and a feed-level end-to-end over the recovery state machine. Live capture is blocked on an Ed25519 API key, which stream-sbe.binance.com requires even for public data. Also: roadmap updated with what the phases measured — the WSL2 sweep (6.3x tail improvement from the OS change alone), the busy-poll live numbers, and one lever measured dead (unchecked digit accumulation in parse_fixed: same-state A/B showed noise; reverted, recorded, so nobody re-optimizes it on vibes). 354/354 tests Windows, 353/353 Linux. --- CHANGELOG.md | 21 ++ LATENCY-ROADMAP.md | 60 +++-- include/crossbook/venues/binance_sbe.hpp | 312 +++++++++++++++++++++++ tests/CMakeLists.txt | 1 + tests/test_binance_sbe.cpp | 280 ++++++++++++++++++++ 5 files changed, 657 insertions(+), 17 deletions(-) create mode 100644 include/crossbook/venues/binance_sbe.hpp create mode 100644 tests/test_binance_sbe.cpp diff --git a/CHANGELOG.md b/CHANGELOG.md index dc1411e..4abb170 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,27 @@ 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 diff --git a/LATENCY-ROADMAP.md b/LATENCY-ROADMAP.md index 354fe84..5e8785c 100644 --- a/LATENCY-ROADMAP.md +++ b/LATENCY-ROADMAP.md @@ -74,31 +74,57 @@ 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. -**Phase 2 — transport for latency.** The transport is a blocking `recv` with -a 1 s timeout and takes no timestamps. In order: a busy-poll read mode -(non-blocking socket, spin on an isolated core); `SO_TIMESTAMPING` -kernel/NIC receive timestamps threaded into the event, so measurement starts -at the wire rather than after recv + TLS + reassembly; and a revised -`Transport` read contract — the current copy-in `read(buf, len)` cannot -express zero-copy completion. Skip kTLS: RX-path p99 regressions +*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.** Binance spot SBE is the only major -binary L2 diff-depth feed today and turns the ~1.9 us JSON decode into -struct-field reads; it also exercises the venue-decoder seam properly. -Deribit SBE multicast follows if derivatives matter — the only feed anywhere -that removes TLS entirely. Kraken's lever is placement plus FIX L3, not +**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; SWAR digit parsing in `parse_fixed`; 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. +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 diff --git a/include/crossbook/venues/binance_sbe.hpp b/include/crossbook/venues/binance_sbe.hpp new file mode 100644 index 0000000..44a2dfb --- /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/tests/CMakeLists.txt b/tests/CMakeLists.txt index 9e7a039..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 diff --git a/tests/test_binance_sbe.cpp b/tests/test_binance_sbe.cpp new file mode 100644 index 0000000..3eea122 --- /dev/null +++ b/tests/test_binance_sbe.cpp @@ -0,0 +1,280 @@ +// 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); +} From 2ee20fd5db9e113353d8be60486cbc19d992848a Mon Sep 17 00:00:00 2001 From: Joshua Dardashti <135767837+jdardash@users.noreply.github.com> Date: Sun, 2 Aug 2026 16:01:51 -0700 Subject: [PATCH 10/11] fix(portability): macOS spells htonl as a macro, and clang-format gets its wrap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The loopback test qualified htonl/ntohs with :: — a parse error on macOS, where both are macros. And the SBE files now carry the exact formatting clang-format wanted for the lines it flagged. Caught by the matrix, as designed: MSVC, GCC 13, and clang 14 all accepted both. --- include/crossbook/venues/binance_sbe.hpp | 4 +- tests/test_binance_sbe.cpp | 60 +++++++++++------------- tests/test_transport.cpp | 6 ++- 3 files changed, 33 insertions(+), 37 deletions(-) diff --git a/include/crossbook/venues/binance_sbe.hpp b/include/crossbook/venues/binance_sbe.hpp index 44a2dfb..5328465 100644 --- a/include/crossbook/venues/binance_sbe.hpp +++ b/include/crossbook/venues/binance_sbe.hpp @@ -198,8 +198,8 @@ class BinanceSbeDecoder { 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) { + [[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; diff --git a/tests/test_binance_sbe.cpp b/tests/test_binance_sbe.cpp index 3eea122..a9b4958 100644 --- a/tests/test_binance_sbe.cpp +++ b/tests/test_binance_sbe.cpp @@ -53,14 +53,13 @@ struct Level { /// 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") { + 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.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); @@ -103,7 +102,9 @@ std::string snapshot_frame(std::int64_t event_us, std::int64_t book_id, std::int return w.bytes; } -BinanceSbeDecoder btcusdt() { return BinanceSbeDecoder(InstrumentSpec{"BTCUSDT", 2, 8}); } +BinanceSbeDecoder btcusdt() { + return BinanceSbeDecoder(InstrumentSpec{"BTCUSDT", 2, 8}); +} } // namespace @@ -134,29 +135,25 @@ TEST_CASE("a coarser exponent rescales exactly", "[venues][binance][sbe]") { // 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 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]") { +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 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]") { +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"); @@ -179,12 +176,11 @@ TEST_CASE("an SBE depth snapshot carries its book id as both sequence ends", REQUIRE(msg.levels.size() == 2); } -TEST_CASE("other templates on the socket are routing, not failure", - "[venues][binance][sbe]") { +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(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); @@ -204,11 +200,10 @@ TEST_CASE("a foreign schema id is loudly malformed", "[venues][binance][sbe]") { CHECK(msg.error == DecodeError::kMalformed); } -TEST_CASE("truncation anywhere is malformed, never a partial decode", - "[venues][binance][sbe]") { +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}}); + 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) { @@ -225,24 +220,24 @@ TEST_CASE("a newer minor version's extra bytes are skipped, not misread", // 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(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.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(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(20); // Empty ask group, same stride. w.u16(0); w.str8("BTCUSDT"); @@ -258,8 +253,7 @@ TEST_CASE("a newer minor version's extra bytes are skipped, not misread", 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); + 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}}); diff --git a/tests/test_transport.cpp b/tests/test_transport.cpp index f8e63d9..a06f476 100644 --- a/tests/test_transport.cpp +++ b/tests/test_transport.cpp @@ -47,14 +47,16 @@ class LoopbackListener { listen_fd_ = ::socket(AF_INET, SOCK_STREAM, 0); ::sockaddr_in addr{}; addr.sin_family = AF_INET; - addr.sin_addr.s_addr = ::htonl(INADDR_LOOPBACK); + // 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); + port_ = ntohs(addr.sin_port); } ~LoopbackListener() { From f2bfd7dd3ce8c9b7f97cc4c0b6f41701ac160c24 Mon Sep 17 00:00:00 2001 From: Joshua Dardashti <135767837+jdardash@users.noreply.github.com> Date: Sun, 2 Aug 2026 16:15:37 -0700 Subject: [PATCH 11/11] style(test): the exact wrap and comment alignment clang-format 18 wants --- tests/test_transport.cpp | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/tests/test_transport.cpp b/tests/test_transport.cpp index a06f476..982fdc2 100644 --- a/tests/test_transport.cpp +++ b/tests/test_transport.cpp @@ -156,8 +156,7 @@ TEST_CASE("a nonzero timeout still blocks and still delivers", "[transport]") { CHECK(got == 4); } -TEST_CASE("the kernel receive timestamp is exposed where the platform has one", - "[transport]") { +TEST_CASE("the kernel receive timestamp is exposed where the platform has one", "[transport]") { LoopbackListener listener; detail::TcpSocket sock; std::string error; @@ -188,7 +187,7 @@ TEST_CASE("the kernel receive timestamp is exposed where the platform has one", 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 < 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