RemoteClockOffsetEstimator class

worker/include/RTC/RTP/Packet.hpp

@@ -838,6 +838,22 @@ namespace RTC
 				return this->payloadDescriptorHandler->GetTemporalLayer();
 			}
 
+			/**
+			 * Capture time (in ms) of the packet.
+			 */
+			std::optional<uint64_t> GetCaptureMs() const
+			{
+				return this->captureMs;
+			}
+
+			/**
+			 * Set the capture time (in ms) of the packet.
+			 */
+			void SetCaptureMs(uint64_t captureMs)
+			{
+				this->captureMs = captureMs;
+			}
+
 		private:
 			/**
 			 * @remarks
@@ -960,6 +976,8 @@ namespace RTC
 			RTP::HeaderExtensionIds headerExtensionIds{};
 			// Codec related.
 			std::shared_ptr<Codecs::PayloadDescriptorHandler> payloadDescriptorHandler;
+			// Capture time of the packet.
+			std::optional<uint64_t> captureMs;
 		};
 	} // namespace RTP
 } // namespace RTC

worker/include/RTC/RTP/RtpStreamSend.hpp

@@ -14,9 +14,9 @@ namespace RTC
 		{
 		public:
 			// Maximum retransmission buffer size for video (ms).
-			static const uint32_t MaxRetransmissionDelayForVideoMs;
+			static constexpr uint32_t MaxRetransmissionDelayForVideoMs{ 2000 };
 			// Maximum retransmission buffer size for audio (ms).
-			static const uint32_t MaxRetransmissionDelayForAudioMs;
+			static constexpr uint32_t MaxRetransmissionDelayForAudioMs{ 1000 };
 
 		public:
 			enum class ReceivePacketResult : uint8_t

worker/include/RTC/RemoteClockOffsetEstimator.hpp

@@ -0,0 +1,87 @@
+#ifndef MS_RTC_REMOTE_CLOCK_OFFSET_ESTIMATOR_HPP
+#define MS_RTC_REMOTE_CLOCK_OFFSET_ESTIMATOR_HPP
+
+#include "common.hpp"
+#include <vector>
+
+namespace RTC
+{
+	/**
+	 * Estimates the offset between the wall clock of a remote sender, as reported in
+	 * the NTP field of the RTCP Sender Reports it sends, and mediasoup's own monotonic
+	 * clock. Both are expressed in milliseconds, so the estimated offset satisfies:
+	 *
+	 *   localMs = remoteMs + offsetMs
+	 *
+	 * Each sample is the difference between the arrival time of a Sender Report and
+	 * the NTP value it carries, so it holds the clock offset plus the one way delay
+	 * of that Sender Report. That delay is removed with half of the RTT when known,
+	 * and the median of a sliding window is taken so that transient delay spikes are
+	 * rejected.
+	 *
+	 * A single instance is meant to be shared by all the RTP streams of a given
+	 * CNAME. Those streams come from the same machine and hence from the same wall
+	 * clock, and using a different offset for each of them would reintroduce the
+	 * very inter stream skew this is meant to remove.
+	 *
+	 * @remarks
+	 * - Based on the RemoteNtpTimeEstimator class of libwebrtc.
+	 */
+	class RemoteClockOffsetEstimator
+	{
+	public:
+		/**
+		 * Number of most recent samples the median is computed over.
+		 */
+		static constexpr size_t WindowSize{ 7 };
+		/**
+		 * Number of samples required before an offset is reported.
+		 */
+		static constexpr size_t MinSampleCount{ 3 };
+
+	public:
+		RemoteClockOffsetEstimator();
+
+	public:
+		/**
+		 * Feed a received RTCP Sender Report.
+		 *
+		 * @param remoteNtpMs - NTP field of the Sender Report, in milliseconds.
+		 * @param localArrivalMs - Our local time at which the Sender Report arrived.
+		 * @param rttMs - RTT towards the sender, or 0 if not known yet.
+		 */
+		void AddSenderReport(uint64_t remoteNtpMs, uint64_t localArrivalMs, uint32_t rttMs);
+
+		/**
+		 * The estimated offset, or no value while less than `MinSampleCount` samples
+		 * have been gathered.
+		 */
+		std::optional<int64_t> GetOffsetMs() const
+		{
+			return this->offsetMs;
+		}
+
+		/**
+		 * Translate a time expressed in the remote sender's wall clock into our own
+		 * monotonic clock. Returns no value if there is no offset yet or if the given
+		 * time does not map into our clock.
+		 */
+		std::optional<uint64_t> RemoteMsToLocalMs(uint64_t remoteMs) const;
+
+		void Reset();
+
+	private:
+		void UpdateOffsetMs();
+
+	private:
+		// Most recent samples, oldest first.
+		std::vector<int64_t> samples;
+		// Arrival time of the last accepted Sender Report, so that all the Sender
+		// Reports of a same compound packet produce a single sample.
+		uint64_t lastLocalArrivalMs{ 0 };
+		// Median of the samples in the window.
+		std::optional<int64_t> offsetMs;
+	};
+} // namespace RTC
+
+#endif

worker/meson.build

@@ -124,6 +124,7 @@ common_sources = [
   'src/RTC/PortManager.cpp',
   'src/RTC/Producer.cpp',
   'src/RTC/RateCalculator.cpp',
+  'src/RTC/RemoteClockOffsetEstimator.cpp',
   'src/RTC/Router.cpp',
   'src/RTC/RtcLogger.cpp',
   'src/RTC/RtpListener.cpp',
@@ -429,6 +430,7 @@ test_sources = [
   'test/src/RTC/TestNackGenerator.cpp',
   'test/src/RTC/TestPortManager.cpp',
   'test/src/RTC/TestRateCalculator.cpp',
+  'test/src/RTC/TestRemoteClockOffsetEstimator.cpp',
   'test/src/RTC/TestRtpEncodingParameters.cpp',
   'test/src/RTC/TestSeqManager.cpp',
   'test/src/RTC/TestSubchannelsCodec.cpp',

worker/src/RTC/RTP/Packet.cpp

@@ -391,6 +391,11 @@ namespace RTC
 			MS_DUMP_CLEAN(indentation, "  padding length: %" PRIu8, GetPaddingLength());
 			MS_DUMP_CLEAN(indentation, "  padded to 4 bytes: %s", IsPaddedTo4Bytes() ? "yes" : "no");
 
+			if (GetCaptureMs())
+			{
+				MS_DUMP_CLEAN(indentation, "  capture time (ms):%" PRIu64, GetCaptureMs().value());
+			}
+
 			if (this->payloadDescriptorHandler)
 			{
 				MS_DUMP_CLEAN(indentation + 1, "<PayloadDescriptorHandler>");
@@ -422,6 +427,9 @@ namespace RTC
 			// Clone extension ids.
 			clonedPacket->headerExtensionIds = this->headerExtensionIds;
 
+			// Clone capture time.
+			clonedPacket->captureMs = this->captureMs;
+
 			// Assign the payload descriptor handler.
 			clonedPacket->payloadDescriptorHandler = this->payloadDescriptorHandler;
 

worker/src/RTC/RTP/RtpStreamSend.cpp

@@ -21,11 +21,6 @@ namespace RTC
 		  MaxRequestedPackets + 1);
 		static constexpr uint32_t DefaultRtt{ 100u };
 
-		/* Class Static. */
-
-		const uint32_t RtpStreamSend::MaxRetransmissionDelayForVideoMs{ 2000u };
-		const uint32_t RtpStreamSend::MaxRetransmissionDelayForAudioMs{ 1000u };
-
 		/* Instance methods. */
 
 		RtpStreamSend::RtpStreamSend(

worker/src/RTC/RemoteClockOffsetEstimator.cpp

@@ -0,0 +1,106 @@
+#define MS_CLASS "RTC::RemoteClockOffsetEstimator"
+// #define MS_LOG_DEV_LEVEL 3
+
+#include "RTC/RemoteClockOffsetEstimator.hpp"
+#include "Logger.hpp"
+
+namespace RTC
+{
+	/* Instance methods. */
+
+	RemoteClockOffsetEstimator::RemoteClockOffsetEstimator()
+	{
+		MS_TRACE();
+
+		this->samples.reserve(RemoteClockOffsetEstimator::WindowSize);
+	}
+
+	void RemoteClockOffsetEstimator::AddSenderReport(
+	  uint64_t remoteNtpMs, uint64_t localArrivalMs, uint32_t rttMs)
+	{
+		MS_TRACE();
+
+		// Ignore Sender Reports with no NTP timestamp.
+		if (remoteNtpMs == 0)
+		{
+			MS_DEBUG_DEV("ignoring Sender Report with no NTP timestamp");
+
+			return;
+		}
+
+		// Ignore a Sender Report belonging to a compound packet already accounted
+		// for. Otherwise a single delayed compound packet would contribute as many
+		// samples as streams it reports about, and hence bias the median.
+		if (localArrivalMs == this->lastLocalArrivalMs)
+		{
+			return;
+		}
+
+		this->lastLocalArrivalMs = localArrivalMs;
+
+		// The sample holds the clock offset plus the one way delay of this Sender
+		// Report. Assume a symmetric path and remove half of the RTT.
+		const int64_t sample = static_cast<int64_t>(localArrivalMs) -
+		                       static_cast<int64_t>(remoteNtpMs) - (static_cast<int64_t>(rttMs) / 2);
+
+		if (this->samples.size() == RemoteClockOffsetEstimator::WindowSize)
+		{
+			this->samples.erase(this->samples.begin());
+		}
+
+		this->samples.push_back(sample);
+
+		UpdateOffsetMs();
+	}
+
+	std::optional<uint64_t> RemoteClockOffsetEstimator::RemoteMsToLocalMs(uint64_t remoteMs) const
+	{
+		MS_TRACE();
+
+		if (!this->offsetMs.has_value())
+		{
+			return std::nullopt;
+		}
+
+		const int64_t localMs = static_cast<int64_t>(remoteMs) + this->offsetMs.value();
+
+		// The given time does not map into our clock, so the input is bogus.
+		if (localMs < 0)
+		{
+			MS_WARN_2TAGS(
+			  rtp, rtcp, "remote time does not map into our clock [remoteMs:%" PRIu64 "]", remoteMs);
+
+			return std::nullopt;
+		}
+
+		return static_cast<uint64_t>(localMs);
+	}
+
+	void RemoteClockOffsetEstimator::Reset()
+	{
+		MS_TRACE();
+
+		this->samples.clear();
+		this->lastLocalArrivalMs = 0;
+		this->offsetMs.reset();
+	}
+
+	void RemoteClockOffsetEstimator::UpdateOffsetMs()
+	{
+		MS_TRACE();
+
+		if (this->samples.size() < RemoteClockOffsetEstimator::MinSampleCount)
+		{
+			return;
+		}
+
+		// Take the median of the window. While the window is not full its size may
+		// be even, in which case the upper of the two middle samples is taken.
+		std::vector<int64_t> sortedSamples(this->samples);
+		const auto middle = sortedSamples.begin() + (sortedSamples.size() / 2);
+
+		std::nth_element(sortedSamples.begin(), middle, sortedSamples.end());
+
+		this->offsetMs = *middle;
+	}
+} // namespace RTC

worker/test/src/RTC/RTP/TestPacket.cpp

@@ -1059,12 +1059,16 @@ SCENARIO("RTP Packet", "[serializable][rtp][packet]")
 		  /*paddingLength*/ 0);
 
 		REQUIRE(packet->IsPaddedTo4Bytes() == true);
+		REQUIRE(packet->GetCaptureMs() == std::nullopt);
 
 		packet->SetPayloadType(100);
 		packet->SetMarker(true);
 		packet->SetSequenceNumber(12345);
 		packet->SetTimestamp(987654321);
 		packet->SetSsrc(1234567890);
+		packet->SetCaptureMs(99998888);
+
+		REQUIRE(packet->GetCaptureMs() == 99998888);
 
 		std::vector<RTC::RTP::Packet::Extension> extensions;
 
@@ -1204,6 +1208,7 @@ SCENARIO("RTP Packet", "[serializable][rtp][packet]")
 		REQUIRE(extensionLen == 3);
 
 		REQUIRE(packet->IsPaddedTo4Bytes() == true);
+		REQUIRE(packet->GetCaptureMs() == 99998888);
 
 		/* Clone it. */
 
@@ -1252,6 +1257,7 @@ SCENARIO("RTP Packet", "[serializable][rtp][packet]")
 		REQUIRE(extensionLen == 3);
 
 		REQUIRE(packet->IsPaddedTo4Bytes() == true);
+		REQUIRE(packet->GetCaptureMs() == 99998888);
 
 		/* Set payload. */
 

worker/test/src/RTC/TestPortManager.cpp

@@ -10,7 +10,7 @@
 // merged unrelated bindings. This scenario locks down the post-fix
 // behavior: distinct tuples produce distinct keys, equal tuples produce equal
 // keys.
-SCENARIO("PortManager", "[rtc][portmanager]")
+SCENARIO("PortManager", "[portmanager]")
 {
 	// Helper: build an IPv4 `sockaddr_storage` from a dotted-quad string + port=0.
 	auto makeV4 = [](const char* dottedQuad)