Add: distributed worker runtime, group task, and fork+shm chip isolation

[hw-native-sys-bot]

Summary

• Add distributed scheduling engine (src/common/distributed/): TensorMap dependency tracking, ring-buffer back-pressure, scope lifetime, Orchestrator/Scheduler/WorkerThread model
• Add group task: submit N args for N workers on 1 DAG node, completion aggregation via sub_complete_count
• Fork+shm ChipWorker process isolation (DistChipProcess): each chip runs in its own forked process with isolated address space, eliminating sim global-state crashes when multiple chips execute concurrently
• Python scope context manager (with hw.scope():) replaces scope_begin()/scope_end()
• DistSubWorker: fork/shm mailbox for GIL-free Python callable execution
• DeviceRunner changed to thread_local for multi-ChipWorker thread safety
• ChipWorker implements IWorker for uniform scheduling interface
• Python bindings (nanobind) and Worker/HostWorker wrappers
• Move tests/ut/*.py to tests/ut/py/ for consistent test layout
• docs/distributed_level_runtime.md: level model, scheduling engine, process/thread model, unified API
• docs/sim_multi_device_isolation.md: root cause analysis of sim multi-device crashes and fix

Testing

• C++ unit tests: 5 suites pass (tensormap, ring, scope, orchestrator, scheduler incl. 3 group tests)
• Python unit tests: 146 tests pass
• ST case 1: L2 single chip — correct output (47.0)
• ST case 2: L3 ChipTask → SubTask — TensorMap dependency
• ST case 3: L3 group (2 real ChipWorkers) → SubTask — 2 chips concurrent, no crash, correct output
• Hardware tests (if applicable)

[gemini-code-assist]

Code Review

This pull request implements a distributed runtime for L3 orchestration, introducing core C++ components such as the Orchestrator, Scheduler, and TensorMap, along with their corresponding Python bindings. It enables multi-process sub-worker management via fork and shared memory and updates the ChipWorker to support the new scheduling engine. Feedback highlights critical concurrency issues, specifically race conditions in task dependency tracking and data races within the DistTensorMap. Additionally, improvements are suggested for the DistRing retirement logic and the efficiency of the scheduler and sub-worker loops to reduce CPU overhead and latency.

[gemini-code-assist]

In submit(), producers in the COMPLETED state are skipped when adding dependencies. This creates a critical race condition: a producer can transition from COMPLETED to CONSUMED (and have its buffers freed) after the Orchestrator has looked it up but before the consumer task is even dispatched. This leads to use-after-free when the consumer eventually runs. Dependencies must be added for any producer that is not yet CONSUMED, regardless of whether it has already finished execution.

Suggested change

	TaskState ps_state = ps.state.load(std::memory_order_acquire);
	if (ps_state == TaskState::COMPLETED || ps_state == TaskState::CONSUMED) {
	// Producer already finished; no dependency needed.
	continue;
	}
	TaskState ps_state = ps.state.load(std::memory_order_acquire);
	if (ps_state == TaskState::CONSUMED) {
	// Producer already finished and released; no dependency needed.
	continue;
	}

[gemini-code-assist]

The consumption logic in release_ref is incorrect and dangerous. First, it uses released >= total as a threshold, which ignores the implicit 'self' reference (the task must complete itself before being consumed), potentially leading to premature release if scope_end is called before the task finishes. Second, it allows a task to be consumed while in the RUNNING state. If a task is consumed while running, its output buffers are deleted (via DistTaskSlotState::reset), causing a use-after-free in the worker thread. Consumption should only occur when the task is COMPLETED and all references (consumers + scope) are released.

Suggested change

	TaskState cur = s.state.load(std::memory_order_acquire);
	if (released >= total && (cur == TaskState::COMPLETED || cur == TaskState::RUNNING)) {
	on_consumed(slot);
	}
	TaskState cur = s.state.load(std::memory_order_acquire);
	if (released >= total + 1 && cur == TaskState::COMPLETED) {
	on_consumed(slot);
	}

[gemini-code-assist]

The release logic in DistRing is fundamentally broken for out-of-order task completion. It uses a monotonic last_alive_ counter and advances it to the task_id of the released slot even if older tasks are still active. This causes alloc() to incorrectly believe that older slots are available for reuse, leading to slot collisions and use-after-free bugs when tasks complete out of order (which is expected in a multi-worker scheduler). A robust implementation should use a free-list (e.g., a thread-safe queue of available slots) instead of a sliding window with monotonic retirement if out-of-order release is required.

[gemini-code-assist]

The documentation states that DistTensorMap is owned exclusively by the Orchestrator and requires no locking. However, DistWorker::on_consumed (called from the Scheduler thread) invokes orchestrator_.on_consumed, which calls tensormap_->erase_task_outputs. This results in concurrent access to the underlying std::unordered_map from both the Orchestrator and Scheduler threads, leading to data races and undefined behavior. A mutex must be added to protect DistTensorMap operations.

[gemini-code-assist]

The scheduler loop uses wait_for with a hardcoded 1ms timeout to check for new tasks in the ready_queue. This introduces unnecessary latency (up to 1ms per task submission when the scheduler is idle) and causes the scheduler thread to wake up frequently even when there is no work. The scheduler should wait on a condition variable that is notified by both the Orchestrator (when a task is pushed to ready_queue) and the WorkerThreads (when a task completes).

[gemini-code-assist]

The _sub_worker_loop in the forked child process is a tight spin loop that polls the shared memory mailbox without any yielding or sleeping. While this minimizes latency, it will consume 100% of a CPU core per sub-worker even when idle. Given that these are host-side Python workers, adding a small time.sleep(0) or a micro-sleep when the state is _IDLE would significantly reduce CPU waste without a major impact on orchestration latency.

[gemini-code-assist]

When all workers of a specific type are busy, dispatch_ready() pops a task from the ready_queue and immediately pushes it back to the end of the queue. This results in a busy-spin (cycling through the queue every 1ms) when the system is under load. Instead of popping and re-pushing, the scheduler should only pop a task when a compatible worker is known to be idle, or it should wait for a 'worker idle' signal to avoid redundant queue operations.