07 · Queue：把生产速度和消费速度接起来

    async def _start(self, rollout_id):
        self.rollout_id = rollout_id
        self.running = True
        if self.producer is None or self.producer.done():
            if self.producer is not None:
                self.producer.result()
            # Keep completed results bounded even when training stops consuming.
            # At most one queued result plus one result per pool slot stays local.
            self.outputs = asyncio.Queue(maxsize=1)
            self.producer = asyncio.create_task(self._produce())

    async def _produce(self):
        async def generate_slot(group):
            while self.running or group is not None:
                output = await self._generate_group(self.rollout_id, group)
                group = None
                if output is not None:  # Aborted groups have been returned to the durable queue.
                    await self.outputs.put(output)
                del output

    def get_completed_groups(self, limit: int | None = None) -> list[tuple[int, list[Sample]]]:
        """Pop up to ``limit`` completed groups (all of them when ``None``).

        Callers that only need a fixed number of groups must pass ``limit`` —
        anything popped beyond it would otherwise have to be thrown away, and
        these groups are fully generated and reward-scored, with their prompts
        already consumed from ``data_buffer``.
        """
        completed: list[tuple[int, list[Sample]]] = []
        while limit is None or len(completed) < limit:
            try:
                completed.append(self.output_queue.get_nowait())
            except queue.Empty:
                break
        return completed

    def queue_size(self) -> int:
        return self.output_queue.qsize()