Distributed Async Evaluation — Support W=512+ Multi-Worker Scaling

[justinjoy]

Issue #401: Distributed Async Evaluation — Support W=512+ Multi-Worker Scaling

Summary

The wirelog TDD (Timely-Differential Dataflow) multi-worker evaluation pipeline exhibits super-linear slowdown beyond W=4: running with W=8 workers is 8% slower than W=4 (28.3s vs 26.1s), despite doubling the worker count. This violates the expected near-linear scaling behavior and blocks any path to high-parallelism deployment.

The root cause is a combination of barrier-based synchronous coordination, O(W) sequential coordinator exchange, and O(W²) exchange buffer allocation that together create an Amdahl's Law ceiling at ~W=4. The current architecture's serial fraction (estimated 5–15%) makes scaling beyond W=8 physically impossible without fundamental changes.

This umbrella issue tracks the design and implementation of a distributed async evaluation architecture that replaces barrier synchronization with epoch-based convergence, serial exchange with tree-structured parallel merge, and sequential coordination with pipelined compute/exchange overlap. The target is near-linear scaling to W=512 and W=1024.

---

Root Cause Analysis

Observed Performance Data

Workers	Time (s)	Speedup vs W=1	Speedup vs W=4	Status
W=1	752.4	1.0×	—	Baseline (sequential path)
W=4	26.1	28.8×	1.0×	Working (post-#404 fixes)
W=8	28.3	26.6×	0.92×	REGRESSING

Note: The W=1 → W=4 gap (28.8×) far exceeds the 4× expected from parallelism alone. This is because the TDD/BDX multi-worker path uses a fundamentally more efficient incremental evaluation strategy (delta-only BDX) than the W=1 sequential path (col_eval_stratum). The W=8 vs W=4 regression is overhead layered on top of this faster algorithm.

Three Bottlenecks

Bottleneck	Current	At W=512	Code Location
Barrier synchronization	wl_workqueue_wait_all() blocks until slowest worker finishes	Straggler latency × 512 iterations	workqueue.c:200-220
O(W²) exchange buffers	exchange_bufs[W][W] allocates W² col_rel_t* slots	262,144 slots (512²)	eval.c:2425-2446
O(W) sequential exchange	tdd_bdx_exchange_deltas runs single-threaded on coordinator	512 serial delta unions per sub-pass	eval.c:3581-3767

Amdahl's Law Analysis

T(W) = T_seq + T_par/W + T_sync(W) + T_overhead(W)

Current serial fraction (estimated): 5-15%
  - At f=10%: max theoretical speedup = 1/f = 10× (regardless of W)
  - At W=512 with f=10%: T(512) ≈ T_seq (parallel portion negligible)

Target serial fraction: <2%
  - At f=2%: max theoretical speedup = 50×
  - Enables meaningful scaling to W=512+

Additional Contributing Factors

• O(N log N) coordinator IDB re-sort per iteration (eval.c:3661-3666, flagged as TODO(#390)): grows unboundedly as VarPointsTo accumulates millions of tuples
• Workqueue ring capacity: WL_WQ_RING_CAP = 256 (workqueue.c:30) — physically cannot enqueue W=512 work items
• Single mutex + thundering herd: cond_broadcast wakes ALL W workers simultaneously, serializing on one mutex (workqueue.c:78-108)
• EDB replication: All W workers read identical large EDB tables (e.g., Var_Type: 947k rows), multiplying memory bandwidth pressure with no parallelism benefit
• Memory budget over-partitioning: budget = total / (W+1) (session.c:546) — at W=8, each party gets only 8.3% of RAM, triggering premature backpressure

---

Solution Architecture

Current vs Proposed

CURRENT (Barrier-Based Synchronous):

  for each iteration:
    for each sub-pass:
      ┌─────────────────────────────────────────┐
      │ 1. RESET worker contexts     O(W×nrels) │
      │ 2. DISPATCH: submit W tasks  O(W) mutex │
      │ 3. BARRIER: wait_all()       BLOCKING   │
      │ 4. CONVERGENCE CHECK         O(W) scan  │
      │ 5. EXCHANGE: serial merge    O(W×D)     │
      └─────────────────────────────────────────┘

PROPOSED (Distributed Async):

  Epoch-based execution:
    ┌──────────────────────────────────────────────────────┐
    │ Workers compute continuously (no barrier)            │
    │   → push delta to lock-free SPSC queue (per-worker)  │
    │   → atomically increment epoch counter               │
    │                                                      │
    │ Merge tree reduces deltas: O(log₂W) depth            │
    │   → parallel binary merge at each level               │
    │   → zero-copy shared views between levels             │
    │                                                      │
    │ Coordinator distributes merged delta (pipelined)      │
    │   → workers start next epoch while merge proceeds     │
    │                                                      │
    │ Convergence: atomic epoch counter (no global lock)    │
    └──────────────────────────────────────────────────────┘

Key Architectural Changes

Component	Current	Proposed	Complexity Change
Synchronization	pthread_barrier / wl_workqueue_wait_all	Atomic epoch counter	O(W) → O(1) amortized
Delta exchange	Sequential coordinator merge	Tree-structured parallel merge	O(W) depth → O(log₂W) depth
Worker communication	Single mutex + cond_broadcast	Lock-free SPSC ring buffers (per-worker)	~50-100ns → ~10-20ns per op
Exchange buffers	exchange_bufs[W][W] O(W²)	Per-worker delta slots O(W)	W²→W memory
Coordinator IDB sort	O(N log N) re-sort per iteration	O(N+D) sorted merge-append	Eliminates redundant work
Compute/exchange	Sequential (compute → wait → exchange)	Pipelined (overlap compute and merge)	Latency hiding

Lock-Free Protocol (High-Level)

Per-worker SPSC ring buffer (implemented: wirelog/util/lockfree_queue.h):
  - Enqueue: load head(acquire), write slot, store tail(release)
  - Dequeue: load tail(acquire), read slot, store head(release)
  - No CAS, no mutex — 2 atomic ops per operation
  - ABA-free: monotone integer cursors, no pointer comparison

MPMC wrapper: W independent SPSC queues, coordinator polls round-robin.
  - Zero contention between producers (each owns a queue)
  - Coordinator dequeue_all: O(W × N) scan

Epoch convergence:
  - Workers: atomic_fetch_add(&epoch.workers_done, 1)
  - Last worker (prev+1 == W): check any_new_at_epoch flag
  - If no new tuples: set converged = true (no barrier needed)

---

Implementation Roadmap

Phase 0: Compatibility & Foundation (3 atomic commits)

Goal: Remove scaling walls without changing the synchronization model. All changes backward-compatible with W=1 fast path.

Commit 1: Extend WL_WQ_RING_CAP

• File: wirelog/workqueue.c:30
• Change: WL_WQ_RING_CAP from 256 → 1024 (or dynamic allocation)
• Why: Current cap physically prevents W>256 from enqueuing work items
• Test: Verify W=512 work items can be submitted without overflow

Commit 2: Sorted merge-insert optimization (TODO #390)

• File: wirelog/columnar/eval.c:3661-3666
• Change: Replace O(N log N) tdd_dedup_rel(coord_idb) with O(N+D) merge-append that maintains sorted order incrementally
• Why: Eliminates the single largest per-iteration coordinator cost
• Test: DOOP convergence correctness at W=4, verify iteration count unchanged

Commit 3: Cache-line padding for worker state

• Files: wirelog/columnar/eval.c (worker context struct), wirelog/workqueue.c (work item struct)
• Change: Add __attribute__((aligned(64))) or manual padding to prevent false sharing between worker-local state and coordinator state
• Why: At W=8+, false sharing on adjacent cache lines causes coherency traffic spikes
• Test: perf stat cache-miss comparison W=4 vs W=8

Phase 1: Async Delta Slot + Epoch Convergence

Goal: Remove barrier synchronization. Workers submit deltas asynchronously; coordinator detects convergence via atomic epoch counters.

• Integrate lock-free MPMC queue (wirelog/util/lockfree_queue.h, already implemented)
• Replace wl_workqueue_wait_all() barrier with epoch-counter convergence detection
• Workers push delta to per-worker SPSC queue after each sub-pass
• Coordinator polls queues and merges deltas (still serial at this phase)
• Validation: DOOP W=4 correctness (same tuple count, same iteration count)
• Target: Eliminate barrier idle time; W=8 should match or beat W=4

Phase 2: Tree-Structured Parallel Merge

Goal: Replace O(W) serial delta merge with O(log₂W) parallel tree reduction.

Level 0 (leaves):    W₀  W₁  W₂  W₃  W₄  W₅  W₆  W₇  ... W₅₁₁
                      ╲ ╱    ╲ ╱    ╲ ╱    ╲ ╱              ╲ ╱
Level 1 (W/2):       M₀₁   M₂₃   M₄₅   M₆₇    ...       M₅₁₀₋₅₁₁
                       ╲   ╱       ╲   ╱
Level 2 (W/4):       M₀₋₃      M₄₋₇         ...
                         ...
Level 9 (root):      M₀₋₅₁₁    ← final merged delta

• Each level's merges are independent → parallel execution using workqueue
• Dedup at each level prevents exponential intermediate growth
• W=512: 9 levels, 511 merge operations (vs 512 serial)
• Validation: W=16 scaling (expect near-linear improvement over W=8)

Phase 3: Pipelined Compute/Exchange Overlap

Goal: Workers start next epoch while coordinator merges current epoch's deltas.

• Async double-buffer: workers write to epoch e slot while coordinator merges epoch e-1
• Speculative execution: workers proceed with stale delta (safe under monotonicity — stale delta ⊆ fresh delta)
• Validation: W=64 scaling, measure compute/exchange overlap ratio
• Semi-naive correctness invariant: Speculative execution with stale deltas may produce redundant tuples (re-derived from previous iterations), but never incorrect tuples. Dedup at merge eliminates duplicates. Monotonicity of Datalog ensures old_delta ⊆ new_delta, so workers using stale deltas compute a superset that converges to the same fixpoint.

Phase 4: Long-term Optimizations

Goal: Production-grade scaling to W=512+.

• Partitioned IDB (if needed): Hash-partition large IDB tables across workers to reduce memory replication
• NUMA-aware scheduling: Pin worker threads to NUMA nodes; allocate worker memory from local node
• EDB partitioning for BDX mode: Hash-partition large EDB tables (e.g., Var_Type 947k rows) by EXCHANGE key instead of full replication
• Cooperative scheduling: Worker-driven coordination to replace coordinator bottleneck

---

Deliverables Checklist

• Phase 0: 3 atomic commits (ring cap, sorted merge, cache-line padding)
• Phase 0: Code review + architect sign-off
• Phase 0: Full regression suite passing (112+ tests)
• Phase 1: Async delta slot + epoch convergence implementation
• Phase 1: DOOP W=4 correctness validation (same tuple count as baseline)
• Phase 1: TSan clean at W=8
• Phase 2: Tree-structured parallel merge implementation
• Phase 2: W=16 scaling validation (near-linear improvement)
• Phase 3: Pipelined compute/exchange overlap
• Phase 3: W=64 scaling validation
• Phase 4: Production optimization + W=512 target

---

Performance Targets

Phase	W	Target Time	Speedup vs W=1	Speedup vs Current W=4
Baseline	1	752s	1×	—
Baseline	4	26.1s	28.8×	1.0×
Current	8	28.3s	26.6×	0.92× (regression)
Phase 1	8	~26s	28.9×	1.0×
Phase 2	32	~7s	107×	3.7×
Phase 3	128	~2s	376×	13×
Phase 4	512	<1s	752×+	26×+

Note: W=1→W=4 speedup (28.8×) includes algorithmic improvement from BDX incremental evaluation, not just parallelism. Targets for W>8 assume the parallel merge tree and pipelining unlock true parallelism on top of the BDX algorithmic advantage.

---

Technical Notes

Semi-Naive Correctness Under Async Execution

Speculative execution with stale deltas preserves semi-naive correctness:

• Monotonicity: Datalog is monotone — stale_delta ⊆ fresh_delta (adding facts never removes derived facts)
• Redundant but never incorrect: Workers using stale deltas may re-derive tuples from previous iterations, but dedup at merge eliminates duplicates
• Convergence: The fixpoint is unique regardless of evaluation order; async execution changes the path but not the destination
• Formal invariant: At every epoch boundary, the union of all worker-produced tuples is a subset of the minimal model

Thread Safety

• Lock-free synchronization via atomic epoch counter (TSan validated for SPSC queue)
• Per-worker SPSC queues eliminate cross-worker contention entirely
• Coordinator is sole consumer — no multi-consumer race conditions
• Memory ordering: acquire/release semantics on cursor updates; relaxed for diagnostic counters

Backward Compatibility

• W=1 fast path fully preserved: All async machinery is bypassed when num_workers == 1
• Existing API unchanged: wl_session_snapshot() remains the public entry point
• Graceful degradation: If lock-free queue is unavailable (build config), falls back to current barrier-based path

---

Dependencies & Blockers

• Lock-free MPMC queue implementation (wirelog/util/lockfree_queue.h, wirelog/util/lockfree_queue.c)
• Async exchange architecture design (see protocol specification above)
• Root cause analysis and W=4/W=8 profiling (see Root Cause Analysis section)
• Phase 0 code review (in progress on feature/issue-401-distributed-async-exchange)
• DOOP W>1 EOVERFLOW regression fix (current build fails at W=4 and W=8 — likely interaction with fix: DOOP benchmark fails with multi-worker execution (W=4, W=8) #404 commits)
• TDD W=1 sequential path performance validation

Related Issues

• DOOP performance does not scale beyond w4: recursive strata run single-threaded #390: Sorted merge TODO — O(N log N) → O(N+D) merge-append (addressed in Phase 0, Commit 2)
• feat(#390): enable recursive TDD scaling beyond w4 with BDX mode #398: num_workers cap — ring capacity limitation (addressed in Phase 0, Commit 1)
• fix: DOOP benchmark fails with multi-worker execution (W=4, W=8) #404: join_output_limit regression — TDD worker join truncation fixes (prerequisite: must be stable before Phase 1)

---

Discussion Points

1. W=512 realism: Is W=512 a realistic deployment target, or should we cap at W=256? The tree merge overhead at W=512 (9 levels) vs W=256 (8 levels) is marginal, but memory bandwidth may become the true bottleneck before W=512.

2. Partitioned IDB timing: Should hash-partitioned IDB (Phase 4) be pulled into Phase 2 to improve memory locality earlier? Trade-off: more complex merge logic vs better per-worker cache utilization.

3. Cooperative scheduling: Should Phase 3 explore worker-driven coordination (workers decide when to proceed) vs coordinator-driven (coordinator signals epoch transitions)? Cooperative scheduling eliminates the coordinator as a bottleneck but complicates convergence detection.

4. EDB partitioning for DOOP: DOOP's large EDB tables (Var_Type 947k rows, Var_DeclaringMethod 947k rows) are fully replicated to all workers. Hash-partitioning EDB by EXCHANGE key would reduce memory bandwidth at W=8+, but requires changes to tdd_init_workers_hybrid (eval.c:3199-3228). Is this worth the complexity?

5. Speculative execution safety: The monotonicity argument for stale-delta correctness assumes pure Datalog (no negation across strata). Should we add a runtime assertion that speculative execution is only enabled for non-negated strata?

[justinjoy]

Team Review: Architect, Critic, Executor Analysis

Summary

Three specialists have completed a comprehensive review of Issue #407's distributed async evaluation roadmap:

1. Architect (Design Review): Lock-free protocol is sound; Phase 0 foundation is mechanically correct; 6 recommendations for P0-P3.
2. Critic (Strategic Review): REVISE VERDICT — 9 findings including 2 CRITICAL blockers (W=256 hard limit vs W=512+ goal, and zero integration of lock-free queue with evaluation loop).
3. Executor (Task Decomposition): Provided detailed Phase 0-4 task breakdown with file locations, complexity, acceptance criteria, and parallelization strategy.

---

Key Findings

From Architect Review

Strengths:

• SPSC design with acquire/release memory ordering is correct
• Cache-line padding prevents false sharing (though Apple Silicon fix needed)
• Dynamic ring capacity enables W=512 task submission
• Sorted merge-insert achieves O(N+D) optimization

Critical Recommendations:

1. Add epoch field to wl_delta_msg_t before integration
2. Fix dequeue_all fairness (currently sequential drain, should be round-robin)
3. Design epoch convergence protocol before queue integration (hardest design problem)
4. Eliminate O(W²) exchange buffer matrix → O(W) per-worker slots
5. Fix Apple Silicon cache-line padding (64B → 128B)
6. Defer speculative execution until Phases 1-3 proven

---

From Critic Review (VERDICT: REVISE)

CRITICAL Findings (blocks execution):

1. W=512+ goal contradicts WL_MAX_WORKERS_HARD_LIMIT = 256

   • Session layer clamps at 256 (session.c:412)
   • Workqueue/lock-free-queue engineered for W=512+
   • Stated goal is unreachable without raising the hard limit
   • Action: Add "raise WL_MAX_WORKERS_HARD_LIMIT to 512+" as explicit Phase 0 task OR revise goal to W=256

2. Lock-free queue has ZERO integration with evaluation loop

   • wl_mpmc_enqueue/dequeue not called from eval.c
   • Current exchange path entirely synchronous (barrier → serial union → partition → broadcast)
   • "Distributed async exchange" doesn't exist yet
   • Action: Must add concrete control flow design before Phase 1

MAJOR Findings:

3. Amdahl's Law analysis is flawed — serial fraction grows with W, not shrinks

   • Exchange cost is O(total_tuples) regardless of W
   • More workers → smaller per-worker compute → larger serial fraction percentage
   • 5-15% estimate is unrealistic at W=256+
   • Action: Model serial fraction as f(W), measure at W=4, W=16, W=64

4. No Phase 0 performance measurements

   • Sorted merge and cache-line padding include no benchmark results
   • Existing suite (bench_flowlog) could measure exchange phase time
   • Action: Run CSPA/DOOP before/after Phase 0, report serial fraction

5. test_workqueue_capacity.c is dead code

   • References removed WL_WQ_RING_CAP macro
   • Disabled in meson.build but file remains
   • Dynamic capacity formula has no test
   • Action: Delete or rewrite to test max(256, next_pow2(num_workers * 2))

6. MPMC naming misleads about thread safety

   • Actual design is SPSC (single-consumer only)
   • "MPMC" suggests any thread can dequeue → data race if misused
   • Action: Rename to wl_mpsc_queue_t or add prominent consumer-thread warning

MINOR Findings:

• tdd_dedup_rel after tdd_sorted_merge_append may be redundant (line 3754)
• MSVC atomic shim uses compiler barrier only (safe on x86, unsafe on ARM Windows)
• Memory cost at W=512: 20MB/worker × 512 = 10GB infrastructure (no scaling plan)

Missing from Roadmap:

• Concrete integration plan (how queue replaces barrier)
• Distributed exchange design (tree-reduce or pairwise merge)
• Memory scaling strategy
• Correctness oracle test (async exchange ≡ sync exchange)

---

From Executor Review (Task Decomposition)

Phase 0 (COMPLETED): 3 commits, ~130 LOC

• 0.1: Dynamic ring capacity ✓
• 0.2: O(N+D) sorted merge ✓
• 0.3: Cache-line padding ✓

Phase 1 (NOT STARTED): 6 tasks, ~350 LOC, Medium risk

• 1.1: Epoch counter + convergence state
• 1.2: Worker push delta via lock-free queue
• 1.3: Replace barrier with coordinator poll loop
• 1.4: Create MPMC queue, wire into coordinator
• 1.5: W=1 fast path preservation
• 1.6: Regression test (async ≡ barrier convergence)
• Dependencies: 1.1 ∥ 1.4 → 1.2 → 1.3 → 1.5 → 1.6
• Critical Path: Phase 1 is highest ROI (eliminates barrier idle time)

Phase 2 (DESIGN): 5 tasks, ~425 LOC, Medium risk

• 2.1: Binary merge tree data structure
• 2.2: Parallel merge-dedup kernel
• 2.3: Integrate tree merge into coordinator exchange
• 2.4: Eliminate O(W²) exchange buffer matrix
• 2.5: Update meson.build
• Dependencies: 2.1 ∥ 2.2 → 2.3 → 2.4

Phase 3 (DESIGN): 4 tasks, ~345 LOC, High risk

• 3.1: Double-buffer epoch delta slots
• 3.2: Speculative worker execution (stale delta)
• 3.3: Coordinator pipeline (compute/exchange overlap)
• 3.4: Monotonicity assertion for safety
• Dependencies: 3.1 → 3.2 → {3.3 ∥ 3.4}

Phase 4 (FUTURE): 4 tasks, ~900 LOC, High risk

• 4.1: Hash-partitioned IDB
• 4.2: NUMA-aware scheduling
• 4.3: EDB partitioning for BDX
• 4.4: Cooperative scheduling (worker-driven)

Parallelization Summary:

• Phase 0: All tasks parallel (all done)
• Phase 1: 1.1 ∥ 1.4 can run parallel; rest sequential
• Phase 2: 2.1 ∥ 2.2 ∥ 2.5 parallel; then 2.3 → 2.4
• Phase 3: 3.1 → 3.2 → (3.3 ∥ 3.4)
• Phase 4: 4.1 ∥ 4.2 parallel; then 4.3, 4.4

---

Recommendations (Consensus)

For Immediate Action:

1. Raise or justify W=256 hard limit (critical blocker)
2. Design Phase 1 integration (lock-free queue → evaluation loop control flow)
3. Add epoch field to wl_delta_msg_t (zero-cost, required for Phase 1)
4. Benchmark Phase 0 (measure exchange phase time before/after)
5. Model Amdahl's Law as f(W) (measure serial fraction at W=4, 16, 64)
6. Delete or rewrite test_workqueue_capacity.c (dead code cleanup)

For Phase 1 Execution:

• Prerequisite: Must validate DOOP W=4/W=8 stability (resolve EOVERFLOW blocker from fix: DOOP benchmark fails with multi-worker execution (W=4, W=8) #404)
• Lock-free queue is ready; synchronous baseline is proven
• Critical: epoch convergence protocol must produce identical results to barrier path
• Recommendation: Start with 1.1 + 1.4 (parallel), then proceed sequentially 1.2 → 1.3 → 1.5 → 1.6

For Phase 2+:

• Phase 2 (tree-reduce) is the actual scaling lever — don't skip
• Phase 3 (pipelining) only matters if Phase 2 reduces serial fraction below ~5%
• Phase 4 (NUMA/partitioning) is future optimization — defer until W=256+ validates

---

Reference Links

Architect findings: Lock-free design, memory ordering, SPSC correctness, Phase validation (Tasks 1.1-1.6, 2.1-2.4, 3.1-3.4 with file/line references)

Critic findings: 9 findings (2 CRITICAL, 6 MAJOR, 1 MINOR) with evidence, impact analysis, and fixes

Executor findings: Phase 0-4 detailed task breakdown with dependencies, parallelization, complexity estimates, acceptance criteria

---

Next Steps

1. Resolve CRITICAL findings (W hard limit, integration design)
2. Create issue sub-tasks for Phase 1 (Tasks 1.1-1.6)
3. Benchmark Phase 0 changes (exchange phase time)
4. Validate DOOP W=4 correctness (prerequisite for Phase 1)
5. Execute Phase 1 with epoch convergence correctness oracle

[justinjoy]

Architect Review: Issue #407 Distributed Async Evaluation Design

Summary

I reviewed the entire feature branch including lock-free queue, workqueue, TDD coordinator, and all tests. The lock-free SPSC design is mechanically sound and the Phase 0 foundation is correct, but there are critical architectural risks and a fundamental contradiction between the stated goal and current code limits.

---

Analysis: Lock-Free Queue Design ✅

Strengths:

• SPSC (single-producer, single-consumer) ring design is optimal — no CAS, no contention
• Memory ordering correct: acquire-release pairs form minimum necessary fence
• Cache-line padding prevents false sharing (though Apple Silicon needs 128B not 64B)
• Monotonic cursor wrapping is ABA-free
• TSan-clean SPSC pattern with C11 atomics

Risks:

• dequeue_all fairness concern: drains worker 0 completely before worker 1 (impacts epoch ordering)
• MSVC shim uses compiler barrier only (safe on x86 TSO, unsafe on ARM without MemoryBarrier)
• MPMC naming misleads about thread safety (actual design is MPSC)

---

Phase Validation

Phase 0 (DONE): Ring capacity, sorted merge, cache-line padding all correct individually

• But: Zero integration with evaluation loop yet
• But: Sorted merge benefit unvalidated (no perf measurements)

Phase 1 (NOT STARTED): Replace barrier with epoch-based async

• Critical dependency: Must design convergence protocol BEFORE integration
• Risk: Convergence detection changes fundamentally with async epochs

Phase 2 (DESIGN): Tree-structured parallel merge

• Prerequisite: Workers need peer-to-peer merge capability (not just coordinator-mediated)
• Trade-off: Helps gather but not scatter

Phase 3 (DESIGN): Pipelined compute/exchange overlap

• Highest risk: Requires session snapshot/restore for rollback
• Correctness: Semi-naive evaluation tolerates stale deltas under monotonicity

---

6 Recommendations (Priority Order)

P0: Add epoch field to wl_delta_msg_t (zero-cost now)

• File: lockfree_queue.h:59-63
• Why: Without this, epoch-based convergence detection cannot use the queue
• Cost: 4 bytes (padding fills it)

P0: Fix dequeue_all fairness

• File: lockfree_queue.c:274-286
• Change: Round-robin drain instead of sequential
• Why: Prevents epoch-ordering violations when workers produce at different rates

P1: Design epoch convergence protocol (before integration)

• Must specify: When do workers enqueue? When does coordinator poll? What replaces barrier?
• Must validate: Async convergence produces identical results to barrier path

P1: Eliminate O(W²) exchange buffer matrix

• Replace exchange_bufs[W][W] with per-worker outbound arrays exchange_out[dst]
• Impact: W=512 scales to manageable memory (not O(W²))

P2: Fix Apple Silicon cache-line padding

• 64B padding leaves tail/head on same 128B line on M-series
• Add #ifdef aarch64 128B padding

P3: Defer speculative execution

• Requires copy-on-write session snapshots (don't exist)
• Defer until Phases 1-3 deliver measurable improvement

---

Trade-offs

Option	Pros	Cons
A: Incremental async (Phase 1 only)	Lower risk, proves epoch protocol	Still serial exchange
B: Full async + tree merge (Phase 1+2)	Eliminates both bottlenecks	Complex, harder to debug
C: Keep sync, optimize only	Lowest risk	Doesn't solve Amdahl's Law limit

Recommendation: Option A first, then B if A proves insufficient.

---

Critical Path Items

1. Protocol design (epoch convergence) — blocking Phase 1 integration
2. Correctness oracle (async = barrier) — must be validated before Phase 1 merge
3. W=512 vs W=256 scaling target — clarify in issue scope

The infrastructure is well-built, but integration is the hard part.

[justinjoy]

Team Review: Architecture + Strategy + Execution Consensus

Review Team

• Architect (opus): Design validation, protocol correctness, risk identification
• Critic (opus): Strategic roadmap review, assumption validation, scope challenge
• Executor (opus): Task decomposition, implementation complexity, parallelization analysis

---

Team Consensus: REVISE with Conditions

Overall Verdict: The Phase 0 foundation (lock-free queue, ring capacity, sorted merge) is mechanically sound, BUT the roadmap has two CRITICAL blockers that must be resolved before Phase 1 can proceed.

---

Two CRITICAL Blockers (Must Fix First)

1. W=256 Hard Limit vs W=512+ Goal (Contradiction)

Evidence:

• Issue Distributed Async Evaluation — Support W=512+ Multi-Worker Scaling #407 states: "support W=512+ workers"
• Code fact: WL_MAX_WORKERS_HARD_LIMIT = 256 at session.c:412
• Result: Any request above W=256 is clamped; W=512 is unreachable

Action Required:

• Option A: Raise WL_MAX_WORKERS_HARD_LIMIT to 512+ with memory budget analysis (20MB/worker × 512 = 10GB)
• Option B: Revise Issue Distributed Async Evaluation — Support W=512+ Multi-Worker Scaling #407 goal to W=256 and document why that's sufficient

Verdict: Pick one; document the decision in issue comments. Current state is self-contradictory.

---

2. Lock-Free Queue Has Zero Integration (Async Doesn't Exist Yet)

Evidence:

• Lock-free MPMC queue is fully implemented and tested (commits f6ea9a8, b0487a5)
• Current TDD exchange (eval.c:3668-3756) remains entirely synchronous: barrier → serial union → partition → broadcast
• Zero calls to wl_mpsc_enqueue or wl_mpsc_dequeue in eval.c
• Result: "Distributed async exchange" infrastructure exists but is not integrated; async isn't happening

What Must Happen:
Before any Phase 1 implementation begins, we need a concrete integration design that answers:

1. When do workers enqueue deltas? (After computing delta_rels[ri]?)
2. When does coordinator poll the queue? (In the BARRIER replacement loop?)
3. What replaces wl_workqueue_wait_all()? (Poll loop with epoch counter?)
4. How is convergence detected without barrier? (Atomic epoch flags?)
5. What's the correctness oracle? (Async convergence ≡ barrier convergence?)

Verdict: Design document required (pseudocode + control flow diagram) before Phase 1 implementation. Architect, Executor, and implementation lead must reach consensus on the protocol.

---

Six MAJOR Issues (Pre-Phase 1 Cleanup)

3. Amdahl's Law Analysis is Flawed

• Current assumption: 5-15% static serial fraction
• Reality: Exchange cost is O(total_tuples), independent of W
• As W grows: per-worker compute shrinks, serial fraction increases
• Required fix: Model serial fraction as f(W), measure at W=4/16/64, validate assumptions

4. Phase 0 Has No Performance Validation

• Sorted merge claims O(N+D) vs O(N log N) — no measurements
• Cache-line padding claims 2-10× improvement — no benchmarks
• Required: CSPA/DOOP before/after Phase 0, measure exchange phase time specifically

5. MPMC Naming Misleads About Thread Safety

• Actual design: W independent SPSC rings (single-consumer only)
• Naming "MPMC" suggests any thread can dequeue → data race if misused
• Required: Rename to wl_mpsc_queue_t or add prominent "consumer-thread only" warning

6. test_workqueue_capacity.c is Dead Code

• References removed WL_WQ_RING_CAP macro (no longer exists post-a5a28b9)
• File disabled in meson.build but remains in tree
• Dynamic capacity formula has no test
• Required: Rewrite test for max(256, next_pow2(num_workers * 2)) logic

7. Amdahl Analysis Missing Serial Fraction Measurements

• Without real W=4/8/16 data, we can't validate whether exchange is the real bottleneck
• Required: Instrument tdd_bdx_exchange_deltas and measure ratio at multiple W

8. W=512 Memory Budget Unaddressed

• 20MB per worker × 512 = 10GB infrastructure cost
• No plan to reduce per-worker allocation at high W
• Required: Memory scaling strategy or realistic W target

---

Team Recommendations (Priority Order)

Immediate (Before Any Phase 1 Work)

1. Resolve W=256 vs W=512+ contradiction (Task Add stratification and SCC detection for Datalog programs #4)
2. Write Phase 1 integration design document (Task Add DD Plan Translator: IR to DD operator graph #5)
3. Add epoch field to wl_delta_msg_t (zero-cost, unblocks Phase 1) (Task Add Logic Fusion optimization pass (Phase 1) #6)
4. Benchmark Phase 0 changes (validate sorted merge + padding) (Task Differential Dataflow Homebrew formula 제출이 필요합니다 #7)
5. Measure serial fraction as f(W) (validate Amdahl's Law) (Task Add C-to-Rust FFI marshalling layer for DD execution #8)
6. Fix test_workqueue_capacity.c (dead code cleanup) (Task Add Rust DD executor with interpreter-based dataflow #9)
7. Rename MPMC → MPSC (clarity + safety) (Task Add Rust DD executor with C→Rust FFI pipeline #10)

Then (Before Phase 1 Validation)

8. Validate DOOP W=4/W=8 stability (prerequisite for correctness oracle) (Task Add CLI driver for executing .dl Datalog files #11)

Finally (Phase 1 Implementation, Sequential)

9. Tasks Add LGPL-3.0 license headers to Rust source files #12-17: Epoch convergence → poll loop → fast path → regression test

---

What Must NOT Happen

❌ Do not start Phase 1 implementation until:

• CRITICAL blockers Restructure CI pipeline: staged workflow with static analysis reports #1 and Restructure CI pipeline: staged workflow with static analysis reports #2 are resolved
• Integration design document is approved
• DOOP W=4/W=8 baseline is stable
• Correctness oracle is specified

---

Phase Execution Timeline (If All Prerequisites Met)

Phase	Complexity	Critical Path	ROI
Phase 0 (DONE)	Low	None (complete)	Removed scaling wall
Phase 1	Medium	Epoch convergence protocol	Eliminates barrier idle
Phase 2	Medium	Tree-reduce implementation	Eliminates serial merge
Phase 3	High	Pipelined state machine	Latency hiding
Phase 4	High	NUMA + partitioning	Production scaling

---

Next Conversation

1. Team closes on W=256 vs W=512+ decision
2. Architect + Executor design Phase 1 integration protocol
3. Executor writes detailed task list for Phase 1
4. Execute Phase 1 with correctness oracle validation
5. Measure Phase 1 scaling (W=4 vs W=8 vs W=16)
6. Decide: Is Phase 2 (tree-reduce) necessary, or did Phase 1 solve the regression?

Signed: Architect, Critic, Executor Team
Date: 2026-03-31
Status: REVISE (resolved blockers → ACCEPT)

[justinjoy]

Team Safety Review: Hard Limit Removal Impact (Commit 172a4a7)

VERDICT: REVISE — The hard limit removal achieves the W=512+ goal but replaces a tight safety net with a dangerous footgun. Two critical infrastructure scaling issues discovered.

---

Critical Issues (blocks shipping current change)

1. W×W Exchange Buffer Causes Quadratic Memory Blowup

• File: wirelog/columnar/eval.c:2514-2521
• Issue: tdd_alloc_exchange_bufs allocates W×W pointer matrix
• Impact:
  • W=512: 2MB pointers (fine)
  • W=10,000: 800MB pointers ALONE, before any data
  • W=100,000: 80GB pointers → instant OOM
• Problem: Documentation says "20MB per worker" (linear), but W² exchange cost is hidden
• Why shipped: Exchange buffer cost not mentioned in session.c:402 comment

2. Integer Overflow in Workqueue Ring Capacity

• File: wirelog/workqueue.c:153
• Issue: uint32_t min_cap = num_workers * 2u; wraps when num_workers > 2.1B
• Impact: Potential cap=0 → calloc(0) → heap corruption
• Why critical: Now that UINT32_MAX is allowed, this code path is reachable

---

Major Issues (causes rework)

3. Memory Ledger Does NOT Gate Session Creation

• File: session.c:516-550 (ledger allocated AFTER threads/workqueue)
• Issue: Ledger only tracks runtime allocations, not infrastructure allocations
• Impact: W=10,000 allocates ~200GB infrastructure before ledger is consulted
• Consequence: "operator responsible for memory budgeting" is aspirational — no programmatic enforcement

4. Per-Worker Budget Starves at High W

• File: session.c:543
• Formula: budget = total / (num_workers + 1)
• Example: 32GB system with W=512 → 48MB per worker (8MB arena + 4MB pool + 36MB headroom = too tight)
• Impact: Diminishing returns undocumented

5. Thread Creation Fails Silently at High W

• File: workqueue.c:173-187
• Issue: Creates OS threads in loop with no ulimit check
• Impact: At typical ulimit -u=4096, thread creation fails after ~1000 threads, partial allocations leaked
• Consequence: Expensive failure path with unclear error message

---

Architect's Recommendation

Replace hard limit with RAM-proportional cap:

/* Compute dynamic maximum from physical RAM */
uint64_t phys = col_detect_physical_memory();
uint32_t ram_cap = (phys > 0)
    ? (uint32_t)(phys / (20ULL * 1024 * 1024))  /* 20MB per worker */
    : 256u;  /* fallback */
if (cap > ram_cap) cap = ram_cap;

This allows W=512+ on capable hardware while preventing absurd values.

---

Critic's Recommendation

Option B Hybrid — Dynamic soft limit + documented override:

1. Compute max_safe_W = min(sqrt(RAM/8), RAM/(20MB×2), 4096)
2. Use as default cap BUT allow override via env var (with warning)
3. Add W² cost to documentation
4. Fix integer overflow in workqueue.c:153

Example with 64GB system:

• max_safe_W = min(sqrt(64GB/8), 64GB/(40MB), 4096)
• = min(2896, 1600, 4096) = 1600 workers allowed by default
• Operator can override: WIRELOG_MAX_WORKERS=10000 → warns "requires ~800GB exchange buffer, only 64GB available"

---

Recommended Fix (Do NOT Revert)

The workqueue dynamic ring (commit a5a28b9) is good. Just add:

1. Pre-flight memory check at session creation (before workqueue_create):

uint64_t infra_estimate = (W * 20) + (W * W * 8 / (1024*1024)); /* W² exchange */
if (infra_estimate > available_ram * 0.5) {
    fprintf(stderr, "...\n");  /* clear error */
    return WL_ERROR_MEMORY;
}

2. Integer overflow guard (workqueue.c:153):

if (num_workers > UINT32_MAX / 2u) return NULL;  /* overflow */
uint32_t min_cap = num_workers * 2u;

3. Dynamic soft cap (session.c:420):

uint32_t dynamic_cap = compute_safe_worker_count(physical_ram);
if (cap > dynamic_cap) {
    fprintf(stderr, "[wirelog] clamping W=%u -> %u (memory estimate)\n", cap, dynamic_cap);
    cap = dynamic_cap;
}

4. Document W² cost (session.c:402 comment):

* Per-worker: 20MB infrastructure + 8B per other worker (exchange matrix)
* At W=512: 20MB × 512 + 8B × 512² = 10.2GB + 2MB ≈ 10.2GB

---

Status

• ✅ Hard limit removal unblocks W=512+ (correct goal)
• ❌ No guardrails for infrastructure OOM (current danger)
• 🔧 Recommend adding dynamic cap + pre-flight checks (low effort, high safety gain)

Next step: Revert to WL_MAX_WORKERS_HARD_LIMIT=256 temporarily, then add dynamic cap above it (safer iterative fix).