Worker: Get rid of RTC::RTP::SharedPacket class

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ### NEXT
 
+- Worker: Get rid of `RTC::RTP::SharedPacket` class ([PR #2012](https://github.com/versatica/mediasoup/pull/2012)).
+
 ### 3.31.0
 
 - Fix simulcast RTP timestamp jumps on layer switch and frozen retransmission buffer ([PR #2002](https://github.com/versatica/mediasoup/pull/2002)).

rust/CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ### NEXT
 
+- Worker: Get rid of `RTC::RTP::SharedPacket` class ([PR #2012](https://github.com/versatica/mediasoup/pull/2012)).
+
 ### 0.32.0
 
 - Fix simulcast RTP timestamp jumps on layer switch and frozen retransmission buffer ([PR #2002](https://github.com/versatica/mediasoup/pull/2002)).

worker/fuzzer/src/RTC/RTP/FuzzerRetransmissionBuffer.cpp

@@ -31,13 +31,14 @@ void FuzzerRtcRtpRetransmissionBuffer::Fuzz(const uint8_t* data, size_t len)
 
 	while (len >= 4)
 	{
-		const RTC::RTP::SharedPacket sharedPacket;
-
 		// Set 'random' sequence number and timestamp.
 		packet->SetSequenceNumber(Utils::Byte::Get2Bytes(data, offset));
 		packet->SetTimestamp(Utils::Byte::Get4Bytes(data, offset));
 
-		retransmissionBuffer.Insert(packet, sharedPacket, nowMs);
+		if (auto* item = retransmissionBuffer.Insert(packet, nowMs))
+		{
+			item->sharedPacket = RTC::RTP::ClonePacket(packet);
+		}
 
 		// Let some 'random' time pass and check whether the packet is too old.
 		nowMs += static_cast<int64_t>(Utils::Crypto::GetRandomUInt<uint64_t>(0, 1000));

worker/fuzzer/src/RTC/RTP/FuzzerRtpStreamSend.cpp

@@ -48,13 +48,16 @@ void FuzzerRtcRtpStreamSend::Fuzz(const uint8_t* data, size_t len)
 
 	while (len >= 12)
 	{
-		const RTC::RTP::SharedPacket sharedPacket;
-
 		// Set 'random' sequence number and timestamp.
 		packet->SetSequenceNumber(Utils::Byte::Get2Bytes(data, offset));
 		packet->SetTimestamp(Utils::Byte::Get4Bytes(data, offset + 2));
 
-		stream->ReceivePacket(packet, sharedPacket);
+		const auto result = stream->ReceivePacket(packet);
+
+		if (result.retransmissionBufferItem)
+		{
+			result.retransmissionBufferItem->sharedPacket = RTC::RTP::ClonePacket(packet);
+		}
 
 		// Feed a 'random' Receiver Report, which is what the loss of the stream is
 		// worked out from.

worker/include/RTC/RTP/RetransmissionBuffer.hpp

@@ -22,6 +22,10 @@ namespace RTC
 			{
 				/**
 				 * Original packet.
+				 *
+				 * @remarks
+				 * - Insert() leaves it empty and returns the item so that the caller
+				 *   fills it once the packet has been restored to its original state.
 				 */
 				RTP::SharedPacket sharedPacket{ nullptr };
 				/**
@@ -59,15 +63,26 @@ namespace RTC
 			};
 
 		private:
-			static Item* CreateItem(RTP::Packet* packet, const RTP::SharedPacket& sharedPacket, int64_t nowMs);
+			static Item* CreateItem(RTP::Packet* packet, int64_t nowMs);
 
 		public:
 			RetransmissionBuffer(uint16_t maxItems, int64_t maxRetransmissionDelayMs, uint32_t clockRate);
 			~RetransmissionBuffer();
 
 			void Dump(int indentation = 0) const;
 			Item* Get(uint16_t seq) const;
-			bool Insert(RTP::Packet* packet, const RTP::SharedPacket& sharedPacket, int64_t nowMs);
+			/**
+			 * Insert the given packet into the buffer.
+			 *
+			 * @returns The inserted item, whose `sharedPacket` is still empty, or
+			 *   nullptr if the packet was not inserted.
+			 *
+			 * @remarks
+			 * - The returned item is owned by the buffer and is only valid until the
+			 *   next call that may evict items (Insert(), Clear() or the buffer
+			 *   destruction), so the caller must fill it right away.
+			 */
+			Item* Insert(RTP::Packet* packet, int64_t nowMs);
 			void Clear();
 			/**
 			 * Whether the given stored item was stored longer than the maximum

worker/include/RTC/RTP/RtpStreamSend.hpp

@@ -104,11 +104,25 @@ namespace RTC
 			};
 
 		public:
-			enum class ReceivePacketResult : uint8_t
+			struct ReceivePacketResult
 			{
-				DISCARDED,
-				ACCEPTED_AND_NOT_STORED,
-				ACCEPTED_AND_STORED
+				/**
+				 * Whether the packet was accepted by the stream.
+				 */
+				bool accepted{ false };
+				/**
+				 * The retransmission buffer item the packet info was stored into, or
+				 * nullptr if it was not stored (because the packet was discarded or
+				 * because NACK is not enabled in this stream).
+				 *
+				 * @remarks
+				 * - Its `sharedPacket` is still empty and must be filled by the caller
+				 *   once the packet has been restored to its original state.
+				 * - The item is owned by the retransmission buffer and is only valid
+				 *   until the next call that may evict items from it, so the caller
+				 *   must fill it right away.
+				 */
+				RTP::RetransmissionBuffer::Item* retransmissionBufferItem{ nullptr };
 			};
 
 		public:
@@ -152,7 +166,7 @@ namespace RTC
 
 			void SetRtx(uint8_t payloadType, uint32_t ssrc) override;
 
-			ReceivePacketResult ReceivePacket(RTP::Packet* packet, const RTP::SharedPacket& sharedPacket);
+			ReceivePacketResult ReceivePacket(RTP::Packet* packet);
 
 			void ReceiveNack(RTC::RTCP::FeedbackRtpNackPacket* nackPacket);
 

worker/include/RTC/RTP/SharedPacket.hpp

@@ -8,96 +8,31 @@ namespace RTC
 {
 	namespace RTP
 	{
-		class SharedPacket
-		{
-		public:
-#ifdef MS_DUMP_RTP_SHARED_PACKET_MEMORY_USAGE
-			static thread_local uint64_t allocatedMemory;
-#endif
-
-		public:
-			/**
-			 * Empty constructor.
-			 */
-			SharedPacket();
-
-			/**
-			 * Constructor with RTP packet pointer. If a packet is given it's internally
-			 * cloned.
-			 */
-			explicit SharedPacket(RTP::Packet* packet);
-
-			/**
-			 * Copy constructor.
-			 *
-			 * @remarks
-			 * No need to declare it but let's be explicit.
-			 */
-			SharedPacket(const SharedPacket&) = default;
-
-			/**
-			 * Copy assignment operator.
-			 *
-			 * @remarks
-			 * - No need to declare it but let's be explicit.
-			 *
-			 * @throws MediasoupError if the packet is too big.
-			 */
-			SharedPacket& operator=(const SharedPacket&) = default;
-
-			/**
-			 * Destructor.
-			 */
-			~SharedPacket() = default;
-
-		public:
-			void Dump(int indentation = 0) const;
-
-			bool HasPacket() const
-			{
-				return this->sharedPtr->get() != nullptr;
-			}
-
-			RTP::Packet* GetPacket() const
-			{
-				return this->sharedPtr->get();
-			}
-
-			/**
-			 * Assign given packet (could be nullptr). If packet is given it's internally
-			 * cloned.
-			 *
-			 * @throws MediasoupError if the packet is too big.
-			 */
-			void Assign(RTP::Packet* packet);
-
-#ifdef MS_TEST
-			/**
-			 * Resets the internal packet to nullptr.
-			 *
-			 * @remarks
-			 * This affects to ALL copies of this SharedPacket object.
-			 */
-			void Reset();
-#endif
-
-			/**
-			 * Assert that RTP packet contained in this SharedPacket is a clone of the
-			 * given other packet (or there is no packet inside and no other packet has
-			 * been given).
-			 */
-			void AssertSamePacket(const RTP::Packet* otherPacket) const;
-
-		private:
-			void StorePacket(RTP::Packet* packet);
-
-		private:
-			// NOTE: This needs to be a shared pointer that holds an unique pointer.
-			// Otherwise, when copying/storing the shared pointer in other locations
-			// (buffers, etc), reseting its internal value wouldn't affect other copies
-			// of the shared pointer.
-			std::shared_ptr<std::unique_ptr<RTP::Packet>> sharedPtr;
-		};
+		/**
+		 * Shared ownership of a cloned RTP packet.
+		 */
+		using SharedPacket = std::shared_ptr<RTP::Packet>;
+
+		/**
+		 * Clone the given RTP packet into a newly allocated buffer and return shared
+		 * ownership of the clone.
+		 *
+		 * @remarks
+		 * The buffer allocated for the clone is deallocated by the clone itself
+		 * (via its buffer released listener) once it's destroyed, so the caller just
+		 * needs to drop every reference to the returned SharedPacket.
+		 */
+		SharedPacket ClonePacket(const RTP::Packet* packet);
+
+		/**
+		 * Assert that the given SharedPacket holds a clone of the given RTP packet.
+		 *
+		 * @remarks
+		 * A SharedPacket corresponds to a single packet received by a Producer, so
+		 * this must be asserted wherever an already existing clone (made by another
+		 * Consumer of the same Producer) is reused rather than a new one made.
+		 */
+		void AssertSamePacket(const SharedPacket& sharedPacket, const RTP::Packet* otherPacket);
 	} // namespace RTP
 } // namespace RTC
 

worker/src/RTC/Consumer.cpp

@@ -1375,10 +1375,9 @@ namespace RTC
 			  origTimestamp);
 		}
 
-		const RTC::RTP::RtpStreamSend::ReceivePacketResult result =
-		  rtpStream->ReceivePacket(packet, sharedPacket);
+		const auto result = rtpStream->ReceivePacket(packet);
 
-		if (result != RTC::RTP::RtpStreamSend::ReceivePacketResult::DISCARDED)
+		if (result.accepted)
 		{
 			if (rtpSeqManager->GetMaxOutput() == packet->GetSequenceNumber())
 			{
@@ -1418,11 +1417,27 @@ namespace RTC
 		// Restore the original payload if needed.
 		packet->RestorePayload();
 
-		// If sharedPacket doesn't have a packet inside and it has been stored we
-		// need to clone the packet into it.
-		if (!sharedPacket.HasPacket() && result == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED)
-		{
-			sharedPacket.Assign(packet);
+		// If the packet was accepted and the info has been stored into
+		// the retransmission buffer then we must hand a clone of it over to the
+		// stream. Clone it only once only if needed.
+		//
+		// NOTE: This must happen here, once the packet has been restored above, so
+		// that the clone holds the original packet and not the one rewritten for
+		// this Consumer (the stream reapplies its own values and payload encoding on
+		// every retransmission).
+		if (result.retransmissionBufferItem)
+		{
+			if (!sharedPacket)
+			{
+				sharedPacket = RTC::RTP::ClonePacket(packet);
+			}
+
+			MS_ASSERT(
+			  !result.retransmissionBufferItem->sharedPacket,
+			  "retransmission buffer item already holds a packet [seq:%" PRIu16 "]",
+			  result.retransmissionBufferItem->sequenceNumber);
+
+			result.retransmissionBufferItem->sharedPacket = sharedPacket;
 		}
 
 		// If sent packet was the first packet of a key frame, let's send buffered
@@ -1432,12 +1447,12 @@ namespace RTC
 		{
 			// NOTE: Only send buffered packets if the first packet containing the
 			// key frame was sent.
-			if (result != RTC::RTP::RtpStreamSend::ReceivePacketResult::DISCARDED)
+			if (result.accepted)
 			{
 				for (auto& kv : *targetLayerRetransmissionBuffer)
 				{
 					auto& bufferedSharedPacket = kv.second;
-					auto* bufferedPacket       = bufferedSharedPacket.GetPacket();
+					auto* bufferedPacket       = bufferedSharedPacket.get();
 
 					if (bufferedPacket->GetSequenceNumber() > origSeq)
 					{
@@ -1882,15 +1897,15 @@ namespace RTC
 
 		// Store original packet into the buffer. Only clone once and only if
 		// necessary.
-		if (!sharedPacket.HasPacket())
+		if (!sharedPacket)
 		{
-			sharedPacket.Assign(packet);
+			sharedPacket = RTC::RTP::ClonePacket(packet);
 		}
-		// Assert that, if sharedPacket was already filled, both packet and
-		// sharedPacket are the very same RTP packet.
+		// Assert that the clone already made by another Consumer of the same
+		// Producer is a clone of this very same packet.
 		else
 		{
-			sharedPacket.AssertSamePacket(packet);
+			RTC::RTP::AssertSamePacket(sharedPacket, packet);
 		}
 
 		targetLayerRetransmissionBuffer[packet->GetSequenceNumber()] = sharedPacket;