图解大模型计算加速系列：vLLM源码解析2，调度器策略(Scheduler)

在本系列对vLLM的介绍中，都会按照 “宏观（图解） -> 细节（配合源码）” 的方式，先理清vLLM在这里想做什么事，为什么要这么做，然后再一起来看各小块的代码实现。

【全文目录如下】  
【0】前期提要与本期导览  
【1】入口函数  
【2】SequenceGroup  
2.1 原生输入  
2.2 SequenceGroup的作用  
2.3 SequenceGroup的结构  
【3】add_request()：将seq_group添加进调度器waiting队列  
【4】step()：调度器策略  
4.1 调度器结构  
4.2 整体调度流程  
4.3 _passed_delay：判断调度waiting队列的时间点  
4.4 can_allocate：能否为seq_group分配物理块做prefill  
4.5 can_append_slot：能否为seq_group分配物理块做decode  
4.6 allocate与append_slot：为seq_group分配物理块  
4.7 preempt：抢占策略  
4.8 调度器核心代码  
【5】总结

【#】前期提要与本期导览

在上一篇关于vLLM代码整体架构的文章中，提到过无论是“离线批处理（同步）”还是“在线流式服务（异步）”，它们都采用了同一个推理内核引擎LLMEngine，其整体架构如下：

其中：

在每1个推理阶段中，调度器（Scheduler） 决定哪些请求可以参与推理，并为这些请求做好逻辑块->物理块的映射。
在每1个推理阶段中，分布式执行者 （图中Distributed Workers部分，根据代码，我们将其命名为model_executor会更加合适）接收调度器传来的这些请求，分发到各个worker上去做推理。Worker中的CacheEngine负责实际管理KV Cache；Worker中的model负责加载模型、实行推理，PagedAttention相关的实现和调用就在model下。

这里，每1个推理阶段的定义是：prefill算1个推理阶段，每个decode各算1个推理阶段。在本文中，我们统一用step来表示“1个推理阶段”。

在本文中，我们会详细解读调度器（Scheduler）全部细节；
在下一篇文章中，我们会详细解读块管理（blockmanager）的全部细节，并以parallel sampling，beam search和prefix caching为例，将上图左半部分全部串一遍
在后续文章中，我们会来解读上图右半部分细节（还没来得及拆逻辑，暂时不知道会写几篇）

由于块管理者和调度器在代码上逻辑层层嵌套，所以为了不影响大家对调度器的理解，涉及到块管理者的部分，本文也会给出尽量简明清晰的说明。

【1】入口函数

在源码架构篇中我们提过，本系列的介绍思路是：以“离线批处理”作为入口，详细解说内核引擎LLMEngine的各块细节。在此基础上再来看“在线流式服务”的运作流程。所以现在，先来回顾下离线批处理的调用方式：

from vllm import LLM, SamplingParams  # batch prompts  
prompts = ["Hello, my name is",  "The president of the United States is",  "The capital of France is",  "The future of AI is",]  # 采样参数  
sampling_params = SamplingParams(temperature=0.8, top_p=0.95)  # 初始化vLLM offline batched inference实例，并加载指定模型  
llm = LLM(model="facebook/opt-125m")  # 推理  
outputs = llm.generate(prompts, sampling_params)  # 对每一条prompt，打印其推理结果  
for output in outputs:  prompt = output.prompt  generated_text = output.outputs[0].text  print(f"Prompt: {prompt!r}, Generated text: {generated_text!r}")

有两点需要注意：

llm = LLM(model="facebook/opt-125m")：实例化了一个离线批处理的vLLM对象。其本质是实例化了一个内核引擎LLMEngine对象。在执行这个步骤时，LLMEngine会执行一次模拟实验（profiling），来判断需要在gpu上预留多少的显存空间给KV Cache block（模拟实验的流程参见源码篇1的3.2节，TODO，大家可以对照着来读源码，本文不再涉及这块源码细节）。

推理入口在第24行outputs = llm.generate(prompts, sampling_params)。现在我们进入LLM类下，来看这个generate函数，代码如下：

# vllm/entrypoints/llm.py  
class LLM:  """An LLM for generating texts from given prompts and sampling parameters.  ...  """  def __init__(  self,  model: str,  tokenizer: Optional[str] = None,  tokenizer_mode: str = "auto",  trust_remote_code: bool = False,  tensor_parallel_size: int = 1,  dtype: str = "auto",  quantization: Optional[str] = None,  revision: Optional[str] = None,  tokenizer_revision: Optional[str] = None,  seed: int = 0,  gpu_memory_utilization: float = 0.9,  swap_space: int = 4,  enforce_eager: bool = False,  max_context_len_to_capture: int = 8192,  disable_custom_all_reduce: bool = True,  **kwargs,  ) -> None:  ...  # ==============================================================================  # 使用配置好的engine参数，初始化LLMEngine实例  # ==============================================================================  self.llm_engine = LLMEngine.from_engine_args(  engine_args, usage_context=UsageContext.LLM_CLASS)  # ==============================================================================  # 用于全局唯一的request_id，  # 在vLLM中内核引擎的处理中，1个prompt视为1个request，分配全局唯一的request_id  # ==============================================================================  self.request_counter = Counter()  ...  def generate(  self,  prompts: Optional[Union[str, List[str]]] = None,   sampling_params: Optional[SamplingParams] = None,  prompt_token_ids: Optional[List[List[int]]] = None,   use_tqdm: bool = True,  lora_request: Optional[LoRARequest] = None,  multi_modal_data: Optional[MultiModalData] = None,  ) -> List[RequestOutput]:  """Generates the completions for the input prompts.  NOTE: This class automatically batches the given prompts, considering  the memory constraint. For the best performance, put all of your prompts  into a single list and pass it to this method.  Args:  prompts: prompts可以是str，也可以是list[str]  sampling_params: 采样超参，例如温度、top_k等；如果为None则使用vLLM默认的参数  prompt_token_ids: prompt对应的token_id，如果没有提供的话，vllm会调用tokenizer进行                               转换  use_tqdm: 是否要展示process bar  lora_request: 如果想请求特定的lora_adapter，可以将它的path等信息包装在该请求中,  但vLLM建议尽量不要使用这种方式，因为私有的lora adapter可能会带来一些  安全性的问题          multi_modal_data: 多模态相关的数据  Returns:  A list of `RequestOutput` objects containing the generated  completions in the same order as the input prompts.  """  if prompts is None and prompt_token_ids is None:  raise ValueError("Either prompts or prompt_token_ids must be "  "provided.")  if isinstance(prompts, str):  # Convert a single prompt to a list.  prompts = [prompts]  if (prompts is not None and prompt_token_ids is not None  and len(prompts) != len(prompt_token_ids)):  raise ValueError("The lengths of prompts and prompt_token_ids "  "must be the same.")  if sampling_params is None:  # Use default sampling params.  sampling_params = SamplingParams()  if multi_modal_data:  multi_modal_data.data = multi_modal_data.data.to(torch.float16)  # ============================================================================  # 将request添加到engine中  # 在vLLM内核运算逻辑中，1个prompt算1个request，需要有1个全局唯一的request_id  # ============================================================================  num_requests = len(prompts) if prompts is not None else len(  prompt_token_ids)  for i in range(num_requests):  prompt = prompts[i] if prompts is not None else None  token_ids = None if prompt_token_ids is None else prompt_token_ids[  i]  # =======================================================================  # 将每个prompt添加进LLMEngine中，_add_request具体做了以下几件事：  # - 将每个prompt处理成特定的输入类型（SequenceGroup实例，后文会细说）  # - 将每个prompt加入Scheduler的waiting队列，等待处理  # =======================================================================  self._add_request(  prompt,  sampling_params,  token_ids,  lora_request=lora_request,  # Get ith image while maintaining the batch dim.  multi_modal_data=MultiModalData(  type=multi_modal_data.type,  data=multi_modal_data.data[i].unsqueeze(0))  if multi_modal_data else None,  )  # ============================================================================  # 把这个batch的所有prompt都添加完后，执行推理，详情参见_run_engine  # ============================================================================  return self._run_engine(use_tqdm)  def _add_request(  self,  prompt: Optional[str],  sampling_params: SamplingParams,  prompt_token_ids: Optional[List[int]],  lora_request: Optional[LoRARequest] = None,  multi_modal_data: Optional[MultiModalData] = None,  ) -> None:  # 每个prompt赋1个request_id  request_id = str(next(self.request_counter))  self.llm_engine.add_request(request_id,  prompt,  sampling_params,  prompt_token_ids,  lora_request=lora_request,  multi_modal_data=multi_modal_data)  def _run_engine(self, use_tqdm: bool) -> List[RequestOutput]:  # Initialize tqdm.  if use_tqdm:  num_requests = self.llm_engine.get_num_unfinished_requests()  pbar = tqdm(total=num_requests,  desc="Processed prompts",  dynamic_ncols=True)  # ===========================================================================  # 如果当前调度器中还有没完成推理的请求（调度器中waiting/running/swapped任一队列非空）  # ===========================================================================  outputs: List[RequestOutput] = []  while self.llm_engine.has_unfinished_requests():  # =========================================================================  # 执行1次推理调度（step），决定哪些请求的数据可以参与到这次推理中  # =========================================================================  step_outputs = self.llm_engine.step()  for output in step_outputs:  # =====================================================================  # 如果本step后，有请求已经完成了推理，就将推理结果装进outputs中  # =====================================================================  if output.finished:  outputs.append(output)  if use_tqdm:  pbar.update(1)  if use_tqdm:  pbar.close()  # Sort the outputs by request ID.  # This is necessary because some requests may be finished earlier than  # its previous requests.  outputs = sorted(outputs, key=lambda x: int(x.request_id))  return outputs

总结来说，当调用outputs = llm.generate(prompts, sampling_params)时，它实际做了两件事情：

1）_add_request：将输入数据传给LLMEngine，它具体做了如下事情：

#把每1个prompt包装成一个SequenceGroup对象。从客户端角度看，1个请求可能包含多个prompts，例如离线批处理场景下你可以将1个batch理解成1个请求；但是从LLMEngine的角度看，1个prompt是1个请求，所以它会对输入数据进行预处理。在后文对SequenceGroup的讲解中，我们会来看vLLM这样做的意义。

#把包装成SequenceGroup对象的数据加入调度器（Scheduler）的waiting队列，等待处理。 这一块相关的细节，我们放在后文说。

2）_run_engine：执行推理。只要调度器的waiting/running/swapped队列非空，就认为此时这批batch还没有做完推理，这时就会调用LLMEngine的step()函数，来完成1次调度以决定要送哪些数据去做推理。

所以，想要知道调度器的运作流程，只要从**LLMEngine的add_request()和step()两个函数入手就好了。不过在正式进入这两个函数的讲解之前，先来看和输入数据一个问题：**

为什么要把每个prompt都包装成一个SequenceGroup实例？SequenceGroup又长什么样呢？

【2】SequenceGroup

2.1 原生输入

在一般的推理场景中，通常给模型传1个prompt及相关的采样参数，让模型来做推理。此时你的输入可能长下面这样：

("To be or not to be,",  
SamplingParams(temperature=0.8, top_k=5, presence_penalty=0.2)),

但在其余的场景中，模型decoding的策略可能更加复杂，例如：

Parallel Sampling：你传给模型1个prompt，希望模型基于这个这个prompt，给出n种不同的output
Beam Search：你传给模型1个prompt，在采用Beam Search时，每个推理阶段你都会产出top k个output，其中k被称为Beam width（束宽）。

这些情况下，你传给模型的输入可能长下面这样：

# Parallel Sampling  
("What is the meaning of life?",  
SamplingParams(n=2, temperature=0.8, top_p=0.95, frequency_penalty=0.1))  # Beam Search (best_of = 束宽)  
("It is only with the heart that one can see rightly",  
SamplingParams(n=3, best_of=3, use_beam_search=True, temperature=0.0)),

【备注：SamplingParams遵从OpenAI API范式，对其中各种参数的解释可参见OpenAI官方文档】

总结来说，可能出现"1个prompt -> 多个outputs"的情况。那是否能设计一种办法，对1个prompt下所有的outputs进行集中管理，来方便vLLM更好做推理呢？

2.2 SequenceGroup的作用

"1个prompt -> 多个outputs"这样的结构组成一个SequenceGroup实例。

其中每组"prompt -> output"组成一个序列（seq，属于Sequence实例），每个seq下有若干状态(status)属性，包括：

FINISHED_STOPPED：正常执行完毕，例如碰到符号，该seq的推理正常结束了
FINISHED_LENGTH_CAPPED：因为seq的长度达到最大长度限制，而结束推理
FINISHED_ABORTED：因不正常状态，而被终止的推理。例如客户端断开连接，则服务器会终止相关seq的推理
FINISHED_IGNORED：因prompt过长而被终止执行的推理。本质上也是受到长度限制
WAITING：正在waiting队列中。waiting队列中的序列都没有做过prefill。
RUNNING：正在running队列中，即已经开始做推理。
SWAPPED：正在swapped队列中，表示此时gpu资源不足，相关的seq_group被抢占，导致其暂停推理，相关的KV block被置换到cpu上（swap out），等待gpu资源充足时再置换回来重新计算（swap in）。
若干和Finish相关的状态，表示该seq推理已经结束，具体包括：**

在vLLM中有一个重要假设：一个seq_group中的所有seq共享1个prompt。

来通过一个具体的例子，更好感受一下SequenceGroup的作用：

在推理开始之前，这个seq_group下只有1条seq，它就是prompt，状态为waiting。

在第1个推理阶段，调度器选中了这个seq_group，由于它的采样参数中n = 4，所以在做完prefill之后，它会生成4个seq，它们的状态都是running。

在若干个推理阶段后，gpu上的资源不够了，这个seq_group不幸被调度器抢占（preemption），它相关的KV block也被swap out到cpu上。此时所有seq的状态变为swapped。这里要注意，当一个seq_group被抢占时，对它的处理有两种方式：Swap：如果该seq_group下的seq数量 > 1，此时会采取swap策略，即把seq_group下【所有】seq的KV block从gpu上卸载到cpu上。（seq数量比较多，直接把算出的KV block抛弃，比较可惜）。Recomputation：如果该seq_group下的seq数量 = 1，此时会采取recomputation策略，即把该seq_group相关的物理块都释放掉，然后将它重新放回waiting队列中。等下次它被选中推理时，就是从prefill阶段开始重新推理了，因此被称为“重计算”。（seq数量少，重新计算KV block的成本不高）。

【注意，并不是每个seq_group都会经历抢占，具体要看调度器策略和gpu资源使用情况】

又过了若干个推理阶段，gpu上的资源又充足了，此时执行swap in操作，将卸载到cpu上的KV block重新读到gpu上，继续对该seq_group做推理，此时seq的状态又变为running。

又过了若干个推理阶段，该seq_group中有1个seq已经推理完成了，它的状态就被标记为finish，此后这条已经完成的seq将不参与调度。

又过了若干个推理阶段，这个seq_group下所有的seq都已经完成推理了，这样就可以把它作为最终output返回了。

相信通过这个例子，已经能更好理解为什么vLLM要把1个prompt包装成SequenceGroup实例了。接下来我们就来看SequenceGroup实例的具体结构。

2.3 SequenceGroup的结构

SequenceGroup相关的脚本在vllm/sequence.py中，下图给出了SequenceGroup的结构图解（仅列出重要的属性和方法）：

（1）结构总述

SequenceGroup:

self.seqs_dict：{seq_id: seq}，其中每个seq是一个Sequence对象。正如我们前文介绍的那样，一个seq_group下包含若干seqs
self.sampling_params：采样参数
self.metrics：记录该seq_group相关的指标，例如该seq_group是什么时候被加入LLMEngine的（arrival_time），该seq_group第一次被调度器选中调度是什么时候等等。调度器在选择时，会参考seq_groups们的这些指标来做决策。
get_max_num_running_steps：该seq_group在剩余生命周期内并行running的最大seq数量。“剩余生命周期”指从此刻一直到seq_group中所有的seq都做完推理。举个例子来说，我们看2.2节配图中倒数第3个时刻，此时这个seq_group内所有的seq都还没结束推理，所以若调用这个方法，则返回值为4；再看倒数第2个时刻，此时有1个seq已经完成了推理，所以若调用这个方法，则返回值为3。在后续调度策略代码中，我们将经常看到这个方法被调用，目的是用于估计若当前对一个seq_group做推理，它将消耗多少gpu资源。

来详细看下get_max_num_running_steps代码实现（一切尽在注释中）：

    def get_max_num_running_seqs(self) -> int:  """The maximum number of sequences running in parallel in the remaining  lifetime of the request.  返回请求在其剩余生命周期中并行运行的最大序列数。  """  # ============================================================================  # 若采用beam search，每1个推理阶段都是best_of（束宽）个seq在running  # ============================================================================  if self.sampling_params.use_beam_search:  return self.sampling_params.best_of  # ============================================================================  # 如果不采用beam search  # ============================================================================  else:  # =========================================================================  # 此时best_of默认和n一致，即表示我们希望1个prompt产出n个outputs。因此理论上，这个  # seq_group下会维护best_of个seq（这就是self.num_seqs()的返回值）。  # 如果出现best_of > self.num_seqs()的情况，说明该seq_group刚从waiting变成running  # 准备做推理（参考2.2节配图中左侧第1个时刻），此时对于这个seq_group来说，  # 其剩余生命周期并行运行的最大seq数量为best_of  # =========================================================================  if self.sampling_params.best_of > self.num_seqs():  # At prompt stage, the sequence group is not yet filled up  # and only have one sequence running. However, in the  # generation stage, we will have `best_of` sequences running.  return self.sampling_params.best_of  # =========================================================================  # 其余时刻（例如2.2节配图中非左侧第1个时刻的所有时刻）下，我们就返回这个seq_group中  # 未完成推理的seq数量。根据2.2节介绍，我们知道一个seq的完成状态有四种：  #   SequenceStatus.FINISHED_STOPPED,  #   SequenceStatus.FINISHED_LENGTH_CAPPED,  #   SequenceStatus.FINISHED_ABORTED,  #   SequenceStatus.FINISHED_IGNORED  # =========================================================================  return self.num_unfinished_seqs()

Sequence:

对于一个seq，我们重点来看它的属性self.logical_token_blocks（逻辑块）和方法_append_tokens_to_blocks（生成逻辑块的方法）。在vLLM中，每个seq都单独维护一份属于自己的逻辑块，不同的逻辑块可以指向同一个物理块（此刻你一定很关心逻辑块和物理块是如何做映射的，我们会循序渐进地讲解这点，现在你可以先忽略映射方法，把目光聚焦于“一个seq的逻辑块长什么样，怎么初始化它的逻辑块”）

（2）1个逻辑块的结构

先来回答“1个逻辑块长什么样”这个问题，逻辑块定义的代码比较简单，所以我们直接看代码（一切尽在注释中），代码路径vllm/block.py

class LogicalTokenBlock:  """A block that stores a contiguous chunk of tokens from left to right.  Logical blocks are used to represent the states of the corresponding  physical blocks in the KV cache.  KV cache的逻辑块  """  def __init__(  self,  block_number: int, # 逻辑块的序号  block_size: int, # 每个逻辑块中有多少个槽位（默认为16）  ) -> None:  self.block_number = block_number  self.block_size = block_size  # 逻辑块刚初始化时，将其中的每个token_id都初始化为_BLANK_TOKEN_ID（-1）  self.token_ids = [_BLANK_TOKEN_ID] * block_size   # 当前逻辑块中已经装下的token的数量  self.num_tokens = 0  def is_empty(self) -> bool:  """判断当前逻辑块是为空"""  return self.num_tokens == 0  def get_num_empty_slots(self) -> int:  """当前逻辑块的空余槽位"""  return self.block_size - self.num_tokens  def is_full(self) -> bool:  """判断当前逻辑块是否已经被装满"""  return self.num_tokens == self.block_size  def append_tokens(self, token_ids: List[int]) -> None:  """将给定的一些token_ids装入当前逻辑块中"""  # 给定的token_ids的长度必须 <= 当前逻辑块剩余的槽位  assert len(token_ids) <= self.get_num_empty_slots()  # 当前逻辑块第一个空槽的序号  curr_idx = self.num_tokens  # 将这些tokens装进去  self.token_ids[curr_idx:curr_idx + len(token_ids)] = token_ids  # 更新当前逻辑块中tokens的数量  self.num_tokens += len(token_ids)  def get_token_ids(self) -> List[int]:  """获取当前逻辑块中所有被装满的位置的token_ids"""  return self.token_ids[:self.num_tokens]  def get_last_token_id(self) -> int:  """获取当前逻辑块所所有被装满的位置的最后一个token_id"""  assert self.num_tokens > 0  return self.token_ids[self.num_tokens - 1]

（3）再回到Sequence上来

知道了每个逻辑块的结构，我们现在来回答“怎么给一个seq分配逻辑块”这个问题，也就是回到2.3（1）中Sequence的_append_tokens_to_blocks方法上来：当一个seq只有prompt时，这个方法负责给prompt分配逻辑块；当这个seq开始产出output时，这个方法负责给每一个新生成的token分配逻辑块，整个过程如下图（图片来自vLLM论文，大家忽略图中block_table的部分）：

代码如下（一切尽在注释中，/vllm/sequence.py）:

    def _append_tokens_to_blocks(self, token_ids: List[int]) -> None:  """  将token_ids动态填入逻辑块列表中  Args:  token_ids: prompt部分的token_ids  """  cursor = 0  # 遍历prompt token_ids中的每一个token_id  while cursor < len(token_ids):  # 如果当前逻辑块列表（logical_token_blocks）为空  if not self.logical_token_blocks:  # 则先append一个逻辑块，该逻辑块index为0，大小为16，其中的每一个token_id为-1  self._append_logical_block()  # 取出逻辑块列表中的最后一个逻辑块  last_block = self.logical_token_blocks[-1]  # 如果这最后一个逻辑块中已经没有槽位  if last_block.is_full():  # 那么再append一个逻辑块，其大小为16，其中每一个token_id为-1  self._append_logical_block()  # 把这个新append的逻辑块取出来  last_block = self.logical_token_blocks[-1]  # 检查当前取出的逻辑块中空槽位的数量  num_empty_slots = last_block.get_num_empty_slots()  # 用当前的token_ids填充空槽位，直到无法填满为止  last_block.append_tokens(token_ids[cursor:cursor +  num_empty_slots])  cursor += num_empty_slots

好，到目前为止，就把vLLM对输入数据做预处理的部分介绍完了，简单总结下：

在vLLM内部计算逻辑中，1个prompt是1个request
每个prompt将被包装成一个SequenceGroup实例提供给调度器做调度
1个SequenceGroup实例下维护着若干个Sequence实例，对应着“1个prompt -> 多个outputs"这种更一般性的解码场景。
1个Sequence实例下维护着属于自己的逻辑块列表，数据类型为List[LogicalTokenBlock]

【3】add_request()：将seq_group添加进调度器waiting队列

写了这么多，你是不是已经忘记上面都说了些什么了，不要紧，我们快速回顾下：

首先，我们明确了vLLM最重要的推理内核引擎是LLMEngine
LLMEngine下有两个最重要的方法：add_request()和step()
add_request()负责将每个prompt都包装成一个SequenceGroup对象，送入调度器的waiting队列中等待调度
step()负责执行1次推理过程，在这个过程中，调度器首先决定哪些seq_group可以被送去推理，然后model_executor负责实际执行推理。

现在，在知道SequenceGroup相关定义的基础上，我们可以来看add_request()了，我们直接来看代码（一切尽在注释中，为了方便阅读，代码有所省略）：

# vllm/engine/llm_engine.py  def add_request(  self,  request_id: str, # 每个请求的唯一id  prompt: Optional[str], # prompt（文字版）  sampling_params: SamplingParams, # 用于采样的参数（温度、topk等）  prompt_token_ids: Optional[List[int]] = None, # prompt（input_ids版）  arrival_time: Optional[float] = None, # 请求到达的时间。如果是None，则用当前系统时间  lora_request: Optional[LoRARequest] = None,  # 如果是用lora模型做推理，相关的lora请求  multi_modal_data: Optional[MultiModalData] = None, # 每个请求的多模态数据  ) -> None:  """  将request添加给LLMEngine  Args:  request_id: 在vLLM内部，1条prompt算1个请求，会附给1个请求id  prompt: prompt（文字版）  sampling_params: 采样参数（温度、topk等）  prompt_token_ids: prompt（token_id版），没有提供的话vLLM会调用tokenizer来做  arrival_time: 请求到达的时间。如果是None，则用当前系统时间  multi_modal_data: 多模态数据（暂时忽略不看）  """  ...  # ============================================================================  # 设置该请求的到达时间  # ============================================================================  if arrival_time is None:  arrival_time = time.time()  ...  # 每个KV cache block的大小（默认为16）  block_size = self.cache_config.block_size  # 当前seq的id（见后文讲解）  seq_id = next(self.seq_counter)  # 获取用于表示<eos>的token_id  eos_token_id = self.tokenizer.get_lora_tokenizer(  lora_request).eos_token_id  # ============================================================================  # 为当前序列创建Sequence对象，在Sequence对象中也包括对当前序列逻辑块们的管理  # ============================================================================  seq = Sequence(seq_id, prompt, prompt_token_ids, block_size,  eos_token_id, lora_request)  ...  # ============================================================================  # 每个prompt被包装成一个SequenceGroup实例  # ============================================================================  seq_group = SequenceGroup(request_id, [seq], sampling_params,  arrival_time, lora_request, multi_modal_data)  # ============================================================================  # 将seq_group中所有序列添加进scheduler的self.waiting队列中  # self.waiting是一个双端队列实例，我们可以在队列的两端进行插入/删除操作  # ============================================================================  self.scheduler.add_seq_group(seq_group)

【4】step()：调度器策略

现在所有的seq_group都已经被送入调度器（Scheduler）的waiting队列中了，接下来就来看，在1个推理阶段中，调度器是通过什么策略来决定要送哪些seq_group去做推理的，这也是vLLM难啃的硬骨头之一。

调度器相关的代码都在 vllm/core/scheduler.py中，由于代码逻辑嵌套比较复杂，所以依然先通过图解的方式把整个调度流程介绍一遍，然后再看关键的源码细节。

4.1 调度器结构

vLLM调度器维护的重要属性如上图所示：

self.waiting, self.running, self.swapped：这三个都是python的**deque()**实例（双端队列，允许你从队列两侧添加或删除元素）。
waiting队列用于存放所有还未开始做推理的seq_group，“未开始”指连prefill阶段都没有经历过。所以waiting队列中的seq_group只有一个seq，即是原始的prompt。
running队列用于存放当前正在做推理的seq_group。更准确地说，它存放的是上1个推理阶段被送去做推理的seq_group们，在开始新一轮推理阶段时，调度器会根据本轮的筛选结果，更新running队列，即决定本轮要送哪些seq_group去做推理。
swapped队列用于存放被抢占的seq_group。在2.2节中我们有提过，若一个seq_group被抢占，调度器会对它执行swap或recomputation操作，分别对应着将它送去swapped队列或waiting队列，在后文我们会详细分析抢占处理的代码
self.policy：是vLLM自定义的一个Policy实例，目标是根据调度器总策略（FCFS，First Come First Serve，先来先服务）原则，对各个队列里的seq_group按照其arrival time进行排序。相关代码比较好读，所以这里我们只概述它的作用，后续不再介绍它的代码实现。
self.prev_time：上一次调度发起的时间点，初始化为0。我们知道每执行1次推理阶段前，调度器都要做一次调度，这个变量存放的就是上次调度发起的时间点。
self.prev_prompt：取值为True/False，初始化为False。若上一次调度时，调度器有从waiting队列中取出seq_group做推理，即为True，否则为False。
self.last_prompt_latency：记录“当前调度时刻（now） - 最后一次有从waiting队列中取数做推理的那个调度时刻”的差值（并不是每一次调度时，调度器一定都会从waiting队列中取seq_group，它可能依旧继续对running队列中的数据做推理），初始化为0。

目前你可能很难明白这三个属性的作用，不要着急，在后文讲解具体调度流程时，我们会再来看它们。这里只需记住它们的定义即可。

1).BlockManager：物理块管理器。这也是vLLM自定义的一个class。截止本文写作时，vLLM提供了BlockSpaceManagerV1和BlockSpaceManagerV2两个版本的块管理器。V1是vLLM默认的版本，V2是改进版本（但还没开发完，例如不支持prefix caching等功能）。所以本文依然基于****BlockSpaceManagerV1进行讲解。物理块管理器这个class下又维护着两个重要属性：

2).BlockAllocator：物理块分配者，负责实际为seq做物理块的分配、释放、拷贝等操作。这也是我们后文要解读的对象。其下又分成self.gpu_allocator和self.cpu_allocator两种类型，分别管理gpu和cpu上的物理块。

3).self.block_tables：负责维护每个seq下的物理块列表，本质上它是一个字典，形式如{seq_id: List[PhysicalTokenBlock]}。注意，这里维护者【所有】seq_group下seq的物理块，而不是单独某一个seq的。因为整个调度器都是全局的，其下的BlockManager自然也是全局的。

读到这里，你还记得2.3节中我们曾介绍过，每个Sequence实例中维护着属于这个seq的逻辑块吗？而从self.block_tables中，又能根据seq_id找到这个seq对应的物理块。这就实现了“逻辑块 -> 物理块”的映射。在刚开始读代码的时候，很多朋友从直觉上都会觉得BlockManager就是用来存储逻辑块和物理块映射的，其实它只负责管理和分配物理块，映射关系潜藏在seq中。理解这点对理解代码非常重要。

现在，就把调度器（Scheduler）的结构理清了。我知道你肯定还有很多疑惑。所以马上来看调度策略的具体流程：“对于装在waiting、running、swapped队列中的那些seq_group，是根据什么规则决定本次推理阶段该送谁去推理呢？”

4.2 整体调度流程

上图刻画了某次调度步骤中三个队列的情况，再复习一下：

1.waiting队列中的数据都没有做过prefill，每个seq_group下只有1个seq（prompt）

2.running队列中存放着上一个推理阶段被送去做推理的所有seq_group

3.swapped队列中存放着之前调度阶段中被抢占的seq_group

running队列中的seq_group不一定能继续在本次调度中被选中做推理，这是因为gpu上KV cache的使用情况一直在变动，以及waiting队列中持续有新的请求进来的原因。所以调度策略的职责就是要根据这些变动，对送入模型做推理的数据做动态规划。

根据源码，我将vLLM调度步骤整理成上述流程图。看着有点复杂是吧，不要担心，我们这就来拆解它。

总结来说：

如果当前swapped队列为空，那就去检查是否能从waiting队列中调度seq_group，直到不满足调度条件为止（gpu空间不足，或waiting队列已为空等）。此时，1个推理阶段中，所有的seq_group都处在prefill阶段。
如果当前swapped队列非空，或者无法从waiting队列中调度任何seq_group时：
检查是否能从running队列中调度seq_group，直到不满足调度条件为止。
若本次无新的被抢占的seq_group，且swapped队列非空，就检查是否能从swapped队列中调度seq_group，直到不满足调度条件为止。

此时，1个推理阶段中，所有的seq_group要么全来自running队列，要么来自running + swapped队列，它们都处在decode阶段。

至此我们要记住vLLM调度中非常重要的一点：在1个推理阶段中，所有的seq_group要么全部处在prefill阶段。要么全部处在decode阶段。

你可能想问：为什么要以swapped是否非空为判断入口呢？

这是因为，如果当前调度步骤中swapped队列非空，说明在之前的调度步骤中这些可怜的seq_group因为资源不足被抢占，而停滞了推理。所以根据FCFS规则，当gpu上有充足资源时，我们应该先考虑它们，而不是考虑waiting队列中新来的那些seq_group。

同理，在图中你会发现，当我们进入对running队列的调度时（图中红色分支），我们会根据“本次调度是否有新的被抢占的seq_group”，来决定要不要调度swapped队列中的数据。这个理由也很简单：在本次调度中，我就是因为考虑到gpu空间不足的风险，我才新抢占了一批序列。既然存在这个风险，我就最好不要再去已有的swapped队列中继续调度seq_group了。

到这里，我们已经把整个调度流程的关键点给说完了。接下来，我们会配合源码，对上图中的细节进行介绍。

4.3 _passed_delay：判断调度waiting队列的时间点

在4.2的流程图中，4会看到进入waiting循环的判断条件之一是：waiting队列是否达到调度间隔阈值。这是个什么东西？又为什么要设置这样一个阈值呢？

模型在做推理时，waiting队列中是源源不断有seq_group进来的，一旦vLLM选择调度waiting队列，它就会停下对running/swapped中seq_group的decode处理，转而去做waiting中seq_group的prefill，也即vLLM必须在新来的seq_group和已经在做推理的seq_group间取得一种均衡：既不能完全不管新来的请求，也不能耽误正在做推理的请求。所以“waiting队列调度间隔阈值”就是来控制这种均衡的：

调度间隔设置得太小，每次调度都只关心waiting中的新请求，这样发送旧请求的用户就迟迟得不到反馈结果。且此时waiting队列中积累的新请求数量可能比较少，不利于做batching，浪费了并发处理的能力。
调度间隔设置得太大，waiting中的请求持续挤压，同样对vLLM推理的整体吞吐有影响。

那这个阈值在代码中是怎么控制的呢？还记得4.1中我们画Scheduler的结构图时有三个乍一看比较难懂的属性吗（见下图），它们就是用来控制这个阈值的：

vllm/core/scheduler.py脚本的_passed_delay()函数写了阈值判断的相关逻辑，直接看代码（一切尽在注释中）:

    def _passed_delay(self, now: float) -> bool:  """  判断当下是否可以从waiting队列中调度新请求  这个函数确保了在调度过程中不会频繁地处理新来的seq_group  Args:  now: 当前调度时间点  """  # =============================================================================  # self.prev_prompt: True/False，记录上一次调度步骤中，是否选择了从waiting队列中做调度  # self.prev_time：上次调度步骤时间点（不管是从哪个队列中调度，每次调度都会记录下时间点）  # 若上个调度步骤中，我们选择从waiting队列中做调度，则计算两个调度时刻的间隔  # ==============================================================================  if self.prev_prompt:  self.last_prompt_latency = now - self.prev_time  # =============================================================================  # 用当前调度时间更新prev_time  # 由于目前还不知道本次是否会从waiting队列中调度，因此prev_prompt先设为False  # =============================================================================  self.prev_time, self.prev_prompt = now, False  # =============================================================================  # Delay scheduling prompts to let waiting queue fill up  # delay_factor：用户配置的，用于调整调度间隔阈值的因子。大于0则意味着用户想开启阈值判断  # =============================================================================  if self.scheduler_config.delay_factor > 0 and self.waiting:  # =========================================================================  # 计算在waiting队列中，最早到达的seq_group的到达时间  # =========================================================================  earliest_arrival_time = min(  [e.metrics.arrival_time for e in self.waiting])  # =========================================================================  # now - earliest_arrival_time：最早到达waiting队列的seq_group当前“实际”等待的时间  # delay_factor*last_prompt_latency：最早到达waiting队列的请求当前“应该”等待的时间  # 只要前者比后者大，或者此时running队列中根本没有请求在跑，就可以进行对waiting做调度  # =========================================================================  passed_delay = (  (now - earliest_arrival_time) >  (self.scheduler_config.delay_factor * self.last_prompt_latency)  or not self.running)  # =============================================================================  # 如果你不想开启阈值判断，那就直接返回True  # =============================================================================  else:  passed_delay = True  return passed_delay

4.4 can_allocate：能否为seq_group分配物理块做prefill

通过了调度时间阈值的判断条件，现在我们顺利从waiting中取出一个seq_group，我们将对它进行prefill操作。所以这里我们必须先判断：gpu上是否有充足的空间为该seq_group分配物理块做prefill，根据4.1中绘制的调度器结构，这个操作当然是由我们的self.block_manager来做。

判断的入口代码为can_allocate = self.block_manager.can_allocate(seq_group)，配合上面图例，我们直接来看can_allocate函数的代码，（一切尽在注释中）:

    # vllm/core/block_manager_v1.py  def can_allocate(self, seq_group: SequenceGroup) -> AllocStatus:  """  确实是否可以给这个seq_group分配物理块，返回结果有三种情况：  - AllocStatus.NEVER：不分配；  - AllocStatus.OK：可以分配；  - AllocStatus.LATER：延迟分配  """  # FIXME(woosuk): Here we assume that all sequences in the group share  # the same prompt. This may not be true for preempted sequences.  # (这里我们假设一个seq_group下的所有序列的prompt都是相同的)  # ===========================================================================  # 取出这个seq_group下处于waiting状态的序列  # ===========================================================================  seq = seq_group.get_seqs(status=SequenceStatus.WAITING)[0]  # ===========================================================================  # 取出这个seq所有的逻辑块  # ===========================================================================  num_required_blocks = len(seq.logical_token_blocks)  # ===========================================================================  # block上的滑动窗口（可暂时假设其值为None，先忽略不看  # ===========================================================================  if self.block_sliding_window is not None:  num_required_blocks = min(num_required_blocks,  self.block_sliding_window)  # ===========================================================================  # 计算当前所有可用的物理块数量，List[PhysicalTokenBlock]  # ===========================================================================  num_free_gpu_blocks = self.gpu_allocator.get_num_free_blocks()  # ===========================================================================  # Use watermark to avoid frequent cache eviction.  # 决定是否能为当前seq分配物理块  # ===========================================================================  # 如果设备中所有的物理块数量 - 该seq实际需要的物理块数量 < 水位线block数量，则不分配  # （说明当前seq太长了）  if (self.num_total_gpu_blocks - num_required_blocks <  self.watermark_blocks):  return AllocStatus.NEVER  # 如果设备中可用的物理块数量 - 该seq实际需要的block数量 >= 水位线block数量，则分配  if num_free_gpu_blocks - num_required_blocks >= self.watermark_blocks:  return AllocStatus.OK  # 否则，现在不能分配，但可以延迟分配  else:  return AllocStatus.LATER

对上述代码做一些额外的说明：

代码第32行：num_free_gpu_blocks = self.gpu_allocator.get_num_free_blocks()。这里是在统计当前gpu上所有可用的物理块数数量（忘记gpu_allocator是什么的朋友，可以再回顾下4.1的调度器结构图）。在vLLM中，gpu_allocator的类型有两种：

1.CachedBlockAllocator：按照prefix caching的思想来分配和管理物理块。在原理篇中，我们提过又些prompts中可能含有类似system message（例如，“假设你是一个能提供帮助的行车导航”）E）等prefix信息，带有这些相同prefix信息的prompt完全可以共享用于存放prefix的物理块，这样既节省显存，也不用再对prefix做推理。

2.UncachedBlockAllocator：正常分配和管理物理块，没有额外实现prefix caching的功能。

关于这两种allocator的具体实现方式，我们将放在源码解读第3篇块管理来做讲解。这里大家只要明白大致定义即可，并不影响我们对调度策略的解读。

self.watermark_blocks：水位线block数量，它起的是一个预警和缓冲的作用，防止在1次调度中把gpu上预留给KV Cache的显存空间打得过满，出现一些意外风险（毕竟这个预留的显存空间也是我们估计出来的）。

NEVER和LATER的区别：这两者的相同之处在于，都是因为当前显存空间不够，而无法继续调度seq_group。区别在于，NEVER是因为这条seq实在太长（即prompt太长），长到动用了gpu上所有的block（num_total_gpu_blocks）都无法处理它，所以后续步骤中我们会直接把这个seq标记为完成，不再处理它；而LATER是因为之前可能已经调度了很多seq_group，它们占据了相当一部分显存空间，导致gpu上剩余的可用block（num_free_gpu_blocks）无法再处理它，所以我们延迟处理。

4.5 can_append_slot：能否为seq_group分配物理块做decode

回顾4.2调度器的流程图，你会看到我们从running队列中调度seq_group时，我们也会判断是否能为该seq_group分配物理块。但这时，我们的物理块空间是用来做decode的（给每个seq分配1个token的位置），而不是用来做prefill的（给每个seq分配若干个token的位置），所以这里我们采取的是另一种判断方法can_append_slot。

更具体来说，running队列中seq_group下的n个seqs在上1个推理阶段共生成了n个token。在本次调度中，我们要先为这n个token分配物理块空间，用于存放它们在本次调度中即将产生的KV值。

好，再回到这个seq_group的n个seqs上来，我们知道：

当往1个seq的物理块上添加1个token时，可能有两种情况：之前的物理块满了，所以我新开1个物理块给它。之前的物理块没满，我直接添加在最后一个物理块的空槽位上。所以，对于1个seq来说，最坏的情况就是添加1个物理块；对于n个seqs来说，最坏的情况就是添加n个物理块(想想原理篇中讲过的copy-on-write机制)。

对于1个seq_group，除了那些标记为“finish”的seq外，其余seqs要么一起送去推理，要么一起不送去推理。即它们是集体行动的

所以，判断能否对一个正在running的seq_group继续做推理的最保守的方式，就是判断当前可用的物理块数量是否至少为n。

直接看代码（一切尽在注释中，编辑器有问题，大家看截图吧）：

4.6 allocate与append_slot：为seq_group分配物理块

当我们判断当前有充足的gpu KV Cache空间给对应的seq_group做新一轮推理时，我们就可以实际给它分配物理块了。这一块的内容涉及的细节太多（不同的prefix caching方式，逻辑块到物理块的映射，物理块释放，物理块的refcount即copy-on-write机制等等），所以我们把这部分留在源码解读3：块管理中来详细说明。

跳过这块并不影响大家对调度器策略的解读。、

4.7 preempt：抢占策略

纵观4.2的调度流程，现在我们只剩1个重点没讲了：抢占策略。

其实在2.2介绍SequenceGroup时，已经提到了抢占策略的核心逻辑，这里再复制一遍：

在若干个推理阶段后，gpu上的资源不够了，这个seq_group不幸被调度器抢占（preemption），它相关的KV block也被swap out到cpu上。此时所有seq的状态变为swapped。这里要注意，当一个seq_group被抢占时，对它的处理有两种方式：

Swap：如果该seq_group剩余生命周期中并行运行的最大seq数量 > 1，此时会采取swap策略，即把seq_group下【所有】seq的KV block从gpu上卸载到cpu上。（seq数量比较多，直接把算出的KV block抛弃，比较可惜）
Recomputation：如果该seq_group剩余生命周期中并行运行的最大seq数量 = 1，此时会采取recomputation策略，即把该seq_group相关的物理块都释放掉，然后将它重新放回waiting队列中(放在最前面)。等下次它被选中推理时，就是从prefill阶段开始重新推理了，因此被称为“重计算”。（seq数量少，重新计算KV block的成本不高）

对“最大生命周期…”这里有疑惑的朋友，回顾下本文2.3（1）。

我们直接来看代码（一切尽在注释中）

    # vllm/core/scheduler.py  def _preempt(  self,  seq_group: SequenceGroup, # 被抢占的seq_group  blocks_to_swap_out: Dict[int, int],  preemption_mode: Optional[PreemptionMode] = None,  ) -> None:  """  对被抢占的seq_group进行处理，包括修改其下seq状态，做好gpu到cpu块之间的映射等  """  # If preemption mode is not specified, we determine the mode as follows:  # We use recomputation by default since it incurs lower overhead than  # swapping. However, when the sequence group has multiple sequences  # (e.g., beam search), recomputation is not currently supported. In  # such a case, we use swapping instead.  # FIXME(woosuk): This makes our scheduling policy a bit bizarre.  # As swapped sequences are prioritized over waiting sequences,  # sequence groups with multiple sequences are implicitly prioritized  # over sequence groups with a single sequence.  # TODO(woosuk): Support recomputation for sequence groups with multiple  # sequences. This may require a more sophisticated CUDA kernel.  # 如果没有指定被抢占的类型  if preemption_mode is None:  # 如果这个seq_group在剩余生命周期中并行运行的最大seq数为1  if seq_group.get_max_num_running_seqs() == 1:  # 就将抢占类型定位“recompute”  preemption_mode = PreemptionMode.RECOMPUTE  # 否则定为swap  else:  preemption_mode = PreemptionMode.SWAP  # =======================================================================  # 如果抢占类型是“RECOMPUTE”  # 则去除该seq对对应物理块的引用，同时将该seq状态改为running，放入waiting队列最前面  # （详情参见self._preempt_by_recompute）  # =======================================================================  if preemption_mode == PreemptionMode.RECOMPUTE:  self._preempt_by_recompute(seq_group)  # =======================================================================  # 如果抢占类型是“SWAP“  # 详情参见self._preempt_by_swap）  # =======================================================================  elif preemption_mode == PreemptionMode.SWAP:  self._preempt_by_swap(seq_group, blocks_to_swap_out)  else:  raise AssertionError("Invalid preemption mode.")  def _preempt_by_recompute(  self,  seq_group: SequenceGroup,  ) -> None:  # 获取这个seq_group下正在running的所有seqs，  # preemption_mode是RECOMPUTE时需要满足正在running的seqs数量为1  seqs = seq_group.get_seqs(status=SequenceStatus.RUNNING)  assert len(seqs) == 1  for seq in seqs:  # 将这条seq的状态从running改成waiting（后续这条seq就要重计算了）  seq.status = SequenceStatus.WAITING  # 释放这条seq对应的物理块  # 即将对应物理块的引用-1，如果此时引用数量为0，说明对应物理块完全自由了，需要再将其放入自由物理块列表中  self.free_seq(seq)  # 因为这条seq需要重计算了，所以将其data对象下_num_computed_tokens设置为0  seq.reset_state_for_recompute()  # NOTE: For FCFS, we insert the preempted sequence group to the front  # of the waiting queue.  # 将被抢占，且未来需要重计算的序列，放到waiting队列的最前面  self.waiting.appendleft(seq_group)  def _preempt_by_swap(  self,  seq_group: SequenceGroup,  blocks_to_swap_out: Dict[int, int],  ) -> None:  # ======================================================================  # - 释放该seq_group下所有seq的物理块，并为其分配对应的cpu物理块，  # - 将seq的状态从running改成swapped  # ======================================================================  self._swap_out(seq_group, blocks_to_swap_out)  # ======================================================================  # 在scheduler的swapped队列中添加该seq_group  # ======================================================================  self.swapped.append(seq_group)  def _swap_out(  self,  seq_group: SequenceGroup, # 需要被swap到cpu上的seq_group  blocks_to_swap_out: Dict[int, int],  ) -> None:  # ======================================================================  # 检查是否可以将当前seq_group对应的物理块swap到cpu上  # 可以的条件：当前seq_group占用的gpu物理块数量 <= cpu上可用的物理块数量  # ======================================================================  if not self.block_manager.can_swap_out(seq_group):  # FIXME(woosuk): Abort the sequence group instead of aborting the  # entire engine.  raise RuntimeError(  "Aborted due to the lack of CPU swap space. Please increase "  "the swap space to avoid this error.")  # ======================================================================  # 释放该seq_group下所有seq的gpu物理块，并为其创建对应的cpu块  # mapping：{gpu物理块id：cpu物理块id}  # ======================================================================  mapping = self.block_manager.swap_out(seq_group)  blocks_to_swap_out.update(mapping)  # ======================================================================  # 修改该seq_group下所有seq的状态：从running改成swapped  # ======================================================================  for seq in seq_group.get_seqs(status=SequenceStatus.RUNNING):  seq.status = SequenceStatus.SWAPPED

额外说明的一点是，一旦你决定执行swap out操作，你就做做好gpu物理块->cpu物理块之间的映射，这样等之后你想swap in时，你才知道去cpu上的哪里把这些物理块找回来。

swap更多的细节也会涉及到blockmanager，所以遗留的细节，也放在第三篇中说。

4.8 调度器核心代码

有了以上的基础（真是庞大的逻辑），现在终于能来看调度器中关于一次调度策略的核心代码了，大家可以配合4.2流程图阅读，一切尽在注释中～

    # vllm/core/scheduler.py  def _schedule(self) -> SchedulerOutputs:  """  """  # ==============================================================================  # blocks_to_swap_in：{cpu物理块id: gpu物理块id}  # blocks_to_swap_out：{gpu物理块id: cpu物理块id}  # blocks_to_copy: {旧物理块id：[由旧物理块copy-on-write而来的新物理块id]}  # ==============================================================================  blocks_to_swap_in: Dict[int, int] = {}  blocks_to_swap_out: Dict[int, int] = {}  blocks_to_copy: Dict[int, List[int]] = {}  # ==============================================================================  # Fix the current time.  # 获取当下时间  # ==============================================================================  now = time.time()  # ==============================================================================  # Join waiting sequences if possible.  # 如果swapped队列为空  # ==============================================================================  if not self.swapped:  # ==========================================================================  # ignored_seq_groups：记录因太长（所需的blocks和总blocks之间的差值超过阈值了），  # 而无法继续做生成的seq_group，这些seq_group中的seq状态都会被标记为  # FINISHED_IGNORED，表示直接不处理他们  # ==========================================================================  ignored_seq_groups: List[SequenceGroup] = []  # ==========================================================================  # 记录本次被调度的seq_group  # ==========================================================================  scheduled: List[SequenceGroup] = []  # ==========================================================================  # The total number of sequences on the fly, including the  # requests in the generation phase.  # 计算Scheduler running队列中还没有执行完的seq数量  # ==========================================================================  num_curr_seqs = sum(seq_group.get_max_num_running_seqs()  for seq_group in self.running)  curr_loras = set(  seq_group.lora_int_id  for seq_group in self.running) if self.lora_enabled else None  # ==========================================================================  # Optimization: We do not sort the waiting queue since the preempted  # sequence groups are added to the front and the new sequence groups  # are added to the back.  # lora相关的，可以暂时不看  # ==========================================================================  leftover_waiting_sequences = deque()  # ==========================================================================  # 本次调度处理的token总数  # ==========================================================================  num_batched_tokens = 0  # ==========================================================================  # 开启新一次调度（while循环不结束意味着本次调度不结束，  # 跳出while循环时意味着本次调度结束了）  # 开启新一次调度的条件：当waiting队列中有等待处理的请求，且当前时刻可以处理请求  # ==========================================================================  while self._passed_delay(now) and self.waiting:  # =====================================================================  # 取出waiting队列中的第一个请求，也即最早到达的请求（seq_group）  # =====================================================================  seq_group = self.waiting[0]  # =====================================================================  # 统计该seq_group中s处于waiting的seq的数量  # =====================================================================  waiting_seqs = seq_group.get_seqs(  status=SequenceStatus.WAITING)  # =====================================================================  # 从waiting队列中取出来的seq_group，其seq数量一定是1  # =====================================================================  assert len(waiting_seqs) == 1, (  "Waiting sequence group should have only one prompt "  "sequence.")  # =====================================================================  # 获取该seq的序列长度（如果该seq_group来自之前被抢占的请求，  # 那么这个长度不仅包括prompt，  # 还包括output）  # =====================================================================                   num_prefill_tokens = waiting_seqs[0].get_len()  # =====================================================================  # 如果从waiting队列中取出的这条seq的长度 > 每次调度能处理的最大序列长度，  # 那么就打印警告信息，同时把这条seq的状态置为FINISHED_IGNORED，  # 并将对应seq_group装入ignored_seq_groups中，  # 然后将其从waiting列表中移除，不再处理  # =====================================================================  if num_prefill_tokens > self.prompt_limit:  logger.warning(  f"Input prompt ({num_prefill_tokens} tokens) is too "  f"long and exceeds limit of {self.prompt_limit}")  for seq in waiting_seqs:  seq.status = SequenceStatus.FINISHED_IGNORED  ignored_seq_groups.append(seq_group)  self.waiting.popleft()  continue  # =====================================================================  # If the sequence group cannot be allocated, stop.  # 决定是否能给当前seq_group分配物理块  # can_allocate返回值可能有三种：  #       AllocStatus.NEVER：不分配；  #       AllocStatus.OK：可以分配；  #       AllocStatus.LATER：延迟分配  # =====================================================================  can_allocate = self.block_manager.can_allocate(seq_group)  # 若是延迟分配，则说明现在没有足够的block空间，所以跳出while循环（不继续对waiting队列中的数据做处理了）  if can_allocate == AllocStatus.LATER:  break  # 如果不分配，说明seq长得超出了vLLM的处理范围，则后续也不再处理它，直接将该seq状态标记为FINISHED_IGNORED  elif can_allocate == AllocStatus.NEVER:  logger.warning(  f"Input prompt ({num_prefill_tokens} tokens) is too "  f"long and exceeds the capacity of block_manager")  for seq in waiting_seqs:  seq.status = SequenceStatus.FINISHED_IGNORED  ignored_seq_groups.append(seq_group) # 记录因为太长而无法处理的seq_group  self.waiting.popleft() # 将该seq_group从waiting队列中移除  continue  # =====================================================================                 # lora推理相关的部分，可暂时忽略  # =====================================================================  lora_int_id = 0  if self.lora_enabled:  lora_int_id = seq_group.lora_int_id  if (lora_int_id > 0 and lora_int_id not in curr_loras  and len(curr_loras) >= self.lora_config.max_loras):  # We don't have a space for another LoRA, so  # we ignore this request for now.  leftover_waiting_sequences.appendleft(seq_group)  self.waiting.popleft()  continue  # =====================================================================  # If the number of batched tokens exceeds the limit, stop.  # max_num_batched_tokens：单次调度中最多处理的token数量  # num_batched_tokens：本次调度中累计处理的token数量  # 如果后者 > 前者，则结束本次调度  # =====================================================================  num_batched_tokens += num_prefill_tokens  if (num_batched_tokens >  self.scheduler_config.max_num_batched_tokens):  break  # =====================================================================  # The total number of sequences in the RUNNING state should not  # exceed the maximum number of sequences.  # num_new_seqs: 当前seq_group中状态为“未执行完”的序列的数量  # num_curr_seqs：当前调度轮次中，状态为"未执行完“的序列总数  # 如果超过了我们对单次调度能执行的序列总数的阈值，就结束本次调度  # =====================================================================  num_new_seqs = seq_group.get_max_num_running_seqs()  if (num_curr_seqs + num_new_seqs >  self.scheduler_config.max_num_seqs): # 单次迭代中最多处理多少个序列  break  if lora_int_id > 0:  curr_loras.add(lora_int_id)  # =====================================================================  # 走到这一步时，说明当前seq_group已经通过上述种种验证，可以被加入本次调度中执行了  # 先将其从waiting队列中移出  # =====================================================================  self.waiting.popleft()  # =====================================================================  # 为当前seq_group分配物理块  # =====================================================================  self._allocate(seq_group)  # =====================================================================  # 将当前seq_group放入running队列中  # =====================================================================  self.running.append(seq_group)  # =====================================================================  # 记录本次调度累计处理的序列数量  # =====================================================================  num_curr_seqs += num_new_seqs  # =====================================================================  # 记录本次被调度的seq_group  # =====================================================================  scheduled.append(  ScheduledSequenceGroup(  seq_group=seq_group,  token_chunk_size=num_prefill_tokens))  # =====================================================================  # 和lora相关的操作，暂时忽略  # =====================================================================  self.waiting.extendleft(leftover_waiting_sequences)  # =====================================================================  # 如果本次有被调度的seq_group（scheduled非空）  # 或者本次有被设置为不再处理的seq_group(ignored_seq_groups非空)  # 就将其包装成SchedulerOutputs对象  # =====================================================================  if scheduled or ignored_seq_groups:  self.prev_prompt = True  scheduler_outputs = SchedulerOutputs(  scheduled_seq_groups=scheduled,  prompt_run=True,  num_batched_tokens=num_batched_tokens,  blocks_to_swap_in=blocks_to_swap_in,  blocks_to_swap_out=blocks_to_swap_out,  blocks_to_copy=blocks_to_copy,  ignored_seq_groups=ignored_seq_groups,  )  return scheduler_outputs  # ==============================================================================  # NOTE(woosuk): Preemption happens only when there is no available slot  # to keep all the sequence groups in the RUNNING state.  # In this case, the policy is responsible for deciding which sequence  # groups to preempt.  # 如果swap队列非空，且本次没有新的需要被发起推理的seq_group,  # 则对running队列中的seq_group，  # 按照 "当前时间-该seq_group到达时间" ，从早到晚排列running队列中的seq_group  # ==============================================================================  self.running = self.policy.sort_by_priority(now, self.running)  # ==============================================================================  # Reserve new token slots for the running sequence groups.  # 初始化一个新的running队列（deque（））  # 初始化一个抢占列表  # ==============================================================================  running: Deque[SequenceGroup] = deque()  preempted: List[SequenceGroup] = []  # ==============================================================================  # 当running队列非空时  # ==============================================================================  while self.running:  # 取出running队列中最早到来的seq_group  seq_group = self.running.popleft()  # =====================================================================  # 对于running队列中这个最早到来的seq_group，检查对于其中的每一个seq，  # 是否能至少分配一个物理块给它，如果不能的话  # （说明要执行抢占操作了，否则马上会没有资源让这个最早到达的seq_group做完推理）：  # （注意，这里用了while...else，如果while条件正常结束，则进入else内容；  #  如果被break，则不会执行else）  # =====================================================================  while not self.block_manager.can_append_slot(seq_group):  # =====================================================================  # 如果从running队列中取出最早达到的seq_group后，running队列还是非空  # =====================================================================  if self.running:  # ==============================================================  # 抢占running队列中最晚到来的seq_group（可怜的被害者）  # ==============================================================  victim_seq_group = self.running.pop()  # ==============================================================  # 一个seq_group被抢占后，有2中处理方式：  # - 如果该seq_group下只有一个seq，执行【重计算】，  #   将其从running队列中移除，并清空它的物理块，  #   将其seq的状态从running->waiting，并加入waiting队列。后面将重新计算  #  # - 如果该seq_group下有多个seq，执行【swap】，  #   清空它的gpu物理块，并为这些物理块做好cpu物理块映射，  #   这些seq的block_table字典中（{seq_id: block_table}）的block_table  #   从gpu物理块改成cpu物理块  #   将其seqs状态从running -> swapped，加入swapped队列  # ==============================================================  self._preempt(victim_seq_group, blocks_to_swap_out)  preempted.append(victim_seq_group)  # ==============================================================  # 如果除这个最早到来的seq_group外，running队列中再没有别的seq_group了，  # 且此时又没有足够的空间留给这个最早来的seq_group做推理了，那么只能抢占它  # ==============================================================  else:  # 那就只能抢占这个最早到达的seq_group了  # No other sequence groups can be preempted.  # Preempt the current sequence group.  self._preempt(seq_group, blocks_to_swap_out)  preempted.append(seq_group)  break  # ==============================================================  # 如果此时有足够的空间给running队列中最早来的seq_group做推理了  # ==============================================================  else:  # ==============================================================  #  Append new slots to the sequence group.  # seq_group里的每个seq正常做推理。假设现在每个seq正常生成一个token，我们需要根据每个seq当前  # 维护的最后一个物理块的情况，决定是否需要分配新的物理块，决定的结果可能如下：  # - 物理块refcount = 1，且有充足槽位，则无需分配新物理块  # - 物理块refcount = 1，且无充足槽位，分配新的物理块  # - 物理块refcount > 1, 采用copy-on-write机制，分配新物理块，对该seq，  #                      用新物理块替换掉其block_table中维护的最后一个物理块  #                     （称为旧物理块）。释放旧物理块（令其refcount-1）。  #                      同时记录下新旧物理块之间的映射，  #   blocks_to_copy：{旧物理块id：[由旧物理块copy-on-write而来的新物理块id]}  # ==============================================================  self._append_slot(seq_group, blocks_to_copy)  # ==============================================================  # 自定义的running队列中添加这个seq_group  # ==============================================================  running.append(seq_group)  # ==============================================================================  # 最终还能在running队列中运行的seq_group  # ==============================================================================  self.running = running  # ==============================================================================  # Swap in the sequence groups in the SWAPPED state if possible.  # 对于swapped队列中的seq_group，按照到达时间从早到晚排序  # ==============================================================================  self.swapped = self.policy.sort_by_priority(now, self.swapped)  # ==============================================================================  # 如果本次调度没有新安排的被抢占的seq_group(即preempted为空)  # ==============================================================================  if not preempted:  # ==============================================================  # 计算running队列中，所有seq_group下，“到生命周期结束为止最多运行的seq数量”的总和  # ==============================================================  num_curr_seqs = sum(seq_group.get_max_num_running_seqs()  for seq_group in self.running)  # ==============================================================  # lora部分，暂时忽略  # ==============================================================  curr_loras = set(  seq_group.lora_int_id  for seq_group in self.running) if self.lora_enabled else None  # ==============================================================  # lora相关的，可以暂时不看  # ==============================================================  leftover_swapped = deque()  # ==============================================================  # 当swapped队列非空时  # ==============================================================  while self.swapped:  # ==============================================================  # 取出swap队列中最早被抢占的seq_group  # ==============================================================  seq_group = self.swapped[0]  # ==============================================================  # lora相关，暂时不看  # ==============================================================  lora_int_id = 0  if self.lora_enabled:  lora_int_id = seq_group.lora_int_id  if (lora_int_id > 0 and lora_int_id not in curr_loras  and len(curr_loras) >= self.lora_config.max_loras):  # We don't have a space for another LoRA, so  # we ignore this request for now.  leftover_swapped.appendleft(seq_group)  self.swapped.popleft()  continue  # ==============================================================  # If the sequence group cannot be swapped in, stop.  # 判断一个被swap的seq_group现在是否能重新running起来  # 【判断条件】：  # 当前gpu上可用的物理块数量 - 重新跑起这个seq_group需要的物理块数量   # >= 水位线物理块数量  # 其中：  # 后者 = 在被swap之前它已经使用的物理块数量（去重过了）   #       + 若能再次跑起来它至少需要的物理块数量  #（假设每个seq至少需要1个物理块）  # ==============================================================  # 如果不能，则意味着当前没有充足资源处理swap队列中的seq_group，则直接跳出循环  if not self.block_manager.can_swap_in(seq_group):  break  # ==============================================================  # The total number of sequences in the RUNNING state should not  # exceed the maximum number of sequences.  # 如果对于swap队列中的这个seq_group，当前gpu上有充足资源可以让它重新跑起来的话：  # ==============================================================  # 取出这个seq_group在剩余生命周期内将并行运行的最大序列数  num_new_seqs = seq_group.get_max_num_running_seqs()  # 如果已超过一次调度中能处理的最大序列数，则不再对该seq_group进行处理  if (num_curr_seqs + num_new_seqs >  self.scheduler_config.max_num_seqs):  break  # lora部分暂时不看  if lora_int_id > 0:  curr_loras.add(lora_int_id)  # ==============================================================  # 走到这一步，说明可以对swapped队列中的这个seq_group做相关处理了，  # 先把它从队列中移出去  # ==============================================================  self.swapped.popleft()  # ==============================================================  # 将该seq_group下所有cpu块置换回gpu块，  # 并将其下每个seq的状态从swapped改成running  # ==============================================================  self._swap_in(seq_group, blocks_to_swap_in)  # ==============================================================  # 假设其正常做推理了，假设现在生成了一个token，要如何分配物理块（参见上面注释）  # ==============================================================  self._append_slot(seq_group, blocks_to_copy)  num_curr_seqs += num_new_seqs  self.running.append(seq_group)  self.swapped.extendleft(leftover_swapped)  # ==============================================================================  # 如果本次调度有新安排的被抢占的seq_group(即preempted不为空)，那就准备将最终的running队列  # 作为scheduleroutputs返回  # ==============================================================================  # Each sequence in the generation phase only takes one token slot.  # Therefore, the number of batched tokens is equal to the number of  # sequences in the RUNNING state.  # 由于每个seq一次只生成1个token，因此num_batched_tokens = 状态为running的seq数量  num_batched_tokens = sum(  seq_group.num_seqs(status=SequenceStatus.RUNNING)  for seq_group in self.running)  # ==============================================================================  # 构建Schduleroutputs  # ==============================================================================  scheduler_outputs = SchedulerOutputs(  scheduled_seq_groups=[  ScheduledSequenceGroup(seq_group=running_group,  token_chunk_size=1)  for running_group in self.running  ],  prompt_run=False,  num_batched_tokens=num_batched_tokens,  blocks_to_swap_in=blocks_to_swap_in,  blocks_to_swap_out=blocks_to_swap_out,  blocks_to_copy=blocks_to_copy,  ignored_seq_groups=[],  )  return scheduler_outputs

在本文中，我们：

#从vLLM批处理的入口函数开始，介绍了其推理内核LLMEngine的两个重要函数add_request()和step()

#在LLMEngine开始处理请求前（实例化阶段），它会先做一次模拟实验，来估计gpu上需要预留多少显存给KV Cache block。

#当LLMEngine开始处理请求时(add_request)，它会把每个prompt当成一个请求，同时把它包装成一个SequenceGroup对象。

#当LLMEngine开始执行1次调度时（step），调度器策略(Scheduler)会根据实际gpu上KV Cache block的使用情况等要素，来选择要送哪些seq_group去做新一轮推理。注意，在1次推理中，所有seq_group要么一起做prefill，要么一起做decode。

到目前为止，遗留以下问题：

vLLM的物理块管理（block manager）的细节，包括物理块结构，逻辑块-物理块映射，物理块新增与释放，prefix caching等等

step()其余步骤：调度器只是决定了要送哪些seq_group去做推理，但是“每1个推理阶段结束后，如何根据output更新seq_group，将其送入下一次调度”这块不是调度器的职责，也是本文没涉及到的。

将在本系列后续的文章中，对块管理做详细讲解。在这之后，会分别以parallel sampling和beam search这两种decode方式为例，把整个流程传一遍，一起来更好理解vLLM背后的运作逻辑。