[P/D] Refactor mooncake connector sender thread using async coroutines

[dtcccc]

Purpose

This is a separate PR for #31034 to help review.
This PR refactored sender thread using async coroutines. All related data are in the same thread so that we can drop their locks. This makes the sender thread simple and easy to maintain.

Test Plan

Test Result

---

Essential Elements of an Effective PR Description Checklist

• The purpose of the PR, such as "Fix some issue (link existing issues this PR will resolve)".
• The test plan, such as providing test command.
• The test results, such as pasting the results comparison before and after, or e2e results
• (Optional) The necessary documentation update, such as updating supported_models.md and examples for a new model.
• (Optional) Release notes update. If your change is user facing, please update the release notes draft in the Google Doc.

---

Note

^{Cursor Bugbot is generating a summary for commit a64dc77. Configure here.}

[gemini-code-assist]

Code Review

This pull request refactors the Mooncake connector's sender thread to use asyncio coroutines. This is a significant improvement that simplifies the code by removing manual lock management and leveraging modern asynchronous patterns. The state related to sending operations is now managed within a single event loop, which is a robust way to prevent race conditions. The changes are well-structured and follow best practices for integrating asyncio with threads and blocking operations. However, I found a critical issue where a silent failure can occur if a requested transfer ID is not found, potentially leading to data inconsistencies. I've provided a suggestion to fix this by raising an exception instead of returning silently.

[gemini-code-assist]

Returning silently when a request ID is not found can lead to silent failures. The client that initiated the transfer will receive a TRANS_DONE message, making it believe the transfer was successful, while it was actually aborted. This can lead to data inconsistencies or hangs. It's better to raise an exception to signal an error condition, which will then be caught by the _sender_worker and result in a TRANS_ERROR message being sent back to the client.

Suggested change

	return
	raise ValueError(f"Request {req_id} not found in reqs_need_send")

[NickLucche]

Thanks a lot for separating this change @dtcccc , this looks like good!
Only left a few minor comments for style+clarity

[NickLucche]

LGTM

[cursor]

Race condition between async scheduling and sender workers

Medium Severity

The refactoring replaces synchronous lock-protected updates with asyncio.run_coroutine_threadsafe, which schedules record_send_reqs but returns immediately before execution. Since _sender_worker tasks are already running on sender_loop and processing incoming ZMQ requests, if a decoder request arrives before record_send_reqs has executed, send_kv_to_decode will fail to find the request in reqs_need_send, log a warning, and return early without transferring data. The old code used threading.Lock to ensure the dict was updated before being read. The new code provides no such guarantee.

Additional Locations (1)

• vllm/distributed/kv_transfer/kv_connector/v1/mooncake_connector.py#L568-L573