Add: dispatch MPI worker groups through mailbox

docs/mpi-l3-mailbox.md

@@ -0,0 +1,132 @@
+# MPI L3 group mailbox protocol
+
+An MPI L3 group has one L4-owned, named shared-memory mailbox. Only local MPI
+rank 0 opens that mailbox. All ranks participate in the same ordered
+`dispatch_comm` collectives, while Global CommDomain descriptor exchange uses
+a separate `domain_comm`.
+
+```text
+L4 / MpiGroupMailboxEndpoint
+          |
+          | named SharedMemory (one request lane)
+          v
+local MPI rank 0 / L3
+          |
+          | dispatch_comm Bcast + Gather
+          v
+all MPI ranks / L3 -> each rank's local L2 workers
+```
+
+MPI groups never start or connect Simpler command/health TCP sockets. Ordinary
+`RemoteWorkerSpec` workers still use `RemoteL3SocketTransport` and keep their
+existing command and health lanes.
+
+## Startup and shutdown
+
+1. L4 creates the mailbox and writes its name, protocol version, size, and
+   world size into the group manifest.
+2. L4 starts `mpirun` as a new process group and monitors that direct child.
+3. Each rank creates and initializes its own L3 `Worker`.
+4. All ranks complete a readiness `allgather`.
+5. Rank 0 reopens the mailbox by name and publishes `READY`. Other ranks never
+   map it.
+6. L4 attaches every stable MPI worker id to the same
+   `MpiGroupMailboxChannel`; it does not create `RemoteL3Endpoint` sockets.
+7. Shutdown is a mailbox `SHUTDOWN` request, followed by one MPI broadcast.
+   Each rank closes its inner worker, communicators are freed, `mpirun` exits,
+   and L4 unlinks the mailbox and manifest directory.
+
+If startup, a collective, the mailbox, or `mpirun` fails, the group becomes
+terminal. Runtime timeout also kills the complete `mpirun` process group.
+There is no TCP fallback.
+
+## Envelope and state
+
+Protocol version 1 has a fixed 256-byte header and two 16 MiB payload regions.
+The header contains:
+
+- magic `SMPIBOX\0`
+- protocol version and layout size
+- MPI world size
+- group state: `INITIALIZING`, `READY`, `TERMINAL`, or `CLOSED`
+- request state: `IDLE`, `REQUEST_READY`, `TASK_ACCEPTED`, `TASK_DONE`,
+  `TASK_FAILED`, `SHUTDOWN_READY`, or `SHUTDOWN_DONE`
+- monotonic mailbox `sequence_id`
+- opcode: `TASK`, `CONTROL`, `PING`, or `SHUTDOWN`
+- target: `GROUP`, `RANK`, or `PER_RANK`
+- target rank, payload count, and byte lengths
+
+Rank 0 copies the complete request to private memory before publishing
+`TASK_ACCEPTED`. It publishes `TASK_DONE` only after gathering every rank's
+status. Duplicate or decreasing sequence ids make the group terminal.
+
+Every gathered error contains `rank`, `error_type`, and `message`. Any target
+rank failure fails the group operation. A broken command processor or
+collective is terminal; an ordinary task/control application error is returned
+to L4 and the communicator may be reused.
+
+## Target and API semantics
+
+- `orch.submit_next_level(..., worker=id)` remains a directed rank operation.
+  Every MPI rank receives the envelope in collective order, but only the
+  selected rank executes it.
+- `orch.submit_next_level_group(args_list, workers=...)` remains one DAG node.
+  When `workers` is the complete MPI group, C++ batches all members into one
+  `PER_RANK` mailbox request. Rank `workers[i]` uses `args_list[i]`.
+- A subset group remains supported as ordered directed requests. It is not
+  silently widened to the complete MPI group.
+- Group-wide controls use one `GROUP` mailbox request.
+
+The existing remote task codec is reused. It serializes scalar values, tensor
+metadata, inline host payloads, and `RemoteTensorRef` descriptors. Bare host or
+child virtual addresses without a valid remote sidecar are rejected before
+execution; a pointer value is never forwarded as if it were meaningful on
+another rank. `PYTHON_SERIALIZED` callable payloads remain unsupported by the
+underlying Remote L3 protocol; `PYTHON_IMPORT` and inline `CHIP_CALLABLE`
+registration are supported.
+
+## Remote protocol audit and MPI mapping
+
+The wire `FrameType` values remain unchanged:
+
+| Existing frame | MPI mailbox mapping |
+| -------------- | ------------------- |
+| `HELLO` / ready | rank-local initialization, readiness `allgather`, then rank 0 publishes mailbox `READY` |
+| `TASK` | `TASK`; directed `RANK`, or one full-group `PER_RANK` vector |
+| `CONTROL` / `CONTROL_REPLY` | `CONTROL`; directed except the group-wide controls below |
+| `COMPLETION` | gathered per-rank status; selected/per-rank replies returned to L4 |
+| `HEALTH` | `PING` to `GROUP`, gathered before success |
+| `SHUTDOWN` | `SHUTDOWN` to `GROUP`, gathered before `SHUTDOWN_DONE` |
+
+All existing remote controls use the mailbox path:
+
+| Number | Control | MPI target |
+| -----: | ------- | ---------- |
+| 1 | `UNREGISTER_CALLABLE` | directed rank |
+| 2 | `PREPARE_REGISTER_CALLABLE` | directed rank |
+| 3 | `COMMIT_REGISTER_CALLABLE` | directed rank |
+| 4 | `ABORT_REGISTER_CALLABLE` | directed rank |
+| 5 | `PREPARE_CALLABLE` | directed rank |
+| 6 | `ALLOC_REMOTE_BUFFER` | directed rank |
+| 7 | `FREE_REMOTE_BUFFER` | directed rank |
+| 8 | `COPY_TO_REMOTE` | directed rank |
+| 9 | `COPY_FROM_REMOTE` | directed rank |
+| 10 | `EXPORT_BUFFER` | directed rank |
+| 11 | `IMPORT_BUFFER` | directed rank |
+| 12 | `RELEASE_IMPORT` | directed rank |
+| 13 | `COMM_INIT` | directed rank |
+| 14 | `ALLOC_DOMAIN` prepare/import/commit/abort | one group request; descriptor work uses `domain_comm` |
+| 15 | `RELEASE_DOMAIN` | one group request |
+| 16 | `COPY_TO_DOMAIN` | directed rank |
+| 17 | `COPY_FROM_DOMAIN` | directed rank |
+
+Remote control number 18 is intentionally not assigned. The local hierarchical
+protocol keeps number 18 for committed-device-memory control.
+
+## Threading
+
+Only the main dispatcher thread calls MPI. The existing command processor runs
+on a rank-local thread over an in-memory, socket-shaped queue. Global
+CommDomain operations cross back to the dispatcher through a queue and
+`threading.Event`, so they use `domain_comm` on the MPI-owning thread. This
+design does not require `MPI_THREAD_MULTIPLE`.

docs/remote-l3-worker-design.md

@@ -9,6 +9,7 @@ Detailed protocol, buffer, transport, and rollout notes live in:
 
 - [protocol.md](remote-l3-worker-design/protocol.md)
 - [buffers-and-transports.md](remote-l3-worker-design/buffers-and-transports.md)
+- [MPI L3 group mailbox](mpi-l3-mailbox.md)
 - [implementation-plan.md](remote-l3-worker-design/implementation-plan.md)
 - [pr-split-and-audit-plan.md][split-audit-plan]
 
@@ -63,6 +64,10 @@ Implemented:
 - Socket-backed simulation remote sessions via `simpler-remote-worker` and
   `simpler-remote-l3-session`, including `HELLO READY`, TASK/COMPLETION,
   CONTROL/CONTROL_REPLY, SHUTDOWN, and an independent health lane.
+- MPI L3 groups use one rank-0 named shared-memory mailbox plus ordered MPI
+  collectives for task, control, health, Global CommDomain, error, and shutdown
+  handling. They do not create Simpler command or health TCP sockets; ordinary
+  non-MPI Remote L3 sessions retain the socket transport.
 - Simulation remote buffer allocation, copy, export, import, release-import,
   imported-handle scheduling eligibility, and deferred owner free.
 - Registry-scope-aware remote callable manifest/control install for dispatcher

python/bindings/worker_bind.h

@@ -445,6 +445,20 @@ inline void bind_worker(nb::module_ &m) {
             nb::arg("health_host"), nb::arg("health_port"), nb::arg("attach_timeout_s") = 30.0,
             nb::arg("runtime_timeout_s") = 30.0, "Register a REMOTE_L3 endpoint after the session reports HELLO READY."
         )
+        .def(
+            "add_mpi_group_mailbox",
+            [](Worker &self, const std::vector<int32_t> &worker_ids, const std::vector<uint64_t> &session_ids,
+               uint64_t mailbox_ptr, size_t mailbox_bytes, int mpirun_pid, double runtime_timeout_s) {
+                nb::gil_scoped_release release;
+                self.add_mpi_group_mailbox(
+                    worker_ids, session_ids, reinterpret_cast<void *>(mailbox_ptr), mailbox_bytes, mpirun_pid,
+                    runtime_timeout_s
+                );
+            },
+            nb::arg("worker_ids"), nb::arg("session_ids"), nb::arg("mailbox_ptr"), nb::arg("mailbox_bytes"),
+            nb::arg("mpirun_pid"), nb::arg("runtime_timeout_s") = 30.0,
+            "Register one shared-memory MPI group endpoint for each worker id."
+        )
 
         // Release the GIL while starting the Scheduler thread so another Python
         // thread can run during it — e.g. a concurrent close() observing

python/simpler/global_comm_smoke.py

@@ -0,0 +1,167 @@
+# Copyright (c) PyPTO Contributors.
+# This program is free software, you can redistribute it and/or modify it under the terms and conditions of
+# CANN Open Software License Agreement Version 2.0 (the "License").
+# Please refer to the License for details. You may not use this file except in compliance with the License.
+# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
+# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
+# See LICENSE in the root of the software repository for the full text of the License.
+# -----------------------------------------------------------------------------------------------------------
+"""Rank-local L3 callbacks used by the A3 MPI compute + TLOAD smoke."""
+
+from __future__ import annotations
+
+from .task_interface import CallConfig, DataType, TaskArgs, Tensor, TensorArgType
+
+_SMOKE_COUNT = 256
+
+
+def _digest_from_scalars(args: TaskArgs, start: int) -> bytes:
+    return b"".join(int(args.scalar(start + i)).to_bytes(8, "little") for i in range(4))
+
+
+def remote_compute_orch(orch, args: TaskArgs, cfg: CallConfig) -> None:
+    """Submit one local vector-add task from an MPI L3 rank to its selected L2."""
+    from .remote_l3_session import get_inner_handle  # noqa: PLC0415
+
+    if args.scalar_count() != 6:
+        raise ValueError("remote compute task expects domain_id, local_worker_id, and four digest scalars")
+    domain_id = int(args.scalar(0))
+    local_worker_id = int(args.scalar(1))
+    chip_handle = get_inner_handle(_digest_from_scalars(args, 2).hex())
+    context = orch.get_global_domain(domain_id)[local_worker_id]
+
+    chip_args = TaskArgs()
+    for buffer_name in ("lhs", "rhs"):
+        chip_args.add_tensor(
+            Tensor.make(
+                data=context.buffer_ptrs[buffer_name],
+                shapes=(_SMOKE_COUNT,),
+                dtype=DataType.FLOAT32,
+                child_memory=True,
+            ),
+            TensorArgType.INPUT,
+        )
+    chip_args.add_tensor(
+        Tensor.make(
+            data=context.buffer_ptrs["input"],
+            shapes=(_SMOKE_COUNT,),
+            dtype=DataType.FLOAT32,
+            child_memory=True,
+        ),
+        TensorArgType.OUTPUT_EXISTING,
+    )
+    orch.submit_next_level(chip_handle, chip_args, cfg, worker=local_worker_id)
+
+
+def remote_rank_orch(orch, args: TaskArgs, cfg: CallConfig) -> None:
+    """Submit the peer-TLOAD kernel from an MPI L3 rank to its selected L2."""
+    from .remote_l3_session import get_inner_handle  # noqa: PLC0415
+
+    if args.scalar_count() != 6:
+        raise ValueError("global TLOAD remote task expects domain_id, local_worker_id, and four digest scalars")
+    domain_id = int(args.scalar(0))
+    local_worker_id = int(args.scalar(1))
+    chip_handle = get_inner_handle(_digest_from_scalars(args, 2).hex())
+    context = orch.get_global_domain(domain_id)[local_worker_id]
+
+    chip_args = TaskArgs()
+    chip_args.add_tensor(
+        Tensor.make(
+            data=context.buffer_ptrs["input"],
+            shapes=(_SMOKE_COUNT,),
+            dtype=DataType.FLOAT32,
+            child_memory=True,
+        ),
+        TensorArgType.INPUT,
+    )
+    chip_args.add_tensor(
+        Tensor.make(
+            data=context.buffer_ptrs["result"],
+            shapes=(_SMOKE_COUNT,),
+            dtype=DataType.FLOAT32,
+            child_memory=True,
+        ),
+        TensorArgType.OUTPUT_EXISTING,
+    )
+    chip_args.add_scalar(context.domain_size)
+    chip_args.add_scalar(context.device_ctx)
+    orch.submit_next_level(chip_handle, chip_args, cfg, worker=local_worker_id)
+
+
+def remote_compute_group_orch(orch, args: TaskArgs, cfg: CallConfig) -> None:
+    """Drive every L2 owned by one MPI L3 rank with distinct domain buffers."""
+    from .remote_l3_session import get_inner_handle  # noqa: PLC0415
+
+    if args.scalar_count() != 6:
+        raise ValueError("MPI compute group task expects domain_id, local worker count, and four digest scalars")
+    domain_id = int(args.scalar(0))
+    local_worker_count = int(args.scalar(1))
+    chip_handle = get_inner_handle(_digest_from_scalars(args, 2).hex())
+    domain = orch.get_global_domain(domain_id)
+    group_args = []
+    workers = []
+    for local_worker_id in range(local_worker_count):
+        context = domain[local_worker_id]
+        chip_args = TaskArgs()
+        for buffer_name in ("lhs", "rhs"):
+            chip_args.add_tensor(
+                Tensor.make(
+                    data=context.buffer_ptrs[buffer_name],
+                    shapes=(_SMOKE_COUNT,),
+                    dtype=DataType.FLOAT32,
+                    child_memory=True,
+                ),
+                TensorArgType.INPUT,
+            )
+        chip_args.add_tensor(
+            Tensor.make(
+                data=context.buffer_ptrs["input"],
+                shapes=(_SMOKE_COUNT,),
+                dtype=DataType.FLOAT32,
+                child_memory=True,
+            ),
+            TensorArgType.OUTPUT_EXISTING,
+        )
+        group_args.append(chip_args)
+        workers.append(local_worker_id)
+    orch.submit_next_level_group(chip_handle, group_args, cfg, workers=workers)
+
+
+def remote_rank_group_orch(orch, args: TaskArgs, cfg: CallConfig) -> None:
+    """Drive peer TLOAD on every L2 owned by one MPI L3 rank."""
+    from .remote_l3_session import get_inner_handle  # noqa: PLC0415
+
+    if args.scalar_count() != 6:
+        raise ValueError("MPI TLOAD group task expects domain_id, local worker count, and four digest scalars")
+    domain_id = int(args.scalar(0))
+    local_worker_count = int(args.scalar(1))
+    chip_handle = get_inner_handle(_digest_from_scalars(args, 2).hex())
+    domain = orch.get_global_domain(domain_id)
+    group_args = []
+    workers = []
+    for local_worker_id in range(local_worker_count):
+        context = domain[local_worker_id]
+        chip_args = TaskArgs()
+        chip_args.add_tensor(
+            Tensor.make(
+                data=context.buffer_ptrs["input"],
+                shapes=(_SMOKE_COUNT,),
+                dtype=DataType.FLOAT32,
+                child_memory=True,
+            ),
+            TensorArgType.INPUT,
+        )
+        chip_args.add_tensor(
+            Tensor.make(
+                data=context.buffer_ptrs["result"],
+                shapes=(_SMOKE_COUNT,),
+                dtype=DataType.FLOAT32,
+                child_memory=True,
+            ),
+            TensorArgType.OUTPUT_EXISTING,
+        )
+        chip_args.add_scalar(context.domain_size)
+        chip_args.add_scalar(context.device_ctx)
+        group_args.append(chip_args)
+        workers.append(local_worker_id)
+    orch.submit_next_level_group(chip_handle, group_args, cfg, workers=workers)