MediaPipe 实时流机制详解：时间戳边界、实时调度与 Calculator 生命周期

MediaPipe 的计算图常用于处理交互式应用中的视频/音频实时流。本篇围绕 MediaPipe 框架概念文档 Real-time Streams 展开，讲清实时时间戳的约定、默认调度时机、"时间戳边界（timestamp bound）"如何解决下游节点阻塞问题、三种边界传播工具（SetNextTimestampBound、SetTimestampOffset、SetProcessTimestampBounds）的用法与差异，以及 Calculator::Open/Calculator::Close 的调度规则与 TimestampOffset 带来的收尾限制。读完后，你将能够编写符合实时图要求的 Calculator，避免实时应用中常见的"下游永远等不到上游"的队列积压问题。

实时时间戳：单调递增的微秒时间

MediaPipe 计算图常被用于为交互式应用处理视频或音频帧流。框架对流中时间戳的硬性要求只有一条：同一包流上的连续 Packet 必须分配单调递增的时间戳。在此基础上，实时 Calculator 与图约定俗成地以每帧的录制时间（recording time）或显示时间（presentation time）作为时间戳，时间戳表示自 Jan/1/1970:00:00:00 起的微秒数。这一约定使得来自不同源（摄像头、屏幕、文件解码器等）的 Packet 能够以全局一致的顺序被处理。

从源码可以印证这一约定。timestamp.h 的头部注释明确写道："MediaPipe timestamps are in units of microseconds"（MediaPipe 时间戳单位为微秒），并定义了 Timestamp 类的一组特殊值：

• Unset()：默认初始化值，一般不合法；
• Unstarted()：任何合法时间戳之前的值，即 Open() 调用时的输入时间戳；
• PreStream() / PostStream()：表示流前"头数据"与流级"总结"Packet，若发送则必须是该流上唯一的此类 Packet；
• Min() / Max()：Process() 中可见的最小/最大区间时间戳；
• Done()：所有合法时间戳之后的值，即 Close() 调用时的输入时间戳。

Timestamp 还提供 Microseconds()、Milliseconds()、Seconds()、FromSeconds()、FromMilliseconds()、FromMicroseconds() 等转换接口（见 timestamp.h）。需要注意：区间值之间的算术运算会自动钳制在 [Min(), Max()] 范围内，且不能用构造函数直接创建特殊值（会触发 ABSL_CHECK），必须通过上述静态方法创建。

实时调度：何时触发 Calculator 执行

默认情况下，每个 Calculator 在其某个时间戳的全部输入 Packet 齐备时立即运行。这通常发生在两个条件同时满足时：

1. 该 Calculator 已完成上一个帧的处理；
2. 为它产生输入的每一条流上的 Calculator 都已完成当前帧的处理。

MediaPipe 调度器在条件满足后立刻调用该 Calculator。这一机制建立在去中心化的执行模型之上：图中没有全局时钟，不同节点可以同时处理不同时间戳的数据（流水线化），但时间戳在此充当同步键。详细的调度机制（就绪判定、调度队列、执行器）见同目录的 Synchronization。

这里的关键推论是：如果某条输入流在时间戳 T 上既没有 Packet 也没有任何"不会再有 Packet 到达"的信号，下游节点就永远无法确认 T 时刻的输入已完整，也就永远不会被调度——这正是下面"时间戳边界"要解决的问题。

时间戳边界（Timestamp Bounds）：让下游知道"不会再来了"

当一个 Calculator 对某个时间戳不产生任何输出 Packet 时，它可以转而输出一个 "timestamp bound"（时间戳边界），声明"该时间戳上不会有 Packet 产生"。这个声明必不可少：它允许下游 Calculator 在该时间戳上照常运行，即使某些流在该时间戳上没有 Packet。对于交互式实时图，这一点尤为关键——每个 Calculator 必须尽早开始处理，任何无谓的等待都会直接体现为用户可感知的延迟。

原文档给出的经典反例：

node {
   calculator: "A"
   input_stream: "alpha_in"
   output_stream: "alpha"
}
node {
   calculator: "B"
   input_stream: "alpha"
   input_stream: "foo"
   output_stream: "beta"
}

假设在时间戳 T 上，节点 A 没有在输出流 alpha 上发送 Packet。节点 B 在 foo 上收到了 T 的 Packet，正在等待 alpha 上 T 的 Packet。如果 A 没有为 alpha 发送时间戳边界更新，B 就会一直等待 alpha 上的 Packet 到达；与此同时，foo 的 Packet 队列会不断积压 T、T+1 等后续时间戳的 Packet——队列膨胀、延迟上升，直到可能触发流控丢弃（见 Synchronization 文档 中关于 max_queue_size 背压与 FlowLimiterCalculator 丢包的说明）。

每个流都维护一个时间戳边界：该流上允许出现新 Packet 的最低可能时间戳。当一个带时间戳 T 的 Packet 到达时，边界自动前进到 T+1，反映单调性要求，从而让框架可以确定不会再有更早时间戳的 Packet 到来。相应地，流的生产者也可以显式地把边界推进得比最后一个 Packet 所暗示的更靠前，提供"更紧的边界"，让下游更早地确定其输入。

显式输出时间戳边界的 API

Calculator 通过 CalculatorContext::Outputs() 和 OutputStream::Add 在流上输出 Packet；而要改为输出时间戳边界，则使用 CalculatorContext::Outputs() 与 CalculatorContext::SetNextTimestampBound。所给定的边界就是该输出流上下一个 Packet 允许的最低时间戳。当不输出任何 Packet 时，典型的写法是：

cc->Outputs().Tag("output_frame").SetNextTimestampBound(
  cc->InputTimestamp().NextAllowedInStream());

其中 Timestamp::NextAllowedInStream() 返回该流上的下一个合法时间戳。例如 Timestamp(1).NextAllowedInStream() == Timestamp(2)。该函数声明于 timestamp.h，其实现见 timestamp.cc；若当前时间戳之后已不允许再放 Packet（如处于 PostStream），它返回 OneOverPostStream()。

三种时间戳边界传播工具

要用于实时图的 Calculator，需要基于输入边界推导输出边界，以便下游节点能被及时调度。

常见模式：输出时间戳与输入保持一致

一个非常常见的模式是：Calculator 输出的 Packet 与输入 Packet 使用相同的时间戳。此时只需在每次 Calculator::Process 调用中都输出 Packet，就足以隐式地推进输出边界（每次 Add 都会把边界带到 T+1），无需额外工作。

但 Calculator 并不被要求遵循该模式——只要求输出时间戳单调递增。因此某些 Calculator 必须显式计算时间戳边界。MediaPipe 提供三类工具：

1. SetNextTimestampBound()：手工指定边界

用于为输出流指定时间戳边界 t + 1：

cc->Outputs.Tag("OUT").SetNextTimestampBound(t.NextAllowedInStream());

等价做法是产出一个时间戳为 t 的空 Packet（Packet()），同样表示边界推进到 t + 1：

cc->Outputs.Tag("OUT").Add(Packet(), t);

输入流上的边界则由该输入流上"最近收到的 Packet 或空 Packet"的时间戳读出：

Timestamp bound = cc->Inputs().Tag("IN").Value().Timestamp();

2. SetTimestampOffset()：自动复制输入边界

cc->SetTimestampOffset(0);

该设置会自动把输入流上的时间戳边界偏移一个固定量后复制到输出流。它的优势在于：即使只有边界到达、Calculator::Process 从未被调用，边界也会被自动传播。在合同层面，CalculatorContract::SetTimestampOffset 接受 TimestampDiff 偏移量（见 calculator_contract.h 中 SetProcessTimestampBounds 与 SetTimestampOffset 的注释：要求自定义边界计算的 Calculator 应改用 SetProcessTimestampBounds）。

3. SetProcessTimestampBounds(true)：在"settled 时间戳"上调用 Process

cc->SetProcessTimestampBounds(true);

开启后，每当出现一个新的 settled 时间戳（settled timestamp 定义为：低于当前时间戳边界的最新最高时间戳，即该时间戳在各输入流上的状态已不可逆地确定——要么有 Packet，要么确定不会有），框架就会调用一次 Calculator::Process；否则（默认情况下）Process 只在一个或多个 Packet 实际到达时才被调用。

这一设置让 Calculator 可以自行计算并传播边界，即使只有输入边界被更新。它既能复制 SetTimestampOffset(0) 的效果，也能计算考虑了额外因素的边界。用代码等价实现 SetTimestampOffset(0) 的示例（来自原文档）：

absl::Status Open(CalculatorContext* cc) {
  cc->SetProcessTimestampBounds(true);
}

absl::Status Process(CalculatorContext* cc) {
  cc->Outputs.Tag("OUT").SetNextTimestampBound(
      cc->InputTimestamp().NextAllowedInStream());
}

三种方式的选择可以概括为：输出与输入同时间戳 → 什么都不用做，每次 Process 输出即可；边界 = 输入边界 + 固定偏移 → 用 SetTimestampOffset（最省心，Process 不被调用也能传播）；边界计算复杂、依赖运行时状态 → 用 SetProcessTimestampBounds(true) 加手工 SetNextTimestampBound。

Calculator::Open 与 Calculator::Close 的调度时机

Calculator::Open 在所有必需的输入 side-packet 都已被产生之后调用。side-packet 可以由外部应用提供，也可以由图内部的"side-packet calculator"产生：

• 从图外部指定：使用 API 的 CalculatorGraph::Initialize 和 CalculatorGraph::StartRun 传入 side-packet；
• 由图内 Calculator 产生：使用 CalculatorGraphConfig::OutputSidePackets 与 OutputSidePacket::Set。

（side-packet 携带单个时间戳不确定的 Packet，用于传递整段生命周期内保持不变的配置数据，如 框架概念总览 所述。）

Calculator::Close 在所有输入流都变为 Done（被关闭或时间戳边界达到 Timestamp::Done）时调用。原文档特别强调：如果图先完成了所有待执行的 Calculator、整体进入 Done 状态，而某些流尚未 Done，MediaPipe 仍会调用其余的 Calculator::Close，保证每个 Calculator 都有机会产出最终输出。

TimestampOffset 对 Close 的影响：为什么不能在 Close 里发 summary Packet

这是一个容易踩坑的收尾细节：声明了 SetTimestampOffset(0) 的 Calculator，按设计会在所有输入流都达到 Timestamp::Done 时令其所有输出流也达到 Timestamp::Done——即"不会再有任何输出可能"。因此这样的 Calculator 无法在 Calculator::Close 期间再发出任何 Packet。

如果 Calculator 需要在 Close 阶段产出一个 summary Packet（例如统计整个流的汇总结果），其 Process 必须把时间戳边界推到至少还留有一个可用时间戳（如 Timestamp::Max）的程度。这意味着此类 Calculator 通常不能依赖 SetTimestampOffset(0)，而必须用 SetNextTimestampBound() 显式指定边界，把 Max 留出来给 Close 阶段使用。这与 timestamp.h 注释中 PostStream() 语义（"该总结时间戳出现在所有区间时间戳之后，且若发送则必须是流上唯一的 Packet"）相一致。

小结与实操要点

• 时间戳 = 微秒，惯例取自帧的录制/显示时间，唯一硬约束是流内单调递增；
• 下游阻塞的根因是上游"静默跳过"某时间戳而不发边界；边界信号让调度器确认 settled 时间戳，从而及时触发下游；
• 输出边界三件套：每次 Process 都输出（同时间戳模式）、SetNextTimestampBound / 空 Packet（显式推进）、SetTimestampOffset(0)（自动复制，Process 不被调用也能传播）、SetProcessTimestampBounds(true)（自定义计算）；
• Open 等待全部 side-packet，Close 等待输入流全部 Done，但图整体 Done 时框架会补齐剩余 Close 调用；
• 需要在 Close 发 summary Packet 的 Calculator 不能依赖 SetTimestampOffset(0)，要显式管理边界、预留 Timestamp::Max。

想进一步了解调度器如何判定节点就绪、settled 时间戳在默认输入策略中的作用、以及 max_queue_size 背压与 FlowLimiterCalculator 丢包策略，可继续阅读 Synchronization；Packet 与时间戳的数据结构细节参见 packets.md、timestamp.h 与 timestamp.cc。