[feat] update rtp packet history module

src/channel/rtp_channel/rtp_video_receiver.h

@@ -6,7 +6,7 @@
 #include <queue>
 #include <set>
 
-#include "clock.h"
+#include "api/clock/clock.h"
 #include "fec_decoder.h"
 #include "io_statistics.h"
 #include "nack_requester.h"

src/channel/rtp_channel/rtp_video_sender.cpp

@@ -11,8 +11,12 @@
 
 RtpVideoSender::RtpVideoSender() {}
 
-RtpVideoSender::RtpVideoSender(std::shared_ptr<IOStatistics> io_statistics)
-    : ssrc_(GenerateUniqueSsrc()), io_statistics_(io_statistics) {
+RtpVideoSender::RtpVideoSender(std::shared_ptr<webrtc::Clock> clock,
+                               std::shared_ptr<IOStatistics> io_statistics)
+    : ssrc_(GenerateUniqueSsrc()),
+      clock_(clock),
+      io_statistics_(io_statistics),
+      rtp_packet_history_(std::make_unique<RtpPacketHistory>(clock)) {
   SetPeriod(std::chrono::milliseconds(5));
 #ifdef SAVE_RTP_SENT_STREAM
   file_rtp_sent_ = fopen("rtp_sent_stream.h264", "w+b");
@@ -67,11 +71,12 @@ int RtpVideoSender::SendRtpPacket(std::shared_ptr<RtpPacket> rtp_packet) {
   }
 
   if (on_sent_packet_func_) {
-    webrtc::RtpPacketToSend* rtp_packet_to_send =
-        dynamic_cast<webrtc::RtpPacketToSend*>(rtp_packet.get());
+    std::shared_ptr<webrtc::RtpPacketToSend> rtp_packet_to_send =
+        std::dynamic_pointer_cast<webrtc::RtpPacketToSend>(rtp_packet);
     rtp_packet_to_send->set_transport_sequence_number(transport_seq_++);
     rtp_packet_to_send->set_packet_type(webrtc::RtpPacketMediaType::kVideo);
     on_sent_packet_func_(*rtp_packet_to_send);
+    rtp_packet_history_->AddPacket(rtp_packet_to_send, clock_->CurrentTime());
   }
 
   if (0 != data_send_func_((const char*)rtp_packet->Buffer().data(),

src/channel/rtp_channel/rtp_video_sender.h

@@ -7,6 +7,7 @@
 #include "ringbuffer.h"
 #include "rtcp_sender_report.h"
 #include "rtp_packet.h"
+#include "rtp_packet_history.h"
 #include "rtp_packet_to_send.h"
 #include "rtp_statistics.h"
 #include "thread_base.h"
@@ -14,7 +15,8 @@
 class RtpVideoSender : public ThreadBase {
  public:
   RtpVideoSender();
-  RtpVideoSender(std::shared_ptr<IOStatistics> io_statistics);
+  RtpVideoSender(std::shared_ptr<webrtc::Clock> clock,
+                 std::shared_ptr<IOStatistics> io_statistics);
   virtual ~RtpVideoSender();
 
  public:
@@ -41,8 +43,10 @@ class RtpVideoSender : public ThreadBase {
 
  private:
   uint32_t ssrc_ = 0;
+  std::shared_ptr<webrtc::Clock> clock_ = nullptr;
   std::unique_ptr<RtpStatistics> rtp_statistics_ = nullptr;
   std::shared_ptr<IOStatistics> io_statistics_ = nullptr;
+  std::unique_ptr<RtpPacketHistory> rtp_packet_history_ = nullptr;
   uint32_t last_send_bytes_ = 0;
   uint32_t last_send_rtcp_sr_packet_ts_ = 0;
   uint32_t total_rtp_payload_sent_ = 0;

src/channel/video_channel_receive.h

@@ -7,7 +7,7 @@
 #ifndef _VIDEO_CHANNEL_RECEIVE_H_
 #define _VIDEO_CHANNEL_RECEIVE_H_
 
-#include "clock.h"
+#include "api/clock/clock.h"
 #include "ice_agent.h"
 #include "rtp_video_receiver.h"
 

src/channel/video_channel_send.cpp

@@ -18,7 +18,8 @@ VideoChannelSend::VideoChannelSend(
       clock_(clock){};
 
 void VideoChannelSend::Initialize(rtp::PAYLOAD_TYPE payload_type) {
-  rtp_video_sender_ = std::make_unique<RtpVideoSender>(ice_io_statistics_);
+  rtp_video_sender_ =
+      std::make_unique<RtpVideoSender>(clock_, ice_io_statistics_);
   rtp_packetizer_ =
       RtpPacketizer::Create(payload_type, rtp_video_sender_->GetSsrc());
   rtp_video_sender_->SetSendDataFunc(

src/channel/video_channel_send.h

@@ -7,9 +7,9 @@
 #ifndef _VIDEO_CHANNEL_SEND_H_
 #define _VIDEO_CHANNEL_SEND_H_
 
+#include "api/clock/clock.h"
 #include "api/transport/network_types.h"
 #include "api/units/timestamp.h"
-#include "clock.h"
 #include "congestion_control.h"
 #include "congestion_control_feedback.h"
 #include "ice_agent.h"

src/common/api/clock/clock.cc

@@ -8,7 +8,7 @@
  *  be found in the AUTHORS file in the root of the source tree.
  */
 
-#include "clock.h"
+#include "api/clock/clock.h"
 
 #include "rtc_base/time_utils.h"
 

src/common/api/clock/clock.h

@@ -16,8 +16,8 @@
 #include <atomic>
 #include <memory>
 
+#include "api/ntp/ntp_time.h"
 #include "api/units/timestamp.h"
-#include "ntp_time.h"
 
 namespace webrtc {
 

src/common/api/ntp/ntp_time_util.cc

@@ -8,7 +8,7 @@
  *  be found in the AUTHORS file in the root of the source tree.
  */
 
-#include "ntp_time_util.h"
+#include "api/ntp/ntp_time_util.h"
 
 #include <algorithm>
 #include <cstdint>

src/common/api/ntp/ntp_time_util.h

@@ -13,8 +13,8 @@
 
 #include <stdint.h>
 
+#include "api/ntp/ntp_time.h"
 #include "api/units/time_delta.h"
-#include "ntp_time.h"
 #include "rtc_base/numerics/safe_conversions.h"
 
 namespace webrtc {

src/qos/congestion_control_feedback_generator.cc

@@ -17,13 +17,13 @@
 #include <utility>
 #include <vector>
 
+#include "api/clock/clock.h"
+#include "api/ntp/ntp_time_util.h"
 #include "api/units/data_rate.h"
 #include "api/units/data_size.h"
 #include "api/units/time_delta.h"
 #include "api/units/timestamp.h"
-#include "clock.h"
 #include "congestion_control_feedback.h"
-#include "ntp_time_util.h"
 #include "rtcp_packet.h"
 #include "rtp_packet_received.h"
 

src/qos/congestion_control_feedback_generator.h

@@ -15,11 +15,11 @@
 #include <memory>
 #include <optional>
 
+#include "api/clock/clock.h"
 #include "api/units/data_rate.h"
 #include "api/units/data_size.h"
 #include "api/units/time_delta.h"
 #include "api/units/timestamp.h"
-#include "clock.h"
 #include "congestion_control_feedback_tracker.h"
 #include "rtp_packet_received.h"
 #include "rtp_transport_feedback_generator.h"

src/qos/nack_requester.h

@@ -13,8 +13,8 @@
 #include <set>
 #include <vector>
 
+#include "api/clock/clock.h"
 #include "api/units/timestamp.h"
-#include "clock.h"
 #include "histogram.h"
 #include "module_common_types.h"
 #include "rtc_base/numerics/sequence_number_util.h"

src/qos/remote_bitrate_estimator.h

@@ -16,9 +16,9 @@
 #include <cstdint>
 #include <vector>
 
+#include "api/clock/clock.h"
 #include "api/units/data_rate.h"
 #include "api/units/time_delta.h"
-#include "clock.h"
 #include "module_common_types.h"
 #include "rtp_packet_received.h"
 

src/qos/remote_bitrate_estimator_abs_send_time.h

@@ -20,11 +20,11 @@
 #include <vector>
 
 #include "aimd_rate_control.h"
+#include "api/clock/clock.h"
 #include "api/units/data_rate.h"
 #include "api/units/data_size.h"
 #include "api/units/time_delta.h"
 #include "api/units/timestamp.h"
-#include "clock.h"
 #include "inter_arrival.h"
 #include "overuse_detector.h"
 #include "overuse_estimator.h"

src/qos/transport_feedback_adapter.cc

@@ -18,11 +18,11 @@
 #include <utility>
 #include <vector>
 
+#include "api/ntp/ntp_time_util.h"
 #include "api/transport/ecn_marking.h"
 #include "api/units/time_delta.h"
 #include "congestion_control_feedback.h"
 #include "log.h"
-#include "ntp_time_util.h"
 #include "rtp_packet_to_send.h"
 
 namespace webrtc {

src/rtp/rtp_packet/rtp_packet_history.cpp

@@ -1,23 +1,105 @@
 #include "rtp_packet_history.h"
 
+#include "log.h"
 #include "sequence_number_compare.h"
 
-RtpPacketHistory::RtpPacketHistory() {}
+RtpPacketHistory::RtpPacketHistory(std::shared_ptr<webrtc::Clock> clock)
+    : clock_(clock),
+      rtt_(webrtc::TimeDelta::MinusInfinity()),
+      number_to_store_(0),
+      packets_inserted_(0) {}
 
 RtpPacketHistory::~RtpPacketHistory() {}
 
-void RtpPacketHistory::AddPacket(std::shared_ptr<RtpPacketToSend> rtp_packet,
-                                 Timestamp send_time) {
-  rtp_packet_history_.push_back(
-      {rtp_packet, send_time, GetPacketIndex(rtp_packet->SequenceNumber())});
+void RtpPacketHistory::SetRtt(webrtc::TimeDelta rtt) {
+  rtt_ = rtt;
+  RemoveDeadPackets();
+}
+
+void RtpPacketHistory::AddPacket(
+    std::shared_ptr<webrtc::RtpPacketToSend> rtp_packet,
+    webrtc::Timestamp send_time) {
+  RemoveDeadPackets();
+  const uint16_t rtp_seq_no = rtp_packet->SequenceNumber();
+  int packet_index = GetPacketIndex(rtp_packet->SequenceNumber());
+  if (packet_index >= 0 &&
+      static_cast<size_t>(packet_index) < rtp_packet_history_.size() &&
+      rtp_packet_history_[packet_index].rtp_packet != nullptr) {
+    LOG_WARN("Duplicate packet inserted: {}", rtp_seq_no);
+    // Remove previous packet to avoid inconsistent state.
+    RemovePacket(packet_index);
+    packet_index = GetPacketIndex(rtp_seq_no);
+  }
+
+  // Packet to be inserted ahead of first packet, expand front.
+  for (; packet_index < 0; ++packet_index) {
+    rtp_packet_history_.emplace_front();
+  }
+  // Packet to be inserted behind last packet, expand back.
+  while (static_cast<int>(rtp_packet_history_.size()) <= packet_index) {
+    rtp_packet_history_.emplace_back();
+  }
+
+  rtp_packet_history_[packet_index] = {rtp_packet, send_time,
+                                       packets_inserted_++};
+}
+
+void RtpPacketHistory::RemoveDeadPackets() {
+  webrtc::Timestamp now = clock_->CurrentTime();
+  webrtc::TimeDelta packet_duration =
+      rtt_.IsFinite()
+          ? (std::max)(kMinPacketDurationRtt * rtt_, kMinPacketDuration)
+          : kMinPacketDuration;
+  while (!rtp_packet_history_.empty()) {
+    if (rtp_packet_history_.size() >= kMaxCapacity) {
+      // We have reached the absolute max capacity, remove one packet
+      // unconditionally.
+      RemovePacket(0);
+      continue;
+    }
+
+    const RtpPacketToSendInfo& stored_packet = rtp_packet_history_.front();
+
+    if (stored_packet.send_time + packet_duration > now) {
+      // Don't cull packets too early to avoid failed retransmission requests.
+      return;
+    }
+
+    if (rtp_packet_history_.size() >= number_to_store_ ||
+        stored_packet.send_time +
+                (packet_duration * kPacketCullingDelayFactor) <=
+            now) {
+      // Too many packets in history, or this packet has timed out. Remove it
+      // and continue.
+      RemovePacket(0);
+    } else {
+      // No more packets can be removed right now.
+      return;
+    }
+  }
+}
+
+std::shared_ptr<webrtc::RtpPacketToSend> RtpPacketHistory::RemovePacket(
+    int packet_index) {
+  // Move the packet out from the StoredPacket container.
+  std::shared_ptr<webrtc::RtpPacketToSend> rtp_packet =
+      std::move(rtp_packet_history_[packet_index].rtp_packet);
+  if (packet_index == 0) {
+    while (!rtp_packet_history_.empty() &&
+           rtp_packet_history_.front().rtp_packet == nullptr) {
+      rtp_packet_history_.pop_front();
+    }
+  }
+
+  return rtp_packet;
 }
 
 int RtpPacketHistory::GetPacketIndex(uint16_t sequence_number) const {
-  if (packet_history_.empty()) {
+  if (rtp_packet_history_.empty()) {
     return 0;
   }
 
-  int first_seq = packet_history_.front().packet_->SequenceNumber();
+  int first_seq = rtp_packet_history_.front().rtp_packet->SequenceNumber();
   if (first_seq == sequence_number) {
     return 0;
   }

src/rtp/rtp_packet/rtp_packet_history.h

@@ -8,31 +8,58 @@
 #define _RTP_PACKET_HISTORY_H_
 
 #include <deque>
+#include <memory>
 
+#include "api/clock/clock.h"
 #include "rtp_packet_to_send.h"
 
 class RtpPacketHistory {
  public:
-  RtpPacketHistory();
+  static constexpr size_t kMaxCapacity = 600;
+  // Maximum number of entries in prioritized queue of padding packets.
+  static constexpr size_t kMaxPaddingHistory = 63;
+  // Don't remove packets within max(1 second, 3x RTT).
+  static constexpr webrtc::TimeDelta kMinPacketDuration =
+      webrtc::TimeDelta::Seconds(1);
+  static constexpr int kMinPacketDurationRtt = 3;
+  // With kStoreAndCull, always remove packets after 3x max(1000ms, 3x rtt).
+  static constexpr int kPacketCullingDelayFactor = 3;
+
+ public:
+  RtpPacketHistory(std::shared_ptr<webrtc::Clock> clock);
   ~RtpPacketHistory();
 
-  void AddPacket(std::shared_ptr<RtpPacketToSend> rtp_packet,
-                 Timestamp send_time);
+ public:
+  void SetRtt(webrtc::TimeDelta rtt);
+  void AddPacket(std::shared_ptr<webrtc::RtpPacketToSend> rtp_packet,
+                 webrtc::Timestamp send_time);
+  void RemoveDeadPackets();
 
  private:
+  std::shared_ptr<webrtc::RtpPacketToSend> RemovePacket(int packet_index);
   int GetPacketIndex(uint16_t sequence_number) const;
 
-  return packet_index;
-}
+ private:
+  struct RtpPacketToSendInfo {
+    RtpPacketToSendInfo() = default;
+    RtpPacketToSendInfo(std::shared_ptr<webrtc::RtpPacketToSend> rtp_packet,
+                        webrtc::Timestamp send_time, uint64_t index)
+        : rtp_packet(rtp_packet), send_time(send_time), index(index) {}
+    RtpPacketToSendInfo(RtpPacketToSendInfo&&) = default;
+    RtpPacketToSendInfo& operator=(RtpPacketToSendInfo&&) = default;
+    ~RtpPacketToSendInfo() = default;
 
-private : struct RtpPacketToSendInfo {
-  std::shared_ptr<RtpPacketToSend> rtp_packet;
-  Timestamp send_time;
-  uint64_t index;
+    std::shared_ptr<webrtc::RtpPacketToSend> rtp_packet;
+    webrtc::Timestamp send_time = webrtc::Timestamp::Zero();
+    uint64_t index;
+  };
+
+ private:
+  std::shared_ptr<webrtc::Clock> clock_;
+  std::deque<RtpPacketToSendInfo> rtp_packet_history_;
+  uint64_t packets_inserted_;
+  webrtc::TimeDelta rtt_;
+  size_t number_to_store_;
 };
 
-private:
-std::deque<std::shared_ptr<RtpPacketToSend>> rtp_packet_history_;
-}
-
 #endif

src/rtp/rtp_packetizer/rtp_packetizer_h264.cpp

@@ -125,11 +125,11 @@ std::vector<std::shared_ptr<RtpPacket>> RtpPacketizerH264::BuildNalu(
 
   if (use_rtp_packet_to_send) {
     std::shared_ptr<webrtc::RtpPacketToSend> rtp_packet =
-        std::make_unique<webrtc::RtpPacketToSend>();
+        std::make_shared<webrtc::RtpPacketToSend>();
     rtp_packet->Build(rtp_packet_frame_.data(), rtp_packet_frame_.size());
     rtp_packets.emplace_back(std::move(rtp_packet));
   } else {
-    std::shared_ptr<RtpPacket> rtp_packet = std::make_unique<RtpPacket>();
+    std::shared_ptr<RtpPacket> rtp_packet = std::make_shared<RtpPacket>();
     rtp_packet->Build(rtp_packet_frame_.data(), rtp_packet_frame_.size());
     rtp_packets.emplace_back(std::move(rtp_packet));
   }
@@ -223,11 +223,11 @@ std::vector<std::shared_ptr<RtpPacket>> RtpPacketizerH264::BuildFua(
 
     if (use_rtp_packet_to_send) {
       std::shared_ptr<webrtc::RtpPacketToSend> rtp_packet =
-          std::make_unique<webrtc::RtpPacketToSend>();
+          std::make_shared<webrtc::RtpPacketToSend>();
       rtp_packet->Build(rtp_packet_frame_.data(), rtp_packet_frame_.size());
       rtp_packets.emplace_back(std::move(rtp_packet));
     } else {
-      std::shared_ptr<RtpPacket> rtp_packet = std::make_unique<RtpPacket>();
+      std::shared_ptr<RtpPacket> rtp_packet = std::make_shared<RtpPacket>();
       rtp_packet->Build(rtp_packet_frame_.data(), rtp_packet_frame_.size());
       rtp_packets.emplace_back(std::move(rtp_packet));
     }

src/transport/ice_transport.cpp

@@ -331,6 +331,13 @@ bool IceTransport::HandleNack(const webrtc::rtcp::CommonHeader &rtcp_block,
   //   rtcp_packet_info->nack.emplace(std::move(nack));
   // }
 
+  // int64_t rtt = rtt_ms();
+  // if (rtt == 0) {
+  //   if (std::optional<TimeDelta> average_rtt = rtcp_receiver_.AverageRtt()) {
+  //     rtt = average_rtt->ms();
+  //   }
+  // }
+
   LOG_INFO("Nack [{}]", nack.packet_ids().size());
 
   return true;

src/transport/ice_transport_controller.h

@@ -7,13 +7,13 @@
 #ifndef _ICE_TRANSPORT_CONTROLLER_H_
 #define _ICE_TRANSPORT_CONTROLLER_H_
 
+#include "api/clock/clock.h"
 #include "api/transport/network_types.h"
 #include "api/units/timestamp.h"
 #include "audio_channel_receive.h"
 #include "audio_channel_send.h"
 #include "audio_decoder.h"
 #include "audio_encoder.h"
-#include "clock.h"
 #include "congestion_control.h"
 #include "congestion_control_feedback.h"
 #include "data_channel_receive.h"

xmake.lua

@@ -47,7 +47,9 @@ target("common")
     "src/common/rtc_base/network/*.cc",
     "src/common/rtc_base/numerics/*.cc",
     "src/common/api/units/*.cc",
-    "src/common/api/transport/*.cc")
+    "src/common/api/transport/*.cc",
+    "src/common/api/clock/*.cc",
+    "src/common/api/ntp/*.cc")
     if not is_os("windows") then
         remove_files("src/common/rtc_base/win32.cc")
     end