diff --git a/context/code/zset-ffi.md b/context/code/zset-ffi.md index d9020e95..37fc794e 100644 --- a/context/code/zset-ffi.md +++ b/context/code/zset-ffi.md @@ -75,7 +75,8 @@ link, so they call the `extern "C"` functions as plain functions. shared cell rather than panicking. 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`, so the caller's `ctx` must be usable from every + 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 diff --git a/context/code/zset.md b/context/code/zset.md index 28e74b4c..62d66441 100644 --- a/context/code/zset.md +++ b/context/code/zset.md @@ -169,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. @@ -231,8 +231,8 @@ 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 `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.** @@ -246,8 +246,8 @@ 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.** `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 @@ -284,24 +284,30 @@ threads are out of scope, and nothing in `zset` spawns one. `Send` and `mpsc` co 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). It is not `Sync` by structure (boxed nodes, no lock); no test pins - that, because a negative needs a compile-fail test and the comment cap forbids a fence. + 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 (an `Any` type check; `tests/cost.rs` counts clones, which it leaves as they were, - and time was not measured). + 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. + 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, so `an_eval_borrows_its_input_instead_of_cloning_it` - still holds. + `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.md b/context/test-evidence/zset.md index f8aaf7b4..1b01ca13 100644 --- a/context/test-evidence/zset.md +++ b/context/test-evidence/zset.md @@ -607,15 +607,28 @@ so the file never runs; the same file's other tests move a circuit for real. 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's arena, which a thread-local would do. +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. -### a_handle_from_another_circuit_is_refused +### every_call_that_takes_a_handle_refuses_a_foreign_one -Ids index an arena, so a foreign handle used to be an out-of-range or mistyped slot. It is now -refused by name ("handle from another circuit"); `should_panic(expected)` pins the message, -so a downcast panic does not satisfy it. +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 8ae19684..37a1f9b7 100644 --- a/crates/zset-ffi/include/zset_ffi.h +++ b/crates/zset-ffi/include/zset_ffi.h @@ -20,7 +20,9 @@ * - 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, 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 + * 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/src/circuit.rs b/crates/zset/src/circuit.rs index 08e7f0d2..fdaa41e6 100644 --- a/crates/zset/src/circuit.rs +++ b/crates/zset/src/circuit.rs @@ -86,7 +86,7 @@ impl Slot { } /// The index of one value in a circuit's arena, typed so a read cannot name the wrong type. -struct Id(usize, PhantomData T>); +struct Id(usize, PhantomData T>); impl Clone for Id { fn clone(&self) -> Self { @@ -114,26 +114,18 @@ impl Arena { self.0[id.0].downcast_mut().expect("zset: arena type") } - /// One value to read and another to write, which are never the same entry. + /// One value to read and another, allocated after it, to write. fn read_write(&mut self, r: Id, w: Id) -> (&R, &mut W) { - assert_ne!( - r.0, w.0, - "zset: an entry cannot be read and written at once" + 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"; - if r.0 < w.0 { - let (lo, hi) = self.0.split_at_mut(w.0); - ( - lo[r.0].downcast_ref().expect(ty), - hi[0].downcast_mut().expect(ty), - ) - } else { - let (lo, hi) = self.0.split_at_mut(r.0); - ( - hi[0].downcast_ref().expect(ty), - lo[w.0].downcast_mut().expect(ty), - ) - } + ( + lo[r.0].downcast_ref().expect(ty), + hi[0].downcast_mut().expect(ty), + ) } } diff --git a/crates/zset/tests/cost.rs b/crates/zset/tests/cost.rs index 17e0e630..0af9ad1f 100644 --- a/crates/zset/tests/cost.rs +++ b/crates/zset/tests/cost.rs @@ -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)] diff --git a/crates/zset/tests/threads.rs b/crates/zset/tests/threads.rs index f165e4b5..a90bace4 100644 --- a/crates/zset/tests/threads.rs +++ b/crates/zset/tests/threads.rs @@ -95,10 +95,77 @@ fn a_transaction_begun_on_one_thread_is_rolled_back_on_another() { assert_eq!(c.ticks(), 1); } -#[test] -#[should_panic(expected = "handle from another circuit")] -fn a_handle_from_another_circuit_is_refused() { - let (mut a, mut b) = (Circuit::new(), Circuit::new()); - let (input, _) = a.input::(); - b.push(&input, Z::singleton(1, 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" + ); + } }