尝试为当前 token 预占一个生成槽位。
(self, token_id: int, for_image_generation: bool, for_video_generation: bool)
| 87 | return 0, None |
| 88 | |
| 89 | async def _reserve_slot(self, token_id: int, for_image_generation: bool, for_video_generation: bool) -> bool: |
| 90 | """尝试为当前 token 预占一个生成槽位。""" |
| 91 | if not self.concurrency_manager: |
| 92 | return True |
| 93 | |
| 94 | if for_image_generation: |
| 95 | return await self.concurrency_manager.acquire_image(token_id) |
| 96 | |
| 97 | if for_video_generation: |
| 98 | return await self.concurrency_manager.acquire_video(token_id) |
| 99 | |
| 100 | return True |
| 101 | |
| 102 | async def _select_round_robin(self, tokens: list[dict], scenario: str) -> Optional[dict]: |
| 103 | """Select candidate in round-robin order for the given scenario.""" |
no test coverage detected