diff --git a/docs/CN/source/tutorial/api_server_args.rst b/docs/CN/source/tutorial/api_server_args.rst index 746827b10..e4fe2a92a 100644 --- a/docs/CN/source/tutorial/api_server_args.rst +++ b/docs/CN/source/tutorial/api_server_args.rst @@ -92,26 +92,52 @@ PD 分离模式参数 推理进度健康检查：当仍有在途请求，且整个 PD Master 连续 ``HEALTH_TIMEOUT`` 秒 没有任何请求成功返回 token 时，接口将返回 HTTP 503。 -.. 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 秒。本地请求限流默认关闭。 +.. option:: --disable_pd_node_self_request_limit + + 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 表示永久等待。 + ``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 节点并重试；最长探测周期由 + ``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`` 控制，默认 36 秒。cache 命中提权还要求输入 + token 数达到 ``LIGHTLLM_PD_CACHE_HIGH_PRIORITY_MIN_PROMPT_TOKENS`` 配置的门槛（默认 4096），避免短请求仅因 + cache 命中率高而提升优先级。 + + 启动示例： + + .. 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 ... .. option:: --disable_pd_cache_high_priority - 禁止 PD Master 将预计输入 cache 命中率高且命中记录仍然新鲜的首段请求提升为高优先级。 + 禁止 PD Master 将输入足够长、预计输入 cache 命中率高且命中记录仍然新鲜的首段请求提升为高优先级。 该参数不影响 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 81b3bb752..1d4cc1285 100644 --- a/docs/EN/source/tutorial/api_server_args.rst +++ b/docs/EN/source/tutorial/api_server_args.rst @@ -95,29 +95,60 @@ 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 - - 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 - 16 seconds. Local request limiting is disabled by default. +.. option:: --disable_pd_node_self_request_limit + + 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. 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. 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 + 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 + 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. + + Startup example: + + .. 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 ... .. 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. + 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 bfb54b4ba..e6c077dbe 100644 --- a/lightllm/server/api_cli.py +++ b/lightllm/server/api_cli.py @@ -77,19 +77,26 @@ 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 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." + "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( "--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( 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/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/manager.py b/lightllm/server/httpserver/manager.py index 9e7532df4..cea3cb6fc 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_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,19 @@ 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, - ) + # 多机 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 try: @@ -803,14 +789,12 @@ 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) - ): + # 多机 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 " - 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(): @@ -1120,20 +1104,16 @@ 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 中超时。""" 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 35a0ecfa8..96f0d7203 100644 --- a/lightllm/server/httpserver_for_pd_master/manager.py +++ b/lightllm/server/httpserver_for_pd_master/manager.py @@ -25,7 +25,10 @@ from lightllm.utils.error_utils import ClientDisconnected, ServerBusyError from lightllm.utils.envs_utils import ( get_pd_cache_high_priority_max_age_seconds, - get_pd_high_priority_request_timeout_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 from .pd_selector import PDSelectionExtraInfo, create_selector @@ -51,10 +54,15 @@ 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 生效；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() 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 +206,7 @@ async def _generate( request, start_time, origin_group_request_id + choice_index, + input_token_num, ) ) @@ -219,7 +228,68 @@ async def _generate_one( request: Request, start_time: float, 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 + + # 发起下一次尝试前检查客户端连接，避免为已断开的请求继续占用 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: " + 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 @@ -243,6 +313,13 @@ 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 ) history_gen_token_strs = [] @@ -261,16 +338,18 @@ 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 节点仅在自身开启 - # 本地限流时使用该值，未开启限流时仍保持无限等待。 - if sampling_params.pd_high_priority_request: - sampling_params.pd_high_priority_request_time_out_seconds = ( - self.pd_high_priority_request_time_out_seconds - ) + # 仅在 Master 开启限流时下发资源等待超时。续跑分段已经产生了部分结果， + # 使用独立配置的等待时间，提高请求最终完成的成功率。 + if self.enable_pd_node_self_request_limit: + 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 0c4f28475..592e04fa4 100644 --- a/lightllm/utils/envs_utils.py +++ b/lightllm/utils/envs_utils.py @@ -307,27 +307,33 @@ 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)) +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) -def get_pd_node_router_wait_timeout_seconds() -> int: - """请求进入 Router 后等待进入推理系统的最长时间，单位为秒。""" - return int(os.getenv("LIGHTLLM_PD_NODE_ROUTER_WAIT_TIMEOUT_SECONDS", 20)) +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_high_priority_request_timeout_seconds() -> int: - """PD Master 为高优先级请求设置的等待时间下限，单位为秒。""" - return int(os.getenv("LIGHTLLM_PD_HIGH_PRIORITY_REQUEST_TIMEOUT_SECONDS", 60)) +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", 120))) + + +@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) 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..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,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 ClientDisconnected, ServerBusyError def _manager() -> HttpServerManagerForPDMaster: @@ -20,8 +21,12 @@ 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.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 manager.disable_pd_cache_high_priority = False return manager @@ -125,7 +130,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", @@ -185,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() @@ -195,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 @@ -236,10 +257,12 @@ async def wait_to_token_package(_p_node, _d_node, _start_time, prompt, params, * MagicMock(), 0, 800, + 0, ): 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 @@ -339,6 +362,7 @@ async def failing_wait_to_token_package(*_args, **_kwargs): MagicMock(), 0, 800, + 0, ): pass @@ -412,6 +436,7 @@ async def wait_to_token_package( MagicMock(), 0, 800, + 0, ): results.append(result) @@ -479,6 +504,7 @@ async def wait_to_token_package(_p_node, _d_node, _start_time, _prompt, sampling MagicMock(), 0, 800, + 0, ): pass @@ -492,21 +518,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 +584,7 @@ async def wait_to_token_package(_p_node, _d_node, _start_time, _prompt, sampling MagicMock(), 0, 800, + input_token_num, ): pass @@ -563,10 +593,12 @@ 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(): +@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.pd_high_priority_request_time_out_seconds = 90 + 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() manager.abort = AsyncMock() @@ -579,16 +611,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", @@ -599,6 +633,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", @@ -607,10 +642,175 @@ async def wait_to_token_package(_p_node, _d_node, _start_time, _prompt, sampling MagicMock(), 0, 800, + 8192, ): pass - assert captured_timeout_seconds == [90] + assert captured_request_settings == [(False, expected_timeout)] + + 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 + request = MagicMock() + request.is_disconnected = AsyncMock(return_value=False) + results = [ + result + async for result in manager._generate_one( + "prompt", + SamplingParams(), + MagicMock(), + request, + 0, + 800, + 0, + ) + ] + + 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() + 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)) @@ -644,6 +844,7 @@ async def wait_to_token_package(*_args, **_kwargs): MagicMock(), 0, 800, + 0, ) assert (await generator.__anext__())[1] == "first" 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..3cf20573a 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,67 +72,73 @@ 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) @@ -143,15 +148,15 @@ def test_high_priority_router_wait_uses_master_timeout( 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 748f62466..a3c95b270 100644 --- a/unit_tests/server/httpserver/test_pd_master_cached_tokens.py +++ b/unit_tests/server/httpserver/test_pd_master_cached_tokens.py @@ -17,8 +17,12 @@ 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.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 mgr.disable_pd_cache_high_priority = False mgr.running_request_count = 0 counter = [0] diff --git a/unit_tests/server/httpserver/test_running_request_lifecycle.py b/unit_tests/server/httpserver/test_running_request_lifecycle.py index bf154826e..45671e53c 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,10 @@ def _multimodal_params(): return SimpleNamespace(audios=[], images=[], verify_and_preload=AsyncMock()) +def _req_status(reqs): + 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 +203,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 +237,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 +267,43 @@ 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]) + req_status.event.set() + request = SimpleNamespace(is_disconnected=AsyncMock(return_value=True)) + + with patch("lightllm.server.httpserver.manager.time.monotonic", return_value=2): + # 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=[], + 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 +311,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..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_defaults_to_disabled_and_can_be_enabled(): +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 f3c7d96dd..0cac7caa9 100644 --- a/unit_tests/utils/test_envs_utils.py +++ b/unit_tests/utils/test_envs_utils.py @@ -1,15 +1,17 @@ from lightllm.utils.envs_utils import ( get_pd_cache_high_priority_max_age_seconds, - get_pd_node_router_wait_timeout_seconds, - get_pd_node_shm_req_alloc_timeout_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, ) -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() @@ -23,37 +25,73 @@ 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_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_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_node_shm_req_alloc_timeout_seconds() == 20 + assert get_pd_cache_high_priority_min_prompt_tokens() == 4096 - get_pd_node_shm_req_alloc_timeout_seconds.cache_clear() + get_pd_cache_high_priority_min_prompt_tokens.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_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_node_shm_req_alloc_timeout_seconds() == 30 + assert get_pd_cache_high_priority_min_prompt_tokens() == 2048 - get_pd_node_shm_req_alloc_timeout_seconds.cache_clear() + get_pd_cache_high_priority_min_prompt_tokens.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() +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_router_wait_timeout_seconds() == 20 + assert get_pd_node_resource_wait_timeout_seconds() == 10 - get_pd_node_router_wait_timeout_seconds.cache_clear() + get_pd_node_resource_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() +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_router_wait_timeout_seconds() == 45 + assert get_pd_node_resource_wait_timeout_seconds() == 30 - get_pd_node_router_wait_timeout_seconds.cache_clear() + 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() + + 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()