08 · 线程池：把阻塞工作移出事件循环

        self._lock = threading.RLock()
        self._reader = ThreadPoolExecutor(max_workers=1, thread_name_prefix="queue-input")
        self._stop = threading.Event()
        self._heartbeat_error = None
        filter_path = getattr(args, "buffer_filter_path", None)
        if filter_path is not None:
            raise ValueError(
                "--buffer-filter-path is not supported by straw; the queue prioritizes ready groups, partials, "
                "then fresh prompts, using older weight versions first within each stage"
            )
        self._heartbeats = threading.Thread(target=self._heartbeat_loop, name="queue-leases", daemon=True)
        self._heartbeats.start()

    async def get_samples_async(self, num_samples):
        """Run blocking queue/file reads on the reader thread, keeping asyncio responsive."""
        return await asyncio.get_running_loop().run_in_executor(self._reader, self.get_samples, num_samples)

async def run_rollout_io(args, function, *values, **kwargs):
    """Bound the executor backlog before submitting encoding/sync work."""
    loop = asyncio.get_running_loop()
    key = (os.getpid(), loop)
    if key not in _async_limits:
        _async_limits[key] = asyncio.Semaphore(getattr(args, "rollout_io_concurrency", 4))
    async with _async_limits[key]:
        pending = asyncio.create_task(asyncio.to_thread(function, *values, **kwargs))
        try:
            return await asyncio.shield(pending)
        except asyncio.CancelledError:
            # Repeated cancellation still cannot stop a filesystem commit.
            # Keep capacity and input ownership until the accepted write finishes.
            while not pending.done():
                try:
                    await asyncio.shield(pending)
                except asyncio.CancelledError:
                    pass
            pending.result()
            raise