market-data-pipeline

License: MIT

A low-latency market data processing framework in C++20: lock-free SPSC queue, fixed-point normalization, and push-to-dispatch latency measurement.

At a glance

FeedGenerator ──push──▶ SPSC ring ──pop──▶ normalize ──▶ Strategy
   (producer)              │                (consumer)
                     CPU-pinned threads

Build & run

cmake -B build -DCMAKE_BUILD_TYPE=Release && cmake --build build -j
./build/benchmark          # throughput + queue microbenches
ctest --test-dir build -V  # unit tests

Linux latency (pinned):

export MDP_PRODUCER_CPU=0 MDP_CONSUMER_CPU=1
taskset -c 0,1 ./build/benchmark

Design choices

1. SPSC, not MPSC — one producer / one consumer; no CAS on the fast path
2. Cache-line isolation — head/tail on separate lines; acquire/release atomics
3. Three separate properties — capacity (throughput), intrinsic latency (low backlog), overload queueing (full-blast p50/p99)

Details: SPSC memory model · docs/design.md · Benchmark Results

---

Architecture

Pipeline

flowchart LR
  subgraph PROD["Producer thread · owns FeedGenerator · MDP_PRODUCER_CPU"]
    FG["FeedGenerator<br/>make_message()"]
    TS0["t0 = now_ns()<br/>stamp RawMessage.timestamp_ns"]
    PUSH["SPSC::push(RawMessage)"]
    FG --> TS0 --> PUSH
  end

  subgraph Q["Shared memory · lock-free SPSC ring<br/>capacity 65536 · head/tail on separate cache lines"]
    RING[("RawMessage slots<br/>producer writes · consumer reads")]
  end

  subgraph CONS["Consumer thread · owns FeedHandler + Strategy · MDP_CONSUMER_CPU"]
    POP["SPSC::pop(raw)"]
    NORM["normalize(raw) → MarketEvent"]
    TS1["t1 = now_ns()<br/>latency = t1 − t0  (optional)"]
    STRAT["Strategy::onMarketEvent(event)"]
    POP --> NORM --> TS1 --> STRAT
  end

  PUSH -->|"by value into ring"| RING
  RING -->|"move out of ring"| POP

Loading

Ownership & flow

Piece	Owner	Notes
FeedGenerator	Producer thread	Synthetic messages; stamps timestamp_ns immediately before push; optional max_in_flight pacing for low-backlog latency
SPSCQueue	Shared (no lock)	Producer owns tail; consumer owns head; one unused slot
FeedHandler	Consumer thread	pop → normalize → optional latency record → strategy
Strategy	Consumer thread	Virtual callback; no I/O on hot path
Latency histogram	Consumer thread	Written only by consumer; read after join

Timestamp locations: t0 = now_ns() after make_message(), immediately before push. t1 = now_ns() after normalize(), before strategy. See Measurement Methodology for the exact include/exclude list.

Producer Thread                     Consumer Thread
────────────────                    ───────────────────────────────────
FeedGenerator                       FeedHandler
  make_message()                      pop(raw)
    stamp timestamp_ns = now_ns()  ──>  normalize(raw) → MarketEvent
  queue.push(raw)                     record latency (optional)
                                      Strategy::onMarketEvent(event)

CPU affinity

Benchmarks and latency measurements assume producer and consumer run on dedicated cores to avoid scheduler interference and cache migration:

Producer Core (CPU 2)
        ↓
   SPSC Queue
        ↓
Consumer Core (CPU 3)

Pin threads in code (Linux) and restrict the process with taskset:

export MDP_PRODUCER_CPU=2
export MDP_CONSUMER_CPU=3
taskset -c 2,3 ./build/pipeline
taskset -c 2,3 ./build/benchmark

MDP_PRODUCER_CPU / MDP_CONSUMER_CPU call pthread_setaffinity_np at thread start. taskset keeps the process off other cores. Use both for stable latency numbers. Unset env vars => no in-code pinning (macOS ignores them).

Layout

include/mdp/     Public headers (.hpp)
src/             Implementation (.cpp)
tests/           Unit tests
docs/            Design notes

Components

File	Responsibility
include/mdp/spsc_queue.hpp	Lock-free single-producer/single-consumer ring buffer
include/mdp/messages.hpp	RawMessage (exchange-native) and MarketEvent (canonical) types
include/mdp/pipeline_constants.hpp	Shared kQueueCapacity (65536)
include/mdp/pipeline_runner.hpp	Orchestration; defines Queue alias wiring messages + SPSC
include/mdp/cpu_affinity.hpp	Optional per-thread CPU pinning (MDP_PRODUCER_CPU, MDP_CONSUMER_CPU)
include/mdp/timestamp.hpp	now_ns(); rdtsc_begin()/rdtsc_end() on x86-64 (fenced TSC)
include/mdp/latency_stats.hpp	100 ns-resolution histogram, p50/p99/p99.9 reporting
include/mdp/normalizer.hpp	Stateless normalize(): raw → canonical fixed-point
include/mdp/strategy.hpp	Strategy base, NoOpStrategy, StatsStrategy
src/feed_generator.cpp	Synthetic feed: Add/Cancel/Modify/Trade in realistic proportions
src/feed_handler.cpp	Consumer: drain queue → normalize → dispatch

---

SPSC memory model

This is the core of the project. Implementation: include/mdp/spsc_queue.hpp. Deeper notes: docs/design.md.

Why SPSC

Exactly one producer (FeedGenerator) and one consumer (FeedHandler). That constraint removes all CAS from the fast path: each thread exclusively owns one index, so neither needs to compete for a slot. MPSC would need atomic CAS on the tail under multi-producer contention — the wrong structure for a single feed-handler thread.

Ownership (important)

Index	Owner	Role
tail	Producer only	Advanced after a successful write into buf_[tail]
head	Consumer only	Advanced after a successful read from buf_[head]

(If a checklist says “producer owns head,” that’s swapped — this codebase matches the usual SPSC convention: producer publishes via tail, consumer consumes via head.)

Each side also keeps a local cached snapshot of the other index (cached_head / cached_tail) so it rarely touches the remote cache line.

Atomics

prod_.tail  : std::atomic<std::size_t>   // written by producer, read by consumer
cons_.head  : std::atomic<std::size_t>   // written by consumer, read by producer
cached_*    : plain std::size_t          // thread-local; not shared
buf_        : T[Capacity]                // published via release on the index

Local index loads use memory_order_relaxed (only this thread stores them). Cross-thread publication uses acquire/release (below).

Acquire / release

push:  write buf_[tail]  →  tail.store(next, release)
pop:   head ==? acquire-load(tail)  →  read buf_[head]  →  head.store(next, release)

• Release-store on tail: makes the buffer write visible to the consumer before it observes the new index.
• Acquire-load on tail (consumer): ensures the slot is read only after seeing that publish.
• Symmetric acquire on head for the producer’s full-check.
• seq_cst is intentionally avoided — we only need that pair’s happens-before, not a global total order.

Cache-line separation

tail (+ cached_head) and head (+ cached_tail) live on separate 64-byte-aligned pads. If they shared a line, every producer tail store would invalidate the consumer’s line (and vice versa) — false sharing. Cached snapshots cut how often each side loads the remote atomic.

Full / empty detection

Power-of-two capacity; indices wrap with a bitmask (& (Capacity - 1)).

Condition	Meaning
head == tail	Empty
(tail + 1) & mask == head	Full

One slot is left unused so full and empty are distinguishable. Max in-flight = Capacity − 1 (65535 for 65536). Push/pop return false when full/empty (non-blocking); the pipeline producer applies spin backpressure (next section).

Queue-full behaviour (backpressure)

What this project does: when push returns false (ring full), the producer spins with cpu_pause() (PAUSE / YIELD) until space appears. It does not drop, overwrite, or block on a mutex/condvar.

// FeedGenerator::run()
msg.timestamp_ns = now_ns();
while (!queue_.push(msg))
    cpu_pause();

Policy	This repo?	Effect on market data
Spin / wait (backpressure)	Yes	Preserves every message and FIFO order; producer stalls; latency samples include wait time after t0
Drop newest	No	Loses updates; catastrophic for gap-sensitive feeds (e.g. incremental book) unless recovery exists
Drop oldest / overwrite	No	Silent loss of earlier state; worse for sequenced feeds
Block (sleep / futex)	No	Would yield the core; adds wakeup jitter — avoided on the hot path here
Load-shed / secondary path	No	Valid production pattern (e.g. snapshot rebuild) — out of scope

Why spin here: this is a synthetic benchmark pipeline. The goal is ordered delivery + lowest jitter while measuring push-to-pre-callback. Spinning keeps the producer on-core and makes backpressure explicit in the latency histogram (p50/p99 ≈ full-queue drain when the consumer can’t keep up).

What you’d choose in production depends on feed semantics:

• Gap-sensitive incremental feeds (ITCH-style add/cancel/modify): usually must not drop; prefer sizing the ring + consumer so full is rare, then backpressure or fail loud + recovery (re-subscribe / snapshot).
• Conflatable quotes: sometimes drop/coalesce is acceptable.
• Hard real-time with isolated cores: spin or short adaptive pause is common; sleeping is rare on the critical path.

The SPSC API stays policy-agnostic (push → bool). Policy lives in the producer loop — swap the while (!push) pause for drop/overwrite/block without changing the queue.

---

Build

Requires: CMake ≥ 3.20, a C++20 compiler (GCC 11+, Clang 14+), pthreads.

Linux:

cmake -B build -DCMAKE_BUILD_TYPE=Release
cmake --build build -j$(nproc)

macOS:

cmake -B build -DCMAKE_BUILD_TYPE=Release
cmake --build build -j$(sysctl -n hw.ncpu)

Debug build with ThreadSanitizer (Linux only — TSan disabled on Apple Clang):

cmake -B build-debug -DCMAKE_BUILD_TYPE=Debug
cmake --build build-debug -j$(nproc)          # Linux
cmake --build build-debug -j$(sysctl -n hw.ncpu) # macOS

---

Run

# Pipeline: 2M messages, print latency report
./build/pipeline

# Custom message count
./build/pipeline 5000000

# Benchmark suite (throughput + queue microbenches)
./build/benchmark

# Tests
ctest --test-dir build -V

Collecting benchmark numbers

Run from the repo root after a Release build:

# Record environment (paste into Benchmark Results)
uname -a
c++ --version          # macOS / Clang
# g++ --version        # Linux / GCC

# Full benchmark table output (each throughput config runs 3×; median reported)
./build/benchmark

Linux (recommended for latency percentiles):

export MDP_PRODUCER_CPU=2
export MDP_CONSUMER_CPU=3
taskset -c 2,3 ./build/benchmark

macOS: throughput numbers are usable for local iteration. Latency percentiles are not printed by ./benchmark / ./pipeline on non-Linux — use the Linux Benchmark Results for p50/p99/min.

Recommended Linux setup for stable numbers

# Pin producer to core 2, consumer to core 3 (same NUMA node, separate physical cores)
export MDP_PRODUCER_CPU=2
export MDP_CONSUMER_CPU=3
taskset -c 2,3 ./build/benchmark

Optional for more stable benchmark results:

# Disable Intel TurboBoost for consistent clock
echo 1 | sudo tee /sys/devices/system/cpu/intel_pstate/no_turbo

# Disable CPU frequency scaling
sudo cpupower frequency-set -g performance

---

Test Coverage

The SPSC queue is tested as production infrastructure, not a coding exercise. Each item below maps to a test in tests/test_spsc_queue.cpp.

• ✓ Empty queue semantics — test_empty_and_full, test_pop_empty
• ✓ Full queue semantics — test_empty_and_full
• ✓ FIFO ordering — test_fifo_order
• ✓ Wraparound correctness — test_wraparound_half, test_wraparound_crossing
• ✓ 1M message concurrent producer-consumer validation — test_concurrent_correctness
• ✓ Raw market-data message transport — test_rawmessage_roundtrip
• ✓ Capacity invariants — test_capacity

Additional stress coverage:

• ✓ Empty → full → empty transitions — test_empty_and_full
• ✓ Fill-drain repeat (100k rounds) — test_fill_drain_repeat
• ✓ Head/tail crossing while non-empty — test_wraparound_crossing
• ✓ Tiny-capacity stress (capacity 4/8) — test_tiny_capacity_stress, test_tiny_capacity_concurrent
• ✓ Bursty producer (exchange-like traffic patterns) — test_bursty_producer

Run the full suite:

./build/test_spsc

---

Interesting Failure Modes Tested

These tests target the boundary conditions where ring-buffer bugs actually appear — more meaningful than a raw pass count:

• Full → empty transitions
• Head/tail crossing while non-empty
• Tiny-capacity wraparound stress
• Bursty producer traffic
• Long-running fill/drain cycles

---

Benchmark Results

Benchmarks report three separate systems properties. Do not compare overload p50 to low-backlog p50 as if they measure the same thing.

Linux (pinned) — Ubuntu 24.04 LTS, DigitalOcean Basic 2 vCPU / 2 GB (BLR1), CPU model DO-Regular (QEMU/KVM, 2 cores @ ~2.0 GHz), kernel 6.8.0-124-generic, GCC 13.3.0, Release (-O3 -march=native), CLOCK_MONOTONIC_RAW, MDP_PRODUCER_CPU=0 / MDP_CONSUMER_CPU=1, taskset -c 0,1.

A. Capacity

Metric	Value
Pipeline throughput (NoOp, full-blast, 5M msgs, median of 3)	5.5 M msg/s
Queue push/pop (1 thread)	176 M msg/s
Queue push/pop (1 thread, TSC)	~11 cycles/msg
Queue push/pop (2 threads)	244 M msg/s

B. Intrinsic latency (low backlog)

Same push→pre-callback probe; producer waits until the queue is empty (max_in_flight=1) before each stamp/push. 1,000,000 messages.

Metric	Value
min	131 ns
p50	300 ns
p99	500 ns
p99.9	1,300 ns
max	92 µs (rare outlier)

p50/p99 sit near min — this is transit + normalize + two clocks, not queue wait.

C. Queueing under overload

Same probe at full blast (max_in_flight=0). Producer outruns consumer; ring fills. 5,000,000 messages (median-throughput latency trial among 3 interleaved with capacity runs).

Metric	Value
min	316 ns
p50	11.7 ms
p99	11.7 ms
p99.9	11.7 ms

p50≈p99≈full-ring drain time (~65K / ~5.5 M msg/s ≈ 12 ms). Min still sees occasional near-empty samples.

One-line story: ~5.5 M msg/s capacity · ~300 ns intrinsic p50 · ~12 ms overload p50.

To reproduce (Linux):

cmake -B build -DCMAKE_BUILD_TYPE=Release
cmake --build build -j$(nproc)
export MDP_PRODUCER_CPU=0
export MDP_CONSUMER_CPU=1
taskset -c 0,1 ./build/benchmark

macOS (dev machine) — Apple M1, Clang 14, Release, no pinning. Throughput only; latency sections skipped in ./benchmark.

Metric	Value
Pipeline throughput (NoOp)	~10 M msg/s
Queue push/pop microbench (1 thread)	~60 M msg/s
Queue push/pop microbench (2 threads)	~50 M msg/s

---

Measurement Methodology

What latency means (definition)

Probe name: push-to-pre-callback (also called push-to-dispatch in comments).

make_message()          ← NOT in latency
t0 = now_ns()           ← START  (stamp RawMessage.timestamp_ns)
  maybe spin if queue full (backpressure wait IS in latency)
  SPSC::push(msg)
  … message may sit in the ring (queue wait IS in latency) …
  SPSC::pop(raw)
  normalize(raw) → MarketEvent
t1 = now_ns()           ← END    (after normalize, before strategy)
  Strategy::onMarketEvent(event)  ← NOT in latency
latency = t1 − t0

Code locations: stamp in FeedGenerator::run() immediately before push; sample in FeedHandler::run() immediately after normalize(), before onMarketEvent.

Included vs excluded

In the latency sample?	Step
No	Message construction (make_message / LCG fields)
No	Producer thread scheduling before t0
Yes	t0 → push (incl. spin-wait if queue is full)
Yes	Time sitting in the SPSC ring (often dominates p50/p99 under load)
Yes	pop + normalize
Yes	Two now_ns() / clock_gettime calls (~tens of ns each on Linux)
No	Strategy::onMarketEvent body
No	Consumer work after t1

So it is not “full pipeline end-to-end,” and it is not “queue hop only.” It is:

time from “about to publish” until “normalized event ready for strategy.”

Under sustained overload (max_in_flight=0) when the producer outruns the consumer, most of that time is queue wait — hence p50/p99 ≈ full-ring drain time (~12 ms here).

Under low backlog (max_in_flight=1) the producer waits until the queue is empty before each stamp/push, so the same probe reports intrinsic latency (p50 ~300 ns here). That is how “min 131 ns vs overload p50 12 ms” is not a contradiction — different occupancy regimes.

Load regimes

Regime	max_in_flight	What you learn	Benchmark
Full-blast capacity	0	How fast the pipeline can go	[1]
Intrinsic latency	1	Transit + normalize without queue wait	[6]
Overload queueing	0 + latency hist	Backlog / drain behaviour	[3]
Queue transport only	n/a	SPSC alone	[4] [5]

Clock

• Linux: CLOCK_MONOTONIC_RAW via now_ns()
• macOS: CLOCK_MONOTONIC (percentiles not printed; see gating below)
• If t0 > t1 (skew / step), the sample is dropped (unsigned wrap would poison the histogram)

Latency recording lives in FeedHandler / PipelineRunner, not in any Strategy — same probe for NoOpStrategy and StatsStrategy.

Other measurements (do not confuse)

Output	What it is
[1] Pipeline throughput	Capacity — full-blast wall-clock M msg/s (headline)
[3] Overload latency	Same probe, full-blast occupancy — queue wait dominates p50/p99
[6] Low-backlog latency	Same probe, max_in_flight=1 — intrinsic transit
[4] / [5] Queue microbench	Push/pop only — no generator/normalize/strategy; [4] also TSC

TSC microbench ([4]) uses fenced reads: rdtsc_begin() (LFENCE; RDTSC) / rdtsc_end() (RDTSCP; LFENCE). Never mix TSC cycles with now_ns() deltas. Pipeline latency percentiles print only on Linux.

Reproducibility

uname -a
g++ --version   # or c++ --version on macOS
lscpu | grep 'Model name'

Pin with MDP_PRODUCER_CPU / MDP_CONSUMER_CPU + taskset on Linux before latency runs.

Design Decisions

Start with SPSC memory model above (including queue-full behaviour). See docs/design.md for the rest:

• Why a bounded ring buffer / backpressure
• CPU affinity (taskset + per-thread pin)
• Why not mutexes
• Fixed-point price representation
• Virtual dispatch vs CRTP in Strategy

---

What's Not Here (Yet)

• Dual-thread queue-only latency mode — [6] covers low-backlog pipeline latency; a stamp-on-push / sample-on-pop queue-only histogram is still optional
• pybind11 bindings — planned; will expose FeedHandler and MarketEvent to Python for researcher use
• Replay mode — read a recorded binary feed file and replay at configurable speed
• Multiple strategies — strategy chain / fan-out to multiple callbacks
• Real exchange feed — ITCH 5.0 or CME MDP3 parser as a drop-in FeedGenerator replacement

---

Resume Bullet

Built a low-latency market data processing framework in C++20 featuring a lock-free SPSC ring buffer with cache-line isolation, per-thread CPU pinning, and separated capacity / intrinsic-latency / overload-queueing benchmarks. Achieved ~5.5 M msg/s pipeline throughput, ~300 ns low-backlog p50, and ~12 ms overload p50 on a pinned 2-vCPU Linux VM.