diff --git a/context/code/zset-ffi.md b/context/code/zset-ffi.md index 64ed53cd..37fc794e 100644 --- a/context/code/zset-ffi.md +++ b/context/code/zset-ffi.md @@ -59,7 +59,7 @@ link, so they call the `extern "C"` functions as plain functions. its return value (`flat_map`/`index_by` non-zero, `keep` and `expired` negative), and the step then fails with `ZS_ERR_CALLBACK`. A callback must return normally. A Rust panic across `extern "C"` aborts, and a Lisp `handler-case` outside `zs_circuit_step` that unwinds - through the Rust frames skips destructors and leaves a `RefCell` inside `zset` borrowed. + through the Rust frames skips destructors and leaves the circuit half way through a step. The ABI cannot stop the unwinding, so it detects it: `ZsCircuit::in_step` is set for the duration of `step` and `rebuild_step`, and cleared only when they return. Every entry point first reads it, so a circuit whose frames were skipped answers `ZS_ERR_POISONED` @@ -73,9 +73,16 @@ link, so they call the `extern "C"` functions as plain functions. return the failure value (`tests/smoke.lisp` does). `index_by` needs exactly one emitted key and a keyed stream valid pairs, else `ZS_ERR_BAD_VALUE`; these faults are recorded in a shared cell rather than panicking. -10. **Thread confinement.** A circuit and every rebuild job made from it share an `Rc` - (`Input`), so they are one group: keep the group on one thread at a time. Z-sets are - separate handles. The error message is per thread. +10. **Threads** (`zset.md`, "Threads"). A `ZsCircuit` may be moved between threads and is used + by one at a time; it is `Send` and not `Sync`, with no lock. Callbacks run on whichever + thread calls `step` or `rebuild_step`, with that thread's thread-local state (Lisp dynamic + bindings, `errno`, `zs_last_error`), so the caller's `ctx` must be usable from every + thread that may step (the wrapper `Ctx` is `unsafe impl Send` on that contract). The + first-fault cell is the one lock, and it is the FFI's own. A `ZsRebuild` no longer shares + a reference count with its circuit (the handles are ids), so it is a separate object, and + a `zs_rebuild_step` call uses both. The caller still keeps each to one thread at a time. + Z-sets are separate handles. The error message is per thread + (`tests/threads_abi.rs`: callbacks ran on the thread that stepped). 11. **The header is hand-written** (cbindgen is not a dependency of this repo). `tests/header.rs` keeps it honest. It maps each Rust parameter and return type of every exported function (including the `unary_op!` expansions) to its C spelling, through a table that panics on diff --git a/context/code/zset.md b/context/code/zset.md index ae025393..62d66441 100644 --- a/context/code/zset.md +++ b/context/code/zset.md @@ -128,13 +128,14 @@ Clause 1's deferred scope, built without ADR 008. The ADR is not edited; this is Decisions, where the ADR left a choice: -1. **Wiring is apart from logic.** `circuit.rs` is the only file with `Rc`, `RefCell` and +1. **Wiring is apart from logic.** `circuit.rs` is the only file with the arena and `Box`. `ops.rs` holds each stateful operator as a plain struct with a pure `step(delta) -> delta`, testable and usable without a circuit. 2. **Scheduling is creation order.** A node is built from streams that already exist, so creation order is a topological order, and the circuit has no cycles: the ADR has no recursion operator, and `integrate` is its own stateful node, not a delayed feedback loop. - `delay` is a node with one held Z-set. Not `Send`; one thread per circuit (clause 1). + `delay` is a node with one held Z-set. `Send`, not `Sync`: one thread steps a circuit at a + time (clause 1; "Threads" below). 3. **Incremental means the integral of the output is the operator of the integral of the input.** That is the property every operator is tested against, over random delta streams. `distinct` and `aggregate` are not linear, so they keep the integrated input and emit the @@ -168,8 +169,8 @@ Decisions, where the ADR left a choice: host's. **`current()` is the whole tick.** A stream keeps the latest piece, which downstream - operators read, and the whole value of the tick, which `Stream::current` reads: pieces - within one tick are absorbed into it, so `integrate.current()` after a chunked load is the + operators read, and the whole value of the tick, which `Circuit::current(&s)` reads: pieces + within one tick are absorbed into it, so `current` of an integrate stream after a chunked load is the full integral and a continued `delay` still reads the previous tick's input. `Output` sums pieces as before. @@ -230,10 +231,10 @@ clause's wording stands; this is how the crate reads it). its map (`a_panicking_value_clone_leaves_a_memtrace_unchanged`). 5. **The edge's two copies.** An integrate stream's piece is also its output, so a second copy (the whole-tick value) only exists while a tick is split. `Slot::whole` is `None` otherwise - and `current()` reads the piece. `piece()` lends a borrow rather than a clone - (improvise-bw4p.11). An integrate stream now holds its integral twice (the node's `acc` - and the stream's piece), not three times; sharing those two needs an `Rc` in the slot and - was not done. + and `Circuit::current` reads the piece. An eval reads its input's piece in the arena by + reference rather than cloning it (improvise-bw4p.11). An integrate stream now holds its integral twice (the node's `acc` + and the stream's piece), not three times; sharing those two needs shared ownership in the + slot and was not done. 6. **What a transaction does not undo.** - A closure's captured state (a `map` function that counts, an FFI callback's side effects). Operators that keep state keep it in the crate's own types for this reason. @@ -245,10 +246,10 @@ clause's wording stands; this is how the crate reads it). can overflow the pending delta before any step runs (`p1_a_chunk_whose_push_panics_in_a_txn_is_counted`). A later count makes `interrupted` true for good, finished or not (`rollback_rebuild.rs`: begun before the transaction and continued inside, finished inside, a first chunk that panicked, and a dead job revived by a later step). - - An `Output::take` made inside the transaction: a rollback subtracts the step's output - from the sink, which a `take` has emptied. Take after `commit`; `zset-ffi` refuses to take + - A `Circuit::take` of an output made inside the transaction: a rollback subtracts the step's + output from the sink, which a `take` has emptied. Take after `commit`; `zset-ffi` refuses to take while a transaction is open. -7. **Pushes.** `Input::push` is not logged (a handle has no circuit). `rollback` puts the +7. **Pushes.** `Circuit::push` is not logged. `rollback` puts the pending delta back to what it was at `begin`, so a push made after `begin` is dropped, and one made before it is pending again. Operators built after `begin` start logging at once and go back to empty; they are not removed. @@ -262,10 +263,51 @@ circuit unusable: drop it and build another. ## Rebuilds and circuits (improvise-bw4p.12) -A `Rebuild` takes its circuit's identity from its `Input`: the circuit holds an `Rc<()>` and every -`Input` it makes holds a clone, so `Rebuild::belongs_to(&circuit)` is a pointer comparison and no -global counter is needed. `Rebuild::step` on another circuit panics, where it used to report +A `Rebuild` takes its circuit's identity from its `Input`: every handle a circuit makes holds the +circuit's token (a process-wide counter since the arena, `Rc::ptr_eq` before), so +`Rebuild::belongs_to(&circuit)` is a comparison. `Rebuild::step` on another circuit panics, where it used to report `Done` and land its chunk in that circuit's next tick. An interrupted job (decision 8 above) can never finish: `interrupted` stays true however many steps follow, because a split tick cannot be mended and a rollback that took a chunk is counted forever (decision 6 above). Drop it and start another job from the same snapshot; the host keeps the snapshot. + +## Threads (user ruling 2026-10-11; ADR 009 clause 1 and section 9) + +The graph is network-agnostic and multi-threaded from the start, and **multi-threading is +scheduling**: a circuit stays synchronous and single-threaded, and a host moves whole circuits +between threads. Shared-nothing: one circuit per runnable unit, joined only by edges (the edge +adapter, a later PR, is the one thing that crosses threads). Splitting one circuit's keys across +threads (data parallelism) is not done and not precluded: its exchange step would be the same +edge. This is a ruling recorded here, not an ADR amendment. Native threads are assumed; wasm +threads are out of scope, and nothing in `zset` spawns one. `Send` and `mpsc` compile on wasm32. + +1. **`Circuit` is `Send` and not `Sync`, with no lock anywhere in it.** It moves between + threads, one at a time (`tests/threads.rs`: a circuit built on one thread, stepped on a new + thread each tick, equals a circuit that never moved; a transaction begun on one thread is + rolled back on another). Not `Sync` is pinned by `a_circuit_is_not_sync`, an ordinary test + using the ambiguous-impl trick: it fails to build (E0283) if `Sync` is ever implemented. +2. **The `Rc` graph is an arena.** What made the circuit `!Send` was the sharing + between the circuit and the handles it hands out. A circuit now owns every edge's slot, every + input's pending delta and every sink's sum in one `Vec>`; a node holds + typed ids, and `eval`, `begin`, `commit` and `rollback` take the arena. A downcast on every + access is the price. Measured on an M-series Mac under load, a release build of a 1001-node + chain (map, filter, distinct, delay) over 3000 ticks, best of 5, alternating with main: main + 127k-131k ticks/s, this 118k-122k, so 7-9% slower per tick (about 1 ns per node per tick). + The reviewer saw 10-25% on a more loaded run. Not reduced. +3. **Handles are plain ids, used through the circuit.** `Stream`, `Input` and `Output` are + `Copy`, `Send` and `Sync`, and carry their circuit's token. The calls moved onto the circuit: + `Circuit::current(&stream)`, `Circuit::push(&input, delta)`, `Circuit::take(&output)`. A + handle from another circuit panics ("handle from another circuit"), where before it was + quietly wrong. A consequence: a closure can no longer read a stream during a step, since it + has no circuit; the old `b1_...` case does not compile, so it is gone. The handle types are + invariant in their element (`PhantomData T>`), as main's `Rc>` was, so a + stream of `for<'a> fn(&'a u8)` cannot be coerced into a stream of `fn(&'static u8)`, where + `current` would panic on a failed downcast. Nine call sites check the token, and + `every_call_that_takes_a_handle_refuses_a_foreign_one` has one case for each. +4. **User closures are `Send + 'static`;** keys and values are `Send`. A custom `Trace` handed to + `join_indexed_in` or `aggregate_in` must be `Send`. `Piecewise` requires `Send + 'static`. +5. **The sink does not clone.** Its node reads the input edge and writes the sum through + `Arena::read_write`, two disjoint entries (the read one is always allocated first). + `a_sink_adds_its_input_with_one_clone_per_key` pins it over `Counted` keys. +6. **Handing a circuit to a thread is the host's business** (`std::thread` lives in host crates). + Whoever owns the circuit pushes, steps and takes; a handle is a number and needs no thread. diff --git a/context/test-evidence/zset-ffi.md b/context/test-evidence/zset-ffi.md index b7c34f1c..8358fb22 100644 --- a/context/test-evidence/zset-ffi.md +++ b/context/test-evidence/zset-ffi.md @@ -221,3 +221,15 @@ A job whose first chunk failed inside the transaction resumed without that chunk The same path through the ABI: `zs_rebuild_step` returns `ZS_ERR_OVERFLOW` from the chunk's push, `zs_circuit_rollback` succeeds, and the next `zs_rebuild_step` was a silent success; it is now `ZS_ERR_INTERRUPTED`. Rival: the mark before the push removed fails `[f5]`. + +### the_opaque_handles_are_send + +`ZsCircuit` and `ZsRebuild` are `Send` (the engine's fault cell is an `Arc` and the +callback context an `unsafe impl Send` wrapper). Rival: restoring the `Rc` fault cell, or +dropping `unsafe impl Send for Ctx`, fails `engine.rs` to build (`*mut c_void cannot be sent`). + +### a_circuit_stepped_on_two_threads_runs_its_callbacks_on_the_thread_that_steps + +A filter callback records its thread; the circuit is stepped on the test thread, moved, and +stepped on another. The ids differ and match the stepping threads, so the callback is not +pinned to the thread that built the circuit. The second thread's push is seen by its own step. diff --git a/context/test-evidence/zset.md b/context/test-evidence/zset.md index 1605fa25..1b01ca13 100644 --- a/context/test-evidence/zset.md +++ b/context/test-evidence/zset.md @@ -552,9 +552,9 @@ A first chunk that panicked left `began_at` unset, and the job resumed without i now recorded before the chunk runs. Rival: recording it never (the `get_or_insert` removed) fails all four; a rival recording it after the chunk was not run separately. -### b1_a_closure_that_reads_a_stream_during_eval_does_not_panic +b1 (a closure reading a stream during an eval; the test is removed) -The borrowed `piece()` does not conflict with a closure that reads streams during an eval, +Superseded by the arena (PR dbsp/zset-send): a stream is read through `Circuit::current`, so a closure cannot read one during a step, and the case no longer compiles. It used to check that the borrowed `piece()` did not conflict with a closure that read streams, including its own source, inside a transaction. It passed before and after. ### c1_a_clone_panic_anywhere_in_a_step_is_rolled_back @@ -596,3 +596,39 @@ Delta review of #355 at b2445fed. `Rebuild::step` takes the chunk from its itera it before stepping; a push that overflows the pending delta panicked with no step run, so `undone` was not bumped and the resumed job silently dropped the chunk (key 1's `+1`). The step is now marked before the push. Rival: that mark removed fails `[p1]`. + +### a_circuit_and_every_handle_it_hands_out_is_send + +The Rc/RefCell graph made `Circuit` `!Send`. Rival: an `Rc<()>` field on `Circuit` fails the +build of `tests/threads.rs` with three `Rc<()> cannot be sent between threads safely` errors, +so the file never runs; the same file's other tests move a circuit for real. + +### a_circuit_built_on_one_thread_steps_on_another_and_returns + +A circuit moved to a new thread for each of nine ticks equals one that never moved, over a +pipeline with `integrate`, `distinct`, `aggregate`, `filter` and `delay`. It would fail if a +node kept state outside the circuit (a thread-local, say), which would not move with it. + +### a_transaction_begun_on_one_thread_is_rolled_back_on_another + +`begin` and a step on one thread, `rollback` on another: the stream, the sink and `ticks` are +as at `begin`. The undo log lives in the nodes, which move with the circuit. + +### every_call_that_takes_a_handle_refuses_a_foreign_one + +Ids index an arena, so a foreign handle from a circuit of the same shape reads or wires the +other circuit's slot. Ten cases (`current`, `take`, `push`, a unary operator, both operands of a +binary one, `output`, `delay`, `integrate`, `distinct`) each need the panic text "handle from +another circuit". The first draft pinned one site in nine. Rivals, each failing it: the check +removed from `current`, `take`, `push`, `unary_with`, `binary_with` (each operand), `output`, +`delay` or `integrate`. + +### a_circuit_is_not_sync + +The ambiguous-impl trick: the test does not build (E0283) if `Circuit` implements `Sync`. +Rival: `unsafe impl Sync for Circuit {}` fails the build of `tests/threads.rs`. + +### a_sink_adds_its_input_with_one_clone_per_key + +A sink over `Counted` keys clones each key once (the insert into its sum). Rival: a sink that +clones its input piece first, as the reviewer injected, makes `N` more. diff --git a/crates/zset-ffi/include/zset_ffi.h b/crates/zset-ffi/include/zset_ffi.h index 93857477..37a1f9b7 100644 --- a/crates/zset-ffi/include/zset_ffi.h +++ b/crates/zset-ffi/include/zset_ffi.h @@ -17,9 +17,12 @@ * - The library never returns memory the caller must free, except handles. * - Weights are int64_t. A weight that would leave int64_t is ZS_ERR_OVERFLOW, never a * wrong value, and never an unwinding exception. - * - Not thread-safe. A ZsCircuit and every ZsRebuild made from it form ONE group (they share - * a reference count): keep the whole group on one thread at a time. Z-sets are separate - * handles. Streams, inputs and sinks are uint32_t ids local to their circuit; a sink id is + * - Not thread-safe, but movable. A ZsCircuit, a ZsRebuild and a ZsZSet may each be handed to + * another thread; use each from one thread at a time (a zs_rebuild_step call uses both its + * job and its circuit). Callbacks run on whichever thread calls zs_circuit_step or + * zs_rebuild_step, with THAT thread's thread-local state (Lisp dynamic bindings, errno, + * zs_last_error), so a ctx must be usable from every thread that may step. + * - Streams, inputs and sinks are uint32_t ids local to their circuit; a sink id is * never a stream id (they are refused as ZS_ERR_BAD_HANDLE), but an id from another circuit * is not detected. A ZsRebuild is stepped only on the circuit that made it (else * ZS_ERR_BAD_HANDLE), even if another circuit now lives at the old address. diff --git a/crates/zset-ffi/src/engine.rs b/crates/zset-ffi/src/engine.rs index 7fb643ce..1dc899ed 100644 --- a/crates/zset-ffi/src/engine.rs +++ b/crates/zset-ffi/src/engine.rs @@ -2,10 +2,9 @@ //! `zset` are caught and classified here. See `context/code/zset-ffi.md`. use std::any::Any; -use std::cell::Cell; use std::ffi::c_void; use std::panic::{AssertUnwindSafe, catch_unwind}; -use std::rc::Rc; +use std::sync::{Arc, Mutex, MutexGuard}; use zset::{Circuit, Input, Output, Progress, Rebuild, Stream, ZSet}; pub type Bytes = Vec; @@ -130,19 +129,30 @@ pub type KeepFn = extern "C" fn(*mut c_void, *const u8, usize) -> i32; /// Returns 0 (still time), 1 (expired), or negative to fail the step. pub type ExpiredFn = extern "C" fn(*mut c_void) -> i32; -/// A caller's context pointer; the circuit never dereferences it. +/// A caller's context pointer, never dereferenced here. `Send` on the caller's contract that +/// it is usable from any thread that steps (`context/code/zset-ffi.md`, decision 10). #[derive(Clone, Copy)] struct Ctx(*mut c_void); -type Fault = Rc>>; +// SAFETY: see the type's contract; the pointer is only ever handed back to the caller. +unsafe impl Send for Ctx {} + +impl Ctx { + /// A method, so a closure captures the whole `Ctx` and not its (non-`Send`) field. + fn get(self) -> *mut c_void { + self.0 + } +} + +type Fault = Arc>>; + +fn fault_lock(fault: &Fault) -> MutexGuard<'_, Option<(Code, &'static str)>> { + fault.lock().unwrap_or_else(|e| e.into_inner()) +} /// Records a fault unless one is already recorded: the first one is the cause. fn note(fault: &Fault, code: Code, msg: &'static str) { - if let Some(first) = fault.take() { - fault.set(Some(first)); - } else { - fault.set(Some((code, msg))); - } + fault_lock(fault).get_or_insert((code, msg)); } extern "C" fn collect(ctx: *mut c_void, p: *const u8, n: usize) { @@ -161,7 +171,7 @@ extern "C" fn collect(ctx: *mut c_void, p: *const u8, n: usize) { fn images(f: FlatMapFn, ctx: Ctx, key: &Bytes, fault: &Fault) -> Vec { let mut out: Vec = Vec::new(); let rc = f( - ctx.0, + ctx.get(), key.as_ptr(), key.len(), collect, @@ -227,7 +237,7 @@ impl Engine { streams: Vec::new(), inputs: Vec::new(), sinks: Vec::new(), - fault: Rc::default(), + fault: Arc::default(), poisoned: false, txn: false, } @@ -236,7 +246,7 @@ impl Engine { fn stream(&self, id: u32) -> Res> { self.streams .get(id as usize) - .cloned() + .copied() .ok_or_else(|| Fail::new(Code::BadHandle, "no such stream")) } @@ -307,9 +317,9 @@ impl Engine { pub fn filter(&mut self, s: u32, keep: KeepFn, ctx: *mut c_void) -> Res { self.alive()?; let s = self.stream(s)?; - let (ctx, fault) = (Ctx(ctx), Rc::clone(&self.fault)); + let (ctx, fault) = (Ctx(ctx), Arc::clone(&self.fault)); let out = self.circuit.filter(&s, move |k: &Bytes| { - match keep(ctx.0, k.as_ptr(), k.len()) { + match keep(ctx.get(), k.as_ptr(), k.len()) { 0 => false, n if n > 0 => true, _ => { @@ -324,7 +334,7 @@ impl Engine { pub fn flat_map(&mut self, s: u32, f: FlatMapFn, ctx: *mut c_void) -> Res { self.alive()?; let s = self.stream(s)?; - let (ctx, fault) = (Ctx(ctx), Rc::clone(&self.fault)); + let (ctx, fault) = (Ctx(ctx), Arc::clone(&self.fault)); let out = self .circuit .flat_map(&s, move |k: &Bytes| images(f, ctx, k, &fault)); @@ -335,7 +345,7 @@ impl Engine { pub fn index_by(&mut self, s: u32, f: FlatMapFn, ctx: *mut c_void) -> Res { self.alive()?; let s = self.stream(s)?; - let (ctx, fault) = (Ctx(ctx), Rc::clone(&self.fault)); + let (ctx, fault) = (Ctx(ctx), Arc::clone(&self.fault)); let out = self.circuit.flat_map(&s, move |k: &Bytes| { let mut gs = images(f, ctx, k, &fault); if gs.len() != 1 { @@ -359,7 +369,7 @@ impl Engine { /// Splits `pair(k, v)` elements; a malformed one records a fault and is dropped. fn keyed(&mut self, s: &Stream) -> Stream> { - let fault = Rc::clone(&self.fault); + let fault = Arc::clone(&self.fault); self.circuit.flat_map(s, move |b: &Bytes| match unpair(b) { Some((k, v)) => vec![(k.to_vec(), v.to_vec())], None => { @@ -378,7 +388,7 @@ impl Engine { self.alive()?; let (a, b) = (self.stream(a)?, self.stream(b)?); let (ka, kb) = (self.keyed(&a), self.keyed(&b)); - let fault = Rc::clone(&self.fault); + let fault = Arc::clone(&self.fault); let j = self .circuit .join_indexed(&ka, &kb, move |k: &Bytes, x: &Bytes, y: &Bytes| { @@ -396,7 +406,7 @@ impl Engine { self.alive()?; let s = self.stream(s)?; let keyed = self.keyed(&s); - let fault = Rc::clone(&self.fault); + let fault = Arc::clone(&self.fault); let folded = self .circuit .aggregate(&keyed, move |_, vals: &[(Bytes, i64)]| { @@ -427,7 +437,7 @@ impl Engine { } } }); - let fault = Rc::clone(&self.fault); + let fault = Arc::clone(&self.fault); let out = self .circuit .flat_map(&folded, move |(k, r): &(Bytes, i64)| { @@ -445,26 +455,27 @@ impl Engine { fn input_of(&self, id: u32) -> Res> { self.inputs .get(id as usize) - .and_then(|i| i.clone()) + .and_then(|i| *i) .ok_or_else(|| Fail::new(Code::BadHandle, "that stream is not an input")) } pub fn push(&mut self, input: u32, z: &BZ) -> Res<()> { self.alive()?; - self.input_of(input)?.push(z.clone()); + let input = self.input_of(input)?; + self.circuit.push(&input, z.clone()); Ok(()) } /// Runs `f` on the circuit; a panic or a recorded fault poisons the engine. fn guarded(&mut self, f: impl FnOnce(&mut Circuit) -> T) -> Res { self.alive()?; - self.fault.set(None); + *fault_lock(&self.fault) = None; match catch_unwind(AssertUnwindSafe(|| f(&mut self.circuit))) { Err(payload) => { self.poisoned = true; Err(classify(payload)) } - Ok(v) => match self.fault.take() { + Ok(v) => match fault_lock(&self.fault).take() { Some((code, msg)) => { self.poisoned = true; Err(Fail::new(code, msg)) @@ -529,7 +540,8 @@ impl Engine { self.sinks .get((sink & !SINK_BIT) as usize) .filter(|_| sink & SINK_BIT != 0) - .map(|s| s.take()) + .copied() + .map(|s| self.circuit.take(&s)) .ok_or_else(|| Fail::new(Code::BadHandle, "no such output")) } @@ -551,7 +563,7 @@ impl Engine { "the rebuild was interrupted by a circuit step before it finished", )); } - let fault = Rc::clone(&self.fault); + let fault = Arc::clone(&self.fault); let progress = self.guarded(|c| { job.step(c, || match expired { None => false, diff --git a/crates/zset-ffi/tests/threads_abi.rs b/crates/zset-ffi/tests/threads_abi.rs new file mode 100644 index 00000000..87ff79c9 --- /dev/null +++ b/crates/zset-ffi/tests/threads_abi.rs @@ -0,0 +1,98 @@ +//! A circuit moves between threads, one at a time, and its callbacks run on whichever thread +//! steps it (`context/code/zset-ffi.md`, decision 10). Evidence: `context/test-evidence/zset-ffi.md`. + +use std::ffi::c_void; +use std::ptr::null_mut; +use std::sync::Mutex; +use std::thread::{self, ThreadId}; +use zset_ffi::*; + +const OK: i32 = 0; + +fn is_send() {} + +#[test] +fn the_opaque_handles_are_send() { + is_send::(); + is_send::(); +} + +/// The caller's context: which threads ran the callback. +struct Seen(Mutex>); + +extern "C" fn keep_all(ctx: *mut c_void, _: *const u8, _: usize) -> i32 { + // SAFETY: `ctx` is the `Seen` the test keeps alive across every step. + unsafe { &*(ctx as *const Seen) } + .0 + .lock() + .unwrap() + .push(thread::current().id()); + 1 +} + +/// A raw handle moved to another thread; the test is the one-at-a-time owner. +struct Moved(*mut ZsCircuit); +unsafe impl Send for Moved {} + +fn push_key(c: *mut ZsCircuit, input: u32, key: &[u8]) { + let mut z = null_mut(); + assert_eq!(unsafe { zs_zset_new(&mut z) }, OK); + assert_eq!(unsafe { zs_zset_insert(z, key.as_ptr(), key.len(), 1) }, OK); + assert_eq!(unsafe { zs_circuit_push(c, input, z) }, OK); + unsafe { zs_zset_free(z) }; +} + +fn step_and_count(c: *mut ZsCircuit, sink: u32) -> usize { + assert_eq!(unsafe { zs_circuit_step(c) }, OK); + let mut out = null_mut(); + assert_eq!(unsafe { zs_circuit_take(c, sink, &mut out) }, OK); + let mut n = 0; + let mut buf = [0u8; 64]; + assert_eq!( + unsafe { zs_zset_encode(out, buf.as_mut_ptr(), buf.len(), &mut n) }, + OK + ); + unsafe { zs_zset_free(out) }; + n / 13 +} + +#[test] +fn a_circuit_stepped_on_two_threads_runs_its_callbacks_on_the_thread_that_steps() { + let seen = Seen(Mutex::new(Vec::new())); + let ctx = &seen as *const Seen as *mut c_void; + let (mut c, mut input, mut kept, mut sink) = (null_mut(), 0, 0, 0); + assert_eq!(unsafe { zs_circuit_new(&mut c) }, OK); + assert_eq!(unsafe { zs_circuit_input(c, &mut input) }, OK); + assert_eq!( + unsafe { zs_circuit_filter(c, input, Some(keep_all), ctx, &mut kept) }, + OK + ); + assert_eq!(unsafe { zs_circuit_output(c, kept, &mut sink) }, OK); + + push_key(c, input, b"a"); + assert_eq!(step_and_count(c, sink), 1); + let here = thread::current().id(); + + let moved = Moved(c); + let there = thread::spawn(move || { + let moved = moved; + push_key(moved.0, input, b"b"); + let n = step_and_count(moved.0, sink); + (n, thread::current().id(), moved) + }) + .join() + .unwrap(); + assert_eq!( + there.0, 1, + "[threads] the second thread's step saw its own push" + ); + assert_ne!(there.1, here); + + let ids = seen.0.lock().unwrap().clone(); + assert_eq!( + ids, + vec![here, there.1], + "[threads] each callback ran where it was stepped" + ); + unsafe { zs_circuit_free(there.2.0) }; +} diff --git a/crates/zset/src/circuit.rs b/crates/zset/src/circuit.rs index 8aee5bbe..fdaa41e6 100644 --- a/crates/zset/src/circuit.rs +++ b/crates/zset/src/circuit.rs @@ -1,12 +1,13 @@ -//! The synchronous, single-threaded circuit. This module is the wiring (`Rc`, `RefCell`, -//! boxed nodes); the operators' logic is in `ops.rs` and `lib.rs`. +//! The synchronous circuit: `Send`, never `Sync`, no locks; it owns every edge's value in an +//! arena and its handles are ids. See `context/code/zset.md`, "Threads". use crate::ops::{ AggregateState, DistinctState, JoinState, Log, Transactional, Undoable, add_logged, unadd, }; use crate::{MemTrace, Trace, ZSet}; -use std::cell::{Ref, RefCell}; -use std::rc::Rc; +use std::any::Any; +use std::marker::PhantomData; +use std::sync::atomic::{AtomicU64, Ordering}; /// What an edge carries: the latest piece, and the whole tick's value, kept only while a /// tick is split (`Circuit::step_continue`); `None` means the piece is the whole. @@ -30,7 +31,7 @@ impl Default for Slot { } /// Values whose pieces of one tick add up to the tick's value. -pub trait Piecewise: Default + Clone { +pub trait Piecewise: Default + Clone + Send + 'static { fn absorb(&mut self, piece: &Self); } @@ -38,174 +39,227 @@ impl Piecewise for () { fn absorb(&mut self, _: &Self) {} } -impl Piecewise for ZSet { +impl Piecewise for ZSet { fn absorb(&mut self, piece: &Self) { *self += piece; } } -/// The value an edge carries during the current step. -pub struct Stream(Rc>>); - -impl Stream { - fn piece(&self) -> Ref<'_, T> { - Ref::map(self.0.borrow(), |s| &s.piece) - } - - fn publish(&self, v: T, cont: bool) { - let mut slot = self.0.borrow_mut(); - let slot = &mut *slot; +impl Slot { + fn publish(&mut self, v: T, cont: bool) { if cont { - if slot.open && slot.saved.is_none() { - slot.saved = Some((slot.piece.clone(), slot.whole.clone())); + if self.open && self.saved.is_none() { + self.saved = Some((self.piece.clone(), self.whole.clone())); } - let mut whole = slot.whole.take().unwrap_or_else(|| slot.piece.clone()); + let mut whole = self.whole.take().unwrap_or_else(|| self.piece.clone()); whole.absorb(&v); - slot.whole = Some(whole); - slot.piece = v; + self.whole = Some(whole); + self.piece = v; } else { - let piece = std::mem::replace(&mut slot.piece, v); - let whole = slot.whole.take(); - if slot.open && slot.saved.is_none() { - slot.saved = Some((piece, whole)); + let piece = std::mem::replace(&mut self.piece, v); + let whole = self.whole.take(); + if self.open && self.saved.is_none() { + self.saved = Some((piece, whole)); } } } } -impl Stream { - fn begin(&self) { - let mut slot = self.0.borrow_mut(); - slot.open = true; - slot.saved = None; +impl Slot { + fn begin(&mut self) { + self.open = true; + self.saved = None; } - fn commit(&self) { - let mut slot = self.0.borrow_mut(); - slot.open = false; - slot.saved = None; + fn commit(&mut self) { + self.open = false; + self.saved = None; } - fn rollback(&self) { - let mut slot = self.0.borrow_mut(); - if let Some((piece, whole)) = slot.saved.take() { - slot.piece = piece; - slot.whole = whole; + fn rollback(&mut self) { + if let Some((piece, whole)) = self.saved.take() { + self.piece = piece; + self.whole = whole; } - slot.open = false; + self.open = false; } } +/// The index of one value in a circuit's arena, typed so a read cannot name the wrong type. +struct Id(usize, PhantomData T>); + +impl Clone for Id { + fn clone(&self) -> Self { + *self + } +} + +impl Copy for Id {} + +/// Every edge's value, every input's pending delta and every output's sum, by id. +#[derive(Default)] +struct Arena(Vec>); + +impl Arena { + fn alloc(&mut self, v: T) -> Id { + self.0.push(Box::new(v)); + Id(self.0.len() - 1, PhantomData) + } + + fn get(&self, id: Id) -> &T { + self.0[id.0].downcast_ref().expect("zset: arena type") + } + + fn get_mut(&mut self, id: Id) -> &mut T { + self.0[id.0].downcast_mut().expect("zset: arena type") + } + + /// One value to read and another, allocated after it, to write. + fn read_write(&mut self, r: Id, w: Id) -> (&R, &mut W) { + let (lo, hi) = self.0.split_at_mut(w.0); + assert!( + r.0 < w.0, + "zset: the read entry comes before the written one" + ); + let ty = "zset: arena type"; + ( + lo[r.0].downcast_ref().expect(ty), + hi[0].downcast_mut().expect(ty), + ) + } +} + +/// Circuit identities; a handle remembers its circuit so a foreign one is refused. +static NEXT_CIRCUIT: AtomicU64 = AtomicU64::new(1); + +type Edge = Id>; + +/// The value an edge carries during the current step. Read it with `Circuit::current`. +pub struct Stream { + id: Edge, + circuit: u64, +} + impl Clone for Stream { fn clone(&self) -> Self { - Stream(Rc::clone(&self.0)) + *self } } -impl Stream { - /// The value of the last tick, all its pieces together (the zero before the first). - pub fn current(&self) -> T { - let slot = self.0.borrow(); - slot.whole.as_ref().unwrap_or(&slot.piece).clone() - } -} +impl Copy for Stream {} -/// Where a caller feeds deltas; they enter the circuit at the next `step`. It remembers -/// which circuit made it, so a `Rebuild` can refuse a circuit that is not its own. -pub struct Input(Rc>>, Rc<()>); +/// Where a caller feeds deltas, with `Circuit::push`; they enter the circuit at the next +/// `step`. It remembers which circuit made it, so a `Rebuild` can refuse another. +pub struct Input { + pending: Id>, + circuit: u64, +} impl Clone for Input { fn clone(&self) -> Self { - Input(Rc::clone(&self.0), Rc::clone(&self.1)) + *self } } -impl Input { - pub fn push(&self, delta: ZSet) { - *self.0.borrow_mut() += delta; - } -} +impl Copy for Input {} -/// A sink that sums every step's delta until it is taken. Take inside no transaction: a +/// A sink that sums every step's delta until `Circuit::take`. Take inside no transaction: a /// rollback subtracts the step's output, which a `take` in between has already removed. -pub struct Output(Rc>>); +pub struct Output { + sum: Id>, + circuit: u64, +} -impl Output { - pub fn take(&self) -> ZSet { - std::mem::take(&mut *self.0.borrow_mut()) +impl Clone for Output { + fn clone(&self) -> Self { + *self } } -trait Node: Transactional { +impl Copy for Output {} + +trait Node: Send { /// `cont` is true for a later piece of a tick that has already begun. - fn eval(&mut self, cont: bool); + fn eval(&mut self, a: &mut Arena, cont: bool); + fn begin(&mut self, a: &mut Arena); + fn commit(&mut self, a: &mut Arena); + fn rollback(&mut self, a: &mut Arena); } /// One input, one output, and the state `f` may change. struct Unary { - src: Stream, - out: Stream, + src: Edge, + out: Edge, state: S, f: F, } -impl Transactional for Unary { - fn begin(&mut self) { - self.state.begin(); - self.out.begin(); - } - - fn commit(&mut self) { - self.state.commit(); - self.out.commit(); - } - - fn rollback(&mut self) { - self.state.rollback(); - self.out.rollback(); - } -} - -impl O> Node - for Unary +impl Node for Unary +where + I: Piecewise, + O: Piecewise, + S: Transactional + Send, + F: FnMut(&mut S, &I) -> O + Send, { - fn eval(&mut self, cont: bool) { - let v = (self.f)(&mut self.state, &self.src.piece()); - self.out.publish(v, cont); + fn eval(&mut self, a: &mut Arena, cont: bool) { + let v = (self.f)(&mut self.state, &a.get(self.src).piece); + a.get_mut(self.out).publish(v, cont); + } + + fn begin(&mut self, a: &mut Arena) { + self.state.begin(); + a.get_mut(self.out).begin(); + } + + fn commit(&mut self, a: &mut Arena) { + self.state.commit(); + a.get_mut(self.out).commit(); + } + + fn rollback(&mut self, a: &mut Arena) { + self.state.rollback(); + a.get_mut(self.out).rollback(); } } struct Binary { - a: Stream, - b: Stream, - out: Stream, + a: Edge, + b: Edge, + out: Edge, state: S, f: F, } -impl Transactional for Binary { - fn begin(&mut self) { - self.state.begin(); - self.out.begin(); - } - - fn commit(&mut self) { - self.state.commit(); - self.out.commit(); - } - - fn rollback(&mut self) { - self.state.rollback(); - self.out.rollback(); - } -} - -impl O> Node - for Binary +impl Node for Binary +where + A: Piecewise, + B: Piecewise, + O: Piecewise, + S: Transactional + Send, + F: FnMut(&mut S, &A, &B) -> O + Send, { - fn eval(&mut self, cont: bool) { - let v = (self.f)(&mut self.state, &self.a.piece(), &self.b.piece()); - self.out.publish(v, cont); + fn eval(&mut self, ar: &mut Arena, cont: bool) { + let v = (self.f)( + &mut self.state, + &ar.get(self.a).piece, + &ar.get(self.b).piece, + ); + ar.get_mut(self.out).publish(v, cont); + } + + fn begin(&mut self, a: &mut Arena) { + self.state.begin(); + a.get_mut(self.out).begin(); + } + + fn commit(&mut self, a: &mut Arena) { + self.state.commit(); + a.get_mut(self.out).commit(); + } + + fn rollback(&mut self, a: &mut Arena) { + self.state.rollback(); + a.get_mut(self.out).rollback(); } } @@ -246,40 +300,37 @@ impl Transactional for Held { /// out(t) = in(t - 1). A continued piece adds to what is held and emits nothing. struct Delay { - src: Stream>, - out: Stream>, + src: Edge>, + out: Edge>, state: Held, } -impl Transactional for Delay { - fn begin(&mut self) { - self.state.begin(); - self.out.begin(); - } - - fn commit(&mut self) { - self.state.commit(); - self.out.commit(); - } - - fn rollback(&mut self) { - self.state.rollback(); - self.out.rollback(); - } -} - -impl Node for Delay { - fn eval(&mut self, cont: bool) { - let d = self.src.piece(); +impl Node for Delay { + fn eval(&mut self, a: &mut Arena, cont: bool) { + let d = &a.get(self.src).piece; self.state.save(); let v = if cont { - self.state.held += &*d; + self.state.held += d; ZSet::new() } else { std::mem::replace(&mut self.state.held, d.clone()) }; - drop(d); - self.out.publish(v, cont); + a.get_mut(self.out).publish(v, cont); + } + + fn begin(&mut self, a: &mut Arena) { + self.state.begin(); + a.get_mut(self.out).begin(); + } + + fn commit(&mut self, a: &mut Arena) { + self.state.commit(); + a.get_mut(self.out).commit(); + } + + fn rollback(&mut self, a: &mut Arena) { + self.state.rollback(); + a.get_mut(self.out).rollback(); } } @@ -306,105 +357,112 @@ impl Transactional for Acc { /// out(t) = in(0) + .. + in(t). A continued piece emits only its own input. struct Integrate { - src: Stream>, - out: Stream>, + src: Edge>, + out: Edge>, state: Acc, } -impl Transactional for Integrate { - fn begin(&mut self) { - self.state.begin(); - self.out.begin(); - } - - fn commit(&mut self) { - self.state.commit(); - self.out.commit(); - } - - fn rollback(&mut self) { - self.state.rollback(); - self.out.rollback(); - } -} - -impl Node for Integrate { - fn eval(&mut self, cont: bool) { - let d = self.src.piece(); - add_logged(&mut self.state.acc, &d, &mut self.state.log); +impl Node for Integrate { + fn eval(&mut self, a: &mut Arena, cont: bool) { + let d = &a.get(self.src).piece; + add_logged(&mut self.state.acc, d, &mut self.state.log); let v = if cont { d.clone() } else { self.state.acc.clone() }; - drop(d); - self.out.publish(v, cont); + a.get_mut(self.out).publish(v, cont); + } + + fn begin(&mut self, a: &mut Arena) { + self.state.begin(); + a.get_mut(self.out).begin(); + } + + fn commit(&mut self, a: &mut Arena) { + self.state.commit(); + a.get_mut(self.out).commit(); + } + + fn rollback(&mut self, a: &mut Arena) { + self.state.rollback(); + a.get_mut(self.out).rollback(); } } /// Moves what was pushed into the circuit's stream. Inside a transaction it remembers what /// was pending, so a rollback gives back the delta the failed step took. struct Source { - pending: Rc>>, - out: Stream>, + pending: Id>, + out: Edge>, saved: Option>, } -impl Transactional for Source { - fn begin(&mut self) { - self.saved = Some(self.pending.borrow().clone()); - self.out.begin(); +impl Node for Source { + fn eval(&mut self, a: &mut Arena, cont: bool) { + let v = std::mem::take(a.get_mut(self.pending)); + a.get_mut(self.out).publish(v, cont); } - fn commit(&mut self) { + fn begin(&mut self, a: &mut Arena) { + self.saved = Some(a.get(self.pending).clone()); + a.get_mut(self.out).begin(); + } + + fn commit(&mut self, a: &mut Arena) { self.saved = None; - self.out.commit(); + a.get_mut(self.out).commit(); } - fn rollback(&mut self) { + fn rollback(&mut self, a: &mut Arena) { if let Some(saved) = self.saved.take() { - *self.pending.borrow_mut() = saved; + *a.get_mut(self.pending) = saved; } - self.out.rollback(); + a.get_mut(self.out).rollback(); } } -impl Node for Source { - fn eval(&mut self, cont: bool) { - let v = std::mem::take(&mut *self.pending.borrow_mut()); - self.out.publish(v, cont); - } -} - -/// The sink's sum, and the log that takes a transaction's additions back. +/// Adds each tick to the output's sum, with the log that takes a transaction's additions +/// back. Its own edge carries nothing. struct Sink { - sum: Rc>>, + src: Edge>, + out: Edge<()>, + sum: Id>, log: Log<(K, i64)>, } -impl Transactional for Sink { - fn begin(&mut self) { +impl Node for Sink { + fn eval(&mut self, a: &mut Arena, cont: bool) { + let (src, sum) = a.read_write(self.src, self.sum); + add_logged(sum, &src.piece, &mut self.log); + a.get_mut(self.out).publish((), cont); + } + + fn begin(&mut self, a: &mut Arena) { self.log.begin(); + a.get_mut(self.out).begin(); } - fn commit(&mut self) { + fn commit(&mut self, a: &mut Arena) { self.log.commit(); + a.get_mut(self.out).commit(); } - fn rollback(&mut self) { - let sum = &mut *self.sum.borrow_mut(); + fn rollback(&mut self, a: &mut Arena) { + let sum = a.get_mut(self.sum); self.log.rollback(|e| unadd(sum, e)); + a.get_mut(self.out).rollback(); } } -/// Nodes run in creation order, which is a topological order because a node can only -/// be built from streams that already exist. There are no cycles to schedule. -#[derive(Default)] +/// Nodes run in creation order, a topological order: a node is built from existing streams. +/// `Send` (`tests/threads.rs`) and, having no lock, not `Sync`. pub struct Circuit { nodes: Vec>, + arena: Arena, ticks: u64, - /// This circuit's identity: every `Input` it makes holds a clone. - token: Rc<()>, + /// This circuit's identity: every handle it makes holds it. + token: u64, /// `Some(ticks)` as of `begin`, while a transaction is open. open: Option, /// A step or piece ran since `begin`. @@ -414,6 +472,20 @@ pub struct Circuit { undone: u64, } +impl Default for Circuit { + fn default() -> Self { + Circuit { + nodes: Vec::new(), + arena: Arena::default(), + ticks: 0, + token: NEXT_CIRCUIT.fetch_add(1, Ordering::Relaxed), + open: None, + stepped: false, + undone: 0, + } + } +} + type Z = ZSet; impl Circuit { @@ -448,7 +520,7 @@ impl Circuit { self.open = Some(self.ticks); self.stepped = false; for n in &mut self.nodes { - n.begin(); + n.begin(&mut self.arena); } } @@ -459,7 +531,7 @@ impl Circuit { "zset: no transaction is open on this circuit" ); for n in &mut self.nodes { - n.commit(); + n.commit(&mut self.arena); } } @@ -471,7 +543,7 @@ impl Circuit { .take() .expect("zset: no transaction is open on this circuit"); for n in &mut self.nodes { - n.rollback(); + n.rollback(&mut self.arena); } self.ticks = ticks; self.undone += u64::from(self.stepped); @@ -481,103 +553,147 @@ impl Circuit { fn run(&mut self, cont: bool) { self.stepped |= self.open.is_some(); for n in &mut self.nodes { - n.eval(cont); + n.eval(&mut self.arena, cont); } } + #[track_caller] + fn mine(&self, circuit: u64) { + assert_eq!( + circuit, self.token, + "zset: a handle from another circuit was used here" + ); + } + /// Adds a node; one built inside a transaction starts recording at once. fn add(&mut self, mut node: Box) { if self.open.is_some() { - node.begin(); + node.begin(&mut self.arena); } self.nodes.push(node); } - fn unary_with( + fn edge(&mut self) -> Stream { + let id = self.arena.alloc(Slot::::default()); + Stream { + id, + circuit: self.token, + } + } + + /// The value of the last tick, all its pieces together (the zero before the first). + pub fn current(&self, s: &Stream) -> T { + self.mine(s.circuit); + let slot = self.arena.get(s.id); + slot.whole.as_ref().unwrap_or(&slot.piece).clone() + } + + /// Feeds a delta to `input`; it enters at the next `step`. + pub fn push(&mut self, input: &Input, delta: ZSet) { + self.mine(input.circuit); + *self.arena.get_mut(input.pending) += delta; + } + + /// Takes the sum of the deltas `output` has seen since the last take. + pub fn take(&mut self, output: &Output) -> ZSet { + self.mine(output.circuit); + std::mem::take(self.arena.get_mut(output.sum)) + } + + fn unary_with( &mut self, src: &Stream, state: S, - f: impl FnMut(&mut S, &I) -> O + 'static, + f: impl FnMut(&mut S, &I) -> O + Send + 'static, ) -> Stream { - let out = Stream(Rc::default()); + self.mine(src.circuit); + let out = self.edge::(); self.add(Box::new(Unary { - src: src.clone(), - out: out.clone(), + src: src.id, + out: out.id, state, f, })); out } - fn unary( + fn unary( &mut self, src: &Stream, - mut f: impl FnMut(&I) -> O + 'static, + mut f: impl FnMut(&I) -> O + Send + 'static, ) -> Stream { self.unary_with(src, (), move |_, d| f(d)) } - fn binary_with< - A: Piecewise + 'static, - B: Piecewise + 'static, - O: Piecewise + 'static, - S: Transactional + 'static, - >( + fn binary_with( &mut self, a: &Stream, b: &Stream, state: S, - f: impl FnMut(&mut S, &A, &B) -> O + 'static, + f: impl FnMut(&mut S, &A, &B) -> O + Send + 'static, ) -> Stream { - let out = Stream(Rc::default()); + self.mine(a.circuit); + self.mine(b.circuit); + let out = self.edge::(); self.add(Box::new(Binary { - a: a.clone(), - b: b.clone(), - out: out.clone(), + a: a.id, + b: b.id, + out: out.id, state, f, })); out } - fn binary( + fn binary( &mut self, a: &Stream, b: &Stream, - mut f: impl FnMut(&A, &B) -> O + 'static, + mut f: impl FnMut(&A, &B) -> O + Send + 'static, ) -> Stream { self.binary_with(a, b, (), move |_, x, y| f(x, y)) } - pub fn input(&mut self) -> (Input, Stream>) { - let pending: Rc>> = Rc::default(); - let out = Stream(Rc::default()); + pub fn input(&mut self) -> (Input, Stream>) { + let pending = self.arena.alloc(Z::::default()); + let out = self.edge::>(); self.add(Box::new(Source { - pending: Rc::clone(&pending), - out: out.clone(), + pending, + out: out.id, saved: None, })); - (Input(pending, Rc::clone(&self.token)), out) + ( + Input { + pending, + circuit: self.token, + }, + out, + ) } - pub fn output(&mut self, s: &Stream>) -> Output { - let sink = Output(Rc::default()); - let state = Sink { - sum: Rc::clone(&sink.0), + pub fn output(&mut self, s: &Stream>) -> Output { + self.mine(s.circuit); + let sum = self.arena.alloc(Z::::default()); + let out = self.edge::<()>(); + self.add(Box::new(Sink { + src: s.id, + out: out.id, + sum, log: Log::default(), - }; - self.unary_with(s, state, |sink, d: &Z| { - add_logged(&mut sink.sum.borrow_mut(), d, &mut sink.log); - }); - sink + })); + Output { + sum, + circuit: self.token, + } } /// out(t) = in(t - 1); out(0) is zero. - pub fn delay(&mut self, s: &Stream>) -> Stream> { - let out = Stream(Rc::default()); + pub fn delay(&mut self, s: &Stream>) -> Stream> { + self.mine(s.circuit); + let out = self.edge::>(); self.add(Box::new(Delay { - src: s.clone(), - out: out.clone(), + src: s.id, + out: out.id, state: Held { held: ZSet::new(), open: false, @@ -588,11 +704,12 @@ impl Circuit { } /// out(t) = in(0) + .. + in(t). - pub fn integrate(&mut self, s: &Stream>) -> Stream> { - let out = Stream(Rc::default()); + pub fn integrate(&mut self, s: &Stream>) -> Stream> { + self.mine(s.circuit); + let out = self.edge::>(); self.add(Box::new(Integrate { - src: s.clone(), - out: out.clone(), + src: s.id, + out: out.id, state: Acc { acc: ZSet::new(), log: Log::default(), @@ -602,12 +719,15 @@ impl Circuit { } /// out(t) = in(t) - in(t - 1). - pub fn differentiate(&mut self, s: &Stream>) -> Stream> { + pub fn differentiate( + &mut self, + s: &Stream>, + ) -> Stream> { let prev = self.delay(s); self.binary(s, &prev, |a: &Z, b: &Z| a - b) } - pub fn plus( + pub fn plus( &mut self, a: &Stream>, b: &Stream>, @@ -615,46 +735,46 @@ impl Circuit { self.binary(a, b, |a: &Z, b: &Z| a + b) } - pub fn map( + pub fn map( &mut self, s: &Stream>, - mut f: impl FnMut(&K) -> L + 'static, + mut f: impl FnMut(&K) -> L + Send + 'static, ) -> Stream> { self.unary(s, move |d: &Z| d.map(&mut f)) } - pub fn filter( + pub fn filter( &mut self, s: &Stream>, - mut keep: impl FnMut(&K) -> bool + 'static, + mut keep: impl FnMut(&K) -> bool + Send + 'static, ) -> Stream> { self.unary(s, move |d: &Z| d.filter(&mut keep)) } /// Each key becomes any number of images, each with the key's weight. pub fn flat_map< - K: Ord + Clone + 'static, - L: Ord + Clone + 'static, + K: Ord + Clone + Send + 'static, + L: Ord + Clone + Send + 'static, I: IntoIterator, >( &mut self, s: &Stream>, - mut f: impl FnMut(&K) -> I + 'static, + mut f: impl FnMut(&K) -> I + Send + 'static, ) -> Stream> { self.unary(s, move |d: &Z| d.flat_map(&mut f)) } /// Keys each element by `key`, giving the `(key, element)` pairs a join reads. - pub fn index_by( + pub fn index_by( &mut self, s: &Stream>, - mut key: impl FnMut(&K) -> G + 'static, + mut key: impl FnMut(&K) -> G + Send + 'static, ) -> Stream> { self.unary(s, move |d: &Z| d.map(|k| (key(k), k.clone()))) } /// Incremental `distinct`. - pub fn distinct(&mut self, s: &Stream>) -> Stream> { + pub fn distinct(&mut self, s: &Stream>) -> Stream> { self.unary_with(s, DistinctState::new(), |st, d: &Z| st.step(d)) } @@ -663,13 +783,13 @@ impl Circuit { &mut self, a: &Stream>, b: &Stream>, - f: impl FnMut(&K, &A, &B) -> O + 'static, + f: impl FnMut(&K, &A, &B) -> O + Send + 'static, ) -> Stream> where - K: Ord + Clone + 'static, - A: Ord + Clone + 'static, - B: Ord + Clone + 'static, - O: Ord + Clone + 'static, + K: Ord + Clone + Send + 'static, + A: Ord + Clone + Send + 'static, + B: Ord + Clone + Send + 'static, + O: Ord + Clone + Send + 'static, { self.join_indexed_in(a, b, MemTrace::new(), MemTrace::new(), f) } @@ -681,15 +801,15 @@ impl Circuit { b: &Stream>, left: TA, right: TB, - mut f: impl FnMut(&K, &A, &B) -> O + 'static, + mut f: impl FnMut(&K, &A, &B) -> O + Send + 'static, ) -> Stream> where - K: Ord + Clone + 'static, - A: Ord + Clone + 'static, - B: Ord + Clone + 'static, - O: Ord + Clone + 'static, - TA: Trace + 'static, - TB: Trace + 'static, + K: Ord + Clone + Send + 'static, + A: Ord + Clone + Send + 'static, + B: Ord + Clone + Send + 'static, + O: Ord + Clone + Send + 'static, + TA: Trace + Send + 'static, + TB: Trace + Send + 'static, { let st = JoinState::new(Undoable::new(left), Undoable::new(right)); self.binary_with(a, b, st, move |st, da: &Z<(K, A)>, db: &Z<(K, B)>| { @@ -701,12 +821,12 @@ impl Circuit { pub fn aggregate( &mut self, s: &Stream>, - f: impl FnMut(&K, &[(V, i64)]) -> Option + 'static, + f: impl FnMut(&K, &[(V, i64)]) -> Option + Send + 'static, ) -> Stream> where - K: Ord + Clone + 'static, - V: Ord + Clone + 'static, - R: Ord + Clone + 'static, + K: Ord + Clone + Send + 'static, + V: Ord + Clone + Send + 'static, + R: Ord + Clone + Send + 'static, { self.aggregate_in(s, MemTrace::new(), f) } @@ -716,13 +836,13 @@ impl Circuit { &mut self, s: &Stream>, groups: T, - mut f: impl FnMut(&K, &[(V, i64)]) -> Option + 'static, + mut f: impl FnMut(&K, &[(V, i64)]) -> Option + Send + 'static, ) -> Stream> where - K: Ord + Clone + 'static, - V: Ord + Clone + 'static, - R: Ord + Clone + 'static, - T: Trace + 'static, + K: Ord + Clone + Send + 'static, + V: Ord + Clone + Send + 'static, + R: Ord + Clone + Send + 'static, + T: Trace + Send + 'static, { let st = AggregateState::new(Undoable::new(groups)); self.unary_with(s, st, move |st, d: &Z<(K, V)>| st.step(d, &mut f)) @@ -748,7 +868,7 @@ pub struct Rebuild { undone_at: Option, } -impl Rebuild { +impl Rebuild { /// `chunk` keys per circuit step; zero is read as one. pub fn new(input: Input, snapshot: ZSet, chunk: usize) -> Self { Rebuild { @@ -762,7 +882,7 @@ impl Rebuild { /// Whether `circuit` is the one that made this job's input. pub fn belongs_to(&self, circuit: &Circuit) -> bool { - Rc::ptr_eq(&self.input.1, &circuit.token) + self.input.circuit == circuit.token } /// Whether a `Circuit::step` split this job's tick, or a rollback undid a step after it @@ -790,7 +910,7 @@ impl Rebuild { return Progress::Done; } circuit.stepped |= circuit.open.is_some(); - self.input.push(chunk); + circuit.push(&self.input, chunk); if self.began_at.is_some() { circuit.step_continue(); } else { diff --git a/crates/zset/tests/circuit.rs b/crates/zset/tests/circuit.rs index 7d38b1fd..550b3a44 100644 --- a/crates/zset/tests/circuit.rs +++ b/crates/zset/tests/circuit.rs @@ -4,7 +4,8 @@ use proptest::prelude::*; use std::cell::Cell; use std::ops::Bound; -use std::rc::Rc; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use zset::{ AggregateState, Circuit, MemTrace, Progress, Rebuild, Stream, Trace, ZSet, differentiate, integrate, @@ -33,8 +34,8 @@ fn matches_recompute( build: impl FnOnce(&mut Circuit, &Stream>) -> Stream>, reference: impl Fn(&Z) -> Z, ) where - K: Ord + Clone + std::fmt::Debug + 'static, - L: Ord + Clone + std::fmt::Debug + 'static, + K: Ord + Clone + std::fmt::Debug + Send + 'static, + L: Ord + Clone + std::fmt::Debug + Send + 'static, { let mut c = Circuit::new(); let (input, s) = c.input::(); @@ -42,10 +43,10 @@ fn matches_recompute( let sink = c.output(&out); let (mut fed, mut got) = (Z::new(), Z::new()); for (i, d) in steps.iter().enumerate() { - input.push(d.clone()); + c.push(&input, d.clone()); c.step(); fed += d; - got += sink.take(); + got += c.take(&sink); assert_eq!(got, reference(&fed), "[incremental] after step {i}"); } } @@ -104,10 +105,10 @@ fn join_circuit(da: &[Z<(u8, u8)>], db: &[Z<(u8, u8)>]) -> Vec> let sink = c.output(&j); let mut deltas = Vec::new(); for i in 0..da.len().max(db.len()) { - ia.push(da.get(i).cloned().unwrap_or_default()); - ib.push(db.get(i).cloned().unwrap_or_default()); + c.push(&ia, da.get(i).cloned().unwrap_or_default()); + c.push(&ib, db.get(i).cloned().unwrap_or_default()); c.step(); - deltas.push(sink.take()); + deltas.push(c.take(&sink)); } deltas } @@ -210,10 +211,10 @@ proptest! { let (sink_s, sink_back) = (c.output(&s), c.output(&back)); let (mut seen_s, mut seen_back) = (Vec::new(), Vec::new()); for d in &steps { - input.push(d.clone()); + c.push(&input, d.clone()); c.step(); - seen_s.push(s.current()); - seen_back.push(back.current()); + seen_s.push(c.current(&s)); + seen_back.push(c.current(&back)); } let _ = (sink_s, sink_back); prop_assert_eq!(seen_back, seen_s, "[circuit-i-d]"); @@ -227,9 +228,9 @@ proptest! { let i = c.integrate(&s); let back = c.differentiate(&i); for (n, d) in steps.iter().enumerate() { - input.push(d.clone()); + c.push(&input, d.clone()); c.step(); - prop_assert_eq!(back.current(), d.clone(), "[circuit-d-i] step {}", n); + prop_assert_eq!(c.current(&back), d.clone(), "[circuit-d-i] step {}", n); } } @@ -240,9 +241,9 @@ proptest! { let d = c.delay(&s); let mut prev = Z::new(); for (n, step) in steps.iter().enumerate() { - input.push(step.clone()); + c.push(&input, step.clone()); c.step(); - prop_assert_eq!(d.current(), prev.clone(), "[delay] step {}", n); + prop_assert_eq!(c.current(&d), prev.clone(), "[delay] step {}", n); prev = step.clone(); } } @@ -263,9 +264,9 @@ proptest! { let (input, s) = whole.input::(); let out = build(&mut whole, &s); let sink = whole.output(&out); - input.push(snapshot.clone()); + whole.push(&input, snapshot.clone()); whole.step(); - let want = sink.take(); + let want = whole.take(&sink); let mut c = Circuit::new(); let (input, s) = c.input::(); @@ -282,7 +283,7 @@ proptest! { left.set(left.get() - 1); left.get() == 0 }); - got += sink.take(); + got += c.take(&sink); if progress == Progress::Done { break; } } prop_assert_eq!(got, want, "[rebuild]"); @@ -315,9 +316,9 @@ fn a_rebuild_that_is_out_of_time_reports_pending_after_one_chunk() { let sink = c.output(&s); let mut job = Rebuild::new(input, (0..6u8).map(|k| (k, 1)).collect(), 2); assert_eq!(job.step(&mut c, || true), Progress::Pending); - assert_eq!(sink.take().len(), 2, "exactly one chunk ran"); + assert_eq!(c.take(&sink).len(), 2, "exactly one chunk ran"); assert_eq!(job.step(&mut c, || false), Progress::Done); - assert_eq!(sink.take().len(), 4); + assert_eq!(c.take(&sink).len(), 4); assert_eq!(job.step(&mut c, || true), Progress::Done); } @@ -332,20 +333,20 @@ fn an_unbounded_range_that_is_empty_reads_nothing() { #[test] fn the_join_reads_state_through_the_trace_it_was_given() { - struct Counting(MemTrace, Rc>); + struct Counting(MemTrace, Arc); impl Trace for Counting { fn apply(&mut self, k: &u8, v: &u8, w: i64) { self.0.apply(k, v, w) } fn lookup(&self, k: &u8) -> Vec<(u8, i64)> { - self.1.set(self.1.get() + 1); + self.1.fetch_add(1, Ordering::Relaxed); self.0.lookup(k) } fn range(&self, lo: Bound<&u8>, hi: Bound<&u8>) -> Vec<(u8, u8, i64)> { self.0.range(lo, hi) } } - let looked = Rc::new(Cell::new(0)); + let looked = Arc::new(AtomicUsize::new(0)); let mut c = Circuit::new(); let (ia, a) = c.input::<(u8, u8)>(); let (ib, b) = c.input::<(u8, u8)>(); @@ -356,11 +357,11 @@ fn the_join_reads_state_through_the_trace_it_was_given() { MemTrace::new(), |k, x, y| (*k, *x, *y), ); - ia.push(Z::singleton((1, 1), 1)); - ib.push(Z::singleton((1, 2), 1)); + c.push(&ia, Z::singleton((1, 1), 1)); + c.push(&ib, Z::singleton((1, 2), 1)); c.step(); - assert_eq!(j.current(), Z::singleton((1, 1, 2), 1)); - assert!(looked.get() > 0); + assert_eq!(c.current(&j), Z::singleton((1, 1, 2), 1)); + assert!(looked.load(Ordering::Relaxed) > 0); } #[test] @@ -398,9 +399,9 @@ fn loaded(snapshot: &Z, chunk: usize, budget: usize) -> (Z, Z, Z let (input, s) = c.input::(); let (d, i, df) = (c.delay(&s), c.integrate(&s), c.differentiate(&s)); let sinks = (c.output(&d), c.output(&i), c.output(&df)); - input.push(Z::singleton(9, 1)); + c.push(&input, Z::singleton(9, 1)); c.step(); - let _ = (sinks.0.take(), sinks.1.take(), sinks.2.take()); + let _ = (c.take(&sinks.0), c.take(&sinks.1), c.take(&sinks.2)); let mut job = Rebuild::new(input, snapshot.clone(), chunk); for _ in 0..50 { let left = Cell::new(budget); @@ -412,7 +413,7 @@ fn loaded(snapshot: &Z, chunk: usize, budget: usize) -> (Z, Z, Z break; } } - (sinks.0.take(), sinks.1.take(), sinks.2.take()) + (c.take(&sinks.0), c.take(&sinks.1), c.take(&sinks.2)) } #[test] @@ -432,13 +433,17 @@ fn an_empty_snapshot_is_still_one_tick() { let mut c = Circuit::new(); let (input, s) = c.input::(); let d = c.delay(&s); - input.push(Z::singleton(1, 1)); + c.push(&input, Z::singleton(1, 1)); c.step(); let mut job = Rebuild::new(input, Z::new(), 3); assert_eq!(job.step(&mut c, || false), Progress::Done); - assert_eq!(d.current().len(), 1, "the empty load ticked the clock once"); + assert_eq!( + c.current(&d).len(), + 1, + "the empty load ticked the clock once" + ); assert_eq!(job.step(&mut c, || false), Progress::Done); - assert_eq!(d.current().len(), 1, "a finished job does not tick again"); + assert_eq!(c.current(&d).len(), 1, "a finished job does not tick again"); } #[test] @@ -460,21 +465,21 @@ proptest! { let (input, s) = c.input::(); let out = temporal(&mut c, &s); let sink = c.output(&out); - input.push(prior.clone()); + c.push(&input, prior.clone()); c.step(); - let _ = sink.take(); - let input_again = input.clone(); + let _ = c.take(&sink); + let input_again = input; let mut job = Rebuild::new(input, snapshot.clone(), chunk); for _ in 0..50 { let left = Cell::new(budget); if job.step(&mut c, || { left.set(left.get() - 1); left.get() == 0 }) == Progress::Done { break; } } - let at = sink.take(); - input_again.push(Z::singleton(5, 1)); + let at = c.take(&sink); + c.push(&input_again, Z::singleton(5, 1)); c.step(); - let next = sink.take(); + let next = c.take(&sink); c.step(); - (at, next, sink.take()) + (at, next, c.take(&sink)) }; prop_assert_eq!(run(chunk, budget), run(100, 100), "[temporal-rebuild]"); } diff --git a/crates/zset/tests/clone_panic.rs b/crates/zset/tests/clone_panic.rs index fae17b24..a50b01d8 100644 --- a/crates/zset/tests/clone_panic.rs +++ b/crates/zset/tests/clone_panic.rs @@ -55,7 +55,7 @@ fn rig(which: u8) -> Rig { let mut c = Circuit::new(); let (a, sa) = c.input::(); let (b, sb) = c.input::(); - let mut streams = vec![sa.clone(), sb.clone()]; + let mut streams = vec![sa, sb]; if which & 1 != 0 { streams.push(c.integrate(&sa)); } @@ -84,14 +84,20 @@ impl Rig { fn view(&self) -> String { format!( "{:?} {:?}", - self.streams.iter().map(|s| s.current()).collect::>(), - self.agg.current() + self.streams + .iter() + .map(|s| self.c.current(s)) + .collect::>(), + self.c.current(&self.agg) ) } - fn drain(&self) -> String { + fn drain(&mut self) -> String { format!( "{:?}", - self.sinks.iter().map(|s| s.take()).collect::>() + self.sinks + .iter() + .map(|s| self.c.take(s)) + .collect::>() ) } } @@ -107,9 +113,9 @@ fn clone_panic_sweep(which: u8) -> Vec { let mut failures = vec![]; for n in 0..400u32 { let (mut x, mut y) = (rig(which), rig(which)); - for r in [&x, &y] { - r.a.push(bz(&[(1, 1, 1), (2, 2, 1)])); - r.b.push(bz(&[(1, 3, 1), (3, 3, 1)])); + for r in [&mut x, &mut y] { + r.c.push(&r.a, bz(&[(1, 1, 1), (2, 2, 1)])); + r.c.push(&r.b, bz(&[(1, 3, 1), (3, 3, 1)])); } x.c.step(); y.c.step(); @@ -117,8 +123,8 @@ fn clone_panic_sweep(which: u8) -> Vec { let (v0, d0) = (x.view(), x.drain()); y.drain(); x.c.begin(); - x.a.push(bz(&[(1, 5, 1), (4, 4, 1), (2, 2, -1)])); - x.b.push(bz(&[(4, 6, 1), (1, 3, -1)])); + x.c.push(&x.a, bz(&[(1, 5, 1), (4, 4, 1), (2, 2, -1)])); + x.c.push(&x.b, bz(&[(4, 6, 1), (1, 3, -1)])); CLONE_TRIP.with(|t| t.set(Some(n))); let r = catch_unwind(AssertUnwindSafe(|| x.c.step())); let fired = CLONE_TRIP.with(|t| t.get()).is_none(); @@ -135,16 +141,13 @@ fn clone_panic_sweep(which: u8) -> Vec { )); continue; } - for r in [&x, &y] { + for r in [&mut x, &mut y] { // The keys the failed step touched come back, so state leaked by one shows. - r.a.push(bz(&[ - (1, 7, 1), - (4, 1, 1), - (1, 5, 1), - (4, 4, 1), - (2, 2, -1), - ])); - r.b.push(bz(&[(4, 2, 1), (1, 9, 1)])); + r.c.push( + &r.a, + bz(&[(1, 7, 1), (4, 1, 1), (1, 5, 1), (4, 4, 1), (2, 2, -1)]), + ); + r.c.push(&r.b, bz(&[(4, 2, 1), (1, 9, 1)])); } x.c.step(); y.c.step(); diff --git a/crates/zset/tests/cost.rs b/crates/zset/tests/cost.rs index 7cce4d98..0af9ad1f 100644 --- a/crates/zset/tests/cost.rs +++ b/crates/zset/tests/cost.rs @@ -53,7 +53,7 @@ fn an_eval_borrows_its_input_instead_of_cloning_it() { let (input, s) = c.input::(); let m = c.map(&s, |k| k.0); let sink = c.output(&m); - input.push(keys(N)); + c.push(&input, keys(N)); let before = clones(); c.step(); assert_eq!( @@ -61,7 +61,7 @@ fn an_eval_borrows_its_input_instead_of_cloning_it() { 0, "[piece] a step over {N} keys cloned keys" ); - assert_eq!(sink.take().len(), N as usize); + assert_eq!(c.take(&sink).len(), N as usize); } #[test] @@ -70,7 +70,7 @@ fn an_integrate_stream_holds_its_integral_twice_when_the_tick_is_whole() { let mut c = Circuit::new(); let (input, s) = c.input::(); let i = c.integrate(&s); - input.push(keys(N)); + c.push(&input, keys(N)); c.step(); // The input's piece (N), the integral (N) and the integrate stream's piece (N). assert_eq!( @@ -82,6 +82,22 @@ fn an_integrate_stream_holds_its_integral_twice_when_the_tick_is_whole() { assert_eq!(live(), base, "[leak] every key is dropped with the circuit"); } +#[test] +fn a_sink_adds_its_input_with_one_clone_per_key() { + let mut c = Circuit::new(); + let (input, s) = c.input::(); + let sink = c.output(&s); + c.push(&input, keys(N)); + let before = clones(); + c.step(); + assert_eq!( + clones() - before, + N as usize, + "[sink] one clone per key it sums" + ); + assert_eq!(c.take(&sink).len(), N as usize); +} + type E = (Counted, u8); #[derive(Clone, Copy)] @@ -105,25 +121,25 @@ fn clones_of_one_step(n: u32, mode: Mode) -> usize { let sinks = (c.output(&d), c.output(&dl), c.output(&j), c.output(&g)); let rows = |lo: u32, hi: u32| -> ZSet { (lo..hi).map(|i| ((Counted::new(i), 0), 1)).collect() }; - a.push(rows(0, n)); - b.push(rows(0, n)); + c.push(&a, rows(0, n)); + c.push(&b, rows(0, n)); c.step(); for z in [&sinks.0, &sinks.1] { - z.take(); + c.take(z); } - sinks.2.take(); - sinks.3.take(); + c.take(&sinks.2); + c.take(&sinks.3); // A small step after the load: a `delay` holds the previous tick's input, so a // transaction opened right after a big load would copy that load once. - a.push(rows(n, n + 2)); - b.push(rows(n, n + 2)); + c.push(&a, rows(n, n + 2)); + c.push(&b, rows(n, n + 2)); c.step(); let before = clones(); if !matches!(mode, Mode::Plain) { c.begin(); } - a.push(rows(n + 2, n + 4)); - b.push(rows(n + 2, n + 4)); + c.push(&a, rows(n + 2, n + 4)); + c.push(&b, rows(n + 2, n + 4)); c.step(); match mode { Mode::Plain => {} @@ -148,13 +164,13 @@ fn clones_of_one_continue(n: u32, txn: bool) -> usize { let mut c = Circuit::new(); let (input, s) = c.input::(); let _i = c.integrate(&s); - input.push((0..n).map(|k| (Counted::new(k), 1)).collect()); + c.push(&input, (0..n).map(|k| (Counted::new(k), 1)).collect()); c.step(); let before = clones(); if txn { c.begin(); } - input.push((n..n + 1).map(|k| (Counted::new(k), 1)).collect()); + c.push(&input, (n..n + 1).map(|k| (Counted::new(k), 1)).collect()); c.step_continue(); if txn { c.commit(); diff --git a/crates/zset/tests/rebuild.rs b/crates/zset/tests/rebuild.rs index 85958653..a1631691 100644 --- a/crates/zset/tests/rebuild.rs +++ b/crates/zset/tests/rebuild.rs @@ -12,7 +12,9 @@ fn pairs() -> impl Strategy> { } /// A pipeline using every operator; returns the input and a closure that drains all sinks. -fn build(c: &mut Circuit) -> (Input, Box String>) { +type Drain = Box String>; + +fn build(c: &mut Circuit) -> (Input, Drain) { let (input, s) = c.input::(); let d = c.delay(&s); let i = c.integrate(&s); @@ -40,18 +42,18 @@ fn build(c: &mut Circuit) -> (Input, Box String>) { c.output(&ddist), c.output(&fm), ); - let drain = move || { + let drain = move |c: &mut Circuit| { format!( "d={:?} i={:?} df={:?} j={:?} ij={:?} agg={:?} dagg={:?} ddist={:?} fm={:?}", - sinks.0.take(), - sinks.1.take(), - sinks.2.take(), - sinks.3.take(), - sinks.4.take(), - sinks.5.take(), - sinks.6.take(), - sinks.7.take(), - sinks.8.take() + c.take(&sinks.0), + c.take(&sinks.1), + c.take(&sinks.2), + c.take(&sinks.3), + c.take(&sinks.4), + c.take(&sinks.5), + c.take(&sinks.6), + c.take(&sinks.7), + c.take(&sinks.8) ) }; (input, Box::new(drain)) @@ -66,10 +68,10 @@ fn run( ) -> (String, String, String) { let mut c = Circuit::new(); let (input, drain) = build(&mut c); - input.push(prior.clone()); + c.push(&input, prior.clone()); c.step(); - let _ = drain(); - let mut job = Rebuild::new(input.clone(), snap.clone(), chunk); + let _ = drain(&mut c); + let mut job = Rebuild::new(input, snap.clone(), chunk); let mut done = false; for _ in 0..50 { let left = Cell::new(budget); @@ -83,12 +85,12 @@ fn run( } } assert!(done, "[rebuild] never finished"); - let at = drain(); - input.push(post.clone()); + let at = drain(&mut c); + c.push(&input, post.clone()); c.step(); - let next = drain(); + let next = drain(&mut c); c.step(); - (at, next, drain()) + (at, next, drain(&mut c)) } proptest! { @@ -113,7 +115,7 @@ fn current_after_a_chunked_rebuild_is_the_whole_ticks_value() { let i = c.integrate(&s); let mut job = Rebuild::new(input, snap.clone(), chunk); while job.step(&mut c, || false) == Progress::Pending {} - i.current() + c.current(&i) }; assert_eq!(cur(1), cur(10), "[current]"); } @@ -168,7 +170,7 @@ fn a_rebuild_stepped_on_a_foreign_circuit_is_refused_loudly() { fn a_rebuild_knows_the_circuit_its_input_came_from() { let (mut a, b) = (Circuit::new(), Circuit::new()); let (input, _s) = a.input::(); - let job = Rebuild::new(input.clone(), ZSet::new(), 1); + let job = Rebuild::new(input, ZSet::new(), 1); assert!( job.belongs_to(&a), "[owner] the circuit that made the input" @@ -182,7 +184,7 @@ fn an_interrupted_rebuild_stays_interrupted_and_a_new_job_loads_the_snapshot() { let mut c = Circuit::new(); let (input, s) = c.input::(); let i = c.integrate(&s); - let mut job = Rebuild::new(input.clone(), snap.clone(), 1); + let mut job = Rebuild::new(input, snap.clone(), 1); assert_eq!(job.step(&mut c, || true), Progress::Pending); c.step(); for _ in 0..3 { @@ -191,5 +193,5 @@ fn an_interrupted_rebuild_stays_interrupted_and_a_new_job_loads_the_snapshot() { } let mut fresh = Rebuild::new(input, snap, 1); assert_eq!(fresh.step(&mut c, || false), Progress::Done); - assert_eq!(i.current().len(), 3); + assert_eq!(c.current(&i).len(), 3); } diff --git a/crates/zset/tests/rollback.rs b/crates/zset/tests/rollback.rs index 46ffd72e..46c7153a 100644 --- a/crates/zset/tests/rollback.rs +++ b/crates/zset/tests/rollback.rs @@ -14,8 +14,8 @@ struct Rig { c: Circuit, a: Input

, b: Input

, - view: Box String>, - drain: Box String>, + view: Box String>, + drain: Box String>, } fn rig() -> Rig { @@ -44,34 +44,34 @@ fn rig() -> Rig { c.output(&ag), c.output(&pl), ); - let (s1, s2) = (sa.clone(), sb.clone()); - let view = move || { + let (s1, s2) = (sa, sb); + let view = move |c: &Circuit| { format!( "{:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?}", - s1.current(), - s2.current(), - ia.current(), - da.current(), - df.current(), - ds.current(), - j.current(), - ij.current(), - ag.current(), - pl.current(), - s1.current().len(), + c.current(&s1), + c.current(&s2), + c.current(&ia), + c.current(&da), + c.current(&df), + c.current(&ds), + c.current(&j), + c.current(&ij), + c.current(&ag), + c.current(&pl), + c.current(&s1).len(), ) }; - let drain = move || { + let drain = move |c: &mut Circuit| { format!( "{:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?}", - sinks.0.take(), - sinks.1.take(), - sinks.2.take(), - sinks.3.take(), - sinks.4.take(), - sinks.5.take(), - sinks.6.take(), - sinks.7.take(), + c.take(&sinks.0), + c.take(&sinks.1), + c.take(&sinks.2), + c.take(&sinks.3), + c.take(&sinks.4), + c.take(&sinks.5), + c.take(&sinks.6), + c.take(&sinks.7), ) }; Rig { @@ -85,8 +85,8 @@ fn rig() -> Rig { impl Rig { fn tick(&mut self, a: &Z

, b: &Z

) { - self.a.push(a.clone()); - self.b.push(b.clone()); + self.c.push(&self.a, a.clone()); + self.c.push(&self.b, b.clone()); self.c.step(); } } @@ -123,31 +123,31 @@ proptest! { for (a, b) in pre.iter().chain([&(preload, Z::new())]) { x.tick(a, b); y.tick(a, b); - prop_assert_eq!((x.view)(), (y.view)(), "[pre] state"); - prop_assert_eq!((x.drain)(), (y.drain)(), "[pre] output"); + prop_assert_eq!((x.view)(&x.c), (y.view)(&y.c), "[pre] state"); + prop_assert_eq!((x.drain)(&mut x.c), (y.drain)(&mut y.c), "[pre] output"); } - x.a.push(carry.clone()); - y.a.push(carry); - let (view_before, drain_before) = ((x.view)(), (x.drain)()); + x.c.push(&x.a, carry.clone()); + y.c.push(&y.a, carry); + let (view_before, drain_before) = ((x.view)(&x.c), (x.drain)(&mut x.c)); x.c.begin(); let mut a = bad.0.clone(); if fail { a += poison(); } - x.a.push(a); - x.b.push(bad.1.clone()); + x.c.push(&x.a, a); + x.c.push(&x.b, bad.1.clone()); let outcome = catch_unwind(AssertUnwindSafe(|| x.c.step())); prop_assert_eq!(outcome.is_err(), fail, "[panic] the poison step panics, the others do not"); x.c.rollback(); - prop_assert_eq!((x.view)(), view_before, "[rollback] every stream's current()"); - prop_assert_eq!((x.drain)(), drain_before, "[rollback] every sink"); + prop_assert_eq!((x.view)(&x.c), view_before, "[rollback] every stream's current()"); + prop_assert_eq!((x.drain)(&mut x.c), drain_before, "[rollback] every sink"); for (i, (a, b)) in post.iter().enumerate() { x.tick(a, b); y.tick(a, b); - prop_assert_eq!((x.view)(), (y.view)(), "[post] state, tick {}", i); - prop_assert_eq!((x.drain)(), (y.drain)(), "[post] output, tick {}", i); + prop_assert_eq!((x.view)(&x.c), (y.view)(&y.c), "[post] state, tick {}", i); + prop_assert_eq!((x.drain)(&mut x.c), (y.drain)(&mut y.c), "[post] output, tick {}", i); } } @@ -168,8 +168,8 @@ proptest! { x.c.commit(); } y.tick(a, b); - prop_assert_eq!((x.view)(), (y.view)(), "[commit] state, tick {}", i); - prop_assert_eq!((x.drain)(), (y.drain)(), "[commit] output, tick {}", i); + prop_assert_eq!((x.view)(&x.c), (y.view)(&y.c), "[commit] state, tick {}", i); + prop_assert_eq!((x.drain)(&mut x.c), (y.drain)(&mut y.c), "[commit] output, tick {}", i); } } } @@ -180,14 +180,14 @@ fn an_operator_built_inside_a_transaction_goes_back_to_empty() { let (input, s) = c.input::(); c.begin(); let i = c.integrate(&s); - input.push([(1u8, 1)].into_iter().collect()); + c.push(&input, [(1u8, 1)].into_iter().collect()); c.step(); c.rollback(); - input.push([(2u8, 1)].into_iter().collect()); + c.push(&input, [(2u8, 1)].into_iter().collect()); c.step(); let want: Z = [(2u8, 1)].into_iter().collect(); assert_eq!( - i.current(), + c.current(&i), want, "[built] the integral forgot the rolled-back tick" ); @@ -202,24 +202,24 @@ fn rollback_undoes_newest_first_so_no_undo_leaves_i64() { let (i, d) = (c.integrate(&s), c.distinct(&s)); let g = c.aggregate(&s, |_, vals| Some(vals.len() as i64)); let one = |w: i64| -> Z

{ [((1, 1), w)].into_iter().collect() }; - input.push(one(i64::MAX)); + c.push(&input, one(i64::MAX)); c.step(); - let before = (i.current(), d.current(), g.current()); + let before = (c.current(&i), c.current(&d), c.current(&g)); c.begin(); for w in [-1, 1] { - input.push(one(w)); + c.push(&input, one(w)); c.step(); } c.rollback(); assert_eq!( - (i.current(), d.current(), g.current()), + (c.current(&i), c.current(&d), c.current(&g)), before, "[order] streams" ); - input.push(one(-1)); + c.push(&input, one(-1)); c.step(); assert_eq!( - i.current(), + c.current(&i), one(i64::MAX - 1), "[order] the integral is back at MAX first" ); @@ -237,10 +237,10 @@ fn a_minimum_weight_applied_to_every_kind_of_state_is_taken_back() { let right: Z

= [((1, 2), 1)].into_iter().collect(); x.tick(&Z::new(), &right); y.tick(&Z::new(), &right); - assert_eq!((x.view)(), (y.view)(), "[min] state"); + assert_eq!((x.view)(&x.c), (y.view)(&y.c), "[min] state"); assert_eq!( - (x.drain)(), - (y.drain)(), + (x.drain)(&mut x.c), + (y.drain)(&mut y.c), "[min] output: the join saw no left side" ); } @@ -251,7 +251,7 @@ fn rollback_undoes_the_clock() { let (input, _s) = c.input::(); c.step(); c.begin(); - input.push([(1u8, 1)].into_iter().collect()); + c.push(&input, [(1u8, 1)].into_iter().collect()); c.step(); assert_eq!(c.ticks(), 2); c.rollback(); @@ -287,23 +287,27 @@ fn a_rollback_across_a_split_tick_restores_the_whole_value() { y.tick(&first, &Z::new()); // The tick is already split when the transaction opens, so every edge holds a `whole`. for r in [&mut x, &mut y] { - r.a.push([((3, 1), 1)].into_iter().collect()); + r.c.push(&r.a, [((3, 1), 1)].into_iter().collect()); r.c.step_continue(); } x.c.begin(); for k in [4u8, 5] { - x.a.push([((k, 1), 1)].into_iter().collect()); + x.c.push(&x.a, [((k, 1), 1)].into_iter().collect()); x.c.step_continue(); } x.c.rollback(); - assert_eq!((x.view)(), (y.view)(), "[split] state after the rollback"); assert_eq!( - (x.drain)(), - (y.drain)(), + (x.view)(&x.c), + (y.view)(&y.c), + "[split] state after the rollback" + ); + assert_eq!( + (x.drain)(&mut x.c), + (y.drain)(&mut y.c), "[split] output after the rollback" ); let next: Z

= [((5, 1), 1)].into_iter().collect(); x.tick(&next, &Z::new()); y.tick(&next, &Z::new()); - assert_eq!((x.view)(), (y.view)(), "[split] the tick after"); + assert_eq!((x.view)(&x.c), (y.view)(&y.c), "[split] the tick after"); } diff --git a/crates/zset/tests/rollback_model.rs b/crates/zset/tests/rollback_model.rs index afb09121..b12e70cc 100644 --- a/crates/zset/tests/rollback_model.rs +++ b/crates/zset/tests/rollback_model.rs @@ -120,7 +120,7 @@ impl World { struct S<'a>(&'a [Stream]); impl S<'_> { fn s(&self, i: usize) -> Stream { - self.0[i % self.0.len()].clone() + self.0[i % self.0.len()] } } let this = S(st); @@ -199,10 +199,10 @@ impl World { self.sinks.push(o); } fn view(&self) -> Vec { - self.streams.iter().map(|s| s.current()).collect() + self.streams.iter().map(|s| self.c.current(s)).collect() } - fn drain(&self) -> Vec { - self.sinks.iter().map(|s| s.take()).collect() + fn drain(&mut self) -> Vec { + self.sinks.iter().map(|s| self.c.take(s)).collect() } } @@ -274,7 +274,7 @@ fn act() -> impl Strategy { fn apply_plain(w: &mut World, a: &Act) { match a { - Act::Push(b, z) => w.inputs[*b as usize].push(z.clone()), + Act::Push(b, z) => w.c.push(&w.inputs[*b as usize], z.clone()), Act::Step(_) => w.c.step(), Act::Cont(_) => w.c.step_continue(), Act::Build(op) => w.build(op), @@ -282,7 +282,7 @@ fn apply_plain(w: &mut World, a: &Act) { } } -fn check(x: &World, y: &World, at: &str) -> Result<(), TestCaseError> { +fn check(x: &mut World, y: &mut World, at: &str) -> Result<(), TestCaseError> { prop_assert_eq!(x.c.ticks(), y.c.ticks(), "ticks {}", at); prop_assert_eq!(x.view(), y.view(), "view {}", at); prop_assert_eq!(x.drain(), y.drain(), "sinks {}", at); @@ -327,7 +327,7 @@ fn run(ops: &[Op], acts: &[Act]) -> Result<(), TestCaseError> { if steps_in_txn > 1 { MULTI_STEP_TXN.fetch_add(1, Relaxed); } - check(&x, &y, &at)?; + check(&mut x, &mut y, &at)?; } Act::Commit | Act::Rollback if open => { x.c.rollback(); @@ -348,7 +348,7 @@ fn run(ops: &[Op], acts: &[Act]) -> Result<(), TestCaseError> { y.build(&op); } } - check(&x, &y, &at)?; + check(&mut x, &mut y, &at)?; } Act::Commit | Act::Rollback => {} _ if open && failed => {} // only rollback after a failed step @@ -388,7 +388,7 @@ fn run(ops: &[Op], acts: &[Act]) -> Result<(), TestCaseError> { return Ok(()); } apply_plain(&mut y, a); - check(&x, &y, &at)?; + check(&mut x, &mut y, &at)?; } } } @@ -404,23 +404,23 @@ fn run(ops: &[Op], acts: &[Act]) -> Result<(), TestCaseError> { } else { RB_CLEAN.fetch_add(1, Relaxed); } - check(&x, &y, "final rollback")?; + check(&mut x, &mut y, "final rollback")?; } // ticks after: both behave the same on a fresh tick for i in 0..3u8 { let z: Z = [((i, i), 1), ((i + 1, 0), -1)].into_iter().collect(); let r = catch_unwind(AssertUnwindSafe(|| { - x.inputs[0].push(z.clone()); - x.inputs[1].push(z.clone()); + x.c.push(&x.inputs[0], z.clone()); + x.c.push(&x.inputs[1], z.clone()); x.c.step() })); if r.is_err() { return Ok(()); } - y.inputs[0].push(z.clone()); - y.inputs[1].push(z); + y.c.push(&y.inputs[0], z.clone()); + y.c.push(&y.inputs[1], z); y.c.step(); - check(&x, &y, &format!("post tick {i}"))?; + check(&mut x, &mut y, &format!("post tick {i}"))?; } Ok(()) } diff --git a/crates/zset/tests/rollback_rebuild.rs b/crates/zset/tests/rollback_rebuild.rs index 88326bc6..4f6ea41d 100644 --- a/crates/zset/tests/rollback_rebuild.rs +++ b/crates/zset/tests/rollback_rebuild.rs @@ -30,7 +30,7 @@ fn r1_rebuild_begun_before_txn_continued_inside_then_rolled_back_reports_interru let (input, s) = c.input::(); let i = c.integrate(&s); let snap = z(&[(1, 1), (2, 1), (3, 1), (4, 1)]); - let mut job = Rebuild::new(input.clone(), snap.clone(), 1); + let mut job = Rebuild::new(input, snap.clone(), 1); assert_eq!(job.step(&mut c, || true), Progress::Pending); c.begin(); assert_eq!(job.step(&mut c, || true), Progress::Pending); @@ -40,7 +40,7 @@ fn r1_rebuild_begun_before_txn_continued_inside_then_rolled_back_reports_interru let mut finished = None; if !interrupted { while job.step(&mut c, || false) == Progress::Pending {} - finished = Some(i.current()); + finished = Some(c.current(&i)); } assert!( interrupted, @@ -55,7 +55,7 @@ fn r2_rebuild_finished_inside_a_rolled_back_txn_is_reported_done_with_nothing_lo let (input, s) = c.input::(); let i = c.integrate(&s); let snap = z(&[(1, 1), (2, 1), (3, 1)]); - let mut job = Rebuild::new(input.clone(), snap.clone(), 1); + let mut job = Rebuild::new(input, snap.clone(), 1); c.begin(); assert_eq!(job.step(&mut c, || false), Progress::Done); c.rollback(); @@ -63,7 +63,7 @@ fn r2_rebuild_finished_inside_a_rolled_back_txn_is_reported_done_with_nothing_lo assert!( interrupted, "[r2] after rollback: not interrupted, integral {:?}", - i.current() + c.current(&i) ); } @@ -74,7 +74,7 @@ fn r3_an_interrupted_rebuild_resurrects_after_rollback_and_a_later_step() { let i = c.integrate(&s); c.step(); // ticks 1 let snap = z(&[(1, 1), (2, 1), (3, 1), (4, 1)]); - let mut job = Rebuild::new(input.clone(), snap.clone(), 1); + let mut job = Rebuild::new(input, snap.clone(), 1); c.begin(); assert_eq!(job.step(&mut c, || true), Progress::Pending); // began_at = 2 c.rollback(); // ticks back to 1 @@ -84,7 +84,7 @@ fn r3_an_interrupted_rebuild_resurrects_after_rollback_and_a_later_step() { let mut got = None; if !dead_after_step { while job.step(&mut c, || false) == Progress::Pending {} - got = Some(i.current()); + got = Some(c.current(&i)); } assert!( dead_after_step, @@ -98,10 +98,10 @@ fn r4_rebuild_whose_first_chunk_panics_in_a_txn_then_rolled_back() { let mut c = Circuit::new(); let (input, s) = c.input::(); let i = c.integrate(&s); - input.push(z(&[(1, i64::MAX)])); + c.push(&input, z(&[(1, i64::MAX)])); c.step(); let snap = z(&[(1, 1), (2, 1), (3, 1)]); - let mut job = Rebuild::new(input.clone(), snap.clone(), 1); + let mut job = Rebuild::new(input, snap.clone(), 1); c.begin(); let r = catch_unwind(AssertUnwindSafe(|| job.step(&mut c, || false))); assert!(r.is_err()); @@ -109,9 +109,9 @@ fn r4_rebuild_whose_first_chunk_panics_in_a_txn_then_rolled_back() { let interrupted = job.interrupted(&c); let mut got = None; if !interrupted { - input.push(z(&[(1, -5)])); // make room so the retry does not overflow + c.push(&input, z(&[(1, -5)])); // make room so the retry does not overflow while job.step(&mut c, || false) == Progress::Pending {} - got = Some(i.current()); + got = Some(c.current(&i)); } assert!( interrupted, @@ -120,36 +120,13 @@ fn r4_rebuild_whose_first_chunk_panics_in_a_txn_then_rolled_back() { ); } -#[test] -fn b1_a_closure_that_reads_a_stream_during_eval_does_not_panic() { - let mut c = Circuit::new(); - let (input, s) = c.input::(); - let s2 = s.clone(); - let i = c.integrate(&s); - let i2 = i.clone(); - // map reads its own source and another node's output during eval - let m = c.map(&s, move |k| { - k.wrapping_add(s2.current().len() as u8 + i2.current().len() as u8) - }); - let m2 = m.clone(); - // a later node reads an earlier node's output inside its own eval - let _f = c.filter(&i, move |_| m2.current().len() < 1000); - input.push(z(&[(1, 1), (2, 1)])); - c.step(); - c.begin(); - input.push(z(&[(3, 1)])); - c.step_continue(); - c.rollback(); - c.step(); -} - #[test] fn p1_a_chunk_whose_push_panics_in_a_txn_is_counted() { let mut c = Circuit::new(); let (input, s) = c.input::(); let i = c.integrate(&s); - input.push(z(&[(1, i64::MAX)])); - let mut job = Rebuild::new(input.clone(), z(&[(1, 1), (2, 1)]), 1); + c.push(&input, z(&[(1, i64::MAX)])); + let mut job = Rebuild::new(input, z(&[(1, 1), (2, 1)]), 1); c.begin(); let r = catch_unwind(AssertUnwindSafe(|| job.step(&mut c, || false))); assert!(r.is_err(), "the push overflows"); @@ -158,7 +135,7 @@ fn p1_a_chunk_whose_push_panics_in_a_txn_is_counted() { let mut got = None; if !interrupted { while job.step(&mut c, || false) == Progress::Pending {} - got = Some(i.current()); + got = Some(c.current(&i)); } assert!( interrupted, diff --git a/crates/zset/tests/threads.rs b/crates/zset/tests/threads.rs new file mode 100644 index 00000000..a90bace4 --- /dev/null +++ b/crates/zset/tests/threads.rs @@ -0,0 +1,171 @@ +//! A circuit moves between threads, one at a time (`context/code/zset.md`, "Threads"). +//! Evidence: `context/test-evidence/zset.md`. + +use std::thread; +use zset::{Circuit, Input, Output, Rebuild, Stream, ZSet}; + +type Z = ZSet; + +fn is_send() {} + +#[test] +fn a_circuit_and_every_handle_it_hands_out_is_send() { + is_send::(); + is_send::>>(); + is_send::>(); + is_send::>(); + is_send::>(); +} + +type Outs = (Output<(u8, i64)>, Output<(u8, u8)>); + +fn delta(n: u8) -> Z<(u8, u8)> { + [((n % 5, n), 1), (((n + 1) % 5, n / 2), -1)] + .into_iter() + .collect() +} + +/// The pipeline's outputs after `ticks` steps, collecting what `take` returns each time. +fn collect(c: &mut Circuit, i: &Input<(u8, u8)>, outs: &Outs, n: u8) -> String { + c.push(i, delta(n)); + c.step(); + format!("{:?} {:?}", c.take(&outs.0), c.take(&outs.1)) +} + +#[test] +fn a_circuit_built_on_one_thread_steps_on_another_and_returns() { + let build = || { + let mut c = Circuit::new(); + let (input, s) = c.input::<(u8, u8)>(); + let i = c.integrate(&s); + let d = c.distinct(&i); + let g = c.aggregate(&d, |_, vals| { + Some(vals.iter().map(|(v, w)| i64::from(*v) * w).sum::()) + }); + let f = c.filter(&s, |(k, _)| k % 2 == 0); + let m = c.delay(&f); + let outs = (c.output(&g), c.output(&m)); + (c, input, outs) + }; + let (mut one, i1, o1) = build(); + let want: Vec = (0..9).map(|n| collect(&mut one, &i1, &o1, n)).collect(); + + let (mut c, input, outs) = build(); + let mut got = Vec::new(); + for n in 0..9 { + // Each tick runs on a fresh thread: the circuit is moved in and moved out again. + let (back, line) = thread::spawn(move || { + let line = collect(&mut c, &input, &outs, n); + (c, line) + }) + .join() + .unwrap(); + c = back; + got.push(line); + } + assert_eq!(got, want, "[threads] each tick on its own thread"); +} + +#[test] +fn a_transaction_begun_on_one_thread_is_rolled_back_on_another() { + let mut c = Circuit::new(); + let (input, s) = c.input::(); + let i = c.integrate(&s); + let out = c.output(&i); + c.push(&input, Z::singleton(1, 1)); + c.step(); + let before = (c.current(&i), c.take(&out)); + c.begin(); + c.push(&input, Z::singleton(2, 1)); + c.step(); + let c = thread::spawn(move || { + let mut c = c; + c.rollback(); + c + }) + .join() + .unwrap(); + let mut c = c; + assert_eq!(c.current(&i), before.0, "[txn] the stream is as it was"); + assert_eq!( + c.take(&out), + Z::new(), + "[txn] the sink holds nothing of the step" + ); + assert_eq!(c.ticks(), 1); +} + +trait AmbiguousIfSync { + fn some_item() {} +} +impl AmbiguousIfSync<()> for T {} +struct IsSync; +impl AmbiguousIfSync for T {} + +// Fails to build (E0283) if `Circuit` becomes `Sync`. +#[test] +fn a_circuit_is_not_sync() { + let _ = >::some_item; +} + +type Site = fn(&mut Circuit, &Input, &Stream>, &Output); + +/// Two circuits of one shape, so a foreign handle indexes a live slot of the same type. +fn twin() -> (Circuit, Input, Stream>, Output) { + let mut c = Circuit::new(); + let (i, s) = c.input::(); + let o = c.output(&s); + (c, i, s, o) +} + +fn panic_text(f: impl FnOnce() + std::panic::UnwindSafe) -> Option { + let e = std::panic::catch_unwind(f).err()?; + e.downcast_ref::() + .cloned() + .or_else(|| e.downcast_ref::<&str>().map(|s| s.to_string())) +} + +#[test] +fn every_call_that_takes_a_handle_refuses_a_foreign_one() { + let sites: [(&str, Site); 10] = [ + ("current", |c, _, s, _| { + let _ = c.current(s); + }), + ("take", |c, _, _, o| drop(c.take(o))), + ("push", |c, i, _, _| c.push(i, Z::new())), + ("unary (map)", |c, _, s, _| { + let _ = c.map(s, |k| *k); + }), + ("binary, first operand", |c, _, s, _| { + let (_, own) = c.input::(); + let _ = c.plus(s, &own); + }), + ("binary, second operand", |c, _, s, _| { + let (_, own) = c.input::(); + let _ = c.plus(&own, s); + }), + ("output", |c, _, s, _| { + let _ = c.output(s); + }), + ("delay", |c, _, s, _| { + let _ = c.delay(s); + }), + ("integrate", |c, _, s, _| { + let _ = c.integrate(s); + }), + ("distinct", |c, _, s, _| { + let _ = c.distinct(s); + }), + ]; + for (name, site) in sites { + let (mut mine, ..) = twin(); + let (_, fi, fs, fo) = twin(); + let text = panic_text(std::panic::AssertUnwindSafe(|| { + site(&mut mine, &fi, &fs, &fo) + })); + assert!( + text.is_some_and(|t| t.contains("handle from another circuit")), + "[foreign] {name} accepted a handle from another circuit" + ); + } +}