From a4d9d8d24220112173e072bd6a4f6d44b3c816d9 Mon Sep 17 00:00:00 2001 From: zclawz Date: Sun, 12 Apr 2026 17:16:18 +0000 Subject: [PATCH 1/3] chore: scaffold parallel onBlock task structure (issue #719) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Annotate the two independent parallel tasks within onBlock: - Task A: verifySignatures (XMSS sig verification, CPU-bound) - Task B: apply_transition (state transition, independent of A) Both are currently sequential. This commit marks the barrier point (forkchoice import) and the task dispatch sites with TODOs and design comments so the actual threading implementation has clear anchors to build on. Phase 1 goal: dispatch A and B to spawned threads concurrently, join at the barrier, then proceed to forkchoice import only if both succeed. Phase 2 will parallelize post-state compute segments. Execution/proof verification (Phase 3) is future work. Closes #719 (partial — scaffolding only) --- pkgs/node/src/chain.zig | 34 +++++++++++++++++++++++++++++++--- 1 file changed, 31 insertions(+), 3 deletions(-) diff --git a/pkgs/node/src/chain.zig b/pkgs/node/src/chain.zig index 52d914853..975d0e25b 100644 --- a/pkgs/node/src/chain.zig +++ b/pkgs/node/src/chain.zig @@ -817,7 +817,22 @@ pub const BeamChain = struct { // import block assuming it is gossip validated or synced // this onBlock corresponds to spec's forkchoice's onblock with some functionality split between this and - // our implemented forkchoice's onblock. this is to parallelize "apply transition" with other verifications + // our implemented forkchoice's onblock. + // + // Parallelization plan (tracked in https://github.com/blockblaz/zeam/issues/719): + // + // Phase 1 – Parallel validations (this PR scaffolds; full impl in follow-up): + // a) verifySignatures — CPU-bound XMSS sig verification, independent of STF + // b) apply_transition — state transition, independent of sig verification result + // Both are dispatched to spawned threads; forkchoice import is gated on both completing + // successfully. + // + // Phase 2 – Parallel post-state compute: + // Independent segments of state computation identified and parallelised. + // + // Phase 3 – Execution / proof verification (future): + // Will be parallelised once Phase 1/2 are stable. + // // Returns a list of missing block roots that need to be fetched from the network pub fn onBlock(self: *Self, signedBlock: types.SignedBlock, blockInfo: CachedProcessedBlockInfo) ![]types.Root { const onblock_timer = zeam_metrics.chain_onblock_duration_seconds.start(); @@ -843,11 +858,20 @@ pub const BeamChain = struct { // If anything below fails, deinit interior first (LIFO: deinit runs before destroy above). errdefer cpost_state.deinit(); - // 2. verify XMSS signatures (independent step; placed before STF for now, parallelizable later) + // 2. verify XMSS signatures — parallel task A (issue #719) + // Currently sequential; TODO: dispatch to a spawned thread concurrently with task B below. + // Inputs: pre_state (read-only), signedBlock (read-only), public_key_cache (read-only) + // Output: error | void + // Barrier: both tasks A and B must succeed before forkchoice import proceeds. // Use public key cache to avoid repeated SSZ deserialization of validator public keys try stf.verifySignatures(self.allocator, pre_state, &signedBlock, &self.public_key_cache); - // 3. apply state transition assuming signatures are valid (STF does not re-verify) + // 3. apply state transition — parallel task B (issue #719) + // Currently sequential (runs after task A); TODO: dispatch concurrently with task A above. + // Inputs: pre_state (read-only clone → cpost_state), block (read-only) + // Output: error | cpost_state mutated in place + // Barrier: forkchoice import is gated on tasks A + B both completing successfully. + // Note: STF runs with validSignatures=true; actual sig verification is task A's responsibility. try stf.apply_transition(self.allocator, cpost_state, block, .{ .logger = self.stf_logger, .validSignatures = true, @@ -868,6 +892,10 @@ pub const BeamChain = struct { var missing_roots: std.ArrayList(types.Root) = .empty; errdefer missing_roots.deinit(self.allocator); + // Barrier point (issue #719): tasks A (sig verification) and B (state transition) must both + // have completed successfully before reaching forkchoice import below. + // Once parallelised, a std.Thread.WaitGroup (or equivalent) will be joined here. + // 3. fc onblock if the block was not pre added by the block production const fcBlock = self.forkChoice.getBlock(block_root) orelse fcprocessing: { const freshFcBlock = try self.forkChoice.onBlock(block, post_state, .{ From 1f7723938bbbb8bef3ecf9ae2a6ab9765dd4e05f Mon Sep 17 00:00:00 2001 From: zclawz Date: Sun, 12 Apr 2026 17:41:49 +0000 Subject: [PATCH 2/3] feat(node): parallelize verifySignatures + apply_transition in onBlock (#719) Spawn two threads inside the computedstate block: - Task A: stf.verifySignatures (XMSS sig verification, read-only inputs) - Task B: stf.apply_transition (state transition, writes to cpost_state only) Both run concurrently. The main thread joins both before proceeding to forkchoice import, acting as the barrier described in #719. self.allocator is wrapped in std.heap.ThreadSafeAllocator so both tasks can allocate safely. If std.Thread.spawn fails for either task, it falls back to inline (sequential) execution, preserving correctness. Also removes the now-redundant scaffold barrier comment. --- pkgs/node/src/chain.zig | 95 +++++++++++++++++++++++++++++++---------- 1 file changed, 72 insertions(+), 23 deletions(-) diff --git a/pkgs/node/src/chain.zig b/pkgs/node/src/chain.zig index 975d0e25b..459bcf5a5 100644 --- a/pkgs/node/src/chain.zig +++ b/pkgs/node/src/chain.zig @@ -858,25 +858,78 @@ pub const BeamChain = struct { // If anything below fails, deinit interior first (LIFO: deinit runs before destroy above). errdefer cpost_state.deinit(); - // 2. verify XMSS signatures — parallel task A (issue #719) - // Currently sequential; TODO: dispatch to a spawned thread concurrently with task B below. - // Inputs: pre_state (read-only), signedBlock (read-only), public_key_cache (read-only) - // Output: error | void - // Barrier: both tasks A and B must succeed before forkchoice import proceeds. - // Use public key cache to avoid repeated SSZ deserialization of validator public keys - try stf.verifySignatures(self.allocator, pre_state, &signedBlock, &self.public_key_cache); - - // 3. apply state transition — parallel task B (issue #719) - // Currently sequential (runs after task A); TODO: dispatch concurrently with task A above. - // Inputs: pre_state (read-only clone → cpost_state), block (read-only) - // Output: error | cpost_state mutated in place - // Barrier: forkchoice import is gated on tasks A + B both completing successfully. - // Note: STF runs with validSignatures=true; actual sig verification is task A's responsibility. - try stf.apply_transition(self.allocator, cpost_state, block, .{ - .logger = self.stf_logger, - .validSignatures = true, - .rootToSlotCache = &self.root_to_slot_cache, - }); + // 2+3. verify XMSS signatures (task A) and apply state transition (task B) in parallel (issue #719) + + // Wrap allocator in thread-safe adapter so both threads can allocate concurrently. + var ts_alloc = std.heap.ThreadSafeAllocator{ .child_allocator = self.allocator }; + const thread_alloc = ts_alloc.allocator(); + + // Task A: verify XMSS signatures + const SigVerifyTask = struct { + allocator: Allocator, + pre_state: *const types.BeamState, + signed_block: *const types.SignedBlock, + pubkey_cache: *xmss.PublicKeyCache, + err: ?anyerror = null, + + fn run(task: *@This()) void { + stf.verifySignatures(task.allocator, task.pre_state, task.signed_block, task.pubkey_cache) catch |e| { + task.err = e; + }; + } + }; + var sig_task = SigVerifyTask{ + .allocator = thread_alloc, + .pre_state = pre_state, + .signed_block = &signedBlock, + .pubkey_cache = &self.public_key_cache, + }; + + // Task B: apply state transition + const StfTask = struct { + allocator: Allocator, + post_state: *types.BeamState, + block: types.BeamBlock, + stf_logger: zeam_utils.ModuleLogger, + root_to_slot_cache: *types.RootToSlotCache, + err: ?anyerror = null, + + fn run(task: *@This()) void { + stf.apply_transition(task.allocator, task.post_state, task.block, .{ + .logger = task.stf_logger, + .validSignatures = true, + .rootToSlotCache = task.root_to_slot_cache, + }) catch |e| { + task.err = e; + }; + } + }; + var stf_task = StfTask{ + .allocator = thread_alloc, + .post_state = cpost_state, + .block = block, + .stf_logger = self.stf_logger, + .root_to_slot_cache = &self.root_to_slot_cache, + }; + + // Spawn threads; fall back to inline execution if spawn fails. + const sig_thread = std.Thread.spawn(.{}, SigVerifyTask.run, .{&sig_task}) catch null_thread: { + SigVerifyTask.run(&sig_task); + break :null_thread null; + }; + const stf_thread = std.Thread.spawn(.{}, StfTask.run, .{&stf_task}) catch null_thread: { + StfTask.run(&stf_task); + break :null_thread null; + }; + + // Join successfully spawned threads (barrier). + if (sig_thread) |t| t.join(); + if (stf_thread) |t| t.join(); + + // Propagate any errors from parallel tasks. + if (sig_task.err) |e| return e; + if (stf_task.err) |e| return e; + break :computedstate cpost_state; }; // If post_state was freshly allocated above and a later step errors (e.g. forkChoice.onBlock, @@ -892,10 +945,6 @@ pub const BeamChain = struct { var missing_roots: std.ArrayList(types.Root) = .empty; errdefer missing_roots.deinit(self.allocator); - // Barrier point (issue #719): tasks A (sig verification) and B (state transition) must both - // have completed successfully before reaching forkchoice import below. - // Once parallelised, a std.Thread.WaitGroup (or equivalent) will be joined here. - // 3. fc onblock if the block was not pre added by the block production const fcBlock = self.forkChoice.getBlock(block_root) orelse fcprocessing: { const freshFcBlock = try self.forkChoice.onBlock(block, post_state, .{ From 9b8e6b424e2d9ff71242baefa8351ea52a0fb8a6 Mon Sep 17 00:00:00 2001 From: zclawz Date: Sun, 12 Apr 2026 18:48:16 +0000 Subject: [PATCH 3/3] fix: resolve CI failure - use per-task arena for SigVerifyTask to avoid OOM on macOS Task A (XMSS sig verification) and Task B (STF) previously shared the same chain allocator via ThreadSafeAllocator, causing peak memory to double vs the sequential baseline (both tasks allocate concurrently). Give Task A its own temporary ArenaAllocator so its allocations are freed when it completes. Task B continues to use the chain allocator for cpost_state, which must outlive the task. Fixes the consistent SIGKILL (OOM) on the macos-latest CI runner. --- pkgs/node/src/chain.zig | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/pkgs/node/src/chain.zig b/pkgs/node/src/chain.zig index 459bcf5a5..d2867b5c7 100644 --- a/pkgs/node/src/chain.zig +++ b/pkgs/node/src/chain.zig @@ -860,9 +860,11 @@ pub const BeamChain = struct { // 2+3. verify XMSS signatures (task A) and apply state transition (task B) in parallel (issue #719) - // Wrap allocator in thread-safe adapter so both threads can allocate concurrently. - var ts_alloc = std.heap.ThreadSafeAllocator{ .child_allocator = self.allocator }; - const thread_alloc = ts_alloc.allocator(); + // Task A gets its own temporary arena so its allocations are freed after it completes, + // keeping peak memory equivalent to the sequential baseline (avoids OOM on CI). + // Task B uses the chain allocator directly since its output (cpost_state) must outlive the task. + var sig_arena = std.heap.ArenaAllocator.init(self.allocator); + defer sig_arena.deinit(); // Task A: verify XMSS signatures const SigVerifyTask = struct { @@ -879,7 +881,7 @@ pub const BeamChain = struct { } }; var sig_task = SigVerifyTask{ - .allocator = thread_alloc, + .allocator = sig_arena.allocator(), .pre_state = pre_state, .signed_block = &signedBlock, .pubkey_cache = &self.public_key_cache, @@ -905,7 +907,7 @@ pub const BeamChain = struct { } }; var stf_task = StfTask{ - .allocator = thread_alloc, + .allocator = self.allocator, .post_state = cpost_state, .block = block, .stf_logger = self.stf_logger,