Skip to content

GitLab

Unverified Commit c6b440e4 authored by Keiven C's avatar Keiven C Committed by GitHub
Browse files

test: add dynamic port allocation for fault_tolerant test execution (#4835) 


Signed-off-by: default avatarKeiven Chang <keivenchang@users.noreply.github.com>
Co-authored-by: default avatarKeiven Chang <keivenchang@users.noreply.github.com>
parent 111e08c5
Show whitespace changes
Inline Side-by-side
Showing with 1104 additions and 234 deletions
tests/conftest.py
View file @ c6b440e4
# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
import os
......@@ -26,6 +14,14 @@ from filelock import FileLock
from tests.utils.constants import TEST_MODELS
from tests.utils.managed_process import ManagedProcess
from tests.utils.port_utils import (
allocate_port,
allocate_ports,
deallocate_port,
deallocate_ports,
)
_logger = logging.getLogger(__name__)
def pytest_configure(config):
......@@ -249,43 +245,118 @@ def pytest_runtestloop(session):
class EtcdServer(ManagedProcess):
def __init__(self, request, port=2379, timeout=300):
# Allocate free ports if port is 0
use_random_port = port == 0
if use_random_port:
# Need two ports: client port and peer port for parallel execution
# Start from 2380 (etcd default 2379 + 1)
port, peer_port = allocate_ports(2, 2380)
else:
peer_port = None
self.port = port
self.peer_port = peer_port # Store for cleanup
self.use_random_port = use_random_port # Track if we allocated the port
port_string = str(port)
etcd_env = os.environ.copy()
etcd_env["ALLOW_NONE_AUTHENTICATION"] = "yes"
data_dir = tempfile.mkdtemp(prefix="etcd_")
command = [
"etcd",
"--listen-client-urls",
f"http://0.0.0.0:{port_string}",
"--advertise-client-urls",
f"http://0.0.0.0:{port_string}",
]
# Add peer port configuration only for random ports (parallel execution)
if peer_port is not None:
peer_port_string = str(peer_port)
command.extend(
[
"--listen-peer-urls",
f"http://0.0.0.0:{peer_port_string}",
"--initial-advertise-peer-urls",
f"http://localhost:{peer_port_string}",
"--initial-cluster",
f"default=http://localhost:{peer_port_string}",
]
)
command.extend(
[
"--data-dir",
data_dir,
]
)
super().__init__(
env=etcd_env,
command=command,
timeout=timeout,
display_output=False,
terminate_existing=not use_random_port, # Disabled for parallel test execution with random ports
health_check_ports=[port],
data_dir=data_dir,
log_dir=request.node.name,
)
def __exit__(self, exc_type, exc_val, exc_tb):
"""Release allocated ports when server exits."""
try:
# Only deallocate ports that were dynamically allocated (not default ports)
if self.use_random_port:
ports_to_release = [self.port]
if self.peer_port is not None:
ports_to_release.append(self.peer_port)
deallocate_ports(ports_to_release)
except Exception as e:
logging.warning(f"Failed to release EtcdServer port: {e}")
return super().__exit__(exc_type, exc_val, exc_tb)
class NatsServer(ManagedProcess):
def __init__(self, request, port=4222, timeout=300):
# Allocate a free port if port is 0
use_random_port = port == 0
if use_random_port:
# Start from 4223 (nats-server default 4222 + 1)
port = allocate_port(4223)
self.port = port
self.use_random_port = use_random_port # Track if we allocated the port
data_dir = tempfile.mkdtemp(prefix="nats_")
command = ["nats-server", "-js", "--trace", "--store_dir", data_dir]
command = [
"nats-server",
"-js",
"--trace",
"--store_dir",
data_dir,
"-p",
str(port),
]
super().__init__(
command=command,
timeout=timeout,
display_output=False,
terminate_existing=not use_random_port, # Disabled for parallel test execution with random ports
data_dir=data_dir,
health_check_ports=[port],
log_dir=request.node.name,
)
def __exit__(self, exc_type, exc_val, exc_tb):
"""Release allocated port when server exits."""
try:
# Only deallocate ports that were dynamically allocated (not default ports)
if self.use_random_port:
deallocate_port(self.port)
except Exception as e:
logging.warning(f"Failed to release NatsServer port: {e}")
return super().__exit__(exc_type, exc_val, exc_tb)
class SharedManagedProcess:
"""Base class for ManagedProcess with file-based reference counting for multi-process sharing."""
......@@ -445,7 +516,10 @@ def runtime_services(request, store_kv, request_plane):
- If store_kv != "etcd", etcd is not started (returns None)
- If request_plane != "nats", NATS is not started (returns None)
Returns a tuple of (nats_process, etcd_process) where each has a .port attribute.
"""
# Port cleanup is now handled in NatsServer and EtcdServer __exit__ methods
if request_plane == "nats" and store_kv == "etcd":
with NatsServer(request) as nats_process:
with EtcdServer(request) as etcd_process:
......@@ -460,6 +534,49 @@ def runtime_services(request, store_kv, request_plane):
yield None, None
@pytest.fixture()
def runtime_services_dynamic_ports(request, store_kv, request_plane):
"""Provide NATS and Etcd servers with truly dynamic ports per test.
This fixture actually allocates dynamic ports by passing port=0 to the servers.
It also sets the NATS_SERVER and ETCD_ENDPOINTS environment variables so that
Dynamo processes can find the services on the dynamic ports.
- If store_kv != "etcd", etcd is not started (returns None)
- If request_plane != "nats", NATS is not started (returns None)
Returns a tuple of (nats_process, etcd_process) where each has a .port attribute.
"""
import os
# Port cleanup is now handled in NatsServer and EtcdServer __exit__ methods
if request_plane == "nats" and store_kv == "etcd":
with NatsServer(request, port=0) as nats_process:
with EtcdServer(request, port=0) as etcd_process:
# Set environment variables for Rust/Python runtime to use. Note that xdist (parallel execution)
# will launch isolated tests in a new process, so no need to worry about environment pollution.
os.environ["NATS_SERVER"] = f"nats://localhost:{nats_process.port}"
os.environ["ETCD_ENDPOINTS"] = f"http://localhost:{etcd_process.port}"
yield nats_process, etcd_process
# No test should rely on these variables after the test, but clean up just in case.
os.environ.pop("NATS_SERVER", None)
os.environ.pop("ETCD_ENDPOINTS", None)
elif request_plane == "nats":
with NatsServer(request, port=0) as nats_process:
os.environ["NATS_SERVER"] = f"nats://localhost:{nats_process.port}"
yield nats_process, None
os.environ.pop("NATS_SERVER", None)
elif store_kv == "etcd":
with EtcdServer(request, port=0) as etcd_process:
os.environ["ETCD_ENDPOINTS"] = f"http://localhost:{etcd_process.port}"
yield None, etcd_process
os.environ.pop("ETCD_ENDPOINTS", None)
else:
yield None, None
@pytest.fixture(scope="session")
def runtime_services_session(request, tmp_path_factory):
"""Session-scoped fixture that provides shared NATS and etcd instances for all tests.
......
tests/fault_tolerance/cancellation/test_sglang.py
View file @ c6b440e4
# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
"""
Test Execution Times (Last Run: 2025-12-09):
- test_request_cancellation_sglang_aggregated: ~46s (gpu_1)
- test_request_cancellation_sglang_decode_cancel: ~60s (gpu_2, estimate)
- Total: 46.06s (0:00:46) for aggregated test only
"""
import logging
import os
import shutil
......@@ -15,9 +22,9 @@ from tests.fault_tolerance.cancellation.utils import (
send_cancellable_request,
)
from tests.utils.constants import FAULT_TOLERANCE_MODEL_NAME
from tests.utils.engine_process import FRONTEND_PORT
from tests.utils.managed_process import ManagedProcess
from tests.utils.payloads import check_health_generate, check_models_api
from tests.utils.port_utils import allocate_port, deallocate_port
logger = logging.getLogger(__name__)
......@@ -32,12 +39,20 @@ pytestmark = [
class DynamoWorkerProcess(ManagedProcess):
"""Process manager for Dynamo worker with SGLang backend"""
def __init__(self, request, mode: str = "agg"):
def __init__(
self,
request,
system_port: int,
frontend_port: int,
mode: str = "agg",
):
"""
Initialize SGLang worker process.
Args:
request: pytest request object
system_port: Port for system metrics server
frontend_port: Port where frontend is running
mode: One of "agg", "prefill", "decode"
"""
command = [
......@@ -66,7 +81,7 @@ class DynamoWorkerProcess(ManagedProcess):
"--disaggregation-mode",
mode,
"--disaggregation-bootstrap-port",
"12345",
"12345", # TODO: use dynamic port allocation
"--host",
"0.0.0.0",
"--disaggregation-transfer-backend",
......@@ -74,20 +89,19 @@ class DynamoWorkerProcess(ManagedProcess):
]
)
# Configure health check based on worker type
if mode in ["prefill", "decode"]:
# Prefill and decode workers check their own status endpoint
health_check_urls = [
(f"http://localhost:{FRONTEND_PORT}/v1/models", check_models_api),
(f"http://localhost:{FRONTEND_PORT}/health", check_health_generate),
(f"http://localhost:{system_port}/health", self.is_ready)
]
else:
# Aggregated workers check both system status and frontend
health_check_urls = [
(f"http://localhost:{system_port}/health", self.is_ready),
(f"http://localhost:{frontend_port}/v1/models", check_models_api),
(f"http://localhost:{frontend_port}/health", check_health_generate),
]
# Set port based on worker type
if mode == "prefill":
port = "8082"
health_check_urls = [(f"http://localhost:{port}/health", self.is_ready)]
elif mode == "decode":
port = "8081"
health_check_urls = [(f"http://localhost:{port}/health", self.is_ready)]
else: # agg (aggregated mode)
port = "8081"
# Set environment variables
env = os.environ.copy()
......@@ -100,7 +114,8 @@ class DynamoWorkerProcess(ManagedProcess):
# intermittent failures
env["DYN_HEALTH_CHECK_ENABLED"] = "false"
env["DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS"] = '["generate"]'
env["DYN_SYSTEM_PORT"] = port
env["DYN_SYSTEM_PORT"] = str(system_port)
env["DYN_HTTP_PORT"] = str(frontend_port)
# Set GPU assignment for disaggregated mode (like disagg.sh)
if mode == "decode":
......@@ -138,10 +153,17 @@ class DynamoWorkerProcess(ManagedProcess):
)
self.mode = mode
self.system_port = system_port
def __exit__(self, exc_type, exc_val, exc_tb):
"""Release allocated port when worker exits."""
try:
# system_port is a required parameter, always set in __init__
deallocate_port(self.system_port)
except Exception as e:
logging.warning(f"Failed to release SGLang worker port: {e}")
def get_pid(self):
"""Get the PID of the worker process"""
return self.proc.pid if self.proc else None
return super().__exit__(exc_type, exc_val, exc_tb)
def is_ready(self, response) -> bool:
"""Check the health of the worker process"""
......@@ -164,7 +186,9 @@ class DynamoWorkerProcess(ManagedProcess):
@pytest.mark.gpu_1
@pytest.mark.xfail(strict=False)
@pytest.mark.parametrize("request_plane", ["nats", "tcp"], indirect=True)
def test_request_cancellation_sglang_aggregated(request, runtime_services):
def test_request_cancellation_sglang_aggregated(
request, runtime_services_dynamic_ports
):
"""
End-to-end test for request cancellation functionality in aggregated mode.
......@@ -172,16 +196,35 @@ def test_request_cancellation_sglang_aggregated(request, runtime_services):
the system properly handles the cancellation and cleans up resources
on the worker side in aggregated (agg) mode.
Tests 3 cancellation scenarios:
1. Completion request
2. Chat completion request
3. Chat completion request (streaming)
Timing (Last Run: 2025-12-09): ~46s total
- Engine initialization: ~14s
- Testing 3 scenarios: ~30s (~10s each)
- Teardown: ~2s
TODO: Test is currently flaky/failing due to SGLang limitations with prefill cancellation.
See: https://github.com/sgl-project/sglang/issues/11139
"""
logger.info("Sanity check if latest test is getting executed")
# Step 1: Start the frontend
# Allocate ports to avoid conflicts with parallel tests
system_port = allocate_port(9100)
# Step 1: Start the frontend (allocates its own port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start an aggregated worker
with DynamoWorkerProcess(request, mode="agg") as worker:
with DynamoWorkerProcess(
request,
system_port=system_port,
frontend_port=frontend.frontend_port,
mode="agg",
) as worker:
logger.info(f"Aggregated Worker PID: {worker.get_pid()}")
# TODO: Why wait after worker ready fixes frontend 404 / 500 flakiness?
time.sleep(2)
......@@ -202,7 +245,9 @@ def test_request_cancellation_sglang_aggregated(request, runtime_services):
logger.info(f"Testing {description.lower()}...")
# Send the request (non-blocking)
cancellable_req = send_cancellable_request(request_type)
cancellable_req = send_cancellable_request(
frontend.frontend_port, request_type
)
# Poll for "New Request ID" pattern (Dynamo context ID)
request_id, worker_log_offset = poll_for_pattern(
......@@ -259,7 +304,9 @@ def test_request_cancellation_sglang_aggregated(request, runtime_services):
],
indirect=True,
)
def test_request_cancellation_sglang_decode_cancel(request, runtime_services):
def test_request_cancellation_sglang_decode_cancel(
request, runtime_services_dynamic_ports
):
"""
End-to-end test for request cancellation during decode phase.
......@@ -268,18 +315,37 @@ def test_request_cancellation_sglang_decode_cancel(request, runtime_services):
on both the prefill and decode workers in a disaggregated setup.
Note: This test requires 2 GPUs to run decode and prefill workers on separate GPUs.
Timing (Last Run: 2025-12-09): ~60s total (estimated)
- Engine initialization: ~20s (decode + prefill workers)
- Testing stream cancellation during decode: ~38s
- Teardown: ~2s
"""
# Step 1: Start the frontend
# Allocate ports to avoid conflicts with parallel tests
decode_system_port = allocate_port(9100)
prefill_system_port = allocate_port(9200)
# Step 1: Start the frontend (allocates its own port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start the decode worker
with DynamoWorkerProcess(request, mode="decode") as decode_worker:
with DynamoWorkerProcess(
request,
system_port=decode_system_port,
frontend_port=frontend.frontend_port,
mode="decode",
) as decode_worker:
logger.info(f"Decode Worker PID: {decode_worker.get_pid()}")
# Step 3: Start the prefill worker
with DynamoWorkerProcess(request, mode="prefill") as prefill_worker:
with DynamoWorkerProcess(
request,
system_port=prefill_system_port,
frontend_port=frontend.frontend_port,
mode="prefill",
) as prefill_worker:
logger.info(f"Prefill Worker PID: {prefill_worker.get_pid()}")
# TODO: Why wait after worker ready fixes frontend 404 / 500 flakiness?
......@@ -291,7 +357,9 @@ def test_request_cancellation_sglang_decode_cancel(request, runtime_services):
)
# Send streaming request (non-blocking)
cancellable_req = send_cancellable_request("chat_completion_stream")
cancellable_req = send_cancellable_request(
frontend.frontend_port, "chat_completion_stream"
)
# Poll for "New Request ID" pattern in decode worker (Dynamo context ID)
request_id, decode_log_offset = poll_for_pattern(
......
tests/fault_tolerance/cancellation/test_trtllm.py
View file @ c6b440e4
# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
"""
Test Execution Times (Last Run: 2025-12-09):
- test_request_cancellation_trtllm_aggregated: ~45s (gpu_1)
- test_request_cancellation_trtllm_decode_cancel: ~115s (gpu_1)
- test_request_cancellation_trtllm_prefill_cancel: ~115s (gpu_1)
- test_request_cancellation_trtllm_kv_transfer_cancel: ~115s (gpu_1, xfail)
- Total: ~390s (0:06:30)
"""
import logging
import os
import shutil
......@@ -15,9 +24,9 @@ from tests.fault_tolerance.cancellation.utils import (
send_cancellable_request,
)
from tests.utils.constants import FAULT_TOLERANCE_MODEL_NAME
from tests.utils.engine_process import FRONTEND_PORT
from tests.utils.managed_process import ManagedProcess
from tests.utils.payloads import check_health_generate, check_models_api
from tests.utils.port_utils import allocate_port, deallocate_port
logger = logging.getLogger(__name__)
......@@ -33,14 +42,24 @@ pytestmark = [
class DynamoWorkerProcess(ManagedProcess):
"""Process manager for Dynamo worker with TensorRT-LLM backend"""
def __init__(self, request, mode: str = "prefill_and_decode"):
def __init__(
self,
request,
frontend_port: int,
mode: str = "prefill_and_decode",
):
"""
Initialize TensorRT-LLM worker process.
Args:
request: pytest request object
frontend_port: Port for the frontend server
mode: One of "prefill_and_decode", "prefill", "decode"
"""
# Allocate system port for this worker
system_port = allocate_port(9100)
self.system_port = system_port
self.frontend_port = frontend_port
# Prefill workers require migration_limit=0 (no KV cache migration support)
migration_limit = "0" if mode == "prefill" else "3"
......@@ -71,19 +90,15 @@ class DynamoWorkerProcess(ManagedProcess):
]
health_check_urls = [
(f"http://localhost:{FRONTEND_PORT}/v1/models", check_models_api),
(f"http://localhost:{FRONTEND_PORT}/health", check_health_generate),
(f"http://localhost:{frontend_port}/v1/models", check_models_api),
(f"http://localhost:{frontend_port}/health", check_health_generate),
]
# Set port based on worker type
if mode == "prefill":
port = "8082"
health_check_urls = [(f"http://localhost:{port}/health", self.is_ready)]
elif mode == "decode":
port = "8081"
health_check_urls = [(f"http://localhost:{port}/health", self.is_ready)]
else: # prefill_and_decode
port = "8081"
# Set health check based on worker type
if mode in ["prefill", "decode"]:
health_check_urls = [
(f"http://localhost:{system_port}/health", self.is_ready)
]
# Set environment variables
env = os.environ.copy()
......@@ -96,7 +111,7 @@ class DynamoWorkerProcess(ManagedProcess):
# intermittent failures
env["DYN_HEALTH_CHECK_ENABLED"] = "false"
env["DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS"] = '["generate"]'
env["DYN_SYSTEM_PORT"] = port
env["DYN_SYSTEM_PORT"] = str(system_port)
# Set log directory based on worker type
log_dir = f"{request.node.name}_{mode}_worker"
......@@ -121,10 +136,6 @@ class DynamoWorkerProcess(ManagedProcess):
self.mode = mode
def get_pid(self):
"""Get the PID of the worker process"""
return self.proc.pid if self.proc else None
def is_ready(self, response) -> bool:
"""Check the health of the worker process"""
try:
......@@ -141,24 +152,47 @@ class DynamoWorkerProcess(ManagedProcess):
)
return False
def __exit__(self, exc_type, exc_val, exc_tb):
"""Release allocated port when worker exits."""
try:
# system_port is always allocated in __init__
deallocate_port(self.system_port)
except Exception as e:
logging.warning(f"Failed to release TRT-LLM worker port: {e}")
return super().__exit__(exc_type, exc_val, exc_tb)
@pytest.mark.timeout(140) # 3x average
@pytest.mark.parametrize("request_plane", ["nats", "tcp"], indirect=True)
def test_request_cancellation_trtllm_aggregated(request, runtime_services):
def test_request_cancellation_trtllm_aggregated(
request, runtime_services_dynamic_ports
):
"""
End-to-end test for request cancellation functionality in aggregated mode.
This test verifies that when a request is cancelled by the client,
the system properly handles the cancellation and cleans up resources
on the worker side in aggregated (prefill_and_decode) mode.
on the worker side in aggregated (prefill_and_decode) mode. Tests three scenarios:
1. Completion request
2. Chat completion request (non-streaming)
3. Chat completion request (streaming)
Timing (Last Run: 2025-12-09): ~45s total
- Engine initialization: ~27s (frontend + worker)
- Testing 3 scenarios: ~15s (~5s each)
- Teardown: ~3s
"""
# Step 1: Start the frontend
# Step 1: Start the frontend (allocates its own frontend_port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start an aggregated worker
with DynamoWorkerProcess(request, mode="prefill_and_decode") as worker:
# Step 2: Start a single worker (allocates its own system_port)
with DynamoWorkerProcess(
request, frontend.frontend_port, mode="prefill_and_decode"
) as worker:
logger.info(f"Aggregated Worker PID: {worker.get_pid()}")
# TODO: Why wait after worker ready fixes frontend 404 / 500 flakiness?
......@@ -180,7 +214,9 @@ def test_request_cancellation_trtllm_aggregated(request, runtime_services):
logger.info(f"Testing {description.lower()}...")
# Send the request (non-blocking)
cancellable_req = send_cancellable_request(request_type)
cancellable_req = send_cancellable_request(
frontend.frontend_port, request_type
)
# Poll for "New Request ID" pattern
request_id, worker_log_offset = poll_for_pattern(
......@@ -227,25 +263,36 @@ def test_request_cancellation_trtllm_aggregated(request, runtime_services):
],
indirect=True,
)
def test_request_cancellation_trtllm_decode_cancel(request, runtime_services):
def test_request_cancellation_trtllm_decode_cancel(
request, runtime_services_dynamic_ports
):
"""
End-to-end test for request cancellation during decode phase with unified frontend.
This test verifies that when a request is cancelled by the client during the decode phase,
the system properly handles the cancellation and cleans up resources
on the decode worker side in a disaggregated setup.
Timing (Last Run: 2025-12-09): ~115s total (2 workers at 45% GPU each)
- Engine initialization: ~92s (frontend: 2s, prefill worker: 45s, decode worker: 45s sequential)
- Testing stream cancellation during decode: ~20s
- Teardown: ~3s
"""
# Step 1: Start the frontend
# Step 1: Start the frontend (allocates its own frontend_port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start the prefill worker
with DynamoWorkerProcess(request, mode="prefill") as prefill_worker:
# Step 2: Start the prefill worker (allocates its own system_port)
with DynamoWorkerProcess(
request, frontend.frontend_port, mode="prefill"
) as prefill_worker:
logger.info(f"Prefill Worker PID: {prefill_worker.get_pid()}")
# Step 3: Start the decode worker
with DynamoWorkerProcess(request, mode="decode") as decode_worker:
# Step 3: Start the decode worker (allocates its own system_port)
with DynamoWorkerProcess(
request, frontend.frontend_port, mode="decode"
) as decode_worker:
logger.info(f"Decode Worker PID: {decode_worker.get_pid()}")
# TODO: Why wait after worker ready fixes frontend 404 / 500 flakiness?
......@@ -257,7 +304,9 @@ def test_request_cancellation_trtllm_decode_cancel(request, runtime_services):
)
# Send streaming request (non-blocking)
cancellable_req = send_cancellable_request("chat_completion_stream")
cancellable_req = send_cancellable_request(
frontend.frontend_port, "chat_completion_stream"
)
# Poll for "Prefill Request ID" pattern in prefill worker (frontend routes here first)
request_id, prefill_log_offset = poll_for_pattern(
......@@ -309,25 +358,36 @@ def test_request_cancellation_trtllm_decode_cancel(request, runtime_services):
],
indirect=True,
)
def test_request_cancellation_trtllm_prefill_cancel(request, runtime_services):
def test_request_cancellation_trtllm_prefill_cancel(
request, runtime_services_dynamic_ports
):
"""
End-to-end test for request cancellation during prefill phase with unified frontend.
This test verifies that when a request is cancelled by the client during the prefill phase,
the system properly handles the cancellation and cleans up resources on the prefill worker.
Since the request is cancelled before prefill completes, the decode worker never receives it.
Timing (Last Run: 2025-12-09): ~115s total (2 workers at 45% GPU each)
- Engine initialization: ~92s (frontend: 2s, prefill worker: 45s, decode worker: 45s sequential)
- Testing cancellation during prefill: ~20s
- Teardown: ~3s
"""
# Step 1: Start the frontend
# Step 1: Start the frontend (allocates its own frontend_port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start the prefill worker
with DynamoWorkerProcess(request, mode="prefill") as prefill_worker:
# Step 2: Start the prefill worker (allocates its own system_port)
with DynamoWorkerProcess(
request, frontend.frontend_port, mode="prefill"
) as prefill_worker:
logger.info(f"Prefill Worker PID: {prefill_worker.get_pid()}")
# Step 3: Start the decode worker
with DynamoWorkerProcess(request, mode="decode") as decode_worker:
# Step 3: Start the decode worker (allocates its own system_port)
with DynamoWorkerProcess(
request, frontend.frontend_port, mode="decode"
) as decode_worker:
logger.info(f"Decode Worker PID: {decode_worker.get_pid()}")
# TODO: Why wait after worker ready fixes frontend 404 / 500 flakiness?
......@@ -340,7 +400,7 @@ def test_request_cancellation_trtllm_prefill_cancel(request, runtime_services):
# Send request with long prompt (non-blocking)
cancellable_req = send_cancellable_request(
"completion", use_long_prompt=True
frontend.frontend_port, "completion", use_long_prompt=True
)
# Poll for "Prefill Request ID" pattern in prefill worker (frontend routes here first)
......@@ -395,24 +455,35 @@ def test_request_cancellation_trtllm_prefill_cancel(request, runtime_services):
reason="May fail due to unknown reason with TRT-LLM or backend implementation",
strict=False,
)
def test_request_cancellation_trtllm_kv_transfer_cancel(request, runtime_services):
def test_request_cancellation_trtllm_kv_transfer_cancel(
request, runtime_services_dynamic_ports
):
"""
End-to-end test for request cancellation during prefill to decode KV transfer phase.
This test verifies that when a request is cancelled by the client during the KV transfer phase,
the system properly handles the cancellation and cleans up resources on the workers.
Timing (Last Run: 2025-12-09): ~115s total (2 workers at 45% GPU each)
- Engine initialization: ~92s (frontend: 2s, prefill worker: 45s, decode worker: 45s sequential)
- Testing KV transfer cancellation: ~20s
- Teardown: ~3s
"""
# Step 1: Start the frontend
# Step 1: Start the frontend (allocates its own frontend_port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start the prefill worker
with DynamoWorkerProcess(request, mode="prefill") as prefill_worker:
# Step 2: Start the prefill worker (allocates its own system_port)
with DynamoWorkerProcess(
request, frontend.frontend_port, mode="prefill"
) as prefill_worker:
logger.info(f"Prefill Worker PID: {prefill_worker.get_pid()}")
# Step 3: Start the decode worker
with DynamoWorkerProcess(request, mode="decode") as decode_worker:
# Step 3: Start the decode worker (allocates its own system_port)
with DynamoWorkerProcess(
request, frontend.frontend_port, mode="decode"
) as decode_worker:
logger.info(f"Decode Worker PID: {decode_worker.get_pid()}")
# TODO: Why wait after worker ready fixes frontend 404 / 500 flakiness?
......@@ -425,7 +496,7 @@ def test_request_cancellation_trtllm_kv_transfer_cancel(request, runtime_service
# Send request with long prompt
cancellable_req = send_cancellable_request(
"completion", use_long_prompt=True
frontend.frontend_port, "completion", use_long_prompt=True
)
# Poll for "Prefill Request ID" pattern in prefill worker
......@@ -466,7 +537,9 @@ def test_request_cancellation_trtllm_kv_transfer_cancel(request, runtime_service
)
# Verify the workers are still functional
cancellable_req = send_cancellable_request("chat_completion_stream")
cancellable_req = send_cancellable_request(
frontend.frontend_port, "chat_completion_stream"
)
_, decode_log_offset = poll_for_pattern(
process=decode_worker,
pattern="Decode Request ID: ",
......
tests/fault_tolerance/cancellation/test_vllm.py
View file @ c6b440e4
# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
"""
Test Execution Times (Last Run: 2025-12-09):
- test_request_cancellation_vllm_aggregated: ~55s (gpu_1)
- test_request_cancellation_vllm_decode_cancel: ~53s (gpu_2)
- test_request_cancellation_vllm_prefill_cancel: ~53s (gpu_2)
- Total: 161.65s (0:02:41)
"""
import logging
import os
import shutil
......@@ -14,9 +22,9 @@ from tests.fault_tolerance.cancellation.utils import (
send_cancellable_request,
)
from tests.utils.constants import FAULT_TOLERANCE_MODEL_NAME
from tests.utils.engine_process import FRONTEND_PORT
from tests.utils.managed_process import ManagedProcess
from tests.utils.payloads import check_health_generate, check_models_api
from tests.utils.port_utils import allocate_port, deallocate_port
logger = logging.getLogger(__name__)
......@@ -32,7 +40,17 @@ pytestmark = [
class DynamoWorkerProcess(ManagedProcess):
"""Process manager for Dynamo worker with vLLM backend"""
def __init__(self, request, is_prefill: bool = False):
def __init__(
self,
request,
frontend_port: int,
is_prefill: bool = False,
):
# Allocate system port for this worker
system_port = allocate_port(9100)
self.system_port = system_port
self.frontend_port = frontend_port
command = [
"python3",
"-m",
......@@ -48,21 +66,20 @@ class DynamoWorkerProcess(ManagedProcess):
"3",
]
# Set port based on worker type
port = "8082" if is_prefill else "8081"
# Configure health check based on worker type
if is_prefill:
# Prefill workers check their own status endpoint
command.append("--is-prefill-worker")
health_check_urls = [(f"http://localhost:{port}/health", self.is_ready)]
health_check_urls = [
(f"http://localhost:{system_port}/health", self.is_ready)
]
else:
# Decode workers should also check their own status endpoint first,
# then verify the frontend sees the model
health_check_urls = [
(f"http://localhost:{port}/health", self.is_ready),
(f"http://localhost:{FRONTEND_PORT}/v1/models", check_models_api),
(f"http://localhost:{FRONTEND_PORT}/health", check_health_generate),
(f"http://localhost:{system_port}/health", self.is_ready),
(f"http://localhost:{frontend_port}/v1/models", check_models_api),
(f"http://localhost:{frontend_port}/health", check_health_generate),
]
# Set environment variables
......@@ -76,13 +93,16 @@ class DynamoWorkerProcess(ManagedProcess):
# intermittent failures
env["DYN_HEALTH_CHECK_ENABLED"] = "false"
env["DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS"] = '["generate"]'
env["DYN_SYSTEM_PORT"] = port
env["DYN_SYSTEM_PORT"] = str(system_port)
env["DYN_HTTP_PORT"] = str(frontend_port)
# Set KV event port and NIXL side channel port only for prefill worker
# to avoid conflicts with decode worker
if is_prefill:
env["DYN_VLLM_KV_EVENT_PORT"] = "20082"
env["VLLM_NIXL_SIDE_CHANNEL_PORT"] = "5601"
env["DYN_VLLM_KV_EVENT_PORT"] = "20082" # TODO: use dynamic port allocation
env[
"VLLM_NIXL_SIDE_CHANNEL_PORT"
] = "5601" # TODO: use dynamic port allocation
# Set log directory based on worker type
worker_type = "prefill_worker" if is_prefill else "worker"
......@@ -119,6 +139,16 @@ class DynamoWorkerProcess(ManagedProcess):
"""Get the PID of the worker process"""
return self.proc.pid if self.proc else None
def __exit__(self, exc_type, exc_val, exc_tb):
"""Release allocated port when worker exits."""
try:
# system_port is always allocated in __init__
deallocate_port(self.system_port)
except Exception as e:
logging.warning(f"Failed to release vLLM worker port: {e}")
return super().__exit__(exc_type, exc_val, exc_tb)
def is_ready(self, response) -> bool:
"""Check the health of the worker process"""
try:
......@@ -137,7 +167,7 @@ class DynamoWorkerProcess(ManagedProcess):
@pytest.mark.timeout(110) # 3x average
@pytest.mark.parametrize("request_plane", ["nats", "tcp"], indirect=True)
def test_request_cancellation_vllm_aggregated(request, runtime_services):
def test_request_cancellation_vllm_aggregated(request, runtime_services_dynamic_ports):
"""
End-to-end test for request cancellation functionality in aggregated mode.
......@@ -147,14 +177,19 @@ def test_request_cancellation_vllm_aggregated(request, runtime_services):
1. Completion request
2. Chat completion request (non-streaming)
3. Chat completion request (streaming)
Timing (Last Run: 2025-12-09): ~55s total
- Engine initialization: ~15s
- Testing 3 scenarios: ~38s (~12s each)
- Teardown: ~2s
"""
# Step 1: Start the frontend
# Step 1: Start the frontend (allocates its own frontend_port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start a single worker
with DynamoWorkerProcess(request) as worker:
# Step 2: Start a single worker (allocates its own system_port)
with DynamoWorkerProcess(request, frontend.frontend_port) as worker:
logger.info(f"Worker PID: {worker.get_pid()}")
# Step 3: Test request cancellation with polling approach
......@@ -173,7 +208,9 @@ def test_request_cancellation_vllm_aggregated(request, runtime_services):
logger.info(f"Testing {description.lower()}...")
# Send the request (non-blocking)
cancellable_req = send_cancellable_request(request_type)
cancellable_req = send_cancellable_request(
frontend.frontend_port, request_type
)
# Poll for "Decode Request ID" pattern (vLLM v2 pattern)
request_id, worker_log_offset = poll_for_pattern(
......@@ -221,7 +258,7 @@ def test_request_cancellation_vllm_aggregated(request, runtime_services):
indirect=True,
)
def test_request_cancellation_vllm_decode_cancel(
request, runtime_services, set_ucx_tls_no_mm
request, runtime_services_dynamic_ports, set_ucx_tls_no_mm
):
"""
End-to-end test for request cancellation during decode phase.
......@@ -229,18 +266,27 @@ def test_request_cancellation_vllm_decode_cancel(
This test verifies that when a request is cancelled by the client during the decode phase,
the system properly handles the cancellation and cleans up resources
on the decode worker side in a disaggregated setup.
Timing (Last Run: 2025-12-09): ~53s total (requires 2 GPUs)
- Engine initialization: ~23s (decode + prefill workers)
- Testing stream cancellation during decode: ~28s
- Teardown: ~2s
"""
# Step 1: Start the frontend
# Step 1: Start the frontend (allocates its own frontend_port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start the prefill worker
with DynamoWorkerProcess(request, is_prefill=True) as prefill_worker:
# Step 2: Start the prefill worker (allocates its own system_port)
with DynamoWorkerProcess(
request, frontend.frontend_port, is_prefill=True
) as prefill_worker:
logger.info(f"Prefill Worker PID: {prefill_worker.get_pid()}")
# Step 3: Start the decode worker
with DynamoWorkerProcess(request, is_prefill=False) as decode_worker:
# Step 3: Start the decode worker (allocates its own system_port)
with DynamoWorkerProcess(
request, frontend.frontend_port, is_prefill=False
) as decode_worker:
logger.info(f"Decode Worker PID: {decode_worker.get_pid()}")
# Step 4: Test request cancellation for streaming scenario
......@@ -249,7 +295,9 @@ def test_request_cancellation_vllm_decode_cancel(
)
# Send streaming request (non-blocking)
cancellable_req = send_cancellable_request("chat_completion_stream")
cancellable_req = send_cancellable_request(
frontend.frontend_port, "chat_completion_stream"
)
# Poll for "Decode Request ID" pattern in decode worker (vLLM v2 pattern)
request_id, decode_log_offset = poll_for_pattern(
......@@ -302,7 +350,7 @@ def test_request_cancellation_vllm_decode_cancel(
indirect=True,
)
def test_request_cancellation_vllm_prefill_cancel(
request, runtime_services, set_ucx_tls_no_mm
request, runtime_services_dynamic_ports, set_ucx_tls_no_mm
):
"""
End-to-end test for request cancellation during prefill phase.
......@@ -310,18 +358,27 @@ def test_request_cancellation_vllm_prefill_cancel(
This test verifies that when a request is cancelled by the client during the prefill phase,
the system properly handles the cancellation and cleans up resources
on both the decode and prefill workers in a disaggregated setup.
Timing (Last Run: 2025-12-09): ~53s total (requires 2 GPUs)
- Engine initialization: ~23s (decode + prefill workers)
- Testing cancellation during prefill: ~28s
- Teardown: ~2s
"""
# Step 1: Start the frontend
# Step 1: Start the frontend (allocates its own frontend_port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start the prefill worker
with DynamoWorkerProcess(request, is_prefill=True) as prefill_worker:
# Step 2: Start the prefill worker (allocates its own system_port)
with DynamoWorkerProcess(
request, frontend.frontend_port, is_prefill=True
) as prefill_worker:
logger.info(f"Prefill Worker PID: {prefill_worker.get_pid()}")
# Step 3: Start the decode worker
with DynamoWorkerProcess(request, is_prefill=False) as decode_worker:
# Step 3: Start the decode worker (allocates its own system_port)
with DynamoWorkerProcess(
request, frontend.frontend_port, is_prefill=False
) as decode_worker:
logger.info(f"Decode Worker PID: {decode_worker.get_pid()}")
# Step 4: Test request cancellation during prefill phase
......@@ -333,7 +390,7 @@ def test_request_cancellation_vllm_prefill_cancel(
# Send request with long prompt (non-blocking)
cancellable_req = send_cancellable_request(
"completion", use_long_prompt=True
frontend.frontend_port, "completion", use_long_prompt=True
)
# Poll for "Prefill Request ID" pattern in prefill worker (vLLM v2 pattern)
......
tests/fault_tolerance/cancellation/utils.py
View file @ c6b440e4
......@@ -8,14 +8,14 @@ import shutil
import socket
import threading
import time
from typing import Any, Callable, Dict
from typing import Any, Callable, Dict, cast
import pytest
import requests
from tests.utils.constants import FAULT_TOLERANCE_MODEL_NAME
from tests.utils.engine_process import FRONTEND_PORT
from tests.utils.managed_process import ManagedProcess
from tests.utils.port_utils import allocate_port, deallocate_port
logger = logging.getLogger(__name__)
......@@ -24,7 +24,11 @@ class DynamoFrontendProcess(ManagedProcess):
"""Process manager for Dynamo frontend"""
def __init__(self, request):
command = ["python", "-m", "dynamo.frontend"]
# Allocate frontend port
frontend_port = allocate_port(8100)
self.frontend_port = frontend_port
command = ["python", "-m", "dynamo.frontend", "--http-port", str(frontend_port)]
env = os.environ.copy()
env["DYN_REQUEST_PLANE"] = request.getfixturevalue("request_plane")
......@@ -51,10 +55,20 @@ class DynamoFrontendProcess(ManagedProcess):
command=command,
env=env,
display_output=True,
terminate_existing=True,
terminate_existing=False, # Don't terminate other processes of the same name, we'll only terminate our own PID
log_dir=log_dir,
)
def __exit__(self, exc_type, exc_val, exc_tb):
"""Release allocated port when frontend exits."""
try:
# frontend_port is always allocated in __init__
deallocate_port(self.frontend_port)
except Exception as e:
logger.warning(f"Failed to release frontend port: {e}")
return super().__exit__(exc_type, exc_val, exc_tb)
class CancellableRequest:
"""A wrapper for a single request that can be explicitly cancelled.
......@@ -174,12 +188,15 @@ class CancellableRequest:
return self.response
def send_completion_request(prompt: str, max_tokens: int) -> CancellableRequest:
def send_completion_request(
prompt: str, max_tokens: int, frontend_port: int
) -> CancellableRequest:
"""Send a completion request to the frontend
Args:
prompt: The prompt for completion
max_tokens: Maximum tokens to generate
frontend_port: Port where the frontend is running
Returns:
A CancellableRequest object that can be explicitly cancelled
......@@ -199,7 +216,7 @@ def send_completion_request(prompt: str, max_tokens: int) -> CancellableRequest:
# Return a cancellable request object
cancellable_req = CancellableRequest()
cancellable_req.post(
f"http://localhost:{FRONTEND_PORT}/v1/completions",
f"http://localhost:{frontend_port}/v1/completions",
headers=headers,
json=payload,
)
......@@ -207,13 +224,14 @@ def send_completion_request(prompt: str, max_tokens: int) -> CancellableRequest:
def send_chat_completion_request(
prompt: str, max_tokens: int, stream: bool = False
prompt: str, max_tokens: int, frontend_port: int, stream: bool = False
) -> CancellableRequest:
"""Send a chat completion request to the frontend
Args:
prompt: The prompt for chat completion
max_tokens: Maximum tokens to generate
frontend_port: Port where the frontend is running
stream: Whether to stream the response
Returns:
......@@ -235,7 +253,7 @@ def send_chat_completion_request(
# Return a cancellable request object
cancellable_req = CancellableRequest()
cancellable_req.post(
f"http://localhost:{FRONTEND_PORT}/v1/chat/completions",
f"http://localhost:{frontend_port}/v1/chat/completions",
headers=headers,
json=payload,
stream=stream,
......@@ -244,12 +262,14 @@ def send_chat_completion_request(
def send_cancellable_request(
frontend_port: int,
request_type: str = "completion",
use_long_prompt: bool = False,
) -> CancellableRequest:
"""Send a request that can be manually cancelled.
Args:
frontend_port: Port where the frontend is running
request_type: Type of request - "completion", "chat_completion", or "chat_completion_stream"
use_long_prompt: Whether to use an extremely long prompt
......@@ -261,11 +281,11 @@ def send_cancellable_request(
prompt += " Make sure it is" + " long" * 16000 + "!"
if request_type == "completion":
return send_completion_request(prompt, 16384)
return send_completion_request(prompt, 16384, frontend_port)
elif request_type == "chat_completion":
return send_chat_completion_request(prompt, 16384, stream=False)
return send_chat_completion_request(prompt, 16384, frontend_port, stream=False)
elif request_type == "chat_completion_stream":
return send_chat_completion_request(prompt, 16384, stream=True)
return send_chat_completion_request(prompt, 16384, frontend_port, stream=True)
else:
raise ValueError(f"Unknown request type: {request_type}")
......@@ -283,12 +303,15 @@ def read_streaming_responses(
Raises:
pytest.fail if stream ends before expected_count responses
"""
response = cancellable_req.get_response()
if not response or response.status_code != 200:
response_raw = cancellable_req.get_response()
if response_raw is None:
pytest.fail("Failed to get streaming response: response is None")
if response_raw.status_code != 200:
pytest.fail(
f"Failed to get streaming response: status_code={response.status_code if response else 'None'}"
f"Failed to get streaming response: status_code={response_raw.status_code}"
)
response = cast(requests.Response, response_raw) # Type narrowing after checks
response_count = 0
for line in response.iter_lines():
response_count += 1
......
tests/fault_tolerance/etcd_ha/test_sglang.py
View file @ c6b440e4
# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
# TODO: Update to use dynamic port allocation (allocate_free_port) for parallel execution
# Currently uses hardcoded ports: FRONTEND_PORT (8000), system ports (8081, 8082)
# See tests/fault_tolerance/migration/test_sglang.py for dynamic port pattern
import logging
import os
import shutil
......
tests/fault_tolerance/migration/test_sglang.py
View file @ c6b440e4
# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
"""
Test Execution Times (Last Run: 2025-12-09):
- test_request_migration_sglang_worker_failure: ~58s (gpu_1)
- test_request_migration_sglang_graceful_shutdown: ~58s (gpu_1, skipped)
- test_no_request_migration_sglang_worker_failure: ~38s (gpu_1)
- test_no_request_migration_sglang_graceful_shutdown: ~38s (gpu_1, skipped)
- Total: 115.71s (0:01:55) for enabled tests
"""
import logging
import os
import shutil
......@@ -8,9 +17,9 @@ import shutil
import pytest
from tests.utils.constants import FAULT_TOLERANCE_MODEL_NAME
from tests.utils.engine_process import FRONTEND_PORT
from tests.utils.managed_process import ManagedProcess, terminate_process_tree
from tests.utils.payloads import check_models_api
from tests.utils.port_utils import allocate_port, deallocate_port
# Import utilities from the refactored utils module
from .utils import (
......@@ -35,8 +44,16 @@ pytestmark = [
class DynamoWorkerProcess(ManagedProcess):
"""Process manager for Dynamo worker with SGLang backend"""
def __init__(self, request, worker_id: str, migration_limit: int = 3):
def __init__(
self,
request,
worker_id: str,
system_port: int,
frontend_port: int,
migration_limit: int = 3,
):
self.worker_id = worker_id
self.system_port = system_port
command = [
"python3",
......@@ -66,7 +83,8 @@ class DynamoWorkerProcess(ManagedProcess):
# intermittent failures
env["DYN_HEALTH_CHECK_ENABLED"] = "false"
env["DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS"] = '["generate"]'
env["DYN_SYSTEM_PORT"] = f"808{worker_id[-1]}"
env["DYN_SYSTEM_PORT"] = str(system_port)
env["DYN_HTTP_PORT"] = str(frontend_port)
# TODO: Have the managed process take a command name explicitly to distinguish
# between processes started with the same command.
......@@ -84,8 +102,8 @@ class DynamoWorkerProcess(ManagedProcess):
command=command,
env=env,
health_check_urls=[
(f"http://localhost:{FRONTEND_PORT}/v1/models", check_models_api),
(f"http://localhost:808{worker_id[-1]}/health", self.is_ready),
(f"http://localhost:{frontend_port}/v1/models", check_models_api),
(f"http://localhost:{system_port}/health", self.is_ready),
],
timeout=300,
display_output=True,
......@@ -95,9 +113,15 @@ class DynamoWorkerProcess(ManagedProcess):
log_dir=log_dir,
)
def get_pid(self):
"""Get the PID of the worker process"""
return self.proc.pid if self.proc else None
def __exit__(self, exc_type, exc_val, exc_tb):
"""Release allocated port when worker exits."""
try:
# system_port is a required parameter, always set in __init__
deallocate_port(self.system_port)
except Exception as e:
logging.warning(f"Failed to release SGLang worker port: {e}")
return super().__exit__(exc_type, exc_val, exc_tb)
def is_ready(self, response) -> bool:
"""Check the health of the worker process"""
......@@ -127,7 +151,7 @@ class DynamoWorkerProcess(ManagedProcess):
indirect=True,
)
def test_request_migration_sglang_worker_failure(
request, runtime_services, set_ucx_tls_no_mm
request, runtime_services_dynamic_ports, set_ucx_tls_no_mm
):
"""
End-to-end test for worker fault tolerance with migration support using SGLang.
......@@ -135,21 +159,42 @@ def test_request_migration_sglang_worker_failure(
This test verifies that when a worker is killed during request processing,
the system can handle the failure gracefully and migrate the request to
another worker.
Timing (Last Run: 2025-12-09): ~58s total
- Engine initialization: ~22s (Worker1: 12s, Worker2: 10s)
- Test execution (request + migration): ~21s
- Teardown: ~15s
"""
# Step 1: Start the frontend
# Allocate ports to avoid conflicts with parallel tests
worker1_system_port = allocate_port(9100)
worker2_system_port = allocate_port(9200)
# Step 1: Start the frontend (allocates its own port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start 2 workers sequentially
with DynamoWorkerProcess(request, "worker1") as worker1:
with DynamoWorkerProcess(
request,
"worker1",
system_port=worker1_system_port,
frontend_port=frontend.frontend_port,
) as worker1:
logger.info(f"Worker 1 PID: {worker1.get_pid()}")
with DynamoWorkerProcess(request, "worker2") as worker2:
with DynamoWorkerProcess(
request,
"worker2",
system_port=worker2_system_port,
frontend_port=frontend.frontend_port,
) as worker2:
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
# Step 3: Send the request
request_thread, response_list = start_completion_request()
request_thread, response_list = start_completion_request(
frontend.frontend_port
)
# Step 4: Use polling to determine which worker received the request
worker, worker_name = determine_request_receiving_worker(
......@@ -169,6 +214,7 @@ def test_request_migration_sglang_worker_failure(
verify_migration_occurred(frontend)
@pytest.mark.timeout(235) # 3x average
@pytest.mark.skip(reason="SGLang graceful shutdown not yet implemented")
@pytest.mark.parametrize(
"request_plane",
......@@ -182,7 +228,7 @@ def test_request_migration_sglang_worker_failure(
indirect=True,
)
def test_request_migration_sglang_graceful_shutdown(
request, runtime_services, set_ucx_tls_no_mm
request, runtime_services_dynamic_ports, set_ucx_tls_no_mm
):
"""
End-to-end test for worker fault tolerance with graceful shutdown and migration support using SGLang.
......@@ -192,21 +238,42 @@ def test_request_migration_sglang_graceful_shutdown(
the request to another worker. Unlike the abrupt kill test, this simulates a more
controlled shutdown scenario where the worker has time to clean up and notify the
system about its shutdown.
Timing (Last Run: 2025-12-09): ~58s total (estimated, similar to worker_failure)
- Engine initialization: ~22s (Worker1: 12s, Worker2: 10s)
- Test execution (request + graceful shutdown + migration): ~21s
- Teardown: ~15s
"""
# Step 1: Start the frontend
# Allocate ports to avoid conflicts with parallel tests
worker1_system_port = allocate_port(9100)
worker2_system_port = allocate_port(9200)
# Step 1: Start the frontend (allocates its own port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start 2 workers sequentially
with DynamoWorkerProcess(request, "worker1") as worker1:
with DynamoWorkerProcess(
request,
"worker1",
system_port=worker1_system_port,
frontend_port=frontend.frontend_port,
) as worker1:
logger.info(f"Worker 1 PID: {worker1.get_pid()}")
with DynamoWorkerProcess(request, "worker2") as worker2:
with DynamoWorkerProcess(
request,
"worker2",
system_port=worker2_system_port,
frontend_port=frontend.frontend_port,
) as worker2:
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
# Step 3: Send the request
request_thread, response_list = start_completion_request()
request_thread, response_list = start_completion_request(
frontend.frontend_port
)
# Step 4: Use polling to determine which worker received the request
worker, worker_name = determine_request_receiving_worker(
......@@ -241,7 +308,7 @@ def test_request_migration_sglang_graceful_shutdown(
indirect=True,
)
def test_no_request_migration_sglang_worker_failure(
request, runtime_services, set_ucx_tls_no_mm
request, runtime_services_dynamic_ports, set_ucx_tls_no_mm
):
"""
End-to-end test for worker fault tolerance with migration disabled using SGLang.
......@@ -249,21 +316,44 @@ def test_no_request_migration_sglang_worker_failure(
This test verifies that when migration is disabled (migration_limit=0) and a worker
is killed during request processing, the request fails as expected without migration.
This is the opposite behavior of test_request_migration_sglang_worker_failure.
Timing (Last Run: 2025-12-09): ~38s total
- Engine initialization: ~23s (Worker1: 13s, Worker2: 10s)
- Test execution (failure validation): <1s
- Teardown: ~15s
"""
# Step 1: Start the frontend
# Allocate ports to avoid conflicts with parallel tests
worker1_system_port = allocate_port(9100)
worker2_system_port = allocate_port(9200)
# Step 1: Start the frontend (allocates its own port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start 2 workers sequentially with migration disabled
with DynamoWorkerProcess(request, "worker1", migration_limit=0) as worker1:
with DynamoWorkerProcess(
request,
"worker1",
system_port=worker1_system_port,
frontend_port=frontend.frontend_port,
migration_limit=0,
) as worker1:
logger.info(f"Worker 1 PID: {worker1.get_pid()}")
with DynamoWorkerProcess(request, "worker2", migration_limit=0) as worker2:
with DynamoWorkerProcess(
request,
"worker2",
system_port=worker2_system_port,
frontend_port=frontend.frontend_port,
migration_limit=0,
) as worker2:
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
# Step 3: Send the request
request_thread, response_list = start_completion_request()
request_thread, response_list = start_completion_request(
frontend.frontend_port
)
# Step 4: Use polling to determine which worker received the request
worker, worker_name = determine_request_receiving_worker(
......@@ -299,6 +389,7 @@ def test_no_request_migration_sglang_worker_failure(
), f"Unexpected migration message: {e}"
@pytest.mark.timeout(135) # 3x average
@pytest.mark.skip(reason="SGLang graceful shutdown not yet implemented")
@pytest.mark.parametrize(
"request_plane",
......@@ -312,7 +403,7 @@ def test_no_request_migration_sglang_worker_failure(
indirect=True,
)
def test_no_request_migration_sglang_graceful_shutdown(
request, runtime_services, set_ucx_tls_no_mm
request, runtime_services_dynamic_ports, set_ucx_tls_no_mm
):
"""
End-to-end test for worker fault tolerance with graceful shutdown and migration disabled using SGLang.
......@@ -321,21 +412,44 @@ def test_no_request_migration_sglang_graceful_shutdown(
receives a graceful shutdown signal (SIGTERM) during request processing, the request
fails as expected without migration. This is the opposite behavior of
test_request_migration_sglang_graceful_shutdown.
Timing (Last Run: 2025-12-09): ~38s total (estimated, similar to no_migration_worker_failure)
- Engine initialization: ~23s (Worker1: 13s, Worker2: 10s)
- Test execution (graceful shutdown + failure validation): <1s
- Teardown: ~15s
"""
# Step 1: Start the frontend
# Allocate ports to avoid conflicts with parallel tests
worker1_system_port = allocate_port(9100)
worker2_system_port = allocate_port(9200)
# Step 1: Start the frontend (allocates its own port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start 2 workers sequentially with migration disabled
with DynamoWorkerProcess(request, "worker1", migration_limit=0) as worker1:
with DynamoWorkerProcess(
request,
"worker1",
system_port=worker1_system_port,
frontend_port=frontend.frontend_port,
migration_limit=0,
) as worker1:
logger.info(f"Worker 1 PID: {worker1.get_pid()}")
with DynamoWorkerProcess(request, "worker2", migration_limit=0) as worker2:
with DynamoWorkerProcess(
request,
"worker2",
system_port=worker2_system_port,
frontend_port=frontend.frontend_port,
migration_limit=0,
) as worker2:
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
# Step 3: Send the request
request_thread, response_list = start_completion_request()
request_thread, response_list = start_completion_request(
frontend.frontend_port
)
# Step 4: Use polling to determine which worker received the request
worker, worker_name = determine_request_receiving_worker(
......
tests/fault_tolerance/migration/test_trtllm.py
View file @ c6b440e4
# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
"""
Test Execution Times (Last Run: 2025-12-09):
- test_request_migration_trtllm_worker_failure: ~95s (gpu_1)
- test_request_migration_trtllm_graceful_shutdown: ~95s (gpu_1, skipped)
- test_no_request_migration_trtllm_worker_failure: ~60s (gpu_1)
- test_no_request_migration_trtllm_graceful_shutdown: ~60s (gpu_1, skipped)
- Total: ~155s (0:02:35) for enabled tests
"""
import logging
import os
import shutil
......@@ -8,9 +17,9 @@ import shutil
import pytest
from tests.utils.constants import FAULT_TOLERANCE_MODEL_NAME
from tests.utils.engine_process import FRONTEND_PORT
from tests.utils.managed_process import ManagedProcess, terminate_process_tree
from tests.utils.payloads import check_models_api
from tests.utils.port_utils import allocate_port, deallocate_port
# Import utilities from the refactored utils module
from .utils import (
......@@ -35,8 +44,19 @@ pytestmark = [
class DynamoWorkerProcess(ManagedProcess):
"""Process manager for Dynamo worker with TRT-LLM backend"""
def __init__(self, request, worker_id: str, migration_limit: int = 3):
def __init__(
self,
request,
worker_id: str,
frontend_port: int,
migration_limit: int = 3,
):
self.worker_id = worker_id
self.frontend_port = frontend_port
# Allocate system port for this worker
system_port = allocate_port(9100)
self.system_port = system_port
command = [
"python3",
......@@ -64,7 +84,7 @@ class DynamoWorkerProcess(ManagedProcess):
# intermittent failures
env["DYN_HEALTH_CHECK_ENABLED"] = "false"
env["DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS"] = '["generate"]'
env["DYN_SYSTEM_PORT"] = f"808{worker_id[-1]}"
env["DYN_SYSTEM_PORT"] = str(system_port)
# TODO: Have the managed process take a command name explicitly to distinguish
# between processes started with the same command.
......@@ -82,8 +102,8 @@ class DynamoWorkerProcess(ManagedProcess):
command=command,
env=env,
health_check_urls=[
(f"http://localhost:{FRONTEND_PORT}/v1/models", check_models_api),
(f"http://localhost:808{worker_id[-1]}/health", self.is_ready),
(f"http://localhost:{frontend_port}/v1/models", check_models_api),
(f"http://localhost:{system_port}/health", self.is_ready),
],
timeout=300,
display_output=True,
......@@ -91,9 +111,15 @@ class DynamoWorkerProcess(ManagedProcess):
log_dir=log_dir,
)
def get_pid(self):
"""Get the PID of the worker process"""
return self.proc.pid if self.proc else None
def __exit__(self, exc_type, exc_val, exc_tb):
"""Release allocated port when worker exits."""
try:
# system_port is always allocated in __init__
deallocate_port(self.system_port)
except Exception as e:
logging.warning(f"Failed to release TRT-LLM worker port: {e}")
return super().__exit__(exc_type, exc_val, exc_tb)
def is_ready(self, response) -> bool:
"""Check the health of the worker process"""
......@@ -123,7 +149,7 @@ class DynamoWorkerProcess(ManagedProcess):
indirect=True,
)
def test_request_migration_trtllm_worker_failure(
request, runtime_services, set_ucx_tls_no_mm
request, runtime_services_dynamic_ports, set_ucx_tls_no_mm
):
"""
End-to-end test for worker fault tolerance with migration support using TRT-LLM.
......@@ -131,21 +157,30 @@ def test_request_migration_trtllm_worker_failure(
This test verifies that when a worker is killed during request processing,
the system can handle the failure gracefully and migrate the request to
another worker.
Timing (Last Run: 2025-12-09): ~95s total (2 workers at 45% GPU each)
- Engine initialization: ~52s (frontend: 2s, worker1: 25s, worker2: 25s sequential)
- Test execution (request + migration): ~40s
- Teardown: ~3s
"""
# Step 1: Start the frontend
# Step 1: Start the frontend (allocates its own frontend_port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start 2 workers sequentially
with DynamoWorkerProcess(request, "worker1") as worker1:
with DynamoWorkerProcess(request, "worker1", frontend.frontend_port) as worker1:
logger.info(f"Worker 1 PID: {worker1.get_pid()}")
with DynamoWorkerProcess(request, "worker2") as worker2:
with DynamoWorkerProcess(
request, "worker2", frontend.frontend_port
) as worker2:
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
# Step 3: Send the request
request_thread, response_list = start_completion_request()
request_thread, response_list = start_completion_request(
frontend.frontend_port
)
# Step 4: Use polling to determine which worker received the request
worker, worker_name = determine_request_receiving_worker(
......@@ -178,7 +213,7 @@ def test_request_migration_trtllm_worker_failure(
indirect=True,
)
def test_request_migration_trtllm_graceful_shutdown(
request, runtime_services, set_ucx_tls_no_mm
request, runtime_services_dynamic_ports, set_ucx_tls_no_mm
):
"""
End-to-end test for worker fault tolerance with graceful shutdown and migration support using TRT-LLM.
......@@ -188,21 +223,30 @@ def test_request_migration_trtllm_graceful_shutdown(
the request to another worker. Unlike the abrupt kill test, this simulates a more
controlled shutdown scenario where the worker has time to clean up and notify the
system about its shutdown.
Timing (Last Run: 2025-12-09): ~95s total (2 workers at 45% GPU each)
- Engine initialization: ~52s (frontend: 2s, worker1: 25s, worker2: 25s sequential)
- Test execution (request + graceful migration): ~40s
- Teardown: ~3s
"""
# Step 1: Start the frontend
# Step 1: Start the frontend (allocates its own frontend_port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start 2 workers sequentially
with DynamoWorkerProcess(request, "worker1") as worker1:
with DynamoWorkerProcess(request, "worker1", frontend.frontend_port) as worker1:
logger.info(f"Worker 1 PID: {worker1.get_pid()}")
with DynamoWorkerProcess(request, "worker2") as worker2:
with DynamoWorkerProcess(
request, "worker2", frontend.frontend_port
) as worker2:
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
# Step 3: Send the request
request_thread, response_list = start_completion_request()
request_thread, response_list = start_completion_request(
frontend.frontend_port
)
# Step 4: Use polling to determine which worker received the request
worker, worker_name = determine_request_receiving_worker(
......@@ -237,7 +281,7 @@ def test_request_migration_trtllm_graceful_shutdown(
indirect=True,
)
def test_no_request_migration_trtllm_worker_failure(
request, runtime_services, set_ucx_tls_no_mm
request, runtime_services_dynamic_ports, set_ucx_tls_no_mm
):
"""
End-to-end test for worker fault tolerance with migration disabled using TRT-LLM.
......@@ -245,21 +289,38 @@ def test_no_request_migration_trtllm_worker_failure(
This test verifies that when migration is disabled (migration_limit=0) and a worker
is killed during request processing, the request fails as expected without migration.
This is the opposite behavior of test_request_migration_trtllm_worker_failure.
Timing (Last Run: 2025-12-09): ~60s total (2 workers at 45% GPU each)
- Engine initialization: ~52s (frontend: 2s, worker1: 25s, worker2: 25s sequential)
- Test execution (request failure): ~6s
- Teardown: ~2s
"""
# Step 1: Start the frontend
# Step 1: Start the frontend (allocates its own frontend_port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start 2 workers sequentially with migration disabled
with DynamoWorkerProcess(request, "worker1", migration_limit=0) as worker1:
with DynamoWorkerProcess(
request,
"worker1",
frontend.frontend_port,
migration_limit=0,
) as worker1:
logger.info(f"Worker 1 PID: {worker1.get_pid()}")
with DynamoWorkerProcess(request, "worker2", migration_limit=0) as worker2:
with DynamoWorkerProcess(
request,
"worker2",
frontend.frontend_port,
migration_limit=0,
) as worker2:
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
# Step 3: Send the request
request_thread, response_list = start_completion_request()
request_thread, response_list = start_completion_request(
frontend.frontend_port
)
# Step 4: Use polling to determine which worker received the request
worker, worker_name = determine_request_receiving_worker(
......@@ -308,7 +369,7 @@ def test_no_request_migration_trtllm_worker_failure(
indirect=True,
)
def test_no_request_migration_trtllm_graceful_shutdown(
request, runtime_services, set_ucx_tls_no_mm
request, runtime_services_dynamic_ports, set_ucx_tls_no_mm
):
"""
End-to-end test for worker fault tolerance with graceful shutdown and migration disabled using TRT-LLM.
......@@ -317,21 +378,38 @@ def test_no_request_migration_trtllm_graceful_shutdown(
receives a graceful shutdown signal (SIGTERM) during request processing, the request
fails as expected without migration. This is the opposite behavior of
test_request_migration_trtllm_graceful_shutdown.
Timing (Last Run: 2025-12-09): ~60s total (2 workers at 45% GPU each)
- Engine initialization: ~52s (frontend: 2s, worker1: 25s, worker2: 25s sequential)
- Test execution (graceful shutdown failure): ~6s
- Teardown: ~2s
"""
# Step 1: Start the frontend
# Step 1: Start the frontend (allocates its own frontend_port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start 2 workers sequentially with migration disabled
with DynamoWorkerProcess(request, "worker1", migration_limit=0) as worker1:
with DynamoWorkerProcess(
request,
"worker1",
frontend.frontend_port,
migration_limit=0,
) as worker1:
logger.info(f"Worker 1 PID: {worker1.get_pid()}")
with DynamoWorkerProcess(request, "worker2", migration_limit=0) as worker2:
with DynamoWorkerProcess(
request,
"worker2",
frontend.frontend_port,
migration_limit=0,
) as worker2:
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
# Step 3: Send the request
request_thread, response_list = start_completion_request()
request_thread, response_list = start_completion_request(
frontend.frontend_port
)
# Step 4: Use polling to determine which worker received the request
worker, worker_name = determine_request_receiving_worker(
......
tests/fault_tolerance/migration/test_vllm.py
View file @ c6b440e4
# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
"""
Test Execution Times (Last Run: 2025-12-09):
- test_request_migration_vllm_worker_failure: ~90s (gpu_1)
- test_request_migration_vllm_graceful_shutdown: ~80s (gpu_1)
- test_no_request_migration_vllm_worker_failure: ~75s (gpu_1)
- test_no_request_migration_vllm_graceful_shutdown: ~75s (gpu_1)
- Total: 318.73s (0:05:18)
"""
import logging
import os
import shutil
......@@ -8,9 +17,9 @@ import shutil
import pytest
from tests.utils.constants import FAULT_TOLERANCE_MODEL_NAME
from tests.utils.engine_process import FRONTEND_PORT
from tests.utils.managed_process import ManagedProcess, terminate_process_tree
from tests.utils.payloads import check_models_api
from tests.utils.port_utils import allocate_port, deallocate_port
# Import utilities from the refactored utils module
from .utils import (
......@@ -35,8 +44,19 @@ pytestmark = [
class DynamoWorkerProcess(ManagedProcess):
"""Process manager for Dynamo worker with vLLM backend"""
def __init__(self, request, worker_id: str, migration_limit: int = 3):
def __init__(
self,
request,
worker_id: str,
frontend_port: int,
migration_limit: int = 3,
):
self.worker_id = worker_id
self.frontend_port = frontend_port
# Allocate system port for this worker
system_port = allocate_port(9100)
self.system_port = system_port
command = [
"python3",
......@@ -57,8 +77,12 @@ class DynamoWorkerProcess(ManagedProcess):
env = os.environ.copy()
env["DYN_REQUEST_PLANE"] = request.getfixturevalue("request_plane")
env["DYN_VLLM_KV_EVENT_PORT"] = f"2008{worker_id[-1]}"
env["VLLM_NIXL_SIDE_CHANNEL_PORT"] = f"560{worker_id[-1]}"
env[
"DYN_VLLM_KV_EVENT_PORT"
] = f"2008{worker_id[-1]}" # TODO: use dynamic port allocation
env[
"VLLM_NIXL_SIDE_CHANNEL_PORT"
] = f"560{worker_id[-1]}" # TODO: use dynamic port allocation
env["DYN_LOG"] = "debug"
# Disable canary health check - these tests expect full control over requests
......@@ -67,7 +91,8 @@ class DynamoWorkerProcess(ManagedProcess):
# intermittent failures
env["DYN_HEALTH_CHECK_ENABLED"] = "false"
env["DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS"] = '["generate"]'
env["DYN_SYSTEM_PORT"] = f"808{worker_id[-1]}"
env["DYN_SYSTEM_PORT"] = str(system_port)
env["DYN_HTTP_PORT"] = str(frontend_port)
# TODO: Have the managed process take a command name explicitly to distinguish
# between processes started with the same command.
......@@ -85,8 +110,8 @@ class DynamoWorkerProcess(ManagedProcess):
command=command,
env=env,
health_check_urls=[
(f"http://localhost:{FRONTEND_PORT}/v1/models", check_models_api),
(f"http://localhost:808{worker_id[-1]}/health", self.is_ready),
(f"http://localhost:{frontend_port}/v1/models", check_models_api),
(f"http://localhost:{system_port}/health", self.is_ready),
],
timeout=300,
display_output=True,
......@@ -96,9 +121,15 @@ class DynamoWorkerProcess(ManagedProcess):
log_dir=log_dir,
)
def get_pid(self):
"""Get the PID of the worker process"""
return self.proc.pid if self.proc else None
def __exit__(self, exc_type, exc_val, exc_tb):
"""Release allocated port when worker exits."""
try:
# system_port is always allocated in __init__
deallocate_port(self.system_port)
except Exception as e:
logging.warning(f"Failed to release vLLM worker port: {e}")
return super().__exit__(exc_type, exc_val, exc_tb)
def is_ready(self, response) -> bool:
"""Check the health of the worker process"""
......@@ -128,7 +159,7 @@ class DynamoWorkerProcess(ManagedProcess):
indirect=True,
)
def test_request_migration_vllm_worker_failure(
request, runtime_services, set_ucx_tls_no_mm
request, runtime_services_dynamic_ports, set_ucx_tls_no_mm
):
"""
End-to-end test for worker fault tolerance with migration support.
......@@ -136,21 +167,30 @@ def test_request_migration_vllm_worker_failure(
This test verifies that when a worker is killed during request processing,
the system can handle the failure gracefully and migrate the request to
another worker.
Timing (Last Run: 2025-12-09): ~90s total
- Engine initialization: ~40s (Worker1: 20s, Worker2: 20s)
- Test execution (request + migration): ~48s
- Teardown: ~2s
"""
# Step 1: Start the frontend
# Step 1: Start the frontend (allocates its own frontend_port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start 2 workers sequentially
with DynamoWorkerProcess(request, "worker1") as worker1:
# Step 2: Start 2 workers sequentially (each allocates its own system_port)
with DynamoWorkerProcess(request, "worker1", frontend.frontend_port) as worker1:
logger.info(f"Worker 1 PID: {worker1.get_pid()}")
with DynamoWorkerProcess(request, "worker2") as worker2:
with DynamoWorkerProcess(
request, "worker2", frontend.frontend_port
) as worker2:
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
# Step 3: Send the request
request_thread, response_list = start_completion_request()
request_thread, response_list = start_completion_request(
frontend.frontend_port
)
# Step 4: Use polling to determine which worker received the request
worker, worker_name = determine_request_receiving_worker(
......@@ -183,7 +223,7 @@ def test_request_migration_vllm_worker_failure(
indirect=True,
)
def test_request_migration_vllm_graceful_shutdown(
request, runtime_services, set_ucx_tls_no_mm
request, runtime_services_dynamic_ports, set_ucx_tls_no_mm
):
"""
End-to-end test for worker fault tolerance with graceful shutdown and migration support.
......@@ -193,21 +233,30 @@ def test_request_migration_vllm_graceful_shutdown(
the request to another worker. Unlike the abrupt kill test, this simulates a more
controlled shutdown scenario where the worker has time to clean up and notify the
system about its shutdown.
Timing (Last Run: 2025-12-09): ~80s total
- Engine initialization: ~40s (Worker1: 20s, Worker2: 20s)
- Test execution (graceful shutdown + migration): ~38s
- Teardown: ~2s
"""
# Step 1: Start the frontend
# Step 1: Start the frontend (allocates its own frontend_port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start 2 workers sequentially
with DynamoWorkerProcess(request, "worker1") as worker1:
# Step 2: Start 2 workers sequentially (each allocates its own system_port)
with DynamoWorkerProcess(request, "worker1", frontend.frontend_port) as worker1:
logger.info(f"Worker 1 PID: {worker1.get_pid()}")
with DynamoWorkerProcess(request, "worker2") as worker2:
with DynamoWorkerProcess(
request, "worker2", frontend.frontend_port
) as worker2:
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
# Step 3: Send the request
request_thread, response_list = start_completion_request()
request_thread, response_list = start_completion_request(
frontend.frontend_port
)
# Step 4: Use polling to determine which worker received the request
worker, worker_name = determine_request_receiving_worker(
......@@ -242,7 +291,7 @@ def test_request_migration_vllm_graceful_shutdown(
indirect=True,
)
def test_no_request_migration_vllm_worker_failure(
request, runtime_services, set_ucx_tls_no_mm
request, runtime_services_dynamic_ports, set_ucx_tls_no_mm
):
"""
End-to-end test for worker fault tolerance with migration disabled.
......@@ -250,21 +299,32 @@ def test_no_request_migration_vllm_worker_failure(
This test verifies that when migration is disabled (migration_limit=0) and a worker
is killed during request processing, the request fails as expected without migration.
This is the opposite behavior of test_request_migration_vllm_worker_failure.
Timing (Last Run: 2025-12-09): ~75s total
- Engine initialization: ~40s (Worker1: 20s, Worker2: 20s)
- Test execution (failure validation): ~33s
- Teardown: ~2s
"""
# Step 1: Start the frontend
# Step 1: Start the frontend (allocates its own frontend_port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start 2 workers sequentially with migration disabled
with DynamoWorkerProcess(request, "worker1", migration_limit=0) as worker1:
# Step 2: Start 2 workers sequentially with migration disabled (each allocates its own system_port)
with DynamoWorkerProcess(
request, "worker1", frontend.frontend_port, migration_limit=0
) as worker1:
logger.info(f"Worker 1 PID: {worker1.get_pid()}")
with DynamoWorkerProcess(request, "worker2", migration_limit=0) as worker2:
with DynamoWorkerProcess(
request, "worker2", frontend.frontend_port, migration_limit=0
) as worker2:
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
# Step 3: Send the request
request_thread, response_list = start_completion_request()
request_thread, response_list = start_completion_request(
frontend.frontend_port
)
# Step 4: Use polling to determine which worker received the request
worker, worker_name = determine_request_receiving_worker(
......@@ -313,7 +373,7 @@ def test_no_request_migration_vllm_worker_failure(
indirect=True,
)
def test_no_request_migration_vllm_graceful_shutdown(
request, runtime_services, set_ucx_tls_no_mm
request, runtime_services_dynamic_ports, set_ucx_tls_no_mm
):
"""
End-to-end test for worker fault tolerance with graceful shutdown and migration disabled.
......@@ -322,21 +382,32 @@ def test_no_request_migration_vllm_graceful_shutdown(
receives a graceful shutdown signal (SIGTERM) during request processing, the request
fails as expected without migration. This is the opposite behavior of
test_request_migration_vllm_graceful_shutdown.
Timing (Last Run: 2025-12-09): ~75s total
- Engine initialization: ~40s (Worker1: 20s, Worker2: 20s)
- Test execution (graceful shutdown validation): ~33s
- Teardown: ~2s
"""
# Step 1: Start the frontend
# Step 1: Start the frontend (allocates its own frontend_port)
with DynamoFrontendProcess(request) as frontend:
logger.info("Frontend started successfully")
# Step 2: Start 2 workers sequentially with migration disabled
with DynamoWorkerProcess(request, "worker1", migration_limit=0) as worker1:
# Step 2: Start 2 workers sequentially with migration disabled (each allocates its own system_port)
with DynamoWorkerProcess(
request, "worker1", frontend.frontend_port, migration_limit=0
) as worker1:
logger.info(f"Worker 1 PID: {worker1.get_pid()}")
with DynamoWorkerProcess(request, "worker2", migration_limit=0) as worker2:
with DynamoWorkerProcess(
request, "worker2", frontend.frontend_port, migration_limit=0
) as worker2:
logger.info(f"Worker 2 PID: {worker2.get_pid()}")
# Step 3: Send the request
request_thread, response_list = start_completion_request()
request_thread, response_list = start_completion_request(
frontend.frontend_port
)
# Step 4: Use polling to determine which worker received the request
worker, worker_name = determine_request_receiving_worker(
......
tests/fault_tolerance/migration/utils.py
View file @ c6b440e4
......@@ -11,8 +11,8 @@ import pytest
import requests
from tests.utils.constants import FAULT_TOLERANCE_MODEL_NAME
from tests.utils.engine_process import FRONTEND_PORT
from tests.utils.managed_process import ManagedProcess
from tests.utils.port_utils import allocate_port, deallocate_port
logger = logging.getLogger(__name__)
......@@ -21,7 +21,19 @@ class DynamoFrontendProcess(ManagedProcess):
"""Process manager for Dynamo frontend"""
def __init__(self, request):
command = ["python", "-m", "dynamo.frontend", "--router-mode", "round-robin"]
# Allocate frontend port
frontend_port = allocate_port(8100)
self.frontend_port = frontend_port
command = [
"python",
"-m",
"dynamo.frontend",
"--router-mode",
"round-robin",
"--http-port",
str(frontend_port),
]
env = os.environ.copy()
env["DYN_REQUEST_PLANE"] = request.getfixturevalue("request_plane")
......@@ -47,15 +59,28 @@ class DynamoFrontendProcess(ManagedProcess):
command=command,
env=env,
display_output=True,
terminate_existing=True,
terminate_existing=False, # Don't terminate other processes of the same name, we'll only terminate our own PID
log_dir=log_dir,
)
def __exit__(self, exc_type, exc_val, exc_tb):
"""Release allocated port when frontend exits."""
try:
# frontend_port is always allocated in __init__
deallocate_port(self.frontend_port)
except Exception as e:
logger.warning(f"Failed to release frontend port: {e}")
return super().__exit__(exc_type, exc_val, exc_tb)
def start_completion_request() -> tuple:
def start_completion_request(frontend_port: int) -> tuple:
"""
Start a long-running completion request in a separate thread.
Args:
frontend_port: Port where the frontend is running
Returns:
tuple: (request_thread, response_list)
"""
......@@ -80,7 +105,7 @@ def start_completion_request() -> tuple:
try:
response = requests.post(
f"http://localhost:{FRONTEND_PORT}/v1/completions",
f"http://localhost:{frontend_port}/v1/completions",
headers=headers,
json=payload,
timeout=timeout,
......
tests/serve/test_trtllm.py
View file @ c6b440e4
......@@ -81,6 +81,7 @@ trtllm_configs = {
pytest.mark.gpu_1,
pytest.mark.pre_merge,
pytest.mark.trtllm,
pytest.mark.skip(reason="unstable"),
pytest.mark.timeout(
480
), # 3x measured time (103.66s) + download time (150s)
......
tests/utils/managed_process.py
View file @ c6b440e4
......@@ -559,6 +559,10 @@ class ManagedProcess:
hasattr(self, "proc") and self.proc is not None and self.proc.poll() is None
)
def get_pid(self) -> int | None:
"""Get the PID of the managed process."""
return self.proc.pid if self.proc else None
def subprocesses(self) -> list[psutil.Process]:
"""Find child processes of the current process."""
if (
......@@ -605,10 +609,6 @@ class DynamoFrontendProcess(ManagedProcess):
log_dir=log_dir,
)
def get_pid(self) -> int | None:
"""Get the PID of the worker process"""
return self.proc.pid if self.proc else None
def main():
with ManagedProcess(
......
tests/utils/port_utils.py 0 → 100644
View file @ c6b440e4
# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
"""
Port allocation utilities for tests.
Port allocation with flock-based locking to prevent race conditions in parallel tests.
"""
import fcntl
import inspect
import json
import os
import random
import socket
import tempfile
import time
from pathlib import Path
# Port allocation lock file
_PORT_LOCK_FILE = Path(tempfile.gettempdir()) / "pytest_port_allocations.lock"
_PORT_REGISTRY_FILE = Path(tempfile.gettempdir()) / "pytest_port_allocations.json"
# Port range for allocation (i16 range for Rust compatibility)
# TODO: Get Rust backend to use u16 instead of i16 so we can use full 1024-65535 range
_PORT_MIN = 1024
_PORT_MAX = 32767
def _load_port_registry() -> dict:
"""Load the port registry from disk.
Returns:
dict: Port registry mapping port numbers (as strings) to allocation info.
Example: {
"30001": {
"timestamp": 1732647123.456,
"caller_file": "/workspace/tests/test_foo.py",
"caller_function": "test_bar",
"caller_line": 42
}
}
"""
if not _PORT_REGISTRY_FILE.exists():
return {}
try:
with open(_PORT_REGISTRY_FILE, "r") as f:
return json.load(f)
except (json.JSONDecodeError, OSError):
return {}
def _save_port_registry(registry: dict) -> None:
"""Save the port registry to disk."""
with open(_PORT_REGISTRY_FILE, "w") as f:
json.dump(registry, f)
def _cleanup_stale_allocations(registry: dict, max_age: float = 900.0) -> dict:
"""Remove port allocations older than max_age seconds."""
current_time = time.time()
cleaned = {}
for port, info in registry.items():
# Handle both old format (timestamp only) and new format (dict with timestamp)
if isinstance(info, dict):
timestamp = info.get("timestamp", 0)
else:
timestamp = info
if current_time - timestamp < max_age:
cleaned[str(port)] = info
return cleaned
def allocate_ports(count: int, start_port: int) -> list[int]:
"""Find and return available ports in i16 range with flock-based locking.
Uses file locking (flock) to prevent race conditions when multiple test processes
allocate ports simultaneously.
Port range is limited to i16 (1024-32767) due to Rust backend expecting i16.
Searches from a random offset (start_port + random(100)) and walks up incrementally.
Wraps around to _PORT_MIN (1024) when exceeding _PORT_MAX. Retries up to 100 times.
Args:
count: Number of unique ports to allocate
start_port: Starting port number for allocation (required)
Returns:
list[int]: List of available port numbers
"""
# Get caller information for debugging
caller_file = "unknown"
caller_function = "unknown"
caller_line = 0
frame = inspect.currentframe()
if frame and frame.f_back:
caller_frame = frame.f_back
caller_info = inspect.getframeinfo(caller_frame)
caller_function = caller_frame.f_code.co_name
caller_file = caller_info.filename
caller_line = caller_info.lineno
# Validate start_port is in valid i16 range. Note that <1024 is reserved for system services (root only)
if start_port < _PORT_MIN or start_port > _PORT_MAX:
raise ValueError(
f"start_port must be between {_PORT_MIN} and {_PORT_MAX}, got {start_port}"
)
# Ensure lock file exists and is writable
_PORT_LOCK_FILE.parent.mkdir(parents=True, exist_ok=True)
_PORT_LOCK_FILE.touch(exist_ok=True)
if not os.access(_PORT_LOCK_FILE, os.W_OK):
raise PermissionError(
f"Port allocation lock file is not writable: {_PORT_LOCK_FILE}"
)
with open(_PORT_LOCK_FILE, "r+") as lock_file:
# Acquire exclusive lock
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
try:
# Load registry and clean up stale allocations
registry = _load_port_registry()
registry = _cleanup_stale_allocations(registry)
allocated_ports = set(int(p) for p in registry.keys())
ports: list[int] = []
# Start searching from desired port + random offset
current_port = start_port + random.randint(0, 100)
if current_port > _PORT_MAX:
current_port = _PORT_MIN + (current_port - _PORT_MAX - 1)
# Retry limit
max_retries = 100
attempts = 0
while len(ports) < count and attempts < max_retries:
attempts += 1
# Try current port
port = current_port
# Increment and wrap around to _PORT_MIN
current_port += 1
if current_port > _PORT_MAX:
current_port = _PORT_MIN
# Skip if already allocated or in our current list
if port in allocated_ports or port in ports:
continue
# Try to bind to verify it's actually free
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(("", port))
sock.close()
ports.append(port)
registry[str(port)] = {
"timestamp": time.time(),
"caller_file": caller_file,
"caller_function": caller_function,
"caller_line": caller_line,
}
except OSError:
continue
if len(ports) < count:
raise RuntimeError(
f"Could not find {count} available ports after {max_retries} retries"
)
# Save updated registry
_save_port_registry(registry)
return ports
finally:
# Release lock
fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
def allocate_port(start_port: int) -> int:
"""Find and return a single available port in i16 range.
Args:
start_port: Starting port number for allocation (required)
Returns:
int: An available port number between start_port and 32767 (i16 max)
"""
return allocate_ports(1, start_port)[0]
def deallocate_ports(ports: list[int]) -> None:
"""Release previously allocated ports back to the pool.
Args:
ports: List of port numbers to release
"""
if not ports:
return
# Ensure lock file exists
_PORT_LOCK_FILE.parent.mkdir(parents=True, exist_ok=True)
_PORT_LOCK_FILE.touch(exist_ok=True)
with open(_PORT_LOCK_FILE, "r+") as lock_file:
# Acquire exclusive lock
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
try:
# Load registry
registry = _load_port_registry()
# Remove the specified ports
for port in ports:
registry.pop(str(port), None)
# Save updated registry
_save_port_registry(registry)
finally:
# Release lock
fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
def deallocate_port(port: int) -> None:
"""Release a previously allocated port back to the pool.
Args:
port: Port number to release
"""
deallocate_ports([port])
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment