diff --git a/conftest.py b/conftest.py index 924e29cf3..1a20887ec 100644 --- a/conftest.py +++ b/conftest.py @@ -165,6 +165,13 @@ def pytest_addoption(parser): default=False, help="Enable dep_gen capture (SubmitTrace ring, first round only)", ) + parser.addoption( + "--enable-dfx-backpressure", + action="store_true", + default=False, + help="DFX backpressure: block-on-contention instead of dropping records " + "(currently wired for L2 swimlane). Requires a diagnostic (e.g. --enable-l2-swimlane).", + ) parser.addoption( "--enable-pmu", nargs="?", diff --git a/python/bindings/task_interface.cpp b/python/bindings/task_interface.cpp index 43c2e6e9c..eb1434829 100644 --- a/python/bindings/task_interface.cpp +++ b/python/bindings/task_interface.cpp @@ -780,6 +780,15 @@ NB_MODULE(_task_interface, m) { c.enable_scope_stats = v ? 1 : 0; } ) + .def_prop_rw( + "enable_dfx_backpressure", + [](const CallConfig &c) { + return static_cast(c.enable_dfx_backpressure); + }, + [](CallConfig &c, bool v) { + c.enable_dfx_backpressure = v ? 1 : 0; + } + ) .def_prop_rw( "output_prefix", [](const CallConfig &c) -> std::string { @@ -802,7 +811,8 @@ NB_MODULE(_task_interface, m) { << ", 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"); + << ", enable_scope_stats=" << (self.enable_scope_stats ? "True" : "False") + << ", enable_dfx_backpressure=" << (self.enable_dfx_backpressure ? "True" : "False"); if (self.runtime_env.any()) { append_ring_values(os, "runtime_env.ring_task_window", true, self.runtime_env.ring_task_window); append_ring_values(os, "runtime_env.ring_heap", true, self.runtime_env.ring_heap); diff --git a/python/simpler/worker.py b/python/simpler/worker.py index c0ebba1d0..c14a077cb 100644 --- a/python/simpler/worker.py +++ b/python/simpler/worker.py @@ -141,13 +141,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 -# 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. +# 8 int32 (block_dim, aicpu_thread_num, enable_l2_swimlane, enable_dump_tensor, +# enable_pmu, enable_dep_gen, enable_scope_stats, enable_dfx_backpressure) + +# 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). @@ -1399,6 +1400,7 @@ def _read_config_from_mailbox(buf: memoryview) -> CallConfig: pmu, dep_gen, scope_stats, + dfx_backpressure, *ring_values, prefix_bytes, ) = _CFG_FMT.unpack_from(buf, _OFF_CONFIG) @@ -1413,6 +1415,7 @@ def _read_config_from_mailbox(buf: memoryview) -> CallConfig: cfg.enable_pmu = pmu cfg.enable_dep_gen = bool(dep_gen) cfg.enable_scope_stats = bool(scope_stats) + cfg.enable_dfx_backpressure = bool(dfx_backpressure) cfg.runtime_env.ring_task_window = ring_task_window cfg.runtime_env.ring_heap = ring_heap cfg.runtime_env.ring_dep_pool = ring_dep_pool diff --git a/simpler_setup/scene_test.py b/simpler_setup/scene_test.py index f7ba66bad..30ccf4dd7 100644 --- a/simpler_setup/scene_test.py +++ b/simpler_setup/scene_test.py @@ -682,6 +682,7 @@ def run_class_cases( # noqa: PLR0913 -- shared layer-5 entry; kwargs mirror CLI enable_pmu, enable_dep_gen, enable_scope_stats, + enable_dfx_backpressure=False, enable_swimlane_overhead=False, ): """Execute a pre-filtered list of cases for one class (layers 5-6). @@ -714,6 +715,7 @@ def run_class_cases( # noqa: PLR0913 -- shared layer-5 entry; kwargs mirror CLI enable_pmu=enable_pmu, enable_dep_gen=enable_dep_gen, enable_scope_stats=enable_scope_stats, + enable_dfx_backpressure=enable_dfx_backpressure, output_prefix=str(prefix) if diagnostics_on else "", ) finally: @@ -902,6 +904,7 @@ def _build_config( enable_pmu=0, enable_dep_gen=False, enable_scope_stats=False, + enable_dfx_backpressure=False, *, output_prefix="", ): @@ -928,6 +931,7 @@ def _build_config( config.enable_pmu = enable_pmu # 0=disabled, >0=enabled with event type config.enable_dep_gen = enable_dep_gen config.enable_scope_stats = enable_scope_stats + config.enable_dfx_backpressure = enable_dfx_backpressure # `output_prefix` is required by CallConfig::validate() whenever any # diagnostic flag is enabled. Caller threads it down from the per-case # directory built by _build_output_prefix(). @@ -965,6 +969,7 @@ def _run_and_validate( # noqa: PLR0913 -- threads CLI diagnostic flags + case c enable_pmu=0, enable_dep_gen=False, enable_scope_stats=False, + enable_dfx_backpressure=False, output_prefix="", ): if self._st_level == 2: @@ -979,6 +984,7 @@ def _run_and_validate( # noqa: PLR0913 -- threads CLI diagnostic flags + case c enable_pmu=enable_pmu, enable_dep_gen=enable_dep_gen, enable_scope_stats=enable_scope_stats, + enable_dfx_backpressure=enable_dfx_backpressure, output_prefix=output_prefix, ) elif self._st_level == 3: @@ -994,6 +1000,7 @@ def _run_and_validate( # noqa: PLR0913 -- threads CLI diagnostic flags + case c enable_pmu=enable_pmu, enable_dep_gen=enable_dep_gen, enable_scope_stats=enable_scope_stats, + enable_dfx_backpressure=enable_dfx_backpressure, output_prefix=output_prefix, ) @@ -1009,6 +1016,7 @@ def _run_and_validate_l2( # noqa: PLR0913 -- threads CLI diagnostic flags + cas enable_pmu=0, enable_dep_gen=False, enable_scope_stats=False, + enable_dfx_backpressure=False, output_prefix="", ): params = case.get("params", {}) @@ -1061,6 +1069,7 @@ def _run_and_validate_l2( # noqa: PLR0913 -- threads CLI diagnostic flags + cas enable_pmu=enable_pmu, enable_dep_gen=enable_dep_gen, enable_scope_stats=enable_scope_stats, + enable_dfx_backpressure=enable_dfx_backpressure, output_prefix=output_prefix, ) @@ -1083,6 +1092,7 @@ def _run_and_validate_l3( # noqa: PLR0913 -- threads CLI diagnostic flags + L3 enable_pmu=0, enable_dep_gen=False, enable_scope_stats=False, + enable_dfx_backpressure=False, output_prefix="", ): # Defensive belt-and-braces: the pytest dispatcher and run_module both @@ -1141,6 +1151,7 @@ def _run_and_validate_l3( # noqa: PLR0913 -- threads CLI diagnostic flags + L3 enable_pmu=enable_pmu, enable_dep_gen=enable_dep_gen, enable_scope_stats=enable_scope_stats, + enable_dfx_backpressure=enable_dfx_backpressure, output_prefix=output_prefix, ) @@ -1188,6 +1199,7 @@ def test_run(self, st_platform, st_worker, request): enable_pmu = request.config.getoption("--enable-pmu", default=0) enable_dep_gen = self._effective_enable_dep_gen(request, warn=True) enable_scope_stats = request.config.getoption("--enable-scope-stats", default=False) + enable_dfx_backpressure = request.config.getoption("--enable-dfx-backpressure", default=False) enable_swimlane_overhead = request.config.getoption("--enable-swimlane-overhead", default=False) if rounds > 1: if enable_l2_swimlane: @@ -1243,6 +1255,7 @@ def test_run(self, st_platform, st_worker, request): enable_pmu=enable_pmu, enable_dep_gen=enable_dep_gen, enable_scope_stats=enable_scope_stats, + enable_dfx_backpressure=enable_dfx_backpressure, enable_swimlane_overhead=enable_swimlane_overhead, ) diff --git a/src/a2a3/platform/include/common/dep_gen.h b/src/a2a3/platform/include/common/dep_gen.h index fb89a3d51..3328ed413 100644 --- a/src/a2a3/platform/include/common/dep_gen.h +++ b/src/a2a3/platform/include/common/dep_gen.h @@ -45,6 +45,7 @@ #include #include "arg_direction.h" // CORE_MAX_TENSOR_ARGS +#include "common/dfx_backpressure_device.h" #include "common/platform_config.h" // ============================================================================= @@ -300,7 +301,8 @@ struct DepGenDataHeader { volatile uint32_t queue_heads[PLATFORM_MAX_AICPU_THREADS]; // Host reads (consumer) volatile uint32_t queue_tails[PLATFORM_MAX_AICPU_THREADS]; // AICPU writes (producer) uint32_t num_instances; // Always 1 for now - uint32_t _pad[3]; + // DFX backpressure coordination (unified across all DFX subsystems). + DfxBackpressureHeader backpressure; } __attribute__((aligned(64))); // ============================================================================= diff --git a/src/a2a3/platform/include/common/l2_swimlane_profiling.h b/src/a2a3/platform/include/common/l2_swimlane_profiling.h index b552df3f3..5d5930312 100644 --- a/src/a2a3/platform/include/common/l2_swimlane_profiling.h +++ b/src/a2a3/platform/include/common/l2_swimlane_profiling.h @@ -60,6 +60,7 @@ #include #include "common/core_type.h" +#include "common/dfx_backpressure_device.h" #include "common/platform_config.h" // ============================================================================= @@ -433,6 +434,9 @@ struct L2SwimlaneDataHeader { uint32_t num_orch_phase_threads; // Number of orch-phase pools the AICPU initialized uint32_t num_phase_cores; // Number of valid entries in core_to_thread (0 = unset) int8_t core_to_thread[PLATFORM_MAX_CORES]; // core_id → scheduler thread index (-1 = unassigned) + + // DFX backpressure coordination (unified across all DFX subsystems). + DfxBackpressureHeader backpressure; } __attribute__((aligned(64))); // ABI lock for the merged header. The phase metadata fields and the diff --git a/src/a2a3/platform/include/common/platform_config.h b/src/a2a3/platform/include/common/platform_config.h index 78c7a88b9..3aa1bc110 100644 --- a/src/a2a3/platform/include/common/platform_config.h +++ b/src/a2a3/platform/include/common/platform_config.h @@ -139,7 +139,7 @@ constexpr int PLATFORM_PROF_BUFFERS_PER_CORE = 8; * Mirrors PLATFORM_PROF_BUFFERS_PER_CORE in role; smaller because AICore records * are slim (32 B each) and the buffer is also smaller per the rotation design. */ -constexpr int PLATFORM_AICORE_BUFFERS_PER_CORE = 4; +constexpr int PLATFORM_AICORE_BUFFERS_PER_CORE = 5; // >= SLOT_COUNT + 1 (DFX release floor) /** * Host preallocation count per AICPU thread for the two phase pools, split per @@ -155,6 +155,29 @@ constexpr int PLATFORM_AICORE_BUFFERS_PER_CORE = 4; constexpr int PLATFORM_PROF_SCHED_BUFFERS_PER_THREAD = 6; constexpr int PLATFORM_PROF_ORCH_BUFFERS_PER_THREAD = 8; +// DFX global-sync backpressure release floor: every pool that opts +// into block-on-contention must hold BUFFERS >= SLOT_COUNT + 1, so a freeze can +// reach "free_queue full (== SLOT_COUNT)" and release — a lane always holds 1 +// active buffer, capping the free_queue at BUFFERS - 1. Enforced at compile time +// so a capacity tweak below the floor fails the build instead of hanging a +// freeze forever at runtime. +static_assert( + PLATFORM_PROF_BUFFERS_PER_CORE >= PLATFORM_PROF_SLOT_COUNT + 1, + "PLATFORM_PROF_BUFFERS_PER_CORE must be >= PLATFORM_PROF_SLOT_COUNT + 1 (DFX backpressure release floor)" +); +static_assert( + PLATFORM_AICORE_BUFFERS_PER_CORE >= PLATFORM_PROF_SLOT_COUNT + 1, + "PLATFORM_AICORE_BUFFERS_PER_CORE must be >= PLATFORM_PROF_SLOT_COUNT + 1 (DFX backpressure release floor)" +); +static_assert( + PLATFORM_PROF_SCHED_BUFFERS_PER_THREAD >= PLATFORM_PROF_SLOT_COUNT + 1, + "PLATFORM_PROF_SCHED_BUFFERS_PER_THREAD must be >= PLATFORM_PROF_SLOT_COUNT + 1 (DFX backpressure release floor)" +); +static_assert( + PLATFORM_PROF_ORCH_BUFFERS_PER_THREAD >= PLATFORM_PROF_SLOT_COUNT + 1, + "PLATFORM_PROF_ORCH_BUFFERS_PER_THREAD must be >= PLATFORM_PROF_SLOT_COUNT + 1 (DFX backpressure release floor)" +); + /** * Ready queue capacity for performance data collection. * Queue holds ReadyQueueEntry structs for buffers ready to be read by Host. @@ -223,6 +246,14 @@ constexpr int PLATFORM_DUMP_BUFFERS_PER_THREAD = 8; */ constexpr int PLATFORM_DUMP_SLOT_COUNT = 4; +// DFX backpressure release floor — see the swimlane block above. +// Only tensor_dump's METADATA channel opts into block-on-contention; the payload +// arena is separate (circular overwrite, no free_queue) and carries no floor. +static_assert( + PLATFORM_DUMP_BUFFERS_PER_THREAD >= PLATFORM_DUMP_SLOT_COUNT + 1, + "PLATFORM_DUMP_BUFFERS_PER_THREAD must be >= PLATFORM_DUMP_SLOT_COUNT + 1 (DFX backpressure release floor)" +); + /** * Expected average tensor size in bytes. * Used together with BUFFERS_PER_THREAD and RECORDS_PER_BUFFER to compute @@ -270,7 +301,13 @@ constexpr int PLATFORM_PMU_SLOT_COUNT = 4; * borrowed). Was 4 (a 4× over-provision). See * docs/dfx/dfx-buffer-capacity-audit.md. */ -constexpr int PLATFORM_PMU_BUFFERS_PER_CORE = 2; +constexpr int PLATFORM_PMU_BUFFERS_PER_CORE = 5; // >= SLOT_COUNT + 1 (DFX release floor) + +// DFX backpressure release floor — see the swimlane block above. +static_assert( + PLATFORM_PMU_BUFFERS_PER_CORE >= PLATFORM_PMU_SLOT_COUNT + 1, + "PLATFORM_PMU_BUFFERS_PER_CORE must be >= PLATFORM_PMU_SLOT_COUNT + 1 (DFX backpressure release floor)" +); /** * Ready queue capacity for PMU data collection. @@ -333,7 +370,13 @@ constexpr int PLATFORM_DEP_GEN_SLOT_COUNT = 4; * device is bounded by SLOT_COUNT, and the drop is rate-bound anyway) — see the * RECORDS_PER_BUFFER comment. dep_gen is opt-in (--enable-dep-gen). */ -constexpr int PLATFORM_DEP_GEN_BUFFERS_PER_INSTANCE = 4; +constexpr int PLATFORM_DEP_GEN_BUFFERS_PER_INSTANCE = 5; // >= SLOT_COUNT + 1 (DFX release floor) + +// DFX backpressure release floor — see the swimlane block above. +static_assert( + PLATFORM_DEP_GEN_BUFFERS_PER_INSTANCE >= PLATFORM_DEP_GEN_SLOT_COUNT + 1, + "PLATFORM_DEP_GEN_BUFFERS_PER_INSTANCE must be >= PLATFORM_DEP_GEN_SLOT_COUNT + 1 (DFX backpressure release floor)" +); /** * Ready queue capacity for dep_gen (per AICPU thread). dep_gen is single- @@ -361,7 +404,7 @@ constexpr int PLATFORM_SCOPE_STATS_RECORDS_PER_BUFFER = 512; /** * SPSC free_queue slot count for scope_stats buffers (Host→Device hand-off depth). */ -constexpr int PLATFORM_SCOPE_STATS_SLOT_COUNT = 8; +constexpr int PLATFORM_SCOPE_STATS_SLOT_COUNT = 4; // aligned with other DFX subsystems /** * Pre-allocated ScopeStatsBuffer count per orchestrator instance. @@ -371,7 +414,14 @@ constexpr int PLATFORM_SCOPE_STATS_SLOT_COUNT = 8; * be borrowed). Was 8 (an 8× over-provision). See * docs/dfx/dfx-buffer-capacity-audit.md. */ -constexpr int PLATFORM_SCOPE_STATS_BUFFERS_PER_INSTANCE = 4; +constexpr int PLATFORM_SCOPE_STATS_BUFFERS_PER_INSTANCE = 5; // >= SLOT_COUNT + 1 (DFX release floor) + +// DFX backpressure release floor — see the swimlane block above. +static_assert( + PLATFORM_SCOPE_STATS_BUFFERS_PER_INSTANCE >= PLATFORM_SCOPE_STATS_SLOT_COUNT + 1, + "PLATFORM_SCOPE_STATS_BUFFERS_PER_INSTANCE must be >= PLATFORM_SCOPE_STATS_SLOT_COUNT + 1 (DFX backpressure " + "release floor)" +); /** * Ready queue capacity for scope_stats (per AICPU thread). scope_stats is diff --git a/src/a2a3/platform/include/common/pmu_profiling.h b/src/a2a3/platform/include/common/pmu_profiling.h index a5e2baaa3..79511b071 100644 --- a/src/a2a3/platform/include/common/pmu_profiling.h +++ b/src/a2a3/platform/include/common/pmu_profiling.h @@ -31,6 +31,7 @@ #include #include "common/core_type.h" +#include "common/dfx_backpressure_device.h" #include "common/platform_config.h" // DAV_2201 hardware counter count. @@ -228,7 +229,8 @@ struct PmuDataHeader { volatile uint32_t queue_tails[PLATFORM_MAX_AICPU_THREADS]; // AICPU writes (producer) uint32_t num_cores; uint32_t event_type; // PmuEventType value, written by host at init - uint32_t pad[2]; + // DFX backpressure coordination (unified across all DFX subsystems). + DfxBackpressureHeader backpressure; } __attribute__((aligned(64))); // ============================================================================= diff --git a/src/a2a3/platform/include/common/tensor_dump.h b/src/a2a3/platform/include/common/tensor_dump.h index 3be0952f1..06d471f7d 100644 --- a/src/a2a3/platform/include/common/tensor_dump.h +++ b/src/a2a3/platform/include/common/tensor_dump.h @@ -43,6 +43,7 @@ #include #include +#include "common/dfx_backpressure_device.h" #include "common/platform_config.h" #include "host_build_graph/runtime/pto_runtime2_types.h" @@ -191,8 +192,9 @@ struct DumpBufferState { volatile uint64_t arena_base; // Device pointer to this thread's arena volatile uint64_t arena_size; // Arena size in bytes volatile uint64_t arena_write_offset; // Monotonic write cursor (host computes % arena_size) + volatile uint64_t epoch_start_offset; // DFX arena barrier: write cursor at start of current epoch volatile uint32_t dropped_record_count; // Records dropped before host export - uint8_t pad1[28]; // Pad to 256 bytes + uint8_t pad1[20]; // Pad to 256 bytes } __attribute__((aligned(64))); static_assert(sizeof(DumpBufferState) == 256, "DumpBufferState must be 256 bytes"); @@ -253,6 +255,8 @@ struct DumpDataHeader { uint64_t arena_size_per_thread; uint32_t magic; uint32_t dump_tensor_level; // DumpTensorLevel: 0=off, 1=partial, 2=full, 3=full_json_only + // DFX backpressure coordination (unified across all DFX subsystems). + DfxBackpressureHeader backpressure; } __attribute__((aligned(64))); // ============================================================================= diff --git a/src/a2a3/platform/include/host/pmu_collector.h b/src/a2a3/platform/include/host/pmu_collector.h index b1742790a..831c4c07c 100644 --- a/src/a2a3/platform/include/host/pmu_collector.h +++ b/src/a2a3/platform/include/host/pmu_collector.h @@ -206,7 +206,7 @@ class PmuCollector : public profiling_common::ProfilerBaseevent_type = static_cast(event_type); hdr->num_cores = static_cast(num_cores); + hdr->backpressure.block_on_contention = + static_cast(block_on_contention); // freeze_active/contended stay 0 (memset above) // ---- Allocate per-core PmuBuffers and populate free_queues + recycled pool ---- const size_t buf_size = sizeof(PmuBuffer); diff --git a/src/a2a3/platform/sim/host/device_runner.cpp b/src/a2a3/platform/sim/host/device_runner.cpp index 7461ed096..b6b6fdd98 100644 --- a/src/a2a3/platform/sim/host/device_runner.cpp +++ b/src/a2a3/platform/sim/host/device_runner.cpp @@ -729,7 +729,8 @@ int DeviceRunner::init_l2_swimlane(int num_aicore, int aicpu_thread_num, int dev }; int rc = l2_swimlane_collector_.initialize( - num_aicore, aicpu_thread_num, device_id, l2_swimlane_level_, alloc_cb, nullptr, free_cb, output_prefix_ + num_aicore, aicpu_thread_num, device_id, l2_swimlane_level_, dfx_backpressure_, alloc_cb, nullptr, free_cb, + output_prefix_ ); if (rc != 0) { return rc; @@ -753,7 +754,7 @@ int DeviceRunner::init_tensor_dump(Runtime &runtime, int device_id) { }; int rc = dump_collector_.initialize( - num_dump_threads, device_id, alloc_cb, nullptr, free_cb, output_prefix_, dump_tensor_level_ + num_dump_threads, device_id, dfx_backpressure_, alloc_cb, nullptr, free_cb, output_prefix_, dump_tensor_level_ ); if (rc != 0) { return rc; @@ -773,7 +774,9 @@ int DeviceRunner::init_pmu( return mem_alloc_.free(dev_ptr); }; - int rc = pmu_collector_.init(num_cores, num_threads, csv_path, event_type, alloc_cb, nullptr, free_cb, -1); + int rc = pmu_collector_.init( + num_cores, num_threads, csv_path, event_type, dfx_backpressure_, alloc_cb, nullptr, free_cb, -1 + ); if (rc != 0) { return rc; } @@ -790,7 +793,7 @@ int DeviceRunner::init_dep_gen(int num_threads, int /*device_id*/) { return mem_alloc_.free(dev_ptr); }; - int rc = dep_gen_collector_.init(num_threads, alloc_cb, nullptr, free_cb, -1); + int rc = dep_gen_collector_.init(num_threads, dfx_backpressure_, alloc_cb, nullptr, free_cb, -1); if (rc != 0) { return rc; } @@ -807,7 +810,9 @@ int DeviceRunner::init_scope_stats(int num_threads) { return mem_alloc_.free(dev_ptr); }; - int rc = scope_stats_collector_.init(num_threads, alloc_cb, /*register_cb=*/nullptr, free_cb, /*device_id=*/-1); + int rc = scope_stats_collector_.init( + num_threads, dfx_backpressure_, alloc_cb, /*register_cb=*/nullptr, free_cb, /*device_id=*/-1 + ); if (rc != 0) { return rc; } diff --git a/src/a5/platform/include/common/dep_gen.h b/src/a5/platform/include/common/dep_gen.h index d73518cf8..3122ebe60 100644 --- a/src/a5/platform/include/common/dep_gen.h +++ b/src/a5/platform/include/common/dep_gen.h @@ -45,6 +45,7 @@ #include #include "arg_direction.h" // CORE_MAX_TENSOR_ARGS +#include "common/dfx_backpressure_device.h" #include "common/platform_config.h" // ============================================================================= @@ -300,6 +301,8 @@ struct DepGenDataHeader { volatile uint32_t queue_heads[PLATFORM_MAX_AICPU_THREADS]; // Host reads (consumer) volatile uint32_t queue_tails[PLATFORM_MAX_AICPU_THREADS]; // AICPU writes (producer) uint32_t num_instances; // Always 1 for now + // DFX backpressure coordination (unified across all DFX subsystems). + DfxBackpressureHeader backpressure; uint32_t _pad[3]; } __attribute__((aligned(64))); diff --git a/src/a5/platform/include/common/l2_swimlane_profiling.h b/src/a5/platform/include/common/l2_swimlane_profiling.h index a9cbffa6f..884ed38a2 100644 --- a/src/a5/platform/include/common/l2_swimlane_profiling.h +++ b/src/a5/platform/include/common/l2_swimlane_profiling.h @@ -60,6 +60,7 @@ #include #include "common/core_type.h" +#include "common/dfx_backpressure_device.h" #include "common/platform_config.h" // ============================================================================= @@ -433,6 +434,9 @@ struct L2SwimlaneDataHeader { uint32_t num_orch_phase_threads; // Number of orch-phase pools the AICPU initialized uint32_t num_phase_cores; // Number of valid entries in core_to_thread (0 = unset) int8_t core_to_thread[PLATFORM_MAX_CORES]; // core_id → scheduler thread index (-1 = unassigned) + + // DFX backpressure coordination (unified across all DFX subsystems). + DfxBackpressureHeader backpressure; } __attribute__((aligned(64))); // ABI lock for the merged header. The phase metadata fields and the diff --git a/src/a5/platform/include/common/pmu_profiling.h b/src/a5/platform/include/common/pmu_profiling.h index 680c81c83..4bae66ae9 100644 --- a/src/a5/platform/include/common/pmu_profiling.h +++ b/src/a5/platform/include/common/pmu_profiling.h @@ -35,6 +35,7 @@ #include #include "common/core_type.h" +#include "common/dfx_backpressure_device.h" #include "common/platform_config.h" /** @@ -267,6 +268,8 @@ struct PmuDataHeader { volatile uint32_t queue_tails[PLATFORM_MAX_AICPU_THREADS]; // AICPU writes (producer) uint32_t num_cores; uint32_t event_type; // PmuEventType value, written by host at init + // DFX backpressure coordination (unified across all DFX subsystems). + DfxBackpressureHeader backpressure; uint32_t pad[2]; } __attribute__((aligned(64))); diff --git a/src/a5/platform/include/common/tensor_dump.h b/src/a5/platform/include/common/tensor_dump.h index dd68fbe87..89244bd97 100644 --- a/src/a5/platform/include/common/tensor_dump.h +++ b/src/a5/platform/include/common/tensor_dump.h @@ -47,6 +47,7 @@ #include #include +#include "common/dfx_backpressure_device.h" #include "common/platform_config.h" #include "host_build_graph/runtime/pto_runtime2_types.h" @@ -195,8 +196,9 @@ struct DumpBufferState { volatile uint64_t arena_base; // Device pointer to this thread's arena volatile uint64_t arena_size; // Arena size in bytes volatile uint64_t arena_write_offset; // Monotonic write cursor (host computes % arena_size) + volatile uint64_t epoch_start_offset; // DFX arena barrier: write cursor at start of current epoch volatile uint32_t dropped_record_count; // Records dropped before host export - uint8_t pad1[84]; // Pad to 256 bytes (172 + 84 = 256) + uint8_t pad1[76]; // Pad to 256 bytes (180 + 76 = 256) } __attribute__((aligned(64))); static_assert(sizeof(DumpBufferState) == 256, "DumpBufferState must be 256 bytes"); @@ -257,6 +259,8 @@ struct DumpDataHeader { uint64_t arena_size_per_thread; uint32_t magic; uint32_t dump_tensor_level; // DumpTensorLevel: 0=off, 1=partial, 2=full, 3=full_json_only + // DFX backpressure coordination (unified across all DFX subsystems). + DfxBackpressureHeader backpressure; } __attribute__((aligned(64))); // ============================================================================= diff --git a/src/a5/platform/include/host/pmu_collector.h b/src/a5/platform/include/host/pmu_collector.h index b42467aa1..b54637765 100644 --- a/src/a5/platform/include/host/pmu_collector.h +++ b/src/a5/platform/include/host/pmu_collector.h @@ -214,7 +214,7 @@ class PmuCollector : public profiling_common::ProfilerBase(pmu_collector_.get_pmu_shm_device_ptr()); @@ -802,7 +803,9 @@ int DeviceRunner::init_scope_stats(int num_threads, int device_id) { // a5: register_cb=nullptr, so the collector mallocs a host shadow per // device buffer + rtMemcpy's the zeroed shadow to device (see // ProfilerBase::alloc_paired_buffer). No halHostRegister on a5. - int rc = scope_stats_collector_.init(num_threads, prof_alloc_cb, /*register_cb=*/nullptr, prof_free_cb, device_id); + int rc = scope_stats_collector_.init( + num_threads, dfx_backpressure_, prof_alloc_cb, /*register_cb=*/nullptr, prof_free_cb, device_id + ); if (rc != 0) { return rc; } @@ -815,7 +818,9 @@ int DeviceRunner::init_dep_gen(int num_threads, int device_id) { // a5: register_cb=nullptr, so the collector mallocs a host shadow per // device buffer + rtMemcpy's the zeroed shadow to device. No // halHostRegister on a5 (matches PMU / L2 swimlane / dump collectors). - int rc = dep_gen_collector_.init(num_threads, prof_alloc_cb, /*register_cb=*/nullptr, prof_free_cb, device_id); + int rc = dep_gen_collector_.init( + num_threads, dfx_backpressure_, prof_alloc_cb, /*register_cb=*/nullptr, prof_free_cb, device_id + ); if (rc != 0) { return rc; } diff --git a/src/a5/platform/shared/host/pmu_collector.cpp b/src/a5/platform/shared/host/pmu_collector.cpp index 8f04b2221..a6499b410 100644 --- a/src/a5/platform/shared/host/pmu_collector.cpp +++ b/src/a5/platform/shared/host/pmu_collector.cpp @@ -43,7 +43,7 @@ PmuCollector::~PmuCollector() { stop(); } // --------------------------------------------------------------------------- int PmuCollector::init( - int num_cores, int num_threads, const std::string &csv_path, PmuEventType event_type, + int num_cores, int num_threads, const std::string &csv_path, PmuEventType event_type, bool block_on_contention, const PmuAllocCallback &alloc_cb, PmuRegisterCallback register_cb, const PmuFreeCallback &free_cb, int device_id ) { if (num_cores <= 0 || num_threads <= 0 || alloc_cb == nullptr || free_cb == nullptr) { @@ -94,6 +94,8 @@ int PmuCollector::init( PmuDataHeader *hdr = get_pmu_header(shm_host_local); hdr->event_type = static_cast(event_type); hdr->num_cores = static_cast(num_cores); + hdr->backpressure.block_on_contention = + static_cast(block_on_contention); // freeze_active/contended stay 0 (memset above) // ---- Allocate the per-core ring-address table for AICore. The ring // table is filled fully by the host. AICore resolves its own PMU MMIO diff --git a/src/a5/platform/sim/host/device_runner.cpp b/src/a5/platform/sim/host/device_runner.cpp index 26f7adff3..b9bc95354 100644 --- a/src/a5/platform/sim/host/device_runner.cpp +++ b/src/a5/platform/sim/host/device_runner.cpp @@ -700,8 +700,8 @@ void DeviceRunner::finalize_collectors() { int DeviceRunner::init_l2_swimlane(int num_aicore, int aicpu_thread_num, int device_id) { int rc = l2_swimlane_collector_.initialize( - num_aicore, aicpu_thread_num, device_id, l2_swimlane_level_, prof_alloc_cb, /*register_cb=*/nullptr, - prof_free_cb, output_prefix_ + num_aicore, aicpu_thread_num, device_id, l2_swimlane_level_, dfx_backpressure_, prof_alloc_cb, + /*register_cb=*/nullptr, prof_free_cb, output_prefix_ ); if (rc == 0) { kernel_args_.l2_swimlane_data_base = @@ -716,8 +716,8 @@ int DeviceRunner::init_tensor_dump(Runtime &runtime, int device_id) { int num_dump_threads = runtime.get_aicpu_thread_num(); int rc = dump_collector_.initialize( - num_dump_threads, device_id, prof_alloc_cb, /*register_cb=*/nullptr, prof_free_cb, output_prefix_, - dump_tensor_level_ + num_dump_threads, device_id, dfx_backpressure_, prof_alloc_cb, /*register_cb=*/nullptr, prof_free_cb, + output_prefix_, dump_tensor_level_ ); if (rc != 0) { return rc; @@ -731,7 +731,8 @@ int DeviceRunner::init_pmu( int num_cores, int num_threads, const std::string &csv_path, PmuEventType event_type, int /*device_id*/ ) { int rc = pmu_collector_.init( - num_cores, num_threads, csv_path, event_type, prof_alloc_cb, /*register_cb=*/nullptr, prof_free_cb, + num_cores, num_threads, csv_path, event_type, dfx_backpressure_, prof_alloc_cb, /*register_cb=*/nullptr, + prof_free_cb, /*device_id=*/-1 ); if (rc == 0) { @@ -747,7 +748,7 @@ int DeviceRunner::init_scope_stats(int num_threads) { // device buffer; sim's profiling_copy_* are plain memcpys, so the dev/host // shadow path collapses to one allocation pair without address-space tricks. int rc = scope_stats_collector_.init( - num_threads, prof_alloc_cb, /*register_cb=*/nullptr, prof_free_cb, /*device_id=*/-1 + num_threads, dfx_backpressure_, prof_alloc_cb, /*register_cb=*/nullptr, prof_free_cb, /*device_id=*/-1 ); if (rc != 0) { return rc; @@ -760,8 +761,9 @@ int DeviceRunner::init_scope_stats(int num_threads) { int DeviceRunner::init_dep_gen(int num_threads, int /*device_id*/) { // a5 sim: register_cb=nullptr; sim's profiling_copy_* are plain memcpys, so // the dev/host shadow path collapses to one allocation pair. - int rc = - dep_gen_collector_.init(num_threads, prof_alloc_cb, /*register_cb=*/nullptr, prof_free_cb, /*device_id=*/-1); + int rc = dep_gen_collector_.init( + num_threads, dfx_backpressure_, prof_alloc_cb, /*register_cb=*/nullptr, prof_free_cb, /*device_id=*/-1 + ); if (rc != 0) { return rc; } diff --git a/src/common/platform/include/aicpu/profiler_device_engine.h b/src/common/platform/include/aicpu/profiler_device_engine.h index d761e0186..9ace618cb 100644 --- a/src/common/platform/include/aicpu/profiler_device_engine.h +++ b/src/common/platform/include/aicpu/profiler_device_engine.h @@ -14,6 +14,7 @@ #include #include "aicpu/device_time.h" +#include "common/dfx_backpressure_device.h" #include "common/memory_barrier.h" #include "common/platform_config.h" @@ -37,6 +38,8 @@ struct DeviceProfilerEngine { return false; } + dfx_backpressure::peer_freeze_barrier(header); + bool contended_signalled = false; const uint64_t start = get_sys_cnt_aicpu(); do { uint32_t current_tail = header->queue_tails[thread_idx]; @@ -47,6 +50,13 @@ struct DeviceProfilerEngine { *head_out = current_head; return true; } + // Block mode (DFX backpressure): mark contention so the host opens a + // global freeze that parks peer lanes too, and never give up — spin + // until the host drains. Drop mode: give up after the bounded wait. + if (dfx_backpressure::block_on_contention(header)) { + dfx_backpressure::mark_contended(header, &contended_signalled); + continue; + } if (Module::kBackpressureWaitCycles == 0) { break; } @@ -58,11 +68,14 @@ struct DeviceProfilerEngine { return false; } - static bool wait_for_free_queue_entry(FreeQueue *free_queue, uint32_t *head_out, uint32_t *tail_out) { + static bool + wait_for_free_queue_entry(DataHeader *header, FreeQueue *free_queue, uint32_t *head_out, uint32_t *tail_out) { if (free_queue == nullptr) { return false; } + dfx_backpressure::peer_freeze_barrier(header); + bool contended_signalled = false; const uint64_t start = get_sys_cnt_aicpu(); do { uint32_t head = free_queue->head; @@ -73,6 +86,13 @@ struct DeviceProfilerEngine { rmb(); // acquire: order the tail read above before the caller's buffer_ptrs read return true; } + // Block mode (DFX backpressure): mark contention so the host opens a + // global freeze that parks peer lanes too, and never give up — spin + // until the host drains. Drop mode: give up after the bounded wait. + if (dfx_backpressure::block_on_contention(header)) { + dfx_backpressure::mark_contended(header, &contended_signalled); + continue; + } if (Module::kBackpressureWaitCycles == 0) { break; } @@ -108,7 +128,7 @@ struct DeviceProfilerEngine { FreeQueue *free_queue = Module::free_queue(state); uint32_t head = 0; uint32_t tail = 0; - if (!wait_for_free_queue_entry(free_queue, &head, &tail)) { + if (!wait_for_free_queue_entry(Module::header(ctx), free_queue, &head, &tail)) { return nullptr; } diff --git a/src/common/platform/include/common/dfx_backpressure_device.h b/src/common/platform/include/common/dfx_backpressure_device.h new file mode 100644 index 000000000..5897a310a --- /dev/null +++ b/src/common/platform/include/common/dfx_backpressure_device.h @@ -0,0 +1,125 @@ +/* + * 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. + * ----------------------------------------------------------------------------------------------------------- + */ + +/** + * @file dfx_backpressure_device.h + * @brief Unified DFX backpressure (block-on-contention + global-sync freeze) + * coordination — shared layout + AICPU-side primitives. + * + * `DfxBackpressureHeader` is the per-subsystem coordination block. Every DFX + * subsystem's DataHeader embeds ONE `DfxBackpressureHeader backpressure;` member + * instead of re-declaring the three fields. It is physically per-subsystem (each + * subsystem has its own shared-memory region) — that is the per-subsystem peer + * group: swimlane's freeze is independent of pmu's. The struct *definition* is + * shared here; the *instances* are necessarily separate. + * + * The two device-side reactions (park at the gate / raise the leader signal) + * live here too. The host side of the mechanism (the freeze open/release state + * machine) lives once in `ProfilerAlgorithms::update_backpressure_freeze` + * (host/profiler_base.h). Every DFX subsystem carries the header block and runs + * the machinery; it is idle unless a run sets `block_on_contention`. + * + * See docs/dfx/global-backpressure-design.md. + */ + +#ifndef SRC_COMMON_PLATFORM_INCLUDE_COMMON_DFX_BACKPRESSURE_DEVICE_H_ +#define SRC_COMMON_PLATFORM_INCLUDE_COMMON_DFX_BACKPRESSURE_DEVICE_H_ + +#include + +#include "common/memory_barrier.h" + +// Per-subsystem backpressure coordination block, embedded once in every DFX +// DataHeader. block_on_contention: host writes at init (1 = block instead of +// drop). freeze_active: host opens/clears a global freeze; device parks at its +// gate while set. contended: device raises on real contention (leader signal); +// host consumes + clears. +struct DfxBackpressureHeader { + uint32_t block_on_contention; // host → device (init) + volatile uint32_t freeze_active; // host → device + volatile uint32_t contended; // device → host +}; + +namespace dfx_backpressure { + +// Bare CPU spin-relax hint for the park loops below: aarch64 `yield` on real +// AICPU silicon, `pause` on the x86_64 host that runs the sim AICPU build. +// Never an OS yield — these loops sit on the AICPU hot path. +inline void spin_relax() { +#if defined(__aarch64__) + __asm__ volatile("yield"); +#elif defined(__x86_64__) + __builtin_ia32_pause(); +#endif +} + +// Is block-on-contention enabled for this subsystem? Reads the host-written init +// field straight from the header (no device-side bool cache needed — it is +// non-volatile and set once per launch, so the read is a plain L1-cached load). +// Null-safe: a not-yet-resolved / early-aborted header means drop mode. +template +inline bool block_on_contention(const Header *header) { + return header != nullptr && header->backpressure.block_on_contention != 0; +} + +// Global-sync peer freeze. Called at every buffer-switch gate: if block mode is +// on AND the host has opened a freeze (`freeze_active != 0`), park here — even +// when this instance still has buffers — until the host clears it. This turns a +// per-instance block (which sparsifies one lane and misleads bottleneck reading) +// into a common-mode, lane-aligned gap. Bare spin hint, never an OS yield. +// No-op in drop mode or before the header resolves. +template +inline void peer_freeze_barrier(const Header *header) { + if (!block_on_contention(header)) { + return; + } + while (header->backpressure.freeze_active != 0) { + spin_relax(); + } + rmb(); // acquire: order host's pre-release queue writes before the reads that follow the gate +} + +// Block mode leader signal. Called (once per wait, via the `signalled` guard) +// when a lane hits real buffer contention, so the host opens a global freeze +// that parks the peer lanes too. Idempotent sticky flag; the host consumes and +// clears it. +template +inline void mark_contended(Header *header, bool *signalled) { + if (!*signalled && header != nullptr) { + header->backpressure.contended = 1; + *signalled = true; + } +} + +// Block mode leader park. Called right after mark_contended() by a lane whose +// own reclaim depends on the host completing an open→drain→release cycle (only +// tensor_dump's arena barrier today — engine-gate leaders instead spin on a real +// free slot). Blocks until the host has both opened the freeze covering this +// contention AND released it. Uses the DISJUNCTION (contended || freeze_active), +// not freeze_active alone: the host opens the freeze (freeze_active=1) before +// consuming contended (=0), so from mark_contended() to release the predicate is +// continuously true — there is no (contended=0, freeze_active=0) window for the +// leader to escape through before the host has reacted. Bare spin hint (never an +// OS yield). No-op in drop mode. +template +inline void wait_for_release(const Header *header) { + if (!block_on_contention(header)) { + return; + } + while (header->backpressure.contended != 0 || header->backpressure.freeze_active != 0) { + spin_relax(); + } + rmb(); // acquire: order host's pre-release writes before the leader's post-release reads +} + +} // namespace dfx_backpressure + +#endif // SRC_COMMON_PLATFORM_INCLUDE_COMMON_DFX_BACKPRESSURE_DEVICE_H_ diff --git a/src/common/platform/include/common/scope_stats.h b/src/common/platform/include/common/scope_stats.h index 0605bc97e..f6cfde675 100644 --- a/src/common/platform/include/common/scope_stats.h +++ b/src/common/platform/include/common/scope_stats.h @@ -47,6 +47,7 @@ #include #include +#include "common/dfx_backpressure_device.h" #include "common/platform_config.h" #define PTO2_SCOPE_STATS_MAX_RING_DEPTH 4 @@ -116,7 +117,7 @@ struct ScopeStatsFreeQueue { volatile uint64_t buffer_ptrs[PLATFORM_SCOPE_STATS_SLOT_COUNT]; volatile uint32_t head; // Consumer read position (Device increments) volatile uint32_t tail; // Producer write position (Host increments) - uint32_t _pad[14]; + uint32_t _pad[22]; // pad to 128 B (4-slot buffer_ptrs=32 + head/tail=8 + 88) } __attribute__((aligned(64))); // Per-instance buffer state. current_buf_ptr / current_buf_seq / the counters @@ -153,6 +154,8 @@ struct ScopeStatsDataHeader { volatile uint32_t queue_heads[PLATFORM_MAX_AICPU_THREADS]; // Host reads (consumer) volatile uint32_t queue_tails[PLATFORM_MAX_AICPU_THREADS]; // AICPU writes (producer) uint32_t num_instances; // Always 1 for now + // DFX backpressure coordination (unified across all DFX subsystems). + DfxBackpressureHeader backpressure; // Per-ring static capacities — written once by AICPU at orchestrator init // (scope_stats_set_ring_capacity / scope_stats_set_tensormap_capacity). diff --git a/src/common/platform/include/host/buffer_pool_manager.h b/src/common/platform/include/host/buffer_pool_manager.h index a667867a0..84aef5fc0 100644 --- a/src/common/platform/include/host/buffer_pool_manager.h +++ b/src/common/platform/include/host/buffer_pool_manager.h @@ -436,6 +436,13 @@ class BufferPoolManager { shard.cv.notify_one(); } + // Inc/dec the delivered-minus-done counter (see collector_inflight()). Inc at + // device-RQ pop time, dec on drop (process_entry) or collect + // (notify_copy_done) — exactly one dec per inc ⇒ never negative. The pop-time + // inc timing is load-bearing; rationale at the try_pop_aicpu_entry call site. + void collector_inflight_inc() { collector_inflight_.fetch_add(1, std::memory_order_release); } + void collector_inflight_dec() { collector_inflight_.fetch_sub(1, std::memory_order_release); } + bool try_pop_ready(ReadyBufferInfo &out, int shard_index = 0) { auto &shard = ready_shards_[normalize_shard(shard_index)]; std::scoped_lock lock(shard.mutex); @@ -466,8 +473,13 @@ class BufferPoolManager { void notify_copy_done(void *dev_ptr, int kind, int shard_index = 0) { auto &shard = done_shards_[normalize_shard(shard_index)]; - std::scoped_lock lock(shard.mutex); - shard.queue.push(DoneInfo{dev_ptr, kind}); + { + std::scoped_lock lock(shard.mutex); + shard.queue.push(DoneInfo{dev_ptr, kind}); + } + // Collect path: this buffer's on_buffer_collected has returned (arena + // bytes pulled for tensor_dump). Balances the pop-time inc. + collector_inflight_dec(); } // ------------------------------------------------------------------------- @@ -656,6 +668,15 @@ class BufferPoolManager { void *shared_mem_host() const { return shared_mem_host_; } int device_id() const { return device_id_; } + // Buffers delivered to the collector (push_to_ready) minus buffers the + // collector has finished (notify_copy_done). Zero ⇒ collector quiesced: the + // mgmt→collector hand-off is empty AND no on_buffer_collected is in flight. + // Used by DFX backpressure's per-subsystem release predicate — tensor_dump + // gates arena reclaim on this (its on_buffer_collected pulls arena bytes, + // which RQ-empty alone does not guarantee are drained). Cheap generic + // counter; subsystems that don't read it pay only one atomic inc/dec/buffer. + uint64_t collector_inflight() const { return collector_inflight_.load(std::memory_order_acquire); } + private: struct ReadyQueueShard { std::mutex mutex; @@ -686,6 +707,9 @@ class BufferPoolManager { // collector → mgmt std::vector done_shards_; + // delivered-to-collector minus collector-done; see collector_inflight(). + std::atomic collector_inflight_{0}; + // Host-side pointer mappings are shared across all collector shards. mutable std::mutex mapping_mutex_; static constexpr size_t kFreeQueueLockStripes = 64; diff --git a/src/common/platform/include/host/dep_gen_collector.h b/src/common/platform/include/host/dep_gen_collector.h index dddab9cd4..3b36561f5 100644 --- a/src/common/platform/include/host/dep_gen_collector.h +++ b/src/common/platform/include/host/dep_gen_collector.h @@ -185,8 +185,8 @@ class DepGenCollector : public profiling_common::ProfilerBasequeue_heads[q] = next_head; @@ -383,6 +389,7 @@ struct ProfilerAlgorithms { // is intentionally not used here — see buffer_pool_manager.h. if (mgr.write_range_to_device(&header->queue_heads[q], sizeof(header->queue_heads[q])) != 0) { header->queue_heads[q] = old_head; + mgr.collector_inflight_dec(); // pop rolled back — undo the pop-time inc LOG_ERROR("%s: failed to advance ready_queue head for thread %d", Module::kSubsystemName, q); return false; } @@ -399,13 +406,21 @@ struct ProfilerAlgorithms { // shadow seen by the collector then matches what the device wrote. template static void process_entry(Mgr &mgr, DataHeader *header, int q, const ReadyEntry &entry) { + // Every path out of this function must balance the pop-time + // collector_inflight_inc() from try_pop_aicpu_entry: drop paths dec here, + // the deliver path hands off to the collector which decs in + // notify_copy_done. auto site_opt = Module::resolve_entry(mgr.shared_mem_host(), header, q, entry); - if (!site_opt.has_value()) return; + if (!site_opt.has_value()) { + mgr.collector_inflight_dec(); // dropped: not delivered to collector + return; + } auto &site = *site_opt; site.info.host_buffer_ptr = mgr.resolve_host_ptr(site.info.dev_buffer_ptr); if (site.info.host_buffer_ptr == nullptr) { // resolve_host_ptr already logged. Drop rather than deliver null. + mgr.collector_inflight_dec(); return; } // a5: pull buffer contents from device into the host shadow before @@ -414,10 +429,19 @@ struct ProfilerAlgorithms { LOG_ERROR( "%s: failed to copy ready buffer from device (kind=%d, thread=%d)", Module::kSubsystemName, site.kind, q ); + mgr.collector_inflight_dec(); return; } - (void)top_up_free_queue(mgr, site.kind, *site.free_queue, site.buffer_size, q); + // Drain-driven free_queue top-up. Suppressed while a DFX backpressure + // freeze is open: during a freeze the single replenish thread owns + // free_queue top-up (update_backpressure_freeze), so a drain thread must + // not also write it. Safe hand-off: a freeze only reaches its replenish + // point once RQ is empty, by which time drain threads have long observed + // freeze_active=1 and stopped here — the two writers never overlap. + if (header->backpressure.freeze_active == 0) { + (void)top_up_free_queue(mgr, site.kind, *site.free_queue, site.buffer_size, q); + } mgr.push_to_ready(site.info, q); } @@ -441,6 +465,84 @@ struct ProfilerAlgorithms { return pushed; } + // DFX backpressure global-sync freeze state machine (host-owned). Runs each + // mgmt tick for every DFX subsystem; idle unless a run enables + // block_on_contention. The device sets + // header->backpressure.contended on real buffer contention; the host opens a freeze + // (freeze_active=1) so every AICPU lane parks at its next buffer-switch gate, + // then releases (freeze_active=0) once all lanes are fully drained + refilled + // — one lane-aligned common-mode gap instead of scattered per-lane + // sparseness. Only meaningful in block mode (block_on_contention set). + // + // freeze_active is host→device, contended / queue_tails are device→host. Host + // writes go through write_range_to_device and device→host reads through + // read_range_from_device so a5 (non-SVM) sees them; a2a3 (SVM) short-circuits + // both to no-ops since the header already lives in shared device memory. + // extra_release_ready is the per-subsystem release predicate + // (Derived::backpressure_release_ready(), default true). The caller + // (mgmt_replenish_loop) evaluates it because this static has no Derived. + // tensor_dump uses it to hold the release until its collector has pulled all + // arena bytes (RQ-empty alone does not imply that — see buffer_pool_manager + // collector_inflight()). + template + static void update_backpressure_freeze(Mgr &mgr, DataHeader *header, bool extra_release_ready = true) { + if (header->backpressure.block_on_contention == 0) return; + + if (header->backpressure.freeze_active == 0) { + // No freeze open: a lane signalled real contention → open one so all + // peers park at their next gate (aligned common-mode gap). Refresh the + // device→host contended flag first (a5 non-SVM; no-op on a2a3). + if (mgr.read_range_from_device(&header->backpressure.contended, sizeof(header->backpressure.contended)) != + 0) { + return; + } + if (header->backpressure.contended != 0) { + // Open the freeze BEFORE consuming the contended signal, so the + // disjunction (contended || freeze_active) stays true across the + // whole handoff. A leader parked in wait_for_release therefore + // never observes the (0,0) escape window. On a5 the two writes are + // sequential DMAs, so freeze_active lands on device first. + header->backpressure.freeze_active = 1; + wmb(); + mgr.write_range_to_device( + &header->backpressure.freeze_active, sizeof(header->backpressure.freeze_active) + ); + header->backpressure.contended = 0; + mgr.write_range_to_device(&header->backpressure.contended, sizeof(header->backpressure.contended)); + } + return; + } + + // Freeze open. Release only when every lane is parked (all per-thread + // ready queues drained) AND every free_queue is refilled to kSlotCount + // (max runway, so resumed lanes don't immediately re-stall — one clean + // gap). Order matters: wait for RQ-empty FIRST. Once RQ is empty the + // device is frozen and the drain threads are idle, so it is safe for + // THIS thread to top up the free queues (drain-driven top_up in + // process_entry has stopped — no concurrent writer to free_queue). That + // top_up is also what makes FQ-full reachable during a freeze, since the + // split-mgmt replenish loop otherwise only drains done→recycled. + // Refresh the whole device→host produce-position array in one DMA read + // (a5 non-SVM; no-op on a2a3). queue_heads is the host's own drain + // position (local), so only queue_tails needs the round-trip. + if (mgr.read_range_from_device(&header->queue_tails[0], sizeof(header->queue_tails)) != 0) { + return; + } + for (int q = 0; q < PLATFORM_MAX_AICPU_THREADS; q++) { + if (header->queue_heads[q] != header->queue_tails[q]) return; + } + replenish_free_queues(mgr, header); + bool all_full = true; + Module::for_each_instance(mgr.shared_mem_host(), header, [&](int, FreeQueue *fq, size_t) { + if (fq->tail - fq->head != Module::kSlotCount) all_full = false; + }); + if (all_full && extra_release_ready) { + header->backpressure.freeze_active = 0; + wmb(); + mgr.write_range_to_device(&header->backpressure.freeze_active, sizeof(header->backpressure.freeze_active)); + } + } + private: template static auto refresh_replenish_metadata(Mgr &mgr, DataHeader *header, int) @@ -564,6 +666,16 @@ class ProfilerBase { ProfilerBase(const ProfilerBase &) = delete; ProfilerBase &operator=(const ProfilerBase &) = delete; + // DFX backpressure per-subsystem release predicate (CRTP hook). Default: + // the freeze may release as soon as the framework's RQ-empty + FQ-full hold. + // A subsystem whose collector owns a separate, independently-overwritten + // region (only tensor_dump today: the payload arena) MUST override this to + // return false until its collector has drained that region, else the device + // resumes and overwrites not-yet-pulled data. Called on the mgmt_replenish + // thread inside the freeze loop — overrides must be cheap, non-blocking, and + // read only atomics. + bool backpressure_release_ready() const { return true; } + private: friend Derived; ProfilerBase() = default; @@ -932,9 +1044,20 @@ class ProfilerBase { } void mgmt_replenish_loop() { + DataHeader *header = Module::header_from_shm(manager_.shared_mem_host()); while (mgmt_running_.load(std::memory_order_relaxed)) { size_t drained = manager_.drain_done_into_recycled(); + // DFX backpressure global-sync freeze state machine. Runs on this + // single replenish thread — the state machine must be single-writer + // for freeze_active. Idle unless a run enables block_on_contention. + // Per-subsystem release predicate (CRTP): default true, only + // tensor_dump overrides it (gate release on collector-quiesce); + // evaluated here because update_backpressure_freeze is a static with + // no Derived, and must be cheap + non-blocking (freeze hot loop). + const bool extra_release_ready = static_cast(this)->backpressure_release_ready(); + ProfilerAlgorithms::update_backpressure_freeze(manager_, header, extra_release_ready); + if (drained == 0) { std::this_thread::sleep_for(std::chrono::microseconds(10)); } @@ -992,6 +1115,10 @@ class ProfilerBase { } void consume(const ReadyBufferInfo &info, int shard_index) { + // on_buffer_collected must finish pulling device data into host memory + // before notify_copy_done decs collector_inflight — the DFX backpressure + // release proxy (inflight==0 ⇒ arena drained) depends on this order. + // Keep the arena pull synchronous, or move its completion ahead of dec. if constexpr (ProfilerDerivedShardAwareCollector::value) { static_cast(this)->on_buffer_collected(info, shard_index); } else { diff --git a/src/common/platform/include/host/scope_stats_collector.h b/src/common/platform/include/host/scope_stats_collector.h index d5fb65f52..81fe11432 100644 --- a/src/common/platform/include/host/scope_stats_collector.h +++ b/src/common/platform/include/host/scope_stats_collector.h @@ -156,8 +156,8 @@ class ScopeStatsCollector : public profiling_common::ProfilerBase #include "aicpu/profiler_device_engine.h" +#include "common/dfx_backpressure_device.h" #include "common/memory_barrier.h" #include "common/platform_config.h" #include "common/unified_log.h" @@ -591,6 +592,60 @@ void dump_args_init(int num_dump_threads) { memset(s_buffers_flushed, 0, sizeof(s_buffers_flushed)); } +// DFX arena backpressure barrier (block mode only). Pre-write predictive lap +// check: if writing `copy_bytes` this epoch would lap past epoch_start and +// overwrite this epoch's not-yet-drained payloads, flush the current metadata +// buffer `buf` (making its records — hence their arena payloads — drainable), +// raise the leader signal, park until the host has fully drained + released, +// then open a fresh epoch and re-acquire a buffer. Pre-empts the overwrite +// (host-side `overwritten` never fires); NOT a post-hoc detection. Single tensor +// > arena is already clamped to arena_size/2 by the caller, so it always fits a +// fresh epoch. Only THIS lane (which flushes) resets its epoch; peer lanes +// parked by the global freeze do not. Hot path: no logging, and the park spins +// on a bare `yield` hint — never an OS yield / syscall. +// +// Returns the buffer to write into: `buf` unchanged when no barrier fires, +// otherwise the freshly re-acquired buffer — or nullptr if none is available +// (caller should account_dropped_records + return). +static DumpMetaBuffer * +dfx_arena_barrier(int thread_idx, DumpBufferState *state, uint64_t copy_bytes, DumpMetaBuffer *buf) { + if (copy_bytes == 0 || !dfx_backpressure::block_on_contention(s_dump_header)) { + return buf; + } + uint64_t epoch_bytes = state->arena_write_offset - state->epoch_start_offset; + if (epoch_bytes + copy_bytes <= state->arena_size) { + return buf; // fits in the current epoch — no lap + } + // (1) Flush the current (accumulated) metadata buffer so its records — hence + // their arena payloads — are drainable before we park. In block mode enqueue + // spins until ready_queue has room (host drain is independent), so it makes + // progress. If it ever fails to publish, we must NOT reset the epoch (those + // payloads would never drain yet get overwritten): drop and bail instead. + if (buf != nullptr && buf->count > 0) { + uint64_t buf_addr = reinterpret_cast(buf); + if (enqueue_dump_ready_buffer(thread_idx, buf_addr, state->current_buf_seq) != 0) { + return nullptr; // could not flush → caller drops, epoch left intact + } + s_current_dump_buf[thread_idx] = nullptr; + state->current_buf_ptr = 0; + wmb(); + } + // (2) Raise the leader signal so the host opens a global freeze, then (3) park + // until that freeze is opened AND released. wait_for_release spins on + // (contended || freeze_active) so there is no escape before the host reacts. + bool signalled = false; + dfx_backpressure::mark_contended(s_dump_header, &signalled); + dfx_backpressure::wait_for_release(s_dump_header); + // (4) Whole arena is now consumed → open a fresh epoch (this lane only) and + // re-acquire a metadata buffer for the pending write. + state->epoch_start_offset = state->arena_write_offset; + DumpMetaBuffer *nb = s_current_dump_buf[thread_idx]; + if (nb == nullptr) { + nb = try_pop_dump_meta_buffer(thread_idx, state, state->current_buf_seq); + } + return nb; +} + int dump_arg_record(int thread_idx, const TensorDumpInfo &info) { if (s_dump_header == nullptr) { return -1; @@ -645,6 +700,15 @@ int dump_arg_record(int thread_idx, const TensorDumpInfo &info) { truncated = true; } + // DFX arena backpressure: if this write would lap the arena, flush + park + + // reset the epoch and re-acquire the buffer (all inside the helper). No-op in + // drop mode; returns the buffer to write into, or nullptr if none is free. + buf = dfx_arena_barrier(thread_idx, state, copy_bytes, buf); + if (buf == nullptr) { + account_dropped_records(state, 1); + return -1; + } + uint64_t offset = state->arena_write_offset; state->arena_write_offset = offset + copy_bytes; diff --git a/src/common/platform/shared/host/dep_gen_collector.cpp b/src/common/platform/shared/host/dep_gen_collector.cpp index 70198c70d..1057e24a5 100644 --- a/src/common/platform/shared/host/dep_gen_collector.cpp +++ b/src/common/platform/shared/host/dep_gen_collector.cpp @@ -42,7 +42,7 @@ DepGenCollector::~DepGenCollector() { stop(); } // --------------------------------------------------------------------------- int DepGenCollector::init( - int num_threads, const DepGenAllocCallback &alloc_cb, DepGenRegisterCallback register_cb, + int num_threads, bool block_on_contention, const DepGenAllocCallback &alloc_cb, DepGenRegisterCallback register_cb, const DepGenFreeCallback &free_cb, int device_id ) { if (initialized_) { @@ -86,6 +86,8 @@ int DepGenCollector::init( std::memset(shm_host_local, 0, shm_size); DepGenDataHeader *hdr = get_dep_gen_header(shm_host_local); hdr->num_instances = static_cast(num_instances); + hdr->backpressure.block_on_contention = + static_cast(block_on_contention); // freeze_active/contended stay 0 (memset above) // ---- Allocate DepGenBuffers, populate free_queue + recycled pool ---- const size_t buf_size = sizeof(DepGenBuffer); diff --git a/src/common/platform/shared/host/l2_swimlane_collector.cpp b/src/common/platform/shared/host/l2_swimlane_collector.cpp index fd9da2fef..463946b1b 100644 --- a/src/common/platform/shared/host/l2_swimlane_collector.cpp +++ b/src/common/platform/shared/host/l2_swimlane_collector.cpp @@ -51,7 +51,7 @@ L2SwimlaneCollector::~L2SwimlaneCollector() { } int L2SwimlaneCollector::initialize( - int num_aicore, int aicpu_thread_num, int device_id, L2SwimlaneLevel l2_swimlane_level, + int num_aicore, int aicpu_thread_num, int device_id, L2SwimlaneLevel l2_swimlane_level, bool block_on_contention, const L2SwimlaneAllocCallback &alloc_cb, L2SwimlaneRegisterCallback register_cb, const L2SwimlaneFreeCallback &free_cb, const std::string &output_prefix ) { @@ -77,6 +77,7 @@ int L2SwimlaneCollector::initialize( num_aicore_ = num_aicore; aicpu_thread_num_ = aicpu_thread_num; l2_swimlane_level_ = l2_swimlane_level; + block_on_contention_ = block_on_contention; output_prefix_ = output_prefix; total_perf_collected_ = 0; total_sched_phase_collected_ = 0; @@ -155,6 +156,8 @@ int L2SwimlaneCollector::initialize( header->num_cores = num_aicore; header->l2_swimlane_level = static_cast(l2_swimlane_level_); + header->backpressure.block_on_contention = + static_cast(block_on_contention_); // freeze_active/contended stay 0 (memset above) // Phase metadata: must be zero-initialized here. alloc_cb returns // uninitialized device memory; AICPU only writes these fields when // phase init runs (level >= SCHED_PHASES). Without zeroing, lower diff --git a/src/common/platform/shared/host/scope_stats_collector.cpp b/src/common/platform/shared/host/scope_stats_collector.cpp index c41248acf..fd35eff7a 100644 --- a/src/common/platform/shared/host/scope_stats_collector.cpp +++ b/src/common/platform/shared/host/scope_stats_collector.cpp @@ -48,8 +48,8 @@ ScopeStatsCollector::~ScopeStatsCollector() { stop(); } // --------------------------------------------------------------------------- int ScopeStatsCollector::init( - int num_threads, const ScopeStatsAllocCallback &alloc_cb, ScopeStatsRegisterCallback register_cb, - const ScopeStatsFreeCallback &free_cb, int device_id + int num_threads, bool block_on_contention, const ScopeStatsAllocCallback &alloc_cb, + ScopeStatsRegisterCallback register_cb, const ScopeStatsFreeCallback &free_cb, int device_id ) { if (num_threads <= 0 || alloc_cb == nullptr || free_cb == nullptr) { LOG_ERROR("ScopeStatsCollector::init: invalid arguments"); @@ -92,6 +92,8 @@ int ScopeStatsCollector::init( std::memset(shm_host_local, 0, shm_size); ScopeStatsDataHeader *hdr = get_scope_stats_header(shm_host_local); hdr->num_instances = static_cast(num_instances); + hdr->backpressure.block_on_contention = + static_cast(block_on_contention); // freeze_active/contended stay 0 (memset above) const size_t buf_size = sizeof(ScopeStatsBuffer); ScopeStatsBufferState *state = get_scope_stats_buffer_state(shm_host_local, 0); diff --git a/src/common/platform/shared/host/tensor_dump_collector.cpp b/src/common/platform/shared/host/tensor_dump_collector.cpp index daa1f710c..effc834ed 100644 --- a/src/common/platform/shared/host/tensor_dump_collector.cpp +++ b/src/common/platform/shared/host/tensor_dump_collector.cpp @@ -48,8 +48,9 @@ TensorDumpCollector::~TensorDumpCollector() { stop(); } int TensorDumpCollector::initialize( - int num_dump_threads, int device_id, const DumpAllocCallback &alloc_cb, DumpRegisterCallback register_cb, - const DumpFreeCallback &free_cb, const std::string &output_prefix, DumpTensorLevel dump_tensor_level + int num_dump_threads, int device_id, bool block_on_contention, const DumpAllocCallback &alloc_cb, + DumpRegisterCallback register_cb, const DumpFreeCallback &free_cb, const std::string &output_prefix, + DumpTensorLevel dump_tensor_level ) { if (shm_host_ != nullptr) { LOG_ERROR("TensorDumpCollector already initialized"); @@ -92,6 +93,8 @@ int TensorDumpCollector::initialize( header->num_dump_threads = static_cast(num_dump_threads); header->records_per_buffer = PLATFORM_DUMP_RECORDS_PER_BUFFER; header->dump_tensor_level = static_cast(dump_tensor_level); + header->backpressure.block_on_contention = + static_cast(block_on_contention); // freeze_active/contended stay 0 (memset above) uint64_t arena_size = calc_dump_arena_size(); header->arena_size_per_thread = arena_size; @@ -112,6 +115,7 @@ int TensorDumpCollector::initialize( state->arena_base = reinterpret_cast(ai.dev_ptr); state->arena_size = arena_size; state->arena_write_offset = 0; + state->epoch_start_offset = 0; // DFX arena barrier: epoch starts at offset 0 state->dropped_record_count = 0; LOG_INFO_V0( diff --git a/src/common/platform/sim/host/device_runner_base.cpp b/src/common/platform/sim/host/device_runner_base.cpp index d69a25803..0ca813e55 100644 --- a/src/common/platform/sim/host/device_runner_base.cpp +++ b/src/common/platform/sim/host/device_runner_base.cpp @@ -581,6 +581,7 @@ void SimDeviceRunnerBase::apply_call_config(const CallConfig &config) { // a2a3 and a5 override set_dep_gen_enabled; an arch without dep_gen no-ops. set_dep_gen_enabled(config.enable_dep_gen != 0); set_scope_stats_enabled(config.enable_scope_stats != 0); + set_dfx_backpressure_enabled(config.enable_dfx_backpressure != 0); set_output_prefix(config.output_prefix); } diff --git a/src/common/platform/sim/host/device_runner_base.h b/src/common/platform/sim/host/device_runner_base.h index 911ad53e2..1a43be928 100644 --- a/src/common/platform/sim/host/device_runner_base.h +++ b/src/common/platform/sim/host/device_runner_base.h @@ -159,6 +159,7 @@ class SimDeviceRunnerBase : public L3L2OrchCommBackend { pmu_event_type_ = resolve_pmu_event_type(enable_pmu); } void set_scope_stats_enabled(bool enable) { enable_scope_stats_ = enable; } + void set_dfx_backpressure_enabled(bool enable) { dfx_backpressure_ = enable; } // Diagnostic artifact root directory (CallConfig::validate() enforces non-empty // upstream when any diagnostic is enabled). void set_output_prefix(const char *prefix) { output_prefix_ = (prefix != nullptr) ? prefix : ""; } @@ -320,7 +321,8 @@ class SimDeviceRunnerBase : public L3L2OrchCommBackend { bool enable_scope_stats_{false}; L2SwimlaneLevel l2_swimlane_level_{L2SwimlaneLevel::DISABLED}; // resolved from set_l2_swimlane_enabled() PmuEventType pmu_event_type_{PmuEventType::PIPE_UTILIZATION}; // resolved from set_pmu_enabled() - std::string output_prefix_{}; // diagnostic artifact root directory + bool dfx_backpressure_{false}; // resolved from CallConfig::enable_dfx_backpressure (block-on-contention) + std::string output_prefix_{}; // diagnostic artifact root directory private: void *l3_l2_allocate_region_bytes(uint64_t bytes) override; diff --git a/src/common/task_interface/call_config.h b/src/common/task_interface/call_config.h index 7845c9632..6d275d5ac 100644 --- a/src/common/task_interface/call_config.h +++ b/src/common/task_interface/call_config.h @@ -115,7 +115,12 @@ struct CallConfig { int32_t enable_pmu = 0; // 0 = disabled; >0 = enabled, value selects event type int32_t enable_dep_gen = 0; int32_t enable_scope_stats = 0; // writes /scope_stats/scope_stats.jsonl - RuntimeEnv runtime_env; // per-task PTO2_RING_* overrides + // DFX backpressure (block-on-contention): when set, DFX collectors block the + // device until the host drains instead of dropping records on buffer + // contention. Opt-in modifier — only meaningful while a diagnostic is + // enabled; does not itself require output_prefix. Default off = drop mode. + int32_t enable_dfx_backpressure = 0; + RuntimeEnv runtime_env; // per-task PTO2_RING_* overrides char output_prefix[1024] = {}; bool diagnostics_any() const noexcept { @@ -143,6 +148,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/tests/st/a2a3/host_build_graph/prepared_callable/test_prepared_callable.py b/tests/st/a2a3/host_build_graph/prepared_callable/test_prepared_callable.py index 0bbe3feb1..7e23280aa 100644 --- a/tests/st/a2a3/host_build_graph/prepared_callable/test_prepared_callable.py +++ b/tests/st/a2a3/host_build_graph/prepared_callable/test_prepared_callable.py @@ -127,6 +127,7 @@ def _run_and_validate_l2( # noqa: PLR0913 enable_pmu=0, enable_dep_gen=False, enable_scope_stats=False, + enable_dfx_backpressure=False, output_prefix="", ): params = case.get("params", {}) diff --git a/tests/st/a2a3/tensormap_and_ringbuffer/prepared_callable/test_prepared_callable.py b/tests/st/a2a3/tensormap_and_ringbuffer/prepared_callable/test_prepared_callable.py index a4461ed4f..63b320df1 100644 --- a/tests/st/a2a3/tensormap_and_ringbuffer/prepared_callable/test_prepared_callable.py +++ b/tests/st/a2a3/tensormap_and_ringbuffer/prepared_callable/test_prepared_callable.py @@ -112,6 +112,7 @@ def _run_and_validate_l2( # noqa: PLR0913 enable_pmu=0, enable_dep_gen=False, enable_scope_stats=False, + enable_dfx_backpressure=False, output_prefix="", ): params = case.get("params", {}) diff --git a/tests/st/a5/host_build_graph/prepared_callable/test_prepared_callable.py b/tests/st/a5/host_build_graph/prepared_callable/test_prepared_callable.py index f4f67cd94..18debfe04 100644 --- a/tests/st/a5/host_build_graph/prepared_callable/test_prepared_callable.py +++ b/tests/st/a5/host_build_graph/prepared_callable/test_prepared_callable.py @@ -117,6 +117,7 @@ def _run_and_validate_l2( # noqa: PLR0913 enable_pmu=0, enable_dep_gen=False, enable_scope_stats=False, + enable_dfx_backpressure=False, output_prefix="", ): params = case.get("params", {}) diff --git a/tests/st/a5/tensormap_and_ringbuffer/prepared_callable/test_prepared_callable.py b/tests/st/a5/tensormap_and_ringbuffer/prepared_callable/test_prepared_callable.py index 37d818b5d..8cfe57e6d 100644 --- a/tests/st/a5/tensormap_and_ringbuffer/prepared_callable/test_prepared_callable.py +++ b/tests/st/a5/tensormap_and_ringbuffer/prepared_callable/test_prepared_callable.py @@ -113,6 +113,7 @@ def _run_and_validate_l2( # noqa: PLR0913 enable_pmu=0, enable_dep_gen=False, enable_scope_stats=False, + enable_dfx_backpressure=False, output_prefix="", ): params = case.get("params", {}) diff --git a/tests/ut/cpp/common/test_scope_stats_collector.cpp b/tests/ut/cpp/common/test_scope_stats_collector.cpp index ed47b845e..27265cce8 100644 --- a/tests/ut/cpp/common/test_scope_stats_collector.cpp +++ b/tests/ut/cpp/common/test_scope_stats_collector.cpp @@ -57,7 +57,7 @@ std::string read_file(const std::filesystem::path &path) { TEST(ScopeStatsCollectorTest, ReconcileRecoversUnflushedCurrentBuffer) { ScopeStatsCollector collector; - ASSERT_EQ(collector.init(1, test_alloc, nullptr, test_free, 0), 0); + ASSERT_EQ(collector.init(1, false, test_alloc, nullptr, test_free, 0), 0); auto *header = get_scope_stats_header(collector.get_scope_stats_shm_device_ptr()); auto *state = get_scope_stats_buffer_state(collector.get_scope_stats_shm_device_ptr(), 0); 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..ba080c16f 100644 --- a/tests/ut/py/test_chip_worker.py +++ b/tests/ut/py/test_chip_worker.py @@ -338,6 +338,7 @@ def test_config_roundtrip(self): cfg.enable_pmu = 5 cfg.enable_dep_gen = True cfg.enable_scope_stats = True + cfg.enable_dfx_backpressure = True cfg.runtime_env.ring_task_window = [16, 32, 128, 256] cfg.runtime_env.ring_heap = [1024, 2048, 4096, 8192] cfg.runtime_env.ring_dep_pool = [64, 128, 256, 512] @@ -354,6 +355,7 @@ def test_config_roundtrip(self): cfg.enable_pmu, int(cfg.enable_dep_gen), int(cfg.enable_scope_stats), + int(cfg.enable_dfx_backpressure), *cfg.runtime_env.ring_task_window, *cfg.runtime_env.ring_heap, *cfg.runtime_env.ring_dep_pool, @@ -368,6 +370,7 @@ def test_config_roundtrip(self): assert decoded.enable_pmu == 5 assert decoded.enable_dep_gen is True assert decoded.enable_scope_stats is True + assert decoded.enable_dfx_backpressure is True assert decoded.runtime_env.ring_task_window == [16, 32, 128, 256] assert decoded.runtime_env.ring_heap == [1024, 2048, 4096, 8192] assert decoded.runtime_env.ring_dep_pool == [64, 128, 256, 512]