16 · Condition、RLock 与 Timer

        groups = []
        dropped_count = 0
        drop_reasons = Counter()
        with self.condition:
            if self.error is not None:
                raise self.error
            if self.demand:
                raise RuntimeError("A rollout batch is already being collected")
            # The previous batch is durable before the next collection starts.
            self.collecting.clear()
            self.rollout_id = rollout_id
            self.demand = self.args.rollout_batch_size
            self.prefetch = min(prefetch, self.capacity)
            self.paused = False
            if self.thread is None:
                self.thread = threading.Thread(target=self._run, name="rollout-collect", daemon=True)
                self.thread.start()
            self.condition.notify_all()
            while self.demand:
                if self.error is not None:
                    raise self.error
                if not self.ready:
                    self.condition.wait()
                    continue
                group, verdict = self.ready.popleft()

    def pause(self, *, drain=True):
        """Stop admission; optionally wait for in-flight results to become durable."""
        with self.condition:
            was_paused = self.paused
            self.paused = True
            self.condition.notify_all()
            while self.error is None:
                unsettled = bool(self.controls) or any(
                    self.configurations.get(worker) != (True, self.rollout_id) for worker in self.running_workers
                )
                if drain:
                    unsettled |= bool(self.pending or self.running_workers)
                if not unsettled:
                    break
                self.condition.wait()
            if self.error is not None:
                raise self.error
            return was_paused

    def close(self):
        with self.condition:
            self.closed = True
            # Unblock consumers and pause() callers waiting for in-flight work.
            if self.error is None:
                self.error = RuntimeError("Rollout scheduler is closed")
            self.condition.notify_all()
        if self.thread is not None:
            self.thread.join()

    def pause(self, *, drain=True):
        with self.condition:
            previous = self.paused
            self.paused = True
            while self.running and self.error is None and (not self.admission_paused or (drain and self.active)):
                self.condition.wait()
            if self.error is not None:
                raise self.error
            return previous

        self.lock = threading.RLock()
        self.not_empty = threading.Condition(self.lock)
        self.total_written = 0
        self.total_read = 0
        self.task_type = task_type

    def write(self, data):
        with self.lock:
            self.buffer.append(data)
            self.total_written += 1
            self.not_empty.notify_all()
        return data

    def read(self):
        with self.not_empty:
            if len(self.buffer) == 0:
                return {"data": [], "meta_info": {}}

            # Don't clear temp_data for regular read operations
            result = self.buffer.get()
            self.total_read += len(result["data"])
            return result

    def wait(self, predicate):
        with self.condition:
            ready = self.condition.wait_for(lambda: self.error or self.done or predicate(self.events), timeout=180)
            if self.error:
                raise self.error
            assert ready and predicate(self.events), f"Probe failed on {self.url}: {self.events[-3:]}"

def get_accelerator() -> Accelerator:
    global _ACCELERATOR
    if _ACCELERATOR is not None:
        return _ACCELERATOR
    with _SELECTION_LOCK:
        if _ACCELERATOR is not None:
            return _ACCELERATOR
        _register_builtin_backends()
        requested = _requested_name()

def _register_builtin_backends() -> None:
    if "cuda" not in _REGISTRY:
        register_accelerator("cuda", CUDAAccelerator, _cuda_available, priority=100, communication_backends=("nccl",))
    if "musa" not in _REGISTRY:
        register_accelerator(
            "musa", MUSAAccelerator, is_musa_available, priority=200, communication_backends=("mccl",)
        )
    if "npu" not in _REGISTRY:
        register_accelerator("npu", NPUAccelerator, is_npu_available, priority=250, communication_backends=("hccl",))
    if "supa" not in _REGISTRY:
        register_accelerator(
            "supa", SUPAAccelerator, is_supa_available, priority=150, communication_backends=("bccl",)
        )

def register_accelerator(
    name: str,
    factory: Callable[[], Accelerator],
    is_available: Callable[[], bool] | None = None,
    priority: int = 0,
    communication_backends: tuple[str, ...] = (),
) -> None:
    """Register a lazily constructed backend without changing Slime core."""
    normalized = name.strip().lower()
    if not normalized or normalized in {"auto", "none"}:
        raise ValueError("Accelerator name must be a non-empty backend name")
    with _SELECTION_LOCK:
        _REGISTRY[normalized] = _BackendRegistration(
            factory=factory,
            is_available=is_available or (lambda: factory().is_available()),
            priority=priority,
            communication_backends=tuple(name.lower() for name in communication_backends),
        )

    def restore():
        with path.open("r+b") as f:
            f.seek(segment.offset)
            f.write(header)
            f.flush()
            os.fsync(f.fileno())

    timer = threading.Timer(0.05, restore)
    timer.start()
    try:
        if loader == "plain":
            result = transport.unpack_rollout_payload(ref)
        elif loader == "async":
            result = asyncio.run(transport.unpack_published_payload(ref))
        else:
            result = ref.load()
        assert result == value
    finally:
        timer.join()
    assert "Straw extent visibility retry" in capfd.readouterr().err