[Serve][LLM] Support concurrent streaming in SGLang completions() endpoint

[Truc54]

Description

This PR addresses issue #63901 by refactoring the streaming path in SGLangServer.completions() to process concurrent generation prompts in parallel instead of sequentially.

Why is this needed?

In the current implementation of the SGLang server engine, completing multiple prompts with stream=True runs a sequential for loop over the prompts. This means prompt i must stream to completion before prompt i+1 starts generating. This sequential bottleneck dramatically limits throughput and increases request latency under concurrent load.

Solution:

• Refactored the request.stream branch in SGLangServer.completions() to execute _stream_generate concurrently for all prompts using asyncio.create_task().
• Collected yielded text delta chunks from all concurrent tasks into a single shared asyncio.Queue.
• Drained the queue in the main async generator and yielded SSE chunks as they arrive, enabling natural interleaving.
• Added proper exception propagation from workers to the main generator.
• Added a finally block to cancel outstanding tasks if the generator is closed or an error occurs, preventing task leaks.

Related issues

Closes #63901

Additional information

Implementation details:

1. Concurrency (asyncio.Queue): Parallelized completion streams using an asynchronous producer-consumer queue pattern.
2. Robustness & Cleanup: Active tasks are tracked and canceled in case of early client disconnection or errors.
3. Tests: Added a new CPU-based mock unit test suite in python/ray/llm/tests/serve/cpu/deployments/test_sglang_server.py that utilizes a meta-path importer hook to bypass native compilation dependencies.

Both tests pass successfully:

• test_concurrent_streaming_completions (Verifies chunks are properly interleaved and delta-decoded).
• test_concurrent_streaming_completions_exception_handling (Verifies exception propagation and task cancellation).

[gemini-code-assist]

Code Review

This pull request refactors the SGLang engine's streaming completions to process multiple prompts concurrently using an asyncio queue and background producer tasks, rather than sequentially. It also adds comprehensive unit tests to verify concurrent streaming and exception handling. The review feedback suggests awaiting the cancelled background tasks in the finally block to prevent potential resource leaks or unhandled task exception warnings.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

When cancelling background tasks in the finally block, it is important to await them (e.g., using asyncio.gather with return_exceptions=True). Otherwise, the tasks might continue running their cleanup/finally blocks in the background after the generator has returned, which can lead to resource leaks, race conditions, or "Task exception was never retrieved" warnings if any task failed.

Suggested change

	finally:
	for task in tasks:
	if not task.done():
	task.cancel()
	finally:
	for task in tasks:
	if not task.done():
	task.cancel()
	if tasks:
	await asyncio.gather(*tasks, return_exceptions=True)

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

^{Reviewed by Cursor Bugbot for commit 5ac4f52. Configure here.}

[cursor]

None sentinel masks producer failures

Medium Severity

Each streaming producer always enqueues None in finally after the try/except, but only Exception subclasses are forwarded on the queue. Failures such as asyncio.CancelledError or KeyboardInterrupt bypass except Exception, yet still emit the completion sentinel, so the main loop can finish as if every prompt completed when a worker actually aborted.

 

^{Reviewed by Cursor Bugbot for commit 5ac4f52. Configure here.}