fix(vllm): make threaded HTTP generation async-safe

[jepio]

What does this PR do ?

Prevents the embedded vLLM HTTP server from sharing AsyncLLM request state across two event loops.

Bug

NeMo-RL creates AsyncLLM on the Ray worker loop, then passes the same object to Uvicorn on another thread and event loop. An HTTP generate() call waits on a vLLM RequestOutputCollector event from Uvicorn's loop. vLLM's output handler sets that event from the Ray worker loop.

asyncio.Event is not thread-safe. The two threads can mutate its waiter deque at the same time. This kills the vLLM output handler and fails every request waiting on that engine replica.

AsyncLLM output_handler failed.
Traceback (most recent call last):
  File ".../vllm/v1/engine/async_llm.py", line 695, in output_handler
    processed_outputs = output_processor.process_outputs(
  File ".../vllm/v1/engine/output_processor.py", line 681, in process_outputs
    req_state.queue.put(request_output)
  File ".../vllm/v1/engine/output_processor.py", line 66, in put
    self.ready.set()
  File "/usr/lib/python3.12/asyncio/locks.py", line 189, in set
    for fut in self._waiters:
RuntimeError: deque mutated during iteration

Fix

Uvicorn remains on its own thread. Generation and cancellation run on the event loop that owns AsyncLLM. Stable setup data stays on the HTTP thread. Sparse refit records the owner loop during worker initialization. The old input-socket lock is no longer needed.

Issues

Closes: #1857

Usage

No user-facing changes.

Before your PR is "Ready for review"

Pre checks:

• Make sure you read and followed Contributor guidelines
• Did you write any new necessary tests?
• Did you run the unit tests and functional tests locally? Visit our Testing Guide for how to run tests
• Did you add or update any necessary documentation? Visit our Document Development Guide for how to write, build and test the docs.

Additional Information

Testing completed:

• pre-commit checks passed: Ruff, formatting, Pyrefly, and file checks.
• Nine focused unit tests passed on Python 3.13. A broader run passed 27 tests before reaching a GPU-only test in the CPU VM.
• An amplified A/B reproduced deque mutated during iteration in the old code after 114 requests. The fix completed 128 requests, five training steps, and five refits.
• An unamplified stress run completed 128 requests, five steps, and five refits without an error.
• The performance A/B completed 95,713 requests with no failures. Request rate changed by -0.17%. Median and p95 improved. Middle-run p99 rose from 1.54 seconds to 2.08 seconds.

Timing probes placed the added tail latency before vLLM admitted each request. The longest wait for generate() to begin on the Ray worker loop was 1.70--1.72 seconds, which matched HTTP p99. Submitting the cross-thread callback was cheap; the callback waited for the Ray worker loop to schedule it. After a refit, vLLM processed queued outputs on that loop before it admitted replacement HTTP requests. The old code admitted those requests on Uvicorn's loop, but that was the unsafe cross-loop access that caused the crash.

[copy-pr-bot]

This pull request requires additional validation before any workflows can run on NVIDIA's runners.

Pull request vetters can view their responsibilities here.

Contributors can view more details about this message here.

[jepio]

/ok to test 350b1d8

[jepio]

/ok to test ed8ce9c

[jepio]

Reviewed by a coordinated team of specialized agents (rl-expert, bug-finder, test-agent, design-reviewer) with an adversarial devil's-advocate pass on every finding, plus direct verification against the real vLLM 0.25.1 source and local test runs (44/44 relevant tests pass; linters pass on all 4 changed files).

This is a solid, well-tested fix for a real, reproduced concurrency bug. All 6 comments below are low/medium-severity suggestions — none are blockers.

Generated by Claude Code

[jepio]

/ok to test d0ba2a7

[jepio]

I reproduced this with the full NeMo-RL GRPO path in nvcr.io/nvidian/nemo-rl:nightly on one four-GPU host.

Run the same test at the PR base (f49e41dda4c55ae6ca9f69476c07a648c6c35806) and the PR head (d0ba2a718d830b66ed70ffee48445c4d5bd462c4). Place this file at /tmp/vllm-deque-repro/sitecustomize.py inside the container:

"""Test-only scheduler hook for vLLM's cross-loop collector race."""

import asyncio
import dis
import os
import sys
import threading
import time


_ORIGINAL_EVENT_SET = asyncio.Event.set
_EVENT_SET_CODE = _ORIGINAL_EVENT_SET.__code__
_EVENT_SET_FOR_ITER_OFFSET = next(
    instruction.offset
    for instruction in dis.get_instructions(asyncio.Event.set)
    if instruction.opname == "FOR_ITER"
)
_DELAY_S = float(os.environ.get("VLLM_DEQUE_RACE_DELAY_S", "0.005"))
_REPORTED_HIT = False


def _event_set_trace(frame, event, _arg):
    global _REPORTED_HIT
    if (
        event == "line"
        and frame.f_code is _EVENT_SET_CODE
        and frame.f_lasti == _EVENT_SET_FOR_ITER_OFFSET
    ):
        fut = frame.f_locals["fut"]
        event_object = frame.f_locals["self"]
        waiter_loop = fut.get_loop()
        try:
            current_loop = asyncio.get_running_loop()
        except RuntimeError:
            current_loop = None
        foreign_loop = waiter_loop is not current_loop
        if not _REPORTED_HIT:
            _REPORTED_HIT = True
            print(
                "VLLM_DEQUE_RACE_HOOK_HIT "
                f"pid={os.getpid()} thread={threading.get_ident()} "
                f"foreign_loop={str(foreign_loop).lower()} delay_s={_DELAY_S}",
                flush=True,
            )
        if foreign_loop:
            waiter_loop.call_soon_threadsafe(lambda: None)
            deadline = time.monotonic() + 1.0
            while fut in event_object._waiters and time.monotonic() < deadline:
                time.sleep(0.0005)
        else:
            time.sleep(_DELAY_S)
    return _event_set_trace


def _global_trace(frame, event, _arg):
    if event == "call" and frame.f_code is _EVENT_SET_CODE:
        return _event_set_trace
    return None


def _instrumented_event_set(event_self):
    caller = sys._getframe(1)
    collector = caller.f_locals.get("self")
    if type(collector).__name__ != "RequestOutputCollector":
        return _ORIGINAL_EVENT_SET(event_self)

    previous_trace = sys.gettrace()
    sys.settrace(_global_trace)
    try:
        return _ORIGINAL_EVENT_SET(event_self)
    finally:
        sys.settrace(previous_trace)


asyncio.Event.set = _instrumented_event_set

From the NeMo-RL checkout, run:

set -u

RUN_DIR=$(mktemp -d)
TRAIN_LOG="${RUN_DIR}/train.log"
HTTP_LOG="${RUN_DIR}/http.log"

export PYTHONUNBUFFERED=1
export RAY_DEDUP_LOGS=0
export VLLM_DEQUE_RACE_DELAY_S=0.005
export PYTHONPATH="/tmp/vllm-deque-repro${PYTHONPATH:+:${PYTHONPATH}}"

uv run --frozen examples/run_grpo.py \
    --config examples/configs/grpo_math_1B.yaml \
    policy.generation.vllm_cfg.async_engine=true \
    +policy.generation.vllm_cfg.expose_http_server=true \
    policy.generation.colocated.enabled=false \
    policy.generation.colocated.resources.num_nodes=1 \
    policy.generation.colocated.resources.gpus_per_node=2 \
    grpo.async_grpo.enabled=true \
    grpo.async_grpo.in_flight_weight_updates=true \
    loss_fn.use_importance_sampling_correction=true \
    cluster.num_nodes=1 \
    cluster.gpus_per_node=4 \
    grpo.max_num_steps=5 \
    grpo.val_period=1000 \
    checkpointing.enabled=false \
    logger.log_dir="${RUN_DIR}" \
    logger.wandb_enabled=false \
    >"${TRAIN_LOG}" 2>&1 &
TRAIN_PID=$!

BASE_URL=
for _ in $(seq 1 180); do
    BASE_URL=$(grep -m1 -Eo 'Starting server on http://[^ ]+/v1' "${TRAIN_LOG}" | sed 's/^Starting server on //' || true)
    [[ -n "${BASE_URL}" ]] && break
    kill -0 "${TRAIN_PID}" 2>/dev/null || break
    sleep 5
done
test -n "${BASE_URL}"
export BASE_URL

seq 1 128 | xargs -I{} -P 16 sh -c '
    code=$(curl -sS --max-time 180 -o /dev/null -w "%{http_code}" \
        -X POST "${BASE_URL}/chat/completions" \
        -H "Content-Type: application/json" \
        -d "{\"model\":\"Qwen/Qwen2.5-1.5B\",\"messages\":[{\"role\":\"user\",\"content\":\"Give one short sentence about event loops. Request {}.\"}],\"max_tokens\":16,\"temperature\":1.0,\"top_p\":1.0}") || exit 1
    test "${code}" = 200 && echo OK
' >"${HTTP_LOG}" 2>&1 &
HTTP_PID=$!

wait "${HTTP_PID}"; HTTP_RC=$?
wait "${TRAIN_PID}"; TRAIN_RC=$?

echo "http_rc=${HTTP_RC} train_rc=${TRAIN_RC} completed=$(grep -c '^OK$' "${HTTP_LOG}" || true)"
grep -E 'VLLM_DEQUE_RACE_HOOK_HIT|AsyncLLM output_handler failed|deque mutated during iteration|Performing policy generation refit' "${TRAIN_LOG}"

The hook does not invent a new state. It wraps Event.set() only when vLLM's real RequestOutputCollector calls it. It pauses the setter while the real waiter runs on its owner loop. Both deque operations already occur in the unmodified serving path; the hook only makes their overlap reliable.

Results:

• Base: the hook found a foreign-loop waiter. The run failed after 114 successful HTTP requests with RuntimeError: deque mutated during iteration from AsyncLLM.output_handler.
• PR: all 128 HTTP requests, five training steps, and five refits completed. The hook ran, found no foreign-loop waiter, and the deque error did not occur.

[jepio]

/ok to test d0ba2a7

[yuki-97]

Scope: the vLLM async HTTP-generation loop fix — _AsyncLLMHTTPClient, the post_init_async startup move, the sparse-refit loop rebinding, and the shutdown() reorder.

Checked:

• Cross-loop safety — traced RequestOutputCollector.put/get and asyncio.Event.set/wait against pinned vLLM 0.25.1 and CPython 3.13; the marshaled path puts both ends of the event on the engine loop.
• The _install_engine_input_socket_lock removal — enumerated every remaining thread that can reach the engine (uvicorn, sparse-refit pools, ZMQ relay, metrics logger, asyncio.to_thread) plus vLLM's own client threads; no cross-thread input_socket sender survives.
• Proxy surface completeness — every engine_client member the two mounted routes can reach in pinned vLLM 0.25.1.
• Startup / shutdown ordering — every URL reader runs after _post_init(), and the uvicorn join is bounded caller-side by WorkerGroup.shutdown(timeout=30.0) plus an unconditional ray.kill.

Remaining comments are three nits: two on the readability of the proxy's contract, one on a gap in the happy-path proxy test.

[jepio]

/ok to test 734175b

[jepio]

/ok to test 734175b

[jepio]

@yuki-97 incorporated changes based on your review