diff --git a/CHANGELOG.md b/CHANGELOG.md index 45e0e4958c..715d72d23c 100644 --- a/CHANGELOG.md +++ b/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)). diff --git a/rust/CHANGELOG.md b/rust/CHANGELOG.md index 18e0739cf3..29cf179958 100644 --- a/rust/CHANGELOG.md +++ b/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)). diff --git a/worker/fuzzer/src/RTC/RTP/FuzzerRetransmissionBuffer.cpp b/worker/fuzzer/src/RTC/RTP/FuzzerRetransmissionBuffer.cpp index 5a871a46c5..f87e352983 100644 --- a/worker/fuzzer/src/RTC/RTP/FuzzerRetransmissionBuffer.cpp +++ b/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(Utils::Crypto::GetRandomUInt(0, 1000)); diff --git a/worker/fuzzer/src/RTC/RTP/FuzzerRtpStreamSend.cpp b/worker/fuzzer/src/RTC/RTP/FuzzerRtpStreamSend.cpp index 68c3714a53..3590df898a 100644 --- a/worker/fuzzer/src/RTC/RTP/FuzzerRtpStreamSend.cpp +++ b/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. diff --git a/worker/include/RTC/RTP/RetransmissionBuffer.hpp b/worker/include/RTC/RTP/RetransmissionBuffer.hpp index a64b0825e4..1dd10767e2 100644 --- a/worker/include/RTC/RTP/RetransmissionBuffer.hpp +++ b/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,7 +63,7 @@ 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); @@ -67,7 +71,18 @@ namespace RTC 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 diff --git a/worker/include/RTC/RTP/RtpStreamSend.hpp b/worker/include/RTC/RTP/RtpStreamSend.hpp index 26217da31f..61e0f7868b 100644 --- a/worker/include/RTC/RTP/RtpStreamSend.hpp +++ b/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); diff --git a/worker/include/RTC/RTP/SharedPacket.hpp b/worker/include/RTC/RTP/SharedPacket.hpp index f736a72df5..63f83ea7ea 100644 --- a/worker/include/RTC/RTP/SharedPacket.hpp +++ b/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> sharedPtr; - }; + /** + * Shared ownership of a cloned RTP packet. + */ + using SharedPacket = std::shared_ptr; + + /** + * 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 diff --git a/worker/src/RTC/Consumer.cpp b/worker/src/RTC/Consumer.cpp index b9035de992..197d637d29 100644 --- a/worker/src/RTC/Consumer.cpp +++ b/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; diff --git a/worker/src/RTC/RTP/RetransmissionBuffer.cpp b/worker/src/RTC/RTP/RetransmissionBuffer.cpp index b4a7ee68d6..d282cd1a2c 100644 --- a/worker/src/RTC/RTP/RetransmissionBuffer.cpp +++ b/worker/src/RTC/RTP/RetransmissionBuffer.cpp @@ -12,23 +12,16 @@ namespace RTC { /* Class methods. */ - RetransmissionBuffer::Item* RetransmissionBuffer::CreateItem( - RTP::Packet* packet, const RTP::SharedPacket& sharedPacket, int64_t nowMs) + RetransmissionBuffer::Item* RetransmissionBuffer::CreateItem(RTP::Packet* packet, int64_t nowMs) { MS_TRACE(); - // Store original packet and some extra info into the item. + // Store some extra info into the item. // - // NOTE: sharedPacket could be empty at this point but it's ok since the - // Consumer will fill it. - // - // NOTE: Members are initialized here (rather than assigned after default - // constructing the item) so that `sharedPacket` is copy constructed and - // hence just shares the already existing shared state. Default - // constructing it would allocate a new shared state that would be - // immediately released by the assignment. + // NOTE: `sharedPacket` is left empty here. Insert() returns the item so + // that the Consumer fills it right after, once the packet has been + // restored to its original state. return new Item{ - .sharedPacket = sharedPacket, .encoder = packet->GetPayloadEncoder(), .ssrc = packet->GetSsrc(), .sequenceNumber = packet->GetSequenceNumber(), @@ -119,9 +112,11 @@ namespace RTC * We discard the packet if too old and also discard it if its timestamp does * not properly fit (by ensuring that elements in the buffer are not only * ordered by increasing seq but also that their timestamp are incremental). + * + * @returns The inserted item (with no packet in it yet) or nullptr if the + * packet was not inserted. */ - bool RetransmissionBuffer::Insert( - RTP::Packet* packet, const RTP::SharedPacket& sharedPacket, int64_t nowMs) + RetransmissionBuffer::Item* RetransmissionBuffer::Insert(RTP::Packet* packet, int64_t nowMs) { MS_TRACE(); @@ -131,14 +126,18 @@ namespace RTC MS_DEBUG_DEV("packet [seq:%" PRIu16 ", timestamp:%" PRIu32 "]", seq, timestamp); + Item* item{ nullptr }; + // Buffer is empty, so just insert new item. if (this->buffer.empty()) { MS_DEBUG_DEV("buffer empty [seq:%" PRIu16 ", timestamp:%" PRIu32 "]", seq, timestamp); - this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, sharedPacket, nowMs)); + item = RetransmissionBuffer::CreateItem(packet, nowMs); + + this->buffer.push_back(item); - return true; + return item; } auto* oldestItem = GetOldest(); @@ -160,9 +159,11 @@ namespace RTC Clear(); - this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, sharedPacket, nowMs)); + item = RetransmissionBuffer::CreateItem(packet, nowMs); - return true; + this->buffer.push_back(item); + + return item; } // Clear too old packets in the buffer. @@ -187,9 +188,11 @@ namespace RTC seq, timestamp); - this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, sharedPacket, nowMs)); + item = RetransmissionBuffer::CreateItem(packet, nowMs); + + this->buffer.push_back(item); - return true; + return item; } oldestItem = GetOldest(); @@ -224,9 +227,11 @@ namespace RTC Clear(); - this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, sharedPacket, nowMs)); + item = RetransmissionBuffer::CreateItem(packet, nowMs); - return true; + this->buffer.push_back(item); + + return item; } MS_WARN_TAG( @@ -237,7 +242,7 @@ namespace RTC seq, timestamp); - return false; + return nullptr; } // Calculate how many blank slots it would be necessary to add when @@ -287,7 +292,9 @@ namespace RTC } // Push the packet, which becomes the newest one in the buffer. - this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, sharedPacket, nowMs)); + item = RetransmissionBuffer::CreateItem(packet, nowMs); + + this->buffer.push_back(item); } // Packet arrived out order and its seq is less than seq of the oldest // stored packet, so will become the oldest one in the buffer. @@ -307,7 +314,7 @@ namespace RTC seq, timestamp); - return false; + return nullptr; } // Ensure that the timestamp of the packet is equal or less than the @@ -322,7 +329,7 @@ namespace RTC seq, timestamp); - return false; + return nullptr; } // Calculate how many blank slots it would be necessary to add when @@ -341,7 +348,7 @@ namespace RTC seq, timestamp); - return false; + return nullptr; } // Push blank slots to the front. @@ -351,7 +358,9 @@ namespace RTC } // Insert the packet, which becomes the oldest one in the buffer. - this->buffer.push_front(RetransmissionBuffer::CreateItem(packet, sharedPacket, nowMs)); + item = RetransmissionBuffer::CreateItem(packet, nowMs); + + this->buffer.push_front(item); } // Otherwise packet must be inserted between oldest and newest stored items // so there is already an allocated slot for it. @@ -365,16 +374,14 @@ namespace RTC // Let's check if an item already exist in same position. If so, assume // it's duplicated. - const auto* item = Get(seq); - - if (item) + if (Get(seq)) { MS_DEBUG_DEV( "packet already in the buffer, discarding [seq:%" PRIu16 ", timestamp:%" PRIu32 "]", seq, timestamp); - return false; + return nullptr; } // idx is the intended position of the received packet in the buffer. @@ -407,7 +414,7 @@ namespace RTC seq, timestamp); - return false; + return nullptr; } } @@ -438,12 +445,13 @@ namespace RTC seq, timestamp); - return false; + return nullptr; } } // Store the packet. - this->buffer[idx] = RetransmissionBuffer::CreateItem(packet, sharedPacket, nowMs); + item = RetransmissionBuffer::CreateItem(packet, nowMs); + this->buffer[idx] = item; } MS_ASSERT( @@ -452,7 +460,9 @@ namespace RTC this->buffer.size(), this->maxItems); - return true; + MS_ASSERT(item != nullptr, "inserted item doesn't exist"); + + return item; } void RetransmissionBuffer::Clear() diff --git a/worker/src/RTC/RTP/RtpStreamSend.cpp b/worker/src/RTC/RTP/RtpStreamSend.cpp index 368e694209..80745b0e31 100644 --- a/worker/src/RTC/RTP/RtpStreamSend.cpp +++ b/worker/src/RTC/RTP/RtpStreamSend.cpp @@ -99,8 +99,7 @@ namespace RTC this->rtxSeq = Utils::Crypto::GetRandomUInt(0, 0xFFFF); } - RtpStreamSend::ReceivePacketResult RtpStreamSend::ReceivePacket( - RTP::Packet* packet, const RTP::SharedPacket& sharedPacket) + RtpStreamSend::ReceivePacketResult RtpStreamSend::ReceivePacket(RTP::Packet* packet) { MS_TRACE(); @@ -110,12 +109,12 @@ namespace RTC // Call the parent method. if (!RtpStream::ReceiveStreamPacket(packet)) { - return ReceivePacketResult::DISCARDED; + return { .accepted = false }; } UpdateUnsentSeqNumbers(packet->GetSequenceNumber()); - bool stored{ false }; + RTP::RetransmissionBuffer::Item* retransmissionBufferItem{ nullptr }; // If NACK is enabled, store the packet into the buffer. if (this->retransmissionBuffer) @@ -132,17 +131,17 @@ namespace RTC packet->GetSsrc(), packet->GetSequenceNumber()); - return ReceivePacketResult::DISCARDED; + return { .accepted = false }; } - stored = this->retransmissionBuffer->Insert(packet, sharedPacket, this->shared->GetTimeMs()); + retransmissionBufferItem = + this->retransmissionBuffer->Insert(packet, this->shared->GetTimeMs()); } // Increase transmission counter. this->transmissionCounter.Update(packet); - return stored ? ReceivePacketResult::ACCEPTED_AND_STORED - : ReceivePacketResult::ACCEPTED_AND_NOT_STORED; + return { .accepted = true, .retransmissionBufferItem = retransmissionBufferItem }; } void RtpStreamSend::ReceiveNack(RTC::RTCP::FeedbackRtpNackPacket* nackPacket) @@ -609,14 +608,14 @@ namespace RTC item->sentTimes++; MS_ASSERT( - item->sharedPacket.HasPacket(), + item->sharedPacket, "stored item doesn't contain a packet [ssrc:%" PRIu32 ", seq:%" PRIu16 ", timestamp:%" PRIu32 "]", item->ssrc, item->sequenceNumber, item->timestamp); - auto* const packet = item->sharedPacket.GetPacket(); + auto* const packet = item->sharedPacket.get(); // Keep the values of the original packet received by the Consumer. const auto origSsrc = packet->GetSsrc(); diff --git a/worker/src/RTC/RTP/SharedPacket.cpp b/worker/src/RTC/RTP/SharedPacket.cpp index b93b6ff7a8..c61ffe3dbd 100644 --- a/worker/src/RTC/RTP/SharedPacket.cpp +++ b/worker/src/RTC/RTP/SharedPacket.cpp @@ -15,8 +15,12 @@ namespace RTC // When cloning a RTP packet, a buffer is allocated for it and its length is // the length of the packet plus this value (in bytes). static constexpr size_t PacketBufferLengthIncrement{ 100 }; - // Callback to pass to every cloned RTP packet to deallocate its buffer once - // the packet releases its buffer (for example when the packet is destroyed). +#ifdef MS_DUMP_RTP_SHARED_PACKET_MEMORY_USAGE + // Total memory currently allocated by cloned RTP packets. + static thread_local uint64_t allocatedMemory{ 0 }; +#endif + // Callback given to every cloned RTP packet to deallocate the buffer we + // allocated for it. static thread_local Serializable::BufferReleasedListener PacketBufferReleasedListener = // NOLINTNEXTLINE(misc-unused-parameters, readability-non-const-parameter) [](const Serializable* packet, uint8_t* buffer) @@ -25,7 +29,7 @@ namespace RTC ::operator delete[](buffer, std::align_val_t{ 4 }); #ifdef MS_DUMP_RTP_SHARED_PACKET_MEMORY_USAGE - SharedPacket::allocatedMemory -= packet->GetBufferLength(); + allocatedMemory -= packet->GetBufferLength(); MS_DUMP_CLEAN( 0, @@ -34,78 +38,51 @@ namespace RTC "]", Logger::Pid, packet->GetBufferLength(), - SharedPacket::allocatedMemory); + allocatedMemory); #endif }; - /* Class variables. */ - -#ifdef MS_DUMP_RTP_SHARED_PACKET_MEMORY_USAGE - thread_local uint64_t SharedPacket::allocatedMemory{ 0 }; -#endif - - /* Instance methods. */ - - SharedPacket::SharedPacket() - : sharedPtr(std::make_shared>(nullptr)) - { - MS_TRACE(); - } - - SharedPacket::SharedPacket(RTP::Packet* packet) - : sharedPtr(std::make_shared>(nullptr)) - { - MS_TRACE(); + /* Namespace methods. */ - if (packet) - { - StorePacket(packet); - } - } - - void SharedPacket::Dump(int indentation) const + SharedPacket ClonePacket(const RTP::Packet* packet) { MS_TRACE(); - MS_DUMP_CLEAN(indentation, ""); - MS_DUMP_CLEAN(indentation, " has packet: %s", HasPacket() ? "yes" : "no"); - if (HasPacket()) - { - const auto* packet = GetPacket(); + const size_t bufferLength = packet->GetLength() + PacketBufferLengthIncrement; - packet->Dump(indentation + 1); - } - MS_DUMP_CLEAN(indentation, ""); - } + // NOTE: Buffer must be 4-byte aligned since RTP packet parsing casts it to + // structs (e.g. FixedHeader, HeaderExtension) that require 4-byte alignment. + auto* buffer = static_cast(::operator new[](bufferLength, std::align_val_t{ 4 })); + auto* clonedPacket = packet->Clone(buffer, bufferLength); - void SharedPacket::Assign(RTP::Packet* packet) - { - MS_TRACE(); + // Set a listener in the packet to deallocate the buffer above once the + // packet releases it. + clonedPacket->SetBufferReleasedListener(std::addressof(PacketBufferReleasedListener)); - if (packet) - { - StorePacket(packet); - } - else - { - this->sharedPtr->reset(nullptr); - } - } +#ifdef MS_DUMP_RTP_SHARED_PACKET_MEMORY_USAGE + allocatedMemory += bufferLength; -#ifdef MS_TEST - void SharedPacket::Reset() - { - MS_TRACE(); + MS_DUMP_CLEAN( + 0, + "[worker.pid:%" PRIu64 + "] [RTC::RTP::SharedPacket] memory allocated [packet buffer:%zu, total allocated memory:%" PRIu64 + "]", + Logger::Pid, + clonedPacket->GetBufferLength(), + allocatedMemory); +#endif - this->sharedPtr->reset(nullptr); + // NOTE: The buffer is freed by PacketBufferReleasedListener above and not + // by a custom deleter of the shared pointer. Deleting the cloned packet is + // hence all this shared pointer must do, so the default deleter is used. + return SharedPacket{ clonedPacket }; } -#endif - void SharedPacket::AssertSamePacket(const RTP::Packet* otherPacket) const + void AssertSamePacket(const SharedPacket& sharedPacket, const RTP::Packet* otherPacket) { MS_TRACE(); - const auto* packet = GetPacket(); + const auto* packet = sharedPacket.get(); if (!packet && !otherPacket) { @@ -135,7 +112,7 @@ namespace RTC MS_ASSERT( packet->GetTimestamp() == otherPacket->GetTimestamp(), - "timestamp %" PRIu16 " in packet in sharedPacket != timestamp %" PRIu16 " in otherPacket", + "timestamp %" PRIu32 " in packet in sharedPacket != timestamp %" PRIu32 " in otherPacket", packet->GetTimestamp(), otherPacket->GetTimestamp()); @@ -146,36 +123,5 @@ namespace RTC otherPacket->GetLength()); } } - - void SharedPacket::StorePacket(RTP::Packet* packet) - { - MS_TRACE(); - - const size_t bufferLength = packet->GetLength() + PacketBufferLengthIncrement; - - // NOTE: Buffer must be 4-byte aligned since RTP packet parsing casts it to - // structs (e.g. FixedHeader, HeaderExtension) that require 4-byte alignment. - auto* buffer = static_cast(::operator new[](bufferLength, std::align_val_t{ 4 })); - auto* clonedPacket = packet->Clone(buffer, bufferLength); - - // Set a listener in the packet to deallocate its buffer once the packet - // is destroyed or releases its internal buffer. - clonedPacket->SetBufferReleasedListener(std::addressof(PacketBufferReleasedListener)); - - this->sharedPtr->reset(clonedPacket); - -#ifdef MS_DUMP_RTP_SHARED_PACKET_MEMORY_USAGE - SharedPacket::allocatedMemory += bufferLength; - - MS_DUMP_CLEAN( - 0, - "[worker.pid:%" PRIu64 - "] [RTC::RTP::SharedPacket] memory allocated [packet buffer:%zu, total allocated memory:%" PRIu64 - "]", - Logger::Pid, - clonedPacket->GetBufferLength(), - SharedPacket::allocatedMemory); -#endif - } } // namespace RTP } // namespace RTC diff --git a/worker/src/RTC/Router.cpp b/worker/src/RTC/Router.cpp index efb5b00c6a..44c04a148a 100644 --- a/worker/src/RTC/Router.cpp +++ b/worker/src/RTC/Router.cpp @@ -679,10 +679,10 @@ namespace RTC if (!consumers.empty()) { - // Cloned ref-counted packet that will be filled for as long as needed - // avoiding multiple allocations unless absolutely necessary. - // Clone only happens if needed and only once. - RTC::RTP::SharedPacket sharedPacket; + // Cloned ref-counted packet shared by all Consumers that need to store + // this packet, avoiding multiple allocations. Clone only happens if + // needed and only once. + RTC::RTP::SharedPacket sharedPacket{ nullptr }; for (auto* consumer : consumers) { @@ -698,9 +698,9 @@ namespace RTC // Assert that, if this Consumer filled sharedPacket or it was already // filled, both packet and sharedPacket are the very same RTP packet. - if (sharedPacket.HasPacket()) + if (sharedPacket) { - sharedPacket.AssertSamePacket(packet); + RTC::RTP::AssertSamePacket(sharedPacket, packet); } } } diff --git a/worker/test/src/RTC/RTP/TestRetransmissionBuffer.cpp b/worker/test/src/RTC/RTP/TestRetransmissionBuffer.cpp index 1c32993388..8219a2ca87 100644 --- a/worker/test/src/RTC/RTP/TestRetransmissionBuffer.cpp +++ b/worker/test/src/RTC/RTP/TestRetransmissionBuffer.cpp @@ -42,9 +42,12 @@ SCENARIO("RTP RetransmissionBuffer", "[rtp][rtx]") packet->SetSequenceNumber(seq); packet->SetTimestamp(timestamp); - const RTC::RTP::SharedPacket sharedPacket; + auto* item = RTC::RTP::RetransmissionBuffer::Insert(packet.get(), nowMs); - RTC::RTP::RetransmissionBuffer::Insert(packet.get(), sharedPacket, nowMs); + if (item) + { + item->sharedPacket = RTC::RTP::ClonePacket(packet.get()); + } } void AssertBuffer(std::vector verificationBuffer) diff --git a/worker/test/src/RTC/RTP/TestRtpStreamSend.cpp b/worker/test/src/RTC/RTP/TestRtpStreamSend.cpp index ab39df5296..34509969f3 100644 --- a/worker/test/src/RTC/RTP/TestRtpStreamSend.cpp +++ b/worker/test/src/RTC/RTP/TestRtpStreamSend.cpp @@ -67,7 +67,7 @@ SCENARIO("RtpStreamSend", "[rtp][rtcp][nack][rtpstream][rtpstreamsend]") std::vector> streams, RTC::RTP::Packet* packet) -> void { - RTC::RTP::SharedPacket sharedPacket; + RTC::RTP::SharedPacket sharedPacket{ nullptr }; for (const auto& kv : streams) { @@ -77,18 +77,21 @@ SCENARIO("RtpStreamSend", "[rtp][rtcp][nack][rtpstream][rtpstreamsend]") packet->SetSsrc(ssrc); - const auto result = stream->ReceivePacket(packet, sharedPacket); + const auto result = stream->ReceivePacket(packet); packet->SetSsrc(origSsrc); // NOTE: Here we must replicate the behaviour of Consumer::sendRtpPacket() - // in which, if the shared packet has been stored and it didn't contain the - // packet yet, we fill it with a cloned packet. - if ( - result == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED && - !sharedPacket.HasPacket()) + // in which, if the packet has been stored, we hand a cloned packet (cloned + // just once and shared by all streams) over to the stream. + if (result.retransmissionBufferItem) { - sharedPacket.Assign(packet); + if (!sharedPacket) + { + sharedPacket = RTC::RTP::ClonePacket(packet); + } + + result.retransmissionBufferItem->sharedPacket = sharedPacket; } } }; @@ -960,16 +963,17 @@ SCENARIO("RtpStreamSend", "[rtp][rtcp][nack][rtpstream][rtpstreamsend]") REQUIRE(context1.GetCurrentSpatialLayer() == 0); REQUIRE(context1.GetCurrentTemporalLayer() == 0); - RTC::RTP::SharedPacket sharedPacket; - packet3->SetSsrc(params1.ssrc); // Whenever packet3 is Nacked on stream1, it must always be set a // 00000001 (S0_T1) active decode target bitmas. - const auto result = stream1->ReceivePacket(packet3.get(), sharedPacket); + const auto result = stream1->ReceivePacket(packet3.get()); + + REQUIRE(result.accepted); + REQUIRE(result.retransmissionBufferItem); - REQUIRE(result == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED); + const RTC::RTP::SharedPacket sharedPacket = RTC::RTP::ClonePacket(packet3.get()); - sharedPacket.Assign(packet3.get()); + result.retransmissionBufferItem->sharedPacket = sharedPacket; // Process the third packet with context2 and verify current spatial layers. forwarded = packet3->ProcessPayload(std::addressof(context2), marker); @@ -982,7 +986,12 @@ SCENARIO("RtpStreamSend", "[rtp][rtcp][nack][rtpstream][rtpstreamsend]") // Whenever packet3 is Nacked on stream2, it must always be set a // 00000011 (S0_T1) active decode target bitmas. - stream2->ReceivePacket(packet3.get(), sharedPacket); + const auto result2 = stream2->ReceivePacket(packet3.get()); + + if (result2.retransmissionBufferItem) + { + result2.retransmissionBufferItem->sharedPacket = sharedPacket; + } // Create a NACK item that requests the third packet. RTC::RTCP::FeedbackRtpNackPacket nackPacket(0, params1.ssrc); @@ -1704,15 +1713,15 @@ SCENARIO("RtpStreamSend", "[rtp][rtcp][nack][rtpstream][rtpstreamsend]") const auto stream = std::make_unique( std::addressof(testRtpStreamListener), std::addressof(shared), params, mid); - const RTC::RTP::SharedPacket sharedPacket; + auto result = stream->ReceivePacket(packet.get()); - auto result = stream->ReceivePacket(packet.get(), sharedPacket); + REQUIRE(result.accepted); + REQUIRE(result.retransmissionBufferItem); - REQUIRE(result == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED); + result = stream->ReceivePacket(packet.get()); - result = stream->ReceivePacket(packet.get(), sharedPacket); - - REQUIRE(result == RTC::RTP::RtpStreamSend::ReceivePacketResult::DISCARDED); + REQUIRE(!result.accepted); + REQUIRE(!result.retransmissionBufferItem); } SECTION("packets are not discarded a seq cycle after the timestamp goes back") @@ -1735,17 +1744,16 @@ SCENARIO("RtpStreamSend", "[rtp][rtcp][nack][rtpstream][rtpstreamsend]") const auto stream = std::make_unique( std::addressof(testRtpStreamListener), std::addressof(shared), params, mid); - const RTC::RTP::SharedPacket sharedPacket; - // Packets stored before the timestamp goes back. for (uint16_t seq{ 1000 }; seq < 1010; ++seq) { packet->SetSequenceNumber(seq); packet->SetTimestamp(1000000000 + seq); - const auto result = stream->ReceivePacket(packet.get(), sharedPacket); + const auto result = stream->ReceivePacket(packet.get()); - REQUIRE(result == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED); + REQUIRE(result.accepted); + REQUIRE(result.retransmissionBufferItem); } // The timestamp goes back much further than MaxRetransmissionDelayForVideoMs @@ -1758,15 +1766,16 @@ SCENARIO("RtpStreamSend", "[rtp][rtcp][nack][rtpstream][rtpstreamsend]") packet->SetSequenceNumber(seq); packet->SetTimestamp(timestamp++); - stream->ReceivePacket(packet.get(), sharedPacket); + stream->ReceivePacket(packet.get()); } packet->SetSequenceNumber(1000); packet->SetTimestamp(timestamp); - const auto result = stream->ReceivePacket(packet.get(), sharedPacket); + const auto result = stream->ReceivePacket(packet.get()); - REQUIRE(result == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED); + REQUIRE(result.accepted); + REQUIRE(result.retransmissionBufferItem); } SECTION("Sender Report RTP timestamp is based on the capture instant when known") @@ -2297,9 +2306,9 @@ SCENARIO("RtpStreamSend", "[rtp][rtcp][nack][rtpstream][rtpstreamsend]") packet->SetSsrc(1111); - const RTC::RTP::SharedPacket sharedPacket(packet.get()); + const RTC::RTP::SharedPacket sharedPacket = RTC::RTP::ClonePacket(packet.get()); - stream1->ReceivePacket(packet.get(), sharedPacket); + stream1->ReceivePacket(packet.get()); } std::chrono::duration dur = std::chrono::system_clock::now() - start; @@ -2313,7 +2322,7 @@ SCENARIO("RtpStreamSend", "[rtp][rtcp][nack][rtpstream][rtpstreamsend]") for (size_t i = 0; i < iterations; i++) { - const RTC::RTP::SharedPacket sharedPacket; + const RTC::RTP::SharedPacket sharedPacket{ nullptr }; // Create packet. const std::unique_ptr packet( @@ -2321,7 +2330,7 @@ SCENARIO("RtpStreamSend", "[rtp][rtcp][nack][rtpstream][rtpstreamsend]") packet->SetSsrc(1111); - stream2->ReceivePacket(packet.get(), sharedPacket); + stream2->ReceivePacket(packet.get()); } dur = std::chrono::system_clock::now() - start; diff --git a/worker/test/src/RTC/RTP/TestSharedPacket.cpp b/worker/test/src/RTC/RTP/TestSharedPacket.cpp index cbb868f4e8..4b57e37499 100644 --- a/worker/test/src/RTC/RTP/TestSharedPacket.cpp +++ b/worker/test/src/RTC/RTP/TestSharedPacket.cpp @@ -30,56 +30,47 @@ SCENARIO("RTP SharedPacket", "[rtp][sharedpacket]") { RTC::RTP::SharedPacket sharedPacket; - REQUIRE(!sharedPacket.HasPacket()); - REQUIRE(sharedPacket.GetPacket() == nullptr); + REQUIRE(!sharedPacket); + REQUIRE(sharedPacket.get() == nullptr); - sharedPacket.Assign(packetA); + sharedPacket = RTC::RTP::ClonePacket(packetA); - REQUIRE(sharedPacket.HasPacket()); - compareRtpPackets(sharedPacket.GetPacket(), packetA); + REQUIRE(sharedPacket); + compareRtpPackets(sharedPacket.get(), packetA); - sharedPacket.Reset(); + sharedPacket.reset(); - REQUIRE(!sharedPacket.HasPacket()); - REQUIRE(sharedPacket.GetPacket() == nullptr); + REQUIRE(!sharedPacket); + REQUIRE(sharedPacket.get() == nullptr); delete packetA; delete packetB; } - SECTION("constructor with packet and copy constructor") + SECTION("clone packet and copy constructor") { - // Create sharedPacket1 using constructor with a packet. - RTC::RTP::SharedPacket sharedPacket1(packetA); + // Create sharedPacket1 holding a clone of packetA. + RTC::RTP::SharedPacket sharedPacket1 = RTC::RTP::ClonePacket(packetA); - REQUIRE(sharedPacket1.HasPacket()); - compareRtpPackets(sharedPacket1.GetPacket(), packetA); + REQUIRE(sharedPacket1); + compareRtpPackets(sharedPacket1.get(), packetA); // Create sharedPacket2 using copy constructor. RTC::RTP::SharedPacket sharedPacket2(sharedPacket1); - REQUIRE(sharedPacket2.HasPacket()); - compareRtpPackets(sharedPacket2.GetPacket(), packetA); + REQUIRE(sharedPacket2); + compareRtpPackets(sharedPacket2.get(), packetA); + REQUIRE(sharedPacket1.get() == sharedPacket2.get()); - sharedPacket2.Assign(packetB); + RTC::RTP::AssertSamePacket(sharedPacket1, sharedPacket1.get()); + RTC::RTP::AssertSamePacket(sharedPacket1, sharedPacket2.get()); + RTC::RTP::AssertSamePacket(sharedPacket2, sharedPacket2.get()); + RTC::RTP::AssertSamePacket(sharedPacket2, sharedPacket1.get()); - REQUIRE(sharedPacket1.HasPacket()); - compareRtpPackets(sharedPacket1.GetPacket(), packetB); - REQUIRE(sharedPacket2.HasPacket()); - compareRtpPackets(sharedPacket2.GetPacket(), packetB); - REQUIRE(sharedPacket1.GetPacket() == sharedPacket2.GetPacket()); + sharedPacket1.reset(); - sharedPacket1.AssertSamePacket(sharedPacket1.GetPacket()); - sharedPacket1.AssertSamePacket(sharedPacket2.GetPacket()); - sharedPacket2.AssertSamePacket(sharedPacket2.GetPacket()); - sharedPacket2.AssertSamePacket(sharedPacket1.GetPacket()); - - sharedPacket1.Reset(); - - REQUIRE(!sharedPacket1.HasPacket()); - REQUIRE(sharedPacket1.GetPacket() == nullptr); - REQUIRE(!sharedPacket2.HasPacket()); - REQUIRE(sharedPacket2.GetPacket() == nullptr); + REQUIRE(!sharedPacket1); + REQUIRE(sharedPacket1.get() == nullptr); delete packetA; delete packetB; @@ -87,38 +78,29 @@ SCENARIO("RTP SharedPacket", "[rtp][sharedpacket]") SECTION("copy assignment operator") { - RTC::RTP::SharedPacket sharedPacket1(packetA); + RTC::RTP::SharedPacket sharedPacket1 = RTC::RTP::ClonePacket(packetA); - REQUIRE(sharedPacket1.HasPacket()); - compareRtpPackets(sharedPacket1.GetPacket(), packetA); + REQUIRE(sharedPacket1); + compareRtpPackets(sharedPacket1.get(), packetA); RTC::RTP::SharedPacket sharedPacket2; // Fill sharedPacket2 using copy assignment operator. sharedPacket2 = sharedPacket1; - REQUIRE(sharedPacket2.HasPacket()); - compareRtpPackets(sharedPacket2.GetPacket(), packetA); - - sharedPacket2.Assign(packetB); - - REQUIRE(sharedPacket1.HasPacket()); - compareRtpPackets(sharedPacket1.GetPacket(), packetB); - REQUIRE(sharedPacket2.HasPacket()); - compareRtpPackets(sharedPacket2.GetPacket(), packetB); - REQUIRE(sharedPacket1.GetPacket() == sharedPacket2.GetPacket()); + REQUIRE(sharedPacket2); + compareRtpPackets(sharedPacket2.get(), packetA); + REQUIRE(sharedPacket1.get() == sharedPacket2.get()); - sharedPacket1.AssertSamePacket(sharedPacket1.GetPacket()); - sharedPacket1.AssertSamePacket(sharedPacket2.GetPacket()); - sharedPacket2.AssertSamePacket(sharedPacket2.GetPacket()); - sharedPacket2.AssertSamePacket(sharedPacket1.GetPacket()); + RTC::RTP::AssertSamePacket(sharedPacket1, sharedPacket1.get()); + RTC::RTP::AssertSamePacket(sharedPacket1, sharedPacket2.get()); + RTC::RTP::AssertSamePacket(sharedPacket2, sharedPacket2.get()); + RTC::RTP::AssertSamePacket(sharedPacket2, sharedPacket1.get()); - sharedPacket1.Reset(); + sharedPacket1.reset(); - REQUIRE(!sharedPacket1.HasPacket()); - REQUIRE(sharedPacket1.GetPacket() == nullptr); - REQUIRE(!sharedPacket2.HasPacket()); - REQUIRE(sharedPacket2.GetPacket() == nullptr); + REQUIRE(!sharedPacket1); + REQUIRE(sharedPacket1.get() == nullptr); delete packetA; delete packetB; @@ -126,15 +108,15 @@ SCENARIO("RTP SharedPacket", "[rtp][sharedpacket]") SECTION("assign nullptr") { - RTC::RTP::SharedPacket sharedPacket(packetA); + RTC::RTP::SharedPacket sharedPacket = RTC::RTP::ClonePacket(packetA); - REQUIRE(sharedPacket.HasPacket()); - compareRtpPackets(sharedPacket.GetPacket(), packetA); + REQUIRE(sharedPacket); + compareRtpPackets(sharedPacket.get(), packetA); - sharedPacket.Assign(nullptr); + sharedPacket = nullptr; - REQUIRE(!sharedPacket.HasPacket()); - REQUIRE(sharedPacket.GetPacket() == nullptr); + REQUIRE(!sharedPacket); + REQUIRE(sharedPacket.get() == nullptr); delete packetA; delete packetB; diff --git a/worker/test/src/RTC/TestConsumer.cpp b/worker/test/src/RTC/TestConsumer.cpp index 8093545ed2..73125cdf5b 100644 --- a/worker/test/src/RTC/TestConsumer.cpp +++ b/worker/test/src/RTC/TestConsumer.cpp @@ -248,7 +248,7 @@ SCENARIO("Consumer with SimpleProducerStreamManager", "[rtp][consumer]") { Fixture fixture; auto* packet = RTC::RTP::Packet::Parse(buffer, originalPacketLength + 64); - RTC::RTP::SharedPacket sharedPacket(packet); + RTC::RTP::SharedPacket sharedPacket{ nullptr }; packet->SetPayloadType(payloadType); @@ -267,7 +267,7 @@ SCENARIO("Consumer with SimpleProducerStreamManager", "[rtp][consumer]") dynamic_cast(fixture.consumer.get())->TransportConnected(); auto* packet = RTC::RTP::Packet::Parse(buffer, originalPacketLength + 64); - RTC::RTP::SharedPacket sharedPacket(packet); + RTC::RTP::SharedPacket sharedPacket{ nullptr }; packet->SetPayloadType(payloadType + 1); @@ -285,7 +285,7 @@ SCENARIO("Consumer with SimpleProducerStreamManager", "[rtp][consumer]") dynamic_cast(fixture.consumer.get())->TransportConnected(); auto* packet = RTC::RTP::Packet::Parse(buffer, originalPacketLength + 0); - RTC::RTP::SharedPacket sharedPacket(packet); + RTC::RTP::SharedPacket sharedPacket{ nullptr }; packet->SetPayloadType(payloadType + 1); @@ -303,30 +303,32 @@ SCENARIO("Consumer with SimpleProducerStreamManager", "[rtp][consumer]") dynamic_cast(fixture.consumer.get())->TransportConnected(); auto* packet = RTC::RTP::Packet::Parse(buffer, originalPacketLength + 64); - RTC::RTP::SharedPacket sharedPacket(packet); + RTC::RTP::SharedPacket sharedPacket{ nullptr }; uint16_t seq{ 1 }; + // NOTE: The Router creates a new shared packet for every packet received + // from the Producer, so we must reset it before every send (the Consumer + // fills it with a clone of the packet if it stores it). packet->SetSequenceNumber(seq++); packet->SetPayloadType(payloadType); - sharedPacket.Assign(packet); fixture.consumer->SendRtpPacket(packet, sharedPacket); packet->SetSequenceNumber(seq++); - sharedPacket.Assign(packet); + sharedPacket.reset(); fixture.consumer->SendRtpPacket(packet, sharedPacket); packet->SetSequenceNumber(seq++); - sharedPacket.Assign(packet); + sharedPacket.reset(); fixture.consumer->SendRtpPacket(packet, sharedPacket); packet->SetSequenceNumber(seq++); // Remove the payload so it won't be sent. packet->RemovePayload(); - sharedPacket.Assign(packet); + sharedPacket.reset(); fixture.consumer->SendRtpPacket(packet, sharedPacket);