webrtc FEC 协议

参考：
https://www.cnblogs.com/ishen/p/15333271.html
https://zhuanlan.zhihu.com/p/603421239

1. 生成

1.1 等待并筹齐多个原始包

webrtc 会等待筹齐多个 rtp 包后，再统一生成冗余包，参看 UlpfecGenerator::AddPacketAndGenerateFec() 函数:

void UlpfecGenerator::AddPacketAndGenerateFec(const RtpPacketToSend& packet) {
  ...

  const bool complete_frame = packet.Marker();
  if (media_packets_.size() < kUlpfecMaxMediaPackets) {
    // 构造 fec packet 放入等待队列中
    // Our packet masks can only protect up to |kUlpfecMaxMediaPackets| packets.
    auto fec_packet = std::make_unique<ForwardErrorCorrection::Packet>();
    fec_packet->data = packet.Buffer();
    media_packets_.push_back(std::move(fec_packet));
  }

  // 这里每帧计数加1
  if (complete_frame) {
    ++num_protected_frames_;
  }

  // 动态计算得到 fec 配置参数
  auto params = CurrentParams();

  // 判断是否满足 fec 的条件
  // 1. 必须从 complete_frame 开始
  // 2. 包数量大于等于 params.max_fec_frames 阈值
  // 3. 或者 目前冗余度低于设定的冗余度，同时还筹齐了最少 4 个包
  if (complete_frame &&
      (num_protected_frames_ >= params.max_fec_frames ||
       (ExcessOverheadBelowMax() && MinimumMediaPacketsReached()))) {
    // 开始 fec 编码
    fec_->EncodeFec(media_packets_, params.fec_rate, kNumImportantPackets,
                    kUseUnequalProtection, params.fec_mask_type,
                    &generated_fec_packets_);
  }
  ...
}

需要注意的是：

• complete_frame 变量的存在，表明如果一个编码帧生成了多个 rtp 包，那么 fec 不会分开他们，同一帧的多个 rtp 包总是被一个 fec 保护

1.2 重新累积

每 k 个数据包生成 fec 后，需要重新累积，参看 UlpfecGenerator::ResetState() 函数：

void UlpfecGenerator::ResetState() {
  // 清空原始数据包
  media_packets_.clear();
  last_media_packet_.reset();
  generated_fec_packets_.clear();
  num_protected_frames_ = 0;
  media_contains_keyframe_ = false;
}

1.3 算法原理

一个待发送的 rtp 包，可以看做 [1, n] 即 1 行 n 列形式的数据，其中 n 是字节数，有 k 个原始数据包，那么可以得到 D = [k, n] 的数据。
fec 生成可以看做一个生成矩阵 G 乘以 D，现在假设 G = [q, k]，那么 G * D = F，即 F = [q, n]。
以上公式可以描述为：k 个原始数据包，生成了 q 个冗余包，其中 q <= k。
那么对于 webrtc fec，G 生成矩阵是如何设计的呢？实际上 G 只由 0 和 1 构成，表示选择或不选择，且 G * D 的矩阵加法变为 xor 异或运算，解决数据溢出的问题。

1.4 掩码表选择

上一节说了生成矩阵的选择，在 webrtc 的实现中，预先配置了大量生成矩阵(或者称为掩码)，例如 fec_private_tables_random.cc 文件内：

...

#define kMaskRandom8_4 \
  0x45, 0x00, \
  0xb4, 0x00, \
  0x6a, 0x00, \
  0x89, 0x00
...

#define kMaskRandom10_3 \
  0xa4, 0x40, \
  0xc9, 0x00, \
  0x52, 0x80

...

这里生成矩阵每一行由 16 进制数表示，kMaskRandom8_4 表示 8 个原始包生成 4 个 fec 包。
实际上 webrtc 对 fec 丢包模型有两种生成矩阵配置表，即突发丢包和随机丢包，对应不同掩码表。这两种配置表数值设计的规则是什么，暂时还不清楚

掩码表的选择，在 ForwardErrorCorrection::EncodeFec() 函数中：

int ForwardErrorCorrection::EncodeFec(const PacketList& media_packets,
                                      uint8_t protection_factor,
                                      int num_important_packets,
                                      bool use_unequal_protection,
                                      FecMaskType fec_mask_type,
                                      std::list<Packet*>* fec_packets) {
  // 原始包数量
  const size_t num_media_packets = media_packets.size();
  // 根据原始包数量和实时配置的冗余率，得到 fec 包数量
  int num_fec_packets = NumFecPackets(num_media_packets, protection_factor);
  // 找到掩码表
  internal::GeneratePacketMasks(num_media_packets, num_fec_packets,
                                num_important_packets, use_unequal_protection,
                                &mask_table, packet_masks_);
  ...
}

1.5 计算生成

现在生成矩阵和原始数据都有了，只需要按矩阵乘法生成 fec 即可，参看 ForwardErrorCorrection::GenerateFecPayloads() 函数：

void ForwardErrorCorrection::GenerateFecPayloads(
    const PacketList& media_packets,
    size_t num_fec_packets) {
  // 遍历生成每个冗余包
  for (size_t i = 0; i < num_fec_packets; ++i) {
    ...
    // 对于每个冗余包，都需要遍历一遍原始数据包，根据掩码表确定哪些数据包需要参与本轮 fec 包生成
    while (media_packets_it != media_packets.end()) {
      Packet* const media_packet = media_packets_it->get();
      // Should `media_packet` be protected by `fec_packet`?
      // 查掩码表，如果为 1，表示当前数据包要参与当前 fec 包生成
      if (packet_masks_[pkt_mask_idx] & (1 << (7 - media_pkt_idx))) {
        ...
        XorHeaders(*media_packet, fec_packet);
        XorPayloads(*media_packet, media_payload_length, fec_header_size,
                    fec_packet);
      }
      media_packets_it++;
      ...
    }
  }
}

对于数据包的 header，webrtc 内部会 xor 前 12 字节(webrtc 不会包含 csrc)，参看 ForwardErrorCorrection::XorHeaders() 函数：

void ForwardErrorCorrection::XorHeaders(const Packet& src, Packet* dst) {
  uint8_t* dst_data = dst->data.MutableData();
  const uint8_t* src_data = src.data.cdata();
  // XOR the first 2 bytes of the header: V, P, X, CC, M, PT fields.
  dst_data[0] ^= src_data[0];
  dst_data[1] ^= src_data[1];

  // XOR the length recovery field.
  uint8_t src_payload_length_network_order[2];
  ByteWriter<uint16_t>::WriteBigEndian(src_payload_length_network_order,
                                       src.data.size() - kRtpHeaderSize);
  dst_data[2] ^= src_payload_length_network_order[0];
  dst_data[3] ^= src_payload_length_network_order[1];

  // XOR the 5th to 8th bytes of the header: the timestamp field.
  dst_data[4] ^= src_data[4];
  dst_data[5] ^= src_data[5];
  dst_data[6] ^= src_data[6];
  dst_data[7] ^= src_data[7];

  // Skip the 9th to 12th bytes of the header.
}

对于数据包的 payload，webrtc 内部实际上是从 12 字节的 header 后开始 xor，即 header extension 也会被保护，参看 ForwardErrorCorrection::XorPayloads() 函数:

void ForwardErrorCorrection::XorPayloads(const Packet& src,
                                         size_t payload_length,
                                         size_t dst_offset,
                                         Packet* dst) {
  // 如果当前数据包比较长，fec 包需要扩容
  if (dst_offset + payload_length > dst->data.size()) {
    size_t old_size = dst->data.size();
    size_t new_size = dst_offset + payload_length;
    dst->data.SetSize(new_size);
    memset(dst->data.MutableData() + old_size, 0, new_size - old_size);
  }
  uint8_t* dst_data = dst->data.MutableData();
  const uint8_t* src_data = src.data.cdata();
  // 注意 payload_length 即数据包减去 12 字节后的结果
  for (size_t i = 0; i < payload_length; ++i) {
    dst_data[dst_offset + i] ^= src_data[kRtpHeaderSize + i];
  }
}

1.6 ulp fec 和 flex fec 有什么区别

实际上 webrtc 生成 fec 的代码，即 ForwardErrorCorrection 类，对 ulp 和 flex 都是相同的，即 ulp fec 只存在 level 0 级的全量保护。
唯一区别就是两者生成的 fec 头部格式不同：

std::unique_ptr<ForwardErrorCorrection> ForwardErrorCorrection::CreateUlpfec(
    uint32_t ssrc) {
  std::unique_ptr<FecHeaderReader> fec_header_reader(new UlpfecHeaderReader());
  std::unique_ptr<FecHeaderWriter> fec_header_writer(new UlpfecHeaderWriter());
  return std::unique_ptr<ForwardErrorCorrection>(new ForwardErrorCorrection(
      std::move(fec_header_reader), std::move(fec_header_writer), ssrc, ssrc));
}

std::unique_ptr<ForwardErrorCorrection> ForwardErrorCorrection::CreateFlexfec(
    uint32_t ssrc,
    uint32_t protected_media_ssrc) {
  std::unique_ptr<FecHeaderReader> fec_header_reader(new FlexfecHeaderReader());
  std::unique_ptr<FecHeaderWriter> fec_header_writer(new FlexfecHeaderWriter());
  return std::unique_ptr<ForwardErrorCorrection>(new ForwardErrorCorrection(
      std::move(fec_header_reader), std::move(fec_header_writer), ssrc,
      protected_media_ssrc));
}

如上可以看到，两者只是在 fec_header_reader 和 fec_header_writer 上不一样。同时 ulp fec 与原始数据流共用同一个 ssrc、序列号，只是 pt 值不同；flex fec 有自己独立的 ssrc、序列号、pt 值。

2. 打包传输

2.1 ulp fec 打包方式

在 UlpfecGenerator::GetFecPackets() 函数中，会先将生成的 fec 包打包为 red 封装格式，然后打包为 rtp 格式(为什么不单独传输而是要用 red 封装？暂时不清楚原因)：

std::vector<std::unique_ptr<RtpPacketToSend>> UlpfecGenerator::GetFecPackets() {
  ...
  // 遍历生成的 fec 包
  for (const auto* fec_packet : generated_fec_packets_) {
    std::unique_ptr<RtpPacketToSend> red_packet =
        std::make_unique<RtpPacketToSend>(*last_media_packet_);

    // red 打包，具体参看源码
    // 实际上这里 red 打包格式很简单，只使用了一个 block header，即一个 F + PT 的结构
    ...

    fec_packets.push_back(std::move(red_packet));
  }

  ResetState();

  ...

  return fec_packets;
}

注意这里会调用 ResetState() 清空状态，参考前文。

2.2 flex fec 打包方式

在 FlexfecSender::GetFecPackets() 函数中，会将生成的 fec 包打包为 rtp，并且需要注意的是其拥有与原始包独立的 ssrc、pt、seq 值等字段:

std::vector<std::unique_ptr<RtpPacketToSend>> FlexfecSender::GetFecPackets() {
  ...
  // 遍历生成的 fec 包
  for (const auto* fec_packet : ulpfec_generator_.generated_fec_packets_) {
    std::unique_ptr<RtpPacketToSend> fec_packet_to_send(
        new RtpPacketToSend(&rtp_header_extension_map_));

    // rtp 打包，具体参看源码
    ...
    
    fec_packets_to_send.push_back(std::move(fec_packet_to_send));
  }

  if (!fec_packets_to_send.empty()) {
    ulpfec_generator_.ResetState();
  }

  ...

  return fec_packets_to_send;
}

注意这里也会调用 ResetState() 清空状态，参考前文。

3. 丢包恢复

在接收到数据包后，会来到 ForwardErrorCorrection::DecodeFec() 函数：

void ForwardErrorCorrection::DecodeFec(const ReceivedPacket& received_packet,
                                       RecoveredPacketList* recovered_packets) {
  ...

  InsertPacket(received_packet, recovered_packets);
  AttemptRecovery(recovered_packets);
}

ForwardErrorCorrection::InsertPacket() 会将原始数据包插入到 recovered_packets 队列，并将 fec 包插入到 received_fec_packets_ 队列:

void ForwardErrorCorrection::InsertPacket(
    const ReceivedPacket& received_packet,
    RecoveredPacketList* recovered_packets) {
  ...

  if (received_packet.is_fec) {
    InsertFecPacket(*recovered_packets, received_packet);
  } else {
    InsertMediaPacket(recovered_packets, received_packet);
  }

  DiscardOldRecoveredPackets(recovered_packets);
}

同时在调用 InsertFecPacket() 和 InsertMediaPacket() 函数的时候，会根据 fec 包的 fec header 中的 mask 字段，将 fec 包与其保护的原始数据包对应起来，以 ForwardErrorCorrection::UpdateCoveringFecPackets() 函数为例：

void ForwardErrorCorrection::UpdateCoveringFecPackets(
    const RecoveredPacket& packet) {
  for (auto& fec_packet : received_fec_packets_) {
    // Is this FEC packet protecting the media packet `packet`?
    auto protected_it = absl::c_lower_bound(
        fec_packet->protected_packets, &packet, SortablePacket::LessThan());
    if (protected_it != fec_packet->protected_packets.end() &&
        (*protected_it)->seq_num == packet.seq_num) {
      // Found an FEC packet which is protecting `packet`.
      (*protected_it)->pkt = packet.pkt;
    }
  }
}

之后就是调用 ForwardErrorCorrection::AttemptRecovery() 函数试图恢复缺失包了：

void ForwardErrorCorrection::AttemptRecovery(
    RecoveredPacketList* recovered_packets) {
  // 遍历收到的 fec 包列表
  auto fec_packet_it = received_fec_packets_.begin();
  while (fec_packet_it != received_fec_packets_.end()) {
    // Search for each FEC packet's protected media packets.
    int packets_missing = NumCoveredPacketsMissing(**fec_packet_it);

    // We can only recover one packet with an FEC packet.
    if (packets_missing == 1) {
      // Recovery possible.
      // 缺失包恢复
      std::unique_ptr<RecoveredPacket> recovered_packet(new RecoveredPacket());
      recovered_packet->pkt = nullptr;
      if (!RecoverPacket(**fec_packet_it, recovered_packet.get())) {
        // Can't recover using this packet, drop it.
        fec_packet_it = received_fec_packets_.erase(fec_packet_it);
        continue;
      }
      auto* recovered_packet_ptr = recovered_packet.get();
      // 将恢复的缺失包，加入到原始包队列
      recovered_packets->push_back(std::move(recovered_packet));
      recovered_packets->sort(SortablePacket::LessThan());
      // 更新 fec 和原始包的关联关系
      UpdateCoveringFecPackets(*recovered_packet_ptr);
      DiscardOldRecoveredPackets(recovered_packets);
      // 当前 fec 包已经不需要了
      fec_packet_it = received_fec_packets_.erase(fec_packet_it);

      // A packet has been recovered. We need to check the FEC list again, as
      // this may allow additional packets to be recovered.
      // Restart for first FEC packet.
      // 由于新的缺失包的恢复，可能前面的 fec 包也能进行丢包恢复了，所以这里重新开始遍历
      fec_packet_it = received_fec_packets_.begin();

      ...
    }
  }
}

可以看到，只有 packets_missing == 1 即当前 fec 包对应保护的原始包，只缺失一个时，才能恢复，这与 XOR 算法吻合。
而 NumCoveredPacketsMissing() 函数就是根据 fec 包和保护的原始包的对应关系，判断缺失包的数量，逻辑比较简单，这里不用展示了。