From 49f64d43f92b9c4078923e43e5ff0cf87909d702 Mon Sep 17 00:00:00 2001 From: wangzaijun Date: Thu, 3 Sep 2026 09:55:40 +0000 Subject: [PATCH 01/12] feat(pd): avoid prioritizing short cache-hit requests --- docs/CN/source/tutorial/api_server_args.rst | 6 +++-- docs/EN/source/tutorial/api_server_args.rst | 10 +++++--- .../httpserver_for_pd_master/manager.py | 12 ++++++--- lightllm/utils/envs_utils.py | 6 +++++ .../test_pd_master_multi_choice.py | 25 ++++++++++++++----- .../test_pd_master_cached_tokens.py | 1 + unit_tests/utils/test_envs_utils.py | 19 ++++++++++++++ 7 files changed, 64 insertions(+), 15 deletions(-) diff --git a/docs/CN/source/tutorial/api_server_args.rst b/docs/CN/source/tutorial/api_server_args.rst index 746827b10..536901764 100644 --- a/docs/CN/source/tutorial/api_server_args.rst +++ b/docs/CN/source/tutorial/api_server_args.rst @@ -104,11 +104,13 @@ PD 分离模式参数 ``pd_high_priority_request_time_out_seconds`` 下发一个统一的超时时间下限。P/D 节点分别取 该值与本地 ``shm_req``、Router 超时的较大值；该字段为 0 时不延长本地超时。PD Master 下发值由 ``LIGHTLLM_PD_HIGH_PRIORITY_REQUEST_TIMEOUT_SECONDS`` 控制，默认 60 秒。cache 命中记录允许提升优先级的 - 最大年龄由 ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MAX_AGE_SECONDS`` 控制，默认 16 秒。本地请求限流默认关闭。 + 最大年龄由 ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MAX_AGE_SECONDS`` 控制，默认 16 秒。cache 命中提权还要求输入 + token 数达到 ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MIN_PROMPT_TOKENS`` 配置的门槛（默认 4096），避免短请求仅因 + cache 命中率高而提升优先级。本地请求限流默认关闭。 .. option:: --disable_pd_cache_high_priority - 禁止 PD Master 将预计输入 cache 命中率高且命中记录仍然新鲜的首段请求提升为高优先级。 + 禁止 PD Master 将输入足够长、预计输入 cache 命中率高且命中记录仍然新鲜的首段请求提升为高优先级。 该参数不影响 PD Decode 容量不足后的分段续跑请求；续跑请求仍保持高优先级。默认不启用， 即默认允许新鲜高 cache 命中请求提升优先级。 diff --git a/docs/EN/source/tutorial/api_server_args.rst b/docs/EN/source/tutorial/api_server_args.rst index 81b3bb752..d3db0970e 100644 --- a/docs/EN/source/tutorial/api_server_args.rst +++ b/docs/EN/source/tutorial/api_server_args.rst @@ -109,13 +109,15 @@ PD disaggregation Mode Parameters ``shm_req`` or Router timeout; zero does not extend the local timeout. The value supplied by PD Master is controlled by ``LIGHTLLM_PD_HIGH_PRIORITY_REQUEST_TIMEOUT_SECONDS`` and defaults to 60 seconds. The maximum cache-record age eligible for promotion is controlled by ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MAX_AGE_SECONDS`` and defaults to - 16 seconds. Local request limiting is disabled by default. + 16 seconds. Cache-hit promotion also requires at least the number of input tokens configured by + ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MIN_PROMPT_TOKENS`` (4096 by default), so short requests do not gain priority + solely from a high cache-hit rate. Local request limiting is disabled by default. .. option:: --disable_pd_cache_high_priority - Disable PD Master from promoting first-segment requests whose estimated input cache hit rate is high and whose - cache record is still fresh. This does not affect segmented continuation requests after PD Decode capacity - exhaustion; continuation requests remain high priority. Disabled by default, so fresh high-cache-hit requests + Disable PD Master from promoting sufficiently long first-segment requests whose estimated input cache hit rate + is high and whose cache record is still fresh. This does not affect segmented continuation requests after PD + Decode capacity exhaustion; continuation requests remain high priority. Disabled by default, so eligible requests are promoted unless this option is set. .. option:: --config_server_host diff --git a/lightllm/server/httpserver_for_pd_master/manager.py b/lightllm/server/httpserver_for_pd_master/manager.py index 35a0ecfa8..8e17237ac 100644 --- a/lightllm/server/httpserver_for_pd_master/manager.py +++ b/lightllm/server/httpserver_for_pd_master/manager.py @@ -25,6 +25,7 @@ from lightllm.utils.error_utils import ClientDisconnected, ServerBusyError from lightllm.utils.envs_utils import ( get_pd_cache_high_priority_max_age_seconds, + get_pd_cache_high_priority_min_prompt_tokens, get_pd_high_priority_request_timeout_seconds, ) from lightllm.utils.shm_port_args import get_shm_port_args @@ -55,6 +56,7 @@ def __init__( # P/D 节点传递有限的等待时间，避免资源异常时永久占用请求链路。 self.pd_high_priority_request_time_out_seconds = get_pd_high_priority_request_timeout_seconds() self.pd_cache_high_priority_max_age_seconds = get_pd_cache_high_priority_max_age_seconds() + self.pd_cache_high_priority_min_prompt_tokens = get_pd_cache_high_priority_min_prompt_tokens() self.disable_pd_cache_high_priority = args.disable_pd_cache_high_priority self.tokenizer = get_tokenizer(args.model_dir, args.tokenizer_mode, trust_remote_code=args.trust_remote_code) @@ -198,6 +200,7 @@ async def _generate( request, start_time, origin_group_request_id + choice_index, + input_token_num, ) ) @@ -219,6 +222,7 @@ async def _generate_one( request: Request, start_time: float, origin_request_id: int, + input_token_num: int, ): block_group_request_id = origin_request_id p_node = None @@ -243,6 +247,7 @@ async def _generate_one( and selection_extra_info.estimated_cache_hit_rate > 0.8 and cache_age_seconds is not None and cache_age_seconds <= self.pd_cache_high_priority_max_age_seconds + and input_token_num >= self.pd_cache_high_priority_min_prompt_tokens ) history_gen_token_strs = [] @@ -261,9 +266,10 @@ async def _generate_one( sampling_params.group_request_id = block_group_request_id logger.info(f"pd log gen sub req id {block_group_request_id} for main req id {origin_request_id}") sampling_params.max_new_tokens = remaining_max_new_tokens - # 首段仅在预计输入 cache 命中率高于 0.8 且命中记录仍在有效时间窗内时 - # 提升优先级，避免为可能已被 P 节点淘汰的陈旧 KV cache 插队。第二段及 - # 后续分段仍统一使用高优先级，避免因临时资源紧张导致分段续跑失败。 + # 首段仅在输入达到长度门槛、预计 cache 命中率高于 0.8 且命中记录仍在 + # 有效时间窗内时提升优先级，避免短请求或可能已被 P 节点淘汰的陈旧 + # KV cache 插队。第二段及后续分段仍统一使用高优先级，避免因临时资源 + # 紧张导致分段续跑失败。 sampling_params.pd_high_priority_request = segment_index > 0 or has_fresh_high_cache_hit # 为高优先级请求下发较长的有限等待时间；P/D 节点仅在自身开启 # 本地限流时使用该值，未开启限流时仍保持无限等待。 diff --git a/lightllm/utils/envs_utils.py b/lightllm/utils/envs_utils.py index 0c4f28475..d7713aae8 100644 --- a/lightllm/utils/envs_utils.py +++ b/lightllm/utils/envs_utils.py @@ -330,6 +330,12 @@ def get_pd_cache_high_priority_max_age_seconds() -> int: return max(0, int(os.getenv("LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MAX_AGE_SECONDS", 16))) +@lru_cache(maxsize=None) +def get_pd_cache_high_priority_min_prompt_tokens() -> int: + """cache 命中请求提升为 PD 高优先级时要求的最小 prompt token 数。""" + return max(0, int(os.getenv("LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MIN_PROMPT_TOKENS", 4096))) + + @lru_cache(maxsize=None) def get_lightllm_url_pool_maxsize() -> int: return int(os.getenv("LIGHTLLM_URL_POOL_MAXSIZE", 512)) diff --git a/test/test_pd_selector/test_pd_master_multi_choice.py b/test/test_pd_selector/test_pd_master_multi_choice.py index a44eccba9..53360fb34 100644 --- a/test/test_pd_selector/test_pd_master_multi_choice.py +++ b/test/test_pd_selector/test_pd_master_multi_choice.py @@ -22,6 +22,7 @@ def _manager() -> HttpServerManagerForPDMaster: manager.tokens = MagicMock(return_value=2) manager.pd_high_priority_request_time_out_seconds = 60 manager.pd_cache_high_priority_max_age_seconds = 60 + manager.pd_cache_high_priority_min_prompt_tokens = 4096 manager.disable_pd_cache_high_priority = False return manager @@ -125,7 +126,9 @@ async def generate_one( child_request, start_time, origin_request_id, + input_token_num, ): + assert input_token_num == 2 yield ( origin_request_id, "choice-0", @@ -236,6 +239,7 @@ async def wait_to_token_package(_p_node, _d_node, _start_time, prompt, params, * MagicMock(), 0, 800, + 0, ): results.append(result) @@ -339,6 +343,7 @@ async def failing_wait_to_token_package(*_args, **_kwargs): MagicMock(), 0, 800, + 0, ): pass @@ -412,6 +417,7 @@ async def wait_to_token_package( MagicMock(), 0, 800, + 0, ): results.append(result) @@ -479,6 +485,7 @@ async def wait_to_token_package(_p_node, _d_node, _start_time, _prompt, sampling MagicMock(), 0, 800, + 0, ): pass @@ -492,21 +499,24 @@ async def wait_to_token_package(_p_node, _d_node, _start_time, _prompt, sampling "estimated_cache_hit_rate", "cache_age_seconds", "disable_pd_cache_high_priority", + "input_token_num", "expected_high_priority", ), [ - (0.8, 0.0, False, False), - (0.81, 0.0, False, True), - (0.81, 60.0, False, True), - (0.81, 60.1, False, False), - (0.81, None, False, False), - (0.81, 0.0, True, False), + (0.8, 0.0, False, 4096, False), + (0.81, 0.0, False, 4095, False), + (0.81, 0.0, False, 4096, True), + (0.81, 60.0, False, 4096, True), + (0.81, 60.1, False, 4096, False), + (0.81, None, False, 4096, False), + (0.81, 0.0, True, 4096, False), ], ) def test_pd_master_promotes_only_fresh_high_estimated_cache_hit( estimated_cache_hit_rate, cache_age_seconds, disable_pd_cache_high_priority, + input_token_num, expected_high_priority, ): async def run(): @@ -555,6 +565,7 @@ async def wait_to_token_package(_p_node, _d_node, _start_time, _prompt, sampling MagicMock(), 0, 800, + input_token_num, ): pass @@ -607,6 +618,7 @@ async def wait_to_token_package(_p_node, _d_node, _start_time, _prompt, sampling MagicMock(), 0, 800, + 8192, ): pass @@ -644,6 +656,7 @@ async def wait_to_token_package(*_args, **_kwargs): MagicMock(), 0, 800, + 0, ) assert (await generator.__anext__())[1] == "first" diff --git a/unit_tests/server/httpserver/test_pd_master_cached_tokens.py b/unit_tests/server/httpserver/test_pd_master_cached_tokens.py index 748f62466..ebecbd47e 100644 --- a/unit_tests/server/httpserver/test_pd_master_cached_tokens.py +++ b/unit_tests/server/httpserver/test_pd_master_cached_tokens.py @@ -19,6 +19,7 @@ def _make_manager(monkeypatch): mgr.args = SimpleNamespace(disable_pd_master_decode_capacity_limit=True) mgr.pd_high_priority_request_time_out_seconds = 60 mgr.pd_cache_high_priority_max_age_seconds = 60 + mgr.pd_cache_high_priority_min_prompt_tokens = 8192 mgr.disable_pd_cache_high_priority = False mgr.running_request_count = 0 counter = [0] diff --git a/unit_tests/utils/test_envs_utils.py b/unit_tests/utils/test_envs_utils.py index f3c7d96dd..da853acda 100644 --- a/unit_tests/utils/test_envs_utils.py +++ b/unit_tests/utils/test_envs_utils.py @@ -1,5 +1,6 @@ from lightllm.utils.envs_utils import ( get_pd_cache_high_priority_max_age_seconds, + get_pd_cache_high_priority_min_prompt_tokens, get_pd_node_router_wait_timeout_seconds, get_pd_node_shm_req_alloc_timeout_seconds, ) @@ -23,6 +24,24 @@ def test_pd_cache_high_priority_max_age_reads_environment_variable(monkeypatch): get_pd_cache_high_priority_max_age_seconds.cache_clear() +def test_pd_cache_high_priority_min_prompt_tokens_defaults_to_4096(monkeypatch): + monkeypatch.delenv("LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MIN_PROMPT_TOKENS", raising=False) + get_pd_cache_high_priority_min_prompt_tokens.cache_clear() + + assert get_pd_cache_high_priority_min_prompt_tokens() == 4096 + + get_pd_cache_high_priority_min_prompt_tokens.cache_clear() + + +def test_pd_cache_high_priority_min_prompt_tokens_reads_environment_variable(monkeypatch): + monkeypatch.setenv("LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MIN_PROMPT_TOKENS", "2048") + get_pd_cache_high_priority_min_prompt_tokens.cache_clear() + + assert get_pd_cache_high_priority_min_prompt_tokens() == 2048 + + get_pd_cache_high_priority_min_prompt_tokens.cache_clear() + + def test_pd_node_shm_req_alloc_timeout_defaults_to_20_seconds(monkeypatch): monkeypatch.delenv("LIGHTLLM_PD_NODE_SHM_REQ_ALLOC_TIMEOUT_SECONDS", raising=False) get_pd_node_shm_req_alloc_timeout_seconds.cache_clear() From 988f0a659cf18dec36e91142334aefedc18ff1ac Mon Sep 17 00:00:00 2001 From: wangzaijun Date: Thu, 3 Sep 2026 10:31:46 +0000 Subject: [PATCH 02/12] fix --- lightllm/server/httpserver_for_pd_master/manager.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/lightllm/server/httpserver_for_pd_master/manager.py b/lightllm/server/httpserver_for_pd_master/manager.py index 8e17237ac..f4d631576 100644 --- a/lightllm/server/httpserver_for_pd_master/manager.py +++ b/lightllm/server/httpserver_for_pd_master/manager.py @@ -247,6 +247,12 @@ async def _generate_one( and selection_extra_info.estimated_cache_hit_rate > 0.8 and cache_age_seconds is not None and cache_age_seconds <= self.pd_cache_high_priority_max_age_seconds + # TODO: 更细粒度的策略可由每个 P 节点维护待调度及尚未完成请求的 token_len 队列， + # 并向 PD Master 周期上报排队 token 总量、最长请求长度和最老请求等待时间等摘要。 + # PD Master 可用 input_token_num * (1 - estimated_cache_hit_rate) 估算新请求剩余的 + # Prefill 工作量；当目标 P 节点存在长请求时，允许剩余工作量很小的高 cache 命中短请求 + # 提升优先级，以较低额外成本改善其 TTFT。该策略只重排尚未执行的请求，不尝试抢占 + # 已在 GPU 上运行的请求，并应通过最大连续插队次数或最长等待时间防止长请求饥饿。 and input_token_num >= self.pd_cache_high_priority_min_prompt_tokens ) From 13a964aa805c4b030593cf7b71c26320f8da0e70 Mon Sep 17 00:00:00 2001 From: wangzaijun Date: Fri, 4 Sep 2026 03:08:59 +0000 Subject: [PATCH 03/12] fix(pd): centralize node resource wait timeout --- docs/CN/source/tutorial/api_server_args.rst | 32 +++++--- docs/EN/source/tutorial/api_server_args.rst | 34 +++++--- lightllm/server/api_cli.py | 5 +- lightllm/server/core/objs/sampling_params.py | 10 +-- lightllm/server/httpserver/manager.py | 62 +++++--------- .../httpserver_for_pd_master/manager.py | 19 +++-- lightllm/utils/envs_utils.py | 18 +---- .../test_pd_master_multi_choice.py | 22 ++--- .../test_pd_node_request_limit.py | 80 ++++++++++--------- .../test_pd_master_cached_tokens.py | 2 +- .../test_running_request_lifecycle.py | 62 ++++++++++---- unit_tests/server/test_pd_master_mode.py | 2 +- unit_tests/utils/test_envs_utils.py | 41 +++------- 13 files changed, 198 insertions(+), 191 deletions(-) diff --git a/docs/CN/source/tutorial/api_server_args.rst b/docs/CN/source/tutorial/api_server_args.rst index 536901764..fa78ab611 100644 --- a/docs/CN/source/tutorial/api_server_args.rst +++ b/docs/CN/source/tutorial/api_server_args.rst @@ -94,19 +94,27 @@ PD 分离模式参数 .. option:: --enable_pd_node_self_request_limit - 在 Prefill/Decode 节点上启用本地请求限流。PD Master 当前不执行请求准入限流。 - HTTP server 申请本地 ``shm_req`` 对象的超时时间由 - ``LIGHTLLM_PD_NODE_SHM_REQ_ALLOC_TIMEOUT_SECONDS`` 控制（默认 20 秒）；请求进入 Router 后等待 - 进入推理系统的超时时间由 ``LIGHTLLM_PD_NODE_ROUTER_WAIT_TIMEOUT_SECONDS`` 控制（默认 20 秒）。 - 超时会导致 ``Server is busy``；其中已进入 Router 但仍未进入推理系统的请求会主动标记为 aborted， - 由 PD Master 转换为 HTTP 429。未开启限流时请求会持续等待资源；PD 高优先级请求 - （分段续跑请求，或预计输入 cache 命中率高于 0.8 且命中记录仍然新鲜的请求）由 PD Master 通过 - ``pd_high_priority_request_time_out_seconds`` 下发一个统一的超时时间下限。P/D 节点分别取 - 该值与本地 ``shm_req``、Router 超时的较大值；该字段为 0 时不延长本地超时。PD Master 下发值由 - ``LIGHTLLM_PD_HIGH_PRIORITY_REQUEST_TIMEOUT_SECONDS`` 控制，默认 60 秒。cache 命中记录允许提升优先级的 - 最大年龄由 ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MAX_AGE_SECONDS`` 控制，默认 16 秒。cache 命中提权还要求输入 + 启用由 PD Master 统一管理的 P/D 节点资源等待限流。该参数只需要在 PD Master 启动时设置， + 不需要在 Prefill/Decode 节点上设置。开启后，PD Master 通过 + ``pd_node_resource_wait_timeout_seconds`` 为所有请求下发统一的资源等待上限；P/D 节点只负责按下发值 + 控制本地 ``shm_req`` 申请和 Router 等待进入推理系统，不读取本地限流开关或超时配置，也不根据请求 + 是否为高优先级改变超时。该值由 PD Master 上的 + ``LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS`` 控制，默认 10 秒；设置为 -1 表示永久等待。 + 设置为非负数时，超时会导致 ``Server is busy``； + 其中已进入 Router 但仍未进入推理系统的请求会主动标记为 aborted，由 PD Master 转换为 HTTP 429。 + 未设置该启动参数时，PD Master 统一下发 -1，即所有 P/D 节点永久等待。 + 多机 TP 场景仅由 master 节点执行超时判断，slave 节点永久等待。cache 命中记录允许提升优先级的最大年龄由 + ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MAX_AGE_SECONDS`` 控制，默认 16 秒。cache 命中提权还要求输入 token 数达到 ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MIN_PROMPT_TOKENS`` 配置的门槛（默认 4096），避免短请求仅因 - cache 命中率高而提升优先级。本地请求限流默认关闭。 + cache 命中率高而提升优先级。 + + 启动示例： + + .. code-block:: bash + + LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS=10 \ + python -m lightllm.server.api_server --run_mode pd_master \ + --enable_pd_node_self_request_limit ... .. option:: --disable_pd_cache_high_priority diff --git a/docs/EN/source/tutorial/api_server_args.rst b/docs/EN/source/tutorial/api_server_args.rst index d3db0970e..9bafcae7b 100644 --- a/docs/EN/source/tutorial/api_server_args.rst +++ b/docs/EN/source/tutorial/api_server_args.rst @@ -97,21 +97,29 @@ PD disaggregation Mode Parameters .. option:: --enable_pd_node_self_request_limit - Enable local request limiting on Prefill/Decode nodes. PD Master does not currently perform request admission - limiting. The local ``shm_req`` allocation timeout is controlled by - ``LIGHTLLM_PD_NODE_SHM_REQ_ALLOC_TIMEOUT_SECONDS`` (20 seconds by default), while the timeout from Router entry - to inference entry is controlled by ``LIGHTLLM_PD_NODE_ROUTER_WAIT_TIMEOUT_SECONDS`` (20 seconds by default). - A timeout reports ``Server is busy``; a request that has entered the Router but not inference is proactively - marked aborted, and PD Master converts this to HTTP 429. Requests continue waiting when local admission is - disabled. For PD high-priority requests (segmented continuation requests, or requests whose estimated input - cache hit rate is above 0.8 and whose cache record is still fresh), PD Master supplies a shared timeout floor through - ``pd_high_priority_request_time_out_seconds``. Each P/D node uses the greater of this value and its local - ``shm_req`` or Router timeout; zero does not extend the local timeout. The value supplied by PD Master is controlled by - ``LIGHTLLM_PD_HIGH_PRIORITY_REQUEST_TIMEOUT_SECONDS`` and defaults to 60 seconds. The maximum cache-record age - eligible for promotion is controlled by ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MAX_AGE_SECONDS`` and defaults to + Enable PD Master-managed resource wait limiting for P/D nodes. Set this option only when starting PD Master; + it is not needed on Prefill or Decode nodes. Once enabled, PD Master supplies + ``pd_node_resource_wait_timeout_seconds`` for every request. P/D nodes only enforce the received value for local + ``shm_req`` allocation and the wait from Router entry to inference entry; they do not read local limiting switches + or timeout settings, and request priority does not alter the timeout. The value is + controlled on PD Master by ``LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS`` and defaults to 10 seconds; set it + to -1 to wait indefinitely. When set to a non-negative value, a timeout reports ``Server is busy``; a request that has + entered the Router but not inference is proactively marked aborted, and PD Master converts this to HTTP 429. + Without this startup option, PD Master sends -1 and all P/D nodes wait indefinitely. + In multi-node TP deployments, only the master node evaluates the timeout; slave nodes wait indefinitely. + The maximum cache-record age eligible for promotion is controlled by + ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MAX_AGE_SECONDS`` and defaults to 16 seconds. Cache-hit promotion also requires at least the number of input tokens configured by ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MIN_PROMPT_TOKENS`` (4096 by default), so short requests do not gain priority - solely from a high cache-hit rate. Local request limiting is disabled by default. + solely from a high cache-hit rate. + + Startup example: + + .. code-block:: bash + + LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS=10 \ + python -m lightllm.server.api_server --run_mode pd_master \ + --enable_pd_node_self_request_limit ... .. option:: --disable_pd_cache_high_priority diff --git a/lightllm/server/api_cli.py b/lightllm/server/api_cli.py index bfb54b4ba..fa19be299 100644 --- a/lightllm/server/api_cli.py +++ b/lightllm/server/api_cli.py @@ -80,8 +80,9 @@ def add_cli_args(parser: argparse.ArgumentParser) -> argparse.ArgumentParser: "--enable_pd_node_self_request_limit", action="store_true", help=( - "Enable local request limiting on Prefill/Decode nodes by enforcing shm_req allocation and Router " - "scheduling wait timeouts. PD Master admission limiting is not currently enabled. Default: disabled." + "Enable PD Master-managed resource wait limiting for Prefill/Decode nodes. Configure this option only " + "on PD Master; it sends all timeout details to P/D nodes, which only enforce the received values. " + "Default: disabled." ), ) parser.add_argument( diff --git a/lightllm/server/core/objs/sampling_params.py b/lightllm/server/core/objs/sampling_params.py index a7b4547d8..6f96a1ed4 100644 --- a/lightllm/server/core/objs/sampling_params.py +++ b/lightllm/server/core/objs/sampling_params.py @@ -297,9 +297,9 @@ class SamplingParams(ctypes.Structure): # 由 PD Master 为分段续跑或预计 cache 命中率较高的请求设置，表示请求需 # 以高优先级插入 Router 调度队列。 ("pd_high_priority_request", ctypes.c_bool), - # PD 高优先级请求在开启本地限流的 P/D 节点上的等待时间下限。节点分别取 - # 该值与本地超时的较大值，用于 shm_req 申请和 Router 等待进入推理系统。 - ("pd_high_priority_request_time_out_seconds", ctypes.c_int), + # P/D 节点的资源等待超时，由 PD Master 下发。非负值用于控制 shm_req 申请和 + # Router 等待进入推理系统的时限；负数表示永久等待。 + ("pd_node_resource_wait_timeout_seconds", ctypes.c_int), ("suggested_dp_index", ctypes.c_int), # suggest dp index, deepseekv2 dp mode, use to suggest used dp_index # in pd split mode, use to keep the id of pd master ("pd_master_node_id", NodeUUId), @@ -345,7 +345,7 @@ def init(self, tokenizer, **kwargs): self.group_request_id = kwargs.get("group_request_id", -1) # 这两个字段是 PD Master 的内部调度信息，不能由外部请求参数开启或修改。 self.pd_high_priority_request = False - self.pd_high_priority_request_time_out_seconds = 0 + self.pd_node_resource_wait_timeout_seconds = -1 self.suggested_dp_index = kwargs.get("suggested_dp_index", -1) self.skip_special_tokens = kwargs.get("skip_special_tokens", SKIP_SPECIAL_TOKENS) @@ -513,7 +513,7 @@ def to_dict(self): "invalid_token_ids": self.invalid_token_ids.to_list(), "group_request_id": self.group_request_id, "pd_high_priority_request": self.pd_high_priority_request, - "pd_high_priority_request_time_out_seconds": self.pd_high_priority_request_time_out_seconds, + "pd_node_resource_wait_timeout_seconds": self.pd_node_resource_wait_timeout_seconds, "skip_special_tokens": self.skip_special_tokens, "add_special_tokens": self.add_special_tokens, "add_spaces_between_special_tokens": self.add_spaces_between_special_tokens, diff --git a/lightllm/server/httpserver/manager.py b/lightllm/server/httpserver/manager.py index 9e7532df4..237c050e9 100644 --- a/lightllm/server/httpserver/manager.py +++ b/lightllm/server/httpserver/manager.py @@ -36,11 +36,7 @@ from .manager_ext import HttpRlManagerHelper from lightllm.utils.statics_utils import MovingAverage from lightllm.utils.config_utils import get_vocab_size -from lightllm.utils.envs_utils import ( - get_pd_node_router_wait_timeout_seconds, - get_pd_node_shm_req_alloc_timeout_seconds, - get_unique_server_name, -) +from lightllm.utils.envs_utils import get_env_start_args, get_unique_server_name from lightllm.utils.shm_port_args import get_shm_port_args from lightllm.utils.error_utils import ( ClientDisconnected, @@ -126,13 +122,6 @@ def __init__( self.pd_mode: NodeRole = NodeRole(self.args.run_mode) assert self.pd_mode in [NodeRole.NORMAL, NodeRole.P, NodeRole.D] - # HTTP server 只负责在本地 shm_req 或 Router 等待过久时快速返回繁忙，PD Master 负责 QPS 准入限流。 - # 该开关控制 P/D 节点是否启用这两类本地等待超时；多机 TP 从节点不独立拒绝请求。 - self.pd_node_request_limit_enabled: bool = ( - self.args.enable_pd_node_self_request_limit and self.pd_mode.is_P_or_D() and not self.is_multinode_tp_slave - ) - self.pd_node_shm_req_alloc_timeout_seconds = get_pd_node_shm_req_alloc_timeout_seconds() - self.pd_node_router_wait_timeout_seconds = get_pd_node_router_wait_timeout_seconds() self.id_gen = ReqIDGenerator() self.first_time_costs = MovingAverage() self.per_token_costs = MovingAverage() @@ -442,12 +431,12 @@ async def generate( await self._register_running_request() running_request_registered = True - # 申请资源并存储。PD 高优先级请求仍以更短的间隔抢占资源；开启本地限流时， - # 使用 PD Master 下发的较长超时时间，避免资源异常时一直等待。 + # 申请资源并存储。PD 高优先级请求仍以更短的间隔重试；资源等待上限 + # 完全由 PD Master 下发，与请求优先级无关。 alloced_req_indexes = await self._alloc_shm_req_indexes( sampling_params.n, pd_high_priority_request=sampling_params.pd_high_priority_request, - pd_high_priority_request_time_out_seconds=sampling_params.pd_high_priority_request_time_out_seconds, + pd_node_resource_wait_timeout_seconds=sampling_params.pd_node_resource_wait_timeout_seconds, ) req_objs: List[Req] = [] for i, req_index in enumerate(alloced_req_indexes): @@ -558,22 +547,17 @@ async def _alloc_shm_req_indexes( self, req_num: int, pd_high_priority_request: bool = False, - pd_high_priority_request_time_out_seconds: int = 0, + pd_node_resource_wait_timeout_seconds: int = -1, ) -> List[int]: """为一个请求申请全部 shm_req 索引，申请失败时回滚已分配的索引。 - 未开启本地限流时无限等待。开启限流后，普通请求使用节点的 shm_req 申请 - 超时时间；高优先级请求取本地超时与 PD Master 下发值中的较大值。 + PD Master 下发非负值时启用资源等待超时，负数表示无限等待。多机 TP slave + 不独立限流，由 master 节点统一判断。请求优先级只影响重试间隔，不影响超时值。 """ alloced_req_indexes = [] alloc_timeout_seconds = None - if self.pd_node_request_limit_enabled: - alloc_timeout_seconds = self.pd_node_shm_req_alloc_timeout_seconds - if pd_high_priority_request: - alloc_timeout_seconds = max( - alloc_timeout_seconds, - pd_high_priority_request_time_out_seconds, - ) + if not self.is_multinode_tp_slave and pd_node_resource_wait_timeout_seconds >= 0: + alloc_timeout_seconds = pd_node_resource_wait_timeout_seconds alloc_deadline = time.monotonic() + alloc_timeout_seconds if alloc_timeout_seconds is not None else None try: @@ -803,14 +787,11 @@ async def _wait_to_token_package( except asyncio.TimeoutError: pass - if ( - self.pd_node_request_limit_enabled - and is_first_token - and req_status.has_timed_out_waiting_for_inference(self.pd_node_router_wait_timeout_seconds) - ): + if is_first_token and req_status.has_timed_out_waiting_for_inference(): + resource_wait_timeout_seconds = sampling_params.pd_node_resource_wait_timeout_seconds raise ServerBusyError( f"PD {self.args.run_mode} node is busy: request did not enter inference " - f"within {self.pd_node_router_wait_timeout_seconds} seconds" + f"within {resource_wait_timeout_seconds} seconds" ) if request is not None and await request.is_disconnected(): @@ -1112,6 +1093,8 @@ class ReqStatus: def __init__(self, group_request_id, multimodal_params, req_objs: List[Req], start_time) -> None: self.lock = asyncio.Lock() self.event = asyncio.Event() + args = get_env_start_args() + self.is_multinode_tp_slave = args.dp == 1 and args.nnodes > 1 and args.node_rank > 0 self.group_req_objs = GroupReqObjs( group_req_id=group_request_id, multimodal_params=multimodal_params, @@ -1120,20 +1103,19 @@ def __init__(self, group_request_id, multimodal_params, req_objs: List[Req], sta ) self.out_token_info_list = [] - def has_timed_out_waiting_for_inference(self, timeout_seconds: float) -> bool: - """判断请求组是否已在 Router 中等待进入推理系统超时。""" + def has_timed_out_waiting_for_inference(self) -> bool: + """按 PD Master 下发的资源等待上限判断请求是否在 Router 中超时。""" + # 多机 TP slave 只跟随 master 执行，不能独立判定超时并中止请求。 + if self.is_multinode_tp_slave: + return False current_time = time.monotonic() reqs = self.group_req_objs.shm_req_objs # 组内任一请求已经进入新 batch，说明整个请求组已经开始执行，不能再按 Router 等待超时清理。 if any(req.infer_start_time > 0 for req in reqs): return False - # 高优先级请求取本地 Router 超时与 PD Master 下发值中的较大值，既保证 - # 它比普通请求拥有更充足的等待机会，也避免资源异常时永久滞留。 - if any(req.sample_params.pd_high_priority_request for req in reqs): - timeout_seconds = max( - timeout_seconds, - reqs[0].sample_params.pd_high_priority_request_time_out_seconds, - ) + timeout_seconds = reqs[0].sample_params.pd_node_resource_wait_timeout_seconds + if timeout_seconds < 0: + return False for req in reqs: if req.router_arrival_time > 0 and current_time - req.router_arrival_time >= timeout_seconds: diff --git a/lightllm/server/httpserver_for_pd_master/manager.py b/lightllm/server/httpserver_for_pd_master/manager.py index f4d631576..189b15549 100644 --- a/lightllm/server/httpserver_for_pd_master/manager.py +++ b/lightllm/server/httpserver_for_pd_master/manager.py @@ -26,7 +26,7 @@ from lightllm.utils.envs_utils import ( get_pd_cache_high_priority_max_age_seconds, get_pd_cache_high_priority_min_prompt_tokens, - get_pd_high_priority_request_timeout_seconds, + get_pd_node_resource_wait_timeout_seconds, ) from lightllm.utils.shm_port_args import get_shm_port_args from .pd_selector import PDSelectionExtraInfo, create_selector @@ -52,9 +52,11 @@ def __init__( self.health_timeout = int(os.getenv("HEALTH_TIMEOUT", "200")) self.latest_success_infer_time = time.time() self.running_request_count = 0 - # 高优先级请求仍可比普通请求等待更久，但通过请求参数向开启本地限流的 - # P/D 节点传递有限的等待时间，避免资源异常时永久占用请求链路。 - self.pd_high_priority_request_time_out_seconds = get_pd_high_priority_request_timeout_seconds() + # 限流开关只在 PD Master 生效。开启后由 Master 统一下发资源等待上限； + # 未开启时下发 -1，P/D 节点不读取本地开关或超时配置，只执行收到的值。 + self.pd_node_resource_wait_timeout_seconds = ( + get_pd_node_resource_wait_timeout_seconds() if args.enable_pd_node_self_request_limit else -1 + ) self.pd_cache_high_priority_max_age_seconds = get_pd_cache_high_priority_max_age_seconds() self.pd_cache_high_priority_min_prompt_tokens = get_pd_cache_high_priority_min_prompt_tokens() self.disable_pd_cache_high_priority = args.disable_pd_cache_high_priority @@ -277,12 +279,9 @@ async def _generate_one( # KV cache 插队。第二段及后续分段仍统一使用高优先级，避免因临时资源 # 紧张导致分段续跑失败。 sampling_params.pd_high_priority_request = segment_index > 0 or has_fresh_high_cache_hit - # 为高优先级请求下发较长的有限等待时间；P/D 节点仅在自身开启 - # 本地限流时使用该值，未开启限流时仍保持无限等待。 - if sampling_params.pd_high_priority_request: - sampling_params.pd_high_priority_request_time_out_seconds = ( - self.pd_high_priority_request_time_out_seconds - ) + # 资源等待超时与请求优先级相互独立；P/D 节点直接使用 PD Master + # 下发值，负数表示持续等待资源。 + sampling_params.pd_node_resource_wait_timeout_seconds = self.pd_node_resource_wait_timeout_seconds # 分段请求始终复用循环外选定的 P 节点；这里只按每段实际发送的 # prompt 更新该节点的在途 prefill 负载，不会重新选点。 diff --git a/lightllm/utils/envs_utils.py b/lightllm/utils/envs_utils.py index d7713aae8..0f7120370 100644 --- a/lightllm/utils/envs_utils.py +++ b/lightllm/utils/envs_utils.py @@ -307,21 +307,9 @@ def _get_mtp_draft_backbone_layer_num(draft_model_dir: str) -> int: @lru_cache(maxsize=None) -def get_pd_node_shm_req_alloc_timeout_seconds() -> int: - """PD 节点申请 ``shm_req`` 对象的最长等待时间，单位为秒。""" - return int(os.getenv("LIGHTLLM_PD_NODE_SHM_REQ_ALLOC_TIMEOUT_SECONDS", 20)) - - -@lru_cache(maxsize=None) -def get_pd_node_router_wait_timeout_seconds() -> int: - """请求进入 Router 后等待进入推理系统的最长时间，单位为秒。""" - return int(os.getenv("LIGHTLLM_PD_NODE_ROUTER_WAIT_TIMEOUT_SECONDS", 20)) - - -@lru_cache(maxsize=None) -def get_pd_high_priority_request_timeout_seconds() -> int: - """PD Master 为高优先级请求设置的等待时间下限，单位为秒。""" - return int(os.getenv("LIGHTLLM_PD_HIGH_PRIORITY_REQUEST_TIMEOUT_SECONDS", 60)) +def get_pd_node_resource_wait_timeout_seconds() -> int: + """P/D 节点的资源等待超时，单位为秒；负数表示永久等待。""" + return int(os.getenv("LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS", 10)) @lru_cache(maxsize=None) diff --git a/test/test_pd_selector/test_pd_master_multi_choice.py b/test/test_pd_selector/test_pd_master_multi_choice.py index 53360fb34..d3b62704c 100644 --- a/test/test_pd_selector/test_pd_master_multi_choice.py +++ b/test/test_pd_selector/test_pd_master_multi_choice.py @@ -20,7 +20,7 @@ def _manager() -> HttpServerManagerForPDMaster: manager.metric_client = MagicMock() manager._log_req_header = AsyncMock() manager.tokens = MagicMock(return_value=2) - manager.pd_high_priority_request_time_out_seconds = 60 + manager.pd_node_resource_wait_timeout_seconds = -1 manager.pd_cache_high_priority_max_age_seconds = 60 manager.pd_cache_high_priority_min_prompt_tokens = 4096 manager.disable_pd_cache_high_priority = False @@ -574,10 +574,10 @@ async def wait_to_token_package(_p_node, _d_node, _start_time, _prompt, sampling asyncio.run(asyncio.wait_for(run(), timeout=2)) -def test_pd_master_sets_high_priority_timeout(): +def test_pd_master_sets_resource_wait_timeout_independently_of_priority(): async def run(): manager = _manager() - manager.pd_high_priority_request_time_out_seconds = 90 + manager.pd_node_resource_wait_timeout_seconds = 90 manager.id_gen.generate_id.return_value = 808 manager.remove_req = AsyncMock() manager.abort = AsyncMock() @@ -590,16 +590,18 @@ async def run(): return_value=( p_node, d_node, - PDSelectionExtraInfo( - estimated_cache_hit_rate=0.81, - cache_last_insert_time=time.monotonic(), - ), + PDSelectionExtraInfo(), ) ) - captured_timeout_seconds = [] + captured_request_settings = [] async def wait_to_token_package(_p_node, _d_node, _start_time, _prompt, sampling_params, *_args): - captured_timeout_seconds.append(sampling_params.pd_high_priority_request_time_out_seconds) + captured_request_settings.append( + ( + sampling_params.pd_high_priority_request, + sampling_params.pd_node_resource_wait_timeout_seconds, + ) + ) yield ( sampling_params.group_request_id, "x", @@ -622,7 +624,7 @@ async def wait_to_token_package(_p_node, _d_node, _start_time, _prompt, sampling ): pass - assert captured_timeout_seconds == [90] + assert captured_request_settings == [(False, 90)] asyncio.run(asyncio.wait_for(run(), timeout=2)) diff --git a/test/test_pd_selector/test_pd_node_request_limit.py b/test/test_pd_selector/test_pd_node_request_limit.py index 9e071cfa9..a591bdc13 100644 --- a/test/test_pd_selector/test_pd_node_request_limit.py +++ b/test/test_pd_selector/test_pd_node_request_limit.py @@ -25,26 +25,25 @@ def set_value(self, value: int): def _manager() -> HttpServerManager: manager = HttpServerManager.__new__(HttpServerManager) manager.args = SimpleNamespace(run_mode="decode", running_max_req_size=64) - manager.pd_node_request_limit_enabled = False + manager.is_multinode_tp_slave = False manager._run_reqs_count_lock = asyncio.Lock() manager.run_reqs_count_mark = FakeSharedInt() manager.latest_success_infer_time_mark = FakeSharedInt() - manager.pd_node_shm_req_alloc_timeout_seconds = 20 manager.shm_req_manager = MagicMock() manager.shm_req_manager.async_release_req_index = AsyncMock() return manager -def test_pd_high_priority_timeout_is_internal_and_defaults_to_zero(): +def test_pd_node_resource_wait_timeout_is_internal_and_defaults_to_waiting_forever(): sampling_params = SamplingParams() sampling_params.init( None, pd_high_priority_request=True, - pd_high_priority_request_time_out_seconds=99, + pd_node_resource_wait_timeout_seconds=99, ) assert sampling_params.pd_high_priority_request is False - assert sampling_params.pd_high_priority_request_time_out_seconds == 0 + assert sampling_params.pd_node_resource_wait_timeout_seconds == -1 def test_shm_req_partial_allocations_are_released_on_failure(): @@ -60,7 +59,7 @@ async def run(): asyncio.run(run()) -def test_shm_req_allocation_waits_forever_when_local_limit_is_disabled(): +def test_shm_req_allocation_waits_forever_when_timeout_is_negative(): async def run(): manager = _manager() manager.shm_req_manager.async_alloc_req_index = AsyncMock(side_effect=[None, None, 3]) @@ -73,85 +72,92 @@ async def run(): asyncio.run(run()) -def test_high_priority_shm_req_allocation_uses_shorter_backoff_even_with_local_limit(): +def test_multinode_tp_slave_does_not_apply_shm_req_allocation_timeout(): async def run(): manager = _manager() - manager.pd_node_request_limit_enabled = True - manager.shm_req_manager.async_alloc_req_index = AsyncMock(side_effect=[None, 3]) + manager.is_multinode_tp_slave = True + manager.shm_req_manager.async_alloc_req_index = AsyncMock(side_effect=[None, None, 3]) with patch("lightllm.server.httpserver.manager.asyncio.sleep", new=AsyncMock()) as sleep: - assert await manager._alloc_shm_req_indexes(1, pd_high_priority_request=True) == [3] + assert await manager._alloc_shm_req_indexes(1, pd_node_resource_wait_timeout_seconds=0) == [3] - assert sleep.await_args_list[0].args[0] == pytest.approx(0.1 * 0.2) + assert sleep.await_count == 2 asyncio.run(run()) -def test_high_priority_shm_req_allocation_uses_master_timeout_with_local_limit(): +def test_high_priority_shm_req_allocation_uses_shorter_backoff_even_with_local_limit(): async def run(): manager = _manager() - manager.pd_node_request_limit_enabled = True - manager.shm_req_manager.async_alloc_req_index = AsyncMock(return_value=None) + manager.shm_req_manager.async_alloc_req_index = AsyncMock(side_effect=[None, 3]) - with ( - patch("lightllm.server.httpserver.manager.time.monotonic", side_effect=[100, 161]), - pytest.raises(ServerBusyError, match="within 60 seconds"), - ): - await manager._alloc_shm_req_indexes( - 1, - pd_high_priority_request=True, - pd_high_priority_request_time_out_seconds=60, + with patch("lightllm.server.httpserver.manager.asyncio.sleep", new=AsyncMock()) as sleep: + assert ( + await manager._alloc_shm_req_indexes( + 1, + pd_high_priority_request=True, + pd_node_resource_wait_timeout_seconds=60, + ) + == [3] ) + assert sleep.await_args_list[0].args[0] == pytest.approx(0.1 * 0.2) + asyncio.run(run()) -def test_high_priority_shm_req_allocation_does_not_shorten_local_timeout(): +@pytest.mark.parametrize("pd_high_priority_request", [False, True]) +def test_shm_req_allocation_uses_master_timeout_independently_of_priority(pd_high_priority_request): async def run(): manager = _manager() - manager.pd_node_request_limit_enabled = True - manager.pd_node_shm_req_alloc_timeout_seconds = 80 manager.shm_req_manager.async_alloc_req_index = AsyncMock(return_value=None) with ( - patch("lightllm.server.httpserver.manager.time.monotonic", side_effect=[100, 181]), - pytest.raises(ServerBusyError, match="within 80 seconds"), + patch("lightllm.server.httpserver.manager.time.monotonic", side_effect=[100, 161]), + pytest.raises(ServerBusyError, match="within 60 seconds"), ): await manager._alloc_shm_req_indexes( 1, - pd_high_priority_request=True, - pd_high_priority_request_time_out_seconds=60, + pd_high_priority_request=pd_high_priority_request, + pd_node_resource_wait_timeout_seconds=60, ) asyncio.run(run()) @pytest.mark.parametrize( - ("infer_start_time", "local_timeout_seconds", "high_priority_timeout_seconds", "expected"), - [(0, 20, 0, True), (0, 20, 60, True), (0, 80, 60, False), (1, 20, 60, False)], + ("infer_start_time", "pd_high_priority_request", "resource_wait_timeout_seconds", "expected"), + [ + (0, False, -1, False), + (0, False, 60, True), + (0, True, 60, True), + (0, True, 80, False), + (1, False, 60, False), + ], ) -def test_high_priority_router_wait_uses_master_timeout( +def test_router_wait_uses_master_timeout_independently_of_priority( infer_start_time, - local_timeout_seconds, - high_priority_timeout_seconds, + pd_high_priority_request, + resource_wait_timeout_seconds, expected, ): req_status = ReqStatus.__new__(ReqStatus) + req_status.is_multinode_tp_slave = False req_status.group_req_objs = SimpleNamespace( shm_req_objs=[ SimpleNamespace( infer_start_time=infer_start_time, router_arrival_time=100, sample_params=SimpleNamespace( - pd_high_priority_request=True, - pd_high_priority_request_time_out_seconds=high_priority_timeout_seconds, + pd_high_priority_request=pd_high_priority_request, + pd_node_resource_wait_timeout_seconds=resource_wait_timeout_seconds, ), ) ] ) with patch("lightllm.server.httpserver.manager.time.monotonic", return_value=161): - assert req_status.has_timed_out_waiting_for_inference(local_timeout_seconds) is expected + assert req_status.has_timed_out_waiting_for_inference() is expected def test_pd_high_priority_request_is_inserted_before_first_normal_request(): diff --git a/unit_tests/server/httpserver/test_pd_master_cached_tokens.py b/unit_tests/server/httpserver/test_pd_master_cached_tokens.py index ebecbd47e..b514a8429 100644 --- a/unit_tests/server/httpserver/test_pd_master_cached_tokens.py +++ b/unit_tests/server/httpserver/test_pd_master_cached_tokens.py @@ -17,7 +17,7 @@ def _make_manager(monkeypatch): monkeypatch.setattr(SamplingParams, "from_buffer_copy", classmethod(lambda cls, other: copy.copy(other))) mgr = object.__new__(HttpServerManagerForPDMaster) mgr.args = SimpleNamespace(disable_pd_master_decode_capacity_limit=True) - mgr.pd_high_priority_request_time_out_seconds = 60 + mgr.pd_node_resource_wait_timeout_seconds = -1 mgr.pd_cache_high_priority_max_age_seconds = 60 mgr.pd_cache_high_priority_min_prompt_tokens = 8192 mgr.disable_pd_cache_high_priority = False diff --git a/unit_tests/server/httpserver/test_running_request_lifecycle.py b/unit_tests/server/httpserver/test_running_request_lifecycle.py index bf154826e..6f4ae60e1 100644 --- a/unit_tests/server/httpserver/test_running_request_lifecycle.py +++ b/unit_tests/server/httpserver/test_running_request_lifecycle.py @@ -8,7 +8,7 @@ from lightllm.server.core.objs import SamplingParams from lightllm.server.httpserver.manager import HttpServerManager, ReqStatus from lightllm.server.pd_io_struct import NodeRole, ObjType -from lightllm.utils.error_utils import PDPrefillNodeStopGenToken, ServerBusyError +from lightllm.utils.error_utils import ClientDisconnected, PDPrefillNodeStopGenToken, ServerBusyError class _ValueMark: @@ -25,15 +25,11 @@ def set_value(self, value): def _make_manager(mode: NodeRole): manager = HttpServerManager.__new__(HttpServerManager) manager.args = SimpleNamespace( - enable_pd_node_self_request_limit=False, run_mode=mode.value, running_max_req_size=2, ) manager.pd_mode = mode manager.is_multinode_tp_slave = False - manager.pd_node_request_limit_enabled = False - manager.pd_node_shm_req_alloc_timeout_seconds = 20 - manager.pd_node_router_wait_timeout_seconds = 20 manager.alloc_req_id = MagicMock(return_value=123) manager.is_multinode_tp_master = False manager.rl_controller = None @@ -58,6 +54,7 @@ def _sampling_params(): sampling_params.group_request_id = 123 sampling_params.n = 1 sampling_params.max_new_tokens = 1 + sampling_params.pd_node_resource_wait_timeout_seconds = 20 return sampling_params @@ -65,6 +62,12 @@ def _multimodal_params(): return SimpleNamespace(audios=[], images=[], verify_and_preload=AsyncMock()) +def _req_status(reqs, *, dp=1, nnodes=1, node_rank=0): + start_args = SimpleNamespace(dp=dp, nnodes=nnodes, node_rank=node_rank) + with patch("lightllm.server.httpserver.manager.get_env_start_args", return_value=start_args): + return ReqStatus(123, None, reqs, 0) + + async def _drain_generate(manager, sampling_params, multimodal_params, websocket=None, pd_event=None): async for _ in manager.generate( prompt="prompt", @@ -202,7 +205,6 @@ async def run(): def test_pd_node_returns_busy_when_shm_req_allocation_times_out(mode): async def run(): manager = _make_manager(mode) - manager.pd_node_request_limit_enabled = True manager.shm_req_manager = SimpleNamespace( async_alloc_req_index=AsyncMock(return_value=None), async_release_req_index=AsyncMock(), @@ -237,15 +239,17 @@ async def run(): def test_pd_node_returns_busy_while_first_token_request_waits_in_router(mode): async def run(): manager = _make_manager(mode) - manager.pd_node_request_limit_enabled = True req = SimpleNamespace( request_id=123, is_aborted=False, router_arrival_time=0, infer_start_time=0, - sample_params=SimpleNamespace(pd_high_priority_request=False), + sample_params=SimpleNamespace( + pd_high_priority_request=False, + pd_node_resource_wait_timeout_seconds=20, + ), ) - req_status = ReqStatus(123, None, [req], 0) + req_status = _req_status([req]) req_status.event.set() sampling_params = _sampling_params() @@ -265,13 +269,42 @@ async def run(): asyncio.run(run()) -def test_httpserver_keeps_started_and_high_priority_request_groups(): +def test_multinode_tp_slave_does_not_apply_router_wait_timeout(): + async def run(): + manager = _make_manager(NodeRole.D) + manager.is_multinode_tp_slave = True + req = SimpleNamespace( + router_arrival_time=1.0, + infer_start_time=0.0, + sample_params=SimpleNamespace(pd_node_resource_wait_timeout_seconds=0), + ) + req_status = _req_status([req], nnodes=2, node_rank=1) + req_status.event.set() + request = SimpleNamespace(is_disconnected=AsyncMock(return_value=True)) + + with patch("lightllm.server.httpserver.manager.time.monotonic", return_value=2): + assert req_status.has_timed_out_waiting_for_inference() is False + output_generator = manager._wait_to_token_package( + start_time=0, + prompt_ids=[], + group_request_id=123, + sampling_params=_sampling_params(), + req_status=req_status, + request=request, + ) + with pytest.raises(ClientDisconnected): + await output_generator.__anext__() + + asyncio.run(run()) + + +def test_httpserver_keeps_started_requests_and_requests_with_remaining_master_timeout(): waiting_req = SimpleNamespace( router_arrival_time=1.0, infer_start_time=0.0, sample_params=SimpleNamespace( pd_high_priority_request=False, - pd_high_priority_request_time_out_seconds=60, + pd_node_resource_wait_timeout_seconds=60, ), ) started_req = SimpleNamespace( @@ -279,14 +312,13 @@ def test_httpserver_keeps_started_and_high_priority_request_groups(): infer_start_time=2.0, sample_params=SimpleNamespace(pd_high_priority_request=False), ) - req_status = ReqStatus(123, None, [waiting_req, started_req], 0) + req_status = _req_status([waiting_req, started_req]) with patch("lightllm.server.httpserver.manager.time.monotonic", return_value=30): - assert req_status.has_timed_out_waiting_for_inference(20) is False + assert req_status.has_timed_out_waiting_for_inference() is False started_req.infer_start_time = 0.0 - waiting_req.sample_params.pd_high_priority_request = True - assert req_status.has_timed_out_waiting_for_inference(20) is False + assert req_status.has_timed_out_waiting_for_inference() is False def test_pd_node_self_request_limit_releases_partially_allocated_shm_reqs(): diff --git a/unit_tests/server/test_pd_master_mode.py b/unit_tests/server/test_pd_master_mode.py index 1c704f6bc..86bc48855 100644 --- a/unit_tests/server/test_pd_master_mode.py +++ b/unit_tests/server/test_pd_master_mode.py @@ -10,7 +10,7 @@ from lightllm.server.httpserver_for_pd_master.manager import HttpServerManagerForPDMaster, PDManager -def test_pd_node_self_request_limit_cli_defaults_to_disabled_and_can_be_enabled(): +def test_pd_node_self_request_limit_cli_is_configured_on_pd_master(): parser = make_argument_parser() assert parser.parse_args([]).enable_pd_node_self_request_limit is False diff --git a/unit_tests/utils/test_envs_utils.py b/unit_tests/utils/test_envs_utils.py index da853acda..57baf0539 100644 --- a/unit_tests/utils/test_envs_utils.py +++ b/unit_tests/utils/test_envs_utils.py @@ -1,8 +1,7 @@ from lightllm.utils.envs_utils import ( get_pd_cache_high_priority_max_age_seconds, get_pd_cache_high_priority_min_prompt_tokens, - get_pd_node_router_wait_timeout_seconds, - get_pd_node_shm_req_alloc_timeout_seconds, + get_pd_node_resource_wait_timeout_seconds, ) @@ -42,37 +41,19 @@ def test_pd_cache_high_priority_min_prompt_tokens_reads_environment_variable(mon get_pd_cache_high_priority_min_prompt_tokens.cache_clear() -def test_pd_node_shm_req_alloc_timeout_defaults_to_20_seconds(monkeypatch): - monkeypatch.delenv("LIGHTLLM_PD_NODE_SHM_REQ_ALLOC_TIMEOUT_SECONDS", raising=False) - get_pd_node_shm_req_alloc_timeout_seconds.cache_clear() +def test_pd_node_resource_wait_timeout_defaults_to_10_seconds(monkeypatch): + monkeypatch.delenv("LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS", raising=False) + get_pd_node_resource_wait_timeout_seconds.cache_clear() - assert get_pd_node_shm_req_alloc_timeout_seconds() == 20 + assert get_pd_node_resource_wait_timeout_seconds() == 10 - get_pd_node_shm_req_alloc_timeout_seconds.cache_clear() + get_pd_node_resource_wait_timeout_seconds.cache_clear() -def test_pd_node_shm_req_alloc_timeout_reads_environment_variable(monkeypatch): - monkeypatch.setenv("LIGHTLLM_PD_NODE_SHM_REQ_ALLOC_TIMEOUT_SECONDS", "30") - get_pd_node_shm_req_alloc_timeout_seconds.cache_clear() +def test_pd_node_resource_wait_timeout_reads_environment_variable(monkeypatch): + monkeypatch.setenv("LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS", "30") + get_pd_node_resource_wait_timeout_seconds.cache_clear() - assert get_pd_node_shm_req_alloc_timeout_seconds() == 30 + assert get_pd_node_resource_wait_timeout_seconds() == 30 - get_pd_node_shm_req_alloc_timeout_seconds.cache_clear() - - -def test_pd_node_router_wait_timeout_defaults_to_20_seconds(monkeypatch): - monkeypatch.delenv("LIGHTLLM_PD_NODE_ROUTER_WAIT_TIMEOUT_SECONDS", raising=False) - get_pd_node_router_wait_timeout_seconds.cache_clear() - - assert get_pd_node_router_wait_timeout_seconds() == 20 - - get_pd_node_router_wait_timeout_seconds.cache_clear() - - -def test_pd_node_router_wait_timeout_reads_environment_variable(monkeypatch): - monkeypatch.setenv("LIGHTLLM_PD_NODE_ROUTER_WAIT_TIMEOUT_SECONDS", "45") - get_pd_node_router_wait_timeout_seconds.cache_clear() - - assert get_pd_node_router_wait_timeout_seconds() == 45 - - get_pd_node_router_wait_timeout_seconds.cache_clear() + get_pd_node_resource_wait_timeout_seconds.cache_clear() From 347c97a35e610994e657d6a2aef846b37798f389 Mon Sep 17 00:00:00 2001 From: wangzaijun Date: Fri, 4 Sep 2026 05:49:05 +0000 Subject: [PATCH 04/12] docs(pd): explain TP slave timeout handling --- lightllm/server/httpserver/manager.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/lightllm/server/httpserver/manager.py b/lightllm/server/httpserver/manager.py index 237c050e9..370eea628 100644 --- a/lightllm/server/httpserver/manager.py +++ b/lightllm/server/httpserver/manager.py @@ -556,6 +556,8 @@ async def _alloc_shm_req_indexes( """ alloced_req_indexes = [] alloc_timeout_seconds = None + # 多机 TP 各 rank 必须保持请求执行一致；slave 若按本地计时独立超时退出，可能导致 + # master/其他 rank 继续进入 collective 而发生状态不一致或阻塞，因此超时由 master 统一决策。 if not self.is_multinode_tp_slave and pd_node_resource_wait_timeout_seconds >= 0: alloc_timeout_seconds = pd_node_resource_wait_timeout_seconds alloc_deadline = time.monotonic() + alloc_timeout_seconds if alloc_timeout_seconds is not None else None From 7ea97a534a03315a60786f62cce9cfb8a908ee51 Mon Sep 17 00:00:00 2001 From: wangzaijun Date: Fri, 4 Sep 2026 05:56:13 +0000 Subject: [PATCH 05/12] fix(pd): separate request limit flag from timeout --- .../server/httpserver_for_pd_master/manager.py | 15 +++++++-------- .../test_pd_master_multi_choice.py | 10 +++++++--- 2 files changed, 14 insertions(+), 11 deletions(-) diff --git a/lightllm/server/httpserver_for_pd_master/manager.py b/lightllm/server/httpserver_for_pd_master/manager.py index 189b15549..4488d924a 100644 --- a/lightllm/server/httpserver_for_pd_master/manager.py +++ b/lightllm/server/httpserver_for_pd_master/manager.py @@ -52,11 +52,9 @@ def __init__( self.health_timeout = int(os.getenv("HEALTH_TIMEOUT", "200")) self.latest_success_infer_time = time.time() self.running_request_count = 0 - # 限流开关只在 PD Master 生效。开启后由 Master 统一下发资源等待上限； - # 未开启时下发 -1，P/D 节点不读取本地开关或超时配置，只执行收到的值。 - self.pd_node_resource_wait_timeout_seconds = ( - get_pd_node_resource_wait_timeout_seconds() if args.enable_pd_node_self_request_limit else -1 - ) + # 限流开关只在 PD Master 生效；P/D 节点不读取本地开关或超时配置，只执行 Master 下发的值。 + self.enable_pd_node_self_request_limit = args.enable_pd_node_self_request_limit + self.pd_node_resource_wait_timeout_seconds = get_pd_node_resource_wait_timeout_seconds() self.pd_cache_high_priority_max_age_seconds = get_pd_cache_high_priority_max_age_seconds() self.pd_cache_high_priority_min_prompt_tokens = get_pd_cache_high_priority_min_prompt_tokens() self.disable_pd_cache_high_priority = args.disable_pd_cache_high_priority @@ -279,9 +277,10 @@ async def _generate_one( # KV cache 插队。第二段及后续分段仍统一使用高优先级，避免因临时资源 # 紧张导致分段续跑失败。 sampling_params.pd_high_priority_request = segment_index > 0 or has_fresh_high_cache_hit - # 资源等待超时与请求优先级相互独立；P/D 节点直接使用 PD Master - # 下发值，负数表示持续等待资源。 - sampling_params.pd_node_resource_wait_timeout_seconds = self.pd_node_resource_wait_timeout_seconds + # 资源等待超时与请求优先级相互独立；仅在 Master 开启限流时下发配置值， + # 否则保留 SamplingParams 的默认值（负数表示持续等待资源）。 + if self.enable_pd_node_self_request_limit: + sampling_params.pd_node_resource_wait_timeout_seconds = self.pd_node_resource_wait_timeout_seconds # 分段请求始终复用循环外选定的 P 节点；这里只按每段实际发送的 # prompt 更新该节点的在途 prefill 负载，不会重新选点。 diff --git a/test/test_pd_selector/test_pd_master_multi_choice.py b/test/test_pd_selector/test_pd_master_multi_choice.py index d3b62704c..cfdf089d6 100644 --- a/test/test_pd_selector/test_pd_master_multi_choice.py +++ b/test/test_pd_selector/test_pd_master_multi_choice.py @@ -20,7 +20,8 @@ def _manager() -> HttpServerManagerForPDMaster: manager.metric_client = MagicMock() manager._log_req_header = AsyncMock() manager.tokens = MagicMock(return_value=2) - manager.pd_node_resource_wait_timeout_seconds = -1 + manager.enable_pd_node_self_request_limit = False + manager.pd_node_resource_wait_timeout_seconds = 10 manager.pd_cache_high_priority_max_age_seconds = 60 manager.pd_cache_high_priority_min_prompt_tokens = 4096 manager.disable_pd_cache_high_priority = False @@ -574,9 +575,11 @@ async def wait_to_token_package(_p_node, _d_node, _start_time, _prompt, sampling asyncio.run(asyncio.wait_for(run(), timeout=2)) -def test_pd_master_sets_resource_wait_timeout_independently_of_priority(): +@pytest.mark.parametrize(("enable_limit", "expected_timeout"), [(False, -1), (True, 90)]) +def test_pd_master_sets_resource_wait_timeout_when_enabled(enable_limit, expected_timeout): async def run(): manager = _manager() + manager.enable_pd_node_self_request_limit = enable_limit manager.pd_node_resource_wait_timeout_seconds = 90 manager.id_gen.generate_id.return_value = 808 manager.remove_req = AsyncMock() @@ -612,6 +615,7 @@ async def wait_to_token_package(_p_node, _d_node, _start_time, _prompt, sampling manager._wait_to_token_package = wait_to_token_package sampling_params = SamplingParams() + sampling_params.pd_node_resource_wait_timeout_seconds = -1 sampling_params.max_new_tokens = 1 async for _ in manager._generate_one( "prompt", @@ -624,7 +628,7 @@ async def wait_to_token_package(_p_node, _d_node, _start_time, _prompt, sampling ): pass - assert captured_request_settings == [(False, 90)] + assert captured_request_settings == [(False, expected_timeout)] asyncio.run(asyncio.wait_for(run(), timeout=2)) From b9c81ae0b87b2ed7364f4dc6e61c9fc4ee15c0d8 Mon Sep 17 00:00:00 2001 From: wangzaijun Date: Fri, 4 Sep 2026 06:31:05 +0000 Subject: [PATCH 06/12] feat(pd): retry busy nodes during request limiting --- docs/CN/source/tutorial/api_server_args.rst | 17 ++- docs/EN/source/tutorial/api_server_args.rst | 20 ++- lightllm/server/api_cli.py | 8 +- lightllm/server/core/objs/start_args_type.py | 2 +- .../httpserver_for_pd_master/manager.py | 55 +++++++- lightllm/utils/envs_utils.py | 8 +- .../test_pd_master_multi_choice.py | 130 +++++++++++++++++- .../test_pd_master_cached_tokens.py | 2 + unit_tests/server/test_pd_master_mode.py | 8 +- unit_tests/utils/test_envs_utils.py | 23 +++- 10 files changed, 245 insertions(+), 28 deletions(-) diff --git a/docs/CN/source/tutorial/api_server_args.rst b/docs/CN/source/tutorial/api_server_args.rst index fa78ab611..540b78a82 100644 --- a/docs/CN/source/tutorial/api_server_args.rst +++ b/docs/CN/source/tutorial/api_server_args.rst @@ -92,19 +92,22 @@ PD 分离模式参数 推理进度健康检查：当仍有在途请求，且整个 PD Master 连续 ``HEALTH_TIMEOUT`` 秒 没有任何请求成功返回 token 时，接口将返回 HTTP 503。 -.. option:: --enable_pd_node_self_request_limit +.. option:: --disable_pd_node_self_request_limit - 启用由 PD Master 统一管理的 P/D 节点资源等待限流。该参数只需要在 PD Master 启动时设置， - 不需要在 Prefill/Decode 节点上设置。开启后，PD Master 通过 + P/D 节点资源等待限流默认启用，并由 PD Master 统一管理。该参数只在需要关闭此功能时设置，且只需添加到 + PD Master 的启动参数中，不需要在 Prefill/Decode 节点上设置。默认情况下，PD Master 通过 ``pd_node_resource_wait_timeout_seconds`` 为所有请求下发统一的资源等待上限；P/D 节点只负责按下发值 控制本地 ``shm_req`` 申请和 Router 等待进入推理系统，不读取本地限流开关或超时配置，也不根据请求 是否为高优先级改变超时。该值由 PD Master 上的 ``LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS`` 控制，默认 10 秒；设置为 -1 表示永久等待。 设置为非负数时，超时会导致 ``Server is busy``； 其中已进入 Router 但仍未进入推理系统的请求会主动标记为 aborted，由 PD Master 转换为 HTTP 429。 - 未设置该启动参数时，PD Master 统一下发 -1，即所有 P/D 节点永久等待。 + 本功能启用时，PD Master 收到 ``Server is busy`` 会重新选择 P/D 节点并重试；最长探测周期由 + ``LIGHTLLM_PD_NODE_BUSY_RETRY_TIMEOUT_SECONDS`` 控制，默认 120 秒。若请求已经向客户端输出 token， + 则不再从头重试，以免产生重复内容。设置 ``--disable_pd_node_self_request_limit`` 后，PD Master 不再下发 + 有限的资源等待时间；P/D 节点永久等待，其他原因产生的 ``Server is busy`` 也会直接返回，不触发重试。 多机 TP 场景仅由 master 节点执行超时判断，slave 节点永久等待。cache 命中记录允许提升优先级的最大年龄由 - ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MAX_AGE_SECONDS`` 控制，默认 16 秒。cache 命中提权还要求输入 + ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MAX_AGE_SECONDS`` 控制，默认 36 秒。cache 命中提权还要求输入 token 数达到 ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MIN_PROMPT_TOKENS`` 配置的门槛（默认 4096），避免短请求仅因 cache 命中率高而提升优先级。 @@ -113,8 +116,8 @@ PD 分离模式参数 .. code-block:: bash LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS=10 \ - python -m lightllm.server.api_server --run_mode pd_master \ - --enable_pd_node_self_request_limit ... + LIGHTLLM_PD_NODE_BUSY_RETRY_TIMEOUT_SECONDS=120 \ + python -m lightllm.server.api_server --run_mode pd_master ... .. option:: --disable_pd_cache_high_priority diff --git a/docs/EN/source/tutorial/api_server_args.rst b/docs/EN/source/tutorial/api_server_args.rst index 9bafcae7b..0102f7ce0 100644 --- a/docs/EN/source/tutorial/api_server_args.rst +++ b/docs/EN/source/tutorial/api_server_args.rst @@ -95,21 +95,27 @@ PD disaggregation Mode Parameters the endpoints return HTTP 503 if no request on the PD Master successfully returns a token for ``HEALTH_TIMEOUT`` consecutive seconds. -.. option:: --enable_pd_node_self_request_limit +.. option:: --disable_pd_node_self_request_limit - Enable PD Master-managed resource wait limiting for P/D nodes. Set this option only when starting PD Master; - it is not needed on Prefill or Decode nodes. Once enabled, PD Master supplies + P/D-node resource wait limiting is enabled by default and managed centrally by PD Master. Set this option only + when disabling the feature, and only when starting PD Master; it is not needed on Prefill or Decode nodes. + By default, PD Master supplies ``pd_node_resource_wait_timeout_seconds`` for every request. P/D nodes only enforce the received value for local ``shm_req`` allocation and the wait from Router entry to inference entry; they do not read local limiting switches or timeout settings, and request priority does not alter the timeout. The value is controlled on PD Master by ``LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS`` and defaults to 10 seconds; set it to -1 to wait indefinitely. When set to a non-negative value, a timeout reports ``Server is busy``; a request that has entered the Router but not inference is proactively marked aborted, and PD Master converts this to HTTP 429. - Without this startup option, PD Master sends -1 and all P/D nodes wait indefinitely. + While this feature is enabled, PD Master selects P/D nodes again and retries after receiving ``Server is busy``. + The maximum probing period is controlled by ``LIGHTLLM_PD_NODE_BUSY_RETRY_TIMEOUT_SECONDS`` and defaults to + 120 seconds. Once response tokens have been streamed to the client, the request is not restarted because doing so + would duplicate output. With ``--disable_pd_node_self_request_limit``, PD Master no longer supplies a finite + resource wait timeout; all P/D nodes wait indefinitely, and a ``Server is busy`` raised for another reason is + returned immediately without retrying. In multi-node TP deployments, only the master node evaluates the timeout; slave nodes wait indefinitely. The maximum cache-record age eligible for promotion is controlled by ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MAX_AGE_SECONDS`` and defaults to - 16 seconds. Cache-hit promotion also requires at least the number of input tokens configured by + 36 seconds. Cache-hit promotion also requires at least the number of input tokens configured by ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MIN_PROMPT_TOKENS`` (4096 by default), so short requests do not gain priority solely from a high cache-hit rate. @@ -118,8 +124,8 @@ PD disaggregation Mode Parameters .. code-block:: bash LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS=10 \ - python -m lightllm.server.api_server --run_mode pd_master \ - --enable_pd_node_self_request_limit ... + LIGHTLLM_PD_NODE_BUSY_RETRY_TIMEOUT_SECONDS=120 \ + python -m lightllm.server.api_server --run_mode pd_master ... .. option:: --disable_pd_cache_high_priority diff --git a/lightllm/server/api_cli.py b/lightllm/server/api_cli.py index fa19be299..c79f59c18 100644 --- a/lightllm/server/api_cli.py +++ b/lightllm/server/api_cli.py @@ -77,12 +77,12 @@ def add_cli_args(parser: argparse.ArgumentParser) -> argparse.ArgumentParser: ), ) parser.add_argument( - "--enable_pd_node_self_request_limit", + "--disable_pd_node_self_request_limit", action="store_true", help=( - "Enable PD Master-managed resource wait limiting for Prefill/Decode nodes. Configure this option only " - "on PD Master; it sends all timeout details to P/D nodes, which only enforce the received values. " - "Default: disabled." + "Disable PD Master-managed resource wait limiting and retries for requests rejected as server busy. " + "Configure this option only on PD Master. By default, PD Master sends timeout details to P/D nodes, " + "which only enforce the received values, and retries busy requests." ), ) parser.add_argument( diff --git a/lightllm/server/core/objs/start_args_type.py b/lightllm/server/core/objs/start_args_type.py index 2d0712e60..9c89975de 100644 --- a/lightllm/server/core/objs/start_args_type.py +++ b/lightllm/server/core/objs/start_args_type.py @@ -23,7 +23,7 @@ class StartArgs: pd_master_ip: str = field(default="0.0.0.0") pd_master_port: int = field(default=1212) pd_master_mode: str = field(default="elastic") - enable_pd_node_self_request_limit: bool = field(default=False) + disable_pd_node_self_request_limit: bool = field(default=False) disable_pd_cache_high_priority: bool = field(default=False) pd_trans_mode: str = field(default="nccl", metadata={"choices": ["nccl", "nixl"]}) config_server_host: str = field(default=None) diff --git a/lightllm/server/httpserver_for_pd_master/manager.py b/lightllm/server/httpserver_for_pd_master/manager.py index 4488d924a..31d26250e 100644 --- a/lightllm/server/httpserver_for_pd_master/manager.py +++ b/lightllm/server/httpserver_for_pd_master/manager.py @@ -26,6 +26,7 @@ from lightllm.utils.envs_utils import ( get_pd_cache_high_priority_max_age_seconds, get_pd_cache_high_priority_min_prompt_tokens, + get_pd_node_busy_retry_timeout_seconds, get_pd_node_resource_wait_timeout_seconds, ) from lightllm.utils.shm_port_args import get_shm_port_args @@ -53,8 +54,9 @@ def __init__( self.latest_success_infer_time = time.time() self.running_request_count = 0 # 限流开关只在 PD Master 生效；P/D 节点不读取本地开关或超时配置，只执行 Master 下发的值。 - self.enable_pd_node_self_request_limit = args.enable_pd_node_self_request_limit + self.enable_pd_node_self_request_limit = not args.disable_pd_node_self_request_limit self.pd_node_resource_wait_timeout_seconds = get_pd_node_resource_wait_timeout_seconds() + self.pd_node_busy_retry_timeout_seconds = get_pd_node_busy_retry_timeout_seconds() self.pd_cache_high_priority_max_age_seconds = get_pd_cache_high_priority_max_age_seconds() self.pd_cache_high_priority_min_prompt_tokens = get_pd_cache_high_priority_min_prompt_tokens() self.disable_pd_cache_high_priority = args.disable_pd_cache_high_priority @@ -224,6 +226,57 @@ async def _generate_one( origin_request_id: int, input_token_num: int, ): + """节点繁忙时重新选择 P/D 节点，并在配置的探测周期内重试。""" + retry_start_time = time.monotonic() + has_yielded_result = False + + while True: + try: + generator = self._generate_one_attempt( + prompt, + origin_sampling_params, + multimodal_params, + request, + start_time, + origin_request_id, + input_token_num, + ) + async with aclosing(generator): + async for result in generator: + has_yielded_result = True + yield result + return + except ServerBusyError: + # 关闭节点自限流时，不启用与该策略配套的 busy 重试，直接透传异常。 + if not self.enable_pd_node_self_request_limit: + raise + + # 已向客户端输出 token 后不能从头生成，否则会产生重复内容。 + elapsed_seconds = time.monotonic() - retry_start_time + if has_yielded_result or elapsed_seconds >= self.pd_node_busy_retry_timeout_seconds: + raise + logger.warning( + f"group_request_id: {origin_request_id} PD node is busy, retrying with another node; " + f"elapsed: {elapsed_seconds:.3f}s, retry timeout: " + f"{self.pd_node_busy_retry_timeout_seconds}s" + ) + + async def _generate_one_attempt( + self, + prompt: str, + origin_sampling_params: SamplingParams, + multimodal_params: MultimodalParams, + request: Request, + start_time: float, + origin_request_id: int, + input_token_num: int, + ): + """执行一次完整的单 choice 生成尝试。 + + 本函数负责选择 P/D 节点、执行所有分段生成，以及在结束或异常时清理请求和节点负载； + 它不处理重试。若节点返回 ``ServerBusyError``，异常会在本次清理完成后交给 + ``_generate_one``，由外层决定是否重新选择节点并发起下一次尝试。 + """ block_group_request_id = origin_request_id p_node = None d_node = None diff --git a/lightllm/utils/envs_utils.py b/lightllm/utils/envs_utils.py index 0f7120370..2fd397583 100644 --- a/lightllm/utils/envs_utils.py +++ b/lightllm/utils/envs_utils.py @@ -312,10 +312,16 @@ def get_pd_node_resource_wait_timeout_seconds() -> int: return int(os.getenv("LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS", 10)) +@lru_cache(maxsize=None) +def get_pd_node_busy_retry_timeout_seconds() -> int: + """PD Master 收到节点繁忙错误后的最长重试时间，单位为秒。""" + return max(0, int(os.getenv("LIGHTLLM_PD_NODE_BUSY_RETRY_TIMEOUT_SECONDS", 120))) + + @lru_cache(maxsize=None) def get_pd_cache_high_priority_max_age_seconds() -> int: """cache 命中请求提升为 PD 高优先级时允许的最大缓存年龄，单位为秒。""" - return max(0, int(os.getenv("LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MAX_AGE_SECONDS", 16))) + return max(0, int(os.getenv("LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MAX_AGE_SECONDS", 36))) @lru_cache(maxsize=None) diff --git a/test/test_pd_selector/test_pd_master_multi_choice.py b/test/test_pd_selector/test_pd_master_multi_choice.py index cfdf089d6..9129ebd62 100644 --- a/test/test_pd_selector/test_pd_master_multi_choice.py +++ b/test/test_pd_selector/test_pd_master_multi_choice.py @@ -10,6 +10,7 @@ from lightllm.server.httpserver.manager import HttpServerManager from lightllm.server.httpserver_for_pd_master.manager import HttpServerManagerForPDMaster from lightllm.server.httpserver_for_pd_master.pd_selector import PDSelectionExtraInfo +from lightllm.utils.error_utils import ServerBusyError def _manager() -> HttpServerManagerForPDMaster: @@ -20,8 +21,9 @@ def _manager() -> HttpServerManagerForPDMaster: manager.metric_client = MagicMock() manager._log_req_header = AsyncMock() manager.tokens = MagicMock(return_value=2) - manager.enable_pd_node_self_request_limit = False + manager.enable_pd_node_self_request_limit = True manager.pd_node_resource_wait_timeout_seconds = 10 + manager.pd_node_busy_retry_timeout_seconds = 120 manager.pd_cache_high_priority_max_age_seconds = 60 manager.pd_cache_high_priority_min_prompt_tokens = 4096 manager.disable_pd_cache_high_priority = False @@ -633,6 +635,132 @@ async def wait_to_token_package(_p_node, _d_node, _start_time, _prompt, sampling asyncio.run(asyncio.wait_for(run(), timeout=2)) +def test_pd_master_retries_generate_one_when_node_is_busy(): + async def run(): + manager = _manager() + manager.enable_pd_node_self_request_limit = True + attempt_count = 0 + + async def generate_one_attempt(*_args): + nonlocal attempt_count + attempt_count += 1 + if attempt_count == 1: + raise ServerBusyError("node is busy") + yield 800, "ok", {}, FinishStatus(FinishStatus.FINISHED_STOP) + + manager._generate_one_attempt = generate_one_attempt + results = [ + result + async for result in manager._generate_one( + "prompt", + SamplingParams(), + MagicMock(), + MagicMock(), + 0, + 800, + 0, + ) + ] + + assert attempt_count == 2 + assert [result[1] for result in results] == ["ok"] + + asyncio.run(asyncio.wait_for(run(), timeout=2)) + + +def test_pd_master_does_not_retry_busy_error_when_self_limit_is_disabled(): + async def run(): + manager = _manager() + manager.enable_pd_node_self_request_limit = False + attempt_count = 0 + + async def generate_one_attempt(*_args): + nonlocal attempt_count + attempt_count += 1 + raise ServerBusyError("node is busy") + yield + + manager._generate_one_attempt = generate_one_attempt + with pytest.raises(ServerBusyError, match="node is busy"): + async for _ in manager._generate_one( + "prompt", + SamplingParams(), + MagicMock(), + MagicMock(), + 0, + 800, + 0, + ): + pass + + assert attempt_count == 1 + + asyncio.run(asyncio.wait_for(run(), timeout=2)) + + +def test_pd_master_stops_busy_retry_when_probe_period_expires(): + async def run(): + manager = _manager() + manager.enable_pd_node_self_request_limit = True + manager.pd_node_busy_retry_timeout_seconds = 0 + attempt_count = 0 + + async def generate_one_attempt(*_args): + nonlocal attempt_count + attempt_count += 1 + raise ServerBusyError("node is busy") + yield + + manager._generate_one_attempt = generate_one_attempt + with pytest.raises(ServerBusyError, match="node is busy"): + async for _ in manager._generate_one( + "prompt", + SamplingParams(), + MagicMock(), + MagicMock(), + 0, + 800, + 0, + ): + pass + + assert attempt_count == 1 + + asyncio.run(asyncio.wait_for(run(), timeout=2)) + + +def test_pd_master_does_not_retry_busy_error_after_streaming_output(): + async def run(): + manager = _manager() + manager.enable_pd_node_self_request_limit = True + attempt_count = 0 + results = [] + + async def generate_one_attempt(*_args): + nonlocal attempt_count + attempt_count += 1 + yield 800, "visible", {}, FinishStatus() + raise ServerBusyError("node is busy") + + manager._generate_one_attempt = generate_one_attempt + with pytest.raises(ServerBusyError, match="node is busy"): + async for result in manager._generate_one( + "prompt", + SamplingParams(), + MagicMock(), + MagicMock(), + 0, + 800, + 0, + ): + results.append(result) + + assert attempt_count == 1 + assert [result[1] for result in results] == ["visible"] + + asyncio.run(asyncio.wait_for(run(), timeout=2)) + + def test_pd_master_releases_prefill_load_when_stream_is_closed(): async def run(): manager = _manager() diff --git a/unit_tests/server/httpserver/test_pd_master_cached_tokens.py b/unit_tests/server/httpserver/test_pd_master_cached_tokens.py index b514a8429..666900a38 100644 --- a/unit_tests/server/httpserver/test_pd_master_cached_tokens.py +++ b/unit_tests/server/httpserver/test_pd_master_cached_tokens.py @@ -17,7 +17,9 @@ def _make_manager(monkeypatch): monkeypatch.setattr(SamplingParams, "from_buffer_copy", classmethod(lambda cls, other: copy.copy(other))) mgr = object.__new__(HttpServerManagerForPDMaster) mgr.args = SimpleNamespace(disable_pd_master_decode_capacity_limit=True) + mgr.enable_pd_node_self_request_limit = True mgr.pd_node_resource_wait_timeout_seconds = -1 + mgr.pd_node_busy_retry_timeout_seconds = 120 mgr.pd_cache_high_priority_max_age_seconds = 60 mgr.pd_cache_high_priority_min_prompt_tokens = 8192 mgr.disable_pd_cache_high_priority = False diff --git a/unit_tests/server/test_pd_master_mode.py b/unit_tests/server/test_pd_master_mode.py index 86bc48855..6e1d006e0 100644 --- a/unit_tests/server/test_pd_master_mode.py +++ b/unit_tests/server/test_pd_master_mode.py @@ -10,12 +10,12 @@ from lightllm.server.httpserver_for_pd_master.manager import HttpServerManagerForPDMaster, PDManager -def test_pd_node_self_request_limit_cli_is_configured_on_pd_master(): +def test_pd_node_self_request_limit_cli_defaults_to_enabled_and_can_be_disabled(): parser = make_argument_parser() - assert parser.parse_args([]).enable_pd_node_self_request_limit is False - assert parser.parse_args(["--enable_pd_node_self_request_limit"]).enable_pd_node_self_request_limit is True - assert StartArgs().enable_pd_node_self_request_limit is False + assert parser.parse_args([]).disable_pd_node_self_request_limit is False + assert parser.parse_args(["--disable_pd_node_self_request_limit"]).disable_pd_node_self_request_limit is True + assert StartArgs().disable_pd_node_self_request_limit is False def test_pd_cache_high_priority_cli_defaults_to_enabled_and_can_be_disabled(): diff --git a/unit_tests/utils/test_envs_utils.py b/unit_tests/utils/test_envs_utils.py index 57baf0539..0bdbb61df 100644 --- a/unit_tests/utils/test_envs_utils.py +++ b/unit_tests/utils/test_envs_utils.py @@ -1,15 +1,16 @@ from lightllm.utils.envs_utils import ( get_pd_cache_high_priority_max_age_seconds, get_pd_cache_high_priority_min_prompt_tokens, + get_pd_node_busy_retry_timeout_seconds, get_pd_node_resource_wait_timeout_seconds, ) -def test_pd_cache_high_priority_max_age_defaults_to_16_seconds(monkeypatch): +def test_pd_cache_high_priority_max_age_defaults_to_36_seconds(monkeypatch): monkeypatch.delenv("LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MAX_AGE_SECONDS", raising=False) get_pd_cache_high_priority_max_age_seconds.cache_clear() - assert get_pd_cache_high_priority_max_age_seconds() == 16 + assert get_pd_cache_high_priority_max_age_seconds() == 36 get_pd_cache_high_priority_max_age_seconds.cache_clear() @@ -57,3 +58,21 @@ def test_pd_node_resource_wait_timeout_reads_environment_variable(monkeypatch): assert get_pd_node_resource_wait_timeout_seconds() == 30 get_pd_node_resource_wait_timeout_seconds.cache_clear() + + +def test_pd_node_busy_retry_timeout_defaults_to_120_seconds(monkeypatch): + monkeypatch.delenv("LIGHTLLM_PD_NODE_BUSY_RETRY_TIMEOUT_SECONDS", raising=False) + get_pd_node_busy_retry_timeout_seconds.cache_clear() + + assert get_pd_node_busy_retry_timeout_seconds() == 120 + + get_pd_node_busy_retry_timeout_seconds.cache_clear() + + +def test_pd_node_busy_retry_timeout_reads_environment_variable(monkeypatch): + monkeypatch.setenv("LIGHTLLM_PD_NODE_BUSY_RETRY_TIMEOUT_SECONDS", "60") + get_pd_node_busy_retry_timeout_seconds.cache_clear() + + assert get_pd_node_busy_retry_timeout_seconds() == 60 + + get_pd_node_busy_retry_timeout_seconds.cache_clear() From bbb64658c7a9948b1cf246af4a5931c79d17a2a3 Mon Sep 17 00:00:00 2001 From: wangzaijun Date: Fri, 4 Sep 2026 06:43:19 +0000 Subject: [PATCH 07/12] feat(pd): extend continuation resource wait timeout --- docs/CN/source/tutorial/api_server_args.rst | 8 ++++++-- docs/EN/source/tutorial/api_server_args.rst | 8 ++++++-- .../httpserver_for_pd_master/manager.py | 13 ++++++++++--- lightllm/utils/envs_utils.py | 6 ++++++ .../test_pd_master_multi_choice.py | 18 +++++++++++++++++- .../test_pd_master_cached_tokens.py | 1 + unit_tests/utils/test_envs_utils.py | 19 +++++++++++++++++++ 7 files changed, 65 insertions(+), 8 deletions(-) diff --git a/docs/CN/source/tutorial/api_server_args.rst b/docs/CN/source/tutorial/api_server_args.rst index 540b78a82..586ace215 100644 --- a/docs/CN/source/tutorial/api_server_args.rst +++ b/docs/CN/source/tutorial/api_server_args.rst @@ -97,9 +97,12 @@ PD 分离模式参数 P/D 节点资源等待限流默认启用，并由 PD Master 统一管理。该参数只在需要关闭此功能时设置，且只需添加到 PD Master 的启动参数中，不需要在 Prefill/Decode 节点上设置。默认情况下，PD Master 通过 ``pd_node_resource_wait_timeout_seconds`` 为所有请求下发统一的资源等待上限；P/D 节点只负责按下发值 - 控制本地 ``shm_req`` 申请和 Router 等待进入推理系统，不读取本地限流开关或超时配置，也不根据请求 - 是否为高优先级改变超时。该值由 PD Master 上的 + 控制本地 ``shm_req`` 申请和 Router 等待进入推理系统，不读取本地限流开关或超时配置。首段的等待上限由 + PD Master 上的 ``LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS`` 控制，默认 10 秒；设置为 -1 表示永久等待。 + ``segment_index > 0`` 的续跑分段使用独立的等待上限，该值由 + ``LIGHTLLM_PD_NODE_CONTINUATION_RESOURCE_WAIT_TIMEOUT_SECONDS`` 控制，默认 60 秒，以提高已产生部分结果的 + 请求最终完成的成功率。 设置为非负数时，超时会导致 ``Server is busy``； 其中已进入 Router 但仍未进入推理系统的请求会主动标记为 aborted，由 PD Master 转换为 HTTP 429。 本功能启用时，PD Master 收到 ``Server is busy`` 会重新选择 P/D 节点并重试；最长探测周期由 @@ -116,6 +119,7 @@ PD 分离模式参数 .. code-block:: bash LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS=10 \ + LIGHTLLM_PD_NODE_CONTINUATION_RESOURCE_WAIT_TIMEOUT_SECONDS=60 \ LIGHTLLM_PD_NODE_BUSY_RETRY_TIMEOUT_SECONDS=120 \ python -m lightllm.server.api_server --run_mode pd_master ... diff --git a/docs/EN/source/tutorial/api_server_args.rst b/docs/EN/source/tutorial/api_server_args.rst index 0102f7ce0..b5c782330 100644 --- a/docs/EN/source/tutorial/api_server_args.rst +++ b/docs/EN/source/tutorial/api_server_args.rst @@ -102,9 +102,12 @@ PD disaggregation Mode Parameters By default, PD Master supplies ``pd_node_resource_wait_timeout_seconds`` for every request. P/D nodes only enforce the received value for local ``shm_req`` allocation and the wait from Router entry to inference entry; they do not read local limiting switches - or timeout settings, and request priority does not alter the timeout. The value is + or timeout settings. The first segment's timeout is controlled on PD Master by ``LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS`` and defaults to 10 seconds; set it - to -1 to wait indefinitely. When set to a non-negative value, a timeout reports ``Server is busy``; a request that has + to -1 to wait indefinitely. Continuation segments with ``segment_index > 0`` use a separate timeout controlled by + ``LIGHTLLM_PD_NODE_CONTINUATION_RESOURCE_WAIT_TIMEOUT_SECONDS`` and defaults to 60 seconds, improving the chance + that requests which have already produced partial results complete successfully. When set to a non-negative value, + a timeout reports ``Server is busy``; a request that has entered the Router but not inference is proactively marked aborted, and PD Master converts this to HTTP 429. While this feature is enabled, PD Master selects P/D nodes again and retries after receiving ``Server is busy``. The maximum probing period is controlled by ``LIGHTLLM_PD_NODE_BUSY_RETRY_TIMEOUT_SECONDS`` and defaults to @@ -124,6 +127,7 @@ PD disaggregation Mode Parameters .. code-block:: bash LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS=10 \ + LIGHTLLM_PD_NODE_CONTINUATION_RESOURCE_WAIT_TIMEOUT_SECONDS=60 \ LIGHTLLM_PD_NODE_BUSY_RETRY_TIMEOUT_SECONDS=120 \ python -m lightllm.server.api_server --run_mode pd_master ... diff --git a/lightllm/server/httpserver_for_pd_master/manager.py b/lightllm/server/httpserver_for_pd_master/manager.py index 31d26250e..ba569f48b 100644 --- a/lightllm/server/httpserver_for_pd_master/manager.py +++ b/lightllm/server/httpserver_for_pd_master/manager.py @@ -27,6 +27,7 @@ get_pd_cache_high_priority_max_age_seconds, get_pd_cache_high_priority_min_prompt_tokens, get_pd_node_busy_retry_timeout_seconds, + get_pd_node_continuation_resource_wait_timeout_seconds, get_pd_node_resource_wait_timeout_seconds, ) from lightllm.utils.shm_port_args import get_shm_port_args @@ -56,6 +57,9 @@ def __init__( # 限流开关只在 PD Master 生效；P/D 节点不读取本地开关或超时配置，只执行 Master 下发的值。 self.enable_pd_node_self_request_limit = not args.disable_pd_node_self_request_limit self.pd_node_resource_wait_timeout_seconds = get_pd_node_resource_wait_timeout_seconds() + self.pd_node_continuation_resource_wait_timeout_seconds = ( + get_pd_node_continuation_resource_wait_timeout_seconds() + ) self.pd_node_busy_retry_timeout_seconds = get_pd_node_busy_retry_timeout_seconds() self.pd_cache_high_priority_max_age_seconds = get_pd_cache_high_priority_max_age_seconds() self.pd_cache_high_priority_min_prompt_tokens = get_pd_cache_high_priority_min_prompt_tokens() @@ -330,10 +334,13 @@ async def _generate_one_attempt( # KV cache 插队。第二段及后续分段仍统一使用高优先级，避免因临时资源 # 紧张导致分段续跑失败。 sampling_params.pd_high_priority_request = segment_index > 0 or has_fresh_high_cache_hit - # 资源等待超时与请求优先级相互独立；仅在 Master 开启限流时下发配置值， - # 否则保留 SamplingParams 的默认值（负数表示持续等待资源）。 + # 仅在 Master 开启限流时下发资源等待超时。续跑分段已经产生了部分结果， + # 使用独立配置的等待时间，提高请求最终完成的成功率。 if self.enable_pd_node_self_request_limit: - sampling_params.pd_node_resource_wait_timeout_seconds = self.pd_node_resource_wait_timeout_seconds + resource_wait_timeout_seconds = self.pd_node_resource_wait_timeout_seconds + if segment_index > 0: + resource_wait_timeout_seconds = self.pd_node_continuation_resource_wait_timeout_seconds + sampling_params.pd_node_resource_wait_timeout_seconds = resource_wait_timeout_seconds # 分段请求始终复用循环外选定的 P 节点；这里只按每段实际发送的 # prompt 更新该节点的在途 prefill 负载，不会重新选点。 diff --git a/lightllm/utils/envs_utils.py b/lightllm/utils/envs_utils.py index 2fd397583..4405e822a 100644 --- a/lightllm/utils/envs_utils.py +++ b/lightllm/utils/envs_utils.py @@ -312,6 +312,12 @@ def get_pd_node_resource_wait_timeout_seconds() -> int: return int(os.getenv("LIGHTLLM_PD_NODE_RESOURCE_WAIT_TIMEOUT_SECONDS", 10)) +@lru_cache(maxsize=None) +def get_pd_node_continuation_resource_wait_timeout_seconds() -> int: + """P/D 节点处理续跑分段时的资源等待超时，单位为秒。""" + return max(0, int(os.getenv("LIGHTLLM_PD_NODE_CONTINUATION_RESOURCE_WAIT_TIMEOUT_SECONDS", 60))) + + @lru_cache(maxsize=None) def get_pd_node_busy_retry_timeout_seconds() -> int: """PD Master 收到节点繁忙错误后的最长重试时间，单位为秒。""" diff --git a/test/test_pd_selector/test_pd_master_multi_choice.py b/test/test_pd_selector/test_pd_master_multi_choice.py index 9129ebd62..8d48f5d5c 100644 --- a/test/test_pd_selector/test_pd_master_multi_choice.py +++ b/test/test_pd_selector/test_pd_master_multi_choice.py @@ -23,6 +23,7 @@ def _manager() -> HttpServerManagerForPDMaster: manager.tokens = MagicMock(return_value=2) manager.enable_pd_node_self_request_limit = True manager.pd_node_resource_wait_timeout_seconds = 10 + manager.pd_node_continuation_resource_wait_timeout_seconds = 60 manager.pd_node_busy_retry_timeout_seconds = 120 manager.pd_cache_high_priority_max_age_seconds = 60 manager.pd_cache_high_priority_min_prompt_tokens = 4096 @@ -191,9 +192,21 @@ async def generate_one(*_args, **_kwargs): asyncio.run(asyncio.wait_for(run(), timeout=2)) -def test_pd_master_hides_capacity_finish_token_and_continues_next_segment(): +@pytest.mark.parametrize( + ("initial_timeout", "continuation_timeout", "expected_timeouts"), + [ + (10, 60, [10, 60]), + (90, 60, [90, 60]), + (-1, 60, [-1, 60]), + ], +) +def test_pd_master_hides_capacity_finish_token_and_continues_next_segment( + initial_timeout, continuation_timeout, expected_timeouts +): async def run(): manager = _manager() + manager.pd_node_resource_wait_timeout_seconds = initial_timeout + manager.pd_node_continuation_resource_wait_timeout_seconds = continuation_timeout manager.id_gen.generate_id.side_effect = [808, 816] manager.remove_req = AsyncMock() manager.abort = AsyncMock() @@ -201,10 +214,12 @@ async def run(): d_node = MagicMock() manager.select_p_d_node = AsyncMock(return_value=(p_node, d_node, PDSelectionExtraInfo())) segment_index = 0 + resource_wait_timeouts = [] async def wait_to_token_package(_p_node, _d_node, _start_time, prompt, params, *_args): nonlocal segment_index segment_index += 1 + resource_wait_timeouts.append(params.pd_node_resource_wait_timeout_seconds) if segment_index == 1: assert prompt == "prompt" assert params.max_new_tokens == 4 @@ -247,6 +262,7 @@ async def wait_to_token_package(_p_node, _d_node, _start_time, prompt, params, * results.append(result) assert segment_index == 2 + assert resource_wait_timeouts == expected_timeouts assert [result[1] for result in results] == ["visible", "continued"] assert all(result[3].status != FinishStatus.FINISHED_PD_DECODE_CAPACITY for result in results) assert results[-1][3].status == FinishStatus.FINISHED_STOP diff --git a/unit_tests/server/httpserver/test_pd_master_cached_tokens.py b/unit_tests/server/httpserver/test_pd_master_cached_tokens.py index 666900a38..a3c95b270 100644 --- a/unit_tests/server/httpserver/test_pd_master_cached_tokens.py +++ b/unit_tests/server/httpserver/test_pd_master_cached_tokens.py @@ -19,6 +19,7 @@ def _make_manager(monkeypatch): mgr.args = SimpleNamespace(disable_pd_master_decode_capacity_limit=True) mgr.enable_pd_node_self_request_limit = True mgr.pd_node_resource_wait_timeout_seconds = -1 + mgr.pd_node_continuation_resource_wait_timeout_seconds = 60 mgr.pd_node_busy_retry_timeout_seconds = 120 mgr.pd_cache_high_priority_max_age_seconds = 60 mgr.pd_cache_high_priority_min_prompt_tokens = 8192 diff --git a/unit_tests/utils/test_envs_utils.py b/unit_tests/utils/test_envs_utils.py index 0bdbb61df..0cac7caa9 100644 --- a/unit_tests/utils/test_envs_utils.py +++ b/unit_tests/utils/test_envs_utils.py @@ -2,6 +2,7 @@ get_pd_cache_high_priority_max_age_seconds, get_pd_cache_high_priority_min_prompt_tokens, get_pd_node_busy_retry_timeout_seconds, + get_pd_node_continuation_resource_wait_timeout_seconds, get_pd_node_resource_wait_timeout_seconds, ) @@ -60,6 +61,24 @@ def test_pd_node_resource_wait_timeout_reads_environment_variable(monkeypatch): get_pd_node_resource_wait_timeout_seconds.cache_clear() +def test_pd_node_continuation_resource_wait_timeout_defaults_to_60_seconds(monkeypatch): + monkeypatch.delenv("LIGHTLLM_PD_NODE_CONTINUATION_RESOURCE_WAIT_TIMEOUT_SECONDS", raising=False) + get_pd_node_continuation_resource_wait_timeout_seconds.cache_clear() + + assert get_pd_node_continuation_resource_wait_timeout_seconds() == 60 + + get_pd_node_continuation_resource_wait_timeout_seconds.cache_clear() + + +def test_pd_node_continuation_resource_wait_timeout_reads_environment_variable(monkeypatch): + monkeypatch.setenv("LIGHTLLM_PD_NODE_CONTINUATION_RESOURCE_WAIT_TIMEOUT_SECONDS", "90") + get_pd_node_continuation_resource_wait_timeout_seconds.cache_clear() + + assert get_pd_node_continuation_resource_wait_timeout_seconds() == 90 + + get_pd_node_continuation_resource_wait_timeout_seconds.cache_clear() + + def test_pd_node_busy_retry_timeout_defaults_to_120_seconds(monkeypatch): monkeypatch.delenv("LIGHTLLM_PD_NODE_BUSY_RETRY_TIMEOUT_SECONDS", raising=False) get_pd_node_busy_retry_timeout_seconds.cache_clear() From bcfdddc006b93634a637f783cb89cd6e66468fe1 Mon Sep 17 00:00:00 2001 From: wangzaijun Date: Fri, 4 Sep 2026 06:48:55 +0000 Subject: [PATCH 08/12] refactor(pd): move TP slave timeout guard to manager --- lightllm/server/httpserver/manager.py | 14 +++++++------- .../test_pd_selector/test_pd_node_request_limit.py | 1 - .../httpserver/test_running_request_lifecycle.py | 11 +++++------ 3 files changed, 12 insertions(+), 14 deletions(-) diff --git a/lightllm/server/httpserver/manager.py b/lightllm/server/httpserver/manager.py index 370eea628..692275ae8 100644 --- a/lightllm/server/httpserver/manager.py +++ b/lightllm/server/httpserver/manager.py @@ -36,7 +36,7 @@ from .manager_ext import HttpRlManagerHelper from lightllm.utils.statics_utils import MovingAverage from lightllm.utils.config_utils import get_vocab_size -from lightllm.utils.envs_utils import get_env_start_args, get_unique_server_name +from lightllm.utils.envs_utils import get_unique_server_name from lightllm.utils.shm_port_args import get_shm_port_args from lightllm.utils.error_utils import ( ClientDisconnected, @@ -789,7 +789,12 @@ async def _wait_to_token_package( except asyncio.TimeoutError: pass - if is_first_token and req_status.has_timed_out_waiting_for_inference(): + # 多机 TP slave 只跟随 master 执行，不能独立判定超时并中止请求。 + if ( + is_first_token + and not self.is_multinode_tp_slave + and req_status.has_timed_out_waiting_for_inference() + ): resource_wait_timeout_seconds = sampling_params.pd_node_resource_wait_timeout_seconds raise ServerBusyError( f"PD {self.args.run_mode} node is busy: request did not enter inference " @@ -1095,8 +1100,6 @@ class ReqStatus: def __init__(self, group_request_id, multimodal_params, req_objs: List[Req], start_time) -> None: self.lock = asyncio.Lock() self.event = asyncio.Event() - args = get_env_start_args() - self.is_multinode_tp_slave = args.dp == 1 and args.nnodes > 1 and args.node_rank > 0 self.group_req_objs = GroupReqObjs( group_req_id=group_request_id, multimodal_params=multimodal_params, @@ -1107,9 +1110,6 @@ def __init__(self, group_request_id, multimodal_params, req_objs: List[Req], sta def has_timed_out_waiting_for_inference(self) -> bool: """按 PD Master 下发的资源等待上限判断请求是否在 Router 中超时。""" - # 多机 TP slave 只跟随 master 执行，不能独立判定超时并中止请求。 - if self.is_multinode_tp_slave: - return False current_time = time.monotonic() reqs = self.group_req_objs.shm_req_objs # 组内任一请求已经进入新 batch，说明整个请求组已经开始执行，不能再按 Router 等待超时清理。 diff --git a/test/test_pd_selector/test_pd_node_request_limit.py b/test/test_pd_selector/test_pd_node_request_limit.py index a591bdc13..3cf20573a 100644 --- a/test/test_pd_selector/test_pd_node_request_limit.py +++ b/test/test_pd_selector/test_pd_node_request_limit.py @@ -142,7 +142,6 @@ def test_router_wait_uses_master_timeout_independently_of_priority( expected, ): req_status = ReqStatus.__new__(ReqStatus) - req_status.is_multinode_tp_slave = False req_status.group_req_objs = SimpleNamespace( shm_req_objs=[ SimpleNamespace( diff --git a/unit_tests/server/httpserver/test_running_request_lifecycle.py b/unit_tests/server/httpserver/test_running_request_lifecycle.py index 6f4ae60e1..45671e53c 100644 --- a/unit_tests/server/httpserver/test_running_request_lifecycle.py +++ b/unit_tests/server/httpserver/test_running_request_lifecycle.py @@ -62,10 +62,8 @@ def _multimodal_params(): return SimpleNamespace(audios=[], images=[], verify_and_preload=AsyncMock()) -def _req_status(reqs, *, dp=1, nnodes=1, node_rank=0): - start_args = SimpleNamespace(dp=dp, nnodes=nnodes, node_rank=node_rank) - with patch("lightllm.server.httpserver.manager.get_env_start_args", return_value=start_args): - return ReqStatus(123, None, reqs, 0) +def _req_status(reqs): + return ReqStatus(123, None, reqs, 0) async def _drain_generate(manager, sampling_params, multimodal_params, websocket=None, pd_event=None): @@ -278,12 +276,13 @@ async def run(): infer_start_time=0.0, sample_params=SimpleNamespace(pd_node_resource_wait_timeout_seconds=0), ) - req_status = _req_status([req], nnodes=2, node_rank=1) + req_status = _req_status([req]) req_status.event.set() request = SimpleNamespace(is_disconnected=AsyncMock(return_value=True)) with patch("lightllm.server.httpserver.manager.time.monotonic", return_value=2): - assert req_status.has_timed_out_waiting_for_inference() is False + # ReqStatus 只判断时间条件；是否允许 slave 执行超时策略由 manager 在调用处决定。 + assert req_status.has_timed_out_waiting_for_inference() is True output_generator = manager._wait_to_token_package( start_time=0, prompt_ids=[], From 099c81c6b0c18930bebe5f9b4736bd8f2e133ec7 Mon Sep 17 00:00:00 2001 From: wangzaijun Date: Fri, 4 Sep 2026 07:21:46 +0000 Subject: [PATCH 09/12] docs(pd): clarify cache priority scheduling tradeoffs --- docs/CN/source/tutorial/api_server_args.rst | 9 +++++++++ docs/EN/source/tutorial/api_server_args.rst | 11 +++++++++++ lightllm/server/api_cli.py | 10 ++++++++-- 3 files changed, 28 insertions(+), 2 deletions(-) diff --git a/docs/CN/source/tutorial/api_server_args.rst b/docs/CN/source/tutorial/api_server_args.rst index 586ace215..e4fe2a92a 100644 --- a/docs/CN/source/tutorial/api_server_args.rst +++ b/docs/CN/source/tutorial/api_server_args.rst @@ -129,6 +129,15 @@ PD 分离模式参数 该参数不影响 PD Decode 容量不足后的分段续跑请求；续跑请求仍保持高优先级。默认不启用， 即默认允许新鲜高 cache 命中请求提升优先级。 + 建议只在 PD Master 上配置该参数。当单个 P 节点的 GPU cache、CPU cache 和 disk cache 总容量相对于 + 请求工作集较小时，高负载下后到的请求容易快速淘汰已有 cache，使原本可以命中 cache 的请求退化为 + 重新执行 Prefill，进而显著降低 Prefill 效率。此时建议保留默认的高优先级策略，让预计 cache 命中率高的 + 请求提前进入推理，尽量在 cache 被淘汰前完成复用。 + + 该策略会改变排队顺序，因此普通请求（未达到 cache 命中率、cache 年龄或最小 prompt token 数门槛的请求） + 的首字延迟可能升高。如果 P 节点 cache 容量充足、系统负载较低，或者业务更重视调度公平性和普通请求的 + 首字延迟，可以设置 ``--disable_pd_cache_high_priority`` 关闭该策略。 + .. option:: --config_server_host 配置服务器模式下的主机地址 diff --git a/docs/EN/source/tutorial/api_server_args.rst b/docs/EN/source/tutorial/api_server_args.rst index b5c782330..1d4cc1285 100644 --- a/docs/EN/source/tutorial/api_server_args.rst +++ b/docs/EN/source/tutorial/api_server_args.rst @@ -138,6 +138,17 @@ PD disaggregation Mode Parameters Decode capacity exhaustion; continuation requests remain high priority. Disabled by default, so eligible requests are promoted unless this option is set. + Configure this option only on PD Master. When a Prefill node's combined GPU, CPU, and disk cache capacity is small + relative to its request working set, later requests can quickly evict reusable cache entries under high load. + Requests that could otherwise hit the cache must then repeat Prefill computation, which can significantly reduce + Prefill efficiency. In this situation, keep the default high-priority policy enabled so requests with a high + estimated cache hit rate can run earlier and reuse their cache entries before eviction. + + This policy changes queue ordering and may increase time to first token (TTFT) for ordinary requests that do not + meet the cache-hit-rate, cache-age, or minimum-prompt-token thresholds. Consider setting + ``--disable_pd_cache_high_priority`` when Prefill cache capacity is sufficient and cache churn is low, or when + scheduling fairness and ordinary-request TTFT are more important than preserving cache-hit efficiency. + .. option:: --config_server_host Host address in configuration server mode diff --git a/lightllm/server/api_cli.py b/lightllm/server/api_cli.py index c79f59c18..e6c077dbe 100644 --- a/lightllm/server/api_cli.py +++ b/lightllm/server/api_cli.py @@ -89,8 +89,14 @@ def add_cli_args(parser: argparse.ArgumentParser) -> argparse.ArgumentParser: "--disable_pd_cache_high_priority", action="store_true", help=( - "Disable promoting first-segment PD Master requests with a fresh high cache-hit estimate. " - "Segmented continuation requests remain high priority. Default: disabled." + "Disable PD Master's high-priority scheduling for first-segment requests with a fresh, high " + "cache-hit estimate. Keep this policy enabled when a Prefill node's combined GPU, CPU, and disk " + "cache is small relative to its workload: under high load, ordinary scheduling can evict reusable " + "cache entries before they are consumed and significantly reduce Prefill efficiency. The policy " + "lets eligible cache-hit requests run earlier, but may increase TTFT for ordinary requests. " + "Consider disabling it only when scheduling fairness or ordinary-request latency is more important, " + "or when cache capacity is sufficient and cache churn is low. Segmented continuation requests remain " + "high priority. Configure this option only on PD Master. The policy is enabled by default." ), ) parser.add_argument( From b75707a701b32134b1f4d7e7671a607407de77af Mon Sep 17 00:00:00 2001 From: wangzaijun Date: Fri, 4 Sep 2026 07:22:43 +0000 Subject: [PATCH 10/12] tune(pd): extend cache priority max age --- lightllm/utils/envs_utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lightllm/utils/envs_utils.py b/lightllm/utils/envs_utils.py index 4405e822a..592e04fa4 100644 --- a/lightllm/utils/envs_utils.py +++ b/lightllm/utils/envs_utils.py @@ -327,7 +327,7 @@ def get_pd_node_busy_retry_timeout_seconds() -> int: @lru_cache(maxsize=None) def get_pd_cache_high_priority_max_age_seconds() -> int: """cache 命中请求提升为 PD 高优先级时允许的最大缓存年龄，单位为秒。""" - return max(0, int(os.getenv("LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MAX_AGE_SECONDS", 36))) + return max(0, int(os.getenv("LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MAX_AGE_SECONDS", 120))) @lru_cache(maxsize=None) From 120b42d60831bcf282a5c63758c09c3b955846ff Mon Sep 17 00:00:00 2001 From: wangzaijun Date: Fri, 4 Sep 2026 07:43:33 +0000 Subject: [PATCH 11/12] fix(pd): stop busy retries after client disconnect --- .../httpserver_for_pd_master/manager.py | 9 ++++ .../test_pd_master_multi_choice.py | 42 ++++++++++++++++++- 2 files changed, 49 insertions(+), 2 deletions(-) diff --git a/lightllm/server/httpserver_for_pd_master/manager.py b/lightllm/server/httpserver_for_pd_master/manager.py index ba569f48b..96f0d7203 100644 --- a/lightllm/server/httpserver_for_pd_master/manager.py +++ b/lightllm/server/httpserver_for_pd_master/manager.py @@ -259,6 +259,15 @@ async def _generate_one( elapsed_seconds = time.monotonic() - retry_start_time if has_yielded_result or elapsed_seconds >= self.pd_node_busy_retry_timeout_seconds: raise + + # 发起下一次尝试前检查客户端连接，避免为已断开的请求继续占用 P/D 节点资源。 + if await request.is_disconnected(): + disconnect_reason = "_generate_one busy retry check network disconnected" + logger.warning(f"group_request_id: {origin_request_id} {disconnect_reason}") + raise ClientDisconnected( + group_request_id=origin_request_id, + reason=disconnect_reason, + ) logger.warning( f"group_request_id: {origin_request_id} PD node is busy, retrying with another node; " f"elapsed: {elapsed_seconds:.3f}s, retry timeout: " diff --git a/test/test_pd_selector/test_pd_master_multi_choice.py b/test/test_pd_selector/test_pd_master_multi_choice.py index 8d48f5d5c..38f43232f 100644 --- a/test/test_pd_selector/test_pd_master_multi_choice.py +++ b/test/test_pd_selector/test_pd_master_multi_choice.py @@ -10,7 +10,7 @@ from lightllm.server.httpserver.manager import HttpServerManager from lightllm.server.httpserver_for_pd_master.manager import HttpServerManagerForPDMaster from lightllm.server.httpserver_for_pd_master.pd_selector import PDSelectionExtraInfo -from lightllm.utils.error_utils import ServerBusyError +from lightllm.utils.error_utils import ClientDisconnected, ServerBusyError def _manager() -> HttpServerManagerForPDMaster: @@ -665,13 +665,15 @@ async def generate_one_attempt(*_args): yield 800, "ok", {}, FinishStatus(FinishStatus.FINISHED_STOP) manager._generate_one_attempt = generate_one_attempt + request = MagicMock() + request.is_disconnected = AsyncMock(return_value=False) results = [ result async for result in manager._generate_one( "prompt", SamplingParams(), MagicMock(), - MagicMock(), + request, 0, 800, 0, @@ -679,11 +681,47 @@ async def generate_one_attempt(*_args): ] assert attempt_count == 2 + request.is_disconnected.assert_awaited_once() assert [result[1] for result in results] == ["ok"] asyncio.run(asyncio.wait_for(run(), timeout=2)) +def test_pd_master_stops_busy_retry_when_client_disconnects(): + async def run(): + manager = _manager() + attempt_count = 0 + + async def generate_one_attempt(*_args): + nonlocal attempt_count + attempt_count += 1 + raise ServerBusyError("node is busy") + yield + + manager._generate_one_attempt = generate_one_attempt + request = MagicMock() + request.is_disconnected = AsyncMock(side_effect=[False, True]) + + with pytest.raises(ClientDisconnected) as exc_info: + async for _ in manager._generate_one( + "prompt", + SamplingParams(), + MagicMock(), + request, + 0, + 800, + 0, + ): + pass + + assert attempt_count == 2 + assert request.is_disconnected.await_count == 2 + assert exc_info.value.group_request_id == 800 + assert exc_info.value.reason == "_generate_one busy retry check network disconnected" + + asyncio.run(asyncio.wait_for(run(), timeout=2)) + + def test_pd_master_does_not_retry_busy_error_when_self_limit_is_disabled(): async def run(): manager = _manager() From ac2791d86efb227b175e6a3ab315af2e1fe72a46 Mon Sep 17 00:00:00 2001 From: wangzaijun Date: Fri, 4 Sep 2026 07:45:43 +0000 Subject: [PATCH 12/12] fix --- lightllm/server/httpserver/manager.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/lightllm/server/httpserver/manager.py b/lightllm/server/httpserver/manager.py index 692275ae8..cea3cb6fc 100644 --- a/lightllm/server/httpserver/manager.py +++ b/lightllm/server/httpserver/manager.py @@ -790,11 +790,7 @@ async def _wait_to_token_package( pass # 多机 TP slave 只跟随 master 执行，不能独立判定超时并中止请求。 - if ( - is_first_token - and not self.is_multinode_tp_slave - and req_status.has_timed_out_waiting_for_inference() - ): + if is_first_token and not self.is_multinode_tp_slave and req_status.has_timed_out_waiting_for_inference(): resource_wait_timeout_seconds = sampling_params.pd_node_resource_wait_timeout_seconds raise ServerBusyError( f"PD {self.args.run_mode} node is busy: request did not enter inference "