Lesson 66 of 71
Project: Market Data Handler
A market data handler is the component that receives an exchange feed, decodes it, checks that no message is lost, and keeps an up-to-date book of prices for the strategies that read it. This project builds one in four stages, each a complete program. It reuses the header and flyweight style from Binary Protocols and Encoding, the measurement methods from Measuring Latency, and the techniques from Allocation-Free and Cache-Friendly Code.
Problem statement
An exchange publishes one instrument's market data as a stream of binary messages. Each message carries a sequence number and either sets the total quantity at one price level or is a heartbeat. Messages can be lost or duplicated in transit. When a message is lost, the handler cannot trust its book until it receives a snapshot: a message that lists every non-empty price level as of a given sequence number. The feed here is generated in-process from a script, so every run produces the same messages, the same gaps, and the same output.
Requirements
| # | Requirement |
|---|---|
| R1 | Decode messages from a ByteBuffer with flyweights; no object per message |
| R2 | Detect gaps and duplicates by sequence number |
| R3 | Keep a price-level book: total quantity per price, best bid, best ask |
| R4 | On a gap, buffer incoming messages, wait for a snapshot, load it, replay the buffered messages newer than the snapshot |
| R5 | Record pipeline latency in an HdrHistogram |
| R6 | Allocate zero bytes per message after warmup, checked with ThreadMXBean |
| Stage | Adds | Requirements |
|---|---|---|
| 1 | Message format, encoder and decoder flyweights, gap detector | R1, R2 |
| 2 | Array-based price-level book | R3 |
| 3 | Recovery state machine with a pending buffer and snapshots | R4 |
| 4 | Generated feed of 100,000 messages, histograms, allocation check | R5, R6 |
The book in this project stores only the total quantity at each price, which is what a market-by-price feed sends. A book that tracks individual orders in FIFO queues is built in Project: Limit Order Book.
Stage 1: message format, flyweights, and gap detection
The message format extends the header from the binary protocols lesson with a send timestamp, used in Stage 4. All fields are little-endian at fixed offsets.
| Offset | Field | Type | Notes |
|---|---|---|---|
| 0 | length | u16 | Whole message, header included |
| 2 | type | u8 | 1 = LEVEL, 2 = HEARTBEAT, 3 = SNAPSHOT |
| 3 | seq | u64 | Sequence number |
| 11 | sendTime | u64 | Exchange send time in nanoseconds |
| 19 | side | u8 | LEVEL only: 'B' or 'A' |
| 20 | price | i64 | LEVEL only: price in ticks |
| 28 | qty | i64 | LEVEL only: new total quantity, 0 removes the level |
A LEVEL message is an incremental update: it states the new total at one price, not a change relative to the old total. Applying the same LEVEL message twice gives the same book, which makes duplicates harmless once detected. The encoder writes at the buffer's position and advances it; the decoder reads with absolute offsets and never moves the position.
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
public class FeedDecoder {
static final int LENGTH = 0, TYPE = 2, SEQ = 3, SEND_TIME = 11, HEADER = 19;
static final int SIDE = 19, PRICE = 20, QTY = 28, LEVEL_LENGTH = 36;
static final byte LEVEL = 1, HEARTBEAT = 2;
static final byte BID = 'B', ASK = 'A';
static final class Encoder {
private ByteBuffer buf;
Encoder wrap(ByteBuffer buf) { this.buf = buf; return this; }
private void header(int length, byte type, long seq, long sendTime) {
int at = buf.position();
buf.putShort(at + LENGTH, (short) length).put(at + TYPE, type)
.putLong(at + SEQ, seq).putLong(at + SEND_TIME, sendTime);
}
void level(long seq, long sendTime, byte side, long price, long qty) {
int at = buf.position();
header(LEVEL_LENGTH, LEVEL, seq, sendTime);
buf.put(at + SIDE, side).putLong(at + PRICE, price).putLong(at + QTY, qty);
buf.position(at + LEVEL_LENGTH);
}
void heartbeat(long seq, long sendTime) {
header(HEADER, HEARTBEAT, seq, sendTime);
buf.position(buf.position() + HEADER);
}
}
static final class Decoder {
private ByteBuffer buf;
private int at;
Decoder wrap(ByteBuffer buf, int offset) { this.buf = buf; this.at = offset; return this; }
int length() { return Short.toUnsignedInt(buf.getShort(at + LENGTH)); }
byte type() { return buf.get(at + TYPE); }
long seq() { return buf.getLong(at + SEQ); }
long sendTime() { return buf.getLong(at + SEND_TIME); }
char side() { return (char) buf.get(at + SIDE); }
long price() { return buf.getLong(at + PRICE); }
long qty() { return buf.getLong(at + QTY); }
}
static final class GapDetector {
static final int OK = 0, DUPLICATE = 1, GAP = 2;
long expected = 1;
int check(long seq) {
if (seq < expected) return DUPLICATE;
int result = seq == expected ? OK : GAP;
expected = seq + 1;
return result;
}
}
public static void main(String[] args) {
ByteBuffer wire = ByteBuffer.allocateDirect(1024).order(ByteOrder.LITTLE_ENDIAN);
Encoder enc = new Encoder().wrap(wire);
enc.level(1, 1_000, BID, 10_000, 500);
enc.level(2, 2_000, ASK, 10_002, 300);
enc.level(3, 3_000, BID, 9_999, 200);
enc.level(2, 2_000, ASK, 10_002, 300); // duplicate of seq 2
enc.heartbeat(4, 4_000);
enc.level(7, 7_000, BID, 10_000, 0); // seq 5 and 6 lost
int limit = wire.position();
Decoder msg = new Decoder();
GapDetector gaps = new GapDetector();
for (int offset = 0; offset < limit; offset += msg.length()) {
msg.wrap(wire, offset);
long before = gaps.expected;
switch (gaps.check(msg.seq())) {
case GapDetector.DUPLICATE -> System.out.println("seq " + msg.seq() + " duplicate, dropped");
case GapDetector.GAP -> System.out.println("seq " + msg.seq() + " gap, missing "
+ before + ".." + (msg.seq() - 1));
default -> {
if (msg.type() == LEVEL) {
System.out.println("seq " + msg.seq() + " LEVEL " + msg.side() + " "
+ msg.qty() + " @ " + msg.price());
} else {
System.out.println("seq " + msg.seq() + " HEARTBEAT");
}
}
}
}
System.out.println("bytes on wire: " + limit + ", next expected: " + gaps.expected);
}
}seq 1 LEVEL B 500 @ 10000
seq 2 LEVEL A 300 @ 10002
seq 3 LEVEL B 200 @ 9999
seq 2 duplicate, dropped
seq 4 HEARTBEAT
seq 7 gap, missing 5..6
bytes on wire: 199, next expected: 8The decoder reads length first, so the loop can step to the next message without knowing the body layout. The gap detector is three comparisons and one field: a sequence number below expected is a duplicate, one above it is a gap. In this stage the gap is only reported. Stage 3 shows why the message after a gap cannot be applied directly: the lost messages may have changed price levels that the later message does not mention.
Stage 2: a price-level book from incremental updates
The book keeps two long arrays, one per side, indexed by price - basePrice. A tick is the smallest price step the instrument allows, so prices are whole numbers of ticks and every price in the range has one array slot. Setting a level is one array store. The best bid is the highest index with a quantity on the bid side, and the best ask is the lowest index with a quantity on the ask side. Both are kept as fields and rescanned only when the best level is removed.
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.Arrays;
public class LevelBook {
static final int LENGTH = 0, TYPE = 2, SEQ = 3, SEND_TIME = 11, HEADER = 19;
static final int SIDE = 19, PRICE = 20, QTY = 28, LEVEL_LENGTH = 36;
static final byte LEVEL = 1, BID = 'B', ASK = 'A';
static final class Decoder {
private ByteBuffer buf;
private int at;
Decoder wrap(ByteBuffer buf, int offset) { this.buf = buf; this.at = offset; return this; }
int length() { return Short.toUnsignedInt(buf.getShort(at + LENGTH)); }
byte type() { return buf.get(at + TYPE); }
long seq() { return buf.getLong(at + SEQ); }
byte side() { return buf.get(at + SIDE); }
long price() { return buf.getLong(at + PRICE); }
long qty() { return buf.getLong(at + QTY); }
}
static void level(ByteBuffer buf, long seq, byte side, long price, long qty) {
int at = buf.position();
buf.putShort(at + LENGTH, (short) LEVEL_LENGTH).put(at + TYPE, LEVEL).putLong(at + SEQ, seq)
.putLong(at + SEND_TIME, seq * 1_000).put(at + SIDE, side)
.putLong(at + PRICE, price).putLong(at + QTY, qty);
buf.position(at + LEVEL_LENGTH);
}
static final class Book {
final long base;
final long[] bids, asks;
int bestBid = -1, bestAsk;
int rejected;
Book(long base, int levels) {
this.base = base;
bids = new long[levels];
asks = new long[levels];
bestAsk = levels;
}
void set(byte side, long price, long qty) {
long i = price - base;
if (i < 0 || i >= bids.length) { rejected++; return; }
int idx = (int) i;
if (side == BID) {
bids[idx] = qty;
if (qty > 0) { if (idx > bestBid) bestBid = idx; }
else if (idx == bestBid) { while (bestBid >= 0 && bids[bestBid] == 0) bestBid--; }
} else {
asks[idx] = qty;
if (qty > 0) { if (idx < bestAsk) bestAsk = idx; }
else if (idx == bestAsk) { while (bestAsk < asks.length && asks[bestAsk] == 0) bestAsk++; }
}
}
long bestBidPrice() { return bestBid < 0 ? -1 : base + bestBid; }
long bestAskPrice() { return bestAsk == asks.length ? -1 : base + bestAsk; }
void clear() {
Arrays.fill(bids, 0);
Arrays.fill(asks, 0);
bestBid = -1;
bestAsk = asks.length;
}
String top() {
return bestBid < 0 || bestAsk == asks.length ? "bid " + bestBidPrice() + " ask " + bestAskPrice()
: bids[bestBid] + " @ " + bestBidPrice() + " | " + asks[bestAsk] + " @ " + bestAskPrice()
+ " spread " + (bestAsk - bestBid);
}
void printDepth(int n) {
for (int k = 0, i = bestAsk + n - 1; k < n; k++, i--) {
if (i < asks.length && asks[i] > 0) System.out.println(" ask " + (base + i) + " x " + asks[i]);
}
for (int k = 0, i = bestBid; k < n && i >= 0; i--) {
if (bids[i] > 0) { System.out.println(" bid " + (base + i) + " x " + bids[i]); k++; }
}
}
}
public static void main(String[] args) {
ByteBuffer wire = ByteBuffer.allocateDirect(1024).order(ByteOrder.LITTLE_ENDIAN);
long seq = 1;
level(wire, seq++, BID, 10_000, 500);
level(wire, seq++, ASK, 10_002, 300);
level(wire, seq++, BID, 9_999, 200);
level(wire, seq++, ASK, 10_003, 700);
level(wire, seq++, BID, 10_001, 100); // new best bid
level(wire, seq++, ASK, 10_002, 0); // best ask removed
level(wire, seq++, BID, 10_000, 650); // level replaced, not added to
level(wire, seq++, ASK, 20_000, 100); // outside the book's range
int limit = wire.position();
Book book = new Book(9_990, 32);
LevelBook.Decoder msg = new LevelBook.Decoder();
for (int offset = 0; offset < limit; offset += msg.length()) {
msg.wrap(wire, offset);
book.set(msg.side(), msg.price(), msg.qty());
System.out.println("seq " + msg.seq() + ": " + book.top());
}
System.out.println("depth:");
book.printDepth(3);
System.out.println("rejected: " + book.rejected);
}
}seq 1: bid 10000 ask -1
seq 2: 500 @ 10000 | 300 @ 10002 spread 2
seq 3: 500 @ 10000 | 300 @ 10002 spread 2
seq 4: 500 @ 10000 | 300 @ 10002 spread 2
seq 5: 100 @ 10001 | 300 @ 10002 spread 1
seq 6: 100 @ 10001 | 700 @ 10003 spread 2
seq 7: 100 @ 10001 | 700 @ 10003 spread 2
seq 8: 100 @ 10001 | 700 @ 10003 spread 2
depth:
ask 10003 x 700
bid 10001 x 100
bid 10000 x 650
bid 9999 x 200
rejected: 1Seq 6 removes the best ask, so the book scans upward from index 12 and finds 10,003. Seq 7 shows the market-by-price rule: the new total 650 replaces 500. The update at 20,000 falls outside the 32-tick window and is counted, not stored; a production book sizes the window from the instrument's price limits and recenters it when the price moves, which is one of the extension tasks.
The arrays give O(1) updates and keep neighbouring prices in neighbouring memory, which the CPU caches lesson explains. The rescan after removing the best level is O(gap to the next level), which is short in a liquid market. A TreeMap<Long, Long> would give O(log n) for every update and allocate entries and boxed keys.
Stage 3: gap recovery with snapshots
After a gap, the book is wrong in an unknown way, so the handler must not publish it. The handler moves through two states:
| State | On LEVEL or HEARTBEAT | On SNAPSHOT |
|---|---|---|
| LIVE | Check sequence; apply, drop duplicate, or start recovery | Ignore |
| RECOVERING | Copy the message into the pending buffer | Load the book, set expected = snapshot.seq + 1, replay pending messages, go LIVE |
The pending buffer is a pre-allocated ByteBuffer. Each message is copied into it with the absolute put(int index, ByteBuffer src, int offset, int length) added in Java 16, so buffering creates no objects. A SNAPSHOT message uses the same header, then a u16 entry count at offset 19, then 17-byte entries (side, price, qty) from offset 21. Its seq is the last incremental sequence number already included in it. On replay, pending messages at or below that number are skipped because the snapshot already contains them.
The script also keeps a publisher book, the book the exchange itself holds, updated with every message including the lost ones. The last line compares the two.
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.Arrays;
public class GapRecovery {
static final int LENGTH = 0, TYPE = 2, SEQ = 3, SEND_TIME = 11, HEADER = 19;
static final int SIDE = 19, PRICE = 20, QTY = 28, LEVEL_LENGTH = 36;
static final int COUNT = 19, ENTRIES = 21, ENTRY = 17;
static final byte LEVEL = 1, HEARTBEAT = 2, SNAPSHOT = 3, BID = 'B', ASK = 'A';
static final class Decoder {
private ByteBuffer buf;
private int at;
Decoder wrap(ByteBuffer buf, int offset) { this.buf = buf; this.at = offset; return this; }
int length() { return Short.toUnsignedInt(buf.getShort(at + LENGTH)); }
byte type() { return buf.get(at + TYPE); }
long seq() { return buf.getLong(at + SEQ); }
byte side() { return buf.get(at + SIDE); }
long price() { return buf.getLong(at + PRICE); }
long qty() { return buf.getLong(at + QTY); }
int count() { return Short.toUnsignedInt(buf.getShort(at + COUNT)); }
byte entrySide(int i) { return buf.get(at + ENTRIES + i * ENTRY); }
long entryPrice(int i) { return buf.getLong(at + ENTRIES + i * ENTRY + 1); }
long entryQty(int i) { return buf.getLong(at + ENTRIES + i * ENTRY + 9); }
}
static final class Book {
final long base;
final long[] bids, asks;
int bestBid = -1, bestAsk;
Book(long base, int levels) { this.base = base; bids = new long[levels]; asks = new long[levels]; bestAsk = levels; }
void set(byte side, long price, long qty) {
long i = price - base;
if (i < 0 || i >= bids.length) return;
int idx = (int) i;
if (side == BID) {
bids[idx] = qty;
if (qty > 0) { if (idx > bestBid) bestBid = idx; }
else if (idx == bestBid) { while (bestBid >= 0 && bids[bestBid] == 0) bestBid--; }
} else {
asks[idx] = qty;
if (qty > 0) { if (idx < bestAsk) bestAsk = idx; }
else if (idx == bestAsk) { while (bestAsk < asks.length && asks[bestAsk] == 0) bestAsk++; }
}
}
void clear() { Arrays.fill(bids, 0); Arrays.fill(asks, 0); bestBid = -1; bestAsk = asks.length; }
boolean sameAs(Book o) { return Arrays.equals(bids, o.bids) && Arrays.equals(asks, o.asks); }
String top() {
return (bestBid < 0 ? "-" : bids[bestBid] + " @ " + (base + bestBid)) + " | "
+ (bestAsk == asks.length ? "-" : asks[bestAsk] + " @ " + (base + bestAsk));
}
// Writes a SNAPSHOT message of every non-empty level at the buffer's position.
void writeSnapshot(ByteBuffer out, long seq) {
int at = out.position(), n = 0;
for (int i = 0; i < bids.length; i++) {
if (bids[i] > 0) writeEntry(out, at, n++, BID, base + i, bids[i]);
if (asks[i] > 0) writeEntry(out, at, n++, ASK, base + i, asks[i]);
}
int length = ENTRIES + n * ENTRY;
out.putShort(at + LENGTH, (short) length).put(at + TYPE, SNAPSHOT).putLong(at + SEQ, seq)
.putLong(at + SEND_TIME, 0).putShort(at + COUNT, (short) n);
out.position(at + length);
}
private static void writeEntry(ByteBuffer out, int at, int n, byte side, long price, long qty) {
int e = at + ENTRIES + n * ENTRY;
out.put(e, side).putLong(e + 1, price).putLong(e + 9, qty);
}
}
static final class Handler {
final Book book = new Book(9_990, 32);
final Decoder msg = new Decoder(), replayMsg = new Decoder();
final ByteBuffer pending = ByteBuffer.allocateDirect(4096).order(ByteOrder.LITTLE_ENDIAN);
int pendingLen;
long expected = 1;
boolean recovering;
int gaps, duplicates, buffered, replayed, skipped;
void onMessage(ByteBuffer buf, int offset) {
msg.wrap(buf, offset);
if (msg.type() == SNAPSHOT) { onSnapshot(msg); return; }
if (recovering) { buffer(buf, offset, msg.length()); return; }
long seq = msg.seq();
if (seq < expected) { duplicates++; return; }
if (seq > expected) {
gaps++;
System.out.println("gap at seq " + seq + ", expected " + expected + ": RECOVERING");
recovering = true;
book.clear();
pendingLen = 0;
buffer(buf, offset, msg.length());
return;
}
apply(msg);
}
private void apply(Decoder m) {
if (m.type() == LEVEL) book.set(m.side(), m.price(), m.qty());
expected = m.seq() + 1;
}
private void buffer(ByteBuffer buf, int offset, int length) {
pending.put(pendingLen, buf, offset, length);
pendingLen += length;
buffered++;
}
private void onSnapshot(Decoder s) {
if (!recovering) return;
book.clear();
for (int i = 0; i < s.count(); i++) book.set(s.entrySide(i), s.entryPrice(i), s.entryQty(i));
expected = s.seq() + 1;
recovering = false;
System.out.println("snapshot seq " + s.seq() + " loaded, " + s.count() + " levels: LIVE");
for (int off = 0; off < pendingLen; off += replayMsg.length()) {
replayMsg.wrap(pending, off);
if (replayMsg.seq() < expected) { skipped++; continue; }
if (replayMsg.seq() > expected) { gaps++; recovering = true; book.clear(); break; }
apply(replayMsg);
replayed++;
}
pendingLen = 0;
}
}
static void level(ByteBuffer buf, long seq, byte side, long price, long qty) {
int at = buf.position();
buf.putShort(at + LENGTH, (short) LEVEL_LENGTH).put(at + TYPE, LEVEL).putLong(at + SEQ, seq)
.putLong(at + SEND_TIME, seq * 1_000).put(at + SIDE, side)
.putLong(at + PRICE, price).putLong(at + QTY, qty);
buf.position(at + LEVEL_LENGTH);
}
public static void main(String[] args) {
ByteBuffer wire = ByteBuffer.allocateDirect(4096).order(ByteOrder.LITTLE_ENDIAN);
ByteBuffer lost = ByteBuffer.allocateDirect(1024).order(ByteOrder.LITTLE_ENDIAN);
ByteBuffer snapshotChannel = ByteBuffer.allocateDirect(1024).order(ByteOrder.LITTLE_ENDIAN);
Book publisher = new Book(9_990, 32);
Decoder d = new Decoder();
long[][] script = {
{1, BID, 10_000, 500}, {2, ASK, 10_002, 300}, {3, BID, 9_999, 200}, {4, ASK, 10_003, 700},
{5, BID, 10_001, 100}, {6, ASK, 10_002, 0}, // lost on the wire
{7, BID, 10_000, 650}, {8, ASK, 10_004, 50}, {9, BID, 10_001, 0}, {10, ASK, 10_003, 400}
};
for (long[] s : script) {
boolean dropped = s[0] == 5 || s[0] == 6;
ByteBuffer target = dropped ? lost : wire;
int at = target.position();
level(target, s[0], (byte) s[1], s[2], s[3]);
d.wrap(target, at);
publisher.set(d.side(), d.price(), d.qty());
if (s[0] == 8) publisher.writeSnapshot(snapshotChannel, 8); // snapshot as of seq 8
if (s[0] == 9) { // snapshot arrives after seq 9
snapshotChannel.flip();
wire.put(snapshotChannel);
}
}
level(wire, 4, ASK, 10_003, 700); // late duplicate
int limit = wire.position();
Handler h = new Handler();
for (int offset = 0; offset < limit; offset += d.length()) {
d.wrap(wire, offset);
h.onMessage(wire, offset);
if (d.type() == LEVEL) {
System.out.println("seq " + d.seq() + (h.recovering ? " buffered" : ": " + h.book.top()));
}
}
System.out.println("gaps " + h.gaps + ", duplicates " + h.duplicates + ", buffered " + h.buffered
+ ", replayed " + h.replayed + ", skipped " + h.skipped);
System.out.println("book matches publisher: " + h.book.sameAs(publisher));
}
}seq 1: 500 @ 10000 | -
seq 2: 500 @ 10000 | 300 @ 10002
seq 3: 500 @ 10000 | 300 @ 10002
seq 4: 500 @ 10000 | 300 @ 10002
gap at seq 7, expected 5: RECOVERING
seq 7 buffered
seq 8 buffered
seq 9 buffered
snapshot seq 8 loaded, 5 levels: LIVE
seq 10: 650 @ 10000 | 400 @ 10003
seq 4: 650 @ 10000 | 400 @ 10003
gaps 1, duplicates 1, buffered 3, replayed 1, skipped 2
book matches publisher: trueMessages 7, 8, and 9 arrive while the handler is recovering, so they are copied into the pending buffer. The snapshot reflects the publisher's book after seq 8, including the effect of the lost messages 5 and 6. Replay skips 7 and 8, which the snapshot already contains, and applies 9. Applying 7 again would not corrupt a market-by-price book, because LEVEL sets a total, but skipping keeps the rule simple and also holds for feeds whose messages are deltas.
The sequence of steps matters for correctness:
- The handler clears the book at the gap, so no reader sees a book built on missing data.
- It keeps buffering until the snapshot arrives, because the snapshot may be older than the newest incremental message.
- It replays only messages with
seq > snapshot.seq, in order, and stops if the pending buffer itself has a hole.
In a real feed the handler sends a snapshot or retransmission request when it enters RECOVERING; here the script plays the role of the snapshot channel. The case where the pending buffer fills before the snapshot arrives is not handled; the extension tasks list it.
Stage 4: a measured, allocation-free pipeline
The last stage runs the Stage 3 handler on a generated feed of 100,000 messages. A linear congruential generator (LCG), a formula that computes the next pseudo-random number from the previous one, makes every run identical. Bids use the lower 32 ticks of a 64-tick window and asks the upper 32, so the book never crosses. Every 10,000 messages, three messages are dropped, the snapshot is taken five messages later, and it arrives two messages after that, so each recovery skips and replays buffered messages.
Two histograms are recorded:
| Histogram | Value per message | Printed as |
|---|---|---|
| Feed latency | Receive time minus sendTime, both synthetic and generated with the feed |
Exact percentiles, because the values are deterministic |
| Handler time | System.nanoTime around onMessage |
Checks only, because the values depend on the machine |
Each Histogram is created with a fixed highest trackable value, so recordValue writes into existing arrays and does not allocate. Values above the limit would throw, so handler times are capped first. After 20 warmup passes, one more pass is measured with getThreadAllocatedBytes.
import java.lang.management.ManagementFactory;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.Arrays;
import org.HdrHistogram.Histogram;
public class MarketDataHandler {
static final int LENGTH = 0, TYPE = 2, SEQ = 3, SEND_TIME = 11, HEADER = 19;
static final int SIDE = 19, PRICE = 20, QTY = 28, LEVEL_LENGTH = 36;
static final int COUNT = 19, ENTRIES = 21, ENTRY = 17;
static final byte LEVEL = 1, HEARTBEAT = 2, SNAPSHOT = 3, BID = 'B', ASK = 'A';
static final long BASE = 10_000;
static final int LEVELS = 64;
static final class Decoder {
private ByteBuffer buf;
private int at;
Decoder wrap(ByteBuffer buf, int offset) { this.buf = buf; this.at = offset; return this; }
int length() { return Short.toUnsignedInt(buf.getShort(at + LENGTH)); }
byte type() { return buf.get(at + TYPE); }
long seq() { return buf.getLong(at + SEQ); }
long sendTime() { return buf.getLong(at + SEND_TIME); }
byte side() { return buf.get(at + SIDE); }
long price() { return buf.getLong(at + PRICE); }
long qty() { return buf.getLong(at + QTY); }
int count() { return Short.toUnsignedInt(buf.getShort(at + COUNT)); }
byte entrySide(int i) { return buf.get(at + ENTRIES + i * ENTRY); }
long entryPrice(int i) { return buf.getLong(at + ENTRIES + i * ENTRY + 1); }
long entryQty(int i) { return buf.getLong(at + ENTRIES + i * ENTRY + 9); }
}
static final class Book {
final long base;
final long[] bids, asks;
int bestBid = -1, bestAsk;
Book(long base, int levels) { this.base = base; bids = new long[levels]; asks = new long[levels]; bestAsk = levels; }
void set(byte side, long price, long qty) {
long i = price - base;
if (i < 0 || i >= bids.length) return;
int idx = (int) i;
if (side == BID) {
bids[idx] = qty;
if (qty > 0) { if (idx > bestBid) bestBid = idx; }
else if (idx == bestBid) { while (bestBid >= 0 && bids[bestBid] == 0) bestBid--; }
} else {
asks[idx] = qty;
if (qty > 0) { if (idx < bestAsk) bestAsk = idx; }
else if (idx == bestAsk) { while (bestAsk < asks.length && asks[bestAsk] == 0) bestAsk++; }
}
}
void clear() { Arrays.fill(bids, 0); Arrays.fill(asks, 0); bestBid = -1; bestAsk = asks.length; }
boolean sameAs(Book o) {
return Arrays.equals(bids, o.bids) && Arrays.equals(asks, o.asks)
&& bestBid == o.bestBid && bestAsk == o.bestAsk;
}
void writeSnapshot(ByteBuffer out, long seq, long sendTime) {
int at = out.position(), n = 0;
for (int i = 0; i < bids.length; i++) {
if (bids[i] > 0) writeEntry(out, at, n++, BID, base + i, bids[i]);
if (asks[i] > 0) writeEntry(out, at, n++, ASK, base + i, asks[i]);
}
int length = ENTRIES + n * ENTRY;
out.putShort(at + LENGTH, (short) length).put(at + TYPE, SNAPSHOT).putLong(at + SEQ, seq)
.putLong(at + SEND_TIME, sendTime).putShort(at + COUNT, (short) n);
out.position(at + length);
}
private static void writeEntry(ByteBuffer out, int at, int n, byte side, long price, long qty) {
int e = at + ENTRIES + n * ENTRY;
out.put(e, side).putLong(e + 1, price).putLong(e + 9, qty);
}
}
static final class Handler {
final Book book = new Book(BASE, LEVELS);
final Decoder msg = new Decoder(), replayMsg = new Decoder();
final ByteBuffer pending = ByteBuffer.allocateDirect(64 * 1024).order(ByteOrder.LITTLE_ENDIAN);
int pendingLen;
long expected;
boolean recovering;
long gaps, duplicates, snapshots, replayed, skipped, updates;
void reset() {
book.clear();
pendingLen = 0;
expected = 1;
recovering = false;
gaps = duplicates = snapshots = replayed = skipped = updates = 0;
}
void onMessage(ByteBuffer buf, int offset) {
msg.wrap(buf, offset);
if (msg.type() == SNAPSHOT) { onSnapshot(msg); return; }
if (recovering) { buffer(buf, offset, msg.length()); return; }
long seq = msg.seq();
if (seq < expected) { duplicates++; return; }
if (seq > expected) {
gaps++;
recovering = true;
book.clear();
pendingLen = 0;
buffer(buf, offset, msg.length());
return;
}
apply(msg);
}
private void apply(Decoder m) {
if (m.type() == LEVEL) { book.set(m.side(), m.price(), m.qty()); updates++; }
expected = m.seq() + 1;
}
private void buffer(ByteBuffer buf, int offset, int length) {
pending.put(pendingLen, buf, offset, length);
pendingLen += length;
}
private void onSnapshot(Decoder s) {
if (!recovering) return;
snapshots++;
book.clear();
for (int i = 0; i < s.count(); i++) book.set(s.entrySide(i), s.entryPrice(i), s.entryQty(i));
expected = s.seq() + 1;
recovering = false;
for (int off = 0; off < pendingLen; off += replayMsg.length()) {
replayMsg.wrap(pending, off);
if (replayMsg.seq() < expected) { skipped++; continue; }
if (replayMsg.seq() > expected) { gaps++; recovering = true; book.clear(); break; }
apply(replayMsg);
replayed++;
}
pendingLen = 0;
}
}
// Generated feed: the wire bytes, a synthetic receive time per message, and the publisher's book.
static final class Feed {
final ByteBuffer wire = ByteBuffer.allocateDirect(8 * 1024 * 1024).order(ByteOrder.LITTLE_ENDIAN);
final long[] receiveTime = new long[200_000];
final Book publisher = new Book(BASE, LEVELS);
int messages, limit;
long state = 42;
int next(int bound) {
state = state * 6364136223846793005L + 1442695040888963407L;
return (int) ((state >>> 33) % bound);
}
Feed(int total) {
ByteBuffer lost = ByteBuffer.allocateDirect(1024).order(ByteOrder.LITTLE_ENDIAN);
ByteBuffer snapshotChannel = ByteBuffer.allocateDirect(4096).order(ByteOrder.LITTLE_ENDIAN);
Decoder d = new Decoder();
for (long seq = 1; seq <= total; seq++) {
long sendTime = seq * 10_000;
boolean dropped = seq % 10_000 >= 1 && seq % 10_000 <= 3 && seq > 3;
ByteBuffer target = dropped ? lost.clear() : wire;
int at = target.position();
if (next(10) == 0) {
target.putShort(at + LENGTH, (short) HEADER).put(at + TYPE, HEARTBEAT)
.putLong(at + SEQ, seq).putLong(at + SEND_TIME, sendTime).position(at + HEADER);
} else {
boolean bid = next(2) == 0;
long price = BASE + (bid ? next(32) : 32 + next(32));
long qty = next(5) * 100L;
target.putShort(at + LENGTH, (short) LEVEL_LENGTH).put(at + TYPE, LEVEL)
.putLong(at + SEQ, seq).putLong(at + SEND_TIME, sendTime).put(at + SIDE, bid ? BID : ASK)
.putLong(at + PRICE, price).putLong(at + QTY, qty).position(at + LEVEL_LENGTH);
d.wrap(target, at);
publisher.set(d.side(), d.price(), d.qty());
}
if (!dropped) receiveTime[messages++] = sendTime + delay();
if (seq % 10_000 == 8 && seq > 8) publisher.writeSnapshot(snapshotChannel.clear(), seq, sendTime);
if (seq % 10_000 == 10 && seq > 10) {
snapshotChannel.flip();
wire.put(snapshotChannel);
receiveTime[messages++] = sendTime + delay();
}
}
limit = wire.position();
}
// Synthetic network delay: 20 to 25 microseconds, with a 300 microsecond spike every 1,000 messages.
long delay() {
return 20_000 + next(5_000) + (messages % 1_000 == 999 ? 300_000 : 0);
}
}
static void runPass(Feed feed, Handler h, Decoder d, Histogram feedLatency, Histogram handlerTime) {
h.reset();
feedLatency.reset();
handlerTime.reset();
int i = 0;
for (int offset = 0; offset < feed.limit; offset += d.length(), i++) {
d.wrap(feed.wire, offset);
long start = System.nanoTime();
h.onMessage(feed.wire, offset);
long took = System.nanoTime() - start;
feedLatency.recordValue(feed.receiveTime[i] - d.sendTime());
handlerTime.recordValue(Math.min(took, 1_000_000_000L));
}
}
public static void main(String[] args) {
Feed feed = new Feed(100_000);
Handler h = new Handler();
Decoder d = new Decoder();
Histogram feedLatency = new Histogram(10_000_000L, 3);
Histogram handlerTime = new Histogram(1_000_000_000L, 3);
for (int pass = 0; pass < 20; pass++) runPass(feed, h, d, feedLatency, handlerTime);
com.sun.management.ThreadMXBean threads =
(com.sun.management.ThreadMXBean) ManagementFactory.getThreadMXBean();
long id = Thread.currentThread().threadId();
long before = threads.getThreadAllocatedBytes(id);
runPass(feed, h, d, feedLatency, handlerTime);
long allocated = threads.getThreadAllocatedBytes(id) - before;
System.out.println("messages on wire: " + feed.messages + ", wire bytes: " + feed.limit);
System.out.println("updates " + h.updates + ", gaps " + h.gaps + ", snapshots " + h.snapshots
+ ", skipped " + h.skipped + ", replayed " + h.replayed + ", duplicates " + h.duplicates);
System.out.println("book matches publisher: " + h.book.sameAs(feed.publisher) + ", live: " + !h.recovering);
System.out.println("best bid " + (BASE + h.book.bestBid) + ", best ask " + (BASE + h.book.bestAsk));
System.out.println("feed latency ns: p50 " + feedLatency.getValueAtPercentile(50)
+ ", p99 " + feedLatency.getValueAtPercentile(99)
+ ", p99.9 " + feedLatency.getValueAtPercentile(99.9)
+ ", max " + feedLatency.getMaxValue());
System.out.println("handler samples = messages: " + (handlerTime.getTotalCount() == feed.messages));
System.out.println("handler p50 <= p99 <= max: "
+ (handlerTime.getValueAtPercentile(50) <= handlerTime.getValueAtPercentile(99)
&& handlerTime.getValueAtPercentile(99) <= handlerTime.getMaxValue()));
System.out.println("heap bytes allocated in measured pass: " + allocated);
}
}messages on wire: 99982, wire bytes: 3436833
updates 89921, gaps 9, snapshots 9, skipped 45, replayed 18, duplicates 0
book matches publisher: true, live: true
best bid 10030, best ask 10032
feed latency ns: p50 22527, p99 24959, p99.9 44639, max 325119
handler samples = messages: true
handler p50 <= p99 <= max: true
heap bytes allocated in measured pass: 0The pipeline has three parts on the hot path: the decoder flyweight, the handler with its book and pending buffer, and the two histograms. None of them creates an object per message: the flyweight reads from the buffer, the book stores into long arrays, the pending buffer is a pre-allocated direct buffer, the counters are primitive fields, and the histograms record into their own count arrays. The measured pass therefore reports zero bytes. Printing to System.out allocates, which is why all printing happens after the measurement ends.
The feed latency line shows how a tail forms. The base delay of 20 to 25 microseconds sets p50 and p99. The nine snapshot messages carry the send time of the moment the snapshot was taken, 20 microseconds before they are sent, so they appear near 45 microseconds and p99.9 lands on them. The spike hits exactly 100 messages, 0.1% of the stream, so all of them sit above the 99.9th percentile and only max shows the 325 microsecond value: a percentile reports nothing about events rarer than its own tail, which is why max is always printed next to it. An average of these values would sit near 23 microseconds and hide the spike, which is the point made in Measuring Latency. The handler time is printed only as checks because it changes with the CPU, the JIT, and other load on the machine; on a real system it would be reported as percentiles, and the gap between the two histograms shows how much of the end-to-end latency belongs to the handler.
The book matches the publisher's book after 100,000 messages and nine recoveries, which checks requirements R2 to R4 together: any mistake in gap handling, snapshot loading, or replay would leave a different book.
Parts of the handler
| Part | Class or field | Techniques | Lesson that teaches them |
|---|---|---|---|
| Message format | Offsets, Decoder |
Fixed-length fields, little-endian, length in header | Binary Protocols |
| Decoding | Decoder.wrap |
Flyweight over a direct ByteBuffer, absolute reads |
ByteBuffer and Direct Buffers |
| Gap detection | expected |
Sequence numbers, duplicates dropped | Binary Protocols |
| Book | Book |
Arrays indexed by price ticks, cached best levels | CPU Caches and Locality |
| Recovery | recovering, pending, onSnapshot |
State machine, pre-allocated buffer, replay after snapshot | This lesson |
| Measurement | Histogram, ThreadMXBean |
HdrHistogram, allocation per pass | Measuring Latency, Allocation-Free Code |
Extension tasks
- Handle a full pending buffer: stop buffering, count the overflow, and wait for a snapshot newer than the last discarded message.
- Send an explicit snapshot request when the handler enters RECOVERING, and time out and re-request if no snapshot arrives.
- Add A/B line arbitration: feed the same stream twice with different losses and keep the first copy of each sequence number, so most gaps never reach recovery.
- Recenter the price window when a level arrives outside it, copying the arrays once instead of rejecting the update.
- Publish top-of-book changes to a strategy thread through a single-producer ring buffer from the Disruptor lesson.
- Replace the per-level book with the order-by-order book from Project: Limit Order Book and a market-by-order message type (add, modify, delete by order id).
- Read the feed from a UDP multicast socket from TCP, UDP, Multicast instead of the in-process buffer.
Interview points
- A market data handler must never apply an incremental update after a gap. The lost messages may have changed levels that later messages do not mention, so the book is unknown until a snapshot arrives.
- Recovery buffers incremental messages during the wait, loads the snapshot, and replays only messages with a sequence number above the snapshot's.
- Market-by-price LEVEL messages that carry the new total are idempotent; delta messages are not, which is why replay must skip messages the snapshot already contains.
- A flyweight decoder reads fields at fixed offsets from the receive buffer, so decoding a message allocates nothing.
- An array book indexed by price ticks gives O(1) updates; the best level is cached and rescanned only when it is emptied.
getThreadAllocatedBytesaround a warmed-up pass gives a direct check that the steady state allocates zero bytes.- HdrHistogram records into pre-allocated arrays when created with a fixed range; a value above the range throws, so inputs are capped.
- Machine-dependent timings are reported as percentiles in production and as checks in tests.
Summary
- The feed format has a 19-byte header (length, type, sequence number, send time) followed by a fixed body per type.
- The gap detector compares each sequence number with the expected one: lower is a duplicate, higher is a gap.
- The book stores total quantity per price in two
longarrays and caches the best bid and best ask. - On a gap the handler clears the book, buffers messages into a pre-allocated
ByteBuffer, and waits for a snapshot. - After loading a snapshot it skips buffered messages at or below the snapshot's sequence number and replays the rest in order.
- The generated 100,000-message feed with nine gaps ends with a book equal to the publisher's.
- A spike on exactly 0.1% of messages stays above p99.9 and appears only in
max; handler times are printed as checks. - The measured pass allocates zero heap bytes after warmup.