diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md b/src/a2a3/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md index 7b49a7e1d4..f81f993910 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md @@ -446,7 +446,7 @@ The orchestrator completes fanout wiring before publishing a task to the ready q - Checks `task_state >= COMPLETED` (early-finished optimization) - If not completed: prepends consumer to producer's `fanout_head` via `dep_pool.prepend` - **Releases** `fanout_lock` -3. Atomically releases the +1 redundance + early_finished count via `fanin_refcount.fetch_add` +3. Atomically releases the +1 redundance + completed-fanin count via `fanin_refcount.fetch_add` 4. If all deps satisfied: pushes task to the routed ready queue Zero-fanin tasks and tasks whose claimed producers are already completed skip dep_pool entry allocation and publish directly to the routed ready queue. @@ -683,11 +683,13 @@ reserved core slot and a launch-visible payload before a consumer can pre-occupy `next_block_idx` records reservation progress; `published_block_count` independently establishes publication and early-candidate readiness. -The producer's `dispatch_propagated` claim and fanout snapshot are serialized under -`fanout_lock` with consumer wiring. An edge already in the snapshot is counted by scheduler -propagation; wiring seeds an edge added after the claim and enqueues the consumer if that seed -completes `dispatch_fanin`. This gives each eligible producer-consumer edge exactly one -early-candidate contribution regardless of which side acquires the lock first. +The producer's slot-local dispatch-propagated bit in `ready_state` and fanout snapshot are +serialized under `fanout_lock` with consumer wiring. Wiring already locks and reads the +producer's 64-byte `PTO2TaskSlotState`, so testing this bit does not read a producer payload +cache line. An edge already in the snapshot is counted by scheduler propagation; wiring seeds +an edge added after the claim and enqueues the consumer if that seed completes +`dispatch_fanin`. This gives each eligible producer-consumer edge exactly one early-candidate +contribution regardless of which side acquires the lock first. #### MIX per-core placement diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp index 5aa82dadfc..5eb2c09580 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp @@ -389,7 +389,7 @@ void PTO2OrchestratorState::wire_fanin_task(PTO2TaskSlotState &slot_state, int32 PTO2TaskPayload *payload = slot_state.payload; slot_state.fanin_count = wfanin + 1; - int32_t early_finished = 0; + int32_t completed_fanin = 0; int32_t early_propagated = 0; bool early_disqualified = false; for_each_fanin_slot_state(*payload, [&](PTO2TaskSlotState *producer) { @@ -399,12 +399,12 @@ void PTO2OrchestratorState::wire_fanin_task(PTO2TaskSlotState &slot_state, int32 early_disqualified = true; } if (pstate >= PTO2_TASK_COMPLETED) { - early_finished++; + completed_fanin++; } else { producer->fanout_head = rss.dep_pool.prepend(producer->fanout_head, &slot_state); // The marker shares fanout_lock with propagation's snapshot. A set // marker means this edge is outside that snapshot and needs a seed. - if (producer->payload->dispatch_propagated.load(std::memory_order_acquire) != 0) { + if (!early_disqualified && producer->has_dispatch_propagated()) { early_propagated++; } } @@ -414,17 +414,17 @@ void PTO2OrchestratorState::wire_fanin_task(PTO2TaskSlotState &slot_state, int32 // Completed producers and edges outside a producer's one-shot propagation // snapshot both need wiring-time contributions. Seed only when every // producer is codegen-flagged; one unflagged producer disqualifies the task. - int32_t early_seed = early_finished + early_propagated; + int32_t early_seed = completed_fanin + early_propagated; if (!early_disqualified && early_seed != 0) { int32_t dispatch_fanin = payload->dispatch_fanin.fetch_add(early_seed, std::memory_order_acq_rel) + early_seed; // A fully pre-completed fanin routes normally. If any producer was live, // the exact-full increment must enqueue the early candidate. - if (early_finished != payload->fanin_actual_count && dispatch_fanin == payload->fanin_actual_count) { + if (completed_fanin != payload->fanin_actual_count && dispatch_fanin == payload->fanin_actual_count) { sched->try_enqueue_early_dispatch_candidate(slot_state); } } - int32_t init_rc = early_finished + 1; + int32_t init_rc = completed_fanin + 1; int32_t new_rc = slot_state.fanin_refcount.fetch_add(init_rc, std::memory_order_acq_rel) + init_rc; mark_dep_pool_position(slot_state); if (new_rc >= slot_state.fanin_count) { diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2_types.h b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2_types.h index 09276b0722..1a947a9666 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2_types.h +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2_types.h @@ -306,7 +306,6 @@ struct PTO2TaskPayload { // mask and a late stager rings only its remaining bits. A sync_start consumer // preserves the mask for rendezvous counting and its single launch pass. std::atomic early_dispatch_state{0}; - std::atomic dispatch_propagated{0}; // PRODUCER side: once-guard for fanout propagation // The launch owner publishes COMPLETE only after all owned doorbells are // visible, keeping fanout private until every gated block has launched. std::atomic early_dispatch_launch_state{PTO2_EARLY_DISPATCH_LAUNCH_NONE}; @@ -403,13 +402,12 @@ struct PTO2TaskPayload { // one of ITS producers is flagged (propagate_dispatch_fanin bumps // dispatch_fanin and may CAS early_dispatch_state on any consumer, independent of the // consumer's own hint). So they MUST be zeroed here unconditionally. - // Publication, propagation, and launch fields share this same - // per-submit lifetime and are reset here too. + // Publication and launch fields share this same per-submit lifetime + // and are reset here too. early_dispatch_state.store(PTO2_EARLY_DISPATCH_NONE, std::memory_order_relaxed); for (int w = 0; w < PTO2_EARLY_DISPATCH_CORE_MASK_WORDS; w++) staged_core_mask[w].store(0, std::memory_order_relaxed); dispatch_fanin.store(0, std::memory_order_relaxed); - dispatch_propagated.store(0, std::memory_order_relaxed); published_block_count.store(0, std::memory_order_relaxed); early_dispatch_launch_state.store(PTO2_EARLY_DISPATCH_LAUNCH_NONE, std::memory_order_relaxed); running_slot_count.store(0, std::memory_order_relaxed); @@ -442,7 +440,7 @@ static_assert( * Per-task slot scheduling state (scheduler-private, NOT in shared memory) * * Consolidates all hot-path scheduling fields into a single cache-friendly - * structure (32 bytes = half a cache line). Accessing any field of a task's + * structure (64 bytes = one cache line). Accessing any field of a task's * slot state brings all related fields into the same cache line. * * Concurrency notes: @@ -480,6 +478,12 @@ enum PTO2DeferredCompletionFlag : uint8_t { PTO2_SUBTASK_DEFERRED = 4, }; +enum PTO2DispatchPropagationFlag : uint8_t { + PTO2_DISPATCH_PROPAGATED = 8, +}; + +static_assert((PTO2_DISPATCH_PROPAGATED & (PTO2_READY_CLAIMED | PTO2_COMPLETION_DONE | PTO2_SUBTASK_DEFERRED)) == 0); + struct alignas(64) PTO2TaskSlotState { // Fanout lock + list (accessed together under lock in on_task_complete) std::atomic fanout_lock; // Per-task spinlock (0=unlocked, 1=locked) @@ -514,6 +518,8 @@ struct alignas(64) PTO2TaskSlotState { // slot_state (not payload) so fanin walks read the already-hot producer // slot_state cache line. bool allow_early_resolve{false}; + // Concurrent lifecycle updates preserve unrelated bits. Slot reuse clears + // the byte only after the previous task lifetime is quiescent. std::atomic ready_state{PTO2_READY_UNCLAIMED}; int32_t dep_pool_mark{0}; // Dep pool top after Orch-side wiring @@ -560,6 +566,19 @@ struct alignas(64) PTO2TaskSlotState { task = t; } + // Lock-free callers use this only as a fast-path hint. A false result is + // rechecked by try_mark_dispatch_propagated() while holding fanout_lock. + bool has_dispatch_propagated() const { + return (ready_state.load(std::memory_order_acquire) & PTO2_DISPATCH_PROPAGATED) != 0; + } + + // The propagation owner holds fanout_lock from this claim through its + // fanout snapshot so wiring can classify late edges exactly once. + bool try_mark_dispatch_propagated() { + return (ready_state.fetch_or(PTO2_DISPATCH_PROPAGATED, std::memory_order_acq_rel) & PTO2_DISPATCH_PROPAGATED) == + 0; + } + void mark_completed() { task_state.store(PTO2_TASK_COMPLETED, std::memory_order_release); ready_state.fetch_or(PTO2_COMPLETION_DONE, std::memory_order_release); diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h index 1d02c32aac..c3941aeeb1 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h @@ -923,9 +923,9 @@ struct PTO2SchedulerState { // The once-claim and fanout snapshot share fanout_lock with wiring, so a // set marker tells wiring that its new edge is outside this snapshot. // The unlocked precheck avoids fanout-lock traffic after propagation. - if (p.payload->dispatch_propagated.load(std::memory_order_acquire) != 0) return; + if (p.has_dispatch_propagated()) return; p.lock_fanout(); - if (p.payload->dispatch_propagated.exchange(1, std::memory_order_acq_rel) != 0) { + if (!p.try_mark_dispatch_propagated()) { p.unlock_fanout(); return; } diff --git a/src/a5/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md b/src/a5/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md index d98f6a84af..01c4b09b09 100644 --- a/src/a5/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md +++ b/src/a5/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md @@ -446,7 +446,7 @@ The orchestrator completes fanout wiring before publishing a task to the ready q - Checks `task_state >= COMPLETED` (early-finished optimization) - If not completed: prepends consumer to producer's `fanout_head` via `dep_pool.prepend` - **Releases** `fanout_lock` -3. Atomically releases the +1 redundance + early_finished count via `fanin_refcount.fetch_add` +3. Atomically releases the +1 redundance + completed-fanin count via `fanin_refcount.fetch_add` 4. If all deps satisfied: pushes task to the routed ready queue Zero-fanin tasks and tasks whose claimed producers are already completed skip dep_pool entry allocation and publish directly to the routed ready queue. diff --git a/src/a5/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp b/src/a5/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp index 69d3493aa6..9cfcb49234 100644 --- a/src/a5/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp +++ b/src/a5/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp @@ -376,19 +376,19 @@ void PTO2OrchestratorState::wire_fanin_task(PTO2TaskSlotState &slot_state, int32 PTO2TaskPayload *payload = slot_state.payload; slot_state.fanin_count = wfanin + 1; - int32_t early_finished = 0; + int32_t completed_fanin = 0; for_each_fanin_slot_state(*payload, [&](PTO2TaskSlotState *producer) { producer->lock_fanout(); int32_t pstate = producer->task_state.load(std::memory_order_acquire); if (pstate >= PTO2_TASK_COMPLETED) { - early_finished++; + completed_fanin++; } else { producer->fanout_head = rss.dep_pool.prepend(producer->fanout_head, &slot_state); } producer->unlock_fanout(); }); - int32_t init_rc = early_finished + 1; + int32_t init_rc = completed_fanin + 1; int32_t new_rc = slot_state.fanin_refcount.fetch_add(init_rc, std::memory_order_acq_rel) + init_rc; mark_dep_pool_position(slot_state); if (new_rc >= slot_state.fanin_count) { diff --git a/tests/ut/cpp/a2a3/test_wiring.cpp b/tests/ut/cpp/a2a3/test_wiring.cpp index a085e2bdf9..72ef904e39 100644 --- a/tests/ut/cpp/a2a3/test_wiring.cpp +++ b/tests/ut/cpp/a2a3/test_wiring.cpp @@ -174,7 +174,7 @@ TEST_F(WiringTest, WireTaskAllProducersEarlyFinished) { // fanin_count = 2 + 1 = 3 EXPECT_EQ(task_slot.fanin_count, 3); - // early_finished = 2, init_rc = 2 + 1 = 3, so refcount should hit fanin_count + // completed_fanin = 2, init_rc = 2 + 1 = 3, so refcount should hit fanin_count EXPECT_GE(task_slot.fanin_refcount.load(), task_slot.fanin_count); // Task should be in ready queue @@ -210,7 +210,7 @@ TEST_F(WiringTest, WireTaskProducersPendingTaskNotReady) { // fanin_count = 3 (2 + 1) EXPECT_EQ(task_slot.fanin_count, 3); - // early_finished = 0, init_rc = 1 -> not ready + // completed_fanin = 0, init_rc = 1 -> not ready EXPECT_EQ(task_slot.fanin_refcount.load(), 1); EXPECT_LT(task_slot.fanin_refcount.load(), task_slot.fanin_count); @@ -226,6 +226,26 @@ TEST_F(WiringTest, WireTaskProducersPendingTaskNotReady) { EXPECT_EQ(producer_slots[1].fanout_head->slot_state, &task_slot); } +TEST_F(WiringTest, DispatchPropagationMarkerPreservesReadyStateFlagsAndResets) { + alignas(64) PTO2TaskSlotState producer; + init_slot(producer, PTO2_TASK_PENDING, 1, 1); + + EXPECT_FALSE(producer.has_dispatch_propagated()); + EXPECT_TRUE(producer.try_mark_dispatch_propagated()); + EXPECT_TRUE(sched.try_claim_ready_once(producer)); + producer.mark_completed(); + producer.mark_any_subtask_deferred(); + EXPECT_FALSE(producer.try_mark_dispatch_propagated()); + EXPECT_EQ( + producer.ready_state.load(std::memory_order_relaxed), + PTO2_READY_CLAIMED | PTO2_COMPLETION_DONE | PTO2_SUBTASK_DEFERRED | PTO2_DISPATCH_PROPAGATED + ); + + producer.reset_for_reuse(); + EXPECT_FALSE(producer.has_dispatch_propagated()); + EXPECT_EQ(producer.ready_state.load(std::memory_order_relaxed), PTO2_READY_UNCLAIMED); +} + // ============================================================================= // Orch-side wiring: mixed early-finished and pending producers // ============================================================================= @@ -253,7 +273,7 @@ TEST_F(WiringTest, WireTaskMixedProducerStates) { // fanin_count = 4 (3 + 1) EXPECT_EQ(task_slot.fanin_count, 4); - // early_finished = 2 (COMPLETED + CONSUMED), init_rc = 3 + // completed_fanin = 2 (COMPLETED + CONSUMED), init_rc = 3 // Not yet 4 -> not ready (one producer still running) EXPECT_EQ(task_slot.fanin_refcount.load(), 3); @@ -321,7 +341,7 @@ TEST_F(WiringTest, WireTaskUnflaggedPrecompletedProducerDoesNotSeed) { wire_fanin(task_slot, 1); EXPECT_EQ(payload.dispatch_fanin.load(), 0); // NOT seeded (direct-only) - // Readiness is unaffected: early_finished(1) + 1 == fanin_count(2) -> ready. + // Readiness is unaffected: completed_fanin(1) + 1 == fanin_count(2) -> ready. EXPECT_GE(task_slot.fanin_refcount.load(), task_slot.fanin_count); } @@ -375,12 +395,12 @@ TEST_F(WiringTest, EarlyDispatchWaitsForAllProducerBlocksPublished) { sched.record_published_blocks(producer, 2); sched.propagate_dispatch_fanin(producer); EXPECT_EQ(payload.dispatch_fanin.load(), 0); - EXPECT_EQ(producer.payload->dispatch_propagated.load(), 0); + EXPECT_FALSE(producer.has_dispatch_propagated()); sched.record_published_blocks(producer, 1); sched.propagate_dispatch_fanin(producer); EXPECT_EQ(payload.dispatch_fanin.load(), payload.fanin_actual_count); - EXPECT_EQ(producer.payload->dispatch_propagated.load(), 1); + EXPECT_TRUE(producer.has_dispatch_propagated()); sched.propagate_dispatch_fanin(producer); EXPECT_EQ(payload.dispatch_fanin.load(), payload.fanin_actual_count); @@ -396,7 +416,7 @@ TEST_F(WiringTest, LateWiredFullyPublishedProducerStillSeedsEarlyDispatch) { producer.allow_early_resolve = true; sched.record_published_blocks(producer, producer.logical_block_num); sched.propagate_dispatch_fanin(producer); - ASSERT_EQ(producer.payload->dispatch_propagated.load(), 1); + ASSERT_TRUE(producer.has_dispatch_propagated()); init_slot(consumer, PTO2_TASK_PENDING, 0, 1); consumer_payload.fanin_actual_count = 1; @@ -582,18 +602,18 @@ TEST_F(WiringTest, EarlyDispatchFanoutWaitsForDoorbellPass) { producer.payload->early_dispatch_state.store(PTO2_EARLY_DISPATCH_STAGING, std::memory_order_relaxed); sched.propagate_dispatch_fanin(producer); EXPECT_EQ(consumer.payload->dispatch_fanin.load(), 0); - EXPECT_EQ(producer.payload->dispatch_propagated.load(), 0); + EXPECT_FALSE(producer.has_dispatch_propagated()); producer.payload->early_dispatch_launch_state.store(PTO2_EARLY_DISPATCH_LAUNCH_RINGING); producer.payload->early_dispatch_state.store(PTO2_EARLY_DISPATCH_DISPATCHED, std::memory_order_release); sched.propagate_dispatch_fanin(producer); EXPECT_EQ(consumer.payload->dispatch_fanin.load(), 0); - EXPECT_EQ(producer.payload->dispatch_propagated.load(), 0); + EXPECT_FALSE(producer.has_dispatch_propagated()); producer.payload->early_dispatch_launch_state.store(PTO2_EARLY_DISPATCH_LAUNCH_COMPLETE, std::memory_order_seq_cst); sched.propagate_dispatch_fanin(producer); EXPECT_EQ(consumer.payload->dispatch_fanin.load(), 1); - EXPECT_EQ(producer.payload->dispatch_propagated.load(), 1); + EXPECT_TRUE(producer.has_dispatch_propagated()); } TEST_F(WiringTest, LateStagerRingsCapturedDoorbellAfterTableReuse) { @@ -636,13 +656,13 @@ TEST_F(WiringTest, EarlyDispatchReleaseConsumesDoorbellMask) { EXPECT_EQ(task.payload->staged_core_mask[0].load(), 0); EXPECT_EQ(get_test_reg_stub_base_addr(), reg_addr); EXPECT_EQ(get_test_reg_stub_value(), (static_cast(token) << 32) | token); - EXPECT_EQ(task.payload->dispatch_propagated.load(), 0); + EXPECT_FALSE(task.has_dispatch_propagated()); sched.record_published_blocks(task, 1); // The staging path must retry even though its earlier state snapshot was // STAGING; release already missed this final publication count. sched.propagate_dispatch_fanin(task); - EXPECT_EQ(task.payload->dispatch_propagated.load(), 1); + EXPECT_TRUE(task.has_dispatch_propagated()); sched.early_dispatch_doorbell_table[core_id].addr = 0x11110000; sched.early_dispatch_doorbell_table[core_id].token = 12; @@ -711,7 +731,7 @@ TEST_F(WiringTest, SyncStartDrainFinalizeRetriesProducerFirstRendezvous) { sync_consumer.payload->running_slot_count.store(2, std::memory_order_seq_cst); EXPECT_TRUE(sched.retry_sync_start_rendezvous_after_drain(sync_consumer)); EXPECT_EQ(sync_consumer.payload->early_dispatch_launch_state.load(), PTO2_EARLY_DISPATCH_LAUNCH_COMPLETE); - EXPECT_EQ(sync_consumer.payload->dispatch_propagated.load(), 1); + EXPECT_TRUE(sync_consumer.has_dispatch_propagated()); EXPECT_EQ(downstream.payload->dispatch_fanin.load(), 1); EXPECT_FALSE(sched.retry_sync_start_rendezvous_after_drain(sync_consumer)); @@ -900,8 +920,8 @@ TEST_F(WiringTest, UnflaggedProducerDoesNotPropagate) { sched.propagate_dispatch_fanin(producer); - EXPECT_EQ(cons_payload.dispatch_fanin.load(), 0); // no bump - EXPECT_EQ(prod_payload.dispatch_propagated.load(), 0); // gate returned before once-guard + EXPECT_EQ(cons_payload.dispatch_fanin.load(), 0); // no bump + EXPECT_FALSE(producer.has_dispatch_propagated()); // gate returned before once-guard } TEST_F(WiringTest, FlaggedPrecompletedCreatorTransparentToEarlyDispatch) { diff --git a/tests/ut/cpp/a5/test_wiring.cpp b/tests/ut/cpp/a5/test_wiring.cpp index f9cc21b9cc..62723ac39f 100644 --- a/tests/ut/cpp/a5/test_wiring.cpp +++ b/tests/ut/cpp/a5/test_wiring.cpp @@ -156,7 +156,7 @@ TEST_F(WiringTest, WireTaskAllProducersEarlyFinished) { // fanin_count = 2 + 1 = 3 EXPECT_EQ(task_slot.fanin_count, 3); - // early_finished = 2, init_rc = 2 + 1 = 3, so refcount should hit fanin_count + // completed_fanin = 2, init_rc = 2 + 1 = 3, so refcount should hit fanin_count EXPECT_GE(task_slot.fanin_refcount.load(), task_slot.fanin_count); // Task should be in ready queue @@ -192,7 +192,7 @@ TEST_F(WiringTest, WireTaskProducersPendingTaskNotReady) { // fanin_count = 3 (2 + 1) EXPECT_EQ(task_slot.fanin_count, 3); - // early_finished = 0, init_rc = 1 -> not ready + // completed_fanin = 0, init_rc = 1 -> not ready EXPECT_EQ(task_slot.fanin_refcount.load(), 1); EXPECT_LT(task_slot.fanin_refcount.load(), task_slot.fanin_count); @@ -235,7 +235,7 @@ TEST_F(WiringTest, WireTaskMixedProducerStates) { // fanin_count = 4 (3 + 1) EXPECT_EQ(task_slot.fanin_count, 4); - // early_finished = 2 (COMPLETED + CONSUMED), init_rc = 3 + // completed_fanin = 2 (COMPLETED + CONSUMED), init_rc = 3 // Not yet 4 -> not ready (one producer still running) EXPECT_EQ(task_slot.fanin_refcount.load(), 3);