feat(dfx): unified block-on-contention backpressure across DFX subsystems

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="?",

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<bool>(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);

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

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,
         )
 

src/a2a3/platform/include/common/dep_gen.h

@@ -45,6 +45,7 @@
 #include <cstdint>
 
 #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)));
 
 // =============================================================================

src/a2a3/platform/include/common/l2_swimlane_profiling.h

@@ -60,6 +60,7 @@
 #include <vector>
 
 #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

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

src/a2a3/platform/include/common/pmu_profiling.h

@@ -31,6 +31,7 @@
 #include <cstddef>
 
 #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)));
 
 // =============================================================================

src/a2a3/platform/include/common/tensor_dump.h

@@ -43,6 +43,7 @@
 #include <cstddef>
 #include <cstdint>
 
+#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)));
 
 // =============================================================================

src/a2a3/platform/include/host/pmu_collector.h

@@ -206,7 +206,7 @@ class PmuCollector : public profiling_common::ProfilerBase<PmuCollector, PmuModu
      * @return 0 on success, non-zero on failure
      */
     int 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
     );