fiddlerwoaroof/improvise
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #362 from fiddlerwoaroof/dbsp/zset-send

Browse Source 
refactor(zset): Circuit is Send (arena, id handles)
This commit is contained in:
Ed Langley
2026-10-10 21:52:06 -07:00
committed by GitHub
parent 668235ad9b ce6e36112d
commit 08bacbe265
16 changed files with 1039 additions and 531 deletions
Download Patch File Download Diff File Expand all files Collapse all files
context/code/zset-ffi.md
+11 -4
View File
@@ -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
context/code/zset.md
+56 -14
View File
@@ -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<dyn Node>`. `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<RefCell>` 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<Box<dyn Any + Send>>`; 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<fn(T) -> T>`), as main's `Rc<RefCell<T>>` 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.
context/test-evidence/zset-ffi.md
+12
View File
@@ -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<Mutex>` and the
callback context an `unsafe impl Send` wrapper). Rival: restoring the `Rc<Cell>` 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.
context/test-evidence/zset.md
+38 -2
View File
@@ -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.
crates/zset-ffi/include/zset_ffi.h
+6 -3
View File
@@ -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.
crates/zset-ffi/src/engine.rs
+38 -26
View File
@@ -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<u8>;
@@ -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<Cell<Option<(Code, &'static str)>>>;
// 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<Mutex<Option<(Code, &'static str)>>>;
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<Bytes> {
let mut out: Vec<Bytes> = 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<Stream<BZ>> {
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<u32> {
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<u32> {
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<u32> {
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<BZ>) -> Stream<ZSet<(Bytes, Bytes)>> {
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<Input<Bytes>> {
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<T>(&mut self, f: impl FnOnce(&mut Circuit) -> T) -> Res<T> {
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,
crates/zset-ffi/tests/threads_abi.rs
+98
View File
@@ -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<T: Send>() {}
#[test]
fn the_opaque_handles_are_send() {
is_send::<ZsCircuit>();
is_send::<ZsRebuild>();
}
/// The caller's context: which threads ran the callback.
struct Seen(Mutex<Vec<ThreadId>>);
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) };
}
crates/zset/src/circuit.rs
+398 -278
View File
File diff suppressed because it is too large Load Diff
crates/zset/tests/circuit.rs
+45 -40
View File
@@ -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<K, L>(
build: impl FnOnce(&mut Circuit, &Stream<Z<K>>) -> Stream<Z<L>>,
reference: impl Fn(&Z<K>) -> Z<L>,
) 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::<K>();
@@ -42,10 +43,10 @@ fn matches_recompute<K, L>(
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<Z<(u8, u8, u8)>>
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::<u8>();
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::<u8>();
@@ -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<u8, u8>, Rc<Cell<usize>>);
struct Counting(MemTrace<u8, u8>, Arc<AtomicUsize>);
impl Trace<u8, u8> 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<u8>, chunk: usize, budget: usize) -> (Z<u8>, Z<u8>, Z<u8>
let (input, s) = c.input::<u8>();
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<u8>, chunk: usize, budget: usize) -> (Z<u8>, Z<u8>, Z<u8>
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::<u8>();
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::<u8>();
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]");
}
crates/zset/tests/clone_panic.rs
+22 -19
View File
@@ -55,7 +55,7 @@ fn rig(which: u8) -> Rig {
let mut c = Circuit::new();
let (a, sa) = c.input::<B>();
let (b, sb) = c.input::<B>();
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::<Vec<_>>(),
self.agg.current()
self.streams
.iter()
.map(|s| self.c.current(s))
.collect::<Vec<_>>(),
self.c.current(&self.agg)
)
}
fn drain(&self) -> String {
fn drain(&mut self) -> String {
format!(
"{:?}",
self.sinks.iter().map(|s| s.take()).collect::<Vec<_>>()
self.sinks
.iter()
.map(|s| self.c.take(s))
.collect::<Vec<_>>()
)
}
}
@@ -107,9 +113,9 @@ fn clone_panic_sweep(which: u8) -> Vec<String> {
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<String> {
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<String> {
));
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();
crates/zset/tests/cost.rs
+30 -14
View File
@@ -53,7 +53,7 @@ fn an_eval_borrows_its_input_instead_of_cloning_it() {
let (input, s) = c.input::<Counted>();
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::<Counted>();
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::<Counted>();
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<E> { (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::<Counted>();
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();
crates/zset/tests/rebuild.rs
+24 -22
View File
@@ -12,7 +12,9 @@ fn pairs() -> impl Strategy<Value = Z<u8>> {
}
/// A pipeline using every operator; returns the input and a closure that drains all sinks.
fn build(c: &mut Circuit) -> (Input<u8>, Box<dyn Fn() -> String>) {
type Drain = Box<dyn Fn(&mut Circuit) -> String>;
fn build(c: &mut Circuit) -> (Input<u8>, Drain) {
let (input, s) = c.input::<u8>();
let d = c.delay(&s);
let i = c.integrate(&s);
@@ -40,18 +42,18 @@ fn build(c: &mut Circuit) -> (Input<u8>, Box<dyn Fn() -> 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::<u8>();
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::<u8>();
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);
}
crates/zset/tests/rollback.rs
+62 -58
View File
@@ -14,8 +14,8 @@ struct Rig {
c: Circuit,
a: Input<P>,
b: Input<P>,
view: Box<dyn Fn() -> String>,
drain: Box<dyn Fn() -> String>,
view: Box<dyn Fn(&Circuit) -> String>,
drain: Box<dyn Fn(&mut Circuit) -> 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<P>, b: &Z<P>) {
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::<u8>();
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<u8> = [(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<P> { [((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<P> = [((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::<u8>();
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<P> = [((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");
}
crates/zset/tests/rollback_model.rs
+15 -15
View File
@@ -120,7 +120,7 @@ impl World {
struct S<'a>(&'a [Stream<Z>]);
impl S<'_> {
fn s(&self, i: usize) -> Stream<Z> {
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<Z> {
self.streams.iter().map(|s| s.current()).collect()
self.streams.iter().map(|s| self.c.current(s)).collect()
}
fn drain(&self) -> Vec<Z> {
self.sinks.iter().map(|s| s.take()).collect()
fn drain(&mut self) -> Vec<Z> {
self.sinks.iter().map(|s| self.c.take(s)).collect()
}
}
@@ -274,7 +274,7 @@ fn act() -> impl Strategy<Value = Act> {
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(())
}
crates/zset/tests/rollback_rebuild.rs
+13 -36
View File
@@ -30,7 +30,7 @@ fn r1_rebuild_begun_before_txn_continued_inside_then_rolled_back_reports_interru
let (input, s) = c.input::<u8>();
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::<u8>();
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::<u8>();
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::<u8>();
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::<u8>();
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,
crates/zset/tests/threads.rs
+171
View File
@@ -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<K> = ZSet<K>;
fn is_send<T: Send>() {}
#[test]
fn a_circuit_and_every_handle_it_hands_out_is_send() {
is_send::<Circuit>();
is_send::<Stream<Z<u8>>>();
is_send::<Input<u8>>();
is_send::<Output<u8>>();
is_send::<Rebuild<u8>>();
}
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::<i64>())
});
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<String> = (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::<u8>();
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<A> {
fn some_item() {}
}
impl<T: ?Sized> AmbiguousIfSync<()> for T {}
struct IsSync;
impl<T: ?Sized + Sync> AmbiguousIfSync<IsSync> for T {}
// Fails to build (E0283) if `Circuit` becomes `Sync`.
#[test]
fn a_circuit_is_not_sync() {
let _ = <Circuit as AmbiguousIfSync<_>>::some_item;
}
type Site = fn(&mut Circuit, &Input<u8>, &Stream<Z<u8>>, &Output<u8>);
/// Two circuits of one shape, so a foreign handle indexes a live slot of the same type.
fn twin() -> (Circuit, Input<u8>, Stream<Z<u8>>, Output<u8>) {
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let o = c.output(&s);
(c, i, s, o)
}
fn panic_text(f: impl FnOnce() + std::panic::UnwindSafe) -> Option<String> {
let e = std::panic::catch_unwind(f).err()?;
e.downcast_ref::<String>()
.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::<u8>();
let _ = c.plus(s, &own);
}),
("binary, second operand", |c, _, s, _| {
let (_, own) = c.input::<u8>();
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"
);
}
}