From 992349b2f7be9049aab6ae74044d978cd807f1c9 Mon Sep 17 00:00:00 2001 From: Yunnglin Date: Mon, 14 Sep 2026 11:00:16 +0800 Subject: [PATCH 1/4] fix: serialize concurrent vllm engine startup --- .../sampler/vllm_sampler/vllm_sampler.py | 28 ++++++++++- tests/sampler/test_vllm_startup_lock.py | 50 +++++++++++++++++++ 2 files changed, 76 insertions(+), 2 deletions(-) create mode 100644 tests/sampler/test_vllm_startup_lock.py diff --git a/src/twinkle/sampler/vllm_sampler/vllm_sampler.py b/src/twinkle/sampler/vllm_sampler/vllm_sampler.py index 7877ddb5..40306911 100644 --- a/src/twinkle/sampler/vllm_sampler/vllm_sampler.py +++ b/src/twinkle/sampler/vllm_sampler/vllm_sampler.py @@ -3,9 +3,11 @@ import atexit import numpy as np import os +import tempfile import threading +from contextlib import contextmanager from copy import copy -from typing import Any, Dict, List, Optional, Type, Union +from typing import Any, Dict, Iterator, List, Optional, Type, Union from twinkle import DeviceMesh, get_logger, remote_class, remote_function, requires from twinkle.checkpoint_engine import CheckpointEngineMixin @@ -19,6 +21,27 @@ logger = get_logger() +@contextmanager +def _vllm_engine_startup_lock(lock_path: str | None = None) -> Iterator[None]: + """Serialize local vLLM engine startup across sampler actors. + + vLLM selects the TCP port for its internal TP workers with a + check-then-bind sequence. Two samplers created concurrently in one pod + can select the same port, causing one engine to fail with ``EADDRINUSE``. + The lock is held only until the engine reports ready and is automatically + released if the actor exits. + """ + import fcntl + + path = lock_path or os.path.join(tempfile.gettempdir(), 'twinkle-vllm-engine-init.lock') + with open(path, 'a+', encoding='utf-8') as lock_file: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + + def _convert_ndarray_to_list(obj: Any) -> Any: if isinstance(obj, np.ndarray): return obj.tolist() @@ -103,7 +126,8 @@ def __init__(self, model_id: str, engine_args: Dict[str, Any] = None, device_mes # Create engine in the background event loop so all async operations # (including vLLM's internal background tasks) run in the same loop - self.engine: VLLMEngine = self._run_in_loop(self._create_engine_async(VLLMEngine, model_id, engine_kwargs)) + with _vllm_engine_startup_lock(): + self.engine: VLLMEngine = self._run_in_loop(self._create_engine_async(VLLMEngine, model_id, engine_kwargs)) # fix: On NPU, monkey_patch_model can trigger Triton compatibility errors and abort sampler init. # fix: Explicitly skip this patch on NPU and keep it for non-NPU paths only. # NPU platform may trigger triton errors with monkey_patch_model diff --git a/tests/sampler/test_vllm_startup_lock.py b/tests/sampler/test_vllm_startup_lock.py new file mode 100644 index 00000000..1cc6e7ba --- /dev/null +++ b/tests/sampler/test_vllm_startup_lock.py @@ -0,0 +1,50 @@ +import multiprocessing +import os + +import pytest + +from twinkle.sampler.vllm_sampler.vllm_sampler import _vllm_engine_startup_lock + + +def _hold_startup_lock(lock_path: str, acquired, release) -> None: + with _vllm_engine_startup_lock(lock_path): + acquired.set() + if not release.wait(timeout=5): + raise TimeoutError('test did not release vLLM startup lock') + + +def _acquire_startup_lock(lock_path: str, acquired) -> None: + with _vllm_engine_startup_lock(lock_path): + acquired.set() + + +@pytest.mark.skipif(os.name == 'nt', reason='vLLM startup lock requires fcntl') +def test_vllm_engine_startup_is_serialized(tmp_path): + """Concurrent local sampler actors must not race vLLM's free-port probe.""" + context = multiprocessing.get_context('spawn') + lock_path = str(tmp_path / 'vllm-engine-init.lock') + first_acquired = context.Event() + release_first = context.Event() + second_acquired = context.Event() + first = context.Process(target=_hold_startup_lock, args=(lock_path, first_acquired, release_first)) + second = context.Process(target=_acquire_startup_lock, args=(lock_path, second_acquired)) + + try: + first.start() + assert first_acquired.wait(timeout=5) + + second.start() + assert not second_acquired.wait(timeout=0.2) + + release_first.set() + assert second_acquired.wait(timeout=5) + finally: + release_first.set() + for process in (first, second): + process.join(timeout=5) + if process.is_alive(): + process.terminate() + process.join(timeout=5) + + assert first.exitcode == 0 + assert second.exitcode == 0 From 9b20ff0c622b28fb7c126658891eee5305a395dc Mon Sep 17 00:00:00 2001 From: Yunnglin Date: Mon, 14 Sep 2026 11:28:25 +0800 Subject: [PATCH 2/4] refactor: reuse shared posix file lock --- src/twinkle/sampler/vllm_sampler/vllm_sampler.py | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/src/twinkle/sampler/vllm_sampler/vllm_sampler.py b/src/twinkle/sampler/vllm_sampler/vllm_sampler.py index 40306911..6e28d809 100644 --- a/src/twinkle/sampler/vllm_sampler/vllm_sampler.py +++ b/src/twinkle/sampler/vllm_sampler/vllm_sampler.py @@ -17,6 +17,7 @@ from twinkle.patch.vllm_lora_weights import VLLMLoraWeights from twinkle.sampler.base import Sampler from twinkle.utils import Platform +from twinkle.utils.parallel import PosixFileLock logger = get_logger() @@ -31,15 +32,9 @@ def _vllm_engine_startup_lock(lock_path: str | None = None) -> Iterator[None]: The lock is held only until the engine reports ready and is automatically released if the actor exits. """ - import fcntl - path = lock_path or os.path.join(tempfile.gettempdir(), 'twinkle-vllm-engine-init.lock') - with open(path, 'a+', encoding='utf-8') as lock_file: - fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) - try: - yield - finally: - fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + with PosixFileLock(path): + yield def _convert_ndarray_to_list(obj: Any) -> Any: From ef7a54400a04357e4818d008bf07d183e1227c17 Mon Sep 17 00:00:00 2001 From: Yunnglin Date: Mon, 14 Sep 2026 11:35:03 +0800 Subject: [PATCH 3/4] refactor: inline vllm startup lock --- .../sampler/vllm_sampler/vllm_sampler.py | 21 ++----------------- tests/sampler/test_vllm_startup_lock.py | 6 +++--- 2 files changed, 5 insertions(+), 22 deletions(-) diff --git a/src/twinkle/sampler/vllm_sampler/vllm_sampler.py b/src/twinkle/sampler/vllm_sampler/vllm_sampler.py index 6e28d809..51db9d21 100644 --- a/src/twinkle/sampler/vllm_sampler/vllm_sampler.py +++ b/src/twinkle/sampler/vllm_sampler/vllm_sampler.py @@ -3,11 +3,9 @@ import atexit import numpy as np import os -import tempfile import threading -from contextlib import contextmanager from copy import copy -from typing import Any, Dict, Iterator, List, Optional, Type, Union +from typing import Any, Dict, List, Optional, Type, Union from twinkle import DeviceMesh, get_logger, remote_class, remote_function, requires from twinkle.checkpoint_engine import CheckpointEngineMixin @@ -22,21 +20,6 @@ logger = get_logger() -@contextmanager -def _vllm_engine_startup_lock(lock_path: str | None = None) -> Iterator[None]: - """Serialize local vLLM engine startup across sampler actors. - - vLLM selects the TCP port for its internal TP workers with a - check-then-bind sequence. Two samplers created concurrently in one pod - can select the same port, causing one engine to fail with ``EADDRINUSE``. - The lock is held only until the engine reports ready and is automatically - released if the actor exits. - """ - path = lock_path or os.path.join(tempfile.gettempdir(), 'twinkle-vllm-engine-init.lock') - with PosixFileLock(path): - yield - - def _convert_ndarray_to_list(obj: Any) -> Any: if isinstance(obj, np.ndarray): return obj.tolist() @@ -121,7 +104,7 @@ def __init__(self, model_id: str, engine_args: Dict[str, Any] = None, device_mes # Create engine in the background event loop so all async operations # (including vLLM's internal background tasks) run in the same loop - with _vllm_engine_startup_lock(): + with PosixFileLock('/tmp/twinkle-vllm-engine-init.lock'): self.engine: VLLMEngine = self._run_in_loop(self._create_engine_async(VLLMEngine, model_id, engine_kwargs)) # fix: On NPU, monkey_patch_model can trigger Triton compatibility errors and abort sampler init. # fix: Explicitly skip this patch on NPU and keep it for non-NPU paths only. diff --git a/tests/sampler/test_vllm_startup_lock.py b/tests/sampler/test_vllm_startup_lock.py index 1cc6e7ba..49d11c75 100644 --- a/tests/sampler/test_vllm_startup_lock.py +++ b/tests/sampler/test_vllm_startup_lock.py @@ -3,18 +3,18 @@ import pytest -from twinkle.sampler.vllm_sampler.vllm_sampler import _vllm_engine_startup_lock +from twinkle.utils.parallel import PosixFileLock def _hold_startup_lock(lock_path: str, acquired, release) -> None: - with _vllm_engine_startup_lock(lock_path): + with PosixFileLock(lock_path): acquired.set() if not release.wait(timeout=5): raise TimeoutError('test did not release vLLM startup lock') def _acquire_startup_lock(lock_path: str, acquired) -> None: - with _vllm_engine_startup_lock(lock_path): + with PosixFileLock(lock_path): acquired.set() From ef6b95088659b53b0e5144d7578467205c78082c Mon Sep 17 00:00:00 2001 From: Yunnglin Date: Mon, 14 Sep 2026 12:34:19 +0800 Subject: [PATCH 4/4] fix: narrow vllm startup lock scope --- src/twinkle/sampler/vllm_sampler/vllm_engine.py | 10 ++++++---- src/twinkle/sampler/vllm_sampler/vllm_sampler.py | 4 +--- tests/sampler/test_vllm_startup_lock.py | 7 +++++-- 3 files changed, 12 insertions(+), 9 deletions(-) diff --git a/src/twinkle/sampler/vllm_sampler/vllm_engine.py b/src/twinkle/sampler/vllm_sampler/vllm_engine.py index 9f444728..27dd4eba 100644 --- a/src/twinkle/sampler/vllm_sampler/vllm_engine.py +++ b/src/twinkle/sampler/vllm_sampler/vllm_engine.py @@ -13,6 +13,7 @@ from twinkle.sampler.base_engine import BaseSamplerEngine from twinkle.utils import Platform from twinkle.utils.framework import Torch +from twinkle.utils.parallel import PosixFileLock from twinkle.utils.zmq_utils import configure_zmq_socket, get_timeout_s_from_env logger = get_logger() @@ -346,10 +347,11 @@ def _create_engine(self): engine_args = AsyncEngineArgs(**filtered_engine_config) vllm_config = engine_args.create_engine_config(usage_context=UsageContext.OPENAI_API_SERVER) - engine = AsyncLLM.from_vllm_config( - vllm_config=vllm_config, - usage_context=UsageContext.OPENAI_API_SERVER, - ) + with PosixFileLock('/tmp/twinkle-vllm-engine-init.lock'): + engine = AsyncLLM.from_vllm_config( + vllm_config=vllm_config, + usage_context=UsageContext.OPENAI_API_SERVER, + ) logger.info(f'VLLMEngine initialized: model={self.model_id}') return engine diff --git a/src/twinkle/sampler/vllm_sampler/vllm_sampler.py b/src/twinkle/sampler/vllm_sampler/vllm_sampler.py index 51db9d21..7877ddb5 100644 --- a/src/twinkle/sampler/vllm_sampler/vllm_sampler.py +++ b/src/twinkle/sampler/vllm_sampler/vllm_sampler.py @@ -15,7 +15,6 @@ from twinkle.patch.vllm_lora_weights import VLLMLoraWeights from twinkle.sampler.base import Sampler from twinkle.utils import Platform -from twinkle.utils.parallel import PosixFileLock logger = get_logger() @@ -104,8 +103,7 @@ def __init__(self, model_id: str, engine_args: Dict[str, Any] = None, device_mes # Create engine in the background event loop so all async operations # (including vLLM's internal background tasks) run in the same loop - with PosixFileLock('/tmp/twinkle-vllm-engine-init.lock'): - self.engine: VLLMEngine = self._run_in_loop(self._create_engine_async(VLLMEngine, model_id, engine_kwargs)) + self.engine: VLLMEngine = self._run_in_loop(self._create_engine_async(VLLMEngine, model_id, engine_kwargs)) # fix: On NPU, monkey_patch_model can trigger Triton compatibility errors and abort sampler init. # fix: Explicitly skip this patch on NPU and keep it for non-NPU paths only. # NPU platform may trigger triton errors with monkey_patch_model diff --git a/tests/sampler/test_vllm_startup_lock.py b/tests/sampler/test_vllm_startup_lock.py index 49d11c75..d07ad74a 100644 --- a/tests/sampler/test_vllm_startup_lock.py +++ b/tests/sampler/test_vllm_startup_lock.py @@ -13,7 +13,8 @@ def _hold_startup_lock(lock_path: str, acquired, release) -> None: raise TimeoutError('test did not release vLLM startup lock') -def _acquire_startup_lock(lock_path: str, acquired) -> None: +def _acquire_startup_lock(lock_path: str, started, acquired) -> None: + started.set() with PosixFileLock(lock_path): acquired.set() @@ -25,15 +26,17 @@ def test_vllm_engine_startup_is_serialized(tmp_path): lock_path = str(tmp_path / 'vllm-engine-init.lock') first_acquired = context.Event() release_first = context.Event() + second_started = context.Event() second_acquired = context.Event() first = context.Process(target=_hold_startup_lock, args=(lock_path, first_acquired, release_first)) - second = context.Process(target=_acquire_startup_lock, args=(lock_path, second_acquired)) + second = context.Process(target=_acquire_startup_lock, args=(lock_path, second_started, second_acquired)) try: first.start() assert first_acquired.wait(timeout=5) second.start() + assert second_started.wait(timeout=5) assert not second_acquired.wait(timeout=0.2) release_first.set()