From 66f94b763afd2d28ee870b36a8813d532f0da4d4 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 1 Sep 2026 16:52:55 -0400 Subject: [PATCH 1/3] ddir-server: make the control plane parallel and timer-free Replay one worker-0 FIFO through Timely, keep response handles local, and use logical progress for tick, peek, and tail completion. Remove periodic sleeps and physical worker barriers so idle workers park in the scheduler and command completion follows dataflow progress. Keep diagnostics opt-in because its logging dataflow is itself scheduler-active, and drain received control containers into the local queue without cloning their records. --- interactive/examples/ddir_server.rs | 14 +- interactive/server/Cargo.toml | 1 + interactive/server/README.md | 41 +- interactive/server/demo/two_sessions.py | 1 - interactive/server/src/cmd.rs | 167 +++++++- interactive/server/src/loop_.rs | 542 +++++++++++++----------- interactive/server/src/main.rs | 201 +++++---- interactive/server/tests/multiworker.rs | 164 +++++++ interactive/src/server.rs | 17 +- 9 files changed, 791 insertions(+), 357 deletions(-) create mode 100644 interactive/server/tests/multiworker.rs diff --git a/interactive/examples/ddir_server.rs b/interactive/examples/ddir_server.rs index 3ff3e7a7e..6a679880b 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,10 @@ fn dispatch(cmd: &Command, server: &mut Server, worker: &mut Worker) -> bool { } } } - Command::Tick => { - server.tick(worker); + 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 +208,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 +297,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..b1254f759 100644 --- a/interactive/server/README.md +++ b/interactive/server/README.md @@ -1,16 +1,29 @@ # 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). + +Worker 0 admits one FIFO control stream and broadcasts it to every worker. +Because that 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 the broadcast 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 +40,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 +49,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` 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..67394fea3 100644 --- a/interactive/server/src/cmd.rs +++ b/interactive/server/src/cmd.rs @@ -11,11 +11,13 @@ //! ` load begin` opens; subsequent lines are //! literal program text terminated by ` end-load`. -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, OuterTime}; pub type ReqId = String; @@ -115,11 +117,150 @@ 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::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. @@ -620,6 +761,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(); diff --git a/interactive/server/src/loop_.rs b/interactive/server/src/loop_.rs index d8d7f2329..23a068a37 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 broadcast delivers that order 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, 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::{Broadcast, Input as VecInput}; +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,347 @@ 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, + }, + SessionEnded(ConnectionId), + Shutdown, +} + pub fn run_worker( worker: &mut Worker, - requests: Receiver, - session_ends: Receiver, + events: Option>, ) { - 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); + // 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 input = worker.dataflow::(|scope| { + let (input, stream) = scope.new_input::(); + stream + .broadcast() + .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::new(); 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, + } => { + let response = if worker.index() == 0 { + responses.remove(&token) + } else { + None + }; + dispatch( + command, + connection_id, + &reqid, + response, + &mut server, + &mut tails, + worker, + &mut shutdown, + ); + } + 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, + }); + 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, ) { - 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::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 +387,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 +449,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 +457,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..0f66d3baa 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,34 @@ 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 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 +68,98 @@ 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 (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); - // 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 +168,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 +182,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 +208,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 +236,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 +257,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 +288,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 +304,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 +321,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 +359,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 +408,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..1492e4699 --- /dev/null +++ b/interactive/server/tests/multiworker.rs @@ -0,0 +1,164 @@ +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}"); + } + } +} + +#[test] +fn commands_replay_on_four_workers_without_duplicating_input() { + 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") + // 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;\nr0 end-load\n", + ); + request(&mut writer, &mut reader, "r1", "r1 feed world 0 7 val=9\n"); + request(&mut writer, &mut reader, "r2", "r2 tick\n"); + let rows = request(&mut writer, &mut reader, "r3", "r3 peek rows\n"); + assert_eq!(rows, vec!["diff=1 key=Tuple([Int(7)]) val=Tuple([Int(9)])"]); + + 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(); +} diff --git a/interactive/src/server.rs b/interactive/src/server.rs index 84e086102..367bcde0d 100644 --- a/interactive/src/server.rs +++ b/interactive/src/server.rs @@ -10,8 +10,9 @@ //! The server executes a [`Command`] — already parsed, lowered, and validated. //! Programs are parsed *off the worker threads* (on the intake side) and shipped //! here as `scope_ir::Program`s; a malformed program is rejected before it ever -//! reaches a worker, so bad input can't panic the computation. `Command` is -//! serializable precisely so it can ride a timely `Sequencer` to every worker. +//! reaches a worker, so bad input can't panic the computation. [`Command`] is +//! serializable so worker 0 can broadcast one ordered command stream to the +//! whole worker group. //! //! # The two binding points //! @@ -184,9 +185,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 +201,8 @@ pub enum Command { time: Option, diff: Diff, }, - /// Close the current epoch and run to quiescence. - Tick, + /// 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 +221,6 @@ pub enum Command { }, /// Print the registry (worker 0). List, - /// Print the command help (worker 0). - Help, /// Stop the server. Exit, } From 1ba7467247582cea1988431d768cffcaf2ab3b95 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 1 Sep 2026 16:54:24 -0400 Subject: [PATCH 2/3] ddir-server: make Corgi a live rendering backend Select one renderer for the server while retaining the current row-speaking registry as an explicit transition boundary. Corgi remains columnar within each installed program. Preserve DDIR's signed integer semantics for min by sign-swizzling Corgi's candidate leaves in an order-only columnar view before segmented sorting. The selected rows remain in their original columns, with no Value conversion. --- interactive/server/README.md | 10 ++++- interactive/server/src/loop_.rs | 5 ++- interactive/server/src/main.rs | 7 ++- interactive/server/tests/multiworker.rs | 34 ++++++++++++--- interactive/src/backend/corgi.rs | 4 +- interactive/src/corgi/reduce.rs | 53 ++++++++++++++++------- interactive/src/ir.rs | 6 ++- interactive/src/server.rs | 41 +++++++++++++++++- interactive/tests/corgi_backend.rs | 8 ++++ interactive/tests/explain.rs | 7 ++- interactive/tests/programs/signed_min.ddp | 5 +++ 11 files changed, 145 insertions(+), 35 deletions(-) create mode 100644 interactive/tests/programs/signed_min.ddp diff --git a/interactive/server/README.md b/interactive/server/README.md index b1254f759..8d31ca413 100644 --- a/interactive/server/README.md +++ b/interactive/server/README.md @@ -11,7 +11,15 @@ 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). +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 broadcasts it to every worker. Because that one source already defines a total order, command coordination diff --git a/interactive/server/src/loop_.rs b/interactive/server/src/loop_.rs index 23a068a37..9fa70f6aa 100644 --- a/interactive/server/src/loop_.rs +++ b/interactive/server/src/loop_.rs @@ -9,7 +9,7 @@ use std::rc::Rc; use std::sync::mpsc::{Receiver, Sender, TryRecvError}; use differential_dataflow::operators::arrange::ShutdownButton; -use interactive::server::{Command as ServerCommand, 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; @@ -48,6 +48,7 @@ enum Work { pub fn run_worker( worker: &mut Worker, events: Option>, + backend: RenderBackend, ) { // Logging a park wakes the diagnostics dataflow, whose scheduling logs can // in turn wake it again. Keep idle servers genuinely idle unless an @@ -88,7 +89,7 @@ pub fn run_worker( // 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::new(); + let mut server = Server::with_backend(backend); let mut tails: HashMap = HashMap::new(); let mut responses: HashMap> = HashMap::new(); let mut next_token = 0u64; diff --git a/interactive/server/src/main.rs b/interactive/server/src/main.rs index 0f66d3baa..268807321 100644 --- a/interactive/server/src/main.rs +++ b/interactive/server/src/main.rs @@ -33,6 +33,7 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use cmd::{prepare, ConnectionId, LineParser, Request}; +use interactive::server::RenderBackend; use timely::scheduling::activate::SyncActivations; enum ControlEvent { @@ -76,6 +77,10 @@ fn main() { }) .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 (event_tx, event_rx) = channel::(); let (activation_tx, activation_rx) = sync_channel(1); @@ -99,7 +104,7 @@ fn main() { } else { None }; - control_loop::run_worker(worker, events); + control_loop::run_worker(worker, events, backend); // The live server intentionally keeps installed dataflows and // diagnostics around. Once every worker observes shutdown, remove diff --git a/interactive/server/tests/multiworker.rs b/interactive/server/tests/multiworker.rs index 1492e4699..50b0b9d1e 100644 --- a/interactive/server/tests/multiworker.rs +++ b/interactive/server/tests/multiworker.rs @@ -77,8 +77,7 @@ fn request_observing( } } -#[test] -fn commands_replay_on_four_workers_without_duplicating_input() { +fn assert_backend(backend: &str) { let listeners: Vec<_> = (0..3) .map(|_| TcpListener::bind("127.0.0.1:0").unwrap()) .collect(); @@ -90,6 +89,7 @@ fn commands_replay_on_four_workers_without_duplicating_input() { 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") @@ -123,12 +123,24 @@ fn commands_replay_on_four_workers_without_duplicating_input() { &mut writer, &mut reader, "r0", - "r0 load world begin\nlet rows = input 0;\nexport \"rows\" = rows;\nr0 end-load\n", + "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 7 val=9\n"); - request(&mut writer, &mut reader, "r2", "r2 tick\n"); - let rows = request(&mut writer, &mut reader, "r3", "r3 peek rows\n"); - assert_eq!(rows, vec!["diff=1 key=Tuple([Int(7)]) val=Tuple([Int(9)])"]); + request(&mut writer, &mut reader, "r2", "r2 feed world 0 7 val=3\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"); @@ -162,3 +174,13 @@ fn commands_replay_on_four_workers_without_duplicating_input() { 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>; @@ -304,17 +329,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, } } @@ -427,6 +460,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(); @@ -464,7 +498,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)) 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; From 77e0018387556b772dc2817883184b42ae27768f Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Tue, 1 Sep 2026 16:55:51 -0400 Subject: [PATCH 3/3] ddir-server: admit input batches atomically Add a typed FeedBatch command and a bounded begin/end text protocol so one request can stage many rows for one input at the current epoch. Parse and validate the complete body before it enters the ordered command stream, partition it by body position before Timely exchange, and apply each worker's shard without advancing time. Treat this row-speaking form as a compatibility ingress rather than the intended high-volume data path. Drain exchanged command containers without cloning their shards, and document that admission succeeds before a later tick establishes visibility and completion. --- interactive/examples/ddir_server.rs | 7 + interactive/server/README.md | 60 +++++-- interactive/server/src/cmd.rs | 229 ++++++++++++++++++++++-- interactive/server/src/loop_.rs | 78 +++++++- interactive/server/tests/multiworker.rs | 8 +- interactive/src/server.rs | 82 ++++++++- interactive/tests/server_batch.rs | 44 +++++ 7 files changed, 468 insertions(+), 40 deletions(-) create mode 100644 interactive/tests/server_batch.rs diff --git a/interactive/examples/ddir_server.rs b/interactive/examples/ddir_server.rs index 6a679880b..67bf91d89 100644 --- a/interactive/examples/ddir_server.rs +++ b/interactive/examples/ddir_server.rs @@ -180,6 +180,13 @@ fn dispatch(cmd: &Command, server: &mut Server, worker: &mut Worker) -> bool { } } } + 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); diff --git a/interactive/server/README.md b/interactive/server/README.md index 8d31ca413..6d552e850 100644 --- a/interactive/server/README.md +++ b/interactive/server/README.md @@ -21,17 +21,19 @@ 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 broadcasts it to every worker. -Because that 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. +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 the broadcast 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. +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 @@ -71,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 @@ -100,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/src/cmd.rs b/interactive/server/src/cmd.rs index 67394fea3..df8a06ec6 100644 --- a/interactive/server/src/cmd.rs +++ b/interactive/server/src/cmd.rs @@ -7,9 +7,9 @@ //! - `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::any::{type_name_of_val, Any}; use std::collections::{BTreeMap, HashMap}; @@ -17,7 +17,7 @@ use std::panic::{catch_unwind, AssertUnwindSafe}; use interactive::ir::{eval, Diff, Value}; use interactive::scope_ir::{Program, Source}; -use interactive::server::{Command as ServerCommand, OuterTime}; +use interactive::server::{Command as ServerCommand, InputUpdate, OuterTime}; pub type ReqId = String; @@ -71,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 { @@ -188,6 +194,15 @@ pub fn prepare(command: Cmd) -> Result { 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 { .. } => { @@ -270,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 @@ -298,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, @@ -310,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() } } @@ -323,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() } } @@ -333,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 @@ -373,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; } @@ -414,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() } } @@ -432,6 +531,10 @@ enum ParseOutcome { bindings: BTreeMap, explain: bool, }, + BeginFeed { + prog: String, + input: usize, + }, } /// A `` for `feed`: a comma-separated integer row → `Tuple`, `_` or @@ -469,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" => { @@ -569,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 @@ -871,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 9fa70f6aa..231848dfe 100644 --- a/interactive/server/src/loop_.rs +++ b/interactive/server/src/loop_.rs @@ -1,5 +1,5 @@ //! Multi-worker live control loop. Worker 0 admits one FIFO command stream; -//! a single-source Timely broadcast delivers that order to every worker. +//! a single-source Timely exchange delivers one ordered record to every worker. //! No external clock or distributed sequencing protocol participates in //! command ordering. @@ -13,7 +13,7 @@ use interactive::server::{Command as ServerCommand, OuterTime, RenderBackend, Se 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::vec::{Broadcast, Input as VecInput}; +use timely::dataflow::operators::vec::{Input as VecInput, Map}; use timely::dataflow::operators::{CapabilitySet, Exchange, Inspect, Probe}; use timely::worker::Worker; @@ -40,11 +40,63 @@ enum Work { 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, events: Option>, @@ -74,10 +126,13 @@ pub fn run_worker( 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 - .broadcast() + .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(..)); @@ -105,6 +160,7 @@ pub fn run_worker( reqid, command, connection_id, + logical_count, } => { let response = if worker.index() == 0 { responses.remove(&token) @@ -120,6 +176,7 @@ pub fn run_worker( &mut tails, worker, &mut shutdown, + logical_count, ); } Work::SessionEnded(connection) => { @@ -158,6 +215,7 @@ pub fn run_worker( reqid, command: kind, connection_id, + logical_count: None, }); admitted = true; } @@ -213,6 +271,7 @@ fn dispatch( tails: &mut HashMap, worker: &mut Worker, shutdown: &mut bool, + logical_count: Option, ) { let result = match command { Err(error) => Err(error), @@ -259,6 +318,19 @@ fn dispatch( }; result.map(|()| format!("fed {:?} input {} at t={}", prog, input, server.epoch())) } + 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)), diff --git a/interactive/server/tests/multiworker.rs b/interactive/server/tests/multiworker.rs index 50b0b9d1e..539ff94ea 100644 --- a/interactive/server/tests/multiworker.rs +++ b/interactive/server/tests/multiworker.rs @@ -125,8 +125,12 @@ fn assert_backend(backend: &str) { "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 7 val=9\n"); - request(&mut writer, &mut reader, "r2", "r2 feed world 0 7 val=3\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!( diff --git a/interactive/src/server.rs b/interactive/src/server.rs index 294bc8428..4ccdb9b98 100644 --- a/interactive/src/server.rs +++ b/interactive/src/server.rs @@ -75,6 +75,16 @@ pub enum RenderBackend { Corgi, } +/// One row in the server's compatibility input path. The enclosing command +/// supplies the program, positional input, and current server epoch once for +/// the whole batch. Native bulk ingress need not materialize this row form. +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] +pub struct InputUpdate { + pub key: Value, + pub val: Value, + pub diff: Diff, +} + impl std::str::FromStr for RenderBackend { type Err = String; @@ -226,6 +236,12 @@ pub enum Command { time: Option, diff: Diff, }, + /// 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. @@ -649,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 { @@ -671,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/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),] + ); + }); +}