fix: /pause_generation with --tokenizer-worker-num > 1

python/sglang/srt/entrypoints/http_server.py

@@ -145,6 +145,7 @@
 from sglang.srt.managers.multi_tokenizer_mixin import (
     MultiTokenizerRouter,
     TokenizerWorker,
+    create_shared_pause_flag,
     get_main_process_id,
     monkey_patch_uvicorn_multiprocessing,
     read_from_shared_memory,
@@ -602,7 +603,6 @@ async def model_info():
         "has_audio_understanding": model_config.is_audio_understandable_model,
         "model_type": getattr(model_config.hf_config, "model_type", None),
         "architectures": getattr(model_config.hf_config, "architectures", None),
-        "weight_version": _global_state.tokenizer_manager.server_args.weight_version,
         # "hf_config": model_config.hf_config.to_dict(),
     }
     return result
@@ -1946,7 +1946,7 @@ def _execute_server_warmup(server_args: ServerArgs):
             _global_state.tokenizer_manager.server_status = ServerStatus.Up
 
         else:
-            logger.info(f"Start of pd disaggregation warmup ...")
+            logger.info("Start of pd disaggregation warmup ...")
             json_data = {
                 "sampling_params": {
                     "temperature": 0.0,
@@ -2188,6 +2188,7 @@ def _setup_and_run_http_server(
         multi_tokenizer_args_shm = write_data_for_multi_tokenizer(
             port_args, server_args, scheduler_infos[0]
         )
+        pause_flag_shm = create_shared_pause_flag(os.getpid())
 
     try:
         # Update logging configs
@@ -2289,6 +2290,8 @@ async def _run_with_ssl_refresh():
         if server_args.tokenizer_worker_num > 1:
             if multi_tokenizer_args_shm is not None:
                 multi_tokenizer_args_shm.unlink()
+            pause_flag_shm.close()
+            pause_flag_shm.unlink()
             if _global_state is not None:
                 _global_state.tokenizer_manager.socket_mapping.clear_all_sockets()
 

python/sglang/srt/managers/multi_tokenizer_mixin.py

@@ -404,6 +404,9 @@ def __init__(
             self.send_to_scheduler, 2
         )
 
+        main_pid = get_main_process_id()
+        self._init_shared_pause(get_shared_pause_shm_name(main_pid))
+
     def _attach_multi_http_worker_info(self, req: Union[BaseReq, BaseBatchReq]):
 
         if isinstance(req, BaseReq):
@@ -477,6 +480,28 @@ def read_from_shared_memory(name: str) -> Any:
         raise FileNotFoundError(f"Shared memory {name} not found")
 
 
+SHARED_PAUSE_SHM_PREFIX = "sglang_pause_"
+
+
+def get_shared_pause_shm_name(pid: int) -> str:
+    return f"{SHARED_PAUSE_SHM_PREFIX}{pid}"
+
+
+def create_shared_pause_flag(pid: int) -> shared_memory.SharedMemory:
+    """Returns a SharedMemory object, which the caller must close and unlink on shutdown."""
+    name = get_shared_pause_shm_name(pid)
+    try:
+        shm = shared_memory.SharedMemory(name=name)
+        if shm.size < 1:
+            shm.close()
+            shm.unlink()
+            shm = shared_memory.SharedMemory(create=True, size=1, name=name)
+    except FileNotFoundError:
+        shm = shared_memory.SharedMemory(create=True, size=1, name=name)
+    shm.buf[0] = 0
+    return shm
+
+
 def write_data_for_multi_tokenizer(
     port_args: PortArgs, server_args: ServerArgs, scheduler_info: Dict
 ):

python/sglang/srt/managers/tokenizer_control_mixin.py

@@ -425,15 +425,8 @@ async def update_weights_from_distributed(
         if obj.abort_all_requests:
             self.abort_request(abort_all=True)
 
-        # Hold is_pause_cond while updating to prevent unpause from racing.
-        async with self.is_pause_cond:
-            is_paused = self.is_pause
-            if is_paused:
-                results = await self.update_weights_from_distributed_communicator(obj)
-
-        if not is_paused:
-            async with self.model_update_lock.writer_lock:
-                results = await self.update_weights_from_distributed_communicator(obj)
+        async with self._ensure_paused_or_model_locked():
+            results = await self.update_weights_from_distributed_communicator(obj)
 
         success, message = FanOutCommunicator.merge_results(results)
         if success and obj.weight_version is not None:
@@ -483,14 +476,8 @@ async def update_weights_from_tensor(
         if obj.abort_all_requests:
             self.abort_request(abort_all=True)
 
-        async with self.is_pause_cond:
-            is_paused = self.is_pause
-            if is_paused:
-                results = await self.update_weights_from_tensor_communicator(obj)
-
-        if not is_paused:
-            async with self.model_update_lock.writer_lock:
-                results = await self.update_weights_from_tensor_communicator(obj)
+        async with self._ensure_paused_or_model_locked():
+            results = await self.update_weights_from_tensor_communicator(obj)
 
         success, message = FanOutCommunicator.merge_results(results)
         if success and obj.weight_version is not None:
@@ -512,17 +499,9 @@ async def update_weights_from_ipc(
                 self.server_args.dp_size == 1 or self.server_args.enable_dp_attention
             ), "dp_size must be 1 or dp attention must be enabled for update weights from IPC"
             logger.info("Starting IPC weight update")
-
-            async with self.is_pause_cond:
-                is_paused = self.is_pause
-                if is_paused:
-                    result = (await self.update_weights_from_ipc_communicator(obj))[0]
-                    success, message = result.success, result.message
-
-            if not is_paused:
-                async with self.model_update_lock.writer_lock:
-                    result = (await self.update_weights_from_ipc_communicator(obj))[0]
-                    success, message = result.success, result.message
+            async with self._ensure_paused_or_model_locked():
+                result = (await self.update_weights_from_ipc_communicator(obj))[0]
+                success, message = result.success, result.message
         except Exception as e:
             error_msg = f"IPC weight update failed: {str(e)}"
             logger.error(error_msg)

python/sglang/srt/managers/tokenizer_manager.py

@@ -25,10 +25,11 @@
 import sys
 import threading
 from collections import deque
-from contextlib import nullcontext
+from contextlib import asynccontextmanager, nullcontext
 from datetime import datetime
 from enum import Enum
 from http import HTTPStatus
+from multiprocessing import shared_memory
 from typing import Any, Awaitable, Dict, List, Optional, Tuple, Union
 
 import fastapi
@@ -106,7 +107,7 @@
     get_or_create_event_loop,
     kill_process_tree,
 )
-from sglang.srt.utils.aio_rwlock import RWLock
+from sglang.srt.utils.aio_rwlock import RWCondition, RWLock
 from sglang.srt.utils.hf_transformers_utils import (
     get_processor,
     get_tokenizer,
@@ -415,7 +416,99 @@ def init_weight_update(self):
             None
         )
         self.is_pause = False
-        self.is_pause_cond = asyncio.Condition()
+        self.is_pause_cond = RWCondition()
+        self._is_pause_shm = None
+
+    def _init_shared_pause(self, shm_name: str):
+        """Attach to shared pause flag for multi-worker coordination.
+
+        In multi-tokenizer mode, each worker process has its own is_pause flag
+        and RWCondition, which cannot be shared across processes. This method
+        connects to a shared memory byte so that pause/continue from any
+        worker is visible to all workers. A background task polls the shared
+        flag and updates the local is_pause + is_pause_cond. Note that the
+        consistency of is_pause is reduced to eventual consistency in this
+        case.
+        """
+        self._is_pause_shm = shared_memory.SharedMemory(name=shm_name)
+        self.is_pause = bool(self._is_pause_shm.buf[0])
+        self.asyncio_tasks.add(asyncio.create_task(self._poll_shared_pause()))
+
+    async def _poll_shared_pause(self):
+        """Background task that polls shared memory and syncs local is_pause."""
+        while True:
+            await asyncio.sleep(0.1)
+            if bool(self._is_pause_shm.buf[0]) != self.is_pause:
+                async with self.is_pause_cond.writer_lock:
+                    shm_paused = bool(self._is_pause_shm.buf[0])
+                    if shm_paused != self.is_pause:
+                        self.is_pause = shm_paused
+                        self.is_pause_cond.notify_all()
+
+    @asynccontextmanager
+    async def _ensure_paused_or_model_locked(self):
+        """
+        Context manager: within the context, either the engine is paused OR the
+        model_update_lock writer is held.
+
+        Acquires is_pause_cond as a reader, which blocks writers
+        (continue_generation, pause_generation) but not other readers
+        (send_request).  This prevents continue_generation from unpausing
+        while a weight update is in-flight (TOCTOU fix from #22304).
+
+        NOTE: with --tokenizer-worker-num > 1, is_pause_cond is process-local
+        so it cannot block continue_generation in another worker process.
+
+        NOTE: with --tokenizer-worker-num > 1, model_update_lock is
+        process-local so it cannot block model read/update in another worker
+        process.
+        """
+        await self.is_pause_cond.acquire_reader()
+        pause_reader_held = True
+        try:
+            if self.is_pause:
+                yield
+                return
+
+            async def wait_until_paused():
+                await self.is_pause_cond.wait_for_reader(lambda: self.is_pause)
+
+            lock_task = asyncio.create_task(self.model_update_lock.acquire_writer())
+            pause_task = asyncio.create_task(wait_until_paused())
+            lock_acquired = False
+            try:
+                _, pending = await asyncio.wait(
+                    [lock_task, pause_task], return_when=asyncio.FIRST_COMPLETED
+                )
+                for t in pending:
+                    t.cancel()
+                for t in pending:
+                    try:
+                        await t
+                    except asyncio.CancelledError:
+                        pass
+                if lock_task.done() and not lock_task.cancelled():
+                    await lock_task
+                    lock_acquired = True
+                if pause_task.done() and not pause_task.cancelled():
+                    await pause_task
+                if lock_acquired:
+                    await self.is_pause_cond.release_reader()
+                    pause_reader_held = False
+                yield
+            finally:
+                for t in (lock_task, pause_task):
+                    if not t.done():
+                        t.cancel()
+                        try:
+                            await t
+                        except asyncio.CancelledError:
+                            pass
+                if lock_acquired:
+                    await self.model_update_lock.release_writer()
+        finally:
+            if pause_reader_held:
+                await self.is_pause_cond.release_reader()
 
     def init_lora(self):
         # LoRA
@@ -543,8 +636,8 @@ async def generate_request(
         # Log the request
         self.request_logger.log_received_request(obj, self.tokenizer, request)
 
-        async with self.is_pause_cond:
-            await self.is_pause_cond.wait_for(lambda: not self.is_pause)
+        async with self.is_pause_cond.reader_lock:
+            await self.is_pause_cond.wait_for_reader(lambda: not self.is_pause)
 
         async with self.model_update_lock.reader_lock:
             await self._validate_and_resolve_lora(obj)
@@ -1468,24 +1561,29 @@ def abort_request(self, rid: str = "", abort_all: bool = False):
             )
 
     async def pause_generation(self, obj: PauseGenerationReqInput):
-        async with self.is_pause_cond:
+        async with self.is_pause_cond.writer_lock:
             self.is_pause = True
+            if self._is_pause_shm is not None:
+                self._is_pause_shm.buf[0] = 1
+            self.is_pause_cond.notify_all()
             if obj.mode != "abort":
-                await self.send_to_scheduler.send_pyobj(obj)
+                self.send_to_scheduler.send_pyobj(obj)
             else:
                 # we are using the model_update_lock to check if there is still on-going requests.
                 while True:
                     # TODO: maybe make it async instead of fire-and-forget
                     self.abort_request(abort_all=True)
-                    is_locked = await self.model_update_lock.is_locked()
+                    is_locked = self.model_update_lock.is_locked()
                     if not is_locked:
                         break
                     await asyncio.sleep(1.0)
 
     async def continue_generation(self, obj: ContinueGenerationReqInput):
-        async with self.is_pause_cond:
+        async with self.is_pause_cond.writer_lock:
             self.is_pause = False
-            await self.send_to_scheduler.send_pyobj(obj)
+            if self._is_pause_shm is not None:
+                self._is_pause_shm.buf[0] = 0
+            self.send_to_scheduler.send_pyobj(obj)
             self.is_pause_cond.notify_all()
 
     async def update_weights_from_disk(
@@ -1503,14 +1601,7 @@ async def update_weights_from_disk(
         if obj.abort_all_requests:
             self.abort_request(abort_all=True)
 
-        # Immediately update the weights if the engine is in paused state
-        async with self.is_pause_cond:
-            is_paused = self.is_pause
-
-        lock_context = (
-            self.model_update_lock.writer_lock if not is_paused else nullcontext()
-        )
-        async with lock_context:
+        async with self._ensure_paused_or_model_locked():
             success, message, num_paused_requests = (
                 await self._wait_for_model_update_from_disk(obj)
             )