调度类参数：vllm-ascend 怎么"决定每一步算什么"

参数组 1/8 · 调度 代码基线 a71b766ce（2026-09-23）
讲解范围：vllm_ascend/ascend_config.py 中 scheduler_config 子配置 （enable_balance_scheduling / recompute_scheduler_enable / short_request_first_config / profiling_chunk_config / batch_job_sched_config / dyntra_lb_config）+ 顶层 enable_reduce_sample、mega_moe_max_tokens， 以及它们的 8 个实现文件（BalanceScheduler、ShortRequestFirstRequestQueue、ProfilingChunkScheduler、 RecomputeScheduler、BatchJobAwareScheduler、DyntraLBDPEngineCoreProc、sampler 分支、mega_moe 算子加载）。

① 这段代码是干嘛的

② 它在系统中的位置

③ 概念爬坡：从"发个请求"走到"这 8 个参数"

④ 主干流程：一个参数从"命令行"走到"生效"

vllm_ascend/ascend_config.py · AscendConfig 默认值示例 · L391-423            "scheduler_config": {                       # ← 6 个调度子配置都住这里
                "enable_balance_scheduling": false,     # ⚡多实例均衡调度（→5.1）
                "recompute_scheduler_enable": false,    # ⚡重计算调度（→5.5）
                "short_request_first_config": {
                    "enabled": false,
                    "threshold": 256,                   # ↳ prompt ≤256 token 算"短请求"
                    "long_max_wait_ms": 0.0             # ↳ 长请求最多被晾这么久（防饿死）
                },
                "profiling_chunk_config": {
                    "enabled": false,
                    "smooth_factor": 1.0,
                    "min_chunk": 4096,
                    "need_timing": null,
                    "max_fit_chunk": 30
                },
                "batch_job_sched_config": {
                    "enabled": false,
                    "max_jobs": 20,
                    "reserve_margin_blocks": 2,
                    "reserve_max_blocks": 8,
                    "low_available_tokens_threshold": 4096,
                    "short_decode_token_threshold": 32
                },
                "dyntra_lb_config": {
                    "enabled": false,
                    ...
                    "bubble_threshold": 5.0,            # ↳ 气泡超过均值 5 倍才触发迁移
                    "long_req_block_threshold": 700,
                },
            },

⑤ 关键节点逐个展开：8 个参数，8 个"治病"的故事

5.1 enable_balance_scheduling —— 多实例"各调各的"，先让大家互相看见

vllm_ascend/patch/platform/patch_balance_schedule.py · BalanceScheduler.balance_gather · L146-162    def balance_gather(self):
        """All-gather per-rank running counts into ``self.balance_queue``.

        ... This MUST be invoked on every rank every non-idle iteration, including
        dummy-batch iterations where a drained rank never enters ``schedule()`` --
        ``all_gather`` is a collective, so any rank skipping it deadlocks
        the busy ranks.  ← ⚡集体操作纪律：缺席即死锁
        """
        if not self._balance_enabled or self.dp_group is None:
            return
        running_tensor = torch.tensor([len(self.running)], dtype=torch.int, device="cpu")
        dist.all_gather(self.balance_queue, running_tensor, group=self.dp_group)  # →NPU/HCCL 集合通信

5.2 dyntra_lb_config —— 均衡的进阶版：按 KV 块算"气泡"，把请求搬去闲卡

vllm_ascend/patch/platform/patch_dyntra_lb_core.py · DyntraLBDPEngineCoreProc._balance_load · L121-136class DyntraLBDPEngineCoreProc(DPEngineCoreProc):
    @staticmethod
    def _balance_load(
        requests_by_rank: Sequence[tuple[Sequence[int], int]],  # ↳ 各卡（请求块列表, 总块数）
        dev_num: int,
        max_num_seqs: int | None = None,
        max_iters: int = 1000,
        threshold: float = 5,          # ↳ 即 bubble_threshold，气泡倍数阈值
    ) -> list[_Modification]:
        modifications: list[_Modification] = [{"out_blk": [], "in_blk": [], "freeze": False} for _ in range(dev_num)]

        def calc_bubble(cards_info: Sequence[_CardData]) -> float:  # ⚡气泡 = 块数不均的程度
            tot_blks = [c["tot_blk"] for c in cards_info]
            max_blk = max(tot_blks)
            avg_blk = sum(tot_blks) / dev_num

5.3 short_request_first_config —— 三个队列，让短请求插队但不让长请求饿死

vllm_ascend/core/short_request_first_scheduler.py · ShortRequestFirstRequestQueue · L148-183（节选）    def _classify_queue(self, request: Request) -> RequestQueue:
        if self.immediate_predicate is not None and self.immediate_predicate(request):
            return self._immediate_queue
        if request.num_prompt_tokens <= self.threshold:  # ⚡threshold=256：短请求判据
            return self._short_queue
        return self._long_queue

    def _long_head_wait_ms(self) -> float | None:
        if self.long_max_wait_ms <= 0 or not self._long_queue:
            return None
        head = self._long_queue.peek_request()
        enqueued_at = self._long_enqueue_at.get(head.request_id)
        if enqueued_at is None:
            return None
        return (time.monotonic() - enqueued_at) * 1000.0  # ↳ 长请求队头已等多久

    def _select_schedulable_queue(self) -> RequestQueue | None:
        if self._immediate_queue:
            return self._immediate_queue            # ⚡1. 插单最优先
        if self._long_queue:
            wait_ms = self._long_head_wait_ms()
            if wait_ms is not None and wait_ms >= self.long_max_wait_ms and self._short_queue:
                return self._long_queue             # ⚡2. 防饿死：长请求等满时限，插队
        if self._short_queue:
            return self._short_queue                # ⚡3. 默认短先行
        if self._long_queue:
            return self._long_queue
        return None

5.4 profiling_chunk_config —— chunk 切多大？用真实计时说话

vllm_ascend/core/scheduler_profiling_chunk.py · ProfilingChunkScheduler.run_profiling_chunk_init · L127-135    def run_profiling_chunk_init(self, model_executor) -> None:
        """Profile prefill latency using real model forward passes.

        Called by EngineCore after model_executor is ready.  Collects latency
        samples at different chunk sizes and fits the quadratic model.
        """                                    # ⚡不是估算，是真跑模型计时
        if self._profiling_initialized:
            return
        self._profiling_initialized = True
        if model_executor is None:
            logger.warning("[ProfilingChunk] No model_executor provided, skipping profiling")
            return

5.5 recompute_scheduler_enable —— 抢占时，先想办法卸载，卸不动才重算

vllm_ascend/core/recompute_scheduler.py · RecomputeScheduler · L85-103class RecomputeScheduler(Scheduler):
    """Use vLLM scheduling with best-effort decode-side preemption offload.

    This keeps a local copy of vLLM's schedule() only to pad the first decode
    request for stable Ascend speculative-decode graph shapes. Preempted KV is
    offloaded when possible; otherwise the request is sent back to P to redo
    prefill.                      ← ⚡卸载优先，重算兜底（名字的由来）
    """

    prefill_capacity_bound: bool

    def _apply_load_balance_modifications(self) -> None:
        """Hook for DyntraLB to apply cross-rank scheduling decisions."""  # ⚡5.2 的迁移在此落地

    def _can_admit_waiting_request(self, request: Request) -> bool:
        """Hook for DyntraLB to filter waiting-request admission."""
        return True

5.6 batch_job_sched_config —— 在线流量与批量作业混部，给离线任务"记账"

vllm_ascend/core/batch_job_aware_scheduler.py · JobDecodeEstimator · L136-148    def predict(self, job_name: str) -> int:
        """Return the predicted decode length for *job*."""
        job_stat = self._job_stats.get(job_name)
        if job_stat is None:
            return self._COLD_START_DEFAULT_DECODE   # ⚡没见过的新作业：冷启动默认值
        return int(job_stat.ewma_average)            # ↳ 历史 EWMA 均值

    def observe(self, job_name: str, decode_len: int) -> None:
        """Record one decode‑length observation for *job*.

        This method updates the job statistics using the EWMA (Exponentially
        Weighted Moving Average) formula for online learning.
        """                                          # ⚡每结束一个请求就回填，越学越准

5.7 mega_moe_max_tokens —— MoE 融合算子的"缓冲区容量申报"

vllm_ascend/ops/fused_moe/moe_comm_method.py · MoE 初始化与缓冲分配 · L270-275、L334-343        self.enable_fused_mc2 = get_ascend_config().enable_fused_mc2
        if self.enable_fused_mc2 == 1 and is_mega_moe_supported():  # ⚡前置：融合 MC2 + 硬件支持
            self.mega_moe_symm_buffer = None
            self.get_symm_buffer_for_mega_moe, self.mega_moe = moe_utils.load_cann_mega_moe_ops()  # →NPU

            # P nodes and PD-mixed nodes use the configured value. ...
            max_recv_token_num = get_ascend_config().mega_moe_max_tokens  # ⚡申报值
            # absolute_safe_max_recv_token_num is the max value required by mega moe api
            if max_recv_token_num > absolute_safe_max_recv_token_num:
                max_recv_token_num = absolute_safe_max_recv_token_num   # ↳ 超上限被夹住
            logger.warning_once(
                "MegaMoe symm buffer: max_recv_token_num is set from "
                "mega_moe_max_tokens=%d ... If the actual per-rank received token count after "
                "dispatch exceeds this value, precision degradation will ...")  # ⚡超发=精度风险

5.8 enable_reduce_sample —— 采样阶段的"分片去重"

vllm_ascend/sample/sampler.py · 采样分支 · L169-177    if get_ascend_config().enable_reduce_sample:  # ⚡实验特性开关
        tp_group = get_tp_group()
        B, V_local = logits.shape     # ↳ 本卡只持有词表分片 V_local
        rank = tp_group.rank_in_group

        if top_k is None or (p is None and k is None):
            k_for_topk = V_local      # ↳ 不限 k 时，分片全量即是上限
        else:
            k_for_topk = min(top_k, V_local)  # ⚡去重：k 不会超过本分片宽度

⑥ 关键数据结构：调度器是一棵"可组合"的类树

⑦ 边界与异常：这些参数"不能一起吃"

vllm_ascend/ascend_config.py · derive_and_validate · L807-820        if self.enable_reduce_sample:
            logger.warning_once(
                "enable_reduce_sample is an experimental feature. Use with caution.")  # ⚡实验特性
            if (...细粒度 TP 相关字段...):
                raise ValueError(
                    "enable_reduce_sample is incompatible with "
                    ...                                     # ⚡红线一：与细粒度 TP 互斥
            if (...PD 分离场景...):
                ...raise ValueError(
                    "enable_reduce_sample is not supported on PD-disaggregated "
                    "scenarios. Please disable enable_reduce_sample.")  # ⚡红线二：PD 不支持

⑧ 小结：一张图带走

自检表