Merge dbsp/source-trait into dbsp/model-over-source
# Conflicts: # context/test-evidence/core.md
This commit is contained in:
Claude (dbsp-sources)
@@ -1086,6 +1086,45 @@ fn a_reply_held_across_a_host_restart_cannot_pull_the_client_back() {
|
||||
assert_eq!(state, Synced::Settled);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_late_reply_from_a_host_the_client_never_joined_cannot_strand_it() {
|
||||
let window = Window {
|
||||
top: 0,
|
||||
left: 0,
|
||||
rows: 1,
|
||||
cols: 1,
|
||||
};
|
||||
let mut first = host_instance(1);
|
||||
let mut client = EdgeClient::new(session());
|
||||
client.track(shape(), window);
|
||||
client.sync(&mut Flaky::new(&mut first), &shape(), 6);
|
||||
|
||||
let mut second = host_instance(2);
|
||||
let id = connect(&mut second);
|
||||
send(&mut second, id, vec![set("South", 1.0)]);
|
||||
let mut lost = Flaky::scripted(&mut second, vec![Some(Fault::DropReply)]);
|
||||
client.connection_lost();
|
||||
client.sync(&mut lost, &shape(), 6);
|
||||
let late = lost.replies.remove(0);
|
||||
|
||||
let mut third = host_instance(3);
|
||||
client.connection_lost();
|
||||
client.sync(&mut Flaky::new(&mut third), &shape(), 6);
|
||||
let Reply::Correction(late) = late else {
|
||||
panic!("the lost reply was a correction")
|
||||
};
|
||||
client.apply(&late);
|
||||
client.connection_lost();
|
||||
|
||||
let (state, _) = client.sync(&mut Flaky::new(&mut third), &shape(), 6);
|
||||
assert_eq!(
|
||||
state,
|
||||
Synced::Settled,
|
||||
"the live host's answers must be taken"
|
||||
);
|
||||
assert_eq!(client.held(&shape()), Some(&drawn(&third, &window)));
|
||||
}
|
||||
|
||||
// ── the whole thing, at random ─────────────────────────────────────────────
|
||||
|
||||
mod random {
|
||||
@@ -1265,6 +1304,39 @@ mod random {
|
||||
deliver_all(client, ¬ices);
|
||||
}
|
||||
|
||||
/// Epochs the client has applied a correction from, oldest first.
|
||||
fn note_epoch(client: &EdgeClient, epochs: &mut Vec<(u64, u64)>) {
|
||||
if let Some(w) = client.watermark(&shape()) {
|
||||
let now = (w.host, w.generation);
|
||||
if !epochs.contains(&now) {
|
||||
epochs.push(now);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A late reply arrives out of band. One from an epoch the client was in and has left is refused;
|
||||
/// one from an epoch it never joined may legitimately move it, and the next answer brings it back.
|
||||
fn replay(
|
||||
client: &mut EdgeClient,
|
||||
c: &improvise_protocol::delivery::Correction,
|
||||
epochs: &mut Vec<(u64, u64)>,
|
||||
) -> Result<(), TestCaseError> {
|
||||
let theirs = (c.through.host, c.through.generation);
|
||||
let current = client.watermark(&shape()).map(|w| (w.host, w.generation));
|
||||
let left = epochs.contains(&theirs) && current != Some(theirs);
|
||||
let (outcome, _) = client.apply(c);
|
||||
if left {
|
||||
prop_assert_eq!(
|
||||
outcome,
|
||||
Outcome::Duplicate,
|
||||
"a reply from the left epoch {:?}",
|
||||
theirs
|
||||
);
|
||||
}
|
||||
note_epoch(client, epochs);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
proptest! {
|
||||
#[test]
|
||||
fn a_corrected_state_is_the_target_and_correcting_it_again_is_the_zero_zset(
|
||||
@@ -1306,9 +1378,7 @@ mod random {
|
||||
}
|
||||
if let Some(i) = weather.replay.filter(|_| !old.is_empty()) {
|
||||
match &old[i % old.len()] {
|
||||
Reply::Correction(c) => {
|
||||
client.apply(c);
|
||||
}
|
||||
Reply::Correction(c) => replay(&mut client, c, &mut epochs)?,
|
||||
Reply::Notices(list) => deliver_all(&mut client, list),
|
||||
}
|
||||
}
|
||||
@@ -1316,20 +1386,10 @@ mod random {
|
||||
let mut edge = Flaky::scripted(&mut host, script);
|
||||
client.sync(&mut edge, &shape(), 6);
|
||||
old.append(&mut edge.replies);
|
||||
note_epoch(&client, &mut epochs);
|
||||
let moved_host = matches!(op, Op::Restart | Op::Swap);
|
||||
if let (true, Some(Reply::Correction(c))) = (moved_host, old.get(step % old.len().max(1))) {
|
||||
client.apply(c);
|
||||
}
|
||||
if let Some(w) = client.watermark(&shape()) {
|
||||
let now = (w.host, w.generation);
|
||||
let seen_at = epochs.iter().position(|e| *e == now);
|
||||
prop_assert!(
|
||||
seen_at.is_none_or(|i| i + 1 == epochs.len()),
|
||||
"the client went back to a superseded epoch {now:?} after {epochs:?}"
|
||||
);
|
||||
if seen_at.is_none() {
|
||||
epochs.push(now);
|
||||
}
|
||||
replay(&mut client, c, &mut epochs)?;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -51,6 +51,23 @@ impl Numbers {
|
||||
numbers
|
||||
}
|
||||
|
||||
/// Numbers from what an engine computed: how many, their exact sum and the least and
|
||||
/// greatest that are not NaN. Extremes that contradict the count or are NaN are dropped.
|
||||
pub fn from_parts(
|
||||
count: u64,
|
||||
sum: ExactSum,
|
||||
least: Option<f64>,
|
||||
greatest: Option<f64>,
|
||||
) -> Self {
|
||||
let held = |x: Option<f64>| x.filter(|x| count > 0 && !x.is_nan());
|
||||
Numbers {
|
||||
count,
|
||||
sum,
|
||||
least: held(least),
|
||||
greatest: held(greatest),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn add(&mut self, x: f64) {
|
||||
self.count += 1;
|
||||
self.sum.add(x, 1);
|
||||
@@ -104,9 +121,9 @@ fn merged(a: Option<f64>, b: Option<f64>, pick: fn(f64, f64) -> f64) -> Option<f
|
||||
}
|
||||
}
|
||||
|
||||
/// The values that are not numbers, as far as unanimity needs them.
|
||||
/// What the values that are not numbers agree on: nothing held, one value, or disagreement.
|
||||
#[derive(Debug, Clone, Default, PartialEq)]
|
||||
enum Others {
|
||||
pub enum Agreement {
|
||||
#[default]
|
||||
None,
|
||||
One(CellValue),
|
||||
@@ -118,7 +135,7 @@ enum Others {
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct Summary {
|
||||
numbers: Numbers,
|
||||
others: Others,
|
||||
others: Agreement,
|
||||
}
|
||||
|
||||
impl Summary {
|
||||
@@ -128,6 +145,11 @@ impl Summary {
|
||||
summary
|
||||
}
|
||||
|
||||
/// A summary from what an engine computed: the numbers, and what the rest agree on.
|
||||
pub fn from_parts(numbers: Numbers, others: Agreement) -> Self {
|
||||
Summary { numbers, others }
|
||||
}
|
||||
|
||||
pub fn add(&mut self, value: &CellValue) {
|
||||
match value.as_f64() {
|
||||
Some(x) => self.numbers.add(x),
|
||||
@@ -137,18 +159,18 @@ impl Summary {
|
||||
|
||||
fn add_other(&mut self, value: &CellValue) {
|
||||
self.others = match std::mem::take(&mut self.others) {
|
||||
Others::None => Others::One(value.clone()),
|
||||
Others::One(held) if held == *value => Others::One(held),
|
||||
_ => Others::Mixed,
|
||||
Agreement::None => Agreement::One(value.clone()),
|
||||
Agreement::One(held) if held == *value => Agreement::One(held),
|
||||
_ => Agreement::Mixed,
|
||||
};
|
||||
}
|
||||
|
||||
pub fn merge(&mut self, other: &Summary) {
|
||||
self.numbers.merge(&other.numbers);
|
||||
match &other.others {
|
||||
Others::None => {}
|
||||
Others::One(v) => self.add_other(v),
|
||||
Others::Mixed => self.others = Others::Mixed,
|
||||
Agreement::None => {}
|
||||
Agreement::One(v) => self.add_other(v),
|
||||
Agreement::Mixed => self.others = Agreement::Mixed,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -159,8 +181,8 @@ impl Summary {
|
||||
|
||||
pub fn reduce(&self, func: &AggFunc) -> Option<CellValue> {
|
||||
match (&self.others, self.numbers.count()) {
|
||||
(Others::None, _) => self.numbers.reduce(func).map(CellValue::Number),
|
||||
(Others::One(only), 0) => Some(only.clone()),
|
||||
(Agreement::None, _) => self.numbers.reduce(func).map(CellValue::Number),
|
||||
(Agreement::One(only), 0) => Some(only.clone()),
|
||||
_ => Some(CellValue::Error("mixed".to_string())),
|
||||
}
|
||||
}
|
||||
@@ -210,7 +232,10 @@ impl CellSource for DataStore {
|
||||
&'a self,
|
||||
partial: &'a [(String, String)],
|
||||
) -> Box<dyn Iterator<Item = CellKey> + 'a> {
|
||||
Box::new(self.cells(partial).map(|(key, _)| key))
|
||||
match partial.is_empty() {
|
||||
true => Box::new(self.iter_cells().map(|(key, _)| key)),
|
||||
false => Box::new(self.matching_cells(partial).into_iter().map(|(key, _)| key)),
|
||||
}
|
||||
}
|
||||
|
||||
fn cells<'a>(
|
||||
@@ -559,4 +584,61 @@ mod tests {
|
||||
"no cell is keyed by R=E alone"
|
||||
);
|
||||
}
|
||||
|
||||
/// The parts an engine would compute for `values`, found without `Summary::of`.
|
||||
fn parts_of(values: &[CellValue]) -> Summary {
|
||||
let numbers: Vec<f64> = values.iter().filter_map(|v| v.as_f64()).collect();
|
||||
let mut sum = ExactSum::new();
|
||||
numbers.iter().for_each(|x| sum.add(*x, 1));
|
||||
let finite = || numbers.iter().copied().filter(|x| !x.is_nan());
|
||||
let least = finite().min_by(f64::total_cmp);
|
||||
let greatest = finite().max_by(f64::total_cmp);
|
||||
let rest: Vec<&CellValue> = values.iter().filter(|v| v.as_f64().is_none()).collect();
|
||||
let agreement = match rest.split_first() {
|
||||
None => Agreement::None,
|
||||
Some((first, others)) if others.iter().all(|v| v == first) => {
|
||||
Agreement::One((*first).clone())
|
||||
}
|
||||
Some(_) => Agreement::Mixed,
|
||||
};
|
||||
let parts = Numbers::from_parts(numbers.len() as u64, sum, least, greatest);
|
||||
Summary::from_parts(parts, agreement)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_summary_built_from_an_engines_parts_reduces_as_one_built_from_the_values() {
|
||||
for values in sequences(3) {
|
||||
let refs: Vec<&CellValue> = values.iter().collect();
|
||||
for func in &FUNCS {
|
||||
let want = reference(&refs, func);
|
||||
let got = parts_of(&values).reduce(func);
|
||||
assert!(
|
||||
same(&want, &got),
|
||||
"{func:?} of {values:?}: {want:?} != {got:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parts_that_contradict_themselves_are_made_consistent() {
|
||||
let none = Numbers::from_parts(0, ExactSum::new(), Some(1.0), Some(f64::NAN));
|
||||
assert_eq!(none.reduce(&AggFunc::Min), None, "no numbers, no extremes");
|
||||
let nan = Numbers::from_parts(2, ExactSum::new(), Some(f64::NAN), None);
|
||||
let min = nan.reduce(&AggFunc::Min).unwrap();
|
||||
assert!(min.is_nan(), "a NaN is never an extreme");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn keys_of_a_store_are_its_cells_keys_without_the_values() {
|
||||
let rows = rows();
|
||||
let store = store_of(&rows);
|
||||
let all: Vec<CellKey> = store.keys(&[]).collect();
|
||||
assert_eq!(all, rows.iter().map(|(k, _)| k.clone()).collect::<Vec<_>>());
|
||||
let east = vec![pair("R", "E")];
|
||||
let under: Vec<CellKey> = store.keys(&east).collect();
|
||||
let want: Vec<CellKey> = store.cells(&east).map(|c| c.0).collect();
|
||||
assert_eq!(under, want);
|
||||
assert!(!under.is_empty());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -243,11 +243,26 @@ impl EdgeClient {
|
||||
}
|
||||
}
|
||||
|
||||
/// A correction. It applies only at the watermark it starts from: `held + (target - held)`.
|
||||
/// A correction that arrives on its own: it applies only at the watermark it starts from,
|
||||
/// as `held + (target - held)`, and never from an epoch the client has left.
|
||||
pub fn apply(&mut self, c: &Correction) -> (Outcome, Vec<Event>) {
|
||||
self.apply_with(c, false)
|
||||
}
|
||||
|
||||
/// The reply to a request this client just made. The host that answers is the live one, so
|
||||
/// an epoch the client had written off is live again (`context/code/delivery.md` section 4).
|
||||
pub fn apply_answer(&mut self, c: &Correction) -> (Outcome, Vec<Event>) {
|
||||
self.apply_with(c, true)
|
||||
}
|
||||
|
||||
fn apply_with(&mut self, c: &Correction, answer: bool) -> (Outcome, Vec<Event>) {
|
||||
let Some(t) = self.shapes.get_mut(&c.shape) else {
|
||||
return (Outcome::Untracked, Vec::new());
|
||||
};
|
||||
if answer {
|
||||
t.left
|
||||
.retain(|e| *e != (c.through.host, c.through.generation));
|
||||
}
|
||||
let target = decode(&c.target);
|
||||
let older = |s: Option<Stamp>| {
|
||||
s.is_some_and(|s| s.host == c.through.host && c.through.generation < s.generation)
|
||||
@@ -347,7 +362,7 @@ impl EdgeClient {
|
||||
return (Synced::Settled, events);
|
||||
};
|
||||
match edge.call(request) {
|
||||
Ok(Reply::Correction(c)) => events.extend(self.apply(&c).1),
|
||||
Ok(Reply::Correction(c)) => events.extend(self.apply_answer(&c).1),
|
||||
Ok(Reply::Notices(list)) => {
|
||||
list.iter()
|
||||
.for_each(|n| events.extend(self.accept(n, Some(shape))));
|
||||
@@ -387,7 +402,7 @@ impl EdgeClient {
|
||||
Ok(Reply::Notices(list)) => list
|
||||
.iter()
|
||||
.for_each(|n| events.extend(self.accept(n, Some(shape)))),
|
||||
Ok(Reply::Correction(c)) => events.extend(self.apply(&c).1),
|
||||
Ok(Reply::Correction(c)) => events.extend(self.apply_answer(&c).1),
|
||||
Err(EdgeError::Failed) => {
|
||||
self.request_failed(shape);
|
||||
return (Synced::Failed, events);
|
||||
|
||||
@@ -47,6 +47,7 @@ extern "C" {
|
||||
#define ZS_ERR_POISONED 7
|
||||
#define ZS_ERR_CALLBACK 8
|
||||
#define ZS_ERR_INTERRUPTED 9
|
||||
#define ZS_ERR_TRANSACTION 10
|
||||
|
||||
#define ZS_AGG_COUNT 0
|
||||
#define ZS_AGG_SUM 1
|
||||
@@ -131,8 +132,27 @@ int32_t zs_circuit_aggregate(ZsCircuit *c, uint32_t stream, int32_t kind, uint32
|
||||
int32_t zs_circuit_push(ZsCircuit *c, uint32_t input, const ZsZSet *z);
|
||||
|
||||
/* One clock tick. After a failure in a step the circuit is poisoned: every later call that
|
||||
* uses it returns ZS_ERR_POISONED, except the free. */
|
||||
* uses it returns ZS_ERR_POISONED, except the free, and zs_circuit_rollback of a transaction
|
||||
* the failed step ran in. */
|
||||
int32_t zs_circuit_step(ZsCircuit *c);
|
||||
|
||||
/* Transactions: undo a failed command. zs_circuit_begin opens one; zs_circuit_commit keeps
|
||||
* what its steps did; zs_circuit_rollback puts the circuit's state back to what it was at
|
||||
* begin, and revives a circuit that a failed step in the transaction poisoned. The cost of a
|
||||
* transaction is the keys its steps touch (plus a delay's held input), not the circuit's size.
|
||||
* - At most one is open. begin on an open one, and commit or rollback on none, is
|
||||
* ZS_ERR_TRANSACTION. begin on a poisoned circuit is ZS_ERR_POISONED; so is commit after a
|
||||
* failed step (roll back instead). A poison from outside a transaction is permanent.
|
||||
* - Pushes made after begin are dropped by a rollback; what was pending at begin is pending
|
||||
* again. zs_circuit_take is ZS_ERR_TRANSACTION while one is open: take after commit.
|
||||
* - Operators built inside a transaction stay, and their state goes back to empty.
|
||||
* - What is outside the circuit is not undone: a callback's own side effects, and a
|
||||
* ZsRebuild stepped in the transaction (it reports ZS_ERR_INTERRUPTED afterwards; start
|
||||
* another job). */
|
||||
int32_t zs_circuit_begin(ZsCircuit *c);
|
||||
int32_t zs_circuit_commit(ZsCircuit *c);
|
||||
int32_t zs_circuit_rollback(ZsCircuit *c);
|
||||
|
||||
int32_t zs_circuit_take(ZsCircuit *c, uint32_t sink, ZsZSet **out);
|
||||
|
||||
/* Resumable snapshot load: chunk keys per piece (0 reads as 1). The whole load is one
|
||||
|
||||
@@ -40,6 +40,7 @@ codes! {
|
||||
Poisoned = 7 => "ZS_ERR_POISONED",
|
||||
Callback = 8 => "ZS_ERR_CALLBACK",
|
||||
Interrupted = 9 => "ZS_ERR_INTERRUPTED",
|
||||
Transaction = 10 => "ZS_ERR_TRANSACTION",
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
@@ -209,6 +210,8 @@ pub struct Engine {
|
||||
sinks: Vec<Output<Bytes>>,
|
||||
fault: Fault,
|
||||
poisoned: bool,
|
||||
/// A transaction is open (`begin`, and not yet `commit` or `rollback`).
|
||||
txn: bool,
|
||||
}
|
||||
|
||||
impl Default for Engine {
|
||||
@@ -226,6 +229,7 @@ impl Engine {
|
||||
sinks: Vec::new(),
|
||||
fault: Rc::default(),
|
||||
poisoned: false,
|
||||
txn: false,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -474,8 +478,54 @@ impl Engine {
|
||||
self.guarded(|c| c.step())
|
||||
}
|
||||
|
||||
fn no_transaction() -> Fail {
|
||||
Fail::new(Code::Transaction, "no transaction is open")
|
||||
}
|
||||
|
||||
/// Opens a transaction: the circuit logs what its steps change.
|
||||
pub fn begin(&mut self) -> Res<()> {
|
||||
self.alive()?;
|
||||
if self.txn {
|
||||
return Err(Fail::new(
|
||||
Code::Transaction,
|
||||
"a transaction is already open",
|
||||
));
|
||||
}
|
||||
self.circuit.begin();
|
||||
self.txn = true;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Keeps the transaction's steps. A transaction whose step failed cannot be committed.
|
||||
pub fn commit(&mut self) -> Res<()> {
|
||||
if !self.txn {
|
||||
return Err(Self::no_transaction());
|
||||
}
|
||||
self.alive()?;
|
||||
self.circuit.commit();
|
||||
self.txn = false;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Undoes the transaction's steps and clears the poison a failed step left in it.
|
||||
pub fn rollback(&mut self) -> Res<()> {
|
||||
if !self.txn {
|
||||
return Err(Self::no_transaction());
|
||||
}
|
||||
self.circuit.rollback();
|
||||
self.poisoned = false;
|
||||
self.txn = false;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn take(&mut self, sink: u32) -> Res<BZ> {
|
||||
self.alive()?;
|
||||
if self.txn {
|
||||
return Err(Fail::new(
|
||||
Code::Transaction,
|
||||
"a transaction is open: commit or roll back before taking output",
|
||||
));
|
||||
}
|
||||
self.sinks
|
||||
.get((sink & !SINK_BIT) as usize)
|
||||
.filter(|_| sink & SINK_BIT != 0)
|
||||
|
||||
@@ -499,6 +499,30 @@ pub unsafe extern "C" fn zs_circuit_step(c: *mut ZsCircuit) -> i32 {
|
||||
})
|
||||
}
|
||||
|
||||
/// # Safety
|
||||
/// `c` is a live circuit or null.
|
||||
#[unsafe(no_mangle)]
|
||||
pub unsafe extern "C" fn zs_circuit_begin(c: *mut ZsCircuit) -> i32 {
|
||||
// SAFETY: the caller's contract.
|
||||
guard(|| unsafe { circuit(c)? }.engine().begin())
|
||||
}
|
||||
|
||||
/// # Safety
|
||||
/// As `zs_circuit_begin`.
|
||||
#[unsafe(no_mangle)]
|
||||
pub unsafe extern "C" fn zs_circuit_commit(c: *mut ZsCircuit) -> i32 {
|
||||
// SAFETY: the caller's contract.
|
||||
guard(|| unsafe { circuit(c)? }.engine().commit())
|
||||
}
|
||||
|
||||
/// # Safety
|
||||
/// As `zs_circuit_begin`.
|
||||
#[unsafe(no_mangle)]
|
||||
pub unsafe extern "C" fn zs_circuit_rollback(c: *mut ZsCircuit) -> i32 {
|
||||
// SAFETY: the caller's contract.
|
||||
guard(|| unsafe { circuit(c)? }.engine().rollback())
|
||||
}
|
||||
|
||||
/// # Safety
|
||||
/// `c` is a live circuit or null; `out` is null or writable.
|
||||
#[unsafe(no_mangle)]
|
||||
|
||||
@@ -965,3 +965,94 @@ fn a_host_step_between_rebuild_calls_is_interrupted_not_a_poison() {
|
||||
unsafe { zs_rebuild_free(job) };
|
||||
unsafe { zs_zset_free(snapshot) };
|
||||
}
|
||||
|
||||
/// An integrate over `k` held at `i64::MAX` (taken), and its sink.
|
||||
fn at_max() -> (C, u32, u32) {
|
||||
let c = C::new();
|
||||
let i = c.input();
|
||||
let mut int = 0;
|
||||
assert_eq!(unsafe { zs_circuit_integrate(c.0, i, &mut int) }, OK);
|
||||
let s = c.sink(int);
|
||||
c.push(i, &[(b"k", i64::MAX)]);
|
||||
assert_eq!(c.run(s), one(&[(b"k", i64::MAX)]));
|
||||
(c, i, s)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_rollback_after_a_failed_step_revives_the_circuit_in_its_old_state() {
|
||||
let (c, i, s) = at_max();
|
||||
assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
|
||||
c.push(i, &[(b"a", 1), (b"k", 1)]);
|
||||
assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Overflow as i32);
|
||||
assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Poisoned as i32);
|
||||
assert_eq!(
|
||||
unsafe { zs_circuit_commit(c.0) },
|
||||
Code::Poisoned as i32,
|
||||
"a failed step cannot be committed"
|
||||
);
|
||||
assert_eq!(unsafe { zs_circuit_rollback(c.0) }, OK);
|
||||
c.push(i, &[(b"k", -1)]);
|
||||
assert_eq!(
|
||||
c.run(s),
|
||||
one(&[(b"k", i64::MAX - 1)]),
|
||||
"[rollback] neither `a` nor the failed +1 is in the integral"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_committed_transaction_keeps_its_steps_and_a_new_one_may_follow() {
|
||||
let (c, i, s) = at_max();
|
||||
assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
|
||||
c.push(i, &[(b"k", -1)]);
|
||||
assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
|
||||
assert_eq!(unsafe { zs_circuit_commit(c.0) }, OK);
|
||||
assert_eq!(c.take(s), one(&[(b"k", i64::MAX - 1)]));
|
||||
assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
|
||||
assert_eq!(unsafe { zs_circuit_rollback(c.0) }, OK);
|
||||
c.push(i, &[(b"k", -1)]);
|
||||
assert_eq!(c.run(s), one(&[(b"k", i64::MAX - 2)]));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_failed_callback_step_is_rolled_back_too() {
|
||||
let c = C::new();
|
||||
let i = c.input();
|
||||
let (mut f, mut int) = (0, 0);
|
||||
assert_eq!(unsafe { zs_circuit_integrate(c.0, i, &mut int) }, OK);
|
||||
let s = c.sink(int);
|
||||
assert_eq!(
|
||||
unsafe { zs_circuit_filter(c.0, i, Some(fails), null_mut(), &mut f) },
|
||||
OK
|
||||
);
|
||||
assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
|
||||
c.push(i, &[(b"k", 1)]);
|
||||
assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Callback as i32);
|
||||
assert_eq!(unsafe { zs_circuit_rollback(c.0) }, OK);
|
||||
assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK, "usable again");
|
||||
assert_eq!(unsafe { zs_circuit_commit(c.0) }, OK);
|
||||
assert_eq!(
|
||||
c.take(s),
|
||||
vec![],
|
||||
"[callback] the failed step left no output"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn transaction_misuse_is_a_code_and_changes_nothing() {
|
||||
let (c, i, s) = at_max();
|
||||
let t = Code::Transaction as i32;
|
||||
assert_eq!(unsafe { zs_circuit_commit(c.0) }, t);
|
||||
assert_eq!(unsafe { zs_circuit_rollback(c.0) }, t);
|
||||
assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
|
||||
assert_eq!(unsafe { zs_circuit_begin(c.0) }, t);
|
||||
let mut z = null_mut();
|
||||
assert_eq!(unsafe { zs_circuit_take(c.0, s, &mut z) }, t);
|
||||
assert!(z.is_null(), "nothing written");
|
||||
assert!(last_error().contains("roll back"), "{}", last_error());
|
||||
assert_eq!(unsafe { zs_circuit_commit(c.0) }, OK);
|
||||
// A poison from outside any transaction is not undone by a rollback.
|
||||
c.push(i, &[(b"k", 1)]);
|
||||
assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Overflow as i32);
|
||||
assert_eq!(unsafe { zs_circuit_rollback(c.0) }, t);
|
||||
assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Poisoned as i32);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,193 @@
|
||||
//! Transactions through the ABI: a rebuild across a rollback, callbacks, and poison.
|
||||
//! From the review of #355. Evidence: `context/test-evidence/zset-ffi.md`.
|
||||
|
||||
use std::ffi::c_void;
|
||||
use std::ptr::null_mut;
|
||||
use zset_ffi::engine::Code;
|
||||
use zset_ffi::*;
|
||||
|
||||
const OK: i32 = 0;
|
||||
|
||||
fn zset(entries: &[(&[u8], i64)]) -> *mut ZsZSet {
|
||||
let mut z = null_mut();
|
||||
assert_eq!(unsafe { zs_zset_new(&mut z) }, OK);
|
||||
for (k, w) in entries {
|
||||
assert_eq!(unsafe { zs_zset_insert(z, k.as_ptr(), k.len(), *w) }, OK);
|
||||
}
|
||||
z
|
||||
}
|
||||
|
||||
fn weight(z: *const ZsZSet, k: &[u8]) -> i64 {
|
||||
let mut w = 0;
|
||||
assert_eq!(
|
||||
unsafe { zs_zset_weight(z, k.as_ptr(), k.len(), &mut w) },
|
||||
OK
|
||||
);
|
||||
w
|
||||
}
|
||||
|
||||
fn rig() -> (*mut ZsCircuit, u32, u32) {
|
||||
let mut c = null_mut();
|
||||
assert_eq!(unsafe { zs_circuit_new(&mut c) }, OK);
|
||||
let mut i = 0;
|
||||
assert_eq!(unsafe { zs_circuit_input(c, &mut i) }, OK);
|
||||
let mut int = 0;
|
||||
assert_eq!(unsafe { zs_circuit_integrate(c, i, &mut int) }, OK);
|
||||
let mut s = 0;
|
||||
assert_eq!(unsafe { zs_circuit_output(c, int, &mut s) }, OK);
|
||||
(c, i, s)
|
||||
}
|
||||
|
||||
extern "C" fn out_of_time(_: *mut c_void) -> i32 {
|
||||
1
|
||||
}
|
||||
|
||||
fn take(c: *mut ZsCircuit, s: u32) -> *mut ZsZSet {
|
||||
let mut z = null_mut();
|
||||
assert_eq!(unsafe { zs_circuit_take(c, s, &mut z) }, OK);
|
||||
z
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn f1_rebuild_begun_before_a_txn_continued_inside_and_rolled_back() {
|
||||
let (c, i, s) = rig();
|
||||
let snap = zset(&[(b"a", 1), (b"b", 1), (b"c", 1), (b"d", 1)]);
|
||||
let mut job = null_mut();
|
||||
assert_eq!(unsafe { zs_rebuild_new(c, i, snap, 1, &mut job) }, OK);
|
||||
let mut done = 0;
|
||||
assert_eq!(
|
||||
unsafe { zs_rebuild_step(job, c, Some(out_of_time), null_mut(), &mut done) },
|
||||
OK
|
||||
);
|
||||
assert_eq!(unsafe { zs_circuit_begin(c) }, OK);
|
||||
assert_eq!(
|
||||
unsafe { zs_rebuild_step(job, c, Some(out_of_time), null_mut(), &mut done) },
|
||||
OK
|
||||
);
|
||||
assert_eq!(unsafe { zs_circuit_rollback(c) }, OK);
|
||||
let rc = unsafe { zs_rebuild_step(job, c, None, null_mut(), &mut done) };
|
||||
let z = take(c, s);
|
||||
let got = (
|
||||
rc,
|
||||
done,
|
||||
weight(z, b"a"),
|
||||
weight(z, b"b"),
|
||||
weight(z, b"c"),
|
||||
weight(z, b"d"),
|
||||
);
|
||||
unsafe { zs_zset_free(z) };
|
||||
unsafe { zs_rebuild_free(job) };
|
||||
unsafe { zs_zset_free(snap) };
|
||||
unsafe { zs_circuit_free(c) };
|
||||
assert_eq!(
|
||||
got.0,
|
||||
Code::Interrupted as i32,
|
||||
"[f1] header: a rebuild stepped in a rolled-back txn reports ZS_ERR_INTERRUPTED; got (rc, done, a, b, c, d) = {got:?}"
|
||||
);
|
||||
}
|
||||
|
||||
extern "C" fn rolls_back(ctx: *mut c_void, _: *const u8, _: usize) -> i32 {
|
||||
let (c, seen) = unsafe { &mut *(ctx as *mut (*mut ZsCircuit, i32)) };
|
||||
*seen = unsafe { zs_circuit_rollback(*c) };
|
||||
1
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn f2_a_rollback_from_inside_a_callback_is_refused_and_the_txn_stays_open() {
|
||||
let (c, i, s) = rig();
|
||||
let mut state: (*mut ZsCircuit, i32) = (c, 0);
|
||||
let mut f = 0;
|
||||
let ctx = (&mut state as *mut (*mut ZsCircuit, i32)).cast();
|
||||
assert_eq!(
|
||||
unsafe { zs_circuit_filter(c, i, Some(rolls_back), ctx, &mut f) },
|
||||
OK
|
||||
);
|
||||
assert_eq!(unsafe { zs_circuit_begin(c) }, OK);
|
||||
let z = zset(&[(b"k", 1)]);
|
||||
assert_eq!(unsafe { zs_circuit_push(c, i, z) }, OK);
|
||||
assert_eq!(unsafe { zs_circuit_step(c) }, OK);
|
||||
assert_eq!(
|
||||
state.1,
|
||||
Code::Poisoned as i32,
|
||||
"[f2] rollback inside a callback"
|
||||
);
|
||||
assert_eq!(
|
||||
unsafe { zs_circuit_commit(c) },
|
||||
OK,
|
||||
"[f2] txn still open, commit works"
|
||||
);
|
||||
let out = take(c, s);
|
||||
assert_eq!(weight(out, b"k"), 1);
|
||||
unsafe { zs_zset_free(out) };
|
||||
unsafe { zs_zset_free(z) };
|
||||
unsafe { zs_circuit_free(c) };
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn f3_begin_on_a_circuit_poisoned_outside_a_txn_is_poisoned() {
|
||||
let (c, i, _s) = rig();
|
||||
let z = zset(&[(b"k", i64::MAX)]);
|
||||
assert_eq!(unsafe { zs_circuit_push(c, i, z) }, OK);
|
||||
assert_eq!(unsafe { zs_circuit_step(c) }, OK);
|
||||
assert_eq!(unsafe { zs_circuit_push(c, i, z) }, OK);
|
||||
assert_eq!(unsafe { zs_circuit_step(c) }, Code::Overflow as i32);
|
||||
assert_eq!(unsafe { zs_circuit_begin(c) }, Code::Poisoned as i32);
|
||||
assert_eq!(unsafe { zs_circuit_rollback(c) }, Code::Transaction as i32);
|
||||
assert_eq!(unsafe { zs_circuit_commit(c) }, Code::Transaction as i32);
|
||||
unsafe { zs_zset_free(z) };
|
||||
unsafe { zs_circuit_free(c) };
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn f4_a_failed_rebuild_step_in_a_txn_then_rollback_then_resume() {
|
||||
let (c, i, s) = rig();
|
||||
let big = zset(&[(b"a", i64::MAX)]);
|
||||
assert_eq!(unsafe { zs_circuit_push(c, i, big) }, OK);
|
||||
assert_eq!(unsafe { zs_circuit_step(c) }, OK);
|
||||
let o = take(c, s);
|
||||
unsafe { zs_zset_free(o) };
|
||||
let snap = zset(&[(b"a", 1), (b"b", 1), (b"c", 1)]);
|
||||
let mut job = null_mut();
|
||||
assert_eq!(unsafe { zs_rebuild_new(c, i, snap, 1, &mut job) }, OK);
|
||||
assert_eq!(unsafe { zs_circuit_begin(c) }, OK);
|
||||
let mut done = 0;
|
||||
assert_eq!(
|
||||
unsafe { zs_rebuild_step(job, c, None, null_mut(), &mut done) },
|
||||
Code::Overflow as i32
|
||||
);
|
||||
assert_eq!(unsafe { zs_circuit_rollback(c) }, OK);
|
||||
let rc = unsafe { zs_rebuild_step(job, c, None, null_mut(), &mut done) };
|
||||
unsafe { zs_rebuild_free(job) };
|
||||
unsafe { zs_zset_free(snap) };
|
||||
unsafe { zs_zset_free(big) };
|
||||
unsafe { zs_circuit_free(c) };
|
||||
assert_eq!(
|
||||
rc,
|
||||
Code::Interrupted as i32,
|
||||
"[f4] header: a rebuild stepped in a rolled-back txn reports ZS_ERR_INTERRUPTED; got rc={rc} done={done}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn f5_a_chunk_whose_push_overflows_in_a_txn_then_rollback_is_interrupted() {
|
||||
let (c, i, _s) = rig();
|
||||
let pending = zset(&[(b"a", i64::MAX)]);
|
||||
assert_eq!(unsafe { zs_circuit_push(c, i, pending) }, OK);
|
||||
let snap = zset(&[(b"a", 1), (b"b", 1)]);
|
||||
let mut job = null_mut();
|
||||
assert_eq!(unsafe { zs_rebuild_new(c, i, snap, 1, &mut job) }, OK);
|
||||
assert_eq!(unsafe { zs_circuit_begin(c) }, OK);
|
||||
let mut done = 0;
|
||||
assert_eq!(
|
||||
unsafe { zs_rebuild_step(job, c, None, null_mut(), &mut done) },
|
||||
Code::Overflow as i32,
|
||||
"the chunk's push overflows the pending delta"
|
||||
);
|
||||
assert_eq!(unsafe { zs_circuit_rollback(c) }, OK);
|
||||
let rc = unsafe { zs_rebuild_step(job, c, None, null_mut(), &mut done) };
|
||||
unsafe { zs_rebuild_free(job) };
|
||||
unsafe { zs_zset_free(snap) };
|
||||
unsafe { zs_zset_free(pending) };
|
||||
unsafe { zs_circuit_free(c) };
|
||||
assert_eq!(rc, Code::Interrupted as i32, "[f5] got rc={rc} done={done}");
|
||||
}
|
||||
@@ -48,6 +48,8 @@
|
||||
(defcfun "zs_circuit_filter" :int32 (c :pointer) (stream :uint32) (keep :pointer) (ctx :pointer) (out :pointer))
|
||||
(defcfun "zs_circuit_push" :int32 (c :pointer) (input :uint32) (z :pointer))
|
||||
(defcfun "zs_circuit_step" :int32 (c :pointer))
|
||||
(defcfun "zs_circuit_begin" :int32 (c :pointer))
|
||||
(defcfun "zs_circuit_rollback" :int32 (c :pointer))
|
||||
|
||||
(defcallback keep-or-fail :int32 ((ctx :pointer) (key :pointer) (len :size))
|
||||
(declare (ignore ctx key len))
|
||||
@@ -61,9 +63,13 @@
|
||||
(check "filter" (zs-circuit-filter c (mem-ref id :uint32) (callback keep-or-fail) (null-pointer) id) 0)
|
||||
(check "zset" (zs-zset-new zout) 0)
|
||||
(key-call #'zs-zset-insert (mem-ref zout :pointer) "k" 1)
|
||||
(check "begin a transaction" (zs-circuit-begin c) 0)
|
||||
(check "push" (zs-circuit-push c 0 (mem-ref zout :pointer)) 0)
|
||||
(check "step reports the callback's failure" (zs-circuit-step c) 8)
|
||||
(check "the circuit is dead afterwards" (zs-circuit-step c) 7)
|
||||
(check "rollback revives it" (zs-circuit-rollback c) 0)
|
||||
(check "a transaction opens again" (zs-circuit-begin c) 0)
|
||||
(check "and a second begin is refused" (zs-circuit-begin c) 10)
|
||||
(zs-zset-free (mem-ref zout :pointer))
|
||||
(zs-circuit-free c)))
|
||||
(format t "zset smoke: all ok~%")
|
||||
|
||||
+394
-62
@@ -1,23 +1,30 @@
|
||||
//! The synchronous, single-threaded circuit. This module is the wiring (`Rc`, `RefCell`,
|
||||
//! boxed nodes); the operators' logic is in `ops.rs` and `lib.rs`.
|
||||
|
||||
use crate::ops::{AggregateState, DistinctState, JoinState};
|
||||
use crate::ops::{
|
||||
AggregateState, DistinctState, JoinState, Log, Transactional, Undoable, add_logged, unadd,
|
||||
};
|
||||
use crate::{MemTrace, Trace, ZSet};
|
||||
use std::cell::RefCell;
|
||||
use std::cell::{Ref, RefCell};
|
||||
use std::rc::Rc;
|
||||
|
||||
/// What an edge carries: the latest piece, and the whole value of the current tick.
|
||||
/// They differ only while a tick arrives in pieces (`Circuit::step_continue`).
|
||||
/// What an edge carries: the latest piece, and the whole tick's value, kept only while a
|
||||
/// tick is split (`Circuit::step_continue`); `None` means the piece is the whole.
|
||||
pub(crate) struct Slot<T> {
|
||||
piece: T,
|
||||
whole: T,
|
||||
whole: Option<T>,
|
||||
/// Inside a transaction: the `(piece, whole)` the first publish replaced.
|
||||
saved: Option<(T, Option<T>)>,
|
||||
open: bool,
|
||||
}
|
||||
|
||||
impl<T: Default> Default for Slot<T> {
|
||||
fn default() -> Self {
|
||||
Slot {
|
||||
piece: T::default(),
|
||||
whole: T::default(),
|
||||
whole: None,
|
||||
saved: None,
|
||||
open: false,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -41,18 +48,51 @@ impl<K: Ord + Clone> Piecewise for ZSet<K> {
|
||||
pub struct Stream<T>(Rc<RefCell<Slot<T>>>);
|
||||
|
||||
impl<T: Piecewise> Stream<T> {
|
||||
fn piece(&self) -> T {
|
||||
self.0.borrow().piece.clone()
|
||||
fn piece(&self) -> Ref<'_, T> {
|
||||
Ref::map(self.0.borrow(), |s| &s.piece)
|
||||
}
|
||||
|
||||
fn publish(&self, v: T, cont: bool) {
|
||||
let mut slot = self.0.borrow_mut();
|
||||
let slot = &mut *slot;
|
||||
if cont {
|
||||
slot.whole.absorb(&v);
|
||||
if slot.open && slot.saved.is_none() {
|
||||
slot.saved = Some((slot.piece.clone(), slot.whole.clone()));
|
||||
}
|
||||
let mut whole = slot.whole.take().unwrap_or_else(|| slot.piece.clone());
|
||||
whole.absorb(&v);
|
||||
slot.whole = Some(whole);
|
||||
slot.piece = v;
|
||||
} else {
|
||||
slot.whole = v.clone();
|
||||
let piece = std::mem::replace(&mut slot.piece, v);
|
||||
let whole = slot.whole.take();
|
||||
if slot.open && slot.saved.is_none() {
|
||||
slot.saved = Some((piece, whole));
|
||||
}
|
||||
}
|
||||
slot.piece = v;
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Stream<T> {
|
||||
fn begin(&self) {
|
||||
let mut slot = self.0.borrow_mut();
|
||||
slot.open = true;
|
||||
slot.saved = None;
|
||||
}
|
||||
|
||||
fn commit(&self) {
|
||||
let mut slot = self.0.borrow_mut();
|
||||
slot.open = false;
|
||||
slot.saved = None;
|
||||
}
|
||||
|
||||
fn rollback(&self) {
|
||||
let mut slot = self.0.borrow_mut();
|
||||
if let Some((piece, whole)) = slot.saved.take() {
|
||||
slot.piece = piece;
|
||||
slot.whole = whole;
|
||||
}
|
||||
slot.open = false;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -65,16 +105,18 @@ impl<T> Clone for Stream<T> {
|
||||
impl<T: Clone> Stream<T> {
|
||||
/// The value of the last tick, all its pieces together (the zero before the first).
|
||||
pub fn current(&self) -> T {
|
||||
self.0.borrow().whole.clone()
|
||||
let slot = self.0.borrow();
|
||||
slot.whole.as_ref().unwrap_or(&slot.piece).clone()
|
||||
}
|
||||
}
|
||||
|
||||
/// Where a caller feeds deltas; they enter the circuit at the next `step`.
|
||||
pub struct Input<K>(Rc<RefCell<ZSet<K>>>);
|
||||
/// Where a caller feeds deltas; they enter the circuit at the next `step`. It remembers
|
||||
/// which circuit made it, so a `Rebuild` can refuse a circuit that is not its own.
|
||||
pub struct Input<K>(Rc<RefCell<ZSet<K>>>, Rc<()>);
|
||||
|
||||
impl<K> Clone for Input<K> {
|
||||
fn clone(&self) -> Self {
|
||||
Input(Rc::clone(&self.0))
|
||||
Input(Rc::clone(&self.0), Rc::clone(&self.1))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -84,7 +126,8 @@ impl<K: Ord> Input<K> {
|
||||
}
|
||||
}
|
||||
|
||||
/// A sink that sums every step's delta until it is taken.
|
||||
/// A sink that sums every step's delta until it is taken. Take inside no transaction: a
|
||||
/// rollback subtracts the step's output, which a `take` in between has already removed.
|
||||
pub struct Output<K>(Rc<RefCell<ZSet<K>>>);
|
||||
|
||||
impl<K: Ord> Output<K> {
|
||||
@@ -93,78 +136,237 @@ impl<K: Ord> Output<K> {
|
||||
}
|
||||
}
|
||||
|
||||
trait Node {
|
||||
trait Node: Transactional {
|
||||
/// `cont` is true for a later piece of a tick that has already begun.
|
||||
fn eval(&mut self, cont: bool);
|
||||
}
|
||||
|
||||
struct Unary<I, O, F> {
|
||||
/// One input, one output, and the state `f` may change.
|
||||
struct Unary<I, O, S, F> {
|
||||
src: Stream<I>,
|
||||
out: Stream<O>,
|
||||
state: S,
|
||||
f: F,
|
||||
}
|
||||
|
||||
impl<I: Piecewise, O: Piecewise, F: FnMut(&I) -> O> Node for Unary<I, O, F> {
|
||||
impl<I, O, S: Transactional, F> Transactional for Unary<I, O, S, F> {
|
||||
fn begin(&mut self) {
|
||||
self.state.begin();
|
||||
self.out.begin();
|
||||
}
|
||||
|
||||
fn commit(&mut self) {
|
||||
self.state.commit();
|
||||
self.out.commit();
|
||||
}
|
||||
|
||||
fn rollback(&mut self) {
|
||||
self.state.rollback();
|
||||
self.out.rollback();
|
||||
}
|
||||
}
|
||||
|
||||
impl<I: Piecewise, O: Piecewise, S: Transactional, F: FnMut(&mut S, &I) -> O> Node
|
||||
for Unary<I, O, S, F>
|
||||
{
|
||||
fn eval(&mut self, cont: bool) {
|
||||
let v = (self.f)(&self.src.piece());
|
||||
let v = (self.f)(&mut self.state, &self.src.piece());
|
||||
self.out.publish(v, cont);
|
||||
}
|
||||
}
|
||||
|
||||
struct Binary<A, B, O, F> {
|
||||
struct Binary<A, B, O, S, F> {
|
||||
a: Stream<A>,
|
||||
b: Stream<B>,
|
||||
out: Stream<O>,
|
||||
state: S,
|
||||
f: F,
|
||||
}
|
||||
|
||||
impl<A: Piecewise, B: Piecewise, O: Piecewise, F: FnMut(&A, &B) -> O> Node for Binary<A, B, O, F> {
|
||||
impl<A, B, O, S: Transactional, F> Transactional for Binary<A, B, O, S, F> {
|
||||
fn begin(&mut self) {
|
||||
self.state.begin();
|
||||
self.out.begin();
|
||||
}
|
||||
|
||||
fn commit(&mut self) {
|
||||
self.state.commit();
|
||||
self.out.commit();
|
||||
}
|
||||
|
||||
fn rollback(&mut self) {
|
||||
self.state.rollback();
|
||||
self.out.rollback();
|
||||
}
|
||||
}
|
||||
|
||||
impl<A: Piecewise, B: Piecewise, O: Piecewise, S: Transactional, F: FnMut(&mut S, &A, &B) -> O> Node
|
||||
for Binary<A, B, O, S, F>
|
||||
{
|
||||
fn eval(&mut self, cont: bool) {
|
||||
let v = (self.f)(&self.a.piece(), &self.b.piece());
|
||||
let v = (self.f)(&mut self.state, &self.a.piece(), &self.b.piece());
|
||||
self.out.publish(v, cont);
|
||||
}
|
||||
}
|
||||
|
||||
/// What a `Delay` holds, and what it held when a transaction began.
|
||||
struct Held<K> {
|
||||
held: ZSet<K>,
|
||||
open: bool,
|
||||
saved: Option<ZSet<K>>,
|
||||
}
|
||||
|
||||
impl<K: Ord + Clone> Held<K> {
|
||||
/// Notes the value to go back to, once per transaction. `held` is one tick's delta.
|
||||
fn save(&mut self) {
|
||||
if self.open && self.saved.is_none() {
|
||||
self.saved = Some(self.held.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<K> Transactional for Held<K> {
|
||||
fn begin(&mut self) {
|
||||
self.open = true;
|
||||
self.saved = None;
|
||||
}
|
||||
|
||||
fn commit(&mut self) {
|
||||
self.open = false;
|
||||
self.saved = None;
|
||||
}
|
||||
|
||||
fn rollback(&mut self) {
|
||||
if let Some(held) = self.saved.take() {
|
||||
self.held = held;
|
||||
}
|
||||
self.open = false;
|
||||
}
|
||||
}
|
||||
|
||||
/// out(t) = in(t - 1). A continued piece adds to what is held and emits nothing.
|
||||
struct Delay<K> {
|
||||
src: Stream<ZSet<K>>,
|
||||
out: Stream<ZSet<K>>,
|
||||
held: ZSet<K>,
|
||||
state: Held<K>,
|
||||
}
|
||||
|
||||
impl<K> Transactional for Delay<K> {
|
||||
fn begin(&mut self) {
|
||||
self.state.begin();
|
||||
self.out.begin();
|
||||
}
|
||||
|
||||
fn commit(&mut self) {
|
||||
self.state.commit();
|
||||
self.out.commit();
|
||||
}
|
||||
|
||||
fn rollback(&mut self) {
|
||||
self.state.rollback();
|
||||
self.out.rollback();
|
||||
}
|
||||
}
|
||||
|
||||
impl<K: Ord + Clone> Node for Delay<K> {
|
||||
fn eval(&mut self, cont: bool) {
|
||||
let d = self.src.piece();
|
||||
self.state.save();
|
||||
let v = if cont {
|
||||
self.held += &d;
|
||||
self.state.held += &*d;
|
||||
ZSet::new()
|
||||
} else {
|
||||
std::mem::replace(&mut self.held, d)
|
||||
std::mem::replace(&mut self.state.held, d.clone())
|
||||
};
|
||||
drop(d);
|
||||
self.out.publish(v, cont);
|
||||
}
|
||||
}
|
||||
|
||||
/// The integral, and the log that takes a transaction's additions back.
|
||||
struct Acc<K> {
|
||||
acc: ZSet<K>,
|
||||
log: Log<(K, i64)>,
|
||||
}
|
||||
|
||||
impl<K: Ord> Transactional for Acc<K> {
|
||||
fn begin(&mut self) {
|
||||
self.log.begin();
|
||||
}
|
||||
|
||||
fn commit(&mut self) {
|
||||
self.log.commit();
|
||||
}
|
||||
|
||||
fn rollback(&mut self) {
|
||||
let acc = &mut self.acc;
|
||||
self.log.rollback(|e| unadd(acc, e));
|
||||
}
|
||||
}
|
||||
|
||||
/// out(t) = in(0) + .. + in(t). A continued piece emits only its own input.
|
||||
struct Integrate<K> {
|
||||
src: Stream<ZSet<K>>,
|
||||
out: Stream<ZSet<K>>,
|
||||
acc: ZSet<K>,
|
||||
state: Acc<K>,
|
||||
}
|
||||
|
||||
impl<K: Ord> Transactional for Integrate<K> {
|
||||
fn begin(&mut self) {
|
||||
self.state.begin();
|
||||
self.out.begin();
|
||||
}
|
||||
|
||||
fn commit(&mut self) {
|
||||
self.state.commit();
|
||||
self.out.commit();
|
||||
}
|
||||
|
||||
fn rollback(&mut self) {
|
||||
self.state.rollback();
|
||||
self.out.rollback();
|
||||
}
|
||||
}
|
||||
|
||||
impl<K: Ord + Clone> Node for Integrate<K> {
|
||||
fn eval(&mut self, cont: bool) {
|
||||
let d = self.src.piece();
|
||||
self.acc += &d;
|
||||
let v = if cont { d } else { self.acc.clone() };
|
||||
add_logged(&mut self.state.acc, &d, &mut self.state.log);
|
||||
let v = if cont {
|
||||
d.clone()
|
||||
} else {
|
||||
self.state.acc.clone()
|
||||
};
|
||||
drop(d);
|
||||
self.out.publish(v, cont);
|
||||
}
|
||||
}
|
||||
|
||||
/// Moves what was pushed into the circuit's stream.
|
||||
/// Moves what was pushed into the circuit's stream. Inside a transaction it remembers what
|
||||
/// was pending, so a rollback gives back the delta the failed step took.
|
||||
struct Source<K> {
|
||||
pending: Rc<RefCell<ZSet<K>>>,
|
||||
out: Stream<ZSet<K>>,
|
||||
saved: Option<ZSet<K>>,
|
||||
}
|
||||
|
||||
impl<K: Ord + Clone> Transactional for Source<K> {
|
||||
fn begin(&mut self) {
|
||||
self.saved = Some(self.pending.borrow().clone());
|
||||
self.out.begin();
|
||||
}
|
||||
|
||||
fn commit(&mut self) {
|
||||
self.saved = None;
|
||||
self.out.commit();
|
||||
}
|
||||
|
||||
fn rollback(&mut self) {
|
||||
if let Some(saved) = self.saved.take() {
|
||||
*self.pending.borrow_mut() = saved;
|
||||
}
|
||||
self.out.rollback();
|
||||
}
|
||||
}
|
||||
|
||||
impl<K: Ord + Clone> Node for Source<K> {
|
||||
@@ -174,12 +376,42 @@ impl<K: Ord + Clone> Node for Source<K> {
|
||||
}
|
||||
}
|
||||
|
||||
/// The sink's sum, and the log that takes a transaction's additions back.
|
||||
struct Sink<K> {
|
||||
sum: Rc<RefCell<ZSet<K>>>,
|
||||
log: Log<(K, i64)>,
|
||||
}
|
||||
|
||||
impl<K: Ord> Transactional for Sink<K> {
|
||||
fn begin(&mut self) {
|
||||
self.log.begin();
|
||||
}
|
||||
|
||||
fn commit(&mut self) {
|
||||
self.log.commit();
|
||||
}
|
||||
|
||||
fn rollback(&mut self) {
|
||||
let sum = &mut *self.sum.borrow_mut();
|
||||
self.log.rollback(|e| unadd(sum, e));
|
||||
}
|
||||
}
|
||||
|
||||
/// Nodes run in creation order, which is a topological order because a node can only
|
||||
/// be built from streams that already exist. There are no cycles to schedule.
|
||||
#[derive(Default)]
|
||||
pub struct Circuit {
|
||||
nodes: Vec<Box<dyn Node>>,
|
||||
ticks: u64,
|
||||
/// This circuit's identity: every `Input` it makes holds a clone.
|
||||
token: Rc<()>,
|
||||
/// `Some(ticks)` as of `begin`, while a transaction is open.
|
||||
open: Option<u64>,
|
||||
/// A step or piece ran since `begin`.
|
||||
stepped: bool,
|
||||
/// Rollbacks that undid a step or piece. It never goes down, unlike `ticks`, so a
|
||||
/// `Rebuild` can tell that a rollback took some of its chunks.
|
||||
undone: u64,
|
||||
}
|
||||
|
||||
type Z<K> = ZSet<K>;
|
||||
@@ -206,21 +438,103 @@ impl Circuit {
|
||||
self.run(true);
|
||||
}
|
||||
|
||||
/// Opens a transaction: state changes are logged until `commit` or `rollback`, at the
|
||||
/// cost of the keys touched. Panics if one is open. See `context/code/zset.md`.
|
||||
pub fn begin(&mut self) {
|
||||
assert!(
|
||||
self.open.is_none(),
|
||||
"zset: a transaction is already open on this circuit"
|
||||
);
|
||||
self.open = Some(self.ticks);
|
||||
self.stepped = false;
|
||||
for n in &mut self.nodes {
|
||||
n.begin();
|
||||
}
|
||||
}
|
||||
|
||||
/// Keeps what the transaction did. Panics if none is open.
|
||||
pub fn commit(&mut self) {
|
||||
assert!(
|
||||
self.open.take().is_some(),
|
||||
"zset: no transaction is open on this circuit"
|
||||
);
|
||||
for n in &mut self.nodes {
|
||||
n.commit();
|
||||
}
|
||||
}
|
||||
|
||||
/// Puts the state back to what it was at `begin`, even after a step panicked half way;
|
||||
/// pushes since are dropped. Panics if none is open.
|
||||
pub fn rollback(&mut self) {
|
||||
let ticks = self
|
||||
.open
|
||||
.take()
|
||||
.expect("zset: no transaction is open on this circuit");
|
||||
for n in &mut self.nodes {
|
||||
n.rollback();
|
||||
}
|
||||
self.ticks = ticks;
|
||||
self.undone += u64::from(self.stepped);
|
||||
self.stepped = false;
|
||||
}
|
||||
|
||||
fn run(&mut self, cont: bool) {
|
||||
self.stepped |= self.open.is_some();
|
||||
for n in &mut self.nodes {
|
||||
n.eval(cont);
|
||||
}
|
||||
}
|
||||
|
||||
/// Adds a node; one built inside a transaction starts recording at once.
|
||||
fn add(&mut self, mut node: Box<dyn Node>) {
|
||||
if self.open.is_some() {
|
||||
node.begin();
|
||||
}
|
||||
self.nodes.push(node);
|
||||
}
|
||||
|
||||
fn unary_with<I: Piecewise + 'static, O: Piecewise + 'static, S: Transactional + 'static>(
|
||||
&mut self,
|
||||
src: &Stream<I>,
|
||||
state: S,
|
||||
f: impl FnMut(&mut S, &I) -> O + 'static,
|
||||
) -> Stream<O> {
|
||||
let out = Stream(Rc::default());
|
||||
self.add(Box::new(Unary {
|
||||
src: src.clone(),
|
||||
out: out.clone(),
|
||||
state,
|
||||
f,
|
||||
}));
|
||||
out
|
||||
}
|
||||
|
||||
fn unary<I: Piecewise + 'static, O: Piecewise + 'static>(
|
||||
&mut self,
|
||||
src: &Stream<I>,
|
||||
f: impl FnMut(&I) -> O + 'static,
|
||||
mut f: impl FnMut(&I) -> O + 'static,
|
||||
) -> Stream<O> {
|
||||
self.unary_with(src, (), move |_, d| f(d))
|
||||
}
|
||||
|
||||
fn binary_with<
|
||||
A: Piecewise + 'static,
|
||||
B: Piecewise + 'static,
|
||||
O: Piecewise + 'static,
|
||||
S: Transactional + 'static,
|
||||
>(
|
||||
&mut self,
|
||||
a: &Stream<A>,
|
||||
b: &Stream<B>,
|
||||
state: S,
|
||||
f: impl FnMut(&mut S, &A, &B) -> O + 'static,
|
||||
) -> Stream<O> {
|
||||
let out = Stream(Rc::default());
|
||||
self.nodes.push(Box::new(Unary {
|
||||
src: src.clone(),
|
||||
self.add(Box::new(Binary {
|
||||
a: a.clone(),
|
||||
b: b.clone(),
|
||||
out: out.clone(),
|
||||
state,
|
||||
f,
|
||||
}));
|
||||
out
|
||||
@@ -230,33 +544,30 @@ impl Circuit {
|
||||
&mut self,
|
||||
a: &Stream<A>,
|
||||
b: &Stream<B>,
|
||||
f: impl FnMut(&A, &B) -> O + 'static,
|
||||
mut f: impl FnMut(&A, &B) -> O + 'static,
|
||||
) -> Stream<O> {
|
||||
let out = Stream(Rc::default());
|
||||
self.nodes.push(Box::new(Binary {
|
||||
a: a.clone(),
|
||||
b: b.clone(),
|
||||
out: out.clone(),
|
||||
f,
|
||||
}));
|
||||
out
|
||||
self.binary_with(a, b, (), move |_, x, y| f(x, y))
|
||||
}
|
||||
|
||||
pub fn input<K: Ord + Clone + 'static>(&mut self) -> (Input<K>, Stream<Z<K>>) {
|
||||
let pending: Rc<RefCell<Z<K>>> = Rc::default();
|
||||
let out = Stream(Rc::default());
|
||||
self.nodes.push(Box::new(Source {
|
||||
self.add(Box::new(Source {
|
||||
pending: Rc::clone(&pending),
|
||||
out: out.clone(),
|
||||
saved: None,
|
||||
}));
|
||||
(Input(pending), out)
|
||||
(Input(pending, Rc::clone(&self.token)), out)
|
||||
}
|
||||
|
||||
pub fn output<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Output<K> {
|
||||
let sink = Output(Rc::default());
|
||||
let acc = Rc::clone(&sink.0);
|
||||
self.unary(s, move |d: &Z<K>| {
|
||||
*acc.borrow_mut() += d;
|
||||
let state = Sink {
|
||||
sum: Rc::clone(&sink.0),
|
||||
log: Log::default(),
|
||||
};
|
||||
self.unary_with(s, state, |sink, d: &Z<K>| {
|
||||
add_logged(&mut sink.sum.borrow_mut(), d, &mut sink.log);
|
||||
});
|
||||
sink
|
||||
}
|
||||
@@ -264,10 +575,14 @@ impl Circuit {
|
||||
/// out(t) = in(t - 1); out(0) is zero.
|
||||
pub fn delay<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Stream<Z<K>> {
|
||||
let out = Stream(Rc::default());
|
||||
self.nodes.push(Box::new(Delay {
|
||||
self.add(Box::new(Delay {
|
||||
src: s.clone(),
|
||||
out: out.clone(),
|
||||
held: ZSet::new(),
|
||||
state: Held {
|
||||
held: ZSet::new(),
|
||||
open: false,
|
||||
saved: None,
|
||||
},
|
||||
}));
|
||||
out
|
||||
}
|
||||
@@ -275,10 +590,13 @@ impl Circuit {
|
||||
/// out(t) = in(0) + .. + in(t).
|
||||
pub fn integrate<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Stream<Z<K>> {
|
||||
let out = Stream(Rc::default());
|
||||
self.nodes.push(Box::new(Integrate {
|
||||
self.add(Box::new(Integrate {
|
||||
src: s.clone(),
|
||||
out: out.clone(),
|
||||
acc: ZSet::new(),
|
||||
state: Acc {
|
||||
acc: ZSet::new(),
|
||||
log: Log::default(),
|
||||
},
|
||||
}));
|
||||
out
|
||||
}
|
||||
@@ -337,8 +655,7 @@ impl Circuit {
|
||||
|
||||
/// Incremental `distinct`.
|
||||
pub fn distinct<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Stream<Z<K>> {
|
||||
let mut st = DistinctState::new();
|
||||
self.unary(s, move |d: &Z<K>| st.step(d))
|
||||
self.unary_with(s, DistinctState::new(), |st, d: &Z<K>| st.step(d))
|
||||
}
|
||||
|
||||
/// The bilinear delta join over in-memory traces.
|
||||
@@ -374,8 +691,8 @@ impl Circuit {
|
||||
TA: Trace<K, A> + 'static,
|
||||
TB: Trace<K, B> + 'static,
|
||||
{
|
||||
let mut st = JoinState::new(left, right);
|
||||
self.binary(a, b, move |da: &Z<(K, A)>, db: &Z<(K, B)>| {
|
||||
let st = JoinState::new(Undoable::new(left), Undoable::new(right));
|
||||
self.binary_with(a, b, st, move |st, da: &Z<(K, A)>, db: &Z<(K, B)>| {
|
||||
st.step(da, db, &mut f)
|
||||
})
|
||||
}
|
||||
@@ -407,8 +724,8 @@ impl Circuit {
|
||||
R: Ord + Clone + 'static,
|
||||
T: Trace<K, V> + 'static,
|
||||
{
|
||||
let mut st = AggregateState::new(groups);
|
||||
self.unary(s, move |d: &Z<(K, V)>| st.step(d, &mut f))
|
||||
let st = AggregateState::new(Undoable::new(groups));
|
||||
self.unary_with(s, st, move |st, d: &Z<(K, V)>| st.step(d, &mut f))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -420,13 +737,15 @@ pub enum Progress {
|
||||
}
|
||||
|
||||
/// A snapshot load as a sliced job: one logical tick (`context/code/zset.md`). A pending
|
||||
/// job owns the clock: a `Circuit::step` before it is Done makes `step` panic.
|
||||
/// job owns the clock: a `Circuit::step` before it is Done makes `step` panic, for good.
|
||||
pub struct Rebuild<K> {
|
||||
input: Input<K>,
|
||||
rest: std::collections::btree_map::IntoIter<K, i64>,
|
||||
chunk: usize,
|
||||
/// The circuit's tick count right after this job's first chunk, once it has run one.
|
||||
began_at: Option<u64>,
|
||||
/// The circuit's `undone` count when this job first ran, even if that chunk panicked.
|
||||
undone_at: Option<u64>,
|
||||
}
|
||||
|
||||
impl<K: Ord> Rebuild<K> {
|
||||
@@ -437,27 +756,40 @@ impl<K: Ord> Rebuild<K> {
|
||||
rest: snapshot.into_iter(),
|
||||
chunk: chunk.max(1),
|
||||
began_at: None,
|
||||
undone_at: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether a `Circuit::step` since this job began has split its tick. A finished job
|
||||
/// cannot be interrupted.
|
||||
/// Whether `circuit` is the one that made this job's input.
|
||||
pub fn belongs_to(&self, circuit: &Circuit) -> bool {
|
||||
Rc::ptr_eq(&self.input.1, &circuit.token)
|
||||
}
|
||||
|
||||
/// Whether a `Circuit::step` split this job's tick, or a rollback undid a step after it
|
||||
/// first ran (finished or not). An interrupted job never finishes: start another.
|
||||
pub fn interrupted(&self, circuit: &Circuit) -> bool {
|
||||
self.rest.len() > 0 && self.began_at.is_some_and(|t| t != circuit.ticks())
|
||||
let split = self.rest.len() > 0 && self.began_at.is_some_and(|t| t != circuit.ticks());
|
||||
split || self.undone_at.is_some_and(|u| u != circuit.undone)
|
||||
}
|
||||
|
||||
/// Runs chunks until done or `expired()`, at least one per call. A predicate, not an
|
||||
/// `Instant` (wasm32). An empty snapshot takes its one tick; a finished job none.
|
||||
pub fn step(&mut self, circuit: &mut Circuit, mut expired: impl FnMut() -> bool) -> Progress {
|
||||
assert!(
|
||||
self.belongs_to(circuit),
|
||||
"zset: the rebuild's input belongs to another circuit"
|
||||
);
|
||||
self.undone_at.get_or_insert(circuit.undone);
|
||||
assert!(
|
||||
!self.interrupted(circuit),
|
||||
"zset: the rebuild was interrupted by a Circuit::step before it finished"
|
||||
"zset: the rebuild was interrupted by a Circuit::step or a rollback before it finished"
|
||||
);
|
||||
loop {
|
||||
let chunk: ZSet<K> = self.rest.by_ref().take(self.chunk).collect();
|
||||
if chunk.is_zero() && self.began_at.is_some() {
|
||||
return Progress::Done;
|
||||
}
|
||||
circuit.stepped |= circuit.open.is_some();
|
||||
self.input.push(chunk);
|
||||
if self.began_at.is_some() {
|
||||
circuit.step_continue();
|
||||
|
||||
@@ -91,7 +91,7 @@ impl<K: Ord> ZSet<K> {
|
||||
self.adjust(key, weight, false);
|
||||
}
|
||||
|
||||
fn adjust(&mut self, key: K, weight: i64, subtract: bool) {
|
||||
pub(crate) fn adjust(&mut self, key: K, weight: i64, subtract: bool) {
|
||||
match self.weights.entry(key) {
|
||||
Entry::Vacant(e) => {
|
||||
let from_zero = if subtract { negated(weight) } else { weight };
|
||||
|
||||
+183
-1
@@ -4,15 +4,95 @@
|
||||
use crate::{Trace, ZSet, plus, times};
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
/// State a transaction can undo (`Circuit::begin`): `begin` starts recording, `commit`
|
||||
/// forgets the record, `rollback` replays it backwards. The cost is the delta, not the state.
|
||||
pub(crate) trait Transactional {
|
||||
fn begin(&mut self);
|
||||
fn commit(&mut self);
|
||||
fn rollback(&mut self);
|
||||
}
|
||||
|
||||
impl Transactional for () {
|
||||
fn begin(&mut self) {}
|
||||
fn commit(&mut self) {}
|
||||
fn rollback(&mut self) {}
|
||||
}
|
||||
|
||||
/// An undo log that records only while a transaction is open.
|
||||
#[derive(Clone, Debug)]
|
||||
pub(crate) struct Log<E> {
|
||||
on: bool,
|
||||
entries: Vec<E>,
|
||||
}
|
||||
|
||||
impl<E> Default for Log<E> {
|
||||
fn default() -> Self {
|
||||
Log {
|
||||
on: false,
|
||||
entries: Vec::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<E> Log<E> {
|
||||
pub(crate) fn begin(&mut self) {
|
||||
self.on = true;
|
||||
self.entries.clear();
|
||||
}
|
||||
|
||||
pub(crate) fn commit(&mut self) {
|
||||
self.on = false;
|
||||
self.entries = Vec::new();
|
||||
}
|
||||
|
||||
/// Builds the entry if a transaction is open. Build it before the change it describes,
|
||||
/// so a panicking `Clone` cannot leave a change unlogged.
|
||||
pub(crate) fn entry(&self, entry: impl FnOnce() -> E) -> Option<E> {
|
||||
self.on.then(entry)
|
||||
}
|
||||
|
||||
/// Logs an entry made by `entry`, once its change has been made.
|
||||
pub(crate) fn push(&mut self, entry: Option<E>) {
|
||||
self.entries.extend(entry);
|
||||
}
|
||||
|
||||
/// Hands the entries to `undo`, newest first, and closes the log.
|
||||
pub(crate) fn rollback(&mut self, mut undo: impl FnMut(E)) {
|
||||
self.on = false;
|
||||
for e in std::mem::take(&mut self.entries).into_iter().rev() {
|
||||
undo(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// `z += d`, one key at a time, logging each key that was applied. A key whose addition
|
||||
/// overflows panics before it changes anything, so the log is exactly what was applied.
|
||||
pub(crate) fn add_logged<K: Ord + Clone>(z: &mut ZSet<K>, d: &ZSet<K>, log: &mut Log<(K, i64)>) {
|
||||
for (k, w) in d.iter() {
|
||||
let entry = log.entry(|| (k.clone(), w));
|
||||
z.insert(k.clone(), w);
|
||||
log.push(entry);
|
||||
}
|
||||
}
|
||||
|
||||
/// Takes one logged addition back. The result is the weight before it, which was in range.
|
||||
pub(crate) fn unadd<K: Ord>(z: &mut ZSet<K>, (k, w): (K, i64)) {
|
||||
z.adjust(k, w, true);
|
||||
}
|
||||
|
||||
/// Incremental `distinct`: out = distinct(held + d) - distinct(held), per touched key.
|
||||
#[derive(Clone, Debug, Default)]
|
||||
pub struct DistinctState<K> {
|
||||
held: ZSet<K>,
|
||||
log: Log<(K, i64)>,
|
||||
}
|
||||
|
||||
impl<K: Ord + Clone> DistinctState<K> {
|
||||
pub fn new() -> Self {
|
||||
DistinctState { held: ZSet::new() }
|
||||
DistinctState {
|
||||
held: ZSet::new(),
|
||||
log: Log::default(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn step(&mut self, d: &ZSet<K>) -> ZSet<K> {
|
||||
@@ -21,12 +101,84 @@ impl<K: Ord + Clone> DistinctState<K> {
|
||||
let before = self.held.weight(k);
|
||||
let after = plus(before, w);
|
||||
out.insert(k.clone(), i64::from(after > 0) - i64::from(before > 0));
|
||||
let entry = self.log.entry(|| (k.clone(), w));
|
||||
self.held.insert(k.clone(), w);
|
||||
self.log.push(entry);
|
||||
}
|
||||
out
|
||||
}
|
||||
}
|
||||
|
||||
impl<K: Ord> Transactional for DistinctState<K> {
|
||||
fn begin(&mut self) {
|
||||
self.log.begin();
|
||||
}
|
||||
|
||||
fn commit(&mut self) {
|
||||
self.log.commit();
|
||||
}
|
||||
|
||||
fn rollback(&mut self) {
|
||||
let held = &mut self.held;
|
||||
self.log.rollback(|e| unadd(held, e));
|
||||
}
|
||||
}
|
||||
|
||||
/// A trace that logs what it applies, so a transaction can take it back.
|
||||
#[derive(Clone, Debug)]
|
||||
pub(crate) struct Undoable<T, K, V> {
|
||||
inner: T,
|
||||
log: Log<(K, V, i64)>,
|
||||
}
|
||||
|
||||
impl<T, K, V> Undoable<T, K, V> {
|
||||
pub(crate) fn new(inner: T) -> Self {
|
||||
Undoable {
|
||||
inner,
|
||||
log: Log::default(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Trace<K, V>, K: Clone, V: Clone> Trace<K, V> for Undoable<T, K, V> {
|
||||
fn apply(&mut self, k: &K, v: &V, w: i64) {
|
||||
let entry = self.log.entry(|| (k.clone(), v.clone(), w));
|
||||
self.inner.apply(k, v, w);
|
||||
self.log.push(entry);
|
||||
}
|
||||
|
||||
fn lookup(&self, k: &K) -> Vec<(V, i64)> {
|
||||
self.inner.lookup(k)
|
||||
}
|
||||
|
||||
fn range(&self, lo: std::ops::Bound<&K>, hi: std::ops::Bound<&K>) -> Vec<(K, V, i64)> {
|
||||
self.inner.range(lo, hi)
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Trace<K, V>, K, V> Transactional for Undoable<T, K, V> {
|
||||
fn begin(&mut self) {
|
||||
self.log.begin();
|
||||
}
|
||||
|
||||
fn commit(&mut self) {
|
||||
self.log.commit();
|
||||
}
|
||||
|
||||
fn rollback(&mut self) {
|
||||
let inner = &mut self.inner;
|
||||
self.log.rollback(|(k, v, w)| match w.checked_neg() {
|
||||
Some(n) => inner.apply(&k, &v, n),
|
||||
// -i64::MIN is 2^63: add it as i64::MAX then 1. The weight before the apply
|
||||
// was non-negative, so neither step leaves i64.
|
||||
None => {
|
||||
inner.apply(&k, &v, i64::MAX);
|
||||
inner.apply(&k, &v, 1);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/// The bilinear delta join of two keyed streams:
|
||||
/// out = da x b_prev + a_prev x db + da x db, never a rescan of either side.
|
||||
#[derive(Clone, Debug, Default)]
|
||||
@@ -75,6 +227,22 @@ impl<TA, TB> JoinState<TA, TB> {
|
||||
}
|
||||
}
|
||||
|
||||
impl<TA: Transactional, TB: Transactional> Transactional for JoinState<TA, TB> {
|
||||
fn begin(&mut self) {
|
||||
self.left.begin();
|
||||
self.right.begin();
|
||||
}
|
||||
|
||||
fn commit(&mut self) {
|
||||
self.left.commit();
|
||||
self.right.commit();
|
||||
}
|
||||
|
||||
fn rollback(&mut self) {
|
||||
self.left.rollback();
|
||||
self.right.rollback();
|
||||
}
|
||||
}
|
||||
/// Incremental group-by: `f` folds one key's integrated values to at most one result.
|
||||
/// An empty group has no row, and `f` is not asked about it. Only touched keys are re-folded: out = -(k, f(old)) + (k, f(new)).
|
||||
#[derive(Clone, Debug, Default)]
|
||||
@@ -122,3 +290,17 @@ impl<T> AggregateState<T> {
|
||||
out
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Transactional> Transactional for AggregateState<T> {
|
||||
fn begin(&mut self) {
|
||||
self.groups.begin();
|
||||
}
|
||||
|
||||
fn commit(&mut self) {
|
||||
self.groups.commit();
|
||||
}
|
||||
|
||||
fn rollback(&mut self) {
|
||||
self.groups.rollback();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,7 +7,8 @@ use std::ops::Bound;
|
||||
/// An integrated `ZSet<(K, V)>`, indexed by `K`. Reads return owned values, so an
|
||||
/// implementation may page or compress what it holds. See `context/code/zset.md`.
|
||||
pub trait Trace<K, V> {
|
||||
/// Adds `w` to the weight of `(k, v)`.
|
||||
/// Adds `w` to the weight of `(k, v)`. A panic must leave the trace as it was: a
|
||||
/// transaction (`Circuit::begin`) logs an apply only once it has returned.
|
||||
fn apply(&mut self, k: &K, v: &V, w: i64);
|
||||
/// The values at `k` with their non-zero weights, ascending by value.
|
||||
fn lookup(&self, k: &K) -> Vec<(V, i64)>;
|
||||
@@ -52,10 +53,19 @@ fn is_empty_range<K: Ord>(lo: Bound<&K>, hi: Bound<&K>) -> bool {
|
||||
|
||||
impl<K: Ord + Clone, V: Ord + Clone> Trace<K, V> for MemTrace<K, V> {
|
||||
fn apply(&mut self, k: &K, v: &V, w: i64) {
|
||||
let group = self.groups.entry(k.clone()).or_default();
|
||||
group.insert(v.clone(), w);
|
||||
if group.is_zero() {
|
||||
self.groups.remove(k);
|
||||
// Every clone is made before the map changes, so a panicking one changes nothing.
|
||||
match self.groups.get_mut(k) {
|
||||
Some(group) => {
|
||||
group.insert(v.clone(), w);
|
||||
if group.is_zero() {
|
||||
self.groups.remove(k);
|
||||
}
|
||||
}
|
||||
None if w != 0 => {
|
||||
let group = ZSet::singleton(v.clone(), w);
|
||||
self.groups.insert(k.clone(), group);
|
||||
}
|
||||
None => {}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,187 @@
|
||||
//! A panicking `Clone` anywhere in a transactional step is rolled back, and `MemTrace::apply`
|
||||
//! survives one. Evidence: `context/test-evidence/zset.md`.
|
||||
|
||||
use std::cell::Cell;
|
||||
use std::panic::{AssertUnwindSafe, catch_unwind};
|
||||
use zset::{Circuit, Input, MemTrace, Output, Stream, Trace, ZSet};
|
||||
|
||||
type Z<K> = ZSet<K>;
|
||||
|
||||
fn quiet() {
|
||||
static O: std::sync::Once = std::sync::Once::new();
|
||||
O.call_once(|| {
|
||||
let prev = std::panic::take_hook();
|
||||
std::panic::set_hook(Box::new(move |info| {
|
||||
let s = format!("{info}");
|
||||
if !(s.contains("tripped") || s.contains("overflow")) {
|
||||
prev(info);
|
||||
}
|
||||
}));
|
||||
});
|
||||
}
|
||||
|
||||
thread_local! { static CLONE_TRIP: Cell<Option<u32>> = const { Cell::new(None) }; }
|
||||
|
||||
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
|
||||
struct Boom(u8);
|
||||
|
||||
impl Clone for Boom {
|
||||
fn clone(&self) -> Self {
|
||||
CLONE_TRIP.with(|t| {
|
||||
if let Some(n) = t.get() {
|
||||
if n == 0 {
|
||||
t.set(None);
|
||||
panic!("clone tripped");
|
||||
}
|
||||
t.set(Some(n - 1));
|
||||
}
|
||||
});
|
||||
Boom(self.0)
|
||||
}
|
||||
}
|
||||
|
||||
type B = (Boom, Boom);
|
||||
|
||||
struct Rig {
|
||||
c: Circuit,
|
||||
a: Input<B>,
|
||||
b: Input<B>,
|
||||
streams: Vec<Stream<Z<B>>>,
|
||||
agg: Stream<Z<(Boom, u8)>>,
|
||||
sinks: Vec<Output<B>>,
|
||||
}
|
||||
|
||||
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()];
|
||||
if which & 1 != 0 {
|
||||
streams.push(c.integrate(&sa));
|
||||
}
|
||||
if which & 2 != 0 {
|
||||
streams.push(c.distinct(&sa));
|
||||
}
|
||||
if which & 4 != 0 {
|
||||
streams.push(c.join_indexed(&sa, &sb, |k, x, y| (Boom(k.0), Boom(x.0 ^ y.0))));
|
||||
}
|
||||
if which & 8 != 0 {
|
||||
streams.push(c.delay(&sa));
|
||||
}
|
||||
let agg = c.aggregate(&sa, |_, vals: &[(Boom, i64)]| Some(vals.len() as u8));
|
||||
let sinks = streams.iter().map(|s| c.output(s)).collect();
|
||||
Rig {
|
||||
c,
|
||||
a,
|
||||
b,
|
||||
streams,
|
||||
agg,
|
||||
sinks,
|
||||
}
|
||||
}
|
||||
|
||||
impl Rig {
|
||||
fn view(&self) -> String {
|
||||
format!(
|
||||
"{:?} {:?}",
|
||||
self.streams.iter().map(|s| s.current()).collect::<Vec<_>>(),
|
||||
self.agg.current()
|
||||
)
|
||||
}
|
||||
fn drain(&self) -> String {
|
||||
format!(
|
||||
"{:?}",
|
||||
self.sinks.iter().map(|s| s.take()).collect::<Vec<_>>()
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
fn bz(v: &[(u8, u8, i64)]) -> Z<B> {
|
||||
v.iter()
|
||||
.map(|(k, x, w)| ((Boom(*k), Boom(*x)), *w))
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn clone_panic_sweep(which: u8) -> Vec<String> {
|
||||
quiet();
|
||||
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)]));
|
||||
}
|
||||
x.c.step();
|
||||
y.c.step();
|
||||
x.drain();
|
||||
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)]));
|
||||
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();
|
||||
CLONE_TRIP.with(|t| t.set(None));
|
||||
if !fired {
|
||||
assert!(r.is_ok());
|
||||
break;
|
||||
}
|
||||
x.c.rollback();
|
||||
let (v1, d1) = (x.view(), x.drain());
|
||||
if v1 != v0 || d1 != d0 {
|
||||
failures.push(format!(
|
||||
"n={n}: rollback state differs\n view {v1}\n want {v0}\n drain {d1}"
|
||||
));
|
||||
continue;
|
||||
}
|
||||
for r in [&x, &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)]));
|
||||
}
|
||||
x.c.step();
|
||||
y.c.step();
|
||||
if x.view() != y.view() || x.drain() != y.drain() {
|
||||
failures.push(format!("n={n}: post tick differs"));
|
||||
}
|
||||
}
|
||||
failures
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn c1_a_clone_panic_anywhere_in_a_step_is_rolled_back() {
|
||||
let mut all = vec![];
|
||||
for which in [1u8, 2, 4, 8, 15] {
|
||||
for f in clone_panic_sweep(which) {
|
||||
all.push(format!("which={which} {f}"));
|
||||
}
|
||||
}
|
||||
assert!(
|
||||
all.is_empty(),
|
||||
"{} failures:\n{}",
|
||||
all.len(),
|
||||
all.join("\n")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_panicking_value_clone_leaves_a_memtrace_unchanged() {
|
||||
quiet();
|
||||
let mut t: MemTrace<u8, Boom> = MemTrace::new();
|
||||
CLONE_TRIP.with(|c| c.set(Some(0)));
|
||||
let r = catch_unwind(AssertUnwindSafe(|| t.apply(&1, &Boom(1), 1)));
|
||||
CLONE_TRIP.with(|c| c.set(None));
|
||||
assert!(r.is_err());
|
||||
assert_eq!(
|
||||
t.keys(),
|
||||
0,
|
||||
"[c2] MemTrace::apply panicked but left a group"
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,173 @@
|
||||
//! What the circuit copies and holds, counted in key clones and live keys, never in time.
|
||||
//! Evidence: `context/test-evidence/zset.md`.
|
||||
|
||||
use std::cell::Cell;
|
||||
use zset::{Circuit, ZSet};
|
||||
|
||||
thread_local! {
|
||||
static CLONES: Cell<usize> = const { Cell::new(0) };
|
||||
static LIVE: Cell<isize> = const { Cell::new(0) };
|
||||
}
|
||||
|
||||
/// A key that counts its clones and how many of it are alive on this thread.
|
||||
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
|
||||
struct Counted(u32);
|
||||
|
||||
impl Counted {
|
||||
fn new(n: u32) -> Self {
|
||||
LIVE.with(|l| l.set(l.get() + 1));
|
||||
Counted(n)
|
||||
}
|
||||
}
|
||||
|
||||
impl Clone for Counted {
|
||||
fn clone(&self) -> Self {
|
||||
CLONES.with(|c| c.set(c.get() + 1));
|
||||
Counted::new(self.0)
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Counted {
|
||||
fn drop(&mut self) {
|
||||
LIVE.with(|l| l.set(l.get() - 1));
|
||||
}
|
||||
}
|
||||
|
||||
fn clones() -> usize {
|
||||
CLONES.with(Cell::get)
|
||||
}
|
||||
|
||||
fn live() -> isize {
|
||||
LIVE.with(Cell::get)
|
||||
}
|
||||
|
||||
fn keys(n: u32) -> ZSet<Counted> {
|
||||
(0..n).map(|i| (Counted::new(i), 1)).collect()
|
||||
}
|
||||
|
||||
const N: u32 = 100;
|
||||
|
||||
#[test]
|
||||
fn an_eval_borrows_its_input_instead_of_cloning_it() {
|
||||
let mut c = Circuit::new();
|
||||
let (input, s) = c.input::<Counted>();
|
||||
let m = c.map(&s, |k| k.0);
|
||||
let sink = c.output(&m);
|
||||
input.push(keys(N));
|
||||
let before = clones();
|
||||
c.step();
|
||||
assert_eq!(
|
||||
clones() - before,
|
||||
0,
|
||||
"[piece] a step over {N} keys cloned keys"
|
||||
);
|
||||
assert_eq!(sink.take().len(), N as usize);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_integrate_stream_holds_its_integral_twice_when_the_tick_is_whole() {
|
||||
let base = live();
|
||||
let mut c = Circuit::new();
|
||||
let (input, s) = c.input::<Counted>();
|
||||
let i = c.integrate(&s);
|
||||
input.push(keys(N));
|
||||
c.step();
|
||||
// The input's piece (N), the integral (N) and the integrate stream's piece (N).
|
||||
assert_eq!(
|
||||
live() - base,
|
||||
3 * N as isize,
|
||||
"[whole] live keys after one step"
|
||||
);
|
||||
drop((c, input, s, i));
|
||||
assert_eq!(live(), base, "[leak] every key is dropped with the circuit");
|
||||
}
|
||||
|
||||
type E = (Counted, u8);
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
enum Mode {
|
||||
Plain,
|
||||
Commit,
|
||||
Rollback,
|
||||
}
|
||||
|
||||
/// Every stateful operator over `Counted` keys, loaded with `n` keys; then one step of a
|
||||
/// two-key delta in `mode`. Returns the key clones that step (and `begin`/`end`) made.
|
||||
fn clones_of_one_step(n: u32, mode: Mode) -> usize {
|
||||
let mut c = Circuit::new();
|
||||
let (a, sa) = c.input::<E>();
|
||||
let (b, sb) = c.input::<E>();
|
||||
let d = c.distinct(&sa);
|
||||
let dl = c.delay(&sa);
|
||||
let j = c.join_indexed(&sa, &sb, |k, x, y| (k.0, *x, *y));
|
||||
let g = c.aggregate(&sa, |_, vals| Some(vals.len() as i64));
|
||||
let _ig = c.integrate(&sa);
|
||||
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.step();
|
||||
for z in [&sinks.0, &sinks.1] {
|
||||
z.take();
|
||||
}
|
||||
sinks.2.take();
|
||||
sinks.3.take();
|
||||
// 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.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.step();
|
||||
match mode {
|
||||
Mode::Plain => {}
|
||||
Mode::Commit => c.commit(),
|
||||
Mode::Rollback => c.rollback(),
|
||||
}
|
||||
clones() - before
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_transaction_costs_the_keys_a_step_touches_and_not_the_size_of_the_circuit() {
|
||||
for mode in [Mode::Commit, Mode::Rollback] {
|
||||
let extra = |n| clones_of_one_step(n, mode) - clones_of_one_step(n, Mode::Plain);
|
||||
let (small, large) = (extra(10), extra(300));
|
||||
assert!(small > 0, "[cost] the log copies the keys it records");
|
||||
assert_eq!(small, large, "[cost] the overhead grew with the state");
|
||||
}
|
||||
}
|
||||
|
||||
/// Clones made by one 1-key `step_continue` over an integral of `n` keys.
|
||||
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.step();
|
||||
let before = clones();
|
||||
if txn {
|
||||
c.begin();
|
||||
}
|
||||
input.push((n..n + 1).map(|k| (Counted::new(k), 1)).collect());
|
||||
c.step_continue();
|
||||
if txn {
|
||||
c.commit();
|
||||
}
|
||||
clones() - before
|
||||
}
|
||||
|
||||
/// A limit, pinned so the docs cannot drift from it: a continued piece clones the edge's
|
||||
/// whole value once per transaction, which is the size of the state.
|
||||
#[test]
|
||||
fn a_split_tick_in_a_transaction_costs_the_size_of_the_edge() {
|
||||
let extra = |n| clones_of_one_continue(n, true) - clones_of_one_continue(n, false);
|
||||
let (small, large) = (extra(10), extra(300));
|
||||
assert!(large >= 300, "[split] the edge's value is cloned: {large}");
|
||||
assert!(large > small, "[split] and it grows with the state");
|
||||
}
|
||||
@@ -154,3 +154,42 @@ fn a_finished_rebuild_and_an_interrupted_check_do_not_trip_on_later_ticks() {
|
||||
assert!(!job.interrupted(&c));
|
||||
assert_eq!(job.step(&mut c, || true), Progress::Done);
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[should_panic(expected = "another circuit")]
|
||||
fn a_rebuild_stepped_on_a_foreign_circuit_is_refused_loudly() {
|
||||
let (mut a, mut b) = (Circuit::new(), Circuit::new());
|
||||
let (input, _s) = a.input::<u8>();
|
||||
let mut job = Rebuild::new(input, [(1u8, 1)].into_iter().collect(), 1);
|
||||
job.step(&mut b, || false);
|
||||
}
|
||||
|
||||
#[test]
|
||||
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);
|
||||
assert!(
|
||||
job.belongs_to(&a),
|
||||
"[owner] the circuit that made the input"
|
||||
);
|
||||
assert!(!job.belongs_to(&b), "[owner] another circuit");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_interrupted_rebuild_stays_interrupted_and_a_new_job_loads_the_snapshot() {
|
||||
let snap: Z<u8> = [(1, 1), (2, 1), (3, 1)].into_iter().collect();
|
||||
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);
|
||||
assert_eq!(job.step(&mut c, || true), Progress::Pending);
|
||||
c.step();
|
||||
for _ in 0..3 {
|
||||
assert!(job.interrupted(&c), "[forever] steps never un-interrupt it");
|
||||
c.step();
|
||||
}
|
||||
let mut fresh = Rebuild::new(input, snap, 1);
|
||||
assert_eq!(fresh.step(&mut c, || false), Progress::Done);
|
||||
assert_eq!(i.current().len(), 3);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
# Seeds for failure cases proptest has generated in the past. It is
|
||||
# automatically read and these particular cases re-run before any
|
||||
# novel cases are generated.
|
||||
#
|
||||
# It is recommended to check this file in to source control so that
|
||||
# everyone who runs the test benefits from these saved cases.
|
||||
cc b611223baeb182175151c758f4f6877b5897e5b067f9e5010d14e20b1c6c4418 # shrinks to pre = [], post = [(ZSet { weights: {} }, ZSet { weights: {} })], carry = ZSet { weights: {} }, bad = (ZSet { weights: {} }, ZSet { weights: {} }), fail = false
|
||||
cc 71d785a9a4f55f12c26afef2c65babad2f29c1f2a30becf55936928d570254b4 # shrinks to pre = [], post = [(ZSet { weights: {} }, ZSet { weights: {} })], carry = ZSet { weights: {} }, bad = (ZSet { weights: {} }, ZSet { weights: {} }), fail = false
|
||||
@@ -0,0 +1,309 @@
|
||||
//! `Circuit::begin`/`rollback`/`commit` (ADR 009 clause 11): a failed step is undone, and
|
||||
//! the circuit then behaves as if the step never ran. Evidence: `context/test-evidence/zset.md`.
|
||||
|
||||
use proptest::prelude::*;
|
||||
use std::panic::{AssertUnwindSafe, catch_unwind};
|
||||
use zset::{Circuit, Input, ZSet};
|
||||
|
||||
type Z<K> = ZSet<K>;
|
||||
type P = (u8, u8);
|
||||
|
||||
/// A circuit using every stateful operator, and two closures: every stream's
|
||||
/// `current()`, and every sink drained.
|
||||
struct Rig {
|
||||
c: Circuit,
|
||||
a: Input<P>,
|
||||
b: Input<P>,
|
||||
view: Box<dyn Fn() -> String>,
|
||||
drain: Box<dyn Fn() -> String>,
|
||||
}
|
||||
|
||||
fn rig() -> Rig {
|
||||
let mut c = Circuit::new();
|
||||
let (a, sa) = c.input::<P>();
|
||||
let (b, sb) = c.input::<P>();
|
||||
let ia = c.integrate(&sa);
|
||||
let da = c.delay(&sa);
|
||||
let df = c.differentiate(&sa);
|
||||
let ds = c.distinct(&sa);
|
||||
let j = c.join_indexed(&sa, &sb, |k, x, y| (*k, *x, *y));
|
||||
let ij = c.integrate(&j);
|
||||
let ag = c.aggregate(&sa, |_, vals| {
|
||||
Some(vals.iter().fold(0i64, |s, (v, w)| {
|
||||
s.wrapping_add(i64::from(*v).wrapping_mul(*w))
|
||||
}))
|
||||
});
|
||||
let pl = c.plus(&ia, &sb);
|
||||
let sinks = (
|
||||
c.output(&ia),
|
||||
c.output(&da),
|
||||
c.output(&df),
|
||||
c.output(&ds),
|
||||
c.output(&j),
|
||||
c.output(&ij),
|
||||
c.output(&ag),
|
||||
c.output(&pl),
|
||||
);
|
||||
let (s1, s2) = (sa.clone(), sb.clone());
|
||||
let view = move || {
|
||||
format!(
|
||||
"{:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?}",
|
||||
s1.current(),
|
||||
s2.current(),
|
||||
ia.current(),
|
||||
da.current(),
|
||||
df.current(),
|
||||
ds.current(),
|
||||
j.current(),
|
||||
ij.current(),
|
||||
ag.current(),
|
||||
pl.current(),
|
||||
s1.current().len(),
|
||||
)
|
||||
};
|
||||
let drain = move || {
|
||||
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(),
|
||||
)
|
||||
};
|
||||
Rig {
|
||||
c,
|
||||
a,
|
||||
b,
|
||||
view: Box::new(view),
|
||||
drain: Box::new(drain),
|
||||
}
|
||||
}
|
||||
|
||||
impl Rig {
|
||||
fn tick(&mut self, a: &Z<P>, b: &Z<P>) {
|
||||
self.a.push(a.clone());
|
||||
self.b.push(b.clone());
|
||||
self.c.step();
|
||||
}
|
||||
}
|
||||
|
||||
fn delta() -> impl Strategy<Value = Z<P>> {
|
||||
prop::collection::vec(((0u8..4, 0u8..3), -3i64..=3), 0..6).prop_map(|p| p.into_iter().collect())
|
||||
}
|
||||
|
||||
fn ticks(n: std::ops::Range<usize>) -> impl Strategy<Value = Vec<(Z<P>, Z<P>)>> {
|
||||
prop::collection::vec((delta(), delta()), n)
|
||||
}
|
||||
|
||||
/// The key that sits near `i64::MAX` once preloaded, so a later `+5` overflows it.
|
||||
const BIG: P = (9, 9);
|
||||
|
||||
fn poison() -> Z<P> {
|
||||
// `(0, 0)` sorts first, so operators have applied it before `BIG` overflows.
|
||||
[((0, 0), 1), (BIG, 5)].into_iter().collect()
|
||||
}
|
||||
|
||||
proptest! {
|
||||
/// A step that fails half way, then a rollback, leaves the circuit as it was; the ticks
|
||||
/// after run as if the failed step had never been attempted.
|
||||
#[test]
|
||||
fn a_rolled_back_step_leaves_the_circuit_as_it_was_and_the_ticks_after_unchanged(
|
||||
pre in ticks(0..4),
|
||||
post in ticks(1..4),
|
||||
carry in delta(),
|
||||
bad in (delta(), delta()),
|
||||
fail in any::<bool>(),
|
||||
) {
|
||||
let (mut x, mut y) = (rig(), rig());
|
||||
let preload: Z<P> = [(BIG, i64::MAX - 1)].into_iter().collect();
|
||||
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");
|
||||
}
|
||||
x.a.push(carry.clone());
|
||||
y.a.push(carry);
|
||||
let (view_before, drain_before) = ((x.view)(), (x.drain)());
|
||||
|
||||
x.c.begin();
|
||||
let mut a = bad.0.clone();
|
||||
if fail {
|
||||
a += poison();
|
||||
}
|
||||
x.a.push(a);
|
||||
x.b.push(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");
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
/// `commit` keeps what the steps did; a transaction that commits is not observable.
|
||||
#[test]
|
||||
fn a_committed_transaction_equals_the_same_steps_outside_one(
|
||||
steps in ticks(1..6),
|
||||
split in 0usize..6,
|
||||
) {
|
||||
let (mut x, mut y) = (rig(), rig());
|
||||
for (i, (a, b)) in steps.iter().enumerate() {
|
||||
let open = i >= split;
|
||||
if open {
|
||||
x.c.begin();
|
||||
}
|
||||
x.tick(a, b);
|
||||
if open {
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_operator_built_inside_a_transaction_goes_back_to_empty() {
|
||||
let mut c = Circuit::new();
|
||||
let (input, s) = c.input::<u8>();
|
||||
c.begin();
|
||||
let i = c.integrate(&s);
|
||||
input.push([(1u8, 1)].into_iter().collect());
|
||||
c.step();
|
||||
c.rollback();
|
||||
input.push([(2u8, 1)].into_iter().collect());
|
||||
c.step();
|
||||
let want: Z<u8> = [(2u8, 1)].into_iter().collect();
|
||||
assert_eq!(
|
||||
i.current(),
|
||||
want,
|
||||
"[built] the integral forgot the rolled-back tick"
|
||||
);
|
||||
}
|
||||
|
||||
/// One key at `MAX` that a transaction lowers by one and raises again. Undone oldest first,
|
||||
/// the first undo would take it to `MAX + 1`; newest first, nothing leaves `i64`.
|
||||
#[test]
|
||||
fn rollback_undoes_newest_first_so_no_undo_leaves_i64() {
|
||||
let mut c = Circuit::new();
|
||||
let (input, s) = c.input::<P>();
|
||||
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.step();
|
||||
let before = (i.current(), d.current(), g.current());
|
||||
c.begin();
|
||||
for w in [-1, 1] {
|
||||
input.push(one(w));
|
||||
c.step();
|
||||
}
|
||||
c.rollback();
|
||||
assert_eq!(
|
||||
(i.current(), d.current(), g.current()),
|
||||
before,
|
||||
"[order] streams"
|
||||
);
|
||||
input.push(one(-1));
|
||||
c.step();
|
||||
assert_eq!(
|
||||
i.current(),
|
||||
one(i64::MAX - 1),
|
||||
"[order] the integral is back at MAX first"
|
||||
);
|
||||
}
|
||||
|
||||
/// `i64::MIN` has no negation: a trace undoes it in two applications.
|
||||
#[test]
|
||||
fn a_minimum_weight_applied_to_every_kind_of_state_is_taken_back() {
|
||||
let mut x = rig();
|
||||
let mut y = rig();
|
||||
let min: Z<P> = [((1, 1), i64::MIN)].into_iter().collect();
|
||||
x.c.begin();
|
||||
x.tick(&min, &Z::new());
|
||||
x.c.rollback();
|
||||
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.drain)(),
|
||||
(y.drain)(),
|
||||
"[min] output: the join saw no left side"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rollback_undoes_the_clock() {
|
||||
let mut c = Circuit::new();
|
||||
let (input, _s) = c.input::<u8>();
|
||||
c.step();
|
||||
c.begin();
|
||||
input.push([(1u8, 1)].into_iter().collect());
|
||||
c.step();
|
||||
assert_eq!(c.ticks(), 2);
|
||||
c.rollback();
|
||||
assert_eq!(c.ticks(), 1, "[ticks] the rolled-back step is not counted");
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[should_panic(expected = "already open")]
|
||||
fn a_second_begin_is_refused_loudly() {
|
||||
let mut c = Circuit::new();
|
||||
c.begin();
|
||||
c.begin();
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[should_panic(expected = "no transaction is open")]
|
||||
fn a_rollback_with_nothing_open_is_refused_loudly() {
|
||||
Circuit::new().rollback();
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[should_panic(expected = "no transaction is open")]
|
||||
fn a_commit_with_nothing_open_is_refused_loudly() {
|
||||
Circuit::new().commit();
|
||||
}
|
||||
|
||||
/// A tick in two pieces, rolled back: the edge's whole value comes back, not the last piece.
|
||||
#[test]
|
||||
fn a_rollback_across_a_split_tick_restores_the_whole_value() {
|
||||
let (mut x, mut y) = (rig(), rig());
|
||||
let first: Z<P> = [((1, 1), 2), ((2, 1), 1)].into_iter().collect();
|
||||
x.tick(&first, &Z::new());
|
||||
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.step_continue();
|
||||
}
|
||||
x.c.begin();
|
||||
for k in [4u8, 5] {
|
||||
x.a.push([((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)(),
|
||||
"[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");
|
||||
}
|
||||
@@ -0,0 +1,462 @@
|
||||
//! A model-based property over every public operator: a rollback leaves the circuit equal to one
|
||||
//! that never ran the step. From the review of #355. Evidence: `context/test-evidence/zset.md`.
|
||||
|
||||
use proptest::prelude::*;
|
||||
use proptest::test_runner::{Config, TestRunner};
|
||||
use std::cell::Cell;
|
||||
use std::collections::BTreeMap;
|
||||
use std::ops::Bound;
|
||||
use std::panic::{AssertUnwindSafe, catch_unwind};
|
||||
use std::sync::atomic::{AtomicUsize, Ordering::Relaxed};
|
||||
use zset::{Circuit, Input, Output, Stream, Trace, ZSet};
|
||||
|
||||
type P = (u8, u8);
|
||||
type Z = ZSet<P>;
|
||||
|
||||
thread_local! { static TRIP: Cell<Option<u32>> = const { Cell::new(None) }; }
|
||||
|
||||
fn maybe_trip() {
|
||||
TRIP.with(|t| {
|
||||
if let Some(n) = t.get() {
|
||||
if n == 0 {
|
||||
t.set(None);
|
||||
panic!("tripped");
|
||||
}
|
||||
t.set(Some(n - 1));
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct VecTrace<K, V>(BTreeMap<(K, V), i64>);
|
||||
|
||||
impl<K: Ord + Clone, V: Ord + Clone> Trace<K, V> for VecTrace<K, V> {
|
||||
fn apply(&mut self, k: &K, v: &V, w: i64) {
|
||||
maybe_trip();
|
||||
let key = (k.clone(), v.clone());
|
||||
let cur = self.0.get(&key).copied().unwrap_or(0);
|
||||
let n = cur.checked_add(w).expect("overflow in VecTrace");
|
||||
if n == 0 {
|
||||
self.0.remove(&key);
|
||||
} else {
|
||||
self.0.insert(key, n);
|
||||
}
|
||||
}
|
||||
fn lookup(&self, k: &K) -> Vec<(V, i64)> {
|
||||
self.0
|
||||
.iter()
|
||||
.filter(|((kk, _), _)| kk == k)
|
||||
.map(|((_, v), w)| (v.clone(), *w))
|
||||
.collect()
|
||||
}
|
||||
fn range(&self, lo: Bound<&K>, hi: Bound<&K>) -> Vec<(K, V, i64)> {
|
||||
self.0
|
||||
.iter()
|
||||
.filter(|((k, _), _)| std::ops::RangeBounds::contains(&(lo, hi), k))
|
||||
.map(|((k, v), w)| (k.clone(), v.clone(), *w))
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
enum Op {
|
||||
Integrate(usize),
|
||||
Delay(usize),
|
||||
Diff(usize),
|
||||
Distinct(usize),
|
||||
Plus(usize, usize),
|
||||
Map(usize),
|
||||
Filter(usize),
|
||||
FlatMap(usize),
|
||||
IndexBy(usize),
|
||||
Join(usize, usize),
|
||||
JoinCustom(usize, usize),
|
||||
Agg(usize),
|
||||
AggCustom(usize),
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
enum Act {
|
||||
Push(bool, Z),
|
||||
Step(Option<u32>),
|
||||
Cont(Option<u32>),
|
||||
Begin,
|
||||
Commit,
|
||||
Rollback,
|
||||
Build(Op),
|
||||
}
|
||||
|
||||
struct World {
|
||||
c: Circuit,
|
||||
inputs: [Input<P>; 2],
|
||||
streams: Vec<Stream<Z>>,
|
||||
sinks: Vec<Output<P>>,
|
||||
}
|
||||
|
||||
fn agg_f(_: &u8, vals: &[(u8, i64)]) -> Option<u8> {
|
||||
maybe_trip();
|
||||
Some(
|
||||
vals.iter()
|
||||
.fold(0u8, |s, (v, w)| s.wrapping_add(v.wrapping_mul(*w as u8))),
|
||||
)
|
||||
}
|
||||
|
||||
impl World {
|
||||
fn new() -> Self {
|
||||
let mut c = Circuit::new();
|
||||
let (a, sa) = c.input::<P>();
|
||||
let (b, sb) = c.input::<P>();
|
||||
let sinks = vec![c.output(&sa), c.output(&sb)];
|
||||
World {
|
||||
c,
|
||||
inputs: [a, b],
|
||||
streams: vec![sa, sb],
|
||||
sinks,
|
||||
}
|
||||
}
|
||||
fn build(&mut self, op: &Op) {
|
||||
let c = &mut self.c;
|
||||
let st = &self.streams;
|
||||
struct S<'a>(&'a [Stream<Z>]);
|
||||
impl S<'_> {
|
||||
fn s(&self, i: usize) -> Stream<Z> {
|
||||
self.0[i % self.0.len()].clone()
|
||||
}
|
||||
}
|
||||
let this = S(st);
|
||||
let out: Stream<Z> = match *op {
|
||||
Op::Integrate(i) => {
|
||||
let s = this.s(i);
|
||||
c.integrate(&s)
|
||||
}
|
||||
Op::Delay(i) => {
|
||||
let s = this.s(i);
|
||||
c.delay(&s)
|
||||
}
|
||||
Op::Diff(i) => {
|
||||
let s = this.s(i);
|
||||
c.differentiate(&s)
|
||||
}
|
||||
Op::Distinct(i) => {
|
||||
let s = this.s(i);
|
||||
c.distinct(&s)
|
||||
}
|
||||
Op::Plus(i, j) => {
|
||||
let (x, y) = (this.s(i), this.s(j));
|
||||
c.plus(&x, &y)
|
||||
}
|
||||
Op::Map(i) => {
|
||||
let s = this.s(i);
|
||||
c.map(&s, |p: &P| {
|
||||
maybe_trip();
|
||||
(p.1, p.0 % 4)
|
||||
})
|
||||
}
|
||||
Op::Filter(i) => {
|
||||
let s = this.s(i);
|
||||
c.filter(&s, |p: &P| {
|
||||
maybe_trip();
|
||||
p.0.is_multiple_of(2)
|
||||
})
|
||||
}
|
||||
Op::FlatMap(i) => {
|
||||
let s = this.s(i);
|
||||
c.flat_map(&s, |p: &P| {
|
||||
maybe_trip();
|
||||
vec![*p, (p.0, p.1.wrapping_add(1) % 4)]
|
||||
})
|
||||
}
|
||||
Op::IndexBy(i) => {
|
||||
let s = this.s(i);
|
||||
let ix = c.index_by(&s, |p: &P| p.1);
|
||||
c.map(&ix, |(g, p): &(u8, P)| (*g, p.0))
|
||||
}
|
||||
Op::Join(i, j) => {
|
||||
let (x, y) = (this.s(i), this.s(j));
|
||||
c.join_indexed(&x, &y, |k, a, b| (*k, (a ^ b) % 4))
|
||||
}
|
||||
Op::JoinCustom(i, j) => {
|
||||
let (x, y) = (this.s(i), this.s(j));
|
||||
c.join_indexed_in(
|
||||
&x,
|
||||
&y,
|
||||
VecTrace::default(),
|
||||
VecTrace::default(),
|
||||
|k, a, b| (*k, a.wrapping_add(*b) % 4),
|
||||
)
|
||||
}
|
||||
Op::Agg(i) => {
|
||||
let s = this.s(i);
|
||||
c.aggregate(&s, agg_f)
|
||||
}
|
||||
Op::AggCustom(i) => {
|
||||
let s = this.s(i);
|
||||
c.aggregate_in(&s, VecTrace::default(), agg_f)
|
||||
}
|
||||
};
|
||||
let o = self.c.output(&out);
|
||||
self.streams.push(out);
|
||||
self.sinks.push(o);
|
||||
}
|
||||
fn view(&self) -> Vec<Z> {
|
||||
self.streams.iter().map(|s| s.current()).collect()
|
||||
}
|
||||
fn drain(&self) -> Vec<Z> {
|
||||
self.sinks.iter().map(|s| s.take()).collect()
|
||||
}
|
||||
}
|
||||
|
||||
static CASES: AtomicUsize = AtomicUsize::new(0);
|
||||
static RB_AFTER_PANIC: AtomicUsize = AtomicUsize::new(0);
|
||||
static RB_CLEAN: AtomicUsize = AtomicUsize::new(0);
|
||||
static RB_AFTER_CONT: AtomicUsize = AtomicUsize::new(0);
|
||||
static COMMITS: AtomicUsize = AtomicUsize::new(0);
|
||||
static BUILT_IN_TXN: AtomicUsize = AtomicUsize::new(0);
|
||||
static DOUBLE_BEGIN: AtomicUsize = AtomicUsize::new(0);
|
||||
static OUTSIDE_PANIC_STOP: AtomicUsize = AtomicUsize::new(0);
|
||||
static MULTI_STEP_TXN: AtomicUsize = AtomicUsize::new(0);
|
||||
static TRIPS: AtomicUsize = AtomicUsize::new(0);
|
||||
|
||||
fn weight() -> impl Strategy<Value = i64> {
|
||||
prop_oneof![
|
||||
8 => -3i64..=3,
|
||||
1 => Just(i64::MAX),
|
||||
1 => Just(i64::MIN),
|
||||
1 => Just(i64::MAX / 2 + 1),
|
||||
1 => Just(-(i64::MAX / 2) - 1),
|
||||
]
|
||||
}
|
||||
|
||||
fn delta() -> impl Strategy<Value = Z> {
|
||||
prop::collection::vec(((0u8..4, 0u8..4), weight()), 0..5).prop_map(|v| {
|
||||
let mut z = Z::new();
|
||||
for (k, w) in v {
|
||||
// build without overflowing: replace if it would
|
||||
if z.weight(&k).checked_add(w).is_some() {
|
||||
z.insert(k, w);
|
||||
}
|
||||
}
|
||||
z
|
||||
})
|
||||
}
|
||||
|
||||
fn op() -> impl Strategy<Value = Op> {
|
||||
let i = 0usize..64;
|
||||
prop_oneof![
|
||||
i.clone().prop_map(Op::Integrate),
|
||||
i.clone().prop_map(Op::Delay),
|
||||
i.clone().prop_map(Op::Diff),
|
||||
i.clone().prop_map(Op::Distinct),
|
||||
(i.clone(), i.clone()).prop_map(|(a, b)| Op::Plus(a, b)),
|
||||
i.clone().prop_map(Op::Map),
|
||||
i.clone().prop_map(Op::Filter),
|
||||
i.clone().prop_map(Op::FlatMap),
|
||||
i.clone().prop_map(Op::IndexBy),
|
||||
(i.clone(), i.clone()).prop_map(|(a, b)| Op::Join(a, b)),
|
||||
(i.clone(), i.clone()).prop_map(|(a, b)| Op::JoinCustom(a, b)),
|
||||
i.clone().prop_map(Op::Agg),
|
||||
i.prop_map(Op::AggCustom),
|
||||
]
|
||||
}
|
||||
|
||||
fn act() -> impl Strategy<Value = Act> {
|
||||
let trip = prop_oneof![2 => Just(None), 1 => (0u32..40).prop_map(Some)];
|
||||
prop_oneof![
|
||||
4 => (any::<bool>(), delta()).prop_map(|(b, z)| Act::Push(b, z)),
|
||||
4 => trip.clone().prop_map(Act::Step),
|
||||
2 => trip.prop_map(Act::Cont),
|
||||
2 => Just(Act::Begin),
|
||||
1 => Just(Act::Commit),
|
||||
2 => Just(Act::Rollback),
|
||||
1 => op().prop_map(Act::Build),
|
||||
]
|
||||
}
|
||||
|
||||
fn apply_plain(w: &mut World, a: &Act) {
|
||||
match a {
|
||||
Act::Push(b, z) => w.inputs[*b as usize].push(z.clone()),
|
||||
Act::Step(_) => w.c.step(),
|
||||
Act::Cont(_) => w.c.step_continue(),
|
||||
Act::Build(op) => w.build(op),
|
||||
_ => unreachable!(),
|
||||
}
|
||||
}
|
||||
|
||||
fn check(x: &World, y: &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);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn run(ops: &[Op], acts: &[Act]) -> Result<(), TestCaseError> {
|
||||
CASES.fetch_add(1, Relaxed);
|
||||
let (mut x, mut y) = (World::new(), World::new());
|
||||
for o in ops {
|
||||
x.build(o);
|
||||
y.build(o);
|
||||
}
|
||||
let mut open = false;
|
||||
let mut failed = false;
|
||||
let mut buf: Vec<Act> = vec![];
|
||||
let mut steps_in_txn = 0;
|
||||
let mut cont_in_txn = false;
|
||||
for (n, a) in acts.iter().enumerate() {
|
||||
let at = format!("after act {n} {a:?}");
|
||||
match a {
|
||||
Act::Begin => {
|
||||
if open {
|
||||
let r = catch_unwind(AssertUnwindSafe(|| x.c.begin()));
|
||||
prop_assert!(r.is_err(), "double begin must panic");
|
||||
DOUBLE_BEGIN.fetch_add(1, Relaxed);
|
||||
} else {
|
||||
x.c.begin();
|
||||
open = true;
|
||||
failed = false;
|
||||
steps_in_txn = 0;
|
||||
cont_in_txn = false;
|
||||
}
|
||||
}
|
||||
Act::Commit if open && !failed => {
|
||||
x.c.commit();
|
||||
open = false;
|
||||
for b in buf.drain(..) {
|
||||
apply_plain(&mut y, &b);
|
||||
}
|
||||
COMMITS.fetch_add(1, Relaxed);
|
||||
if steps_in_txn > 1 {
|
||||
MULTI_STEP_TXN.fetch_add(1, Relaxed);
|
||||
}
|
||||
check(&x, &y, &at)?;
|
||||
}
|
||||
Act::Commit | Act::Rollback if open => {
|
||||
x.c.rollback();
|
||||
open = false;
|
||||
if failed {
|
||||
RB_AFTER_PANIC.fetch_add(1, Relaxed);
|
||||
} else {
|
||||
RB_CLEAN.fetch_add(1, Relaxed);
|
||||
}
|
||||
if cont_in_txn {
|
||||
RB_AFTER_CONT.fetch_add(1, Relaxed);
|
||||
}
|
||||
if steps_in_txn > 1 {
|
||||
MULTI_STEP_TXN.fetch_add(1, Relaxed);
|
||||
}
|
||||
for b in buf.drain(..) {
|
||||
if let Act::Build(op) = b {
|
||||
y.build(&op);
|
||||
}
|
||||
}
|
||||
check(&x, &y, &at)?;
|
||||
}
|
||||
Act::Commit | Act::Rollback => {}
|
||||
_ if open && failed => {} // only rollback after a failed step
|
||||
Act::Step(t) | Act::Cont(t) if open => {
|
||||
TRIP.with(|c| c.set(*t));
|
||||
let r = catch_unwind(AssertUnwindSafe(|| apply_plain(&mut x, a)));
|
||||
let tripped = TRIP.with(|c| c.get()).is_none() && t.is_some();
|
||||
TRIP.with(|c| c.set(None));
|
||||
steps_in_txn += 1;
|
||||
if matches!(a, Act::Cont(_)) {
|
||||
cont_in_txn = true;
|
||||
}
|
||||
if r.is_err() {
|
||||
failed = true;
|
||||
if tripped {
|
||||
TRIPS.fetch_add(1, Relaxed);
|
||||
}
|
||||
} else {
|
||||
buf.push(a.clone());
|
||||
}
|
||||
}
|
||||
_ if open => {
|
||||
if matches!(a, Act::Build(_)) {
|
||||
BUILT_IN_TXN.fetch_add(1, Relaxed);
|
||||
}
|
||||
let r = catch_unwind(AssertUnwindSafe(|| apply_plain(&mut x, a)));
|
||||
if r.is_err() {
|
||||
failed = true;
|
||||
} else {
|
||||
buf.push(a.clone());
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
let r = catch_unwind(AssertUnwindSafe(|| apply_plain(&mut x, a)));
|
||||
if r.is_err() {
|
||||
OUTSIDE_PANIC_STOP.fetch_add(1, Relaxed);
|
||||
return Ok(());
|
||||
}
|
||||
apply_plain(&mut y, a);
|
||||
check(&x, &y, &at)?;
|
||||
}
|
||||
}
|
||||
}
|
||||
if open {
|
||||
x.c.rollback();
|
||||
for b in buf.drain(..) {
|
||||
if let Act::Build(op) = b {
|
||||
y.build(&op);
|
||||
}
|
||||
}
|
||||
if failed {
|
||||
RB_AFTER_PANIC.fetch_add(1, Relaxed);
|
||||
} else {
|
||||
RB_CLEAN.fetch_add(1, Relaxed);
|
||||
}
|
||||
check(&x, &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.step()
|
||||
}));
|
||||
if r.is_err() {
|
||||
return Ok(());
|
||||
}
|
||||
y.inputs[0].push(z.clone());
|
||||
y.inputs[1].push(z);
|
||||
y.c.step();
|
||||
check(&x, &y, &format!("post tick {i}"))?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reviewer_rollback_matches_a_circuit_that_never_ran_the_step() {
|
||||
let cases: u32 = std::env::var("ATTACK_CASES")
|
||||
.ok()
|
||||
.and_then(|s| s.parse().ok())
|
||||
.unwrap_or(256);
|
||||
std::panic::set_hook(Box::new(|_| {}));
|
||||
let mut runner = TestRunner::new(Config {
|
||||
cases,
|
||||
..Config::default()
|
||||
});
|
||||
let strat = (
|
||||
prop::collection::vec(op(), 1..12),
|
||||
prop::collection::vec(act(), 1..40),
|
||||
);
|
||||
let r = runner.run(&strat, |(ops, acts)| run(&ops, &acts));
|
||||
eprintln!(
|
||||
"cases={} rb_after_panic={} rb_clean={} rb_after_cont={} commits={} built_in_txn={} double_begin={} outside_panic_stop={} multi_step_txn={} trips={}",
|
||||
CASES.load(Relaxed),
|
||||
RB_AFTER_PANIC.load(Relaxed),
|
||||
RB_CLEAN.load(Relaxed),
|
||||
RB_AFTER_CONT.load(Relaxed),
|
||||
COMMITS.load(Relaxed),
|
||||
BUILT_IN_TXN.load(Relaxed),
|
||||
DOUBLE_BEGIN.load(Relaxed),
|
||||
OUTSIDE_PANIC_STOP.load(Relaxed),
|
||||
MULTI_STEP_TXN.load(Relaxed),
|
||||
TRIPS.load(Relaxed)
|
||||
);
|
||||
let _ = std::panic::take_hook();
|
||||
if let Err(e) = r {
|
||||
eprintln!("FAILURE: {e:?}");
|
||||
panic!("attack failed");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,167 @@
|
||||
//! A `Rebuild` stepped across a rollback, and closures that read streams during an eval.
|
||||
//! Evidence: `context/test-evidence/zset.md`.
|
||||
|
||||
use std::panic::{AssertUnwindSafe, catch_unwind};
|
||||
use zset::{Circuit, Progress, Rebuild, ZSet};
|
||||
|
||||
type Z<K> = ZSet<K>;
|
||||
|
||||
fn quiet() {
|
||||
static O: std::sync::Once = std::sync::Once::new();
|
||||
O.call_once(|| {
|
||||
let prev = std::panic::take_hook();
|
||||
std::panic::set_hook(Box::new(move |info| {
|
||||
let s = format!("{info}");
|
||||
if !(s.contains("tripped") || s.contains("overflow")) {
|
||||
prev(info);
|
||||
}
|
||||
}));
|
||||
});
|
||||
}
|
||||
|
||||
fn z(keys: &[(u8, i64)]) -> Z<u8> {
|
||||
quiet();
|
||||
keys.iter().copied().collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn r1_rebuild_begun_before_txn_continued_inside_then_rolled_back_reports_interrupted() {
|
||||
let mut c = Circuit::new();
|
||||
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);
|
||||
assert_eq!(job.step(&mut c, || true), Progress::Pending);
|
||||
c.begin();
|
||||
assert_eq!(job.step(&mut c, || true), Progress::Pending);
|
||||
c.rollback();
|
||||
let interrupted = job.interrupted(&c);
|
||||
// Docs (zset.md Transactions 6, header): the job reports itself dead after a rollback.
|
||||
let mut finished = None;
|
||||
if !interrupted {
|
||||
while job.step(&mut c, || false) == Progress::Pending {}
|
||||
finished = Some(i.current());
|
||||
}
|
||||
assert!(
|
||||
interrupted,
|
||||
"[r1] job not interrupted after rollback; finishing it gives {:?}, want {:?}",
|
||||
finished, snap
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn r2_rebuild_finished_inside_a_rolled_back_txn_is_reported_done_with_nothing_loaded() {
|
||||
let mut c = Circuit::new();
|
||||
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);
|
||||
c.begin();
|
||||
assert_eq!(job.step(&mut c, || false), Progress::Done);
|
||||
c.rollback();
|
||||
let interrupted = job.interrupted(&c);
|
||||
assert!(
|
||||
interrupted,
|
||||
"[r2] after rollback: not interrupted, integral {:?}",
|
||||
i.current()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn r3_an_interrupted_rebuild_resurrects_after_rollback_and_a_later_step() {
|
||||
let mut c = Circuit::new();
|
||||
let (input, s) = c.input::<u8>();
|
||||
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);
|
||||
c.begin();
|
||||
assert_eq!(job.step(&mut c, || true), Progress::Pending); // began_at = 2
|
||||
c.rollback(); // ticks back to 1
|
||||
assert!(job.interrupted(&c), "[r3] dead right after rollback");
|
||||
c.step(); // ticks 2 again
|
||||
let dead_after_step = job.interrupted(&c);
|
||||
let mut got = None;
|
||||
if !dead_after_step {
|
||||
while job.step(&mut c, || false) == Progress::Pending {}
|
||||
got = Some(i.current());
|
||||
}
|
||||
assert!(
|
||||
dead_after_step,
|
||||
"[r3] interrupted job came back to life after a later step; finishing it gives {:?}, want {:?}",
|
||||
got, snap
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
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.step();
|
||||
let snap = z(&[(1, 1), (2, 1), (3, 1)]);
|
||||
let mut job = Rebuild::new(input.clone(), snap.clone(), 1);
|
||||
c.begin();
|
||||
let r = catch_unwind(AssertUnwindSafe(|| job.step(&mut c, || false)));
|
||||
assert!(r.is_err());
|
||||
c.rollback();
|
||||
let interrupted = job.interrupted(&c);
|
||||
let mut got = None;
|
||||
if !interrupted {
|
||||
input.push(z(&[(1, -5)])); // make room so the retry does not overflow
|
||||
while job.step(&mut c, || false) == Progress::Pending {}
|
||||
got = Some(i.current());
|
||||
}
|
||||
assert!(
|
||||
interrupted,
|
||||
"[r4] after a panicked first chunk + rollback, the job is not interrupted; finishing gives {:?}",
|
||||
got
|
||||
);
|
||||
}
|
||||
|
||||
#[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.begin();
|
||||
let r = catch_unwind(AssertUnwindSafe(|| job.step(&mut c, || false)));
|
||||
assert!(r.is_err(), "the push overflows");
|
||||
c.rollback();
|
||||
let interrupted = job.interrupted(&c);
|
||||
let mut got = None;
|
||||
if !interrupted {
|
||||
while job.step(&mut c, || false) == Progress::Pending {}
|
||||
got = Some(i.current());
|
||||
}
|
||||
assert!(
|
||||
interrupted,
|
||||
"[p1] the chunk left the job unstepped; finishing it gives {got:?}"
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user