Add TMR prepared state pipeline - #1588
Conversation
- Separate control traffic from two identified task frames - Accept a queued frame while one sequential child executor is active - Bound parent dispatch capacity and fill every available frame - Retire publish ordering across every pre-publish failure path - Reject oversized argument blobs before publishing mailbox state - Preserve single-frame fallbacks on unsupported local backends - Keep failure-path tests and host-buffer cleanup terminal-safe - Release the GIL around native execution so admission remains live
📝 WalkthroughWalkthroughThis change adds generation-safe pipeline leases, whole-run FIFO admission, prepared-run runtime APIs, image-aware AICore stream reuse, and a two-frame mailbox protocol with prepare/activate states. Scheduler, worker, binding, runtime, documentation, and test coverage are updated accordingly. ChangesPrepared pipeline execution
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related issues
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: 12
🧹 Nitpick comments (10)
tests/ut/py/test_callable_identity.py (1)
604-604: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse the shared pipeline-depth constant instead of the literal.
In this test path, the default depth comes from
PTO_PIPELINE_MAX_DEPTH, so this assertion should reference the exported worker constant rather than hard-coding2.Suggested change
- assert fake_c_worker.pipeline_depth == 2 + assert fake_c_worker.pipeline_depth == worker_mod.PTO_PIPELINE_MAX_DEPTH🤖 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/py/test_callable_identity.py` at line 604, Update the assertion for fake_c_worker.pipeline_depth to compare against the exported PTO_PIPELINE_MAX_DEPTH constant instead of the hard-coded literal 2, preserving the existing default-depth validation.tests/ut/cpp/hierarchical/test_orchestrator.cpp (1)
752-772: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueFatal assertions can leave a pending
std::asyncfuture, turning a failed invariant into a CI hang. In each site, aASSERT_*onwait_for(...) == readyreturns from the test body while the async task may still be blocked, and~std::futurefromstd::asyncthen blocks forever.
tests/ut/cpp/hierarchical/test_orchestrator.cpp#L752-L772: unconditionally drain the remaining lease (consumeprepared) before the fatal assertion onthird.tests/ut/cpp/hierarchical/test_orchestrator.cpp#L805-L817: unconditionally consumeactivebefore the fatal assertion onreplacement.tests/ut/cpp/hierarchical/test_scheduler.cpp#L555-L566: unblock the second dispatch (writeTASK_ACTIVE) before the fatal assertion, or downgrade it toEXPECT_*plus an explicit unblock.🤖 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 752 - 772, Prevent fatal assertions from destroying still-blocked std::async futures: in tests/ut/cpp/hierarchical/test_orchestrator.cpp lines 752-772, always consume prepared via its task slot before the fatal third readiness assertion; in tests/ut/cpp/hierarchical/test_orchestrator.cpp lines 805-817, always consume active before the fatal replacement readiness assertion; in tests/ut/cpp/hierarchical/test_scheduler.cpp lines 555-566, write TASK_ACTIVE to unblock the second dispatch before the fatal assertion, or replace it with an EXPECT_* followed by explicit unblocking.src/common/hierarchical/scheduler.cpp (1)
365-402: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider extracting the shared preparable-eligibility predicate.
preparable_work_ready(),dispatch_preparable_next_level_group(), anddispatch_preparable_next_level_singles()each re-implement the same READY/run_id/worker_type/group/diagnostics/supports_prepare_activate/capacity/ACK checks. Any future change has to be made in three places consistently, and a divergence silently turns the CV predicate into a spurious wakeup source or a missed wakeup. A singlebool preparable_slot_eligible(TaskSlot, RunId, RunId)helper used by all three would remove that risk.Also applies to: 414-476
🤖 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 365 - 402, Extract the shared eligibility checks from preparable_work_ready(), dispatch_preparable_next_level_group(), and dispatch_preparable_next_level_singles() into a preparable_slot_eligible(TaskSlot, RunId, RunId) helper. Have all three callers use this helper for the READY, run, worker type, group, diagnostics, supports_prepare_activate, capacity, and active-run ACK checks, preserving their existing group/single-specific behavior.src/common/platform/onboard/host/c_api_shared.cpp (1)
681-688: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
format_prepared_attrsemits duplicated values under different keys.
request=/run=both printidentity.run_id, andepoch=/generation=both printidentity.generation. Traces will look like they carry four independent identity fields when only two exist — drop the aliases (or populate them from real fields).♻️ Proposed trim
- attrs, attrs_size, "request=%llu run=%llu epoch=%llu slot=%u generation=%llu dispatch=%llu", - static_cast<unsigned long long>(identity.run_id), static_cast<unsigned long long>(identity.run_id), - static_cast<unsigned long long>(identity.generation), identity.slot_id, - static_cast<unsigned long long>(identity.generation), static_cast<unsigned long long>(identity.dispatch_id) + attrs, attrs_size, "run=%llu slot=%u generation=%llu dispatch=%llu", + static_cast<unsigned long long>(identity.run_id), identity.slot_id, + static_cast<unsigned long long>(identity.generation), static_cast<unsigned long long>(identity.dispatch_id)🤖 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/c_api_shared.cpp` around lines 681 - 688, Update format_prepared_attrs to stop emitting duplicate identity values under request/run and epoch/generation aliases; retain only keys backed by distinct PreparedRunIdentity fields, while preserving the existing slot and dispatch attributes.src/common/hierarchical/worker_manager.h (2)
148-156: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winAdd the symmetric bound assert for the control-frame region.
MAILBOX_OFF_TASK_ARGS_BLOBis now guarded against the frame-protocol trailer, butMAILBOX_OFF_CONTROL_CALLABLE_HASH(+CALLABLE_HASH_DIGEST_SIZE) has no equivalent check even though the control frame shares these offsets. It fits comfortably today; the assert is what keeps a futureCTRL_SHM_NAME_BYTESor digest-size change from silently overwriting the trailer.🛡️ Proposed assert
static_assert( MAILBOX_OFF_TASK_ARGS_BLOB < MAILBOX_OFF_FRAME_PROTOCOL, "mailbox task-args region must precede the frame protocol trailer" ); +static_assert( + MAILBOX_OFF_CONTROL_CALLABLE_HASH + static_cast<ptrdiff_t>(CALLABLE_HASH_DIGEST_SIZE) <= + MAILBOX_OFF_FRAME_PROTOCOL, + "mailbox control region must precede the frame protocol trailer" +);🤖 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.h` around lines 148 - 156, Add a static_assert next to the existing task-args bound check that verifies MAILBOX_OFF_CONTROL_CALLABLE_HASH plus CALLABLE_HASH_DIGEST_SIZE remains before MAILBOX_OFF_FRAME_PROTOCOL. Keep the existing offset and capacity definitions unchanged.
97-97: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDeclare
MAILBOX_TASK_PROTOCOL_VERSIONasuint64_tto match its wire slot.The field occupies the 8-byte
MAILBOX_OFF_FRAME_PROTOCOLslot and is unpacked as"=Q"on the Python side, sorun_two_framehas to widen it into a localuint64_t protocolbefore thememcpy. Declaring the constant at wire width documents the slot and drops the conversion step.🤖 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.h` at line 97, Change MAILBOX_TASK_PROTOCOL_VERSION from uint32_t to uint64_t so it matches the 8-byte MAILBOX_OFF_FRAME_PROTOCOL wire slot and the Python "=Q" unpacking; update run_two_frame to use the widened uint64_t protocol directly in the memcpy without an intermediate conversion.python/simpler/worker.py (2)
1564-1569: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
publish_native_acceptanceis never passedFalse.Both call sites (the single-frame
_run_mailbox_loopdefault and the two-framehandle_task(...)at Line 1975) useTrue, so the flag and the two conditionals it guards at Lines 1609–1610 are currently unreachable configurability. Dropping it keeps the dispatch path to one shape until a caller actually needs the other.Also applies to: 1972-1976
🤖 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 1564 - 1569, Remove the unused publish_native_acceptance parameter from handle_task and delete the conditionals it controls around the native acceptance publication. Update both handle_task call sites, including _run_mailbox_loop and the two-frame dispatch, so they use the simplified signature without passing True.
198-206: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winConsider exporting the frame-trailer offsets from the bindings instead of recomputing them here.
These five offsets,
_TASK_ACCEPTED,_TASK_PROTOCOL_VERSIONand_MAILBOX_ARGS_CAPACITYare hand-mirrored fromworker_manager.h(MAILBOX_OFF_FRAME_*,MAILBOX_TASK_ACCEPTED,MAILBOX_TASK_PROTOCOL_VERSION,MAILBOX_ARGS_CAPACITY). They agree today, but a future C++ layout change would silently produce garbage reads on the child side rather than a build/import error —MAILBOX_FRAME_SIZEwas exported precisely to avoid that. Exporting the trailer constants the same way (and asserting them in_assert_bindings_match_source_tree) removes the drift surface.🤖 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 198 - 206, The worker bindings currently hand-mirror frame trailer constants, risking drift from the C++ layout. Export MAILBOX_OFF_FRAME_*, MAILBOX_TASK_ACCEPTED, MAILBOX_TASK_PROTOCOL_VERSION, and MAILBOX_ARGS_CAPACITY from the bindings, replace the local values used by python/simpler/worker.py, and extend _assert_bindings_match_source_tree to validate them against worker_manager.h.src/common/hierarchical/worker_manager.cpp (1)
606-893: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftConsider a
fail_endpoint(msg)helper to collapse the repeated poison/notify/release/return block.
run_two_framehas thirteen return points, and eight of them repeat the identical five-line sequence: setendpoint_poisoned_,publish_cv_.notify_all(), setoutcome/error_message,release_claim(),return completion. Every one of them is load-bearing — droppingrelease_claim()on a single path permanently strands a frame, and dropping thenotify_all()strands every publish-sequence waiter. A single helper (or an RAII guard for the claim, givenrelease_claimis already idempotent) makes that invariant structural rather than something each new early-return has to remember.🤖 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 606 - 893, Refactor the repeated poisoned-endpoint failure paths in LocalMailboxEndpoint::run_two_frame by introducing a local fail_endpoint helper that marks endpoint_poisoned_, notifies publish_cv_, assigns the failure outcome and message, releases the claimed frame, and returns completion. Replace the existing matching early-return blocks with this helper while preserving non-poisoning failures and the catch-path cleanup behavior.src/common/worker/chip_worker.cpp (1)
252-255: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider extracting the function-pointer teardown into one helper.
The prepared-run pointers are now the fourth group duplicated across three rollback blocks and
finalize(). Every future ABI symbol has to be added in all of them, and a miss leaves a dangling pointer into adlclosed SO. A singleclear_runtime_bindings()called from each site removes that class of drift.Also applies to: 302-305, 404-407
🤖 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/worker/chip_worker.cpp` around lines 252 - 255, Extract the prepared-run function-pointer teardown from finalize() and the rollback blocks into a shared clear_runtime_bindings() helper. Move all assignments for supports_prepared_run_fn_, prepare_run_fn_, execute_prepared_fn_, and abort_prepared_fn_ into that helper, then call it at each existing teardown site, including the locations around the referenced rollback blocks.
🤖 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 `@python/bindings/task_interface.cpp`:
- Around line 1489-1499: Update the prepare-with-lease binding around
prepare_with_lease so the prepared runtime owns a persistent copy of storage
rather than receiving the stack-local &storage. Ensure that the owned args
remain valid through execute_prepared or abort_prepared, then are released when
the prepared run is consumed or aborted.
In `@python/simpler/worker.py`:
- Around line 1898-1901: Update all four abort_prepared cleanup blocks in the
worker flow to record exceptions instead of silently passing, using the module’s
existing stderr/error-reporting mechanism. Preserve the cleanup attempt and
subsequent fail_frame behavior, while including the abort_prepared exception
details so failed prepared-state reclamation is diagnosable and Ruff S110 is
resolved.
- Around line 1845-1863: Add a short yield to admission_loop after each scan
when no frame is in an actionable state, using time.sleep(0) or the existing
minimal sleep convention. Track whether the scan found a frame in _TASK_READY,
_PREPARE_READY, or _ACTIVATE, and preserve immediate polling when actionable
work is present while preventing idle full-core spinning.
- Around line 1939-1942: Guard the body of the daemon admission_loop so
exceptions from identity/frame parsing or failure reporting are captured as a
terminal action instead of terminating the producer silently. Preserve the
executor’s existing wait predicate and enqueue a shutdown action carrying the
recorded exception cause; when handling that shutdown action, pass the cause to
_write_error before returning.
In `@src/common/hierarchical/orchestrator.cpp`:
- Around line 345-354: Update the failure handling loop in cancel_unstarted_run
so failure_message is written only after successfully transitioning a PENDING or
READY slot to FAILED. Use the slot-state mutex synchronization path established
by slot_state and the corresponding completion/poison handling, ensuring
competing writers and readers cannot race.
In `@src/common/hierarchical/scheduler.cpp`:
- Around line 404-412: Update dispatch_preparable_next_level_group() and
dispatch_preparable_next_level_singles() so each preparable run’s staged-slot
availability is checked per worker before staging. Ensure group selection is
completed or isolated before singles are dispatched, preventing a group and
single from the same preparable FIFO from staging the same run in one
dispatch_ready() pass and preserving dispatch_prepared()’s one-staged-run
assertion.
In `@src/common/platform/onboard/host/device_runner_base.cpp`:
- Line 254: Synchronize all accesses to the single-entry prebuilt arena cache
used by simpler_prepare_run and simpler_execute_prepared, since
selected_arena_bank() is thread-local and the existing shared
g_prepared_run_gate lock does not protect cache mutation. Add and consistently
acquire a mutex around prebuilt_runtime_arena_cache_* reads, rebuilds, and
clears, preserving the existing bank-0 behavior.
In `@src/common/platform/onboard/host/device_runner_base.h`:
- Around line 415-416: Update the documentation for activate_launch_shape to
clearly state that the prepared Runtime’s geometry is latched immediately before
the device launch submit step, replacing the truncated “Device S.” wording.
In `@src/common/worker/chip_worker.cpp`:
- Around line 669-689: Update ChipWorker::abort_prepared to claim the matching
PREPARED slot while holding prepared_slots_mu_ before invoking
abort_prepared_fn_. Transition it to an ABORTING state or clear it atomically
under the lock so concurrent aborts and execute_prepared cannot reuse the slot;
retain the identity validation and ensure the claimed slot is finalized
consistently after the backend call.
- Around line 636-658: Ensure pre-ABI failures in the prepared execution flow
abort the runtime before clearing its slot state. Update the
set_task_accepted_state_fn_ failure path and the select_slot_resources exception
path to invoke abort_prepared_fn_ for the corresponding runtime buffer, while
preserving the existing slot reset and exception propagation behavior; do not
add abort handling to the successful execute_prepared_fn_ path where the ABI
owns runtime destruction.
In `@tests/st/a2a3/host_build_graph/run_stream_reuse/test_run_stream_reuse.py`:
- Around line 136-139: Update the docstrings in the Python wrapper definitions
around worker stream reuse, specifically in simpler.worker and
bindings.task_interface, to describe flat-stream semantics:
ChipWorker::run_stream_set_create_count advances when the loaded image changes,
while the AICore stream is reused across runs with the same image. Remove
wording that says the counter advances once per run, keeping the behavior and
API unchanged.
In
`@tests/st/a2a3/host_build_graph/worker_async_endpoint/test_worker_async_endpoint.py`:
- Around line 113-122: Throttle the observation loop around
saw_active_and_accepted with the suite’s established 1 ms sleep so polling does
not starve dispatch. Make the assertion tolerate run.done occurring before a
sample by preserving success when both frames complete normally, and increase
_CHAIN_LENGTH from 64 to a longer value such as the FIFO suite’s 4096 to widen
the active window for sampling.
---
Nitpick comments:
In `@python/simpler/worker.py`:
- Around line 1564-1569: Remove the unused publish_native_acceptance parameter
from handle_task and delete the conditionals it controls around the native
acceptance publication. Update both handle_task call sites, including
_run_mailbox_loop and the two-frame dispatch, so they use the simplified
signature without passing True.
- Around line 198-206: The worker bindings currently hand-mirror frame trailer
constants, risking drift from the C++ layout. Export MAILBOX_OFF_FRAME_*,
MAILBOX_TASK_ACCEPTED, MAILBOX_TASK_PROTOCOL_VERSION, and MAILBOX_ARGS_CAPACITY
from the bindings, replace the local values used by python/simpler/worker.py,
and extend _assert_bindings_match_source_tree to validate them against
worker_manager.h.
In `@src/common/hierarchical/scheduler.cpp`:
- Around line 365-402: Extract the shared eligibility checks from
preparable_work_ready(), dispatch_preparable_next_level_group(), and
dispatch_preparable_next_level_singles() into a
preparable_slot_eligible(TaskSlot, RunId, RunId) helper. Have all three callers
use this helper for the READY, run, worker type, group, diagnostics,
supports_prepare_activate, capacity, and active-run ACK checks, preserving their
existing group/single-specific behavior.
In `@src/common/hierarchical/worker_manager.cpp`:
- Around line 606-893: Refactor the repeated poisoned-endpoint failure paths in
LocalMailboxEndpoint::run_two_frame by introducing a local fail_endpoint helper
that marks endpoint_poisoned_, notifies publish_cv_, assigns the failure outcome
and message, releases the claimed frame, and returns completion. Replace the
existing matching early-return blocks with this helper while preserving
non-poisoning failures and the catch-path cleanup behavior.
In `@src/common/hierarchical/worker_manager.h`:
- Around line 148-156: Add a static_assert next to the existing task-args bound
check that verifies MAILBOX_OFF_CONTROL_CALLABLE_HASH plus
CALLABLE_HASH_DIGEST_SIZE remains before MAILBOX_OFF_FRAME_PROTOCOL. Keep the
existing offset and capacity definitions unchanged.
- Line 97: Change MAILBOX_TASK_PROTOCOL_VERSION from uint32_t to uint64_t so it
matches the 8-byte MAILBOX_OFF_FRAME_PROTOCOL wire slot and the Python "=Q"
unpacking; update run_two_frame to use the widened uint64_t protocol directly in
the memcpy without an intermediate conversion.
In `@src/common/platform/onboard/host/c_api_shared.cpp`:
- Around line 681-688: Update format_prepared_attrs to stop emitting duplicate
identity values under request/run and epoch/generation aliases; retain only keys
backed by distinct PreparedRunIdentity fields, while preserving the existing
slot and dispatch attributes.
In `@src/common/worker/chip_worker.cpp`:
- Around line 252-255: Extract the prepared-run function-pointer teardown from
finalize() and the rollback blocks into a shared clear_runtime_bindings()
helper. Move all assignments for supports_prepared_run_fn_, prepare_run_fn_,
execute_prepared_fn_, and abort_prepared_fn_ into that helper, then call it at
each existing teardown site, including the locations around the referenced
rollback blocks.
In `@tests/ut/cpp/hierarchical/test_orchestrator.cpp`:
- Around line 752-772: Prevent fatal assertions from destroying still-blocked
std::async futures: in tests/ut/cpp/hierarchical/test_orchestrator.cpp lines
752-772, always consume prepared via its task slot before the fatal third
readiness assertion; in tests/ut/cpp/hierarchical/test_orchestrator.cpp lines
805-817, always consume active before the fatal replacement readiness assertion;
in tests/ut/cpp/hierarchical/test_scheduler.cpp lines 555-566, write TASK_ACTIVE
to unblock the second dispatch before the fatal assertion, or replace it with an
EXPECT_* followed by explicit unblocking.
In `@tests/ut/py/test_callable_identity.py`:
- Line 604: Update the assertion for fake_c_worker.pipeline_depth to compare
against the exported PTO_PIPELINE_MAX_DEPTH constant instead of the hard-coded
literal 2, preserving the existing default-depth validation.
🪄 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: ab67315d-725f-410e-ac84-b0da89488e6d
📒 Files selected for processing (41)
docs/task-flow.mdpython/bindings/task_interface.cpppython/bindings/worker_bind.hpython/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/docs/RUNTIME_LOGIC.mdsrc/a2a3/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/include/host/run_stream_slots.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/worker/chip_worker.cppsrc/common/worker/chip_worker.hsrc/common/worker/pipeline_slot_pool.hsrc/common/worker/pto_runtime_c_api.htests/st/a2a3/host_build_graph/run_stream_reuse/test_run_stream_reuse.pytests/st/a2a3/host_build_graph/worker_async_endpoint/kernels/orchestration/long_vector_orch.cpptests/st/a2a3/host_build_graph/worker_async_endpoint/test_worker_async_endpoint.pytests/st/a2a3/host_build_graph/worker_async_fifo/kernels/orchestration/pipelined_vector_orch.cpptests/st/a2a3/host_build_graph/worker_async_fifo/test_worker_async_fifo.pytests/ut/cpp/CMakeLists.txttests/ut/cpp/common/test_trb_runtime_temp_buffer.cpptests/ut/cpp/hierarchical/test_orchestrator.cpptests/ut/cpp/hierarchical/test_pipeline_contract.cpptests/ut/cpp/hierarchical/test_run_stream_slots.cpptests/ut/cpp/hierarchical/test_scheduler.cpptests/ut/py/test_callable_identity.pytests/ut/py/test_worker/test_host_worker.py
| [](ChipWorker &self, int32_t callable_id, uint64_t args_blob_ptr, size_t blob_capacity, | ||
| const CallConfig &config, uint32_t pipeline_slot, uint64_t pipeline_generation, uint64_t run_id, | ||
| uint64_t dispatch_id) { | ||
| nb::gil_scoped_release release; | ||
| TaskArgsView view = read_blob(reinterpret_cast<const uint8_t *>(args_blob_ptr), blob_capacity); | ||
| ChipStorageTaskArgs storage = view_to_chip_storage(view); | ||
| self.prepare_with_lease( | ||
| callable_id, &storage, config, PipelineSlotLease{pipeline_slot, 0, pipeline_generation}, run_id, | ||
| dispatch_id | ||
| ); | ||
| }, |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
fd -t f 'chip_worker.(h|cpp)' src | xargs -r ast-grep outline --items all
rg -nP -C25 '\bprepare_with_lease\s*\(' --type=cpp --type=cc srcRepository: hw-native-sys/simpler
Length of output: 8886
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== chip_worker.h prepare/execute declarations =="
rg -n -C 40 'execute_prepared|prepare_with_lease|PreparedSlot|prepared_slots_' src/common/worker/chip_worker.h src/common/worker/chip_worker.cpp
echo
echo "== task_interface binding slice =="
sed -n '1450,1535p' python/bindings/task_interface.cppRepository: hw-native-sys/simpler
Length of output: 50377
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== source files mentioning SimplerPrepareRunFn/simpler_prepare_run =="
rg -n -C 12 'SimplerPrepareRunFn|simpler_prepare_run|prepare_with_lease|PreparedSlot' src python bindings || true
echo
echo "== exact chip_worker.cpp implementation slices =="
sed -n '542,690p' src/common/worker/chip_worker.cppRepository: hw-native-sys/simpler
Length of output: 45358
Keep the args lifetime matching the prepared-run contract.
prepare_with_lease is defined to construct a live runtime object bound with these args, persist it until execute_prepared/abort_prepared, then consume/destroy it. Passing this lambda’s stack-local &storage violates that contract and makes the later prepared execution read freed memory; store the args on a heap/runtime-owned object, or otherwise ensure the prepared runtime owns a copy until execute/abort completion.
🤖 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/bindings/task_interface.cpp` around lines 1489 - 1499, Update the
prepare-with-lease binding around prepare_with_lease so the prepared runtime
owns a persistent copy of storage rather than receiving the stack-local
&storage. Ensure that the owned args remain valid through execute_prepared or
abort_prepared, then are released when the prepared run is consumed or aborted.
| def admission_loop() -> None: | ||
| nonlocal control_queued | ||
| while not stop_admission.is_set(): | ||
| control_state = _mailbox_load_i32(state_addr) | ||
| with action_cv: | ||
| if control_state == _SHUTDOWN: | ||
| actions.append(("shutdown", -1, None)) | ||
| action_cv.notify() | ||
| return | ||
| if control_state == _CONTROL_REQUEST and not control_queued: | ||
| control_queued = True | ||
| actions.append(("control", -1, None)) | ||
| action_cv.notify() | ||
|
|
||
| for index, (frame_buf, frame_addr) in enumerate(zip(frame_bufs, frame_addrs, strict=True)): | ||
| frame_state_addr = frame_addr + _OFF_STATE | ||
| frame_state = _mailbox_load_i32(frame_state_addr) | ||
| if frame_state not in (_TASK_READY, _PREPARE_READY, _ACTIVATE): | ||
| continue |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
The admission thread spins with no yield, so the child now burns a full core and contends for the GIL with the executor.
admission_loop polls the control word and both frame state words in a tight while with no sleep/yield. The single-frame loop could justify that as the task's own latency path, but this is now a second Python thread running for the child's entire lifetime alongside the executor thread; every GIL slice it takes delays the executor's Python-level phases (identity reads, _write_error, state stores) between device calls. A very short time.sleep(0)/sleep(1e-5) on an idle scan (no frame in an actionable state) keeps the hot handoff fast while giving back the core.
🤖 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 1845 - 1863, Add a short yield to
admission_loop after each scan when no frame is in an actionable state, using
time.sleep(0) or the existing minimal sleep convention. Track whether the scan
found a frame in _TASK_READY, _PREPARE_READY, or _ACTIVATE, and preserve
immediate polling when actionable work is present while preventing idle
full-core spinning.
| try: | ||
| abort_prepared(prepared_identity) | ||
| except Exception: # noqa: BLE001 | ||
| pass |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Silently swallowed abort_prepared failures leave unreclaimable prepared state with no diagnostic.
The same try/except: pass appears four times (Lines 1898–1901, 1917–1920, 1959–1962, 1983–1986). abort_prepared is the only path that reclaims an unpublished prepared run; if it throws, the device-side epoch leaks and nothing records why — the subsequent fail_frame message describes the identity mismatch, not the failed reclaim. Log the exception (stderr, as the rest of this module does) or fold it into the frame's error message. Flagged by Ruff (S110).
🧰 Tools
🪛 Ruff (0.16.0)
[error] 1900-1901: try-except-pass detected, consider logging the exception
(S110)
🤖 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 1898 - 1901, Update all four
abort_prepared cleanup blocks in the worker flow to record exceptions instead of
silently passing, using the module’s existing stderr/error-reporting mechanism.
Preserve the cleanup attempt and subsequent fail_frame behavior, while including
the abort_prepared exception details so failed prepared-state reclamation is
diagnosable and Ruff S110 is resolved.
Source: Linters/SAST tools
| while True: | ||
| with action_cv: | ||
| action_cv.wait_for(lambda: bool(actions)) | ||
| kind, index, identity = actions.pop(0) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
An unhandled exception in admission_loop wedges the child forever.
admission_loop is a daemon thread whose body is not guarded: read_identity, _mailbox_load_i32, fail_frame/_write_error and _PIPELINE_LEASE_FMT.unpack_from can all raise (e.g. a struct.error on a corrupted frame). If it dies, the thread disappears silently and this wait_for(lambda: bool(actions)) blocks with no timeout and no producer left to notify it. The child stays alive, so the parent's check_child_death liveness sampling never trips either, and the dispatch hangs until the whole tree is torn down.
Wrap the loop body so a producer failure becomes a terminal action the executor observes.
🛡️ Proposed fix: surface a producer crash instead of hanging
def admission_loop() -> None:
nonlocal control_queued
- while not stop_admission.is_set():
+ try:
+ admission_scan(),
+ except BaseException as exc: # noqa: BLE001 -- producer death must not wedge the executor
+ with action_cv:
+ admission_error.append(_format_exc(f"chip_process dev={device_id} admission", exc))
+ actions.append(("shutdown", -1, None))
+ action_cv.notify()and make the executor's predicate lambda: bool(actions) unchanged while ("shutdown", ...) now also carries the recorded cause into _write_error before returning.
🤖 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 1939 - 1942, Guard the body of the
daemon admission_loop so exceptions from identity/frame parsing or failure
reporting are captured as a terminal action instead of terminating the producer
silently. Preserve the executor’s existing wait predicate and enqueue a shutdown
action carrying the recorded exception cause; when handling that shutdown
action, pass the cause to _write_error before returning.
| for (TaskSlot slot : slots) { | ||
| TaskSlotState &state = slot_state(slot); | ||
| TaskState current = state.state.load(std::memory_order_acquire); | ||
| if (current == TaskState::PENDING || current == TaskState::READY) { | ||
| state.failure_message = message; | ||
| (void)state.state.compare_exchange_strong( | ||
| current, TaskState::FAILED, std::memory_order_release, std::memory_order_acquire | ||
| ); | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
rg -nP -C6 '\bfailure_message\b' src/common/hierarchical
ast-grep run --pattern 'struct TaskSlotState { $$$ }' --lang cpp src/common/hierarchicalRepository: hw-native-sys/simpler
Length of output: 7926
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== relevant orchestrator sections =="
sed -n '190,370p' src/common/hierarchical/orchestrator.cpp
sed -n '620,715p' src/common/hierarchical/orchestrator.cpp
echo "== task state declaration and state field =="
rg -n -C4 'enum class TaskState|struct TaskSlotState|compare_exchange_strong|task_state|slot_state' src/common/hierarchical/types* src/common/hierarchical/scheduler.cpp
echo "== scheduler state transitions around failure =="
sed -n '220,310p' src/common/hierarchical/scheduler.cppRepository: hw-native-sys/simpler
Length of output: 27094
Synchronize failure_message on live task slots.
failure_message is assigned from cancel_unstarted_run() before the CAS and under no slot lock, while Scheduler::on_task_complete() and poison_task() can also write failure_message after transitioning slots to FAILED. These unsynchronized string writes/reads on live task slots can race; assign the message only after winning the transition and publish it through the slot’s established slot-state mutex path.
🤖 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 345 - 354, Update the
failure handling loop in cancel_unstarted_run so failure_message is written only
after successfully transitioning a PENDING or READY slot to FAILED. Use the
slot-state mutex synchronization path established by slot_state and the
corresponding completion/poison handling, ensuring competing writers and readers
cannot race.
| /** Latch a prepared Runtime's geometry immediately before Device S. */ | ||
| void activate_launch_shape(const Runtime &runtime); |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Truncated doc comment.
"immediately before Device S." reads as a cut-off sentence; state what is latched and before which step (e.g. "immediately before the device launch submit").
🤖 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 415 -
416, Update the documentation for activate_launch_shape to clearly state that
the prepared Runtime’s geometry is latched immediately before the device launch
submit step, replacing the truncated “Device S.” wording.
| int rc = -1; | ||
| if (accepted_state != nullptr && set_task_accepted_state_fn_ != nullptr) { | ||
| int bind_rc = set_task_accepted_state_fn_(device_ctx_, accepted_state, accepted_value); | ||
| if (bind_rc != 0) { | ||
| std::scoped_lock lk(prepared_slots_mu_); | ||
| prepared_slots_[lease.slot_id] = PreparedSlot{}; | ||
| throw std::runtime_error("set_task_accepted_state_ctx failed with code " + std::to_string(bind_rc)); | ||
| } | ||
| } | ||
| auto clear_accepted_state = [&]() { | ||
| if (accepted_state != nullptr && set_task_accepted_state_fn_ != nullptr) { | ||
| (void)set_task_accepted_state_fn_(device_ctx_, nullptr, 0); | ||
| } | ||
| }; | ||
| try { | ||
| (void)select_slot_resources(lease.slot_id); | ||
| rc = execute_prepared_fn_(device_ctx_, runtime_bufs_[lease.slot_id].data(), callable_id, &config, &identity); | ||
| } catch (...) { | ||
| clear_accepted_state(); | ||
| std::scoped_lock lk(prepared_slots_mu_); | ||
| prepared_slots_[lease.slot_id] = PreparedSlot{}; | ||
| throw; | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Prepared Runtime leaks when execution fails before entering the ABI.
If set_task_accepted_state_fn_ fails (Line 639) or select_slot_resources throws (Line 651), the slot is reset to PreparedSlot{} without calling abort_prepared_fn_. The Runtime that simpler_prepare_run constructed in runtime_bufs_[slot] is still live, but the identity is now gone from prepared_slots_, so neither abort_prepared nor the finalize() sweep can ever reclaim it — its tensor leases and arena state are held until the process exits. Only the execute_prepared_fn_ return path is safe, because the ABI destroys the Runtime itself.
🛡️ Abort the prepared runtime on the pre-ABI failure paths
int rc = -1;
if (accepted_state != nullptr && set_task_accepted_state_fn_ != nullptr) {
int bind_rc = set_task_accepted_state_fn_(device_ctx_, accepted_state, accepted_value);
if (bind_rc != 0) {
+ (void)abort_prepared_fn_(device_ctx_, runtime_bufs_[lease.slot_id].data(), &identity);
std::scoped_lock lk(prepared_slots_mu_);
prepared_slots_[lease.slot_id] = PreparedSlot{};
throw std::runtime_error("set_task_accepted_state_ctx failed with code " + std::to_string(bind_rc));
}
}
@@
+ bool entered_abi = false;
try {
(void)select_slot_resources(lease.slot_id);
+ entered_abi = true;
rc = execute_prepared_fn_(device_ctx_, runtime_bufs_[lease.slot_id].data(), callable_id, &config, &identity);
} catch (...) {
clear_accepted_state();
+ if (!entered_abi) {
+ (void)abort_prepared_fn_(device_ctx_, runtime_bufs_[lease.slot_id].data(), &identity);
+ }
std::scoped_lock lk(prepared_slots_mu_);
prepared_slots_[lease.slot_id] = PreparedSlot{};
throw;
}📝 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.
| int rc = -1; | |
| if (accepted_state != nullptr && set_task_accepted_state_fn_ != nullptr) { | |
| int bind_rc = set_task_accepted_state_fn_(device_ctx_, accepted_state, accepted_value); | |
| if (bind_rc != 0) { | |
| std::scoped_lock lk(prepared_slots_mu_); | |
| prepared_slots_[lease.slot_id] = PreparedSlot{}; | |
| throw std::runtime_error("set_task_accepted_state_ctx failed with code " + std::to_string(bind_rc)); | |
| } | |
| } | |
| auto clear_accepted_state = [&]() { | |
| if (accepted_state != nullptr && set_task_accepted_state_fn_ != nullptr) { | |
| (void)set_task_accepted_state_fn_(device_ctx_, nullptr, 0); | |
| } | |
| }; | |
| try { | |
| (void)select_slot_resources(lease.slot_id); | |
| rc = execute_prepared_fn_(device_ctx_, runtime_bufs_[lease.slot_id].data(), callable_id, &config, &identity); | |
| } catch (...) { | |
| clear_accepted_state(); | |
| std::scoped_lock lk(prepared_slots_mu_); | |
| prepared_slots_[lease.slot_id] = PreparedSlot{}; | |
| throw; | |
| } | |
| int rc = -1; | |
| if (accepted_state != nullptr && set_task_accepted_state_fn_ != nullptr) { | |
| int bind_rc = set_task_accepted_state_fn_(device_ctx_, accepted_state, accepted_value); | |
| if (bind_rc != 0) { | |
| (void)abort_prepared_fn_(device_ctx_, runtime_bufs_[lease.slot_id].data(), &identity); | |
| std::scoped_lock lk(prepared_slots_mu_); | |
| prepared_slots_[lease.slot_id] = PreparedSlot{}; | |
| throw std::runtime_error("set_task_accepted_state_ctx failed with code " + std::to_string(bind_rc)); | |
| } | |
| } | |
| auto clear_accepted_state = [&]() { | |
| if (accepted_state != nullptr && set_task_accepted_state_fn_ != nullptr) { | |
| (void)set_task_accepted_state_fn_(device_ctx_, nullptr, 0); | |
| } | |
| }; | |
| bool entered_abi = false; | |
| try { | |
| (void)select_slot_resources(lease.slot_id); | |
| entered_abi = true; | |
| rc = execute_prepared_fn_(device_ctx_, runtime_bufs_[lease.slot_id].data(), callable_id, &config, &identity); | |
| } catch (...) { | |
| clear_accepted_state(); | |
| if (!entered_abi) { | |
| (void)abort_prepared_fn_(device_ctx_, runtime_bufs_[lease.slot_id].data(), &identity); | |
| } | |
| std::scoped_lock lk(prepared_slots_mu_); | |
| prepared_slots_[lease.slot_id] = PreparedSlot{}; | |
| throw; | |
| } |
🤖 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/worker/chip_worker.cpp` around lines 636 - 658, Ensure pre-ABI
failures in the prepared execution flow abort the runtime before clearing its
slot state. Update the set_task_accepted_state_fn_ failure path and the
select_slot_resources exception path to invoke abort_prepared_fn_ for the
corresponding runtime buffer, while preserving the existing slot reset and
exception propagation behavior; do not add abort handling to the successful
execute_prepared_fn_ path where the ABI owns runtime destruction.
| void ChipWorker::abort_prepared(const PipelineSlotLease &lease, uint64_t run_id, uint64_t dispatch_id) { | ||
| const PreparedRunIdentity identity{run_id, lease.generation, dispatch_id, lease.slot_id, 0}; | ||
| { | ||
| std::scoped_lock lk(prepared_slots_mu_); | ||
| if (lease.slot_id >= prepared_slots_.size()) return; | ||
| const PreparedSlot &slot = prepared_slots_[lease.slot_id]; | ||
| if (slot.state != PreparedSlotState::PREPARED || slot.identity.run_id != run_id || | ||
| slot.identity.generation != lease.generation || slot.identity.dispatch_id != dispatch_id) { | ||
| return; | ||
| } | ||
| } | ||
| (void)select_slot_resources(lease.slot_id); | ||
| int rc = abort_prepared_fn_(device_ctx_, runtime_bufs_[lease.slot_id].data(), &identity); | ||
| { | ||
| std::scoped_lock lk(prepared_slots_mu_); | ||
| prepared_slots_[lease.slot_id] = PreparedSlot{}; | ||
| } | ||
| if (rc != 0) { | ||
| throw std::runtime_error("abort prepared run failed with code " + std::to_string(rc)); | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
abort_prepared releases the lock between the state check and the backend call.
The scoped lock ends at Line 679, then abort_prepared_fn_ runs at Line 681 with the slot still marked PREPARED. Two consequences:
- Two concurrent aborts (for example the
finalize()sweep at Lines 358-378 racing a caller-driven abort for the same identity) both pass the check and both invoke the ABI on the same handle;simpler_abort_preparedrunsr->~Runtime()each time — double destruction. execute_preparedcan flip the same slotPREPARED → ACTIVEin that window, so the abort destroys the Runtime the executor is about to publish.
Claiming the slot under the lock before dropping it (e.g. an ABORTING state, or setting PreparedSlot{} up front and reconstructing on failure) closes both windows.
🔒️ Claim the slot under the lock
void ChipWorker::abort_prepared(const PipelineSlotLease &lease, uint64_t run_id, uint64_t dispatch_id) {
const PreparedRunIdentity identity{run_id, lease.generation, dispatch_id, lease.slot_id, 0};
{
std::scoped_lock lk(prepared_slots_mu_);
if (lease.slot_id >= prepared_slots_.size()) return;
- const PreparedSlot &slot = prepared_slots_[lease.slot_id];
+ PreparedSlot &slot = prepared_slots_[lease.slot_id];
if (slot.state != PreparedSlotState::PREPARED || slot.identity.run_id != run_id ||
slot.identity.generation != lease.generation || slot.identity.dispatch_id != dispatch_id) {
return;
}
+ // Claim the slot so no peer abort or execute can reach this handle.
+ slot = PreparedSlot{};
}
(void)select_slot_resources(lease.slot_id);
int rc = abort_prepared_fn_(device_ctx_, runtime_bufs_[lease.slot_id].data(), &identity);
- {
- std::scoped_lock lk(prepared_slots_mu_);
- prepared_slots_[lease.slot_id] = PreparedSlot{};
- }
if (rc != 0) {
throw std::runtime_error("abort prepared run failed with code " + std::to_string(rc));
}
}📝 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 ChipWorker::abort_prepared(const PipelineSlotLease &lease, uint64_t run_id, uint64_t dispatch_id) { | |
| const PreparedRunIdentity identity{run_id, lease.generation, dispatch_id, lease.slot_id, 0}; | |
| { | |
| std::scoped_lock lk(prepared_slots_mu_); | |
| if (lease.slot_id >= prepared_slots_.size()) return; | |
| const PreparedSlot &slot = prepared_slots_[lease.slot_id]; | |
| if (slot.state != PreparedSlotState::PREPARED || slot.identity.run_id != run_id || | |
| slot.identity.generation != lease.generation || slot.identity.dispatch_id != dispatch_id) { | |
| return; | |
| } | |
| } | |
| (void)select_slot_resources(lease.slot_id); | |
| int rc = abort_prepared_fn_(device_ctx_, runtime_bufs_[lease.slot_id].data(), &identity); | |
| { | |
| std::scoped_lock lk(prepared_slots_mu_); | |
| prepared_slots_[lease.slot_id] = PreparedSlot{}; | |
| } | |
| if (rc != 0) { | |
| throw std::runtime_error("abort prepared run failed with code " + std::to_string(rc)); | |
| } | |
| } | |
| void ChipWorker::abort_prepared(const PipelineSlotLease &lease, uint64_t run_id, uint64_t dispatch_id) { | |
| const PreparedRunIdentity identity{run_id, lease.generation, dispatch_id, lease.slot_id, 0}; | |
| { | |
| std::scoped_lock lk(prepared_slots_mu_); | |
| if (lease.slot_id >= prepared_slots_.size()) return; | |
| PreparedSlot &slot = prepared_slots_[lease.slot_id]; | |
| if (slot.state != PreparedSlotState::PREPARED || slot.identity.run_id != run_id || | |
| slot.identity.generation != lease.generation || slot.identity.dispatch_id != dispatch_id) { | |
| return; | |
| } | |
| // Claim the slot so no peer abort or execute can reach this handle. | |
| slot = PreparedSlot{}; | |
| } | |
| (void)select_slot_resources(lease.slot_id); | |
| int rc = abort_prepared_fn_(device_ctx_, runtime_bufs_[lease.slot_id].data(), &identity); | |
| if (rc != 0) { | |
| throw std::runtime_error("abort prepared run failed with code " + std::to_string(rc)); | |
| } | |
| } |
🤖 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/worker/chip_worker.cpp` around lines 669 - 689, Update
ChipWorker::abort_prepared to claim the matching PREPARED slot while holding
prepared_slots_mu_ before invoking abort_prepared_fn_. Transition it to an
ABORTING state or clear it atomically under the lock so concurrent aborts and
execute_prepared cannot reuse the slot; retain the identity validation and
ensure the claimed slot is finalized consistently after the backend call.
| assert st_worker.run_stream_set_create_count == after_first, ( | ||
| f"{rounds} same-image runs recreated the AICore stream: " | ||
| f"{after_first} -> {st_worker.run_stream_set_create_count}" | ||
| ) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
fd -t f 'worker.py' python | xargs rg -n -A10 'def run_stream_set_create_count'
fd -t f 'task_interface.py' python | xargs rg -n -A4 'def run_stream_set_create_count'Repository: hw-native-sys/simpler
Length of output: 1276
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Candidate files =="
git ls-files | rg '(^|/)(worker\.py|chip_worker\.h|task_interface\.py|test_run_stream_reuse\.py)$'
echo
echo "== Python worker property docstring =="
sed -n '6650,6675p' python/simpler/worker.py
echo
echo "== C++ property declaration/inline definition =="
rg -n -A8 -B3 'run_stream_set_create_count' src/common/worker/chip_worker.h
echo
echo "== Test context =="
sed -n '115,150p' tests/st/a2a3/host_build_graph/run_stream_reuse/test_run_stream_reuse.py
echo
echo "== Other run_stream_set_create_count references =="
rg -n -C3 'run_stream_set_create_count' python src tests | sed -n '1,220p'Repository: hw-native-sys/simpler
Length of output: 23607
Update the Python wrapper docstrings to match the flat-stream semantics.
Both python/simpler/worker.py:6660-6666 and python/bindings/task_interface.cpp:1572-1575 still describe the counter as advancing once per run. Update them to match ChipWorker::run_stream_set_create_count/DeviceRunnerBase, which says the AICore stream is reused while the loaded image is unchanged.
🤖 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/run_stream_reuse/test_run_stream_reuse.py`
around lines 136 - 139, Update the docstrings in the Python wrapper definitions
around worker stream reuse, specifically in simpler.worker and
bindings.task_interface, to describe flat-stream semantics:
ChipWorker::run_stream_set_create_count advances when the loaded image changes,
while the AICore stream is reused across runs with the same image. Remove
wording that says the counter advances once per run, keeping the behavior and
API unchanged.
| saw_active_and_accepted = False | ||
| deadline = time.monotonic() + 10.0 | ||
| while not run.done and time.monotonic() < deadline: | ||
| states = [_mailbox_load_i32(addr) for addr in frame_state_addrs] | ||
| if _TASK_ACTIVE in states and _TASK_ACCEPTED_STATE in states: | ||
| saw_active_and_accepted = True | ||
| break | ||
|
|
||
| run.wait(20.0) | ||
| assert saw_active_and_accepted, "frame B was not accepted while frame A was active" |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Unthrottled poll loop, and the assertion is racy if the run finishes first.
Two things on this observation window:
- The loop body has no
time.sleep, so it spins at full speed (holding the GIL) against the very dispatch it is trying to observe — other polling loops in this suite sleep 1 ms. - If both frames drain before a sample lands in the
(_TASK_ACTIVE, _TASK_ACCEPTED_STATE)pair,run.doneends the loop and line 122 fails even though the protocol behaved correctly. With_CHAIN_LENGTH = 64the first task is short, so this is a plausible flake source; consider a longer chain (the FIFO suite uses 4096) so the active window is wide enough to sample reliably.
♻️ Throttle the poll
while not run.done and time.monotonic() < deadline:
states = [_mailbox_load_i32(addr) for addr in frame_state_addrs]
if _TASK_ACTIVE in states and _TASK_ACCEPTED_STATE in states:
saw_active_and_accepted = True
break
+ time.sleep(0.001)🤖 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_endpoint/test_worker_async_endpoint.py`
around lines 113 - 122, Throttle the observation loop around
saw_active_and_accepted with the suite’s established 1 ms sleep so polling does
not starve dispatch. Make the assertion tolerate run.done occurring before a
sample by preserving success when both frames complete normally, and increase
_CHAIN_LENGTH from 64 to a longer value such as the FIFO suite’s 4096 to widen
the active window for sampling.
1 participant
Stack
Depends on #1587 (
codex/worker-async-b4b-hbg-prepared-epoch).Summary
RuntimeEnvcompatibility key and explicitly reject incompatible prepared runsTASK_ACTIVEno longer masquerades as launch acceptanceNo simulation code was changed or run in this PR.
Validation
Remote CPU (
liteserver-hps-148e, commit3a6d6494source state):881 passed, 6 skipped66/66 passed5/5 passed91 passedQueued A2A3 hardware:
task_20260729_044137_250311224272, exit 0, Ascend 910 A2A3task_20260729_055400_87277419590,1 passed, exit 0task_20260729_055400_87288330884,2 passed, exit 0TMR profiling from the final hardware run:
runner_run: 17.495 mssimpler_prepare: starts 3.139 ms after S1 starts and lasts 0.170 ms