Pauses the LLM engine and aborts all running/inflight requests. Args: control_request: The control request containing pause command Raises: Exception: If pause is not supported in current configuration Exception: If engine worker queue cleanup tim
(self, control_request: ControlRequest)
| 1284 | self.send_response_server.send_response(request_id, [error_result]) |
| 1285 | |
| 1286 | def _control_pause(self, control_request: ControlRequest): |
| 1287 | """Pauses the LLM engine and aborts all running/inflight requests. |
| 1288 | Args: |
| 1289 | control_request: The control request containing pause command |
| 1290 | |
| 1291 | Raises: |
| 1292 | Exception: If pause is not supported in current configuration |
| 1293 | Exception: If engine worker queue cleanup times out |
| 1294 | |
| 1295 | Returns: |
| 1296 | None |
| 1297 | """ |
| 1298 | |
| 1299 | if not envs.ENABLE_V1_KVCACHE_SCHEDULER: |
| 1300 | raise Exception("pause only supported in ENABLE_V1_KVCACHE_SCHEDULER") |
| 1301 | if self.cfg.scheduler_config.name != "local": |
| 1302 | raise Exception(f"pause only supported in local scheduler, current {self.cfg.scheduler_config.name}") |
| 1303 | |
| 1304 | self.llm_logger.info("Start to pause request generation.") |
| 1305 | |
| 1306 | with self._pause_cond: |
| 1307 | if self.is_paused: |
| 1308 | self.llm_logger.info("Engine is already paused, no need to pause again.") |
| 1309 | return |
| 1310 | self.is_paused = True |
| 1311 | |
| 1312 | self.llm_logger.info("Abort running requests.") |
| 1313 | |
| 1314 | self.resource_manager.log_status() |
| 1315 | # preempted all running reqs. preempted reqs will be append to ResourceManager.waiting queue |
| 1316 | timeout, count = 60, 0 |
| 1317 | while self.engine_worker_queue.exist_tasks(): |
| 1318 | time.sleep(0.001) |
| 1319 | count += 1 |
| 1320 | if count >= timeout * 1000: |
| 1321 | break |
| 1322 | if count >= timeout * 1000: |
| 1323 | error_msg = f"Emptying engine worker queue timed out after {timeout} seconds, worker may hanged!" |
| 1324 | self.llm_logger.error(error_msg) |
| 1325 | raise Exception(error_msg) |
| 1326 | running_reqs = self.resource_manager.preempted_all() |
| 1327 | if len(running_reqs) > 0: |
| 1328 | self.llm_logger.info(f"Total {len(running_reqs)} requests need to be aborted.") |
| 1329 | self.resource_manager.get_real_bsz() |
| 1330 | self.engine_worker_queue.put_tasks((running_reqs, self.resource_manager.real_bsz)) |
| 1331 | self.resource_manager.wait_worker_inflight_requests_finish(timeout=60) |
| 1332 | # self.engine_worker_queue.clear_data() |
| 1333 | self.token_processor.clear_data() |
| 1334 | self.resource_manager.log_status() |
| 1335 | |
| 1336 | # abort inflight requests to user |
| 1337 | inflight_requests = self.scheduler.get_inflight_requests() |
| 1338 | self.llm_logger.info(f"Abort inflight requests (total {len(inflight_requests)}).") |
| 1339 | for req in inflight_requests: |
| 1340 | self._send_error_response(req.request_id, "Request is aborted since engine is paused.") |
| 1341 | self.scheduler.reset() |
| 1342 | |
| 1343 | # pause cache transfer |
no test coverage detected