Add: dispatch MPI worker groups through mailbox - #1624
Open
sunkaixuan2018 wants to merge 2 commits into
Open
Conversation
Add static MPI-launched L3 groups with rank-0 manifest broadcast and ready coordination, and extend Global CommDomains across mixed local, TCP-remote, and MPI-backed L3 nodes. Harden the control path with validated control names, bounded MPI collectives and L4 fanout, rollback-safe peer mappings, isolated process cleanup, fixed-port session binding compatibility, diagnostics, documentation, and regression coverage. Co-authored-by: Leaf-Salix <2503954024@qq.com> Co-authored-by: xl <21039015+xl1123@users.noreply.github.com>
5 tasks
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughChangesGlobal CommDomain contracts and backends
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
Sequence Diagram(s)sequenceDiagram
participant Orchestrator
participant Worker
participant Mailbox
participant MpiDispatcher
participant L3Rank
Orchestrator->>Worker: allocate_global_domain
Worker->>Mailbox: publish domain control
Mailbox->>MpiDispatcher: dispatch ordered request
MpiDispatcher->>L3Rank: prepare or import domain
L3Rank-->>MpiDispatcher: descriptor or context result
MpiDispatcher-->>Mailbox: publish completion
Mailbox-->>Worker: return control response
Worker-->>Orchestrator: commit domain view
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1🛠️ Fix failing CI checks 💡
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: 11
🧹 Nitpick comments (19)
python/simpler/global_comm_smoke.py (1)
22-167: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExtract the shared prologue and tensor builders.
The four callbacks repeat the same steps: validate
scalar_count() == 6, importget_inner_handle, decode the digest, and look up the domain. The group variants then repeat the body of their single-worker counterparts inside a loop. Two small helpers, one for the prologue and one per kernel argument shape, would remove most of this duplication and keep the four entry points to a few lines each.♻️ Sketch
+def _resolve(orch, args: TaskArgs, message: str): + from .remote_l3_session import get_inner_handle # noqa: PLC0415 + + if args.scalar_count() != 6: + raise ValueError(message) + return ( + get_inner_handle(_digest_from_scalars(args, 2).hex()), + orch.get_global_domain(int(args.scalar(0))), + int(args.scalar(1)), + ) + + +def _compute_args(context) -> TaskArgs: + chip_args = TaskArgs() + for buffer_name in ("lhs", "rhs"): + chip_args.add_tensor(_domain_tensor(context, buffer_name), TensorArgType.INPUT) + chip_args.add_tensor(_domain_tensor(context, "input"), TensorArgType.OUTPUT_EXISTING) + return chip_args🤖 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/global_comm_smoke.py` around lines 22 - 167, Refactor remote_compute_orch, remote_rank_orch, remote_compute_group_orch, and remote_rank_group_orch to share a helper for importing get_inner_handle, validating the six scalars, decoding the digest, and resolving the domain. Add reusable tensor-argument builders for the compute and TLOAD shapes, then have the group callbacks reuse the corresponding single-worker argument construction inside their loops while preserving worker selection and submission behavior.tools/a3_l4_tcp_smoke/mpirun_compute_then_tload_2x2_smoke.py (1)
51-60: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueOptional: use iterable unpacking instead of concatenation.
Static analysis flags
list(include_dirs) + [str(...)]on line 53. Use unpacking for a more idiomatic construction.♻️ Proposed fix
- kernel_include_dirs = list(include_dirs) + [str(compiler.project_root / "src" / "common")] + kernel_include_dirs = [*include_dirs, str(compiler.project_root / "src" / "common")]🤖 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 `@tools/a3_l4_tcp_smoke/mpirun_compute_then_tload_2x2_smoke.py` around lines 51 - 60, Update the kernel_include_dirs construction in _compile_aiv to use iterable unpacking when combining include_dirs with the common source directory, preserving the existing ordering and values.Source: Linters/SAST tools
python/simpler/mpi_l3_session.py (1)
504-513: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueReplace the
asserton the shutdown payload with an explicit check.Line 507 uses
assert payload is not None. Python removes asserts under-O, and_rewrite_frame_identitywould then fail onNone. Raise an explicit error instead.🤖 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/mpi_l3_session.py` around lines 504 - 513, Replace the assert in the MailboxOpcode.SHUTDOWN branch of the request handling flow with an explicit payload None check that raises an appropriate error before calling _rewrite_frame_identity; preserve the existing shutdown behavior when payload is present.tools/mpi_group_mailbox_smoke.py (1)
34-37: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAdd a sleep to the wait helper.
_wait_untilre-evaluates the predicate with no pause. Each mailbox state read maps to a shared-memory load plus an import lookup, so this pins a core for the whole wait. Sleep for a short interval between checks.♻️ Proposed fix
def _wait_until(predicate, *, deadline: float, label: str) -> None: while not predicate(): if time.monotonic() >= deadline: raise TimeoutError(f"timed out waiting for {label}") + 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 `@tools/mpi_group_mailbox_smoke.py` around lines 34 - 37, Update _wait_until to pause briefly between predicate evaluations, while retaining the existing deadline check and TimeoutError behavior; add the sleep inside the loop after a failed predicate check.tests/ut/py/test_global_comm_domain.py (1)
259-278: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valuePrefer allocated ports over the fixed
19073 + indexrange.The test never binds these endpoints, so it passes today. The file already provides
_free_tcp_ports. Using it removes the fixed range and keeps the endpoint construction consistent across the file.🤖 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_global_comm_domain.py` around lines 259 - 278, Update _failure_injection_worker to obtain two ports through the existing _free_tcp_ports helper, then build each RemoteWorkerSpec endpoint from those allocated ports instead of the fixed 19073 + index range. Preserve the current node ordering and endpoint construction format.tests/ut/cpp/hierarchical/test_remote_endpoint.cpp (1)
659-661: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winBound the state wait so a regression fails instead of hanging.
Line 661 spins until
RequestState::REQUEST_READYwith no deadline. Line 697 uses the same pattern. Ifexchange_group_taskstops publishing a request, the test hangs and the CI job times out with no diagnostic. Add a deadline andFAIL()when it expires.♻️ Proposed change for the helper
-void respond_with_payloads(std::vector<uint8_t> &mailbox, const std::vector<std::vector<uint8_t>> &payloads) { +void wait_for_request_ready(const std::vector<uint8_t> &mailbox) { + using namespace mpi_group_mailbox; + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); + while (mailbox_state(mailbox, OFF_REQUEST_STATE) != static_cast<int32_t>(RequestState::REQUEST_READY)) { + ASSERT_LT(std::chrono::steady_clock::now(), deadline) << "mailbox request was never published"; + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } +} + +void respond_with_payloads(std::vector<uint8_t> &mailbox, const std::vector<std::vector<uint8_t>> &payloads) { using namespace mpi_group_mailbox; - while (mailbox_state(mailbox, OFF_REQUEST_STATE) != static_cast<int32_t>(RequestState::REQUEST_READY)) {} + wait_for_request_ready(mailbox);🤖 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_remote_endpoint.cpp` around lines 659 - 661, Bound the polling loops in both respond_with_payloads and the matching wait near exchange_group_task with a deadline; when RequestState::REQUEST_READY is not observed before expiration, call FAIL() with a diagnostic, while preserving the existing behavior when the state becomes ready.python/simpler/mpi_group_mailbox.py (3)
320-329: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winTruncated error JSON becomes unparsable.
If the encoded rank errors exceed
MAILBOX_ERROR_BYTES, line 324 cuts the JSON mid-document.read_resultthen failsjson.loadsand falls back to a raw replacement decode, so the rank attribution is lost. Truncate each message before encoding instead, so the document stays valid.♻️ Proposed fix
- data = json.dumps([asdict(error) for error in errors], sort_keys=True).encode("utf-8") - if len(data) > MAILBOX_ERROR_BYTES: - data = data[: MAILBOX_ERROR_BYTES - 1] + entries = [asdict(error) for error in errors] + data = json.dumps(entries, sort_keys=True).encode("utf-8") + while len(data) > MAILBOX_ERROR_BYTES and entries: + budget = max(0, len(entries[-1]["message"]) // 2) + if budget == 0: + entries.pop() + else: + entries[-1]["message"] = entries[-1]["message"][:budget] + data = json.dumps(entries, sort_keys=True).encode("utf-8") + data = data[:MAILBOX_ERROR_BYTES]🤖 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/mpi_group_mailbox.py` around lines 320 - 329, Update the error serialization flow in the mailbox failure-writing method so oversized data remains valid JSON. Truncate individual rank-error message fields before `json.dumps`, then encode and write the complete serialized document without slicing the encoded JSON at `MAILBOX_ERROR_BYTES`; preserve rank attribution and the existing state updates.
345-357: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winNarrow the exception scope around the error decode.
Line 353 catches
BaseException, so aKeyboardInterruptorSystemExitraised during the decode is converted into a plain message. Catch the decode and lookup errors only. Ruff also reports BLE001 here.♻️ Proposed fix
- except BaseException: + except (ValueError, TypeError, KeyError, UnicodeDecodeError): message = raw.decode("utf-8", errors="replace") or "MPI group request failed"🤖 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/mpi_group_mailbox.py` around lines 345 - 357, The TASK_FAILED error decoding in the mailbox request handling must not catch control-flow exceptions. Narrow the try/except around json decoding and entry field access to the specific decode and lookup/type errors that can occur, replacing the broad BaseException handler so Ruff BLE001 is resolved while preserving the raw-message fallback.Source: Linters/SAST tools
373-375: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueMove the test-only mutator out of the shipped protocol class.
overwrite_request_payload_for_testwrites arbitrary bytes into the request region with no state or capacity check. It is reachable in production. The single caller istests/ut/py/test_mpi_group_mailbox.pyline 96. Write throughmailbox._bufferfrom the test, or guard the helper with a length check againstMAILBOX_PAYLOAD_BYTES.🤖 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/mpi_group_mailbox.py` around lines 373 - 375, The test-only method overwrite_request_payload_for_test should not remain exposed on the shipped mailbox protocol class. Remove it and update the sole caller in test_mpi_group_mailbox.py to write the payload directly through mailbox._buffer, or otherwise enforce MAILBOX_PAYLOAD_BYTES capacity before writing if the helper must remain.tests/ut/py/test_mpi_group_mailbox.py (1)
25-211: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a capacity-guard case to this suite.
The suite covers targets, sequencing, failure, and shutdown. It does not cover the capacity guard in
_encode_payloads(MAILBOX_PAYLOAD_BYTES) or the truncation path infail_request. Both are wire-protocol limits. Add one test that writes an oversized payload vector and expectsValueError.🤖 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_mpi_group_mailbox.py` around lines 25 - 211, Add a test covering the mailbox payload capacity guard by attempting to write a request whose encoded payloads exceed MAILBOX_PAYLOAD_BYTES and asserting ValueError. Exercise the write_request path and ensure the oversized payload vector is rejected before acceptance or state progression.tools/mpi_l3_group_smoke.py (1)
41-43: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRestore
SIMPLER_MPI_SMOKE_DIRafter the run.Line 42 sets the variable and never removes it. The temporary directory is deleted when the
withblock exits, so the variable then points at a missing path. Ifrunis ever called from another module or twice, the stale value leaks. Usetry/finallyoros.environ.popat the end of the block.🤖 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 `@tools/mpi_l3_group_smoke.py` around lines 41 - 43, Restore the SIMPLER_MPI_SMOKE_DIR environment variable after the temporary-directory run in the surrounding run flow: save any prior value before assigning output_dir, then restore it in a finally block (or remove it when absent) so repeated or nested calls never retain the deleted path.python/simpler/worker.py (2)
3515-3523: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the unreachable node-identity fallback.
_build_remote_manifestruns only for ids inself._remote_worker_ids, and_remote_like_worker_ids()is the union of_remote_worker_idsand_mpi_worker_ids. The condition at Line 3515 is therefore always true, so lines 3520-3523 never execute. Drop the branch, or state the caller contract that makes it reachable.♻️ Proposed simplification
- if worker_id in self._remote_like_worker_ids(): - runtime = self._resolved_global_nodes()[int(worker_id)] - node_rank = runtime.node_rank - node_count = runtime.node_count - global_device_ranks = runtime.global_device_ranks - else: - node_rank = 0 - node_count = 1 - global_device_ranks = spec.global_device_ranks or tuple(range(len(spec.device_ids))) + runtime = self._resolved_global_nodes()[int(worker_id)] + node_rank = runtime.node_rank + node_count = runtime.node_count + global_device_ranks = runtime.global_device_ranks🤖 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 3515 - 3523, Remove the unreachable else fallback in _build_remote_manifest and rely directly on the runtime values from _resolved_global_nodes()[int(worker_id)] for node_rank, node_count, and global_device_ranks. Preserve the existing remote-worker caller contract and eliminate the redundant _remote_like_worker_ids() condition.
7295-7310: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winBind the release to the run that allocated the handle.
_release_global_domain_handlereadsself._building_run_resources, so the pending-release queue it appends to is the run whose graph is being built at release time, not the run that allocated the handle. The local path avoids this:_allocate_domaincaptures the owning_RunResourcesin the_release_fnclosure. For aretain_after_run=Truedomain released inside a later run, the fence that frees it is that later run's fence, and no queue entry exists on the allocating run. Double free is prevented by_release_global_domain_nowmemoization, so this is an ordering concern, not a corruption. Bind the owning resources at allocation for symmetry with_release_domain_handle.♻️ Proposed binding at allocation
- _release_fn=self._release_global_domain_handle, + _release_fn=lambda released, owner=resources: self._release_global_domain_handle(released, owner),Then accept the owning resources explicitly:
- def _release_global_domain_handle(self, handle: GlobalCommDomainHandle) -> None: + def _release_global_domain_handle( + self, handle: GlobalCommDomainHandle, resources: _RunResources | None = None + ) -> None: if self._worker is None: return - resources = self._building_run_resources🤖 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 7295 - 7310, Bind each global domain handle to its allocating _RunResources when created, mirroring _allocate_domain’s _release_fn closure. Update _release_global_domain_handle to accept and use the owning resources rather than reading self._building_run_resources, so pending releases and fences are associated with the allocating run; preserve the existing cleanup and _release_global_domain_now memoization behavior.python/simpler/orchestrator.py (1)
432-441: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winCorrect the docstring: MPI groups are also supported.
The summary line says "without MPI", and the body lists only
Worker.add_workerandWorker.add_remote_worker.Worker._allocate_global_domainalso routes a complete MPI group through_mpi_group_control, anddocs/comm-domain.mddocumentsadd_mpirun_worker_groupmembers. Update the text so users of MPI groups find this API.📝 Proposed docstring fix
- """Create a CommDomain across local and/or remote L3 nodes without MPI. + """Create a CommDomain across local, remote, and MPI-launched L3 nodes. Each member is ``(l3_worker_id, local_l2_worker_id)``. The L3 worker - may have been registered by ``Worker.add_worker`` or - ``Worker.add_remote_worker``. L4 collects every L2 export descriptor, + may have been registered by ``Worker.add_worker``, + ``Worker.add_remote_worker``, or ``Worker.add_mpirun_worker_group``. + L4 collects every L2 export descriptor,🤖 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/orchestrator.py` around lines 432 - 441, Update the docstring for the CommDomain creation method to state that MPI groups are supported, removing the “without MPI” limitation and mentioning MPI group registration alongside Worker.add_worker and Worker.add_remote_worker. Ensure the member description reflects add_mpirun_worker_group usage while preserving the existing lifecycle and commit behavior documentation.src/common/platform_comm/comm_sim.cpp (1)
199-216: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueConsider suppressing implicit moves on
GlobalDomainAllocation.
GlobalDomainAllocationownslocal_baseandshm_nameas raw members and defines a destructor that releases both. The implicitly generated move constructor copieslocal_baseand leaves the source pointer non-null, so a moved-from object wouldmunmapthe same address again. The current code always stores the allocation in astd::unique_ptrand never moves it, so no live path is affected. Deleting the copy/move operations, or reusingGlobalPeerMappingfor the local mapping, removes the hazard for future changes.♻️ Optional hardening
struct GlobalDomainAllocation { + GlobalDomainAllocation() = default; ~GlobalDomainAllocation() { if (local_base != nullptr) { munmap(local_base, mapping_size); } if (!shm_name.empty()) { shm_unlink(shm_name.c_str()); } } + GlobalDomainAllocation(const GlobalDomainAllocation &) = delete; + GlobalDomainAllocation &operator=(const GlobalDomainAllocation &) = delete; + GlobalDomainAllocation(GlobalDomainAllocation &&) = delete; + GlobalDomainAllocation &operator=(GlobalDomainAllocation &&) = delete;🤖 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_comm/comm_sim.cpp` around lines 199 - 216, Make GlobalDomainAllocation non-copyable and non-movable by explicitly deleting its copy and move constructors and assignment operators, preventing duplicated ownership of local_base and shm_name while preserving its current unique_ptr-based usage.src/common/worker/chip_worker.h (1)
161-165: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDocument the new public global-domain APIs.
Every neighbouring
comm_*declaration in this header carries a doc comment that states the ownership and pairing contract. The three new declarations carry none. State whatcomm_global_domain_preparereturns (descriptor bytes, local window base, actual mapping size), that the actual mapping size can exceed the requestedwindow_size, thatcomm_global_domain_importrequires a rank-ordered complete table, and thatcomm_global_domain_releasepairs withprepareand also runs fromfinalize().♻️ Suggested doc comment
+ /// Global CommDomain lifecycle (L4-brokered, independent of the + /// comm_init sessions above). `prepare` creates this rank's local window + /// and returns (descriptor_bytes, local_window_base, mapping_size); the + /// mapping size may exceed `window_size` after backend alignment. + /// `import` takes the complete rank-ordered descriptor table and returns + /// the device CommContext. `release` pairs with `prepare` and is also + /// driven for every tracked domain by `finalize()`. std::tuple<std::vector<uint8_t>, uint64_t, size_t> comm_global_domain_prepare(🤖 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.h` around lines 161 - 165, Document the public APIs comm_global_domain_prepare, comm_global_domain_import, and comm_global_domain_release in chip_worker.h. Specify that prepare returns descriptor bytes, the local window base, and actual mapping size, which may exceed window_size; import requires a complete rank-ordered descriptor table; and release pairs with prepare and is also invoked by finalize().python/simpler/global_comm_domain.py (1)
371-384: 🗄️ Data Integrity & Integration | 🔵 Trivial | 💤 Low valueConsider validating the decoded capability result.
Every other decoder in this module re-validates its fields.
decode_comm_init_resultaccepts anyprofile,max_ranks, anddescriptor_bytes. A peer that reports a different descriptor ABI size passes silently, and the mismatch surfaces later during prepare/import. A cheap check here fails fast.♻️ Optional validation
reader.done("COMM_INIT result") + if profile not in GLOBAL_DOMAIN_PROFILE_IDS: + raise ValueError(f"unsupported global domain profile {profile!r}") + if descriptor_bytes != GLOBAL_DOMAIN_DESCRIPTOR_BYTES or max_ranks == 0 or max_ranks > GLOBAL_DOMAIN_MAX_RANKS: + raise ValueError("global comm init result capability is invalid") result = GlobalCommInitResult(🤖 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/global_comm_domain.py` around lines 371 - 384, Update decode_comm_init_result to validate the decoded profile, max_ranks, and descriptor_bytes before constructing GlobalCommInitResult, reusing the module’s existing validation helpers or conventions. Reject unsupported values, including descriptor ABI sizes that do not match the expected value, while preserving the existing successful decode flow.src/common/worker/chip_worker.cpp (1)
868-875: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winHarden the descriptor ABI and the release-pointer use.
Two points in this block:
- Line 869 calls
comm_global_domain_release_fn_without a null check, unlikecomm_global_domain_releaseat Line 900.init()resolves both symbols with the requiredload_symboland clears both together, so the pointer is non-null whenevercomm_global_domain_prepare_fn_is non-null. The guard is still worth adding for consistency with the surrounding code.- Lines 873-875 ship the raw
CommGlobalDomainDescriptorbytes to Python, which decodes them with the fixed little-endian layout"<IIIIQII256s". Only the total size is asserted (sizeof(...) == 288in both backends). A field reorder that keeps the size constant would silently misdecode. Add per-fieldoffsetofassertions next to the existing size assertion.♻️ Proposed hardening
if (local_window_base == 0 || descriptor.mapping_size == 0) { - comm_global_domain_release_fn_(domain_id); + if (comm_global_domain_release_fn_ != nullptr) { + comm_global_domain_release_fn_(domain_id); + } global_domain_ids_.erase(domain_id); throw std::runtime_error("comm_global_domain_prepare returned an invalid window"); }Add next to the existing size assertion (for example in
src/common/platform_comm/comm.h):static_assert(offsetof(CommGlobalDomainDescriptor, mapping_size) == 16, "descriptor layout changed"); static_assert(offsetof(CommGlobalDomainDescriptor, handle_size) == 24, "descriptor layout changed"); static_assert(offsetof(CommGlobalDomainDescriptor, handle) == 32, "descriptor layout changed");🤖 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 868 - 875, In the invalid-window cleanup within the descriptor preparation flow, guard comm_global_domain_release_fn_ before invoking it, matching the existing comm_global_domain_release handling. Also strengthen CommGlobalDomainDescriptor ABI validation beside its existing size assertion by adding static_assert checks for the mapping_size, handle_size, and handle field offsets required by the Python little-endian decoder.python/bindings/task_interface.cpp (1)
1588-1617: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winConsider releasing the GIL for the global-domain calls.
The three new bindings hold the GIL for the whole native call.
comm_global_domain_preparereserves and maps a VMM/Fabric window, exports a handle, and zeroes the entire window;comm_global_domain_importimports one peer window per rank and copies aCommContextto the device. Both can run for a long time and block every other Python thread in the process. The adjacent device-side helpers in this file already usenb::call_guard<nb::gil_scoped_release>()for exactly this reason, for example_l3_child_onboard_region_createat Line 1839 and_ChipWorker.initat Line 1387.Note that the lambda for
comm_global_domain_prepareconstructsnb::bytesandnb::make_tuplefrom its result, so a whole-lambda call guard is not correct there. Wrap only the native call, or move the Python object construction after the guard scope ends.♻️ Sketch for the prepare binding
[](ChipWorker &self, uint64_t domain_id, uint32_t domain_rank, uint32_t rank_count, size_t window_size, uint32_t profile) { - auto [descriptor, local_window_base, actual_window_size] = - self.comm_global_domain_prepare(domain_id, domain_rank, rank_count, window_size, profile); + std::vector<uint8_t> descriptor; + uint64_t local_window_base = 0; + size_t actual_window_size = 0; + { + nb::gil_scoped_release release; + std::tie(descriptor, local_window_base, actual_window_size) = + self.comm_global_domain_prepare(domain_id, domain_rank, rank_count, window_size, profile); + } return nb::make_tuple(
comm_global_domain_importandcomm_global_domain_releasereturn plain integers, so anb::call_guard<nb::gil_scoped_release>()on the.def(...)is sufficient for those two.🤖 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 1588 - 1617, Release the GIL while executing the native calls in comm_global_domain_prepare, comm_global_domain_import, and comm_global_domain_release. For comm_global_domain_prepare, scope the GIL release only around self.comm_global_domain_prepare so nb::bytes and nb::make_tuple construction still runs with the GIL held; add a whole-binding call guard for the plain-integer import and release methods.
🤖 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/mpi-l3-mailbox.md`:
- Line 1: Add the mpi-l3-mailbox documentation page to the nav configuration in
mkdocs.yml, using the existing title and navigation structure so strict MkDocs
builds include docs/mpi-l3-mailbox.md.
In `@docs/remote-l3-worker-design/implementation-record.md`:
- Around line 95-96: Update the implementation record text to refer to the
shipped A3 Fabric profile as “A3 Fabric V1” rather than “A3 Fabric V2,” matching
the identifiers GLOBAL_DOMAIN_PROFILE_A3_FABRIC and
COMM_GLOBAL_DOMAIN_PROFILE_A3_FABRIC.
In `@python/simpler/global_comm_domain.py`:
- Around line 395-400: Enforce the 64-buffer limit in encode_domain_command
before serializing buffers, using a shared GLOBAL_DOMAIN_MAX_BUFFERS constant.
Update decode_domain_command to use the same constant instead of the existing
hardcoded or unrelated limit, and raise the established validation error when
command.buffers exceeds the bound.
In `@python/simpler/mpi_l3_session.py`:
- Around line 476-495: Add a short sleep/backoff to the mailbox.request_state
polling loop in the MPI session receive flow, and enforce a deadline when the
group remains terminal so it cannot spin indefinitely; also add the same pause
after each request.test() call in the dispatcher loop at
python/simpler/mpi_l3_session.py lines 71-84. Apply both changes in
python/simpler/mpi_l3_session.py:476-495 and
python/simpler/mpi_l3_session.py:71-84, preserving the existing request dispatch
behavior.
- Around line 500-517: Move the _payload_for_rank(request, rank) call inside the
existing try block in the per-rank dispatch logic, before opcode handling.
Preserve the existing MpiRankError conversion so IndexError and ValueError are
captured and included in dispatch_comm.gather rather than escaping the loop.
In `@python/simpler/task_interface.py`:
- Around line 1105-1109: Update TaskInterface.release to set _released before
invoking _release_fn(self), while preserving the existing early return for
already released handles. Keep the callback invocation unchanged so release
failures still propagate, but ensure member() and buffer_range() observe the
handle as released even when the callback raises.
In `@src/common/hierarchical/remote_endpoint.cpp`:
- Around line 904-945: Update both timeout branches in exchange_group_task,
including the dispatch-timeout path after the leader’s run_exchange, to
increment group_departed_ while holding group_mu_ before marking the rendezvous
complete and throwing. Preserve the existing timeout error, terminal marking,
group termination, notification, and exception behavior so the normal reset
logic can clear group_active_ and group_frames_.
- Around line 766-830: Update the polling loop in the request-waiting method
containing OFF_REQUEST_STATE to use bounded backoff instead of continuously
spinning. Add an escalating yield or short sleep on each iteration while
preserving prompt handling of TASK_DONE, SHUTDOWN_DONE, TASK_FAILED, terminal,
and timeout states; reset or initialize the backoff appropriately for each
request.
In `@src/common/platform_comm/comm_sim.cpp`:
- Line 219: Protect accesses to global_domain_allocations in
comm_global_domain_prepare, comm_global_domain_import, and
comm_global_domain_release with a mutex consistently across the sim and HCCL
backends, covering all reads, writes, and erases; alternatively, explicitly
document and enforce the required single-thread caller contract.
In `@task.md`:
- Around line 7-11: Remove personal and internal identifiers from the task
record: replace the branch name, myserver host, and validation host references
in the affected entries with generic placeholders while preserving the remaining
handoff metadata and structure.
In `@tools/a3_l4_tcp_smoke/mpirun_compute_then_tload_2x2_smoke.py`:
- Around line 310-313: Replace the concrete network defaults in
tools/a3_l4_tcp_smoke/mpirun_compute_then_tload_2x2_smoke.py lines 310-313 for
the --host-37, --host-35, --roce-37, and --roce-35 arguments with
documentation-reserved placeholder addresses, or make them required without
defaults. Update the example command in tools/a3_l4_tcp_smoke/README.md lines
17-23 to use the same placeholders.
---
Nitpick comments:
In `@python/bindings/task_interface.cpp`:
- Around line 1588-1617: Release the GIL while executing the native calls in
comm_global_domain_prepare, comm_global_domain_import, and
comm_global_domain_release. For comm_global_domain_prepare, scope the GIL
release only around self.comm_global_domain_prepare so nb::bytes and
nb::make_tuple construction still runs with the GIL held; add a whole-binding
call guard for the plain-integer import and release methods.
In `@python/simpler/global_comm_domain.py`:
- Around line 371-384: Update decode_comm_init_result to validate the decoded
profile, max_ranks, and descriptor_bytes before constructing
GlobalCommInitResult, reusing the module’s existing validation helpers or
conventions. Reject unsupported values, including descriptor ABI sizes that do
not match the expected value, while preserving the existing successful decode
flow.
In `@python/simpler/global_comm_smoke.py`:
- Around line 22-167: Refactor remote_compute_orch, remote_rank_orch,
remote_compute_group_orch, and remote_rank_group_orch to share a helper for
importing get_inner_handle, validating the six scalars, decoding the digest, and
resolving the domain. Add reusable tensor-argument builders for the compute and
TLOAD shapes, then have the group callbacks reuse the corresponding
single-worker argument construction inside their loops while preserving worker
selection and submission behavior.
In `@python/simpler/mpi_group_mailbox.py`:
- Around line 320-329: Update the error serialization flow in the mailbox
failure-writing method so oversized data remains valid JSON. Truncate individual
rank-error message fields before `json.dumps`, then encode and write the
complete serialized document without slicing the encoded JSON at
`MAILBOX_ERROR_BYTES`; preserve rank attribution and the existing state updates.
- Around line 345-357: The TASK_FAILED error decoding in the mailbox request
handling must not catch control-flow exceptions. Narrow the try/except around
json decoding and entry field access to the specific decode and lookup/type
errors that can occur, replacing the broad BaseException handler so Ruff BLE001
is resolved while preserving the raw-message fallback.
- Around line 373-375: The test-only method overwrite_request_payload_for_test
should not remain exposed on the shipped mailbox protocol class. Remove it and
update the sole caller in test_mpi_group_mailbox.py to write the payload
directly through mailbox._buffer, or otherwise enforce MAILBOX_PAYLOAD_BYTES
capacity before writing if the helper must remain.
In `@python/simpler/mpi_l3_session.py`:
- Around line 504-513: Replace the assert in the MailboxOpcode.SHUTDOWN branch
of the request handling flow with an explicit payload None check that raises an
appropriate error before calling _rewrite_frame_identity; preserve the existing
shutdown behavior when payload is present.
In `@python/simpler/orchestrator.py`:
- Around line 432-441: Update the docstring for the CommDomain creation method
to state that MPI groups are supported, removing the “without MPI” limitation
and mentioning MPI group registration alongside Worker.add_worker and
Worker.add_remote_worker. Ensure the member description reflects
add_mpirun_worker_group usage while preserving the existing lifecycle and commit
behavior documentation.
In `@python/simpler/worker.py`:
- Around line 3515-3523: Remove the unreachable else fallback in
_build_remote_manifest and rely directly on the runtime values from
_resolved_global_nodes()[int(worker_id)] for node_rank, node_count, and
global_device_ranks. Preserve the existing remote-worker caller contract and
eliminate the redundant _remote_like_worker_ids() condition.
- Around line 7295-7310: Bind each global domain handle to its allocating
_RunResources when created, mirroring _allocate_domain’s _release_fn closure.
Update _release_global_domain_handle to accept and use the owning resources
rather than reading self._building_run_resources, so pending releases and fences
are associated with the allocating run; preserve the existing cleanup and
_release_global_domain_now memoization behavior.
In `@src/common/platform_comm/comm_sim.cpp`:
- Around line 199-216: Make GlobalDomainAllocation non-copyable and non-movable
by explicitly deleting its copy and move constructors and assignment operators,
preventing duplicated ownership of local_base and shm_name while preserving its
current unique_ptr-based usage.
In `@src/common/worker/chip_worker.cpp`:
- Around line 868-875: In the invalid-window cleanup within the descriptor
preparation flow, guard comm_global_domain_release_fn_ before invoking it,
matching the existing comm_global_domain_release handling. Also strengthen
CommGlobalDomainDescriptor ABI validation beside its existing size assertion by
adding static_assert checks for the mapping_size, handle_size, and handle field
offsets required by the Python little-endian decoder.
In `@src/common/worker/chip_worker.h`:
- Around line 161-165: Document the public APIs comm_global_domain_prepare,
comm_global_domain_import, and comm_global_domain_release in chip_worker.h.
Specify that prepare returns descriptor bytes, the local window base, and actual
mapping size, which may exceed window_size; import requires a complete
rank-ordered descriptor table; and release pairs with prepare and is also
invoked by finalize().
In `@tests/ut/cpp/hierarchical/test_remote_endpoint.cpp`:
- Around line 659-661: Bound the polling loops in both respond_with_payloads and
the matching wait near exchange_group_task with a deadline; when
RequestState::REQUEST_READY is not observed before expiration, call FAIL() with
a diagnostic, while preserving the existing behavior when the state becomes
ready.
In `@tests/ut/py/test_global_comm_domain.py`:
- Around line 259-278: Update _failure_injection_worker to obtain two ports
through the existing _free_tcp_ports helper, then build each RemoteWorkerSpec
endpoint from those allocated ports instead of the fixed 19073 + index range.
Preserve the current node ordering and endpoint construction format.
In `@tests/ut/py/test_mpi_group_mailbox.py`:
- Around line 25-211: Add a test covering the mailbox payload capacity guard by
attempting to write a request whose encoded payloads exceed
MAILBOX_PAYLOAD_BYTES and asserting ValueError. Exercise the write_request path
and ensure the oversized payload vector is rejected before acceptance or state
progression.
In `@tools/a3_l4_tcp_smoke/mpirun_compute_then_tload_2x2_smoke.py`:
- Around line 51-60: Update the kernel_include_dirs construction in _compile_aiv
to use iterable unpacking when combining include_dirs with the common source
directory, preserving the existing ordering and values.
In `@tools/mpi_group_mailbox_smoke.py`:
- Around line 34-37: Update _wait_until to pause briefly between predicate
evaluations, while retaining the existing deadline check and TimeoutError
behavior; add the sleep inside the loop after a failed predicate check.
In `@tools/mpi_l3_group_smoke.py`:
- Around line 41-43: Restore the SIMPLER_MPI_SMOKE_DIR environment variable
after the temporary-directory run in the surrounding run flow: save any prior
value before assigning output_dir, then restore it in a finally block (or remove
it when absent) so repeated or nested calls never retain the deleted path.
🪄 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: a870b115-1c58-4057-9696-095762c08e11
📒 Files selected for processing (50)
docs/comm-domain.mddocs/mpi-l3-mailbox.mddocs/remote-l3-worker-design.mddocs/remote-l3-worker-design/implementation-record.mddocs/remote-l3-worker-design/protocol.mdpython/bindings/CMakeLists.txtpython/bindings/task_interface.cpppython/bindings/worker_bind.hpython/simpler/global_comm_domain.pypython/simpler/global_comm_smoke.pypython/simpler/mpi_group_mailbox.pypython/simpler/mpi_group_smoke.pypython/simpler/mpi_l3_session.pypython/simpler/orchestrator.pypython/simpler/remote_l3_protocol.pypython/simpler/remote_l3_session.pypython/simpler/remote_l3_worker.pypython/simpler/task_interface.pypython/simpler/worker.pysrc/a2a3/platform/onboard/host/comm_hccl.cppsrc/a5/platform/onboard/host/comm_hccl.cppsrc/common/hierarchical/mpi_group_mailbox.hsrc/common/hierarchical/remote_endpoint.cppsrc/common/hierarchical/remote_endpoint.hsrc/common/hierarchical/remote_wire.cppsrc/common/hierarchical/remote_wire.hsrc/common/hierarchical/worker.cppsrc/common/hierarchical/worker.hsrc/common/hierarchical/worker_manager.cppsrc/common/hierarchical/worker_manager.hsrc/common/platform_comm/comm.hsrc/common/platform_comm/comm_sim.cppsrc/common/worker/chip_worker.cppsrc/common/worker/chip_worker.htask.mdtests/ut/cpp/CMakeLists.txttests/ut/cpp/hierarchical/test_remote_endpoint.cpptests/ut/py/test_callable_identity.pytests/ut/py/test_global_comm_domain.pytests/ut/py/test_mpi_group_mailbox.pytests/ut/py/test_mpi_l3_group.pytests/ut/py/test_worker/test_startup_readiness.pytools/a3_l4_tcp_smoke/README.mdtools/a3_l4_tcp_smoke/kernels/aiv/global_tload_kernel.cpptools/a3_l4_tcp_smoke/kernels/aiv/local_add_kernel.cpptools/a3_l4_tcp_smoke/kernels/orchestration/global_tload_orch.cpptools/a3_l4_tcp_smoke/kernels/orchestration/local_add_orch.cpptools/a3_l4_tcp_smoke/mpirun_compute_then_tload_2x2_smoke.pytools/mpi_group_mailbox_smoke.pytools/mpi_l3_group_smoke.py
| @@ -0,0 +1,132 @@ | |||
| # MPI L3 group mailbox protocol | |||
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Add this page to the MkDocs nav.
The docs build reports that docs/mpi-l3-mailbox.md exists but is not listed in the nav configuration. In strict mode this warning can fail the build. Add an entry for this page to mkdocs.yml.
🧰 Tools
🪛 GitHub Actions: docs / 1_build.txt
[warning] 1-1: MkDocs warning: This page exists in the docs directory but is not included in the nav configuration.
🪛 GitHub Actions: docs / build
[warning] 1-1: MkDocs strict mode warning: this page exists in the docs directory but is not included in the nav configuration.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/mpi-l3-mailbox.md` at line 1, Add the mpi-l3-mailbox documentation page
to the nav configuration in mkdocs.yml, using the existing title and navigation
structure so strict MkDocs builds include docs/mpi-l3-mailbox.md.
Source: Pipeline failures
Comment on lines
+95
to
+96
| released or the Worker closes. Sim shm and A3 Fabric V2 use the same | ||
| descriptor ABI. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Correct the Fabric profile version.
The text states "A3 Fabric V2". The implemented profile identifier is a3-fabric-v1 (GLOBAL_DOMAIN_PROFILE_A3_FABRIC in python/simpler/global_comm_domain.py Line 23), and the native constant is COMM_GLOBAL_DOMAIN_PROFILE_A3_FABRIC. Align the document with the shipped profile name.
📝 Proposed fix
- released or the Worker closes. Sim shm and A3 Fabric V2 use the same
- descriptor ABI.
+ released or the Worker closes. The sim shm and `a3-fabric-v1` profiles use
+ the same descriptor ABI.📝 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.
Suggested change
| released or the Worker closes. Sim shm and A3 Fabric V2 use the same | |
| descriptor ABI. | |
| released or the Worker closes. The sim shm and `a3-fabric-v1` profiles use | |
| the same descriptor ABI. |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/remote-l3-worker-design/implementation-record.md` around lines 95 - 96,
Update the implementation record text to refer to the shipped A3 Fabric profile
as “A3 Fabric V1” rather than “A3 Fabric V2,” matching the identifiers
GLOBAL_DOMAIN_PROFILE_A3_FABRIC and COMM_GLOBAL_DOMAIN_PROFILE_A3_FABRIC.
Comment on lines
+395
to
+400
| if len({buffer.name for buffer in command.buffers}) != len(command.buffers): | ||
| raise ValueError("global domain command contains duplicate buffer names") | ||
| if any(not buffer.name or buffer.nbytes <= 0 for buffer in command.buffers): | ||
| raise ValueError("global domain buffers require a name and positive size") | ||
| if sum(buffer.nbytes for buffer in command.buffers) > command.window_size: | ||
| raise ValueError("global domain buffers exceed the requested window") |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Bound the buffer count on the encode side.
decode_domain_command rejects buffer_count > GLOBAL_DOMAIN_MAX_RANKS (Line 453), but encode_domain_command applies no count limit. A command with more than 64 buffers encodes successfully and then fails to decode at the peer. Enforce the same bound when encoding so the producer cannot build an undecodable frame.
🐛 Proposed fix
+ if len(command.buffers) > GLOBAL_DOMAIN_MAX_BUFFERS:
+ raise ValueError("global domain command buffer count exceeds maximum")
if len({buffer.name for buffer in command.buffers}) != len(command.buffers):
raise ValueError("global domain command contains duplicate buffer names")Add the constant and use it in decode_domain_command as well:
GLOBAL_DOMAIN_MAX_BUFFERS = 64🤖 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/global_comm_domain.py` around lines 395 - 400, Enforce the
64-buffer limit in encode_domain_command before serializing buffers, using a
shared GLOBAL_DOMAIN_MAX_BUFFERS constant. Update decode_domain_command to use
the same constant instead of the existing hardcoded or unrelated limit, and
raise the established validation error when command.buffers exceeds the bound.
Comment on lines
+476
to
+495
| while True: | ||
| if rank == 0: | ||
| assert mailbox is not None | ||
| while mailbox.request_state not in ( | ||
| MailboxRequestState.REQUEST_READY, | ||
| MailboxRequestState.SHUTDOWN_READY, | ||
| ): | ||
| if mailbox.group_state is MailboxGroupState.TERMINAL: | ||
| break | ||
| if mailbox.group_state is MailboxGroupState.TERMINAL: | ||
| request = None | ||
| else: | ||
| try: | ||
| request = mailbox.accept_request(last_sequence_id=last_sequence_id) | ||
| last_sequence_id = request.sequence_id | ||
| except BaseException: | ||
| request = None | ||
| else: | ||
| request = None | ||
| request = dispatch_comm.bcast(request, root=0) |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
Busy-wait loops with no backoff in the MPI session. Both wait loops re-check their condition without any pause, so each rank consumes a full core while it waits. The shared root cause is a missing sleep or backoff in the polling loops.
python/simpler/mpi_l3_session.py#L476-L495: add a short sleep inside thewhile mailbox.request_state not in (...)loop on lines 479-484, and add a terminal-state deadline so a stalled group does not spin forever.python/simpler/mpi_l3_session.py#L71-L84: add a short sleep after eachrequest.test()call so the dispatcher thread does not spin until the timeout expires.
🧰 Tools
🪛 Ruff (0.16.0)
[warning] 491-491: Do not catch blind exception: BaseException
(BLE001)
📍 Affects 1 file
python/simpler/mpi_l3_session.py#L476-L495(this comment)python/simpler/mpi_l3_session.py#L71-L84
🤖 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/mpi_l3_session.py` around lines 476 - 495, Add a short
sleep/backoff to the mailbox.request_state polling loop in the MPI session
receive flow, and enforce a deadline when the group remains terminal so it
cannot spin indefinitely; also add the same pause after each request.test() call
in the dispatcher loop at python/simpler/mpi_l3_session.py lines 71-84. Apply
both changes in python/simpler/mpi_l3_session.py:476-495 and
python/simpler/mpi_l3_session.py:71-84, preserving the existing request dispatch
behavior.
Comment on lines
+500
to
+517
| local_reply = b"" | ||
| local_error: MpiRankError | None = None | ||
| payload = _payload_for_rank(request, rank) | ||
| try: | ||
| if request.opcode is MailboxOpcode.PING: | ||
| local_reply = b"" | ||
| elif request.opcode is MailboxOpcode.SHUTDOWN: | ||
| assert payload is not None | ||
| connection.feed(_rewrite_frame_identity(payload, manifest)) | ||
| elif payload is not None: | ||
| local_command_sequence += 1 | ||
| local_reply = connection.exchange( | ||
| _rewrite_frame_identity(payload, manifest, sequence=local_command_sequence) | ||
| ) | ||
| except BaseException as exc: # noqa: BLE001 | ||
| local_error = MpiRankError(rank, type(exc).__name__, str(exc)) | ||
|
|
||
| gathered = dispatch_comm.gather((rank, payload is not None, local_reply, local_error), root=0) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Compute the per-rank payload inside the guarded block.
Line 502 runs outside the try that starts on line 503. _payload_for_rank raises IndexError when request.payloads is shorter than the world size, and raises ValueError for an unknown target (line 330). Both escape the dispatch loop directly into the finally block. The rank then exits without calling dispatch_comm.gather, so every other rank blocks in the gather and the mailbox request never completes. Move the call inside the try so the failure is reported as an MpiRankError.
🐛 Proposed fix
local_reply = b""
local_error: MpiRankError | None = None
- payload = _payload_for_rank(request, rank)
+ payload: bytes | None = None
try:
+ payload = _payload_for_rank(request, rank)
if request.opcode is MailboxOpcode.PING:📝 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.
Suggested change
| local_reply = b"" | |
| local_error: MpiRankError | None = None | |
| payload = _payload_for_rank(request, rank) | |
| try: | |
| if request.opcode is MailboxOpcode.PING: | |
| local_reply = b"" | |
| elif request.opcode is MailboxOpcode.SHUTDOWN: | |
| assert payload is not None | |
| connection.feed(_rewrite_frame_identity(payload, manifest)) | |
| elif payload is not None: | |
| local_command_sequence += 1 | |
| local_reply = connection.exchange( | |
| _rewrite_frame_identity(payload, manifest, sequence=local_command_sequence) | |
| ) | |
| except BaseException as exc: # noqa: BLE001 | |
| local_error = MpiRankError(rank, type(exc).__name__, str(exc)) | |
| gathered = dispatch_comm.gather((rank, payload is not None, local_reply, local_error), root=0) | |
| local_reply = b"" | |
| local_error: MpiRankError | None = None | |
| payload: bytes | None = None | |
| try: | |
| payload = _payload_for_rank(request, rank) | |
| if request.opcode is MailboxOpcode.PING: | |
| local_reply = b"" | |
| elif request.opcode is MailboxOpcode.SHUTDOWN: | |
| assert payload is not None | |
| connection.feed(_rewrite_frame_identity(payload, manifest)) | |
| elif payload is not None: | |
| local_command_sequence += 1 | |
| local_reply = connection.exchange( | |
| _rewrite_frame_identity(payload, manifest, sequence=local_command_sequence) | |
| ) | |
| except BaseException as exc: # noqa: BLE001 | |
| local_error = MpiRankError(rank, type(exc).__name__, str(exc)) | |
| gathered = dispatch_comm.gather((rank, payload is not None, local_reply, local_error), root=0) |
🤖 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/mpi_l3_session.py` around lines 500 - 517, Move the
_payload_for_rank(request, rank) call inside the existing try block in the
per-rank dispatch logic, before opcode handling. Preserve the existing
MpiRankError conversion so IndexError and ValueError are captured and included
in dispatch_comm.gather rather than escaping the loop.
Comment on lines
+766
to
+830
| const Deadline deadline = deadline_from_now(runtime_timeout_s_); | ||
| while (true) { | ||
| const auto state = static_cast<RequestState>(load_i32(OFF_REQUEST_STATE)); | ||
| if (state == RequestState::TASK_DONE) { | ||
| const uint32_t response_count = read_u32(OFF_RESPONSE_COUNT); | ||
| const size_t response_bytes = read_u32(OFF_RESPONSE_BYTES); | ||
| const uint32_t expected_count = target == Target::PER_RANK ? static_cast<uint32_t>(world_size_) : 1U; | ||
| const size_t prefix_bytes = 4 + 4 * static_cast<size_t>(response_count); | ||
| if (response_count != expected_count || response_bytes < prefix_bytes || response_bytes > PAYLOAD_BYTES) { | ||
| mark_terminal("MPI group mailbox returned an invalid response vector"); | ||
| kill_mpirun_group(); | ||
| throw std::runtime_error("MpiGroupMailboxChannel: invalid response vector"); | ||
| } | ||
| uint32_t encoded_count = 0; | ||
| std::memcpy(&encoded_count, mailbox_ + RESPONSE_OFFSET, sizeof(encoded_count)); | ||
| if (encoded_count != response_count) { | ||
| mark_terminal("MPI group mailbox response vector length mismatch"); | ||
| kill_mpirun_group(); | ||
| throw std::runtime_error("MpiGroupMailboxChannel: response vector length mismatch"); | ||
| } | ||
| std::vector<uint32_t> payload_sizes(response_count); | ||
| size_t total_payload_bytes = 0; | ||
| for (uint32_t i = 0; i < response_count; ++i) { | ||
| std::memcpy( | ||
| &payload_sizes[i], mailbox_ + RESPONSE_OFFSET + 4 + 4 * static_cast<size_t>(i), | ||
| sizeof(payload_sizes[i]) | ||
| ); | ||
| total_payload_bytes += payload_sizes[i]; | ||
| } | ||
| if (prefix_bytes + total_payload_bytes != response_bytes) { | ||
| mark_terminal("MPI group mailbox response vector length mismatch"); | ||
| kill_mpirun_group(); | ||
| throw std::runtime_error("MpiGroupMailboxChannel: response vector length mismatch"); | ||
| } | ||
| std::vector<std::vector<uint8_t>> responses; | ||
| responses.reserve(response_count); | ||
| size_t response_offset = RESPONSE_OFFSET + prefix_bytes; | ||
| for (uint32_t payload_size : payload_sizes) { | ||
| std::vector<uint8_t> response(payload_size); | ||
| if (payload_size > 0) { | ||
| std::memcpy(response.data(), mailbox_ + response_offset, payload_size); | ||
| } | ||
| response_offset += payload_size; | ||
| responses.push_back(std::move(response)); | ||
| } | ||
| store_i32(OFF_REQUEST_STATE, static_cast<int32_t>(RequestState::IDLE)); | ||
| return responses; | ||
| } | ||
| if (state == RequestState::SHUTDOWN_DONE) { | ||
| store_i32(OFF_REQUEST_STATE, static_cast<int32_t>(RequestState::IDLE)); | ||
| return {{}}; | ||
| } | ||
| if (state == RequestState::TASK_FAILED || terminal()) { | ||
| const std::string reason = terminal_reason(); | ||
| if (!terminal()) store_i32(OFF_REQUEST_STATE, static_cast<int32_t>(RequestState::IDLE)); | ||
| throw std::runtime_error( | ||
| "MpiGroupMailboxChannel: MPI group request failed" + (reason.empty() ? std::string() : ": " + reason) | ||
| ); | ||
| } | ||
| if (std::chrono::steady_clock::now() >= deadline) { | ||
| mark_terminal("MPI group mailbox request timed out at sequence " + std::to_string(sequence)); | ||
| kill_mpirun_group(); | ||
| throw std::runtime_error("MpiGroupMailboxChannel: request timed out"); | ||
| } | ||
| } |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy lift
Replace the busy-wait poll with a bounded backoff.
The loop polls OFF_REQUEST_STATE with no sleep and no yield. It runs on the dispatcher thread and spins until the reply arrives or the deadline expires. For every task and control this consumes one full core for the whole remote execution time, and up to runtime_timeout_s_ when a rank hangs. The previous TCP transport blocked in the kernel instead. On a host that also runs L3 and L2 worker threads, this steals CPU from the compute path.
Add a short sleep or an escalating yield/sleep backoff inside the loop.
♻️ Proposed backoff
const Deadline deadline = deadline_from_now(runtime_timeout_s_);
+ int spins = 0;
while (true) {
const auto state = static_cast<RequestState>(load_i32(OFF_REQUEST_STATE)); if (std::chrono::steady_clock::now() >= deadline) {
mark_terminal("MPI group mailbox request timed out at sequence " + std::to_string(sequence));
kill_mpirun_group();
throw std::runtime_error("MpiGroupMailboxChannel: request timed out");
}
+ if (++spins < 1024) {
+ std::this_thread::yield();
+ } else {
+ std::this_thread::sleep_for(std::chrono::microseconds(200));
+ }
}📝 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.
Suggested change
| const Deadline deadline = deadline_from_now(runtime_timeout_s_); | |
| while (true) { | |
| const auto state = static_cast<RequestState>(load_i32(OFF_REQUEST_STATE)); | |
| if (state == RequestState::TASK_DONE) { | |
| const uint32_t response_count = read_u32(OFF_RESPONSE_COUNT); | |
| const size_t response_bytes = read_u32(OFF_RESPONSE_BYTES); | |
| const uint32_t expected_count = target == Target::PER_RANK ? static_cast<uint32_t>(world_size_) : 1U; | |
| const size_t prefix_bytes = 4 + 4 * static_cast<size_t>(response_count); | |
| if (response_count != expected_count || response_bytes < prefix_bytes || response_bytes > PAYLOAD_BYTES) { | |
| mark_terminal("MPI group mailbox returned an invalid response vector"); | |
| kill_mpirun_group(); | |
| throw std::runtime_error("MpiGroupMailboxChannel: invalid response vector"); | |
| } | |
| uint32_t encoded_count = 0; | |
| std::memcpy(&encoded_count, mailbox_ + RESPONSE_OFFSET, sizeof(encoded_count)); | |
| if (encoded_count != response_count) { | |
| mark_terminal("MPI group mailbox response vector length mismatch"); | |
| kill_mpirun_group(); | |
| throw std::runtime_error("MpiGroupMailboxChannel: response vector length mismatch"); | |
| } | |
| std::vector<uint32_t> payload_sizes(response_count); | |
| size_t total_payload_bytes = 0; | |
| for (uint32_t i = 0; i < response_count; ++i) { | |
| std::memcpy( | |
| &payload_sizes[i], mailbox_ + RESPONSE_OFFSET + 4 + 4 * static_cast<size_t>(i), | |
| sizeof(payload_sizes[i]) | |
| ); | |
| total_payload_bytes += payload_sizes[i]; | |
| } | |
| if (prefix_bytes + total_payload_bytes != response_bytes) { | |
| mark_terminal("MPI group mailbox response vector length mismatch"); | |
| kill_mpirun_group(); | |
| throw std::runtime_error("MpiGroupMailboxChannel: response vector length mismatch"); | |
| } | |
| std::vector<std::vector<uint8_t>> responses; | |
| responses.reserve(response_count); | |
| size_t response_offset = RESPONSE_OFFSET + prefix_bytes; | |
| for (uint32_t payload_size : payload_sizes) { | |
| std::vector<uint8_t> response(payload_size); | |
| if (payload_size > 0) { | |
| std::memcpy(response.data(), mailbox_ + response_offset, payload_size); | |
| } | |
| response_offset += payload_size; | |
| responses.push_back(std::move(response)); | |
| } | |
| store_i32(OFF_REQUEST_STATE, static_cast<int32_t>(RequestState::IDLE)); | |
| return responses; | |
| } | |
| if (state == RequestState::SHUTDOWN_DONE) { | |
| store_i32(OFF_REQUEST_STATE, static_cast<int32_t>(RequestState::IDLE)); | |
| return {{}}; | |
| } | |
| if (state == RequestState::TASK_FAILED || terminal()) { | |
| const std::string reason = terminal_reason(); | |
| if (!terminal()) store_i32(OFF_REQUEST_STATE, static_cast<int32_t>(RequestState::IDLE)); | |
| throw std::runtime_error( | |
| "MpiGroupMailboxChannel: MPI group request failed" + (reason.empty() ? std::string() : ": " + reason) | |
| ); | |
| } | |
| if (std::chrono::steady_clock::now() >= deadline) { | |
| mark_terminal("MPI group mailbox request timed out at sequence " + std::to_string(sequence)); | |
| kill_mpirun_group(); | |
| throw std::runtime_error("MpiGroupMailboxChannel: request timed out"); | |
| } | |
| } | |
| const Deadline deadline = deadline_from_now(runtime_timeout_s_); | |
| int spins = 0; | |
| while (true) { | |
| const auto state = static_cast<RequestState>(load_i32(OFF_REQUEST_STATE)); | |
| if (state == RequestState::TASK_DONE) { | |
| const uint32_t response_count = read_u32(OFF_RESPONSE_COUNT); | |
| const size_t response_bytes = read_u32(OFF_RESPONSE_BYTES); | |
| const uint32_t expected_count = target == Target::PER_RANK ? static_cast<uint32_t>(world_size_) : 1U; | |
| const size_t prefix_bytes = 4 + 4 * static_cast<size_t>(response_count); | |
| if (response_count != expected_count || response_bytes < prefix_bytes || response_bytes > PAYLOAD_BYTES) { | |
| mark_terminal("MPI group mailbox returned an invalid response vector"); | |
| kill_mpirun_group(); | |
| throw std::runtime_error("MpiGroupMailboxChannel: invalid response vector"); | |
| } | |
| uint32_t encoded_count = 0; | |
| std::memcpy(&encoded_count, mailbox_ + RESPONSE_OFFSET, sizeof(encoded_count)); | |
| if (encoded_count != response_count) { | |
| mark_terminal("MPI group mailbox response vector length mismatch"); | |
| kill_mpirun_group(); | |
| throw std::runtime_error("MpiGroupMailboxChannel: response vector length mismatch"); | |
| } | |
| std::vector<uint32_t> payload_sizes(response_count); | |
| size_t total_payload_bytes = 0; | |
| for (uint32_t i = 0; i < response_count; ++i) { | |
| std::memcpy( | |
| &payload_sizes[i], mailbox_ + RESPONSE_OFFSET + 4 + 4 * static_cast<size_t>(i), | |
| sizeof(payload_sizes[i]) | |
| ); | |
| total_payload_bytes += payload_sizes[i]; | |
| } | |
| if (prefix_bytes + total_payload_bytes != response_bytes) { | |
| mark_terminal("MPI group mailbox response vector length mismatch"); | |
| kill_mpirun_group(); | |
| throw std::runtime_error("MpiGroupMailboxChannel: response vector length mismatch"); | |
| } | |
| std::vector<std::vector<uint8_t>> responses; | |
| responses.reserve(response_count); | |
| size_t response_offset = RESPONSE_OFFSET + prefix_bytes; | |
| for (uint32_t payload_size : payload_sizes) { | |
| std::vector<uint8_t> response(payload_size); | |
| if (payload_size > 0) { | |
| std::memcpy(response.data(), mailbox_ + response_offset, payload_size); | |
| } | |
| response_offset += payload_size; | |
| responses.push_back(std::move(response)); | |
| } | |
| store_i32(OFF_REQUEST_STATE, static_cast<int32_t>(RequestState::IDLE)); | |
| return responses; | |
| } | |
| if (state == RequestState::SHUTDOWN_DONE) { | |
| store_i32(OFF_REQUEST_STATE, static_cast<int32_t>(RequestState::IDLE)); | |
| return {{}}; | |
| } | |
| if (state == RequestState::TASK_FAILED || terminal()) { | |
| const std::string reason = terminal_reason(); | |
| if (!terminal()) store_i32(OFF_REQUEST_STATE, static_cast<int32_t>(RequestState::IDLE)); | |
| throw std::runtime_error( | |
| "MpiGroupMailboxChannel: MPI group request failed" + (reason.empty() ? std::string() : ": " + reason) | |
| ); | |
| } | |
| if (std::chrono::steady_clock::now() >= deadline) { | |
| mark_terminal("MPI group mailbox request timed out at sequence " + std::to_string(sequence)); | |
| kill_mpirun_group(); | |
| throw std::runtime_error("MpiGroupMailboxChannel: request timed out"); | |
| } | |
| if (++spins < 1024) { | |
| std::this_thread::yield(); | |
| } else { | |
| std::this_thread::sleep_for(std::chrono::microseconds(200)); | |
| } | |
| } |
🤖 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/remote_endpoint.cpp` around lines 766 - 830, Update
the polling loop in the request-waiting method containing OFF_REQUEST_STATE to
use bounded backoff instead of continuously spinning. Add an escalating yield or
short sleep on each iteration while preserving prompt handling of TASK_DONE,
SHUTDOWN_DONE, TASK_FAILED, terminal, and timeout states; reset or initialize
the backoff appropriately for each request.
Comment on lines
+904
to
+945
| while (!group_done_) { | ||
| if (group_cv_.wait_until(lock, deadline) == std::cv_status::timeout) { | ||
| group_error_ = std::make_exception_ptr(std::runtime_error("MPI group task batching timed out")); | ||
| group_done_ = true; | ||
| lock.unlock(); | ||
| mark_terminal("MPI group task batching timed out waiting for all rank payloads"); | ||
| kill_mpirun_group(); | ||
| group_cv_.notify_all(); | ||
| throw std::runtime_error("MpiGroupMailboxChannel: group task batching timed out"); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| if (leader) { | ||
| try { | ||
| std::lock_guard<std::mutex> lane_lock(lane_mu_); | ||
| auto replies = | ||
| run_exchange(frames, mpi_group_mailbox::Opcode::TASK, mpi_group_mailbox::Target::PER_RANK, -1); | ||
| std::lock_guard<std::mutex> group_lock(group_mu_); | ||
| group_replies_ = std::move(replies); | ||
| group_done_ = true; | ||
| } catch (...) { | ||
| std::lock_guard<std::mutex> group_lock(group_mu_); | ||
| group_error_ = std::current_exception(); | ||
| group_done_ = true; | ||
| } | ||
| group_cv_.notify_all(); | ||
| } | ||
|
|
||
| std::unique_lock<std::mutex> lock(group_mu_); | ||
| while (!group_done_) { | ||
| if (group_cv_.wait_until(lock, deadline) == std::cv_status::timeout) { | ||
| group_error_ = std::make_exception_ptr(std::runtime_error("MPI group task dispatch timed out")); | ||
| group_done_ = true; | ||
| lock.unlock(); | ||
| mark_terminal("MPI group task dispatch timed out"); | ||
| kill_mpirun_group(); | ||
| group_cv_.notify_all(); | ||
| throw std::runtime_error("MpiGroupMailboxChannel: group task dispatch timed out"); | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Timeout paths leave the group rendezvous permanently active.
Both timeout branches set group_error_ and group_done_, then throw without incrementing group_departed_. The remaining participants wake, copy the error, and each increments group_departed_ once. The total then reaches world_size_ - 1, so the reset block at lines 958-965 never runs. group_active_ stays true, group_frames_ keeps the stale per-rank frames, and any later exchange_group_task with a different task_slot blocks at line 876 until its own deadline.
The group is already marked terminal on these paths, so the practical impact is limited. Still, reset the rendezvous state on the timeout paths so the channel state stays consistent.
🐛 Suggested reset on timeout
while (!group_done_) {
if (group_cv_.wait_until(lock, deadline) == std::cv_status::timeout) {
group_error_ = std::make_exception_ptr(std::runtime_error("MPI group task batching timed out"));
group_done_ = true;
+ ++group_departed_;
lock.unlock();
mark_terminal("MPI group task batching timed out waiting for all rank payloads");
kill_mpirun_group();
group_cv_.notify_all();
throw std::runtime_error("MpiGroupMailboxChannel: group task batching timed out");
}
}Apply the same ++group_departed_ to the dispatch-timeout branch at lines 936-944.
🤖 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/remote_endpoint.cpp` around lines 904 - 945, Update
both timeout branches in exchange_group_task, including the dispatch-timeout
path after the leader’s run_exchange, to increment group_departed_ while holding
group_mu_ before marking the rendezvous complete and throwing. Preserve the
existing timeout error, terminal marking, group termination, notification, and
exception behavior so the normal reset logic can clear group_active_ and
group_frames_.
| }; | ||
|
|
||
| static_assert(sizeof(CommGlobalDomainDescriptor) == 288, "global domain descriptor ABI changed"); | ||
| static std::unordered_map<uint64_t, std::unique_ptr<GlobalDomainAllocation>> global_domain_allocations; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Trace callers of the global-domain entry points and any documented threading contract.
set -euo pipefail
rg -nP -C 6 '\bcomm_global_domain_(prepare|import|release)\b' --type=cpp --type=h
rg -nP -C 6 'global_domain_ids_|comm_global_domain_release_fn_' src/common/worker
rg -n -C 4 'thread' src/common/platform_comm/comm.hRepository: hw-native-sys/simpler
Length of output: 159
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== files =="
git ls-files | rg '(^|/)(comm_sim\.cpp|comm_hccl\.cpp|comm\.h)$|src/common/worker|src/common/platform_comm' | head -200
echo "== target file context =="
if [ -f src/common/platform_comm/comm_sim.cpp ]; then
nl -ba src/common/platform_comm/comm_sim.cpp | sed -n '180,245p'
else
echo "missing src/common/platform_comm/comm_sim.cpp"
fi
echo "== hccl context =="
if [ -f src/a2a3/platform/onboard/host/comm_hccl.cpp ]; then
nl -ba src/a2a3/platform/onboard/host/comm_hccl.cpp | sed -n '70,125p'
else
echo "missing src/a2a3/platform/onboard/host/comm_hccl.cpp"
fi
echo "== global_domain references (all) =="
rg -n -C 4 'global_domain_allocations|comm_global_domain_(prepare|import|release)|global_domain_ids_|comm_global_domain_release_fn_|thread' . --glob '*.{cpp,c,h,hpp}' || trueRepository: hw-native-sys/simpler
Length of output: 616
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Behavioral/static verifier: find all entry points and non-static non-mutex-protected direct users of the mutable global map.
python3 - <<'PY'
from pathlib import Path
import re
files = {Path(p) for p in str(s).splitlines() for s in [__import__('subprocess').run(['git','ls-files'],capture_output=True).stdout.decode().splitlines()]}
for f in files:
if 'src/common/platform_comm/comm_sim.cpp' not in str(f) and 'src/a2a3/platform/onboard/host/comm_hccl.cpp' not in str(f):
continue
text = f.read_text(errors='ignore')
for m in re.finditer(r'comm_global_domain_(prepare|import|release)\b', text):
line=text.count('\n',0,m.start())+1
col=text.count('\t',0,m.start()-text.rfind('\n',0,m.start()))+1
print(f"{f}:{line}:{col}: {m.group(0)}")
print(f"-- {f} --")
for line_no, line in enumerate(text.splitlines(),1):
if 'global_domain_allocations' in line:
print(f"{line_no}: {line}")
PYRepository: hw-native-sys/simpler
Length of output: 267
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== target file context =="
python3 - <<'PY'
from pathlib import Path
for p in ["src/common/platform_comm/comm_sim.cpp", "src/a2a3/platform/onboard/host/comm_hccl.cpp", "src/common/platform_comm/comm.h"]:
path=Path(p)
print(f"-- {p} --")
lines=path.read_text(errors="ignore").splitlines()
for start,end in [(180,245),(70,125),(1,220)]:
print(f"[[ {p}:{start}-{end} ]]")
for i in range(start-1, min(end, len(lines))):
print(f"{i+1}: {lines[i]}")
PY
echo "== global-domain references =="
python3 - <<'PY'
import re
from pathlib import Path
for p in ["src/common/platform_comm/comm_sim.cpp", "src/a2a3/platform/onboard/host/comm_hccl.cpp", "src/common/platform_comm/comm.h"]:
path=Path(p)
text=path.read_text(errors="ignore")
print(f"-- {p} --")
for needle in ["comm_global_domain_prepare", "comm_global_domain_import", "comm_global_domain_release", "global_domain_allocations", "comm_global_domain_release_fn", "global_domain_ids_", "std::mutex", "pthread_mutex", "thread"]:
print(f"needle={needle}")
line_no=1
for line in text.splitlines():
if needle in line:
print(f"{line_no}: {line}")
line_no += 1
PY
echo "== precise caller scan =="
python3 - <<'PY'
import re, subprocess
files=subprocess.check_output(["git","ls-files"]).decode().splitlines()
for f in files:
if not f.endswith((".cpp", ".c", ".h", ".hpp", ".cc", ".cxx", ".hh", ".hxx")):
continue
text=Path(f).read_text(errors="ignore")
for needle in ["comm_global_domain_prepare", "comm_global_domain_import", "comm_global_domain_release"]:
if needle in text:
print(f"-- {f} -- {needle}")
line_no=1
for line in text.splitlines():
if needle in line:
print(f"{line_no}: {line}")
line_no += 1
PYRepository: hw-native-sys/simpler
Length of output: 50377
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
import re
def lines(file, start, end):
text = file.read_text(errors="ignore").splitlines()
return [(i+1, text[i]) for i in range(start-1, min(end, len(text)))]
def scan_file(file, include_text=False):
name = str(file)
text = file.read_text(errors="ignore")
ids = [m.start() for m in re.finditer(r'\bcomm_global_domain_(prepare|import|release)\b', text)]
print(f"-- caller contexts in {name}: matches={len(ids)} --")
if not ids:
return
for m in re.finditer(r'\bcomm_global_domain_(prepare|import|release)\b', text):
line = text.count('\n', 0, m.start()) + 1
print(f"marker line {line}: {text.splitlines()[line-1].strip()}")
# print a compact 50-line window around every marker without repeating overlaps
used = set()
windows = []
for i in ids:
line = text.count('\n', 0, i) + 1
for j in range(max(1, line-25), min(len(text.splitlines()), line+25)+1):
used.add(j)
for j in sorted(used):
windows.append((j, text.splitlines()[j-1]))
for j, line in windows:
print(f"{j}: {line}")
for p in ["src/common/platform_comm/comm_sim.cpp", "src/a2a3/platform/onboard/host/comm_hccl.cpp"]:
scan_file(Path(p))
# Also print exact function bodies for global-domain functions around their declarations.
for p in ["src/common/platform_comm/comm_sim.cpp", "src/a2a3/platform/onboard/host/comm_hccl.cpp"]:
path = Path(p)
text = path.read_text(errors="ignore").splitlines()
print(f"-- function ranges for global-domain in {p} --")
for needle in ["comm_global_domain_prepare", "comm_global_domain_import", "comm_global_domain_release"]:
idx = next((i for i, line in enumerate(text, 1) if needle == line.split("(", 1)[0].removeprefix("extern \"C\" int ").rstrip()), None)
if idx is None:
print(f"{needle}: not found")
continue
paren = text[idx-1].find("(")
if paren == -1:
# include line; find brace through remaining lines
for e in range(idx, len(text)+1):
if "{" in text[e-1]:
idx = e
break
brace_line = next(((idx, text[idx-1].find("{"))) for i in range(idx, len(text)+1) if "{" in text[i-1])
depth = 0
start_line = brace_line[0]
end_line = len(text)
for i in range(brace_line[0], len(text)+1):
for ch in text[i-1]:
if ch == "{":
depth += 1
elif ch == "}":
depth -= 1
if depth == 0:
end_line = i
break
if depth == 0 and end_line == i:
break
print(f"{needle}: {start_line}-{end_line}")
for ln in range(start_line, end_line + 1):
print(f"{ln}: {text[ln-1]}")
# Scans for global_domain_allocations accesses and mutexes in the two files.
for p in ["src/common/platform_comm/comm_sim.cpp", "src/a2a3/platform/onboard/host/comm_hccl.cpp"]:
path = Path(p)
text = path.read_text(errors="ignore").splitlines()
print(f"-- allocations map and synchronization in {p} --")
for i, line in enumerate(text, 1):
if "global_domain_allocations" in line or "std::mutex" in line or "pthread_mutex" in line or "std::lock_guard" in line or "std::scoped_lock" in line:
print(f"{i}: {line}")
PYRepository: hw-native-sys/simpler
Length of output: 31235
Document or serialize global-domain calls.
comm_global_domain_prepare, comm_global_domain_import, and comm_global_domain_release read, write, and erase global_domain_allocations without synchronization in both sim and HCCL backends. If the caller can drive different domain IDs across threads in the same L2 process, add a mutex or document the one-thread contract.
🤖 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_comm/comm_sim.cpp` at line 219, Protect accesses to
global_domain_allocations in comm_global_domain_prepare,
comm_global_domain_import, and comm_global_domain_release with a mutex
consistently across the sim and HCCL backends, covering all reads, writes, and
erases; alternatively, explicitly document and enforce the required
single-thread caller contract.
Comment on lines
+7
to
+11
| - Local Branch: `skx/mpi-mailbox-broadcast` | ||
| - Delivery Branch: `origin/use-mpi-and-remove-socket` | ||
| - Validation Hosts: servers 37 and 35 (external handoff required) | ||
| - Baseline PR Head: `d3d17b21593e642c6c5f3956ab163589cdb2bbe1` | ||
| - Latest Tested Tree: final local branch tree; final commit SHA is reported at handoff |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟡 Minor | ⚡ Quick win
Remove the personal and internal host identifiers from this record.
Line 7 names the local branch skx/mpi-mailbox-broadcast, which carries a personal handle. Line 45 names the host myserver, and lines 9 and 46 name internal validation hosts "servers 37 and 35". Replace these with generic placeholders.
As per coding guidelines: "Do not include private information in documentation or code, including usernames, absolute paths containing usernames, or other personally identifiable information; use relative paths or generic placeholders instead."
Also applies to: 44-48
🤖 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 `@task.md` around lines 7 - 11, Remove personal and internal identifiers from
the task record: replace the branch name, myserver host, and validation host
references in the affected entries with generic placeholders while preserving
the remaining handoff metadata and structure.
Source: Coding guidelines
Comment on lines
+310
to
+313
| parser.add_argument("--host-37", default="120.9.10.37") | ||
| parser.add_argument("--host-35", default="120.9.10.35") | ||
| parser.add_argument("--roce-37", default="10.30.2.1,10.30.2.2") | ||
| parser.add_argument("--roce-35", default="10.30.0.1,10.30.0.2") |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟡 Minor | ⚡ Quick win
Replace hardcoded internal IP addresses with placeholders. Both the script's CLI defaults and the README's example command embed the same concrete, real-looking internal server and RoCE network addresses. Use generic placeholders instead.
tools/a3_l4_tcp_smoke/mpirun_compute_then_tload_2x2_smoke.py#L310-L313: replace the--host-37/--host-35/--roce-37/--roce-35default values with documentation-reserved placeholder addresses (for example, RFC 5737203.0.113.0/24) or make the arguments required with no default.tools/a3_l4_tcp_smoke/README.md#L17-L23: update the example command to use the same placeholder addresses as the script.
As per coding guidelines, **/*: "Do not include private information in documentation or code, including usernames, absolute paths containing usernames, or other personally identifiable information; use relative paths or generic placeholders instead."
📍 Affects 2 files
tools/a3_l4_tcp_smoke/mpirun_compute_then_tload_2x2_smoke.py#L310-L313(this comment)tools/a3_l4_tcp_smoke/README.md#L17-L23
🤖 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 `@tools/a3_l4_tcp_smoke/mpirun_compute_then_tload_2x2_smoke.py` around lines
310 - 313, Replace the concrete network defaults in
tools/a3_l4_tcp_smoke/mpirun_compute_then_tload_2x2_smoke.py lines 310-313 for
the --host-37, --host-35, --roce-37, and --roce-35 arguments with
documentation-reserved placeholder addresses, or make them required without
defaults. Update the example command in tools/a3_l4_tcp_smoke/README.md lines
17-23 to use the same placeholders.
Source: Coding guidelines
sunkaixuan2018
force-pushed
the
use-mpi-and-remove-socket
branch
3 times, most recently
from
August 1, 2026 12:11
8c0a0db to
8e8a7c4
Compare
- Route MPI group task and control traffic through a named rank-0 mailbox - Distribute per-rank payloads and collect ranked results with MPI collectives - Preserve the TCP transport for ordinary non-MPI Remote L3 workers - Add protocol, lifecycle, timeout, compatibility, and smoke coverage
sunkaixuan2018
force-pushed
the
use-mpi-and-remove-socket
branch
from
August 1, 2026 12:22
8e8a7c4 to
a348313
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
mpirunprocess-group cleanupDependency
This PR is stacked on #1623 and uses its latest head as the implementation baseline. It does not rewrite or modify the
add-mpi-runbranch. Until #1623 is merged, GitHub may show both the dependency commit and this PR's mailbox commit.Testing
mpirunintegration on server 37The hardware and MPI integration items remain explicitly unvalidated until the server-37 agent returns logs.