调度算法

在大模型服务系统中，调度算法决定了请求的批处理方式、执行顺序和资源分配，直接影响系统的吞吐量和延迟。本节介绍从静态批处理到动态调度的演进。

7.3.1 Continuous Batching （连续批处理）

问题背景

传统的静态批处理（Static Batching）存在严重问题：

问题：

请求B完成后，GPU空闲等待A和C
新请求D必须等待整个批次完成
GPU利用率低，延迟高

核心思想

Continuous Batching（也称为In-flight Batching或Dynamic Batching）允许： 1. 请求动态加入：新请求可以在任何迭代加入批次 2. 请求动态退出：完成的请求立即释放资源 3. 迭代级调度：每个迭代重新组织批次

关键机制

实现细节

迭代级调度

python

class ContinuousBatcher:
    def __init__(self, model, max_batch_size, max_tokens):
         self.model = model
         self.max_batch_size = max_batch_size
         self.max_tokens = max_tokens
         self.waiting_queue = deque()      # 等待队列
         self.running_batch = []           # 运行中的请求


    def schedule(self):
         """迭代级调度"""
         # 1. 移除完成的请求
         self.running_batch = [req for req in self.running_batch if not req.finished]

         # 2.   尝试添加新请求
         while (len(self.running_batch) < self.max_batch_size and
                 self.waiting_queue):
               new_req = self.waiting_queue.popleft()
               if self.can_fit(new_req):
                  self.running_batch.append(new_req)
               else:
                  self.waiting_queue.appendleft(new_req)
                  break


         # 3.   执行当前批次
         if self.running_batch:
               self.execute_batch(self.running_batch)


    def can_fit(self, request):
         """ 检查是否能容纳新请求"""
         total_tokens = sum(req.num_tokens() for req in self.running_batch)
         total_tokens += request.num_tokens()
         return total_tokens <= self.max_tokens


    def execute_batch(self, batch):
         """执行批次"""
         # 构建输入张量
         input_ids = pad_and_stack([req.get_input() for req in batch])
         attention_mask = create_attention_mask(batch)
         position_ids = create_position_ids(batch)


         #   前向传播
         outputs = self.model(input_ids, attention_mask, position_ids)


         #   分发结果
         for i, req in enumerate(batch):
               req.update(outputs[i])

请求状态管理

python

class Request:
      def __init__(self, prompt, max_new_tokens):
          self.prompt_tokens = tokenize(prompt)
          self.generated_tokens = []
          self.max_new_tokens = max_new_tokens
          self.phase = 'PREFILL'       # PREFILL 或DECODE
          
      def num_tokens(self):
          return len(self.prompt_tokens) + len(self.generated_tokens)

      def is_done(self):
          return (len(self.generated_tokens) >= self.max_new_tokens or
                      self.generated_tokens[-1] == EOS_TOKEN)

      def get_input(self):
          if self.phase == 'PREFILL':
               return self.prompt_tokens
          else:
               return [self.generated_tokens[-1]]

      def update(self, new_token):
          self.generated_tokens.append(new_token)
          self.phase = 'DECODE'

吞吐量优化

批大小权衡

批大小	吞吐量	延迟	GPU利用率
小	低	低	低
中	高	中	高
大	饱和	高	高

最优批大小取决于：

模型大小和内存占用
请求到达模式
延迟SLA要求（Service Level Agreement（服务等级协议 / 服务水平协议））

自适应批大小

python

def adaptive_batch_size(self, current_batch, waiting_requests):
       """  自适应调整批大小"""
       #   基于当前负载和等待队列长度
       queue_length = len(waiting_requests)
       avg_wait_time = self.get_avg_wait_time()

       if queue_length > 10 and avg_wait_time > 5.0:
             #   队列压力大，增加批大小
             return min(self.max_batch_size, len(current_batch) + 2)
       elif avg_wait_time < 1.0 and len(current_batch) > 1:
             #   负载轻，减小批大小以降低延迟
             return max(1, len(current_batch) - 1)

       return len(current_batch)

性能分析

Continuous Batching 相比Static Batching的提升：

指标	Static Batching	Continuous Batching	提升
吞吐量	1.0x	2.0-3.0x	2-3x
平均延迟	1.0x	0.8-1.2x	相当
P99延迟	1.0x	0.5-0.8x	更好
GPU利用率	40-60%	70-90%	显著提升

7.3.2 Chunked Prefill

问题背景

在Continuous Batching中，Prefill阶段（处理输入prompt）和Decode阶段（生成token）存在冲突：

Prefill计算密集：需要处理长序列，占用大量计算资源
Decode内存带宽密集：需要频繁访问KV Cache
混合执行困难：同时执行Prefill和Decode会互相干扰

传统方法： - 方案A：所有请求先完成Prefill，再一起Decode → 新请求等待时间长 - 方案B：每个请求独立Prefill和Decode → 批处理效果差

核心思想

Chunked Prefill 将长Prefill分解为多个小块，与Decode迭代交错执行：

分块策略

将长序列的Prefill分成固定大小的块（如512 tokens）：

\(\text{num\_chunks} = \left\lceil \frac{\text{seq\_len}}{\text{chunk\_size}} \right\rceil\)

每个Prefill块与Decode迭代一起批处理执行。

实现细节

混合批次构建

python

class ChunkedPrefillScheduler:
   def __init__(self, model, chunk_size=512, max_batch_tokens=4096):
       self.model = model
       self.chunk_size = chunk_size
       self.max_batch_tokens = max_batch_tokens
       self.prefill_queue = deque()       # 等待Prefill的请求
       self.decode_queue = deque()        # 等待Decode的请求


   def schedule(self):
       """ 构建混合批次"""
       batch = []
       total_tokens = 0

       # 1. 优先添加Decode请求（低延迟敏感）
       while self.decode_queue and total_tokens < self.max_batch_tokens:
             req = self.decode_queue.popleft()
             batch.append(('DECODE', req))
             total_tokens += 1   # Decode 每次1个token
           
       # 2.   添加Prefill块
       while self.prefill_queue and total_tokens < self.max_batch_tokens:
             req = self.prefill_queue[0]

             #   计算当前Prefill块的大小
             remaining = req.prefill_remaining()
             chunk = min(remaining, self.chunk_size)

             if total_tokens + chunk <= self.max_batch_tokens:
                  batch.append(('PREFILL', req, chunk))
                  total_tokens += chunk
                  req.advance_prefill(chunk)

                  if req.prefill_done():
                     self.prefill_queue.popleft()
                     self.decode_queue.append(req)
             else:
                  break

       return batch


   def execute_batch(self, batch):
       """执行混合批次"""
       # 分离Prefill和Decode
       prefill_items = [item for item in batch if item[0] == 'PREFILL']
       decode_items = [item for item in batch if item[0] == 'DECODE']

       #   构建统一的输入
       input_ids = []
       position_ids = []

       for _, req, chunk in prefill_items:
            input_ids.extend(req.get_prefill_chunk(chunk))
            position_ids.extend(range(req.prefill_start, req.prefill_start + chunk))

       for _, req in decode_items:
            input_ids.append(req.get_last_token())
            position_ids.append(req.get_position())

       #   单次前向
       outputs = self.model(input_ids, position_ids)

       #   分发结果
       idx = 0
       for _, req, chunk in prefill_items:
            req.update_kv_cache(outputs[idx:idx+chunk])
            idx += chunk


       for _, req in decode_items:
            req.append_token(outputs[idx])
            idx += 1

注意力掩码处理

混合批次需要特殊的注意力掩码：

python

def create_mixed_attention_mask(batch):
    """
    创建混合批次的注意力掩码
    Prefill: 因果掩码
    Decode: 全可见（只需要最后一个token的KV）
    """
    total_len = sum(chunk if op == 'PREFILL' else 1 for op, *rest in batch)
    mask = torch.zeros(total_len, total_len)

    pos = 0
    for op, *rest in batch:
          if op == 'PREFILL':
               _, _, chunk = rest
               # Prefill  使用因果掩码
               mask[pos:pos+chunk, pos:pos+chunk] = torch.tril(torch.ones(chunk, chunk))
               pos += chunk
          else:                    
               # Decode 只关注自己
               mask[pos, pos] = 1
               pos += 1


    return mask

延迟优化分析

首Token延迟（TTFT）

策略	TTFT	说明
完整Prefill	高	新请求等待前面所有Prefill完成
Chunked Prefill	低	新请求可以快速加入批次

交错延迟

性能分析

Chunked Prefill的优势：

更好的公平性：短请求不会被长请求阻塞
更高的 GPU利用率：计算和内存带宽操作更好地重叠
更可预测的延迟：避免长Prefill造成的延迟尖峰

实测数据（vLLM）：

在混合工作负载上，P99延迟降低30-50%
吞吐量提升10-20%

7.3.3 SARATHI-Serve 调度

问题背景

现有调度算法往往只关注吞吐量或延迟的单一优化目标。SARATHI-Serve提出了一种吞吐量-延迟联合优化的调度策略。

核心思想

SARATHI-Serve 的核心洞察：

预填充和解耦的权衡：大 batch 提升吞吐量但增加延迟
请求特性感知：不同请求有不同的延迟敏感度
自适应调度：根据系统状态动态调整策略

关键概念

请求分类：

延迟敏感型：交互式应用（如聊天），要求低TTFT
吞吐量敏感型：批处理应用（如文档处理），要求高吞吐

调度模式：

eager模式：低延迟优先，小 batch快速响应
batch模式：高吞吐优先，大batch最大化利用率

调度策略

自适应模式切换

python

class SarathiScheduler:
    def __init__(self, model):
        self.model = model
        self.mode = 'BATCH'     # 或 'EAGER'
        self.latency_threshold = 2.0      # 延迟阈值（秒）
        self.throughput_history = deque(maxlen=100)


    def select_mode(self):
        """ 选择调度模式"""
        current_latency = self.measure_avg_latency()
        queue_length = len(self.waiting_queue)

        if current_latency > self.latency_threshold:
            #   延迟过高，切换到eager模式
            self.mode = 'EAGER'
        elif queue_length > 10:
            #   队列积压，切换到batch模式
            self.mode = 'BATCH'


    def schedule_eager(self):
        """Eager 模式：低延迟优先"""
        batch = []

        #   优先处理延迟敏感请求
        latency_sensitive = [r for r in self.waiting_queue if r.is_latency_sensitive]
        throughput_sensitive = [r for r in self.waiting_queue if not r.is_latency_sensitive]

        #   小batch快速处理
        for req in latency_sensitive[:self.small_batch_size]:
            batch.append(req)

        return batch


    def schedule_batch(self):
        """Batch 模式：高吞吐优先"""
        batch = []
        total_tokens = 0

        #   尽可能填充批次
        for req in self.waiting_queue:
            if total_tokens + req.num_tokens() <= self.max_tokens:
                batch.append(req)
                total_tokens += req.num_tokens()


        return batch

请求优先级

python

class Request:
     def __init__(self, prompt, priority='normal', max_latency=None):
           self.prompt = prompt
           self.priority = priority    # 'high', 'normal', 'low'
           self.max_latency = max_latency
           self.arrival_time = time.time()


     def get_priority_score(self):
           """ 计算优先级分数（越低越优先）"""
           wait_time = time.time() - self.arrival_time

           #   基于优先级的基准分数
           base_score = {'high': 0, 'normal': 10, 'low': 20}[self.priority]

           #   等待时间加权
           urgency = wait_time / self.max_latency if self.max_latency else 0

           return base_score - urgency * 10

吞吐量-延迟权衡

Pareto 最优分析

SARATHI-Serve 探索了吞吐量和延迟的Pareto前沿：

latex

   吞吐量 ↑
         │
         │         ●   最优batch大小
         │        /│\
         │       / │ \
         │     ●   │     ●   过大batch（延迟高）
         │   /     │
         │ ●       │ 过小batch（吞吐低）
         └──────────→ 延迟

动态batch大小

python

  def dynamic_batch_size(self, requests, latency_sla):
       """
       动态确定最优batch大小
       目标：在满足延迟SLA的前提下最大化吞吐量
       """
        #   估计不同batch大小的延迟
       for batch_size in range(1, self.max_batch_size + 1):
             estimated_latency = self.estimate_latency(requests[:batch_size])

             if estimated_latency <= latency_sla:
                best_batch_size = batch_size
             else:
                break
                 
       return best_batch_size


 def estimate_latency(self, batch):
       """估计批次执行延迟"""
       # 基于历史数据和模型特性
       num_tokens = sum(r.num_tokens() for r in batch)

       #   线性模型（简化）
       latency = self.latency_intercept + self.latency_per_token * num_tokens

       return latency

性能分析

SARATHI-Serve 在不同工作负载下的表现：

工作负载	吞吐量提升	P99延迟降低
聊天	1.2x	40%
代码生成	1.5x	25%
文档处理	1.8x	10%
混合	1.4x	30%

调度算法 ​

7.3.1 Continuous Batching （连续批处理） ​

问题背景 ​

核心思想 ​

实现细节 ​

吞吐量优化 ​

性能分析 ​

7.3.2 Chunked Prefill ​

问题背景 ​

核心思想 ​

分块策略 ​

实现细节 ​

注意力掩码处理 ​

延迟优化分析 ​

性能分析 ​

7.3.3 SARATHI-Serve 调度 ​

问题背景 ​

核心思想 ​

关键概念 ​

调度策略 ​

吞吐量-延迟权衡 ​

性能分析 ​

调度算法

7.3.1 Continuous Batching （连续批处理）

问题背景

核心思想

实现细节

吞吐量优化

性能分析

7.3.2 Chunked Prefill

问题背景

核心思想

分块策略

实现细节

注意力掩码处理

延迟优化分析

性能分析

7.3.3 SARATHI-Serve 调度

问题背景

核心思想

关键概念

调度策略

吞吐量-延迟权衡

性能分析