(self, batch_size: int)
| 110 | return experiences |
| 111 | |
| 112 | async def _read_fifo(self, batch_size: int) -> List[Experience]: |
| 113 | exp_list = [] |
| 114 | start_time = time.time() |
| 115 | while len(exp_list) < batch_size: |
| 116 | if self.stopped: |
| 117 | raise StopAsyncIteration() |
| 118 | if time.time() - start_time > self.max_timeout: |
| 119 | self.logger.warning( |
| 120 | f"Max read timeout reached ({self.max_timeout} s), " |
| 121 | f"only got {len(exp_list)} experiences, stopping..." |
| 122 | ) |
| 123 | raise StopAsyncIteration() |
| 124 | |
| 125 | current_offset = self.offset |
| 126 | remaining = batch_size - len(exp_list) |
| 127 | |
| 128 | async def operation(session: AsyncSession): |
| 129 | stmt = ( |
| 130 | select(self.table_model_cls) |
| 131 | .where(self.table_model_cls.id > current_offset) |
| 132 | .order_by(asc(self.table_model_cls.id)) |
| 133 | .limit(remaining) |
| 134 | ) |
| 135 | result = await session.execute(stmt) |
| 136 | meta_rows = result.scalars().all() |
| 137 | if not meta_rows: |
| 138 | return [], None |
| 139 | ids = [row.id for row in meta_rows] |
| 140 | blob_map = await self._fetch_blobs(session, ids) |
| 141 | return ( |
| 142 | self._assemble_experiences(meta_rows, blob_map), |
| 143 | meta_rows[-1].id, |
| 144 | ) |
| 145 | |
| 146 | experiences, next_offset = await async_run_with_retry_session( |
| 147 | self.session, operation, self.max_retry_times, self.max_retry_interval |
| 148 | ) |
| 149 | if next_offset is not None: |
| 150 | self.offset = next_offset |
| 151 | start_time = time.time() |
| 152 | exp_list.extend(experiences) |
| 153 | if len(exp_list) < batch_size: |
| 154 | self.logger.info(f"Waiting for {batch_size - len(exp_list)} more experiences...") |
| 155 | await asyncio.sleep(1) |
| 156 | return exp_list |
| 157 | |
| 158 | async def _read_priority(self, batch_size: int, min_model_version: int = 0) -> List[Experience]: |
| 159 | exp_list = [] |
nothing calls this directly
no test coverage detected