diff --git a/docs/hierarchical_level_runtime.md b/docs/hierarchical_level_runtime.md index bc361ad970..0fc3bda7fa 100644 --- a/docs/hierarchical_level_runtime.md +++ b/docs/hierarchical_level_runtime.md @@ -177,7 +177,7 @@ Communication channels: A higher-level `Worker` can register a lower-level `Worker` as a NEXT_LEVEL child through the same mailbox protocol L3 uses for chip children. The Python `Worker.add_worker(child)` stores an un-init'd child -Worker; on first `run()`, the parent forks a child process that inits the +Worker; on `init()`, the parent forks a child process that inits the inner Worker and enters a mailbox-polling loop (`_child_worker_loop`). ```python @@ -214,7 +214,7 @@ Worker opens its own scope, executes the orch function with its own — recursion is symmetric. **Nested fork ordering**: L3's own children (sub/chip) are forked **inside** -the L4 child process, on L3's first `run()`. This keeps the process tree +the L4 child process, during L3's eager `init()`. This keeps the process tree clean: L4 parent → L3 child → L3's sub/chip grandchildren. **Mode per level is independent**: L3 can use PROCESS (chip children), while diff --git a/docs/task-flow.md b/docs/task-flow.md index 4c16a36350..8ea0c8a40e 100644 --- a/docs/task-flow.md +++ b/docs/task-flow.md @@ -460,13 +460,13 @@ that maps to a Python orchestration function, not a `ChipCallable`. ### Fork sequence -L4's `init()` allocates the L4 Worker's HeapRing (before fork). -On first `run()`, the deferred `_start_hierarchical()`: +L4's `init()` allocates the L4 Worker's HeapRing (before fork), then +`_start_hierarchical()` runs eagerly in the same `init()`: 1. Forks one child process per L3 Worker child 2. **Inside the child**: `inner_worker.init()` creates the L3 Worker - (mmaps L3's own HeapRing), allocates L3's sub/chip mailboxes. L3's - own children are forked lazily on L3's first `run()`. + (mmaps L3's own HeapRing), allocates L3's sub/chip mailboxes, and — + since init is eager — forks L3's own sub/chip children in turn. 3. Child enters `_child_worker_loop(mailbox, registry, inner_worker)` 4. **Parent**: registers each mailbox with L4's Worker via `add_next_level_worker_at(worker_id, mailbox_addr)` @@ -476,11 +476,11 @@ L4 parent process ├─ Worker(4) + HeapRing (MAP_SHARED, inherited by L3 child) └─ fork ──────────────────► L3 child process ├─ inner_worker.init() - │ └─ Worker(3) + L3's own HeapRing + │ ├─ Worker(3) + L3's own HeapRing + │ └─ _start_hierarchical() forks L3's sub children └─ _child_worker_loop(mbox, registry, inner_worker) - └─ on first dispatch: + └─ on dispatch: inner_worker.run(orch_fn, args, cfg) - └─ _start_hierarchical() forks L3's sub children ``` ### Dispatch walkthrough diff --git a/docs/worker-manager.md b/docs/worker-manager.md index 80bc16aa91..36dfafdfcb 100644 --- a/docs/worker-manager.md +++ b/docs/worker-manager.md @@ -263,12 +263,11 @@ When an L4 Worker has L3 Worker children, the fork sequence nests: ```text L4 parent process ├─ _init_hierarchical(): Worker(4) + HeapRing mmap (before fork) - └─ _start_hierarchical() (on first run): + └─ _start_hierarchical() (eager, in init()): ├─ fork L3 child ────────► L3 child process: │ inner_worker.init() ← Worker(3) + L3 HeapRing + │ └─ _start_hierarchical() forks L3's sub/chip children │ _child_worker_loop() - │ └─ on first dispatch: inner_worker.run() - │ └─ _start_hierarchical() forks L3's sub/chip children └─ register mailbox with L4's Worker ``` diff --git a/python/simpler/worker.py b/python/simpler/worker.py index 21e268a44f..960dc17107 100644 --- a/python/simpler/worker.py +++ b/python/simpler/worker.py @@ -233,6 +233,10 @@ def my_l4_orch(orch, args, config): # mapping (the pure-Python blob-rewrite scheme, no runtime C++ change). _BLOB_TENSOR_STRIDE = 128 _BLOB_HEADER_BYTES = 8 +# Byte offset of Tensor.child_memory within the 128 B Tensor (static_assert in +# src/common/task_interface/tensor.h pins it to 43): 1 ⇒ chip-owned device +# pointer (never a host buffer, so never blob-rewritten). +_BLOB_TENSOR_CHILD_MEMORY_OFF = 43 # Layout of the CTRL_COMM_INIT request shm. _COMM_INIT_HEADER = struct.Struct(" None: raise RuntimeError("remote memory APIs require Worker.init() before allocation or copy") if int(worker_id) not in set(self._remote_worker_ids): raise ValueError("remote memory APIs require a remote worker id returned by add_remote_worker") - self._start_hierarchical() if self._worker is None: raise RuntimeError("remote memory APIs require a started hierarchical Worker") @@ -2381,7 +2395,6 @@ def _post_start_register_remote( # noqa: PLR0912 -- two-phase remote register/c handle, _is_new = self._install_registration_locked(reg) return handle - self._start_hierarchical() if self._worker is None: raise RuntimeError("Worker.register(RemoteCallable): hierarchical worker is not started") @@ -2910,7 +2923,6 @@ def _unregister_remote_handle(self, handle: CallableHandle) -> None: errors: list[str] = [] try: if self._initialized: - self._start_hierarchical() assert self._worker is not None for worker_id in worker_ids: try: @@ -3060,17 +3072,16 @@ def init(self, prewarm_config=None) -> None: (``runtime_env.ring_task_window`` / ``ring_heap`` / ``ring_dep_pool``) is built + cached so the first ``run`` with the same sizing skips the (~800ms) cold prebuilt runtime-arena build. - Timing is level-dependent: an L2 worker prewarms here in - ``init``; an L3+ worker has no chip children until the hierarchy - starts on the first ``run``, so it prewarms there — during - hierarchy startup, alongside the callable SO upload and before the - first task dispatches. A no-op for runtimes without a prebuilt - arena (host_build_graph). ``None`` (default) disables prewarm. + Both L2 and L3+ workers prewarm here in ``init``: an L3+ worker + forks its chip children and starts the hierarchy in ``init`` + too, prewarming during that startup — alongside the callable SO + upload and before the first task dispatches. A no-op for runtimes + without a prebuilt arena (host_build_graph). ``None`` (default) + disables prewarm. """ if self._initialized: raise RuntimeError("Worker already initialized") - # Validate up front so a bad config fails here regardless of level, even - # though an L3+ worker does not build the arena until the first run. + # Validate up front so a bad config fails here before any fork. if prewarm_config is not None: prewarm_config.validate() self._prewarm_config = prewarm_config @@ -3080,6 +3091,11 @@ def init(self, prewarm_config=None) -> None: self._init_level2() elif self.level >= 3: self._init_hierarchical() + # Fork the chip/sub children and start the C++ scheduler now, + # so the whole L2 hierarchy is live (children forked, + # ChipWorker.init done, callables uploaded H2D) by the time + # init() returns. + self._start_hierarchical() else: raise ValueError(f"Worker: level {self.level} not supported") except BaseException: @@ -3195,7 +3211,12 @@ def _init_hierarchical(self) -> None: self._hierarchical_started = False def _start_hierarchical(self) -> None: # noqa: PLR0912 -- three parallel fork loops (sub/chip/next) + bootstrap wait + scheduler register/init; branches track the fork order documented in the body - """Fork child processes and start C++ scheduler. Called on first run().""" + """Fork child processes and start C++ scheduler. + + Called once from init() for L3+ workers. The "starting"/"started" + state machine still guards against a concurrent register/unregister + thread racing an in-progress startup. + """ device_ids = self._config.get("device_ids", []) n_sub = self._config.get("num_sub_workers", 0) @@ -3283,10 +3304,10 @@ def _start_hierarchical(self) -> None: # noqa: PLR0912 -- three parallel fork l # Fork next-level Worker children (L4+ with Worker children). # Each child process: init the inner Worker (which mmaps its own - # HeapRing and allocates its own child mailboxes), then enter - # _child_worker_loop. The inner Worker's own children are forked - # lazily on first run() inside _child_worker_loop, so the process - # tree nests correctly: L4 → L3 child → L3's chip/sub children. + # HeapRing, allocates its own child mailboxes, and — since init is + # eager — forks its own chip/sub children in turn), then enter + # _child_worker_loop. The process tree nests correctly: + # L4 → L3 child → L3's chip/sub children. for idx, inner_worker in enumerate(self._next_level_workers): pid = os.fork() if pid == 0: @@ -3365,9 +3386,10 @@ def _abort_hierarchical(self) -> None: """Tear down all forked children + shms after a bootstrap failure. Best-effort: SIGKILL every child we spawned, reap them, then close - and unlink every mailbox. Called only from the init() failure path, - so `dw.init()` has not run and the C++ scheduler is not holding any - mailbox references. + and unlink every mailbox. Called from the init() failure path via + _cleanup_partial_init, which closes the C++ _Worker (stopping the + scheduler + WorkerThreads) *before* this runs, so no C++ thread holds + a mailbox reference when the children are killed. """ pids = list(self._chip_pids) + list(self._sub_pids) + list(self._next_level_pids) for pid in pids: @@ -4030,10 +4052,10 @@ def create_host_buffer(self, nbytes: int) -> HostBuffer: """Allocate a born-shared host buffer, attached into every chip child, that a later ``run()`` reads/writes with **no per-run copy**. - L3 chip children are forked lazily on the first ``run()``; memory created - afterwards is not in their address space. This hands you memory that is - *born* in a shm already attached into every child, so there is nothing to - copy: the child reads and writes the same physical pages the parent sees. + L3 chip children are forked during ``init()``; memory created afterwards + is not in their address space. This hands you memory that is *born* in a + shm already attached into every child, so there is nothing to copy: the + child reads and writes the same physical pages the parent sees. Returns a :class:`HostBuffer` whose ``buffer`` is a ``memoryview`` over that shm. Build a tensor over it with the buffer protocol, framework of @@ -4055,7 +4077,6 @@ def create_host_buffer(self, nbytes: int) -> HostBuffer: raise TypeError("create_host_buffer requires a level >= 3 Worker") if not self._initialized: raise RuntimeError("create_host_buffer requires Worker.init() before allocation") - self._start_hierarchical() if not self._chip_shms: raise RuntimeError("create_host_buffer requires forked chip children (none are configured)") assert self._worker is not None @@ -4339,7 +4360,6 @@ def run(self, callable, args=None, config=None) -> None: self._chip_worker._run_slot(state.slot_id, args, cfg) return None - self._start_hierarchical() assert self._orch is not None assert self._worker is not None # Drop any error stashed by a previous run() so this call starts diff --git a/simpler_setup/scene_test.py b/simpler_setup/scene_test.py index a99e02a1d4..8339c5d3df 100644 --- a/simpler_setup/scene_test.py +++ b/simpler_setup/scene_test.py @@ -306,6 +306,65 @@ def _build_l3_task_args(test_args: TaskArgsBuilder, orch_signature: list): return chip_args, output_names +def _rehost_task_tensors_on_chip_buffers(worker, test_args: TaskArgsBuilder): + """Back each tensor arg with its own ``Worker.create_host_buffer`` region. + + An L3 ``Worker.init()`` forks its chip children, so tensors allocated by + ``generate_args`` afterwards (raw ``share_memory_()``) are not in a child's + address space — dispatching their VA segfaults the child. ``create_host_buffer`` + hands back born-shared memory already attached into every chip child, so a + tensor built over it round-trips with no per-run copy. Each tensor is copied + once into its own such buffer and the builder is mutated in place to point at + it. + + Each tensor gets a **separate** buffer, never one packed region: the + Orchestrator infers task dependencies from tensor address ranges, so distinct + args must occupy distinct, non-adjacent allocations — packing them together + aliases unrelated tensors into one range and corrupts the dependency wiring. + + Returns the ``HostBuffer`` handles to free after the run (via + :func:`_free_chip_host_buffers`). No-op — returns ``[]`` — when the worker has + no chip children (sub-only L3); sub-worker visibility is a separate mechanism. + """ + import torch # noqa: PLC0415 + + if not getattr(worker, "_chip_shms", None): + return [] + handles = [] + for idx, spec in enumerate(test_args.specs): + if not isinstance(spec, Tensor) or not isinstance(spec.value, torch.Tensor): + continue + src = spec.value + if src.numel() == 0: + continue + buf = worker.create_host_buffer(src.numel() * src.element_size()) + hosted = torch.frombuffer(buf.buffer, dtype=src.dtype, count=src.numel()).reshape(src.shape) + hosted.copy_(src) + test_args._specs[idx] = Tensor(spec.name, hosted) + test_args._data[spec.name] = hosted + handles.append(buf) + return handles + + +def _free_chip_host_buffers(worker, test_args: TaskArgsBuilder, handles): + """Drop the host-backed tensor views, then free their buffers. + + A live view keeps the shm pages exported, so the tensors must be released + before ``free_host_buffer`` (which otherwise only warns). Safe to call with an + empty ``handles`` list. + """ + if not handles: + return + import gc # noqa: PLC0415 + + for name in test_args.tensor_names(): + test_args._data[name] = None + test_args._specs = [s for s in test_args._specs if not isinstance(s, Tensor)] + gc.collect() + for buf in handles: + worker.free_host_buffer(buf) + + @contextmanager def _temporary_env(env_updates): """Temporarily set environment variables.""" @@ -1102,58 +1161,65 @@ def _run_and_validate_l3( # noqa: PLR0913 -- threads CLI diagnostic flags + L3 # Build args test_args = self.generate_args(params) + # Worker.init() has already forked the chip children, so tensors built + # here must live in born-shared memory to be visible to them. Rehost onto + # per-tensor create_host_buffer regions; freed in the finally below. + host_bufs = _rehost_task_tensors_on_chip_buffers(worker, test_args) - # Compute golden (unless skip_golden) - golden_args = None - if not skip_golden: - golden_args = test_args.clone() - self.compute_golden(golden_args, params) - - # Save initial tensor values for reset between rounds - all_tensor_names = test_args.tensor_names() - initial_tensors = {} - if rounds > 1: - for name in all_tensor_names: - initial_tensors[name] = getattr(test_args, name).clone() - - # Build CallableNamespace: compiled ChipCallables + sub callable IDs - ns = CallableNamespace({**compiled_callables, **sub_handles}) - - # Get orch function (plain function from CALLABLE) - orch_fn = self.CALLABLE["orchestration"] - - # Execute rounds. As for L2 (see _run_and_validate_l2), per-round timing - # is obtained offline from the `[STRACE]` stderr markers via - # `strace_timing --rounds-table`; the L3 chip children emit their own - # markers (grouped by (pid, inv)), so multi-round works without any - # inline fd capture here. - for round_idx in range(rounds): - if round_idx > 0: - for name, initial in initial_tensors.items(): - getattr(test_args, name).copy_(initial) - - # See _run_and_validate_l2: the per-round masking is dead code - # under the existing upstream gate. Keep parity by passing through. - config = self._build_config( - config_dict, - enable_l2_swimlane=enable_l2_swimlane, - enable_dump_args=enable_dump_args, - enable_pmu=enable_pmu, - enable_dep_gen=enable_dep_gen, - enable_scope_stats=enable_scope_stats, - output_prefix=output_prefix, - ) + try: + # Compute golden (unless skip_golden) + golden_args = None + if not skip_golden: + golden_args = test_args.clone() + self.compute_golden(golden_args, params) + + # Save initial tensor values for reset between rounds + all_tensor_names = test_args.tensor_names() + initial_tensors = {} + if rounds > 1: + for name in all_tensor_names: + initial_tensors[name] = getattr(test_args, name).clone() + + # Build CallableNamespace: compiled ChipCallables + sub callable IDs + ns = CallableNamespace({**compiled_callables, **sub_handles}) + + # Get orch function (plain function from CALLABLE) + orch_fn = self.CALLABLE["orchestration"] + + # Execute rounds. As for L2 (see _run_and_validate_l2), per-round timing + # is obtained offline from the `[STRACE]` stderr markers via + # `strace_timing --rounds-table`; the L3 chip children emit their own + # markers (grouped by (pid, inv)), so multi-round works without any + # inline fd capture here. + for round_idx in range(rounds): + if round_idx > 0: + for name, initial in initial_tensors.items(): + getattr(test_args, name).copy_(initial) + + # See _run_and_validate_l2: the per-round masking is dead code + # under the existing upstream gate. Keep parity by passing through. + config = self._build_config( + config_dict, + enable_l2_swimlane=enable_l2_swimlane, + enable_dump_args=enable_dump_args, + enable_pmu=enable_pmu, + enable_dep_gen=enable_dep_gen, + enable_scope_stats=enable_scope_stats, + output_prefix=output_prefix, + ) - # Orch fn signature: (orch, args, cfg) — inner fn forwards to - # the user's scene orch which takes (orch, callables, task_args, config). - def task_orch(orch, _args, _cfg, _ns=ns, _test_args=test_args, _config=config): - orch_fn(orch, _ns, _test_args, _config) + # Orch fn signature: (orch, args, cfg) — inner fn forwards to + # the user's scene orch which takes (orch, callables, task_args, config). + def task_orch(orch, _args, _cfg, _ns=ns, _test_args=test_args, _config=config): + orch_fn(orch, _ns, _test_args, _config) - with _temporary_env(self._resolve_env()): - worker.run(task_orch) + with _temporary_env(self._resolve_env()): + worker.run(task_orch) - if not skip_golden: - _compare_outputs(test_args, golden_args, all_tensor_names, self.RTOL, self.ATOL) + if not skip_golden: + _compare_outputs(test_args, golden_args, all_tensor_names, self.RTOL, self.ATOL) + finally: + _free_chip_host_buffers(worker, test_args, host_bufs) # ------------------------------------------------------------------ # pytest auto test method diff --git a/tests/st/a2a3/tensormap_and_ringbuffer/test_l3_host_buffer_registration.py b/tests/st/a2a3/tensormap_and_ringbuffer/test_l3_host_buffer_registration.py index 9610eecd77..cee6e29bfc 100644 --- a/tests/st/a2a3/tensormap_and_ringbuffer/test_l3_host_buffer_registration.py +++ b/tests/st/a2a3/tensormap_and_ringbuffer/test_l3_host_buffer_registration.py @@ -9,8 +9,8 @@ # ----------------------------------------------------------------------------------------------------------- """L3 post-fork zero-copy host buffers (issue #1027 #1). -A host tensor created *after* the chip children are forked (lazily on the -first ``Worker.run()``) is not visible to those children: the orch fn runs in +A host tensor created *after* the chip children are forked (during +``Worker.init()``) is not visible to those children: the orch fn runs in the parent and carries a raw parent VA that is unmapped (or stale) in the child. ``Worker.create_host_buffer`` hands back born-shared memory already attached into every chip child, so a tensor built over it with ``torch.frombuffer`` round-trips @@ -97,21 +97,16 @@ class TestPostForkHostBufferZeroCopy(SceneTestCase): {"name": "post_fork_zero_copy", "platforms": ["a2a3sim"]}, ] - def _force_fork(self, worker, chip_handle): - a = torch.full((SIZE,), 2.0, dtype=DTYPE).share_memory_() - b = torch.full((SIZE,), 3.0, dtype=DTYPE).share_memory_() - out = torch.zeros(SIZE, dtype=DTYPE).share_memory_() - worker.run(_one_task_orch(chip_handle, a, b, out), args=None, config=CallConfig()) - assert torch.allclose(out, _golden(a, b), rtol=self.RTOL, atol=self.ATOL) - def test_run(self, st_worker): """Zero-copy: buffers allocated AFTER the fork via ``create_host_buffer``, - filled in place, run, and read back — all without a per-run copy.""" + filled in place, run, and read back — all without a per-run copy. + + ``Worker.init()`` has already forked the chip children, so + ``create_host_buffer`` (which requires forked children) is usable + immediately — no warm-up run to force the fork is needed.""" worker = st_worker chip_handle = type(self)._st_chip_handles["vector"] - self._force_fork(worker, chip_handle) - nbytes = SIZE * DTYPE.itemsize # element count × dtype size, not a magic 4 ba = worker.create_host_buffer(nbytes) bb = worker.create_host_buffer(nbytes) diff --git a/tests/ut/py/test_callable_identity.py b/tests/ut/py/test_callable_identity.py index 5ed5847607..8318732655 100644 --- a/tests/ut/py/test_callable_identity.py +++ b/tests/ut/py/test_callable_identity.py @@ -501,6 +501,9 @@ def fake_open_remote_session(self, *, spec, worker_id, session_id, timeout_s): monkeypatch.setattr(worker_mod, "_Worker", fake_worker_ctor) monkeypatch.setattr(Worker, "_open_remote_session", fake_open_remote_session) + # This test exercises _init_hierarchical's remote-id bookkeeping with a + # fake C worker; stub the eager fork+start so it isn't driven here. + monkeypatch.setattr(Worker, "_start_hierarchical", lambda self: None) worker = Worker(level=4, num_sub_workers=0) try: diff --git a/tests/ut/py/test_worker/test_host_worker.py b/tests/ut/py/test_worker/test_host_worker.py index 912a12310d..df933ef691 100644 --- a/tests/ut/py/test_worker/test_host_worker.py +++ b/tests/ut/py/test_worker/test_host_worker.py @@ -225,32 +225,30 @@ def test_close_releases_registered_callables(self): assert hw._identity_registry == {} assert hw._live_handles == {} - def test_prepare_python_fn_after_init_before_start_succeeds(self): - # init() allocates mailboxes but does not fork children. Python - # callables prepared in this window still land in the startup - # snapshot consumed by the first run(). + def test_prepare_python_fn_before_init_succeeds(self): + # init() is eager (forks children + starts the scheduler), so the + # only local-install window is before init(). A Python callable + # registered here lands in the fork snapshot. hw = Worker(level=3, num_sub_workers=0) - hw.init() try: handle = hw.register(lambda args: None) assert _slot_for(hw, handle) in hw._callable_registry finally: hw.close() - def test_prepare_python_fn_after_init_before_start_does_not_broadcast(self): - class BroadcastTrap: - def broadcast_control_all(self, *args, **kwargs): - raise AssertionError("pre-start Python prepare must not broadcast") - + def test_prepare_python_fn_before_init_does_not_broadcast(self): + # A pre-init register is a local-only install: children are not + # forked yet, so it must not trigger any child broadcast. hw = Worker(level=3, num_sub_workers=1) - hw.init() - real_worker = hw._worker + + def broadcast_trap(*args, **kwargs): + raise AssertionError("pre-init Python register must not broadcast") + + hw._broadcast_py_control = broadcast_trap try: - hw._worker = BroadcastTrap() handle = hw.register(lambda args: None) assert _slot_for(hw, handle) in hw._callable_registry finally: - hw._worker = real_worker hw.close() def test_prepare_python_fn_after_start_no_python_children_raises(self): @@ -373,75 +371,6 @@ def do_unregister(): hw._hierarchical_start_cv.wait = original_wait hw.close() - def test_prepare_blocks_startup_snapshot_from_not_started_window(self): - hw = Worker(level=3, num_sub_workers=0) - hw.init() - - real_registry_lock = hw._registry_lock - register_waiting = threading.Event() - release_register = threading.Event() - startup_snapshot_attempted = threading.Event() - result: list[CallableHandle] = [] - errors: list[BaseException] = [] - - class BlockingRegistryLock: - def __enter__(self): - thread_name = threading.current_thread().name - if thread_name == "register-thread": - register_waiting.set() - if not release_register.wait(timeout=2.0): - raise TimeoutError("test timed out waiting to release register") - elif thread_name == "startup-thread": - startup_snapshot_attempted.set() - return real_registry_lock.__enter__() - - def __exit__(self, exc_type, exc, tb): - return real_registry_lock.__exit__(exc_type, exc, tb) - - def locked(self): - return real_registry_lock.locked() - - hw._registry_lock = BlockingRegistryLock() - - def do_register(): - try: - result.append(hw.register(lambda args: None)) - except BaseException as exc: # noqa: BLE001 - errors.append(exc) - - def do_startup(): - try: - hw._start_hierarchical() - except BaseException as exc: # noqa: BLE001 - errors.append(exc) - - register_thread = threading.Thread(target=do_register, name="register-thread") - startup_thread = threading.Thread(target=do_startup, name="startup-thread") - try: - register_thread.start() - assert register_waiting.wait(timeout=2.0) - - startup_thread.start() - assert not startup_snapshot_attempted.wait(timeout=0.2) - - release_register.set() - register_thread.join(timeout=2.0) - startup_thread.join(timeout=2.0) - - assert not register_thread.is_alive() - assert not startup_thread.is_alive() - assert errors == [] - assert len(result) == 1 - assert _slot_for(hw, result[0]) == 0 - assert startup_snapshot_attempted.is_set() - assert hw._hierarchical_start_state == "started" - finally: - release_register.set() - register_thread.join(timeout=2.0) - startup_thread.join(timeout=2.0) - hw._registry_lock = real_registry_lock - hw.close() - def test_prepare_chip_callable_after_init_no_chips_succeeds(self): # With no chip children (device_ids unset), the C++ broadcast is a # no-op (next_level_threads_ is empty) — exercises the facade path