| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | """Integration tests for shutdown behavior, timeout, and signal handling.""" |
| 4 | |
| 5 | import asyncio |
| 6 | import signal |
| 7 | import subprocess |
| 8 | import sys |
| 9 | import time |
| 10 | from dataclasses import dataclass, field |
| 11 | from pathlib import Path |
| 12 | |
| 13 | import httpx |
| 14 | import openai |
| 15 | import psutil |
| 16 | import pytest |
| 17 | from prometheus_client.parser import text_string_to_metric_families |
| 18 | |
| 19 | from tests.utils import RemoteOpenAIServer |
| 20 | from vllm.utils.network_utils import get_open_port |
| 21 | |
| 22 | MODEL_NAME = "hmellor/tiny-random-LlamaForCausalLM" |
| 23 | |
| 24 | # GPU initialization might take take longer |
| 25 | _SERVER_STARTUP_TIMEOUT = 120 |
| 26 | _PROCESS_EXIT_TIMEOUT = 15 |
| 27 | _SHUTDOWN_DETECTION_TIMEOUT = 10 |
| 28 | _CHILD_CLEANUP_TIMEOUT = 10 |
| 29 | _INFLIGHT_REQUEST_START_TIMEOUT = 5 |
| 30 | _INFLIGHT_REQUEST_POLL_INTERVAL = 0.1 |
| 31 | _ABORT_CLIENT_TIMEOUT = 3 |
| 32 | |
| 33 | |
| 34 | def _get_child_pids(parent_pid: int) -> list[int]: |
| 35 | try: |
| 36 | parent = psutil.Process(parent_pid) |
| 37 | return [c.pid for c in parent.children(recursive=True)] |
| 38 | except psutil.NoSuchProcess: |
| 39 | return [] |
| 40 | |
| 41 | |
| 42 | async def _assert_children_cleaned_up( |
| 43 | child_pids: list[int], |
| 44 | timeout: float = _CHILD_CLEANUP_TIMEOUT, |
| 45 | ): |
| 46 | """Wait for child processes to exit and fail if any remain.""" |
| 47 | if not child_pids: |
| 48 | return |
| 49 | |
| 50 | deadline = time.time() + timeout |
| 51 | while time.time() < deadline: |
| 52 | still_alive = [] |
| 53 | for pid in child_pids: |
| 54 | try: |
| 55 | p = psutil.Process(pid) |
| 56 | if p.is_running() and p.status() != psutil.STATUS_ZOMBIE: |
| 57 | still_alive.append(pid) |
| 58 | except psutil.NoSuchProcess: |
| 59 | pass |
| 60 | if not still_alive: |
| 61 | return |
| 62 | await asyncio.sleep(0.5) |
| 63 | |
| 64 | pytest.fail( |
| 65 | f"Child processes {still_alive} still alive after {timeout}s. " |
| 66 | f"Process cleanup may not be working correctly." |
| 67 | ) |
| 68 | |
| 69 | |
| 70 | @dataclass |
| 71 | class ShutdownState: |
| 72 | got_503: bool = False |
| 73 | got_500: bool = False |
| 74 | requests_after_sigterm: int = 0 |
| 75 | aborted_requests: int = 0 |
| 76 | connection_errors: int = 0 |
| 77 | inflight_requests: int = 0 |
| 78 | stop_requesting: bool = False |
| 79 | errors: list[str] = field(default_factory=list) |
| 80 | |
| 81 | |
| 82 | async def _concurrent_request_loop( |
| 83 | client: openai.AsyncOpenAI, |
| 84 | state: ShutdownState, |
| 85 | sigterm_sent: asyncio.Event | None = None, |
| 86 | concurrency: int = 10, |
| 87 | ): |
| 88 | """Run multiple concurrent requests to keep the server busy.""" |
| 89 | |
| 90 | async def single_request(): |
| 91 | while not state.stop_requesting: |
| 92 | try: |
| 93 | state.inflight_requests += 1 |
| 94 | response = await client.completions.create( |
| 95 | model=MODEL_NAME, |
| 96 | prompt="Write a story: ", |
| 97 | max_tokens=200, |
| 98 | ) |
| 99 | if sigterm_sent is not None and sigterm_sent.is_set(): |
| 100 | state.requests_after_sigterm += 1 |
| 101 | # Check if any choice has finish_reason='abort' |
| 102 | if any(choice.finish_reason == "abort" for choice in response.choices): |
| 103 | state.aborted_requests += 1 |
| 104 | except openai.APIStatusError as e: |
| 105 | if e.status_code == 503: |
| 106 | state.got_503 = True |
| 107 | elif e.status_code == 500: |
| 108 | state.got_500 = True |
| 109 | else: |
| 110 | state.errors.append(f"API error: {e}") |
| 111 | except (openai.APIConnectionError, httpx.RemoteProtocolError): |
| 112 | state.connection_errors += 1 |
| 113 | if sigterm_sent is not None and sigterm_sent.is_set(): |
| 114 | break |
| 115 | except Exception as e: |
| 116 | state.errors.append(f"Unexpected error: {e}") |
| 117 | break |
| 118 | finally: |
| 119 | state.inflight_requests -= 1 |
| 120 | await asyncio.sleep(0.01) |
| 121 | |
| 122 | tasks = [asyncio.create_task(single_request()) for _ in range(concurrency)] |
| 123 | try: |
| 124 | await asyncio.gather(*tasks, return_exceptions=True) |
| 125 | finally: |
| 126 | for t in tasks: |
| 127 | if not t.done(): |
| 128 | t.cancel() |
| 129 | |
| 130 | |
| 131 | def _num_running_requests(metrics_text: str) -> float: |
| 132 | """Sum ``vllm:num_requests_running`` across the server's label sets.""" |
| 133 | total = 0.0 |
| 134 | for family in text_string_to_metric_families(metrics_text): |
| 135 | if family.name == "vllm:num_requests_running": |
| 136 | for sample in family.samples: |
| 137 | if sample.name == "vllm:num_requests_running": |
| 138 | total += sample.value |
| 139 | return total |
| 140 | |
| 141 | |
| 142 | async def _wait_for_running_request( |
| 143 | metrics_url: str, |
| 144 | timeout: float = _INFLIGHT_REQUEST_START_TIMEOUT, |
| 145 | ) -> float: |
| 146 | """Wait until the server reports a running request, returning the delay. |
| 147 | |
| 148 | ``ShutdownState.inflight_requests`` is incremented before the HTTP call is |
| 149 | awaited, so it only shows that a client coroutine started. Draining can |
| 150 | only complete work the engine has already admitted, so synchronize on the |
| 151 | server's own scheduler gauge instead. |
| 152 | """ |
| 153 | start = time.monotonic() |
| 154 | deadline = start + timeout |
| 155 | async with httpx.AsyncClient() as metrics_client: |
| 156 | while time.monotonic() < deadline: |
| 157 | response = await metrics_client.get(metrics_url) |
| 158 | response.raise_for_status() |
| 159 | if _num_running_requests(response.text) > 0: |
| 160 | return time.monotonic() - start |
| 161 | await asyncio.sleep(_INFLIGHT_REQUEST_POLL_INTERVAL) |
| 162 | raise AssertionError(f"No request reached the running state within {timeout}s") |
| 163 | |
| 164 | |
| 165 | @pytest.mark.asyncio |
| 166 | async def test_shutdown_on_engine_failure(tmp_path: Path): |
| 167 | """Verify that API returns connection error when server process is killed. |
| 168 | |
| 169 | Starts a vLLM server, kills it to simulate a crash, then verifies that |
| 170 | subsequent API calls fail appropriately. |
| 171 | """ |
| 172 | port = get_open_port() |
| 173 | |
| 174 | # Redirect server output to a file rather than a pipe: nothing drains the |
| 175 | # pipes while we poll for startup, so a server that writes more than the |
| 176 | # pipe buffer holds would block on write and never become ready. |
| 177 | server_log = tmp_path / "server.log" |
| 178 | with server_log.open("wb") as log_file: |
| 179 | proc = subprocess.Popen( |
| 180 | [ |
| 181 | # dtype, max-len etc set so that this can run in CI |
| 182 | sys.executable, |
| 183 | "-m", |
| 184 | "vllm.entrypoints.launchers.api_server.entry", |
| 185 | "--model", |
| 186 | MODEL_NAME, |
| 187 | "--dtype", |
| 188 | "bfloat16", |
| 189 | "--max-model-len", |
| 190 | "128", |
| 191 | "--enforce-eager", |
| 192 | "--port", |
| 193 | str(port), |
| 194 | "--gpu-memory-utilization", |
| 195 | "0.05", |
| 196 | "--max-num-seqs", |
| 197 | "2", |
| 198 | ], |
| 199 | stdout=log_file, |
| 200 | stderr=subprocess.STDOUT, |
| 201 | preexec_fn=lambda: signal.signal(signal.SIGINT, signal.SIG_IGN), |
| 202 | ) |
| 203 | |
| 204 | # Wait for server startup |
| 205 | start_time = time.time() |
| 206 | client = openai.AsyncOpenAI( |
| 207 | base_url=f"http://localhost:{port}/v1", |
| 208 | api_key="dummy", |
| 209 | max_retries=0, |
| 210 | timeout=10, |
| 211 | ) |
| 212 | |
| 213 | # Poll until server is ready |
| 214 | while time.time() - start_time < _SERVER_STARTUP_TIMEOUT: |
| 215 | try: |
| 216 | await client.completions.create( |
| 217 | model=MODEL_NAME, prompt="Hello", max_tokens=1 |
| 218 | ) |
| 219 | break |
| 220 | except Exception: |
| 221 | time.sleep(0.5) |
| 222 | if proc.poll() is not None: |
| 223 | pytest.fail( |
| 224 | f"Server died during startup: {proc.returncode}\n" |
| 225 | f"{server_log.read_text(errors='replace')}" |
| 226 | ) |
| 227 | else: |
| 228 | proc.terminate() |
| 229 | proc.wait(timeout=_PROCESS_EXIT_TIMEOUT) |
| 230 | pytest.fail( |
| 231 | f"Server failed to start in {_SERVER_STARTUP_TIMEOUT} seconds\n" |
| 232 | f"{server_log.read_text(errors='replace')}" |
| 233 | ) |
| 234 | |
| 235 | # Kill server to simulate crash |
| 236 | proc.terminate() |
| 237 | time.sleep(1) |
| 238 | |
| 239 | # Verify API calls now fail |
| 240 | with pytest.raises((openai.APIConnectionError, openai.APIStatusError)): |
| 241 | await client.completions.create( |
| 242 | model=MODEL_NAME, prompt="This should fail", max_tokens=1 |
| 243 | ) |
| 244 | |
| 245 | return_code = proc.wait(timeout=_PROCESS_EXIT_TIMEOUT) |
| 246 | assert return_code is not None |
| 247 | |
| 248 | |
| 249 | @pytest.mark.asyncio |
| 250 | async def test_wait_timeout_completes_requests(): |
| 251 | """Verify wait timeout: new requests rejected, in-flight requests complete.""" |
| 252 | server_args = [ |
| 253 | "--dtype", |
| 254 | "bfloat16", |
| 255 | "--max-model-len", |
| 256 | "256", |
| 257 | "--enforce-eager", |
| 258 | "--gpu-memory-utilization", |
| 259 | "0.05", |
| 260 | "--max-num-seqs", |
| 261 | "4", |
| 262 | "--shutdown-timeout", |
| 263 | "30", |
| 264 | ] |
| 265 | |
| 266 | with RemoteOpenAIServer(MODEL_NAME, server_args) as remote_server: |
| 267 | client = remote_server.get_async_client() |
| 268 | proc = remote_server.proc |
| 269 | child_pids = _get_child_pids(proc.pid) |
| 270 | |
| 271 | state = ShutdownState() |
| 272 | sigterm_sent = asyncio.Event() |
| 273 | |
| 274 | request_task = asyncio.create_task( |
| 275 | _concurrent_request_loop(client, state, sigterm_sent, concurrency=10) |
| 276 | ) |
| 277 | |
| 278 | try: |
| 279 | # Drain can only complete requests the engine already admitted, so |
| 280 | # wait for the server to report one instead of guessing with a sleep. |
| 281 | admission_delay = await _wait_for_running_request( |
| 282 | remote_server.url_for("metrics") |
| 283 | ) |
| 284 | proc.send_signal(signal.SIGTERM) |
| 285 | sigterm_sent.set() |
| 286 | |
| 287 | await asyncio.wait_for(request_task, timeout=_SHUTDOWN_DETECTION_TIMEOUT) |
| 288 | except asyncio.TimeoutError: |
| 289 | pass |
| 290 | finally: |
| 291 | state.stop_requesting = True |
| 292 | if not request_task.done(): |
| 293 | request_task.cancel() |
| 294 | await asyncio.gather(request_task, return_exceptions=True) |
| 295 | |
| 296 | # wait timeout should complete in-flight requests |
| 297 | assert state.requests_after_sigterm > 0, ( |
| 298 | f"Wait timeout should complete in-flight requests. " |
| 299 | f"admission_delay: {admission_delay:.3f}s, " |
| 300 | f"503: {state.got_503}, 500: {state.got_500}, " |
| 301 | f"conn_errors: {state.connection_errors}, errors: {state.errors}" |
| 302 | ) |
| 303 | # server must stop accepting new requests (503, 500, or connection close) |
| 304 | assert state.got_503 or state.got_500 or state.connection_errors > 0, ( |
| 305 | f"Server should stop accepting requests. " |
| 306 | f"completed: {state.requests_after_sigterm}, errors: {state.errors}" |
| 307 | ) |
| 308 | |
| 309 | await _assert_children_cleaned_up(child_pids) |
| 310 | |
| 311 | |
| 312 | @pytest.mark.asyncio |
| 313 | @pytest.mark.parametrize("wait_for_engine_idle", [0.0, 2.0]) |
| 314 | async def test_abort_timeout_exits_within_cleanup_grace( |
| 315 | wait_for_engine_idle: float, |
| 316 | ): |
| 317 | server_args = [ |
| 318 | "--dtype", |
| 319 | "bfloat16", |
| 320 | "--max-model-len", |
| 321 | "256", |
| 322 | "--enforce-eager", |
| 323 | "--gpu-memory-utilization", |
| 324 | "0.05", |
| 325 | "--max-num-seqs", |
| 326 | "4", |
| 327 | "--shutdown-timeout", |
| 328 | "0", |
| 329 | ] |
| 330 | |
| 331 | with RemoteOpenAIServer(MODEL_NAME, server_args) as remote_server: |
| 332 | proc = remote_server.proc |
| 333 | child_pids = _get_child_pids(proc.pid) |
| 334 | |
| 335 | if wait_for_engine_idle > 0: |
| 336 | client = remote_server.get_async_client() |
| 337 | # Send requests to ensure engine is fully initialized |
| 338 | for _ in range(2): |
| 339 | await client.completions.create( |
| 340 | model=MODEL_NAME, |
| 341 | prompt="Test request: ", |
| 342 | max_tokens=10, |
| 343 | ) |
| 344 | # Wait for engine to become idle |
| 345 | await asyncio.sleep(wait_for_engine_idle) |
| 346 | |
| 347 | proc.send_signal(signal.SIGTERM) |
| 348 | |
| 349 | # A zero request timeout aborts requests immediately, but process |
| 350 | # teardown may still use the platform-specific resource cleanup grace. |
| 351 | termination_timeout = remote_server._get_process_termination_timeout() |
| 352 | try: |
| 353 | proc.wait(timeout=termination_timeout) |
| 354 | except subprocess.TimeoutExpired: |
| 355 | proc.kill() |
| 356 | proc.wait(timeout=5) |
| 357 | pytest.fail( |
| 358 | "Process did not exit within the resource cleanup grace " |
| 359 | f"({termination_timeout}s)" |
| 360 | ) |
| 361 | |
| 362 | assert proc.returncode in (0, -15, None), f"Unexpected: {proc.returncode}" |
| 363 | |
| 364 | await _assert_children_cleaned_up(child_pids) |
| 365 | |
| 366 | |
| 367 | @pytest.mark.asyncio |
| 368 | async def test_wait_timeout_with_short_duration(): |
| 369 | """Verify server exits cleanly with a short wait timeout.""" |
| 370 | wait_timeout = 3 |
| 371 | server_args = [ |
| 372 | "--dtype", |
| 373 | "bfloat16", |
| 374 | "--max-model-len", |
| 375 | "256", |
| 376 | "--enforce-eager", |
| 377 | "--gpu-memory-utilization", |
| 378 | "0.05", |
| 379 | "--max-num-seqs", |
| 380 | "4", |
| 381 | "--shutdown-timeout", |
| 382 | str(wait_timeout), |
| 383 | ] |
| 384 | |
| 385 | with RemoteOpenAIServer(MODEL_NAME, server_args) as remote_server: |
| 386 | client = remote_server.get_async_client() |
| 387 | proc = remote_server.proc |
| 388 | child_pids = _get_child_pids(proc.pid) |
| 389 | |
| 390 | state = ShutdownState() |
| 391 | request_task = asyncio.create_task( |
| 392 | _concurrent_request_loop(client, state, concurrency=3) |
| 393 | ) |
| 394 | |
| 395 | await asyncio.sleep(0.5) |
| 396 | |
| 397 | start_time = time.time() |
| 398 | proc.send_signal(signal.SIGTERM) |
| 399 | |
| 400 | # server should exit within wait_timeout + buffer |
| 401 | max_wait = wait_timeout + 15 |
| 402 | for _ in range(int(max_wait * 10)): |
| 403 | if proc.poll() is not None: |
| 404 | break |
| 405 | time.sleep(0.1) |
| 406 | |
| 407 | exit_time = time.time() - start_time |
| 408 | |
| 409 | state.stop_requesting = True |
| 410 | if not request_task.done(): |
| 411 | request_task.cancel() |
| 412 | await asyncio.gather(request_task, return_exceptions=True) |
| 413 | |
| 414 | if proc.poll() is None: |
| 415 | proc.kill() |
| 416 | proc.wait(timeout=5) |
| 417 | pytest.fail(f"Process did not exit within {max_wait}s after SIGTERM") |
| 418 | |
| 419 | assert exit_time < wait_timeout + 10, ( |
| 420 | f"Took too long to exit ({exit_time:.1f}s), expected <{wait_timeout + 10}s" |
| 421 | ) |
| 422 | assert proc.returncode in (0, -15, None), f"Unexpected: {proc.returncode}" |
| 423 | |
| 424 | await _assert_children_cleaned_up(child_pids) |
| 425 | |
| 426 | |
| 427 | @pytest.mark.asyncio |
| 428 | async def test_abort_timeout_fails_inflight_requests(): |
| 429 | """Verify abort timeout (0) immediately aborts in-flight requests.""" |
| 430 | server_args = [ |
| 431 | "--dtype", |
| 432 | "bfloat16", |
| 433 | "--max-model-len", |
| 434 | "256", |
| 435 | "--enforce-eager", |
| 436 | "--gpu-memory-utilization", |
| 437 | "0.05", |
| 438 | "--max-num-seqs", |
| 439 | "4", |
| 440 | "--shutdown-timeout", |
| 441 | "0", |
| 442 | ] |
| 443 | |
| 444 | with RemoteOpenAIServer(MODEL_NAME, server_args) as remote_server: |
| 445 | client = remote_server.get_async_client(timeout=_ABORT_CLIENT_TIMEOUT) |
| 446 | proc = remote_server.proc |
| 447 | child_pids = _get_child_pids(proc.pid) |
| 448 | |
| 449 | state = ShutdownState() |
| 450 | sigterm_sent = asyncio.Event() |
| 451 | |
| 452 | request_task = asyncio.create_task( |
| 453 | _concurrent_request_loop(client, state, sigterm_sent, concurrency=10) |
| 454 | ) |
| 455 | |
| 456 | deadline = time.time() + _INFLIGHT_REQUEST_START_TIMEOUT |
| 457 | while state.inflight_requests == 0 and time.time() < deadline: |
| 458 | await asyncio.sleep(_INFLIGHT_REQUEST_POLL_INTERVAL) |
| 459 | assert state.inflight_requests > 0 |
| 460 | |
| 461 | proc.send_signal(signal.SIGTERM) |
| 462 | sigterm_sent.set() |
| 463 | |
| 464 | try: |
| 465 | await asyncio.wait_for(request_task, timeout=5) |
| 466 | except asyncio.TimeoutError: |
| 467 | pass |
| 468 | finally: |
| 469 | state.stop_requesting = True |
| 470 | if not request_task.done(): |
| 471 | request_task.cancel() |
| 472 | await asyncio.gather(request_task, return_exceptions=True) |
| 473 | |
| 474 | # With abort timeout (0), requests should be aborted (finish_reason='abort') |
| 475 | # or rejected (connection errors or API errors) |
| 476 | assert ( |
| 477 | state.aborted_requests > 0 |
| 478 | or state.connection_errors > 0 |
| 479 | or state.got_500 |
| 480 | or state.got_503 |
| 481 | ), ( |
| 482 | f"Abort timeout should cause request aborts or failures. " |
| 483 | f"aborted: {state.aborted_requests}, " |
| 484 | f"503: {state.got_503}, 500: {state.got_500}, " |
| 485 | f"conn_errors: {state.connection_errors}, " |
| 486 | f"completed: {state.requests_after_sigterm}" |
| 487 | ) |
| 488 | |
| 489 | # Verify fast shutdown |
| 490 | start_time = time.time() |
| 491 | for _ in range(100): |
| 492 | if proc.poll() is not None: |
| 493 | break |
| 494 | time.sleep(0.1) |
| 495 | |
| 496 | exit_time = time.time() - start_time |
| 497 | assert exit_time < 10, f"Abort timeout shutdown took too long: {exit_time:.1f}s" |
| 498 | |
| 499 | await _assert_children_cleaned_up(child_pids) |
| 500 | |
| 501 | |
| 502 | @pytest.mark.asyncio |
| 503 | async def test_request_rejection_during_shutdown(): |
| 504 | """Verify new requests are rejected with error during shutdown.""" |
| 505 | server_args = [ |
| 506 | "--dtype", |
| 507 | "bfloat16", |
| 508 | "--max-model-len", |
| 509 | "256", |
| 510 | "--enforce-eager", |
| 511 | "--gpu-memory-utilization", |
| 512 | "0.05", |
| 513 | "--max-num-seqs", |
| 514 | "4", |
| 515 | "--shutdown-timeout", |
| 516 | "30", |
| 517 | ] |
| 518 | |
| 519 | with RemoteOpenAIServer(MODEL_NAME, server_args) as remote_server: |
| 520 | client = remote_server.get_async_client() |
| 521 | proc = remote_server.proc |
| 522 | child_pids = _get_child_pids(proc.pid) |
| 523 | |
| 524 | proc.send_signal(signal.SIGTERM) |
| 525 | |
| 526 | await asyncio.sleep(1.0) |
| 527 | |
| 528 | # Try to send new requests - they should be rejected |
| 529 | rejected_count = 0 |
| 530 | for _ in range(10): |
| 531 | try: |
| 532 | await client.completions.create( |
| 533 | model=MODEL_NAME, prompt="Hello", max_tokens=10 |
| 534 | ) |
| 535 | except ( |
| 536 | openai.APIStatusError, |
| 537 | openai.APIConnectionError, |
| 538 | httpx.RemoteProtocolError, |
| 539 | ): |
| 540 | rejected_count += 1 |
| 541 | await asyncio.sleep(0.1) |
| 542 | |
| 543 | assert rejected_count > 0, ( |
| 544 | f"Expected requests to be rejected during shutdown, " |
| 545 | f"but {rejected_count} were rejected out of 10" |
| 546 | ) |
| 547 | |
| 548 | await _assert_children_cleaned_up(child_pids) |
| 549 | |
| 550 | |
| 551 | @pytest.mark.asyncio |
| 552 | async def test_multi_api_server_shutdown(): |
| 553 | """Verify shutdown works with multiple API servers.""" |
| 554 | server_args = [ |
| 555 | "--dtype", |
| 556 | "bfloat16", |
| 557 | "--max-model-len", |
| 558 | "256", |
| 559 | "--enforce-eager", |
| 560 | "--gpu-memory-utilization", |
| 561 | "0.05", |
| 562 | "--max-num-seqs", |
| 563 | "4", |
| 564 | "--shutdown-timeout", |
| 565 | "30", |
| 566 | "--api-server-count", |
| 567 | "2", |
| 568 | ] |
| 569 | |
| 570 | with RemoteOpenAIServer(MODEL_NAME, server_args, auto_port=True) as remote_server: |
| 571 | client = remote_server.get_async_client() |
| 572 | proc = remote_server.proc |
| 573 | child_pids = _get_child_pids(proc.pid) |
| 574 | |
| 575 | assert len(child_pids) >= 2, ( |
| 576 | f"Expected at least 2 child processes, got {len(child_pids)}" |
| 577 | ) |
| 578 | |
| 579 | state = ShutdownState() |
| 580 | sigterm_sent = asyncio.Event() |
| 581 | |
| 582 | # Start concurrent requests across both API servers |
| 583 | request_task = asyncio.create_task( |
| 584 | _concurrent_request_loop(client, state, sigterm_sent, concurrency=8) |
| 585 | ) |
| 586 | |
| 587 | await asyncio.sleep(0.5) |
| 588 | |
| 589 | # Send SIGTERM to parent - should propagate to all children |
| 590 | proc.send_signal(signal.SIGTERM) |
| 591 | sigterm_sent.set() |
| 592 | |
| 593 | try: |
| 594 | await asyncio.wait_for(request_task, timeout=_SHUTDOWN_DETECTION_TIMEOUT) |
| 595 | except asyncio.TimeoutError: |
| 596 | pass |
| 597 | finally: |
| 598 | state.stop_requesting = True |
| 599 | if not request_task.done(): |
| 600 | request_task.cancel() |
| 601 | await asyncio.gather(request_task, return_exceptions=True) |
| 602 | |
| 603 | for _ in range(300): # up to 30 seconds |
| 604 | if proc.poll() is not None: |
| 605 | break |
| 606 | time.sleep(0.1) |
| 607 | |
| 608 | if proc.poll() is None: |
| 609 | proc.kill() |
| 610 | proc.wait(timeout=5) |
| 611 | pytest.fail("Process did not exit after SIGTERM") |
| 612 | |
| 613 | await _assert_children_cleaned_up(child_pids) |