From 23069b569e59151778799a0eb727f648d8f38307 Mon Sep 17 00:00:00 2001 From: chenhuan09 Date: Tue, 2 Jun 2026 19:07:59 +0800 Subject: [PATCH 1/3] limit prefill fetch num --- fastdeploy/engine/common_engine.py | 21 +++++++++++++++++---- fastdeploy/envs.py | 1 + 2 files changed, 18 insertions(+), 4 deletions(-) diff --git a/fastdeploy/engine/common_engine.py b/fastdeploy/engine/common_engine.py index b4a60e93bec..6d4624e33f3 100644 --- a/fastdeploy/engine/common_engine.py +++ b/fastdeploy/engine/common_engine.py @@ -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": + max_inflight_prefill = envs.FD_MAX_INFLIGHT_PREFILL + inflight_prefill = len(self.resource_manager.running) + if inflight_prefill >= max_inflight_prefill: + is_fetching = False + 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 diff --git a/fastdeploy/envs.py b/fastdeploy/envs.py index 5fd73e1962b..8b6bbf46f4b 100644 --- a/fastdeploy/envs.py +++ b/fastdeploy/envs.py @@ -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")), "FD_ENABLE_REQUEST_DISCONNECT_STOP_INFERENCE": lambda: int( os.getenv("FD_ENABLE_REQUEST_DISCONNECT_STOP_INFERENCE", "1") ), From 13b1a95e05246b90f25d955f4dcfa6b25be37205 Mon Sep 17 00:00:00 2001 From: chenhuan09 Date: Wed, 3 Jun 2026 14:55:56 +0800 Subject: [PATCH 2/3] fix unittest --- tests/engine/test_common_engine.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/engine/test_common_engine.py b/tests/engine/test_common_engine.py index 4b539555a34..3edff3eb916 100644 --- a/tests/engine/test_common_engine.py +++ b/tests/engine/test_common_engine.py @@ -338,6 +338,7 @@ class DummyRM: def __init__(self): self.abort_req_ids_set = set() self.waiting = [] + self.running = [] self.real_bsz = 1 self.add_request_in_p = Mock() self.pre_recycle_resource = Mock() From 38ac7e9f9c68a973b806f4617752bf14c244ef23 Mon Sep 17 00:00:00 2001 From: chenhuan09 Date: Wed, 3 Jun 2026 18:41:56 +0800 Subject: [PATCH 3/3] fix unittest --- tests/engine/test_common_engine.py | 131 ++++++++++++++++++++++++++++- 1 file changed, 129 insertions(+), 2 deletions(-) diff --git a/tests/engine/test_common_engine.py b/tests/engine/test_common_engine.py index 3edff3eb916..e1e3cdd5673 100644 --- a/tests/engine/test_common_engine.py +++ b/tests/engine/test_common_engine.py @@ -333,7 +333,7 @@ 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() @@ -344,7 +344,7 @@ def __init__(self): self.pre_recycle_resource = Mock() def available_batch(self): - return 1 + return available_batch def apply_async_preprocess(self, _task): return None @@ -1498,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",