diff --git a/examples/workers/l3/child_memory/README.md b/examples/workers/l3/child_memory/README.md index 627dbc04b5..99459ee4c3 100644 --- a/examples/workers/l3/child_memory/README.md +++ b/examples/workers/l3/child_memory/README.md @@ -16,7 +16,10 @@ pattern in miniature. There is no `orch.free` — the buffer is reclaimed by `DeviceRunner::finalize` when `worker.close()` runs. That asymmetry is the whole point of -`child_memory`: you opt out of the runtime's bookkeeping. +`child_memory`: you opt out of the runtime's **automatic lifecycle** (no +auto-malloc, no auto-free). The pointer is still tracked for **ownership**: its +`(worker_id, ptr)` provenance is validated on every `copy_*` and dispatch, so it +cannot be freed, copied, or run on a worker that did not allocate it. ## Run diff --git a/python/simpler/orchestrator.py b/python/simpler/orchestrator.py index 4affc1a1d5..b013624bdd 100644 --- a/python/simpler/orchestrator.py +++ b/python/simpler/orchestrator.py @@ -194,11 +194,33 @@ def submit_next_level( self._worker._stage_host_buffers_for_chip_submit(c_args) final_worker_ids = _remote_data_eligible_worker_ids(remote_sidecar, eligible_worker_ids) cpp_worker_id = int(worker) - captured_refs = self._worker._capture_remote_sidecar_refs(remote_sidecar) if self._worker is not None else [] + worker = self._worker + # Do the (fallible) kind4 provenance analysis BEFORE capturing remote slot + # refs, so an exception here can never leave captured refs neither + # released nor adopted (which would defer a remote free forever). Capture + # is the last step before the rollback try. + child_ptrs = worker._child_ptrs_in_args(c_args) if worker is not None else [] + prov_guard: Any = contextlib.nullcontext() + candidates: set[int] = set() + if child_ptrs and worker is not None: + candidates = self._child_dispatch_candidates(cpp_worker_id, final_worker_ids) + prov_guard = worker._child_prov_lock + captured_refs = worker._capture_remote_sidecar_refs(remote_sidecar) if worker is not None else [] try: - self._o.submit_next_level( - digest, kind, target_namespace, c_args, cfg, cpp_worker_id, final_worker_ids, remote_sidecar - ) + with prov_guard: + if child_ptrs and worker is not None: + worker._child_prov_check_dispatch(child_ptrs, candidates, api="submit_next_level") + # The child_memory arg resolved to a unique owner; pass that + # worker as the effective affinity so C++ keys the child + # TensorKey by its owner (worker_id), not the raw -1. + # Otherwise the same buffer submitted once as -1 and once as W + # yields two keys (ptr,-1) / (ptr,W) and its dependency is + # missed. The unique target is exactly what the scheduler + # would have picked, so pinning it changes no scheduling. + cpp_worker_id = next(iter(candidates)) + self._o.submit_next_level( + digest, kind, target_namespace, c_args, cfg, cpp_worker_id, final_worker_ids, remote_sidecar + ) except BaseException: if self._worker is not None: self._worker._release_remote_slot_refs(captured_refs) @@ -206,7 +228,23 @@ def submit_next_level( if self._worker is not None: self._worker._adopt_remote_slot_refs(captured_refs) - def submit_next_level_group( + def _child_dispatch_candidates(self, cpp_worker_id: int, eligible_ids: Any) -> set[int]: + """Resolve the set of eligible target workers for a kind4 dispatch. + + A pinned ``worker`` is the sole candidate; an unpinned ``-1`` falls back + to the callable's eligible set, or the full next-level pool when the + callable is unconstrained. ``_child_prov_check_dispatch`` rejects any + result that is not a single unique target. + """ + if cpp_worker_id >= 0: + return {cpp_worker_id} + if eligible_ids: + return {int(w) for w in eligible_ids} + if self._worker is None: + return set() + return set(self._worker._next_level_target_ids()) + + def submit_next_level_group( # noqa: PLR0912 -- linear per-member sidecar + eligibility + kind4-provenance passes, one branch each self, callable_handle: Any, args_list: list, @@ -270,14 +308,67 @@ def submit_next_level_group( else [] ) cpp_worker_ids = worker_ids + # Per-member kind4 dispatch guard: each member's child_memory pointers + # must resolve to that member's unique eligible target and be live there. + # Run this (fallible) analysis BEFORE capturing remote slot refs, so an + # exception here can never strand captured refs outside the rollback try. + worker = self._worker + member_checks: list[tuple[int, list[tuple[int, int]], set[int]]] = [] + if worker is not None: + for g, c_args in enumerate(c_args_list): + child_ptrs = worker._child_ptrs_in_args(c_args) + if not child_ptrs: + continue + worker_pin = cpp_worker_ids[g] if g < len(cpp_worker_ids) else -1 + eligible_g = worker_id_sets[g] if g < len(worker_id_sets) else [] + member_checks.append((g, child_ptrs, self._child_dispatch_candidates(int(worker_pin), eligible_g))) + prov_guard: Any = ( + worker._child_prov_lock if (worker is not None and member_checks) else contextlib.nullcontext() + ) captured_refs: list[Any] = [] if self._worker is not None and remote_sidecars is not None: for sidecar in remote_sidecars: captured_refs.extend(self._worker._capture_remote_sidecar_refs(sidecar)) try: - self._o.submit_next_level_group( - digest, kind, target_namespace, c_args_list, cfg, cpp_worker_ids, worker_id_sets, remote_sidecars - ) + with prov_guard: + if member_checks: + # Materialise a full per-member affinity so each child member's + # resolved owner is the effective affinity C++ keys its child + # TensorKey by (see the single-submit note); non-child members + # keep their original affinity / -1. Only an empty/None + # ``workers`` is padded — a non-empty list must already be one + # per member (C++ enforces this), so padding a short one would + # silently bypass that length check. + if cpp_worker_ids: + if len(cpp_worker_ids) != len(c_args_list): + raise ValueError( + f"submit_next_level_group: workers length {len(cpp_worker_ids)} " + f"!= {len(c_args_list)} args" + ) + cpp_worker_ids = list(cpp_worker_ids) + else: + cpp_worker_ids = [-1] * len(c_args_list) + for g, child_ptrs, candidates in member_checks: + assert worker is not None # member_checks is only populated when worker is present + worker._child_prov_check_dispatch(child_ptrs, candidates, api="submit_next_level_group") + cpp_worker_ids[g] = next(iter(candidates)) + # A group dispatches its members in parallel to distinct workers. + # Two members pinned to the same owner (e.g. two child args on the + # same chip) would give the same affinity twice; the scheduler sees + # that WorkerThread idle for both and serializes them on one thread. + # This rejects that duplicate-pinned case only — full injective + # feasibility (e.g. a wildcard member left with no free worker once + # a child member is pinned) is the scheduler's pre-existing capacity + # concern, not narrowed here. + pinned = [wid for wid in cpp_worker_ids if wid >= 0] + if len(pinned) != len(set(pinned)): + raise ValueError( + f"submit_next_level_group: members resolve to duplicate target workers " + f"{cpp_worker_ids} — a group must dispatch to distinct workers" + ) + self._o.submit_next_level_group( + digest, kind, target_namespace, c_args_list, cfg, cpp_worker_ids, worker_id_sets, remote_sidecars + ) except BaseException: if self._worker is not None: self._worker._release_remote_slot_refs(captured_refs) @@ -422,20 +513,58 @@ def scope(self) -> Iterator[Orchestrator]: self._o.scope_end() def malloc(self, worker_id: int, size: int) -> int: - """Allocate memory on next-level worker *worker_id*. Returns a pointer.""" - return int(self._o.malloc(int(worker_id), int(size))) + """Allocate memory on next-level worker *worker_id*. Returns a pointer. + + This is the single L3 choke for kind4 device memory: ``Worker.malloc`` + also funnels through here, as does a user's direct ``orch.malloc``. The + returned pointer's ``(worker_id, ptr)`` provenance is recorded so a later + free / copy / kind4 dispatch to the wrong worker is rejected. + """ + wid, sz = int(worker_id), int(size) + if self._worker is None: + return int(self._o.malloc(wid, sz)) + with self._worker._child_prov_lock: + ptr = int(self._o.malloc(wid, sz)) + self._worker._child_prov_record_malloc(wid, ptr) + return ptr def free(self, worker_id: int, ptr: int) -> None: """Free memory on next-level worker *worker_id*.""" - self._o.free(int(worker_id), int(ptr)) + wid, p = int(worker_id), int(ptr) + if self._worker is None: + self._o.free(wid, p) + return + with self._worker._child_prov_lock: + # Safety-first commit barrier: revoke provenance BEFORE the native + # free. If the native free succeeds and an async unwind (e.g. a + # KeyboardInterrupt delivered after the binding returns) fires before + # a post-free clear could run, a freed address would stay live and a + # later copy/dispatch would re-authorize it — a UAF. Revoking first + # turns a native-free failure into a terminal leak (recoverable) but + # never re-authorizes a maybe-freed address. + self._worker._child_prov_require_malloc_base(wid, p, api="free") + self._worker._child_prov_clear_malloc(wid, p) + self._o.free(wid, p) def copy_to(self, worker_id: int, dst: int, src: int, size: int) -> None: """Copy *size* bytes from host *src* to worker *dst*.""" - self._o.copy_to(int(worker_id), int(dst), int(src), int(size)) + wid, d = int(worker_id), int(dst) + if self._worker is None: + self._o.copy_to(wid, d, int(src), int(size)) + return + with self._worker._child_prov_lock: + self._worker._child_prov_require_live(wid, d, api="copy_to") + self._o.copy_to(wid, d, int(src), int(size)) def copy_from(self, worker_id: int, dst: int, src: int, size: int) -> None: """Copy *size* bytes from worker *src* to host *dst*.""" - self._o.copy_from(int(worker_id), int(dst), int(src), int(size)) + wid, s = int(worker_id), int(src) + if self._worker is None: + self._o.copy_from(wid, int(dst), s, int(size)) + return + with self._worker._child_prov_lock: + self._worker._child_prov_require_live(wid, s, api="copy_from") + self._o.copy_from(wid, int(dst), s, int(size)) def alloc(self, shape: Sequence[int], dtype: DataType) -> Tensor: """Allocate a runtime-managed intermediate buffer. diff --git a/python/simpler/worker.py b/python/simpler/worker.py index 4e568a3b67..58af03ea68 100644 --- a/python/simpler/worker.py +++ b/python/simpler/worker.py @@ -75,6 +75,7 @@ def my_l4_orch(orch, args, config): import threading import time import uuid +from collections.abc import Sequence from dataclasses import dataclass, field from multiprocessing.shared_memory import SharedMemory from typing import Any, cast @@ -415,6 +416,32 @@ class _RemoteSession: _IdentitySnapshotEntry = tuple[bytes, Any, int, str, str] +class _ChildProvEntry: + """Provenance record for one exact ``(worker_id, device_ptr)`` child pointer. + + Typed rather than a bare presence bit because the same ``(worker_id, ptr)`` + can carry more than one role at once: a ``malloc`` base and a CommDomain + window / carved buffer pointer can legally alias the same device address. + The key is live while ``malloc_owned or domain_allocation_ids``; only an + exact ``malloc`` base is ``free``-able, while a domain pointer is revoked by + its domain's release. Interior pointers are never recorded, so a pointer + that merely lands inside a live allocation has no entry and is rejected. + """ + + __slots__ = ("malloc_owned", "domain_allocation_ids") + + def __init__(self) -> None: + self.malloc_owned: bool = False + self.domain_allocation_ids: set[int] = set() + + def is_live(self) -> bool: + """True iff this entry still carries a role. A role-less entry is dead — + live checks are fail-closed on this, never on key presence alone, so an + entry momentarily left empty (e.g. an interrupted revoke) never + re-authorizes a freed pointer.""" + return self.malloc_owned or bool(self.domain_allocation_ids) + + @dataclass class _HostBufEntry: """Parent-side record for a born-shared post-fork host buffer. @@ -2050,6 +2077,19 @@ def __init__( self._live_l3_l2_regions: list[Any] = [] self._l3_l2_orch_comm_host_buffers: dict[int, int] = {} + # Live-provenance of child (kind4, device) pointers, keyed on the exact + # ``(worker_id, device_ptr)`` composite: a raw device VA is not globally + # unique (two chips can return the same numeric address), so a single + # ptr->worker map would collide. Populated by malloc / allocate_domain, + # consumed by free / copy_to / copy_from and by kind4 argument dispatch + # so a device pointer is never freed, copied, or run on the wrong worker. + # Guarded by ``_child_prov_lock``, which makes each op atomic. Ordering is + # safety-first: malloc records only after the native alloc succeeds, while + # free (and domain release) revokes BEFORE the native free, so an + # interrupted op never leaves a freed address live. Cleared on close(). + self._child_alloc_prov: dict[tuple[int, int], _ChildProvEntry] = {} + self._child_prov_lock = threading.Lock() + # Post-fork zero-copy host buffers (``create_host_buffer``). Keyed by the # born-shared shm's mapped base (== the buffer's data_ptr); each entry maps # a named shm into every local L3 child so memory created after the children @@ -4780,6 +4820,16 @@ def _allocate_domain( # noqa: PLR0912 -- linear input-validation + per-chip shm _release_fn=self._release_domain_handle, ) self._live_domains[name] = handle + # The backend windows are now live: record each chip's window base and + # every carved buffer pointer so a later kind4 (child_memory) dispatch of + # one of them is validated against its owning chip. Revoked by + # _release_domain_now just before the backend free (a commit barrier), + # not by the deferred marker — so the deferred window stays dispatchable. + with self._child_prov_lock: + for chip_idx, ctx in contexts.items(): + self._child_prov_record_domain(chip_idx, int(ctx.local_window_base), allocation_id) + for buf_ptr in ctx.buffer_ptrs.values(): + self._child_prov_record_domain(chip_idx, int(buf_ptr), allocation_id) return handle def _release_domain_handle(self, handle: CommDomainHandle) -> None: @@ -4829,6 +4879,15 @@ def _release_domain_now(self, handle: CommDomainHandle) -> None: deferred-release path and by the abort/close cleanup helpers.""" if self._worker is None: return + # Revoke provenance BEFORE the physical free: once release begins the + # domain's pointers are no longer dispatchable. Revoking after the + # backend free would leave a use-after-free window (a concurrent + # copy/dispatch could still validate the being-freed pointer as live), + # and a partial/failed release would strand a freed pointer as "live" + # forever. Dropping first is the safe direction — a leak (if the backend + # free later fails) is recoverable; a use-after-free is not. + with self._child_prov_lock: + self._child_prov_drop_domain(handle.allocation_id) workers = handle.workers # Release payload is just the fixed header — no rank_ids tail; the # backend looked them up from its own per-allocation record at @@ -4946,6 +5005,136 @@ def _release_all_live_domains(self) -> None: # with in-flight dispatch on the same chip mailbox. # ------------------------------------------------------------------ + # ------------------------------------------------------------------ + # Child (kind4, device) pointer provenance (guard ②) + # + # Every mutator/reader below assumes the caller holds ``_child_prov_lock``, + # so the enclosing op is atomic. Ordering is safety-first: record after a + # successful native alloc; revoke before the native free. + # ------------------------------------------------------------------ + + def _child_prov_record_malloc(self, worker_id: int, ptr: int) -> None: + """Mark ``(worker_id, ptr)`` as a live malloc base (after a successful malloc).""" + entry = self._child_alloc_prov.get((worker_id, ptr)) + if entry is None: + # Fully initialise the role BEFORE inserting, so the dict never holds + # a role-less (dead) entry even if an async unwind lands here. + entry = _ChildProvEntry() + entry.malloc_owned = True + self._child_alloc_prov[(worker_id, ptr)] = entry + else: + entry.malloc_owned = True + + def _child_prov_require_malloc_base(self, worker_id: int, ptr: int, *, api: str) -> None: + """Require ``(worker_id, ptr)`` to be an exact live malloc base (freeable). + + Rejects a wrong-worker pointer, an interior/stale pointer, a double free, + and a CommDomain pointer (which is revoked by its domain's release, never + by ``free``). + """ + entry = self._child_alloc_prov.get((worker_id, ptr)) + if entry is None or not entry.malloc_owned: + raise ValueError( + f"Worker.{api}: device pointer 0x{ptr:x} is not a live malloc base on worker " + f"{worker_id} (wrong worker, already-freed/stale, an interior pointer, or a " + f"CommDomain buffer that must be released via release_domain)" + ) + + def _child_prov_clear_malloc(self, worker_id: int, ptr: int) -> None: + """Revoke the malloc role of ``(worker_id, ptr)`` — called BEFORE the native + free (safety-first), so an interrupted free never leaves the address live.""" + key = (worker_id, ptr) + entry = self._child_alloc_prov.get(key) + if entry is None: + return + if entry.domain_allocation_ids: + entry.malloc_owned = False # still live via a domain — keep the entry + else: + del self._child_alloc_prov[key] # last role — delete directly, no empty state + + def _child_prov_require_live(self, worker_id: int, ptr: int, *, api: str) -> None: + """Require ``(worker_id, ptr)`` to be a live child pointer (malloc or domain).""" + entry = self._child_alloc_prov.get((worker_id, ptr)) + if entry is None or not entry.is_live(): + raise ValueError( + f"Worker.{api}: device pointer 0x{ptr:x} is not a live allocation on worker " + f"{worker_id} (wrong worker, freed/stale, or an interior pointer)" + ) + + def _child_prov_record_domain(self, worker_id: int, ptr: int, allocation_id: int) -> None: + """Record a CommDomain window / buffer pointer at exact ``(worker_id, ptr)``.""" + entry = self._child_alloc_prov.get((worker_id, ptr)) + if entry is None: + entry = _ChildProvEntry() + self._child_alloc_prov[(worker_id, ptr)] = entry + entry.domain_allocation_ids.add(allocation_id) + + def _child_prov_drop_domain(self, allocation_id: int) -> None: + """Drop every pointer recorded by a CommDomain allocation (at the start of + its physical release, before the backend free — see _release_domain_now).""" + for key in list(self._child_alloc_prov): + entry = self._child_alloc_prov[key] + if allocation_id not in entry.domain_allocation_ids: + continue + if entry.malloc_owned or len(entry.domain_allocation_ids) > 1: + entry.domain_allocation_ids.discard(allocation_id) # other roles remain + else: + del self._child_alloc_prov[key] # last role — delete directly, no empty state + + @staticmethod + def _child_ptrs_in_args(args: Any) -> list[tuple[int, int]]: + """Extract ``(device_ptr, arg_index)`` for every child_memory tensor in ``args``.""" + out: list[tuple[int, int]] = [] + for i in range(args.tensor_count()): + tensor = args.tensor(i) + if tensor.child_memory: + out.append((int(tensor.data), i)) + return out + + def _next_level_target_ids(self) -> Sequence[int]: + """The full pool of dispatchable next-level worker ids. + + Chip ids ``0..N`` at L3; the stable ``_next_level_worker_ids`` at L4+ (an + index range would not match the local/remote stable worker ids). + """ + if self._chip_shms: + return range(len(self._chip_shms)) + return self._next_level_worker_ids + + def _child_prov_check_dispatch( + self, child_ptrs: list[tuple[int, int]], candidate_worker_ids: Any, *, api: str + ) -> None: + """Validate every child_memory pointer against its unique target worker. + + A child_memory argument must resolve to exactly one eligible target + worker; ``0`` or ``>= 2`` candidates is ambiguous and rejected (judged on + the resolved eligibility, not the raw ``worker=-1``). The pointer must be + a live allocation on that target — else it is being routed to the wrong + worker, or is stale. + """ + if not child_ptrs: + return + candidates = set(candidate_worker_ids) + if len(candidates) != 1: + arg_index = child_ptrs[0][1] + raise ValueError( + f"orch.{api}: child_memory argument (arg {arg_index}) cannot resolve a unique " + f"target worker (eligible={sorted(candidates)}); pin worker= explicitly" + ) + target = next(iter(candidates)) + for ptr, arg_index in child_ptrs: + entry = self._child_alloc_prov.get((target, ptr)) + if entry is None or not entry.is_live(): + raise ValueError( + f"orch.{api}: child_memory argument (arg {arg_index}, ptr 0x{ptr:x}) is not a " + f"live allocation on target worker {target} (wrong worker, stale, or interior pointer)" + ) + + def _clear_child_prov(self) -> None: + """Drop the whole child-pointer provenance table (close-path hygiene).""" + with self._child_prov_lock: + self._child_alloc_prov.clear() + def _check_chip_worker_id(self, worker_id: int) -> None: """Range-check ``worker_id`` against the L3-level chip mailbox set. @@ -4964,7 +5153,12 @@ def malloc(self, size: int, worker_id: int = 0) -> int: with self._operation_lease("malloc"): if self.level == 2: assert self._chip_worker is not None - return self._chip_worker.malloc(size) + # L2 is a single chip; worker_id is meaningless there, so the + # provenance is keyed on the canonical worker 0. + with self._child_prov_lock: + ptr = self._chip_worker.malloc(size) + self._child_prov_record_malloc(0, int(ptr)) + return ptr self._check_chip_worker_id(worker_id) assert self._orch is not None return self._orch.malloc(worker_id, size) @@ -4974,7 +5168,13 @@ def free(self, ptr: int, worker_id: int = 0) -> None: with self._operation_lease("free"): if self.level == 2: assert self._chip_worker is not None - self._chip_worker.free(ptr) + # Safety-first commit barrier (mirrors Orchestrator.free): revoke + # provenance BEFORE the native free so an async unwind after a + # successful free can never leave a freed address live. + with self._child_prov_lock: + self._child_prov_require_malloc_base(0, int(ptr), api="free") + self._child_prov_clear_malloc(0, int(ptr)) + self._chip_worker.free(ptr) return self._check_chip_worker_id(worker_id) assert self._orch is not None @@ -4985,7 +5185,9 @@ def copy_to(self, dst: int, src: int, size: int, worker_id: int = 0) -> None: with self._operation_lease("copy_to"): if self.level == 2: assert self._chip_worker is not None - self._chip_worker.copy_to(dst, src, size) + with self._child_prov_lock: + self._child_prov_require_live(0, int(dst), api="copy_to") + self._chip_worker.copy_to(dst, src, size) return self._check_chip_worker_id(worker_id) assert self._orch is not None @@ -4996,7 +5198,9 @@ def copy_from(self, dst: int, src: int, size: int, worker_id: int = 0) -> None: with self._operation_lease("copy_from"): if self.level == 2: assert self._chip_worker is not None - self._chip_worker.copy_from(dst, src, size) + with self._child_prov_lock: + self._child_prov_require_live(0, int(src), api="copy_from") + self._chip_worker.copy_from(dst, src, size) return self._check_chip_worker_id(worker_id) assert self._orch is not None @@ -5747,6 +5951,7 @@ def _step(fn) -> None: _step(self._cleanup_l3_l2_regions) if self._live_domains: _step(self._release_all_live_domains) + _step(self._clear_child_prov) _step(self._release_active_remote_slot_refs) _step(self._flush_pending_remote_frees) # Host buffers must be released while the local L3 child mailboxes are diff --git a/tests/ut/py/test_worker/test_child_addr_guard.py b/tests/ut/py/test_worker/test_child_addr_guard.py new file mode 100644 index 0000000000..f4c392ced9 --- /dev/null +++ b/tests/ut/py/test_worker/test_child_addr_guard.py @@ -0,0 +1,618 @@ +# 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. +# ----------------------------------------------------------------------------------------------------------- +"""G2 — device-address guard ②: kind4 (child_memory) pointer provenance. + +Device-free tests that pin the guard's contract: a child device pointer is +tracked by its exact ``(worker_id, ptr)`` provenance so it is never freed, +copied, or dispatched to the wrong next-level worker, and a stale (freed) +pointer is rejected before reuse. Covers every real entry — ``Worker.malloc`` +(L2), ``orch.malloc/copy/free`` (L3, the path that bypasses ``Worker.malloc``), +``submit_next_level`` / ``submit_next_level_group`` dispatch, and CommDomain +window pointers — plus the explicit boundary that strict ABA is NOT covered. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest +import simpler.orchestrator as orch_mod +from _task_interface import DataType, Tensor, TensorArgType +from simpler.orchestrator import Orchestrator +from simpler.task_interface import TaskArgs +from simpler.worker import Worker, _ChildProvEntry, _Lifecycle + + +def _l3() -> Worker: + return Worker(level=3, num_sub_workers=0, platform="a2a3sim", runtime="tensormap_and_ringbuffer") + + +def _child_args(ptr: int, *, n: int = 16) -> TaskArgs: + args = TaskArgs() + args.add_tensor(Tensor.make(ptr, (n,), DataType.FLOAT32, child_memory=True), TensorArgType.OUTPUT_EXISTING) + return args + + +def _record_malloc(w: Worker, worker_id: int, ptr: int) -> None: + with w._child_prov_lock: + w._child_prov_record_malloc(worker_id, ptr) + + +# ---------------------------------------------------------------------------- +# Provenance table — typed exact (worker_id, ptr) entries +# ---------------------------------------------------------------------------- + + +class TestProvenanceTable: + def test_malloc_base_is_live_then_freed(self): + w = _l3() + with w._child_prov_lock: + w._child_prov_record_malloc(0, 0x1000) + w._child_prov_require_live(0, 0x1000, api="copy_to") # no raise + w._child_prov_require_malloc_base(0, 0x1000, api="free") # no raise + w._child_prov_clear_malloc(0, 0x1000) + with pytest.raises(ValueError, match="not a live allocation"): + w._child_prov_require_live(0, 0x1000, api="copy_to") + + def test_free_of_unknown_pointer_rejected(self): + w = _l3() + with w._child_prov_lock, pytest.raises(ValueError, match="not a live malloc base"): + w._child_prov_require_malloc_base(0, 0xDEAD, api="free") + + def test_double_free_stale_before_reuse_rejected(self): + w = _l3() + with w._child_prov_lock: + w._child_prov_record_malloc(0, 0x1000) + w._child_prov_clear_malloc(0, 0x1000) + with pytest.raises(ValueError, match="already-freed/stale"): + w._child_prov_require_malloc_base(0, 0x1000, api="free") + + def test_interior_pointer_is_not_a_live_entry(self): + w = _l3() + with w._child_prov_lock: + w._child_prov_record_malloc(0, 0x1000) + # A pointer physically inside the allocation has no exact entry. + with pytest.raises(ValueError, match="interior"): + w._child_prov_require_live(0, 0x1008, api="copy_to") + + def test_same_numeric_va_on_two_workers_is_independent(self): + # A raw device VA is not globally unique; the composite key keeps two + # chips' identical numeric addresses independent. + w = _l3() + with w._child_prov_lock: + w._child_prov_record_malloc(0, 0x4000) + w._child_prov_record_malloc(1, 0x4000) + w._child_prov_clear_malloc(0, 0x4000) + with pytest.raises(ValueError, match="not a live allocation"): + w._child_prov_require_live(0, 0x4000, api="copy_to") + w._child_prov_require_live(1, 0x4000, api="copy_to") # survives + + def test_domain_pointer_is_not_freeable_but_is_dispatchable(self): + w = _l3() + with w._child_prov_lock: + w._child_prov_record_domain(0, 0x8000, allocation_id=7) + # usable for copy / dispatch + w._child_prov_require_live(0, 0x8000, api="copy_to") + # but not free-able — domains are revoked by release, not free() + with pytest.raises(ValueError, match="CommDomain buffer"): + w._child_prov_require_malloc_base(0, 0x8000, api="free") + w._child_prov_drop_domain(7) + with pytest.raises(ValueError, match="not a live allocation"): + w._child_prov_require_live(0, 0x8000, api="copy_to") + + def test_malloc_and_domain_alias_same_pointer(self): + # A malloc base and a domain buffer can alias the same (worker, ptr). + # Clearing the malloc role leaves it live via the domain; only dropping + # the domain removes it. + w = _l3() + with w._child_prov_lock: + w._child_prov_record_malloc(0, 0x9000) + w._child_prov_record_domain(0, 0x9000, allocation_id=3) + w._child_prov_clear_malloc(0, 0x9000) + w._child_prov_require_live(0, 0x9000, api="copy_to") # still live via domain + w._child_prov_drop_domain(3) + with pytest.raises(ValueError, match="not a live allocation"): + w._child_prov_require_live(0, 0x9000, api="copy_to") + + def test_empty_entry_is_not_live_fail_closed(self): + # A role-less entry (e.g. one left by an interrupted revoke) must never be + # treated as live — every check is fail-closed on the roles, not on the + # key's mere presence. + w = _l3() + w._child_alloc_prov[(0, 0x1000)] = _ChildProvEntry() # both roles empty + with w._child_prov_lock: + with pytest.raises(ValueError, match="not a live allocation"): + w._child_prov_require_live(0, 0x1000, api="copy_to") + with pytest.raises(ValueError, match="not a live malloc base"): + w._child_prov_require_malloc_base(0, 0x1000, api="free") + with pytest.raises(ValueError, match="not a live allocation on target worker 0"): + w._child_prov_check_dispatch([(0x1000, 0)], {0}, api="submit_next_level") + + def test_drop_last_role_deletes_entry_no_empty_state(self): + # Dropping the last role deletes the key outright — it never leaves a + # role-less entry behind (which a later live check could not tell from a + # live one without the fail-closed guard). + w = _l3() + with w._child_prov_lock: + w._child_prov_record_domain(0, 0x5000, allocation_id=1) + w._child_prov_drop_domain(1) + assert (0, 0x5000) not in w._child_alloc_prov + w._child_prov_record_malloc(0, 0x6000) + w._child_prov_clear_malloc(0, 0x6000) + assert (0, 0x6000) not in w._child_alloc_prov + + def test_strict_aba_is_explicitly_not_covered(self): + # Documentation test: after free + re-malloc of the same numeric VA, the + # (worker, ptr) becomes live again and an old handle is indistinguishable + # from the new allocation. Strict ABA is deferred to P1 generation + # handles; this guard only catches stale-BEFORE-reuse. + w = _l3() + with w._child_prov_lock: + w._child_prov_record_malloc(0, 0x1000) + w._child_prov_clear_malloc(0, 0x1000) + w._child_prov_record_malloc(0, 0x1000) # device handed back the same VA + w._child_prov_require_live(0, 0x1000, api="copy_to") # NOT rejected — by design + + +# ---------------------------------------------------------------------------- +# Target resolution + dispatch check +# ---------------------------------------------------------------------------- + + +class TestDispatchResolution: + def test_candidates_pinned_worker(self): + o = Orchestrator(MagicMock(), _l3()) + assert o._child_dispatch_candidates(2, [0, 1, 2]) == {2} + + def test_candidates_wildcard_uses_eligible_set(self): + o = Orchestrator(MagicMock(), _l3()) + assert o._child_dispatch_candidates(-1, [0, 1]) == {0, 1} + + def test_candidates_wildcard_unconstrained_uses_full_pool(self): + w = _l3() + w._chip_shms = [object(), object(), object()] # 3 chips + o = Orchestrator(MagicMock(), w) + assert o._child_dispatch_candidates(-1, []) == {0, 1, 2} + + def test_unique_target_and_live_passes(self): + w = _l3() + _record_malloc(w, 0, 0x1000) + with w._child_prov_lock: + w._child_prov_check_dispatch([(0x1000, 0)], {0}, api="submit_next_level") + + def test_ambiguous_target_rejected(self): + w = _l3() + _record_malloc(w, 0, 0x1000) + with w._child_prov_lock, pytest.raises(ValueError, match="cannot resolve a unique"): + w._child_prov_check_dispatch([(0x1000, 0)], {0, 1}, api="submit_next_level") + + def test_unique_target_but_wrong_worker_rejected(self): + w = _l3() + _record_malloc(w, 0, 0x1000) # lives on worker 0 + with w._child_prov_lock, pytest.raises(ValueError, match="not a live allocation on target worker 1"): + w._child_prov_check_dispatch([(0x1000, 0)], {1}, api="submit_next_level") + + +# ---------------------------------------------------------------------------- +# Orchestrator malloc / free / copy — the L3 choke (also the orch.* bypass) +# ---------------------------------------------------------------------------- + + +class TestOrchestratorMemoryOps: + def test_malloc_records_then_free_clears(self): + w = _l3() + fake = MagicMock() + fake.malloc.return_value = 0x1000 + o = Orchestrator(fake, w) + ptr = o.malloc(0, 64) + assert ptr == 0x1000 + assert (0, 0x1000) in w._child_alloc_prov + o.free(0, ptr) + fake.free.assert_called_once_with(0, 0x1000) + assert (0, 0x1000) not in w._child_alloc_prov + + def test_free_wrong_worker_rejected_without_native_free(self): + w = _l3() + fake = MagicMock() + fake.malloc.return_value = 0x1000 + o = Orchestrator(fake, w) + o.malloc(1, 64) # allocated on worker 1 + with pytest.raises(ValueError, match="not a live malloc base"): + o.free(0, 0x1000) # freed on worker 0 + fake.free.assert_not_called() + + def test_copy_to_requires_live_device_dst(self): + w = _l3() + fake = MagicMock() + fake.malloc.return_value = 0x2000 + o = Orchestrator(fake, w) + with pytest.raises(ValueError, match="not a live allocation"): + o.copy_to(0, 0x2000, 0xABCD, 64) + fake.copy_to.assert_not_called() + o.malloc(0, 64) + o.copy_to(0, 0x2000, 0xABCD, 64) + fake.copy_to.assert_called_once() + + def test_copy_from_requires_live_device_src(self): + w = _l3() + fake = MagicMock() + fake.malloc.return_value = 0x3000 + o = Orchestrator(fake, w) + with pytest.raises(ValueError, match="not a live allocation"): + o.copy_from(0, 0xABCD, 0x3000, 64) + fake.copy_from.assert_not_called() + + def test_isolated_orchestrator_has_no_guard(self): + # An Orchestrator constructed without a Worker back-ref (test isolation) + # must not attempt provenance tracking. + fake = MagicMock() + fake.malloc.return_value = 0x1000 + o = Orchestrator(fake, None) + assert o.malloc(0, 64) == 0x1000 + o.free(0, 0x1000) # no raise, no table + + +# ---------------------------------------------------------------------------- +# submit_next_level / group dispatch guard +# ---------------------------------------------------------------------------- + + +@pytest.fixture +def _fake_handle(monkeypatch): + """Patch _require_handle so submit_* can run device-free with a chosen + eligible set; returns a setter for the eligible worker ids.""" + state = {"eligible": (0,)} + + def _fake(callable_handle, **_kwargs): + return (b"d" * 32, "NEXT_LEVEL", "LOCAL_CHIP", state["eligible"]) + + monkeypatch.setattr(orch_mod, "_require_handle", _fake) + return state + + +class TestSubmitDispatchGuard: + def test_child_arg_to_correct_worker_passes(self, _fake_handle): + w = _l3() + _record_malloc(w, 0, 0x1000) + fake = MagicMock() + o = Orchestrator(fake, w) + o.submit_next_level(object(), _child_args(0x1000), None, worker=0) + fake.submit_next_level.assert_called_once() + + def test_child_arg_to_wrong_worker_rejected(self, _fake_handle): + w = _l3() + w._chip_shms = [object(), object()] + _record_malloc(w, 0, 0x1000) + fake = MagicMock() + o = Orchestrator(fake, w) + with pytest.raises(ValueError, match="not a live allocation on target worker 1"): + o.submit_next_level(object(), _child_args(0x1000), None, worker=1) + fake.submit_next_level.assert_not_called() + + def test_child_arg_wildcard_ambiguous_rejected(self, _fake_handle): + _fake_handle["eligible"] = (0, 1) # two eligible targets + w = _l3() + _record_malloc(w, 0, 0x1000) + fake = MagicMock() + o = Orchestrator(fake, w) + with pytest.raises(ValueError, match="cannot resolve a unique"): + o.submit_next_level(object(), _child_args(0x1000), None, worker=-1) + fake.submit_next_level.assert_not_called() + + def test_child_arg_wildcard_narrowed_passes_resolved_affinity_to_cpp(self, _fake_handle): + # worker=-1 narrowed to a unique owner must be passed to C++ as that + # worker (not the raw -1), so the child TensorKey is keyed by its owner + # and deps against an explicit-worker submit of the same buffer are seen. + _fake_handle["eligible"] = (0,) + w = _l3() + _record_malloc(w, 0, 0x1000) + fake = MagicMock() + o = Orchestrator(fake, w) + o.submit_next_level(object(), _child_args(0x1000), None, worker=-1) + fake.submit_next_level.assert_called_once() + assert fake.submit_next_level.call_args.args[5] == 0 # cpp_worker_id, not -1 + + def test_host_only_args_are_not_guarded(self, _fake_handle): + # A submit with no child_memory tensor never touches provenance. + w = _l3() + fake = MagicMock() + o = Orchestrator(fake, w) + args = TaskArgs() + args.add_tensor(Tensor.make(0, (16,), DataType.FLOAT32, child_memory=False), TensorArgType.INPUT) + o.submit_next_level(object(), args, None, worker=-1) + fake.submit_next_level.assert_called_once() + + def test_group_member_child_arg_wrong_worker_rejected(self, _fake_handle): + w = _l3() + w._chip_shms = [object(), object()] + _record_malloc(w, 0, 0x1000) + fake = MagicMock() + o = Orchestrator(fake, w) + # member 0 carries the child ptr live on worker 0, but is pinned to worker 1 + with pytest.raises(ValueError, match="not a live allocation on target worker 1"): + o.submit_next_level_group(object(), [_child_args(0x1000), TaskArgs()], None, workers=[1, 0]) + fake.submit_next_level_group.assert_not_called() + + def test_group_member_child_arg_correct_worker_passes(self, _fake_handle): + w = _l3() + w._chip_shms = [object(), object()] + _record_malloc(w, 1, 0x1000) + fake = MagicMock() + o = Orchestrator(fake, w) + o.submit_next_level_group(object(), [TaskArgs(), _child_args(0x1000)], None, workers=[0, 1]) + fake.submit_next_level_group.assert_called_once() + + def test_group_child_member_wildcard_passes_resolved_affinity_to_cpp(self, _fake_handle): + # A single-member group (schedulable) whose child member's eligibility + # narrows to a unique owner materialises a full per-member affinity + # pinning it to that owner. + _fake_handle["eligible"] = (1,) + w = _l3() + w._chip_shms = [object(), object()] + _record_malloc(w, 1, 0x1000) + fake = MagicMock() + o = Orchestrator(fake, w) + o.submit_next_level_group(object(), [_child_args(0x1000)], None, workers=None) + fake.submit_next_level_group.assert_called_once() + assert fake.submit_next_level_group.call_args.args[5] == [1] # cpp_worker_ids + + def test_group_child_member_rejects_mismatched_workers_length(self, _fake_handle): + # A non-empty workers list must be one-per-member; a short list must NOT + # be silently padded (that would bypass the C++ length check). + w = _l3() + w._chip_shms = [object(), object()] + _record_malloc(w, 0, 0x1000) + fake = MagicMock() + o = Orchestrator(fake, w) + with pytest.raises(ValueError, match="workers length 1 != 2 args"): + o.submit_next_level_group(object(), [_child_args(0x1000), TaskArgs()], None, workers=[0]) + fake.submit_next_level_group.assert_not_called() + + def test_group_two_child_members_same_owner_rejected(self, _fake_handle): + # Two child members owned by the same worker resolve to the same + # affinity; a group must dispatch to distinct workers, so reject rather + # than silently serialize them on one WorkerThread. + _fake_handle["eligible"] = (0,) + w = _l3() + w._chip_shms = [object(), object()] + _record_malloc(w, 0, 0x1000) + _record_malloc(w, 0, 0x2000) + fake = MagicMock() + o = Orchestrator(fake, w) + with pytest.raises(ValueError, match="distinct workers"): + o.submit_next_level_group(object(), [_child_args(0x1000), _child_args(0x2000)], None, workers=None) + fake.submit_next_level_group.assert_not_called() + + def test_domain_pointer_dispatch_to_owner_then_rejected_after_release(self, _fake_handle): + # CommDomain window pointers enter provenance so they dispatch as kind4 + # to their owning chip; a release revokes them. + w = _l3() + w._chip_shms = [object(), object()] + with w._child_prov_lock: + w._child_prov_record_domain(0, 0x5000, allocation_id=42) + fake = MagicMock() + o = Orchestrator(fake, w) + o.submit_next_level(object(), _child_args(0x5000), None, worker=0) + fake.submit_next_level.assert_called_once() + # wrong chip is rejected + with pytest.raises(ValueError, match="target worker 1"): + o.submit_next_level(object(), _child_args(0x5000), None, worker=1) + # after release the pointer is dead everywhere + with w._child_prov_lock: + w._child_prov_drop_domain(42) + with pytest.raises(ValueError, match="not a live allocation"): + o.submit_next_level(object(), _child_args(0x5000), None, worker=0) + + +# ---------------------------------------------------------------------------- +# L2 Worker.malloc/free/copy path (direct to the single chip) +# ---------------------------------------------------------------------------- + + +class TestL2WorkerPath: + def _l2(self) -> tuple[Worker, MagicMock]: + w = Worker(level=2, platform="a2a3sim", runtime="tensormap_and_ringbuffer", device_id=0) + chip = MagicMock() + chip.malloc.return_value = 0x2000 + w._chip_worker = chip + w._lifecycle = _Lifecycle.READY + return w, chip + + def test_l2_malloc_records_and_free_clears(self): + w, chip = self._l2() + ptr = w.malloc(64) + assert (0, ptr) in w._child_alloc_prov + w.free(ptr) + chip.free.assert_called_once_with(0x2000) + assert (0, ptr) not in w._child_alloc_prov + + def test_l2_free_stale_rejected_without_native_free(self): + w, chip = self._l2() + w.malloc(64) + w.free(0x2000) + chip.free.reset_mock() + with pytest.raises(ValueError, match="already-freed/stale"): + w.free(0x2000) + chip.free.assert_not_called() + + def test_l2_free_revokes_before_native_free(self): + # L2 mirrors the L3 commit barrier: revoke before the native free. + w, chip = self._l2() + w.malloc(64) + seen = {} + chip.free.side_effect = lambda p: seen.__setitem__("live_at_native", (0, 0x2000) in w._child_alloc_prov) + w.free(0x2000) + assert seen["live_at_native"] is False + + def test_l2_copy_to_requires_live_dst(self): + w, chip = self._l2() + with pytest.raises(ValueError, match="not a live allocation"): + w.copy_to(0x2000, 0xABCD, 64) + chip.copy_to.assert_not_called() + w.malloc(64) + w.copy_to(0x2000, 0xABCD, 64) + chip.copy_to.assert_called_once() + + +# ---------------------------------------------------------------------------- +# Provenance transaction failures (record after alloc success; free revokes +# before the native free — safety-first, terminal-leak on failure) +# ---------------------------------------------------------------------------- + + +class TestProvenanceTransactions: + def test_orch_malloc_native_error_records_nothing(self): + # Provenance is recorded only after the backend malloc succeeds. + w = _l3() + fake = MagicMock() + fake.malloc.side_effect = RuntimeError("device OOM") + o = Orchestrator(fake, w) + with pytest.raises(RuntimeError, match="device OOM"): + o.malloc(0, 64) + assert w._child_alloc_prov == {} + + def test_orch_free_revokes_before_native_free(self): + # Safety-first commit barrier: provenance is revoked BEFORE the native + # free, so an async unwind after a successful free cannot leave a freed + # address live. + w = _l3() + fake = MagicMock() + fake.malloc.return_value = 0x1000 + o = Orchestrator(fake, w) + o.malloc(0, 64) + seen = {} + fake.free.side_effect = lambda wid, p: seen.__setitem__("live_at_native", (0, 0x1000) in w._child_alloc_prov) + o.free(0, 0x1000) + assert seen["live_at_native"] is False # already revoked when native free runs + + def test_orch_free_native_error_revokes_provenance_safe_first(self): + # A native free that fails becomes a terminal leak — provenance is + # revoked, never re-authorized. No retry (the address is no longer a + # live malloc base). + w = _l3() + fake = MagicMock() + fake.malloc.return_value = 0x1000 + o = Orchestrator(fake, w) + o.malloc(0, 64) + fake.free.side_effect = RuntimeError("free failed") + with pytest.raises(RuntimeError, match="free failed"): + o.free(0, 0x1000) + assert (0, 0x1000) not in w._child_alloc_prov # revoked (terminal leak) + fake.free.side_effect = None + with pytest.raises(ValueError, match="not a live malloc base"): + o.free(0, 0x1000) + + def test_free_holds_lock_across_native_free(self): + # Deterministic mutual-exclusion check: the native free runs while + # _child_prov_lock is held, so a concurrent free/copy/dispatch cannot + # interleave with a half-completed free. + w = _l3() + fake = MagicMock() + fake.malloc.return_value = 0x1000 + o = Orchestrator(fake, w) + o.malloc(0, 64) + held = {} + + def _sf(wid, p): + acquired = w._child_prov_lock.acquire(blocking=False) + held["locked_during_native"] = not acquired + if acquired: + w._child_prov_lock.release() + + fake.free.side_effect = _sf + o.free(0, 0x1000) + assert held["locked_during_native"] is True + + def test_capture_refs_after_provenance_analysis(self, _fake_handle, monkeypatch): + # blocker: the kind4 provenance analysis must run BEFORE remote slot refs + # are captured, so an analysis failure cannot strand captured refs outside + # the rollback try (deferring a remote free forever). + w = _l3() + o = Orchestrator(MagicMock(), w) + monkeypatch.setattr(w, "_child_ptrs_in_args", MagicMock(side_effect=RuntimeError("boom"))) + cap = MagicMock(return_value=[]) + monkeypatch.setattr(w, "_capture_remote_sidecar_refs", cap) + with pytest.raises(RuntimeError, match="boom"): + o.submit_next_level(object(), _child_args(0x1000), None, worker=0) + cap.assert_not_called() + + def test_l2_malloc_native_error_records_nothing(self): + w = Worker(level=2, platform="a2a3sim", runtime="tensormap_and_ringbuffer", device_id=0) + chip = MagicMock() + chip.malloc.side_effect = RuntimeError("device OOM") + w._chip_worker = chip + w._lifecycle = _Lifecycle.READY + with pytest.raises(RuntimeError, match="device OOM"): + w.malloc(64) + assert w._child_alloc_prov == {} + + +# ---------------------------------------------------------------------------- +# CommDomain physical release — revoke before backend free (commit barrier) +# ---------------------------------------------------------------------------- + + +class TestDomainReleaseOrdering: + def _domain_worker(self): + """A level-3 Worker with a recorded domain and a native worker present.""" + w = _l3() + w._worker = MagicMock() # non-None so _release_domain_now proceeds + with w._child_prov_lock: + w._child_prov_record_domain(0, 0x5000, allocation_id=9) + w._child_prov_record_domain(1, 0x6000, allocation_id=9) + handle = SimpleNamespace(name="d", workers=(0, 1), allocation_id=9) + return w, handle + + def test_release_revokes_provenance_before_backend_free(self, monkeypatch): + w, handle = self._domain_worker() + seen_live_at_dispatch = {} + + def _fake_dispatch(**kwargs): + # At physical-free time the pointers must already be revoked. + seen_live_at_dispatch["still_live"] = (0, 0x5000) in w._child_alloc_prov + + monkeypatch.setattr(w, "_dispatch_control_domain", _fake_dispatch) + w._release_domain_now(handle) # type: ignore[arg-type] + assert seen_live_at_dispatch["still_live"] is False + assert (0, 0x5000) not in w._child_alloc_prov + assert (1, 0x6000) not in w._child_alloc_prov + + def test_release_vs_dispatch_rejects_during_native_release(self, monkeypatch): + # Deterministic domain-release-vs-dispatch: by the time the backend free + # runs, the pointer is already revoked, so a dispatch that lands during + # the native release is rejected — no freed-but-live window. + w, handle = self._domain_worker() + outcome = {} + + def _fake_dispatch(**kwargs): + try: + with w._child_prov_lock: + w._child_prov_check_dispatch([(0x5000, 0)], {0}, api="submit_next_level") + outcome["dispatch"] = "allowed" + except ValueError: + outcome["dispatch"] = "rejected" + + monkeypatch.setattr(w, "_dispatch_control_domain", _fake_dispatch) + w._release_domain_now(handle) # type: ignore[arg-type] + assert outcome["dispatch"] == "rejected" + + def test_release_backend_failure_leaves_provenance_dropped(self, monkeypatch): + # A partial/failed backend release must not restore the pointers to live + # (a leak is safe; a use-after-free is not). + w, handle = self._domain_worker() + + def _boom(**kwargs): + raise RuntimeError("release failed on one chip") + + monkeypatch.setattr(w, "_dispatch_control_domain", _boom) + with pytest.raises(RuntimeError, match="release failed"): + w._release_domain_now(handle) # type: ignore[arg-type] + assert (0, 0x5000) not in w._child_alloc_prov + assert (1, 0x6000) not in w._child_alloc_prov