Add HBG prepared epoch pipeline - #1587
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
📝 WalkthroughWalkthroughThe change adds generation-safe pipeline-slot admission, run-scoped FIFO scheduling, and optional two-frame prepared activation across hierarchical workers. Mailbox layouts, runtime APIs, Python bindings, endpoint dispatch, launch-shape activation, and concurrency tests are updated accordingly. ChangesWhole-run admission and scheduling
Runtime and Python integration
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
🧹 Nitpick comments (10)
src/common/hierarchical/worker_manager.cpp (1)
305-315: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low value
enqueue_dispatchtransiently overshoots capacity.
fetch_addthen rollback means a concurrenthas_capacity()/idle()observer can seeinflight_ == capacity_ + 1before the rollback lands, so a scheduler gate keyed onhas_capacity()may reject a slot that is actually free (and, symmetrically,idle()never reports true spuriously, so the risk is only a missed admission). A CAS loop keeps the counter within bounds.♻️ Bounded reservation
- uint32_t previous = inflight_.fetch_add(1, std::memory_order_acq_rel); - if (previous >= capacity_) { - inflight_.fetch_sub(1, std::memory_order_acq_rel); - throw std::logic_error("WorkerThread::dispatch: endpoint capacity exceeded"); - } + uint32_t previous = inflight_.load(std::memory_order_acquire); + do { + if (previous >= capacity_) { + throw std::logic_error("WorkerThread::dispatch: endpoint capacity exceeded"); + } + } while (!inflight_.compare_exchange_weak( + previous, previous + 1, std::memory_order_acq_rel, std::memory_order_acquire + ));🤖 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 305 - 315, Update WorkerThread::enqueue_dispatch to reserve inflight_ with a compare-exchange loop that only increments when the current value is below capacity_. Remove the fetch_add-and-rollback sequence so concurrent has_capacity() and idle() observers never see inflight_ exceed capacity_; retain the existing exception, dispatch ID assignment, queue insertion, and notification behavior.src/common/hierarchical/worker_manager.h (1)
97-97: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDeclare the protocol version with its wire width.
The frame trailer field is 8 bytes (
worker_manager.cppwrites auint64_t; Python unpacks=Q), so auint32_tconstant forces an implicit widen at every use and invites a 4-bytememcpyif someone passes the constant directly.♻️ Match the wire type
-static constexpr uint32_t MAILBOX_TASK_PROTOCOL_VERSION = 2; +static constexpr uint64_t MAILBOX_TASK_PROTOCOL_VERSION = 2;🤖 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 its declared type matches the 8-byte protocol trailer written by the worker-manager serialization path and read as =Q by Python. Keep its value and existing uses unchanged.python/simpler/worker.py (4)
1753-1761: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
slot_id < task_frame_countis validated against the parameter, but the frame arrays are sized by_TASK_FRAME_COUNT.
frame_bufs/frame_addrsare built withrange(_TASK_FRAME_COUNT)(Line 1740-1742) while identity validation boundsslot_idbytask_frame_count. Atask_frame_count > _TASK_FRAME_COUNTwould admit a slot id that indexes past both lists. Unreachable today (the only caller passes_TASK_FRAME_COUNT), but the two sources of truth should be one.🤖 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 1753 - 1761, The validate_identity function must bound slot_id using the same _TASK_FRAME_COUNT constant used to size frame_bufs and frame_addrs, rather than the task_frame_count parameter. Update that validation while preserving the other identity checks.
1888-1892: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winSilently swallowed
abort_preparedfailures leave no diagnostic trail.All four abort paths use bare
except Exception: pass. An abort that fails means the backend still holds an unpublished prepared run (device arena/slot not reclaimed), which will surface later as an unexplained slot exhaustion or lease mismatch with no clue about the original cause. Emit the exception to stderr like the other best-effort cleanups in this module do.♻️ Suggested logging
- try: - abort_prepared(prepared_identity) - except Exception: # noqa: BLE001 - pass + try: + abort_prepared(prepared_identity) + except Exception as exc: # noqa: BLE001 + sys.stderr.write( + f"chip_process dev={device_id}: abort_prepared failed: {type(exc).__name__}: {exc}\n" + ) + sys.stderr.flush()Also applies to: 1908-1911, 1949-1953, 1973-1977
🤖 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 1888 - 1892, Update all four abort cleanup paths around abort_prepared to preserve best-effort exception handling while emitting each caught exception to stderr, matching the diagnostic pattern used by other best-effort cleanups in the module. Replace the silent pass blocks associated with prepared_identity in each path; do not alter the abort flow or re-raise the failures.Source: Linters/SAST tools
1737-1737: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win
prepared_framesis mutated from both the admission and executor threads without synchronization.The admission thread inserts/pops at Lines 1881/1893/1907 while the executor pops at Lines 1954/1960. Individual dict ops are atomic in CPython, so nothing corrupts today, but the check-then-act pairs (
index in prepared_frames→ insert,get→ compare → pop) are not, and the invariant "one prepared epoch per slot" depends on mailbox-state ordering rather than on any explicit guard. Moving these accesses underaction_cv(already held for the queue) makes the ownership rule enforceable rather than incidental.Also applies to: 1906-1921, 1945-1960
🤖 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 1737, Protect all accesses to prepared_frames in the admission and executor paths with action_cv, including the check-then-insert/pop sequences and get/compare/pop logic around the identified admission and executor operations. Ensure each compound operation executes while holding the condition’s lock, preserving the one-prepared-epoch-per-slot invariant without changing mailbox ordering.
1836-1854: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAdmission thread spins with no yield.
admission_looppolls the control word and both frame states in a tight loop with notime.sleep/backoff. In the forked chip child this burns a core per chip and contends for the GIL with the executor thread between its native calls (the executor only drops the GIL insiderun_from_blob/execute_prepared). A short bounded sleep once no frame is actionable would keep latency while removing the busy-wait.♻️ Suggested bounded poll
def admission_loop() -> None: nonlocal control_queued while not stop_admission.is_set(): + progressed = False control_state = _mailbox_load_i32(state_addr)…and at the end of the frame sweep,
if not progressed: time.sleep(_MAILBOX_POLL_INTERVAL_S).Please confirm what poll cadence the existing
_run_mailbox_loopuses so the two loops stay consistent.🤖 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 1836 - 1854, Update admission_loop to avoid tight polling by tracking whether control or frame admission made progress during each iteration and sleeping for the established mailbox poll interval when none did. Reuse the cadence used by _run_mailbox_loop via _MAILBOX_POLL_INTERVAL_S, while preserving immediate handling of actionable frames and shutdown/control requests.tests/ut/py/test_callable_identity.py (1)
604-604: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAssert against the constant rather than the literal
2.This worker has no
device_ids, so_start_hierarchicalleavesdirect_chip_pipeline_depthatPTO_PIPELINE_MAX_DEPTH. Hard-coding2makes the test fail confusingly if that cap ever changes, and hides what the assertion is actually about (no chips ⇒ no depth negotiation, so the cap is passed through).♻️ Proposed change
- assert fake_c_worker.pipeline_depth == 2 + # No device_ids, so no chip negotiation happens and the cap is passed through. + 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 PTO_PIPELINE_MAX_DEPTH instead of the literal 2, preserving the test’s verification that the no-device path passes through the configured maximum depth.tests/ut/cpp/hierarchical/test_scheduler.cpp (1)
721-722: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valuePrefer a bounded copy into the fixed-size
output_prefix.
std::strcpyis flagged by static analysis;std::snprintf(diagnostic_config.output_prefix, sizeof(diagnostic_config.output_prefix), "%s", "/tmp/simpler-diagnostic-successor")keeps the intent and removes the unbounded-write pattern.🤖 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_scheduler.cpp` around lines 721 - 722, Replace the unbounded std::strcpy assignment to diagnostic_config.output_prefix with a bounded std::snprintf call using sizeof(diagnostic_config.output_prefix), preserving the existing prefix value and fixed-buffer safety.Source: Linters/SAST tools
tests/st/a2a3/host_build_graph/worker_async_fifo/kernels/orchestration/pipelined_vector_orch.cpp (1)
18-20: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDuplicated orchestration kernel differing only in
kChainLength.This file is otherwise identical to
tests/st/a2a3/host_build_graph/worker_async_endpoint/kernels/orchestration/long_vector_orch.cpp; consider a single shared source with the chain length injected as a compile definition.🤖 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/kernels/orchestration/pipelined_vector_orch.cpp` around lines 18 - 20, Consolidate the duplicated orchestration kernel by reusing the shared implementation from long_vector_orch.cpp, and remove the duplicate source-specific logic from the worker_async_fifo path. Inject the differing kChainLength value of 512 through the build configuration as a compile definition, while preserving the existing kernel behavior and constants.tests/st/a2a3/host_build_graph/worker_async_fifo/test_worker_async_fifo.py (1)
161-169: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueCapture the submitter thread's exception for a diagnosable failure.
If
st_worker.submit(third_graph)raises inside the daemon thread, the exception is discarded and the test fails at Line 169 with a misleading "did not enter after the first run freed its slot" message.♻️ Record the failure
- submitter = threading.Thread( - target=lambda: third_result.setdefault("handle", st_worker.submit(third_graph)), daemon=True - ) + def _submit_third(): + try: + third_result["handle"] = st_worker.submit(third_graph) + except BaseException as exc: # noqa: BLE001 + third_result["error"] = exc + + submitter = threading.Thread(target=_submit_third, daemon=True)and assert
third_result.get("error") is Nonebefore Line 176.🤖 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` around lines 161 - 169, Update the third-graph submitter thread around st_worker.submit to catch any exception and store it in third_result["error"] alongside the handle. Before the existing post-release callback assertion, assert that third_result.get("error") is None so submission failures are reported directly while preserving the current admission-capacity checks.
🤖 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 `@src/common/hierarchical/orchestrator.cpp`:
- Around line 355-369: Update Orchestrator::cancel_unstarted_run to avoid
calling try_consume more than once for a FAILED slot whose normal completion
path already consumed it. Gate the failed-slot pass and producer pass using the
existing consumption state, or otherwise track fanout-reference release
explicitly, while preserving cancellation cleanup for unconsumed slots and
preventing the total + 1 threshold from being reached prematurely.
- Around line 345-354: Update the cancellation loop in cancel_unstarted_run so
failure_message is written only after successfully transitioning the slot from
PENDING or READY to FAILED via the existing CAS; do not assign it before the
CAS. Ensure the completion/failure reporting path synchronizes access to
failure_message as needed, using the existing fanout_mu consistently if required
by the current state ownership.
In `@src/common/hierarchical/scheduler.cpp`:
- Around line 185-189: Align the preparable wake predicate in scheduler.cpp
lines 185-189 with dispatcher acceptance by checking supports_prepare_activate,
has_capacity(), and diagnostics_any(), or use a per-run declined latch. In the
group dispatcher at lines 380-386, pop and discard stale non-READY heads before
continuing. In the single-run dispatcher at lines 416-427, discard non-READY
entries and continue; reserve enqueue_ready_cb for genuine run or routing
mismatches.
In `@src/common/hierarchical/worker_manager.cpp`:
- Around line 413-452: Ensure dispatch IDs assigned by enqueue_dispatch cannot
remain unretired when WorkerThread::dispatch_process fails before
endpoint_->run_prepared_with_activation. Add or invoke an endpoint-side
abandonment path for the assigned ID on the null-endpoint, invalid-slot, and
other pre-endpoint failure paths, advancing the same publish sequence used by
retire_publish_sequence; alternatively move ID assignment into the endpoint so
only entered dispatches receive IDs. Preserve normal run_two_frame publication
behavior.
In `@src/common/platform/onboard/host/c_api_shared.cpp`:
- Around line 675-682: Update format_prepared_attrs to remove the request and
epoch trace attributes, since PreparedRunIdentity has no corresponding fields;
emit only run, slot, generation, and dispatch using their actual identity
members.
In `@src/common/worker/chip_worker.cpp`:
- Around line 650-670: Update ChipWorker::abort_prepared to claim the validated
slot under prepared_slots_mu_ before releasing the lock, transitioning it from
PREPARED to the same in-progress state used by execute_prepared. Recheck the
lease identity while claiming, then perform select_slot_resources and
abort_prepared_fn_ only after ownership is acquired, preventing concurrent abort
callers from processing the same slot.
In
`@tests/st/a2a3/host_build_graph/worker_async_endpoint/test_worker_async_endpoint.py`:
- Around line 113-122: Make the frame-state assertion deterministic by blocking
frame A at the appropriate synchronization point, using the release-fence
pattern from test_worker_async_fifo, until frame B reaches _TASK_ACCEPTED_STATE
while A remains _TASK_ACTIVE. Update the polling loop around
saw_active_and_accepted to include a short sleep to avoid busy-spinning and GIL
contention, then release the block before run.wait while preserving the existing
assertion.
In `@tests/ut/cpp/hierarchical/test_orchestrator.cpp`:
- Around line 761-772: Replace the fatal readiness assertion in
tests/ut/cpp/hierarchical/test_orchestrator.cpp:761-772 with a non-fatal check
and unconditionally release the prepared lease via the existing recovery path
before continuing or returning. Apply the same change at
tests/ut/cpp/hierarchical/test_orchestrator.cpp:808-817, ensuring the active
slot is always consumed so replacement can complete. At
tests/ut/cpp/hierarchical/test_scheduler.cpp:541-548, replace ASSERT_NE with
EXPECT_NE and skip recovery dispatch when the prerequisite is unavailable rather
than returning while the future remains outstanding.
In `@tests/ut/cpp/hierarchical/test_scheduler.cpp`:
- Around line 578-590: Bound both mailbox polling loops in the mock child thread
around the child lambda with steady-clock deadlines, and exit the child when
PREPARE_READY or ACTIVATE is not observed before its deadline. Ensure
child.join() can always complete while preserving the existing state transitions
when each expected state arrives.
---
Nitpick comments:
In `@python/simpler/worker.py`:
- Around line 1753-1761: The validate_identity function must bound slot_id using
the same _TASK_FRAME_COUNT constant used to size frame_bufs and frame_addrs,
rather than the task_frame_count parameter. Update that validation while
preserving the other identity checks.
- Around line 1888-1892: Update all four abort cleanup paths around
abort_prepared to preserve best-effort exception handling while emitting each
caught exception to stderr, matching the diagnostic pattern used by other
best-effort cleanups in the module. Replace the silent pass blocks associated
with prepared_identity in each path; do not alter the abort flow or re-raise the
failures.
- Line 1737: Protect all accesses to prepared_frames in the admission and
executor paths with action_cv, including the check-then-insert/pop sequences and
get/compare/pop logic around the identified admission and executor operations.
Ensure each compound operation executes while holding the condition’s lock,
preserving the one-prepared-epoch-per-slot invariant without changing mailbox
ordering.
- Around line 1836-1854: Update admission_loop to avoid tight polling by
tracking whether control or frame admission made progress during each iteration
and sleeping for the established mailbox poll interval when none did. Reuse the
cadence used by _run_mailbox_loop via _MAILBOX_POLL_INTERVAL_S, while preserving
immediate handling of actionable frames and shutdown/control requests.
In `@src/common/hierarchical/worker_manager.cpp`:
- Around line 305-315: Update WorkerThread::enqueue_dispatch to reserve
inflight_ with a compare-exchange loop that only increments when the current
value is below capacity_. Remove the fetch_add-and-rollback sequence so
concurrent has_capacity() and idle() observers never see inflight_ exceed
capacity_; retain the existing exception, dispatch ID assignment, queue
insertion, and notification behavior.
In `@src/common/hierarchical/worker_manager.h`:
- Line 97: Change MAILBOX_TASK_PROTOCOL_VERSION from uint32_t to uint64_t so its
declared type matches the 8-byte protocol trailer written by the worker-manager
serialization path and read as =Q by Python. Keep its value and existing uses
unchanged.
In
`@tests/st/a2a3/host_build_graph/worker_async_fifo/kernels/orchestration/pipelined_vector_orch.cpp`:
- Around line 18-20: Consolidate the duplicated orchestration kernel by reusing
the shared implementation from long_vector_orch.cpp, and remove the duplicate
source-specific logic from the worker_async_fifo path. Inject the differing
kChainLength value of 512 through the build configuration as a compile
definition, while preserving the existing kernel behavior and constants.
In `@tests/st/a2a3/host_build_graph/worker_async_fifo/test_worker_async_fifo.py`:
- Around line 161-169: Update the third-graph submitter thread around
st_worker.submit to catch any exception and store it in third_result["error"]
alongside the handle. Before the existing post-release callback assertion,
assert that third_result.get("error") is None so submission failures are
reported directly while preserving the current admission-capacity checks.
In `@tests/ut/cpp/hierarchical/test_scheduler.cpp`:
- Around line 721-722: Replace the unbounded std::strcpy assignment to
diagnostic_config.output_prefix with a bounded std::snprintf call using
sizeof(diagnostic_config.output_prefix), preserving the existing prefix value
and fixed-buffer safety.
In `@tests/ut/py/test_callable_identity.py`:
- Line 604: Update the assertion for fake_c_worker.pipeline_depth to compare
against PTO_PIPELINE_MAX_DEPTH instead of the literal 2, preserving the test’s
verification that the no-device path passes through the configured maximum
depth.
🪄 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: b2216452-97fc-49eb-a3db-87a32be75912
📒 Files selected for processing (33)
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.hsrc/a2a3/runtime/host_build_graph/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/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/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/hierarchical/test_orchestrator.cpptests/ut/cpp/hierarchical/test_pipeline_contract.cpptests/ut/cpp/hierarchical/test_scheduler.cpptests/ut/py/test_callable_identity.pytests/ut/py/test_worker/test_host_worker.py
| 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
failure_message is written before the CAS, so it is clobbered even when the slot has already started.
cancel_unstarted_run runs on the submitting thread while the scheduler owns READY slots. The message is assigned unconditionally, then the CAS may fail because the scheduler just moved the slot READY→RUNNING. That slot then carries this cancellation text into its own completion/failure reporting, and the unsynchronized std::string assignment races with the scheduler's read/write of the same field on the dispatch path.
Only publish the message on the transition you actually win.
🐛 Proposed fix
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
- );
- }
+ while (current == TaskState::PENDING || current == TaskState::READY) {
+ // Claim the slot first; only the winner owns failure_message.
+ if (state.state.compare_exchange_strong(
+ current, TaskState::RESERVED_FOR_CANCEL, std::memory_order_acq_rel, std::memory_order_acquire
+ )) {
+ state.failure_message = message;
+ state.state.store(TaskState::FAILED, std::memory_order_release);
+ break;
+ }
+ }
}If adding an intermediate state is undesirable, guard the message assignment with the slot's existing fanout_mu and have the failure reader take the same lock.
🤖 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
cancellation loop in cancel_unstarted_run so failure_message is written only
after successfully transitioning the slot from PENDING or READY to FAILED via
the existing CAS; do not assign it before the CAS. Ensure the completion/failure
reporting path synchronizes access to failure_message as needed, using the
existing fanout_mu consistently if required by the current state ownership.
| for (TaskSlot slot : slots) { | ||
| TaskSlotState &state = slot_state(slot); | ||
| TaskState current = state.state.load(std::memory_order_acquire); | ||
| if (current == TaskState::FAILED) try_consume(slot); | ||
| } | ||
| for (TaskSlot slot : slots) { | ||
| TaskSlotState &state = slot_state(slot); | ||
| std::vector<TaskSlot> producers; | ||
| { | ||
| std::lock_guard<std::mutex> lk(state.fanout_mu); | ||
| producers = state.fanin_producers; | ||
| } | ||
| for (TaskSlot producer : producers) | ||
| try_consume(producer); | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
rg -nP -C6 '\btry_consume\s*\(' src/common/hierarchical
rg -nP -C6 'fanout_released' src/common/hierarchicalRepository: hw-native-sys/simpler
Length of output: 16858
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== orchestrator.cpp relevant functions =="
sed -n '300,380p' src/common/hierarchical/orchestrator.cpp
sed -n '640,730p' src/common/hierarchical/orchestrator.cpp
sed -n '1050,1115p' src/common/hierarchical/orchestrator.cpp
echo "== scheduler.cpp relevant functions =="
sed -n '230,350p' src/common/hierarchical/scheduler.cpp
sed -n '335,375p' src/common/hierarchical/scheduler.cpp
echo "== structural occurrences around task states and scope register/deregister =="
rg -n -C4 'TaskState::FAILED|TaskState::COMPLETED|TaskState::RUNNING|scope_[rd]efer_task|register_task|scope_->unregister|unregister_task|on_task_scope|fanout_total|release_ref|scope_ref' src/common/hierarchical *.h 2>/dev/null || trueRepository: hw-native-sys/simpler
Length of output: 38181
Avoid counting an extra fanout release when canceling FAILED tasks.
In Orchestrator::cancel_unstarted_run, the normal completion path already calls try_consume once at completion; this failure loop calls try_consume for every FAILED slot again and then again for each failed producer, so failed slots can reach the total + 1 threshold while consumers are still alive. Gate this extra pass on slots that were not already consumed elsewhere, or track/release fanout refs more explicitly.
🤖 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 355 - 369, Update
Orchestrator::cancel_unstarted_run to avoid calling try_consume more than once
for a FAILED slot whose normal completion path already consumed it. Gate the
failed-slot pass and producer pass using the existing consumption state, or
otherwise track fanout-reference release explicitly, while preserving
cancellation cleanup for unconsumed slots and preventing the total + 1 threshold
from being reached prematurely.
| if (!ready && cfg_.preparable_run_cb) { | ||
| RunId preparable = cfg_.preparable_run_cb(); | ||
| ready = preparable != INVALID_RUN_ID && !cfg_.manager->has_staged_run(preparable) && | ||
| !cfg_.ready_next_level_queues->empty(preparable); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Preparable-dispatch wake condition and decline paths are misaligned, so the scheduler can spin without progress. The wait predicate treats a non-empty preparable queue as dispatchable work, but the preparable dispatchers decline on conditions the predicate never checks (supports_prepare_activate, has_capacity(), diagnostics_any()) and leave the declined entry queued — so completion_cv_.wait returns immediately on every iteration.
src/common/hierarchical/scheduler.cpp#L185-L189: make the predicate agree with dispatcher acceptance, or consult a per-run "declined" latch set by the dispatchers.src/common/hierarchical/scheduler.cpp#L380-L386: pop and drop a stale (non-READY) group head instead of returning with it still at the front.src/common/hierarchical/scheduler.cpp#L416-L427: drop non-READYsingles withcontinueand reserveenqueue_ready_cbfor genuine run/routing mismatches.
📍 Affects 1 file
src/common/hierarchical/scheduler.cpp#L185-L189(this comment)src/common/hierarchical/scheduler.cpp#L380-L386src/common/hierarchical/scheduler.cpp#L416-L427
🤖 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 185 - 189, Align the
preparable wake predicate in scheduler.cpp lines 185-189 with dispatcher
acceptance by checking supports_prepare_activate, has_capacity(), and
diagnostics_any(), or use a per-run declined latch. In the group dispatcher at
lines 380-386, pop and discard stale non-READY heads before continuing. In the
single-run dispatcher at lines 416-427, discard non-READY entries and continue;
reserve enqueue_ready_cb for genuine run or routing mismatches.
| WorkerCompletion WorkerThread::dispatch_process(WorkerDispatch d, const std::function<void()> &on_accept) { | ||
| if (!endpoint_) throw std::runtime_error("WorkerThread::dispatch_process: null endpoint"); | ||
| if (d.prepare_only) { | ||
| TaskSlotState *slot = ring_ == nullptr ? nullptr : ring_->slot_state(d.task_slot); | ||
| if (slot == nullptr) throw std::runtime_error("WorkerThread::dispatch_process: invalid prepared task slot"); | ||
| const RunId run_id = slot->run_id; | ||
| auto clear_staged = [&]() { | ||
| { | ||
| std::lock_guard<std::mutex> lk(activation_mu_); | ||
| if (activation_requested_run_id_ == run_id) activation_requested_run_id_ = INVALID_RUN_ID; | ||
| } | ||
| backend_ready_run_id_.store(INVALID_RUN_ID, std::memory_order_release); | ||
| activated_run_id_.store(INVALID_RUN_ID, std::memory_order_release); | ||
| staged_run_id_.store(INVALID_RUN_ID, std::memory_order_release); | ||
| }; | ||
| try { | ||
| WorkerCompletion completion = endpoint_->run_prepared_with_activation( | ||
| ring_, d, | ||
| [this, run_id]() { | ||
| backend_ready_run_id_.store(run_id, std::memory_order_release); | ||
| }, | ||
| [this, run_id]() { | ||
| std::unique_lock<std::mutex> lk(activation_mu_); | ||
| activation_cv_.wait(lk, [this, run_id] { | ||
| return activation_requested_run_id_ == run_id || shutdown_.load(std::memory_order_acquire); | ||
| }); | ||
| if (shutdown_.load(std::memory_order_acquire)) { | ||
| throw std::runtime_error("prepared activation interrupted by worker shutdown"); | ||
| } | ||
| activation_requested_run_id_ = INVALID_RUN_ID; | ||
| }, | ||
| on_accept | ||
| ); | ||
| clear_staged(); | ||
| return completion; | ||
| } catch (...) { | ||
| clear_staged(); | ||
| throw; | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
A dispatch that fails before entering run_two_frame permanently stalls the endpoint's publish sequence.
dispatch_id is assigned in enqueue_dispatch, but next_publish_id_ is advanced only inside LocalMailboxEndpoint::run_two_frame (via the publish block or retire_publish_sequence). The throws at Line 414 and Line 417 (and any other failure between enqueue_dispatch and the endpoint call) leave that id unretired, so every later dispatch on the same two-frame endpoint blocks forever in retire_publish_sequence's publish_cv_.wait — a hard hang, not a reported failure.
Either retire/skip the id on this path (e.g. an endpoint-side abandon_dispatch(dispatch_id)), or assign dispatch_id inside the endpoint so ids exist only for dispatches that reach the sequence.
🤖 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 413 - 452, Ensure
dispatch IDs assigned by enqueue_dispatch cannot remain unretired when
WorkerThread::dispatch_process fails before
endpoint_->run_prepared_with_activation. Add or invoke an endpoint-side
abandonment path for the assigned ID on the null-endpoint, invalid-slot, and
other pre-endpoint failure paths, advancing the same publish sequence used by
retire_publish_sequence; alternatively move ID assignment into the endpoint so
only entered dispatches receive IDs. Preserve normal run_two_frame publication
behavior.
| static void format_prepared_attrs(const PreparedRunIdentity &identity, char *attrs, size_t attrs_size) { | ||
| std::snprintf( | ||
| 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) | ||
| ); | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
request= and epoch= trace attributes print duplicated values.
Six conversions are fed only four distinct fields: run_id is printed for both request and run, and generation for both epoch and generation. PreparedRunIdentity carries no request/epoch member, so these two keys emit misleading data into every prepare/execute/abort trace span. Drop the phantom keys (or add the real fields to the identity struct).
🐛 Proposed fix
static void format_prepared_attrs(const PreparedRunIdentity &identity, char *attrs, size_t attrs_size) {
std::snprintf(
- 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)
);
}📝 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.
| static void format_prepared_attrs(const PreparedRunIdentity &identity, char *attrs, size_t attrs_size) { | |
| std::snprintf( | |
| 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) | |
| ); | |
| } | |
| static void format_prepared_attrs(const PreparedRunIdentity &identity, char *attrs, size_t attrs_size) { | |
| std::snprintf( | |
| 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 675 - 682,
Update format_prepared_attrs to remove the request and epoch trace attributes,
since PreparedRunIdentity has no corresponding fields; emit only run, slot,
generation, and dispatch using their actual identity members.
| 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
TOCTOU: abort_prepared can double-abort the same slot.
The PREPARED+identity check at Lines 655-659 is made under prepared_slots_mu_, then the lock is released before select_slot_resources / abort_prepared_fn_. Two callers reaching the same identity concurrently — finalize()'s sweep at Lines 369-378 and a mailbox-driven abort, which the header comments describe as separate admission/executor threads — both pass the check and both call abort_prepared_fn_ on runtime_bufs_[lease.slot_id], destroying an already-destroyed Runtime. execute_prepared avoids this by claiming the slot (ACTIVE) under the lock; mirror that here.
🔒️ Claim the slot before aborting
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 a concurrent abort (finalize sweep vs. mailbox
+ // abort) cannot destroy the same Runtime twice.
+ slot.state = PreparedSlotState::ABORTING;
}📝 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 a concurrent abort (finalize sweep vs. mailbox | |
| // abort) cannot destroy the same Runtime twice. | |
| slot.state = PreparedSlotState::ABORTING; | |
| } | |
| (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)); | |
| } | |
| } |
🤖 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 650 - 670, Update
ChipWorker::abort_prepared to claim the validated slot under prepared_slots_mu_
before releasing the lock, transitioning it from PREPARED to the same
in-progress state used by execute_prepared. Recheck the lease identity while
claiming, then perform select_slot_resources and abort_prepared_fn_ only after
ownership is acquired, preventing concurrent abort callers from processing the
same slot.
| 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.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
The observation window makes this assertion timing-dependent.
saw_active_and_accepted requires sampling the instant frame A is ACTIVE and frame B is ACCEPTED. If the run finishes between polls (or the sampling thread is descheduled), run.done ends the loop and Line 122 fails even though the protocol behaved correctly. The loop also spins with no sleep, maximizing GIL contention with the very worker threads it is observing. Consider blocking the first task (as test_worker_async_fifo does with a release fence) so the window is deterministic, and adding a small sleep to the poll.
🤖 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, Make the frame-state assertion deterministic by blocking
frame A at the appropriate synchronization point, using the release-fence
pattern from test_worker_async_fifo, until frame B reaches _TASK_ACCEPTED_STATE
while A remains _TASK_ACTIVE. Update the polling loop around
saw_active_and_accepted to include a short sleep to avoid busy-spinning and GIL
contention, then release the block before run.wait while preserving the existing
assertion.
| bool prepared_consumed_early = false; | ||
| const std::future_status third_status = third.wait_for(std::chrono::seconds(1)); | ||
| EXPECT_EQ(third_status, std::future_status::ready); | ||
| if (third_status != std::future_status::ready) { | ||
| // Returning the remaining lease guarantees the async begin_run is | ||
| // joinable even when this admission invariant fails. | ||
| S(prepared.task_slot).state.store(TaskState::COMPLETED, std::memory_order_release); | ||
| EXPECT_TRUE(orch.on_consumed(prepared.task_slot)); | ||
| prepared_consumed_early = true; | ||
| } | ||
| ASSERT_EQ(third.wait_for(std::chrono::seconds(1)), std::future_status::ready); | ||
| RunId third_id = third.get(); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Fatal assertions run while a blocking future is still outstanding, turning failures into binary hangs. Each of these recovery blocks uses ASSERT_*, which returns from the test body; the associated std::async/std::future task is still parked in begin_run/endpoint.run, so ~future blocks forever instead of reporting the failure.
tests/ut/cpp/hierarchical/test_orchestrator.cpp#L761-L772: make Line 771 non-fatal and unconditionally release the prepared lease before returning.tests/ut/cpp/hierarchical/test_orchestrator.cpp#L808-L817: make Line 816 non-fatal and unconditionally consume the active slot soreplacementcan complete.tests/ut/cpp/hierarchical/test_scheduler.cpp#L541-L548: replaceASSERT_NEat Line 543 withEXPECT_NEand skip the recovery dispatch instead of returning.
📍 Affects 2 files
tests/ut/cpp/hierarchical/test_orchestrator.cpp#L761-L772(this comment)tests/ut/cpp/hierarchical/test_orchestrator.cpp#L808-L817tests/ut/cpp/hierarchical/test_scheduler.cpp#L541-L548
🤖 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 761 - 772,
Replace the fatal readiness assertion in
tests/ut/cpp/hierarchical/test_orchestrator.cpp:761-772 with a non-fatal check
and unconditionally release the prepared lease via the existing recovery path
before continuing or returning. Apply the same change at
tests/ut/cpp/hierarchical/test_orchestrator.cpp:808-817, ensuring the active
slot is always consumed so replacement can complete. At
tests/ut/cpp/hierarchical/test_scheduler.cpp:541-548, replace ASSERT_NE with
EXPECT_NE and skip recovery dispatch when the prerequisite is unavailable rather
than returning while the future remains outstanding.
| std::thread child([&] { | ||
| char *frame = mailbox.data() + MAILBOX_FRAME_SIZE; | ||
| auto *state = reinterpret_cast<volatile int32_t *>(frame + MAILBOX_OFF_STATE); | ||
| while (__atomic_load_n(state, __ATOMIC_ACQUIRE) != static_cast<int32_t>(MailboxState::PREPARE_READY)) { | ||
| std::this_thread::sleep_for(std::chrono::microseconds(50)); | ||
| } | ||
| __atomic_store_n(state, static_cast<int32_t>(MailboxState::BACKEND_READY), __ATOMIC_RELEASE); | ||
| while (__atomic_load_n(state, __ATOMIC_ACQUIRE) != static_cast<int32_t>(MailboxState::ACTIVATE)) { | ||
| std::this_thread::sleep_for(std::chrono::microseconds(50)); | ||
| } | ||
| __atomic_store_n(state, static_cast<int32_t>(MailboxState::TASK_ACTIVE), __ATOMIC_RELEASE); | ||
| __atomic_store_n(state, static_cast<int32_t>(MailboxState::TASK_DONE), __ATOMIC_RELEASE); | ||
| }); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Bound the mock child's mailbox spin loops.
Both while loops poll forever; if the endpoint never publishes PREPARE_READY/ACTIVATE, child.join() at Line 624 hangs the test binary rather than failing it. Add a steady-clock deadline and bail out of the child.
🤖 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_scheduler.cpp` around lines 578 - 590, Bound
both mailbox polling loops in the mock child thread around the child lambda with
steady-clock deadlines, and exit the child when PREPARE_READY or ACTIVATE is not
observed before its deadline. Ensure child.join() can always complete while
preserving the existing state transitions when each expected state arrives.
1 participant
Scope
Stack
Merge order: #1541 -> #1574 -> #1583 -> this PR.
This PR is stacked on #1583. Its B4b-only commit is
fb31c054; until #1583 merges, GitHub also shows the earlier stack commits in the comparison.Validation
880 passed, 6 skippedintests/ut/py.66/66passed.task_20260729_041521_159264010394exited 0 on Ascend910.task_20260729_041556_160985021136exited 0, 2 passed.task_20260729_041631_16284985747exited 0, 1 passed.