diff --git a/python/sglang/srt/disaggregation/cp_per_layer_transfer.py b/python/sglang/srt/disaggregation/cp_per_layer_transfer.py index d6a1a318c..ec3c87e84 100644 --- a/python/sglang/srt/disaggregation/cp_per_layer_transfer.py +++ b/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