Feat: add bounded whole-run FIFO admission - #1541
Conversation
Compute a stable identity from each callable's AICore child binaries and function mapping, then carry it through callable registration. Reuse a slot's AICPU stream independently, but recreate its AICore stream whenever the loaded image changes. Extend hardware coverage with alternating add/sub images and retain same-image reuse checks.
📝 WalkthroughWalkthroughThe change introduces generation-safe pipeline leases, run-scoped FIFO admission, task-acceptance signaling, lease-aware mailbox dispatch, and AICore image-aware stream reuse across C++, Python bindings, device runners, documentation, and tests. ChangesPipeline admission and acceptance
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 9
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/common/hierarchical/orchestrator.cpp (1)
177-206: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winAvoid racing
fail_run_submissionagainst scheduler activation. The run phase is sampled without the completion/run lock, andcancel_unstarted_runis later allowed to mark slotsPENDING/READY/FREEasFAILED; if the closed run has already been activated toEXECUTING, the scheduler can dispatch the same slot asRUNNINGconcurrently, losing one terminal transition. Guard cancellation by the current run phase and use stateful atomic transitions for slot states.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/common/hierarchical/orchestrator.cpp` around lines 177 - 206, Update fail_run_submission and cancel_unstarted_run to synchronize phase inspection with the run/completion state, preventing cancellation from racing scheduler activation. Recheck the current phase while holding the appropriate lock before cancelling, and make each slot-state update use compare-and-exchange transitions that only convert valid non-running states; never mark an EXECUTING/RUNNING slot FAILED or overwrite a concurrent scheduler transition.
🧹 Nitpick comments (5)
src/common/platform/onboard/host/device_runner_base.h (1)
297-320: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDoc block missing the new
aicore_image_hashparam. The@paramlist enumerates every other argument; add a line for the new one so the staging contract stays self-describing (same forrecord_host_orch_callableat Line 332).🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/common/platform/onboard/host/device_runner_base.h` around lines 297 - 320, Update the documentation for record_device_orch_callable to add an `@param` entry describing aicore_image_hash, matching its position and purpose in the function signature. Also add the corresponding parameter documentation to record_host_orch_callable, without changing implementation behavior.src/common/hierarchical/scheduler.cpp (1)
366-369: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueOptional: fold the
active_run_cb ? scoped : unscopedbranching into helpers. The same "resolve active run, bail on invalid, pick scoped vs unscoped queue op" shape is now repeated in all four dispatch paths plus the wait predicate; a pair of small private helpers (e.g.resolve_active_run()returningstd::optional<RunId>andpop_ready(...)) would keep future queue-API changes in one place.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/common/hierarchical/scheduler.cpp` around lines 366 - 369, Consolidate the repeated active-run resolution and scoped/unscoped ready-queue branching across all four dispatch paths and the wait predicate in the scheduler class. Add small private helpers such as resolve_active_run() and pop_ready(...) that preserve invalid-run early returns while centralizing queue API selection, then update the affected paths to use them.python/simpler/task_interface.py (1)
1277-1284: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAdd one-line docstrings to match the neighbouring properties.
Every other property in this class documents its unit/semantics;
pipeline_depthandruntime_slot_countare the odd ones out.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/simpler/task_interface.py` around lines 1277 - 1284, Add concise one-line docstrings to the pipeline_depth and runtime_slot_count properties in the task interface, matching the neighboring properties’ style and documenting each value’s unit or semantics.tests/ut/cpp/hierarchical/test_orchestrator.cpp (1)
656-665: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winA failed
ASSERT_*between here andthird.get()turns the test into a hang.
ASSERT_TRUEreturns from the test body, and~futurefromstd::asyncblocks until the task completes — butbegin_run()only returns once a slot frees, which is exactly what the skipped lines would have done. Consider releasing the admission slot from a scope guard so an assertion failure fails fast instead of wedging the suite.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/ut/cpp/hierarchical/test_orchestrator.cpp` around lines 656 - 665, Ensure the asynchronous third begin_run test always releases an admission slot before any ASSERT_* can exit the test, by adding a scope guard around the first run’s slot ownership and consumption cleanup. Preserve the existing successful-path behavior while making cleanup execute during assertion-induced early returns, so third.get() cannot leave the test hanging.src/common/platform/onboard/host/device_runner_base.cpp (1)
143-150: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winBank validation accepts
< PTO_PIPELINE_MAX_DEPTHbut only two banks exist.
select_arena_bankadmits any id belowPTO_PIPELINE_MAX_DEPTH, while every consumer is a binarybank == 0 ? bank0 : bank1. With today's depth of 2 this is benign, but raising the depth constant would silently alias banks 2..N onto bank1 and share arenas across concurrently admitted runs. Prefer indexing an array of banks, or validate against the actual bank count.♻️ Sketch
- if (bank_id >= PTO_PIPELINE_MAX_DEPTH) { - LOG_ERROR("arena bank %u is outside [0, %u)", bank_id, PTO_PIPELINE_MAX_DEPTH); + if (bank_id >= kArenaBankCount) { + LOG_ERROR("arena bank %u is outside [0, %u)", bank_id, kArenaBankCount); return -1; }with
gm_heap_arena_[kArenaBankCount]etc. so the selection isgm_heap_arena_[g_arena_bank].Also applies to: 193-211
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/common/platform/onboard/host/device_runner_base.cpp` around lines 143 - 150, Update select_arena_bank and the associated arena consumers to validate against the actual number of supported banks rather than PTO_PIPELINE_MAX_DEPTH. Define or reuse a two-bank count, reject IDs outside that range, and replace binary bank-selection logic with indexed access so each accepted bank maps to its own arena without aliasing.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/orchestrator.md`:
- Around line 101-106: The documentation around the “later submit” behavior must
distinguish admission from acceptance: clarify that acceptance unblocks only an
already-admitted successor, while a third submission with bounded depth two
remains blocked in begin_run() until a terminal run releases its lease. Preserve
the existing endpoint acceptance and completion fallback details.
In `@docs/worker-manager.md`:
- Around line 175-181: Update the documentation around the mailbox polling loop
to state that on_accept() is a one-shot callback, invoked only on the first
observation of TASK_ACCEPTED. Document the acceptance_observed guard and
preserve TASK_DONE as the mailbox ownership boundary, so implementations do not
repeatedly decrement per-run acceptance accounting.
In `@python/simpler/worker.py`:
- Line 1544: In the comment near the worker implementation, replace the
ambiguous multiplication symbol “×” in “N×40B” with the ASCII character “x”,
preserving the rest of the comment unchanged.
- Around line 180-186: Update _PIPELINE_LEASE_FMT to use the C ABI’s packed
12-byte layout for PipelineSlotLease: uint32 slot_id, uint32 reserved, and
uint64 generation with no padding between fields. Ensure _OFF_ARGS resolves to
the established MAILBOX_OFF_ARGS offset and remains compatible with the C++
definition in pto_runtime_c_api.h.
In `@src/a2a3/platform/onboard/host/device_runner.cpp`:
- Around line 621-624: Make acceptance durable across the mailbox transition in
the dispatch flow around publish_task_accepted and
LocalMailboxEndpoint::run_with_accept: ensure a parent that observes TASK_DONE
without previously sampling TASK_ACCEPTED still invokes on_accept exactly once.
Use a latched acceptance flag/counter or equivalent state, and preserve the
existing behavior for parents that observe TASK_ACCEPTED first so
decrement_run_accepts is not duplicated.
In `@src/common/hierarchical/orchestrator.cpp`:
- Around line 360-371: Update Orchestrator::decrement_run_accepts to acquire
run->completion_mu before changing pending_accepts and before calling
completion_cv.notify_all(). Keep the decrement, underflow handling, error
recording, and notification behavior intact while ensuring the predicate
transition is synchronized with wait_run_accepted.
In `@src/common/hierarchical/types.cpp`:
- Around line 70-82: Update ReadyQueue’s unscoped access paths, including
try_pop(TaskSlot&) and try_front(TaskSlot&), so they no longer select buckets by
unordered_map iteration order; remove these overloads or add separate
run-insertion-order tracking that preserves FIFO, and update callers such as
try_pop_group(TaskSlot&) and try_pop_single(worker_id, TaskSlot&) accordingly.
In `@src/common/hierarchical/worker_manager.cpp`:
- Around line 397-404: Move the acceptance callback out of the mailbox-locked
section in the worker task flow around read_mailbox_state and
LocalMailboxEndpoint::run. Record that TASK_ACCEPTED was observed while polling,
then invoke on_accept_ only after the mailbox round-trip and mailbox_mu_ have
been released, preserving exactly-once callback behavior.
In `@tests/st/a2a3/host_build_graph/worker_async_fifo/test_worker_async_fifo.py`:
- Line 114: After unpacking tensors into first_a, first_b, first_out, second_a,
second_b, and second_out, explicitly delete those references before
free_host_buffer runs; apply the same cleanup at the corresponding later
unpacking block, while guarding it for early-failure paths where the names may
not be bound.
---
Outside diff comments:
In `@src/common/hierarchical/orchestrator.cpp`:
- Around line 177-206: Update fail_run_submission and cancel_unstarted_run to
synchronize phase inspection with the run/completion state, preventing
cancellation from racing scheduler activation. Recheck the current phase while
holding the appropriate lock before cancelling, and make each slot-state update
use compare-and-exchange transitions that only convert valid non-running states;
never mark an EXECUTING/RUNNING slot FAILED or overwrite a concurrent scheduler
transition.
---
Nitpick comments:
In `@python/simpler/task_interface.py`:
- Around line 1277-1284: Add concise one-line docstrings to the pipeline_depth
and runtime_slot_count properties in the task interface, matching the
neighboring properties’ style and documenting each value’s unit or semantics.
In `@src/common/hierarchical/scheduler.cpp`:
- Around line 366-369: Consolidate the repeated active-run resolution and
scoped/unscoped ready-queue branching across all four dispatch paths and the
wait predicate in the scheduler class. Add small private helpers such as
resolve_active_run() and pop_ready(...) that preserve invalid-run early returns
while centralizing queue API selection, then update the affected paths to use
them.
In `@src/common/platform/onboard/host/device_runner_base.cpp`:
- Around line 143-150: Update select_arena_bank and the associated arena
consumers to validate against the actual number of supported banks rather than
PTO_PIPELINE_MAX_DEPTH. Define or reuse a two-bank count, reject IDs outside
that range, and replace binary bank-selection logic with indexed access so each
accepted bank maps to its own arena without aliasing.
In `@src/common/platform/onboard/host/device_runner_base.h`:
- Around line 297-320: Update the documentation for record_device_orch_callable
to add an `@param` entry describing aicore_image_hash, matching its position and
purpose in the function signature. Also add the corresponding parameter
documentation to record_host_orch_callable, without changing implementation
behavior.
In `@tests/ut/cpp/hierarchical/test_orchestrator.cpp`:
- Around line 656-665: Ensure the asynchronous third begin_run test always
releases an admission slot before any ASSERT_* can exit the test, by adding a
scope guard around the first run’s slot ownership and consumption cleanup.
Preserve the existing successful-path behavior while making cleanup execute
during assertion-induced early returns, so third.get() cannot leave the test
hanging.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 976e895c-87d2-41d9-ab42-f455679ef0bc
📒 Files selected for processing (45)
docs/orchestrator.mddocs/task-flow.mddocs/worker-manager.mdpython/bindings/task_interface.cpppython/bindings/worker_bind.hpython/simpler/orchestrator.pypython/simpler/task_interface.pypython/simpler/worker.pysrc/a2a3/platform/onboard/host/device_runner.cppsrc/a2a3/platform/onboard/host/device_runner.hsrc/a2a3/runtime/host_build_graph/host/runtime_maker.cppsrc/a2a3/runtime/tensormap_and_ringbuffer/host/runtime_maker.cppsrc/a5/runtime/host_build_graph/host/runtime_maker.cppsrc/a5/runtime/tensormap_and_ringbuffer/host/runtime_maker.cppsrc/common/hierarchical/orchestrator.cppsrc/common/hierarchical/orchestrator.hsrc/common/hierarchical/scheduler.cppsrc/common/hierarchical/scheduler.hsrc/common/hierarchical/types.cppsrc/common/hierarchical/types.hsrc/common/hierarchical/worker.cppsrc/common/hierarchical/worker.hsrc/common/hierarchical/worker_manager.cppsrc/common/hierarchical/worker_manager.hsrc/common/platform/onboard/host/c_api_shared.cppsrc/common/platform/onboard/host/device_runner_base.cppsrc/common/platform/onboard/host/device_runner_base.hsrc/common/platform/sim/host/c_api_shared.cppsrc/common/task_interface/chip_callable_layout.hsrc/common/task_interface/prepare_callable_common.hsrc/common/utils/fnv1a_64.hsrc/common/worker/chip_worker.cppsrc/common/worker/chip_worker.hsrc/common/worker/pipeline_contract.hsrc/common/worker/pipeline_slot_pool.hsrc/common/worker/pto_runtime_c_api.htests/st/a2a3/host_build_graph/run_stream_reuse/kernels/aiv/kernel_sub.cpptests/st/a2a3/host_build_graph/run_stream_reuse/test_run_stream_reuse.pytests/st/a2a3/host_build_graph/worker_async_fifo/test_worker_async_fifo.pytests/ut/cpp/hierarchical/test_orchestrator.cpptests/ut/cpp/hierarchical/test_pipeline_contract.cpptests/ut/cpp/hierarchical/test_scheduler.cpptests/ut/cpp/types/test_chip_callable_upload_immutable.cpptests/ut/py/test_callable_identity.pytests/ut/py/test_worker/test_host_worker.py
| blocking path and returns an already-completed handle. At L3 and above, graph | ||
| callbacks remain serialized, but a later submit waits only for every dispatch | ||
| in the prior run to cross its endpoint acceptance boundary. On A2A3 onboard, | ||
| that boundary is after both device kernels are enqueued and before stream | ||
| synchronization. Endpoints without an earlier signal fall back to completion. | ||
| Each run still owns its completion error, keepalives, and cleanup independently. |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Clarify that acceptance only unblocks the admitted successor.
This reads as though every later submit waits only for acceptance. With bounded depth two, the third submission still blocks in begin_run() until a terminal run releases its lease; only an already-admitted successor can build after the prior run is accepted.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/orchestrator.md` around lines 101 - 106, The documentation around the
“later submit” behavior must distinguish admission from acceptance: clarify that
acceptance unblocks only an already-admitted successor, while a third submission
with bounded depth two remains blocked in begin_run() until a terminal run
releases its lease. Preserve the existing endpoint acceptance and completion
fallback details.
| // Poll without sleeping: TASK_ACCEPTED is an intermediate observation and | ||
| // TASK_DONE remains the mailbox ownership boundary. | ||
| while (true) { | ||
| MailboxState state = read_state(mailbox_); | ||
| if (state == MailboxState::TASK_ACCEPTED) on_accept(); | ||
| if (state == MailboxState::TASK_DONE) break; | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Document acceptance as a one-shot callback.
The loop calls on_accept() repeatedly while the mailbox remains TASK_ACCEPTED; the implementation guards this with acceptance_observed. Repeating it would over-decrement per-run acceptance accounting if copied into another endpoint.
Proposed documentation fix
+ bool acceptance_observed = false;
while (true) {
MailboxState state = read_state(mailbox_);
- if (state == MailboxState::TASK_ACCEPTED) on_accept();
+ if (state == MailboxState::TASK_ACCEPTED && !acceptance_observed) {
+ acceptance_observed = true;
+ on_accept();
+ }
if (state == MailboxState::TASK_DONE) break;
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // Poll without sleeping: TASK_ACCEPTED is an intermediate observation and | |
| // TASK_DONE remains the mailbox ownership boundary. | |
| while (true) { | |
| MailboxState state = read_state(mailbox_); | |
| if (state == MailboxState::TASK_ACCEPTED) on_accept(); | |
| if (state == MailboxState::TASK_DONE) break; | |
| } | |
| // Poll without sleeping: TASK_ACCEPTED is an intermediate observation and | |
| // TASK_DONE remains the mailbox ownership boundary. | |
| bool acceptance_observed = false; | |
| while (true) { | |
| MailboxState state = read_state(mailbox_); | |
| if (state == MailboxState::TASK_ACCEPTED && !acceptance_observed) { | |
| acceptance_observed = true; | |
| on_accept(); | |
| } | |
| if (state == MailboxState::TASK_DONE) break; | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/worker-manager.md` around lines 175 - 181, Update the documentation
around the mailbox polling loop to state that on_accept() is a one-shot
callback, invoked only on the first observation of TASK_ACCEPTED. Document the
acceptance_observed guard and preserve TASK_DONE as the mailbox ownership
boundary, so implementations do not repeatedly decrement per-run acceptance
accounting.
| # The generation-safe pipeline lease follows CONFIG. Args start after the | ||
| # lease, 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). | ||
| _OFF_ARGS = (_OFF_CONFIG + _CFG_FMT.size + 7) & ~7 | ||
| _PIPELINE_LEASE_FMT = struct.Struct("=IIQ") | ||
| _OFF_PIPELINE_LEASE = (_OFF_CONFIG + _CFG_FMT.size + 7) & ~7 | ||
| _OFF_ARGS = (_OFF_PIPELINE_LEASE + _PIPELINE_LEASE_FMT.size + 7) & ~7 |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
rg -nP -C3 'MAILBOX_OFF_PIPELINE_LEASE|MAILBOX_OFF_TASK_ARGS_BLOB|MAILBOX_OFF_CONFIG' src/common | head -40
ast-grep run --pattern 'struct PipelineSlotLease { $$$ }' --lang cpp src/common/hierarchical/types.hRepository: hw-native-sys/simpler
Length of output: 4471
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== worker.py relevant lines =="
sed -n '150,195p' python/simpler/worker.py | cat -n
echo
echo "== hierarchical offsets and struct definitions =="
rg -n -C4 'CALLABLE_HASH_DIGEST_SIZE|PipelineSlotLease|struct\s+\w+CallConfig|class CallConfig|struct PipelineSlotLease|pipeline_lease' src/common/hierarchical -g '*.h' -g '*.hpp' -g '*.cpp'
echo
echo "== Python side definitions =="
rg -n -C3 '_OFF_PIPELINE_LEASE|_OFF_ARGS|_CFG_FMT|_PIPELINE_LEASE_FMT|_OFFSET_' python/simpler/worker.py python -g '*.py'Repository: hw-native-sys/simpler
Length of output: 38187
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== exact definitions in worker_manager.h =="
sed -n '1,150p' src/common/hierarchical/worker_manager.h | nl -ba | sed -n '1,150p'
echo
echo "== import of MAILBOX_SIZE/consts in Python worker.py =="
sed -n '1,150p' python/simpler/worker.py | nl -ba | sed -n '1,150p'
echo
echo "== any source generation/special compiler attribute for PipelineSlotLease =="
rg -n -C4 'PipelineSlotLease|struct\s+.*Lease|pragma\s+pack|alignas|__attribute__|packed' src/common src -g '*.h' -g '*.hpp' -g '*.cpp'
echo
echo "== call_config.h wire layout mentions =="
rg -n -C4 'CallConfig|CALLABLE_HASH_DIGEST_SIZE|RUNTIME_ENV_RING_COUNT|pipeline_lease|Mailbox' src/common/hierarchical -g '*.h' -g '*.hpp' -g '*.cpp'Repository: hw-native-sys/simpler
Length of output: 243
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== exact definitions in worker_manager.h =="
sed -n '1,150p' src/common/hierarchical/worker_manager.h | awk '{printf "%6d\t%s\n", NR, $0}'
echo
echo "== import of MAILBOX_SIZE/consts in Python worker.py =="
sed -n '1,150p' python/simpler/worker.py | awk '{printf "%6d\t%s\n", NR, $0}'
echo
echo "== source generation/special compiler attribute for PipelineSlotLease =="
rg -n -C4 'PipelineSlotLease|struct\s+.*Lease|pragma\s+pack|alignas|__attribute__|packed' src/common src -g '*.h' -g '*.hpp' -g '*.cpp'
echo
echo "== call_config.h wire layout mentions =="
rg -n -C4 'CallConfig|CALLABLE_HASH_DIGEST_SIZE|RUNTIME_ENV_RING_COUNT|pipeline_lease|Mailbox' src/common/hierarchical -g '*.h' -g '*.hpp' -g '*.cpp'Repository: hw-native-sys/simpler
Length of output: 50378
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== direct source snippets for relevant definitions =="
sed -n '80,115p' src/common/hierarchical/worker_manager.h
sed -n '175,205p' python/simpler/worker.py
rg -n -C2 'struct PipelineSlotLease|typedef.*PipelineSlotLease|alignas|pragma pack|packed|uint32_t.*uint[0-9]+|static_assert\(sizeof\(PipelineSlotLease\)' src/common/hierarchical src/common/hierarchical/* worker.cc python -g '*.h' -g '*.hpp' -g '*.cpp' -g '*.cc' -g '*.py' | head -120
echo
echo "== deterministic layout comparison from source text =="
python3 - <<'PY'
from pathlib import Path
import re
worker = Path('python/simpler/worker.py').read_text()
mgr = Path('src/common/hierarchical/worker_manager.h').read_text()
# Extract Python struct fmt constants used for layout assertion.
m = re.search(r'_CFG_FMT = struct\.Struct\("([^"]+)"\)', worker)
if not m:
raise SystemExit('CFG_FMT not found')
cfg_fmt = m.group(1)
print('cfg_fmt:', cfg_fmt)
m = re.search(r'_PIPELINE_LEASE_FMT = struct\.Struct\("([^"]+)"\)', worker)
if not m:
raise SystemExit('LEASE_FMT not found')
lease_fmt = m.group(1)
print('lease_fmt:', lease_fmt, 'size:', len(lease_fmt.replace("=I", "4").replace("Q", "8")))
# Extract C++ MAILBOX_OFF_CONFIG and derive calculated MAILBOX_OFF_PIPELINE_LEASE/ARGS from worker_manager.h text.
off_config = re.search(r'MAILBOX_OFF_CONFIG = ([0-9]+);', mgr).group(1)
# Keep exact expression from worker_manager.h; if no full expression, print all near definitions.
print('off_config:', off_config)
# Detect whether C++ PipelineSlotLease has explicit packing. Keep it conservative if source is scarce.
leases = re.findall(r'(?s)struct\s+(?:alignas\([^)]+\)\s+)?PipelineSlotLease\s*\{[^}]+\}', mgr)
packed = re.search(r'(?s)struct\s+PipelineSlotLease\s*\{[^}]+\}\s*__attribute__\(\(packed\)\)', mgr)
print('PipelineSlotLease count:', len(leases))
print('has explicit packed attr:', bool(packed))
if any('alignas' in x or 'pragma' in x for x in leases):
print('PipelineSlotLease alignments:', [len(x) for x in leases])
PYRepository: hw-native-sys/simpler
Length of output: 10021
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== PipelineSlotLease references in src/common/hierarchical =="
rg -n -C3 'PipelineSlotLease|pipeline_lease' src/common/hierarchical -g '*.h' -g '*.hpp' -g '*.cpp' -g '*.cc' | sed -n '1,220p'
echo
echo "== call_config.h and related constants =="
fd -a 'call_config\.(h|cpp|hpp)$' src/common python/simpler | awk '{print}'
sed -n '1,220p' src/common/hierarchical/call_config.h 2>/dev/null || trueRepository: hw-native-sys/simpler
Length of output: 6633
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== task_interface call_config.h =="
sed -n '1,220p' src/common/task_interface/call_config.h
echo
echo "== locate PipelineSlotLease definition =="
rg -n -C6 'PipelineSlotLease\s*(=|:|{)|using PipelineSlotLease|typedef' src python -g '*.h' -g '*.hpp' -g '*.cpp' -g '*.cc' -g '*.py' | sed -n '1,220p'
echo
echo "== include headers in worker_manager.h/types.h relevant to PipelineSlotLease =="
sed -n '1,90p' src/common/hierarchical/worker_manager.h
rg -n 'include|using|struct PipelineSlotLease' src/common/hierarchical/types.h src/common/hierarchical/*.h src/common/task_interface -g '*.h' -g '*.hpp' | sed -n '1,160p'Repository: hw-native-sys/simpler
Length of output: 7595
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== exact PipelineSlotLease definition =="
rg -n -C6 'PipelineSlotLease\s*(=|:|{)|using PipelineSlotLease|typedef' src python -g '*.h' -g '*.hpp' -g '*.cpp' -g '*.cc' -g '*.py' | sed -n '1,240p'
echo
echo "== includes around worker_manager.h lease references =="
sed -n '1,35p' src/common/hierarchical/worker_manager.h
sed -n '1,35p' src/common/hierarchical/types.h
echo
echo "== call_config includes in worker_manager =="
rg -n '`#include`.*call_config|`#include`.*types|using PipelineSlotLease|struct PipelineSlotLease|PipelineSlotLease' src/common/hierarchical/worker_manager.h src/common/hierarchical/worker_manager.cpp src/common -g '*.h' -g '*.hpp' -g '*.cpp' | sed -n '1,160p'Repository: hw-native-sys/simpler
Length of output: 371
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== all PipelineSlotLease definitions/usages =="
rg -n -C4 'PipelineSlotLease' src python -g '*.h' -g '*.hpp' -g '*.cpp' -g '*.cc' -g '*.py' | sed -n '1,260p'
echo
echo "== file excerpts including PipelineSlotLease =="
python3 - <<'PY'
from pathlib import Path
terms = ['PipelineSlotLease', 'struct PipelineSlotLease', 'using PipelineSlotLease', 'typedef']
for p in Path('.').rglob('*'):
if not p.is_file():
continue
if not any(s.endswith(part) for part in ['.h','.hpp','.cpp','.cc','.py']):
continue
try:
text = p.read_text(encoding='utf-8')
except Exception:
continue
if 'PipelineSlotLease' not in text:
continue
lines = text.splitlines()
hits = [i for i,l in enumerate(lines, 1) if 'PipelineSlotLease' in l]
print(f'== {p} ==')
for hit in hits:
start=max(1, hit-3); end=min(len(lines), hit+3)
for n in range(start, end+1):
print(f'{n:6d}\t{lines[n-1]}')
PYRepository: hw-native-sys/simpler
Length of output: 19623
Match PipelineSlotLease to the C++ packed wire size.
src/common/worker/pto_runtime_c_api.h defines the C/C ABI type as uint32_t slot_id, uint32_t reserved, uint64_t generation, giving a packed size of 12 bytes and making _OFF_ARGS = MAILBOX_OFF_ARGS. "_IIQ" is a native-Q struct, so it can insert 4 bytes of padding if struct alignment > 4, making _OFF_ARGS = _OFF_PIPELINE_LEASE + 16 and writing the args blob 4 bytes past the C++ offest. Add an explicit native-4-aligned uint64 type or declare the C ABI form as packed for the wire layout.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@python/simpler/worker.py` around lines 180 - 186, Update _PIPELINE_LEASE_FMT
to use the C ABI’s packed 12-byte layout for PipelineSlotLease: uint32 slot_id,
uint32 reserved, and uint64 generation with no padding between fields. Ensure
_OFF_ARGS resolves to the established MAILBOX_OFF_ARGS offset and remains
compatible with the C++ definition in pto_runtime_c_api.h.
| raise RuntimeError(f"chip_process dev={device_id}: invalid pipeline lease reserved field") | ||
| # Hand the mailbox bytes straight to C++ (zero-copy zero-decode): | ||
| # the blob layout is what `write_blob` already wrote, so re-parsing | ||
| # it in Python is N×40B of avoidable work and a permanent |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Ruff RUF003: ambiguous × in the comment.
Replace with x to keep the lint clean.
✏️ Proposed fix
- # it in Python is N×40B of avoidable work and a permanent
+ # it in Python is N x 40B of avoidable work and a permanent📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| # it in Python is N×40B of avoidable work and a permanent | |
| # it in Python is N x 40B of avoidable work and a permanent |
🧰 Tools
🪛 Ruff (0.15.21)
[warning] 1544-1544: Comment contains ambiguous × (MULTIPLICATION SIGN). Did you mean x (LATIN SMALL LETTER X)?
(RUF003)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@python/simpler/worker.py` at line 1544, In the comment near the worker
implementation, replace the ambiguous multiplication symbol “×” in “N×40B” with
the ASCII character “x”, preserving the rest of the comment unchanged.
Source: Linters/SAST tools
| // Both kernels are enqueued. Publish before reap_run synchronizes either | ||
| // stream; the parent retains mailbox ownership until TASK_DONE. | ||
| publish_task_accepted(); | ||
|
|
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Acceptance is published as a transient mailbox state that a polling parent can miss.
publish_task_accepted() stores TASK_ACCEPTED into the same word the child later overwrites with TASK_DONE. LocalMailboxEndpoint::run_with_accept (src/common/hierarchical/worker_manager.cpp:315-436) only invokes on_accept when it samples TASK_ACCEPTED; the TASK_DONE branch has no fallback. For a short run the parent's poll can go straight from TASK_READY to TASK_DONE, so on_accept never fires and Orchestrator::decrement_run_accepts is never called for that dispatch — pending_accepts stays non-zero and wait_run_accepted only unblocks later via the terminal-phase override.
Suggest making acceptance edge-durable (e.g. a separate accepted flag/counter the parent latches, or fire on_accept unconditionally on the TASK_DONE path when acceptance was never observed).
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/a2a3/platform/onboard/host/device_runner.cpp` around lines 621 - 624,
Make acceptance durable across the mailbox transition in the dispatch flow
around publish_task_accepted and LocalMailboxEndpoint::run_with_accept: ensure a
parent that observes TASK_DONE without previously sampling TASK_ACCEPTED still
invokes on_accept exactly once. Use a latched acceptance flag/counter or
equivalent state, and preserve the existing behavior for parents that observe
TASK_ACCEPTED first so decrement_run_accepts is not duplicated.
| void Orchestrator::decrement_run_accepts(RunId run_id) { | ||
| auto run = find_run(run_id); | ||
| if (run == nullptr) return; | ||
| int32_t remaining = run->pending_accepts.fetch_sub(1, std::memory_order_acq_rel) - 1; | ||
| if (remaining < 0) { | ||
| run->pending_accepts.store(0, std::memory_order_release); | ||
| record_run_error( | ||
| run, std::make_exception_ptr(std::logic_error("Orchestrator: run acceptance count underflow")) | ||
| ); | ||
| } | ||
| if (remaining <= 0) run->completion_cv.notify_all(); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
Lost-wakeup: pending_accepts is mutated outside completion_mu, but wait_run_accepted predicates on it.
wait_run_accepted evaluates acceptance_ready() while holding run->completion_mu. decrement_run_accepts never takes that mutex — it decrements atomically and notifies. If the decrement to 0 and notify_all() land after the waiter evaluated the predicate but before it parks on the condvar, the waiter blocks forever (no later notify is guaranteed for a run whose tasks all accepted but which is not yet terminal).
Interpose the mutex before notifying (compare with finish_run_if_ready, which mutates phase under the lock).
🔒 Proposed fix
if (remaining < 0) {
run->pending_accepts.store(0, std::memory_order_release);
record_run_error(
run, std::make_exception_ptr(std::logic_error("Orchestrator: run acceptance count underflow"))
);
}
- if (remaining <= 0) run->completion_cv.notify_all();
+ if (remaining <= 0) {
+ // Serialize with wait_run_accepted's predicate evaluation so a
+ // decrement that races a parking waiter cannot be missed.
+ { std::lock_guard<std::mutex> lk(run->completion_mu); }
+ run->completion_cv.notify_all();
+ }
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| void Orchestrator::decrement_run_accepts(RunId run_id) { | |
| auto run = find_run(run_id); | |
| if (run == nullptr) return; | |
| int32_t remaining = run->pending_accepts.fetch_sub(1, std::memory_order_acq_rel) - 1; | |
| if (remaining < 0) { | |
| run->pending_accepts.store(0, std::memory_order_release); | |
| record_run_error( | |
| run, std::make_exception_ptr(std::logic_error("Orchestrator: run acceptance count underflow")) | |
| ); | |
| } | |
| if (remaining <= 0) run->completion_cv.notify_all(); | |
| } | |
| void Orchestrator::decrement_run_accepts(RunId run_id) { | |
| auto run = find_run(run_id); | |
| if (run == nullptr) return; | |
| int32_t remaining = run->pending_accepts.fetch_sub(1, std::memory_order_acq_rel) - 1; | |
| if (remaining < 0) { | |
| run->pending_accepts.store(0, std::memory_order_release); | |
| record_run_error( | |
| run, std::make_exception_ptr(std::logic_error("Orchestrator: run acceptance count underflow")) | |
| ); | |
| } | |
| if (remaining <= 0) { | |
| // Serialize with wait_run_accepted's predicate evaluation so a | |
| // decrement that races a parking waiter cannot be missed. | |
| { std::lock_guard<std::mutex> lk(run->completion_mu); } | |
| run->completion_cv.notify_all(); | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/common/hierarchical/orchestrator.cpp` around lines 360 - 371, Update
Orchestrator::decrement_run_accepts to acquire run->completion_mu before
changing pending_accepts and before calling completion_cv.notify_all(). Keep the
decrement, underflow handling, error recording, and notification behavior intact
while ensuring the predicate transition is synchronized with wait_run_accepted.
| bool ReadyQueue::try_pop(TaskSlot &out) { | ||
| std::lock_guard<std::mutex> lk(mu_); | ||
| if (q_.empty()) return false; | ||
| out = q_.front(); | ||
| q_.pop(); | ||
| for (auto it = queues_.begin(); it != queues_.end();) { | ||
| if (it->second.empty()) { | ||
| it = queues_.erase(it); | ||
| continue; | ||
| } | ||
| out = it->second.front(); | ||
| it->second.pop(); | ||
| if (it->second.empty()) queues_.erase(it); | ||
| return true; | ||
| } | ||
| return false; |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Find unscoped queue calls that would retain the nondeterministic behavior.
rg -nP '\.(try_pop|try_front)\s*\(\s*[^,()]+\s*\)' src tests
rg -nP '\.(try_pop_single|try_front_group|try_pop_group)\s*\(\s*[^,()]+\s*\)' src testsRepository: hw-native-sys/simpler
Length of output: 1824
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Relevant file outline:"
ast-grep outline src/common/hierarchical/types.cpp --view expanded || true
echo
echo "types.cpp relevant ranges:"
sed -n '1,230p' src/common/hierarchical/types.cpp | cat -n
echo
echo "Declaration/header hints:"
fd -a 'types\.h|types\.hpp|types\.hh' . | sed 's#^\./##' | while read -r f; do
echo "--- $f"
rg -n "ReadyQueue|NextLevelReadyQueues|try_pop|try_front" "$f" || true
done
echo
echo "All ReadyQueue-related usage in src/tests:"
rg -n "ReadyQueue|NextLevelReadyQueues|try_pop|try_front|try_pop_single|try_front_group|try_pop_group" src testsRepository: hw-native-sys/simpler
Length of output: 50377
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Behavioral probe for the map-based dequeue selection invariant:
# with an unordered_map where values are FIFO queues, iteration order
# can arbitrarily select any non-empty bucket on each unscoped dequeue.
python3 - <<'PY'
from collections import deque
queues = {}
for run in ["run_early", "run_late"]:
queues[run] = deque([run, run, run])
selections = []
while queues:
selected_runs = []
for label, q in queues.items():
if q:
selected_runs.append(label)
q.popleft()
if selected_runs:
selections.append(selected_runs)
else:
break
print("dequeueable buckets observed across deterministic insertion order:", len(set(tuple(s) for s in selections)))
print("sample order:", selections[:5])
PYRepository: hw-native-sys/simpler
Length of output: 324
Preserve FIFO ordering for unscoped ReadyQueue pops.
The unscoped overloads (try_pop(TaskSlot&) and try_front(TaskSlot&)) iterate ReadyQueue::queues_, a std::unordered_map, so they can dequeue from any non-empty run bucket. This is reached by the scheduler’s unscoped Next-level calls, including try_pop_group(TaskSlot&) and try_pop_single(worker_id, TaskSlot&); remove the unscoped overloads or track run insertion order separately.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/common/hierarchical/types.cpp` around lines 70 - 82, Update ReadyQueue’s
unscoped access paths, including try_pop(TaskSlot&) and try_front(TaskSlot&), so
they no longer select buckets by unordered_map iteration order; remove these
overloads or add separate run-insertion-order tracking that preserves FIFO, and
update callers such as try_pop_group(TaskSlot&) and try_pop_single(worker_id,
TaskSlot&) accordingly.
| bool acceptance_observed = false; | ||
| while (true) { | ||
| MailboxState state = read_mailbox_state(); | ||
| if (state == MailboxState::TASK_ACCEPTED && !acceptance_observed) { | ||
| acceptance_observed = true; | ||
| if (on_accept) on_accept(); | ||
| } | ||
| if (state == MailboxState::TASK_DONE) break; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Trace what the on_accept callback wired by WorkerManager::start actually does.
rg -nP -C6 'on_accept' src/common/hierarchical/ | head -100
ast-grep run --pattern 'void Orchestrator::mark_task_accepted($$$) { $$$ }' --lang cpp src/common/hierarchical/orchestrator.cppRepository: hw-native-sys/simpler
Length of output: 8612
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "worker_manager.cpp relevant lines:"
sed -n '360,430p' src/common/hierarchical/worker_manager.cpp | cat -n
echo
echo "all start/on_accept usages:"
rg -n -C4 'start\(|on_accept' src/common/hierarchical src || true
echo
echo "LocalMailboxEndpoint run/lock methods:"
rg -n -C3 'LocalMailboxEndpoint|mailbox_mu_|read_mailbox_state|run_with_accept|dispatch_process|control_' src/common/hierarchical/worker_manager.cpp src/common/hierarchical/worker_manager.hRepository: hw-native-sys/simpler
Length of output: 50378
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "worker_manager.cpp lines 330-425:"
sed -n '330,425p' src/common/hierarchical/worker_manager.cpp | cat -n
echo
echo "worker_manager.cpp lines 100-220:"
sed -n '100,220p' src/common/hierarchical/worker_manager.cpp | cat -n
echo
echo "focused matches within src/common/hierarchical:"
rg -n -C4 '\bstart\s*\(|\bWorkerManager::start\b|\bon_accept\s*=|on_accept_' src/common/hierarchical/worker_manager.* worker_manager.* 2>/dev/null || trueRepository: hw-native-sys/simpler
Length of output: 20663
Move acceptance out of the mailbox lock.
on_accept_ is invoked inside LocalMailboxEndpoint::run() while mailbox_mu_ is held, so any orchestration performed by the callback (for example FIFO activation/state publication) extends the window during which control_* requests for this endpoint may block. Fire the callback once the lock is released, or otherwise separate the callback from the mailbox round-trip.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/common/hierarchical/worker_manager.cpp` around lines 397 - 404, Move the
acceptance callback out of the mailbox-locked section in the worker task flow
around read_mailbox_state and LocalMailboxEndpoint::run. Record that
TASK_ACCEPTED was observed while polling, then invoke on_accept_ only after the
mailbox round-trip and mailbox_mu_ have been released, preserving exactly-once
callback behavior.
| buffer, tensor = self._tensor_from_host_buffer(st_worker, value) | ||
| buffers.append(buffer) | ||
| tensors.append(tensor) | ||
| first_a, first_b, first_out, second_a, second_b, second_out = tensors |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
tensors.clear() does not drop the unpacked tensor references.
Line 114 binds first_a … second_out to the same torch.frombuffer views, and those locals are still live in the frame when free_host_buffer runs, so the shm pages stay exported and each free emits the "still has a live view" warning. Drop the unpacked names too.
🧹 Proposed fix
tensors.clear()
+ first_a = first_b = first_out = second_a = second_b = second_out = None
if all(handle is None or handle.done for handle in handles):(or del first_a, first_b, first_out, second_a, second_b, second_out guarded for the early-failure path)
Also applies to: 178-181
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/st/a2a3/host_build_graph/worker_async_fifo/test_worker_async_fifo.py`
at line 114, After unpacking tensors into first_a, first_b, first_out, second_a,
second_b, and second_out, explicitly delete those references before
free_host_buffer runs; apply the same cleanup at the corresponding later
unpacking block, while guarding it for early-failure paths where the names may
not be bound.
1 participant
Summary
Stack
Depends on #1540. The intended merge order is #1539 -> #1467 -> #1540 -> this PR. Until the earlier PRs land, GitHub shows their ancestor commits in this PR as well; the B2 change itself is commit
ae227665.Validation
Validated from
/data/sunkaixuan/lcw_subdir/simpler-worker-async-batch2onliteserver-hps-148e:task_20260727_075417_7948633725, exit 0.task_20260727_075434_7971071027, 1 passed on device 1. It verifies N+1 graph build-ahead, no N+1 NPU dispatch before N is terminal, third-submit backpressure, slot1 execution, and output correctness.Per the agreed scope, no simulation test was run; GitHub simulation CI remains a later automatic-regression step.
Non-goals
This PR establishes bounded whole-run admission and FIFO dispatch. Prepared-provider overlap and later scheduler/backend split steps remain separate follow-up PRs.