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

Merge dbsp/zset-edge into dbsp/sched

Browse Source 
# Conflicts:
#	context/test-evidence/zset.md
#	crates/zset/src/edge.rs
This commit is contained in:
Claude (dbsp-threads)
2026-10-10 21:33:54 -07:00
committed by Edward Langley
parent 2d3d54bebf 2ee4fee32f
commit 92c7875673
12 changed files with 1057 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
context/code/delivery.md
+1 -1
View File
@@ -243,4 +243,4 @@ pushes `Host::take_notices()`.
- **A pushed filter** is the client's `window`, applied on every request.
- **Hidden-item overlays, pruning queries and column widths** are ADR 008 and clause 10.
**The same edge, generalised in `zset`.** `zset::edge` (`context/code/zset.md`, "Edges") carries Z-set deltas between two circuits with the rules of this file (stamps, coalescing by summing, `target - held`, left epochs) and no log; this layer is unchanged and does not use it.
**The same edge, generalised in `zset`.** The `zset` crate's edge module, `crates/zset/src/edge.rs` (`context/code/zset.md`, "Edges") carries Z-set deltas between two circuits with the rules of this file (stamps, coalescing by summing, `target - held`, token-matched answers) and no log; this layer is unchanged and does not use it.
context/code/zset-ffi.md
+2 -1
View File
@@ -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
context/code/zset.md
+64 -37
View File
@@ -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,25 +284,31 @@ 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<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 (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<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, 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.
@@ -329,37 +335,58 @@ Nothing here spawns a thread; whoever owns each circuit calls `pump` and `poll`.
(`a_full_pipe_coalesces_exactly_and_keeps_the_range`).
4. **Order and gaps.** The receiver applies a delta only if it is in the epoch it joined and
`from == watermark`. `to <= watermark` is a duplicate and changes nothing. Anything else (a
gap, an overlap, an unjoined epoch, a frame that does not decode) sets `stale`, and `stale`
makes the next `poll` send a `Request`.
5. **A receiver always joins by correction, never by a delta.** A new `EdgeSource` is stale. The
sender answers a request with `Correction { epoch, through: seq, target: integral }` and the
receiver pushes `target - held` (delivery's rule), so the sender keeps its integral and the
receiver keeps what it has applied: two copies of the edge's state, one at each end. There is
no shared log as in delivery, so every answer is a full correction. Joining by a first delta
with `from == 0` was tried first, and the differential property found it wrong: a sender that
restarted with a non-empty integral at `seq 0` looks like a fresh one
(`two_circuits_on_an_edge_equal_one_circuit`, shrunk to a `RestartSink` after a rollback-free
history).
6. **A restart is a new epoch.** `EdgeSink::restart(epoch, seq, integral)` keeps the pipe and
codec. The receiver sees a delta from an epoch it has not joined, goes stale, and the answer
moves it onto the new epoch, adding the old one to `left` (the newest 16). Frames from a left
epoch change nothing, so a late reply from a dead sender cannot regress the receiver. A
receiver that restarts (`EdgeSource::restart`) is stale again.
7. **A lost final message is invisible to the receiver.** `connection_lost()` is the host's
gap, an overlap, a frame that does not decode) is a *mark*: `stale` is set and a counter of
troubles moves. `stale` makes the next `poll` send a `Request`.
5. **A receiver joins an incarnation only by the answer to its latest request.** A `Request`
carries a token (the receiver's counter, which `EdgeSource::restart` keeps), and the sender
echoes the token of the last request it read in its `Correction { epoch, through: seq, token,
target: integral }`. The receiver pushes `target - held` (delivery's rule). This carries over
delivery's third fix, "the host that answers a request is live" (`delivery.md` section 4): a
frame from an incarnation the receiver has not joined only marks it stale and never moves the
watermark, unless it is the answer to the request just sent. A reply to an older request, a
duplicate, or a late reply from a dead sender is therefore never a join. A correction from the
joined incarnation applies when it is ahead of the watermark, whoever asked. There is no shared
log as in delivery, so every answer is a full correction. Two earlier designs were wrong, each
found by a property: joining by a first delta with `from == 0` (a restarted sender with a
non-empty integral at `seq 0` looks fresh), and a `left` list of incarnations (a late reply
from an incarnation never joined sent the receiver back and shut the live sender out:
`a_late_reply_from_a_dead_incarnation_changes_nothing`). There is no `left` list now.
6. **What clears `stale`.** Only an answer, and only if it is not behind the watermark
(`an_answer_older_than_the_watermark_does_not_clear_stale`) and no mark has been made since
its request went out (`trouble_seen_after_a_request_keeps_its_answer_from_clearing_stale`);
an answer at the watermark says nothing newer exists
(`an_answer_at_the_watermark_says_nothing_newer_and_clears_stale`).
7. **A restart is a new epoch, and an epoch is never reused.** `EdgeSink::restart(epoch, seq,
integral)` keeps the pipe and codec. The first frame of an unjoined epoch makes the receiver
re-ask once for that epoch, because the request may have died with the sender that read it
(`a_new_epoch_frame_re_asks_once_after_a_lost_request`). A sender that reuses an epoch id at
a lower seq cannot be told from a reordered old frame, so the receiver ignores it as a
duplicate and is stranded until the sender restarts again: the caller must pass a fresh epoch.
A receiver that restarts (`EdgeSource::restart`) is stale again.
8. **A lost final message is invisible to the receiver.** `connection_lost()` is the host's
timeout: it marks the receiver stale and lets it ask again. The host calls it on its own local
monotonic timeout, never by comparing times across machines. It is also what retries a request
whose answer was lost. Requests carry the receiver's stamp and this sender does not read it.
8. **The wire form is a hook.** `EdgeCodec<K>` turns a `Frame<K>` into `Wire` and back, with
monotonic timeout, never by comparing times across machines, and it must be longer than the
round trip, or every answer is one request behind. Requests carry the receiver's stamp and
this sender does not read it.
9. **The wire form is a hook.** `EdgeCodec<K>` turns a `Frame<K>` into `Wire` and back, with
`CodecError` for what does not decode; `Direct` is the identity. `improvise-protocol` is to
implement it for its own key types in the lossless text of ADR 005, and `zset` stays free of
serde. The tests run every history through a hand-written text codec
(`the_same_histories_through_a_text_codec`).
9. **Threads.** `EdgeSink` and `EdgeSource` are `Send` when the pipe and codec are. The real
10. **Threads.** `EdgeSink` and `EdgeSource` are `Send` when the pipe and codec are. The real
two-thread test steps a producer on one thread and a consumer on another with a capacity-2 channel
(`a_circuit_on_each_of_two_threads_equals_one_circuit`). The property test simulates arbitrary
interleavings, losses and restarts by choosing the order of `commit`, `pump`, `poll` and step.
(`a_circuit_on_each_of_two_threads_equals_one_circuit`). Two properties simulate arbitrary
interleavings: `tests/edge.rs` over FIFO pipes that lose frames, and `tests/edge_net.rs` over a
network that drops, duplicates, garbles and reorders in-flight frames both ways, restarts either
end at any seq and times out (capacity 1 to 4). It checks after every action that the
receiver's applied state is the sender's integral at the stamp it claims, and that it never
moves to an older incarnation.
`poll` and `committed_from` panic inside a transaction: a rollback drops what a poll pushed,
and the stamps would not know (`a_poll_inside_a_transaction_is_refused`).
`committed` checks overflow before it changes anything, so a caught panic leaves the sink as
it was. `pump` clones the waiting delta each attempt against a full pipe; that is left.
Native threads are assumed; `mpsc` and `Send` compile on wasm32 and nothing runs there.
10. **Not built.** A sink with a log to answer a watermark with only the moved keys (delivery's
11. **Not built.** A sink with a log to answer a watermark with only the moved keys (delivery's
`Moved` correction). A pipe that blocks or notifies, because `try_send` and `try_recv` are
polling. Key-sharded data parallelism inside a circuit: its exchange step would be this same
edge, and nothing here forbids it.
context/test-evidence/zset.md
+83 -5
View File
@@ -607,14 +607,14 @@ 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,
@@ -664,10 +664,71 @@ why that assertion is there.
A new epoch holding `{1, 2}` plus one command `9`: the receiver applies only `{9}` and stamps
`(2, 1)`. Rival R3 applies the whole target and fails.
### a_late_frame_from_an_epoch_the_receiver_left_changes_nothing
### a_late_reply_from_a_dead_incarnation_changes_nothing
After epoch 2 is joined, a correction from epoch 1 with a larger `through` and a delta of epoch 1
are ignored and ask for nothing. Rival R6 (no `left` check on corrections) fails it.
The reviewer's case: epoch 100's reply to the first request is delayed, the sender restarts as
1001, the receiver asks again and joins it, and then the late reply arrives. The first design
(a `left` list) went back to epoch 100 and then dropped every 1001 frame. Now the late reply is
not the answer to the latest request, so it only marks stale. Rival N2 (a foreign correction
accepted) fails it.
### an_unsolicited_correction_from_another_incarnation_only_marks_stale
A correction with a token nobody asked for, from another epoch, leaves the stamp as it was and
makes the receiver ask. Rival N2 fails it.
### only_the_reply_to_the_latest_request_may_join_a_new_incarnation
Two requests are out (a timeout between them); the reply to the first, from another epoch, does
not join, and the reply to the second does. Rival N1 (any reply while asked is the answer) fails
it.
### an_answer_older_than_the_watermark_does_not_clear_stale
An answer through 1 to a receiver at 3 neither moves it back nor says it is current. Rival N6
(`<=` where `==` belongs) fails it.
### an_answer_at_the_watermark_says_nothing_newer_and_clears_stale
A repeat answer at the watermark clears `stale`; without the line the receiver asks for ever.
### trouble_seen_after_a_request_keeps_its_answer_from_clearing_stale
A gap seen after the request went out, then an answer through the old watermark: still stale.
Rival N5 (no comparison of marks) fails it.
### a_new_epoch_frame_re_asks_once_after_a_lost_request
The sender read a request and restarted before answering. Five frames from the new epoch cause
exactly one more request. Rival N3 (no re-ask) fails the first half and rival N4 (a re-ask for
every foreign frame) the second.
### a_poll_inside_a_transaction_is_refused
A poll outside a transaction followed by a rolled-back consumer step loses nothing; a poll inside
one panics. The reviewer's sequence stranded the receiver at `(1, 1)` holding nothing. Rival N8
(the assertion removed) fails it.
### a_zero_capacity_pipe_is_refused
`channel_pipes(0)` made a rendezvous channel that `try_send` could never use. Rival N10 fails it.
### a_committed_that_overflows_leaves_the_sink_as_it_was
A weight overflow in `committed` no longer leaves a bumped `seq` or half an integral behind.
Rival N9 (mutating first) fails it.
### the_receiver_is_never_wrong_over_a_hostile_network
The reviewer's generator as a permanent property: dropped, duplicated, garbled and reordered
frames in both directions, restarts at any seq, timeouts and receiver restarts, capacity 1 to 4.
After every action the receiver's applied state equals the sender's integral at the stamp it
claims and its epoch never goes down; after the network is reliable it catches up. Run at 20,000
cases three times, all green. Rivals N4 and N7 (a foreign delta moves the receiver) fail it.
### the_reviewers_late_reply_case_is_a_history
The late-reply sequence as a fixed history through the same World.
### an_unreadable_frame_asks_for_a_correction
@@ -694,3 +755,20 @@ the combined circuit's. Rival R1 fails it.
Three corrections waiting, `poll(c, 2)` reads two and the next reads one, so a unit's turn is
bounded by its budget. Rival: the bound removed from the loop fails the first assertion.
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
+3 -1
View File
@@ -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.
crates/zset/src/circuit.rs
+10 -18
View File
@@ -86,7 +86,7 @@ impl<T> Slot<T> {
}
/// The index of one value in a circuit's arena, typed so a read cannot name the wrong type.
struct Id<T>(usize, PhantomData<fn() -> T>);
struct Id<T>(usize, PhantomData<fn(T) -> T>);
impl<T> Clone for Id<T> {
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<R: 'static, W: 'static>(&mut self, r: Id<R>, w: Id<W>) -> (&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),
)
}
}
crates/zset/src/edge.rs
+92 -45
View File
@@ -2,7 +2,6 @@
//! is a channel or a network. See `context/code/zset.md`, "Edges".
use crate::{Circuit, Input, Output, ZSet};
use std::collections::VecDeque;
use std::sync::mpsc;
/// A point on the sender's causal clock: its incarnation, and how many commands it has committed.
@@ -26,13 +25,17 @@ pub struct Delta<K> {
pub struct Correction<K> {
pub epoch: u64,
pub through: u64,
/// The token of the request this answers.
pub token: u64,
pub target: ZSet<K>,
}
/// "Bring me up from here": carried with the receiver's watermark and never read by this sender.
/// "Bring me up from here". `have` is carried and never read by this sender; `token` is echoed
/// by the answer, and only the answer to the latest request may move the receiver to a new epoch.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Request {
pub have: Option<Stamp>,
pub token: u64,
}
/// Everything that crosses an edge, in both directions.
@@ -98,6 +101,7 @@ pub struct ChannelPipe<W> {
/// Two connected ends; each direction holds `cap` items.
pub fn channel_pipes<W: Send>(cap: usize) -> (ChannelPipe<W>, ChannelPipe<W>) {
assert!(cap > 0, "zset: a channel pipe holds at least one item");
let (atx, brx) = mpsc::sync_channel(cap);
let (btx, arx) = mpsc::sync_channel(cap);
(
@@ -142,7 +146,8 @@ pub struct EdgeSink<K, C, P> {
integral: ZSet<K>,
/// The commands `(sent, seq]`, summed; `sent == seq` means nothing waits.
waiting: Delta<K>,
owes_correction: bool,
/// The token of the latest request read and not yet answered.
owed: Option<u64>,
closed: bool,
traffic: u64,
}
@@ -168,7 +173,7 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
to: seq,
delta: ZSet::new(),
},
owes_correction: false,
owed: None,
closed: false,
traffic: 0,
}
@@ -188,7 +193,7 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
/// Whether any committed command has not been sent.
pub fn backlog(&self) -> bool {
self.waiting.to > self.waiting.from || self.owes_correction
self.waiting.to > self.waiting.from || self.owed.is_some()
}
/// The same pipe and codec, after a restart: a new incarnation at `seq` holding `integral`.
@@ -198,6 +203,8 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
/// One committed command changed the output by `delta` (empty is still a command).
pub fn committed(&mut self, delta: ZSet<K>) -> Stamp {
preflight(&self.integral, &delta);
preflight(&self.waiting.delta, &delta);
self.seq += 1;
self.integral += &delta;
self.waiting.delta += delta;
@@ -225,8 +232,8 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
match self.pipe.try_recv() {
Ok(wire) => {
self.traffic += 1;
if let Ok(Frame::Request(_)) = self.codec.decode(wire) {
self.owes_correction = true;
if let Ok(Frame::Request(r)) = self.codec.decode(wire) {
self.owed = Some(r.token);
}
}
Err(Recv::Empty) => break,
@@ -236,10 +243,11 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
if self.closed {
return Flow::Closed;
}
let frame = if self.owes_correction {
let frame = if let Some(token) = self.owed {
Frame::Correction(Correction {
epoch: self.epoch,
through: self.seq,
token,
target: self.integral.clone(),
})
} else if self.waiting.to > self.waiting.from {
@@ -250,7 +258,7 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
match self.pipe.try_send(self.codec.encode(&frame)) {
Ok(()) => {
self.traffic += 1;
self.owes_correction = false;
self.owed = None;
self.waiting = Delta {
epoch: self.epoch,
from: self.seq,
@@ -268,8 +276,12 @@ impl<K: Ord + Clone, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSink<K, C, P> {
}
}
/// How many incarnations a receiver remembers having left.
const LEFT: usize = 16;
/// Panics, before anything changes, if adding `d` to `z` would overflow a weight.
fn preflight<K: Ord>(z: &ZSet<K>, d: &ZSet<K>) {
for (k, w) in d.iter() {
crate::plus(z.weight(k), w);
}
}
/// The receiving end, beside the consumer circuit: it applies deltas to the circuit's input in
/// order, and asks for a correction when the stamps show something missing.
@@ -280,9 +292,14 @@ pub struct EdgeSource<K, C, P> {
joined: Option<u64>,
watermark: u64,
held: ZSet<K>,
left: VecDeque<u64>,
stale: bool,
/// A request is out. `token` is its token, and `marks` counted the troubles seen so far.
asked: bool,
token: u64,
marks: u64,
ask_marks: u64,
/// The foreign incarnation a re-ask was already made for.
noted: Option<u64>,
closed: bool,
traffic: u64,
}
@@ -297,9 +314,12 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
joined: None,
watermark: 0,
held: ZSet::new(),
left: VecDeque::new(),
stale: true,
asked: false,
token: 0,
marks: 0,
ask_marks: 0,
noted: None,
closed: false,
traffic: 0,
}
@@ -325,7 +345,16 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
/// The same pipe and codec, after the consumer lost its state: it holds nothing again.
pub fn restart(self, input: Input<K>) -> Self {
Self::new(self.codec, self.pipe, input)
let token = self.token;
EdgeSource {
token,
..Self::new(self.codec, self.pipe, input)
}
}
/// What has been applied to the circuit's input so far.
pub fn held(&self) -> &ZSet<K> {
&self.held
}
/// The host's timeout or reconnect: ask again from the watermark held.
@@ -334,9 +363,13 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
self.asked = false;
}
/// Applies up to `max` frames to the circuit's input, then asks for a correction if one is
/// wanted. Returns how many frames it read. The host steps the circuit.
/// Applies up to `max` frames, then asks if wanted; returns the frames read. Panics inside
/// a transaction, where a rollback would drop what was pushed.
pub fn poll(&mut self, circuit: &mut Circuit, max: usize) -> usize {
assert!(
!circuit.in_transaction(),
"zset: an edge applies frames only outside a transaction"
);
let mut read = 0;
while read < max && !self.closed {
match self.pipe.try_recv() {
@@ -347,7 +380,7 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
Ok(Frame::Delta(d)) => self.on_delta(circuit, d),
Ok(Frame::Correction(c)) => self.on_correction(circuit, c),
Ok(Frame::Request(_)) => {}
Err(_) => self.stale = true,
Err(_) => self.mark(),
}
}
Err(Recv::Empty) => break,
@@ -358,8 +391,20 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
read
}
fn has_left(&self, epoch: u64) -> bool {
self.left.contains(&epoch)
/// Something is wrong or unknown: ask, and no answer to an earlier request may say otherwise.
fn mark(&mut self) {
self.stale = true;
self.marks += 1;
}
/// A frame from an unjoined epoch never moves the watermark; the first one per epoch
/// re-asks, since the request may have died with the sender that read it.
fn foreign(&mut self, epoch: u64) {
self.mark();
if self.noted != Some(epoch) {
self.noted = Some(epoch);
self.asked = false;
}
}
fn apply(&mut self, circuit: &mut Circuit, delta: ZSet<K>) {
@@ -368,54 +413,56 @@ impl<K: Ord + Clone + Send + 'static, C: EdgeCodec<K>, P: Pipe<C::Wire>> EdgeSou
}
fn on_delta(&mut self, circuit: &mut Circuit, d: Delta<K>) {
if self.has_left(d.epoch) {
if self.joined != Some(d.epoch) {
return self.foreign(d.epoch);
}
if d.to <= self.watermark {
return;
}
let known = self.joined == Some(d.epoch);
if known && d.to <= self.watermark {
return;
}
if known && d.from == self.watermark {
if d.from == self.watermark {
self.watermark = d.to;
self.apply(circuit, d.delta);
} else {
self.stale = true;
self.mark();
}
}
fn on_correction(&mut self, circuit: &mut Circuit, c: Correction<K>) {
if self.has_left(c.epoch) {
return;
}
self.asked = false;
let answer = self.asked && c.token == self.token;
let known = self.joined == Some(c.epoch);
if known && c.through <= self.watermark {
self.stale &= c.through != self.watermark;
return;
if !known && !answer {
return self.foreign(c.epoch);
}
if !known && let Some(old) = self.joined.replace(c.epoch) {
self.left.push_back(old);
if self.left.len() > LEFT {
self.left.pop_front();
}
if answer {
self.asked = false;
self.noted = None;
}
if !known || c.through > self.watermark {
self.joined = Some(c.epoch);
self.watermark = c.through;
let delta = &c.target - &self.held;
self.apply(circuit, delta);
}
if answer && c.through == self.watermark && self.marks == self.ask_marks {
self.stale = false;
}
let delta = &c.target - &self.held;
self.watermark = c.through;
self.stale = false;
self.apply(circuit, delta);
}
fn ask(&mut self) {
if !self.stale || self.asked || self.closed {
return;
}
let wire = self
.codec
.encode(&Frame::Request(Request { have: self.stamp() }));
let token = self.token + 1;
let wire = self.codec.encode(&Frame::Request(Request {
have: self.stamp(),
token,
}));
match self.pipe.try_send(wire) {
Ok(()) => {
self.asked = true;
self.traffic += 1;
self.token = token;
self.ask_marks = self.marks;
}
Err(Refused::Full(_)) => {}
Err(Refused::Closed) => self.closed = true,
crates/zset/tests/cost.rs
+16
View File
@@ -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)]
crates/zset/tests/edge.rs
+297 -114
View File
@@ -1,13 +1,16 @@
//! The edge between two circuits: a producer and a consumer joined by stamped, coalescing
//! deltas equal one circuit running both. Evidence: `context/test-evidence/zset.md`.
mod support;
use proptest::prelude::*;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::thread;
use support::Text;
use zset::{
ChannelPipe, Circuit, CodecError, Correction, Delta, Direct, EdgeCodec, EdgeSink, EdgeSource,
Flow, Frame, Input, Output, Pipe, Recv, Refused, Request, Stamp, ZSet, channel_pipes,
ChannelPipe, Circuit, Correction, Delta, Direct, EdgeCodec, EdgeSink, EdgeSource, Flow, Frame,
Input, Output, Pipe, Recv, Refused, Stamp, ZSet, channel_pipes,
};
type Z = ZSet<u8>;
@@ -98,80 +101,6 @@ fn fault<W: Send>(
)
}
/// A text wire form, to run the same histories through an encoding.
#[derive(Clone, Copy)]
struct Text;
fn zs(z: &Z) -> String {
z.iter()
.map(|(k, w)| format!("{k}:{w}"))
.collect::<Vec<_>>()
.join(",")
}
fn unzs(s: &str) -> Result<Z, CodecError> {
let bad = || CodecError(format!("bad entries {s:?}"));
s.split(',')
.filter(|p| !p.is_empty())
.map(|p| {
let (k, w) = p.split_once(':').ok_or_else(bad)?;
Ok((
k.parse::<u8>().map_err(|_| bad())?,
w.parse::<i64>().map_err(|_| bad())?,
))
})
.collect()
}
impl EdgeCodec<u8> for Text {
type Wire = String;
fn encode(&self, f: &Frame<u8>) -> String {
match f {
Frame::Delta(d) => format!("D {} {} {} {}", d.epoch, d.from, d.to, zs(&d.delta)),
Frame::Correction(c) => format!("C {} {} {}", c.epoch, c.through, zs(&c.target)),
Frame::Request(r) => match r.have {
Some(s) => format!("R {} {}", s.epoch, s.seq),
None => "R".to_string(),
},
}
}
fn decode(&self, w: String) -> Result<Frame<u8>, CodecError> {
let bad = || CodecError(format!("bad frame {w:?}"));
let mut it = w.splitn(5, ' ');
let num = |s: Option<&str>| s.and_then(|s| s.parse::<u64>().ok()).ok_or_else(bad);
match it.next() {
Some("D") => Ok(Frame::Delta(Delta {
epoch: num(it.next())?,
from: num(it.next())?,
to: num(it.next())?,
delta: unzs(it.next().ok_or_else(bad)?)?,
})),
Some("C") => {
let mut it = w.splitn(4, ' ');
it.next();
Ok(Frame::Correction(Correction {
epoch: num(it.next())?,
through: num(it.next())?,
target: unzs(it.next().ok_or_else(bad)?)?,
}))
}
Some("R") => {
let have = match it.next() {
None => None,
e => Some(Stamp {
epoch: num(e)?,
seq: num(it.next())?,
}),
};
Ok(Frame::Request(Request { have }))
}
_ => Err(bad()),
}
}
}
type Sink<C> = EdgeSink<u8, C, FaultPipe<<C as EdgeCodec<u8>>::Wire>>;
type Source<C> = EdgeSource<u8, C, FaultPipe<<C as EdgeCodec<u8>>::Wire>>;
@@ -504,11 +433,8 @@ fn a_duplicate_delta_is_applied_once() {
let (i, s) = c.input::<u8>();
let o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
let join = Frame::Correction(Correction {
epoch: 1,
through: 0,
target: Z::new(),
});
src.poll(&mut c, 10);
let join = answer(&mut b, 1, 0, Z::new());
let d = Frame::Delta(Delta {
epoch: 1,
from: 0,
@@ -543,42 +469,300 @@ fn a_restarted_sender_is_corrected_not_trusted() {
assert_eq!(src.stamp(), Some(Stamp { epoch: 2, seq: 1 }));
}
/// Answers the receiver's latest request in `b` with a correction that echoes its token.
fn answer(b: &mut ChannelPipe<Frame<u8>>, epoch: u64, through: u64, target: Z) -> Frame<u8> {
let mut token = None;
while let Ok(f) = b.try_recv() {
if let Frame::Request(r) = f {
token = Some(r.token);
}
}
Frame::Correction(Correction {
epoch,
through,
token: token.expect("a request was waiting"),
target,
})
}
fn reply(b: &mut ChannelPipe<Frame<u8>>, epoch: u64, through: u64, target: Z) {
let f = answer(b, epoch, through, target);
b.try_send(f).unwrap();
}
#[test]
fn a_late_frame_from_an_epoch_the_receiver_left_changes_nothing() {
fn a_late_reply_from_a_dead_incarnation_changes_nothing() {
let (a, mut b) = channel_pipes::<Frame<u8>>(8);
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
let correction = |epoch, through, target| {
Frame::Correction(Correction {
epoch,
through,
target,
})
};
b.try_send(correction(1, 1, z(&[(1, 1)]))).unwrap();
b.try_send(correction(2, 3, z(&[(1, 1), (5, 1)]))).unwrap();
src.poll(&mut c, 10);
assert_eq!(src.stamp(), Some(Stamp { epoch: 2, seq: 3 }));
// A reply from the dead host that outlived it, and a delta that follows it.
b.try_send(correction(1, 9, z(&[(1, 1), (7, 1)]))).unwrap();
// Epoch 100 reads the first request; its reply is delayed. The sender restarts as 1001.
let late = answer(&mut b, 100, 1, z(&[(9, 1)]));
src.connection_lost();
src.poll(&mut c, 10);
reply(&mut b, 1001, 1, z(&[(5, 1)]));
src.poll(&mut c, 10);
assert_eq!(
src.stamp(),
Some(Stamp {
epoch: 1001,
seq: 1
})
);
b.try_send(late).unwrap();
b.try_send(Frame::Delta(Delta {
epoch: 1,
from: 9,
to: 10,
delta: z(&[(8, 1)]),
epoch: 1001,
from: 1,
to: 2,
delta: z(&[(6, 1)]),
}))
.unwrap();
src.poll(&mut c, 10);
c.step();
assert_eq!(
c.take(&o),
z(&[(1, 1), (5, 1)]),
"[left] epoch 1's late frames are refused"
src.stamp(),
Some(Stamp {
epoch: 1001,
seq: 2
}),
"[late] the reply of an incarnation never joined does not move the receiver back"
);
assert_eq!(
c.take(&o),
z(&[(5, 1), (6, 1)]),
"[late] and 1001 is not shut out"
);
}
#[test]
fn an_unsolicited_correction_from_another_incarnation_only_marks_stale() {
let (a, mut b) = channel_pipes::<Frame<u8>>(8);
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let _o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
src.poll(&mut c, 10);
reply(&mut b, 1, 1, Z::new());
src.poll(&mut c, 10);
assert!(!src.stale());
let wrong = Frame::Correction(Correction {
epoch: 2,
through: 9,
token: 77,
target: z(&[(3, 1)]),
});
b.try_send(wrong).unwrap();
src.poll(&mut c, 10);
assert_eq!(
src.stamp(),
Some(Stamp { epoch: 1, seq: 1 }),
"[foreign] not moved"
);
assert!(src.stale(), "[foreign] but it asks");
}
#[test]
fn only_the_reply_to_the_latest_request_may_join_a_new_incarnation() {
let (a, mut b) = channel_pipes::<Frame<u8>>(8);
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let _o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
src.poll(&mut c, 10);
let late = answer(&mut b, 100, 1, z(&[(9, 1)]));
src.connection_lost();
src.poll(&mut c, 10);
b.try_send(late).unwrap();
src.poll(&mut c, 10);
assert_eq!(
src.stamp(),
None,
"[token] the reply to the first request is not the answer"
);
reply(&mut b, 1001, 1, z(&[(5, 1)]));
src.poll(&mut c, 10);
assert_eq!(
src.stamp(),
Some(Stamp {
epoch: 1001,
seq: 1
})
);
}
#[test]
fn an_answer_older_than_the_watermark_does_not_clear_stale() {
let (a, mut b) = channel_pipes::<Frame<u8>>(8);
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let _o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
src.poll(&mut c, 10);
reply(&mut b, 1, 3, Z::new());
src.poll(&mut c, 10);
src.connection_lost();
src.poll(&mut c, 10);
reply(&mut b, 1, 1, Z::new());
src.poll(&mut c, 10);
assert_eq!(
src.stamp(),
Some(Stamp { epoch: 1, seq: 3 }),
"[older] never moves back"
);
assert!(
src.stale(),
"[older] and it does not say the receiver is current"
);
}
#[test]
fn traffic_never_goes_down_across_a_restart_of_either_end() {
let (mut sink, mut src, mut c, _, _) = mirror(8);
sink.committed(z(&[(1, 1)]));
drive(&mut sink, &mut src, &mut c);
let (s, r) = (sink.traffic(), src.traffic());
assert!(s > 0 && r > 0, "[setup] frames moved both ways");
let sink = sink.restart(2, 0, Z::new());
let mut c2 = Circuit::new();
let (i2, _) = c2.input::<u8>();
let src = src.restart(i2);
assert_eq!((sink.traffic(), src.traffic()), (s, r), "[restart] counters kept");
}
#[test]
fn a_new_epoch_frame_re_asks_once_after_a_lost_request() {
let (a, mut b) = channel_pipes::<Frame<u8>>(64);
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let _o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
src.poll(&mut c, 8);
assert!(
matches!(b.try_recv(), Ok(Frame::Request(_))),
"[first] the request"
);
// The sender read it and restarted before answering.
let mut asks = 0;
for n in 0..5u64 {
b.try_send(Frame::Delta(Delta {
epoch: 2,
from: n,
to: n + 1,
delta: Z::new(),
}))
.unwrap();
src.poll(&mut c, 8);
while let Ok(Frame::Request(_)) = b.try_recv() {
asks += 1;
}
}
assert_eq!(
asks, 1,
"[reask] once for the new epoch, not once per frame"
);
}
#[test]
fn an_answer_at_the_watermark_says_nothing_newer_and_clears_stale() {
let (a, mut b) = channel_pipes::<Frame<u8>>(8);
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let _o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
src.poll(&mut c, 8);
reply(&mut b, 1, 1, Z::new());
src.poll(&mut c, 8);
src.connection_lost();
src.poll(&mut c, 8);
reply(&mut b, 1, 1, Z::new());
src.poll(&mut c, 8);
assert!(
!src.stale(),
"[dup-answer] through == watermark clears stale"
);
}
#[test]
fn trouble_seen_after_a_request_keeps_its_answer_from_clearing_stale() {
let (a, mut b) = channel_pipes::<Frame<u8>>(8);
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let _o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
src.poll(&mut c, 8);
reply(&mut b, 1, 1, Z::new());
src.poll(&mut c, 8);
src.connection_lost();
src.poll(&mut c, 8);
let reply = answer(&mut b, 1, 1, Z::new());
// A gap seen after the request went out: the reply was cut before it.
b.try_send(Frame::Delta(Delta {
epoch: 1,
from: 4,
to: 5,
delta: Z::new(),
}))
.unwrap();
b.try_send(reply).unwrap();
src.poll(&mut c, 8);
assert!(src.stale(), "[marks] the reply predates the gap");
}
#[test]
fn a_poll_inside_a_transaction_is_refused() {
let (a, mut b) = channel_pipes::<Frame<u8>>(8);
let mut c = Circuit::new();
let (i, s) = c.input::<u8>();
let o = c.output(&s);
let mut src = EdgeSource::new(Direct, a, i);
src.poll(&mut c, 8);
reply(&mut b, 1, 1, z(&[(1, 1)]));
// Outside a transaction a rollback of the consumer's step keeps what was polled.
src.poll(&mut c, 8);
c.begin();
c.step();
c.rollback();
c.step();
assert_eq!(
c.take(&o),
z(&[(1, 1)]),
"[txn] a step rolled back after poll loses nothing"
);
c.begin();
let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| src.poll(&mut c, 8)));
assert!(
r.is_err(),
"[txn] a poll inside a transaction would be dropped by rollback"
);
}
#[test]
#[should_panic(expected = "at least one item")]
fn a_zero_capacity_pipe_is_refused() {
let _ = channel_pipes::<u8>(0);
}
#[test]
fn a_committed_that_overflows_leaves_the_sink_as_it_was() {
let (mut sink, mut src, mut c, _, _) = mirror(8);
sink.committed(z(&[(1, i64::MAX)]));
let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
sink.committed(z(&[(2, 1), (1, 1)]))
}));
assert!(r.is_err(), "the weight overflows");
assert_eq!(
sink.stamp(),
Stamp { epoch: 1, seq: 1 },
"[overflow] no stamp consumed"
);
drive(&mut sink, &mut src, &mut c);
assert_eq!(
src.held(),
&z(&[(1, i64::MAX)]),
"[overflow] and the key 2 did not slip in"
);
assert_eq!(src.stamp(), Some(Stamp { epoch: 2, seq: 3 }));
assert!(!src.stale(), "[left] and they do not ask for anything");
}
#[test]
@@ -594,7 +778,7 @@ fn an_unreadable_frame_asks_for_a_correction() {
src.stale(),
"[initial] a new receiver asks for a correction"
);
b.try_send("C 1 1 1:1".to_string()).unwrap();
b.try_send("C 1 1 1 1:1".to_string()).unwrap();
src.poll(&mut c, 10);
assert!(!src.stale());
b.try_send("garbage".to_string()).unwrap();
@@ -608,27 +792,26 @@ fn an_unreadable_frame_asks_for_a_correction() {
fn a_rolled_back_command_does_not_cross() {
let (mut sink, mut src, mut c, out, _) = mirror(8);
let (mut p, p_in, p_out) = producer();
// The host's command fails and rolls back; the sink is not asked.
p.begin();
p.push(&p_in, z(&[(1, 1)]));
p.step();
p.rollback();
assert_eq!(
sink.stamp(),
Stamp { epoch: 1, seq: 0 },
"[stamp] nothing consumed a number"
);
assert!(!sink.backlog(), "[send] nothing waits");
drive(&mut sink, &mut src, &mut c);
assert_eq!(c.take(&out), Z::new());
// The committed retry does cross, and is the first stamp.
// The retry is a different command and commits.
p.begin();
p.push(&p_in, z(&[(1, 1)]));
p.push(&p_in, z(&[(2, 1)]));
p.step();
p.commit();
assert_eq!(
sink.committed_from(&mut p, &p_out),
Stamp { epoch: 1, seq: 1 }
);
drive(&mut sink, &mut src, &mut c);
assert_eq!(
c.take(&out),
z(&[(2, 1)]),
"[retry] only the committed command crossed"
);
}
#[test]
crates/zset/tests/edge_net.rs
+333
View File
@@ -0,0 +1,333 @@
//! An edge over a network that drops, duplicates, garbles and reorders frames.
//! Evidence: `context/test-evidence/zset.md`.
mod support;
use proptest::prelude::*;
use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex};
use support::Text;
use zset::{Circuit, EdgeSink, EdgeSource, Output, Pipe, Recv, Refused, Stamp, ZSet};
type Z = ZSet<u8>;
/// One direction of the network: frames in flight (which the test may reorder, drop, duplicate
/// or garble) and frames delivered (which `try_recv` takes in order).
#[derive(Default)]
struct Lane {
flight: Vec<String>,
ready: VecDeque<String>,
}
#[derive(Default)]
struct Net {
lanes: [Lane; 2],
cap: usize,
}
type Shared = Arc<Mutex<Net>>;
/// One end: sends into lane `out`, receives from the other one.
struct End {
net: Shared,
out: usize,
}
impl Pipe<String> for End {
fn try_send(&mut self, wire: String) -> Result<(), Refused<String>> {
let mut n = self.net.lock().unwrap();
let cap = n.cap;
let lane = &mut n.lanes[self.out];
if lane.flight.len() + lane.ready.len() >= cap {
return Err(Refused::Full(wire));
}
lane.flight.push(wire);
Ok(())
}
fn try_recv(&mut self) -> Result<String, Recv> {
let mut n = self.net.lock().unwrap();
n.lanes[1 - self.out].ready.pop_front().ok_or(Recv::Empty)
}
}
fn net(cap: usize) -> (Shared, End, End) {
let shared: Shared = Arc::new(Mutex::new(Net {
cap,
..Net::default()
}));
let sink_end = End {
net: shared.clone(),
out: 0,
};
let src_end = End {
net: shared.clone(),
out: 1,
};
(shared, sink_end, src_end)
}
fn z(entries: &[(u8, i64)]) -> Z {
entries.iter().copied().collect()
}
struct World {
net: Shared,
sink: EdgeSink<u8, Text, End>,
src: EdgeSource<u8, Text, End>,
cons: Circuit,
out: Output<u8>,
got: Z,
total: Z,
epoch: u64,
seq: u64,
history: HashMap<(u64, u64), Z>,
joined_max: u64,
}
impl World {
fn new(cap: usize) -> Self {
let (shared, a, b) = net(cap);
let mut cons = Circuit::new();
let (i, s) = cons.input::<u8>();
let out = cons.output(&s);
let mut history = HashMap::new();
history.insert((1, 0), Z::new());
World {
net: shared,
sink: EdgeSink::new(Text, a, 1),
src: EdgeSource::new(Text, b, i),
cons,
out,
got: Z::new(),
total: Z::new(),
epoch: 1,
seq: 0,
history,
joined_max: 0,
}
}
fn commit(&mut self, d: Z) {
self.total += &d;
self.seq += 1;
self.sink.committed(d);
self.history
.insert((self.epoch, self.seq), self.total.clone());
}
fn restart_sink(&mut self, seq: u64) {
self.epoch += 1;
self.seq = seq;
let old = std::mem::replace(
&mut self.sink,
EdgeSink::new(
Text,
End {
net: self.net.clone(),
out: 0,
},
0,
),
);
self.sink = old.restart(self.epoch, seq, self.total.clone());
self.history.insert((self.epoch, seq), self.total.clone());
}
fn restart_source(&mut self) {
let mut cons = Circuit::new();
let (i, s) = cons.input::<u8>();
self.out = cons.output(&s);
let old = std::mem::replace(
&mut self.src,
EdgeSource::new(
Text,
End {
net: self.net.clone(),
out: 1,
},
i,
),
);
self.src = old.restart(i);
self.cons = cons;
self.got = Z::new();
self.joined_max = 0;
}
fn step(&mut self) {
self.cons.step();
self.got += self.cons.take(&self.out);
}
fn lane(&mut self, dir: usize, idx: usize, f: impl FnOnce(&mut Lane, usize)) {
let mut n = self.net.lock().unwrap();
let lane = &mut n.lanes[dir];
if !lane.flight.is_empty() {
let at = idx % lane.flight.len();
f(lane, at);
}
}
/// The two safety rules, after every action.
fn check(&mut self, when: &str) {
if let Some(Stamp { epoch, seq }) = self.src.stamp() {
assert!(
epoch >= self.joined_max,
"[older] {when}: the receiver went back from epoch {} to {epoch}",
self.joined_max
);
self.joined_max = epoch;
let want = self.history.get(&(epoch, seq));
assert_eq!(
Some(self.src.held()),
want,
"[claim] {when}: held is not the sender's integral at ({epoch}, {seq})"
);
}
}
fn settle(&mut self) {
for _ in 0..400 {
let silent = {
let mut n = self.net.lock().unwrap();
let silent = n
.lanes
.iter()
.all(|l| l.flight.is_empty() && l.ready.is_empty());
for lane in &mut n.lanes {
let flight = std::mem::take(&mut lane.flight);
lane.ready.extend(flight);
}
silent
};
self.sink.pump();
let read = self.src.poll(&mut self.cons, 8);
self.step();
self.check("settle");
let quiet = {
let n = self.net.lock().unwrap();
n.lanes
.iter()
.all(|l| l.flight.is_empty() && l.ready.is_empty())
};
if quiet
&& !self.sink.backlog()
&& !self.src.stale()
&& self.src.stamp() == Some(self.sink.stamp())
{
break;
}
if silent && read == 0 {
// Nothing was in flight: the reply is lost, not late. Only now does the timeout fire.
self.src.connection_lost();
}
}
assert_eq!(
self.src.stamp(),
Some(self.sink.stamp()),
"[settle] never caught up"
);
self.step();
assert_eq!(self.got, self.total, "[settle] the consumer's integral");
}
}
#[derive(Clone, Debug)]
enum Act {
Commit(Vec<(u8, i64)>),
Pump,
Poll(usize),
Step,
Timeout,
Deliver(usize, usize),
Drop(usize, usize),
Dup(usize, usize),
Garble(usize, usize),
RestartSink(u64),
RestartSource,
}
fn act() -> impl Strategy<Value = Act> {
let lane = 0usize..2;
let at = 0usize..8;
prop_oneof![
5 => prop::collection::vec((0u8..10, -2i64..=2), 0..3).prop_map(Act::Commit),
4 => Just(Act::Pump),
4 => (1usize..4).prop_map(Act::Poll),
2 => Just(Act::Step),
2 => Just(Act::Timeout),
8 => (lane.clone(), at.clone()).prop_map(|(l, i)| Act::Deliver(l, i)),
2 => (lane.clone(), at.clone()).prop_map(|(l, i)| Act::Drop(l, i)),
2 => (lane.clone(), at.clone()).prop_map(|(l, i)| Act::Dup(l, i)),
1 => (lane, at).prop_map(|(l, i)| Act::Garble(l, i)),
1 => (0u64..8).prop_map(Act::RestartSink),
1 => Just(Act::RestartSource),
]
}
fn run(cap: usize, acts: &[Act]) {
let mut w = World::new(cap);
for (n, a) in acts.iter().enumerate() {
match a {
Act::Commit(d) => w.commit(d.iter().copied().collect()),
Act::Pump => {
w.sink.pump();
}
Act::Poll(m) => {
w.src.poll(&mut w.cons, *m);
}
Act::Step => w.step(),
Act::Timeout => w.src.connection_lost(),
Act::Deliver(d, i) => w.lane(*d, *i, |l, at| {
let f = l.flight.remove(at);
l.ready.push_back(f);
}),
Act::Drop(d, i) => w.lane(*d, *i, |l, at| {
l.flight.remove(at);
}),
Act::Dup(d, i) => w.lane(*d, *i, |l, at| {
let f = l.flight[at].clone();
l.flight.push(f);
}),
Act::Garble(d, i) => w.lane(*d, *i, |l, at| l.flight[at] = "garbage".to_string()),
Act::RestartSink(seq) => w.restart_sink(*seq),
Act::RestartSource => w.restart_source(),
}
w.check(&format!("after action {n} ({a:?})"));
}
w.settle();
}
proptest! {
/// Any history over a lossy, duplicating, garbling, reordering network keeps both safety
/// rules at every step, and settles to the sender's integral once the network is reliable.
#[test]
fn the_receiver_is_never_wrong_over_a_hostile_network(
acts in prop::collection::vec(act(), 0..60), cap in 1usize..5,
) {
run(cap, &acts);
}
}
#[test]
fn the_reviewers_late_reply_case_is_a_history() {
// Epoch 1 reads a request, its reply is delayed, the sender restarts, and the late reply comes.
let acts = [
Act::Poll(1),
Act::Pump,
Act::Commit(vec![(1, 1)]),
Act::Deliver(1, 0),
Act::Pump,
Act::RestartSink(3),
Act::Timeout,
Act::Poll(1),
Act::Deliver(1, 0),
Act::Pump,
Act::Deliver(0, 1),
Act::Deliver(0, 0),
Act::Poll(4),
];
run(2, &acts);
let _ = z(&[]);
}
crates/zset/tests/support/mod.rs
+83
View File
@@ -0,0 +1,83 @@
//! A text wire form of `u8`-keyed frames, shared by the edge tests.
use zset::{CodecError, Correction, Delta, EdgeCodec, Frame, Request, Stamp, ZSet};
type Z = ZSet<u8>;
/// A text wire form, to run the same histories through an encoding.
#[derive(Clone, Copy)]
pub struct Text;
pub fn zs(z: &Z) -> String {
z.iter()
.map(|(k, w)| format!("{k}:{w}"))
.collect::<Vec<_>>()
.join(",")
}
pub fn unzs(s: &str) -> Result<Z, CodecError> {
let bad = || CodecError(format!("bad entries {s:?}"));
s.split(',')
.filter(|p| !p.is_empty())
.map(|p| {
let (k, w) = p.split_once(':').ok_or_else(bad)?;
Ok((
k.parse::<u8>().map_err(|_| bad())?,
w.parse::<i64>().map_err(|_| bad())?,
))
})
.collect()
}
impl EdgeCodec<u8> for Text {
type Wire = String;
fn encode(&self, f: &Frame<u8>) -> String {
match f {
Frame::Delta(d) => format!("D {} {} {} {}", d.epoch, d.from, d.to, zs(&d.delta)),
Frame::Correction(c) => {
format!("C {} {} {} {}", c.epoch, c.through, c.token, zs(&c.target))
}
Frame::Request(r) => match r.have {
Some(s) => format!("R {} {} {}", r.token, s.epoch, s.seq),
None => format!("R {}", r.token),
},
}
}
fn decode(&self, w: String) -> Result<Frame<u8>, CodecError> {
let bad = || CodecError(format!("bad frame {w:?}"));
let mut it = w.splitn(5, ' ');
let num = |s: Option<&str>| s.and_then(|s| s.parse::<u64>().ok()).ok_or_else(bad);
match it.next() {
Some("D") => Ok(Frame::Delta(Delta {
epoch: num(it.next())?,
from: num(it.next())?,
to: num(it.next())?,
delta: unzs(it.next().ok_or_else(bad)?)?,
})),
Some("C") => {
let mut it = w.splitn(5, ' ');
it.next();
Ok(Frame::Correction(Correction {
epoch: num(it.next())?,
through: num(it.next())?,
token: num(it.next())?,
target: unzs(it.next().ok_or_else(bad)?)?,
}))
}
Some("R") => {
let token = num(it.next())?;
let have = match it.next() {
None => None,
e => Some(Stamp {
epoch: num(e)?,
seq: num(it.next())?,
}),
};
Ok(Frame::Request(Request { have, token }))
}
_ => Err(bad()),
}
}
}
crates/zset/tests/threads.rs
+73 -6
View File
@@ -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::<u8>();
b.push(&input, Z::singleton(1, 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"
);
}
}