Unified the resource cleanup for aborted and finished requests

Signed-off-by: 盏一 <zhanyi.ww@alibaba-inc.com>

Author: hidva

vllm/v1/engine/async_llm.py

@@ -331,7 +331,7 @@ async def _run_output_handler(self):
                     # 3) Abort any reqs that finished due to stop strings.
                     await self.engine_core.abort_requests_async(
                         processed_outputs.reqs_to_abort)
-                    self.output_processor.handle_abort_reqs(
+                    self.output_processor.free_aborted_reqs(
                         processed_outputs.reqs_to_abort)
 
                 # 4) Logging.
@@ -355,7 +355,7 @@ async def abort(self, request_id: str) -> None:
         # At this point, the abort message has already been sent to EngineCore,
         # so the request status in the Frontend can be removed.
         # For more details, please see: PR #15326
-        self.output_processor.handle_abort_reqs(request_ids)
+        self.output_processor.free_aborted_reqs(request_ids)
 
         if self.log_requests:
             logger.info("Aborted request %s.", request_id)

vllm/v1/engine/output_processor.py

@@ -240,7 +240,7 @@ def abort_requests(
         request_ids: Iterable[str],
     ) -> list[str]:
         request_ids_to_abort = self.flatten_req_to_abort(request_ids)
-        self.handle_abort_reqs(request_ids_to_abort)
+        self.free_aborted_reqs(request_ids_to_abort)
         return request_ids_to_abort
 
     def flatten_req_to_abort(self, req_ids: Iterable[str]) -> list[str]:
@@ -253,7 +253,7 @@ def flatten_req_to_abort(self, req_ids: Iterable[str]) -> list[str]:
                 ret.extend(parent.child_requests)
         return ret
 
-    def handle_abort_reqs(self, req_ids: Iterable[str]):
+    def free_aborted_reqs(self, req_ids: Iterable[str]):
         """
         Handles aborted requests. This method is triggered when the frontend
         detects that a request has ended, such as when the client disconnects
@@ -263,16 +263,21 @@ def handle_abort_reqs(self, req_ids: Iterable[str]):
             req_state = self.request_states.pop(req_id, None)
             if req_state is not None:
                 self.lora_states.abort_request(req_state)
+        # TODO: handle _update_stats_from_finished here
+        # may need to add some parameters, such as iteration_stats and
+        # finished_reason.
         return
 
-    def finish_request(self, request_id: str) -> None:
+    def free_finised_reqs(self, req_ids: Iterable[str]):
         """
         Handle a finished request. This method is called when EngineCore detects
         that the request has ended, and the resources related to the request
         maintained by EngineCore have been released.
         """
-        req_state = self.request_states.pop(request_id)
-        self.lora_states.finish_request(req_state)
+        for req_id in req_ids:
+            req_state = self.request_states.pop(req_id)
+            self.lora_states.finish_request(req_state)
+        # TODO: handle _update_stats_from_finished here
         return
 
     def add_request(
@@ -385,7 +390,7 @@ def process_outputs(
                     # detected stop string, abort needed in EngineCore.
                     reqs_to_abort.append(req_id)
                 else:
-                    self.finish_request(req_id)
+                    self.free_finised_reqs((req_id, ))
 
                 # Track per-request stats
                 self._update_stats_from_finished(req_state, finish_reason,