Compare commits

...
29 changed files with 3860 additions and 68 deletions

No files matched your search

+942
View File
@@ -0,0 +1,942 @@
# Security ID handling in `sparkd`
This document traces how security identifiers are created, translated, persisted,
indexed, and published by the `sparkd` market-data paths. It focuses on the
legacy CME MDP3/OnixS pipeline and the newer generalized market-data pipeline.
## Executive summary
`sparkd` uses two identifiers that must not be confused:
| Identifier | Meaning | Typical field | Lifetime |
|---|---|---|---|
| Exchange/feed ID | The venue's identifier used to route incoming definitions, books, and trades | `FixedSecurity::numeric_id`, `ExchangeKey`, or `domain::Security::security_common.exchange_sec_id` | Feed/session-specific unless the source guarantees stability |
| Spark security key | Spark's internal non-zero identifier used in shared memory, APIs, orders, positions, and downstream services | `FixedSecurity::key`, `api::Security::id()`, `SecurityKey`; historically also called `security_id` | Intended to be stable across restarts |
`invalid_security_key` is `0`; a security with key `0` is not yet publicly
discoverable through the shared-memory consumer path.
The two principal paths differ:
- **MDP3/OnixS:** the definition callback sets only the CME ID in
`numeric_id`. `MDProducer::add_security()` allocates the shared-memory entry
with key `0`. `SecurityIDRecorder`, driven by a 200 ms dispatcher timer,
resolves or inserts a stable key in `key_ids.sqlite`; its callback finally
calls `MDProducerHandle::set_key()`.
- **Generalized/newer pipeline:** `GeneralMarketDataListener::on_securities()`
currently uses the ID already present on the incoming domain security
(`use_given_sec_key == true`). Live adapters normally obtain that ID
synchronously from `IDGenerator` first. `_copy_security_common()` then copies
it to both `FixedSecurity::key` and `numeric_id`, so
`MDProducer::add_security()` publishes it synchronously. Only the listener's
own fallback `IDGenerator` call is bypassed by the current constant.
## End-to-end diagrams
### CME MDP3 / OnixS
```mermaid
sequenceDiagram
participant FE as OnixS feed-engine thread
participant L as MDP3Listener
participant P as MDProducer / shared memory
participant Q as SecurityIDRecorder queues
participant D as sparkd dispatcher thread
participant DB as key_ids.sqlite
FE->>L: instrument definition (CME SecurityID)
L->>L: set_security_common() - numeric_id = CME SecurityID, key remains 0
L->>P: add_security(FixedSecurity)
P->>P: allocate hot + cold entries with key 0
P->>Q: ensure_key(handle)
L->>L: index[CME SecurityID] = handle
Note over FE,L: Books can route by CME ID while the Spark key is pending
D->>Q: every 200 ms - work_spreads(), work_handles(), work_db()
Q->>DB: find (namespace, descriptor), or insert a new row
DB-->>Q: SQLite rowid = Spark security key
Q->>P: handle.set_key(rowid)
P->>P: hot.key first, then cold security.key, then increment security_count
Note over P: The security is now public to shared-memory consumers
```
### Generalized/newer pipeline (current behavior)
```mermaid
sequenceDiagram
participant A as Venue adapter / plugin
participant DB as IDGenerator / key_ids.sqlite
participant G as GeneralMarketDataListener
participant I as GeneralSecurityIndex
participant P as MDProducer / shared memory
A->>DB: get_or_add(venue, type, adapter key)
DB-->>A: SQLite rowid as domain SecurityID
A->>G: on_securities(domain Security list with ID)
G->>G: _find_sec() - use incoming security_common.id
G->>G: _copy_security_common() - key = id, numeric_id = id
G->>P: add_security(FixedSecurity with non-zero key)
P->>P: set_key() during add_security()
G->>I: index[id] = handle
Note over G,P: No asynchronous ID assignment in the current configuration
```
## Data-flow charts
The sequence diagrams above emphasize timing and thread transitions. These
flowcharts emphasize which identifier is carried through each representation
and where data is retained.
### MDP3 / legacy data flow
```mermaid
flowchart TD
CME["CME MDP3 packets<br/>CME SecurityID"] --> ONIXS["OnixS feed engine<br/>decode definition and updates"]
ONIXS --> DEF["MDP3Listener definition callback"]
DEF --> FIXED["FixedSecurity<br/>numeric_id = CME SecurityID<br/>key = 0"]
FIXED --> PRODUCER["MDProducer::add_security"]
PRODUCER --> PENDING["Hot and cold shared memory<br/>allocated but not public<br/>key = 0"]
PRODUCER --> QUEUE["SecurityIDRecorder<br/>pending_handle queue"]
FIXED --> CMEINDEX["CMESecurityIndex<br/>CME SecurityID to handle"]
QUEUE --> TIMER["Dispatcher timer<br/>every 200 ms"]
TIMER --> LEGS{"Spread legs<br/>all resolved?"}
LEGS -- No --> UNRESOLVED["unresolved_legs<br/>exchange IDs awaiting keys"]
UNRESOLVED --> CMEINDEX
CMEINDEX --> LEGS
LEGS -- Yes or outright --> LOGICAL["Logical identity<br/>venue + type + security_desc"]
LOGICAL --> SELECT["SQLite SELECT<br/>key_ids.sqlite"]
SELECT -- Existing row --> ROWID["SQLite rowid<br/>Spark security key"]
SELECT -- Missing row --> INSERT["Batched SQLite INSERT"]
INSERT --> ROWID
ROWID --> SETKEY["MDProducerHandle::set_key"]
SETKEY --> HOT["Write hot-entry key"]
HOT --> COLD["Write cold security key"]
COLD --> COUNT["Increment security_count"]
COUNT --> PUBLIC["Public security<br/>visible to MDConsumer"]
ONIXS --> UPDATE["Book, trade, status, or RFQ<br/>carrying CME SecurityID"]
UPDATE --> CMEINDEX
CMEINDEX --> HANDLE["MDProducerHandle"]
HANDLE --> LIVE["Update live shared-memory data"]
PUBLIC --> LIVE
```
The CME index is usable before the Spark key is assigned, so feed updates can
reach the internal handle while the security is still hidden from consumers.
The persistent rowid crosses the publication boundary only at `set_key()`.
### Generalized/newer data flow
```mermaid
flowchart TD
FEED["Venue feed, Databento,<br/>Python plugin, or replay"] --> ADAPTER["Source adapter<br/>build domain::Security"]
ADAPTER --> IDTYPE{"Live definition or<br/>recorded/pre-keyed input?"}
IDTYPE -- Live --> IDINPUT["Logical identity<br/>venue + type + adapter-chosen name"]
IDINPUT --> MUTEX["IDGenerator global mutex"]
MUTEX --> SELECT["Synchronous SQLite SELECT<br/>key_ids.sqlite"]
SELECT -- Existing row --> ROWID["SQLite rowid"]
SELECT -- Missing row --> INSERT["Synchronous SQLite INSERT<br/>autocommit"]
INSERT --> ROWID
ROWID --> DOMAINID["domain::SecurityID<br/>used as Spark ID"]
IDTYPE -- Recorded or pre-keyed --> DOMAINID
DOMAINID --> DOMAINSEC["SecurityCommon::id = Spark ID"]
DOMAINSEC --> LISTENER["GeneralMarketDataListener::on_securities"]
LISTENER --> CCMAP["cc_sec_id_map<br/>incoming ID to Spark ID"]
LISTENER --> CONVERT["_copy_security_common<br/>FixedSecurity.key = ID<br/>numeric_id = ID"]
CONVERT --> PRODUCER["MDProducer::add_security"]
PRODUCER --> SETKEY["Immediate set_key<br/>no recorder timer"]
SETKEY --> SHM["Hot and cold shared memory<br/>immediately public"]
CONVERT --> GENERALINDEX["GeneralSecurityIndex<br/>ID to handle"]
FEED --> EVENT["Book, trade, or statistics<br/>with exchange/source identity"]
EVENT --> EVENTMAP["Adapter map<br/>source identity to domain::SecurityID"]
EVENTMAP --> EVENTID["Event carries domain::SecurityID"]
EVENT --> PYHOT["Python bridge only<br/>synchronous IDGenerator lookup"]
PYHOT --> EVENTID
PYHOT -. uses same mutex and SQLite path .-> MUTEX
EVENTID --> CCMAP
CCMAP --> GENERALINDEX
GENERALINDEX --> HANDLE["MDProducerHandle"]
HANDLE --> SHMUPDATE["Update book, trade,<br/>statistics, or status"]
SHM --> SHMUPDATE
```
In the normal newer live path, persistence occurs before the definition enters
the generalized listener. The DB call therefore adds synchronous adapter-side
latency, but publication needs no later timer. Most adapters use an in-memory
map for subsequent events; the dotted Python path is the documented exception.
## The publication boundary
`MDProducerHandle::set_key()` is the definitive publication point for the
legacy shared-memory representation. Its write ordering is intentional:
1. Write the key into the hot book entry (`HotEntry` or `HotDepthEntry`).
2. Write the key into the cold security entry.
3. Increment the cold header's public `security_count`.
The cold security key is written last because consumers treat a non-zero cold
key as visibility. At that moment the corresponding hot entry must already be
addressable. `set_key()` only permits the transition from `0` to one key; it is
not a key-update mechanism.
## Identifier representations
The names are historically inconsistent. In particular, many protobuf fields
named `security_id` contain the **Spark key**, not the exchange's ID.
| Layer | Exchange/feed identifier | Spark identifier |
|---|---|---|
| MDP3 wire/OnixS | `securityId()` | not present |
| Legacy fixed security | `FixedSecurity::numeric_id` (`uint64_t`) | `FixedSecurity::key` / `security_key` (`uint32_t`) |
| Legacy API security | venue-specific `exchange_id` | top-level `Security.id()` |
| New domain model | `SecurityCommon::exchange_sec_id` (string) is descriptive; `SecurityCommon::id` is also used to correlate source events | `domain::SecurityID` (`uint64_t`), currently copied into the Spark key |
| Shared-memory hot book | none | `HotEntry::key` or `HotDepthEntry::key` |
| Shared-memory cold metadata | `FixedSecurity::numeric_id` | `FixedSecurity::key` |
| CME runtime lookup | `CMESecurityIndex` key | `SecurityData::handle.key()` |
| General runtime lookup | incoming/canonical `domain::SecurityID` | `GeneralSecurityIndex` key and `handle.key()` (currently the same value) |
The generalized model declares `domain::SecurityID` as 64-bit, but the legacy
shared-memory `security_key` remains 32-bit. `_copy_security_common()` assigns a
64-bit value to both `numeric_id` and the 32-bit `key`. This is safe only while
allocated IDs fit in 32 bits; unlike the legacy lookup path, this assignment has
no explicit checked conversion.
## Which parts use exchange IDs, descriptors, or Spark security IDs?
The three identifier classes serve different jobs:
- **Exchange/external ID:** routes data inside a venue or vendor feed adapter.
- **Descriptor/natural key:** recognizes a definition logically at the
persistence, discovery, or human-facing boundary.
- **Spark security ID/key:** joins the instrument to books, orders, fills,
positions, and services after publication.
The core downstream system does not repeatedly identify a security from its
description. The description is primarily used to obtain a Spark key; after
that boundary, most joins use the numeric Spark key.
### Component-level usage matrix
| Component or data path | Exchange/external ID | Descriptor or name | Spark security ID/key |
|---|---|---|---|
| CME MDP3 definition decoder | Reads `securityId()` and stores it in `FixedSecurity::numeric_id` | Copies definition symbol/tag 55 into `security_desc` | Leaves `FixedSecurity::key` at `0` initially |
| `CMESecurityIndex` | Primary map key; CME ID resolves to a handle | Not used for update routing | Returns `handle.key()` when the Spark key has been assigned |
| MDP3 books, trades, status, statistics, and RFQs | Incoming CME ID selects the handle through `CMESecurityIndex` | Not used to locate each update | The selected handle writes keyed shared-memory/trade data after publication |
| MDP3 spread definitions | Each leg initially contains `legSecurityId()` as `exchange_key` | Parent `security_desc` identifies the spread's DB row | Each leg's Spark key and then the parent Spark key must be resolved before publication |
| Legacy ICE OnixS | Uses ICE `marketId` as `numeric_id` and index key | ICE security description becomes the logical lookup name | Uses the same recorder-assigned Spark key pattern as MDP3 |
| `SecurityIDRecorder` | Does not allocate from the external ID | Uses `(venue, security type, securities::descriptor())` to find/create a row | Treats the SQLite rowid as the resulting Spark key |
| Native generalized adapters | Use exchange strings/numeric IDs to parse definitions and correlate updates | Usually pass a description or exchange instrument name to `IDGenerator` | Put the returned ID into `SecurityCommon::id` and later events |
| Databento adapter | Uses `msg.hd.instrument_id` for feed correlation and its in-memory map | Uses the decimal instrument ID string—not the human description—as the `IDGenerator` name | Caches `instrument_id -> domain::SecurityID` and places that ID on books/trades |
| Python bridge | Reads `exchange_security_id` from definitions, books, and trades | Uses that external-ID string as the `IDGenerator` name | Resolves it to `domain::SecurityID`; currently performs this lookup for every event |
| `IDGenerator` | Does not interpret whether `name` came from an exchange ID | Uses exactly `(venue, type, caller-supplied name)` | Returns the SQLite rowid as a 64-bit ID |
| `GeneralMarketDataListener` | Does not route on `exchange_sec_id` directly | Copies `sec.desc()` into fixed metadata; its disabled fallback allocator would use `sec.desc()` | Uses incoming `SecurityCommon::id`, maps events through `cc_sec_id_map`, and copies the value into both `key` and `numeric_id` |
| `GeneralSecurityIndex` | No separate external-ID concept in the current use-given path | Not used | Maps the generalized ID/Spark key to `MDProducerHandle` |
| `MDProducer` / shared memory | Retains `numeric_id`; in legacy feeds this is the exchange ID, but in the generalized path it is currently the Spark ID | Stores `security_desc` in the cold definition | Uses the key in hot/cold entries; non-zero key is the visibility boundary |
| `MDConsumer` | Not normally used to find live books | Can enumerate/filter converted security definitions | Finds handles and book data by `security_key` |
| `SecurityCache` / `securities.sqlite` | Preserves venue-specific exchange data inside the serialized security blob | Maintains descriptor/name lookup indexes and supports venue/type/name queries | Primary persisted definition key is `api::Security.id()` |
| MBO processor and MBO service | May retain exchange **order** IDs, which are separate from security exchange IDs | Not used to find the instrument | Indexes snapshots and requests by `SecurityKey`; MBO protobuf `security_id` means Spark key |
| Order manager and algorithms | Execution gateways eventually use venue-specific fields from the resolved security to encode orders | Descriptions appear in logs and errors, not as the primary join | Order definitions, subscriptions, book events, native orders, spread legs, and fills use `security_key` / `Security.id()` |
| Position, PNL, and risk services | Not used as their position join | May resolve metadata for display/reporting | Positions, fills, overrides, and PNL handles are keyed by Spark security ID |
| Relay and market-data recordings | Serialized security definitions may preserve venue exchange IDs | Preserve full descriptors in recorded definitions | Books, updates, trades, and replay correlation preserve the recorded Spark key |
| Synthetic security service | No venue exchange ID; synthetics use `VENUE_BTS` | Synthetic name is the recorder's logical descriptor | Recorder assigns the SQLite rowid and synthetic metadata is keyed by it |
### Exchange/external IDs: where they stop
In MDP3 the separation is explicit:
```text
CME SecurityID
-> FixedSecurity.numeric_id
-> CMESecurityIndex lookup
-> MDProducerHandle
-> Spark security_key for downstream publication
```
The external CME ID continues to route feed messages but does not become the
order/position key. This permits the same downstream Spark key to survive a
restart as long as the logical descriptor mapping remains in
`key_ids.sqlite`, even if the feed must rediscover the CME routing ID.
The generalized path collapses more of these roles:
```text
exchange/vendor identity
-> adapter calls IDGenerator
-> domain::SecurityID
-> FixedSecurity.key and FixedSecurity.numeric_id
-> GeneralSecurityIndex and downstream Spark key
```
`SecurityCommon::exchange_sec_id` may still contain the original textual
external identifier, but `GeneralMarketDataListener` does not use it to route
events and does not put it in `numeric_id`. The adapter must translate later
events to `domain::SecurityID` before invoking the listener.
### Descriptor/name users
Descriptors are used in three main places:
1. **Key allocation:** `SecurityIDRecorder` uses the legacy venue-specific
descriptor, while `IDGenerator` uses its caller-provided `name`.
2. **Definition lookup and discovery:** `SecurityCache` supports
venue/type/name lookup; symbol filters and administrative tools use security
metadata to select or display instruments.
3. **Diagnostics:** logs include symbols and descriptions so humans can
understand a numeric key.
Descriptions generally do not drive the high-rate book, trade, order, fill, or
position paths. Those paths use numeric IDs after definition resolution.
It is important that “descriptor” does not mean one universal canonical field:
- Legacy CME uses Globex `security_desc` from tag 55.
- Many generalized adapters use `sec.desc()` or an exchange instrument name.
- Databento uses its numeric instrument ID converted to a string.
- Python uses `exchange_security_id`.
Therefore, two adapters may describe the same economic contract but request
different logical rows and receive different Spark IDs.
### Spark security-ID users
Once a security is published, its Spark key is the dominant system identifier:
```mermaid
flowchart LR
KEY["Spark security key"] --> SHM["Hot/cold shared memory"]
KEY --> MDC["MDConsumer and book subscriptions"]
KEY --> MBO["MBO snapshots and queries"]
KEY --> OMS["Orders, algorithms, and spread legs"]
KEY --> FILLS["Fills and execution records"]
KEY --> POS["Positions, PNL, and risk"]
KEY --> API["Services and client APIs"]
KEY --> REC["Relay and market-data recordings"]
KEY --> SECDB["Serialized security definitions"]
```
Fields named `security_id`, `security_key`, or top-level `Security.id()` in
these downstream areas almost always mean the Spark key. They should not be
compared directly with CME `SecurityID`, Databento `instrument_id`, or another
venue/vendor identifier without going through the relevant adapter/index
mapping.
## Persistence and allocation
### The authoritative key mapping: `key_ids.sqlite`
Both allocators use `SqliteStore("key_ids")`. With the normal configuration the
file is:
```text
<SystemConfig.data_path>/databases/key_ids/key_ids.sqlite
```
The database contains one table and two indexes:
```sql
CREATE TABLE store (ns TEXT, key TEXT, value TEXT);
CREATE UNIQUE INDEX key_index ON store(ns, key);
CREATE INDEX ns_index ON store(ns);
```
The SQLite `rowid` is the generated Spark key. `ns` is normally
`<VENUE>.<SECURITY_TYPE>` (for example `GLBX.FUTURE`), and `key` is the security
descriptor chosen by the caller. `value` is normally empty. Therefore ID
stability depends on all three of these remaining stable:
- the configured `data_path` and preservation of this database;
- the venue/type namespace;
- the descriptor supplied by the feed adapter.
The two allocator implementations share the file and convention:
- `SecurityIDRecorder` (legacy): derives the namespace from legacy API enums and
uses `securities::descriptor()`. For CME this is the Globex
`security_desc`, populated from the definition's symbol/tag 55.
- `IDGenerator` (newer): derives the namespace from new domain enums and uses
the caller's `name`. Most adapters pass `sec.desc()` or an exchange instrument
identifier; Databento deliberately passes the decimal instrument ID, and the
Python bridge uses the exchange security ID.
### Logical-row examples
The following names are illustrative. Production uses the exact string emitted
by the selected feed/adapter, including its case, spacing, and weekly-series
notation.
#### Legacy `SecurityIDRecorder`
For CME, `set_security_common()` copies the MDP3 definition's symbol/tag 55 into
Globex `security_desc`. After converting the fixed security to
`api::Security`, `securities::descriptor()` returns that same
`glbx.security_desc()` string.
| Instrument | Legacy enums | Example descriptor from `security_desc` | Row searched/inserted in `store` |
|---|---|---|---|
| Future | `VENUE_GLBX`, `SECURITYTYPE_FUTURE` | `ESZ6` | `ns='GLBX.FUTURE', key='ESZ6'` |
| Standard option | `VENUE_GLBX`, `SECURITYTYPE_OPTION` | `ESZ6 C6500` | `ns='GLBX.OPTION', key='ESZ6 C6500'` |
| Weekly option | `VENUE_GLBX`, `SECURITYTYPE_OPTION` | `ESZ6-W1 C6500` | `ns='GLBX.OPTION', key='ESZ6-W1 C6500'` |
| Spot from a legacy-compatible venue path | `VENUE_COINBASE`, `SECURITYTYPE_SPOT` | `BTC-USD` | `ns='COINBASE.SPOT', key='BTC-USD'` |
A weekly option does not get a special namespace. Standard and weekly options
are both `SECURITYTYPE_OPTION`, so both live in `GLBX.OPTION`; the exact
`security_desc` distinguishes the weekly series. For example, if the database
assigned rowids `42001` through `42004` to the rows above, those integers—not
the descriptor strings—would be the Spark security keys.
Conceptually, legacy allocation is:
```text
future: lookup_or_insert("GLBX.FUTURE", "ESZ6")
option: lookup_or_insert("GLBX.OPTION", "ESZ6 C6500")
weekly option: lookup_or_insert("GLBX.OPTION", "ESZ6-W1 C6500")
spot: lookup_or_insert("COINBASE.SPOT", "BTC-USD")
```
#### Newer `IDGenerator`
The namespace result is similar, but the third argument is chosen by the
adapter. Examples using a description-oriented adapter are:
```cpp
get_or_add(VENUE_GLBX, SECURITY_TYPE_FUTURE, "ESZ6");
// store: ns="GLBX.FUTURE", key="ESZ6"
get_or_add(VENUE_GLBX, SECURITY_TYPE_OPTION, "ESZ6 C6500");
// store: ns="GLBX.OPTION", key="ESZ6 C6500"
get_or_add(VENUE_GLBX, SECURITY_TYPE_OPTION, "ESZ6-W1 C6500");
// store: ns="GLBX.OPTION", key="ESZ6-W1 C6500"
get_or_add(VENUE_COINBASE, SECURITY_TYPE_SPOT, "BTC-USD");
// store: ns="COINBASE.SPOT", key="BTC-USD"
```
Databento intentionally produces a different logical row even when the human
description is `ESZ6`. If its instrument ID were `12345678`, it would call the
equivalent of:
```cpp
get_or_add(VENUE_GLBX, SECURITY_TYPE_FUTURE, "12345678");
// store: ns="GLBX.FUTURE", key="12345678"
```
Likewise, the Python bridge uses `exchange_security_id`. A spot definition with
`exchange_security_id="BTC-USD"` produces the `COINBASE.SPOT / BTC-USD` row,
while a provider that sends `exchange_security_id="BTCUSD"` produces a
different row and therefore a different Spark key.
These examples show why the tuple is only logical when the adapter's naming
contract is also known. `GLBX.FUTURE / ESZ6` and `GLBX.FUTURE / 12345678` may
describe the same contract to a human but are distinct rows to `key_ids.sqlite`.
Both enum-prefix removals currently produce the same namespace spelling (for
example, `GLBX.OPTION`), so both mechanisms allocate from one global SQLite
rowid space. Live rows cannot share a rowid. The table does not declare
`AUTOINCREMENT`, however, so deleting rows—especially the highest row—can allow
SQLite to reuse a number later. Security-key rows should therefore be treated
as append-only operational data.
`SqliteStore` uses WAL mode and `PRAGMA synchronous = 0`. This favors startup
and feed latency over maximum durability against a host/power failure. The
unique `(ns, key)` index prevents two different rows for the same logical
security.
### Other places the key is stored
| Location | Purpose | Authoritative for allocation? |
|---|---|---|
| Hot and cold shared-memory segments | Live publication and fast consumer lookup | No; rebuilt at process startup |
| `CMESecurityIndex` / `GeneralSecurityIndex` and adapter maps | In-process routing from an incoming ID to a producer handle | No; in-memory only |
| `<data_path>/databases/md/securities.sqlite` | Serialized legacy `api::Security` definitions keyed by Spark key; used by security cache and MDP3 Sunday preload | No; it stores metadata after a key exists |
| Relay/market-data recordings | Preserve the key embedded in recorded securities, books, and trades for deterministic replay | No; replay normally reuses the recorded value |
| Orders, fills, positions, APIs | Reference a security by the published Spark key | No |
The `SecurityIDRecorder::work_db()` comment says a published security is also
recorded in the security database, but that function does **not** write
`securities.sqlite` in the current code. Full security definitions are persisted
separately by `SecurityCache::flush_to_disk()` through
`db::insert_or_replace()`.
## CME MDP3 / OnixS pipeline in detail
### 1. Construction and thread setup
`MDP3Main::PImpl` owns one `CMESecurityIndex`, one `SecurityIDRecorder`, and one
`MDProducer`. It constructs the recorder with:
- a lookup callback that maps `ExchangeKey` through the CME index to the
handle's current Spark key; and
- `Dispatcher::current()`, which drives a 200 ms recorder timer.
In the current `MarketDataGateway` route, `OnixsGlbxSubscription` creates a
dedicated `dispatcher_thread`, constructs and starts `MDP3Main` on that thread,
then calls that thread's dispatcher `run()`. The older `md run onixs` command
uses its command/dispatcher thread similarly.
The OnixS feed engine has its own configured thread pool. Instrument and market
data callbacks therefore originate on OnixS feed-engine threads, while recorder
database work and final key assignment execute on the MDP3 dispatcher thread.
| MDP3 stage | Executing thread | DB call? |
|---|---|---|
| Decode definition and set CME `numeric_id` | OnixS feed-engine callback thread | No |
| Allocate hot/cold entry and queue handle | Same OnixS callback thread | No |
| Add CME-ID index entry | Same OnixS callback thread | No |
| Resolve spread legs | MDP3 dispatcher thread, on 200 ms timer | No |
| Select/insert Spark key | MDP3 dispatcher thread, on 200 ms timer | Yes |
| `handle.set_key()` and public visibility | MDP3 dispatcher thread, in DB callback | No additional call |
### 2. Definition callback: exchange ID is set, Spark key is not
For futures, options, and spreads, `MDP3Listener::set_security_common()` does:
```cpp
fsec.numeric_id = def.securityId();
```
It also copies symbol/descriptor/venue and pricing fields. A newly constructed
`FixedSecurity` already has `key == invalid_security_key` (`0`), and this helper
does not change it.
For a spread, every leg starts with:
```cpp
leg.exchange_key = ExchangeKey(definition_leg.legSecurityId());
leg.key = invalid_security_key;
```
Thus both the parent and legs initially carry exchange-routing identity, but no
Spark identity.
### 3. Shared-memory allocation and exchange indexing
`MDP3Listener::add_security()` performs these operations in order:
1. `_producer.add_security(security)` allocates a hot entry and a variable-sized
cold entry and copies the security into the cold segment.
2. Because the key is `0`, `MDProducer` queues the handle through
`SecurityIDRecorder::ensure_key()`.
3. `_index.add_security(security.numeric_id, handle)` immediately creates the
CME-ID-to-handle mapping.
4. The initial exchange status is written through the handle.
This ordering means incoming books can already be routed by CME ID while the
Spark key is pending. Shared-memory consumers cannot discover the security yet.
Also, `MDProducerHandle::add_trade()` only appends to the public trade queue
when `key() != 0`, so a trade received in this short window updates the cold
last-trade state but is not added to the keyed trade log.
### 4. Recorder tick: spreads, handles, then database
Every 200 ms the recorder calls, in this exact order:
1. `work_spreads()` tries to fill previously unresolved spread leg keys by
asking `CMESecurityIndex` for each leg exchange ID.
2. `work_handles()` drains new handles. Outrights are converted to an API
security and queued for DB lookup immediately. A spread without all leg keys
is retained in `pending_handle` and its leg pointers are registered in
`unresolved_legs`.
3. `work_db()` resolves queued `(venue, type, descriptor)` requests.
Because spread work precedes new-handle work, a newly seen unresolved spread
normally waits until at least the next timer tick to resolve legs. It waits
longer if any outright leg has not yet received its Spark key. A spread's own
key is not requested until every leg key is non-zero.
### 5. Lookup or generation
`work_db()` first queries `(ns, key)`:
- Existing row: use its `rowid` as the key and invoke the callback.
- Missing row: batch it into `insert_many_resolve_ids()`. SQLite assigns a new
rowid inside a transaction, the row object receives that ID, and the callback
runs after the transaction completes.
New inserts are throttled to roughly 200 per timer pass (the current `> 200`
check can admit 201) to avoid delaying heartbeats during startup. The queues are
TBB concurrent queues, but all SQLite access is serialized onto the recorder's
dispatcher thread. `check_access()` asserts this single-thread rule after the
first work pass records the dispatcher thread ID.
### 6. Final assignment and visibility
The queued callback captures a copy of `MDProducerHandle` and calls
`handle.set_key(id)` on the dispatcher thread. `set_key()` writes hot, then cold,
then increments `security_count`. From this point:
- shared-memory scans report the security;
- `CMESecurityIndex::get_key(CME_ID)` returns the Spark key;
- trades and RFQs can be emitted with a non-zero key;
- resolved spreads can copy this key into their leg structures.
### 7. Sunday preload is a fast path
For configured symbols on Sunday, `MDP3Main::prepare_listener()` loads stored
definitions from `securities.sqlite`. It converts them with
`set_security_key=true`, so both the security key and spread leg keys come from
the serialized API security. `MDProducer::add_security()` calls `set_key()`
immediately, and no recorder lookup is required for that preloaded handle. The
CME exchange ID is still restored into `numeric_id`, allowing the normal CME
index to be built.
## Generalized/newer pipeline in detail
### 1. Topology and ownership
`Run::run()` starts a detached thread named `MarketDataGateway`. On that thread,
`run_md_gwy()` obtains the process-wide `IDGenerator`, constructs
`MarketDataGateway`, and starts the configured providers. For each `(venue,
source)` returned by a provider, the gateway creates an `MdSink` containing:
- a venue-local `GeneralSecurityIndex`;
- a default `MDProducer` (not wired to a `SecurityIDRecorder`); and
- a `GeneralMarketDataListener` connecting the source to that producer/index.
The source decides its callback thread. Some sources call the listener while
starting; others own network or replay threads. The generalized listener calls
the producer inline on that source thread. The process-wide `IDGenerator`
serializes its own calls with a mutex, but the listener's transient
`cc_sec_id_map` is an ordinary `unordered_map`, so each source/listener pair is
expected to serialize security/event callbacks.
| Generalized stage | Executing thread | DB call? |
|---|---|---|
| Build/parse domain definition | Source/adaptor thread | No |
| `IDGenerator::get_or_add()` | Same source/adaptor thread | Yes, synchronously |
| `GeneralMarketDataListener::on_securities()` | Same source callback thread | No in current use-given mode |
| Copy ID to `FixedSecurity::key` and `numeric_id` | Same source callback thread | No |
| `MDProducer::add_security()` / `set_key()` | Same source callback thread | No additional call |
| Add `GeneralSecurityIndex` entry | Same source callback thread | No |
There are two important compatibility exceptions inside the provider framework:
- `OnixsGlbxSubscription::start()` ignores the supplied generalized listener
and starts `MDP3Main`; CME/OnixS therefore still follows the asynchronous
legacy flow documented above.
- `EmumbaDeribitSubscription::start()` also ignores that listener and uses
`EmMDListener`, which allocates via `IDGenerator` and writes through its own
producer/index. Its assignment is still synchronous and pre-populated, so it
resembles the newer ID flow even though it does not pass through
`GeneralMarketDataListener`.
### 2. How a new ID is constructed
For a normal live generalized adapter, ID creation happens **in the adapter,
before** `GeneralMarketDataListener::on_securities()`:
```text
logical identity = (domain venue, domain security type, adapter-chosen name)
Spark ID = SQLite rowid for that identity in key_ids.sqlite
```
`IDGenerator::get_or_add(venue, type, name)` formats a namespace by stripping
the enum prefixes:
```text
VENUE_TAIFEX + SECURITY_TYPE_FUTURE -> TAIFEX.FUTURE
```
It then calls `retrieve_one(ns, name)`. If the row exists, it returns that
row's `rowid`. If absent, it inserts `(ns, name, "")`, and SQLite assigns the
next rowid. That integer is wrapped in `domain::SecurityID` and placed in
`SecurityCommon::id`.
The third identity component is adapter-specific, and this matters when trying
to reproduce a mapping:
| Source/adaptor family | Typical `name` sent to `IDGenerator` |
|---|---|
| Binance, Coinbase, Deribit, OKX and similar native/CCAPI paths | Usually `sec.desc()` / exchange instrument name |
| Taifex | product/instrument ID string |
| Databento | `std::to_string(msg.hd.instrument_id)`, not the human-readable description |
| Python bridge | `exchange_security_id` (and it asserts Python supplied ID is zero for definitions) |
| Emumba Deribit | extracted Emumba security descriptor |
| Replay | no new allocation for a recorded definition; it reuses the recorded `SecurityCommon::id` |
So “security name” alone is not a universal logical key. Reproduction requires
the exact venue enum, exact security-type enum, and the exact adapter-specific
name used at insertion time.
### 3. What `GeneralMarketDataListener` does with that ID
`on_securities()` first calls `_find_sec(sec, true)`. Its current, compile-time
behavior is controlled by:
```cpp
const bool use_given_sec_key = true;
```
Consequently, for a newly seen incoming `domain::SecurityID` it does **not**
call the DB. It copies `security_common.id.value()` into its transient
`cc_sec_id_map`. `_add_future()`, `_add_option()`, `_add_spread()`, or
`_add_spot()` then calls `_copy_security_common()`, which sets:
```cpp
to.key = incoming_id;
to.numeric_id = incoming_id;
```
This collapses the generalized source/correlation ID and Spark shared-memory
key into the same number. The source's text `exchange_sec_id` remains on the
domain security but is not copied into `FixedSecurity::numeric_id` by this
path.
`_commit_security()` then:
1. calls `MDProducer::add_security()` with an already non-zero key;
2. causes `MDProducerHandle::set_key()` to run immediately during that call;
3. inserts the handle in `GeneralSecurityIndex` under `numeric_id` (the same
number); and
4. returns with the security already visible in shared memory.
Because the sink's producer has no recorder, supplying `id == 0` would allocate
the shared-memory entry but never assign or publish its key. Generalized live
sources must therefore provide a valid non-zero ID before `on_securities()`.
The listener contains an alternative branch for `use_given_sec_key == false`.
That branch would call `IDGenerator::get_or_add(me->venue, sec.type(),
sec.desc())` itself. It is currently disabled, and `_generate_security_key()` is
also effectively unused. Turning the flag off is not a mechanical change: an
adapter such as Databento allocates using numeric instrument ID while the
listener branch would use `sec.desc()`, potentially creating a second key for
the same instrument.
There is also a spread failure edge case. `_find_sec(sec, true)` records the
parent ID in `cc_sec_id_map` before `_add_spread()` resolves its legs. If any leg
ID is absent, `_add_spread()` logs and returns without committing the spread. A
later repeat of the parent definition is then classified as a duplicate and
ignored, so definition ordering is a hard requirement in the current code.
### 4. How later books, trades, and statistics find the security
Every later domain event carries a `domain::SecurityID`:
- `Book::security_id()`;
- `TradeRecord::sec_id`; or
- `SecurityMarketDataStatistics::security_id`.
The listener first uses `cc_sec_id_map` to translate that event ID to the value
stored in `GeneralSecurityIndex`, then obtains the `MDProducerHandle`. With
`use_given_sec_key == true`, this mapping is numerically identity-to-identity,
but it still serves as the admission check that the definition was seen.
Most high-throughput adapters avoid SQLite on market-data events:
- Databento allocates on the definition and caches
`instrument_id -> domain::SecurityID` in `sec_id_map` for books/trades.
- Taifex keeps the resolved ID in its security-details mapper.
- Other native subscriptions maintain comparable descriptor/ID maps.
The Python bridge is a notable exception: `to_bts_security_()`,
`to_bts_trade_()`, and `to_book_()` all call `get_sec_id_()`, which calls
`IDGenerator::get_id(..., add_if_new=true)`. As written, that performs at least
one SQLite `SELECT` for each converted Python definition, trade, or book. It
returns the existing ID after the first insert, but it is still a DB call and
takes the global `IDGenerator` mutex on every event.
### 5. Persistence timing and latency
The newer allocator is synchronous:
- Existing security: mutex acquisition plus one SQLite `SELECT` on the caller's
thread.
- New security: mutex acquisition, one `SELECT`, then one SQLite `INSERT`
(SQLite autocommit), also on the caller's thread.
- Listener publication: no further DB call under the current
`use_given_sec_key` setting.
The ID therefore exists in `key_ids.sqlite` before the adapter sends the
definition to the listener. There is no intentional 200 ms publication delay,
but the source callback/startup path directly pays DB and mutex latency. Bulk
definition discovery performs one lookup and possibly one insert per security;
`IDGenerator` does not batch those operations.
WAL mode helps concurrency, while `synchronous=0` reduces fsync cost. It does
not eliminate SQL parsing/stepping, mutex contention, page-cache misses, or
occasional file-system work. The impact is usually a startup/definition cost
when adapters cache the result, but it becomes hot-path latency for code such as
the current Python bridge that resolves every event through the DB.
`GeneralMarketDataListener` itself does not persist the full contract definition
to `securities.sqlite`; that remains the separate security-cache/metadata
persistence path described earlier.
## Comparison: MDP3 recorder vs generalized allocator
| Property | MDP3 / `SecurityIDRecorder` | Generalized / `IDGenerator` |
|---|---|---|
| Logical input | venue + type + legacy `security_desc` | venue + type + adapter-chosen name |
| Numeric output | SQLite rowid, checked/read as 32-bit `security_key` | SQLite rowid returned as 64-bit `domain::SecurityID`, then narrowed into 32-bit shared-memory key |
| Allocation point | After definition is copied to shared memory | Normally inside adapter before listener sees definition |
| DB execution | Asynchronous on dispatcher timer | Synchronous on caller/source thread |
| Existing ID cost | One deferred `SELECT` | One immediate `SELECT` per call |
| New ID cost | Deferred, batched transaction (about 200/pass) | Immediate `SELECT` + individual autocommit `INSERT` |
| Time until public | Usually 0–200+ ms; spreads may take extra ticks | Immediate after synchronous lookup/insert completes |
| Feed callback blocking | Queue push only for ID work | Yes, if the adapter invokes `IDGenerator` there |
| Spread legs | Parent waits asynchronously until all leg keys resolve by exchange ID | Definition processing requires every leg's incoming ID already exist in `cc_sec_id_map`; otherwise the spread is aborted |
| Runtime ID relationship | Exchange ID and Spark key remain distinct | Incoming/canonical ID, `numeric_id`, index ID, and Spark key are currently the same number |
| Persistence database | `key_ids.sqlite` | The same `key_ids.sqlite` |
| Batching | Yes for new rows | No |
| Failure behavior | Security has an internal handle with key 0 while normal queued work is pending; DB work does not block the feed callback | DB errors/zero results occur in the synchronous source path; a zero ID can leave an unpublishable entry, with no pending-key state machine |
Neither algorithm generates a UUID or a content hash. Both ultimately use an
opaque, database-assigned SQLite rowid. It is normally increasing and looks like
a sequence, but this schema does not request SQLite `AUTOINCREMENT`. The tuple
used to *look up* the row is logical; the resulting number is not derivable from
that tuple.
## Can another system reconstruct the same keys?
Not from `(venue, type, name)` alone. Another system can reconstruct the
**logical lookup tuple**, but it cannot calculate the numeric Spark key because
the key depends on the historical insertion order of every row in
`key_ids.sqlite`.
For example, two empty systems can insert the exact same `GLBX.FUTURE / ESZ6`
row and receive different IDs if other securities were inserted before it on
one system. They are sequential opaque IDs, not deterministic hashes and not
UUIDs.
Ways to preserve cross-system identity are:
1. Copy/replicate `key_ids.sqlite` (including rowids) and keep descriptor rules
identical. This preserves existing numeric keys but requires coordinated
allocation for new ones.
2. Export an explicit mapping of `(namespace, adapter key) -> Spark key`, then
import rows with their original rowids before either system allocates new
securities.
3. Treat a stable external identity (venue + type + exchange instrument ID, or
another canonical instrument identifier) as the cross-system key and carry a
translation table to each system's local Spark key.
4. Introduce a deterministic ID scheme only as a versioned migration. A hash can
be computed independently, but collision handling, descriptor
canonicalization, aliases/renames, the existing 32-bit ABI, and compatibility
with orders/recordings/databases all need explicit design.
Copying only `securities.sqlite` is insufficient as an allocator strategy. It
contains security blobs keyed by already-assigned IDs, but `key_ids.sqlite` is
the mapping used to find or allocate those IDs. Conversely, copying only
`key_ids.sqlite` preserves the numbers but not all contract metadata.
## Other assignment paths in `sparkd`
The two main pipelines reduce to four ways a producer handle can acquire a key:
1. **Already keyed input.** `MDProducer::add_security()` sees a non-zero
`FixedSecurity::key` and calls `set_key()` immediately. Generalized feeds,
Sunday preload, relay playback, and other replay paths use this mechanism.
2. **Recorder-backed live input.** The input key is zero and the producer owns a
`SecurityIDRecorder`; `ensure_key()` queues asynchronous assignment. CME
MDP3 and legacy OnixS ICE use this mechanism. ICE uses `marketId` as
`numeric_id` and otherwise follows the same recorder/index design.
3. **Service assignment.** `SyntheticSecurityService` calls
`SecurityIDRecorder::assign_key()` directly on a background dispatcher. The
callback receives a copied `api::Security` with its top-level `id` set, then
publishes it to synthetic subscribers. It uses the same `key_ids.sqlite`
rowid allocator.
4. **Unkeyed input with no recorder.** The entry is allocated with key zero and
never becomes public. This is not a valid production assignment path, but it
is the failure mode if a generalized source omits its ID.
Relay and replay deserve care when reasoning about persistence:
- The generalized `MDReplayer` forwards the recorded domain security ID and
emits later books/statistics with the recorded `sec_key`, so the listener's
current use-given mode preserves the recording's numeric identity.
- The legacy relay producer converts a recorded `api::Security` with
`set_security_key=true`, so its shared-memory publication also preserves the
embedded key.
- The legacy relay recorder additionally calls `bulk_insert(..., "REPLAY: ")`
for logical mappings. Those rows use a `REPLAY: ` namespace prefix; this is a
separate mapping population step, not a deterministic derivation of the key.
## Operational implications
- Back up `key_ids.sqlite` and `securities.sqlite` together if numeric ID and
definition continuity matter.
- Do not delete/rebuild `key_ids.sqlite` on one host and assume IDs will match
another host, even when all instruments are identical.
- Before moving a feed between adapters, compare the exact `name` each adapter
supplies to `IDGenerator`. Matching venue/type is insufficient.
- Cache generated IDs in the source after definition processing. Avoid calling
`IDGenerator` from book/trade hot paths; the Python bridge currently violates
this guideline.
- Keep generated rowids below the 32-bit shared-memory ceiling until the
`security_key` ABI is widened or explicit bounds checking is added to the
generalized path.
- Ensure spread legs are announced before generalized spread parents.
- Treat `synchronous=0` as a durability decision. A copied DB should be taken
with SQLite-aware backup/checkpoint handling rather than by casually copying
active WAL files.
Useful inspection queries (run against the configured DB path) are:
```sql
SELECT rowid AS spark_key, ns, key
FROM store
ORDER BY rowid;
SELECT rowid AS spark_key, ns, key
FROM store
WHERE ns = 'GLBX.FUTURE' AND key = '<exact-adapter-key>';
```
## Source map
- [`SecurityIDRecorder.cpp`](../src/lib_spark/spark/md/SecurityIDRecorder.cpp):
legacy queues, 200 ms work ordering, logical row construction, DB callbacks,
and spread-leg resolution.
- [`IDGenerator.cpp`](../src/lib_spark/spark/persistence/IDGenerator.cpp):
synchronous generalized lookup/allocation and process-wide mutex.
- [`SqliteStore.cpp`](../src/lib_spark/spark/persistence/SqliteStore.cpp): DB
location, schema, WAL/synchronous pragmas, individual inserts, and batch
inserts.
- [`MDProducer.cpp`](../src/lib_spark/spark/md/MDProducer.cpp) and
[`MDProducerHandle.cpp`](../src/lib_spark/spark/md/MDProducerHandle.cpp):
shared-memory allocation, recorder handoff, and ordered publication.
- [`FixedSecurity.h`](../src/lib_spark/spark/md/Shm/FixedSecurity.h) and
[`MDStructure.h`](../src/lib_spark/spark/md/Shm/MDStructure.h): identifier
widths and hot/cold storage fields.
- [`MDP3Main.cpp`](../src/app_sparkd/sparkd/md/platforms/mdp3_onixs/MDP3Main.cpp)
and
[`MDP3Listener.h`](../src/app_sparkd/sparkd/md/platforms/mdp3_onixs/MDP3Listener.h):
MDP3 construction, definition callbacks, CME ID extraction, and spread legs.
- [`MDP3Listener.cpp`](../src/app_sparkd/sparkd/md/platforms/mdp3_onixs/MDP3Listener.cpp):
producer/index insertion order.
- [`OnixsGlbxSubscription.cpp`](../src/app_sparkd/sparkd/md/glbx/OnixsGlbxSubscription.cpp):
current dedicated MDP3 dispatcher-thread ownership.
- [`MarketDataGateway.cpp`](../src/app_sparkd/sparkd/md/MarketDataGateway.cpp)
and
[`GeneralMarketDataListener.cpp`](../src/lib_spark/spark/md/GeneralMarketDataListener.cpp):
generalized source/sink topology, incoming-ID mapping, fixed-security
conversion, and publication.
- [`DatabentoDataHandler.cpp`](../src/app_sparkd/sparkd/md/databento/DatabentoDataHandler.cpp),
[`TaifexDataProcessors.h`](../src/app_sparkd/sparkd/md/taifex/TaifexDataProcessors.h),
and
[`PyMarketDataSubscription.cpp`](../src/app_sparkd/sparkd/md/PyMarketDataSubscription.cpp):
representative adapter-side allocation and caching behavior.
- [`SecurityDB.cpp`](../src/lib_spark/spark/md/SecurityDB.cpp) and
[`SecurityCache.cpp`](../src/lib_spark/spark/md/SecurityCache.cpp): separate
persistence of full security definitions.
- [`SecurityIndex.h`](../src/lib_spark/spark/md/SecurityIndex.h),
[`MBOServiceServer.cpp`](../src/lib_spark/spark/md/MBOServiceServer.cpp), and
[`MDConsumer.cpp`](../src/lib_spark/spark/md/MDConsumer.cpp): exchange-index
translation followed by Spark-keyed market-data consumption and MBO access.
- [`OrderFactory.cpp`](../src/lib_spark/spark/orderman/order_factory/OrderFactory.cpp),
[`FillStore.cpp`](../src/lib_spark/spark/orderman/actors/FillStore.cpp), and
[`PositionMap.cpp`](../src/app_sparkd/sparkd/position/PositionMap.cpp):
representative downstream use of Spark security keys.
- [`MDReplayer.cpp`](../src/app_sparkd/sparkd/md/recording/MDReplayer.cpp) and
[`RelayClient.cpp`](../src/lib_spark/spark/md/relay/RelayClient.cpp): recorded
ID reuse.
@@ -0,0 +1,7 @@
SPARKD__CONFIG_DIR=src/app_sparkd/configurations/cme_dev_replay_aws
SPARKD__DATA_DIR=/home/ubuntu/8000/sparkdata/data
SPARKD__IMAGE=sparkd_symbols
SPARKD__PYTHON_REQUIREMENTS=src/app_sparkd/configurations/cme_dev_replay_aws/requirements.txt
SPARKD__RUNTIME_CONF_DIR=/home/ubuntu/8000/sparkdata/conf
SPARKD__SHM_SIZE=512m
SPARKD__MEMORY=1g
@@ -0,0 +1,5 @@
numpy==2.2.6
pybit==5.11.0
requests==2.31.0
websocket-client==1.8.0
websockets==15.0.1
@@ -0,0 +1,83 @@
DeploymentType {
value {
type: "prod"
}
}
PublisherOptions {
value {
network_interface_a: 'ens5'
network_interface_b: 'ens5'
#these get overriden if not entered here
heartbeat_ms: 5000
heartbeat_boot_seconds: 60
skip_options: false
hot_initial: 1024
hot_growth: 1024
cold_initial: 8192
cold_growth: 2048
channel_prefix: ""
disable_heartbeat: false
}
}
CMEOptions {
value {
feed_engine {
name: "mdp3feed"
spin_time: 0
channels_per_thread: 3
data_wait: 10
udp_socket_buffer_size: 268435456
}
channel {
id: "*"
feed_engine: "mdp3feed"
use_tcp_replay: false
recovery_queue_size: 640000
lost_packet_wait_timeout_ms: 10
onixs_log_mode: CME_ONIXS_LOG_MODE_UNSPECIFIED
#max_gap_recovery_cache_size: 4
use_natural_refresh: false
join_snapshot_on_gap: false
}
}
}
Logging {
value {
initial_queue_size: 120001
max_queue_growth: 4
console {
default_level: "Info"
subsystem {
name: "md"
level: "Info"
}
subsystem {
name: "OrderMan"
level: "Info"
}
}
}
}
MarketDataConfig {
value {
glbx {
unique_check_value: 15
use_databento: false
}
# recording {
# venue: VENUE_GLBX
# num_of_seconds_to_run_file_writer: 300
# num_of_seconds_to_run_db_writer : 0
# n_book_levels: 10
# batch_size: 100 #for the db writer
# enable_book_recording: true
# enable_trade_recording: true
# enable_bbo_recording: false
# enable_tbbo_recording: false
# enable_misc_recording: true
#}
}
}
@@ -0,0 +1,61 @@
SymbolLists {
value {
list {
key: "GlbxFutures"
symbol: "ES"
}
# list {
# key: "GlbxOptions"
# symbol: "E1A"
# symbol: "E1B"
# symbol: "E1C"
# symbol: "E1D"
# symbol: "E2A"
# symbol: "E2B"
# symbol: "E2C"
# symbol: "E2D"
# symbol: "E3A"
# symbol: "E3B"
# symbol: "E3C"
# symbol: "E3D"
# symbol: "E4A"
# symbol: "E4B"
# symbol: "E4C"
# symbol: "E4D"
# symbol: "E5A"
# symbol: "E5B"
# symbol: "E5C"
# symbol: "E5D"
# symbol: "EW1"
# symbol: "EW2"
# symbol: "EW3"
# symbol: "EW4"
# symbol: "ESH9"
# symbol: "ESZ6"
# symbol: "ESH0"
# symbol: "ESM6"
# symbol: "ESZ7"
# symbol: "ESH6"
# symbol: "ESM7"
# symbol: "ESZ8"
# symbol: "ESH7"
# symbol: "ESH8"
# symbol: "ESU6"
# symbol: "ESZ5"
# symbol: "ESZ9"
# symbol: "E1D"
# symbol: "ESZ5"
# symbol: "ES"
# }
}
}
Symbols {
value {
venues {
name: "VENUE_GLBX"
future: @SymbolLists.list[GlbxFutures].symbol
# option: @SymbolLists.list[GlbxOptions].symbol
}
}
}
@@ -0,0 +1,7 @@
SPARKD__CONFIG_DIR=src/app_sparkd/configurations/ophir_prod_live_wb
SPARKD__DATA_DIR=/home/dk/8000/sparkdata/data
SPARKD__IMAGE=sparkd_symbols
SPARKD__PYTHON_REQUIREMENTS=src/app_sparkd/configurations/ophir_prod_live_wb/requirements.txt
SPARKD__RUNTIME_CONF_DIR=/home/dk/8000/sparkdata/conf
SPARKD__SHM_SIZE=8g
SPARKD__MEMORY=16g
@@ -0,0 +1,5 @@
numpy==2.2.6
pybit==5.11.0
requests==2.31.0
websocket-client==1.8.0
websockets==15.0.1
@@ -0,0 +1,101 @@
DeploymentType {
value {
type: "prod"
}
}
PublisherOptions {
value {
network_interface_a: 'enp193s0f0np0'
network_interface_b: 'enp193s0f1np1'
#these get overriden if not entered here
heartbeat_ms: 5000
heartbeat_boot_seconds: 60
skip_options: false
hot_initial: 1024
hot_growth: 1024
cold_initial: 8192
cold_growth: 2048
channel_prefix: ""
disable_heartbeat: false
}
}
CMEOptions {
value {
feed_engine {
name: "mdp3feed"
spin_time: 0
channels_per_thread: 3
data_wait: 10
udp_socket_buffer_size: 268435456
}
channel {
id: "*"
feed_engine: "mdp3feed"
use_tcp_replay: false
recovery_queue_size: 640000
lost_packet_wait_timeout_ms: 10
onixs_log_mode: CME_ONIXS_LOG_MODE_UNSPECIFIED
#max_gap_recovery_cache_size: 4
use_natural_refresh: false
join_snapshot_on_gap: false
}
}
}
Logging {
value {
initial_queue_size: 120001
max_queue_growth: 4
console {
default_level: "Info"
subsystem {
name: "md"
level: "Info"
}
subsystem {
name: "OrderMan"
level: "Info"
}
}
}
}
MarketDataConfig {
value {
glbx {
unique_check_value: 15
use_databento: false
}
# recording {
# venue: VENUE_GLBX
# num_of_seconds_to_run_file_writer: 300
# num_of_seconds_to_run_db_writer : 0
# n_book_levels: 10
# batch_size: 100 #for the db writer
# enable_book_recording: true
# enable_trade_recording: true
# enable_bbo_recording: false
# enable_tbbo_recording: false
# enable_misc_recording: true
#}
}
}
OrdermanCfg {
value {
executor {
thread_name: "oman_orders"
poll: true
hosted_actor: "MDSource"
hosted_actor: "OrderManager"
hosted_actor: "ExecutionGateway"
}
executor {
thread_name: "oman_slow"
poll: false
hosted_actor: "*"
}
fallback_cust_order_handling_inst: "Y"
}
}
@@ -0,0 +1,61 @@
SymbolLists {
value {
list {
key: "GlbxFutures"
symbol: "ES"
}
list {
key: "GlbxOptions"
symbol: "E1A"
symbol: "E1B"
symbol: "E1C"
symbol: "E1D"
symbol: "E2A"
symbol: "E2B"
symbol: "E2C"
symbol: "E2D"
symbol: "E3A"
symbol: "E3B"
symbol: "E3C"
symbol: "E3D"
symbol: "E4A"
symbol: "E4B"
symbol: "E4C"
symbol: "E4D"
symbol: "E5A"
symbol: "E5B"
symbol: "E5C"
symbol: "E5D"
symbol: "EW1"
symbol: "EW2"
symbol: "EW3"
symbol: "EW4"
symbol: "ESH9"
symbol: "ESZ6"
symbol: "ESH0"
symbol: "ESM6"
symbol: "ESZ7"
symbol: "ESH6"
symbol: "ESM7"
symbol: "ESZ8"
symbol: "ESH7"
symbol: "ESH8"
symbol: "ESU6"
symbol: "ESZ5"
symbol: "ESZ9"
symbol: "E1D"
symbol: "ESZ5"
symbol: "ES"
}
}
}
Symbols {
value {
venues {
name: "VENUE_GLBX"
future: @SymbolLists.list[GlbxFutures].symbol
option: @SymbolLists.list[GlbxOptions].symbol
}
}
}
@@ -3,9 +3,11 @@
#include "spark/core/LOCK_GUARD.h"
#include "spark/logging/logging.h"
#include "spark/md/IMarketDataListener.h"
#include "spark/md/GeneralMarketDataListener.h"
#include "spark/md/SymbolLoader.h"
#include "spark/os/os.h"
#include "spark/persistence/IDGenerator.h"
#include "spark/orderman/gateways/sim/SimExecutionGatewayFactory.h"
#include "spark/utils/Dispatcher.h"
#include "sparkd/md/platforms/IMDPlatform.h"
#include "sparkd/md/platforms/mdp3_onixs/MDP3Main.h"
@@ -29,6 +31,10 @@ using std::vector;
struct OnixsGlbxSubscription::PImpl {
GlbxMarketDataConfig config;
shared_ptr<IDGenerator> sec_id_store;
// The decision is captured by start(), before the worker thread is created.
// A source never changes output routes while it is running.
shared_ptr<IMarketDataListener> simulation_listener;
bool simulation_enabled{false};
PImpl(shared_ptr<IDGenerator> sec_id_store, GlbxMarketDataConfig config)
: config(std::move(config)), sec_id_store(std::move(sec_id_store)) {}
@@ -66,7 +72,20 @@ struct OnixsGlbxSubscription::PImpl {
FMT_JOIN(symbols, ", "));
vector<shared_ptr<IMDPlatform>> platforms;
auto main = std::make_shared<MDP3Main>();
main->start({symbols.begin(), symbols.end()}, nullptr, nullptr);
if (simulation_enabled) {
bts_assert(simulation_listener != nullptr);
LOG(md, Info,
"GLBX MDP3 route: simulation (SimExecutionGateway is enabled)");
main->start(
{symbols.begin(), symbols.end()}, nullptr, nullptr,
simulation_listener);
} else {
// Keep the production route byte-for-byte on its established
// startup path: MDP3 owns its legacy producer and SHM channels.
LOG(md, Info,
"GLBX MDP3 route: legacy production (SimExecutionGateway is disabled)");
main->start({symbols.begin(), symbols.end()}, nullptr, nullptr);
}
platforms.emplace_back(std::move(main));
dispatcher = &utils::Dispatcher::current();
heartbeat_timer.emplace(*dispatcher, 250, [&]() {
@@ -107,9 +126,24 @@ OnixsGlbxSubscription::OnixsGlbxSubscription(
OnixsGlbxSubscription::~OnixsGlbxSubscription() = default;
void OnixsGlbxSubscription::start(
shared_ptr<IMarketDataListener> /*listener*/) {
shared_ptr<IMarketDataListener> listener) {
std::atomic_thread_fence(std::memory_order_release);
bts_assert(!me_->dispatcher_thread.joinable());
me_->simulation_enabled =
orderman::SimExecutionGatewayFactory::is_sim_execution_on(
api::VENUE_GLBX);
me_->simulation_listener =
me_->simulation_enabled ? std::move(listener) : nullptr;
bts_assert(!me_->simulation_enabled || me_->simulation_listener != nullptr);
if (me_->simulation_enabled) {
auto generalized_listener =
std::dynamic_pointer_cast<GeneralMarketDataListener>(
me_->simulation_listener);
bts_assert_msg(
generalized_listener != nullptr,
"MDP3 simulation requires a GeneralMarketDataListener");
generalized_listener->set_source_security_ids_are_external();
}
me_->dispatcher_thread = std::thread{[&]() { me_->run(); }};
}
@@ -1,30 +1,58 @@
#include "MDP3Listener.h"
#include "spark/md/log_messages/trade_latency_log.h"
#include "spark/config/protos/market_data.pb.h"
#include "spark/config/protos/market_data.pb.h"
#include "spark/domain/Book.h"
#include "spark/domain/Security.h"
#include "spark/domain/TradeRecord.h"
#include "spark/md/protos/co2.pb.h"
#include "spark/utils/datetime.h"
#include "spark/utils/helper_macros.h"
#include <vector>
using namespace std;
using namespace bts::co2;
namespace bts {
namespace md {
using namespace bts::co2;
namespace bts {
namespace md {
namespace {
domain::AggressorSide to_domain_aggressor(onixs::AggressorSide::Enum side) {
if (side == onixs::AggressorSide::Buy) return domain::AggressorSide::Buy;
if (side == onixs::AggressorSide::Sell) return domain::AggressorSide::Sell;
return domain::AggressorSide::None;
}
domain::BookLevel to_domain_book_level(const PriceSize &level) {
if (level.is_nan() || level.size() == 0) return domain::BookLevel::empty();
return {domain::Price::from_double(level.price()),
domain::Quantity(level.size()), domain::Quantity(level.direct_size()), 0};
}
template <size_t N>
void copy_book_side(const std::array<PriceSize, N> &source,
domain::BookSide &destination) {
const auto count = std::min(source.size(), destination.size());
for (size_t i = 0; i < count; ++i) destination[i] = to_domain_book_level(source[i]);
}
} // namespace
MDP3Listener::MDP3Listener(
CMESecurityIndex &index,
MDProducer &producer,
MBOWriter *mbo,
const std::vector<MDProducerChannel> channels,
TradeSummarySignalDispatcher *trade_summary_dispatcher)
MBOWriter *mbo,
const std::vector<MDProducerChannel> channels,
TradeSummarySignalDispatcher *trade_summary_dispatcher,
std::shared_ptr<IMarketDataListener> simulation_listener)
: _index(index), _producer(producer),
_verbose_ticker(config::get<CMEOptions>().verbose_ticker()),
_packet_arrived(), _latest_sequence(0),
// _settlement_recorder(settles),
_symbol_filter_storage(), _symbol_filter(), _channels(channels),
_concurrency_monitor(0), _book_updates(), _mbo(mbo),
_trade_summary_dispatcher(trade_summary_dispatcher) {
_trade_summary_dispatcher(trade_summary_dispatcher),
_simulation_listener(std::move(simulation_listener)) {
_book_updates.reserve(32);
}
@@ -180,16 +208,35 @@ void MDP3Listener::onBookUpdate(
// Trades
// Called from onTradeSummary and onTradeRecovery, ends up calling
// MDProducerHandle::add_trade
void on_trade(
void on_trade(
SecurityData *data,
const onixs::Decimal &price,
int size,
onixs::AggressorSide::Enum agside,
const onixs::Timestamp &onixs_date) {
const onixs::Timestamp &onixs_date,
IMarketDataListener *simulation_listener) {
auto ts = utils::TimeStamp::now();
PriceSize ps = pricesize_of_price_size(
price, data->handle.security().display_factor, size);
auto date = ll_of_timestamp(onixs_date);
if (simulation_listener != nullptr) {
// The MDP3 private handle key is the CME security id. GMDL resolves it
// to the persisted Spark key before it creates its SHM security.
domain::TradeRecord trade;
trade.sec_id = domain::SecurityID(data->handle.key());
trade.time_stamp =
core::TimeStamp(
onixs_date.year(), static_cast<uint8_t>(onixs_date.month()),
onixs_date.day(), onixs_date.hour(), onixs_date.minute(),
onixs_date.second()) +
core::TimeSpan::from_millis(onixs_date.millisecond());
trade.agg_side = to_domain_aggressor(agside);
trade.trade_quantity = domain::Quantity(size);
trade.trade_price = domain::Price::from_double(ps.price());
std::vector<domain::TradeRecord const *> trades{&trade};
simulation_listener->on_trades(trades);
return;
}
auto date = ll_of_timestamp(onixs_date);
log_trade_latency(
TradeLatencyStage::MDP3_RECEIVED, data->handle.security(), date, ps,
map_aggressor(agside), data->electronic_volume + size, false);
@@ -288,8 +335,8 @@ void MDP3Listener::heartbeat() {
// Called by MDP3Main::prepare_listener as part of pre-load. Also called by
// MDP3Listener::onFuture, onOption and onSpread
void MDP3Listener::add_security(
const FixedSecurity &security, co2::TradingStatus status) {
void MDP3Listener::add_security(
FixedSecurity &security, co2::TradingStatus status) {
bts_assert(security.numeric_id != 0);
if (security.type == api::SECURITYTYPE_OPTION) {
const auto &option = static_cast<const FixedOptionSecurity &>(security);
@@ -304,21 +351,67 @@ void MDP3Listener::add_security(
return;
}
}
// Set up the shared memory for this security
auto handle = _producer.add_security(security);
if (_simulation_listener != nullptr) {
// A private producer still needs a stable key for its decoding cache.
// Keep that key as the CME external id; GMDL owns persisted Spark-key
// resolution and the only shared-memory publication.
security.key = security.numeric_id;
if (security.type == api::SECURITYTYPE_SPREAD) {
auto &spread = static_cast<FixedSpreadSecurityL &>(security);
for (size_t i = 0; i < spread.leg_count(); ++i) {
// The private producer validates that spread legs have keys when
// a security becomes visible. At this stage the only valid
// identity is CME's leg SecurityID; GMDL resolves it later.
spread.legs[i].key = spread.legs[i].exchange_key.value();
}
}
}
auto handle = _producer.add_security(security);
// Add it to the index so it can receive messages
_index.add_security(security.numeric_id, handle);
_index.add_security(security.numeric_id, handle);
if (_simulation_listener != nullptr) {
domain::Security domain_security(convert_security(security));
std::vector<domain::Security const *> securities{&domain_security};
_simulation_listener->on_securities(securities);
return;
}
handle.update_exchange_status(utils::TimeStamp::now(), status);
}
// Private functions
void MDP3Listener::flush_books() {
void MDP3Listener::flush_books() {
// This is called at the end of the packet
// Write all book updates to shared memory and reset for the next packet
for (auto &entry : _book_updates) {
uint32_t sequence = _latest_sequence;
for (auto &entry : _book_updates) {
uint32_t sequence = _latest_sequence;
if (_simulation_listener != nullptr) {
// A domain Book has direct and implied quantities per level. Sending
// the merged snapshot once per packet preserves both MDP3 book kinds
// and gives the simulation gateway one matching/publication point.
domain::Book book = domain::Book::empty(
domain::SecurityID(entry.handle.key()));
copy_book_side(entry.bids.first, book.bids());
copy_book_side(entry.asks.first, book.asks());
if (entry.type_flags & book_type_implied) {
for (size_t i = 0; i < entry.bids.second.size() && i < book.bids().size(); ++i) {
const auto &implied = entry.bids.second[i];
if (!implied.is_nan() && implied.size() != 0) {
book.bids()[i] = to_domain_book_level(implied);
}
}
for (size_t i = 0; i < entry.asks.second.size() && i < book.asks().size(); ++i) {
const auto &implied = entry.asks.second[i];
if (!implied.is_nan() && implied.size() != 0) {
book.asks()[i] = to_domain_book_level(implied);
}
}
}
_simulation_listener->on_book(book, _packet_arrived);
continue;
}
if (entry.type_flags == book_type_direct) {
entry.handle.update_direct(
_packet_arrived, entry.bids.first, entry.asks.first, sequence);
@@ -333,7 +426,7 @@ void MDP3Listener::flush_books() {
}
// Reset pending updates
_book_updates.clear();
}
}
void MDP3Listener::update_all_status(CoconutSecurityStatus status) {
_index.walk_security_list(
@@ -5,7 +5,8 @@
#include "spark/config/protos/toggles.pb.h"
#include "spark/io/JsonDumper.h"
#include "spark/io/json_formatter.h"
#include "spark/md/MBOProcessor.h"
#include "spark/md/MBOProcessor.h"
#include "spark/md/IMarketDataListener.h"
#include "spark/md/Shm/FixedSecurity.h"
#include "spark/md/protos/trading_status.h"
#include "spark/orderman/events/TradeSummarySignalDispatcher.h"
@@ -152,9 +153,10 @@ class MDP3Listener : public onixs::HandlerListener,
MDP3Listener(
CMESecurityIndex &index,
MDProducer &,
MBOWriter *mbo,
const std::vector<MDProducerChannel>,
TradeSummarySignalDispatcher *);
MBOWriter *mbo,
const std::vector<MDProducerChannel>,
TradeSummarySignalDispatcher *,
std::shared_ptr<IMarketDataListener> simulation_listener = nullptr);
virtual ~MDP3Listener();
//
@@ -667,7 +669,7 @@ class MDP3Listener : public onixs::HandlerListener,
void heartbeat();
// Add a new security to the index
void add_security(const FixedSecurity &security, co2::TradingStatus);
void add_security(FixedSecurity &security, co2::TradingStatus);
private:
enum SecurityFilterAction { ADD, SKIP, DELETE };
@@ -720,8 +722,9 @@ class MDP3Listener : public onixs::HandlerListener,
};
std::vector<BookData> _book_updates;
MBOWriter *_mbo;
TradeSummarySignalDispatcher *_trade_summary_dispatcher;
};
TradeSummarySignalDispatcher *_trade_summary_dispatcher;
std::shared_ptr<IMarketDataListener> _simulation_listener;
};
//////////////////////////////////////////
// Template Member Function Definitions
@@ -981,12 +984,13 @@ void MDP3Listener::onBookUpdate(
}
}
void on_trade(
void on_trade(
SecurityData *data,
const onixs::Decimal &price,
int size,
onixs::AggressorSide::Enum agside,
const onixs::Timestamp &onixs_date);
const onixs::Timestamp &onixs_date,
IMarketDataListener *simulation_listener);
template <typename TradeRecovery>
void MDP3Listener::onTradeRecovery(
@@ -1001,8 +1005,8 @@ void MDP3Listener::onTradeRecovery(
onixs::Decimal price;
arg.entry().entryPx(price);
on_trade(
data, price, size, onixs::AggressorSide::NoAggressor,
arg.message().transactTime());
data, price, size, onixs::AggressorSide::NoAggressor,
arg.message().transactTime(), _simulation_listener.get());
}
}
@@ -1025,10 +1029,12 @@ void MDP3Listener::onTradeSummary(
for (size_t x = 0; x < sz; ++x) {
const auto &d = details[x];
on_trade(
data, price, d.lastQty(), agside, arg.message().transactTime());
data, price, d.lastQty(), agside, arg.message().transactTime(),
_simulation_listener.get());
}
} else {
on_trade(data, price, size, agside, arg.message().transactTime());
on_trade(data, price, size, agside, arg.message().transactTime(),
_simulation_listener.get());
}
}
}
@@ -36,6 +36,10 @@ struct MDP3Main::PImpl {
CMESecurityIndex index;
SecurityIDRecorder recorder;
MDProducer producer;
// Simulation needs MDP3's per-security decoding state, but it must not
// become a second GLBX shared-memory writer. This producer has no shared
// memory or security mapper; GMDL is the sole public producer in that mode.
MDProducer simulation_producer{MDProducerFlags::None};
std::map<std::string, std::unique_ptr<onixs::FeedEngine>> engines;
std::map<std::string, std::unique_ptr<onixs::Handler>> handlers;
std::vector<std::unique_ptr<MDP3Listener>> listeners;
@@ -178,7 +182,8 @@ onixs::SessionSettings session_settings_for_channel(
void MDP3Main::start(
const std::vector<UserSymbol> &symbols,
MBOWriter *mbo,
TradeSummarySignalDispatcher *tssd) {
TradeSummarySignalDispatcher *tssd,
std::shared_ptr<IMarketDataListener> simulation_listener) {
std::ostringstream oss;
for (const auto &us : symbols) {
oss << us << " ";
@@ -191,7 +196,8 @@ void MDP3Main::start(
me_->engines = create_feed_engines(channels);
for (const auto &channel : channels) {
unique_ptr<onixs::Handler> phandler = prepare_handler(channel.id, mbo);
me_->listeners.push_back(prepare_listener(*phandler, channel, mbo, tssd));
me_->listeners.push_back(prepare_listener(
*phandler, channel, mbo, tssd, simulation_listener));
me_->handlers.insert(std::make_pair(channel.id, std::move(phandler)));
std::stringstream symstr;
for (const auto &sym : channel.symbols) {
@@ -386,19 +392,25 @@ unique_ptr<MDP3Listener> MDP3Main::prepare_listener(
onixs::Handler &handler,
const CMEChannel &channel,
MBOWriter *mbo,
TradeSummarySignalDispatcher *trade_summary_dispatcher) {
TradeSummarySignalDispatcher *trade_summary_dispatcher,
std::shared_ptr<IMarketDataListener> simulation_listener) {
LOG(md, Info, "MDP3Main::prepare_listener channel: {}", channel);
// We need to get a handler to the producer channels so we can produce
// heartbeats
vector<MDProducerChannel> producer_channels;
for (auto &symbol : channel.symbols) {
producer_channels.push_back(me_->producer.get_channel(
symbol.symbol().c_str(), symbol.type(), symbol.exchange()));
MDProducer &producer = simulation_listener ? me_->simulation_producer
: me_->producer;
// Private simulation state does not need heartbeat SHM channels.
if (!simulation_listener) {
for (auto &symbol : channel.symbols) {
producer_channels.push_back(producer.get_channel(
symbol.symbol().c_str(), symbol.type(), symbol.exchange()));
}
}
unique_ptr<MDP3Listener> listener(new MDP3Listener(
me_->index, me_->producer, mbo, producer_channels,
trade_summary_dispatcher));
me_->index, producer, mbo, producer_channels,
trade_summary_dispatcher, std::move(simulation_listener)));
for (auto &symbol : channel.symbols) {
listener->add_filter_symbol(symbol.symbol());
}
@@ -2,6 +2,7 @@
#include "CMESecurityIndex.h"
#include "MDP3Listener.h"
#include "spark/md/IMarketDataListener.h"
#include "spark/md/coconut2.h"
#include "spark/utils/bts_no_copy.h"
#include "onixs.h"
@@ -37,7 +38,8 @@ class MDP3Main : public IMDPlatform {
void start(
const std::vector<UserSymbol> &symbols,
MBOWriter *mbo,
TradeSummarySignalDispatcher *tssd);
TradeSummarySignalDispatcher *tssd,
std::shared_ptr<IMarketDataListener> simulation_listener = nullptr);
void send_heartbeats() override;
void initialize(const std::string &license_store);
@@ -47,7 +49,8 @@ class MDP3Main : public IMDPlatform {
onixs::Handler &handler,
const CMEChannel &channel,
MBOWriter *,
TradeSummarySignalDispatcher *);
TradeSummarySignalDispatcher *,
std::shared_ptr<IMarketDataListener> simulation_listener = nullptr);
private:
struct PImpl;
@@ -231,10 +231,16 @@ void PositionServiceServer::create_manual_fill(
for (auto mf : batch.manual_fill()) {
auto id = utils::gen_id("manual_fills");
mf.set_record_id(id);
struct timeval tv;
gettimeofday(&tv, NULL);
mf.mutable_timestamp()->set_seconds(tv.tv_sec);
mf.mutable_timestamp()->set_nanos(tv.tv_usec * 1000);
// Older clients do not send a timestamp. Retain their existing behavior,
// but preserve a timestamp supplied by newer clients so a manual fill can
// be assigned to its intended trading day.
if (!mf.has_timestamp() ||
(mf.timestamp().seconds() == 0 && mf.timestamp().nanos() == 0)) {
struct timeval tv;
gettimeofday(&tv, NULL);
mf.mutable_timestamp()->set_seconds(tv.tv_sec);
mf.mutable_timestamp()->set_nanos(tv.tv_usec * 1000);
}
me->manual_fills_store->insert(
manual_fills_ns, std::to_string(id), utils::pb::to_string(mf));
@@ -296,6 +302,16 @@ void PositionServiceServer::update_manual_fill(
this_resp->set_ok(false);
this_resp->set_msg("Could not find record");
} else {
// Legacy clients omit timestamps when editing. Do not let that erase
// the trading date that was assigned when the fill was created.
if (!mf.has_timestamp() ||
(mf.timestamp().seconds() == 0 && mf.timestamp().nanos() == 0)) {
ManualFill existing;
if (utils::pb::from_string<ManualFill>(val, &existing) &&
existing.has_timestamp()) {
mf.mutable_timestamp()->CopyFrom(existing.timestamp());
}
}
me->manual_fills_store->update(
manual_fills_ns, id, utils::pb::to_string(mf));
this_resp->set_ok(true);
@@ -28,7 +28,8 @@ namespace bts.shell.manual_fills {
nameof(user),
nameof(side),
nameof(quantity),
nameof(price)
nameof(price),
nameof(fill_date)
}, typeof(ManualFillRow).GetProperty);
readonly ISecurityMetadataStore _sec_meta_store;
@@ -43,6 +44,7 @@ namespace bts.shell.manual_fills {
UserTuple? _user;
Side _side;
double _price;
DateTime _timestamp;
bool _is_selected;
ManualFillData? _data;
IDisposable _committed_snapshot;
@@ -68,6 +70,9 @@ namespace bts.shell.manual_fills {
// TickDisplaySettingsImmutable for qty, obtained via ISecurityMetadataStore.
_quantity.conversion_data = new(1, QuantityConversionOpType.DivideActual);
_price = data.value.manual_fill.Price;
_timestamp = data.value.manual_fill.HasTimestamp
? data.value.manual_fill.Timestamp.to_local_time()
: DateTime.Now;
if (placeholder_key.HasValue) {
_is_dirty = backing(nameof(is_dirty), false);
_editable_item_state = backing(nameof(editable_item_state), EditableItemState.Uncommitted);
@@ -113,6 +118,22 @@ namespace bts.shell.manual_fills {
public TickDisplaySettingsImmutable quantity_display => _quantity_display.Value;
public double price { get => _price; set => _set(ref _price, value); }
/// <summary>
/// The trading date assigned to this fill. The time of day is retained so
/// that a fill entered for today continues to be classified as a day fill.
/// </summary>
public DateTime fill_date {
get => _timestamp.Date;
set {
var timestamp = value.Date + _timestamp.TimeOfDay;
if (_timestamp != timestamp) {
_timestamp = timestamp;
RaisePropertyChanged(nameof(fill_date));
_data = null;
}
}
}
public DateTime? time_entered =>
data.value.manual_fill.HasTimestamp
? data.value.manual_fill.Timestamp.to_local_time()
@@ -157,7 +178,8 @@ namespace bts.shell.manual_fills {
Price = price,
Side = side,
Quantity = quantity,
SecurityKey = security?.Id ?? 0
SecurityKey = security?.Id ?? 0,
Timestamp = Google.Protobuf.WellKnownTypes.Timestamp.FromDateTime(_timestamp.ToUniversalTime())
};
public ManualFillTuple to_tuple() =>
@@ -39,6 +39,8 @@
<ColumnDefinition Width="Auto"/>
<ColumnDefinition Width="{Binding Source={StaticResource HorizontalPush}}"/>
<ColumnDefinition Width="Auto"/>
<ColumnDefinition Width="{Binding Source={StaticResource HorizontalPush}}"/>
<ColumnDefinition Width="Auto"/>
</Grid.ColumnDefinitions>
<Grid.RowDefinitions>
<RowDefinition Height="Auto"/>
@@ -46,7 +48,7 @@
<RowDefinition Height="*"/>
</Grid.RowDefinitions>
<Rectangle
Grid.ColumnSpan="11"
Grid.ColumnSpan="13"
HorizontalAlignment="Stretch"
VerticalAlignment="Stretch"
Fill="{u:BTSColor header}" />
@@ -56,6 +58,7 @@
<TextBlock Grid.Column="6" Text="Side" Style="{StaticResource HeaderText}"/>
<TextBlock Grid.Column="8" Text="Qty" Style="{StaticResource HeaderText}"/>
<TextBlock Grid.Column="10" Text="Price" Style="{StaticResource HeaderText}"/>
<TextBlock Grid.Column="12" Text="Date" Style="{StaticResource HeaderText}"/>
<c:ContractPickerPopup
Grid.Row="2"
@@ -88,6 +91,7 @@
TickDisplaySettings="{Binding quantity_display}"
Value="{Binding display_quantity}" />
<usp:PriceEntryControl Grid.Row="2" Grid.Column="10" MinWidth="55" Value="{Binding price}" ContractSpec="{Binding spec}"/>
<DatePicker Grid.Row="2" Grid.Column="12" SelectedDate="{Binding fill_date}" />
</Grid>
<c:DialogSkirt Grid.Row="1" Margin="0,6,0,0">
<c:DialogSkirt.RightItems>
@@ -272,6 +272,23 @@
</DataGridTemplateColumn.CellEditingTemplate>
</DataGridTemplateColumn>
<!-- Trading Date -->
<DataGridTemplateColumn
c:DataGridEx.CellKey="{x:Static Date, MemberType={u:Nested usp:ManualFillsView.Columns}}"
Header="Date"
SortMemberPath="fill_date">
<DataGridTemplateColumn.CellTemplate>
<DataTemplate>
<TextBlock Text="{Binding fill_date, StringFormat={}{0:d}}" VerticalAlignment="Center" />
</DataTemplate>
</DataGridTemplateColumn.CellTemplate>
<DataGridTemplateColumn.CellEditingTemplate>
<DataTemplate>
<DatePicker SelectedDate="{Binding fill_date, UpdateSourceTrigger=PropertyChanged}" />
</DataTemplate>
</DataGridTemplateColumn.CellEditingTemplate>
</DataGridTemplateColumn>
<!-- Time Entered -->
<DataGridTextColumn
c:DataGridEx.CellKey="{x:Static
@@ -31,6 +31,7 @@ namespace mbts.ui.spark {
Side,
Quantity,
Price,
Date,
TimeEntered
}
+20 -3
View File
@@ -90,9 +90,26 @@
<Setter Property="CaretBrush" Value="{u:BTSBrush UI.TextBox.normal_caret}" />
</Style>
<Style x:Key="TextBox.Dark" TargetType="TextBox" BasedOn="{StaticResource {x:Type TextBox}}" />
<Style x:Key="ListBoxItem.Dark" TargetType="ListBoxItem">
<Style x:Key="TextBox.Dark" TargetType="TextBox" BasedOn="{StaticResource {x:Type TextBox}}" />
<Style TargetType="DatePicker">
<Setter Property="Background" Value="{u:BTSBrush UI.TextBox.normal_background}" />
<Setter Property="Foreground" Value="{u:BTSBrush UI.TextBox.normal_foreground}" />
<Setter Property="BorderBrush" Value="Transparent" />
<Setter Property="BorderThickness" Value="0" />
<Setter Property="FocusVisualStyle" Value="{x:Null}" />
</Style>
<Style TargetType="DatePickerTextBox">
<Setter Property="Background" Value="{u:BTSBrush UI.TextBox.normal_background}" />
<Setter Property="Foreground" Value="{u:BTSBrush UI.TextBox.normal_foreground}" />
<Setter Property="BorderBrush" Value="Transparent" />
<Setter Property="BorderThickness" Value="0" />
<Setter Property="CaretBrush" Value="{u:BTSBrush UI.TextBox.normal_caret}" />
<Setter Property="FocusVisualStyle" Value="{x:Null}" />
</Style>
<Style x:Key="ListBoxItem.Dark" TargetType="ListBoxItem">
<Setter Property="Background" Value="{StaticResource ListBoxItem.Background}" />
<Setter Property="Foreground" Value="{StaticResource ListBoxItem.Foreground}" />
<Setter Property="BorderThickness" Value="0" />
@@ -54,6 +54,7 @@
#include <google/protobuf/timestamp.pb.h>
#include <google/protobuf/util/time_util.h>
#include <memory>
#include <mutex>
#include <optional>
#include <sstream>
#include <string>
@@ -280,8 +281,9 @@ struct GeneralMarketDataListener::PImpl {
// if this is set to true, then the security key passed on in the
// on_securities function is used. If this is not set to true, then a new key
// is generated
const bool use_given_sec_key =
true; // with true, replay works, with orders goign in ((did not push it in, since it was always false - need to test))
// Normally the source provides Spark keys. MDP3 explicitly opts into
// external CME SecurityID resolution before it begins delivery.
bool use_given_sec_key{true};
const std::string cc_name_prefix = "cc_security";
size_t n_bk_updates = 0;
bts::persistence::IDGenerator *id_gen;
@@ -295,6 +297,14 @@ struct GeneralMarketDataListener::PImpl {
// (a flag is used to tell if bts::domain::SecurityID can be used as bts id
// as well)
std::unordered_map<bts::domain::SecurityID, uint64_t> cc_sec_id_map = {};
// CME can define a calendar/inter-commodity spread before every leg
// definition reaches this channel. Retain it until its external leg IDs
// have been resolved to the GMDL security namespace.
std::vector<bts::domain::Security> deferred_spreads;
// MDP3 invokes one listener per CME channel. Serialize only the simulated
// route at the GMDL boundary so definition/index/book updates remain
// ordered with respect to the single GMDL producer.
std::mutex simulation_callback_mutex;
MDArray bids = {};
MDArray asks = {};
IncrementalBookData inc_book_data = {};
@@ -385,9 +395,29 @@ GeneralMarketDataListener::GeneralMarketDataListener(
GeneralMarketDataListener::~GeneralMarketDataListener() {}
void GeneralMarketDataListener::set_source_security_ids_are_external(
bool value) {
// Sources configure this during start-up, before any callback can access
// the index. Keeping it source-scoped avoids changing Databento GLBX.
//currently all other data sources set it to true only mdp3 based cme data sets it as false
me->use_given_sec_key = !value;
}
void GeneralMarketDataListener::on_securities(
const std::vector<domain::Security const *> &complete_security_list) {
std::unique_lock<std::mutex> simulation_lock(
me->simulation_callback_mutex, std::defer_lock);
if (me->sim_ex_gwy != nullptr) simulation_lock.lock();
LOG_SECURITIES_EVENT(Info, complete_security_list.size());
const auto spread_legs_resolved = [this](const domain::Security &sec) {
if (sec.type() != SEC_TYPE(SPREAD)) return true;
const auto &spread = std::get<domain::SpreadSecurity>(sec.security);
return std::all_of(
spread.legs.begin(), spread.legs.end(), [this](const auto &leg) {
return me->cc_sec_id_map.find(leg.security_id) !=
me->cc_sec_id_map.end();
});
};
for (auto s_ptr : complete_security_list) {
const domain::Security &sec = *s_ptr;
CCSecSearchResult search_result = _find_sec(sec, true);
@@ -409,12 +439,28 @@ void GeneralMarketDataListener::on_securities(
_add_option(sec);
break;
case SEC_TYPE(SPREAD):
_add_spread(sec);
if (spread_legs_resolved(sec)) {
_add_spread(sec);
} else {
me->deferred_spreads.push_back(sec);
LOG(md, Info,
"Deferring GLBX spread {} until its CME leg definitions arrive",
sec.desc());
}
break;
default:
break;
}
}
for (auto it = me->deferred_spreads.begin();
it != me->deferred_spreads.end();) {
if (spread_legs_resolved(*it)) {
_add_spread(*it);
it = me->deferred_spreads.erase(it);
} else {
++it;
}
}
}
void GeneralMarketDataListener::on_security_state(
@@ -442,6 +488,9 @@ void GeneralMarketDataListener::on_security_statistics(
void GeneralMarketDataListener::on_book(
const bts::domain::Book &book, std::optional<utils::TimeStamp> time) {
std::unique_lock<std::mutex> simulation_lock(
me->simulation_callback_mutex, std::defer_lock);
if (me->sim_ex_gwy != nullptr) simulation_lock.lock();
LOG_BOOK(Diag, book, BookLog_Status::PROCESSING);
// in case this is called externally in conjunction iwth inc book udpates
me->inc_book_data.overwrite_book(book);
@@ -488,6 +537,9 @@ co2::AggressorSide try_fill_agg_side(
}
void GeneralMarketDataListener::on_trades(
const std::vector<domain::TradeRecord const *> &trades) {
std::unique_lock<std::mutex> simulation_lock(
me->simulation_callback_mutex, std::defer_lock);
if (me->sim_ex_gwy != nullptr) simulation_lock.lock();
utils::Stopwatch sw;
for (auto tp : trades) {
sw.lap();
@@ -504,7 +556,7 @@ void GeneralMarketDataListener::on_trades(
aggressor_side = try_fill_agg_side(tp->trade_price, aggressor_side, bk);
const auto &sec = handle->handle.security();
LOG_TRADE_RECORD(
Info, t, sec.exchange, sec.type, sec.security_desc(),
Diag, t, sec.exchange, sec.type, sec.security_desc(),
TradeRecordLog_Status::PROCESSING);
const auto trade_ts = tp->time_stamp.to_ll();
@@ -790,14 +842,30 @@ void GeneralMarketDataListener::_on_book(
me->n_bk_updates++;
auto handle = _find_handle(book.security_id());
if (handle == nullptr) {
LOG_BOOK(Error, book, BookLog_Status::ABORTED, "Sec handle not found");
const auto mapping = me->cc_sec_id_map.find(book.security_id());
const bool mapped = mapping != me->cc_sec_id_map.end();
LOG_BOOK(
Error, book, domain::DomainSecurityUtils::convert(me->venue),
api::SECURITYTYPE_UNSET, "", BookLog_Status::ABORTED,
fmt::format(
"Sec handle not found; reason={}; mapped_spark_key={}; "
"key_mode={}; registered_mappings={}; deferred_spreads={}; "
"simulation={}",
mapped ? "mapped_key_missing_from_index" : "source_id_not_mapped",
mapped ? std::to_string(mapping->second) : "unmapped",
me->use_given_sec_key ? "source_key" : "persisted_spark_key",
me->cc_sec_id_map.size(), me->deferred_spreads.size(),
me->sim_ex_gwy != nullptr));
return;
}
copy_books(book, me->bids, bts::domain::Side::Value::Buy);
copy_books(book, me->asks, bts::domain::Side::Value::Sell);
_possibly_match_with_sim_orders(
bts::md::SecurityKey(book.security_id().value()));
const auto &sec = handle->handle.security();
// `book.security_id()` is the source identity for MDP3 (the CME
// SecurityID). Simulation orders are keyed by the persisted Spark key
// owned by this GMDL handle.
_possibly_match_with_sim_orders(
bts::md::SecurityKey(handle->handle.key()));
LOG_BOOK(
Diag, book, sec.exchange, sec.type, sec.security_desc(),
BookLog_Status::PROCESSING);
@@ -23,6 +23,9 @@ class GeneralMarketDataListener : public IMarketDataListener {
api::Venue venue,
persistence::IDGenerator *sec_id_store);
virtual ~GeneralMarketDataListener();
// MDP3 identifies securities in callbacks by CME SecurityID, not by a
// persisted Spark key. Sources opt into this before they start.
void set_source_security_ids_are_external(bool value = true);
void on_securities(const std::vector<domain::Security const *>
&complete_security_list) override;
void on_security_state(const domain::SecurityState &security_state) override;
+11 -5
View File
@@ -42,15 +42,21 @@ struct BookLog {
api::Venue venue,
api::SecurityType sec_type,
std::string sec_name,
BookLog_Status status)
BookLog_Status status,
std::string message = "")
: book(std::move(book)), venue(venue), sec_type(sec_type),
sec_name(std::move(sec_name)), status(status), message(),
timestamp(core::TimeStamp::now()) {}
sec_name(std::move(sec_name)), status(status),
message(std::move(message)), timestamp(core::TimeStamp::now()) {}
explicit BookLog(
domain::Book book, BookLog_Status status, std::string message)
: book(std::move(book)), status(status), message(std::move(message)),
timestamp(core::TimeStamp::now()) {}
: BookLog(
std::move(book),
api::VENUE_UNSET,
api::SECURITYTYPE_UNSET,
"",
status,
std::move(message)) {}
};
} // namespace bts::md
+105
View File
@@ -0,0 +1,105 @@
# CME MDP 3.0 packet recorder
`cme_recorder.py` records selected CME multicast flows into a local replay
bundle. It does not change or invoke Spark.
The initial defaults are:
- Feed A on the interface supplied with `--interface`.
- Channels 310, 311, and 323.
- Incremental (`I`), instrument definition (`N`), MBP recovery (`S`), and MBO
recovery (`SMBO`) flows.
- Stop as soon as every definition and recovery flow has completed a gap-free
`TotNumReports` cycle and incremental traffic has been observed.
- Give up after five minutes and create an explicitly incomplete bundle.
## Requirements
- Linux and Python 3.8 or newer.
- `dumpcap`, normally supplied by the Wireshark command-line tools.
- Permission to capture from the CME interface. Run under the site's approved
`dumpcap` capability/group configuration, or run the command with `sudo`.
- Read access to the production `config.xml` and `templates.xml`.
- Enough local disk space for the selected options channels.
The recorder joins each selected multicast group itself. It can therefore run
without Spark, although it is also safe to run while another process has joined
the same groups.
## Validate the plan without recording
Run this first on `arbbts-svr01`:
```sh
python3 tools/cme/cme_recorder.py \
--dry-run \
--interface enp193s0f0np0 \
--cme-data-dir /home/wedbushprod/8000/sparkdata/data/cme_mdp3 \
--channels 310,311,323 \
--feeds A \
--output /home/wedbushprod/cme-recordings/es-smoke
```
The dry run verifies the interface and CME files, resolves all twelve required
multicast flows, and prints the exact capture filter. It creates no output.
## Shortest useful smoke recording
```sh
python3 tools/cme/cme_recorder.py \
--interface enp193s0f0np0 \
--cme-data-dir /home/wedbushprod/8000/sparkdata/data/cme_mdp3 \
--channels 310,311,323 \
--feeds A \
--min-seconds 1 \
--max-seconds 300 \
--output /home/wedbushprod/cme-recordings/es-smoke
```
The output directory must not already exist. The recorder will never overwrite
an existing bundle.
Expected progress messages include:
```text
[DONE] [configuration] validated 12 multicast flows
[DONE] [multicast] joined all 12 required multicast groups
[DONE] [capture] PCAPNG capture started
[INFO] [channel 310 feed A definitions] cycle started; TotNumReports=...
[DONE] [channel 310 feed A definitions] complete gap-free replay cycle received
[DONE] [channel 310 feed A MBP recovery] complete gap-free replay cycle received
[DONE] [channel 310 feed A MBO recovery] complete gap-free replay cycle received
[DONE] [capture] all required CME stages are complete
[DONE] [bundle] local replay bundle ready: ...
```
The final directory contains exactly:
```text
recording.pcapng
config.xml
templates.xml
manifest.json
checksums.sha256
```
`manifest.json` contains completion status and per-flow packet, sequence-gap,
template, and `TotNumReports` counts. A timeout or Ctrl-C still finalizes a local
bundle, but marks it incomplete and returns a non-zero status.
Verify the completed bundle with:
```sh
cd /home/wedbushprod/cme-recordings/es-smoke
sha256sum --check checksums.sha256
```
## Exit status
- `0`: every required stage completed.
- `1`: configuration, permission, dependency, or capture error.
- `2`: timeout/incomplete recording; the bundle was still finalized.
- `130`: interrupted by the operator; the bundle was still finalized.
For the later full recording, increase `--max-seconds` and add a fixed-duration
mode only after the smoke bundle has been successfully replayed.
+133
View File
@@ -0,0 +1,133 @@
# CME MDP 3.0 one-pass replayer
`cme_replayer.py` replays one recorder bundle over IPv4 UDP multicast. This is
the Phase 3 smoke-test version:
- It makes exactly one pass and exits at end-of-file.
- PCAPNG capture timestamps control the original inter-packet timing.
- The CME UDP payload is sent byte-for-byte unchanged.
- It does not rewrite CME timestamps, packet sequence numbers, or `RptSeq`.
- It does not start, stop, or modify Spark/OnixS.
The kernel constructs new Ethernet, IPv4, and UDP headers for the selected EC2
interface. The original CME source IP and UDP source port are therefore not
preserved. OnixS normally subscribes by multicast group, destination port, and
interface; this version is intentionally aimed at validating that path first.
## Requirements
- Linux and Python 3.8 or newer.
- The complete local bundle produced by `cme_recorder.py`.
- An IPv4 address and multicast route on the output interface.
- OnixS configured for the same multicast groups and ports in the bundled
`config.xml`, with its network interface set to the EC2 replay interface.
The replayer uses normal UDP sockets and does not normally need root. It sets
multicast TTL to 1 and enables local multicast loopback, allowing an OnixS
process on the same EC2 host to receive the packets.
Use a dedicated replay interface, subnet, network namespace, or isolated EC2
test network. Do not route replay groups toward a production or CME network.
## Copy the files to EC2
Copy these two repository files through your normal deployment process:
```text
tools/cme/cme_replayer.py
tools/cme/README-replayer.md
```
The replayer has no third-party Python dependencies.
## Preflight without sending packets
Run preflight first. It verifies every bundle checksum, parses the manifest and
`config.xml`, scans the entire PCAPNG file, checks capture timestamp ordering,
and requires packets for every recorded multicast destination.
PCAPNG packets from multiple multicast flows or receive queues can occasionally
appear in file order with a slightly earlier capture timestamp than the packet
before them. Preflight reports the count and maximum regression but does not
reject the recording. Replay preserves file order and sends such a packet
immediately. This affects pacing metadata only; the CME payload remains
unchanged.
```sh
python3 tools/cme/cme_replayer.py \
--dry-run \
--bundle /path/to/es-smoke \
--interface eth0
```
Replace `eth0` with the interface configured for OnixS on EC2. Successful
preflight prints every multicast `group:port` and finishes with:
```text
[DONE] [dry-run] preflight complete; no packets were sent
```
If the initial recorder smoke bundle was finalized before every recovery stage
completed, add `--allow-incomplete`. This should only be used intentionally:
```sh
python3 tools/cme/cme_replayer.py \
--dry-run \
--allow-incomplete \
--bundle /path/to/es-smoke \
--interface eth0
```
## One-pass replay
Start OnixS first and wait until it has joined all configured multicast groups.
Then run:
```sh
python3 tools/cme/cme_replayer.py \
--bundle /path/to/es-smoke \
--interface eth0 \
--start-delay-seconds 2
```
The program validates and scans the bundle before sending anything. It then
waits for the start delay and sends one pass at 1x recorded timing. Expected
lifecycle messages include:
```text
[DONE] [bundle] checksums and manifest validated
[DONE] [capture] validated ... recorded duration=...s
[DONE] [network] validated output interface ...
[INFO] [replay] one-pass replay starts in 2.000s
[DONE] [replay] one-pass replay complete: ...
```
`Ctrl-C` stops the replay cleanly with exit status 130. There is no looping.
## Verify packet delivery independently
On the EC2 host, observe one expected flow on the OnixS interface while replay
is running:
```sh
sudo tcpdump -ni eth0 'dst host 224.0.31.1 and udp dst port 14310'
```
For the first application test, restart OnixS before each replay. Starting the
same recording twice against a continuously running receiver sends the original
sequence numbers again and may be treated as duplicate or stale data.
Because CME timestamps are unchanged, expect historical exchange times and
large reported latency. The purpose of this version is only to establish
whether definitions, recovery data, books, and trades flow through OnixS.
## Exit status
- `0`: preflight or the complete one-pass replay succeeded.
- `1`: bundle, capture, interface, configuration, or socket error.
- `130`: interrupted by the operator.
The completion log reports packets sent, payload bytes, packets sent more than
1 ms late, and maximum scheduling lateness. Python scheduling is suitable for
the initial functional test; consistently high lateness during busy feed bursts
would justify a later C++ replay data path.
+967
View File
@@ -0,0 +1,967 @@
#!/usr/bin/env python3
"""Capture selected CME MDP 3.0 multicast flows into a local replay bundle.
The packet capture is delegated to dumpcap so the data path stays in native
code and the output is PCAPNG. This process also joins and observes each
selected multicast flow so it can stop after complete instrument-definition
and recovery cycles have been received.
"""
from __future__ import annotations
import argparse
import collections
import dataclasses
import datetime as dt
import fcntl
import hashlib
import json
import os
from pathlib import Path
import platform
import re
import select
import shutil
import signal
import socket
import struct
import subprocess
import sys
import threading
import time
import uuid
import xml.etree.ElementTree as ET
from typing import Counter, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple
DEFAULT_CME_DATA_DIR = Path("/home/wedbushprod/8000/sparkdata/data/cme_mdp3")
DEFAULT_CHANNELS = (310, 311, 323)
REQUIRED_FEED_TYPES = ("I", "N", "S", "SMBO")
FEED_LABELS = {
"I": "incremental",
"N": "definitions",
"S": "MBP recovery",
"SMBO": "MBO recovery",
}
# These are fallbacks for older/newer CME schema families. The recorder also
# discovers template IDs by message name from the supplied templates.xml.
FALLBACK_DEFINITION_TEMPLATES = frozenset((27, 29, 41, 54, 55, 56))
FALLBACK_SNAPSHOT_TEMPLATES = frozenset((38, 44, 52, 53))
def utc_now() -> dt.datetime:
return dt.datetime.now(dt.timezone.utc)
def iso_utc(value: Optional[dt.datetime] = None) -> str:
return (value or utc_now()).isoformat(timespec="milliseconds").replace("+00:00", "Z")
class EventLog:
def __init__(self) -> None:
self.events: List[dict] = []
def emit(self, level: str, stage: str, message: str, **details: object) -> None:
event = {
"timestamp_utc": iso_utc(),
"level": level,
"stage": stage,
"message": message,
}
if details:
event["details"] = details
self.events.append(event)
print(
f"{event['timestamp_utc']} [{level}] [{stage}] {message}",
file=sys.stderr,
flush=True,
)
def info(self, stage: str, message: str, **details: object) -> None:
self.emit("INFO", stage, message, **details)
def done(self, stage: str, message: str, **details: object) -> None:
self.emit("DONE", stage, message, **details)
def warning(self, stage: str, message: str, **details: object) -> None:
self.emit("WARN", stage, message, **details)
def error(self, stage: str, message: str, **details: object) -> None:
self.emit("ERROR", stage, message, **details)
def local_name(tag: str) -> str:
return tag.rsplit("}", 1)[-1]
@dataclasses.dataclass(frozen=True)
class Flow:
channel: int
connection_id: str
feed: str
feed_type: str
group: str
port: int
source_ip: Optional[str]
@property
def key(self) -> str:
return f"{self.channel}:{self.connection_id}"
@property
def label(self) -> str:
return f"channel {self.channel} feed {self.feed} {FEED_LABELS[self.feed_type]}"
def child_text(element: ET.Element, name: str) -> str:
for child in element:
if local_name(child.tag) == name:
return (child.text or "").strip()
return ""
def parse_cme_flows(
config_path: Path,
channels: Iterable[int],
feeds: Iterable[str],
include_mbo: bool,
) -> List[Flow]:
wanted_channels = set(channels)
wanted_feeds = {feed.upper() for feed in feeds}
wanted_types = set(REQUIRED_FEED_TYPES)
if not include_mbo:
wanted_types.remove("SMBO")
root = ET.parse(config_path).getroot()
result: List[Flow] = []
found_channels = set()
for channel_element in root.iter():
if local_name(channel_element.tag) != "channel":
continue
channel_text = channel_element.attrib.get("id", "")
if not channel_text.isdigit():
continue
channel = int(channel_text)
if channel not in wanted_channels:
continue
found_channels.add(channel)
for connection in channel_element.iter():
if local_name(connection.tag) != "connection":
continue
connection_id = connection.attrib.get("id", "")
feed = child_text(connection, "feed").upper()
group = child_text(connection, "ip")
port_text = child_text(connection, "port")
source_ip = child_text(connection, "host-ip") or None
feed_type = ""
for child in connection:
if local_name(child.tag) == "type":
feed_type = child.attrib.get("feed-type", "").upper()
break
if (
feed in wanted_feeds
and feed_type in wanted_types
and group
and port_text.isdigit()
):
result.append(
Flow(
channel=channel,
connection_id=connection_id,
feed=feed,
feed_type=feed_type,
group=group,
port=int(port_text),
source_ip=source_ip,
)
)
missing_channels = wanted_channels - found_channels
if missing_channels:
raise ValueError(
"CME config.xml does not contain channel(s): "
+ ", ".join(str(value) for value in sorted(missing_channels))
)
expected_pairs = {
(channel, feed, feed_type)
for channel in wanted_channels
for feed in wanted_feeds
for feed_type in wanted_types
}
actual_pairs = {(flow.channel, flow.feed, flow.feed_type) for flow in result}
missing_flows = expected_pairs - actual_pairs
if missing_flows:
formatted = ", ".join(
f"channel={channel}/feed={feed}/type={feed_type}"
for channel, feed, feed_type in sorted(missing_flows)
)
raise ValueError(f"CME config.xml is missing required multicast flows: {formatted}")
return sorted(result, key=lambda flow: (flow.channel, flow.feed, flow.feed_type))
@dataclasses.dataclass(frozen=True)
class TemplateInfo:
template_id: int
name: str
kind: str
total_reports_offset: int
def parse_template_catalog(templates_path: Path) -> Dict[int, TemplateInfo]:
result: Dict[int, TemplateInfo] = {}
try:
root = ET.parse(templates_path).getroot()
except ET.ParseError:
return result
for message in root.iter():
if local_name(message.tag).lower() != "message":
continue
template_text = message.attrib.get("id", "")
name = message.attrib.get("name", "")
if not template_text.isdigit():
continue
lowered = name.lower()
if "instrumentdefinition" in lowered:
kind = "definition"
fallback_offset = 1
elif "snapshotfullrefresh" in lowered:
kind = "snapshot"
fallback_offset = 4
else:
continue
total_offset = fallback_offset
for field in message.iter():
if field.attrib.get("name", "").lower() == "totnumreports":
offset_text = field.attrib.get("offset", "")
if offset_text.isdigit():
total_offset = int(offset_text)
break
template_id = int(template_text)
result[template_id] = TemplateInfo(
template_id=template_id,
name=name,
kind=kind,
total_reports_offset=total_offset,
)
return result
@dataclasses.dataclass(frozen=True)
class SbeMessage:
template_id: int
schema_id: int
version: int
block_length: int
body: bytes
@dataclasses.dataclass(frozen=True)
class MdpPacket:
sequence: int
sending_time_ns: int
messages: Tuple[SbeMessage, ...]
def parse_mdp_packet(payload: bytes) -> MdpPacket:
if len(payload) < 12:
raise ValueError(f"MDP packet is only {len(payload)} bytes; expected at least 12")
sequence, sending_time_ns = struct.unpack_from("<IQ", payload, 0)
messages: List[SbeMessage] = []
offset = 12
while offset < len(payload):
if len(payload) - offset < 2:
raise ValueError("trailing byte after final SBE message")
message_size = struct.unpack_from("<H", payload, offset)[0]
if message_size < 10:
raise ValueError(f"invalid SBE message size {message_size}")
end = offset + message_size
if end > len(payload):
raise ValueError(
f"SBE message size {message_size} exceeds remaining packet bytes"
)
block_length, template_id, schema_id, version = struct.unpack_from(
"<HHHH", payload, offset + 2
)
body_start = offset + 10
if body_start + block_length > end:
raise ValueError(
f"template {template_id} block length {block_length} exceeds message"
)
messages.append(
SbeMessage(
template_id=template_id,
schema_id=schema_id,
version=version,
block_length=block_length,
body=payload[body_start:end],
)
)
offset = end
return MdpPacket(sequence, sending_time_ns, tuple(messages))
@dataclasses.dataclass
class FlowProgress:
flow: Flow
incremental_packets_required: int
template_catalog: Mapping[int, TemplateInfo]
event_log: EventLog
packets: int = 0
invalid_packets: int = 0
sequence_gaps: int = 0
duplicate_packets: int = 0
last_sequence: Optional[int] = None
first_sequence: Optional[int] = None
first_sending_time_ns: Optional[int] = None
last_sending_time_ns: Optional[int] = None
first_observed_utc: Optional[str] = None
last_observed_utc: Optional[str] = None
messages: int = 0
cycle_messages: int = 0
expected_cycle_messages: Optional[int] = None
complete: bool = False
seen_templates: Counter[int] = dataclasses.field(default_factory=collections.Counter)
def _template_info(self, message: SbeMessage) -> Optional[TemplateInfo]:
configured = self.template_catalog.get(message.template_id)
if configured:
return configured
if message.template_id in FALLBACK_DEFINITION_TEMPLATES:
return TemplateInfo(message.template_id, "fallback", "definition", 1)
if message.template_id in FALLBACK_SNAPSHOT_TEMPLATES:
return TemplateInfo(message.template_id, "fallback", "snapshot", 4)
return None
def _reset_cycle_after_gap(self) -> None:
if self.flow.feed_type in ("N", "S", "SMBO"):
self.cycle_messages = 0
def observe(self, payload: bytes) -> None:
observed = iso_utc()
try:
packet = parse_mdp_packet(payload)
except ValueError as error:
self.invalid_packets += 1
if self.invalid_packets <= 3:
self.event_log.warning(self.flow.label, f"invalid CME packet: {error}")
return
if self.first_observed_utc is None:
self.first_observed_utc = observed
self.first_sequence = packet.sequence
self.first_sending_time_ns = packet.sending_time_ns
self.event_log.info(
self.flow.label,
f"traffic observed on {self.flow.group}:{self.flow.port}",
sequence=packet.sequence,
)
self.last_observed_utc = observed
self.last_sending_time_ns = packet.sending_time_ns
if self.last_sequence is not None:
if packet.sequence == self.last_sequence:
self.duplicate_packets += 1
return
expected = (self.last_sequence + 1) & 0xFFFFFFFF
if packet.sequence != expected:
self.sequence_gaps += 1
self._reset_cycle_after_gap()
self.event_log.warning(
self.flow.label,
"sequence gap; restarting completion count for this cycle",
expected_sequence=expected,
actual_sequence=packet.sequence,
)
self.last_sequence = packet.sequence
self.packets += 1
if self.flow.feed_type == "I":
if self.packets >= self.incremental_packets_required:
self.complete = True
self.event_log.done(
self.flow.label,
f"incremental stage complete after {self.packets} valid packet(s)",
)
return
required_kind = "definition" if self.flow.feed_type == "N" else "snapshot"
for message in packet.messages:
self.seen_templates[message.template_id] += 1
info = self._template_info(message)
if not info or info.kind != required_kind:
continue
total_offset = info.total_reports_offset
if total_offset + 4 > len(message.body):
continue
total = struct.unpack_from("<I", message.body, total_offset)[0]
if total in (0, 0xFFFFFFFF):
continue
if self.expected_cycle_messages is None:
self.expected_cycle_messages = total
self.event_log.info(
self.flow.label,
f"cycle started; TotNumReports={total}",
template_id=message.template_id,
)
elif total != self.expected_cycle_messages:
self.event_log.warning(
self.flow.label,
"TotNumReports changed; restarting completion count",
previous=self.expected_cycle_messages,
current=total,
)
self.expected_cycle_messages = total
self.cycle_messages = 0
self.cycle_messages += 1
self.messages += 1
if self.cycle_messages >= self.expected_cycle_messages:
self.complete = True
self.event_log.done(
self.flow.label,
"complete gap-free replay cycle received",
messages=self.cycle_messages,
total_reports=self.expected_cycle_messages,
)
break
def summary(self) -> dict:
return {
"channel": self.flow.channel,
"connection_id": self.flow.connection_id,
"feed": self.flow.feed,
"feed_type": self.flow.feed_type,
"stage": FEED_LABELS[self.flow.feed_type],
"multicast_group": self.flow.group,
"port": self.flow.port,
"source_ip": self.flow.source_ip,
"complete": self.complete,
"valid_packets_observed": self.packets,
"invalid_packets_observed": self.invalid_packets,
"sequence_gaps_observed": self.sequence_gaps,
"duplicate_packets_observed": self.duplicate_packets,
"first_sequence": self.first_sequence,
"last_sequence": self.last_sequence,
"first_sending_time_ns": self.first_sending_time_ns,
"last_sending_time_ns": self.last_sending_time_ns,
"first_observed_utc": self.first_observed_utc,
"last_observed_utc": self.last_observed_utc,
"cycle_messages_observed": self.cycle_messages,
"expected_cycle_messages": self.expected_cycle_messages,
"template_counts": {
str(key): value for key, value in sorted(self.seen_templates.items())
},
}
def interface_ipv4(interface: str) -> str:
request = struct.pack("256s", interface.encode("ascii")[:15])
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
response = fcntl.ioctl(sock.fileno(), 0x8915, request) # SIOCGIFADDR
return socket.inet_ntoa(response[20:24])
def open_multicast_socket(flow: Flow, interface_ip: str, receive_buffer: int) -> socket.socket:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
try:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, receive_buffer)
# Linux IP_MULTICAST_ALL. It prevents a socket from receiving traffic
# for groups joined by other sockets on the same port.
try:
sock.setsockopt(socket.IPPROTO_IP, 49, 0)
bind_address = ""
except OSError:
# Binding to the multicast group itself provides equivalent flow
# isolation on older Linux kernels without IP_MULTICAST_ALL.
bind_address = flow.group
sock.bind((bind_address, flow.port))
membership = socket.inet_aton(flow.group) + socket.inet_aton(interface_ip)
sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, membership)
sock.setblocking(False)
return sock
except Exception:
sock.close()
raise
def build_capture_filter(flows: Sequence[Flow]) -> str:
return " or ".join(
f"(dst host {flow.group} and udp dst port {flow.port})" for flow in flows
)
class DumpcapProcess:
def __init__(
self,
executable: str,
interface: str,
capture_filter: str,
output_path: Path,
buffer_mib: int,
event_log: EventLog,
) -> None:
self.command = [
executable,
"-q",
"-i",
interface,
"-f",
capture_filter,
"-s",
"0",
"-B",
str(buffer_mib),
"-w",
str(output_path),
]
self.event_log = event_log
self.stderr_lines: List[str] = []
self.packet_count: Optional[int] = None
self.dropped_packet_count: Optional[int] = None
self.process: Optional[subprocess.Popen[str]] = None
self.reader: Optional[threading.Thread] = None
def start(self) -> None:
self.process = subprocess.Popen(
self.command,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
)
assert self.process.stderr is not None
def read_stderr() -> None:
assert self.process is not None and self.process.stderr is not None
for raw_line in self.process.stderr:
line = raw_line.rstrip()
self.stderr_lines.append(line)
match = re.search(r"Packets captured:\s*(\d+)", line)
if match:
self.packet_count = int(match.group(1))
dropped_match = re.search(r"Packets dropped:\s*(\d+)", line)
if dropped_match:
self.dropped_packet_count = int(dropped_match.group(1))
if line and "capturing on" not in line.lower():
self.event_log.info("dumpcap", line)
self.reader = threading.Thread(target=read_stderr, name="dumpcap-stderr", daemon=True)
self.reader.start()
time.sleep(0.3)
if self.process.poll() is not None:
self.wait()
raise RuntimeError(
"dumpcap exited during startup: " + " | ".join(self.stderr_lines[-5:])
)
def stop(self) -> int:
if self.process is None:
return 0
if self.process.poll() is None:
self.process.send_signal(signal.SIGINT)
try:
return self.process.wait(timeout=15)
except subprocess.TimeoutExpired:
self.event_log.warning("dumpcap", "did not stop after SIGINT; terminating")
self.process.terminate()
return self.process.wait(timeout=5)
finally:
if self.reader:
self.reader.join(timeout=2)
def wait(self) -> int:
if self.process is None:
return 0
result = self.process.wait()
if self.reader:
self.reader.join(timeout=2)
return result
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
for block in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def write_checksums(directory: Path, filenames: Sequence[str]) -> None:
output = directory / "checksums.sha256"
with output.open("w", encoding="utf-8", newline="\n") as stream:
for filename in filenames:
stream.write(f"{sha256_file(directory / filename)} {filename}\n")
def parse_csv_ints(value: str) -> Tuple[int, ...]:
try:
result = tuple(sorted({int(item.strip()) for item in value.split(",") if item.strip()}))
except ValueError as error:
raise argparse.ArgumentTypeError("channels must be comma-separated integers") from error
if not result:
raise argparse.ArgumentTypeError("at least one channel is required")
return result
def parse_csv_feeds(value: str) -> Tuple[str, ...]:
result = tuple(sorted({item.strip().upper() for item in value.split(",") if item.strip()}))
if not result or any(item not in ("A", "B") for item in result):
raise argparse.ArgumentTypeError("feeds must be A, B, or A,B")
return result
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Record CME MDP 3.0 multicast into a self-contained local PCAPNG bundle."
)
parser.add_argument("--interface", required=True, help="Linux capture interface name")
parser.add_argument(
"--cme-data-dir",
type=Path,
default=DEFAULT_CME_DATA_DIR,
help=f"directory containing config.xml and templates.xml (default: {DEFAULT_CME_DATA_DIR})",
)
parser.add_argument(
"--channels",
type=parse_csv_ints,
default=DEFAULT_CHANNELS,
help="comma-separated CME channel IDs (default: 310,311,323)",
)
parser.add_argument(
"--feeds",
type=parse_csv_feeds,
default=("A",),
help="A, B, or A,B (default: A)",
)
parser.add_argument(
"--output",
type=Path,
required=True,
help="new local bundle directory; it must not already exist",
)
parser.add_argument(
"--min-seconds",
type=float,
default=1.0,
help="minimum capture duration even if all stages finish (default: 1)",
)
parser.add_argument(
"--max-seconds",
type=float,
default=300.0,
help="stop and create an incomplete bundle after this duration (default: 300)",
)
parser.add_argument(
"--incremental-packets",
type=int,
default=1,
help="valid incremental packets needed per channel (default: 1)",
)
parser.add_argument(
"--skip-mbo",
action="store_true",
help="do not require or capture the SMBO recovery flow",
)
parser.add_argument(
"--capture-buffer-mib",
type=int,
default=1024,
help="dumpcap capture buffer in MiB (default: 1024)",
)
parser.add_argument(
"--monitor-buffer-mib",
type=int,
default=64,
help="requested receive buffer per completion monitor socket (default: 64)",
)
parser.add_argument(
"--dumpcap",
default="dumpcap",
help="dumpcap executable name or path (default: dumpcap)",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="validate files/interface and print the capture plan without capturing",
)
return parser
def validate_args(args: argparse.Namespace) -> None:
if args.min_seconds < 0:
raise ValueError("--min-seconds cannot be negative")
if args.max_seconds <= 0 or args.max_seconds < args.min_seconds:
raise ValueError("--max-seconds must be positive and at least --min-seconds")
if args.incremental_packets <= 0:
raise ValueError("--incremental-packets must be positive")
if args.capture_buffer_mib <= 0 or args.monitor_buffer_mib <= 0:
raise ValueError("capture and monitor buffers must be positive")
if args.output.exists():
raise ValueError(f"output directory already exists: {args.output}")
def print_plan(
args: argparse.Namespace,
interface_ip: str,
flows: Sequence[Flow],
capture_filter: str,
) -> None:
print(f"interface: {args.interface} ({interface_ip})")
print(f"channels: {','.join(str(value) for value in args.channels)}")
print(f"feeds: {','.join(args.feeds)}")
print(f"output: {args.output}")
print("flows:")
for flow in flows:
print(
f" {flow.connection_id}: channel={flow.channel} feed={flow.feed} "
f"type={flow.feed_type} group={flow.group}:{flow.port}"
)
print(f"capture filter: {capture_filter}")
def run_capture(args: argparse.Namespace) -> int:
event_log = EventLog()
validate_args(args)
config_source = args.cme_data_dir / "config.xml"
templates_source = args.cme_data_dir / "templates.xml"
for required in (config_source, templates_source):
if not required.is_file():
raise ValueError(f"required CME file does not exist: {required}")
socket.if_nametoindex(args.interface)
interface_ip = interface_ipv4(args.interface)
flows = parse_cme_flows(
config_source,
args.channels,
args.feeds,
include_mbo=not args.skip_mbo,
)
capture_filter = build_capture_filter(flows)
template_catalog = parse_template_catalog(templates_source)
event_log.done(
"configuration",
f"validated {len(flows)} multicast flows from CME configuration",
channels=list(args.channels),
feeds=list(args.feeds),
)
if args.dry_run:
print_plan(args, interface_ip, flows, capture_filter)
event_log.done("dry-run", "capture plan is valid; no files were created")
return 0
dumpcap = shutil.which(args.dumpcap) if os.sep not in args.dumpcap else args.dumpcap
if not dumpcap or not Path(dumpcap).is_file():
raise ValueError(
"dumpcap was not found; install the Wireshark command-line capture tools "
"or pass --dumpcap /path/to/dumpcap"
)
partial = args.output.parent / f".{args.output.name}.partial-{uuid.uuid4().hex[:8]}"
partial.mkdir(parents=True, exist_ok=False)
recording_path = partial / "recording.pcapng"
shutil.copy2(config_source, partial / "config.xml")
shutil.copy2(templates_source, partial / "templates.xml")
event_log.done("bundle", "copied capture-time CME config.xml and templates.xml")
monitor_sockets: Dict[socket.socket, FlowProgress] = {}
all_sockets: List[socket.socket] = []
progress_objects: List[FlowProgress] = []
dumpcap_process: Optional[DumpcapProcess] = None
started_at = utc_now()
started_monotonic = time.monotonic()
reason = "error"
interrupted = False
previous_handlers: Dict[int, object] = {}
def request_stop(signum: int, _frame: object) -> None:
nonlocal interrupted
interrupted = True
event_log.warning("capture", f"received signal {signum}; finalizing local bundle")
for signum in (signal.SIGINT, signal.SIGTERM):
previous_handlers[signum] = signal.getsignal(signum)
signal.signal(signum, request_stop)
try:
receive_buffer = args.monitor_buffer_mib * 1024 * 1024
for flow in flows:
sock = open_multicast_socket(flow, interface_ip, receive_buffer)
progress = FlowProgress(
flow=flow,
incremental_packets_required=args.incremental_packets,
template_catalog=template_catalog,
event_log=event_log,
)
monitor_sockets[sock] = progress
all_sockets.append(sock)
progress_objects.append(progress)
actual_buffer = sock.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF)
event_log.info(
flow.label,
f"joined {flow.group}:{flow.port}",
monitor_receive_buffer_bytes=actual_buffer,
)
event_log.done("multicast", f"joined all {len(flows)} required multicast groups")
dumpcap_process = DumpcapProcess(
executable=str(dumpcap),
interface=args.interface,
capture_filter=capture_filter,
output_path=recording_path,
buffer_mib=args.capture_buffer_mib,
event_log=event_log,
)
dumpcap_process.start()
# Discard packets queued in the completion-monitor sockets before
# dumpcap became active. The recording may contain a short prefix that
# the monitor ignores, but it can never omit packets used to declare a
# definition/recovery cycle complete.
discarded_before_monitor = 0
for sock in all_sockets:
while True:
try:
sock.recv(65535)
discarded_before_monitor += 1
except BlockingIOError:
break
event_log.done("capture", f"PCAPNG capture started on {args.interface}")
if discarded_before_monitor:
event_log.info(
"capture",
"discarded pre-monitor socket backlog; these packets may still be present "
"as a harmless PCAPNG prefix",
packets=discarded_before_monitor,
)
while True:
elapsed = time.monotonic() - started_monotonic
if interrupted:
reason = "interrupted"
break
if dumpcap_process.process and dumpcap_process.process.poll() is not None:
reason = "dumpcap_exited"
event_log.error("capture", "dumpcap exited unexpectedly")
break
if elapsed >= args.max_seconds:
reason = "timeout"
incomplete = [progress.flow.label for progress in monitor_sockets.values()]
event_log.warning(
"capture",
"maximum duration reached before all stages completed",
incomplete_stages=incomplete,
)
break
if not monitor_sockets and elapsed >= args.min_seconds:
reason = "all_stages_complete"
event_log.done("capture", "all required CME stages are complete")
break
readable, _, _ = select.select(list(monitor_sockets), [], [], 0.5)
for sock in readable:
progress = monitor_sockets.get(sock)
if progress is None:
continue
for _ in range(4096):
try:
payload = sock.recv(65535)
except BlockingIOError:
break
progress.observe(payload)
if progress.complete:
monitor_sockets.pop(sock, None)
break
finally:
for signum, handler in previous_handlers.items():
signal.signal(signum, handler)
if dumpcap_process:
result = dumpcap_process.stop()
if result not in (0, 130, -signal.SIGINT) and reason == "all_stages_complete":
reason = "dumpcap_error"
event_log.error("dumpcap", f"dumpcap exited with status {result}")
for sock in all_sockets:
sock.close()
ended_at = utc_now()
capture_drops = dumpcap_process.dropped_packet_count if dumpcap_process else None
if capture_drops:
event_log.error(
"capture",
"dumpcap reported dropped packets; bundle cannot be considered complete",
dropped_packets=capture_drops,
)
all_complete = (
bool(progress_objects)
and all(item.complete for item in progress_objects)
and not capture_drops
)
capture_status = "complete" if all_complete and reason == "all_stages_complete" else "incomplete"
manifest = {
"bundle_format": "cme-mdp3-pcapng-v1",
"status": capture_status,
"completion_reason": reason,
"started_utc": iso_utc(started_at),
"ended_utc": iso_utc(ended_at),
"wall_duration_seconds": round((ended_at - started_at).total_seconds(), 6),
"host": {
"hostname": socket.gethostname(),
"platform": platform.platform(),
"python": platform.python_version(),
},
"capture": {
"interface": args.interface,
"interface_ipv4": interface_ip,
"channels": list(args.channels),
"feeds": list(args.feeds),
"include_mbo": not args.skip_mbo,
"filter": capture_filter,
"dumpcap_command": dumpcap_process.command if dumpcap_process else None,
"dumpcap_packets_captured": dumpcap_process.packet_count if dumpcap_process else None,
"dumpcap_packets_dropped": capture_drops,
"recording_size_bytes": recording_path.stat().st_size if recording_path.exists() else 0,
},
"stages": [item.summary() for item in progress_objects],
"events": event_log.events,
"files": [
"recording.pcapng",
"config.xml",
"templates.xml",
"manifest.json",
"checksums.sha256",
],
}
manifest_path = partial / "manifest.json"
manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8")
write_checksums(
partial,
("recording.pcapng", "config.xml", "templates.xml", "manifest.json"),
)
event_log.done("checksums", "created SHA-256 checksums for all bundle content")
os.replace(partial, args.output)
event_log.done("bundle", f"local replay bundle ready: {args.output}")
if capture_status == "complete":
return 0
if reason == "interrupted":
return 130
return 2
def main(argv: Optional[Sequence[str]] = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
try:
return run_capture(args)
except (OSError, ValueError, RuntimeError) as error:
print(f"ERROR: {error}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())
+670
View File
@@ -0,0 +1,670 @@
#!/usr/bin/env python3
"""Replay one CME MDP 3.0 PCAPNG bundle over UDP multicast.
Capture timestamps pace the replay. CME packet and message bytes are sent
unchanged. This first version deliberately performs one pass only: it does not
loop, rewrite timestamps, or rewrite sequence numbers.
"""
from __future__ import annotations
import argparse
import dataclasses
import datetime as dt
import fcntl
import hashlib
import itertools
import json
from pathlib import Path
import signal
import socket
import struct
import sys
import time
import xml.etree.ElementTree as ET
from typing import Callable, Dict, Iterable, Iterator, List, Optional, Sequence, Set, Tuple
PCAPNG_SECTION_HEADER = 0x0A0D0D0A
PCAPNG_INTERFACE_DESCRIPTION = 0x00000001
PCAPNG_SIMPLE_PACKET = 0x00000003
PCAPNG_ENHANCED_PACKET = 0x00000006
LINKTYPE_ETHERNET = 1
LINKTYPE_RAW = 101
LINKTYPE_LINUX_SLL = 113
LINKTYPE_LINUX_SLL2 = 276
SUPPORTED_LINK_TYPES = {
LINKTYPE_ETHERNET,
LINKTYPE_RAW,
LINKTYPE_LINUX_SLL,
LINKTYPE_LINUX_SLL2,
}
def iso_utc() -> str:
return dt.datetime.now(dt.timezone.utc).isoformat(timespec="milliseconds").replace(
"+00:00", "Z"
)
class EventLog:
def emit(self, level: str, stage: str, message: str) -> None:
print(f"{iso_utc()} [{level}] [{stage}] {message}", file=sys.stderr, flush=True)
def info(self, stage: str, message: str) -> None:
self.emit("INFO", stage, message)
def done(self, stage: str, message: str) -> None:
self.emit("DONE", stage, message)
def warning(self, stage: str, message: str) -> None:
self.emit("WARN", stage, message)
def local_name(tag: str) -> str:
return tag.rsplit("}", 1)[-1]
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
for block in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def read_checksums(path: Path) -> Dict[str, str]:
result: Dict[str, str] = {}
for line_number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
line = raw_line.strip()
if not line:
continue
parts = line.split()
if len(parts) != 2 or len(parts[0]) != 64:
raise ValueError(f"invalid checksums.sha256 line {line_number}")
digest, filename = parts
if any(character not in "0123456789abcdefABCDEF" for character in digest):
raise ValueError(f"invalid SHA-256 digest on line {line_number}")
if Path(filename).name != filename or filename in result:
raise ValueError(f"unsafe or duplicate checksum filename on line {line_number}")
result[filename] = digest.lower()
if not result:
raise ValueError("checksums.sha256 contains no entries")
return result
def config_destinations(path: Path) -> Set[Tuple[str, int]]:
root = ET.parse(path).getroot()
destinations: Set[Tuple[str, int]] = set()
for connection in root.iter():
if local_name(connection.tag) != "connection":
continue
group = ""
port_text = ""
for child in connection:
name = local_name(child.tag)
if name == "ip":
group = (child.text or "").strip()
elif name == "port":
port_text = (child.text or "").strip()
if group and port_text.isdigit():
try:
if socket.inet_aton(group) and 224 <= int(group.split(".", 1)[0]) <= 239:
destinations.add((group, int(port_text)))
except OSError:
continue
return destinations
@dataclasses.dataclass(frozen=True)
class Bundle:
directory: Path
recording: Path
destinations: Set[Tuple[str, int]]
manifest: dict
def validate_bundle(directory: Path, allow_incomplete: bool) -> Bundle:
directory = directory.resolve()
required = (
"recording.pcapng",
"config.xml",
"templates.xml",
"manifest.json",
"checksums.sha256",
)
for filename in required:
if not (directory / filename).is_file():
raise ValueError(f"bundle is missing {filename}: {directory}")
checksums = read_checksums(directory / "checksums.sha256")
for filename in required[:-1]:
expected = checksums.get(filename)
if expected is None:
raise ValueError(f"checksums.sha256 does not contain {filename}")
actual = sha256_file(directory / filename)
if actual != expected:
raise ValueError(f"SHA-256 mismatch for {filename}")
manifest = json.loads((directory / "manifest.json").read_text(encoding="utf-8"))
if manifest.get("bundle_format") != "cme-mdp3-pcapng-v1":
raise ValueError(f"unsupported bundle format: {manifest.get('bundle_format')!r}")
status = manifest.get("status")
if status != "complete" and not allow_incomplete:
raise ValueError(
f"bundle status is {status!r}; use --allow-incomplete only for an "
"intentional smoke test"
)
configured = config_destinations(directory / "config.xml")
if not configured:
raise ValueError("config.xml contains no IPv4 multicast destinations")
destinations: Set[Tuple[str, int]] = set()
for stage in manifest.get("stages", []):
group = stage.get("multicast_group")
port = stage.get("port")
if isinstance(group, str) and isinstance(port, int):
destinations.add((group, port))
if not destinations:
raise ValueError("manifest contains no recorded multicast stages")
missing = destinations - configured
if missing:
formatted = ", ".join(f"{group}:{port}" for group, port in sorted(missing))
raise ValueError(f"manifest destinations are absent from config.xml: {formatted}")
return Bundle(directory, directory / "recording.pcapng", destinations, manifest)
@dataclasses.dataclass(frozen=True)
class InterfaceDescription:
link_type: int
timestamp_resolution: int = 6
timestamp_resolution_binary: bool = False
timestamp_offset_seconds: int = 0
def timestamp_ns(self, ticks: int) -> int:
if self.timestamp_resolution_binary:
value = (ticks * 1_000_000_000) // (1 << self.timestamp_resolution)
elif self.timestamp_resolution <= 9:
value = ticks * (10 ** (9 - self.timestamp_resolution))
else:
value = ticks // (10 ** (self.timestamp_resolution - 9))
return value + self.timestamp_offset_seconds * 1_000_000_000
@dataclasses.dataclass(frozen=True)
class CapturedUdpPacket:
capture_time_ns: int
source_ip: str
source_port: int
destination_ip: str
destination_port: int
payload: bytes
def parse_options(data: bytes, byte_order: str) -> Iterator[Tuple[int, bytes]]:
offset = 0
while offset + 4 <= len(data):
option_code, option_length = struct.unpack_from(byte_order + "HH", data, offset)
offset += 4
if option_code == 0:
return
end = offset + option_length
if end > len(data):
raise ValueError("PCAPNG option exceeds its block")
yield option_code, data[offset:end]
offset += (option_length + 3) & ~3
def parse_interface_description(data: bytes, byte_order: str) -> InterfaceDescription:
if len(data) < 8:
raise ValueError("truncated PCAPNG interface description block")
link_type = struct.unpack_from(byte_order + "H", data, 0)[0]
if link_type not in SUPPORTED_LINK_TYPES:
raise ValueError(f"unsupported PCAPNG link type {link_type}")
resolution = 6
binary = False
offset_seconds = 0
for code, value in parse_options(data[8:], byte_order):
if code == 9 and len(value) == 1: # if_tsresol
binary = bool(value[0] & 0x80)
resolution = value[0] & 0x7F
elif code == 14 and len(value) == 8: # if_tsoffset
offset_seconds = struct.unpack(byte_order + "q", value)[0]
return InterfaceDescription(link_type, resolution, binary, offset_seconds)
def ipv4_offset(frame: bytes, link_type: int) -> Optional[int]:
if link_type == LINKTYPE_RAW:
return 0 if frame and frame[0] >> 4 == 4 else None
if link_type == LINKTYPE_LINUX_SLL:
if len(frame) < 16 or struct.unpack_from("!H", frame, 14)[0] != 0x0800:
return None
return 16
if link_type == LINKTYPE_LINUX_SLL2:
if len(frame) < 20 or struct.unpack_from("!H", frame, 0)[0] != 0x0800:
return None
return 20
if len(frame) < 14:
return None
protocol = struct.unpack_from("!H", frame, 12)[0]
offset = 14
while protocol in (0x8100, 0x88A8, 0x9100):
if len(frame) < offset + 4:
return None
protocol = struct.unpack_from("!H", frame, offset + 2)[0]
offset += 4
return offset if protocol == 0x0800 else None
def extract_udp(frame: bytes, link_type: int, capture_time_ns: int) -> Optional[CapturedUdpPacket]:
ip_offset = ipv4_offset(frame, link_type)
if ip_offset is None or len(frame) < ip_offset + 20:
return None
version_ihl = frame[ip_offset]
if version_ihl >> 4 != 4:
return None
header_length = (version_ihl & 0x0F) * 4
if header_length < 20 or len(frame) < ip_offset + header_length + 8:
return None
if frame[ip_offset + 9] != socket.IPPROTO_UDP:
return None
fragment = struct.unpack_from("!H", frame, ip_offset + 6)[0]
if fragment & 0x3FFF:
raise ValueError("fragmented IPv4 UDP packets are not supported")
total_length = struct.unpack_from("!H", frame, ip_offset + 2)[0]
if total_length < header_length + 8 or ip_offset + total_length > len(frame):
raise ValueError("truncated IPv4 packet in PCAPNG")
udp_offset = ip_offset + header_length
source_port, destination_port, udp_length = struct.unpack_from("!HHH", frame, udp_offset)
if udp_length < 8 or udp_offset + udp_length > ip_offset + total_length:
raise ValueError("invalid UDP length in PCAPNG")
source_ip = socket.inet_ntoa(frame[ip_offset + 12 : ip_offset + 16])
destination_ip = socket.inet_ntoa(frame[ip_offset + 16 : ip_offset + 20])
payload = frame[udp_offset + 8 : udp_offset + udp_length]
return CapturedUdpPacket(
capture_time_ns,
source_ip,
source_port,
destination_ip,
destination_port,
payload,
)
def iter_pcapng_udp(path: Path) -> Iterator[CapturedUdpPacket]:
byte_order: Optional[str] = None
interfaces: List[InterfaceDescription] = []
with path.open("rb") as stream:
while True:
header = stream.read(8)
if not header:
return
if len(header) != 8:
raise ValueError("truncated PCAPNG block header")
raw_type, raw_length = header[:4], header[4:]
if raw_type == struct.pack("<I", PCAPNG_SECTION_HEADER):
bom = stream.read(4)
if len(bom) != 4:
raise ValueError("truncated PCAPNG section header")
if bom == b"\x4d\x3c\x2b\x1a":
byte_order = "<"
elif bom == b"\x1a\x2b\x3c\x4d":
byte_order = ">"
else:
raise ValueError("invalid PCAPNG byte-order magic")
block_length = struct.unpack(byte_order + "I", raw_length)[0]
if block_length < 28 or block_length % 4:
raise ValueError("invalid PCAPNG section-header length")
remainder = stream.read(block_length - 12)
if len(remainder) != block_length - 12:
raise ValueError("truncated PCAPNG section header")
trailing_length = struct.unpack_from(
byte_order + "I", remainder, len(remainder) - 4
)[0]
if trailing_length != block_length:
raise ValueError("PCAPNG section length trailer mismatch")
interfaces = []
continue
if byte_order is None:
raise ValueError("PCAPNG does not begin with a section header")
block_type = struct.unpack(byte_order + "I", raw_type)[0]
block_length = struct.unpack(byte_order + "I", raw_length)[0]
if block_length < 12 or block_length % 4:
raise ValueError(f"invalid PCAPNG block length {block_length}")
remainder = stream.read(block_length - 8)
if len(remainder) != block_length - 8:
raise ValueError("truncated PCAPNG block")
trailing_length = struct.unpack_from(
byte_order + "I", remainder, len(remainder) - 4
)[0]
if trailing_length != block_length:
raise ValueError("PCAPNG block length trailer mismatch")
data = remainder[:-4]
if block_type == PCAPNG_INTERFACE_DESCRIPTION:
interfaces.append(parse_interface_description(data, byte_order))
elif block_type == PCAPNG_SIMPLE_PACKET:
raise ValueError(
"PCAPNG simple packet blocks have no timestamp and cannot be paced"
)
elif block_type == PCAPNG_ENHANCED_PACKET:
if len(data) < 20:
raise ValueError("truncated PCAPNG enhanced packet block")
interface_id, timestamp_high, timestamp_low, captured_length = struct.unpack_from(
byte_order + "IIII", data, 0
)
if interface_id >= len(interfaces):
raise ValueError(f"PCAPNG packet references unknown interface {interface_id}")
if 20 + captured_length > len(data):
raise ValueError("captured packet exceeds its PCAPNG block")
interface = interfaces[interface_id]
ticks = (timestamp_high << 32) | timestamp_low
timestamp_ns = interface.timestamp_ns(ticks)
frame = data[20 : 20 + captured_length]
packet = extract_udp(frame, interface.link_type, timestamp_ns)
if packet is not None:
yield packet
@dataclasses.dataclass(frozen=True)
class CaptureSummary:
packets: int
bytes: int
first_time_ns: int
last_time_ns: int
destinations_seen: Set[Tuple[str, int]]
timestamp_regressions: int
maximum_timestamp_regression_ns: int
@property
def duration_ns(self) -> int:
return self.last_time_ns - self.first_time_ns
def target_packets(
path: Path, destinations: Set[Tuple[str, int]]
) -> Iterator[CapturedUdpPacket]:
for packet in iter_pcapng_udp(path):
if (packet.destination_ip, packet.destination_port) in destinations:
yield packet
def scan_capture(path: Path, destinations: Set[Tuple[str, int]]) -> CaptureSummary:
packets = 0
payload_bytes = 0
first: Optional[int] = None
last: Optional[int] = None
previous: Optional[int] = None
seen: Set[Tuple[str, int]] = set()
timestamp_regressions = 0
maximum_timestamp_regression_ns = 0
for packet in target_packets(path, destinations):
if previous is not None and packet.capture_time_ns < previous:
timestamp_regressions += 1
maximum_timestamp_regression_ns = max(
maximum_timestamp_regression_ns,
previous - packet.capture_time_ns,
)
previous = packet.capture_time_ns
first = packet.capture_time_ns if first is None else first
last = packet.capture_time_ns if last is None else max(last, packet.capture_time_ns)
packets += 1
payload_bytes += len(packet.payload)
seen.add((packet.destination_ip, packet.destination_port))
if not packets or first is None or last is None:
raise ValueError("recording contains no UDP packets matching the manifest destinations")
missing = destinations - seen
if missing:
formatted = ", ".join(f"{group}:{port}" for group, port in sorted(missing))
raise ValueError(f"recording contains no packets for manifest destination(s): {formatted}")
return CaptureSummary(
packets,
payload_bytes,
first,
last,
seen,
timestamp_regressions,
maximum_timestamp_regression_ns,
)
def interface_ipv4(interface: str) -> str:
request = struct.pack("256s", interface.encode("ascii")[:15])
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
response = fcntl.ioctl(sock.fileno(), 0x8915, request) # SIOCGIFADDR
return socket.inet_ntoa(response[20:24])
class MulticastSender:
def __init__(self, interface_ip: str, send_buffer_bytes: int) -> None:
self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, send_buffer_bytes)
self.socket.setsockopt(
socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton(interface_ip)
)
self.socket.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 1)
self.socket.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 1)
def send(self, packet: CapturedUdpPacket) -> None:
sent = self.socket.sendto(
packet.payload, (packet.destination_ip, packet.destination_port)
)
if sent != len(packet.payload):
raise OSError(f"short UDP send: sent {sent} of {len(packet.payload)} bytes")
def close(self) -> None:
self.socket.close()
@dataclasses.dataclass(frozen=True)
class ReplayResult:
packets: int
bytes: int
late_packets: int
maximum_lateness_ns: int
interrupted: bool
def replay_capture(
packets: Iterable[CapturedUdpPacket],
send: Callable[[CapturedUdpPacket], None],
start_delay_seconds: float,
progress_seconds: float,
event_log: EventLog,
should_stop: Callable[[], bool],
monotonic_ns: Callable[[], int] = time.monotonic_ns,
sleep: Callable[[float], None] = time.sleep,
) -> ReplayResult:
iterator = iter(packets)
try:
first = next(iterator)
except StopIteration as error:
raise ValueError("no target packets to replay") from error
replay_start_ns = monotonic_ns() + int(start_delay_seconds * 1_000_000_000)
capture_start_ns = first.capture_time_ns
packets_sent = 0
bytes_sent = 0
late_packets = 0
maximum_lateness_ns = 0
progress_interval_ns = int(progress_seconds * 1_000_000_000)
next_progress_ns = replay_start_ns + progress_interval_ns
for packet in itertools.chain((first,), iterator):
if should_stop():
return ReplayResult(
packets_sent, bytes_sent, late_packets, maximum_lateness_ns, True
)
target_ns = replay_start_ns + packet.capture_time_ns - capture_start_ns
while True:
remaining_ns = target_ns - monotonic_ns()
if remaining_ns <= 0:
break
# Recalculate against an absolute target after every sleep so
# scheduler delays do not accumulate over the recording.
sleep(min(remaining_ns / 1_000_000_000, 0.050))
if should_stop():
return ReplayResult(
packets_sent, bytes_sent, late_packets, maximum_lateness_ns, True
)
before_send_ns = monotonic_ns()
lateness_ns = max(0, before_send_ns - target_ns)
if lateness_ns > 1_000_000:
late_packets += 1
maximum_lateness_ns = max(maximum_lateness_ns, lateness_ns)
send(packet)
packets_sent += 1
bytes_sent += len(packet.payload)
if progress_interval_ns and before_send_ns >= next_progress_ns:
event_log.info(
"replay",
f"sent {packets_sent} packets ({bytes_sent} payload bytes); "
f"current capture offset={(packet.capture_time_ns - capture_start_ns) / 1e9:.3f}s",
)
next_progress_ns = before_send_ns + progress_interval_ns
return ReplayResult(packets_sent, bytes_sent, late_packets, maximum_lateness_ns, False)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description=(
"Replay one CME MDP 3.0 bundle at its original packet timing without "
"rewriting CME timestamps or sequence numbers."
)
)
parser.add_argument("--bundle", required=True, type=Path, help="recorder bundle directory")
parser.add_argument("--interface", required=True, help="Linux multicast output interface")
parser.add_argument(
"--allow-incomplete",
action="store_true",
help="allow a recorder bundle whose manifest status is not complete",
)
parser.add_argument(
"--start-delay-seconds",
type=float,
default=1.0,
help="delay before sending the first packet, allowing OnixS to join (default: 1)",
)
parser.add_argument(
"--send-buffer-mib",
type=int,
default=16,
help="requested UDP send buffer size in MiB (default: 16)",
)
parser.add_argument(
"--progress-seconds",
type=float,
default=10.0,
help="replay progress log interval; 0 disables it (default: 10)",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="validate and scan the complete bundle without sending packets",
)
return parser
def run(args: argparse.Namespace) -> int:
if args.start_delay_seconds < 0:
raise ValueError("--start-delay-seconds cannot be negative")
if args.send_buffer_mib <= 0:
raise ValueError("--send-buffer-mib must be positive")
if args.progress_seconds < 0:
raise ValueError("--progress-seconds cannot be negative")
event_log = EventLog()
bundle = validate_bundle(args.bundle, args.allow_incomplete)
event_log.done(
"bundle",
f"checksums and manifest validated; {len(bundle.destinations)} multicast destinations",
)
summary = scan_capture(bundle.recording, bundle.destinations)
event_log.done(
"capture",
f"validated {summary.packets} packets and {summary.bytes} payload bytes; "
f"recorded duration={summary.duration_ns / 1e9:.6f}s",
)
if summary.timestamp_regressions:
event_log.warning(
"capture",
f"found {summary.timestamp_regressions} capture-order timestamp regression(s); "
f"maximum={summary.maximum_timestamp_regression_ns / 1e6:.6f}ms. "
"File order will be preserved and those packets will be sent immediately.",
)
socket.if_nametoindex(args.interface)
interface_ip = interface_ipv4(args.interface)
event_log.done("network", f"validated output interface {args.interface} ({interface_ip})")
if args.dry_run:
for group, port in sorted(bundle.destinations):
print(f"{group}:{port}")
event_log.done("dry-run", "preflight complete; no packets were sent")
return 0
interrupted = False
previous_handlers: Dict[int, object] = {}
def request_stop(signum: int, _frame: object) -> None:
nonlocal interrupted
interrupted = True
event_log.warning("replay", f"received signal {signum}; stopping one-pass replay")
for signum in (signal.SIGINT, signal.SIGTERM):
previous_handlers[signum] = signal.getsignal(signum)
signal.signal(signum, request_stop)
sender = MulticastSender(interface_ip, args.send_buffer_mib * 1024 * 1024)
try:
actual_buffer = sender.socket.getsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF)
event_log.info(
"replay",
f"one-pass replay starts in {args.start_delay_seconds:.3f}s; "
"CME payloads and timestamps will not be changed",
)
event_log.info("network", f"actual UDP send buffer={actual_buffer} bytes; multicast TTL=1")
result = replay_capture(
target_packets(bundle.recording, bundle.destinations),
sender.send,
args.start_delay_seconds,
args.progress_seconds,
event_log,
lambda: interrupted,
)
finally:
sender.close()
for signum, handler in previous_handlers.items():
signal.signal(signum, handler)
if result.interrupted:
event_log.warning(
"replay", f"stopped after sending {result.packets} packets and {result.bytes} bytes"
)
return 130
event_log.done(
"replay",
f"one-pass replay complete: {result.packets} packets, {result.bytes} payload bytes, "
f">1ms late={result.late_packets}, "
f"maximum lateness={result.maximum_lateness_ns / 1e6:.3f}ms",
)
return 0
def main(argv: Optional[Sequence[str]] = None) -> int:
args = build_parser().parse_args(argv)
try:
return run(args)
except KeyboardInterrupt:
print("ERROR: interrupted during preflight", file=sys.stderr)
return 130
except (OSError, ValueError, json.JSONDecodeError, ET.ParseError) as error:
print(f"ERROR: {error}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())
+104
View File
@@ -0,0 +1,104 @@
import importlib.util
from pathlib import Path
import struct
import sys
import tempfile
import unittest
MODULE_PATH = Path(__file__).parents[1] / "cme_recorder.py"
SPEC = importlib.util.spec_from_file_location("cme_recorder", MODULE_PATH)
recorder = importlib.util.module_from_spec(SPEC)
assert SPEC.loader is not None
sys.modules[SPEC.name] = recorder
SPEC.loader.exec_module(recorder)
def sbe_message(template_id, body, block_length=None, schema_id=1, version=14):
if block_length is None:
block_length = len(body)
content = struct.pack("<HHHH", block_length, template_id, schema_id, version) + body
return struct.pack("<H", len(content) + 2) + content
def mdp_packet(sequence, *messages):
return struct.pack("<IQ", sequence, 123456789) + b"".join(messages)
class StubLog:
def __init__(self):
self.items = []
def info(self, *args, **kwargs):
self.items.append(("info", args, kwargs))
def done(self, *args, **kwargs):
self.items.append(("done", args, kwargs))
def warning(self, *args, **kwargs):
self.items.append(("warning", args, kwargs))
class RecorderTests(unittest.TestCase):
def test_parse_config_selects_requested_a_flows(self):
xml = """\
<configuration>
<channel id="310"><connections>
<connection id="310IA"><type feed-type="I"/><ip>224.0.31.1</ip><port>14310</port><feed>A</feed><host-ip>1.2.3.4</host-ip></connection>
<connection id="310NA"><type feed-type="N"/><ip>224.0.31.43</ip><port>14310</port><feed>A</feed></connection>
<connection id="310SA"><type feed-type="S"/><ip>224.0.31.22</ip><port>14310</port><feed>A</feed></connection>
<connection id="310SAMBO"><type feed-type="SMBO"/><ip>233.72.75.1</ip><port>23310</port><feed>A</feed></connection>
<connection id="310IB"><type feed-type="I"/><ip>224.0.32.1</ip><port>15310</port><feed>B</feed></connection>
</connections></channel>
</configuration>
"""
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "config.xml"
path.write_text(xml)
flows = recorder.parse_cme_flows(path, (310,), ("A",), include_mbo=True)
self.assertEqual([flow.feed_type for flow in flows], ["I", "N", "S", "SMBO"])
self.assertEqual(flows[0].source_ip, "1.2.3.4")
def test_parse_mdp_packet_reads_multiple_sbe_messages(self):
payload = mdp_packet(
42,
sbe_message(12, b"\x00" * 4),
sbe_message(54, b"\x00" * 16),
)
packet = recorder.parse_mdp_packet(payload)
self.assertEqual(packet.sequence, 42)
self.assertEqual([message.template_id for message in packet.messages], [12, 54])
def test_definition_stage_completes_tot_num_reports_cycle(self):
flow = recorder.Flow(310, "310NA", "A", "N", "224.0.31.43", 14310, None)
progress = recorder.FlowProgress(flow, 1, {}, StubLog())
body = b"\x00" + struct.pack("<I", 3) + b"\x00" * 8
for sequence in range(10, 13):
progress.observe(mdp_packet(sequence, sbe_message(54, body)))
self.assertTrue(progress.complete)
self.assertEqual(progress.expected_cycle_messages, 3)
self.assertEqual(progress.cycle_messages, 3)
def test_gap_restarts_recovery_completion_count(self):
flow = recorder.Flow(310, "310SA", "A", "S", "224.0.31.22", 14310, None)
progress = recorder.FlowProgress(flow, 1, {}, StubLog())
body = b"\x00" * 4 + struct.pack("<I", 2) + b"\x00" * 8
progress.observe(mdp_packet(10, sbe_message(52, body)))
progress.observe(mdp_packet(12, sbe_message(52, body)))
self.assertFalse(progress.complete)
self.assertEqual(progress.sequence_gaps, 1)
self.assertEqual(progress.cycle_messages, 1)
progress.observe(mdp_packet(13, sbe_message(52, body)))
self.assertTrue(progress.complete)
def test_incremental_stage_completes_after_requested_packets(self):
flow = recorder.Flow(311, "311IA", "A", "I", "224.0.31.2", 14311, None)
progress = recorder.FlowProgress(flow, 2, {}, StubLog())
progress.observe(mdp_packet(1))
self.assertFalse(progress.complete)
progress.observe(mdp_packet(2))
self.assertTrue(progress.complete)
if __name__ == "__main__":
unittest.main()
+239
View File
@@ -0,0 +1,239 @@
import hashlib
import importlib.util
import json
from pathlib import Path
import socket
import struct
import sys
import tempfile
import unittest
MODULE_PATH = Path(__file__).parents[1] / "cme_replayer.py"
SPEC = importlib.util.spec_from_file_location("cme_replayer", MODULE_PATH)
replayer = importlib.util.module_from_spec(SPEC)
assert SPEC.loader is not None
sys.modules[SPEC.name] = replayer
SPEC.loader.exec_module(replayer)
def pcapng_block(block_type, body):
padding = b"\x00" * ((-len(body)) % 4)
length = 12 + len(body) + len(padding)
return struct.pack("<II", block_type, length) + body + padding + struct.pack("<I", length)
def ethernet_udp(payload, destination_ip="224.0.31.1", destination_port=14310):
ethernet = b"\x01\x00\x5e\x00\x1f\x01" + b"\x02\x00\x00\x00\x00\x01" + struct.pack("!H", 0x0800)
source_ip = socket.inet_aton("10.10.10.1")
destination = socket.inet_aton(destination_ip)
udp = struct.pack("!HHHH", 12000, destination_port, 8 + len(payload), 0) + payload
ipv4 = struct.pack(
"!BBHHHBBH4s4s",
0x45,
0,
20 + len(udp),
1,
0,
1,
socket.IPPROTO_UDP,
0,
source_ip,
destination,
)
return ethernet + ipv4 + udp
def make_pcapng(packets):
section = pcapng_block(
replayer.PCAPNG_SECTION_HEADER,
struct.pack("<IHHq", 0x1A2B3C4D, 1, 0, -1),
)
# if_tsresol=9 makes the enhanced packet timestamps nanoseconds.
options = struct.pack("<HHB3x", 9, 1, 9) + struct.pack("<HH", 0, 0)
interface = pcapng_block(
replayer.PCAPNG_INTERFACE_DESCRIPTION,
struct.pack("<HHI", replayer.LINKTYPE_ETHERNET, 0, 65535) + options,
)
enhanced = []
for timestamp_ns, frame in packets:
body = struct.pack(
"<IIIII",
0,
timestamp_ns >> 32,
timestamp_ns & 0xFFFFFFFF,
len(frame),
len(frame),
) + frame
enhanced.append(pcapng_block(replayer.PCAPNG_ENHANCED_PACKET, body))
return section + interface + b"".join(enhanced)
def write_bundle(directory, pcapng, status="complete"):
directory = Path(directory)
(directory / "recording.pcapng").write_bytes(pcapng)
(directory / "config.xml").write_text(
"""<configuration><channel id="310"><connections><connection id="310IA">
<type feed-type="I"/><ip>224.0.31.1</ip><port>14310</port><feed>A</feed>
</connection></connections></channel></configuration>""",
encoding="utf-8",
)
(directory / "templates.xml").write_text("<templates/>", encoding="utf-8")
manifest = {
"bundle_format": "cme-mdp3-pcapng-v1",
"status": status,
"stages": [{"multicast_group": "224.0.31.1", "port": 14310}],
}
(directory / "manifest.json").write_text(
json.dumps(manifest) + "\n", encoding="utf-8"
)
with (directory / "checksums.sha256").open("w", encoding="utf-8") as stream:
for filename in ("recording.pcapng", "config.xml", "templates.xml", "manifest.json"):
digest = hashlib.sha256((directory / filename).read_bytes()).hexdigest()
stream.write(f"{digest} {filename}\n")
class StubLog:
def __init__(self):
self.items = []
def info(self, stage, message):
self.items.append((stage, message))
class FakeClock:
def __init__(self):
self.value = 5_000_000_000
def monotonic_ns(self):
return self.value
def sleep(self, seconds):
self.value += int(seconds * 1_000_000_000)
class ReplayerTests(unittest.TestCase):
def test_pcapng_reader_extracts_udp_payload_without_changes(self):
original = struct.pack("<IQ", 42, 123456789) + b"CME payload"
capture = make_pcapng([(1_000_000_000, ethernet_udp(original))])
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "recording.pcapng"
path.write_bytes(capture)
packets = list(replayer.iter_pcapng_udp(path))
self.assertEqual(len(packets), 1)
self.assertEqual(packets[0].capture_time_ns, 1_000_000_000)
self.assertEqual(packets[0].destination_ip, "224.0.31.1")
self.assertEqual(packets[0].destination_port, 14310)
self.assertEqual(packets[0].payload, original)
def test_bundle_validation_checks_manifest_and_hashes(self):
capture = make_pcapng([(1, ethernet_udp(b"one"))])
with tempfile.TemporaryDirectory() as directory:
write_bundle(directory, capture)
bundle = replayer.validate_bundle(Path(directory), allow_incomplete=False)
self.assertEqual(bundle.destinations, {("224.0.31.1", 14310)})
(Path(directory) / "recording.pcapng").write_bytes(b"changed")
with self.assertRaisesRegex(ValueError, "SHA-256 mismatch"):
replayer.validate_bundle(Path(directory), allow_incomplete=False)
def test_incomplete_bundle_requires_explicit_override(self):
capture = make_pcapng([(1, ethernet_udp(b"one"))])
with tempfile.TemporaryDirectory() as directory:
write_bundle(directory, capture, status="incomplete")
with self.assertRaisesRegex(ValueError, "--allow-incomplete"):
replayer.validate_bundle(Path(directory), allow_incomplete=False)
replayer.validate_bundle(Path(directory), allow_incomplete=True)
def test_scan_preserves_duration_and_requires_every_manifest_flow(self):
capture = make_pcapng(
[
(1_000_000_000, ethernet_udp(b"one")),
(1_250_000_000, ethernet_udp(b"two")),
]
)
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "recording.pcapng"
path.write_bytes(capture)
summary = replayer.scan_capture(path, {("224.0.31.1", 14310)})
self.assertEqual(summary.packets, 2)
self.assertEqual(summary.duration_ns, 250_000_000)
self.assertEqual(summary.bytes, 6)
self.assertEqual(summary.timestamp_regressions, 0)
def test_scan_reports_out_of_order_timestamps_without_rejecting_capture(self):
capture = make_pcapng(
[
(1_000_000_000, ethernet_udp(b"one")),
(999_000_000, ethernet_udp(b"two")),
(1_250_000_000, ethernet_udp(b"three")),
]
)
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "recording.pcapng"
path.write_bytes(capture)
summary = replayer.scan_capture(path, {("224.0.31.1", 14310)})
self.assertEqual(summary.packets, 3)
self.assertEqual(summary.duration_ns, 250_000_000)
self.assertEqual(summary.timestamp_regressions, 1)
self.assertEqual(summary.maximum_timestamp_regression_ns, 1_000_000)
def test_replay_uses_capture_offsets_and_sends_original_payloads(self):
packets = [
replayer.CapturedUdpPacket(100, "1.1.1.1", 1, "224.0.31.1", 14310, b"one"),
replayer.CapturedUdpPacket(250_000_100, "1.1.1.1", 1, "224.0.31.1", 14310, b"two"),
]
clock = FakeClock()
sent = []
def send(packet):
sent.append((clock.monotonic_ns(), packet.payload))
result = replayer.replay_capture(
packets,
send,
start_delay_seconds=1.0,
progress_seconds=0,
event_log=StubLog(),
should_stop=lambda: False,
monotonic_ns=clock.monotonic_ns,
sleep=clock.sleep,
)
self.assertEqual(sent, [(6_000_000_000, b"one"), (6_250_000_000, b"two")])
self.assertEqual(result.packets, 2)
self.assertEqual(result.bytes, 6)
self.assertFalse(result.interrupted)
def test_replay_sends_regressed_timestamp_immediately_in_file_order(self):
packets = [
replayer.CapturedUdpPacket(100, "1.1.1.1", 1, "224.0.31.1", 14310, b"one"),
replayer.CapturedUdpPacket(50, "1.1.1.1", 1, "224.0.31.1", 14310, b"two"),
replayer.CapturedUdpPacket(100_000_100, "1.1.1.1", 1, "224.0.31.1", 14310, b"three"),
]
clock = FakeClock()
sent = []
def send(packet):
sent.append((clock.monotonic_ns(), packet.payload))
replayer.replay_capture(
packets,
send,
start_delay_seconds=0,
progress_seconds=0,
event_log=StubLog(),
should_stop=lambda: False,
monotonic_ns=clock.monotonic_ns,
sleep=clock.sleep,
)
self.assertEqual(
sent,
[
(5_000_000_000, b"one"),
(5_000_000_000, b"two"),
(5_100_000_000, b"three"),
],
)
if __name__ == "__main__":
unittest.main()