hyphae 3.1.2

Reactive cells and runtime primitives for rship
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

//! `CellMap` analogue of the join.rs / `join_vec.rs` torn-value repro, under the
//! scheduler's wave-parallel drain.
//!
//! A two-input `CellMap` join (`inner_join`, `left_join`, `multi_left_join`) has
//! a left sink and a right sink, each driven by a distinct height-0 root
//! (`left.diffs_cell` / `right.diffs_cell`). Each sink reads the *sibling*
//! side's rows to build the combined output row. Under wave-parallel draining
//! (a wave of >= `PARALLEL_WAVE_THRESHOLD` same-height ops runs on rayon), the
//! two sinks can run at the literal same instant on different threads; if the
//! emit is not ordered consistently with the sibling read, two concurrent
//! updates to one output key can last-write-wins a stale combined value.
//!
//! To force a genuinely parallel wave we drive PAIRS independent joins in one
//! `batch()`: 8 pairs = 16 height-0 diff-cell ops, comfortably over the
//! threshold, so each join's own left+right ops run concurrently. Every
//! iteration sets a fresh, exactly-known value on both sides of every pair, so
//! the correct settled output is computable and checked strictly — a torn
//! survivor (new value on one side, stale on the other) fails the assert.
#![cfg(feature = "scheduler")]

use hyphae::{CellMap, MapQuery, batch, traits::InnerJoinExt};

const PAIRS: usize = 8;
const ITERATIONS: i64 = 3_000;

#[test]
fn concurrent_inner_joins_never_settle_on_a_torn_value() {
    // Force the parallel drain path: production defaults the wave threshold high.
    hyphae::scheduler::set_wave_threshold_for_test(4);
    let mut lefts = Vec::new();
    let mut rights = Vec::new();
    let mut outputs = Vec::new();
    for _ in 0..PAIRS {
        let l = CellMap::<String, i64>::new();
        let r = CellMap::<String, i64>::new();
        // Seed the shared join key on both sides so the inner join emits.
        l.insert("k".into(), 0);
        r.insert("k".into(), 0);
        let out = l.clone().inner_join(r.clone()).materialize();
        lefts.push(l);
        rights.push(r);
        outputs.push(out);
    }

    for it in 1..=ITERATIONS {
        // Both sides of all PAIRS joins mutated in ONE batch → a single wave of
        // 2*PAIRS height-0 diff ops, wide enough to dispatch in parallel.
        batch(|| {
            for (index, (left, right)) in lefts.iter().zip(&rights).enumerate() {
                let base = it
                    .saturating_mul(1000)
                    .saturating_add(i64::try_from(index).unwrap_or(i64::MAX));
                left.insert("k".into(), base);
                right.insert("k".into(), base.saturating_add(500));
            }
        });

        for (index, output) in outputs.iter().enumerate() {
            let base = it
                .saturating_mul(1000)
                .saturating_add(i64::try_from(index).unwrap_or(i64::MAX));
            assert_eq!(
                output.get_value(&"k".to_string()),
                Some((base, base.saturating_add(500))),
                "inner_join settled on a torn/stale value at iteration {it}, pair {index}"
            );
        }
    }
}