流控与拥塞控制

1. 为什么需要流控？

Aeron 基于 UDP，发送端不知道接收端的处理能力。如果没有流控：

接收端溢出：发送速度超过处理速度，数据堆积丢失
网络拥塞：发送速度超过链路容量，引发大量丢包
内存爆满：接收端重建 buffer 溢出

Aeron 的流控是多层级的：

┌─────────────────────────────────────────────────────┐
│                  发送端流控                           │
│  ┌───────────────────────────────────────────────┐  │
│  │  Publisher Limit (Term Buffer)                │  │
│  │  ← 发布者不能超过 term buffer 中最慢消费者的位置  │  │
│  │  ← 防止覆盖 term buffer 中尚未消费的数据         │  │
│  └───────────────────────────────────────────────┘  │
│  ┌───────────────────────────────────────────────┐  │
│  │  Sender Limit (流控策略)                       │  │
│  │  ← FlowControl 计算允许发送的最大 position       │  │
│  │  ← Min/Max/Tagged 策略                         │  │
│  └───────────────────────────────────────────────┘  │
│  ┌───────────────────────────────────────────────┐  │
│  │  Congestion Control (拥塞窗口)                 │  │
│  │  ← Receiver Window = min(流控limit, 拥塞窗口)   │  │
│  └───────────────────────────────────────────────┘  │
├─────────────────────────────────────────────────────┤
│                  接收端流控                           │
│  ┌───────────────────────────────────────────────┐  │
│  │  Receiver Window (Status Message 中报告)        │  │
│  │  ← 接收端能接受的最大数据量（字节）               │  │
│  │  ← 发送端不能超过 receiver position + window     │  │
│  └───────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────┘

2. FlowControl 策略

FlowControl 定义在 FlowControl.java 接口中。

2.1 MinMulticastFlowControl（默认）

MinMulticastFlowControl.java — 多播场景下以最慢的消费者为基准：

senderLimit = min(所有 receiver 的 position + window)

保证最慢的消费者不会丢失数据
快消费者不受慢消费者影响（数据仍然到达，只是 sender 不会超前太多）
URI 参数：fc=min

2.2 MaxMulticastFlowControl

MaxMulticastFlowControl.java — 以最快的消费者为基准：

senderLimit = max(所有 receiver 的 position + window)

慢消费者将丢失数据（接收端会丢弃超出接收窗口的数据）
适合 NAK 可以补上的场景
URI 参数：fc=max

2.3 TaggedMulticastFlowControl

TaggedMulticastFlowControl.java — 选择性跟踪特定接收者的位置：

senderLimit = min(有特定 tag 的 receiver 的 position + window)

只关心打了 tag 的接收者
URI 参数：fc=tagged

2.4 流控计算流程

在 AbstractMinMulticastFlowControl.java 中：

onStatusMessage(flyweight, receiverAddress, senderLimit, ...)
  ├── 解析 Status Message
  │   ├── consumptionTermId + termOffset → receiverPosition
  │   └── receiverWindow
  ├── 更新 receiver 的 tracked position
  ├── 计算新 senderLimit = min(allReceiverPositions) + window
  ├── 如果 senderLimit 增加 → 可能允许更多发送
  └── 返回更新后的 senderLimit

3. Congestion Control：拥塞控制

CongestionControl.java 接口定义了拥塞控制策略：

CongestionControl.onTrackRebuild(
    newStatusMessagePositionNs,   ← 当前 SM 时间戳
    oldStatusMessagePositionNs,   ← 上次 SM 时间戳
    newStatusMessagePosition,     ← 当前消费 position
    oldStatusMessagePosition,     ← 上次消费 position
    initialTermId,
    positionBitsToShift,
    receiverWindowLength,         ← 当前接收窗口
    timeNs
)
→ return (receiverWindowLength | forceStatusMessage flag)

3.1 StaticWindowCongestionControl

StaticWindowCongestionControl.java — 固定窗口：

receiverWindowLength = constant  ← 不变化

最简单的实现，生产环境默认
适合网络条件稳定的场景

3.2 CubicCongestionControl 和其他

Aeron 支持多种拥塞控制算法（通过插件机制）：

静态窗口
Cubic-like 动态调整
自定义实现

4. Loss Detection & NAK

4.1 LossDetector

LossDetector.java 检测消息丢失：

LossDetector.scan()
  ├── TermGapScanner.scanForGap(termBuffer, ...)
  │   ├── 遍历 term buffer 中的 frame
  │   ├── 检测 sequence gap（termOffset 跳跃）
  │   └── 调用 gapHandler.onGap(termId, termOffset, length)
  ├── 首次检测到的 gap → 记录 activeGap
  ├── 延迟后仍未修复的 gap → 通过 LossHandler 发送 NAK
  └── gap 被修复后 → 清除

延迟发送 NAK 的机制由 FeedbackDelayGenerator.java 控制——避免因乱序到达而非真正丢失导致的误报。

4.2 NAK 发送

在 Receiver.java 中，LossDetector 检测到的 gap 会通过 Status Message 中的 NAK 字段报告给 Sender：

Receiver.doWork()
  └── 对每个 PublicationImage:
      ├── LossDetector.scan() → gap 检测
      ├── 组装 Status Message
      │   ├── consumptionPosition
      │   ├── receiverWindow
      │   └── NAK list (如有 gap)
      └── UDP send() Status Message

4.3 NAK 处理与重传

在 Sender.java 中：

Sender.doWork()
  └── 收到 NAK:
      └── RetransmitHandler
          ├── 从 term buffer 找到对应 frame (termId, termOffset)
          ├── 组装 DataHeader
          ├── UDP send() 重传
          └── 更新 sender statistics

5. 端到端流控流程

Publisher               Sender                  Receiver              Subscriber
  │                       │                       │                     │
  │─ offer(msg) ────────→│                       │                     │
  │  写入 term buffer      │                       │                     │
  │                       │─ Data(term=X,off=Y)─→│                     │
  │                       │                       │─ 写入 Image buffer   │
  │                       │                       │─ subscriber.poll() ─→│
  │                       │                       │  ← 返回 fragment     │
  │                       │←─ Status(pos, win)───│                     │
  │                       │  (含 NAK 如有gap)     │                     │
  │                       │                       │                     │
  │  ← BACK_PRESSURED ───│                       │                     │
  │  (如果 publisher pos  │                       │                     │
  │   超过 senderLimit)   │                       │                     │
  │                       │                       │                     │
  │─ retry offer() ─────→│                       │                     │
  │  (等待消费者推进)      │                       │                     │

6. 关键设计决策

决策	实现
多级流控	Publisher Limit + Sender Limit + Congestion Window
Min Multicast 默认	保证最慢消费者不丢数据
NAK 延迟发送	FeedbackDelayGenerator 避免乱序误报
缓存行 Padding	流控对象使用 LHS/RHS padding 消除伪共享
插件化	FlowControl、CongestionControl 通过 URI 参数选择

7. 关键源文件索引

文件	说明
FlowControl.java	流控策略接口
MinMulticastFlowControl.java	最小多播流控
MaxMulticastFlowControl.java	最大多播流控
TaggedMulticastFlowControl.java	标签多播流控
AbstractMinMulticastFlowControl.java	流控基类实现
CongestionControl.java	拥塞控制接口
StaticWindowCongestionControl.java	固定窗口拥塞控制
LossDetector.java	丢包检测
LossHandler.java	NAK 处理回调
RetransmitHandler.java	重传处理