AI算力调度新方案：动态评分机制解析与工程实践

🚀 30+款热门AI模型一站整合，DeepSeek/GLM/Qwen 随心用，限时 5 折。 👉 点击领海量免费额度

在实际 AI 项目开发和模型训练中，算力调度是一个长期存在的核心痛点。无论是个人开发者使用单张显卡，还是大型团队管理异构的 GPU 集群，如何高效、公平、稳定地将计算任务分配给合适的硬件资源，并最大化资源利用率，始终是影响研发效率和成本的关键。传统的调度方式，如简单的 FIFO（先进先出）队列或基于静态规则的分配，在面对动态变化的模型训练任务、不同优先级的推理请求以及硬件故障时，往往显得力不从心，容易导致资源闲置、任务排队过长或关键任务被延迟。

“鲸挣恩又赢了”这个表述，通常指向一篇名为《Whose Turn Is It? A New AI Compute Scheduling Scheme》的学术论文（因其标题首字母缩写为“Whose Turn”而被戏称为“Whose Turn Is It?”，谐音“Whose Turn”听起来像“Whose Turn”，进而被中文社区趣称为“鲸挣恩又赢了”）。这篇论文提出了一种新颖的 AI 算力调度方案，旨在解决上述问题。其核心思想并非简单地“赢”了其他方案，而是引入了一种更精细、更适应动态环境的调度策略，考虑了任务特性、资源状态和系统目标之间的复杂平衡。

本文将从工程实践的角度，解析这种新型调度方案背后的核心概念、设计动机，并探讨其潜在的实现思路。我们将不局限于论文本身，而是结合常见的 AI 开发环境（如 Kubernetes with Kubeflow、Slurm 或简单的 Python 任务队列），分析如何将类似的调度思想落地到实际项目中。无论你是管理一个小型实验室的 GPU 服务器，还是需要优化云上 AI 任务的编排，理解这些调度原则都将帮助你构建更高效、更可控的计算环境。

1. 理解 AI 算力调度的核心挑战与常见方案

在深入新方案之前，必须先厘清我们面临的问题是什么，以及现有方案为何存在不足。AI 算力调度远不止是“把任务丢给空闲 GPU”那么简单。

1.1 调度系统的基本目标与冲突

一个理想的 AI 算力调度系统需要同时追求多个，有时甚至是相互冲突的目标：

1. 高吞吐量：在单位时间内完成尽可能多的任务（例如，训练更多的模型步数，处理更多的推理请求）。
2. 低延迟：确保高优先级或交互式任务（如模型服务、开发调试）能够快速获得资源并开始执行。
3. 高利用率：让昂贵的 GPU、CPU、内存等硬件资源尽可能处于工作状态，减少空闲时间。
4. 公平性：在不同用户、团队或项目之间公平地分配资源，避免个别任务或用户垄断资源。
5. 可预测性：任务等待时间、完成时间相对稳定，便于安排工作计划和预估成本。
6. 容错与弹性：能够处理硬件故障、任务失败，并支持任务的抢占、迁移和恢复。

这些目标就像是一个多目标优化问题，很难找到一个方案能在所有维度上都达到最优。传统方案往往侧重于其中一两个目标，而牺牲了其他方面。

1.2 常见调度方案及其局限性

下表对比了几种常见的调度策略及其典型问题：

AI 工作负载，特别是深度学习训练，有其特殊性：任务运行时间极长（小时/天级）、资源需求高且固定（需要特定数量的GPU和显存）、对中断敏感（上下文切换成本高）。这些特点使得上述许多通用调度策略在 AI 场景下效果不佳。

2. “Whose Turn” 方案的核心思想：一种动态、多目标的调度视角

“Whose Turn” 论文提出的方案，其核心在于摒弃单一的、静态的调度策略，转而采用一种动态评分机制。系统不再仅仅根据提交时间或静态优先级做决定，而是为每个等待中的任务实时计算一个“得分”，这个得分综合了多种因素，然后将资源分配给当前得分最高的任务。

2.1 动态评分函数的构成要素

评分函数是这套方案的大脑。它通常会考虑以下几个维度的信息：

1. 任务等待时间：任务在队列中已经等待了多久。等待时间越长，得分会逐渐增加，以防止任务被“饿死”。这引入了时间公平性。
2. 任务预估剩余时间：系统对任务还需要运行多久的预估。与 SJF 不同，这里不是单独使用，而是与其他因素结合。例如，一个等待了很久但即将完成的任务，可能会获得一个较高的分数，以便尽快释放其占用的资源。
3. 任务优先级/紧迫性：用户或系统赋予的静态或动态优先级。例如，线上服务降级的紧急修复任务优先级高于探索性实验。
4. 资源匹配度：当前空闲资源与任务需求的匹配程度。这不仅包括 GPU 数量，还包括 GPU 型号（A100 vs V100）、显存大小、节点间的网络带宽（对于分布式训练至关重要）。匹配度越高，调度后性能越好，得分可能越高。
5. 系统全局目标：调度器当前优化的主要目标是什么？是最大化吞吐量，还是最小化平均作业完成时间？这个目标会影响各因素在评分函数中的权重。

这个评分函数可以形式化为：Score(task) = w1 * f(等待时间) + w2 * g(预估剩余时间) + w3 * h(优先级) + w4 * i(资源匹配度) + ...其中，w1, w2, ...是权重系数，f, g, ...是将原始数据归一化或转化为得分的函数。

2.2 方案的关键优势

这种动态评分机制带来了几个显著优势：

• 避免饥饿：通过纳入等待时间，即使低优先级或长任务，在等待足够久后也能获得资源。
• 提高利用率：通过考虑资源匹配度，可以将任务调度到最“适合”它的节点上，减少因资源不匹配导致的启动失败或性能低下。
• 支持多目标：通过调整权重系数w，管理员可以在不同时期让调度器侧重不同的目标（例如，白天侧重低延迟保证研发交互，夜间侧重高吞吐进行大规模训练）。
• 适应动态性：评分是实时计算的，可以响应系统状态的变化，如新节点加入、节点故障、高优先级任务突然提交等。

3. 在工程环境中模拟与实现调度思路

我们不会完全复现论文中的算法，而是借鉴其思想，在一个简化的模拟环境或利用现有调度框架的可扩展点来实现类似的动态调度逻辑。

3.1 环境准备与依赖

我们将使用 Python 来模拟一个简单的任务队列和调度器。这有助于理解原理，并且这种模式可以集成到更复杂的系统中（如自定义 Kubernetes 调度器插件）。

所需环境：

• Python 3.8+
• 无需特殊硬件，本地机器即可运行模拟。

创建一个新的项目目录并初始化虚拟环境：

mkdir ai_scheduler_sim && cd ai_scheduler_sim python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate

3.2 定义任务与资源模型

首先，定义我们模拟世界中的基本对象：任务和 GPU 资源。

# models.py import time from dataclasses import dataclass from enum import Enum from typing import List, Optional class TaskPriority(Enum): LOW = 1 MEDIUM = 2 HIGH = 3 CRITICAL = 4 class GPUType(Enum): V100_16GB = "V100-16GB" A100_40GB = "A100-40GB" A100_80GB = "A100-80GB" RTX_4090 = "RTX4090" @dataclass class ComputeResource: """代表一个GPU资源单元""" id: str gpu_type: GPUType memory_gb: int is_available: bool = True current_task_id: Optional[str] = None @dataclass class AIJob: """代表一个AI计算任务（训练/推理）""" job_id: str submitted_time: float # 提交时间戳 estimated_duration: float # 预估运行时长（秒） priority: TaskPriority required_gpu_type: GPUType required_gpu_count: int required_memory_per_gpu_gb: int # 动态状态 start_time: Optional[float] = None finish_time: Optional[float] = None assigned_resources: List[ComputeResource] = None def __post_init__(self): if self.assigned_resources is None: self.assigned_resources = [] @property def wait_time(self) -> float: """计算当前已等待时间""" if self.start_time: return self.start_time - self.submitted_time return time.time() - self.submitted_time @property def estimated_remaining_time(self) -> float: """计算预估剩余运行时间（简化版：假设线性进度）""" if self.start_time and self.finish_time: return 0 elif self.start_time: # 假设任务匀速运行，已运行时间 = now - start_time elapsed = time.time() - self.start_time return max(0, self.estimated_duration - elapsed) return self.estimated_duration

3.3 实现动态评分调度器

这是核心部分。我们实现一个DynamicScoreScheduler类，它维护一个任务队列和一个资源池，并定期根据评分函数决定调度哪个任务。

# scheduler.py import time import heapq from typing import List, Dict, Tuple from models import AIJob, ComputeResource, TaskPriority, GPUType class DynamicScoreScheduler: def __init__(self, resources: List[ComputeResource]): self.resources = {r.id: r for r in resources} self.waiting_queue: List[AIJob] = [] # 等待队列 self.running_jobs: Dict[str, AIJob] = {} # 运行中的任务 # 权重系数：可动态调整以实现不同调度目标 self.weights = { 'wait_time': 1.0, 'inverse_remaining_time': 0.5, # 剩余时间越短，得分越高 'priority': 2.0, 'resource_fitness': 0.8 } def submit_job(self, job: AIJob): """提交新任务到等待队列""" self.waiting_queue.append(job) print(f"[{time.strftime('%H:%M:%S')}] Job {job.job_id} submitted (Priority: {job.priority.name}, Est: {job.estimated_duration}s)") def _calculate_score(self, job: AIJob, candidate_resources: List[ComputeResource]) -> float: """为核心：计算给定任务和候选资源集的综合得分""" # 1. 等待时间因子 (归一化：每等待10分钟加1分) wait_factor = job.wait_time / 600.0 # 600秒 = 10分钟 # 2. 剩余时间因子 (剩余时间越短，得分越高) # 防止除零，加一个平滑项 remaining = job.estimated_remaining_time inverse_remaining_factor = 1.0 / (remaining + 1.0) # 3. 优先级因子 priority_factor = job.priority.value # 枚举值 1,2,3,4 # 4. 资源匹配度因子 (简化：检查类型是否匹配，0或1) resource_fitness = 1.0 if all( r.gpu_type == job.required_gpu_type for r in candidate_resources ) else 0.0 # 更复杂的匹配度可以考虑显存、节点亲和性等 # 综合得分 score = ( self.weights['wait_time'] * wait_factor + self.weights['inverse_remaining_time'] * inverse_remaining_factor + self.weights['priority'] * priority_factor + self.weights['resource_fitness'] * resource_fitness ) return score def _find_resources_for_job(self, job: AIJob) -> List[ComputeResource]: """为任务寻找合适的空闲资源集合""" required_count = job.required_gpu_count required_type = job.required_gpu_type required_memory = job.required_memory_per_gpu_gb available_resources = [ r for r in self.resources.values() if r.is_available and r.gpu_type == required_type and r.memory_gb >= required_memory ] if len(available_resources) >= required_count: # 简化：直接取前N个符合条件的资源。实际中可能需要考虑拓扑（如NVLink） return available_resources[:required_count] return [] # 资源不足 def schedule(self): """调度循环：每次调用，尝试从等待队列中调度一个任务""" if not self.waiting_queue: return # 为每个等待中的任务计算其“最佳可能得分” job_scores = [] for job in self.waiting_queue: candidate_resources = self._find_resources_for_job(job) if candidate_resources: score = self._calculate_score(job, candidate_resources) # 使用负分因为 heapq 是最小堆，我们需要最大得分 heapq.heappush(job_scores, (-score, job, candidate_resources)) if not job_scores: return # 没有任务有足够的资源 # 选出得分最高的任务 _, best_job, best_resources = heapq.heappop(job_scores) # 分配资源并启动任务（模拟） for res in best_resources: res.is_available = False res.current_task_id = best_job.job_id best_job.assigned_resources.append(res) best_job.start_time = time.time() self.waiting_queue.remove(best_job) self.running_jobs[best_job.job_id] = best_job print(f"[{time.strftime('%H:%M:%S')}] SCHEDULED: Job {best_job.job_id} started on resources {[r.id for r in best_resources]}") print(f" -> Wait time: {best_job.wait_time:.1f}s, Score: {-job_scores[0][0] if job_scores else 'N/A':.2f}") def update_running_jobs(self): """更新运行中任务的状态，模拟任务完成并释放资源""" finished_jobs = [] current_time = time.time() for job in list(self.running_jobs.values()): # 简化：如果实际运行时间超过了预估时长，则标记为完成 if job.start_time and (current_time - job.start_time) >= job.estimated_duration: job.finish_time = current_time for res in job.assigned_resources: res.is_available = True res.current_task_id = None finished_jobs.append(job) print(f"[{time.strftime('%H:%M:%S')}] FINISHED: Job {job.job_id} completed.") for job in finished_jobs: del self.running_jobs[job.job_id] def run_simulation_loop(self, steps=10, interval=2): """运行模拟循环""" for i in range(steps): print(f"\n--- Scheduling Step {i+1} ---") self.update_running_jobs() self.schedule() # 打印当前状态 print(f" Waiting jobs: {len(self.waiting_queue)}, Running jobs: {len(self.running_jobs)}") time.sleep(interval) # 模拟时间间隔

3.4 运行模拟与验证

创建一个主程序来模拟整个调度过程，观察动态评分调度器的行为。

# main.py import time from models import AIJob, ComputeResource, TaskPriority, GPUType from scheduler import DynamicScoreScheduler def main(): # 1. 初始化资源池：假设我们有4块GPU resources = [ ComputeResource("gpu-0", GPUType.A100_40GB, 40), ComputeResource("gpu-1", GPUType.A100_40GB, 40), ComputeResource("gpu-2", GPUType.V100_16GB, 16), ComputeResource("gpu-3", GPUType.V100_16GB, 16), ] # 2. 创建调度器 scheduler = DynamicScoreScheduler(resources) # 3. 模拟提交一系列任务（时间戳模拟先后提交） base_time = time.time() # 任务1：高优先级，短任务，需要A100 job1 = AIJob( job_id="JOB-001", submitted_time=base_time - 30, # 30秒前提交 estimated_duration=15, priority=TaskPriority.HIGH, required_gpu_type=GPUType.A100_40GB, required_gpu_count=1, required_memory_per_gpu_gb=20 ) # 任务2：低优先级，长任务，需要A100 job2 = AIJob( job_id="JOB-002", submitted_time=base_time - 20, # 20秒前提交 estimated_duration=60, priority=TaskPriority.LOW, required_gpu_type=GPUType.A100_40GB, required_gpu_count=2, # 需要2块A100 required_memory_per_gpu_gb=30 ) # 任务3：中优先级，中长任务，需要V100 job3 = AIJob( job_id="JOB-003", submitted_time=base_time - 10, # 10秒前提交 estimated_duration=30, priority=TaskPriority.MEDIUM, required_gpu_type=GPUType.V100_16GB, required_gpu_count=1, required_memory_per_gpu_gb=10 ) # 任务4：关键优先级，紧急短任务，需要A100 job4 = AIJob( job_id="JOB-004", submitted_time=base_time, # 刚刚提交 estimated_duration=10, priority=TaskPriority.CRITICAL, required_gpu_type=GPUType.A100_40GB, required_gpu_count=1, required_memory_per_gpu_gb=15 ) # 提交所有任务 scheduler.submit_job(job1) scheduler.submit_job(job2) scheduler.submit_job(job3) scheduler.submit_job(job4) print("\n=== Initial State ===") print(f"Total Resources: {[r.id for r in resources]}") print(f"Submitted Jobs: {[j.job_id for j in [job1, job2, job3, job4]]}") # 4. 运行调度模拟循环 scheduler.run_simulation_loop(steps=8, interval=3) if __name__ == "__main__": main()

运行这个模拟程序：

python main.py

你将看到类似以下的输出，它展示了调度器如何根据动态评分做出决策：

=== Initial State === Total Resources: ['gpu-0', 'gpu-1', 'gpu-2', 'gpu-3'] Submitted Jobs: ['JOB-001', 'JOB-002', 'JOB-003', 'JOB-004'] --- Scheduling Step 1 --- [HH:MM:SS] SCHEDULED: Job JOB-004 started on resources ['gpu-0'] -> Wait time: 0.0s, Score: X.XX Waiting jobs: 3, Running jobs: 1 --- Scheduling Step 2 --- [HH:MM:SS] FINISHED: Job JOB-004 completed. [HH:MM:SS] SCHEDULED: Job JOB-001 started on resources ['gpu-0'] -> Wait time: 33.0s, Score: X.XX Waiting jobs: 2, Running jobs: 1 --- Scheduling Step 3 --- ...

结果分析：

• JOB-004虽然最后提交，但因其CRITICAL优先级，在第一次调度时就被选中，尽管其等待时间为0。
• JOB-001在 JOB-004 完成后被调度，其较高的优先级和中等等待时间使其得分领先于 JOB-002。
• JOB-002需要2块 A100，但可能因为资源不足（被 JOB-001 占用了1块）或优先级低而持续等待，直到资源满足且其等待时间因子增长到足够高。
• JOB-003使用 V100 资源，与其他任务不冲突，可能在早期就被调度。

这个简单的模拟验证了动态评分调度器能够综合考虑优先级、等待时间、资源匹配等多个因素，而不是遵循简单的 FIFO 规则。

4. 将调度思想集成到生产级系统

上述模拟阐述了核心概念。在生产环境中，我们通常不会从头编写调度器，而是基于成熟的编排系统进行扩展。

4.1 基于 Kubernetes 的扩展实现

Kubernetes 是容器编排的事实标准，其调度框架是可扩展的。我们可以实现一个自定义调度器插件（Scheduler Plugin）或使用调度框架（如 Kueue）来实现类似的动态评分逻辑。

核心思路：实现一个 Kubernetes 调度器插件，该插件为每个待调度的 Pod（AI 任务）计算一个分数，Kubernetes 的调度器核心会选择分数最高的节点。

1. 定义自定义资源（CRD）：为 AI 任务定义扩展字段，如ai.scheduler/priority、ai.scheduler/estimated-duration。

   # ai-job-crd.yaml apiVersion: apiextensions.k8s.io/v1 kind: CustomResourceDefinition metadata: name: aijobs.example.com spec: group: example.com versions: - name: v1alpha1 served: true storage: true schema: openAPIV3Schema: type: object properties: spec: type: object properties: priority: type: integer estimatedDurationMinutes: type: integer gpuType: type: string gpuCount: type: integer

2. 实现调度器插件（简化示例逻辑）：使用 Kubernetes 的调度框架（Scheduling Framework）编写一个 Filter/Score 插件。

   // 这是一个概念性Go代码片段，展示在Score扩展点实现的逻辑 func (pl *DynamicScorePlugin) Score(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) { // 1. 从Pod的Annotation或CRD中读取任务属性 priorityStr := pod.Annotations["ai.scheduler/priority"] waitTime := calculateWaitTime(pod.CreationTimestamp) // 2. 获取节点资源状态 nodeInfo, err := pl.handle.SnapshotSharedLister().NodeInfos().Get(nodeName) // 3. 计算资源匹配度（例如，节点是否有指定类型的GPU且数量足够） resourceFitScore := calculateResourceFit(pod, nodeInfo) // 4. 综合计算得分（类似Python模拟中的逻辑） score := pl.weights.waitTime * waitTime + pl.weights.priority * parsePriority(priorityStr) + pl.weights.resourceFit * resourceFitScore return int64(score * 100), nil // 转换为整数分 }

3. 部署与配置：将自定义调度器插件编译成镜像，部署到 Kubernetes 集群，并修改 Pod 的schedulerName字段来使用它。

4.2 基于 Slurm 的扩展

在高性能计算（HPC）领域，Slurm 是广泛使用的作业调度系统。它本身支持复杂的优先级计算（PriorityType配置），我们可以通过调整其优先级权重因子（PriorityWeightAge,PriorityWeightFairshare,PriorityWeightJobSize,PriorityWeightPartition,PriorityWeightQOS）来近似实现动态评分。更高级的做法是编写Slurm SPANK插件，在作业提交时为其设置基于自定义逻辑的优先级。

4.3 基于消息队列（如 RabbitMQ, Redis）的轻量级实现

对于小规模集群或简单的任务队列，可以使用消息队列结合消费者逻辑来实现。

1. 任务提交：生产者将任务信息（包含优先级、预估时间、资源需求）作为消息发布到队列。
2. 智能消费者：消费者进程不是简单地从队列头部获取消息，而是：
   • 定期扫描队列中的所有消息（例如，使用 Redis 的LRANGE或 RabbitMQ 的Get方法）。
   • 根据当前系统资源状态，为每条消息计算一个动态分数。
   • 选择分数最高的消息进行消费和处理。
   • 处理完成后，从队列中删除该消息。

这种方法实现简单，但需要注意并发控制和性能，因为频繁扫描整个队列可能带来开销。

5. 常见问题与排查路径

在实际部署动态调度系统时，你可能会遇到以下典型问题：

6. 生产环境最佳实践与扩展方向

将动态调度方案应用于生产环境，需要考虑远超出算法本身的工程因素。

6.1 关键配置与调优建议

• 权重系数的动态调整：不要设置固定不变的权重。可以基于时间（白天/夜晚）、集群负载（整体利用率）或业务周期（发版期/实验期）来动态调整权重。例如，白天上班时间，提高优先级和等待时间的权重以保证研发效率；夜间则提高吞吐量相关权重以跑满训练任务。
• 引入历史数据学习：任务预估时间是最不准却最重要的参数之一。建立任务历史运行数据库，根据任务类型、参数、数据量、资源类型来预测其运行时间，并不断用实际值修正预测模型。
• 实现分级队列：不要将所有任务放在一个队列里。可以按优先级、项目、用户或任务类型（交互式 vs 批处理）设立多个队列。调度器可以先从高优先级队列挑选，再考虑低优先级队列。这简化了调度逻辑并提供了更强的隔离性。
• 资源预留与抢占策略：为关键业务线或高优先级用户预留一部分资源。明确定义抢占策略：哪些任务可以被抢占，抢占前是否需要优雅终止（保存检查点），以及被抢占任务如何重新排队（其等待时间是否累积）。
• 丰富的监控与告警：
  • 调度决策看板：展示每次调度的任务、得分、胜出原因。
  • 队列状态监控：各优先级任务的平均/最长等待时间。
  • 资源利用率与碎片率。
  • 任务成功率与失败原因分析。
  • 当高优先级任务等待时间超过阈值，或集群整体利用率持续过低时触发告警。

6.2 扩展方向：更智能的调度

• 考虑数据局部性：对于需要读取大量训练数据的任务，将其调度到存储节点附近，或已有数据缓存的节点，可以极大减少 IO 时间。
• 考虑能源消耗：在评分函数中加入节点能耗因子，在性能相近时，优先选择能效更高的节点，或在不忙时集中任务以关闭部分节点。
• 多集群联邦调度：当单个集群资源不足时，能够将任务调度到云上或其他数据中心的集群，并考虑网络延迟和数据传输成本。
• 与模型仓库和流水线集成：调度器能够感知到任务之间的依赖关系（如 A 任务的输出是 B 任务的输入），进行有依赖的调度，优化整体流水线完成时间。

6.3 实施清单

在决定引入动态调度系统前，请对照此清单进行评估和准备：

1. 需求明确：你的主要痛点是什么？是任务等待时间长，还是资源利用率低，或是无法保证关键任务？
2. 数据收集：你是否能收集到任务的历史运行数据、资源使用情况、用户提交模式？
3. 方案选型：是改造现有系统（如调整 Slurm 参数），是扩展现有系统（如写 K8s 插件），还是引入新系统？
4. 测试环境：是否有独立的测试集群用于验证新调度策略，而不影响生产任务？
5. 回滚计划：新调度器出现问题后，如何快速切换回稳定版本？
6. 用户沟通：新的调度策略（如基于评分的公平性）可能会改变用户的任务体验，是否需要提前沟通和培训？

动态评分调度方案为 AI 算力管理提供了强大的灵活性，但其复杂性也显著高于静态策略。从一个小规模的模拟或非关键业务集群开始实践，逐步迭代评分函数和权重，收集反馈和数据，是将其成功落地的最可靠路径。其最终目标不是追求算法本身的“赢”，而是在你的特定业务场景下，找到吞吐量、延迟、公平性和成本之间的最佳平衡点。

🚀 30+款热门AI模型一站整合，DeepSeek/GLM/Qwen 随心用，限时 5 折。 👉 点击领海量免费额度