host_build_graph: dedicated resolution thread (3S+1P)

src/a2a3/runtime/host_build_graph/aicpu/aicpu_executor.cpp

@@ -288,7 +288,11 @@ int32_t AicpuExecutor::run(Runtime *runtime) {
             LOG_ERROR("Thread %d: rt is null after orchestrator error, skipping dispatch", thread_idx);
         } else {
             sched_ctx_.bind_runtime(rt);
-            int32_t completed = sched_ctx_.resolve_and_dispatch(runtime, thread_idx);
+            // 3S+1P: the last thread is the core-less resolution (P) thread; the
+            // rest are core-owning schedulers (S).
+            int32_t completed = (thread_idx == sched_ctx_.p_thread_idx()) ?
+                                    sched_ctx_.run_resolution_thread(runtime, thread_idx) :
+                                    sched_ctx_.resolve_and_dispatch(runtime, thread_idx);
             if (completed < 0) {
                 LOG_ERROR("Thread %d: Scheduler failed with rc=%d", thread_idx, completed);
                 run_rc = completed;

src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_cold_path.cpp

@@ -700,7 +700,21 @@ void SchedulerContext::handshake_partition(Runtime *runtime, int32_t tidx, int32
 bool SchedulerContext::assign_cores_to_threads() {
     // Cluster-aligned round-robin assignment: cluster ci -> sched thread ci % active_sched_threads_.
     // Each cluster = 1 AIC + 2 adjacent AIV; the triple is always kept together.
-    active_sched_threads_ = aicpu_thread_num_;
+    //
+    // 3S+1P: the last AICPU thread is the core-less resolution thread (P); cores
+    // partition across the remaining (aicpu_thread_num_ - 1) scheduler threads
+    // only, so P never owns a cluster and never polls a COND register. P is
+    // mandatory — like tmr's scheduler + orchestrator split, host_build_graph
+    // needs at least two AICPU threads (one S + one P); one thread cannot own
+    // cores and resolve on a dedicated thread at once.
+    if (aicpu_thread_num_ < 2) {
+        LOG_ERROR(
+            "host_build_graph requires aicpu_thread_num >= 2 (1 scheduler + 1 resolution); got %d", aicpu_thread_num_
+        );
+        return false;
+    }
+    p_thread_idx_ = aicpu_thread_num_ - 1;
+    active_sched_threads_ = aicpu_thread_num_ - 1;
     int32_t cluster_count = aic_count_;
 
     // Max clusters any single sched thread can hold: ceil(cluster_count / active_sched_threads_).
@@ -1040,6 +1054,29 @@ void SchedulerContext::on_orchestration_done(
 
     total_tasks_ = total_tasks;
 
+    // Allocate the per-S CompletedTaskQueues here on the boot leader, before it
+    // releases runtime_init_ready_ — no scheduler thread can push until then.
+    // Completed-but-unresolved tasks in flight are bounded by BOTH the total task
+    // count and the ring's task window (a task must occupy a ring slot to run and
+    // complete), so size to the tighter of the two, rounded up to a power of two
+    // and floored at 256. The window already caps this, so there is no artificial
+    // ceiling and a producer never has to spin on a full queue.
+    uint64_t sp_bound = static_cast<uint64_t>(total_tasks);
+    if (sched_->ring_sched_state.ring != nullptr) {
+        uint64_t window = static_cast<uint64_t>(sched_->ring_sched_state.ring->task_window_mask) + 1;
+        if (window < sp_bound) {
+            sp_bound = window;
+        }
+    }
+    uint64_t sp_cap = 256;
+    while (sp_cap < sp_bound) {
+        sp_cap <<= 1;
+    }
+    for (int32_t t = 0; t < active_sched_threads_; t++) {
+        sp_queues_[t].destroy();  // free a prior run's buffer before re-alloc
+        sp_queues_[t].init(sp_cap);
+    }
+
     // Fold tasks completed inline during orchestration
     int32_t inline_completed = static_cast<int32_t>(rt->orchestrator.inline_completed_tasks);
     if (inline_completed > 0) {

src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_completion.cpp

@@ -82,7 +82,7 @@ SlotTransition SchedulerContext::decide_slot_transition(
 // Complete one slot's task: subtask counting, mixed completion, deferred release, profiling.
 void SchedulerContext::complete_slot_task(
     PTO2TaskSlotState &slot_state, int32_t expected_reg_task_id, [[maybe_unused]] PTO2SubtaskSlot subslot,
-    [[maybe_unused]] int32_t thread_idx, int32_t core_id, Handshake *hank, int32_t &completed_this_turn
+    int32_t thread_idx, int32_t core_id, Handshake *hank, [[maybe_unused]] int32_t &completed_this_turn
 #if SIMPLER_DFX
     ,
     uint64_t dispatch_ts, uint64_t finish_ts
@@ -183,48 +183,15 @@ void SchedulerContext::complete_slot_task(
             );
         }
 #endif
+        // 3S+1P: hand the finished task to the dedicated resolution (P) thread.
+        // P publishes completion_flags, drains the wake list, and advances the
+        // watermark — and owns completed_tasks_, so this scheduler thread neither
+        // resolves nor bumps completed_this_turn. (The Resolve swimlane bar is
+        // emitted by P, not here.)
+        sp_queues_[thread_idx].push(&slot_state);
 #if SIMPLER_DFX
-        // Time Resolve (walk the consumer list, decrement each consumer's
-        // fanin, push the newly-ready ones, ring doorbells for early-dispatch
-        // hits) so it renders as a child bar nested inside this iteration's
-        // Complete bar. The 1 µs floor below filters out the ~88% of tasks
-        // with 1-2 consumers (~500 ns Resolve) so only the long broadcast /
-        // reduction walks stand out on the lane.
-        uint64_t resolve_t0 = (l2_swimlane_level_ >= L2SwimlaneLevel::SCHED_PHASES) ? get_sys_cnt_aicpu() : 0;
-#endif
-        // [[maybe_unused]] silences -Werror=unused-but-set-variable on the
-        // profiling-flags-smoke build path where SIMPLER_DFX is OFF and
-        // the Resolve emit below is excluded.
-        [[maybe_unused]] uint32_t consumers_resolved = 0;
-#if SIMPLER_SCHED_PROFILING
-        // SCHED_PROFILING variant takes thread_idx for its per-thread atomic
-        // counter side-effects (g_sched_*_atomic_count[thread_idx], consumed
-        // by the otc_* log lines). It returns CompletionStats whose
-        // `fanout_edges` is the consumer-walk count.
-        consumers_resolved = sched_->on_task_complete(slot_state, thread_idx).fanout_edges;
-#else
-        consumers_resolved = sched_->on_task_complete(slot_state);
-#endif
-#if SIMPLER_DFX
-        if (resolve_t0 != 0) {
-            uint64_t resolve_t1 = get_sys_cnt_aicpu();
-            // Filter: drop Resolve bars under 1 µs so the lane shows only
-            // resolves that did meaningful work (high consumer counts or
-            // doorbells). 50 cycles @ 50 MHz = 1 µs (PLATFORM_PROF_SYS_CNT_FREQ
-            // is the device sys-cnt frequency).
-            constexpr uint64_t RESOLVE_EMIT_MIN_CYCLES = PLATFORM_PROF_SYS_CNT_FREQ / 1'000'000;  // 1 µs
-            if (resolve_t1 - resolve_t0 >= RESOLVE_EMIT_MIN_CYCLES) {
-                l2_swimlane_aicpu_record_sched_phase(
-                    thread_idx, L2SwimlaneSchedPhaseKind::Resolve, resolve_t0, resolve_t1, l2_swimlane.sched_loop_count,
-                    consumers_resolved
-                );
-            }
-        }
         l2_swimlane.phase_complete_count++;
 #endif
-        // Polling: on_task_complete published completion + drained the wake list
-        // inline; no deferred producer-release step.
-        completed_this_turn++;
     }
 
 #if SIMPLER_DFX

src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_context.h

@@ -37,6 +37,57 @@ class Runtime;
 struct Handshake;
 struct PTO2Runtime;
 
+// SPSC ring carrying completed-but-unresolved task slots from one scheduler (S)
+// thread to the dedicated resolution (P) thread. Whole-graph-resident hbg never
+// reclaims a slot, so the queued PTO2TaskSlotState* stays valid until P reads it.
+// Single producer (the owning S thread) / single consumer (P): plain
+// acquire/release on head/tail, no CAS. Capacity is a power of two sized to the
+// in-flight bound (min of total tasks and the ring window), so a producer does
+// not spin on a full queue in practice; the spin is a correctness backstop only
+// (P drains independently, so it always makes progress). The std::atomic cursors
+// make this type non-copyable / non-movable, so `buf` cannot be double-freed
+// through an accidental copy.
+struct CompletedTaskQueue {
+    PTO2TaskSlotState **buf{nullptr};
+    uint64_t cap{0};
+    uint64_t mask{0};
+    alignas(64) std::atomic<uint64_t> head{0};  // consumer (P) cursor
+    alignas(64) std::atomic<uint64_t> tail{0};  // producer (S) cursor
+
+    void init(uint64_t capacity_pow2) {
+        debug_assert(capacity_pow2 != 0 && (capacity_pow2 & (capacity_pow2 - 1)) == 0);
+        cap = capacity_pow2;
+        mask = capacity_pow2 - 1;
+        buf = new PTO2TaskSlotState *[capacity_pow2];
+        head.store(0, std::memory_order_relaxed);
+        tail.store(0, std::memory_order_relaxed);
+    }
+    void destroy() {
+        delete[] buf;
+        buf = nullptr;
+    }
+    // Producer (S). Spins only if the ring is full — sized so this never fires.
+    void push(PTO2TaskSlotState *s) {
+        uint64_t t = tail.load(std::memory_order_relaxed);
+        while (t - head.load(std::memory_order_acquire) >= cap) {
+            SPIN_WAIT_HINT();
+        }
+        buf[t & mask] = s;
+        tail.store(t + 1, std::memory_order_release);
+    }
+    // Consumer (P). Returns nullptr when empty.
+    PTO2TaskSlotState *pop() {
+        uint64_t h = head.load(std::memory_order_relaxed);
+        if (h == tail.load(std::memory_order_acquire)) {
+            return nullptr;
+        }
+        PTO2TaskSlotState *s = buf[h & mask];
+        head.store(h + 1, std::memory_order_release);
+        return s;
+    }
+    uint64_t size() const { return tail.load(std::memory_order_acquire) - head.load(std::memory_order_acquire); }
+};
+
 /**
  * SchedulerContext: owns all scheduler-side state and methods.
  *
@@ -85,6 +136,16 @@ class SchedulerContext {
     // Main scheduler thread entry: poll completion + dispatch ready tasks.
     int32_t resolve_and_dispatch(Runtime *runtime, int32_t thread_idx);
 
+    // Dedicated resolution (P) thread entry (3S+1P). Owns no cores: drains the
+    // per-S CompletedTaskQueues and runs on_task_complete for each finished task
+    // (completion_flags publish + wake-list drain + watermark advance), making P
+    // the sole producer of the ready queues. Owns completed_tasks_ / termination.
+    int32_t run_resolution_thread(Runtime *runtime, int32_t thread_idx);
+
+    // Index of the dedicated resolution (P) thread — the last AICPU thread.
+    // host_build_graph always reserves it (aicpu_thread_num >= 2 is required).
+    int32_t p_thread_idx() const { return p_thread_idx_; }
+
     // Shutdown AICore registers for this thread's assigned cores.
     // Also runs PMU finalize (SIMPLER_DFX) before deinit when enabled.
     // Orchestrator threads (core_trackers_[thread_idx].core_num() == 0) are a no-op.
@@ -157,6 +218,15 @@ class SchedulerContext {
     int32_t aicpu_thread_num_{0};
     int32_t cores_total_num_{0};
 
+    // --- 3S+1P dedicated resolution thread ---
+    // The AICPU threads split into (aicpu_thread_num_ - 1) core-owning schedulers
+    // (S) plus one core-less resolution thread (P) at index p_thread_idx_ =
+    // aicpu_thread_num_ - 1. host_build_graph requires aicpu_thread_num_ >= 2 (one
+    // S + one P). Each S thread hands its finished tasks to P through
+    // sp_queues_[its_thread_idx].
+    int32_t p_thread_idx_{-1};
+    CompletedTaskQueue sp_queues_[MAX_AICPU_THREADS];
+
     // Cluster-ordered worker_id lists, populated by post_handshake_init().
     int32_t aic_worker_ids_[RUNTIME_MAX_WORKER]{};
     int32_t aiv_worker_ids_[RUNTIME_MAX_WORKER]{};