diff --git a/interactive/examples/ddir_server.rs b/interactive/examples/ddir_server.rs index 3ff3e7a7e..67bf91d89 100644 --- a/interactive/examples/ddir_server.rs +++ b/interactive/examples/ddir_server.rs @@ -124,7 +124,7 @@ fn parse_command(line: &str) -> Result { } Ok(Command::Feed { prog, input, key, val, time, diff }) } - "tick" => Ok(Command::Tick), + "tick" if toks.len() == 1 => Ok(Command::Tick { n: 1 }), "bind" | "unbind" if toks.len() == 4 => { let trace = toks[1].to_string(); let prog = toks[2].to_string(); @@ -145,7 +145,6 @@ fn parse_command(line: &str) -> Result { Ok(Command::Peek { trace, key }) } "list" => Ok(Command::List), - "help" => Ok(Command::Help), "exit" | "quit" => Ok(Command::Exit), other => Err(format!("unknown or malformed command {:?} (try `help`)", other)), } @@ -181,8 +180,17 @@ fn dispatch(cmd: &Command, server: &mut Server, worker: &mut Worker) -> bool { } } } - Command::Tick => { - server.tick(worker); + Command::FeedBatch { prog, input, updates } => { + if w0 { + if let Err(e) = server.feed_batch(prog, *input, updates.clone()) { + println!("error: {}", e); + } + } + } + Command::Tick { n } => { + for _ in 0..*n { + server.tick(worker); + } if w0 { println!("tick -> epoch {}", server.epoch()); } } Command::Drop { name } => match server.drop_program(worker, name) { @@ -207,7 +215,6 @@ fn dispatch(cmd: &Command, server: &mut Server, worker: &mut Worker) -> bool { } } Command::List => if w0 { server.list(); }, - Command::Help => if w0 { print_help(); }, Command::Exit => return false, } true @@ -297,6 +304,10 @@ fn main() { if line.is_empty() || line.starts_with('#') || line.starts_with("--") { continue; } + if line == "help" { + print_help(); + continue; + } match parse_command(line) { Ok(cmd) => { let is_exit = matches!(cmd, Command::Exit); diff --git a/interactive/server/Cargo.toml b/interactive/server/Cargo.toml index 6df438ef8..d4b617152 100644 --- a/interactive/server/Cargo.toml +++ b/interactive/server/Cargo.toml @@ -18,4 +18,5 @@ differential-dataflow = { workspace = true } timely = { workspace = true } interactive = { path = ".." } diagnostics = { path = "../../diagnostics" } +serde = { version = "1.0", features = ["derive"] } tungstenite = "0.26" diff --git a/interactive/server/README.md b/interactive/server/README.md index 1f0e9f1c0..6d552e850 100644 --- a/interactive/server/README.md +++ b/interactive/server/README.md @@ -1,16 +1,39 @@ # Live DDIR server -One long-running timely worker hosts interpreted DDIR dataflows through a +A long-running Timely worker group hosts interpreted DDIR dataflows through a load-run-drop lifecycle. Programs share results by name — each may import -collections that others export — and clients follow along over TCP, -WebSocket, or stdin. +collections that others export — and clients follow along over TCP, WebSocket, +or stdin. Run `cargo run -p ddir-server`, then open `interactive/server/console.html` or connect a line-oriented client to TCP port 7777. The same protocol is available -over WebSocket on port 7778. Set `DDIR_BIND`, `DDIR_WS_BIND`, or -`DDIR_TICK_MS` to change those defaults; `DDIR_TICK_MS=0` disables automatic -progress while subscriptions are active. The current `diagnostics` crate is -connected on `DDIR_DIAG_PORT` (default 51371). +over WebSocket on port 7778. Set `DDIR_BIND` or `DDIR_WS_BIND` to change those +defaults. Diagnostics are disabled by default so an idle server can park; +`DDIR_DIAGNOSTICS=1` enables the diagnostics dataflow and its listener on +`DDIR_DIAG_PORT` (default 51371). `DDIR_WORKERS` selects the number of worker +threads (default 1), and `DDIR_BACKEND=vec|corgi` selects the renderer for installed +programs (default `vec`). + +One backend is selected for the whole server. The current registry is a +transitional row-speaking bridge: inputs and imports convert from `Value` rows +to Corgi columns at a program boundary, and exports convert back before they +become shareable traces. A Corgi program stays columnar between those +boundaries, but a production Corgi server should replace the bridge with native +columnar inputs and traces. + +Worker 0 admits one FIFO control stream and routes one ordered record to every +worker. Small commands are replicated; framed input batches are partitioned by +body position before exchange, so each worker receives only its local typed +shard. Because the one source already defines a total order, command +coordination uses no wall clock or distributed sequencer. Workers execute +commands serially in that order without a physical rendezvous between commands; +Timely progress and probes establish logical completion where it is required. +Response channels remain local to worker 0. + +Transport threads wake worker 0 when they enqueue a control event, and Timely +wakes the other workers when their control record arrives. When neither Timely +nor the control plane has work, workers park through the scheduler; the server +does not poll requests with a periodic sleep. Every request can begin with an arbitrary request id. If omitted, the server generates one. Responses are ` data ...`, followed by ` ok ...` or @@ -27,6 +50,7 @@ pipe-syntax program: export "graph.edges" = edges; graph end-load tail graph.edges + tick A binding may also be spelled as a call, so `edges=random(seed=1,arity=2,range=8,count=12,churn=1)` redirects the local @@ -35,14 +59,11 @@ import named `edges` to the same content-addressed source as the fixed-size window into an infinite hash-derived sequence and replaces `churn` rows on every tick. -Automatic ticking happens only while at least one tail is active. This makes a -live demonstration move without assigning input durability semantics to DDIR. -Explicit `tick [n]` remains available for reproducible sessions. Treat -auto-tick as demo furniture rather than a design commitment: as specified, -observation advances time (an observer effect), and the alternative — that a -watcher must be present to move things along, by ticking or by running a -metronome client whose ticks are ordinary logged commands — may be the better -design once the server has real tenants. +Only an explicit, ordered `tick [n]` closes epochs. Tails observe progress but +do not cause it, and an idle server has no deadline to service. A future +queue-driven commit policy can seal an epoch as soon as the preceding epoch +retires and intents are waiting; that decision should come from logical queue +state, not elapsed wall-clock time. ## Writes: `feed` @@ -52,6 +73,31 @@ pushes one update into a loaded program's positional input, exactly as in the `ddir_server` example (`1,2` → a tuple; `_` → unit; a closed scalar term such as `inject(2,tuple(3,4))` for ADT-shaped rows). +For many updates to one target at the current epoch, frame them as one feed: + + feed world 0 begin + 7 val=9 + 7 val=-3 + 8 val=10 diff=-1 + end-feed + +Each body row is ` [val=] [diff=]`; `time=` is intentionally absent +because the enclosing command supplies one epoch. The complete body is admitted +atomically, and workers divide its rows by body position before introducing +them to the dataflow. Partitioning happens before worker transport, avoiding a +full `Value` batch clone on every worker. Text parsing and the row-to-column +boundary remain visible optimization opportunities rather than hidden protocol +behavior. + +The batch's `ok` means that the complete request was admitted to the ordered +worker stream; it does not wait for every worker to stage its shard. A later +`tick` is the visibility and completion boundary for those rows. + +This text-to-`Value` framing is a convenience and compatibility path, not the +intended representation for high-volume ingestion. Large or already-columnar +payloads should be acquired and partitioned through the data plane, with only a +small descriptor entering the ordered control stream. + The stance on contention: **writes are open; policy lives in the dataflow**. The server does not decide who may write what. Cooperating clients follow a simple protocol — include your id and an ordering epoch in the data — and @@ -81,14 +127,15 @@ and bind the export `f(state) + (seed | negate)` to the feedback input; then perturbations. A bound source cannot be dropped (it holds an importer), nor can the bound target (unbind first). -## One gate +## Intake gates -Loads are cheap to request and costly to render, so intake is bounded: -`DDIR_MAX_PROGRAM_BYTES` (default 65536) — a larger `load` body is swallowed -and rejected with one error, before parsing. This is transport self-defense, -not semantics. There are no ownership or quota gates: sessions are trusted, -and admission policy (auth, quotas, rate limits) belongs in a fronting proxy -if a deployment ever needs one. +Multi-line intake is bounded: `DDIR_MAX_PROGRAM_BYTES` (default 65536) caps a +`load`, and `DDIR_MAX_FEED_BYTES` (default 16 MiB) caps a framed `feed`. An +oversized or malformed body is swallowed through its terminator and rejected +with one error; no partial command reaches a worker. This is transport +self-defense, not semantics. There are no ownership or quota gates: sessions +are trusted, and admission policy (auth, quotas, rate limits) belongs in a +fronting proxy if a deployment ever needs one. ## Demos diff --git a/interactive/server/demo/two_sessions.py b/interactive/server/demo/two_sessions.py index 48100ef27..cc78837e4 100644 --- a/interactive/server/demo/two_sessions.py +++ b/interactive/server/demo/two_sessions.py @@ -73,7 +73,6 @@ def main(): DDIR_BIND=f"127.0.0.1:{PORT}", DDIR_WS_BIND=f"127.0.0.1:{PORT + 1}", DDIR_DIAG_PORT=str(PORT + 2), - DDIR_TICK_MS="0", DDIR_MAX_PROGRAM_BYTES="4096", ) server = subprocess.Popen( diff --git a/interactive/server/src/cmd.rs b/interactive/server/src/cmd.rs index d13298521..df8a06ec6 100644 --- a/interactive/server/src/cmd.rs +++ b/interactive/server/src/cmd.rs @@ -7,15 +7,17 @@ //! - `data ` — one streamed body line (peek/tail batches) //! - `end` — terminator after a stream of `data` lines //! -//! Multi-line bodies (a DDIR program) come via a two-phase upload: -//! ` load begin` opens; subsequent lines are -//! literal program text terminated by ` end-load`. +//! Multi-line bodies use two-phase framing: `load ... begin` accepts literal +//! DDIR through `end-load`, while `feed begin` accepts row updates +//! through `end-feed` and becomes one non-interleavable server command. -use std::collections::BTreeMap; +use std::any::{type_name_of_val, Any}; +use std::collections::{BTreeMap, HashMap}; use std::panic::{catch_unwind, AssertUnwindSafe}; use interactive::ir::{eval, Diff, Value}; -use interactive::server::OuterTime; +use interactive::scope_ir::{Program, Source}; +use interactive::server::{Command as ServerCommand, InputUpdate, OuterTime}; pub type ReqId = String; @@ -69,6 +71,12 @@ pub enum Cmd { time: Option, diff: Diff, }, + /// Atomically stage many rows into one program input at the current epoch. + FeedBatch { + prog: String, + input: usize, + updates: Vec, + }, /// Bind a trace's changes into `prog`'s positional `input`, delivered at /// each tick one epoch delayed — the write path for installed programs. Bind { @@ -115,11 +123,159 @@ pub enum DataflowRef { Name(String), } -/// A parsed request: reqid plus the command (or a parse error). +/// A command ready to broadcast to the worker group. Protocol-only tail +/// lifecycle wraps the typed server command vocabulary rather than leaking +/// response channels into the dataflow. +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] +pub enum PreparedCommand { + Server(ServerCommand), + Tail { name: String }, + Stop { tail_reqid: ReqId }, +} + +/// Parse and lower the expensive parts of a protocol command on its session +/// thread. Every worker receives the same typed program, and malformed DDIR +/// never reaches a Timely worker. +pub fn prepare(command: Cmd) -> Result { + let command = match command { + Cmd::Load { + id_hint, + bindings, + program, + explain, + } => { + if explain { + return Err( + "load --explain is reserved; explanation is not implemented here yet".into(), + ); + } + let mut program = catch_unwind(AssertUnwindSafe(|| { + let statements = interactive::parse::pipe::parse(&program); + interactive::lower::lower_tree(statements) + })) + .map_err(panic_message)?; + apply_bindings(&mut program, &bindings)?; + program.optimize(); + ServerCommand::Install { + name: id_hint, + program, + } + } + Cmd::Drop { target } => ServerCommand::Drop { + name: match target { + DataflowRef::Name(name) => name, + DataflowRef::Id(id) => { + return Err(format!( + "numeric dataflow id {} is no longer exposed; use its name", + id + )) + } + }, + }, + Cmd::List => ServerCommand::List, + Cmd::Peek { name } => ServerCommand::Peek { + trace: name, + key: None, + }, + Cmd::Tail { name } => return Ok(PreparedCommand::Tail { name }), + Cmd::Stop { tail_reqid } => return Ok(PreparedCommand::Stop { tail_reqid }), + Cmd::Feed { + prog, + input, + key, + val, + time, + diff, + } => ServerCommand::Feed { + prog, + input, + key, + val, + time, + diff, + }, + Cmd::FeedBatch { + prog, + input, + updates, + } => ServerCommand::FeedBatch { + prog, + input, + updates, + }, + Cmd::Bind { trace, prog, input } => ServerCommand::Bind { trace, prog, input }, + Cmd::Unbind { trace, prog, input } => ServerCommand::Unbind { trace, prog, input }, + Cmd::Query { .. } => { + return Err("query is reserved for --explain dataflows and is not implemented".into()) + } + Cmd::Tick { n } => ServerCommand::Tick { n }, + Cmd::Exit => ServerCommand::Exit, + }; + Ok(PreparedCommand::Server(command)) +} + +fn apply_bindings( + program: &mut Program, + bindings: &BTreeMap, +) -> Result<(), String> { + for (local, binding) in bindings { + let import = program + .root + .imports + .iter_mut() + .find(|import| import.name == *local) + .ok_or_else(|| format!("binding names no import {:?}", local))?; + import.from = Source::Trace(random_binding(binding)?); + } + Ok(()) +} + +/// Translate the call spelling of a binding (`random(...)`) into its +/// content-addressed source name. +fn random_binding(binding: &str) -> Result { + let Some(body) = binding + .strip_prefix("random(") + .and_then(|s| s.strip_suffix(')')) + else { + return Ok(binding.to_string()); + }; + let mut values: HashMap<&str, &str> = HashMap::new(); + for field in body.split(',') { + let (key, value) = field + .trim() + .split_once('=') + .ok_or_else(|| format!("malformed random field {:?}", field))?; + values.insert(key.trim(), value.trim()); + } + let nodes = values.remove("range").ok_or("random requires range")?; + let edges = values.remove("count").ok_or("random requires count")?; + let arity = values.remove("arity").unwrap_or("2"); + let seed = values.remove("seed").unwrap_or("0"); + let churn = values.remove("churn").unwrap_or("0"); + if !values.is_empty() { + return Err(format!("unknown random fields: {:?}", values.keys())); + } + Ok(format!( + "random:nodes={},edges={},arity={},seed={},churn={}", + nodes, edges, arity, seed, churn + )) +} + +fn panic_message(panic: Box) -> String { + if let Some(s) = panic.downcast_ref::<&str>() { + (*s).to_string() + } else if let Some(s) = panic.downcast_ref::() { + s.clone() + } else { + format!("DDIR parser panicked ({})", type_name_of_val(&panic)) + } +} + +/// A prepared request: reqid plus the command (or a parse/lowering error). #[derive(Debug)] pub struct Request { pub reqid: ReqId, - pub kind: Result, + pub kind: Result, /// Where to route responses for this request (and, for `tail`, all /// subsequent batches until `stop`). Cloned from the per-connection /// outbound sender. @@ -129,14 +285,16 @@ pub struct Request { pub connection_id: ConnectionId, } -/// State carried between lines so the parser can splice a multi-line -/// `load ... begin` body together. The parser hands back either a -/// complete `Request` or `None` (more lines required). +/// State carried between lines so the parser can splice a multi-line load or +/// feed body together. The parser hands back either a complete `Request` or +/// `None` (more lines required). #[derive(Default)] pub struct LineParser { pending_load: Option, + pending_feed: Option, auto_reqid_counter: u64, max_load_bytes: usize, + max_feed_bytes: usize, } /// Tokens that introduce a command. If a line begins with one of these @@ -157,6 +315,15 @@ fn max_program_bytes() -> usize { .unwrap_or(65536) } +/// A framed feed is accumulated on its session thread before admission. Keep +/// that buffering bounded while allowing the 100k-row performance regime. +fn max_feed_bytes() -> usize { + std::env::var("DDIR_MAX_FEED_BYTES") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(16 * 1024 * 1024) +} + struct PendingLoad { reqid: ReqId, id_hint: String, @@ -169,10 +336,22 @@ struct PendingLoad { poisoned: Option, } +struct PendingFeed { + reqid: ReqId, + prog: String, + input: usize, + updates: Vec, + bytes: usize, + /// Malformed and oversized bodies are swallowed through `end-feed`, then + /// reported once; no partial command reaches the worker group. + poisoned: Option, +} + impl LineParser { pub fn new() -> Self { LineParser { max_load_bytes: max_program_bytes(), + max_feed_bytes: max_feed_bytes(), ..Self::default() } } @@ -182,6 +361,7 @@ impl LineParser { fn with_cap(max_load_bytes: usize) -> Self { LineParser { max_load_bytes, + max_feed_bytes: max_load_bytes, ..Self::default() } } @@ -192,6 +372,55 @@ impl LineParser { /// the result with a per-connection response sender to form a /// `Request`. pub fn feed(&mut self, line: &str) -> Option<(ReqId, Result)> { + // A framed feed fixes its target once, then accepts compact rows until + // `end-feed`. Parse and buffer the whole body on the session thread so + // the worker group sees either one complete command or no command. + if let Some(ref mut pending) = self.pending_feed { + let trimmed = line.trim(); + let mut parts = trimmed.split_whitespace(); + let first = parts.next(); + let second = parts.next(); + let third = parts.next(); + let explicit_end = first == Some(pending.reqid.as_str()) + && second == Some("end-feed") + && third.is_none(); + let bare_end = first == Some("end-feed") && second.is_none(); + if explicit_end || bare_end { + let done = self.pending_feed.take().unwrap(); + if let Some(err) = done.poisoned { + return Some((done.reqid, Err(err))); + } + return Some(( + done.reqid, + Ok(Cmd::FeedBatch { + prog: done.prog, + input: done.input, + updates: done.updates, + }), + )); + } + + pending.bytes = pending.bytes.saturating_add(line.len() + 1); + if pending.poisoned.is_none() && pending.bytes > self.max_feed_bytes { + pending.poisoned = Some(format!( + "feed: body exceeds {} bytes (DDIR_MAX_FEED_BYTES)", + self.max_feed_bytes + )); + pending.updates.clear(); + } + if pending.poisoned.is_some() || trimmed.is_empty() || trimmed.starts_with('#') { + return None; + } + match parse_input_update(trimmed) { + Ok(update) => pending.updates.push(update), + Err(error) => { + pending.poisoned = Some(format!("feed row: {error}")); + pending.updates.clear(); + } + } + return None; + } + // Inside a pending load body: every line is literal program text // until ` end-load` or ` end-load`. The id_hint // form is the friendly default when the load was auto-reqid'd @@ -232,8 +461,8 @@ impl LineParser { } let trimmed = line.trim(); - // Blank lines and `#` comments are skipped between commands (inside - // a load body every line is literal program text, handled above). + // Blank lines and `#` comments are skipped between commands. Body + // handling above decides whether they are literal or ignorable. if trimmed.is_empty() || trimmed.starts_with('#') { return None; } @@ -273,13 +502,24 @@ impl LineParser { }); None } + ParseOutcome::BeginFeed { prog, input } => { + self.pending_feed = Some(PendingFeed { + reqid, + prog, + input, + updates: Vec::new(), + bytes: 0, + poisoned: None, + }); + None + } } } - /// True if waiting for a ` end-load`. WS transport uses this - /// to forward blank-line program body content verbatim. + /// True if waiting for the terminator of a multi-line body. WS transport + /// uses this to forward blank-line body content verbatim. pub fn awaiting_body(&self) -> bool { - self.pending_load.is_some() + self.pending_load.is_some() || self.pending_feed.is_some() } } @@ -291,6 +531,10 @@ enum ParseOutcome { bindings: BTreeMap, explain: bool, }, + BeginFeed { + prog: String, + input: usize, + }, } /// A `` for `feed`: a comma-separated integer row → `Tuple`, `_` or @@ -328,6 +572,32 @@ fn parse_value(s: &str) -> Result { }) } +/// Parse one row in a framed feed. Its target and timestamp belong to the +/// enclosing command, which keeps large requests compact and atomic. +fn parse_input_update(line: &str) -> Result { + let mut args = line.split_whitespace(); + let Some(key) = args.next() else { + return Err("expected ` [val=] [diff=]`".into()); + }; + let key = parse_value(key).map_err(|error| format!("key: {error}"))?; + let mut val = Value::unit(); + let mut diff: Diff = 1; + for tok in args { + if let Some(value) = tok.strip_prefix("val=") { + val = parse_value(value).map_err(|error| format!("val: {error}"))?; + } else if let Some(value) = tok.strip_prefix("diff=") { + diff = value + .parse() + .map_err(|_| format!("diff= must be an integer, got {value:?}"))?; + } else if tok.starts_with("time=") { + return Err("time= is not allowed; a framed feed uses the current epoch".into()); + } else { + return Err(format!("unrecognized argument {tok:?}")); + } + } + Ok(InputUpdate { key, val, diff }) +} + fn parse_cmd(cmd: &str, args: &[&str]) -> ParseOutcome { match cmd { "load" => { @@ -428,6 +698,20 @@ fn parse_cmd(cmd: &str, args: &[&str]) -> ParseOutcome { }) } "feed" => { + if let [prog, input, "begin"] = args { + let input = match input.parse() { + Ok(input) => input, + Err(_) => { + return ParseOutcome::Err(format!( + "feed: must be a number, got {input:?}" + )) + } + }; + return ParseOutcome::BeginFeed { + prog: (*prog).to_string(), + input, + }; + } // Syntax: `feed [val=] [time=] [diff=]` // A ``/`` is a comma-separated integer row (`1,2` → tuple; // `_`/empty → unit) or a closed scalar term written without @@ -620,6 +904,24 @@ mod tests { } } + #[test] + fn load_is_lowered_before_worker_admission() { + let mut p = LineParser::new(); + assert!(p.feed("r0 load world begin").is_none()); + assert!(p.feed("let rows = input 0;").is_none()); + assert!(p.feed("export \"rows\" = rows;").is_none()); + let (_, parsed) = p.feed("r0 end-load").expect("load is complete"); + let prepared = prepare(parsed.expect("protocol parse succeeds")) + .expect("DDIR parsing and lowering succeed"); + match prepared { + PreparedCommand::Server(ServerCommand::Install { name, program }) => { + assert_eq!(name, "world"); + assert_eq!(program.root.exports[0].name, "rows"); + } + other => panic!("expected prepared install, got {other:?}"), + } + } + #[test] fn load_explain() { let mut p = LineParser::new(); @@ -712,6 +1014,68 @@ mod tests { assert!(got[5].1.is_err()); } + #[test] + fn framed_feed_becomes_one_typed_command() { + let mut p = LineParser::new(); + assert!(p.feed("r0 feed world 2 begin").is_none()); + assert!(p.awaiting_body()); + assert!(p.feed("# rows use the enclosing command's epoch").is_none()); + assert!(p.feed("1 val=10").is_none()); + assert!(p.feed("2 val=-20 diff=-1").is_none()); + let (reqid, parsed) = p.feed("r0 end-feed").expect("feed is complete"); + assert_eq!(reqid, "r0"); + assert!(!p.awaiting_body()); + + let prepared = prepare(parsed.expect("protocol parse succeeds")).unwrap(); + let PreparedCommand::Server(ServerCommand::FeedBatch { + prog, + input, + updates, + }) = prepared + else { + panic!("expected prepared feed batch, got {prepared:?}"); + }; + assert_eq!(prog, "world"); + assert_eq!(input, 2); + assert_eq!(updates.len(), 2); + assert_eq!(updates[0].key, Value::Tuple(vec![Value::Int(1)])); + assert_eq!(updates[0].val, Value::Tuple(vec![Value::Int(10)])); + assert_eq!(updates[0].diff, 1); + assert_eq!(updates[1].diff, -1); + } + + #[test] + fn malformed_framed_feed_is_rejected_atomically() { + let mut p = LineParser::new(); + assert!(p.feed("r0 feed world 0 begin").is_none()); + assert!(p.feed("1 val=10").is_none()); + assert!(p.feed("2 time=3").is_none()); + // Once poisoned, otherwise command-looking lines are body and cannot + // leak a partial batch or desynchronize the protocol. + assert!(p.feed("r1 list").is_none()); + let (reqid, result) = p.feed("end-feed").expect("bare terminator works"); + assert_eq!(reqid, "r0"); + assert!(result + .as_ref() + .is_err_and(|error| error.contains("uses the current epoch"))); + assert!(matches!(p.feed("r2 list"), Some((_, Ok(Cmd::List))))); + } + + #[test] + fn oversized_framed_feed_is_rejected_cleanly() { + let mut p = LineParser::with_cap(32); + assert!(p.feed("r0 feed world 0 begin").is_none()); + for key in 0..16 { + assert!(p.feed(&format!("{key} val={key}")).is_none()); + } + let (reqid, result) = p.feed("r0 end-feed").expect("feed is complete"); + assert_eq!(reqid, "r0"); + assert!(result + .as_ref() + .is_err_and(|error| error.contains("exceeds 32 bytes"))); + assert!(matches!(p.feed("r1 list"), Some((_, Ok(Cmd::List))))); + } + #[test] fn bind_cmd() { let mut p = LineParser::new(); diff --git a/interactive/server/src/loop_.rs b/interactive/server/src/loop_.rs index d8d7f2329..231848dfe 100644 --- a/interactive/server/src/loop_.rs +++ b/interactive/server/src/loop_.rs @@ -1,20 +1,24 @@ -//! Single-worker live control loop. Network sessions parse commands off-worker; -//! this thread alone owns timely and the DDIR registry. +//! Multi-worker live control loop. Worker 0 admits one FIFO command stream; +//! a single-source Timely exchange delivers one ordered record to every worker. +//! No external clock or distributed sequencing protocol participates in +//! command ordering. -use std::any::{type_name_of_val, Any}; -use std::collections::HashMap; -use std::panic::{catch_unwind, AssertUnwindSafe}; +use std::cell::RefCell; +use std::collections::{HashMap, VecDeque}; +use std::rc::Rc; use std::sync::mpsc::{Receiver, Sender, TryRecvError}; -use std::time::{Duration, Instant}; use differential_dataflow::operators::arrange::ShutdownButton; -use interactive::scope_ir::{Program, Source}; -use interactive::server::{OuterTime, Server}; +use interactive::server::{Command as ServerCommand, OuterTime, RenderBackend, Server}; +use timely::dataflow::channels::pact::Pipeline; +use timely::dataflow::operators::generic::operator::Operator; use timely::dataflow::operators::probe::Handle as ProbeHandle; -use timely::dataflow::operators::CapabilitySet; +use timely::dataflow::operators::vec::{Input as VecInput, Map}; +use timely::dataflow::operators::{CapabilitySet, Exchange, Inspect, Probe}; use timely::worker::Worker; -use crate::cmd::{Cmd, ConnectionId, DataflowRef, Request}; +use crate::cmd::{ConnectionId, PreparedCommand, Request}; +use crate::ControlEvent; struct Tail { dataflow_id: usize, @@ -25,278 +29,420 @@ struct Tail { type TailKey = (ConnectionId, String); +/// A control record that is safe to send through Timely. Response senders stay +/// in worker 0's `responses` map and are recovered with `token` after replay. +/// Parse failures are records too: replaying them preserves each session's +/// position in the order while allowing worker 0 to return the error in place. +#[derive(Clone, serde::Serialize, serde::Deserialize)] +enum Work { + Request { + token: u64, + reqid: String, + command: Result, + connection_id: ConnectionId, + /// Original logical row count retained when a batch is sharded. + logical_count: Option, + }, + SessionEnded(ConnectionId), + Shutdown, +} + +impl Work { + /// Produce exactly one ordered control record per worker. Large input + /// batches are partitioned before exchange; small commands are cloned. + fn distribute(self, peers: usize) -> Vec<(u64, Work)> { + match self { + Work::Request { + token, + reqid, + command: + Ok(PreparedCommand::Server(ServerCommand::FeedBatch { + prog, + input, + updates, + })), + connection_id, + logical_count: _, + } => { + let logical_count = updates.len(); + let mut shards = (0..peers).map(|_| Vec::new()).collect::>(); + for (position, update) in updates.into_iter().enumerate() { + shards[position % peers].push(update); + } + shards + .into_iter() + .enumerate() + .map(|(target, updates)| { + ( + target as u64, + Work::Request { + token, + reqid: reqid.clone(), + command: Ok(PreparedCommand::Server(ServerCommand::FeedBatch { + prog: prog.clone(), + input, + updates, + })), + connection_id, + logical_count: Some(logical_count), + }, + ) + }) + .collect() + } + work => (0..peers) + .map(|target| (target as u64, work.clone())) + .collect(), + } + } +} + pub fn run_worker( worker: &mut Worker, - requests: Receiver, - session_ends: Receiver, + events: Option>, + backend: RenderBackend, ) { - let diagnostics_port = std::env::var("DDIR_DIAG_PORT") - .ok() - .and_then(|port| port.parse().ok()) - .unwrap_or(51371); - let diagnostics = diagnostics::logging::register(worker, false); - let _diagnostics_server = - diagnostics::server::Server::start(diagnostics_port, diagnostics.sink); - let mut server = Server::new(); + // Logging a park wakes the diagnostics dataflow, whose scheduling logs can + // in turn wake it again. Keep idle servers genuinely idle unless an + // operator explicitly requests diagnostics. + let diagnostics_enabled = std::env::var("DDIR_DIAGNOSTICS").as_deref() == Ok("1"); + let (_diagnostics_traces, _diagnostics_server) = if diagnostics_enabled { + let diagnostics_port = std::env::var("DDIR_DIAG_PORT") + .ok() + .and_then(|port| port.parse().ok()) + .unwrap_or(51371); + let diagnostics::logging::LoggingState { traces, sink } = + diagnostics::logging::register(worker, false); + let server = if worker.index() == 0 { + Some(diagnostics::server::Server::start(diagnostics_port, sink)) + } else { + drop(sink); + None + }; + (Some(traces), server) + } else { + (None, None) + }; + + let work_queue = Rc::new(RefCell::new(VecDeque::new())); + let queue_out = work_queue.clone(); + let peers = worker.peers(); + let input = worker.dataflow::(|scope| { + let (input, stream) = scope.new_input::(); + stream + .flat_map(move |work| work.distribute(peers)) + .exchange(|(target, _work)| *target) + .map(|(_target, work)| work) + .sink(Pipeline, "QueueControl", move |(input, _frontier)| { + input.for_each(|_time, data| { + queue_out.borrow_mut().extend(data.drain(..)); + }); + }); + input + }); + // Worker 0 is the only source. A single FIFO source already supplies a + // total order, so a wall-clock sequencer would add machinery, not meaning. + let mut work_input = (worker.index() == 0).then_some(input); + + let mut server = Server::with_backend(backend); let mut tails: HashMap = HashMap::new(); - let tick_ms = std::env::var("DDIR_TICK_MS") - .ok() - .and_then(|s| s.parse().ok()) - .unwrap_or(250u64); - let interval = Duration::from_millis(tick_ms); - let mut last_tick = Instant::now(); + let mut responses: HashMap> = HashMap::new(); + let mut next_token = 0u64; + let mut intake_closed = false; let mut shutdown = false; while !shutdown { - match requests.try_recv() { - Ok(request) => dispatch(request, &mut server, &mut tails, worker, &mut shutdown), - Err(TryRecvError::Disconnected) => break, - Err(TryRecvError::Empty) => { - // Session-end notifications use a separate channel. Only - // consume them after all already-queued commands, so a final - // `stop` followed by `exit` cannot race its own cleanup. - while let Ok(connection) = session_ends.try_recv() { + let next_work = { work_queue.borrow_mut().pop_front() }; + if let Some(work) = next_work { + match work { + Work::Request { + token, + reqid, + command, + connection_id, + logical_count, + } => { + let response = if worker.index() == 0 { + responses.remove(&token) + } else { + None + }; + dispatch( + command, + connection_id, + &reqid, + response, + &mut server, + &mut tails, + worker, + &mut shutdown, + logical_count, + ); + } + Work::SessionEnded(connection) => { stop_connection(connection, &mut tails, worker); } - if tick_ms > 0 && !tails.is_empty() && last_tick.elapsed() >= interval { - tick(&mut server, &mut tails, worker); - last_tick = Instant::now(); - } else { - worker.step(); - std::thread::sleep(Duration::from_millis(5)); + Work::Shutdown => { + shutdown = true; } } + continue; } + + let mut admitted = false; + if let Some(events) = events.as_ref() { + // Bound each intake turn so a continuously busy network cannot + // starve the Timely scheduler that distributes the admitted work. + for _ in 0..1024 { + match events.try_recv() { + Ok(ControlEvent::Request(request)) => { + let Request { + reqid, + kind, + resp, + connection_id, + } = request; + let token = next_token; + next_token = next_token + .checked_add(1) + .expect("control request token overflow"); + responses.insert(token, resp); + work_input + .as_mut() + .expect("worker 0 owns command input") + .send(Work::Request { + token, + reqid, + command: kind, + connection_id, + logical_count: None, + }); + admitted = true; + } + Ok(ControlEvent::SessionEnded(connection)) => { + work_input + .as_mut() + .expect("worker 0 owns command input") + .send(Work::SessionEnded(connection)); + admitted = true; + } + Err(TryRecvError::Empty) => break, + Err(TryRecvError::Disconnected) => { + if !intake_closed { + work_input + .as_mut() + .expect("worker 0 owns command input") + .send(Work::Shutdown); + intake_closed = true; + admitted = true; + } + break; + } + } + } + + if admitted { + work_input + .as_mut() + .expect("worker 0 owns command input") + .flush(); + // Move the newly admitted batch into the distributed stream. + worker.step(); + continue; + } + } + + worker.step_or_park(None); } + + drop(work_input); for (_, tail) in tails.drain() { worker.drop_dataflow(tail.dataflow_id); } } +#[allow(clippy::too_many_arguments)] fn dispatch( - request: Request, + command: Result, + connection_id: ConnectionId, + reqid: &str, + response: Option>, server: &mut Server, tails: &mut HashMap, worker: &mut Worker, shutdown: &mut bool, + logical_count: Option, ) { - let Request { - reqid, - kind, - resp, - connection_id, - } = request; - let result = match kind { - Err(e) => Err(e), - Ok(Cmd::Load { - id_hint, - bindings, - program, - explain, - }) => { - if explain { - Err("load --explain is reserved; explanation is not implemented here yet".into()) - } else { - load(&id_hint, &bindings, &program, server, worker) - .map(|()| format!("installed {:?}", id_hint)) - } - } - Ok(Cmd::Drop { target }) => match name_ref(target) { - Err(e) => Err(e), - Ok(name) => { + let result = match command { + Err(error) => Err(error), + Ok(PreparedCommand::Server(command)) => match command { + ServerCommand::Install { name, program } => server + .install(worker, &name, &program) + .map(|()| format!("installed {:?}", name)), + ServerCommand::Drop { name } => { if tails.values().any(|tail| { server .program_info() .iter() - .find(|p| p.name == name) - .is_some_and(|p| p.exports.contains(&tail.trace)) + .find(|program| program.name == name) + .is_some_and(|program| program.exports.contains(&tail.trace)) }) { Err(format!( "cannot drop {:?}: a tail is reading one of its exports", name )) } else { + // Every worker replays this drop in command order, but + // does not physically rendezvous. The typed process + // allocator safely orphans any in-flight channel data; + // re-check this assumption for zero-copy allocators. server .drop_program(worker, &name) .map(|()| format!("dropped {:?}", name)) } } - }, - Ok(Cmd::Feed { - prog, - input, - key, - val, - time, - diff, - }) => server - .feed(&prog, input, key, val, time, diff) - .map(|()| format!("fed {:?} input {} at t={}", prog, input, server.epoch())), - Ok(Cmd::Bind { trace, prog, input }) => server - .bind(worker, &trace, &prog, input) - .map(|()| format!("bound {:?} -> {:?} input {}", trace, prog, input)), - Ok(Cmd::Unbind { trace, prog, input }) => server - .unbind(worker, &trace, &prog, input) - .map(|()| format!("unbound {:?} -> {:?} input {}", trace, prog, input)), - Ok(Cmd::List) => { - for program in server.program_info() { - send( - &resp, - &reqid, - "data", - format!( - "program name={:?} origin={} inputs={:?} imports={:?} exports={:?}", - program.name, - program.origin, - program.inputs, - program.imports, - program.exports - ), - ); - } - for (name, importers) in server.trace_info() { - send( - &resp, - &reqid, - "data", - format!("trace name={:?} importers={}", name, importers), - ); + ServerCommand::Feed { + prog, + input, + key, + val, + time, + diff, + } => { + // An external row must enter the distributed collection once. + // Its dataflow exchanges will place it on the appropriate peer. + let result = if worker.index() == 0 { + server.feed(&prog, input, key, val, time, diff) + } else { + Ok(()) + }; + result.map(|()| format!("fed {:?} input {} at t={}", prog, input, server.epoch())) } - for (source, target, input) in server.binding_info() { - send( - &resp, - &reqid, - "data", - format!("binding source={:?} target={:?} input={}", source, target, input), - ); + ServerCommand::FeedBatch { + prog, + input, + updates, + } => server.feed_batch(&prog, input, updates).map(|()| { + format!( + "fed {} rows to {:?} input {} at t={}", + logical_count.expect("distributed batch retains its logical row count"), + prog, + input, + server.epoch() + ) + }), + ServerCommand::Bind { trace, prog, input } => server + .bind(worker, &trace, &prog, input) + .map(|()| format!("bound {:?} -> {:?} input {}", trace, prog, input)), + ServerCommand::Unbind { trace, prog, input } => server + .unbind(worker, &trace, &prog, input) + .map(|()| format!("unbound {:?} -> {:?} input {}", trace, prog, input)), + ServerCommand::List => { + if let Some(response) = response.as_ref() { + for program in server.program_info() { + send( + response, + reqid, + "data", + format!( + "program name={:?} origin={} inputs={:?} imports={:?} exports={:?}", + program.name, + program.origin, + program.inputs, + program.imports, + program.exports + ), + ); + } + for (name, importers) in server.trace_info() { + send( + response, + reqid, + "data", + format!("trace name={:?} importers={}", name, importers), + ); + } + for (source, target, input) in server.binding_info() { + send( + response, + reqid, + "data", + format!( + "binding source={:?} target={:?} input={}", + source, target, input + ), + ); + } + } + Ok(format!("t={}", server.epoch())) } - Ok(format!("t={}", server.epoch())) - } - Ok(Cmd::Peek { name }) => match server.snapshot(worker, &name) { - Ok(rows) => { - for (key, val, diff) in rows { - send( - &resp, - &reqid, - "data", - format!("diff={} key={:?} val={:?}", diff, key, val), - ); + ServerCommand::Peek { trace, key } => match server.snapshot(worker, &trace) { + Ok(rows) => { + if let Some(response) = response.as_ref() { + for (row_key, val, diff) in rows { + if key.as_ref().map_or(true, |key| key == &row_key) { + send( + response, + reqid, + "data", + format!("diff={} key={:?} val={:?}", diff, row_key, val), + ); + } + } + } + Ok(format!("t={}", server.epoch())) + } + Err(error) => Err(error), + }, + ServerCommand::Tick { n } => { + for _ in 0..n { + tick(server, tails, worker); } Ok(format!("t={}", server.epoch())) } - Err(e) => Err(e), + ServerCommand::Exit => { + *shutdown = connection_id == 0; + Ok("bye".into()) + } }, - Ok(Cmd::Tail { name }) => start_tail( + Ok(PreparedCommand::Tail { name }) => start_tail( connection_id, - &reqid, + reqid, &name, - resp.clone(), + response.clone(), server, tails, worker, ) .map(|()| format!("tailing {:?} from t={}", name, server.epoch())), - Ok(Cmd::Stop { tail_reqid }) => { + Ok(PreparedCommand::Stop { tail_reqid }) => { let key = (connection_id, tail_reqid.clone()); match tails.remove(&key) { Some(tail) => { worker.drop_dataflow(tail.dataflow_id); - send(&resp, &tail_reqid, "end", String::new()); + if let Some(response) = response.as_ref() { + send(response, &tail_reqid, "end", String::new()); + } Ok(format!("stopped {}", tail_reqid)) } None => Err(format!("no tail {:?} in this session", tail_reqid)), } } - Ok(Cmd::Tick { n }) => { - for _ in 0..n { - tick(server, tails, worker); - } - Ok(format!("t={}", server.epoch())) - } - Ok(Cmd::Query { .. }) => Err( - "query is reserved for --explain dataflows and is not implemented" - .into(), - ), - Ok(Cmd::Exit) => { - *shutdown = connection_id == 0; - Ok("bye".into()) - } }; - match result { - Ok(body) => send(&resp, &reqid, "ok", body), - Err(body) => send(&resp, &reqid, "err", body), - } -} - -fn load( - name: &str, - bindings: &std::collections::BTreeMap, - source: &str, - server: &mut Server, - worker: &mut Worker, -) -> Result<(), String> { - let mut program = catch_unwind(AssertUnwindSafe(|| { - let statements = interactive::parse::pipe::parse(source); - interactive::lower::lower_tree(statements) - })) - .map_err(panic_message)?; - apply_bindings(&mut program, bindings)?; - program.optimize(); - server.install(worker, name, &program) -} - -fn apply_bindings( - program: &mut Program, - bindings: &std::collections::BTreeMap, -) -> Result<(), String> { - for (local, binding) in bindings { - let import = program - .root - .imports - .iter_mut() - .find(|import| import.name == *local) - .ok_or_else(|| format!("binding names no import {:?}", local))?; - import.from = Source::Trace(random_binding(binding)?); - } - Ok(()) -} -/// Translate the call spelling of a binding (`random(...)`) into its -/// content-addressed source name. -fn random_binding(binding: &str) -> Result { - let Some(body) = binding - .strip_prefix("random(") - .and_then(|s| s.strip_suffix(')')) - else { - return Ok(binding.to_string()); - }; - let mut values: HashMap<&str, &str> = HashMap::new(); - for field in body.split(',') { - let (key, value) = field - .trim() - .split_once('=') - .ok_or_else(|| format!("malformed random field {:?}", field))?; - values.insert(key.trim(), value.trim()); - } - let nodes = values.remove("range").ok_or("random requires range")?; - let edges = values.remove("count").ok_or("random requires count")?; - let arity = values.remove("arity").unwrap_or("2"); - let seed = values.remove("seed").unwrap_or("0"); - let churn = values.remove("churn").unwrap_or("0"); - if !values.is_empty() { - return Err(format!("unknown random fields: {:?}", values.keys())); + if let Some(response) = response.as_ref() { + match result { + Ok(body) => send(response, reqid, "ok", body), + Err(body) => send(response, reqid, "err", body), + } } - Ok(format!( - "random:nodes={},edges={},arity={},seed={},churn={}", - nodes, edges, arity, seed, churn - )) } fn start_tail( connection: ConnectionId, reqid: &str, name: &str, - response: Sender, + response: Option>, server: &Server, tails: &mut HashMap, worker: &mut Worker, @@ -314,18 +460,31 @@ fn start_tail( let shutdown = worker.dataflow::(|scope| { let (arranged, shutdown) = trace.import_core(scope.clone(), "TailImport"); arranged - .as_collection(|k, v| (k.clone(), v.clone())) + .as_collection(|key, val| (key.clone(), val.clone())) + .inner + .exchange(|_| 0u64) .inspect(move |((key, val), time, diff)| { - send( - &response, - &tag, - "data", - format!("time={} diff={} key={:?} val={:?}", time, diff, key, val), - ); + if let Some(response) = response.as_ref() { + send( + response, + &tag, + "data", + format!("time={} diff={} key={:?} val={:?}", time, diff, key, val), + ); + } }) .probe_with(&mut probe); shutdown }); + + // A tail's acknowledgement is its initial-replay boundary. Drive the new + // import through the current closed epoch before returning. The response + // channel is FIFO, so every replayed `data` message precedes dispatch's + // terminal `ok` and later commands cannot mistake replay for fresh data. + let epoch = server.epoch(); + while probe.less_than(&epoch) { + worker.step(); + } tails.insert( key, Tail { @@ -363,16 +522,6 @@ fn stop_connection( } } -fn name_ref(target: DataflowRef) -> Result { - match target { - DataflowRef::Name(name) => Ok(name), - DataflowRef::Id(id) => Err(format!( - "numeric dataflow id {} is no longer exposed; use its name", - id - )), - } -} - fn send(sender: &Sender, reqid: &str, kind: &str, body: String) { let suffix = if body.is_empty() { String::new() @@ -381,13 +530,3 @@ fn send(sender: &Sender, reqid: &str, kind: &str, body: String) { }; let _ = sender.send(format!("{} {}{}\n", reqid, kind, suffix)); } - -fn panic_message(panic: Box) -> String { - if let Some(s) = panic.downcast_ref::<&str>() { - (*s).to_string() - } else if let Some(s) = panic.downcast_ref::() { - s.clone() - } else { - format!("DDIR parser panicked ({})", type_name_of_val(&panic)) - } -} diff --git a/interactive/server/src/main.rs b/interactive/server/src/main.rs index 42fd1cca2..268807321 100644 --- a/interactive/server/src/main.rs +++ b/interactive/server/src/main.rs @@ -1,6 +1,6 @@ //! ddir_server entry point. //! -//! v0 is single-binary, single-worker, with three transports: +//! One binary hosts a configurable Timely worker group and three transports: //! - stdin/stdout (always on, process-wide) //! - raw TCP line-protocol on `DDIR_BIND` (default 127.0.0.1:7777) //! - WebSocket on `DDIR_WS_BIND` (default 127.0.0.1:7778) — one WS @@ -14,9 +14,11 @@ //! Threads: //! - main: spawns the worker, the TCP listener, and the stdin session; //! then waits for the worker to finish. -//! - worker: timely worker driving the registry + dispatch loop. -//! - per-session reader: parses lines into commands, tagging each with -//! this session's response sender, and feeds the shared cmd channel. +//! - worker 0: admits the transport FIFO to the distributed control stream. +//! - all workers: replay the same registry + dispatch operations. +//! - per-session reader: parses lines into commands, tags each with this +//! session's response sender, and wakes the worker through the shared +//! control handle. //! - per-session writer: drains the response channel onto the wire. mod cmd; @@ -26,11 +28,35 @@ mod control_loop; use std::io::{BufRead, BufReader, Write}; use std::net::{TcpListener, TcpStream}; use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::mpsc::{channel, Sender}; +use std::sync::mpsc::{channel, sync_channel, SendError, Sender}; use std::sync::{Arc, Mutex}; use std::time::Duration; -use cmd::{ConnectionId, LineParser, Request}; +use cmd::{prepare, ConnectionId, LineParser, Request}; +use interactive::server::RenderBackend; +use timely::scheduling::activate::SyncActivations; + +enum ControlEvent { + Request(Request), + SessionEnded(ConnectionId), +} + +/// Sends one event to the worker and wakes it from Timely's scheduler park. +#[derive(Clone)] +struct ControlHandle { + events: Sender, + activations: SyncActivations, +} + +impl ControlHandle { + fn send(&self, event: ControlEvent) -> Result<(), SendError> { + self.events.send(event)?; + // The empty path schedules no operator. It exists only to make + // `step_or_park` return so the outer control loop can drain `events`. + let _ = self.activations.activate(Vec::new()); + Ok(()) + } +} /// Process-wide allocator for per-session ids. Stdin uses 0; /// subsequent connections get 1, 2, 3, .... @@ -43,84 +69,102 @@ fn alloc_connection_id() -> ConnectionId { fn main() { let bind_addr = std::env::var("DDIR_BIND").unwrap_or_else(|_| "127.0.0.1:7777".to_string()); let ws_bind = std::env::var("DDIR_WS_BIND").unwrap_or_else(|_| "127.0.0.1:7778".to_string()); + let workers = std::env::var("DDIR_WORKERS") + .map(|value| { + value + .parse::() + .expect("DDIR_WORKERS must be a positive integer") + }) + .unwrap_or(1); + assert!(workers > 0, "DDIR_WORKERS must be positive"); + let backend = std::env::var("DDIR_BACKEND") + .unwrap_or_else(|_| "vec".to_string()) + .parse::() + .unwrap_or_else(|error| panic!("DDIR_BACKEND: {error}")); - let (cmd_tx, cmd_rx) = channel::(); - // Session-end notifications: a session's writer thread emits its - // connection_id when the channel closes (the client went away). - // The worker drains this and tears down any tails for that session. - let (session_end_tx, session_end_rx) = channel::(); + let (event_tx, event_rx) = channel::(); + let (activation_tx, activation_rx) = sync_channel(1); + // `execute` invokes one shared closure on every worker. Only worker 0 + // takes the control receiver; the mutex preserves that unique ownership. + let event_rx = Arc::new(Mutex::new(Some(event_rx))); + let guards = { + let event_rx = event_rx.clone(); + timely::execute(timely::Config::process(workers), move |worker| { + let events = if worker.index() == 0 { + activation_tx + .send(worker.activations().borrow().sync()) + .expect("main dropped before worker initialization"); + Some( + event_rx + .lock() + .unwrap() + .take() + .expect("control receiver taken twice"), + ) + } else { + None + }; + control_loop::run_worker(worker, events, backend); - // Spawn the timely worker on its own thread; it owns the registry. - let cmd_rx_cell = Arc::new(Mutex::new(Some(cmd_rx))); - let session_end_rx_cell = Arc::new(Mutex::new(Some(session_end_rx))); - let worker_thread = { - let cmd_rx_cell = cmd_rx_cell.clone(); - let session_end_rx_cell = session_end_rx_cell.clone(); - std::thread::spawn(move || { - timely::execute_directly(move |worker| { - let rx = cmd_rx_cell - .lock() - .unwrap() - .take() - .expect("cmd_rx taken twice"); - let sx = session_end_rx_cell - .lock() - .unwrap() - .take() - .expect("session_end_rx taken twice"); - control_loop::run_worker(worker, rx, sx); - // run_worker returns only on an operator `exit`. The worker - // still holds live dataflows (installed programs, generated - // sources, the diagnostics capture), so `execute_directly`'s - // trailing `while has_dataflows { step_or_park }` loop would - // park forever — and main's joins would never reach its own - // process::exit. Give session writers a beat to flush the - // final `ok bye`, then exit here: the documented - // "let the OS reap the parked listeners" shutdown, actually - // reached. - std::thread::sleep(Duration::from_millis(100)); - std::process::exit(0); - }); + // The live server intentionally keeps installed dataflows and + // diagnostics around. Once every worker observes shutdown, remove + // them so Timely's post-closure drain can finish normally. + for id in worker.installed_dataflows() { + worker.drop_dataflow(id); + } }) + .expect("failed to start Timely workers") + }; + let control = ControlHandle { + events: event_tx, + activations: activation_rx + .recv() + .expect("worker exited before publishing its activator"), }; // TCP listener: each accepted connection spawns its own reader and // writer pair. Failures to bind are non-fatal — the stdin transport // still works. let tcp_handle = { - let session_end_tx = session_end_tx.clone(); - spawn_listener(&bind_addr, "tcp", cmd_tx.clone(), move |stream, cmd_tx| { - run_tcp_session(stream, cmd_tx, session_end_tx.clone()).map_err(|e| e.to_string()) - }) + spawn_listener( + &bind_addr, + "tcp", + control.clone(), + move |stream, control| run_tcp_session(stream, control).map_err(|e| e.to_string()), + ) }; // WebSocket listener on a separate port. Per-connection session uses // a single-thread cooperative read/write loop so the WebSocket isn't // shared across threads (tungstenite::WebSocket isn't easily split). let ws_handle = { - let session_end_tx = session_end_tx.clone(); - spawn_listener(&ws_bind, "ws", cmd_tx.clone(), move |stream, cmd_tx| { - run_ws_session(stream, cmd_tx, session_end_tx.clone()).map_err(|e| e.to_string()) + spawn_listener(&ws_bind, "ws", control.clone(), move |stream, control| { + run_ws_session(stream, control).map_err(|e| e.to_string()) }) }; - // Stdin session: shares the same cmd channel; responses go to stdout + // Stdin session: shares the same control handle; responses go to stdout // via this session's per-connection channel. let stdin_done = std::thread::spawn({ - let cmd_tx = cmd_tx.clone(); - let session_end_tx = session_end_tx.clone(); - move || run_stdin_session(cmd_tx, session_end_tx) + let control = control.clone(); + move || run_stdin_session(control) }); - drop(cmd_tx); - drop(session_end_tx); + drop(control); let _ = stdin_done.join(); - let _ = worker_thread.join(); + let worker_failed = guards.join().into_iter().any(|result| { + if let Err(error) = result { + eprintln!("ddir_server: worker failed: {error}"); + true + } else { + false + } + }); // The listener threads are parked on accept(); without an explicit // shutdown signal, the cleanest exit is to let the OS reap them. let _ = tcp_handle; let _ = ws_handle; - std::process::exit(0); + std::process::exit(i32::from(worker_failed)); } /// Spawn a TcpListener accept loop that hands each accepted stream to a @@ -129,11 +173,11 @@ fn main() { fn spawn_listener( bind: &str, label: &'static str, - cmd_tx: Sender, + control: ControlHandle, session: F, ) -> Option> where - F: Fn(TcpStream, Sender) -> Result<(), String> + Send + Sync + 'static, + F: Fn(TcpStream, ControlHandle) -> Result<(), String> + Send + Sync + 'static, { match TcpListener::bind(bind) { Ok(listener) => { @@ -143,10 +187,10 @@ where for incoming in listener.incoming() { match incoming { Ok(stream) => { - let cmd_tx = cmd_tx.clone(); + let control = control.clone(); let session = session.clone(); std::thread::spawn(move || { - if let Err(e) = session(stream, cmd_tx) { + if let Err(e) = session(stream, control) { eprintln!("ddir_server: {} session ended: {}", label, e); } }); @@ -169,13 +213,12 @@ where /// Run one session against a `BufRead` source and a `Write` sink. Spawns /// the writer pump, then loops on lines, parses each, tags it with this /// session's `connection_id`, and forwards to the worker. On return, -/// announces the session's end via `session_end_tx` so the worker can -/// tear down any tails this session initiated. +/// announces the session's end through the same control FIFO so the workers +/// can tear down any tails this session initiated. fn run_session( input: R, output: W, - cmd_tx: Sender, - session_end_tx: Sender, + control: ControlHandle, connection_id: ConnectionId, on_exit: impl FnOnce() + Send + 'static, ) -> std::io::Result<()> { @@ -198,13 +241,14 @@ fn run_session( }; if let Some((reqid, kind)) = parser.feed(&line) { let is_exit = matches!(kind, Ok(cmd::Cmd::Exit)); + let kind = kind.and_then(prepare); let req = Request { reqid, kind, resp: resp_tx.clone(), connection_id, }; - if cmd_tx.send(req).is_err() { + if control.send(ControlEvent::Request(req)).is_err() { break; } // For stdin, `exit` terminates the whole server; for TCP, it @@ -218,26 +262,22 @@ fn run_session( // Notify the worker BEFORE joining the writer thread: any live tail // operator holds a clone of `resp_tx` inside its inspect closure, // which would otherwise keep `resp_rx` open indefinitely. The worker - // sees the session_end event, auto-stops those tails, which drops + // sees the session-ended event, auto-stops those tails, which drops // the closure (and its resp_tx clone), letting the writer pump exit. drop(resp_tx); - let _ = session_end_tx.send(connection_id); + let _ = control.send(ControlEvent::SessionEnded(connection_id)); let _ = writer_thread.join(); Ok(()) } -fn run_stdin_session(cmd_tx: Sender, session_end_tx: Sender) { +fn run_stdin_session(control: ControlHandle) { let stdin = std::io::stdin(); let stdout = std::io::stdout(); // stdin always gets connection_id 0. - let _ = run_session(stdin.lock(), stdout, cmd_tx, session_end_tx, 0, || {}); + let _ = run_session(stdin.lock(), stdout, control, 0, || {}); } -fn run_tcp_session( - stream: TcpStream, - cmd_tx: Sender, - session_end_tx: Sender, -) -> std::io::Result<()> { +fn run_tcp_session(stream: TcpStream, control: ControlHandle) -> std::io::Result<()> { let connection_id = alloc_connection_id(); let peer = stream .peer_addr() @@ -253,8 +293,7 @@ fn run_tcp_session( run_session( BufReader::new(reader_stream), writer_stream, - cmd_tx, - session_end_tx, + control, connection_id, move || { eprintln!( @@ -270,11 +309,7 @@ fn run_tcp_session( /// and writes (draining the per-connection outbound channel between /// read attempts). tungstenite's `WebSocket` isn't easily split across /// threads, so this is the cleanest shape. -fn run_ws_session( - stream: TcpStream, - cmd_tx: Sender, - session_end_tx: Sender, -) -> Result<(), tungstenite::Error> { +fn run_ws_session(stream: TcpStream, control: ControlHandle) -> Result<(), tungstenite::Error> { let connection_id = alloc_connection_id(); let peer = stream .peer_addr() @@ -291,7 +326,7 @@ fn run_ws_session( "ddir_server: ws client {} (conn={}) handshake failed: {}", peer, connection_id, e ); - let _ = session_end_tx.send(connection_id); + let _ = control.send(ControlEvent::SessionEnded(connection_id)); return Ok(()); } }; @@ -329,13 +364,14 @@ fn run_ws_session( } if let Some((reqid, kind)) = parser.feed(line) { let is_exit = matches!(kind, Ok(cmd::Cmd::Exit)); + let kind = kind.and_then(prepare); let req = Request { reqid, kind, resp: resp_tx.clone(), connection_id, }; - if cmd_tx.send(req).is_err() { + if control.send(ControlEvent::Request(req)).is_err() { should_exit = true; break; } @@ -377,6 +413,6 @@ fn run_ws_session( "ddir_server: ws client {} (conn={}) disconnected", peer, connection_id ); - let _ = session_end_tx.send(connection_id); + let _ = control.send(ControlEvent::SessionEnded(connection_id)); Ok(()) } diff --git a/interactive/server/tests/multiworker.rs b/interactive/server/tests/multiworker.rs new file mode 100644 index 000000000..539ff94ea --- /dev/null +++ b/interactive/server/tests/multiworker.rs @@ -0,0 +1,190 @@ +use std::io::{BufRead, BufReader, Write}; +use std::net::{TcpListener, TcpStream}; +use std::process::{Child, Command, Stdio}; +use std::thread; +use std::time::{Duration, Instant}; + +struct ServerProcess(Child); + +impl ServerProcess { + fn stop(mut self) { + let stdin = self.0.stdin.as_mut().expect("server stdin is piped"); + stdin.write_all(b"exit\n").unwrap(); + stdin.flush().unwrap(); + + let deadline = Instant::now() + Duration::from_secs(10); + loop { + if let Some(status) = self.0.try_wait().unwrap() { + assert!(status.success(), "server exited with {status}"); + return; + } + assert!(Instant::now() < deadline, "server did not exit"); + thread::sleep(Duration::from_millis(10)); + } + } +} + +impl Drop for ServerProcess { + fn drop(&mut self) { + if self.0.try_wait().ok().flatten().is_none() { + let _ = self.0.kill(); + let _ = self.0.wait(); + } + } +} + +fn request( + writer: &mut TcpStream, + reader: &mut BufReader, + reqid: &str, + command: &str, +) -> Vec { + request_observing(writer, reader, reqid, command, |_| {}) +} + +fn request_observing( + writer: &mut TcpStream, + reader: &mut BufReader, + reqid: &str, + command: &str, + mut observe: impl FnMut(&str), +) -> Vec { + writer.write_all(command.as_bytes()).unwrap(); + writer.flush().unwrap(); + let mut data = Vec::new(); + loop { + let mut line = String::new(); + assert_ne!( + reader.read_line(&mut line).unwrap(), + 0, + "server disconnected" + ); + let line = line.trim_end(); + observe(line); + let Some(rest) = line + .strip_prefix(reqid) + .and_then(|line| line.strip_prefix(' ')) + else { + continue; + }; + if let Some(body) = rest.strip_prefix("data ") { + data.push(body.to_string()); + } else if rest == "ok" || rest.starts_with("ok ") { + return data; + } else if rest == "err" || rest.starts_with("err ") { + panic!("request {reqid} failed: {rest}"); + } + } +} + +fn assert_backend(backend: &str) { + let listeners: Vec<_> = (0..3) + .map(|_| TcpListener::bind("127.0.0.1:0").unwrap()) + .collect(); + let ports: Vec<_> = listeners + .iter() + .map(|listener| listener.local_addr().unwrap().port()) + .collect(); + drop(listeners); + + let child = Command::new(env!("CARGO_BIN_EXE_ddir_server")) + .env("DDIR_WORKERS", "4") + .env("DDIR_BACKEND", backend) + // The retired polling/tick knob must not reintroduce wall-clock + // progress if it remains in an old deployment environment. + .env("DDIR_TICK_MS", "1") + .env("DDIR_BIND", format!("127.0.0.1:{}", ports[0])) + .env("DDIR_WS_BIND", format!("127.0.0.1:{}", ports[1])) + .env("DDIR_DIAG_PORT", ports[2].to_string()) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + let server = ServerProcess(child); + + let deadline = Instant::now() + Duration::from_secs(10); + let stream = loop { + match TcpStream::connect(("127.0.0.1", ports[0])) { + Ok(stream) => break stream, + Err(error) => { + assert!(Instant::now() < deadline, "server did not listen: {error}"); + thread::sleep(Duration::from_millis(10)); + } + } + }; + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + let mut writer = stream.try_clone().unwrap(); + let mut reader = BufReader::new(stream); + + request( + &mut writer, + &mut reader, + "r0", + "r0 load world begin\nlet rows = input 0;\nexport \"rows\" = rows;\nexport \"minimum\" = rows | min;\nr0 end-load\n", + ); + request( + &mut writer, + &mut reader, + "r1", + "r1 feed world 0 begin\n7 val=9\n7 val=3\nr1 end-feed\n", + ); + request(&mut writer, &mut reader, "r3", "r3 tick\n"); + let rows = request(&mut writer, &mut reader, "r4", "r4 peek rows\n"); + assert_eq!( + rows, + vec![ + "diff=1 key=Tuple([Int(7)]) val=Tuple([Int(3)])", + "diff=1 key=Tuple([Int(7)]) val=Tuple([Int(9)])", + ] + ); + let minimum = request(&mut writer, &mut reader, "r5", "r5 peek minimum\n"); + assert_eq!( + minimum, + vec!["diff=1 key=Tuple([Int(7)]) val=Tuple([Int(3)])"] + ); + + request(&mut writer, &mut reader, "r6", "r6 tail rows\n"); + request(&mut writer, &mut reader, "r7", "r7 feed world 0 8 val=10\n"); + reader + .get_mut() + .set_read_timeout(Some(Duration::from_millis(100))) + .unwrap(); + let mut unexpected = String::new(); + match reader.read_line(&mut unexpected) { + Err(error) + if error.kind() == std::io::ErrorKind::WouldBlock + || error.kind() == std::io::ErrorKind::TimedOut => {} + result => panic!("tail advanced without an explicit tick: {result:?} {unexpected:?}"), + } + reader + .get_mut() + .set_read_timeout(Some(Duration::from_secs(10))) + .unwrap(); + let mut tail_update = None; + request_observing(&mut writer, &mut reader, "r8", "r8 tick\n", |line| { + if let Some(body) = line.strip_prefix("r6 data ") { + tail_update = Some(body.to_string()); + } + }); + assert_eq!( + tail_update.as_deref(), + Some("time=1 diff=1 key=Tuple([Int(8)]) val=Tuple([Int(10)])") + ); + + drop(reader); + drop(writer); + server.stop(); +} + +#[test] +fn vec_commands_replay_on_four_workers_without_duplicating_input() { + assert_backend("vec"); +} + +#[test] +fn corgi_commands_replay_on_four_workers_without_duplicating_input() { + assert_backend("corgi"); +} diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs index af6e613d6..e940b4feb 100644 --- a/interactive/src/backend/corgi.rs +++ b/interactive/src/backend/corgi.rs @@ -1,7 +1,7 @@ //! The corgi rendering substrate: corgi columns are the native representation on dataflow edges, //! arrangements are chains of sorted columnar chunks (`ChunkSpine`, cursor-less), and -//! scalar logic runs columnar via `eval_graph`. Parallels the row-wise `backend::vec`, which stays -//! the correctness reference. +//! scalar logic runs columnar via `eval_graph`. The row-wise `backend::vec` remains useful for +//! comparison, but its representation choices do not define corgi's physical semantics. //! //! All `Backend` methods are corgi-native: `linear` folds a `LinearOp` chain over each container //! ([`apply_ops`], columnar fast paths with row-wise fallbacks); `arrange` ingests columns without diff --git a/interactive/src/corgi/reduce.rs b/interactive/src/corgi/reduce.rs index 0289de1e6..7d5aa3d0a 100644 --- a/interactive/src/corgi/reduce.rs +++ b/interactive/src/corgi/reduce.rs @@ -14,13 +14,9 @@ //! //! Transcode-free: the real keys/values never leave corgi columns. Ids are resolved to rows by //! integer index (`key_index`/`val_index` → offsets into the concatenated `key_blocks`/`val_blocks` -//! pools), not by carrying `DValue`s. Min/Collect's ordering is corgi's own — one `sort_blocks` per -//! retire orders every bracket's candidates (Min = each block's first, Collect = each block's sorted -//! run, expanded by diff). This uses corgi's STRUCTURAL order, which equals DDIR `Ord` for the -//! non-negative scalar/tuple values these reductions see (all 6 canonical programs); it diverges only -//! for negative ints (corgi's leaf compare is unsigned) and list-valued compares (corgi lists order -//! length-first) — neither arises here. A signed/​list-general order would need a corgi order fix -//! (offset-binary leaf or lex-first lists), not a change here. +//! pools), not by carrying `DValue`s. Min/Collect use corgi's one-pass segmented structural sort. +//! DDIR integers are signed, so Min builds an order-only columnar view with each integer leaf's +//! sign bit swizzled before sorting; the winning row is still gathered from the original columns. //! //! The changed-key restriction is honored by presenting only the changed keys: novel batches are //! read whole (delta-sized), the accumulated history is scanned and filtered to the changed hashes @@ -39,7 +35,7 @@ use differential_dataflow::operators::int_proxy::ProxyBridge; use differential_dataflow::operators::int_proxy::reduce::{ProxyReduceBackend, ReduceInstance, ReduceWindow}; use corgi::arrange::{compare_at, find_ranges, gather, gather_lanes, sort_blocks}; -use corgi::{Bounds, Value as CValue}; +use corgi::{ArithOp, Bounds, NumOp, OpLike, Value as CValue}; use crate::corgi::col_times::ColTime; use crate::corgi::chunk::{columns_to_batch, key_ids, key_lane, CorgiChunk}; @@ -48,6 +44,34 @@ use crate::parse::Reducer; type CBatch = Rc>>; +/// Build a sortable view whose integer leaves have signed `i64` order. +/// +/// DDIR's only scalar is `Int`, transcoded into a Corgi primitive as its raw +/// bits. Corgi's radix sort is unsigned, so XORing each payload leaf's sign bit +/// turns signed order into unsigned order. Sum discriminants remain untouched; +/// only their payload lanes recurse. This consumes freshly gathered candidate +/// columns, allowing Corgi to swizzle their buffers in place when unshared. +fn signed_order_view(value: CValue) -> CValue { + match value { + value @ CValue::Prim(_) => NumOp::from(ArithOp::ToSigned).eval(value), + CValue::Prod(fields) => { + CValue::Prod(fields.into_iter().map(signed_order_view).collect()) + } + CValue::Sum(tags, within, variants) => CValue::Sum( + tags, + within, + variants + .into_iter() + .map(|variant| variant.map(signed_order_view)) + .collect(), + ), + CValue::List(bounds, values) => { + CValue::List(bounds, Box::new(signed_order_view(*values))) + } + CValue::Unit(len) => CValue::Unit(len), + } +} + /// An identity `Hasher` for the id-index maps: their keys are already well-distributed 64-bit /// content hashes (`hash_rows`), so passing the id straight through avoids re-hashing it (siphash /// on `register_keys`/lookups was ~7% of the reduce in profiling). Only `write_u64` is used. @@ -434,12 +458,9 @@ where self.register_vals(col, &out_ids); } Reducer::Min => { - // The DDIR `min` over the values with NON-ZERO net, in corgi's structural order - // (== DDIR `Ord` for the non-negative scalar/tuple values these reductions see; see - // module doc). The sign does not select candidates: `backend::vec` takes `min` over - // every value DD presents, and DD presents every non-zero accumulation. Filtering to - // `> 0` here both dropped all-negative keys and could pick a different minimum when a - // bracket mixed signs. + // The structural minimum over values with NON-ZERO net. The sign does not select + // candidates: DD presents every non-zero accumulation. Filtering to `> 0` here both + // drops all-negative keys and can pick a different minimum when a bracket mixes signs. // Gather all candidates across brackets into one column, segment by // bracket, and one corgi `sort_blocks` gives every bracket's argmin at once // (`perm[block_start]`). The winning ROW is taken columnar and reuses its input value id. @@ -467,14 +488,14 @@ where return (Vec::new(), out_ends); } let cand_col = gather(&self.in_vals, &cand_reps); - let (perm, _) = sort_blocks(&labels, &cand_col); + let (perm, _) = sort_blocks(&labels, &signed_order_view(cand_col)); let min_reps: Vec = block_starts.iter().map(|&lo| cand_reps[perm[lo]]).collect(); let col = gather(&self.in_vals, &min_reps); out_ids = ids(&col); self.register_vals(col, &out_ids); } Reducer::Collect => { - // One row per bracket: the values sorted in corgi structural order (== DDIR `Ord` here), + // One row per bracket: the values sorted in corgi structural order, // each repeated by its diff, as a `List`. One `sort_blocks` orders every bracket's // entries at once; element rows are then taken columnar. Every bracket emits (empty // list if all diffs ≤ 0), matching the row reducer. diff --git a/interactive/src/ir.rs b/interactive/src/ir.rs index ce763357d..3ff615c1d 100644 --- a/interactive/src/ir.rs +++ b/interactive/src/ir.rs @@ -13,8 +13,10 @@ pub type Time = timely::order::Product Result { + match name { + "vec" => Ok(Self::Vec), + "corgi" => Ok(Self::Corgi), + other => Err(format!("backend must be vec or corgi, got {other:?}")), + } + } +} + /// A registered, shareable arrangement: the published form of an `export`, /// arranged by key at the host time so any later install can `import` it. pub type ServerTrace = TraceAgent>; @@ -184,9 +220,9 @@ fn canonical_source_name(name: &str) -> String { /// A unit of server work, already parsed/lowered/validated on the intake side. /// -/// Serializable so it can be circulated to every worker through a timely -/// `Sequencer`; the workers execute it without any further parsing. -#[derive(Clone, serde::Serialize, serde::Deserialize)] +/// Serializable so worker 0 can circulate it to every worker; the workers +/// execute it without any further parsing. +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] pub enum Command { /// Install `program` under `name`. Install { name: String, program: st::Program }, @@ -200,8 +236,14 @@ pub enum Command { time: Option, diff: Diff, }, - /// Close the current epoch and run to quiescence. - Tick, + /// Stage several rows into one positional input at the current epoch. + FeedBatch { + prog: String, + input: usize, + updates: Vec, + }, + /// Close `n` epochs, running to quiescence after each one. + Tick { n: u64 }, /// Drop the named program. Drop { name: String }, /// Snapshot a registered trace (optionally one key) and print it (worker 0). @@ -220,8 +262,6 @@ pub enum Command { }, /// Print the registry (worker 0). List, - /// Print the command help (worker 0). - Help, /// Stop the server. Exit, } @@ -305,17 +345,25 @@ pub struct Server { bindings: Vec, /// The current open epoch; inputs sit here until `tick` closes it. epoch: OuterTime, + /// Rendering substrate for subsequently installed programs. + backend: RenderBackend, } impl Server { /// A fresh server with the host clock at epoch 0. pub fn new() -> Self { + Self::with_backend(RenderBackend::Vec) + } + + /// A fresh server using `backend` for installed DDIR programs. + pub fn with_backend(backend: RenderBackend) -> Self { Server { traces: HashMap::new(), programs: HashMap::new(), importers: HashMap::new(), bindings: Vec::new(), epoch: 0, + backend, } } @@ -428,6 +476,7 @@ impl Server { let probe = ProbeHandle::new(); let root = &prog.root; let traces = &mut self.traces; + let backend = self.backend; // The id this dataflow will get; captured so `drop` can remove it. let dataflow_id = worker.next_dataflow_index(); @@ -465,7 +514,12 @@ impl Server { .iterative::, _, _>(|inner| { let entered: Vec<_> = outer_cols.iter().map(|c| c.clone().enter(inner)).collect(); - let exports = render_tree(root, inner.clone(), 0, entered); + let exports = match backend { + RenderBackend::Vec => render_tree(root, inner.clone(), 0, entered), + RenderBackend::Corgi => { + render_tree_rows(root, inner.clone(), 0, entered) + } + }; exports .into_iter() .map(|c| c.leave(outer)) @@ -611,16 +665,50 @@ impl Server { diff: Diff, ) -> Result<(), String> { let t = time.unwrap_or(self.epoch); - if t < self.epoch { + self.validate_feed(prog, input, t)?; + self.apply_feed(prog, input, key, val, t, diff); + Ok(()) + } + + /// Atomically stage several rows into one input at the current open epoch. + /// + /// The target is validated before its handle changes. The batch does not + /// advance time; all rows become visible together when a later `tick` + /// closes this epoch. + pub fn feed_batch( + &mut self, + prog: &str, + input: usize, + updates: Vec, + ) -> Result<(), String> { + let time = self.epoch; + self.validate_feed(prog, input, time)?; + for update in updates { + self.apply_feed( + prog, + input, + update.key, + update.val, + time, + update.diff, + ); + } + Ok(()) + } + + /// Check everything about an input target that can fail without changing + /// its handle. Both singular and batched feeds validate before applying. + fn validate_feed(&self, prog: &str, input: usize, time: OuterTime) -> Result<(), String> { + if time < self.epoch { return Err(format!( "cannot feed at time {} < current epoch {}", - t, self.epoch + time, self.epoch )); } let prog = canonical_source_name(prog); let installed = self .programs - .get_mut(&prog) + .get(&prog) .ok_or_else(|| format!("no program {:?}", prog))?; if installed.origin != Origin::Program { let kind = if installed.origin == Origin::Clock { @@ -633,12 +721,30 @@ impl Server { prog, kind )); } - let handle = installed + if !installed.inputs.contains_key(&input) { + return Err(format!("program {:?} has no input {}", prog, input)); + } + Ok(()) + } + + /// Apply a feed whose target was already validated. + fn apply_feed( + &mut self, + prog: &str, + input: usize, + key: Value, + val: Value, + time: OuterTime, + diff: Diff, + ) { + let prog = canonical_source_name(prog); + self.programs + .get_mut(&prog) + .expect("feed target was prevalidated") .inputs .get_mut(&input) - .ok_or_else(|| format!("program {:?} has no input {}", prog, input))?; - handle.update_at((key, val), t, diff); - Ok(()) + .expect("feed input was prevalidated") + .update_at((key, val), time, diff); } /// Bind trace `trace` to positional `input` of program `prog`: from now diff --git a/interactive/tests/corgi_backend.rs b/interactive/tests/corgi_backend.rs index 7cc0c2a96..b3ca27b78 100644 --- a/interactive/tests/corgi_backend.rs +++ b/interactive/tests/corgi_backend.rs @@ -47,6 +47,13 @@ fn inputs_for(prog: &str) -> Vec> { rows(&[&[1, 1, 10], &[1, 2, 20], &[2, 1, 30], &[2, 1, 31], &[9, 9, 90]]), rows(&[&[1, 1, 5], &[2, 1, 6], &[3, 3, 7]]), ], + "signed_min" => vec![rows(&[ + &[1, 0], + &[1, -1], + &[1, -3], + &[2, 5], + &[2, -2], + ])], // tour: edges (with a cycle and a chord) + roots. "tour" => vec![ rows(&[&[1, 2], &[2, 3], &[3, 1], &[3, 4], &[5, 2]]), @@ -115,3 +122,4 @@ fn serializing(n: usize) -> timely::Config { #[test] fn case_ops() { assert_backends_agree("case_ops"); } #[test] fn tour() { assert_backends_agree("tour"); } #[test] fn pair_keys() { assert_backends_agree("pair_keys"); } +#[test] fn signed_min() { assert_backends_agree("signed_min"); } diff --git a/interactive/tests/explain.rs b/interactive/tests/explain.rs index 3c2218aab..cf55f1ade 100644 --- a/interactive/tests/explain.rs +++ b/interactive/tests/explain.rs @@ -1,8 +1,8 @@ //! End-to-end semantic tests for the explanation rewrite, built on //! `backend::vec::evaluate` (explicit inputs in, every export out). The -//! sufficiency properties are checked against `vec`, the correctness reference; -//! a final section cross-checks that the rewritten programs render identically -//! on the corgi backend. +//! sufficiency properties use that row execution to evaluate the rewrite; a +//! final section cross-checks the behavior shared by the row and corgi +//! implementations for these programs. //! //! The central property is *sufficiency*: for a query against a program's //! output, the original inputs *restricted to* the demand-sets the rewritten @@ -668,4 +668,3 @@ fn corgi_agrees_on_two_query_explanation() { assert_eq!(qs.len(), 2, "expected at least two scc edges to query"); assert_explained_backends_agree(SCC_ROW, SCC_SHAPES, &inputs, &qs); } - diff --git a/interactive/tests/programs/signed_min.ddp b/interactive/tests/programs/signed_min.ddp new file mode 100644 index 000000000..92b13b8fd --- /dev/null +++ b/interactive/tests/programs/signed_min.ddp @@ -0,0 +1,5 @@ +-- DDIR integers are signed even though Corgi stores their raw bits in u64 +-- leaves. Min must sign-swizzle those leaves for sorting without decoding rows. +let pairs = input 0 | key($0[0] ; $0[1]); + +export "minimum" = pairs | min; diff --git a/interactive/tests/server_batch.rs b/interactive/tests/server_batch.rs new file mode 100644 index 000000000..588672ac1 --- /dev/null +++ b/interactive/tests/server_batch.rs @@ -0,0 +1,44 @@ +use interactive::ir::Value; +use interactive::server::{InputUpdate, Server}; +use interactive::{lower, parse}; + +fn tup(fields: &[i64]) -> Value { + Value::Tuple(fields.iter().copied().map(Value::Int).collect()) +} + +#[test] +fn feed_batch_validates_before_staging_one_epoch() { + timely::execute_directly(|worker| { + let mut program = lower::lower_tree(parse::pipe::parse( + "let rows = input 0; export \"rows\" = rows;", + )); + program.optimize(); + + let mut server = Server::new(); + server.install(worker, "world", &program).unwrap(); + let updates = vec![ + InputUpdate { + key: tup(&[1]), + val: tup(&[10]), + diff: 1, + }, + InputUpdate { + key: tup(&[2]), + val: tup(&[20]), + diff: 1, + }, + ]; + + assert!(server.feed_batch("world", 1, updates.clone()).is_err()); + server.tick(worker); + assert!(server.snapshot(worker, "rows").unwrap().is_empty()); + + server.feed_batch("world", 0, updates).unwrap(); + assert!(server.snapshot(worker, "rows").unwrap().is_empty()); + server.tick(worker); + assert_eq!( + server.snapshot(worker, "rows").unwrap(), + vec![(tup(&[1]), tup(&[10]), 1), (tup(&[2]), tup(&[20]), 1),] + ); + }); +}