From fecb7d05949100a98f5a4ffdaa0da3218da69c1b Mon Sep 17 00:00:00 2001 From: Jie Hao Date: Mon, 26 Jan 2026 17:22:08 -0500 Subject: [PATCH 1/8] feat: request migration for sglang --- components/src/dynamo/sglang/main.py | 106 ++++++++++++------ .../embedding/embedding_handler.py | 4 +- .../sglang/request_handlers/handler_base.py | 57 +++++++++- .../request_handlers/llm/decode_handler.py | 5 +- .../request_handlers/llm/diffusion_handler.py | 9 +- .../request_handlers/llm/prefill_handler.py | 8 +- .../multimodal/encode_worker_handler.py | 8 +- .../multimodal/processor_handler.py | 8 +- .../multimodal/worker_handler.py | 15 ++- 9 files changed, 166 insertions(+), 54 deletions(-) diff --git a/components/src/dynamo/sglang/main.py b/components/src/dynamo/sglang/main.py index 13fa741f7d6f..6d2759d2ea26 100644 --- a/components/src/dynamo/sglang/main.py +++ b/components/src/dynamo/sglang/main.py @@ -51,6 +51,7 @@ async def _handle_non_leader_node( config: SGLang configuration including server args. component: The Dynamo runtime component. generate_endpoint: The Dynamo endpoint for generation requests. + shutdown_event: Event to signal shutdown. """ logging.info( f"Non-leader node detected (node_rank={engine.server_args.node_rank})." @@ -79,34 +80,51 @@ async def worker(): config.dynamo_args.use_kv_events, ) - def signal_handler(): - asyncio.create_task(graceful_shutdown(runtime)) + # Set up signal handlers using signal module to allow chaining + shutdown_event = asyncio.Event() + old_handlers = {} + def signal_handler(signum, frame): + """Handle SIGTERM/SIGINT and chain to previous handlers""" + logging.info(f"Received signal {signum}, initiating graceful shutdown") + # Schedule shutdown in the event loop from the signal handler context + loop.call_soon_threadsafe(shutdown_event.set) + + # Chain to the old handler if it exists and is not default/ignore + old_handler = old_handlers.get(signum) + if old_handler and old_handler not in (signal.SIG_DFL, signal.SIG_IGN): + old_handler(signum, frame) + + runtime.shutdown() + + # Install signal handlers and save old ones for chaining for sig in (signal.SIGTERM, signal.SIGINT): - loop.add_signal_handler(sig, signal_handler) + old_handlers[sig] = signal.signal(sig, signal_handler) - logging.info("Signal handlers will trigger a graceful shutdown of the runtime") + logging.info("Signal handlers set up for graceful shutdown (with chaining)") if config.dynamo_args.embedding_worker: - await init_embedding(runtime, config) + await init_embedding(runtime, config, shutdown_event) elif config.dynamo_args.multimodal_processor: - await init_multimodal_processor(runtime, config) + await init_multimodal_processor(runtime, config, shutdown_event) elif config.dynamo_args.multimodal_encode_worker: - await init_multimodal_encode_worker(runtime, config) + await init_multimodal_encode_worker(runtime, config, shutdown_event) elif config.dynamo_args.multimodal_worker: if config.serving_mode != DisaggregationMode.PREFILL: - await init_multimodal_worker(runtime, config) + await init_multimodal_worker(runtime, config, shutdown_event) else: - await init_multimodal_prefill_worker(runtime, config) + await init_multimodal_prefill_worker(runtime, config, shutdown_event) elif config.dynamo_args.diffusion_worker: - await init_diffusion(runtime, config) + await init_diffusion(runtime, config, shutdown_event) elif config.serving_mode != DisaggregationMode.PREFILL: - await init(runtime, config) + await init(runtime, config, shutdown_event) else: - await init_prefill(runtime, config) + await init_prefill(runtime, config, shutdown_event) -async def init(runtime: DistributedRuntime, config: Config): +async def init( + runtime: DistributedRuntime, config: Config, shutdown_event: asyncio.Event +): server_args, dynamo_args = config.server_args, config.dynamo_args # Prevent SGLang from blocking on non-leader nodes @@ -140,7 +158,7 @@ async def init(runtime: DistributedRuntime, config: Config): ready_event = asyncio.Event() handler = DecodeWorkerHandler( - component, engine, config, publisher, generate_endpoint + component, engine, config, publisher, generate_endpoint, shutdown_event ) handler.register_engine_routes(runtime) @@ -193,7 +211,9 @@ async def init(runtime: DistributedRuntime, config: Config): handler.cleanup() -async def init_prefill(runtime: DistributedRuntime, config: Config): +async def init_prefill( + runtime: DistributedRuntime, config: Config, shutdown_event: asyncio.Event +): server_args, dynamo_args = config.server_args, config.dynamo_args # Prevent SGLang from blocking on non-leader nodes @@ -228,7 +248,7 @@ async def init_prefill(runtime: DistributedRuntime, config: Config): setup_prometheus_registry(engine, generate_endpoint) handler = PrefillWorkerHandler( - component, engine, config, publisher, generate_endpoint + component, engine, config, publisher, generate_endpoint, shutdown_event ) handler.register_engine_routes(runtime) @@ -270,7 +290,9 @@ async def init_prefill(runtime: DistributedRuntime, config: Config): handler.cleanup() -async def init_diffusion(runtime: DistributedRuntime, config: Config): +async def init_diffusion( + runtime: DistributedRuntime, config: Config, shutdown_event: asyncio.Event +): """Initialize diffusion language model worker component""" server_args, dynamo_args = config.server_args, config.dynamo_args @@ -312,7 +334,7 @@ async def init_diffusion(runtime: DistributedRuntime, config: Config): ready_event = asyncio.Event() handler = DiffusionWorkerHandler( - component, engine, config, publisher, generate_endpoint + component, engine, config, publisher, generate_endpoint, shutdown_event ) handler.register_engine_routes(runtime) @@ -355,7 +377,9 @@ async def init_diffusion(runtime: DistributedRuntime, config: Config): handler.cleanup() -async def init_embedding(runtime: DistributedRuntime, config: Config): +async def init_embedding( + runtime: DistributedRuntime, config: Config, shutdown_event: asyncio.Event +): """Initialize embedding worker component""" server_args, dynamo_args = config.server_args, config.dynamo_args @@ -379,7 +403,9 @@ async def init_embedding(runtime: DistributedRuntime, config: Config): # Readiness gate: requests wait until model is registered ready_event = asyncio.Event() - handler = EmbeddingWorkerHandler(component, engine, config, publisher) + handler = EmbeddingWorkerHandler( + component, engine, config, publisher, shutdown_event + ) health_check_payload = SglangHealthCheckPayload( engine, use_text_input=dynamo_args.use_sglang_tokenizer ).to_dict() @@ -417,7 +443,9 @@ async def init_embedding(runtime: DistributedRuntime, config: Config): handler.cleanup() -async def init_multimodal_processor(runtime: DistributedRuntime, config: Config): +async def init_multimodal_processor( + runtime: DistributedRuntime, config: Config, shutdown_event: asyncio.Event +): """Initialize multimodal processor component""" server_args, dynamo_args = config.server_args, config.dynamo_args component = runtime.namespace(dynamo_args.namespace).component( @@ -436,7 +464,9 @@ async def init_multimodal_processor(runtime: DistributedRuntime, config: Config) ready_event = asyncio.Event() - handler = MultimodalProcessorHandler(component, config, encode_worker_client) + handler = MultimodalProcessorHandler( + component, config, encode_worker_client, shutdown_event + ) logging.info("Waiting for Encoder Worker Instances ...") await encode_worker_client.wait_for_instances() @@ -464,7 +494,9 @@ async def init_multimodal_processor(runtime: DistributedRuntime, config: Config) handler.cleanup() -async def init_multimodal_encode_worker(runtime: DistributedRuntime, config: Config): +async def init_multimodal_encode_worker( + runtime: DistributedRuntime, config: Config, shutdown_event: asyncio.Event +): """Initialize multimodal encode worker component""" server_args, dynamo_args = config.server_args, config.dynamo_args @@ -482,7 +514,9 @@ async def init_multimodal_encode_worker(runtime: DistributedRuntime, config: Con .client() ) - handler = MultimodalEncodeWorkerHandler(component, config, pd_worker_client) + handler = MultimodalEncodeWorkerHandler( + component, config, pd_worker_client, shutdown_event + ) await handler.async_init(runtime) await pd_worker_client.wait_for_instances() @@ -502,7 +536,9 @@ async def init_multimodal_encode_worker(runtime: DistributedRuntime, config: Con handler.cleanup() -async def init_multimodal_worker(runtime: DistributedRuntime, config: Config): +async def init_multimodal_worker( + runtime: DistributedRuntime, config: Config, shutdown_event: asyncio.Event +): """Initialize multimodal worker component for aggregated or decode mode""" server_args, dynamo_args = config.server_args, config.dynamo_args @@ -522,9 +558,13 @@ async def init_multimodal_worker(runtime: DistributedRuntime, config: Config): .endpoint("generate") .client() ) - handler = MultimodalWorkerHandler(component, engine, config, prefill_client) + handler = MultimodalWorkerHandler( + component, engine, config, prefill_client, shutdown_event + ) else: - handler = MultimodalWorkerHandler(component, engine, config) + handler = MultimodalWorkerHandler( + component, engine, config, None, shutdown_event + ) await handler.async_init() @@ -565,7 +605,9 @@ async def init_multimodal_worker(runtime: DistributedRuntime, config: Config): handler.cleanup() -async def init_multimodal_prefill_worker(runtime: DistributedRuntime, config: Config): +async def init_multimodal_prefill_worker( + runtime: DistributedRuntime, config: Config, shutdown_event: asyncio.Event +): """Initialize multimodal prefill worker component""" server_args, dynamo_args = config.server_args, config.dynamo_args @@ -577,7 +619,7 @@ async def init_multimodal_prefill_worker(runtime: DistributedRuntime, config: Co generate_endpoint = component.endpoint(dynamo_args.endpoint) - handler = MultimodalPrefillWorkerHandler(component, engine, config) + handler = MultimodalPrefillWorkerHandler(component, engine, config, shutdown_event) await handler.async_init() health_check_payload = SglangPrefillHealthCheckPayload(engine).to_dict() @@ -633,12 +675,6 @@ async def _do_warmup(): logging.warning(f"Prefill warmup failed: {e}") -async def graceful_shutdown(runtime): - logging.info("Received shutdown signal, shutting down DistributedRuntime") - runtime.shutdown() - logging.info("DistributedRuntime shutdown complete") - - def main(): uvloop.run(worker()) diff --git a/components/src/dynamo/sglang/request_handlers/embedding/embedding_handler.py b/components/src/dynamo/sglang/request_handlers/embedding/embedding_handler.py index bb355a11365e..24865f9de3d0 100644 --- a/components/src/dynamo/sglang/request_handlers/embedding/embedding_handler.py +++ b/components/src/dynamo/sglang/request_handlers/embedding/embedding_handler.py @@ -1,6 +1,7 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import asyncio import logging from typing import Optional @@ -20,8 +21,9 @@ def __init__( engine: sgl.Engine, config: Config, publisher: Optional[DynamoSglangPublisher] = None, + shutdown_event: Optional[asyncio.Event] = None, ): - super().__init__(component, engine, config, publisher) + super().__init__(component, engine, config, publisher, None, shutdown_event) logging.info("Embedding worker handler initialized") def cleanup(self): diff --git a/components/src/dynamo/sglang/request_handlers/handler_base.py b/components/src/dynamo/sglang/request_handlers/handler_base.py index db954bf2ae3c..f255cd2d5d86 100644 --- a/components/src/dynamo/sglang/request_handlers/handler_base.py +++ b/components/src/dynamo/sglang/request_handlers/handler_base.py @@ -28,6 +28,7 @@ def __init__( config: Config, publisher: Optional[DynamoSglangPublisher] = None, generate_endpoint=None, + shutdown_event: Optional[asyncio.Event] = None, ) -> None: """Initialize base worker handler. @@ -37,11 +38,13 @@ def __init__( config: SGLang and Dynamo configuration. publisher: Optional metrics publisher for the worker. generate_endpoint: The endpoint handle for discovery registration. + shutdown_event: Optional event to signal shutdown. """ self.component = component self.engine = engine self.config = config self.generate_endpoint = generate_endpoint + self.shutdown_event = shutdown_event if publisher is not None: self.metrics_publisher = publisher.metrics_publisher self.kv_publisher = publisher.kv_publisher @@ -290,12 +293,15 @@ def _get_trace_header(self, context: Context) -> Optional[Dict[str, str]]: async def _handle_cancellation( self, request_id_future: asyncio.Future, context: Context ): - """Background task to handle cancellation by monitoring context state. + """Background task to handle cancellation and shutdown by monitoring both signals. Args: request_id_future: Future that will be set with the SGLang request ID when the first response arrives. context: Context object for cancellation handling. + + Raises: + GeneratorExit: If shutdown event was triggered. """ try: logging.debug(f"Cancellation monitor started for Context: {context.id()}") @@ -307,10 +313,34 @@ async def _handle_cancellation( ) logging.debug(f"Request ID future cancelled for Context: {context.id()}") - await context.async_killed_or_stopped() + # Get the cancellation future + cancellation_future = context.async_killed_or_stopped() + + # Build list of futures/tasks to wait for + wait_for = [cancellation_future] + shutdown_task = None + + if self.shutdown_event: + # Create task for shutdown monitoring and add to wait list + shutdown_task = asyncio.create_task(self.shutdown_event.wait()) + wait_for.append(shutdown_task) + + # Wait for whichever happens first + done, pending = await asyncio.wait( + wait_for, + return_when=asyncio.FIRST_COMPLETED, + ) + + # Cancel the pending task/future + for task in pending: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass logging.info( - f"Cancellation signal received for SGLang Request ID {sglang_request_id}, Context: {context.id()}" + f"Cancellation or shutdown signal received for SGLang Request ID {sglang_request_id}, Context: {context.id()}" ) # Call abort_request on the tokenizer_manager through the engine @@ -329,6 +359,11 @@ async def _handle_cancellation( logging.error( f"SGLang tokenizer_manager not found for abort request: {context.id()}" ) + + # Check which event triggered and raise GeneratorExit if shutdown + if shutdown_task and shutdown_task in done: + raise GeneratorExit("Engine was shut down during token generation") + except asyncio.CancelledError: # Task was cancelled, which is expected when generation completes request_id = "unknown" @@ -347,9 +382,11 @@ async def _cancellation_monitor( self, request_id_future: asyncio.Future, context: Context ) -> AsyncGenerator[asyncio.Task, None]: """ - Context manager for monitoring request cancellation. + Context manager for monitoring request cancellation and shutdown. Automatically creates a background task to monitor for cancellation and - cleans it up when the context exits. + shutdown events, cleaning it up when the context exits. + + If shutdown event was triggered, raises GeneratorExit on exit. Args: request_id_future: Future that will be set with the SGLang request ID @@ -387,6 +424,16 @@ async def _cancellation_monitor( except asyncio.CancelledError: pass else: + # Task completed, check if it was due to shutdown + try: + cancellation_task.result() + except GeneratorExit: + logging.info( + "Decode engine was shut down during token generation DELTEE THIS!!!!!" + ) + raise + except Exception: + pass logging.debug( f"Cancellation monitor task already completed for SGLang Request ID {request_id}, Context: {context.id()}" ) diff --git a/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py b/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py index 959dc8d0a867..377790f1060e 100644 --- a/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py +++ b/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py @@ -4,7 +4,7 @@ import asyncio import logging import time -from typing import Any, AsyncGenerator, Dict +from typing import Any, AsyncGenerator, Dict, Optional import sglang as sgl @@ -24,6 +24,7 @@ def __init__( config: Config, publisher: DynamoSglangPublisher, generate_endpoint=None, + shutdown_event: Optional[asyncio.Event] = None, ) -> None: """Initialize decode worker handler. @@ -32,6 +33,7 @@ def __init__( engine: The SGLang engine instance. config: SGLang and Dynamo configuration. publisher: Metrics publisher for the worker. + shutdown_event: Optional event to signal shutdown. generate_endpoint: The endpoint handle for discovery registration. """ super().__init__( @@ -40,6 +42,7 @@ def __init__( config, publisher, generate_endpoint, + shutdown_event, ) if self.serving_mode == DisaggregationMode.DECODE: logging.info( diff --git a/components/src/dynamo/sglang/request_handlers/llm/diffusion_handler.py b/components/src/dynamo/sglang/request_handlers/llm/diffusion_handler.py index 5e9d4f661281..6096f3d0b30d 100644 --- a/components/src/dynamo/sglang/request_handlers/llm/diffusion_handler.py +++ b/components/src/dynamo/sglang/request_handlers/llm/diffusion_handler.py @@ -1,7 +1,8 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import asyncio import logging -from typing import Any, AsyncGenerator, Dict +from typing import Any, AsyncGenerator, Dict, Optional import sglang as sgl @@ -23,6 +24,7 @@ def __init__( config: Config, publisher: DynamoSglangPublisher = None, generate_endpoint=None, + shutdown_event: Optional[asyncio.Event] = None, ) -> None: """Initialize diffusion worker handler. @@ -32,8 +34,11 @@ def __init__( config: SGLang and Dynamo configuration. publisher: Optional metrics publisher. generate_endpoint: The endpoint handle for discovery. + shutdown_event: Optional event to signal shutdown. """ - super().__init__(component, engine, config, publisher, generate_endpoint) + super().__init__( + component, engine, config, publisher, generate_endpoint, shutdown_event + ) # Validate that diffusion algorithm is configured if ( diff --git a/components/src/dynamo/sglang/request_handlers/llm/prefill_handler.py b/components/src/dynamo/sglang/request_handlers/llm/prefill_handler.py index 7c7dab986392..bbd777f33656 100644 --- a/components/src/dynamo/sglang/request_handlers/llm/prefill_handler.py +++ b/components/src/dynamo/sglang/request_handlers/llm/prefill_handler.py @@ -3,7 +3,7 @@ import asyncio import logging -from typing import Any, AsyncGenerator, Dict +from typing import Any, AsyncGenerator, Dict, Optional import sglang as sgl @@ -23,6 +23,7 @@ def __init__( config: Config, publisher: DynamoSglangPublisher, generate_endpoint=None, + shutdown_event: Optional[asyncio.Event] = None, ) -> None: """Initialize prefill worker handler. @@ -32,10 +33,13 @@ def __init__( config: SGLang and Dynamo configuration. publisher: The SGLang publisher instance. generate_endpoint: The endpoint handle for discovery registration. + shutdown_event: Optional event to signal shutdown. """ self.engine = engine self.bootstrap_host, self.bootstrap_port = self._get_bootstrap_info(self.engine) - super().__init__(component, engine, config, publisher, generate_endpoint) + super().__init__( + component, engine, config, publisher, generate_endpoint, shutdown_event + ) self._consume_tasks = set() logging.info( f"Prefill worker handler initialized - bootstrap host: {self.bootstrap_host}, bootstrap port: {self.bootstrap_port}" diff --git a/components/src/dynamo/sglang/request_handlers/multimodal/encode_worker_handler.py b/components/src/dynamo/sglang/request_handlers/multimodal/encode_worker_handler.py index c30ee4ff010a..2f93bef4c408 100644 --- a/components/src/dynamo/sglang/request_handlers/multimodal/encode_worker_handler.py +++ b/components/src/dynamo/sglang/request_handlers/multimodal/encode_worker_handler.py @@ -1,8 +1,9 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import asyncio import logging -from typing import AsyncIterator +from typing import AsyncIterator, Optional import torch from sglang.srt.parser.conversation import chat_templates @@ -45,8 +46,11 @@ def __init__( component: Component, config: Config, pd_worker_client: Client, + shutdown_event: Optional[asyncio.Event] = None, ) -> None: - super().__init__(component, engine=None, config=config) + super().__init__( + component, engine=None, config=config, shutdown_event=shutdown_event + ) self.pd_worker_client = pd_worker_client self.model = config.server_args.model_path self.served_model_name = config.server_args.served_model_name diff --git a/components/src/dynamo/sglang/request_handlers/multimodal/processor_handler.py b/components/src/dynamo/sglang/request_handlers/multimodal/processor_handler.py index 858e5dd071f6..f83b97b5e437 100644 --- a/components/src/dynamo/sglang/request_handlers/multimodal/processor_handler.py +++ b/components/src/dynamo/sglang/request_handlers/multimodal/processor_handler.py @@ -1,11 +1,12 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import asyncio import json import logging import time import uuid -from typing import Any, Dict +from typing import Any, Dict, Optional from transformers import AutoTokenizer @@ -36,8 +37,11 @@ def __init__( component: Component, config: Config, encode_worker_client: Client, + shutdown_event: Optional[asyncio.Event] = None, ): - super().__init__(component, engine=None, config=config) + super().__init__( + component, engine=None, config=config, shutdown_event=shutdown_event + ) self.encode_worker_client = encode_worker_client self.chat_template = getattr(config.server_args, "chat_template", "qwen2-vl") self.model = config.server_args.model_path diff --git a/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py b/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py index 576169f39492..30e20b5c483c 100644 --- a/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py +++ b/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py @@ -4,7 +4,7 @@ import asyncio import json import logging -from typing import AsyncIterator +from typing import AsyncIterator, Optional import sglang as sgl import torch @@ -236,8 +236,9 @@ def __init__( engine: sgl.Engine, config: Config, prefill_client: Client = None, + shutdown_event: Optional[asyncio.Event] = None, ): - super().__init__(component, engine, config, None) + super().__init__(component, engine, config, None, None, shutdown_event) # Initialize processors self.embeddings_processor = EmbeddingsProcessor() @@ -411,8 +412,14 @@ class MultimodalPrefillWorkerHandler(BaseWorkerHandler): Processes multimodal inputs and coordinates with decode worker. """ - def __init__(self, component: Component, engine: sgl.Engine, config: Config): - super().__init__(component, engine, config) + def __init__( + self, + component: Component, + engine: sgl.Engine, + config: Config, + shutdown_event: Optional[asyncio.Event] = None, + ): + super().__init__(component, engine, config, None, None, shutdown_event) # Initialize processors self.embeddings_processor = EmbeddingsProcessor() From cedb59424aa14934cd883b8bdb9d462a38805484 Mon Sep 17 00:00:00 2001 From: Jie Hao Date: Mon, 26 Jan 2026 17:23:50 -0500 Subject: [PATCH 2/8] update --- components/src/dynamo/sglang/request_handlers/handler_base.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/components/src/dynamo/sglang/request_handlers/handler_base.py b/components/src/dynamo/sglang/request_handlers/handler_base.py index f255cd2d5d86..47cfac1e9ac6 100644 --- a/components/src/dynamo/sglang/request_handlers/handler_base.py +++ b/components/src/dynamo/sglang/request_handlers/handler_base.py @@ -428,9 +428,6 @@ async def _cancellation_monitor( try: cancellation_task.result() except GeneratorExit: - logging.info( - "Decode engine was shut down during token generation DELTEE THIS!!!!!" - ) raise except Exception: pass From b19d96f070c86af4248e1639d0cb864f81403c05 Mon Sep 17 00:00:00 2001 From: Jie Hao Date: Mon, 26 Jan 2026 18:17:06 -0500 Subject: [PATCH 3/8] update --- components/src/dynamo/sglang/main.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/components/src/dynamo/sglang/main.py b/components/src/dynamo/sglang/main.py index 6d2759d2ea26..d24bb4f87e72 100644 --- a/components/src/dynamo/sglang/main.py +++ b/components/src/dynamo/sglang/main.py @@ -91,11 +91,16 @@ def signal_handler(signum, frame): loop.call_soon_threadsafe(shutdown_event.set) # Chain to the old handler if it exists and is not default/ignore - old_handler = old_handlers.get(signum) - if old_handler and old_handler not in (signal.SIG_DFL, signal.SIG_IGN): - old_handler(signum, frame) - - runtime.shutdown() + try: + old_handler = old_handlers.get(signum) + if old_handler and old_handler not in ( + signal.SIG_DFL, + signal.SIG_IGN, + signal.default_int_handler, + ): + old_handler(signum, frame) + finally: + runtime.shutdown() # Install signal handlers and save old ones for chaining for sig in (signal.SIGTERM, signal.SIGINT): From 8733af0ab38d992459376790392404b016ce3364 Mon Sep 17 00:00:00 2001 From: Jie Hao Date: Mon, 26 Jan 2026 19:23:06 -0500 Subject: [PATCH 4/8] update --- .../dynamo/sglang/request_handlers/handler_base.py | 11 +---------- 1 file changed, 1 insertion(+), 10 deletions(-) diff --git a/components/src/dynamo/sglang/request_handlers/handler_base.py b/components/src/dynamo/sglang/request_handlers/handler_base.py index 47cfac1e9ac6..e96efa668595 100644 --- a/components/src/dynamo/sglang/request_handlers/handler_base.py +++ b/components/src/dynamo/sglang/request_handlers/handler_base.py @@ -424,13 +424,4 @@ async def _cancellation_monitor( except asyncio.CancelledError: pass else: - # Task completed, check if it was due to shutdown - try: - cancellation_task.result() - except GeneratorExit: - raise - except Exception: - pass - logging.debug( - f"Cancellation monitor task already completed for SGLang Request ID {request_id}, Context: {context.id()}" - ) + cancellation_task.result() From 79ee486d630327a6750431750cbb785b714c4a01 Mon Sep 17 00:00:00 2001 From: Jie Hao Date: Tue, 3 Feb 2026 17:57:48 -0500 Subject: [PATCH 5/8] update --- .../src/dynamo/common/utils/__init__.py | 10 ++++++-- .../dynamo/common/utils/engine_response.py | 21 +++++++++++++++++ components/src/dynamo/sglang/main.py | 17 ++++++++++++++ .../request_handlers/llm/decode_handler.py | 11 +++++++-- .../multimodal/worker_handler.py | 5 +++- components/src/dynamo/vllm/handlers.py | 23 +++---------------- 6 files changed, 62 insertions(+), 25 deletions(-) create mode 100644 components/src/dynamo/common/utils/engine_response.py diff --git a/components/src/dynamo/common/utils/__init__.py b/components/src/dynamo/common/utils/__init__.py index ad95c1356ac9..95a1f508bc55 100644 --- a/components/src/dynamo/common/utils/__init__.py +++ b/components/src/dynamo/common/utils/__init__.py @@ -14,6 +14,12 @@ - prometheus: Prometheus metrics collection and logging utilities """ -from dynamo.common.utils import endpoint_types, otel_tracing, paths, prometheus +from dynamo.common.utils import ( + endpoint_types, + engine_response, + otel_tracing, + paths, + prometheus, +) -__all__ = ["endpoint_types", "otel_tracing", "paths", "prometheus"] +__all__ = ["endpoint_types", "engine_response", "otel_tracing", "paths", "prometheus"] diff --git a/components/src/dynamo/common/utils/engine_response.py b/components/src/dynamo/common/utils/engine_response.py new file mode 100644 index 000000000000..8caa94d72bb2 --- /dev/null +++ b/components/src/dynamo/common/utils/engine_response.py @@ -0,0 +1,21 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Utilities for engine response processing.""" + +import logging + + +def normalize_finish_reason(finish_reason: str) -> str: + """ + Normalize engine finish reasons to Dynamo-compatible values. + + Engine may return finish reasons that aren't recognized by Dynamo's Rust layer. + This method maps them to compatible values. + [TODO]: Remove this method and add the right code in the Rust layer. + """ + # Map engine's "abort" to Dynamo's "cancelled" + if finish_reason and finish_reason.startswith("abort"): + logging.debug(f"Normalizing finish reason: {finish_reason} to cancelled") + return "cancelled" + return finish_reason diff --git a/components/src/dynamo/sglang/main.py b/components/src/dynamo/sglang/main.py index ef114c4aceef..546ef4916db9 100644 --- a/components/src/dynamo/sglang/main.py +++ b/components/src/dynamo/sglang/main.py @@ -6,6 +6,7 @@ import os import signal import sys +from typing import Any, Callable import sglang as sgl import uvloop @@ -131,6 +132,22 @@ def signal_handler(signum, frame): for sig in (signal.SIGTERM, signal.SIGINT): old_handlers[sig] = signal.signal(sig, signal_handler) + # Override add_signal_handler to prevent SGLang from installing its own signal handlers. + # This ensures that Dynamo retains full control over the graceful shutdown flow, + # rather than allowing SGLang (or other libraries) to overwrite or interfere + # with the signal handlers that Dynamo sets up for SIGTERM/SIGINT. + _orig_add = loop.add_signal_handler + + def watching_add_signal_handler(sig: int, callback: Callable, *args: Any): + if sig in [signal.SIGTERM, signal.SIGINT]: + logging.info( + "SIGTERM/SIGINThandler changed via loop.add_signal_handler() is being suppressed by Dynamo" + ) + else: + return _orig_add(sig, callback, *args) + + loop.add_signal_handler = watching_add_signal_handler # type: ignore[assignment] + logging.info("Signal handlers set up for graceful shutdown (with chaining)") if config.dynamo_args.embedding_worker: diff --git a/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py b/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py index 04a98a353185..9be6bf202004 100644 --- a/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py +++ b/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py @@ -9,6 +9,7 @@ import sglang as sgl from dynamo._core import Component, Context +from dynamo.common.utils.engine_response import normalize_finish_reason from dynamo.sglang.args import Config, DisaggregationMode from dynamo.sglang.publisher import DynamoSglangPublisher from dynamo.sglang.request_handlers.handler_base import BaseWorkerHandler @@ -225,7 +226,9 @@ async def _process_token_stream( out = {} finish_reason = res["meta_info"]["finish_reason"] if finish_reason: - out["finish_reason"] = finish_reason["type"] + out["finish_reason"] = normalize_finish_reason( + finish_reason["type"] + ) # With stream_output=True, output_ids contains only new tokens (disjoint) output_ids = res.get("output_ids", []) @@ -290,7 +293,11 @@ async def _process_text_stream( text = res.get("text", "") finish_reason = res["meta_info"]["finish_reason"] - finish_reason_type = finish_reason["type"] if finish_reason else None + finish_reason_type = ( + normalize_finish_reason(finish_reason["type"]) + if finish_reason + else None + ) next_count = len(text) delta = text[count:] diff --git a/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py b/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py index c29e9bf83217..5101e01e91a2 100644 --- a/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py +++ b/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py @@ -11,6 +11,7 @@ import dynamo.nixl_connect as connect from dynamo._core import Client, Component, Context +from dynamo.common.utils.engine_response import normalize_finish_reason from dynamo.sglang.args import Config, DisaggregationMode from dynamo.sglang.protocol import ( DisaggSglangMultimodalRequest, @@ -153,7 +154,9 @@ async def process_sglang_stream(stream_source) -> AsyncIterator[str]: if finish_reason: output.update( { - "finish_reason": finish_reason.get("type", "stop"), + "finish_reason": normalize_finish_reason( + finish_reason.get("type", "stop") + ), "finished": True, } ) diff --git a/components/src/dynamo/vllm/handlers.py b/components/src/dynamo/vllm/handlers.py index ad83d1503f30..bb71bf4272bb 100644 --- a/components/src/dynamo/vllm/handlers.py +++ b/components/src/dynamo/vllm/handlers.py @@ -22,6 +22,7 @@ from vllm.v1.engine.exceptions import EngineDeadError import dynamo.nixl_connect as nixl_connect +from dynamo.common.utils.engine_response import normalize_finish_reason from dynamo.common.utils.input_params import InputParamManager from dynamo.common.utils.media_nixl import read_decoded_media_via_nixl from dynamo.common.utils.otel_tracing import build_trace_headers @@ -436,20 +437,6 @@ def _get_lora_lock(self, lora_name: str) -> asyncio.Lock: self._lora_load_locks[lora_name] = lock return lock - def _normalize_finish_reason(self, finish_reason: str) -> str: - """ - Normalize vLLM finish reasons to Dynamo-compatible values. - - vLLM may return finish reasons that aren't recognized by Dynamo's Rust layer. - This method maps them to compatible values. - [TODO]: Remove this method and add the right code in the Rust layer. - """ - # Map vLLM's "abort" to Dynamo's "cancelled" - if finish_reason.startswith("abort"): - logging.debug(f"Normalizing finish reason: {finish_reason} to cancelled") - return "cancelled" - return finish_reason - async def load_lora(self, request=None): """ Load a LoRA adapter dynamically into the vLLM's AsyncLLM engine. @@ -1224,9 +1211,7 @@ async def generate_tokens( out["top_logprobs"] = top_logprobs if output.finish_reason: - out["finish_reason"] = self._normalize_finish_reason( - output.finish_reason - ) + out["finish_reason"] = normalize_finish_reason(output.finish_reason) out["completion_usage"] = BaseWorkerHandler._build_completion_usage( request_output=res, embedding_sequence_length=embedding_sequence_length, @@ -1440,9 +1425,7 @@ async def _generate_text_mode(self, request, context, request_id): "role": "assistant", "content": delta_text, }, - "finish_reason": self._normalize_finish_reason( - output.finish_reason - ), + "finish_reason": normalize_finish_reason(output.finish_reason), } chunk = { From 9e77bd914fa01ee0c1df132c0fe5d7563348f646 Mon Sep 17 00:00:00 2001 From: Jie Hao Date: Fri, 6 Feb 2026 22:30:13 -0500 Subject: [PATCH 6/8] update --- components/src/dynamo/sglang/main.py | 150 +++++++++++++++++++-------- 1 file changed, 108 insertions(+), 42 deletions(-) diff --git a/components/src/dynamo/sglang/main.py b/components/src/dynamo/sglang/main.py index c0f8ea7bb359..b915a0ba109e 100644 --- a/components/src/dynamo/sglang/main.py +++ b/components/src/dynamo/sglang/main.py @@ -6,7 +6,8 @@ import os import signal import sys -from typing import Any, Callable +from collections import defaultdict +from typing import Any, Callable, DefaultDict import sglang as sgl import uvloop @@ -83,6 +84,111 @@ async def _handle_non_leader_node( publisher.cleanup() +SignalCallback = Callable[..., Any] + + +def install_graceful_shutdown( + loop: asyncio.AbstractEventLoop, + runtime: Any, + *, + signals: tuple[int, ...] = (signal.SIGTERM, signal.SIGINT), + chain_old_os_handlers: bool = False, +) -> tuple[asyncio.Event, dict[int, list[tuple[SignalCallback, tuple[Any, ...]]]]]: + """ + Set up graceful shutdown + callback chaining. + + What it does: + - Owns OS-level SIGTERM/SIGINT via signal.signal(...) + - Captures (suppresses) loop.add_signal_handler(SIGTERM/SIGINT, ...) registrations + and runs them during shutdown (sync or async) + - Calls runtime.shutdown() during shutdown (sync or async) + - Sets and returns an asyncio.Event you can await to know shutdown was requested + + Returns: + (shutdown_event, deferred_handlers) + """ + shutdown_event = asyncio.Event() + + # Deferred handlers registered via loop.add_signal_handler for these signals + deferred_handlers: DefaultDict[int, list[tuple[SignalCallback, tuple[Any, ...]]]] = defaultdict(list) # type: ignore[assignment] + + # Previous OS handlers (for optional chaining) + old_os_handlers: dict[int, Any] = {} + + shutdown_started = False + + async def _shutdown_sequence(signum: int, frame: Any | None) -> None: + nonlocal shutdown_started + if shutdown_started: + return + shutdown_started = True + + logging.info("Received signal %s, starting graceful shutdown", signum) + shutdown_event.set() + + # Run deferred callbacks in the loop context + for cb, args in list(deferred_handlers.get(signum, [])): + try: + res = cb(*args) + if asyncio.iscoroutine(res): + await res + except Exception: + logging.exception("Deferred signal callback failed: %r", cb) + + # Call runtime.shutdown() (sync or async) + try: + res = runtime.shutdown() + if asyncio.iscoroutine(res): + await res + except Exception: + logging.exception("runtime.shutdown() failed") + + # Optional: chain old OS handler + if chain_old_os_handlers: + old = old_os_handlers.get(signum) + if old and old not in ( + signal.SIG_DFL, + signal.SIG_IGN, + signal.default_int_handler, + ): + try: + old(signum, frame) + except Exception: + logging.exception("Chained old OS handler failed") + + def _schedule_shutdown(signum: int, frame: Any | None) -> None: + def _kick() -> None: + asyncio.create_task(_shutdown_sequence(signum, frame)) + + loop.call_soon_threadsafe(_kick) + + def _os_signal_handler(signum: int, frame: Any) -> None: + # Keep the OS handler tiny; do real work in the loop thread. + _schedule_shutdown(signum, frame) + + # Install OS-level handlers + for sig in signals: + old_os_handlers[sig] = signal.signal(sig, _os_signal_handler) + + # Intercept loop.add_signal_handler for SIGTERM/SIGINT and defer them + orig_add = loop.add_signal_handler + + def watching_add_signal_handler(sig: int, callback: SignalCallback, *args: Any): + if sig in signals: + logging.info( + "Captured loop.add_signal_handler(%s, %r, ...) (deferred).", + sig, + callback, + ) + deferred_handlers[sig].append((callback, args)) + return None + return orig_add(sig, callback, *args) + + loop.add_signal_handler = watching_add_signal_handler # type: ignore[assignment] + + return shutdown_event, deferred_handlers + + async def worker(): config = await parse_args(sys.argv[1:]) dump_config(config.dynamo_args.dump_config_to, config) @@ -113,47 +219,7 @@ async def worker(): ) # Set up signal handlers using signal module to allow chaining - shutdown_event = asyncio.Event() - old_handlers = {} - - def signal_handler(signum, frame): - """Handle SIGTERM/SIGINT and chain to previous handlers""" - logging.info(f"Received signal {signum}, initiating graceful shutdown") - # Schedule shutdown in the event loop from the signal handler context - loop.call_soon_threadsafe(shutdown_event.set) - - # Chain to the old handler if it exists and is not default/ignore - try: - old_handler = old_handlers.get(signum) - if old_handler and old_handler not in ( - signal.SIG_DFL, - signal.SIG_IGN, - signal.default_int_handler, - ): - old_handler(signum, frame) - finally: - runtime.shutdown() - - # Install signal handlers and save old ones for chaining - for sig in (signal.SIGTERM, signal.SIGINT): - old_handlers[sig] = signal.signal(sig, signal_handler) - - # Override add_signal_handler to prevent SGLang from installing its own signal handlers. - # This ensures that Dynamo retains full control over the graceful shutdown flow, - # rather than allowing SGLang (or other libraries) to overwrite or interfere - # with the signal handlers that Dynamo sets up for SIGTERM/SIGINT. - _orig_add = loop.add_signal_handler - - def watching_add_signal_handler(sig: int, callback: Callable, *args: Any): - if sig in [signal.SIGTERM, signal.SIGINT]: - logging.info( - "SIGTERM/SIGINThandler changed via loop.add_signal_handler() is being suppressed by Dynamo" - ) - else: - return _orig_add(sig, callback, *args) - - loop.add_signal_handler = watching_add_signal_handler # type: ignore[assignment] - + shutdown_event, _ = install_graceful_shutdown(loop, runtime) logging.info("Signal handlers set up for graceful shutdown (with chaining)") if config.dynamo_args.image_diffusion_worker: From dafe17ab89564dc62de0914efbc08a25ee7d979c Mon Sep 17 00:00:00 2001 From: Jie Hao Date: Mon, 9 Feb 2026 21:41:32 -0500 Subject: [PATCH 7/8] update --- components/src/dynamo/sglang/main.py | 95 ++++++++++++++++++---------- 1 file changed, 62 insertions(+), 33 deletions(-) diff --git a/components/src/dynamo/sglang/main.py b/components/src/dynamo/sglang/main.py index b915a0ba109e..dbdd49efbb00 100644 --- a/components/src/dynamo/sglang/main.py +++ b/components/src/dynamo/sglang/main.py @@ -2,12 +2,13 @@ # SPDX-License-Identifier: Apache-2.0 import asyncio +import inspect import logging import os import signal import sys from collections import defaultdict -from typing import Any, Callable, DefaultDict +from typing import Any, Awaitable, Callable, DefaultDict import sglang as sgl import uvloop @@ -47,6 +48,8 @@ configure_dynamo_logging() +RUN_DEFERRED_HANDLERS: Callable[[], Awaitable[None]] | None = None + async def _handle_non_leader_node( engine: sgl.Engine, @@ -93,7 +96,7 @@ def install_graceful_shutdown( *, signals: tuple[int, ...] = (signal.SIGTERM, signal.SIGINT), chain_old_os_handlers: bool = False, -) -> tuple[asyncio.Event, dict[int, list[tuple[SignalCallback, tuple[Any, ...]]]]]: +) -> tuple[asyncio.Event, Callable[[], Awaitable[None]]]: """ Set up graceful shutdown + callback chaining. @@ -105,7 +108,7 @@ def install_graceful_shutdown( - Sets and returns an asyncio.Event you can await to know shutdown was requested Returns: - (shutdown_event, deferred_handlers) + (shutdown_event, run_deferred_handlers) """ shutdown_event = asyncio.Event() @@ -116,46 +119,44 @@ def install_graceful_shutdown( old_os_handlers: dict[int, Any] = {} shutdown_started = False + shutdown_signum: int | None = None + deferred_handlers_ran = False + + async def run_deferred_handlers() -> None: + nonlocal deferred_handlers_ran + if not shutdown_started or deferred_handlers_ran: + return + deferred_handlers_ran = True + + signums = ( + [shutdown_signum] + if shutdown_signum is not None + else list(deferred_handlers.keys()) + ) + for sig in signums: + for cb, args in list(deferred_handlers.get(sig, [])): + try: + res = cb(*args) + if inspect.isawaitable(res): + await res + except Exception: + logging.exception("Deferred signal callback failed: %r", cb) async def _shutdown_sequence(signum: int, frame: Any | None) -> None: - nonlocal shutdown_started + nonlocal shutdown_started, shutdown_signum if shutdown_started: return + shutdown_signum = signum shutdown_started = True logging.info("Received signal %s, starting graceful shutdown", signum) shutdown_event.set() - # Run deferred callbacks in the loop context - for cb, args in list(deferred_handlers.get(signum, [])): - try: - res = cb(*args) - if asyncio.iscoroutine(res): - await res - except Exception: - logging.exception("Deferred signal callback failed: %r", cb) - - # Call runtime.shutdown() (sync or async) try: - res = runtime.shutdown() - if asyncio.iscoroutine(res): - await res + runtime.shutdown() except Exception: logging.exception("runtime.shutdown() failed") - # Optional: chain old OS handler - if chain_old_os_handlers: - old = old_os_handlers.get(signum) - if old and old not in ( - signal.SIG_DFL, - signal.SIG_IGN, - signal.default_int_handler, - ): - try: - old(signum, frame) - except Exception: - logging.exception("Chained old OS handler failed") - def _schedule_shutdown(signum: int, frame: Any | None) -> None: def _kick() -> None: asyncio.create_task(_shutdown_sequence(signum, frame)) @@ -186,7 +187,7 @@ def watching_add_signal_handler(sig: int, callback: SignalCallback, *args: Any): loop.add_signal_handler = watching_add_signal_handler # type: ignore[assignment] - return shutdown_event, deferred_handlers + return shutdown_event, run_deferred_handlers async def worker(): @@ -219,7 +220,8 @@ async def worker(): ) # Set up signal handlers using signal module to allow chaining - shutdown_event, _ = install_graceful_shutdown(loop, runtime) + global RUN_DEFERRED_HANDLERS + shutdown_event, RUN_DEFERRED_HANDLERS = install_graceful_shutdown(loop, runtime) logging.info("Signal handlers set up for graceful shutdown (with chaining)") if config.dynamo_args.image_diffusion_worker: @@ -327,9 +329,12 @@ async def init( try: await metrics_task except asyncio.CancelledError: - logging.info("Metrics task succesfully cancelled") + logging.info("Metrics task successfully cancelled") pass handler.cleanup() + if RUN_DEFERRED_HANDLERS is not None: + logging.info("Running deferred handlers") + await RUN_DEFERRED_HANDLERS() async def init_prefill( @@ -410,6 +415,9 @@ async def init_prefill( logging.info("Metrics task successfully cancelled") pass handler.cleanup() + if RUN_DEFERRED_HANDLERS is not None: + logging.info("Running deferred handlers") + await RUN_DEFERRED_HANDLERS() async def init_diffusion( @@ -499,6 +507,9 @@ async def init_diffusion( logging.info("Metrics task successfully cancelled") pass handler.cleanup() + if RUN_DEFERRED_HANDLERS is not None: + logging.info("Running deferred handlers") + await RUN_DEFERRED_HANDLERS() async def init_embedding( @@ -565,6 +576,9 @@ async def init_embedding( logging.info("Metrics task successfully cancelled") pass handler.cleanup() + if RUN_DEFERRED_HANDLERS is not None: + logging.info("Running deferred handlers") + await RUN_DEFERRED_HANDLERS() async def init_image_diffusion(runtime: DistributedRuntime, config: Config): @@ -646,6 +660,9 @@ async def init_image_diffusion(runtime: DistributedRuntime, config: Config): raise finally: handler.cleanup() + if RUN_DEFERRED_HANDLERS is not None: + logging.info("Running deferred handlers") + await RUN_DEFERRED_HANDLERS() async def init_multimodal_processor( @@ -697,6 +714,9 @@ async def init_multimodal_processor( raise finally: handler.cleanup() + if RUN_DEFERRED_HANDLERS is not None: + logging.info("Running deferred handlers") + await RUN_DEFERRED_HANDLERS() async def init_multimodal_encode_worker( @@ -739,6 +759,9 @@ async def init_multimodal_encode_worker( raise finally: handler.cleanup() + if RUN_DEFERRED_HANDLERS is not None: + logging.info("Running deferred handlers") + await RUN_DEFERRED_HANDLERS() async def init_multimodal_worker( @@ -808,6 +831,9 @@ async def init_multimodal_worker( raise finally: handler.cleanup() + if RUN_DEFERRED_HANDLERS is not None: + logging.info("Running deferred handlers") + await RUN_DEFERRED_HANDLERS() async def init_multimodal_prefill_worker( @@ -843,6 +869,9 @@ async def init_multimodal_prefill_worker( raise finally: handler.cleanup() + if RUN_DEFERRED_HANDLERS is not None: + logging.info("Running deferred handlers") + await RUN_DEFERRED_HANDLERS() async def _warmup_prefill_engine(engine: sgl.Engine, server_args) -> None: From 8f1f6daa0abae762bd3552bd52d22607dca3ed8a Mon Sep 17 00:00:00 2001 From: Jie Hao Date: Tue, 10 Feb 2026 13:09:36 -0500 Subject: [PATCH 8/8] update --- components/src/dynamo/sglang/main.py | 1 - 1 file changed, 1 deletion(-) diff --git a/components/src/dynamo/sglang/main.py b/components/src/dynamo/sglang/main.py index dbdd49efbb00..e83dbb660e19 100644 --- a/components/src/dynamo/sglang/main.py +++ b/components/src/dynamo/sglang/main.py @@ -95,7 +95,6 @@ def install_graceful_shutdown( runtime: Any, *, signals: tuple[int, ...] = (signal.SIGTERM, signal.SIGINT), - chain_old_os_handlers: bool = False, ) -> tuple[asyncio.Event, Callable[[], Awaitable[None]]]: """ Set up graceful shutdown + callback chaining.