Skip to content

[PD Disaggregation] Limit prefill fetch num with FD_MAX_INFLIGHT_PREFILL #7981

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cmcamdy merged 3 commits into from
Jun 4, 2026
Merged

[PD Disaggregation] Limit prefill fetch num with FD_MAX_INFLIGHT_PREFILL #7981

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
23069b5
limit prefill fetch num
cmcamdy Jun 2, 2026
13b1a95
fix unittest
cmcamdy Jun 3, 2026
38ac7e9
fix unittest
cmcamdy Jun 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 17 additions & 4 deletions fastdeploy/engine/common_engine.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -912,10 +912,23 @@ def _fetch_request():
with self._pause_cond:
self._pause_cond.wait_for(lambda: not self.is_paused)
nonlocal is_fetching
num_prefill_batch = min(
int(self.resource_manager.available_batch()),
self.cfg.max_prefill_batch,
)
if self.cfg.scheduler_config.splitwise_role == "prefill":
Comment thread
cmcamdy marked this conversation as resolved.
max_inflight_prefill = envs.FD_MAX_INFLIGHT_PREFILL
inflight_prefill = len(self.resource_manager.running)
if inflight_prefill >= max_inflight_prefill:
is_fetching = False

This comment was marked as outdated.

Sign in to view

return
available_for_new = max_inflight_prefill - inflight_prefill
num_prefill_batch = min(
int(self.resource_manager.available_batch()),
self.cfg.max_prefill_batch,
available_for_new,
)
else:
num_prefill_batch = min(
int(self.resource_manager.available_batch()),
self.cfg.max_prefill_batch,
)

if self.cfg.scheduler_config.splitwise_role != "mixed":
max_num_batched_tokens = self.cfg.scheduler_config.max_num_batched_tokens
Expand Down
1 change: 1 addition & 0 deletions fastdeploy/envs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,7 @@ def _validate_split_kv_size(value: int) -> int:
# "Enable FP8 calibration on HPU"
"FD_HPU_MEASUREMENT_MODE": lambda: os.getenv("FD_HPU_MEASUREMENT_MODE", "0"),
"FD_PREFILL_WAIT_DECODE_RESOURCE_SECONDS": lambda: int(os.getenv("FD_PREFILL_WAIT_DECODE_RESOURCE_SECONDS", "30")),
"FD_MAX_INFLIGHT_PREFILL": lambda: int(os.getenv("FD_MAX_INFLIGHT_PREFILL", "20")),

This comment was marked as outdated.

Sign in to view

"FD_ENABLE_REQUEST_DISCONNECT_STOP_INFERENCE": lambda: int(
os.getenv("FD_ENABLE_REQUEST_DISCONNECT_STOP_INFERENCE", "1")
),
Expand Down
132 changes: 130 additions & 2 deletions tests/engine/test_common_engine.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -333,17 +333,18 @@ def get_real_bsz(self):
return DummyRM()

@staticmethod
def _make_v1_prefill_continuous_rm(eng, waiting_async_result=False):
def _make_v1_prefill_continuous_rm(eng, waiting_async_result=False, available_batch=1):
class DummyRM:
def __init__(self):
self.abort_req_ids_set = set()
self.waiting = []
self.running = []
Comment thread
cmcamdy marked this conversation as resolved.
self.real_bsz = 1
self.add_request_in_p = Mock()
self.pre_recycle_resource = Mock()

def available_batch(self):
return 1
return available_batch

def apply_async_preprocess(self, _task):
return None
Expand Down Expand Up @@ -1497,6 +1498,133 @@ def test_schedule_request_to_worker_v1_prefill_decode_alloc_error_safe(self):
eng.resource_manager.add_request_in_p.assert_not_called()
self._detach_finalizer(eng)

def test_schedule_request_to_worker_v1_prefill_max_inflight_skip_fetch(self):
"""When len(running) >= FD_MAX_INFLIGHT_PREFILL, fetch should be skipped."""
cfg = self._make_cfg(
splitwise_role="prefill",
num_gpu_blocks_override=4,
router="0.0.0.0:30000",
kv_cache_ratio=1,
)
eng = self._make_engine(cfg)
self._setup_v1_engine(eng)

eng.scheduler = Mock(get_requests=Mock(return_value=[]), put_results=Mock())
eng.engine_worker_queue = Mock(
exist_tasks=Mock(return_value=False),
get_finished_add_cache_task_req=Mock(return_value=[]),
)

rm = self._make_v1_prefill_continuous_rm(eng, waiting_async_result=False)
rm.running = [Mock() for _ in range(20)] # running == FD_MAX_INFLIGHT_PREFILL default
eng.resource_manager = rm
eng.split_connector = Mock(
send_splitwise_tasks=Mock(),
check_decode_allocated=Mock(return_value=(True, "")),
send_cache_info_to_messager=Mock(),
)

try:
with (
patch("fastdeploy.engine.common_engine.envs.FD_MAX_INFLIGHT_PREFILL", 20),
patch("fastdeploy.engine.common_engine.envs.PREFILL_CONTINUOUS_REQUEST_DECODE_RESOURCES", False),
patch("fastdeploy.engine.common_engine.ThreadPoolExecutor", self._make_dummy_executor(eng)),
patch("fastdeploy.engine.common_engine.time.sleep", lambda *_: None),
):
eng._schedule_request_to_worker_v1()
finally:
eng.running = False

# get_requests should NOT be called because fetch was skipped
eng.scheduler.get_requests.assert_not_called()
self._detach_finalizer(eng)

def test_schedule_request_to_worker_v1_prefill_inflight_constrains_batch(self):
"""When 0 < len(running) < FD_MAX_INFLIGHT_PREFILL, num_prefill_batch is constrained by available_for_new."""
cfg = self._make_cfg(
splitwise_role="prefill",
num_gpu_blocks_override=4,
router="0.0.0.0:30000",
kv_cache_ratio=1,
)
eng = self._make_engine(cfg)
self._setup_v1_engine(eng)

eng.scheduler = Mock(get_requests=Mock(return_value=[]), put_results=Mock())
eng.engine_worker_queue = Mock(
exist_tasks=Mock(return_value=False),
get_finished_add_cache_task_req=Mock(return_value=[]),
)

rm = self._make_v1_prefill_continuous_rm(eng, waiting_async_result=False, available_batch=10)
rm.running = [Mock() for _ in range(18)] # 18 in-flight, max=20, so available_for_new=2
eng.resource_manager = rm
eng.split_connector = Mock(
send_splitwise_tasks=Mock(),
check_decode_allocated=Mock(return_value=(True, "")),
send_cache_info_to_messager=Mock(),
)

try:
with (
patch("fastdeploy.engine.common_engine.envs.FD_MAX_INFLIGHT_PREFILL", 20),
patch("fastdeploy.engine.common_engine.envs.PREFILL_CONTINUOUS_REQUEST_DECODE_RESOURCES", False),
patch("fastdeploy.engine.common_engine.ThreadPoolExecutor", self._make_dummy_executor(eng)),
patch("fastdeploy.engine.common_engine.time.sleep", lambda *_: None),
):
eng._schedule_request_to_worker_v1()
finally:
eng.running = False

# get_requests should be called with batch=2 (available_for_new=20-18)
eng.scheduler.get_requests.assert_called_once()
call_kwargs = eng.scheduler.get_requests.call_args
self.assertEqual(call_kwargs.kwargs.get("batch", call_kwargs[1].get("batch")), 2)
self._detach_finalizer(eng)

def test_schedule_request_to_worker_v1_prefill_inflight_boundary_last_slot(self):
"""When len(running) == max_inflight_prefill - 1, only 1 slot remains (boundary)."""
cfg = self._make_cfg(
splitwise_role="prefill",
num_gpu_blocks_override=4,
router="0.0.0.0:30000",
kv_cache_ratio=1,
)
eng = self._make_engine(cfg)
self._setup_v1_engine(eng)

eng.scheduler = Mock(get_requests=Mock(return_value=[]), put_results=Mock())
eng.engine_worker_queue = Mock(
exist_tasks=Mock(return_value=False),
get_finished_add_cache_task_req=Mock(return_value=[]),
)

rm = self._make_v1_prefill_continuous_rm(eng, waiting_async_result=False, available_batch=10)
rm.running = [Mock() for _ in range(19)] # 19 in-flight, max=20, so available_for_new=1
eng.resource_manager = rm
eng.split_connector = Mock(
send_splitwise_tasks=Mock(),
check_decode_allocated=Mock(return_value=(True, "")),
send_cache_info_to_messager=Mock(),
)

try:
with (
patch("fastdeploy.engine.common_engine.envs.FD_MAX_INFLIGHT_PREFILL", 20),
patch("fastdeploy.engine.common_engine.envs.PREFILL_CONTINUOUS_REQUEST_DECODE_RESOURCES", False),
patch("fastdeploy.engine.common_engine.ThreadPoolExecutor", self._make_dummy_executor(eng)),
patch("fastdeploy.engine.common_engine.time.sleep", lambda *_: None),
):
eng._schedule_request_to_worker_v1()
finally:
eng.running = False

# get_requests should be called with batch=1 (available_for_new=20-19)
eng.scheduler.get_requests.assert_called_once()
call_kwargs = eng.scheduler.get_requests.call_args
self.assertEqual(call_kwargs.kwargs.get("batch", call_kwargs[1].get("batch")), 1)
self._detach_finalizer(eng)

def test_schedule_request_to_worker_v1_decode_preempted_and_errors(self):
cfg = self._make_cfg(
splitwise_role="decode",
Expand Down
Loading