A high-throughput and memory-efficient inference and serving engine for LLMs
Mirrored from https://github.com/tduncan-lum/Nvidia-Optimization-Stack---VLLM.git
Synced Nov 14, 05:30 UTC
[BugFix] Don't cancel asyncio tasks directly from destructors (#22476)
Signed-off-by: Nick Hill <nhill@redhat.com>
Nick Hill committed
Aug 8, 2025 at 01:13 UTC
ccdae737a0c947467488c05f61537e5658fe5064
3 files changed
+23
-14
vllm/utils/__init__.py
+17
-6
@@ -687,19 +687,30 @@ class AsyncMicrobatchTokenizer:
687
max_length = kwargs.get("max_length")
688
689
if not truncation:
690
- return ("encode", add_special_tokens, False, None)
690
+ return "encode", add_special_tokens, False, None
691
692
model_max = getattr(self.tokenizer, "model_max_length", None)
693
if max_length is None or (model_max is not None
694
and max_length == model_max):
695
- return ("encode", add_special_tokens, True, "model_max")
695
+ return "encode", add_special_tokens, True, "model_max"
696
697
- return ("encode", "other")
697
+ return "encode", "other"
698
699
def __del__(self):
700
- for task in self._batcher_tasks:
701
- if not task.done():
702
- task.cancel()
700
+ if ((tasks := getattr(self, "_batcher_tasks", None))
701
+ and (loop := getattr(self, "_loop", None))
702
+ and not loop.is_closed()):
703
+
704
+ def cancel_tasks():
705
+ for task in tasks:
706
+ task.cancel()
707
+
708
+ loop.call_soon_threadsafe(cancel_tasks)
709
+
710
+
711
+def cancel_task_threadsafe(task: Task):
712
+ if task and not task.done() and not (loop := task.get_loop()).is_closed():
713
+ loop.call_soon_threadsafe(task.cancel)
714
715
716
def make_async(
vllm/v1/engine/async_llm.py
+2
-3
@@ -27,7 +27,7 @@ from vllm.transformers_utils.config import (
27
from vllm.transformers_utils.tokenizer import AnyTokenizer
28
from vllm.transformers_utils.tokenizer_group import init_tokenizer_from_configs
29
from vllm.usage.usage_lib import UsageContext
30
-from vllm.utils import Device, cdiv, deprecate_kwargs
30
+from vllm.utils import Device, cancel_task_threadsafe, cdiv, deprecate_kwargs
31
from vllm.v1.engine import EngineCoreRequest
32
from vllm.v1.engine.core_client import EngineCoreClient
33
from vllm.v1.engine.exceptions import EngineDeadError, EngineGenerateError
@@ -219,8 +219,7 @@ class AsyncLLM(EngineClient):
219
if engine_core := getattr(self, "engine_core", None):
220
engine_core.shutdown()
221
222
- if handler := getattr(self, "output_handler", None):
223
- handler.cancel()
222
+ cancel_task_threadsafe(getattr(self, "output_handler", None))
223
224
async def get_supported_tasks(self) -> tuple[SupportedTask, ...]:
225
return await self.engine_core.get_supported_tasks_async()
vllm/v1/engine/core_client.py
+4
-5
@@ -23,7 +23,8 @@ from vllm.config import VllmConfig
23
from vllm.logger import init_logger
24
from vllm.lora.request import LoRARequest
25
from vllm.tasks import SupportedTask
26
-from vllm.utils import get_open_port, get_open_zmq_inproc_path, make_zmq_socket
26
+from vllm.utils import (cancel_task_threadsafe, get_open_port,
27
+ get_open_zmq_inproc_path, make_zmq_socket)
28
from vllm.v1.engine import (EngineCoreOutputs, EngineCoreRequest,
29
EngineCoreRequestType,
30
ReconfigureDistributedRequest, ReconfigureRankType,
@@ -342,10 +343,8 @@ class BackgroundResources:
343
if self.coordinator is not None:
344
self.coordinator.close()
345
345
- if self.output_queue_task is not None:
346
- self.output_queue_task.cancel()
347
- if self.stats_update_task is not None:
348
- self.stats_update_task.cancel()
346
+ cancel_task_threadsafe(self.output_queue_task)
347
+ cancel_task_threadsafe(self.stats_update_task)
348
349
# ZMQ context termination can hang if the sockets
350
# aren't explicitly closed first.