Feat: add generation-safe depth-two pipeline slots - #1540
Conversation
📝 WalkthroughWalkthroughThe change introduces a two-phase run lifecycle with explicit acceptance tracking, generation-safe pipeline-slot leases, per-slot runtime resources, AICore image hashing, and image-aware stream reuse across C++, Python bindings, documentation, and tests. ChangesAcceptance lifecycle
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant PythonWorker
participant ChipWorker
participant DeviceRunner
participant Orchestrator
PythonWorker->>ChipWorker: run_with_lease(slot_id, generation)
ChipWorker->>DeviceRunner: select pipeline slot and arena bank
DeviceRunner->>DeviceRunner: enqueue kernels and publish TASK_ACCEPTED
DeviceRunner->>Orchestrator: mark task accepted
Orchestrator-->>PythonWorker: acceptance wait completes
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: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/common/hierarchical/worker_manager.cpp (1)
279-304: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
accept_once()can crash the worker thread ifon_accept_throws.
accept_once()at line 299 runs unconditionally after thedispatch_processtry/catch, but is itself not exception-protected. Ifon_accept_(d)(here,Orchestrator::mark_task_accepted) throws on this call path (i.e. it wasn't already invoked during dispatch), the exception escapesWorkerThread::loop, the top-level routine of a detachedstd::thread, which callsstd::terminateon an escaping exception — crashing the whole process instead of just failing the one task, unlike every other failure mode here which is converted into a failedWorkerCompletion.🛡️ Proposed fix
WorkerCompletion completion; bool accepted = false; - auto accept_once = [&]() { - if (accepted) return; - accepted = true; - if (on_accept_) on_accept_(d); - }; + auto accept_once = [&]() noexcept { + if (accepted) return; + accepted = true; + if (!on_accept_) return; + try { + on_accept_(d); + } catch (...) { + // Acceptance notification failures must not crash the worker + // thread; dispatch completion is unaffected. + } + };🤖 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 279 - 304, Protect the unconditional accept_once() call in WorkerThread::loop so exceptions from on_accept_ are converted into the current task’s failed WorkerCompletion instead of escaping the worker thread. Reuse the existing ENDPOINT_FAILURE fields and error-message conventions, ensure completion is still finalized through on_complete_, and preserve the existing dispatch_process handling and one-time acceptance behavior.src/a2a3/platform/onboard/host/device_runner.cpp (1)
490-530: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winStale AICore stream handle retained if
rtStreamDestroyfails.When the AICore image hash changes, the retired stream is destroyed at Line 509; on failure the function returns
rcimmediately without clearingstreams.aicore/streams.has_aicore_image. A later call with the same (new) hash will then hit the early-return at Line 504 and reuse a handle that a priorrtStreamDestroycall already attempted to tear down — this contrasts withdestroy_run_stream_sets()(Line 550-558), which unconditionally nulls the handle even when destroy fails.🐛 Proposed fix: always clear retired-stream state before returning
if (streams.aicore != nullptr) { int rc = rtStreamDestroy(streams.aicore); + streams.aicore = nullptr; + streams.has_aicore_image = false; if (rc != 0) { LOG_ERROR("rtStreamDestroy (retired AICore slot %u) failed: %d", slot, rc); - return rc; + return rc; } - streams.aicore = nullptr; - streams.has_aicore_image = false; }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/a2a3/platform/onboard/host/device_runner.cpp` around lines 490 - 530, Update DeviceRunner::ensure_run_stream_set so the retired AICore stream handle and has_aicore_image state are cleared immediately after rtStreamDestroy is attempted, including when destruction fails, before returning the error code. Preserve the existing error logging and stream recreation flow.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/worker-manager.md`:
- Around line 175-181: Update the mailbox example’s preceding function signature
from run(...) to run_with_accept(..., on_accept), then modify the TASK_ACCEPTED
handling in the polling loop so on_accept() is invoked only once while
preserving TASK_DONE as the termination condition.
In `@python/simpler/worker.py`:
- Around line 2124-2147: Update _wait_for_acceptance so the state updates and
waiter notification are protected by a try/finally after
_wait_run_handle_accepted returns. Ensure _accept_wait_in_progress is always
reset and _cv.notify_all() always runs, while preserving _launch_accepted = True
only on successful acceptance and re-raising failures.
In `@src/common/hierarchical/orchestrator.cpp`:
- Line 528: Move increment_run_accepts(run->id, s.group_size()) to immediately
after the poisoned_by_failed_producer branch returns, before the normal
READY/PENDING enqueue path, so producer-poisoned slots are excluded from pending
accepts while dispatched members retain the correct count.
In `@src/common/platform/onboard/host/c_api_shared.cpp`:
- Around line 653-679: Bind g_runner_key to the runner represented by ctx for
the duration of set_task_accepted_state_ctx, select_pipeline_slot_ctx, and
select_arena_bank_ctx, following the guard and cleanup convention used by
simpler_run. Ensure the binding is established before each direct
DeviceRunnerBase call and cleared on every return path, including exceptions.
In `@src/common/worker/chip_worker.cpp`:
- Around line 121-125: Add non-null stub exports for select_pipeline_slot_ctx
and select_arena_bank_ctx in the a5 sim and a5 onboard host runtime C API
implementations. Follow the signatures and no-op behavior of the existing
c_api_shared.cpp implementations so ChipWorker::init() can load both required
symbols successfully.
---
Outside diff comments:
In `@src/a2a3/platform/onboard/host/device_runner.cpp`:
- Around line 490-530: Update DeviceRunner::ensure_run_stream_set so the retired
AICore stream handle and has_aicore_image state are cleared immediately after
rtStreamDestroy is attempted, including when destruction fails, before returning
the error code. Preserve the existing error logging and stream recreation flow.
In `@src/common/hierarchical/worker_manager.cpp`:
- Around line 279-304: Protect the unconditional accept_once() call in
WorkerThread::loop so exceptions from on_accept_ are converted into the current
task’s failed WorkerCompletion instead of escaping the worker thread. Reuse the
existing ENDPOINT_FAILURE fields and error-message conventions, ensure
completion is still finalized through on_complete_, and preserve the existing
dispatch_process handling and one-time acceptance behavior.
🪄 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: de4abed5-9b47-48bc-ae3e-e66e53db0abf
📒 Files selected for processing (39)
docs/orchestrator.mddocs/task-flow.mddocs/worker-manager.mdpython/bindings/task_interface.cpppython/bindings/worker_bind.hpython/simpler/orchestrator.pypython/simpler/task_interface.pypython/simpler/worker.pysrc/a2a3/platform/onboard/host/device_runner.cppsrc/a2a3/platform/onboard/host/device_runner.hsrc/a2a3/runtime/host_build_graph/host/runtime_maker.cppsrc/a2a3/runtime/tensormap_and_ringbuffer/host/runtime_maker.cppsrc/a5/runtime/host_build_graph/host/runtime_maker.cppsrc/a5/runtime/tensormap_and_ringbuffer/host/runtime_maker.cppsrc/common/hierarchical/orchestrator.cppsrc/common/hierarchical/orchestrator.hsrc/common/hierarchical/types.hsrc/common/hierarchical/worker.cppsrc/common/hierarchical/worker_manager.cppsrc/common/hierarchical/worker_manager.hsrc/common/platform/onboard/host/c_api_shared.cppsrc/common/platform/onboard/host/device_runner_base.cppsrc/common/platform/onboard/host/device_runner_base.hsrc/common/platform/sim/host/c_api_shared.cppsrc/common/task_interface/chip_callable_layout.hsrc/common/task_interface/prepare_callable_common.hsrc/common/utils/fnv1a_64.hsrc/common/worker/chip_worker.cppsrc/common/worker/chip_worker.hsrc/common/worker/pipeline_contract.hsrc/common/worker/pipeline_slot_pool.hsrc/common/worker/pto_runtime_c_api.htests/st/a2a3/host_build_graph/run_stream_reuse/kernels/aiv/kernel_sub.cpptests/st/a2a3/host_build_graph/run_stream_reuse/test_run_stream_reuse.pytests/ut/cpp/hierarchical/test_orchestrator.cpptests/ut/cpp/hierarchical/test_pipeline_contract.cpptests/ut/cpp/hierarchical/test_scheduler.cpptests/ut/cpp/types/test_chip_callable_upload_immutable.cpptests/ut/py/test_worker/test_host_worker.py
880cf3f to
ee420e7
Compare
|
Rebased onto Rebase: one commit was dropped, deliberately
enum class MailboxState : int32_t {
...
TASK_ACCEPTED = 8,
};which loses the ACK whenever the child reaches Findings addressed1. Two independent generation authorities (was the blocker). That is squarely on #1541's path — it acquires from Fixed by removing the second authority. 2. The depth-two ST had no teeth. It asserted stream-set counts and generation rejection — both #1467/#1539 mechanisms already on 3. Slot/bank selection was thread-local. 4. 5. 6. Bank 1 silently lost the prebuilt-arena cache. Both Not changed
Full validation table is in the PR description. |
ea9af7f to
6a6f714
Compare
|
Taking the correction: AICore streams are now created fresh for every You're right that the epoch was built on an unproven premise. It assumed that selecting a stream whose recorded image matches is as good as creating one; the only operation we have evidence for is creation. Tracking
On the trace in your finding: it describes TestsThe old assertions pinned the opposite invariant, so they inverted rather than extended:
CostThis partly gives back #1464's persistent-AICore-stream gain, so here is what it measures at, on a 128×128 vector callable, 300 runs after 20 warm-ups:
Read that as an upper bound of roughly +1 ms/run, not a measurement: the two reuse builds differ by 0.78 ms and the fresh samples span 0.77 ms, so this shared box's run-to-run spread is the same size as the effect. The delta is one I think the trade is right — a guarantee that holds by construction beats a sub-noise host-time win — but flagging it explicitly because #1464 is merged and this reverses part of it. Say the word if you would rather keep same-image reuse and accept the narrower guarantee. Also in this pushThe three other findings from the previous round are unchanged and still in: host Validation on |
6a6f714 to
1b2ab0c
Compare
|
Head is [P1] sim declaring K=2 while rejecting slot 1 — took the first of your two options: sim now really implements K=2. [P2] generation filter is not an ownership check — correct, and your timing is exactly the hole: g1 released → g2 leased → g2 has not reached [P2] TMR slot-1 path uncovered — added [P2] stale stream-reuse docs — swept: the contradictory On stream creation and prepare — agreed, no disagreement to record. Two findings that were about On churn — 1092 lines noted. The split you suggest (contract/lease/filter vs platform slot resources + stream realization) is a clean seam; happy to do it before merge if you want it split. Validation on |
1b2ab0c to
3cf4c8d
Compare
Permit A2/A3 host runtimes to declare `pipeline_depth = 2` and give every
resource class the copy count its declaration implies, without admitting a
second run into device execution. `HOST_PER_RUN` and `EXEC_HANDLE` gain one
copy per slot, `DEVICE_SCRATCH` keeps one, and `pipeline_resource_copy_count`
/ `pipeline_resource_slot` derive both from the contract so no caller
hardcodes a replication rule.
GM heap, GM shared memory, and the runtime image are committed together into
one arena bank, so a contract that classifies them inconsistently — or repeats
one of them, where only the first entry is ever read — describes a layout no
runner can build. `has_serviceable_arena_topology` rejects it when the runtime
loads rather than at the first launch that would need the second bank.
Repeated stream kinds select no bank and stay legal.
A run reaches its resources through a `PipelineSlotLease {slot_id,
generation}`. `PipelineSlotPool` is the only mint and the only authority on
ownership: it hands out one lease per free slot, bumps the generation on
reuse, and makes release idempotent for the current generation while
rejecting a stale one. Consumers downstream of the pool cannot re-derive
ownership, so `ChipWorker` carries a `PipelineSlotGenerationFilter` that only
refuses generations older than the newest that slot has presented. The filter
mints nothing: an unleased run selects slot 0 and advances no generation, so a
pool's first lease is still admitted after any number of ordinary runs.
Each slot owns a host `Runtime` staging buffer. That buffer is not the
`RUNTIME_IMAGE` resource — the contract classifies the device-resident image,
while this holds the host-side object a run constructs in place, with its
tensor leases, launch arguments, and validate/finalize state. That is per-run
whatever the device image sharing is, so a runtime whose image is
`DEVICE_SCRATCH` still needs its own buffer per slot.
The A2/A3 runner owns its slot and arena-bank selection as member state, so
two contexts on one thread cannot see each other's choice — a fresh worker's
prewarm lays out its arenas in the bank it will actually serve runs from. The
three pooled device regions become one bank per slot, and the single-entry
prebuilt-arena cache stays owned by bank 0.
AICore streams are outside that lease and outside the slot: each run creates
its own and retires it on every exit path. The instruction cache belongs to the
cores, every slot publishes its image to the same GM code address, and the
platform offers no cache invalidation for code replaced there. Creating a
stream is the only operation known to leave a core free of the previous image's
instructions — selecting an existing one is not — so no record of which image a
stream last ran can make reuse safe, and none is kept. The AICPU stream carries
no such state and stays with its slot for the worker's lifetime.
Simulation implements the same depth, so the contract means the same thing on
both platforms: its runner owns one arena bank and one retained temp buffer per
slot, and a leased slot-1 run there exercises the second copy rather than being
refused.
The generation filter is a replay filter, not an ownership check, and says so:
it rejects a lease whose successor has already presented itself, but one that
was released while its successor has not yet reached the consumer still passes.
Closing that window needs dispatch gated on `PipelineSlotPool::owns`, which
belongs to whoever admits runs.
The ordinary synchronous path stays on slot 0 and the chip child's mailbox
loop passes no lease, so every production run is unleased. No prepared FIFO,
second mailbox frame, or backend publication overlap is enabled here.
A run owns its AICore stream, so a destroy it cannot complete is that run's
failure: the retire path returns the driver's error and keeps the handle, which
both leaves the slot unusable for the next run — the stream may still hold the
previous image's instructions — and leaves teardown something to retry.
Reporting success there would have told the caller a slot is reusable that the
next acquire now refuses. That state machine lives in `RunStreamSlots`, whose
stream creation and destruction are injected, so the failure paths are covered
without a device.
`get_arena_bank_gm_heap_base_ctx`, `get_retained_temp_addr_ctx`, and
`ChipWorker::runtime_buffer_addrs` report which storage a bank and a slot
actually own. Sequential runs re-stage their arguments every round, so correct
output alone survives two slots sharing one buffer; the tests compare addresses
instead.
3cf4c8d to
af454c7
Compare
|
Head is 1.
|
| 位置 | 原先的错误说法 |
|---|---|
docs/task-flow.md |
sim 只支持 slot 0 |
docs/task-flow.md |
runner-scoped retained temporary buffer |
src/common/platform/include/common/host_api.h |
平台只记一个 retained slot；"three per-Worker pooled regions" |
src/a2a3/.../RUNTIME_LOGIC.md、src/a5/.../RUNTIME_LOGIC.md |
runner-scoped retained buffer |
src/common/platform/sim/host/device_runner_base.h |
已不存在的单个 runtime_arena_pool_ |
| HBG 测试类 docstring | 重复运行共享一个 stream set |
rg "runner-scoped retained|single run stream set|reused only while|same-image runs reuse|Simulation serves slot 0" 现在为空。onboard runner header 的「同 image 复用」在上一轮已改。
3. PR 描述
Validation 表已改为 af454c75 的结果（ctest 64/64，含新目标），变异表加了这一轮的 rtStreamDestroy 项。
仍未结案：flat 形态下的段错误
CI 形态 sweep 跑到 L2 tensormap_and_ringbuffer 的 84%（零 FAIL、零段错误）后被 task-submit daemon 重启打断，不是超时也不是测试失败。剩下 16% 正在补跑，结果会单独贴。相关的 main 侧失败已由 #1561 解释（四个 notify demo 硬编码平台与设备 0/1），但那不解释分支侧的段错误，所以我没把它并进去。
2 participants
Summary
Permit A2/A3 host runtimes to declare
pipeline_depth = 2and give every resource class the copy count its declaration implies, without admitting a second run into device execution.HOST_PER_RUNandEXEC_HANDLEget one copy per slot,DEVICE_SCRATCHkeeps one;pipeline_resource_copy_count/pipeline_resource_slotderive both from the contract so no caller hardcodes a replication rule.has_serviceable_arena_topologyrejects a contract that classifies them inconsistently — or repeats one, where only the first entry is read — when the runtime loads, not at the first launch that would need a second bank. Repeated stream kinds select no bank and stay legal.Runtimestaging buffer. That buffer is not theRUNTIME_IMAGEresource: the contract classifies the device-resident image, while this holds the host-side object a run constructs in place, with its tensor leases, launch arguments and validate/finalize state. That is per-run whatever the device image sharing is, so TMR — whose image isDEVICE_SCRATCH— still gets one buffer per slot.RunStreamSlotswith injected create/destroy, so its failure paths are covered without a device.The ordinary synchronous path stays on slot 0, and the chip child's mailbox loop passes no lease, so every production run is unleased. No prepared FIFO, second mailbox frame, or backend publication overlap is enabled here.
Known boundaries
These are deliberate and belong to whole-run admission (#1541) or later, not to B1:
ChipWorkeris downstream ofPipelineSlotPooland never sees an acquire or release, so it keeps a per-slot high-water mark. That rejects a superseded lease, but one released while its successor has not yet reached the consumer still passes. Closing the window needs dispatch gated onPipelineSlotPool::owns. Stated inpipeline_slot_pool.handdocs/task-flow.md.rtStreamCreateis still on the critical path. It runs insideDeviceRunner::run()ahead oflaunch_run, so it overlaps nothing. Moving it into a prepare stage — stream owned by{slot, generation}, destroyed on slot reuse / prepare failure / cancel, ownership re-verified before launch — is async-preparation work.run_from_blobaccepts slot/generation, but_run_chip_main_looppasses neither.Validation
All on
af454c75, rebased ontoupstream/main@40b6329a.pytest tests/utexamples tests/sthost_build_graphhost_build_graph/run_stream_reusetensormap_and_ringbuffer/pipeline_slotsThe CI-shaped onboard sweep reached 84% of the L2
tensormap_and_ringbufferphase with zero failures before the task-submit daemon restarted; the remaining
16% is being re-run and will be posted separately.
Mutation testing
Depth-two tests are checked against deliberate defects rather than assumed to have teeth:
a served bank is uncommitted: 0x12c081600000, 0x0run pipeline lease generation is staleslot 1 reused its stream after another slot published a different imagertStreamDestroydrops the handle and reports successThe last has no end-to-end test. Poisoning the selection needs a context that picks bank ≠ 0 (only an hbg-classified contract does), while observing it needs arenas laid out before any selection (only TMR's
prewarm_config_impldoes; hbg links the weak no-op). That combination needs an hbg worker leasing slot 1 and a TMR worker prewarming in one process and thread, which no scene-test class can construct. The fix is structural — two objects cannot share a member — but no coverage is claimed for it.Cost of fresh streams per run
This gives back part of #1464's persistent-AICore-stream gain. On a 128×128 vector callable, 300 runs after 20 warm-ups:
Read as an upper bound of roughly +1 ms/run, not a measurement: two reuse builds differ by 0.78 ms and the fresh samples span 0.77 ms, so this shared box's spread is the size of the effect. The delta is one
rtStreamCreate+rtStreamDestroyper run — bounded, and proportionally smaller on larger workloads.Stack
PR-B1 of the worker async pipeline. #1541 (bounded whole-run FIFO admission) stacks on this.