当前位置：网站首页>WebRTC源码分析 nack详解

WebRTC源码分析 nack详解

2022-08-10 17:23:00 【音视频开发老舅】

1、Nack过程

1.1 nack是什么

丢包重传(NACK)是抵抗网络错误的重要手段。NACK在接收端检测到数据丢包后，发送NACK报文到发送端；发送端根据NACK报文中的序列号，在发送缓冲区找到对应的数据包，重新发送到接收端。NACK需要发送端，发送缓冲区的支持。

1.2 nack流程

发送端发送rtp，到达接收端时，发现丢包，接收端发送nack请求，发送端会从历史队列中取出数据重发。

2、Nack协议实现

2.1 rfc协议

在rfc4585协议中定义可重传未到达数据的类型有二种：

1）RTPFB：rtp报文丢失重传（nack）。

2）PSFB：指定净荷重传，指定净荷重传里面又分如下三种（关键帧请求）：

1、PLI (Picture Loss Indication) 视频帧丢失重传。

2、SLI (Slice Loss Indication) slice丢失重转。

3、RPSI (Reference Picture Selection Indication)参考帧丢失重传。

在创建视频连接的SDP协议里面，会协商以上述哪种类型进行NACK重转。以webrtc为例，会协商两种NACK，一个rtp报文丢包重传的nack（nack后面不带参数，默认RTPFB）、PLI 视频帧丢失重传的nack。

rtcp包格式

本文福利， C++音视频学习资料包、技术视频，内容包括（音视频开发，面试题，FFmpeg ，webRTC ，rtmp ，hls ，rtsp ，ffplay ，srs）↓↓↓↓↓↓见下面↓↓文章底部点击领取↓↓

nack rtcp报文格式如上图所示，pt=205。Packet identifier(PID) 为丢包起始参考值，Bitmap of Lost Packets(BLP)为16位的bitmap，对应为1的为表示丢包数据，具体如下抓包分析：

Packet identifier(PID)为176。Bitmap of Lost Packets(BLP)：0x6ae1。解析的时候需要按照小模式解析，0x6ae1对应二进制：110101011100001倒过来看1000 0111 0101 0110。按照1bit是丢包，0bit是没有丢包解析，丢失报文序列号分别是：176 177 182 183 184 186 188 190 191与wireshark解析一致，当然pid和blp可以有多个。

3、发送流程

3.1 数据流

以视频为例

3.2 调用堆栈

 H264EncoderImpl::Encode
 VideoStreamEncoder::OnEncodedImage
 VideoSendStreamImpl::OnEncodedImage
 RtpVideoSender::OnEncodedImage
 RTPSenderVideo::SendEncodedImage
 RTPSenderVideo::SendVideo
 RTPSenderVideo::LogAndSendToNetwork
 RTPSender::EnqueuePackets
 pacer //
 RtpSenderEgress::SendPacket
 //放入队列
 RtpPacketHistory::PutRtpPacket

3.3 RtpPacketHistory

RtpPacketHistory 负责缓存历史数据，有nack请求时，从此队列发送

4、视频Nack过程

4.1 rtp接收数据流程

4.2 nack对报序号的管理

接收到rtp数据包后，会调用nack模块对报序号进行记录，同时检测是否需要nack请求
看代码

int NackRequester::OnReceivedPacket(uint16_t seq_num,
                                    bool is_keyframe,
                                    bool is_recovered) {
  RTC_DCHECK_RUN_ON(worker_thread_);
  bool is_retransmitted = true;

  if (!initialized_) {
    newest_seq_num_ = seq_num;
    if (is_keyframe)
      keyframe_list_.insert(seq_num);
    initialized_ = true;
    return 0;
  }

  // Since the `newest_seq_num_` is a packet we have actually received we know
  // that packet has never been Nacked.
  if (seq_num == newest_seq_num_)
    return 0;
  //如果接收的报序号小于之前接收到的，可能是乱序的包，可能是重传包
  //如果nack列表有，则清除
  if (AheadOf(newest_seq_num_, seq_num)) {
    // An out of order packet has been received.
    auto nack_list_it = nack_list_.find(seq_num);
    int nacks_sent_for_packet = 0;
    if (nack_list_it != nack_list_.end()) {
      nacks_sent_for_packet = nack_list_it->second.retries;
      nack_list_.erase(nack_list_it);
    }
    if (!is_retransmitted)
      UpdateReorderingStatistics(seq_num);
    return nacks_sent_for_packet;
  }

  // Keep track of new keyframes.
  //保留最新的关键帧包
  if (is_keyframe)
    keyframe_list_.insert(seq_num);

  // And remove old ones so we don't accumulate keyframes.
  auto it = keyframe_list_.lower_bound(seq_num - kMaxPacketAge);
  if (it != keyframe_list_.begin())
    keyframe_list_.erase(keyframe_list_.begin(), it);

  if (is_recovered) {
    recovered_list_.insert(seq_num);

    // Remove old ones so we don't accumulate recovered packets.
    auto it = recovered_list_.lower_bound(seq_num - kMaxPacketAge);
    if (it != recovered_list_.begin())
      recovered_list_.erase(recovered_list_.begin(), it);

    // Do not send nack for packets recovered by FEC or RTX.
    return 0;
  }

  AddPacketsToNack(newest_seq_num_ + 1, seq_num);
  newest_seq_num_ = seq_num;

  // Are there any nacks that are waiting for this seq_num.
  //获取nack序号，如果有则触发nack
  std::vector<uint16_t> nack_batch = GetNackBatch(kSeqNumOnly);
  if (!nack_batch.empty()) {
    // This batch of NACKs is triggered externally; the initiator can
    // batch them with other feedback messages.
    nack_sender_->SendNack(nack_batch, /*buffering_allowed=*/true);
  }

  return 0;
}

这部分逻辑主要是收到包，查一下是不是乱序的，可能是网络造成乱序，也可能是重发过来的，收到了就把nack list里面的记录删掉

void NackRequester::AddPacketsToNack(uint16_t seq_num_start,
                                     uint16_t seq_num_end) {
  // Called on worker_thread_.
  // Remove old packets.
  //kMaxPacketAge=1000,删除超出队列数量，删除最老的
  auto it = nack_list_.lower_bound(seq_num_end - kMaxPacketAge);
  nack_list_.erase(nack_list_.begin(), it);

  // If the nack list is too large, remove packets from the nack list until
  // the latest first packet of a keyframe. If the list is still too large,
  // clear it and request a keyframe.
  //nack_list 的最大容量为 kMaxNackPackets = 1000, 
  //如果满了会删除最后一个 KeyFrame 之前的所有nacked 序号, 
  //如果删除之后还是满的那么清空 nack_list 并请求KeyFrame
  uint16_t num_new_nacks = ForwardDiff(seq_num_start, seq_num_end);
  if (nack_list_.size() + num_new_nacks > kMaxNackPackets) {
    while (RemovePacketsUntilKeyFrame() &&
           nack_list_.size() + num_new_nacks > kMaxNackPackets) {
    }

    if (nack_list_.size() + num_new_nacks > kMaxNackPackets) {
      nack_list_.clear();
      RTC_LOG(LS_WARNING) << "NACK list full, clearing NACK"
                             " list and requesting keyframe.";
      keyframe_request_sender_->RequestKeyFrame();
      return;
    }
  }

  for (uint16_t seq_num = seq_num_start; seq_num != seq_num_end; ++seq_num) {
    // Do not send nack for packets that are already recovered by FEC or RTX
    if (recovered_list_.find(seq_num) != recovered_list_.end())
      continue;
    NackInfo nack_info(seq_num, seq_num + WaitNumberOfPackets(0.5),
                       clock_->TimeInMilliseconds());
    RTC_DCHECK(nack_list_.find(seq_num) == nack_list_.end());
    nack_list_[seq_num] = nack_info;
  }
}

我们可以看到AddPacketsToNack()函数主要实现了：

nack_list 的最大容量为 kMaxNackPackets = 1000, 如果满了会删除最后一个 KeyFrame 之前的所有nacked 序号, 如果删除之后还是满的那么清空 nack_list 并请求KeyFrame。

本文福利， C++音视频学习资料包、技术视频，内容包括（音视频开发，面试题，FFmpeg ，webRTC ，rtmp ，hls ，rtsp ，ffplay ，srs）↓↓↓↓↓↓见下面↓↓文章底部点击领取↓↓

获取需要nack的

std::vector<uint16_t> NackRequester::GetNackBatch(NackFilterOptions options) {
  // Called on worker_thread_.
  //只考虑根据序列号获取nacklist
  bool consider_seq_num = options != kTimeOnly;
  //只考虑根据时间戳获取nacklist
  bool consider_timestamp = options != kSeqNumOnly;
  //当前时间
  Timestamp now = clock_->CurrentTime();
  std::vector<uint16_t> nack_batch;
  auto it = nack_list_.begin();
  //遍历nack
  while (it != nack_list_.end()) {
    //初始化rtt为重发延时间隔
    TimeDelta resend_delay = TimeDelta::Millis(rtt_ms_);
    //如果使用了nack的配置
    if (backoff_settings_) {
      //设置最大的重发延时间隔
      resend_delay =
          std::max(resend_delay, backoff_settings_->min_retry_interval);
     // 如果重试次数超过了1次，则重新计算幂指后的重发延迟间隔（避免重试频繁）
     //每次延时增大25%，1.25的n次幂
      if (it->second.retries > 1) {
        TimeDelta exponential_backoff =
            std::min(TimeDelta::Millis(rtt_ms_), backoff_settings_->max_rtt) *
            std::pow(backoff_settings_->base, it->second.retries - 1);
        resend_delay = std::max(resend_delay, exponential_backoff);
      }
    }
   // 判断当前包seq_num是否该发送了（即超过了最大发送延迟时间）
    bool delay_timed_out =
        now.ms() - it->second.created_at_time >= send_nack_delay_ms_;
    // 判断基于rtt延迟时间时是否该发送了（即超过了重发延迟时间）
    bool nack_on_rtt_passed =
        now.ms() - it->second.sent_at_time >= resend_delay.ms();
    // 判断基于序列号时是否该发送了（即超过了重发延迟时间）
    bool nack_on_seq_num_passed =
         // 初次发送时有效，避免重复重发
        it->second.sent_at_time == -1 &&
        // 当前包序列号比较老
        AheadOrAt(newest_seq_num_, it->second.send_at_seq_num);
    // 如果该发送了
    if (delay_timed_out && ((consider_seq_num && nack_on_seq_num_passed) ||
                            (consider_timestamp && nack_on_rtt_passed))) {
       // 当前包seq_num加入到nack list
      nack_batch.emplace_back(it->second.seq_num);
      ++it->second.retries; // 累积重试次数
        // 设置发送时间
      it->second.sent_at_time = now.ms();
      // 超过最大重试次数了则从nack_list_移除
      if (it->second.retries >= kMaxNackRetries) {
        RTC_LOG(LS_WARNING) << "Sequence number " << it->second.seq_num
                            << " removed from NACK list due to max retries.";
        it = nack_list_.erase(it);
      } else {
        ++it;
      }
      continue;
    }
    ++it;
  }
  return nack_batch;
}

1、delay_timed_out ：加入nacklist的时间大于要发送nack的延时

2、nack_on_rtt_passed ：该序号上次发送NACK的时间到当前时间要超过前面计算出来的延时。

3：nack_on_seq_num_passed ：确定有最新的包序号比这个大，是被丢失的

4.3 nack的触发时机

逻辑图

有两个地方触发nack，用红方框框出来了

1 当收到rtp数据，nack模块会记录包序号，包类型
2 线程定期检测是否存在丢包，需要nack请求

5、nack响应

5.1 rtcp数据流

5.2 源码

void ModuleRtpRtcpImpl2::OnReceivedNack(
    const std::vector<uint16_t>& nack_sequence_numbers) {
  if (!rtp_sender_)
    return;

  if (!StorePackets() || nack_sequence_numbers.empty()) {
    return;
  }
  // Use RTT from RtcpRttStats class if provided.
  int64_t rtt = rtt_ms();
  if (rtt == 0) {
    rtcp_receiver_.RTT(rtcp_receiver_.RemoteSSRC(), NULL, &rtt, NULL, NULL);
  }
  //取得rtt，把请求和rtt时间调用rtp补包
  rtp_sender_->packet_generator.OnReceivedNack(nack_sequence_numbers, rtt);
}

void RTPSender::OnReceivedNack(
    const std::vector<uint16_t>& nack_sequence_numbers,
    int64_t avg_rtt) {
  //设置历史队列rtt，取包时根据rtt计算
  packet_history_->SetRtt(5 + avg_rtt);
  for (uint16_t seq_no : nack_sequence_numbers) {
    const int32_t bytes_sent = ReSendPacket(seq_no);
    if (bytes_sent < 0) {
      // Failed to send one Sequence number. Give up the rest in this nack.
      RTC_LOG(LS_WARNING) << "Failed resending RTP packet " << seq_no
                          << ", Discard rest of packets.";
      break;
    }
  }
}

int32_t RTPSender::ReSendPacket(uint16_t packet_id) {
  // Try to find packet in RTP packet history. Also verify RTT here, so that we
  // don't retransmit too often.
  absl::optional<RtpPacketHistory::PacketState> stored_packet =
      packet_history_->GetPacketState(packet_id);
  if (!stored_packet || stored_packet->pending_transmission) {
    // Packet not found or already queued for retransmission, ignore.
    return 0;
  }

  const int32_t packet_size = static_cast<int32_t>(stored_packet->packet_size);
  const bool rtx = (RtxStatus() & kRtxRetransmitted) > 0;

  std::unique_ptr<RtpPacketToSend> packet =
      packet_history_->GetPacketAndMarkAsPending(
          packet_id, [&](const RtpPacketToSend& stored_packet) {
            // Check if we're overusing retransmission bitrate.
            // TODO(sprang): Add histograms for nack success or failure
            // reasons.
            std::unique_ptr<RtpPacketToSend> retransmit_packet;
            if (retransmission_rate_limiter_ &&
                !retransmission_rate_limiter_->TryUseRate(packet_size)) {
              return retransmit_packet;
            }
            if (rtx) {
              retransmit_packet = BuildRtxPacket(stored_packet);
            } else {
              retransmit_packet =
                  std::make_unique<RtpPacketToSend>(stored_packet);
            }
            if (retransmit_packet) {
              retransmit_packet->set_retransmitted_sequence_number(
                  stored_packet.SequenceNumber());
            }
            return retransmit_packet;
          });
  if (!packet) {
    return -1;
  }
  packet->set_packet_type(RtpPacketMediaType::kRetransmission);
  packet->set_fec_protect_packet(false);
  std::vector<std::unique_ptr<RtpPacketToSend>> packets;
  packets.emplace_back(std::move(packet));
  paced_sender_->EnqueuePackets(std::move(packets));

  return packet_size;
}

主要看一下RtpPacketHistory::GetPacketAndMarkAsPending 函数
有两个调用：

GetStoredPacket //按照序列号拿到packet
VerifyRtt //距离上次发送，超过rtt时间才能再次发送

bool RtpPacketHistory::VerifyRtt(const RtpPacketHistory::StoredPacket& packet,
                                 int64_t now_ms) const {
  if (packet.send_time_ms_) {
    // Send-time already set, this check must be for a retransmission.
    if (packet.times_retransmitted() > 0 &&
        now_ms < *packet.send_time_ms_ + rtt_ms_) {
      // This packet has already been retransmitted once, and the time since
      // that even is lower than on RTT. Ignore request as this packet is
      // likely already in the network pipe.
      return false;
    }
  }

  return true;
}