fix: match P/D requests by transfer_id in FlagCX connector

[leoda1]

PR Category

Core

PR Type

Bug Fixes

Description

In disaggregated (P/D) serving, the Prefill and Decode engines assign different request_ids to the same logical request. The FlagCX connector previously used request_id as the key for its send/receive bookkeeping, so the Decode side's req_id could never be found in the Prefill side's reqs_need_send map, causing KV transfers to silently fail to match (Request %s not found in reqs_need_send).

This PR introduces a stable, cross-engine transfer_id that both sides agree on, rekeys all connector bookkeeping by it, and hardens the transfer path with proper timeouts.

Related Issues

Changes

• flagcx_connector.py: added a TransferId type and threaded transfer_id through RecvReqMeta, SendBlockMeta, SendReqMeta, and the connector metadata. reqs_to_send / req_blocks now carry (transfer_id, block_ids); the worker looks up send metadata by transfer_id while still reporting completion/timeout under the Prefill request_id (p_req_id) that vLLM expects.
• flagcx_connector.py: guarded against a missing transfer_id — the scheduler logs a warning and skips the request instead of mis-matching.
• flagcx_connector.py: replaced unbounded / hardcoded 60s waits with bounded waits driven by _abort_request_timeout — send_meta.ready.wait() uses a deadline and raises a clear RuntimeError on timeout, and the receive socket RCVTIMEO now scales with _abort_request_timeout.
• examples/disaggregated_serving_xpyd/router.py: inject transfer_id = f"fgx-{request_id}" into kv_transfer_params for both the prefill and decode legs so both engines share the same transfer key.
• vllm_fl/distributed/device_communicators/flagcx.py: fixed ctypes usage for flagcxGetUniqueId() / flagcxCommInitRank() — pass the unique id object directly instead of .contents / ctypes.byref(...).
• vllm_fl/ops/fused_moe/fused_moe_utils.py: provide a default MoE backend priority (TRITON, BATCHED_TRITON) for out-of-tree platforms.

Testing

Checklist

• I have run the existing tests and they pass
• I have added tests for my changes (if applicable)
• I have updated the documentation (if applicable)