Method explore_step

trinity/explorer/explorer.py:315–354

(self)

        return self.config.explorer.name

    async def explore_step(self) -> bool:
        if self.explore_start_time is None:
            self.explore_start_time = time.time()
        try:
            tasks = await self.taskset.read()
        except StopAsyncIteration:
            self.logger.warning("No more tasks to explore. Stop exploring.")
            await self.finish_current_steps()
            await self.save_checkpoint()
            await self.synchronizer.set_explorer_status.remote(
                RunningStatus.STOPPED,
                old_status=RunningStatus.RUNNING,
            )
            await self.shutdown()
            return False
        self.explore_step_num += 1
        if self.rollout_coordinator is None:
            return False
        # FULLY_ASYNC: when the in-flight window is full, wait for the oldest batch to
        # finish before submitting a new one.  This provides rolling backpressure without
        # waiting for a sync-interval boundary.
        if (
            self.sync_style == SyncStyle.FULLY_ASYNC
            and len(self._inflight_train_steps) >= self.config.explorer.max_inflight_batches
        ):
            oldest_step = self._inflight_train_steps.popleft()
            self.logger.debug(
                f"FULLY_ASYNC: at capacity, draining oldest batch (step {oldest_step})."
            )
            await self._finish_explore_step(step=oldest_step)
            self.last_monitored_step = oldest_step
        await self.rollout_coordinator.submit_batch.remote(
            batch_id=self.explore_step_num,
            tasks=tasks,
            batch_type="train",
            min_wait_num=self.min_wait_num,
        )
        if self.sync_style == SyncStyle.FULLY_ASYNC:
            self._inflight_train_steps.append(self.explore_step_num)
        return True

    async def finish_current_steps(self) -> None:
        if self.rollout_coordinator is not None: