[feat] add rtcp thread in rtp video receiver

src/rtcp/rtcp_tcc.cpp

@@ -0,0 +1,30 @@
+#include "rtcp_tcc.h"
+
+RtcpTcc::RtcpTcc() {
+  buffer_ = new uint8_t[DEFAULT_RR_SIZE];
+  size_ = DEFAULT_RR_SIZE;
+}
+
+RtcpTcc::~RtcpTcc() {
+  if (buffer_) {
+    delete buffer_;
+    buffer_ = nullptr;
+  }
+
+  size_ = 0;
+}
+
+void RtcpTcc::SetReportBlock(RtcpReportBlock &rtcp_report_block) {
+  reports_.push_back(rtcp_report_block);
+}
+
+void RtcpTcc::SetReportBlock(std::vector<RtcpReportBlock> &rtcp_report_blocks) {
+  reports_ = rtcp_report_blocks;
+}
+
+const uint8_t *RtcpTcc::Encode() {
+  rtcp_header_.Encode(DEFAULT_RTCP_VERSION, 0, DEFAULT_RR_BLOCK_NUM,
+                      RTCP_TYPE::RR, DEFAULT_RR_SIZE, buffer_);
+
+  return buffer_;
+}

src/rtcp/rtcp_tcc.h

@@ -0,0 +1,77 @@
+/*
+ * @Author: DI JUNKUN
+ * @Date: 2024-12-06
+ * Copyright (c) 2024 by DI JUNKUN, All Rights Reserved.
+ */
+
+#ifndef _RTCP_TCC_H_
+#define _RTCP_TCC_H_
+
+// RR
+//  0                   1                   2                   3
+//  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+// |V=2|P|  FMT=15 |    PT=205     |           length              |
+// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+// |                     SSRC of packet sender                     |
+// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+// |                      SSRC of media source                     |
+// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+// |      base sequence number     |      packet status count      |
+// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+// |                 reference time                | fb pkt. count |
+// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+// |          packet chunk         |         packet chunk          |
+// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+// .                                                               .
+// .                                                               .
+// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+// |         packet chunk          |  recv delta   |  recv delta   |
+// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+// .                                                               .
+// .                                                               .
+// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+// |           recv delta          |  recv delta   | zero padding  |
+// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+
+// RTP transport sequence number
+// 0                   1                   2                   3
+//  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+// |       0xBE    |    0xDE       |           length=1            |
+// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+// |  ID   | L=1   |transport-wide sequence number | zero padding  |
+// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+
+#include <vector>
+
+#include "rtcp_header.h"
+#include "rtcp_typedef.h"
+
+class RtcpTcc {
+ public:
+  RtcpTcc();
+  ~RtcpTcc();
+
+ public:
+  void SetReportBlock(RtcpReportBlock &rtcp_report_block);
+  void SetReportBlock(std::vector<RtcpReportBlock> &rtcp_report_blocks);
+
+ public:
+  const uint8_t *Encode();
+  size_t Decode();
+
+  // Entire RTP buffer
+  const uint8_t *Buffer() const { return buffer_; }
+  size_t Size() const { return size_; }
+
+ private:
+  RtcpHeader rtcp_header_;
+  std::vector<RtcpReportBlock> reports_;
+
+  // Entire RTCP buffer
+  uint8_t *buffer_ = nullptr;
+  size_t size_ = 0;
+};
+
+#endif

src/rtp/rtp_endpoint/rtp_video_receiver.cpp

@@ -8,12 +8,21 @@
 RtpVideoReceiver::RtpVideoReceiver() {}
 
 RtpVideoReceiver::RtpVideoReceiver(std::shared_ptr<IOStatistics> io_statistics)
-    : io_statistics_(io_statistics) {}
+    : io_statistics_(io_statistics) {
+  rtcp_thread_ = std::thread(&RtpVideoReceiver::RtcpThread, this);
+}
 
 RtpVideoReceiver::~RtpVideoReceiver() {
   if (rtp_statistics_) {
     rtp_statistics_->Stop();
   }
+
+  rtcp_stop_.store(true);
+  rtcp_cv_.notify_one();
+
+  if (rtcp_thread_.joinable()) {
+    rtcp_thread_.join();
+  }
 }
 
 void RtpVideoReceiver::InsertRtpPacket(RtpPacket& rtp_packet) {
@@ -299,27 +308,6 @@ bool RtpVideoReceiver::CheckIsAv1FrameCompleted(RtpPacket& rtp_packet) {
   return false;
 }
 
-bool RtpVideoReceiver::Process() {
-  if (!compelete_video_frame_queue_.isEmpty()) {
-    VideoFrame video_frame;
-    compelete_video_frame_queue_.pop(video_frame);
-    if (on_receive_complete_frame_) {
-      // auto now_complete_frame_ts =
-      //     std::chrono::duration_cast<std::chrono::milliseconds>(
-      //         std::chrono::system_clock::now().time_since_epoch())
-      //         .count();
-      // uint32_t duration = now_complete_frame_ts - last_complete_frame_ts_;
-      // LOG_ERROR("Duration {}", duration);
-      // last_complete_frame_ts_ = now_complete_frame_ts;
-
-      on_receive_complete_frame_(video_frame);
-    }
-  }
-
-  std::this_thread::sleep_for(std::chrono::milliseconds(5));
-  return true;
-}
-
 void RtpVideoReceiver::SetSendDataFunc(
     std::function<int(const char*, size_t)> data_send_func) {
   data_send_func_ = data_send_func;
@@ -352,3 +340,46 @@ bool RtpVideoReceiver::CheckIsTimeSendRR() {
     return false;
   }
 }
+
+bool RtpVideoReceiver::Process() {
+  if (!compelete_video_frame_queue_.isEmpty()) {
+    VideoFrame video_frame;
+    compelete_video_frame_queue_.pop(video_frame);
+    if (on_receive_complete_frame_) {
+      // auto now_complete_frame_ts =
+      //     std::chrono::duration_cast<std::chrono::milliseconds>(
+      //         std::chrono::system_clock::now().time_since_epoch())
+      //         .count();
+      // uint32_t duration = now_complete_frame_ts - last_complete_frame_ts_;
+      // LOG_ERROR("Duration {}", duration);
+      // last_complete_frame_ts_ = now_complete_frame_ts;
+
+      on_receive_complete_frame_(video_frame);
+    }
+  }
+
+  std::this_thread::sleep_for(std::chrono::milliseconds(5));
+  return true;
+}
+
+void RtpVideoReceiver::RtcpThread() {
+  while (!rtcp_stop_) {
+    std::unique_lock<std::mutex> lock(rtcp_mtx_);
+    if (rtcp_cv_.wait_for(
+            lock, std::chrono::milliseconds(rtcp_tcc_interval_ms_),
+            [&]() { return send_rtcp_rr_triggered_ || rtcp_stop_; })) {
+      if (rtcp_stop_) break;
+      send_rtcp_rr_triggered_ = false;
+    } else {
+      LOG_ERROR("Send video tcc");
+      auto now = std::chrono::steady_clock::now();
+      auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
+                         now - last_send_rtcp_rr_ts_)
+                         .count();
+      if (elapsed >= rtcp_rr_interval_ms_) {
+        LOG_ERROR("Send video rr [{}]", (void*)this);
+        last_send_rtcp_rr_ts_ = now;
+      }
+    }
+  }
+}

src/rtp/rtp_endpoint/rtp_video_receiver.h

@@ -45,6 +45,7 @@ class RtpVideoReceiver : public ThreadBase {
 
  private:
   bool Process() override;
+  void RtcpThread();
 
  private:
   std::map<uint16_t, RtpPacket> incomplete_frame_list_;
@@ -72,6 +73,16 @@ class RtpVideoReceiver : public ThreadBase {
   // std::map<uint32_t, std::map<uint16_t, RtpPacket>> fec_repair_symbol_list_;
   std::set<uint64_t> incomplete_fec_frame_list_;
   std::map<uint64_t, std::map<uint16_t, RtpPacket>> incomplete_fec_packet_list_;
+
+ private:
+  std::thread rtcp_thread_;
+  std::mutex rtcp_mtx_;
+  std::condition_variable rtcp_cv_;
+  std::chrono::steady_clock::time_point last_send_rtcp_rr_ts_;
+  std::atomic<bool> send_rtcp_rr_triggered_ = false;
+  std::atomic<bool> rtcp_stop_ = false;
+  int rtcp_rr_interval_ms_ = 5000;
+  int rtcp_tcc_interval_ms_ = 200;
 };
 
 #endif

src/statistics/io_statistics.cpp

@@ -155,11 +155,11 @@ void IOStatistics::Stop() {
 }
 
 void IOStatistics::UpdateVideoInboundBytes(uint32_t bytes) {
-  video_inbound_bytes_.fetch_add(bytes, std::memory_order_relaxed);
+  video_inbound_bytes_ += bytes;
 }
 
 void IOStatistics::UpdateVideoOutboundBytes(uint32_t bytes) {
-  video_outbound_bytes_.fetch_add(bytes, std::memory_order_relaxed);
+  video_outbound_bytes_ += bytes;
 }
 
 void IOStatistics::UpdateVideoPacketLossCount(uint16_t seq_num) {
@@ -167,13 +167,11 @@ void IOStatistics::UpdateVideoPacketLossCount(uint16_t seq_num) {
   if (last_v_seq != 0) {
     if (seq_num > last_v_seq) {
       if (seq_num - last_v_seq != 1) {
-        video_rtp_pkt_loss_cnt_.fetch_add(seq_num - last_v_seq - 1,
+        video_rtp_pkt_loss_cnt_ += seq_num - last_v_seq - 1;
-                                          std::memory_order_relaxed);
       }
     } else {
-      video_rtp_pkt_loss_cnt_.fetch_add(
+      video_rtp_pkt_loss_cnt_ +=
-          seq_num + (std::numeric_limits<uint16_t>::max() - last_v_seq) - 1,
+          seq_num + (std::numeric_limits<uint16_t>::max() - last_v_seq) - 1;
-          std::memory_order_relaxed);
     }
   }
 
@@ -181,11 +179,11 @@ void IOStatistics::UpdateVideoPacketLossCount(uint16_t seq_num) {
 }
 
 void IOStatistics::UpdateAudioInboundBytes(uint32_t bytes) {
-  audio_inbound_bytes_.fetch_add(bytes, std::memory_order_relaxed);
+  audio_inbound_bytes_ += bytes;
 }
 
 void IOStatistics::UpdateAudioOutboundBytes(uint32_t bytes) {
-  audio_outbound_bytes_.fetch_add(bytes, std::memory_order_relaxed);
+  audio_outbound_bytes_ += bytes;
 }
 
 void IOStatistics::UpdateAudioPacketLossCount(uint16_t seq_num) {
@@ -193,13 +191,11 @@ void IOStatistics::UpdateAudioPacketLossCount(uint16_t seq_num) {
   if (last_a_seq != 0) {
     if (seq_num > last_a_seq) {
       if (seq_num - last_a_seq != 1) {
-        audio_rtp_pkt_loss_cnt_.fetch_add(seq_num - last_a_seq - 1,
+        audio_rtp_pkt_loss_cnt_ += seq_num - last_a_seq - 1;
-                                          std::memory_order_relaxed);
       }
     } else {
-      audio_rtp_pkt_loss_cnt_.fetch_add(
+      audio_rtp_pkt_loss_cnt_ +=
-          seq_num + (std::numeric_limits<uint16_t>::max() - last_a_seq) - 1,
+          seq_num + (std::numeric_limits<uint16_t>::max() - last_a_seq) - 1;
-          std::memory_order_relaxed);
     }
   }
 
@@ -207,11 +203,11 @@ void IOStatistics::UpdateAudioPacketLossCount(uint16_t seq_num) {
 }
 
 void IOStatistics::UpdateDataInboundBytes(uint32_t bytes) {
-  data_inbound_bytes_.fetch_add(bytes, std::memory_order_relaxed);
+  data_inbound_bytes_ += bytes;
 }
 
 void IOStatistics::UpdateDataOutboundBytes(uint32_t bytes) {
-  data_outbound_bytes_.fetch_add(bytes, std::memory_order_relaxed);
+  data_outbound_bytes_ += bytes;
 }
 
 void IOStatistics::UpdateDataPacketLossCount(uint16_t seq_num) {