Propagate arrival time inside NetEq

Bug: webrtc:341266986
Change-Id: I313ded76b884e9ee0f00f43541c8e9aebc406001
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/351340
Reviewed-by: Jakob Ivarsson‎ <jakobi@webrtc.org>
Commit-Queue: Lionel Koenig <lionelk@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#42381}

api/neteq/BUILD.gn

@@ -23,6 +23,7 @@ rtc_source_set("neteq_api") {
     "../../rtc_base:stringutils",
     "../../system_wrappers:system_wrappers",
     "../audio_codecs:audio_codecs_api",
+    "../units:timestamp",
     "//third_party/abseil-cpp/absl/types:optional",
   ]
 }

api/neteq/neteq.h

@@ -23,6 +23,7 @@
 #include "api/audio_codecs/audio_format.h"
 #include "api/rtp_headers.h"
 #include "api/scoped_refptr.h"
+#include "api/units/timestamp.h"
 
 namespace webrtc {
 
@@ -185,8 +186,10 @@ class NetEq {
 
   // Inserts a new packet into NetEq.
   // Returns 0 on success, -1 on failure.
-  virtual int InsertPacket(const RTPHeader& rtp_header,
+  virtual int InsertPacket(
-                           rtc::ArrayView<const uint8_t> payload) = 0;
+      const RTPHeader& rtp_header,
+      rtc::ArrayView<const uint8_t> payload,
+      Timestamp receive_time = Timestamp::MinusInfinity()) = 0;
 
   // Lets NetEq know that a packet arrived with an empty payload. This typically
   // happens when empty packets are used for probing the network channel, and

audio/channel_receive.cc

@@ -196,8 +196,8 @@ class ChannelReceive : public ChannelReceiveInterface,
  private:
   void ReceivePacket(const uint8_t* packet,
                      size_t packet_length,
-                     const RTPHeader& header)
+                     const RTPHeader& header,
-      RTC_RUN_ON(worker_thread_checker_);
+                     Timestamp receive_time) RTC_RUN_ON(worker_thread_checker_);
   int ResendPackets(const uint16_t* sequence_numbers, int length);
   void UpdatePlayoutTimestamp(bool rtcp, int64_t now_ms)
       RTC_RUN_ON(worker_thread_checker_);
@@ -205,7 +205,8 @@ class ChannelReceive : public ChannelReceiveInterface,
   int GetRtpTimestampRateHz() const;
 
   void OnReceivedPayloadData(rtc::ArrayView<const uint8_t> payload,
-                             const RTPHeader& rtpHeader)
+                             const RTPHeader& rtpHeader,
+                             Timestamp receive_time)
       RTC_RUN_ON(worker_thread_checker_);
 
   void InitFrameTransformerDelegate(
@@ -254,7 +255,6 @@ class ChannelReceive : public ChannelReceiveInterface,
   AudioSinkInterface* audio_sink_ = nullptr;
   AudioLevel _outputAudioLevel;
 
-  Clock* const clock_;
   RemoteNtpTimeEstimator ntp_estimator_ RTC_GUARDED_BY(ts_stats_lock_);
 
   // Timestamp of the audio pulled from NetEq.
@@ -320,7 +320,8 @@ class ChannelReceive : public ChannelReceiveInterface,
 
 void ChannelReceive::OnReceivedPayloadData(
     rtc::ArrayView<const uint8_t> payload,
-    const RTPHeader& rtpHeader) {
+    const RTPHeader& rtpHeader,
+    Timestamp receive_time) {
   if (!playing_) {
     // Avoid inserting into NetEQ when we are not playing. Count the
     // packet as discarded.
@@ -335,7 +336,7 @@ void ChannelReceive::OnReceivedPayloadData(
     // the SourceTracker from updating RtpSource information.
     if (source_tracker_) {
       RtpPacketInfos::vector_type packet_vector = {
-          RtpPacketInfo(rtpHeader, clock_->CurrentTime())};
+          RtpPacketInfo(rtpHeader, receive_time)};
       source_tracker_->OnFrameDelivered(RtpPacketInfos(packet_vector));
     }
 
@@ -343,7 +344,7 @@ void ChannelReceive::OnReceivedPayloadData(
   }
 
   // Push the incoming payload (parsed and ready for decoding) into the ACM
-  if (acm_receiver_.InsertPacket(rtpHeader, payload) != 0) {
+  if (acm_receiver_.InsertPacket(rtpHeader, payload, receive_time) != 0) {
     RTC_DLOG(LS_ERROR) << "ChannelReceive::OnReceivedPayloadData() unable to "
                           "push data to the ACM";
     return;
@@ -372,7 +373,9 @@ void ChannelReceive::InitFrameTransformerDelegate(
       receive_audio_callback = [this](rtc::ArrayView<const uint8_t> packet,
                                       const RTPHeader& header) {
         RTC_DCHECK_RUN_ON(&worker_thread_checker_);
-        OnReceivedPayloadData(packet, header);
+        // TODO(lionelk): Get the receive time.
+        OnReceivedPayloadData(packet, header,
+                              /*receive_time=*/Timestamp::MinusInfinity());
       };
   frame_transformer_delegate_ =
       rtc::make_ref_counted<ChannelReceiveFrameTransformerDelegate>(
@@ -545,7 +548,6 @@ ChannelReceive::ChannelReceive(
                               jitter_buffer_fast_playout,
                               jitter_buffer_min_delay_ms)),
       _outputAudioLevel(),
-      clock_(clock),
       ntp_estimator_(clock),
       playout_timestamp_rtp_(0),
       playout_delay_ms_(0),
@@ -663,12 +665,14 @@ void ChannelReceive::OnRtpPacket(const RtpPacketReceived& packet) {
           rtc::saturated_cast<uint32_t>(packet_copy.payload_type_frequency()),
           header.extension.absolute_capture_time);
 
-  ReceivePacket(packet_copy.data(), packet_copy.size(), header);
+  ReceivePacket(packet_copy.data(), packet_copy.size(), header,
+                packet.arrival_time());
 }
 
 void ChannelReceive::ReceivePacket(const uint8_t* packet,
                                    size_t packet_length,
-                                   const RTPHeader& header) {
+                                   const RTPHeader& header,
+                                   Timestamp receive_time) {
   const uint8_t* payload = packet + header.headerLength;
   RTC_DCHECK_GE(packet_length, header.headerLength);
   size_t payload_length = packet_length - header.headerLength;
@@ -688,7 +692,8 @@ void ChannelReceive::ReceivePacket(const uint8_t* packet,
     const FrameDecryptorInterface::Result decrypt_result =
         frame_decryptor_->Decrypt(
             cricket::MEDIA_TYPE_AUDIO, csrcs,
-            /*additional_data=*/nullptr,
+            /*additional_data=*/
+            nullptr,
             rtc::ArrayView<const uint8_t>(payload, payload_data_length),
             decrypted_audio_payload);
 
@@ -720,7 +725,7 @@ void ChannelReceive::ReceivePacket(const uint8_t* packet,
     frame_transformer_delegate_->Transform(payload_data, header, remote_ssrc_,
                                            mime_type.str());
   } else {
-    OnReceivedPayloadData(payload_data, header);
+    OnReceivedPayloadData(payload_data, header, receive_time);
   }
 }
 

modules/audio_coding/BUILD.gn

@@ -47,6 +47,7 @@ rtc_library("audio_coding") {
     "../../api/audio:audio_frame_api",
     "../../api/audio_codecs:audio_codecs_api",
     "../../api/neteq:neteq_api",
+    "../../api/units:timestamp",
     "../../common_audio",
     "../../common_audio:common_audio_c",
     "../../rtc_base:audio_format_to_string",
@@ -712,6 +713,7 @@ rtc_library("neteq") {
     "../../api/neteq:neteq_api",
     "../../api/neteq:neteq_controller_api",
     "../../api/neteq:tick_timer",
+    "../../api/units:timestamp",
     "../../common_audio",
     "../../common_audio:common_audio_c",
     "../../rtc_base:audio_format_to_string",

modules/audio_coding/acm2/acm_receiver.cc

@@ -20,6 +20,7 @@
 #include "api/audio/audio_frame.h"
 #include "api/audio_codecs/audio_decoder.h"
 #include "api/neteq/neteq.h"
+#include "api/units/timestamp.h"
 #include "modules/audio_coding/acm2/acm_resampler.h"
 #include "modules/audio_coding/acm2/call_statistics.h"
 #include "modules/audio_coding/neteq/default_neteq_factory.h"
@@ -104,7 +105,8 @@ int AcmReceiver::last_output_sample_rate_hz() const {
 }
 
 int AcmReceiver::InsertPacket(const RTPHeader& rtp_header,
-                              rtc::ArrayView<const uint8_t> incoming_payload) {
+                              rtc::ArrayView<const uint8_t> incoming_payload,
+                              Timestamp receive_time) {
   if (incoming_payload.empty()) {
     neteq_->InsertEmptyPacket(rtp_header);
     return 0;
@@ -139,7 +141,7 @@ int AcmReceiver::InsertPacket(const RTPHeader& rtp_header,
     }
   }  // `mutex_` is released.
 
-  if (neteq_->InsertPacket(rtp_header, incoming_payload) < 0) {
+  if (neteq_->InsertPacket(rtp_header, incoming_payload, receive_time) < 0) {
     RTC_LOG(LS_ERROR) << "AcmReceiver::InsertPacket "
                       << static_cast<int>(rtp_header.payloadType)
                       << " Failed to insert packet";

modules/audio_coding/acm2/acm_receiver.h

@@ -26,6 +26,7 @@
 #include "api/audio_codecs/audio_format.h"
 #include "api/neteq/neteq.h"
 #include "api/neteq/neteq_factory.h"
+#include "api/units/timestamp.h"
 #include "modules/audio_coding/acm2/acm_resampler.h"
 #include "modules/audio_coding/acm2/call_statistics.h"
 #include "modules/audio_coding/include/audio_coding_module_typedefs.h"
@@ -68,13 +69,15 @@ class AcmReceiver {
   //                            information about payload type, sequence number,
   //                            timestamp, SSRC and marker bit.
   //   - incoming_payload     : Incoming audio payload.
-  //   - length_payload       : Length of incoming audio payload in bytes.
+  //   - receive_time         : Timestamp when the packet has been seen on the
+  //                            network card.
   //
   // Return value             : 0 if OK.
   //                           <0 if NetEq returned an error.
   //
   int InsertPacket(const RTPHeader& rtp_header,
-                   rtc::ArrayView<const uint8_t> incoming_payload);
+                   rtc::ArrayView<const uint8_t> incoming_payload,
+                   Timestamp receive_time = Timestamp::MinusInfinity());
 
   //
   // Asks NetEq for 10 milliseconds of decoded audio.

modules/audio_coding/neteq/neteq_impl.cc

@@ -193,11 +193,12 @@ NetEqImpl::NetEqImpl(const NetEq::Config& config,
 NetEqImpl::~NetEqImpl() = default;
 
 int NetEqImpl::InsertPacket(const RTPHeader& rtp_header,
-                            rtc::ArrayView<const uint8_t> payload) {
+                            rtc::ArrayView<const uint8_t> payload,
+                            Timestamp receive_time) {
   rtc::MsanCheckInitialized(payload);
   TRACE_EVENT0("webrtc", "NetEqImpl::InsertPacket");
   MutexLock lock(&mutex_);
-  if (InsertPacketInternal(rtp_header, payload) != 0) {
+  if (InsertPacketInternal(rtp_header, payload, receive_time) != 0) {
     return kFail;
   }
   return kOK;
@@ -464,13 +465,12 @@ NetEq::Operation NetEqImpl::last_operation_for_test() const {
 // Methods below this line are private.
 
 int NetEqImpl::InsertPacketInternal(const RTPHeader& rtp_header,
-                                    rtc::ArrayView<const uint8_t> payload) {
+                                    rtc::ArrayView<const uint8_t> payload,
+                                    Timestamp receive_time) {
   if (payload.empty()) {
     RTC_LOG_F(LS_ERROR) << "payload is empty";
     return kInvalidPointer;
   }
-
-  Timestamp receive_time = clock_->CurrentTime();
   stats_->ReceivedPacket();
 
   PacketList packet_list;

modules/audio_coding/neteq/neteq_impl.h

@@ -13,6 +13,7 @@
 
 #include <map>
 #include <memory>
+#include <optional>
 #include <string>
 #include <utility>
 #include <vector>
@@ -24,6 +25,7 @@
 #include "api/neteq/neteq_controller_factory.h"
 #include "api/neteq/tick_timer.h"
 #include "api/rtp_packet_info.h"
+#include "api/units/timestamp.h"
 #include "modules/audio_coding/neteq/audio_multi_vector.h"
 #include "modules/audio_coding/neteq/expand_uma_logger.h"
 #include "modules/audio_coding/neteq/packet.h"
@@ -126,8 +128,10 @@ class NetEqImpl : public webrtc::NetEq {
   NetEqImpl& operator=(const NetEqImpl&) = delete;
 
   // Inserts a new packet into NetEq. Returns 0 on success, -1 on failure.
-  int InsertPacket(const RTPHeader& rtp_header,
+  int InsertPacket(
-                   rtc::ArrayView<const uint8_t> payload) override;
+      const RTPHeader& rtp_header,
+      rtc::ArrayView<const uint8_t> payload,
+      Timestamp receive_time = Timestamp::MinusInfinity()) override;
 
   void InsertEmptyPacket(const RTPHeader& rtp_header) override;
 
@@ -204,7 +208,8 @@ class NetEqImpl : public webrtc::NetEq {
   // above. Returns 0 on success, otherwise an error code.
   // TODO(hlundin): Merge this with InsertPacket above?
   int InsertPacketInternal(const RTPHeader& rtp_header,
-                           rtc::ArrayView<const uint8_t> payload)
+                           rtc::ArrayView<const uint8_t> payload,
+                           Timestamp receive_time)
       RTC_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
   // Returns true if the payload type changed (this should be followed by

modules/audio_coding/neteq/neteq_impl_unittest.cc

@@ -494,7 +494,8 @@ TEST_F(NetEqImplTest, VerifyTimestampPropagation) {
   // Insert one packet.
   clock_.AdvanceTimeMilliseconds(123456);
   Timestamp expected_receive_time = clock_.CurrentTime();
-  EXPECT_EQ(NetEq::kOK, neteq_->InsertPacket(rtp_header, payload));
+  EXPECT_EQ(NetEq::kOK,
+            neteq_->InsertPacket(rtp_header, payload, expected_receive_time));
 
   // Pull audio once.
   const size_t kMaxOutputSize = static_cast<size_t>(10 * kSampleRateHz / 1000);
@@ -588,7 +589,9 @@ TEST_F(NetEqImplTest, ReorderedPacket) {
   // Insert one packet.
   clock_.AdvanceTimeMilliseconds(123456);
   Timestamp expected_receive_time = clock_.CurrentTime();
-  EXPECT_EQ(NetEq::kOK, neteq_->InsertPacket(rtp_header, payload));
+  EXPECT_EQ(NetEq::kOK,
+            neteq_->InsertPacket(rtp_header, payload,
+                                 /*receive_time=*/clock_.CurrentTime()));
 
   // Pull audio once.
   const size_t kMaxOutputSize = static_cast<size_t>(10 * kSampleRateHz / 1000);
@@ -618,14 +621,18 @@ TEST_F(NetEqImplTest, ReorderedPacket) {
   rtp_header.extension.set_audio_level(AudioLevel(/*voice_activity=*/false, 1));
   payload[0] = 1;
   clock_.AdvanceTimeMilliseconds(1000);
-  EXPECT_EQ(NetEq::kOK, neteq_->InsertPacket(rtp_header, payload));
+  EXPECT_EQ(NetEq::kOK,
+            neteq_->InsertPacket(rtp_header, payload,
+                                 /*receive_time=*/clock_.CurrentTime()));
   rtp_header.sequenceNumber += 2;
   rtp_header.timestamp += 2 * kPayloadLengthSamples;
   rtp_header.extension.set_audio_level(AudioLevel(/*voice_activity=*/false, 2));
   payload[0] = 2;
   clock_.AdvanceTimeMilliseconds(2000);
   expected_receive_time = clock_.CurrentTime();
-  EXPECT_EQ(NetEq::kOK, neteq_->InsertPacket(rtp_header, payload));
+  EXPECT_EQ(NetEq::kOK,
+            neteq_->InsertPacket(rtp_header, payload,
+                                 /*receive_time=*/clock_.CurrentTime()));
 
   // Expect only the second packet to be decoded (the one with "2" as the first
   // payload byte).