test(disagg): finish() latency breakdown instrumentation (lever A A5 root-cause)

Log per-request finish breakdown (wait_workers / fallback / wait_rdma) to prove
WHY the per-layer transfer stage (~193ms) doesn't shrink vs the monolithic (~172ms)
— i.e. whether the workers fall behind the forward (submits land late, no overlap)
or the RDMA itself is the cost. INFO-gated.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

python/sglang/srt/disaggregation/cp_per_layer_transfer.py

@@ -19,6 +19,7 @@ from __future__ import annotations
 
 import logging
 import threading
+import time
 from typing import Callable, List, Optional, Tuple
 
 logger = logging.getLogger(__name__)
@@ -141,6 +142,7 @@ class PerLayerTransferContext:
     def finish(self, timeout: float = _FINISH_TIMEOUT_S) -> int:
         """Wait the notifier-fired transfers, synchronously cover any un-fired layers,
         then wait all submitted. Returns 0 on success, -1 on any failure."""
+        t0 = time.perf_counter()
         with self._cond:
             target = self._enqueued
             completed = self._cond.wait_for(
@@ -150,21 +152,33 @@ class PerLayerTransferContext:
                 L for L in range(self.num_layers) if L not in self._submitted_layers
             ]
             failed = self._failed
+            submitted_at_entry = len(self._batch_ids)
+        t1 = time.perf_counter()  # workers drained
         if missing:
-            logger.debug(
-                "[CP_PER_LAYER_TRANSFER] finish: %d notifier-missed layer(s) sent sync",
-                len(missing),
-            )
             for layer_id in missing:
                 with self._cond:
                     if layer_id in self._submitted_layers:
                         continue
                     self._submitted_layers.add(layer_id)
                 self._submit_one(layer_id)  # post-forward: written, no event needed
+        t2 = time.perf_counter()  # fallback submitted
         with self._cond:
             failed = failed or self._failed
             batch_ids = list(self._batch_ids)
         wait_status = self.engine.wait_batch_transfers(batch_ids)
+        t3 = time.perf_counter()  # RDMA complete
+        if logger.isEnabledFor(logging.INFO):
+            logger.info(
+                "[CP_PER_LAYER_TRANSFER] finish breakdown: target=%d submitted_at_entry=%d "
+                "missing=%d wait_workers=%.1fms fallback=%.1fms wait_rdma=%.1fms total=%.1fms",
+                target,
+                submitted_at_entry,
+                len(missing),
+                (t1 - t0) * 1e3,
+                (t2 - t1) * 1e3,
+                (t3 - t2) * 1e3,
+                (t3 - t0) * 1e3,
+            )
         if failed or not completed:
             return -1
         return wait_status