feat(runtime/pto2): predicated dispatch — nullify tasks to retire inline without dispatch

src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto2_dispatch_payload.h

@@ -66,8 +66,9 @@ static_assert(
 // PTO2TaskPayload layout drift fails the build rather than corrupting args[].
 constexpr uint32_t PTO2_TASKPAYLOAD_TENSOR_COUNT_OFFSET = 0;
 constexpr uint32_t PTO2_TASKPAYLOAD_SCALAR_COUNT_OFFSET = 4;
-constexpr uint32_t PTO2_TASKPAYLOAD_TENSORS_OFFSET = 576;
-constexpr uint32_t PTO2_TASKPAYLOAD_SCALARS_OFFSET = 4672;
+// Cache line 9 (byte 576) holds the AICPU-only DispatchPredicate; tensors follow it.
+constexpr uint32_t PTO2_TASKPAYLOAD_TENSORS_OFFSET = 640;
+constexpr uint32_t PTO2_TASKPAYLOAD_SCALARS_OFFSET = 4736;
 constexpr uint32_t PTO2_TASKPAYLOAD_TENSOR_STRIDE = 128;  // sizeof(Tensor)
 
 /**

src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp

@@ -952,6 +952,25 @@ static TaskOutputTensors submit_task_common(
 
     payload.init(args, result, prepared.alloc_result, layout);
     cur_slot_state.set_allow_early_resolve(args.allow_early_resolve());
+
+    // Dispatch predicate: resolve the (tensor, indices) to an absolute GM address
+    // now so the scheduler can read it at the dispatch point with a single load,
+    // no Arg/Tensor access. Both branches write predicate.op explicitly because
+    // payload slots are ring-reused; op == NONE means "always dispatch".
+    {
+        const L0TaskPredicate &pred = args.predicate();
+        if (pred.op != PredicateOp::NONE && pred.operand.tensor != nullptr && pred.operand.tensor->buffer.addr != 0) {
+            uint64_t elem_size = get_element_size(pred.operand.tensor->dtype);
+            uint64_t flat_offset = pred.operand.tensor->compute_flat_offset(pred.operand.indices, pred.operand.ndims);
+            payload.predicate.addr = pred.operand.tensor->buffer.addr + flat_offset * elem_size;
+            payload.predicate.target = pred.target;
+            payload.predicate.elem_size = static_cast<uint8_t>(elem_size);
+            payload.predicate.op = pred.op;
+        } else {
+            payload.predicate.addr = 0;
+            payload.predicate.op = PredicateOp::NONE;
+        }
+    }
 #if SIMPLER_DFX
     if (is_dump_args_enabled()) {
         if (args.scalar_count() > 0) {
@@ -1069,6 +1088,10 @@ TaskOutputTensors PTO2OrchestratorState::submit_task(const MixedKernels &mixed_k
         active_mask.set_sync_start();
     }
 
+    if (args.predicate().op != PredicateOp::NONE) {
+        active_mask.set_has_predicate();
+    }
+
     return submit_task_common(
         orch, args, active_mask, normalized.aic_kernel_id, normalized.aiv0_kernel_id, normalized.aiv1_kernel_id
     );

src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2_types.h

@@ -305,9 +305,15 @@ struct PTO2TaskPayload {
     // reinitialization. READY records that producer release observed OWNER;
     // only cancellation clears OWNER during the current task lifetime.
     std::atomic<uint8_t> early_sync_drain_state{PTO2_EARLY_SYNC_DRAIN_NONE};
-    // === Cache lines 9-72 (4096B) — tensors (alignas(64) forces alignment) ===
+    // === Cache line 9 (byte 576) — dispatch predicate (AICPU-only) ===
+    // Offset is a fixed 576, independent of MAX_TENSOR_ARGS / MAX_SCALAR_ARGS.
+    // AICore never reads it — args are materialized from the tensor_count / tensors
+    // / scalars offsets only. Resolved at submit; evaluated by the scheduler at
+    // dispatch.
+    alignas(64) DispatchPredicate predicate;
+    // === Cache lines 10-73 (4096B) — tensors (alignas(64) forces alignment) ===
     Tensor tensors[MAX_TENSOR_ARGS];
-    // === Cache lines 73-74 (128B) — scalars ===
+    // === Cache lines 74-75 (128B) — scalars ===
     uint64_t scalars[MAX_SCALAR_ARGS];
 
     // Layout verification (size checks that don't need offsetof).
@@ -399,14 +405,20 @@ static_assert(offsetof(PTO2TaskPayload, fanin_spill_pool) == 16, "spill pool poi
 static_assert(
     offsetof(PTO2TaskPayload, fanin_inline_slot_states) == 24, "inline fanin array must follow spill metadata"
 );
-static_assert(offsetof(PTO2TaskPayload, tensors) == 576, "tensors must start at byte 576 (cache line 9)");
 static_assert(
-    offsetof(PTO2TaskPayload, scalars) == 576 + MAX_TENSOR_ARGS * sizeof(Tensor),
+    offsetof(PTO2TaskPayload, predicate) == 576,
+    "dispatch predicate occupies cache line 9 at fixed byte 576 (before tensors, never moves)"
+);
+static_assert(
+    offsetof(PTO2TaskPayload, tensors) == 640, "tensors must start at byte 640 (cache line 10, after predicate)"
+);
+static_assert(
+    offsetof(PTO2TaskPayload, scalars) == 640 + MAX_TENSOR_ARGS * sizeof(Tensor),
     "scalars must immediately follow tensors"
 );
 static_assert(
-    sizeof(PTO2TaskPayload) == 576 + MAX_TENSOR_ARGS * sizeof(Tensor) + MAX_SCALAR_ARGS * sizeof(uint64_t),
-    "PTO2TaskPayload size must stay on the baseline cache-line footprint"
+    sizeof(PTO2TaskPayload) == 640 + MAX_TENSOR_ARGS * sizeof(Tensor) + MAX_SCALAR_ARGS * sizeof(uint64_t),
+    "PTO2TaskPayload size = metadata(576) + predicate cache line(64) + tensors + scalars"
 );
 
 /**

src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_submit_types.h

@@ -39,10 +39,60 @@ enum class PTO2SubtaskSlot : uint8_t {
 /**
  * Subtask mask bits (for ActiveMask)
  */
-inline constexpr uint8_t PTO2_SUBTASK_MASK_AIC = (1u << 0);         // 0x1
-inline constexpr uint8_t PTO2_SUBTASK_MASK_AIV0 = (1u << 1);        // 0x2
-inline constexpr uint8_t PTO2_SUBTASK_MASK_AIV1 = (1u << 2);        // 0x4
-inline constexpr uint8_t PTO2_SUBTASK_FLAG_SYNC_START = (1u << 3);  // 0x8: all blocks must launch atomically
+inline constexpr uint8_t PTO2_SUBTASK_MASK_AIC = (1u << 0);            // 0x1
+inline constexpr uint8_t PTO2_SUBTASK_MASK_AIV0 = (1u << 1);           // 0x2
+inline constexpr uint8_t PTO2_SUBTASK_MASK_AIV1 = (1u << 2);           // 0x4
+inline constexpr uint8_t PTO2_SUBTASK_FLAG_SYNC_START = (1u << 3);     // 0x8: all blocks must launch atomically
+inline constexpr uint8_t PTO2_SUBTASK_FLAG_HAS_PREDICATE = (1u << 4);  // 0x10: task carries a dispatch predicate
+
+// Dispatch-predicate comparison operator. The scheduler evaluates the predicate
+// at the dispatch point — the task is ready (fanin satisfied), so the predicate
+// address' producer has completed and the value read is current, without the
+// wait_for_tensor_ready() stall that get_tensor_data() pays in orchestration.
+// PASS => dispatch normally; FAIL => retire inline via the dep-only path.
+enum class PredicateOp : uint8_t { NONE = 0, EQ, NE, GT, LT, GE, LE };
+
+// Resolved dispatch predicate stored on a task's payload (AICPU-side only): the
+// absolute GM address of the predicate element + the comparison. op == NONE
+// means "no predicate — always dispatch". Populated at submit from an
+// L0TaskPredicate; evaluated by the scheduler at the dispatch point via pass().
+// Layout is 18 bytes (8-aligned).
+struct DispatchPredicate {
+    uint64_t addr{0};      // absolute GM address of the predicate element (0 when op == NONE)
+    int64_t target{0};     // value compared against
+    uint8_t elem_size{0};  // width of the predicate element in bytes (1/2/4/8)
+    PredicateOp op{PredicateOp::NONE};
+
+    // true => dispatch, false => retire inline. Reads elem_size bytes at addr and
+    // sign-extends to 64 bits before comparing to target. Safe to call only when
+    // the owning task is ready (its producer has written the value).
+    bool pass() const {
+        if (op == PredicateOp::NONE) return true;
+        int64_t v = 0;
+        __builtin_memcpy(&v, reinterpret_cast<const void *>(addr), elem_size);
+        uint32_t bits = static_cast<uint32_t>(elem_size) * 8u;
+        if (bits < 64u) {
+            int64_t shift = static_cast<int64_t>(64u - bits);
+            v = (v << shift) >> shift;
+        }
+        switch (op) {
+        case PredicateOp::EQ:
+            return v == target;
+        case PredicateOp::NE:
+            return v != target;
+        case PredicateOp::GT:
+            return v > target;
+        case PredicateOp::LT:
+            return v < target;
+        case PredicateOp::GE:
+            return v >= target;
+        case PredicateOp::LE:
+            return v <= target;
+        default:
+            return true;
+        }
+    }
+};
 
 /**
  * Resource shape — classifies a MixedKernels into one of 3 scheduling buckets.
@@ -86,6 +136,9 @@ class ActiveMask {
 
     bool requires_sync_start() const { return (raw_ & PTO2_SUBTASK_FLAG_SYNC_START) != 0; }
 
+    // True when the task carries a dispatch predicate.
+    bool has_predicate() const { return (raw_ & PTO2_SUBTASK_FLAG_HAS_PREDICATE) != 0; }
+
     PTO2ResourceShape to_shape() const {
         uint8_t cmask = core_mask();
         if (cmask == 0) return PTO2ResourceShape::DUMMY;
@@ -97,6 +150,8 @@ class ActiveMask {
 
     void set_sync_start() { raw_ |= PTO2_SUBTASK_FLAG_SYNC_START; }
 
+    void set_has_predicate() { raw_ |= PTO2_SUBTASK_FLAG_HAS_PREDICATE; }
+
     bool operator==(ActiveMask other) const { return raw_ == other.raw_; }
     bool operator!=(ActiveMask other) const { return raw_ != other.raw_; }
 

src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_types.h

@@ -198,6 +198,27 @@ class TensorRef {
  *   TaskOutputTensors outs = rt_submit_aic_task(kernel_id, args);
  *   const Tensor& y = outs.get_ref(0);
  */
+
+// Operand of a dispatch predicate (L0 layer): locates one element of a tensor —
+// tensor + ndims + indices, mirroring get_tensor_data. The tensor is borrowed and
+// must outlive submit; its buffer must be allocated by then, and its producer must
+// be a dependency of the predicated task so the value is current at dispatch.
+struct L0PredicateOperand {
+    const Tensor *tensor{nullptr};
+    uint32_t ndims{0};
+    uint32_t indices[MAX_TENSOR_DIMS]{};
+};
+
+// Dispatch predicate carried on an Arg: operand OP target (e.g. count[i] > 0).
+// op == NONE means "no predicate — always dispatch". Submit resolves the operand
+// into the payload's DispatchPredicate (an absolute GM address). Read in-process;
+// never crosses the wire.
+struct L0TaskPredicate {
+    L0PredicateOperand operand;
+    PredicateOp op{PredicateOp::NONE};
+    int64_t target{0};
+};
+
 template <size_t MaxT, size_t MaxS>
 struct Arg : TaskArgsTpl<TensorRef, uint64_t, MaxT, MaxS, TensorArgType> {
     using Base = TaskArgsTpl<TensorRef, uint64_t, MaxT, MaxS, TensorArgType>;
@@ -231,6 +252,16 @@ struct Arg : TaskArgsTpl<TensorRef, uint64_t, MaxT, MaxS, TensorArgType> {
     void set_allow_early_resolve(bool v = true) { allow_early_resolve_ = v; }
     bool allow_early_resolve() const { return allow_early_resolve_; }
 
+    // Dispatch predicate (codegen-author set; default op == NONE = always
+    // dispatch). A FALSE result at the dispatch point retires the task inline
+    // through the dep-only path — never dispatched to an AICore — while still
+    // resolving fanin/fanout so consumers unlock. The predicate tensor's producer
+    // MUST be a dependency of this task so the value is current when the task
+    // becomes ready. Read in-process; never crosses the wire.
+    L0TaskPredicate predicate_;
+    void set_predicate(const L0TaskPredicate &pred) { predicate_ = pred; }
+    const L0TaskPredicate &predicate() const { return predicate_; }
+
     void clear() {
         Base::clear();
 #if SIMPLER_DFX
@@ -239,6 +270,7 @@ struct Arg : TaskArgsTpl<TensorRef, uint64_t, MaxT, MaxS, TensorArgType> {
         explicit_deps_ = nullptr;
         explicit_dep_count_ = 0;
         allow_early_resolve_ = false;
+        predicate_ = L0TaskPredicate{};
     }
 
     void reset() {

src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h

@@ -526,12 +526,15 @@ struct PTO2SchedulerState {
     // Inline hot-path methods
     // =========================================================================
 
-    // Route a ready slot to the right global queue. Dummy tasks (empty active_mask)
-    // live in dummy_ready_queue; a ready sync_start cohort goes to the per-shape
-    // ready_sync_queues[] (drained as Tier-0); everything else to ready_queues[].
+    // Route a ready slot to the right global queue. Dep-only tasks — DUMMY-shaped
+    // (empty active_mask) or a task whose dispatch predicate fails — live in
+    // dummy_ready_queue and are retired inline; a ready sync_start cohort goes to
+    // the per-shape ready_sync_queues[] (drained as Tier-0); everything else to
+    // ready_queues[].
     void push_ready_routed(PTO2TaskSlotState *slot_state) {
         PTO2ResourceShape shape = slot_state->active_mask.to_shape();
-        if (shape == PTO2ResourceShape::DUMMY) {
+        if (shape == PTO2ResourceShape::DUMMY ||
+            (slot_state->active_mask.has_predicate() && !slot_state->payload->predicate.pass())) {
             dummy_ready_queue.push(slot_state);
         } else if (slot_state->active_mask.requires_sync_start()) {
             ready_sync_queues[static_cast<int32_t>(shape)].push(slot_state);
@@ -896,6 +899,7 @@ struct PTO2SchedulerState {
         p.unlock_fanout();
         for (; edge != nullptr; edge = edge->next) {
             PTO2TaskSlotState *c = edge->slot_state;
+            if (c->active_mask.has_predicate()) continue;  // predicated consumers never early-dispatch
             // Compare to fanin_actual_count (the real producer-edge count), NOT
             // fanin_count: fanin_count = fanin_actual_count + 1 (a self/wiring +1 that
             // ready_fanin gets but dispatch_fanin does not). dispatch_fanin starts at
@@ -944,6 +948,8 @@ struct PTO2SchedulerState {
     };
 
     inline bool try_early_dispatch_release(PTO2TaskSlotState &slot_state, EarlyDispatchReleaseSink *sink = nullptr) {
+        // A predicated task is evaluated at the ready point, never early-dispatched.
+        if (slot_state.active_mask.has_predicate()) return false;
         // Never staged => CAS NONE->DISPATCHED wins => dispatch normally.
         uint8_t expect = PTO2_EARLY_DISPATCH_NONE;
         if (slot_state.payload->early_dispatch_state.compare_exchange_strong(
@@ -1015,7 +1021,8 @@ struct PTO2SchedulerState {
         }
 
         PTO2ResourceShape shape = slot_state.active_mask.to_shape();
-        if (shape == PTO2ResourceShape::DUMMY) {
+        if (shape == PTO2ResourceShape::DUMMY ||
+            (slot_state.active_mask.has_predicate() && !slot_state.payload->predicate.pass())) {
             dummy_ready_queue.push(&slot_state);
         } else if (slot_state.active_mask.requires_sync_start()) {
             ready_sync_queues[static_cast<int32_t>(shape)].push(&slot_state);
@@ -1055,7 +1062,8 @@ struct PTO2SchedulerState {
         }
 
         PTO2ResourceShape shape = slot_state.active_mask.to_shape();
-        if (shape == PTO2ResourceShape::DUMMY) {
+        if (shape == PTO2ResourceShape::DUMMY ||
+            (slot_state.active_mask.has_predicate() && !slot_state.payload->predicate.pass())) {
             dummy_ready_queue.push(&slot_state, atomic_count, push_wait);
         } else if (slot_state.active_mask.requires_sync_start()) {
             ready_sync_queues[static_cast<int32_t>(shape)].push(&slot_state, atomic_count, push_wait);

src/a5/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp

@@ -935,6 +935,25 @@ static TaskOutputTensors submit_task_common(
     }
 
     payload.init(args, result, prepared.alloc_result, layout);
+
+    // Dispatch predicate: resolve the (tensor, indices) to an absolute GM address
+    // now so the scheduler can read it at the dispatch point with a single load,
+    // no Arg/Tensor access. Both branches write predicate.op explicitly because
+    // payload slots are ring-reused; op == NONE means "always dispatch".
+    {
+        const L0TaskPredicate &pred = args.predicate();
+        if (pred.op != PredicateOp::NONE && pred.operand.tensor != nullptr && pred.operand.tensor->buffer.addr != 0) {
+            uint64_t elem_size = get_element_size(pred.operand.tensor->dtype);
+            uint64_t flat_offset = pred.operand.tensor->compute_flat_offset(pred.operand.indices, pred.operand.ndims);
+            payload.predicate.addr = pred.operand.tensor->buffer.addr + flat_offset * elem_size;
+            payload.predicate.target = pred.target;
+            payload.predicate.elem_size = static_cast<uint8_t>(elem_size);
+            payload.predicate.op = pred.op;
+        } else {
+            payload.predicate.addr = 0;
+            payload.predicate.op = PredicateOp::NONE;
+        }
+    }
 #if SIMPLER_DFX
     if (is_dump_args_enabled()) {
         if (args.scalar_count() > 0) {
@@ -1049,6 +1068,10 @@ TaskOutputTensors PTO2OrchestratorState::submit_task(const MixedKernels &mixed_k
         active_mask.set_sync_start();
     }
 
+    if (args.predicate().op != PredicateOp::NONE) {
+        active_mask.set_has_predicate();
+    }
+
     return submit_task_common(
         orch, args, active_mask, normalized.aic_kernel_id, normalized.aiv0_kernel_id, normalized.aiv1_kernel_id
     );

src/a5/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2_types.h

@@ -220,9 +220,15 @@ struct PTO2TaskPayload {
     int32_t fanin_spill_start{0};   // Linear start index in fanin spill pool (0 = no spill)
     PTO2FaninPool *fanin_spill_pool{nullptr};
     PTO2TaskSlotState *fanin_inline_slot_states[PTO2_FANIN_INLINE_CAP];
-    // === Cache lines 9-72 (4096B) — tensors (alignas(64) forces alignment) ===
+    // === Cache line 9 (byte 576) — dispatch predicate (AICPU-only) ===
+    // Offset is a fixed 576, independent of MAX_TENSOR_ARGS / MAX_SCALAR_ARGS.
+    // AICore never reads it — args are materialized from the tensor_count / tensors
+    // / scalars offsets only. Resolved at submit; evaluated by the scheduler at
+    // dispatch.
+    alignas(64) DispatchPredicate predicate;
+    // === Cache lines 10-73 (4096B) — tensors (alignas(64) forces alignment) ===
     Tensor tensors[MAX_TENSOR_ARGS];
-    // === Cache lines 73-74 (128B) — scalars ===
+    // === Cache lines 74-75 (128B) — scalars ===
     uint64_t scalars[MAX_SCALAR_ARGS];
 
     // Layout verification (size checks that don't need offsetof).
@@ -270,14 +276,20 @@ static_assert(offsetof(PTO2TaskPayload, fanin_spill_pool) == 16, "spill pool poi
 static_assert(
     offsetof(PTO2TaskPayload, fanin_inline_slot_states) == 24, "inline fanin array must follow spill metadata"
 );
-static_assert(offsetof(PTO2TaskPayload, tensors) == 576, "tensors must start at byte 576 (cache line 9)");
 static_assert(
-    offsetof(PTO2TaskPayload, scalars) == 576 + MAX_TENSOR_ARGS * sizeof(Tensor),
+    offsetof(PTO2TaskPayload, predicate) == 576,
+    "dispatch predicate occupies cache line 9 at fixed byte 576 (before tensors, never moves)"
+);
+static_assert(
+    offsetof(PTO2TaskPayload, tensors) == 640, "tensors must start at byte 640 (cache line 10, after predicate)"
+);
+static_assert(
+    offsetof(PTO2TaskPayload, scalars) == 640 + MAX_TENSOR_ARGS * sizeof(Tensor),
     "scalars must immediately follow tensors"
 );
 static_assert(
-    sizeof(PTO2TaskPayload) == 576 + MAX_TENSOR_ARGS * sizeof(Tensor) + MAX_SCALAR_ARGS * sizeof(uint64_t),
-    "PTO2TaskPayload size must stay on the baseline cache-line footprint"
+    sizeof(PTO2TaskPayload) == 640 + MAX_TENSOR_ARGS * sizeof(Tensor) + MAX_SCALAR_ARGS * sizeof(uint64_t),
+    "PTO2TaskPayload size = metadata(576) + predicate cache line(64) + tensors + scalars"
 );
 
 /**