iter embedding-abi-m5.3 (DONE 3/3): time-shard boundary-invisibility proof + friction harvest
Final functional M5 iteration (specae905de, plan 67027ab; on green m5.1204c171+ m5.2b724cd1whose leak-proof resolved viadbd76e5). Proves the chunk/window-boundary-invisibility claim the M4 retirement rests on: - adapter::fold_window — purely additive windowed sibling of fold_symbol (same MidPriceStream+Kernel path; stream_tick_windowed). fold_symbol/MidPriceStream/Kernel byte-untouched. - chrono as a DEV-dep only — the time-shard test derives (y,m)->Unix-ms bounds (data-server's own precedent) and passes raw ms to the bin; the bin is date-math-free; Invariant 1 unaffected. - new single-responsibility [[bin]] timeshard_runner — one thread per disjoint window, each owning its Kernel (Ctx:!Send => compile-time per-thread-ctx). - tests/timeshard.rs — spec §3 strength-ordered: (a) per-shard (acc,n) BIT-EXACT vs single-thread host fold of that exact window in stream order (THE boundary-invisibility proof); (b) Σ partials bit-exact by construction; (c) whole-window within REL_TOL=1e-6 only (cross-shard f64 reassociation is host arithmetic, never a kernel property — bit-exact there would be the bug; tolerance derived from the recursive-summation error bound). Plus the now-deterministic leak-Σ + a journal-only friction timing capture (no bench gate, no timing assertion). Concrete pin: EURUSD 2017-03/04/05 (recon-verified present, contiguous, no gaps; data-server has no month-listing API). RED was a genuine deterministic compile failure (declared-but-absent bin source; Cargo missing-source error vs the plan's predicted env! error — same cause, no false-green; benign plan-text imprecision in Concerns). Boss-verified independently: timeshard determinism 5/5 (no jitter); full ail-embed suite 0 failed AND 0 ignored across all binaries (m5.2 symbol_fan_swarm_leak_free stays un-ignored & green — the saga's resolution intact); isolated embed_rc_global_stats_race still green; AILang workspace data-server count 0; compiler-surface diff empty (zero diff to crates/ailang-*, crates/ail/, runtime/, examples/*.ail, root Cargo.toml, DESIGN.md). Friction-harvest deliverable (P2 flat-array-decision input): host-per-tick-FFI ~= 658 ms / 3,192,562 ticks ~= ~206 ns/tick at real EURUSD volume. M5 functional work complete; milestone closes next via the mandatory audit (which also handles the pre-existing DESIGN.md:2358-2360 "additive M4 concern" drift, out of m5.3 scope). Includes the per-iter journal, stats, and the INDEX.md line.
This commit is contained in:
Generated
+1
@@ -32,6 +32,7 @@ dependencies = [
|
||||
name = "ail-embed"
|
||||
version = "0.0.1"
|
||||
dependencies = [
|
||||
"chrono",
|
||||
"data-server",
|
||||
"tempfile",
|
||||
"zip",
|
||||
|
||||
@@ -30,7 +30,17 @@ data-server = { path = "../../libs/data-server" }
|
||||
[dev-dependencies]
|
||||
zip = "2"
|
||||
tempfile = "3"
|
||||
# Test-only: the time-shard test derives (year,month) -> Unix-ms
|
||||
# window bounds via chrono (data-server's own bound precedent,
|
||||
# data-server/src/lib.rs:696-699) and passes raw ms to the bin, so
|
||||
# the bin itself stays date-math-free. chrono is already transitively
|
||||
# present via data-server; this is the test-direct declaration.
|
||||
chrono = "0.4"
|
||||
|
||||
[[bin]]
|
||||
name = "swarm_runner"
|
||||
path = "src/bin/swarm_runner.rs"
|
||||
|
||||
[[bin]]
|
||||
name = "timeshard_runner"
|
||||
path = "src/bin/timeshard_runner.rs"
|
||||
|
||||
@@ -68,6 +68,25 @@ pub fn fold_symbol(
|
||||
Some(Kernel::new().run(MidPriceStream::new(it).take(max_ticks)))
|
||||
}
|
||||
|
||||
/// Fold the mid-price stream of `symbol` restricted to the inclusive
|
||||
/// Unix-millisecond window `[from_ms, to_ms]` through the M3 kernel.
|
||||
/// `None` if the symbol has no tick stream in that window. The window
|
||||
/// is the bound (no tick cap): the result is `(Σ mid, count)` over
|
||||
/// exactly the records data-server yields for that window, in stream
|
||||
/// order — identical f64-add sequence to a single-thread host fold of
|
||||
/// the same window, hence bit-exact against it. The windowed sibling
|
||||
/// of `fold_symbol`; same `MidPriceStream` + `Kernel` path, only the
|
||||
/// data-server entry differs (`stream_tick_windowed`).
|
||||
pub fn fold_window(
|
||||
server: &Arc<DataServer>,
|
||||
symbol: &str,
|
||||
from_ms: i64,
|
||||
to_ms: i64,
|
||||
) -> Option<(f64, i64)> {
|
||||
let it = server.stream_tick_windowed(symbol, Some(from_ms), Some(to_ms))?;
|
||||
Some(Kernel::new().run(MidPriceStream::new(it)))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::tick_to_px;
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
//! Time-shard swarm harness (spec §3). Run as a subprocess by
|
||||
//! `tests/timeshard.rs` under `AILANG_RC_STATS=1`.
|
||||
//!
|
||||
//! argv: `<data_dir> <symbol> <from:to> [<from:to> ...]` — one
|
||||
//! inclusive Unix-ms window per shard. Date math lives in the test
|
||||
//! (chrono); this bin is date-math-free. Spawns one thread per
|
||||
//! window, each building its OWN `Kernel` (`Ctx: !Send` forces
|
||||
//! one-ctx-per-thread — a shared-ctx design is `error[E0277]`).
|
||||
//! Prints, in shard order, `RESULT <idx> <from> <to> <accbits:016x>
|
||||
//! <n>`, then one `TIMING total_ticks=<Σn> elapsed_ms=<wall>` line
|
||||
//! (the friction-harvest observation; the test records it in the
|
||||
//! close-out journal, it is not asserted).
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
|
||||
use ail_embed::adapter::fold_window;
|
||||
use data_server::DataServer;
|
||||
|
||||
fn main() {
|
||||
let mut args = std::env::args().skip(1);
|
||||
let data_dir = args.next().expect("argv[1] = data dir");
|
||||
let symbol = args.next().expect("argv[2] = symbol");
|
||||
let windows: Vec<(i64, i64)> = args
|
||||
.map(|a| {
|
||||
let (f, t) = a.split_once(':').expect("window arg is from:to");
|
||||
(
|
||||
f.parse::<i64>().expect("from_ms i64"),
|
||||
t.parse::<i64>().expect("to_ms i64"),
|
||||
)
|
||||
})
|
||||
.collect();
|
||||
assert!(
|
||||
windows.len() >= 2,
|
||||
"need >= 2 disjoint windows for a time-shard, got {}",
|
||||
windows.len()
|
||||
);
|
||||
|
||||
let server = Arc::new(DataServer::new(&data_dir));
|
||||
|
||||
let started = Instant::now();
|
||||
let handles: Vec<_> = windows
|
||||
.iter()
|
||||
.copied()
|
||||
.enumerate()
|
||||
.map(|(idx, (from, to))| {
|
||||
let srv = Arc::clone(&server);
|
||||
let sym = symbol.clone();
|
||||
// Each thread builds its own Kernel inside the closure —
|
||||
// Kernel/Ctx are !Send, so this is the ONLY shape that
|
||||
// compiles (one ctx per thread, type-enforced).
|
||||
std::thread::spawn(move || {
|
||||
let (acc, n) = fold_window(&srv, &sym, from, to)
|
||||
.expect("window has a tick stream");
|
||||
(idx, from, to, acc, n)
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
|
||||
let mut results: Vec<(usize, i64, i64, f64, i64)> = handles
|
||||
.into_iter()
|
||||
.map(|h| h.join().expect("shard thread panicked"))
|
||||
.collect();
|
||||
let elapsed_ms = started.elapsed().as_millis();
|
||||
results.sort_by_key(|r| r.0);
|
||||
|
||||
let mut total_ticks: i64 = 0;
|
||||
for (idx, from, to, acc, n) in &results {
|
||||
total_ticks += *n;
|
||||
println!("RESULT {idx} {from} {to} {:016x} {n}", acc.to_bits());
|
||||
}
|
||||
println!("TIMING total_ticks={total_ticks} elapsed_ms={elapsed_ms}");
|
||||
}
|
||||
@@ -0,0 +1,225 @@
|
||||
//! Time-shard boundary-invisibility E2E (spec Architecture §3
|
||||
//! second assertion / Testing strategy §3). One symbol, N disjoint
|
||||
//! single-month windows, one thread per window (each owning its
|
||||
//! `Kernel` — `Ctx: !Send`). Asserts, in spec strength order:
|
||||
//! (a) each shard's `(acc,n)` == a single-thread host fold of THAT
|
||||
//! EXACT window in stream order, BIT-EXACT — the
|
||||
//! boundary-invisibility proof (the kernel result for a window
|
||||
//! is independent of how data-server chunked it);
|
||||
//! (b) the host-summed `(Σacc,Σn)` == the identically-summed
|
||||
//! reference partials, bit-exact by construction;
|
||||
//! (c) the whole-window single-thread total within an
|
||||
//! f64-reassociation tolerance only (cross-shard re-association
|
||||
//! is host arithmetic, NOT a kernel property — asserting it
|
||||
//! bit-exact would be WRONG, per the spec).
|
||||
//! Plus the global leak-Σ (deterministic since dbd76e5) and a
|
||||
//! journal-only friction timing capture (NO bench gate).
|
||||
//!
|
||||
//! Concrete pin: EURUSD, months 2017-03 / 2017-04 / 2017-05 (all
|
||||
//! present on /mnt, contiguous, no gaps — recon-verified). data-server
|
||||
//! has no public month-listing API, so the months are hard-pinned.
|
||||
//! Skips when /mnt absent — data-server's own precedent.
|
||||
|
||||
use std::process::Command;
|
||||
use std::sync::Arc;
|
||||
|
||||
use chrono::{NaiveDate, TimeZone, Utc};
|
||||
use data_server::{DataServer, DEFAULT_DATA_PATH};
|
||||
|
||||
const SYMBOL: &str = "EURUSD";
|
||||
/// Three disjoint single-month windows. Contiguous so the whole-window
|
||||
/// union is exactly 2017-03-01 .. 2017-05-31 23:59:59.999.
|
||||
const SHARDS: [(i32, u32); 3] = [(2017, 3), (2017, 4), (2017, 5)];
|
||||
/// Relative tolerance for the (c) whole-window cross-check. Derivation:
|
||||
/// recursive f64 summation of n terms has error <= (n-1)*eps*Σ|x_i|.
|
||||
/// One EURUSD month ~250k ticks => 3 shards ~750k adds of O(1) mids;
|
||||
/// Σ|x| ~ 8.3e5, (n-1)*eps ~ 8.3e-11 => abs err ~ 6.8e-5, rel ~ 8e-11.
|
||||
/// 1e-6 is ~4 orders of safety yet still catches any real bug
|
||||
/// (a genuine miscount is off by >> 1 ppm).
|
||||
const REL_TOL: f64 = 1e-6;
|
||||
|
||||
fn skip_if_no_data() -> bool {
|
||||
!std::path::Path::new(DEFAULT_DATA_PATH).exists()
|
||||
}
|
||||
|
||||
/// Inclusive Unix-ms bounds for the whole of calendar month `(y, m)`.
|
||||
/// `filter_files` windows at month granularity via
|
||||
/// `unix_ms_to_year_month`, and `filter_chunk` is record-level
|
||||
/// inclusive `[from, to]`; first-instant .. last-instant of the month
|
||||
/// selects exactly that month's file and all its records.
|
||||
fn month_window(y: i32, m: u32) -> (i64, i64) {
|
||||
let from = Utc
|
||||
.from_utc_datetime(
|
||||
&NaiveDate::from_ymd_opt(y, m, 1)
|
||||
.unwrap()
|
||||
.and_hms_opt(0, 0, 0)
|
||||
.unwrap(),
|
||||
)
|
||||
.timestamp_millis();
|
||||
let (ny, nm) = if m == 12 { (y + 1, 1) } else { (y, m + 1) };
|
||||
let next_from = Utc
|
||||
.from_utc_datetime(
|
||||
&NaiveDate::from_ymd_opt(ny, nm, 1)
|
||||
.unwrap()
|
||||
.and_hms_opt(0, 0, 0)
|
||||
.unwrap(),
|
||||
)
|
||||
.timestamp_millis();
|
||||
(from, next_from - 1)
|
||||
}
|
||||
|
||||
/// Single-thread host reference: fold the mid stream over EXACTLY the
|
||||
/// inclusive `[from_ms, to_ms]` window in data-server stream order
|
||||
/// (`acc += mid; n += 1`). Bit-exact with `adapter::fold_window`
|
||||
/// because the f64-add order is identical.
|
||||
fn reference_window(server: &Arc<DataServer>, from_ms: i64, to_ms: i64) -> (f64, i64) {
|
||||
let mut it = server
|
||||
.stream_tick_windowed(SYMBOL, Some(from_ms), Some(to_ms))
|
||||
.expect("EURUSD has a tick stream in the pinned window");
|
||||
let (mut acc, mut n) = (0.0_f64, 0_i64);
|
||||
while let Some(chunk) = it.next_chunk() {
|
||||
for r in chunk.iter() {
|
||||
acc += (r.ask + r.bid) / 2.0;
|
||||
n += 1;
|
||||
}
|
||||
}
|
||||
(acc, n)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn timeshard_boundary_invisibility_and_leak_free() {
|
||||
if skip_if_no_data() {
|
||||
eprintln!("skipping: {DEFAULT_DATA_PATH} absent (mirrors data-server)");
|
||||
return;
|
||||
}
|
||||
|
||||
let windows: Vec<(i64, i64)> =
|
||||
SHARDS.iter().map(|&(y, m)| month_window(y, m)).collect();
|
||||
|
||||
// Spawn the time-shard runner: <data_dir> <symbol> then one
|
||||
// "from:to" arg per shard. It folds each window on its own thread
|
||||
// (own Kernel) and prints `RESULT <idx> <from> <to> <accbits> <n>`
|
||||
// + a final `TIMING total_ticks=<N> elapsed_ms=<M>` line.
|
||||
let mut cmd = Command::new(env!("CARGO_BIN_EXE_timeshard_runner"));
|
||||
cmd.arg(DEFAULT_DATA_PATH).arg(SYMBOL);
|
||||
for (f, t) in &windows {
|
||||
cmd.arg(format!("{f}:{t}"));
|
||||
}
|
||||
let out = cmd
|
||||
.env("AILANG_RC_STATS", "1")
|
||||
.output()
|
||||
.expect("spawn timeshard_runner");
|
||||
let stdout = String::from_utf8_lossy(&out.stdout);
|
||||
let stderr = String::from_utf8_lossy(&out.stderr);
|
||||
assert!(
|
||||
out.status.success(),
|
||||
"timeshard_runner exited {:?}\nstdout:\n{stdout}\nstderr:\n{stderr}",
|
||||
out.status.code()
|
||||
);
|
||||
|
||||
let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
|
||||
|
||||
// Parse RESULT lines into (idx -> (from,to,acc,n)).
|
||||
let mut shards: Vec<(usize, i64, i64, f64, i64)> = Vec::new();
|
||||
for line in stdout.lines().filter(|l| l.starts_with("RESULT ")) {
|
||||
let mut t = line.split_whitespace();
|
||||
let _ = t.next(); // RESULT
|
||||
let idx: usize = t.next().unwrap().parse().unwrap();
|
||||
let from: i64 = t.next().unwrap().parse().unwrap();
|
||||
let to: i64 = t.next().unwrap().parse().unwrap();
|
||||
let acc = f64::from_bits(
|
||||
u64::from_str_radix(t.next().unwrap(), 16).unwrap(),
|
||||
);
|
||||
let n: i64 = t.next().unwrap().parse().unwrap();
|
||||
shards.push((idx, from, to, acc, n));
|
||||
}
|
||||
shards.sort_by_key(|s| s.0);
|
||||
assert_eq!(
|
||||
shards.len(),
|
||||
SHARDS.len(),
|
||||
"expected {} shard RESULT lines, got {}",
|
||||
SHARDS.len(),
|
||||
shards.len()
|
||||
);
|
||||
|
||||
// (a) Per-shard BIT-EXACT vs a single-thread host fold of that
|
||||
// exact window. THE boundary-invisibility proof.
|
||||
let mut sum_acc = 0.0_f64;
|
||||
let mut sum_n = 0_i64;
|
||||
let mut ref_sum_acc = 0.0_f64;
|
||||
let mut ref_sum_n = 0_i64;
|
||||
for (i, &(idx, from, to, acc, n)) in shards.iter().enumerate() {
|
||||
assert_eq!(idx, i, "shard indices contiguous 0..N");
|
||||
assert_eq!((from, to), windows[i], "shard {i} window matches pin");
|
||||
let (r_acc, r_n) = reference_window(&server, from, to);
|
||||
assert_eq!(
|
||||
acc.to_bits(),
|
||||
r_acc.to_bits(),
|
||||
"shard {i} ({from}..{to}): kernel acc BIT-EXACT vs \
|
||||
single-thread host fold of that exact window \
|
||||
(boundary-invisibility)"
|
||||
);
|
||||
assert_eq!(n, r_n, "shard {i}: tick count == reference");
|
||||
assert!(r_n > 0, "shard {i}: non-empty window");
|
||||
sum_acc += acc;
|
||||
sum_n += n;
|
||||
ref_sum_acc += r_acc;
|
||||
ref_sum_n += r_n;
|
||||
}
|
||||
|
||||
// (b) Host-summed partials == identically-summed reference
|
||||
// partials, BIT-EXACT by construction (same values, same
|
||||
// host summation order).
|
||||
assert_eq!(
|
||||
sum_acc.to_bits(),
|
||||
ref_sum_acc.to_bits(),
|
||||
"Σ shard acc bit-exact vs Σ reference partials"
|
||||
);
|
||||
assert_eq!(sum_n, ref_sum_n, "Σ shard n == Σ reference n");
|
||||
|
||||
// (c) Whole-window single-thread total within REL_TOL only —
|
||||
// cross-shard re-association is host arithmetic, NOT a kernel
|
||||
// property; asserting bit-exact here would be WRONG (spec §3).
|
||||
let whole_from = windows.first().unwrap().0;
|
||||
let whole_to = windows.last().unwrap().1;
|
||||
let (whole_acc, whole_n) = reference_window(&server, whole_from, whole_to);
|
||||
assert_eq!(
|
||||
whole_n, sum_n,
|
||||
"whole-window tick count == Σ shard n (windows are an exact \
|
||||
disjoint partition of the union)"
|
||||
);
|
||||
let rel = ((whole_acc - sum_acc) / whole_acc).abs();
|
||||
assert!(
|
||||
rel < REL_TOL,
|
||||
"whole-window total {whole_acc} vs Σ partials {sum_acc}: \
|
||||
relative diff {rel:e} must be < {REL_TOL:e} (f64 \
|
||||
reassociation only — NOT a kernel discrepancy)"
|
||||
);
|
||||
|
||||
// Global leak-Σ across ALL ailang_rc_stats: lines (deterministic
|
||||
// since dbd76e5: atomic global counters + build.rs runtime-track).
|
||||
let mut allocs: u64 = 0;
|
||||
let mut frees: u64 = 0;
|
||||
let mut seen = 0usize;
|
||||
for line in stderr.lines().filter(|l| l.starts_with("ailang_rc_stats:")) {
|
||||
seen += 1;
|
||||
for tok in line.split_whitespace() {
|
||||
if let Some(v) = tok.strip_prefix("allocs=") {
|
||||
allocs += v.parse::<u64>().expect("allocs= u64");
|
||||
} else if let Some(v) = tok.strip_prefix("frees=") {
|
||||
frees += v.parse::<u64>().expect("frees= u64");
|
||||
}
|
||||
}
|
||||
}
|
||||
assert!(seen > 0, "no ailang_rc_stats line; stderr:\n{stderr}");
|
||||
assert_eq!(
|
||||
allocs, frees,
|
||||
"globally leak-free: Σallocs==Σfrees across {seen} stat line(s)"
|
||||
);
|
||||
|
||||
// Friction harvest: journal-only observation, NO assertion (the
|
||||
// value feeds the P2 batch-crossing decision; it is not a gate).
|
||||
if let Some(tl) = stdout.lines().find(|l| l.starts_with("TIMING ")) {
|
||||
eprintln!("m5.3 friction-harvest: {tl} (Σn={sum_n})");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user