Move log_prefill_stats_late to correct location in PP mode (#15946)
This commit is contained in:
@@ -937,8 +937,7 @@ class SchedulerDisaggregationDecodeMixin:
|
|||||||
# 2. decode + prebuilt -> decode + idle (idle forward, prebuilt returns)
|
# 2. decode + prebuilt -> decode + idle (idle forward, prebuilt returns)
|
||||||
# 3. prebuilt + None -> None (None forward, prebuilt returns) + None
|
# 3. prebuilt + None -> None (None forward, prebuilt returns) + None
|
||||||
# 4. prebuilt + decode + None -> idle (idle forward, prebuilt returns) + decode + idle
|
# 4. prebuilt + decode + None -> idle (idle forward, prebuilt returns) + decode + idle
|
||||||
if self.require_mlp_sync:
|
ret = self.maybe_prepare_mlp_sync_batch_and_log_stats(ret)
|
||||||
ret = self.prepare_mlp_sync_batch(ret)
|
|
||||||
|
|
||||||
if ret:
|
if ret:
|
||||||
trace_event_batch("schedule", ret.reqs)
|
trace_event_batch("schedule", ret.reqs)
|
||||||
|
|||||||
@@ -325,14 +325,11 @@ class SchedulerDisaggregationPrefillMixin:
|
|||||||
self.process_prefill_chunk()
|
self.process_prefill_chunk()
|
||||||
|
|
||||||
batch = self.get_new_batch_prefill()
|
batch = self.get_new_batch_prefill()
|
||||||
if self.require_mlp_sync:
|
batch = self.maybe_prepare_mlp_sync_batch_and_log_stats(batch)
|
||||||
batch = self.prepare_mlp_sync_batch(batch)
|
|
||||||
|
|
||||||
if batch:
|
if batch:
|
||||||
trace_event_batch("schedule", batch.reqs)
|
trace_event_batch("schedule", batch.reqs)
|
||||||
|
|
||||||
self.log_prefill_stats_late(batch)
|
|
||||||
|
|
||||||
return batch
|
return batch
|
||||||
|
|
||||||
@torch.no_grad()
|
@torch.no_grad()
|
||||||
@@ -523,7 +520,6 @@ class SchedulerDisaggregationPrefillMixin:
|
|||||||
)
|
)
|
||||||
|
|
||||||
self.maybe_send_health_check_signal()
|
self.maybe_send_health_check_signal()
|
||||||
self.log_prefill_stats_late(batch)
|
|
||||||
|
|
||||||
def process_disagg_prefill_inflight_queue(
|
def process_disagg_prefill_inflight_queue(
|
||||||
self: Scheduler, rids_to_check: Optional[List[str]] = None
|
self: Scheduler, rids_to_check: Optional[List[str]] = None
|
||||||
|
|||||||
@@ -1835,7 +1835,9 @@ class Scheduler(
|
|||||||
# Before merging the new batch into running batch:
|
# Before merging the new batch into running batch:
|
||||||
# 1. All new batches are none -> need_mlp_sync remains true (sync is needed for decode batch).
|
# 1. All new batches are none -> need_mlp_sync remains true (sync is needed for decode batch).
|
||||||
# 2. All new batches are some (prefill / idle) -> we do not need prepare mlp sync one more time.
|
# 2. All new batches are some (prefill / idle) -> we do not need prepare mlp sync one more time.
|
||||||
new_batch = self.prepare_mlp_sync_batch(new_batch)
|
new_batch = self.maybe_prepare_mlp_sync_batch_and_log_stats(
|
||||||
|
new_batch, log_stats=False
|
||||||
|
)
|
||||||
need_mlp_sync = new_batch is None
|
need_mlp_sync = new_batch is None
|
||||||
|
|
||||||
if new_batch is not None:
|
if new_batch is not None:
|
||||||
@@ -1849,15 +1851,14 @@ class Scheduler(
|
|||||||
else:
|
else:
|
||||||
ret = None
|
ret = None
|
||||||
|
|
||||||
# Handle DP attention
|
# Handle DP attention and log stats
|
||||||
if need_mlp_sync:
|
ret = self.maybe_prepare_mlp_sync_batch_and_log_stats(
|
||||||
ret = self.prepare_mlp_sync_batch(ret)
|
ret, need_sync=need_mlp_sync
|
||||||
|
)
|
||||||
|
|
||||||
if ret:
|
if ret:
|
||||||
trace_event_batch("schedule", ret.reqs)
|
trace_event_batch("schedule", ret.reqs)
|
||||||
|
|
||||||
self.log_prefill_stats_late(ret)
|
|
||||||
|
|
||||||
return ret
|
return ret
|
||||||
|
|
||||||
def get_num_allocatable_reqs(self, running_bs):
|
def get_num_allocatable_reqs(self, running_bs):
|
||||||
|
|||||||
@@ -207,6 +207,27 @@ class SchedulerDPAttnMixin:
|
|||||||
offload_tags=self.offload_tags,
|
offload_tags=self.offload_tags,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
def maybe_prepare_mlp_sync_batch_and_log_stats(
|
||||||
|
self: Scheduler,
|
||||||
|
batch: Optional[ScheduleBatch],
|
||||||
|
need_sync: Optional[bool] = None,
|
||||||
|
log_stats: bool = True,
|
||||||
|
) -> Optional[ScheduleBatch]:
|
||||||
|
"""
|
||||||
|
Helper to pair log_prefill_stats with log_prefill_stats_late.
|
||||||
|
Should be called after get_new_batch_prefill() to ensure proper pairing.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
batch: The batch to process
|
||||||
|
need_sync: If specified, overrides self.require_mlp_sync for prepare_mlp_sync_batch decision
|
||||||
|
log_stats: Whether to call log_prefill_stats_late. Set to False for intermediate calls.
|
||||||
|
"""
|
||||||
|
if need_sync if need_sync is not None else self.require_mlp_sync:
|
||||||
|
batch = self.prepare_mlp_sync_batch(batch)
|
||||||
|
if log_stats:
|
||||||
|
self.log_prefill_stats_late(batch)
|
||||||
|
return batch
|
||||||
|
|
||||||
def get_idle_batch(self: Scheduler) -> ScheduleBatch:
|
def get_idle_batch(self: Scheduler) -> ScheduleBatch:
|
||||||
idle_batch = ScheduleBatch.init_new(
|
idle_batch = ScheduleBatch.init_new(
|
||||||
[],
|
[],
|
||||||
|
|||||||
@@ -219,8 +219,7 @@ class SchedulerPPMixin:
|
|||||||
|
|
||||||
self.process_prefill_chunk()
|
self.process_prefill_chunk()
|
||||||
batch = self.get_new_batch_prefill()
|
batch = self.get_new_batch_prefill()
|
||||||
if self.require_mlp_sync:
|
batch = self.maybe_prepare_mlp_sync_batch_and_log_stats(batch)
|
||||||
batch = self.prepare_mlp_sync_batch(batch)
|
|
||||||
self.mbs[mb_id] = batch
|
self.mbs[mb_id] = batch
|
||||||
self.running_mbs[mb_id] = self.running_batch
|
self.running_mbs[mb_id] = self.running_batch
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user