From dd4b554fb475f3935e7b83b006cd1c367938b4f4 Mon Sep 17 00:00:00 2001 From: TaoZQY Date: Wed, 1 Jul 2026 16:25:03 +0800 Subject: [PATCH 1/2] Add strategy2 pipeline configuration plumbing Add pipeline_strategy to CallConfig, worker/runtime C API plumbing, and runtime launch descriptors while preserving upstream worker cleanup behavior. Co-authored-by: Crane-Liu --- python/bindings/task_interface.cpp | 3 +- python/simpler/remote_l3_protocol.py | 1 + python/simpler/worker.py | 9 ++- .../host_build_graph/runtime/runtime.h | 3 + .../runtime/runtime.h | 9 +++ .../runtime/shared/runtime.cpp | 2 + .../host_build_graph/runtime/runtime.h | 3 + .../runtime/runtime.h | 9 +++ .../runtime/shared/runtime.cpp | 2 + src/common/hierarchical/remote_wire.cpp | 2 + .../platform/onboard/host/c_api_shared.cpp | 7 +- src/common/platform/sim/host/c_api_shared.cpp | 7 +- src/common/task_interface/call_config.h | 5 +- src/common/task_interface/pipeline_strategy.h | 79 +++++++++++++++++++ src/common/worker/chip_worker.cpp | 8 +- src/common/worker/chip_worker.h | 4 +- src/common/worker/pto_runtime_c_api.h | 6 +- tests/ut/cpp/types/test_call_config.cpp | 2 +- tests/ut/py/test_chip_worker.py | 3 + tests/ut/py/test_remote_l3_protocol.py | 3 +- tests/ut/py/test_task_interface.py | 3 +- 21 files changed, 146 insertions(+), 24 deletions(-) create mode 100644 src/common/task_interface/pipeline_strategy.h diff --git a/python/bindings/task_interface.cpp b/python/bindings/task_interface.cpp index 43c2e6e9c..0a63f1717 100644 --- a/python/bindings/task_interface.cpp +++ b/python/bindings/task_interface.cpp @@ -713,6 +713,7 @@ NB_MODULE(_task_interface, m) { .def(nb::init<>()) .def_rw("block_dim", &CallConfig::block_dim) .def_rw("aicpu_thread_num", &CallConfig::aicpu_thread_num) + .def_rw("pipeline_strategy", &CallConfig::pipeline_strategy) // runtime_env returns an internal reference so `cfg.runtime_env.ring_heap = X` // writes through to the owning CallConfig (rv_policy::reference_internal). .def_prop_rw( @@ -799,7 +800,7 @@ NB_MODULE(_task_interface, m) { .def("__repr__", [append_ring_values](const CallConfig &self) -> std::string { std::ostringstream os; os << "CallConfig(block_dim=" << self.block_dim << ", aicpu_thread_num=" << self.aicpu_thread_num - << ", enable_l2_swimlane=" << self.enable_l2_swimlane + << ", pipeline_strategy=" << self.pipeline_strategy << ", enable_l2_swimlane=" << self.enable_l2_swimlane << ", enable_dump_tensor=" << self.enable_dump_tensor << ", enable_pmu=" << self.enable_pmu << ", enable_dep_gen=" << (self.enable_dep_gen ? "True" : "False") << ", enable_scope_stats=" << (self.enable_scope_stats ? "True" : "False"); diff --git a/python/simpler/remote_l3_protocol.py b/python/simpler/remote_l3_protocol.py index 661051f64..b484b23fe 100644 --- a/python/simpler/remote_l3_protocol.py +++ b/python/simpler/remote_l3_protocol.py @@ -417,6 +417,7 @@ def decode_call_config(reader: _Reader) -> CallConfig: cfg = CallConfig() cfg.block_dim = reader.i32() cfg.aicpu_thread_num = reader.i32() + cfg.pipeline_strategy = reader.i32() cfg.enable_l2_swimlane = reader.i32() cfg.enable_dump_tensor = reader.i32() cfg.enable_pmu = reader.i32() diff --git a/python/simpler/worker.py b/python/simpler/worker.py index a8c56ae82..e16bab015 100644 --- a/python/simpler/worker.py +++ b/python/simpler/worker.py @@ -140,13 +140,14 @@ def my_l4_orch(orch, args, config): _OFF_CALLABLE = 8 _OFF_CONFIG = 16 # Packed CallConfig wire layout — must match call_config.h byte for byte: -# 7 int32 (block_dim, aicpu_thread_num, enable_l2_swimlane, enable_dump_tensor, -# enable_pmu, enable_dep_gen, enable_scope_stats) + uint64 ring sizing +# 8 int32 (block_dim, aicpu_thread_num, pipeline_strategy, +# enable_l2_swimlane, enable_dump_tensor, enable_pmu, enable_dep_gen, +# enable_scope_stats) + uint64 ring sizing # overrides (3 per-ring arrays of RUNTIME_ENV_RING_COUNT: ring_task_window, # ring_heap, ring_dep_pool) + 1024-byte NUL-terminated output_prefix. Log config # travels separately via ChipWorker.init(log_level, log_info_v) — not on per-task wire. _RUNTIME_ENV_UINT64_FIELD_COUNT = 3 * RUNTIME_ENV_RING_COUNT -_CFG_FMT = struct.Struct("=iiiiiii" + ("Q" * _RUNTIME_ENV_UINT64_FIELD_COUNT) + "1024s") +_CFG_FMT = struct.Struct("=iiiiiiii" + ("Q" * _RUNTIME_ENV_UINT64_FIELD_COUNT) + "1024s") # Args region starts after CONFIG, rounded up to 8 bytes so the first # Tensor.data (uint64_t at OFF_ARGS+8) is 8-byte aligned, avoiding # SIGBUS on strict-alignment platforms (aarch64 atomics, some ARM cores). @@ -1208,6 +1209,7 @@ def _read_config_from_mailbox(buf: memoryview) -> CallConfig: ( block_dim, aicpu_tn, + pipeline_strategy, swl, dt, pmu, @@ -1222,6 +1224,7 @@ def _read_config_from_mailbox(buf: memoryview) -> CallConfig: cfg = CallConfig() cfg.block_dim = block_dim cfg.aicpu_thread_num = aicpu_tn + cfg.pipeline_strategy = pipeline_strategy cfg.enable_l2_swimlane = swl cfg.enable_dump_tensor = int(dt) cfg.enable_pmu = pmu diff --git a/src/a2a3/runtime/host_build_graph/runtime/runtime.h b/src/a2a3/runtime/host_build_graph/runtime/runtime.h index 9f787c8c1..0dc30ab86 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/runtime.h +++ b/src/a2a3/runtime/host_build_graph/runtime/runtime.h @@ -239,6 +239,9 @@ class Runtime { void set_worker_count(int n) { worker_count = n; } int get_aicpu_thread_num() const { return aicpu_thread_num; } void set_aicpu_thread_num(int n) { aicpu_thread_num = n; } + int get_pipeline_strategy() const { return -1; } + void set_pipeline_strategy(int) {} + bool pipeline_defer_submit_disabled() const { return false; } Handshake *get_workers() { return workers; } int32_t get_aicpu_allowed_cpu_count() const { return aicpu_allowed_cpu_count; } void set_aicpu_allowed_cpu_count(int32_t n) { aicpu_allowed_cpu_count = n; } diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/runtime.h b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/runtime.h index 8a41434de..f153c6ec1 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/runtime.h +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/runtime.h @@ -43,6 +43,7 @@ #include "common/l2_swimlane_profiling.h" #include "common/platform_config.h" #include "aicpu/platform_aicpu_affinity.h" // MAX_GATE_THREADS (aicpu_allowed_cpus bound) +#include "pipeline_strategy.h" #include "pto2_dispatch_payload.h" #include "task_args.h" @@ -162,6 +163,8 @@ struct alignas(64) DeviceRuntimeLaunchDesc { // aicpu_thread_num-1 scheduler threads that dispatch tasks to AICore. int aicpu_thread_num; int ready_queue_shards; // Number of ready queue shards (1..MAX_AICPU_THREADS, default MAX-1) + int pipeline_strategy; // -1 = unset baseline, 2 = strategy2 layout + bool pipeline_defer_submit_disabled; // Filter-style affinity gate input (a2a3 onboard). Host fills these // before launch from AICPU OCCUPY, and the device gate keeps threads whose @@ -243,6 +246,12 @@ class Runtime { void set_worker_count(int n) { dev.worker_count = n; } int get_aicpu_thread_num() const { return dev.aicpu_thread_num; } void set_aicpu_thread_num(int n) { dev.aicpu_thread_num = n; } + int get_pipeline_strategy() const { return dev.pipeline_strategy; } + void set_pipeline_strategy(int strategy) { + dev.pipeline_strategy = resolve_pipeline_strategy_with_env(strategy); + dev.pipeline_defer_submit_disabled = resolve_pipeline_defer_submit_disabled_with_env(); + } + bool pipeline_defer_submit_disabled() const { return dev.pipeline_defer_submit_disabled; } Handshake *get_workers() { return dev.workers; } int32_t get_aicpu_allowed_cpu_count() const { return dev.aicpu_allowed_cpu_count; } void set_aicpu_allowed_cpu_count(int32_t n) { dev.aicpu_allowed_cpu_count = n; } diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/shared/runtime.cpp b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/shared/runtime.cpp index 08f86f814..7b2d2a327 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/shared/runtime.cpp +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/shared/runtime.cpp @@ -34,6 +34,8 @@ Runtime::Runtime() { dev.worker_count = 0; dev.aicpu_thread_num = 1; dev.ready_queue_shards = RUNTIME_DEFAULT_READY_QUEUE_SHARDS; + dev.pipeline_strategy = PIPELINE_STRATEGY_UNSET_BASELINE; + dev.pipeline_defer_submit_disabled = false; memset(dev.aicpu_allowed_cpus, 0, sizeof(dev.aicpu_allowed_cpus)); dev.aicpu_allowed_cpu_count = 0; dev.aicpu_launch_count = 0; diff --git a/src/a5/runtime/host_build_graph/runtime/runtime.h b/src/a5/runtime/host_build_graph/runtime/runtime.h index b943ee2f8..29cfbb82b 100644 --- a/src/a5/runtime/host_build_graph/runtime/runtime.h +++ b/src/a5/runtime/host_build_graph/runtime/runtime.h @@ -258,6 +258,9 @@ class Runtime { void set_worker_count(int n) { worker_count = n; } int get_aicpu_thread_num() const { return aicpu_thread_num; } void set_aicpu_thread_num(int n) { aicpu_thread_num = n; } + int get_pipeline_strategy() const { return -1; } + void set_pipeline_strategy(int) {} + bool pipeline_defer_submit_disabled() const { return false; } Handshake *get_workers() { return workers; } int32_t get_aicpu_allowed_cpu_count() const { return aicpu_allowed_cpu_count; } void set_aicpu_allowed_cpu_count(int32_t n) { aicpu_allowed_cpu_count = n; } diff --git a/src/a5/runtime/tensormap_and_ringbuffer/runtime/runtime.h b/src/a5/runtime/tensormap_and_ringbuffer/runtime/runtime.h index 694c3e0f4..2a1dad447 100644 --- a/src/a5/runtime/tensormap_and_ringbuffer/runtime/runtime.h +++ b/src/a5/runtime/tensormap_and_ringbuffer/runtime/runtime.h @@ -42,6 +42,7 @@ #include "common/host_api.h" #include "common/platform_config.h" #include "aicpu/platform_aicpu_affinity.h" // MAX_GATE_THREADS (aicpu_allowed_cpus bound) +#include "pipeline_strategy.h" #include "pto2_dispatch_payload.h" #include "task_args.h" @@ -170,6 +171,8 @@ struct alignas(64) DeviceRuntimeLaunchDesc { // aicpu_thread_num-1 scheduler threads that dispatch tasks to AICore. int aicpu_thread_num; int ready_queue_shards; // Number of ready queue shards (1..MAX_AICPU_THREADS, default MAX-1) + int pipeline_strategy; // -1 = unset baseline, 2 = strategy2 layout + bool pipeline_defer_submit_disabled; // Filter-style affinity gate input (a5 onboard). Host fills before // launch from device-side OCCUPY + DSMI CPU_TOPO via @@ -257,6 +260,12 @@ class Runtime { void set_worker_count(int n) { dev.worker_count = n; } int get_aicpu_thread_num() const { return dev.aicpu_thread_num; } void set_aicpu_thread_num(int n) { dev.aicpu_thread_num = n; } + int get_pipeline_strategy() const { return dev.pipeline_strategy; } + void set_pipeline_strategy(int strategy) { + dev.pipeline_strategy = resolve_pipeline_strategy_with_env(strategy); + dev.pipeline_defer_submit_disabled = resolve_pipeline_defer_submit_disabled_with_env(); + } + bool pipeline_defer_submit_disabled() const { return dev.pipeline_defer_submit_disabled; } Handshake *get_workers() { return dev.workers; } int32_t get_aicpu_allowed_cpu_count() const { return dev.aicpu_allowed_cpu_count; } void set_aicpu_allowed_cpu_count(int32_t n) { dev.aicpu_allowed_cpu_count = n; } diff --git a/src/a5/runtime/tensormap_and_ringbuffer/runtime/shared/runtime.cpp b/src/a5/runtime/tensormap_and_ringbuffer/runtime/shared/runtime.cpp index 08f86f814..7b2d2a327 100644 --- a/src/a5/runtime/tensormap_and_ringbuffer/runtime/shared/runtime.cpp +++ b/src/a5/runtime/tensormap_and_ringbuffer/runtime/shared/runtime.cpp @@ -34,6 +34,8 @@ Runtime::Runtime() { dev.worker_count = 0; dev.aicpu_thread_num = 1; dev.ready_queue_shards = RUNTIME_DEFAULT_READY_QUEUE_SHARDS; + dev.pipeline_strategy = PIPELINE_STRATEGY_UNSET_BASELINE; + dev.pipeline_defer_submit_disabled = false; memset(dev.aicpu_allowed_cpus, 0, sizeof(dev.aicpu_allowed_cpus)); dev.aicpu_allowed_cpu_count = 0; dev.aicpu_launch_count = 0; diff --git a/src/common/hierarchical/remote_wire.cpp b/src/common/hierarchical/remote_wire.cpp index b0cec3181..7e892de4f 100644 --- a/src/common/hierarchical/remote_wire.cpp +++ b/src/common/hierarchical/remote_wire.cpp @@ -310,6 +310,7 @@ std::vector encode_call_config(const CallConfig &config) { std::vector out; put_i32(out, config.block_dim); put_i32(out, config.aicpu_thread_num); + put_i32(out, config.pipeline_strategy); put_i32(out, config.enable_l2_swimlane); put_i32(out, config.enable_dump_tensor); put_i32(out, config.enable_pmu); @@ -323,6 +324,7 @@ CallConfig decode_call_config(const uint8_t *data, size_t size, size_t &offset) CallConfig config{}; config.block_dim = get_i32(data, size, offset); config.aicpu_thread_num = get_i32(data, size, offset); + config.pipeline_strategy = get_i32(data, size, offset); config.enable_l2_swimlane = get_i32(data, size, offset); config.enable_dump_tensor = get_i32(data, size, offset); config.enable_pmu = get_i32(data, size, offset); diff --git a/src/common/platform/onboard/host/c_api_shared.cpp b/src/common/platform/onboard/host/c_api_shared.cpp index 5924c4184..96a3c3d6e 100644 --- a/src/common/platform/onboard/host/c_api_shared.cpp +++ b/src/common/platform/onboard/host/c_api_shared.cpp @@ -461,9 +461,9 @@ static void emit_device_phase_markers(DeviceRunnerBase *runner) { int simpler_run( DeviceContextHandle ctx, RuntimeHandle runtime, int32_t callable_id, const void *args, int block_dim, - int aicpu_thread_num, int enable_l2_swimlane, int enable_dump_tensor, int enable_pmu, int enable_dep_gen, - int enable_scope_stats, const uint64_t *ring_task_window, const uint64_t *ring_heap, const uint64_t *ring_dep_pool, - const char *output_prefix + int aicpu_thread_num, int pipeline_strategy, int enable_l2_swimlane, int enable_dump_tensor, int enable_pmu, + int enable_dep_gen, int enable_scope_stats, const uint64_t *ring_task_window, const uint64_t *ring_heap, + const uint64_t *ring_dep_pool, const char *output_prefix ) { if (ctx == NULL || runtime == NULL) return -1; DeviceRunnerBase *runner = static_cast(ctx); @@ -488,6 +488,7 @@ int simpler_run( if (rc != 0) return rc; Runtime *r = new (runtime) Runtime(); + r->set_pipeline_strategy(pipeline_strategy); // RAII the placement-new'd Runtime so its dtor fires on every exit // (normal returns, the rc-check early-returns below, AND the catch(...) // path). The prior manual `r->~Runtime()` on each return leaked the diff --git a/src/common/platform/sim/host/c_api_shared.cpp b/src/common/platform/sim/host/c_api_shared.cpp index c43b06f8f..8bd7974c3 100644 --- a/src/common/platform/sim/host/c_api_shared.cpp +++ b/src/common/platform/sim/host/c_api_shared.cpp @@ -418,9 +418,9 @@ static void emit_device_phase_markers(SimDeviceRunnerBase *runner) { int simpler_run( DeviceContextHandle ctx, RuntimeHandle runtime, int32_t callable_id, const void *args, int block_dim, - int aicpu_thread_num, int enable_l2_swimlane, int enable_dump_tensor, int enable_pmu, int enable_dep_gen, - int enable_scope_stats, const uint64_t *ring_task_window, const uint64_t *ring_heap, const uint64_t *ring_dep_pool, - const char *output_prefix + int aicpu_thread_num, int pipeline_strategy, int enable_l2_swimlane, int enable_dump_tensor, int enable_pmu, + int enable_dep_gen, int enable_scope_stats, const uint64_t *ring_task_window, const uint64_t *ring_heap, + const uint64_t *ring_dep_pool, const char *output_prefix ) { if (ctx == NULL || runtime == NULL) return -1; SimDeviceRunnerBase *runner = static_cast(ctx); @@ -439,6 +439,7 @@ int simpler_run( try { Runtime *r = new (runtime) Runtime(); + r->set_pipeline_strategy(pipeline_strategy); // RAII the placement-new'd Runtime so its dtor fires on every exit // (normal returns, the rc-check early-returns below, AND the catch(...) // path). Mirrors the onboard c_api_shared fix from PR #928. diff --git a/src/common/task_interface/call_config.h b/src/common/task_interface/call_config.h index 7845c9632..3618c90c4 100644 --- a/src/common/task_interface/call_config.h +++ b/src/common/task_interface/call_config.h @@ -11,7 +11,7 @@ /** * CallConfig — per-NEXT_LEVEL-task config. Carries execution knobs - * (block_dim, aicpu_thread_num), per-task runtime-environment overrides + * (block_dim, aicpu_thread_num, pipeline_strategy), per-task runtime-environment overrides * (`runtime_env.ring_task_window` / `.ring_heap` / `.ring_dep_pool`, each a per-ring array) plus the five parallel * diagnostics sub-features under the profiling umbrella: `enable_l2_swimlane` (swimlane), `enable_dump_tensor`, * `enable_pmu`, `enable_dep_gen`, and `enable_scope_stats`. All five require `output_prefix` because they each write @@ -110,6 +110,7 @@ struct RuntimeEnv { struct CallConfig { int32_t block_dim = 0; // 0 = auto; resolved by DeviceRunner at run() time int32_t aicpu_thread_num = 3; + int32_t pipeline_strategy = -1; // -1 = env/default baseline; 2 = strategy2 layout int32_t enable_l2_swimlane = 0; int32_t enable_dump_tensor = 0; int32_t enable_pmu = 0; // 0 = disabled; >0 = enabled, value selects event type @@ -143,6 +144,6 @@ struct CallConfig { #pragma pack(pop) static_assert(sizeof(RuntimeEnv) == RUNTIME_ENV_UINT64_FIELD_COUNT * sizeof(uint64_t), "RuntimeEnv wire layout drift"); static_assert( - sizeof(CallConfig) == 7 * sizeof(int32_t) + RUNTIME_ENV_UINT64_FIELD_COUNT * sizeof(uint64_t) + 1024, + sizeof(CallConfig) == 8 * sizeof(int32_t) + RUNTIME_ENV_UINT64_FIELD_COUNT * sizeof(uint64_t) + 1024, "CallConfig wire layout drift" ); diff --git a/src/common/task_interface/pipeline_strategy.h b/src/common/task_interface/pipeline_strategy.h new file mode 100644 index 000000000..92242e4ad --- /dev/null +++ b/src/common/task_interface/pipeline_strategy.h @@ -0,0 +1,79 @@ +/* + * Copyright (c) PyPTO Contributors. + * This program is free software, you can redistribute it and/or modify it under the terms and conditions of + * CANN Open Software License Agreement Version 2.0 (the "License"). + * Please refer to the License for details. You may not use this file except in compliance with the License. + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, + * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. + * See LICENSE in the root of the software repository for the full text of the License. + * ----------------------------------------------------------------------------------------------------------- + */ +#ifndef SRC_COMMON_TASK_INTERFACE_PIPELINE_STRATEGY_H_ +#define SRC_COMMON_TASK_INTERFACE_PIPELINE_STRATEGY_H_ + +#include +#include +#include + +enum class PipelineStrategy : int32_t { + BASELINE = 0, + S2_O2_SPLIT_CTRL_STRATEGY2 = 2, +}; + +constexpr int32_t PIPELINE_LAYOUT_MAX_THREADS = 6; +constexpr int32_t PIPELINE_STRATEGY_UNSET_BASELINE = -1; + +struct PipelineLayout { + PipelineStrategy strategy{PipelineStrategy::BASELINE}; + int32_t scheduler_threads{0}; + int32_t orchestrator_threads{1}; + int32_t scheduler_index_by_thread[PIPELINE_LAYOUT_MAX_THREADS]{-1, -1, -1, -1, -1, -1}; + int32_t orchestrator_stage_by_thread[PIPELINE_LAYOUT_MAX_THREADS]{-1, -1, -1, -1, -1, -1}; + const char *name{"baseline"}; + const char *cluster_layout{"baseline"}; +}; + +static inline PipelineLayout resolve_pipeline_layout(int32_t raw_strategy) { + switch (raw_strategy) { + case 2: + return { + PipelineStrategy::S2_O2_SPLIT_CTRL_STRATEGY2, + 2, + 2, + {0, 1, -1, -1, -1, -1}, + {-1, -1, 0, 1, -1, -1}, + "2S2O_split_ctrl_strategy2", + "cluster0=S0,S1;cluster1=O0,O1", + }; + default: + return { + PipelineStrategy::BASELINE, 0, 1, {-1, -1, -1, -1, -1, -1}, + {-1, -1, -1, -1, -1, -1}, "baseline", "baseline", + }; + } +} + +static inline int32_t resolve_pipeline_strategy_with_env(int32_t config_strategy) { + const char *env = std::getenv("SIMPLER_PIPELINE_STRATEGY"); + if (env == nullptr || env[0] == '\0') { + return config_strategy; + } + char *end = nullptr; + long parsed = std::strtol(env, &end, 10); + if (end == env) { + return config_strategy; + } + return static_cast(parsed); +} + +static inline bool resolve_pipeline_defer_submit_disabled_with_env() { + const char *env = std::getenv("SIMPLER_PIPELINE_DEFER_SUBMIT"); + if (env == nullptr || env[0] == '\0') { + return false; + } + return std::strcmp(env, "0") == 0 || std::strcmp(env, "false") == 0 || std::strcmp(env, "FALSE") == 0 || + std::strcmp(env, "off") == 0 || std::strcmp(env, "OFF") == 0 || std::strcmp(env, "no") == 0 || + std::strcmp(env, "NO") == 0; +} + +#endif // SRC_COMMON_TASK_INTERFACE_PIPELINE_STRATEGY_H_ diff --git a/src/common/worker/chip_worker.cpp b/src/common/worker/chip_worker.cpp index 5a51d5a48..de5727459 100644 --- a/src/common/worker/chip_worker.cpp +++ b/src/common/worker/chip_worker.cpp @@ -334,10 +334,10 @@ void ChipWorker::run(int32_t callable_id, const ChipStorageTaskArgs *args, const // Per-stage timing is emitted by the platform as `[STRACE]` log markers, not // returned (see chip_worker.h::run). int rc = run_fn_( - device_ctx_, rt, callable_id, args, config.block_dim, config.aicpu_thread_num, config.enable_l2_swimlane, - config.enable_dump_tensor, config.enable_pmu, config.enable_dep_gen, config.enable_scope_stats, - config.runtime_env.ring_task_window, config.runtime_env.ring_heap, config.runtime_env.ring_dep_pool, - config.output_prefix + device_ctx_, rt, callable_id, args, config.block_dim, config.aicpu_thread_num, config.pipeline_strategy, + config.enable_l2_swimlane, config.enable_dump_tensor, config.enable_pmu, config.enable_dep_gen, + config.enable_scope_stats, config.runtime_env.ring_task_window, config.runtime_env.ring_heap, + config.runtime_env.ring_dep_pool, config.output_prefix ); if (rc != 0) { throw std::runtime_error("run failed with code " + std::to_string(rc)); diff --git a/src/common/worker/chip_worker.h b/src/common/worker/chip_worker.h index 086e5b6b6..03df2e24f 100644 --- a/src/common/worker/chip_worker.h +++ b/src/common/worker/chip_worker.h @@ -146,8 +146,8 @@ class ChipWorker { int (*)(void *, int, const uint8_t *, size_t, const uint8_t *, size_t, const uint8_t *, size_t); using SimplerRegisterCallableFn = int (*)(void *, int32_t, const void *); using SimplerRunFn = int (*)( - void *, void *, int32_t, const void *, int, int, int, int, int, int, int, const uint64_t *, const uint64_t *, - const uint64_t *, const char * + void *, void *, int32_t, const void *, int, int, int, int, int, int, int, int, const uint64_t *, + const uint64_t *, const uint64_t *, const char * ); using SimplerUnregisterCallableFn = int (*)(void *, int32_t); using GetAicpuDlopenCountFn = size_t (*)(void *); diff --git a/src/common/worker/pto_runtime_c_api.h b/src/common/worker/pto_runtime_c_api.h index 380827084..664236f82 100644 --- a/src/common/worker/pto_runtime_c_api.h +++ b/src/common/worker/pto_runtime_c_api.h @@ -197,9 +197,9 @@ int simpler_register_callable(DeviceContextHandle ctx, int32_t callable_id, cons */ int simpler_run( DeviceContextHandle ctx, RuntimeHandle runtime, int32_t callable_id, const void *args, int block_dim, - int aicpu_thread_num, int enable_l2_swimlane, int enable_dump_tensor, int enable_pmu, int enable_dep_gen, - int enable_scope_stats, const uint64_t *ring_task_window, const uint64_t *ring_heap, const uint64_t *ring_dep_pool, - const char *output_prefix + int aicpu_thread_num, int pipeline_strategy, int enable_l2_swimlane, int enable_dump_tensor, int enable_pmu, + int enable_dep_gen, int enable_scope_stats, const uint64_t *ring_task_window, const uint64_t *ring_heap, + const uint64_t *ring_dep_pool, const char *output_prefix ); /** diff --git a/tests/ut/cpp/types/test_call_config.cpp b/tests/ut/cpp/types/test_call_config.cpp index 4a9d5f63e..a9d733c35 100644 --- a/tests/ut/cpp/types/test_call_config.cpp +++ b/tests/ut/cpp/types/test_call_config.cpp @@ -19,7 +19,7 @@ // Wire contract: parent and forked child move CallConfig with one memcpy. TEST(CallConfig, WireLayoutMatchesConstant) { EXPECT_EQ(sizeof(RuntimeEnv), RUNTIME_ENV_UINT64_FIELD_COUNT * sizeof(uint64_t)); - EXPECT_EQ(sizeof(CallConfig), 7 * sizeof(int32_t) + RUNTIME_ENV_UINT64_FIELD_COUNT * sizeof(uint64_t) + 1024); + EXPECT_EQ(sizeof(CallConfig), 8 * sizeof(int32_t) + RUNTIME_ENV_UINT64_FIELD_COUNT * sizeof(uint64_t) + 1024); } TEST(CallConfig, RuntimeEnvDefaultsAreUnset) { diff --git a/tests/ut/py/test_chip_worker.py b/tests/ut/py/test_chip_worker.py index fe6efe4e5..904dd0fa4 100644 --- a/tests/ut/py/test_chip_worker.py +++ b/tests/ut/py/test_chip_worker.py @@ -333,6 +333,7 @@ def test_config_roundtrip(self): cfg = CallConfig() cfg.block_dim = 7 cfg.aicpu_thread_num = 2 + cfg.pipeline_strategy = 2 cfg.enable_l2_swimlane = 3 cfg.enable_dump_tensor = 2 cfg.enable_pmu = 5 @@ -349,6 +350,7 @@ def test_config_roundtrip(self): _OFF_CONFIG, cfg.block_dim, cfg.aicpu_thread_num, + cfg.pipeline_strategy, cfg.enable_l2_swimlane, int(cfg.enable_dump_tensor), cfg.enable_pmu, @@ -363,6 +365,7 @@ def test_config_roundtrip(self): decoded = _read_config_from_mailbox(memoryview(buf)) assert decoded.block_dim == 7 assert decoded.aicpu_thread_num == 2 + assert decoded.pipeline_strategy == 2 assert decoded.enable_l2_swimlane == 3 assert decoded.enable_dump_tensor == 2 assert decoded.enable_pmu == 5 diff --git a/tests/ut/py/test_remote_l3_protocol.py b/tests/ut/py/test_remote_l3_protocol.py index d3853da9d..034d88b7b 100644 --- a/tests/ut/py/test_remote_l3_protocol.py +++ b/tests/ut/py/test_remote_l3_protocol.py @@ -37,7 +37,7 @@ def _oversized_multibyte_error_message(): def test_task_payload_decode_preserves_scope_stats_config(): prefix = b"/tmp/remote-scope" - config = struct.pack(" Date: Wed, 1 Jul 2026 16:25:03 +0800 Subject: [PATCH 2/2] Add strategy2 a2a3 submit pipeline runtime Implement the 2S2O strategy2 submit/dependency pipeline in the a2a3 tensormap runtime and scheduler. Co-authored-by: Crane-Liu --- .../aicpu/aicpu_executor.cpp | 168 +++++- .../runtime/pto_dep_compute.h | 170 ++++-- .../runtime/pto_orchestrator.cpp | 493 +++++++++++++++++- .../runtime/pto_orchestrator.h | 88 ++++ .../runtime/scheduler/pto_scheduler.h | 22 + .../runtime/shared/pto_runtime2_init.cpp | 3 +- 6 files changed, 889 insertions(+), 55 deletions(-) diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/aicpu/aicpu_executor.cpp b/src/a2a3/runtime/tensormap_and_ringbuffer/aicpu/aicpu_executor.cpp index 152f82fc4..7d4131691 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/aicpu/aicpu_executor.cpp +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/aicpu/aicpu_executor.cpp @@ -55,6 +55,7 @@ // CoreCallable for resolved dispatch address #include "callable.h" +#include "pipeline_strategy.h" // Scheduler data structures (CoreExecState, CoreTracker, etc.) #include "scheduler/scheduler_types.h" @@ -115,6 +116,11 @@ struct OrchSoEntry { struct AicpuExecutor { int32_t sched_thread_num_; + int32_t orch_thread_num_; + int32_t primary_orch_thread_idx_; + PipelineLayout pipeline_layout_{}; + int32_t scheduler_index_by_thread_[MAX_AICPU_THREADS]; + int32_t orchestrator_stage_by_thread_[MAX_AICPU_THREADS]; bool serial_orch_sched_{false}; // ===== Thread management state ===== @@ -160,6 +166,10 @@ struct AicpuExecutor { ); int32_t run(Runtime *runtime); void deinit(Runtime *runtime); + void resolve_thread_layout(Runtime *runtime); + int32_t scheduler_idx_for_thread(int32_t thread_idx) const; + int32_t orch_stage_idx_for_thread(int32_t thread_idx) const; + int32_t scheduler_context_idx_for_thread(int32_t thread_idx) const; ~AicpuExecutor() { // Process-wide teardown (the single static instance dies here). Every @@ -186,6 +196,95 @@ static_assert( // ===== AicpuExecutor Method Implementations ===== +void AicpuExecutor::resolve_thread_layout(Runtime *runtime) { + aicpu_thread_num_ = runtime->dev.aicpu_thread_num; + if (aicpu_thread_num_ == 0) aicpu_thread_num_ = 1; + + for (int32_t i = 0; i < MAX_AICPU_THREADS; ++i) { + scheduler_index_by_thread_[i] = -1; + orchestrator_stage_by_thread_[i] = -1; + } + + const int32_t raw_strategy = runtime->dev.pipeline_strategy; + bool use_baseline = raw_strategy < 0; + + if (!use_baseline) { + if (raw_strategy != static_cast(PipelineStrategy::S2_O2_SPLIT_CTRL_STRATEGY2)) { + LOG_WARN("Pipeline strategy %d is not enabled; falling back to baseline layout", raw_strategy); + use_baseline = true; + } else { + pipeline_layout_ = resolve_pipeline_layout(raw_strategy); + int32_t requested_total = pipeline_layout_.scheduler_threads + pipeline_layout_.orchestrator_threads; + if (aicpu_thread_num_ < requested_total) { + LOG_WARN( + "Pipeline strategy %s needs %d AICPU threads, got %d; falling back to baseline layout", + pipeline_layout_.name, requested_total, aicpu_thread_num_ + ); + use_baseline = true; + } + } + } + + if (use_baseline) { + pipeline_layout_ = resolve_pipeline_layout(PIPELINE_STRATEGY_UNSET_BASELINE); + orch_thread_num_ = 1; + sched_thread_num_ = aicpu_thread_num_ - orch_thread_num_; + for (int32_t i = 0; i < sched_thread_num_ && i < MAX_AICPU_THREADS; ++i) { + scheduler_index_by_thread_[i] = i; + } + if (sched_thread_num_ >= 0 && sched_thread_num_ < MAX_AICPU_THREADS) { + orchestrator_stage_by_thread_[sched_thread_num_] = 0; + } + runtime->dev.pipeline_strategy = PIPELINE_STRATEGY_UNSET_BASELINE; + } else { + sched_thread_num_ = pipeline_layout_.scheduler_threads; + orch_thread_num_ = pipeline_layout_.orchestrator_threads; + for (int32_t i = 0; i < aicpu_thread_num_ && i < PIPELINE_LAYOUT_MAX_THREADS; ++i) { + scheduler_index_by_thread_[i] = pipeline_layout_.scheduler_index_by_thread[i]; + orchestrator_stage_by_thread_[i] = pipeline_layout_.orchestrator_stage_by_thread[i]; + } + runtime->dev.pipeline_strategy = raw_strategy; + } + + if (sched_thread_num_ < 0) sched_thread_num_ = 0; + primary_orch_thread_idx_ = -1; + for (int32_t i = 0; i < aicpu_thread_num_ && i < MAX_AICPU_THREADS; ++i) { + if (orchestrator_stage_by_thread_[i] == 0) { + primary_orch_thread_idx_ = i; + break; + } + } + if (primary_orch_thread_idx_ < 0) { + primary_orch_thread_idx_ = sched_thread_num_; + } +} + +int32_t AicpuExecutor::scheduler_idx_for_thread(int32_t thread_idx) const { + if (thread_idx < 0 || thread_idx >= MAX_AICPU_THREADS) { + return -1; + } + return scheduler_index_by_thread_[thread_idx]; +} + +int32_t AicpuExecutor::orch_stage_idx_for_thread(int32_t thread_idx) const { + if (thread_idx < 0 || thread_idx >= MAX_AICPU_THREADS) { + return -1; + } + return orchestrator_stage_by_thread_[thread_idx]; +} + +int32_t AicpuExecutor::scheduler_context_idx_for_thread(int32_t thread_idx) const { + int32_t sched_idx = scheduler_idx_for_thread(thread_idx); + if (sched_idx >= 0) { + return sched_idx; + } + int32_t orch_stage_idx = orch_stage_idx_for_thread(thread_idx); + if (orch_stage_idx >= 0) { + return sched_thread_num_ + orch_stage_idx; + } + return thread_idx; +} + int32_t AicpuExecutor::init(Runtime *runtime) { bool expected = false; if (!initialized_.compare_exchange_strong(expected, true, std::memory_order_acq_rel, std::memory_order_acquire)) { @@ -200,12 +299,7 @@ int32_t AicpuExecutor::init(Runtime *runtime) { return -1; } - // Read execution parameters from runtime. The 0 → 1 fixup runs before the - // sched_thread_num_ derivation so a zero input doesn't leave the scheduler - // count at -1. - aicpu_thread_num_ = runtime->dev.aicpu_thread_num; - if (aicpu_thread_num_ == 0) aicpu_thread_num_ = 1; - sched_thread_num_ = aicpu_thread_num_ - 1; + resolve_thread_layout(runtime); serial_orch_sched_ = runtime->dev.serial_orch_sched; if (aicpu_thread_num_ < 1 || aicpu_thread_num_ > MAX_AICPU_THREADS) { @@ -213,6 +307,11 @@ int32_t AicpuExecutor::init(Runtime *runtime) { init_failed_.store(true, std::memory_order_release); return -1; } + LOG_INFO_V0( + "AicpuExecutor layout: strategy=%s threads=%d sched=%d orch=%d primary_orch=%d clusters=%s", + pipeline_layout_.name, aicpu_thread_num_, sched_thread_num_, orch_thread_num_, primary_orch_thread_idx_, + pipeline_layout_.cluster_layout + ); if (sched_ctx_.init(runtime, aicpu_thread_num_, sched_thread_num_, get_platform_regs()) != 0) { init_failed_.store(true, std::memory_order_release); @@ -379,9 +478,11 @@ int32_t AicpuExecutor::run(Runtime *runtime) { // onboard, where the filter gate already set this same value. platform_aicpu_affinity_set_thread_idx(thread_idx); LOG_INFO_V0("Thread %d: Start (exec_idx=%d)", thread_idx, affinity_exec_idx); + int32_t scheduler_idx = scheduler_idx_for_thread(thread_idx); + int32_t orch_stage_idx = orch_stage_idx_for_thread(thread_idx); // Orchestrator check - if (thread_idx >= sched_thread_num_) { + if (orch_stage_idx == 0) { #if PTO2_PROFILING uint64_t orch_cycle_start = 0; #endif @@ -545,6 +646,15 @@ int32_t AicpuExecutor::run(Runtime *runtime) { // Fill ops / core counts (host can't resolve s_runtime_ops's // device address nor know the SchedulerContext's core fan-out). runtime_finalize_after_wire(rt, sched_ctx_.aic_count(), sched_ctx_.aiv_count()); + bool use_strategy2_submit_ctrl = + pipeline_layout_.strategy == PipelineStrategy::S2_O2_SPLIT_CTRL_STRATEGY2; + bool use_deferred_submit = use_strategy2_submit_ctrl && sched_thread_num_ == 2 && orch_thread_num_ == 2; + if (runtime->dev.pipeline_defer_submit_disabled) { + use_deferred_submit = false; + } + rt->orchestrator.enable_submit_pipeline( + orch_thread_num_, false, use_deferred_submit, use_deferred_submit, use_deferred_submit + ); #if PTO2_PROFILING rt->orchestrator.l2_swimlane_level = get_l2_swimlane_level(); { @@ -598,6 +708,7 @@ int32_t AicpuExecutor::run(Runtime *runtime) { rt_scope_begin(rt); (*p_func)(orch_args_cached_); rt_scope_end(rt); + rt->orchestrator.stop_submit_pipeline(); #if PTO2_PROFILING // Flush the (potentially partially-filled) DepGenBuffer so the host @@ -723,10 +834,42 @@ int32_t AicpuExecutor::run(Runtime *runtime) { #endif // PTO2_ORCH_PROFILING #endif // PTO2_PROFILING LOG_INFO_V0("Thread %d: Orchestrator completed", thread_idx); + } else if (orch_stage_idx > 0) { + LOG_INFO_V0( + "Thread %d: Orchestrator pipeline stage %d waiting for primary orchestrator", thread_idx, orch_stage_idx + ); + while (!runtime_init_ready_.load(std::memory_order_acquire)) { + SPIN_WAIT_HINT(); + } + if (rt != nullptr) { +#if PTO2_PROFILING + if (get_l2_swimlane_level() >= L2SwimlaneLevel::ORCH_PHASES) { + l2_swimlane_aicpu_set_orch_thread_idx(thread_idx); + } + uint64_t orch_stage_cycle_start = get_sys_cnt_aicpu(); + uint64_t orch_stage_active_cycles = rt->orchestrator.run_submit_pipeline_worker(orch_stage_idx, thread_idx); + uint64_t orch_stage_cycle_end = get_sys_cnt_aicpu(); + if (rt->orchestrator.submit_pipeline_work_available.load(std::memory_order_acquire) || + orch_stage_active_cycles > 0) { + LOG_INFO_V9( + "Thread %d: orch_stage_start=%" PRIu64 " orch_stage_end=%" PRIu64 + " orch_stage=%d orch_stage_cost=%.3fus orch_stage_active_cost=%.3fus", + thread_idx, static_cast(orch_stage_cycle_start), + static_cast(orch_stage_cycle_end), orch_stage_idx, + cycles_to_us(orch_stage_cycle_end - orch_stage_cycle_start), cycles_to_us(orch_stage_active_cycles) + ); + } +#else + (void)rt->orchestrator.run_submit_pipeline_worker(orch_stage_idx, thread_idx); +#endif + } else { + LOG_ERROR("Thread %d: rt is null, skipping orchestrator pipeline stage %d", thread_idx, orch_stage_idx); + } + LOG_INFO_V0("Thread %d: Orchestrator pipeline stage %d completed", thread_idx, orch_stage_idx); } // Scheduler thread (orchestrator thread skips dispatch and exits after orchestration) - if (!sched_ctx_.is_completed() && thread_idx < sched_thread_num_) { + if (!sched_ctx_.is_completed() && scheduler_idx >= 0) { // Device orchestration: wait for the primary orchestrator to initialize the SM header while (!runtime_init_ready_.load(std::memory_order_acquire)) { SPIN_WAIT_HINT(); @@ -738,7 +881,7 @@ int32_t AicpuExecutor::run(Runtime *runtime) { if (serial_orch_sched_) { sched_ctx_.wait_for_orchestration_done_before_dispatch(runtime, thread_idx); } - int32_t completed = sched_ctx_.resolve_and_dispatch(runtime, thread_idx); + int32_t completed = sched_ctx_.resolve_and_dispatch(runtime, scheduler_idx); if (completed < 0) { LOG_ERROR("Thread %d: Scheduler failed with rc=%d", thread_idx, completed); run_rc = completed; @@ -799,7 +942,14 @@ void AicpuExecutor::deinit(Runtime *runtime) { aicpu_thread_num_ = 0; sched_thread_num_ = 0; + orch_thread_num_ = 0; + primary_orch_thread_idx_ = -1; + pipeline_layout_ = {}; serial_orch_sched_ = false; + for (int32_t i = 0; i < MAX_AICPU_THREADS; ++i) { + scheduler_index_by_thread_[i] = -1; + orchestrator_stage_by_thread_[i] = -1; + } orch_args_cached_.reset(); // orch_so_table_ entries are intentionally preserved across deinit: they diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_dep_compute.h b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_dep_compute.h index f8392dfbf..e4f415d88 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_dep_compute.h +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_dep_compute.h @@ -65,34 +65,23 @@ struct DepInputs { const PTO2TaskId *explicit_deps; // length = explicit_dep_count (validity checked by caller) }; -/** - * Compute fanin for a task being submitted (STEP 3: Step A creator retention + - * Step B tensormap modifier lookup). - * - * For each non-OUTPUT tensor: - * - If owner_task_id is valid, emit(owner) - * - For INPUT/INOUT (and not manual_dep), tensor_map.lookup(*tensor) and emit - * each matching producer. INOUT+COVERED triggers tensor_map.remove_entry(entry). - * - * @return true on success (or producer-skipped-silently); false if emit signaled - * fatal — caller should propagate (after any fatal bookkeeping done by emit). - */ -template -[[nodiscard]] inline bool -compute_task_fanin(const DepInputs &inputs, PTO2TensorMap &tensor_map, bool in_manual_scope, Emit emit) { +template +[[nodiscard]] inline bool compute_task_fanin_impl( + int32_t tensor_count, TensorAt tensor_at, TagAt tag_at, PTO2TensorMap &tensor_map, bool in_manual_scope, Emit emit +) { if (in_manual_scope) { return true; } - for (int32_t i = 0; i < inputs.tensor_count; i++) { - TensorArgType ptype = inputs.arg_types[i]; + for (int32_t i = 0; i < tensor_count; i++) { + TensorArgType ptype = tag_at(i); if (ptype == TensorArgType::OUTPUT) { // Runtime-created OUTPUT tensors are not looked up in the TensorMap since // they have no dependencies. continue; } - const Tensor *tensor = &inputs.tensors[i].ref(); + const Tensor *tensor = tensor_at(i); // Step A: creator retention — all existing tensors extend their creator lifetime. PTO2TaskId owner = tensor->owner_task_id; @@ -109,7 +98,6 @@ compute_task_fanin(const DepInputs &inputs, PTO2TensorMap &tensor_map, bool in_m if (tensor->manual_dep) { continue; } - bool fatal = false; tensor_map.lookup(*tensor, [&](PTO2TensorMapEntry &entry, OverlapStatus overlap_status) -> bool { if (!emit(entry.producer_task_id)) { @@ -128,23 +116,18 @@ compute_task_fanin(const DepInputs &inputs, PTO2TensorMap &tensor_map, bool in_m return true; } -/** - * Register a task's outputs in the tensormap (STEP 4 in submit_task). - * - * For INOUT and OUTPUT_EXISTING tensors (excluding manual_dep), inserts the - * tensor into tensor_map keyed by its buffer.addr with `task_id` as producer. - * - * No-op when in_manual_scope. - */ -inline void -register_task_outputs(const DepInputs &inputs, PTO2TaskId task_id, PTO2TensorMap &tensor_map, bool in_manual_scope) { +template +inline void register_task_outputs_impl( + int32_t tensor_count, TensorAt tensor_at, TagAt tag_at, PTO2TaskId task_id, PTO2TensorMap &tensor_map, + bool in_manual_scope +) { if (in_manual_scope) { return; } - for (int32_t i = 0; i < inputs.tensor_count; i++) { - TensorArgType ptype = inputs.arg_types[i]; + for (int32_t i = 0; i < tensor_count; i++) { + TensorArgType ptype = tag_at(i); if (ptype == TensorArgType::INOUT || ptype == TensorArgType::OUTPUT_EXISTING) { - const Tensor *tensor = &inputs.tensors[i].ref(); + const Tensor *tensor = tensor_at(i); if (!tensor->manual_dep) { tensor_map.insert(*tensor, task_id); } @@ -153,22 +136,99 @@ register_task_outputs(const DepInputs &inputs, PTO2TaskId task_id, PTO2TensorMap } /** - * Count the tensormap entries register_task_outputs() will insert for this task. + * Compute fanin for a task being submitted (STEP 3: Step A creator retention + + * Step B tensormap modifier lookup). + * + * For each non-OUTPUT tensor: + * - If owner_task_id is valid, emit(owner) + * - For INPUT/INOUT (and not manual_dep), tensor_map.lookup(*tensor) and emit + * each matching producer. INOUT+COVERED triggers tensor_map.remove_entry(entry). + * + * @return true on success (or producer-skipped-silently); false if emit signaled + * fatal — caller should propagate (after any fatal bookkeeping done by emit). + */ +template +[[nodiscard]] inline bool +compute_task_fanin(const DepInputs &inputs, PTO2TensorMap &tensor_map, bool in_manual_scope, Emit emit) { + return compute_task_fanin_impl( + inputs.tensor_count, + [&](int32_t i) { + return &inputs.tensors[i].ref(); + }, + [&](int32_t i) { + return inputs.arg_types[i]; + }, + tensor_map, in_manual_scope, emit + ); +} + +template +[[nodiscard]] inline bool compute_payload_task_fanin( + const PTO2TaskPayload &payload, const TensorArgType *arg_types, PTO2TensorMap &tensor_map, bool in_manual_scope, + Emit emit +) { + return compute_task_fanin_impl( + payload.tensor_count, + [&](int32_t i) { + return &payload.tensors[i]; + }, + [&](int32_t i) { + return arg_types[i]; + }, + tensor_map, in_manual_scope, emit + ); +} + +/** + * Register a task's outputs in the tensormap (STEP 4 in submit_task). + * + * For INOUT and OUTPUT_EXISTING tensors (excluding manual_dep), inserts the + * tensor into tensor_map keyed by its buffer.addr with `task_id` as producer. * - * Mirrors register_task_outputs()'s selection exactly (INOUT / OUTPUT_EXISTING, - * excluding manual_dep), so the returned value is the precise number of - * new_entry() calls that step makes. The orchestrator uses it to reserve pool - * capacity before inserting. Returns 0 in a manual scope (no registration). + * No-op when in_manual_scope. */ -inline int32_t count_registrable_outputs(const DepInputs &inputs, bool in_manual_scope) { +inline void +register_task_outputs(const DepInputs &inputs, PTO2TaskId task_id, PTO2TensorMap &tensor_map, bool in_manual_scope) { + register_task_outputs_impl( + inputs.tensor_count, + [&](int32_t i) { + return &inputs.tensors[i].ref(); + }, + [&](int32_t i) { + return inputs.arg_types[i]; + }, + task_id, tensor_map, in_manual_scope + ); +} + +inline void register_payload_task_outputs( + const PTO2TaskPayload &payload, const TensorArgType *arg_types, PTO2TaskId task_id, PTO2TensorMap &tensor_map, + bool in_manual_scope +) { + register_task_outputs_impl( + payload.tensor_count, + [&](int32_t i) { + return &payload.tensors[i]; + }, + [&](int32_t i) { + return arg_types[i]; + }, + task_id, tensor_map, in_manual_scope + ); +} + +template +inline int32_t +count_registrable_outputs_impl(int32_t tensor_count, TensorAt tensor_at, TagAt tag_at, bool in_manual_scope) { if (in_manual_scope) { return 0; } int32_t needed = 0; - for (int32_t i = 0; i < inputs.tensor_count; i++) { - TensorArgType ptype = inputs.arg_types[i]; + for (int32_t i = 0; i < tensor_count; i++) { + TensorArgType ptype = tag_at(i); if (ptype == TensorArgType::INOUT || ptype == TensorArgType::OUTPUT_EXISTING) { - if (!inputs.tensors[i].ref().manual_dep) { + const Tensor *tensor = tensor_at(i); + if (!tensor->manual_dep) { needed++; } } @@ -176,4 +236,32 @@ inline int32_t count_registrable_outputs(const DepInputs &inputs, bool in_manual return needed; } +inline int32_t count_registrable_outputs(const DepInputs &inputs, bool in_manual_scope) { + return count_registrable_outputs_impl( + inputs.tensor_count, + [&](int32_t i) { + return &inputs.tensors[i].ref(); + }, + [&](int32_t i) { + return inputs.arg_types[i]; + }, + in_manual_scope + ); +} + +inline int32_t count_payload_registrable_outputs( + const PTO2TaskPayload &payload, const TensorArgType *arg_types, bool in_manual_scope +) { + return count_registrable_outputs_impl( + payload.tensor_count, + [&](int32_t i) { + return &payload.tensors[i]; + }, + [&](int32_t i) { + return arg_types[i]; + }, + in_manual_scope + ); +} + #endif // SRC_A2A3_RUNTIME_TENSORMAP_AND_RINGBUFFER_RUNTIME_PTO_DEP_COMPUTE_H_ 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 b7cc58794..d47ea3116 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp @@ -26,6 +26,8 @@ #include #include +#include + #include "aicpu/dep_gen_collector_aicpu.h" #include "common/dep_gen.h" #include "common/unified_log.h" @@ -346,6 +348,8 @@ static bool append_fanin_or_fail( } static void scope_tasks_push(PTO2OrchestratorState *orch, PTO2TaskSlotState *task_slot_state); +static bool +enqueue_scope_end_record_if_supported(PTO2OrchestratorState *orch, PTO2TaskSlotState **task_slot_states, int32_t count); struct PTO2PreparedTask { PTO2TaskId task_id = PTO2TaskId::invalid(); @@ -358,7 +362,8 @@ struct PTO2PreparedTask { static PTO2OutputLayout calculate_output_layout(const L0TaskArgs &args) { PTO2OutputLayout layout; for (int32_t i = 0; i < args.tensor_count(); i++) { - if (args.tag(i) != TensorArgType::OUTPUT) { + TensorArgType tag = args.tag(i); + if (tag != TensorArgType::OUTPUT) { continue; } layout.offsets[i] = layout.total_output_size; @@ -570,7 +575,12 @@ void PTO2OrchestratorState::end_scope() { } if (orch->scheduler && count > 0) { - orch->scheduler->on_scope_end(&orch->scope_tasks[begin], count); + if (!enqueue_scope_end_record_if_supported(orch, &orch->scope_tasks[begin], count)) { + orch->flush_submit_pipeline(); + orch->scheduler->on_scope_end(&orch->scope_tasks[begin], count); + } + } else { + orch->flush_submit_pipeline(); } // Rewind the task buffer — these entries are no longer needed @@ -682,6 +692,429 @@ static bool ensure_tensormap_capacity(PTO2OrchestratorState *orch, int32_t neede return true; } +// ============================================================================= +// Submit Commit Pipeline +// ============================================================================= + +static void submit_pipeline_record_copy(PTO2SubmitCommitRecord &dst, const PTO2SubmitCommitRecord &src) { + dst.kind = src.kind; + dst.payload = src.payload; + dst.slot_state = src.slot_state; + dst.scheduler = src.scheduler; + dst.task_id = src.task_id; + dst.explicit_dep_count = src.explicit_dep_count; + dst.scope_task_count = src.scope_task_count; + dst.in_manual_scope = src.in_manual_scope; + for (int32_t i = 0; i < PTO2_SUBMIT_PIPELINE_EXPLICIT_DEP_CAP; i++) { + dst.explicit_deps[i] = src.explicit_deps[i]; + } + for (int32_t i = 0; i < PTO2_SUBMIT_PIPELINE_SCOPE_INLINE_CAP; i++) { + dst.scope_task_slot_states[i] = src.scope_task_slot_states[i]; + } + for (int32_t i = 0; i < MAX_TENSOR_ARGS; i++) { + dst.arg_types[i] = src.arg_types[i]; + } + for (int32_t i = 0; i < PTO2_SUBTASK_SLOT_COUNT; i++) { + dst.kernel_id[i] = src.kernel_id[i]; + } +} + +static void submit_pipeline_record_reset(PTO2SubmitCommitRecord &record) { + PTO2SubmitCommitRecord empty{}; + submit_pipeline_record_copy(record, empty); +} + +static void submit_pipeline_queue_reset(PTO2SubmitPipelineQueue &queue) { + queue.tail.store(0, std::memory_order_release); + queue.head.store(0, std::memory_order_release); + for (int32_t i = 0; i < PTO2_SUBMIT_PIPELINE_QUEUE_CAP; i++) { + queue.slot_state[i].store(0, std::memory_order_release); + submit_pipeline_record_reset(queue.records[i]); + } +} + +static void submit_pipeline_queue_push(PTO2SubmitPipelineQueue &queue, const PTO2SubmitCommitRecord &record) { + uint64_t seq = queue.tail.load(std::memory_order_acquire); + int32_t slot = static_cast(seq % PTO2_SUBMIT_PIPELINE_QUEUE_CAP); + while (queue.slot_state[slot].load(std::memory_order_acquire) != 0) { + SPIN_WAIT_HINT(); + } + submit_pipeline_record_copy(queue.records[slot], record); + queue.slot_state[slot].store(1, std::memory_order_release); + queue.tail.store(seq + 1, std::memory_order_release); +} + +static bool submit_pipeline_queue_pop(PTO2SubmitPipelineQueue &queue, PTO2SubmitCommitRecord *record) { + uint64_t seq = queue.head.load(std::memory_order_acquire); + uint64_t tail = queue.tail.load(std::memory_order_acquire); + if (seq >= tail) { + return false; + } + + int32_t slot = static_cast(seq % PTO2_SUBMIT_PIPELINE_QUEUE_CAP); + while (queue.slot_state[slot].load(std::memory_order_acquire) != 1) { + SPIN_WAIT_HINT(); + } + submit_pipeline_record_copy(*record, queue.records[slot]); + queue.slot_state[slot].store(0, std::memory_order_release); + queue.head.store(seq + 1, std::memory_order_release); + return true; +} + +static bool submit_pipeline_queue_empty(PTO2SubmitPipelineQueue &queue) { + return queue.head.load(std::memory_order_acquire) >= queue.tail.load(std::memory_order_acquire); +} + +static void mark_submit_pipeline_work_available(PTO2OrchestratorState *orch) { + orch->submit_pipeline_work_available.store(true, std::memory_order_release); + orch->submit_pipeline_control.store(PTO2_SUBMIT_PIPELINE_CONTROL_WORK, std::memory_order_release); +} + +static void signal_scheduler_drain_if_needed(PTO2OrchestratorState *orch, PTO2SubmitCommitRecord &record) { + if (!orch->submit_pipeline_signal_scheduler_drain || record.scheduler == nullptr) { + return; + } + record.scheduler->wiring.orch_needs_drain.store(true, std::memory_order_release); +#if PTO2_PROFILING + orch->submit_pipeline_scheduler_drain_hint_count++; +#endif +} + +static bool +fill_deferred_submit_metadata(PTO2OrchestratorState *orch, const L0TaskArgs &args, PTO2SubmitCommitRecord *record) { + if (args.explicit_dep_count() > PTO2_SUBMIT_PIPELINE_EXPLICIT_DEP_CAP) { + return false; + } + record->explicit_dep_count = static_cast(args.explicit_dep_count()); + record->in_manual_scope = orch->in_manual_scope(); + for (int32_t i = 0; i < args.tensor_count(); i++) { + record->arg_types[i] = args.tag(i); + } + for (int32_t i = 0; i < record->explicit_dep_count; i++) { + record->explicit_deps[i] = args.explicit_dep(static_cast(i)); + } + return true; +} + +static bool enqueue_deferred_submit_record(PTO2OrchestratorState *orch, const PTO2SubmitCommitRecord &record) { + if (!orch->submit_pipeline_defer_dependencies) { + return false; + } +#if PTO2_PROFILING + uint64_t enqueue_start = get_sys_cnt_aicpu(); +#endif + mark_submit_pipeline_work_available(orch); + submit_pipeline_queue_push(orch->submit_pipeline_queues[0], record); +#if PTO2_PROFILING + orch->submit_pipeline_task_enqueue_cycles += get_sys_cnt_aicpu() - enqueue_start; + orch->submit_pipeline_task_enqueue_count++; +#endif + return true; +} + +static bool enqueue_scope_end_record_if_supported( + PTO2OrchestratorState *orch, PTO2TaskSlotState **task_slot_states, int32_t count +) { + if (!orch->submit_pipeline_defer_dependencies || orch->submit_pipeline_commit_stages != 1 || + count > PTO2_SUBMIT_PIPELINE_SCOPE_INLINE_CAP) { + return false; + } + + PTO2SubmitCommitRecord record{}; + record.kind = PTO2SubmitPipelineRecordKind::SCOPE_END; + record.scheduler = orch->scheduler; + record.scope_task_count = count; + for (int32_t i = 0; i < count; i++) { + record.scope_task_slot_states[i] = task_slot_states[i]; + } + +#if PTO2_PROFILING + uint64_t enqueue_start = get_sys_cnt_aicpu(); +#endif + mark_submit_pipeline_work_available(orch); + submit_pipeline_queue_push(orch->submit_pipeline_queues[0], record); +#if PTO2_PROFILING + orch->submit_pipeline_scope_enqueue_cycles += get_sys_cnt_aicpu() - enqueue_start; + orch->submit_pipeline_scope_enqueue_count++; +#endif + return true; +} + +static bool compute_deferred_submit_dependencies(PTO2OrchestratorState *orch, PTO2SubmitCommitRecord &record) { + uint8_t ring_id = record.task_id.ring(); + PTO2RingFlowControl &fc = orch->sm_header->rings[ring_id].fc; + int32_t sm_last_task_alive = fc.last_task_alive.load(std::memory_order_acquire); + orch->tensor_map.sync_tensormap(record.task_id, sm_last_task_alive); + + PTO2FaninBuilder fanin_builder(orch, orch->rings[ring_id].fanin_pool, next_fanin_seen_epoch(orch)); + + for (int32_t i = 0; i < record.explicit_dep_count; i++) { + PTO2TaskId dep_task_id = record.explicit_deps[i]; + if (!dep_task_id.is_valid()) { + orch->report_fatal( + PTO2_ERROR_INVALID_ARGS, __FUNCTION__, "Arg.set_dependencies(...) requires valid task ids" + ); + return false; + } + uint8_t dep_ring_id = dep_task_id.ring(); + PTO2SharedMemoryRingHeader &dep_ring = orch->sm_header->rings[dep_ring_id]; + int32_t dep_local_task_id = static_cast(dep_task_id.local()); + int32_t dep_last_task_alive = dep_ring.fc.last_task_alive.load(std::memory_order_acquire); + if (dep_local_task_id < dep_last_task_alive) { + continue; + } + int32_t dep_slot = dep_ring.get_slot_by_task_id(dep_local_task_id); + PTO2TaskSlotState *producer_slot_state = &dep_ring.get_slot_state_by_slot(dep_slot); + if (!append_fanin_or_fail( + orch, dep_ring_id, dep_slot, producer_slot_state, dep_task_id, &fanin_builder, ring_id + )) { + return false; + } + } + + auto runtime_emit = [&](PTO2TaskId producer_task_id) -> bool { + uint8_t prod_ring = producer_task_id.ring(); + PTO2SharedMemoryRingHeader &producer_ring = orch->sm_header->rings[prod_ring]; + int32_t prod_slot = producer_ring.get_slot_by_task_id(static_cast(producer_task_id.local())); + PTO2TaskSlotState *prod_state = &producer_ring.get_slot_state_by_slot(prod_slot); + return append_fanin_or_fail(orch, prod_ring, prod_slot, prod_state, producer_task_id, &fanin_builder, ring_id); + }; + + if (!compute_payload_task_fanin( + *record.payload, record.arg_types, orch->tensor_map, record.in_manual_scope, runtime_emit + )) { + return false; + } + + int32_t needed = count_payload_registrable_outputs(*record.payload, record.arg_types, record.in_manual_scope); + if (needed > 0 && !ensure_tensormap_capacity(orch, needed)) { + return false; + } + if (needed > 0) { + register_payload_task_outputs( + *record.payload, record.arg_types, record.task_id, orch->tensor_map, record.in_manual_scope + ); +#if PTO2_PROFILING + orch->submit_pipeline_dep_register_count += static_cast(needed); +#endif + } + + int32_t inline_count = std::min(fanin_builder.count, PTO2_FANIN_INLINE_CAP); + record.payload->fanin_actual_count = fanin_builder.count; + record.payload->fanin_spill_start = fanin_builder.spill_start; + record.payload->fanin_spill_pool = &fanin_builder.spill_pool; + for (int32_t i = 0; i < inline_count; i++) { + record.payload->fanin_inline_slot_states[i] = fanin_builder.inline_slots[i]; + } +#if PTO2_PROFILING + orch->submit_pipeline_dep_explicit_count += static_cast(record.explicit_dep_count); + orch->submit_pipeline_dep_fanin_actual_count += static_cast(fanin_builder.count); +#endif + return true; +} + +static void commit_scope_end_record(PTO2OrchestratorState *orch, PTO2SubmitCommitRecord &record) { + if (record.scheduler != nullptr && record.scope_task_count > 0) { + record.scheduler->on_scope_end(record.scope_task_slot_states, record.scope_task_count); + } +#if PTO2_PROFILING + orch->submit_pipeline_scope_record_count++; +#else + (void)orch; +#endif +} + +static void commit_deferred_submit_record(PTO2OrchestratorState *orch, PTO2SubmitCommitRecord &record) { + if (record.payload == nullptr || record.slot_state == nullptr || record.scheduler == nullptr) { + orch->report_fatal(PTO2_ERROR_INVALID_ARGS, __FUNCTION__, "invalid deferred submit record"); + return; + } + +#if PTO2_PROFILING + uint64_t dep_start = get_sys_cnt_aicpu(); +#endif + if (!compute_deferred_submit_dependencies(orch, record)) { + return; + } +#if PTO2_PROFILING + uint64_t dep_end = get_sys_cnt_aicpu(); + orch->submit_pipeline_deferred_dep_cycles += dep_end - dep_start; + uint64_t publish_start = dep_end; +#endif + + uint64_t publish_spins = 0; + if (record.payload->fanin_actual_count == 0) { + publish_spins = record.scheduler->publish_ready_no_fanin(record.slot_state); + } else { + while (!record.scheduler->wiring.queue.push(record.slot_state)) { + publish_spins++; + if (orch->sm_header->orch_error_code.load(std::memory_order_acquire) != PTO2_ERROR_NONE) { + orch->fatal = true; + return; + } + SPIN_WAIT_HINT(); + } + signal_scheduler_drain_if_needed(orch, record); + } + +#if PTO2_PROFILING + uint64_t publish_end = get_sys_cnt_aicpu(); + orch->submit_pipeline_publish_cycles += publish_end - publish_start; + orch->submit_pipeline_publish_spins += publish_spins; + orch->submit_pipeline_deferred_commit_count++; +#else + (void)publish_spins; +#endif +} + +void PTO2OrchestratorState::enable_submit_pipeline( + int32_t orchestrator_threads, bool /*enqueue_submit_records*/, bool defer_submit_dependencies, + bool signal_scheduler_drain, bool compact_deferred_records +) { + submit_pipeline_commit_stages = orchestrator_threads >= 2 ? 1 : 0; + submit_pipeline_enabled = submit_pipeline_commit_stages > 0; + submit_pipeline_defer_dependencies = submit_pipeline_enabled && defer_submit_dependencies; + submit_pipeline_signal_scheduler_drain = + submit_pipeline_defer_dependencies && signal_scheduler_drain && scheduler != nullptr; + submit_pipeline_compact_deferred_records = submit_pipeline_defer_dependencies && compact_deferred_records; + submit_pipeline_stop.store(false, std::memory_order_release); + submit_pipeline_work_available.store(false, std::memory_order_release); + submit_pipeline_control.store(PTO2_SUBMIT_PIPELINE_CONTROL_IDLE, std::memory_order_release); + submit_pipeline_completed.store(0, std::memory_order_release); +#if PTO2_PROFILING + submit_pipeline_task_enqueue_cycles = 0; + submit_pipeline_scope_enqueue_cycles = 0; + submit_pipeline_flush_cycles = 0; + submit_pipeline_deferred_dep_cycles = 0; + submit_pipeline_deferred_fanin_cycles = 0; + submit_pipeline_publish_cycles = 0; + submit_pipeline_scope_release_cycles = 0; + submit_pipeline_publish_spins = 0; + submit_pipeline_scheduler_drain_hint_count = 0; + submit_pipeline_task_enqueue_count = 0; + submit_pipeline_scope_enqueue_count = 0; + submit_pipeline_flush_count = 0; + submit_pipeline_deferred_commit_count = 0; + submit_pipeline_scope_record_count = 0; + submit_pipeline_dep_explicit_count = 0; + submit_pipeline_dep_register_count = 0; + submit_pipeline_dep_fanin_actual_count = 0; +#endif + for (int32_t stage = 0; stage < PTO2_SUBMIT_PIPELINE_MAX_COMMIT_STAGES; stage++) { + submit_pipeline_stage_done[stage].store(false, std::memory_order_release); + submit_pipeline_queue_reset(submit_pipeline_queues[stage]); + } +} + +uint64_t PTO2OrchestratorState::run_submit_pipeline_worker(int32_t stage_idx, int32_t /*phase_thread_idx*/) { + uint64_t active_cycles = 0; + if (!submit_pipeline_enabled || stage_idx != 1) { + while (!submit_pipeline_stop.load(std::memory_order_acquire)) { + SPIN_WAIT_HINT(); + } + return active_cycles; + } + + while (submit_pipeline_control.load(std::memory_order_acquire) == PTO2_SUBMIT_PIPELINE_CONTROL_IDLE) { + SPIN_WAIT_HINT(); + } + + PTO2SubmitPipelineQueue &queue = submit_pipeline_queues[0]; + while (true) { + PTO2SubmitCommitRecord record{}; + if (!submit_pipeline_queue_pop(queue, &record)) { + if (submit_pipeline_stop.load(std::memory_order_acquire) && submit_pipeline_queue_empty(queue)) { + submit_pipeline_stage_done[0].store(true, std::memory_order_release); + return active_cycles; + } + SPIN_WAIT_HINT(); + continue; + } + +#if PTO2_PROFILING + uint64_t start = get_sys_cnt_aicpu(); +#endif + if (record.kind == PTO2SubmitPipelineRecordKind::SCOPE_END) { + commit_scope_end_record(this, record); + } else { + commit_deferred_submit_record(this, record); + } + submit_pipeline_completed.fetch_add(1, std::memory_order_acq_rel); +#if PTO2_PROFILING + active_cycles += get_sys_cnt_aicpu() - start; +#endif + } +} + +void PTO2OrchestratorState::flush_submit_task_batch() {} + +void PTO2OrchestratorState::flush_submit_pipeline() { + if (!submit_pipeline_enabled) { + return; + } + uint64_t target = submit_pipeline_queues[0].tail.load(std::memory_order_acquire); + if (submit_pipeline_completed.load(std::memory_order_acquire) >= target) { + return; + } +#if PTO2_PROFILING + uint64_t flush_start = get_sys_cnt_aicpu(); +#endif + while (submit_pipeline_completed.load(std::memory_order_acquire) < target) { + SPIN_WAIT_HINT(); + } +#if PTO2_PROFILING + submit_pipeline_flush_cycles += get_sys_cnt_aicpu() - flush_start; + submit_pipeline_flush_count++; +#endif +} + +void PTO2OrchestratorState::log_submit_pipeline_diagnostics(int32_t thread_idx) const { +#if PTO2_PROFILING + LOG_INFO_V9( + "Thread %d: orch_pipeline_diag task_enq=%u scope_enq=%u deferred_commit=%u scope_record=%u " + "dep_explicit=%u dep_register=%u dep_fanin=%u drain_hints=%u publish_spins=%" PRIu64 + " enqueue=%.3fus scope_enqueue=%.3fus flush=%.3fus dep=%.3fus publish=%.3fus", + thread_idx, submit_pipeline_task_enqueue_count, submit_pipeline_scope_enqueue_count, + submit_pipeline_deferred_commit_count, submit_pipeline_scope_record_count, submit_pipeline_dep_explicit_count, + submit_pipeline_dep_register_count, submit_pipeline_dep_fanin_actual_count, + submit_pipeline_scheduler_drain_hint_count, static_cast(submit_pipeline_publish_spins), + cycles_to_us(submit_pipeline_task_enqueue_cycles), cycles_to_us(submit_pipeline_scope_enqueue_cycles), + cycles_to_us(submit_pipeline_flush_cycles), cycles_to_us(submit_pipeline_deferred_dep_cycles), + cycles_to_us(submit_pipeline_publish_cycles) + ); +#else + (void)thread_idx; +#endif +} + +void PTO2OrchestratorState::log_four_stage_diagnostics(int32_t thread_idx) const { (void)thread_idx; } + +void PTO2OrchestratorState::log_active_detail_diagnostics( + int32_t thread_idx, uint64_t active_cycles, uint64_t bind_cycles, uint64_t p_bind_cycles, + uint64_t outer_scope_begin_cycles, uint64_t p_func_cycles, uint64_t outer_scope_end_cycles +) const { + (void)thread_idx; + (void)active_cycles; + (void)bind_cycles; + (void)p_bind_cycles; + (void)outer_scope_begin_cycles; + (void)p_func_cycles; + (void)outer_scope_end_cycles; +} + +void PTO2OrchestratorState::log_submit_detail_diagnostics(int32_t thread_idx) const { (void)thread_idx; } + +void PTO2OrchestratorState::stop_submit_pipeline() { + if (!submit_pipeline_enabled) { + submit_pipeline_stop.store(true, std::memory_order_release); + submit_pipeline_control.store(PTO2_SUBMIT_PIPELINE_CONTROL_STOP, std::memory_order_release); + return; + } + flush_submit_pipeline(); + submit_pipeline_stop.store(true, std::memory_order_release); + submit_pipeline_control.store(PTO2_SUBMIT_PIPELINE_CONTROL_STOP, std::memory_order_release); +} + // Shared body for submit_task / submit_dummy_task. Caller has already validated // args.has_error, decided active_mask (empty for dummy), and resolved the per-slot // kernel_ids (all INVALID_KERNEL_ID for dummy). Performs tensormap sync, fanin @@ -743,8 +1176,6 @@ static TaskOutputTensors submit_task_common( } #endif - PTO2FaninBuilder fanin_builder(orch, orch->rings[ring_id].fanin_pool, next_fanin_seen_epoch(orch)); - CYCLE_COUNT_LAP(g_orch_alloc_cycle); #if PTO2_PROFILING @@ -754,6 +1185,60 @@ static TaskOutputTensors submit_task_common( } #endif + bool defer_dependencies = + orch->submit_pipeline_defer_dependencies && args.explicit_dep_count() <= PTO2_SUBMIT_PIPELINE_EXPLICIT_DEP_CAP; + if (defer_dependencies) { + __builtin_prefetch(&task, 1, 1); + task.task_id = task_id; + task.kernel_id[static_cast(PTO2SubtaskSlot::AIC)] = aic_kernel_id; + task.kernel_id[static_cast(PTO2SubtaskSlot::AIV0)] = aiv0_kernel_id; + task.kernel_id[static_cast(PTO2SubtaskSlot::AIV1)] = aiv1_kernel_id; + task.packed_buffer_base = prepared.alloc_result.packed_base; + task.packed_buffer_end = prepared.alloc_result.packed_end; + + payload.fanin_actual_count = 0; + payload.fanin_spill_start = 0; + payload.fanin_spill_pool = &orch->rings[ring_id].fanin_pool; + payload.init(args, result, prepared.alloc_result, layout); +#if PTO2_PROFILING + if (is_dump_args_enabled()) { + if (args.scalar_count() > 0) { + set_dump_args_task_scalar_dtypes( + task_id.raw, static_cast(args.scalar_count()), args.scalar_dtypes() + ); + } + if (args.dump_arg_mask() != 0) { + set_dump_args_task_mask(task_id.raw, args.dump_arg_mask(), args.dump_arg_index_ambiguous_mask()); + } + } +#endif + + PTO2SubmitCommitRecord record{}; + record.kind = PTO2SubmitPipelineRecordKind::TASK_DEFERRED; + record.payload = &payload; + record.slot_state = &cur_slot_state; + record.scheduler = sched; + record.task_id = task_id; + record.kernel_id[static_cast(PTO2SubtaskSlot::AIC)] = aic_kernel_id; + record.kernel_id[static_cast(PTO2SubtaskSlot::AIV0)] = aiv0_kernel_id; + record.kernel_id[static_cast(PTO2SubtaskSlot::AIV1)] = aiv1_kernel_id; + if (fill_deferred_submit_metadata(orch, args, &record)) { + enqueue_deferred_submit_record(orch, record); + CYCLE_COUNT_LAP(g_orch_args_cycle); + CYCLE_COUNT_ORCH_SUBMIT_RECORD(task_id.raw); +#if PTO2_PROFILING + orch->tasks_submitted++; +#if PTO2_ORCH_PROFILING + g_orch_submit_count++; +#endif + g_orch_submit_idx++; +#endif + return result; + } + } + + PTO2FaninBuilder fanin_builder(orch, orch->rings[ring_id].fanin_pool, next_fanin_seen_epoch(orch)); + // === STEP 2: Sync TensorMap validity and optional cleanup === // Read current last_task_alive from shared memory for this ring int32_t sm_last_task_alive = fc.last_task_alive.load(std::memory_order_acquire); diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.h b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.h index 5ceb9af85..2d55c86cb 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.h +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.h @@ -28,6 +28,8 @@ #ifndef PTO_ORCHESTRATOR_H #define PTO_ORCHESTRATOR_H +#include + #include "common/l2_swimlane_profiling.h" #include "utils/device_arena.h" #include "pto_ring_buffer.h" @@ -54,6 +56,42 @@ struct PTO2OrchestratorLayout { uint64_t scope_stack_capacity; }; +constexpr int32_t PTO2_SUBMIT_PIPELINE_QUEUE_CAP = 64; +constexpr int32_t PTO2_SUBMIT_PIPELINE_EXPLICIT_DEP_CAP = 64; +constexpr int32_t PTO2_SUBMIT_PIPELINE_SCOPE_INLINE_CAP = 64; +constexpr int32_t PTO2_SUBMIT_PIPELINE_MAX_COMMIT_STAGES = 1; +constexpr int32_t PTO2_SUBMIT_PIPELINE_CONTROL_IDLE = 0; +constexpr int32_t PTO2_SUBMIT_PIPELINE_CONTROL_WORK = 1; +constexpr int32_t PTO2_SUBMIT_PIPELINE_CONTROL_STOP = 2; +constexpr int32_t PTO2_SUBMIT_PIPELINE_HEAP_GUARD_OUTPUT_BYTES = 64 * 1024; + +enum class PTO2SubmitPipelineRecordKind : int32_t { + TASK_DEFERRED = 0, + SCOPE_END = 1, +}; + +struct PTO2SubmitCommitRecord { + PTO2SubmitPipelineRecordKind kind{PTO2SubmitPipelineRecordKind::TASK_DEFERRED}; + PTO2TaskPayload *payload{nullptr}; + PTO2TaskSlotState *slot_state{nullptr}; + PTO2SchedulerState *scheduler{nullptr}; + PTO2TaskId task_id = PTO2TaskId::invalid(); + PTO2TaskId explicit_deps[PTO2_SUBMIT_PIPELINE_EXPLICIT_DEP_CAP]{}; + PTO2TaskSlotState *scope_task_slot_states[PTO2_SUBMIT_PIPELINE_SCOPE_INLINE_CAP]{}; + TensorArgType arg_types[MAX_TENSOR_ARGS]{}; + int32_t explicit_dep_count{0}; + int32_t scope_task_count{0}; + int32_t kernel_id[PTO2_SUBTASK_SLOT_COUNT]{INVALID_KERNEL_ID, INVALID_KERNEL_ID, INVALID_KERNEL_ID}; + bool in_manual_scope{false}; +}; + +struct PTO2SubmitPipelineQueue { + std::atomic tail{0}; + std::atomic head{0}; + std::atomic slot_state[PTO2_SUBMIT_PIPELINE_QUEUE_CAP]; + PTO2SubmitCommitRecord records[PTO2_SUBMIT_PIPELINE_QUEUE_CAP]; +}; + // ============================================================================= // Orchestrator State // ============================================================================= @@ -92,6 +130,41 @@ struct PTO2OrchestratorState { // In real mode, they communicate via shared memory only PTO2SchedulerState *scheduler; // For simulated mode only + // === SUBMIT COMMIT PIPELINE === + // Strategy2 uses a second orchestrator worker to commit dependency lookup, + // TensorMap registration, and ready/wiring publication after O1 has + // materialized the task payload returned to orchestration. + bool submit_pipeline_enabled{false}; + bool submit_pipeline_defer_dependencies{false}; + bool submit_pipeline_signal_scheduler_drain{false}; + bool submit_pipeline_compact_deferred_records{false}; + int32_t submit_pipeline_commit_stages{0}; + std::atomic submit_pipeline_stop{false}; + std::atomic submit_pipeline_work_available{false}; + std::atomic submit_pipeline_control{PTO2_SUBMIT_PIPELINE_CONTROL_IDLE}; + std::atomic submit_pipeline_completed{0}; + std::atomic submit_pipeline_stage_done[PTO2_SUBMIT_PIPELINE_MAX_COMMIT_STAGES]; + PTO2SubmitPipelineQueue submit_pipeline_queues[PTO2_SUBMIT_PIPELINE_MAX_COMMIT_STAGES]; +#if PTO2_PROFILING + uint64_t submit_pipeline_task_enqueue_cycles{0}; + uint64_t submit_pipeline_scope_enqueue_cycles{0}; + uint64_t submit_pipeline_flush_cycles{0}; + uint64_t submit_pipeline_deferred_dep_cycles{0}; + uint64_t submit_pipeline_deferred_fanin_cycles{0}; + uint64_t submit_pipeline_publish_cycles{0}; + uint64_t submit_pipeline_scope_release_cycles{0}; + uint64_t submit_pipeline_publish_spins{0}; + uint32_t submit_pipeline_scheduler_drain_hint_count{0}; + uint32_t submit_pipeline_task_enqueue_count{0}; + uint32_t submit_pipeline_scope_enqueue_count{0}; + uint32_t submit_pipeline_flush_count{0}; + uint32_t submit_pipeline_deferred_commit_count{0}; + uint32_t submit_pipeline_scope_record_count{0}; + uint32_t submit_pipeline_dep_explicit_count{0}; + uint32_t submit_pipeline_dep_register_count{0}; + uint32_t submit_pipeline_dep_fanin_actual_count{0}; +#endif + // Total core counts set once at executor init; used for submit-time deadlock detection. int32_t total_cluster_count{0}; // AIC cores = MIX clusters int32_t total_aiv_count{0}; // AIV cores (= 2 × clusters on standard hardware) @@ -178,6 +251,21 @@ struct PTO2OrchestratorState { void report_fatal(int32_t error_code, const char *func, const char *fmt, ...); void begin_scope(PTO2ScopeMode mode = PTO2ScopeMode::AUTO); void end_scope(); + void enable_submit_pipeline( + int32_t orchestrator_threads, bool enqueue_submit_records = false, bool defer_submit_dependencies = false, + bool signal_scheduler_drain = false, bool compact_deferred_records = false + ); + uint64_t run_submit_pipeline_worker(int32_t stage_idx, int32_t phase_thread_idx = -1); + void flush_submit_task_batch(); + void flush_submit_pipeline(); + void log_submit_pipeline_diagnostics(int32_t thread_idx) const; + void log_four_stage_diagnostics(int32_t thread_idx) const; + void log_active_detail_diagnostics( + int32_t thread_idx, uint64_t active_cycles, uint64_t bind_cycles, uint64_t p_bind_cycles, + uint64_t outer_scope_begin_cycles, uint64_t p_func_cycles, uint64_t outer_scope_end_cycles + ) const; + void log_submit_detail_diagnostics(int32_t thread_idx) const; + void stop_submit_pipeline(); TaskOutputTensors submit_task(const MixedKernels &mixed_kernels, const L0TaskArgs &args); TaskOutputTensors submit_dummy_task(const L0TaskArgs &args); TaskOutputTensors alloc_tensors(const L0TaskArgs &args); 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 a37eb0d43..6e6711167 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 @@ -846,6 +846,28 @@ struct PTO2SchedulerState { } } + uint64_t publish_ready_no_fanin(PTO2TaskSlotState *slot_state) { + slot_state->fanin_count = 1; + slot_state->fanin_refcount.store(1, std::memory_order_release); + slot_state->dep_pool_mark = 0; + + uint64_t spin_count = 0; + PTO2ResourceShape shape = slot_state->active_mask.to_shape(); + if (shape == PTO2ResourceShape::DUMMY) { + while (!dummy_ready_queue.push(slot_state)) { + spin_count++; + SPIN_WAIT_HINT(); + } + } else { + auto &ready_queue = ready_queues[static_cast(shape)]; + while (!ready_queue.push(slot_state)) { + spin_count++; + SPIN_WAIT_HINT(); + } + } + return spin_count; + } + /** * Wire fanout edges for a single task. Sets fanin_count, acquires each * producer's fanout_lock, allocates dep_pool entries for live producers, diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/shared/pto_runtime2_init.cpp b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/shared/pto_runtime2_init.cpp index 1561acc56..c66313036 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/shared/pto_runtime2_init.cpp +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/shared/pto_runtime2_init.cpp @@ -23,6 +23,7 @@ #include #include +#include #include "pto_orchestrator.h" #include "pto_runtime2.h" @@ -337,7 +338,7 @@ bool PTO2OrchestratorState::init_data_from_layout( const uint64_t heap_sizes[PTO2_MAX_RING_DEPTH], const uint64_t task_window_sizes[PTO2_MAX_RING_DEPTH] ) { auto *orch = this; - *orch = PTO2OrchestratorState{}; + new (orch) PTO2OrchestratorState(); orch->sm_header = reinterpret_cast(sm_dev_base); orch->gm_heap_base = gm_heap;