[PD] NIXL connector: shard by destination (#32025)

python/sglang/srt/disaggregation/nixl/conn.py

@@ -2254,13 +2254,16 @@ class NixlKVManager(CommonKVManager):
         if self.enable_staging:
             self._prefetch_staging_reqs(bootstrap_room)
 
-        # Transfer is async: just enqueue the chunk; the per-queue worker
+        if bootstrap_room not in self.transfer_infos:
-        # (transfer_worker) does the actual gather + RDMA. Routing by
+            # Dummy rank or already cleared; nothing to enqueue.
-        # ``room % N`` keeps every chunk of a given room on the same
+            return None
-        # worker -- and therefore on the same private staging buffer --
+
-        # which is required for the staging ring's offset/watermark
+        # Shard by destination (mirror mooncake): same dst endpoint(s) -> same
-        # state machine to advance correctly.
+        # worker, keeping a room's chunks on one private staging buffer.
-        shard_idx = bootstrap_room % len(self.transfer_queues)
+        session_port_sum = sum(
+            info.dst_port for info in self.transfer_infos[bootstrap_room].values()
+        )
+        shard_idx = session_port_sum % len(self.transfer_queues)
         self.transfer_queues[shard_idx].put(
             TransferKVChunk(
                 room=bootstrap_room,