Network stack and Netty Shuffle

The source notes for this track are currently maintained in Chinese.

预计阅读时间: 90 分钟 前置阅读: doc-02(Shuffle 简介), doc-08(反压) 下一次阅读: doc-10(TM 内存)

1. 网络栈总体架构

1.1 分层模型

┌─────────────────────────────────────────────────────────────────┐
│                    RecordWriter / InputGate                    │
│              (StreamTask 调用层，生产者-消费者 API)             │
├─────────────────────────────────────────────────────────────────┤
│              Serializer / Deserializer                          │
│              (StreamRecord → ByteBuffer / 反向)                 │
├─────────────────────────────────────────────────────────────────┤
│         ResultPartition / ResultSubpartition                    │
│              (Producer: 数据产出与管理)                          │
│         InputGate / InputChannel                                │
│              (Consumer: 数据消费与管理)                          │
├─────────────────────────────────────────────────────────────────┤
│            NetworkBuffer / BufferPool                           │
│              (内存管理: 32KB 页, 池化, 信用额度)                 │
├─────────────────────────────────────────────────────────────────┤
│              Netty (TCP 传输层)                                  │
│      NettyServer (接收连接) + NettyClient (发起连接)            │
│      PartitionRequestQueue (生产者端调度)                        │
│      CreditBasedPartitionRequestClientHandler (消费者端流控)     │
└─────────────────────────────────────────────────────────────────┘

1.2 数据格式

网络栈的快照/数据格式（从 RecordWriter 到 InputGate）：

RecordWriter 序列化格式 (RecordSerializer):
┌────────────────────────────────────────────────────────────────┐
│ Length (4B) │ Type (1B) │ Timestamp (8B) │ Value (Variable)    │
│             │ Record=0   │                 │                    │
│             │ Watermark=1│                 │                    │
│             │ Barrier=2  │                 │                    │
│             │ Latency=3  │                 │                    │
└────────────────────────────────────────────────────────────────┘

NetworkBuffer 格式 (32KB 页):
┌────────────────────────────────────────────────────────────────┐
│ Header │ Payload (RecordWriter 的输出)                          │
└────────────────────────────────────────────────────────────────┘

Netty Message 帧格式:
┌────────────────────────────────────────────────────────────────┐
│ Frame Length (4B) │ Magic Number: 0xBADC0FFE (4B) │ MsgID (1B) │ Payload │
└────────────────────────────────────────────────────────────────┘

1.3 消息类型

// NettyMessage.java — 消息 ID 枚举
BufferResponse        (ID=0)  // 生产者→消费者: 数据缓冲
ErrorResponse         (ID=1)  // 生产者→消费者: 错误
PartitionRequest      (ID=2)  // 消费者→生产者: 请求分区数据 (携带初始信用)
TaskEventRequest      (ID=3)  // 消费者→生产者: 任务事件
CancelPartitionRequest(ID=4)  // 消费者→生产者: 取消分区请求
CloseRequest          (ID=5)  // 消费者→生产者: 关闭连接
AddCredit             (ID=6)  // 消费者→生产者: 增加信用额度
ResumeConsumption     (ID=7)  // 消费者→生产者: (Checkpoint 后)恢复消费
AckAllUserRecords     (ID=8)  // 消费者→生产者: 确认已处理所有用户记录
BacklogAnnouncement   (ID=9)  // 生产者→消费者: 通知积压大小
NewBufferSize         (ID=10) // 消费者→生产者: 反馈建议缓冲大小 (Debloater)
SegmentId             (ID=11) // 消费者→生产者: Segment 标识

2. Producer 端 — RecordWriter → ResultPartition

2.1 RecordWriter 调用链

// RecordWriter.java — emit 核心流程
public void emit(T record, int targetSubpartition) throws IOException {
    // 1. 序列化
    serializer.setPosition(0);
    serializeRecord(serializer, record);  // [Length:4B][Data:String]

    // 2. 写入目标子分区
    targetPartitionWriter.emit(
        serializer.getSharedBuffer(),     // 序列化后的 ByteBuffer
        targetSubpartition                 // 下游目标
    );

    // 3. 如果需要立即 flush (flushAlways=true)
    if (flushAlways) {
        targetPartitionWriter.flush(targetSubpartition);
    }
}

// RecordWriterOutput.java — 事件发送 (Watermark, Barrier)
public void emitWatermark(Watermark watermark) {
    // Watermark 序列化为: [Length][Type=WATERMARK][Timestamp]
    recordWriter.broadcastEvent(watermark);
}

public void broadcastEvent(AbstractEvent event) {
    // 将事件广播到所有子分区 (所有下游 Subtask)
    for (int i = 0; i < numberOfSubpartitions; i++) {
        // 事件序列化 + 写入每个子分区
        targetPartition.emit(event, i);
    }
}

OutputFlusher 守护线程：

// RecordWriter.OutputFlusher — 吞吐与延迟的平衡
class OutputFlusher extends Thread {
    @Override
    public void run() {
        while (isRunning) {
            Thread.sleep(timeout); // 默认 100ms (network.buffer-timeout)
            flushAll();             // 所有子分区的数据都变得可见
        }
    }
}

设计权衡：flushAlways=true 意味着每条记录后都立即 flush → 延迟极低但吞吐差。flushAlways=false 意味着直到 OutputFlusher 醒来或缓冲满时才 flush → 延迟较高但吞吐高。network.buffer-timeout 参数控制这个平衡。

2.2 ResultPartition — 数据产出端

// ResultPartition.java — 核心字段
public abstract class ResultPartition {
    final ResultPartitionType partitionType;   // PIPELINED / BLOCKING
    final BufferPool bufferPool;              // 本地缓冲池
    final ResultPartitionManager partitionManager; // 注册/查找分区

    // 生命周期: PRODUCING → CONSUMING → RELEASED

    // 写入数据
    public void emit(ByteBuffer data, int targetSubpartition) {
        resultSubpartition.add(data);  // 写入子分区
    }

    // 标记完成 (无更多数据)
    public void finish() {
        for (ResultSubpartition sub : subpartitions) {
            sub.finish();  // 注入 EndOfPartitionEvent
        }
    }

    // 可获取缓冲 → isAvailable=true (无反压)
    // 无法获取缓冲 → isAvailable=false (反压)
    public CompletableFuture<?> getAvailableFuture() {
        return bufferPool.getAvailableFuture();
    }
}

2.3 PipelinedSubpartition — 流水线子分区实现

// PipelinedSubpartition.java — 数据队列
public class PipelinedSubpartition extends ResultSubpartition {
    // 双优先级队列: 普通数据 + 优先级事件 (Barrier)
    private final PrioritizedDeque<BufferAndBacklog> buffers;
    private int buffersInBacklog;          // 积压缓冲数
    private boolean isBlocked;             // 被 Checkpoint Barrier 阻塞?
    private int sequenceNumber;            // 单调递增序列号

    // 写入数据
    public void add(BufferConsumer bufferConsumer) {
        synchronized (buffers) {
            if (bufferConsumer.isPriorityEvent()) {
                // Checkpoint Barrier → 优先级队列
                buffers.addPriority(bufferConsumer);
            } else {
                // 普通数据 → 普通队列
                buffers.add(bufferConsumer);
                buffersInBacklog++;
            }
            // 通知消费者有新数据可用
            notifyDataAvailable();
        }
    }

    // 消费者读取数据 (被 Netty PartitionRequestQueue 调用)
    public BufferAndBacklog pollBuffer() {
        synchronized (buffers) {
            if (isBlocked) return null;  // 被 Barrier 阻塞

            Buffer buffer = buffers.poll();
            if (buffer == null) return null;

            // 如果是 Barrier 且 BufferStorageType=BLOCKING:
            //   → isBlocked = true (阻塞上游直到 Checkpoint 完成)

            buffersInBacklog = buffer.isBuffer() ? buffersInBacklog - 1 : buffersInBacklog;
            return new BufferAndBacklog(buffer, buffersInBacklog, nextDataType, sequenceNumber++);
        }
    }
}

Barrier 的阻塞语义：当 PipelinedSubpartition.pollBuffer() 遇到 Checkpoint Barrier 时，它会设置 isBlocked = true。这意味着：后续的普通数据不会被消费（pollBuffer 返回 null），直到消费者完成 Checkpoint 并通过 ResumeConsumption 消息解除阻塞。

3. Consumer 端 — InputGate → InputChannel

3.1 SingleInputGate — 多输入合并

// SingleInputGate.java — 消费端入口
public class SingleInputGate extends InputGate {
    // 有可用数据的通道队列 (按优先级)
    private final PrioritizedDeque<InputChannel> inputChannelsWithData;

    // 跟踪哪些通道已入队 (BitSet 去重)
    private final BitSet enqueuedInputChannelsWithData;

    // 跟踪哪些通道已完成 (收到 EndOfPartitionEvent)
    private final BitSet channelsWithEndOfPartitionEvents;

    // 缓冲池
    private final BufferPool bufferPool;

    // Buffer Debloater — 动态调整缓冲大小
    private final BufferDebloater bufferDebloater;

    // StreamTask 循环中调用
    public Optional<BufferOrEvent> getNext() {
        // 1. 从 inputChannelsWithData 获取有数据的通道
        InputChannel channel = inputChannelsWithData.poll();

        // 2. 从通道读取一个缓冲
        Optional<BufferOrEvent> bufferOrEvent = channel.getNextBuffer();

        // 3. 如果通道还有更多数据 → 重新入队
        if (channel.isMoreAvailable()) {
            enqueueChannel(channel);
        }

        // 4. 如果是 EndOfPartitionEvent → 标记通道完成
        if (bufferOrEvent.isEvent() && isEndOfPartition) {
            channelsWithEndOfPartitionEvents.set(channelIndex);
            // 如果所有通道都完成了 → InputGate 完成
        }

        return bufferOrEvent;
    }
}

3.2 RemoteInputChannel — 信用制流控实现

// RemoteInputChannel.java — Credit 管理的核心
public class RemoteInputChannel extends InputChannel {
    // 未通知给生产者的可用缓冲数 (本地积累的 credit)
    private final AtomicInteger unannouncedCredit = new AtomicInteger(0);

    // 初始独占缓冲数 (networkBuffersPerChannel)
    private final int initialCredit;

    // 已接收但未消费的缓冲队列
    private final PrioritizedDeque<Buffer> receivedBuffers;

    // 单调递增序列号 (检测重排序和数据丢失)
    private int expectedSequenceNumber = 0;

    // 缓冲管理器 (独占缓冲 + 浮动缓冲)
    private final BufferManager bufferManager;

    // ===== 核心方法 =====

    // 接收来自生产者的数据
    public void onBuffer(Buffer buffer, int sequenceNumber, int backlog, int subpartitionId) {
        // 1. 序列号校验
        if (sequenceNumber != expectedSequenceNumber) {
            throw new IllegalStateException("Buffer reordering detected!");
        }
        expectedSequenceNumber++;

        // 2. 如果是阻塞性 Barrier → 释放浮动缓冲 (反压传播)
        if (buffer.getDataType().isBlockingUpstream()) {
            bufferManager.releaseFloatingBuffers();
        }

        // 3. 加入接收队列
        receivedBuffers.add(buffer);

        // 4. 通知 InputGate 有新数据
        notifyDataAvailable();

        // 5. 处理积压信息 (决定是否要求更多缓冲)
        if (backlog >= 0) {
            onSenderBacklog(backlog);
        }
    }

    // 处理积压信息 → 请求浮动缓冲
    public void onSenderBacklog(int backlog) {
        // 需要的缓冲数 = backlog + initialCredit
        int numRequired = backlog + initialCredit;
        // 从 BufferManager 请求浮动缓冲
        bufferManager.requestFloatingBuffers(numRequired);
        // 有新的缓冲可用 → 发送 credit 给生产者
    }

    // 缓冲被消费后 → 归还 credit
    public void notifyBufferAvailable(int numAvailableBuffers) {
        int old = unannouncedCredit.getAndAdd(numAvailableBuffers);
        if (old == 0 && numAvailableBuffers > 0) {
            // 从 0 变为有 credit → 通知生产者
            notifyCreditAvailable();
        }
    }

    // 向生产者发送 AddCredit 消息
    private void notifyCreditAvailable() {
        int credit = unannouncedCredit.getAndSet(0);
        partitionRequestClient.addCredit(this, credit);
    }
}

4. Credit-based Flow Control — 端到端协议

4.1 Credit 协议详解

Credit-based Flow Control 是 Flink 反压机制的基石。它是 接收端驱动 的流控：

协议要素:
- initialCredit: 每个通道初始分配的独占缓冲 (networkBuffersPerChannel)
- floating buffers: 共享池的动态缓冲，按需分配
- unannouncedCredit: 消费者本地已释放但尚未通知生产者的缓冲数
- backlog: 生产者的积压缓冲数 (缓冲区中等待消费的数据量)

Credit 生命周期:
  1. 生产者创建 PipelinedSubpartition
  2. 消费者创建 RemoteInputChannel，请求 initialCredit 个独占缓冲
  3. 消费者发送 PartitionRequest(initialCredit) 给生产者
  4. 生产者有 initialCredit 个 credit → 可以发送 initialCredit 个 buffer
  5. 生产者发送 BufferResponse(buffer, sequenceNumber, backlog)
  6. 消费者收到 buffer → 消费 → 释放 → unannouncedCredit++
  7. 消费者发送 AddCredit(unannouncedCredit) → unannouncedCredit = 0
  8. 生产者收到 AddCredit → credit 增加 → 可以发送更多 buffer

4.2 端到端时序

4.3 独占缓冲 vs 浮动缓冲

独占缓冲 (Exclusive):
  - 每个 RemoteInputChannel 固定分配 initialCredit 个
  - 保证每个 Consumer 最低限度的缓冲 → 防止死锁
  - 即使上游没有数据，这些缓冲也被预留

浮动缓冲 (Floating):
  - 全局共享池中按需分配
  - 根据 backlog 动态分配: numFloating = f(backlog)
  - 有数据积压时 → 分配更多浮动缓冲 → 提高吞吐
  - 无数据积压时 → 释放浮动缓冲 → 节省内存

4.4 生产者端 — PartitionRequestQueue

// PartitionRequestQueue.java — 生产者端的传输调度
public class PartitionRequestQueue {
    // 有可用数据的 reader 队列 (信用额度 > 0 且有数据)
    private final ArrayDeque<SequenceNumberingViewReader> availableReaders;

    // 写出循环
    private void writeAndFlushNextMessageIfPossible() {
        while (true) {
            SequenceNumberingViewReader reader = availableReaders.poll();
            if (reader == null) break;

            // 从子分区读取一个 buffer
            BufferAndBacklog buffer = reader.getNextBuffer();
            if (buffer == null) continue;

            // 构造 BufferResponse
            NettyMessage.BufferResponse response = new NettyMessage.BufferResponse(
                buffer.buffer(),
                reader.getSequenceNumber(),
                buffer.nextDataType(),
                buffer.backlog()
            );

            // 通过 Netty 通道写入
            channel.writeAndFlush(response).addListener(future -> {
                // 写完成后 → 重新入队 (如果还有数据)
                if (reader.hasBuffersAvailable() && reader.hasCredit()) {
                    enqueueAvailableReader(reader);
                }
                // 触发下一个写出
                writeAndFlushNextMessageIfPossible();
            });
        }
    }
}

5. NetworkBuffer 与 BufferPool — 内存管理

5.1 分层缓冲管理

Global:  NetworkBufferPool (JVM 全局单例)
          │  管理所有 32KB NetworkBuffer
          │  总内存 = taskmanager.memory.network.fraction * flink.size
          │
          ├── LocalBufferPool (每个 ResultPartition)
          │   └── currentPoolSize 个 NetworkBuffer
          │
          ├── LocalBufferPool (每个 InputGate)
          │   └── currentPoolSize 个 NetworkBuffer
          │
          └── 浮动缓冲池 (按需分配)

5.2 LocalBufferPool — 动态缓冲分配

// LocalBufferPool.java — 每个 ResultPartition/InputGate 的本地池
public class LocalBufferPool implements BufferPool {
    private final NetworkBufferPool networkBufferPool;  // 全局池
    private final int numRequired;                      // 最小缓冲数

    // 可用缓冲段队列
    private final ArrayDeque<MemorySegment> availableMemorySegments = new ArrayDeque<>();

    // 缓冲可用性 (反压信号)
    private final AvailabilityHelper availabilityHelper = new AvailabilityHelper();

    // 请求缓冲 (ResultPartition 写入时调用)
    public MemorySegment requestMemorySegment(int targetChannel) {
        // 1. 首先从本地池获取
        MemorySegment segment = availableMemorySegments.poll();
        if (segment != null) return segment;

        // 2. 本地池空 → 从全局池请求更多
        segment = networkBufferPool.requestMemorySegment();
        if (segment != null) {
            currentPoolSize++;
            return segment;
        }

        // 3. 全局池空 → 反压!
        availabilityHelper.resetUnavailable();
        return null;
    }

    // 回收缓冲 (InputGate 消费后归还)
    public void recycle(MemorySegment segment) {
        availableMemorySegments.offer(segment);
        availabilityHelper.available();  // 缓冲可用!
    }

    // 反压信号
    public CompletableFuture<?> getAvailableFuture() {
        return availabilityHelper.getAvailableFuture();
    }
}

5.3 透支缓冲 (Overdraft Buffer)

// LocalBufferPool.java — 透支缓冲逻辑
public MemorySegment requestMemorySegment(int targetChannel) {
    // ... 正常获取失败后 ...

    // 透支缓冲: 超出 currentPoolSize 但未超过 maxOverdraftBuffers
    if (numBuffersRequestedForTarget < maxBuffersPerChannel
        && totalRequestedBuffers < currentPoolSize + maxOverdraftBuffersPerGate) {

        segment = networkBufferPool.requestMemorySegment();
        if (segment != null) {
            // 临时"透支"缓冲 → 防止短暂的反压
            return segment;
        }
    }

    // 连透支都不行了 → 真正的反压
    return null;
}

透支缓冲解决了什么问题？

• 短时间的数据尖峰不会立即触发反压
• 例如: 某个 Subpartition 突然有大量数据，但其他 Subpartition 空闲
• 透支缓冲允许借用其他 Subpartition 的缓冲配额

6. BufferDebloater — 动态缓冲大小调整

// BufferDebloater.java — 基于吞吐量动态调整缓冲大小
public class BufferDebloater {
    // 目标缓冲消耗时间 (默认 1ms)
    private final Duration targetBufferDuration;

    // 基于吞吐量计算最优缓冲大小
    public int recalculateBufferSize(int currentBufferSize, long throughput) {
        // 吞吐量 (bytes/ms) * 目标时间 (ms) = 最优缓冲大小 (bytes)
        long optimalSize = (long) (throughput * targetBufferDuration.toMillis());

        // 平滑调整: 避免缓冲大小剧烈变化
        return Math.min(optimalSize, maxBufferSize);
    }

    // 在 BufferResponse 中携带 NewBufferSize 消息给生产者
    // 生产者收到后调整 PipelinedSubpartition 的 bufferSize
    // → 新数据使用新的大缓冲 → 减少网络调用次数 → 提高吞吐
}

7. Shuffle 类型

Local Shuffle vs Remote Shuffle

Local Shuffle (同 TM 内):
  Producer Task ──memcpy──► Consumer Task
  延迟: ~1 μs
  无序列化开销 (对象引用直接传递)

Remote Shuffle (跨 TM):
  Producer Task → RecordWriter → Serializer → Netty → TCP ──►
  Consumer Task ← RecordReader ← Deserializer ← Netty ← TCP
  延迟: ~100 μs - 1 ms
  完整序列化/反序列化

8. 零拷贝优化

Flink 网络栈支持零拷贝传输：

1. Netty 原生的 FileRegion：DefaultFileRegion 直接从文件描述符发送数据，不走用户态拷贝
2. MemorySegment 的 Off-heap 内存：NetworkBuffer 使用堆外 MemorySegment，Netty 可以直接 DMA 到网卡
3. CompositeByteBuf 的 Gather Writes：多个小缓冲组成一个 CompositeByteBuf，Netty 用 scatter-gather I/O 一次性写入 TCP socket

9. 源码导航（完整版）

10. 常见问题 / 面试题

Q1: NetworkBuffer 的 32KB 是怎么来的？为什么不是 64KB？

A: 32KB 是经验值，平衡了以下因素：

• 太小（如 4KB）：Netty 系统调用次数过多，TCP 包头开销占比大（>10%）
• 太大（如 1MB）：单个 buffer 占据过多内存，buffer 池灵活性降低，且 TCP 分段后失去优势
• 32KB：接近 Linux TCP 的默认 tcp_wmem 大小（16-64KB），一次 write() ≈ 一个 TCP segment

Q2: Credit-based 流控和 TCP 流控有什么本质区别？

A: TCP 流控是字节级别的，Flink Credit 是 buffer 级别的：

• TCP: 滑动窗口大小 = 可接收的字节数
• Flink: credit = 可接收的 NetworkBuffer 个数
• 一个 buffer = 一个独立的 32KB 内存段。消费者归还一个 buffer（消费完成）→ credit+1 → 生产者可以复用这个内存段。
• TCP 流控防止网络拥塞，Flink Credit 同时防止网络拥塞和内存耗尽。

Q3: 什么是透支缓冲？什么时候应该增大透支缓冲？

A: 透支缓冲是超出 currentPoolSize 的临时缓冲，用于处理突发流量。当背压指标显示短暂尖峰（< 100ms）而非持续背压时，增大 maxOverdraftBuffersPerGate 可以消除这些尖峰。默认值较小（5），适合大多数场景。

Q4: Local Shuffle 和 Remote Shuffle 在代码中如何区分？

A: SingleInputGate.setupInputChannels() 时，如果发现结果分区在同一个 TM 上且是 PIPELINED 模式，创建 LocalInputChannel。LocalInputChannel 直接通过内存拷贝读取 ResultSubpartition，完全不走 Netty。否则创建 RemoteInputChannel，通过 Netty TCP 读取。

下一步

• 反压机制: doc-08-backpressure.md — 反压如何产生、检测和解决
• TM 内存模型: doc-10-taskmanager-memory.md — Network Memory 在总内存中的位置