From 9732740fb90bb8c54b0f62a9781c0ff7cfd14936 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Luis=20Mill=C3=A1n?= Date: Fri, 23 Oct 2026 11:11:15 +0200 Subject: [PATCH 1/8] Worker: Get rid of RTC::RTP::SharedPacket class `std::shared_ptr>` nests two owning smart pointers to obtain something neither of them is meant for: a mutable slot that every copy of the handle observes. The packet is owned by the inner unique pointer; the outer shared pointer is only there so that filling that slot is visible through the copies already handed out. It costs a shared state that must be allocated before there is any packet to share, plus one extra indirection on every access. That slot was only needed because the shared packet was copied into the retransmission buffer while still empty and had to be filled afterwards anyway, once `RtpStreamSend::ReceivePacket()` told the Consumer that the packet had been stored. Fill the buffer item instead of the shared packet, so a plain `std::shared_ptr` is enough: - `RetransmissionBuffer::Insert()` no longer takes the shared packet, and new `RetransmissionBuffer::StorePacket()` fills the already inserted item. - `RtpStreamSend::StoreSharedPacket()` is called by the Consumer right after `ReceivePacket()` returns `ACCEPTED_AND_STORED`. - `SharedPacket` is now just a typedef plus `ClonePacket()` and `AssertSamePacket()`. Cloning behaviour is unchanged (same condition, same moment and same clone shared among Consumers). This also removes a `malloc`/`free` pair from the per packet path, which the Router was paying for every dispatched RTP packet even when no Consumer ended up storing it. Note: this also makes `RetransmissionBuffer::Item` cheap to default construct, with the previous type every blank slot would have allocated a shared state of its own. Holding the items by value, saving an allocation per stored packet and avoiding a pointer chase when walking the buffer, can now be done as a follow up. --- .../RTC/RTP/FuzzerRetransmissionBuffer.cpp | 7 +- .../src/RTC/RTP/FuzzerRtpStreamSend.cpp | 9 +- .../include/RTC/RTP/RetransmissionBuffer.hpp | 5 +- worker/include/RTC/RTP/RtpStreamSend.hpp | 13 +- worker/include/RTC/RTP/SharedPacket.hpp | 115 +++--------- worker/meson.build | 1 - worker/src/RTC/Consumer.cpp | 39 ++-- worker/src/RTC/RTP/RetransmissionBuffer.cpp | 53 +++--- worker/src/RTC/RTP/RtpStreamSend.cpp | 18 +- worker/src/RTC/RTP/SharedPacket.cpp | 166 ++++++------------ worker/src/RTC/Router.cpp | 12 +- .../src/RTC/RTP/TestRetransmissionBuffer.cpp | 7 +- worker/test/src/RTC/RTP/TestRtpStreamSend.cpp | 54 +++--- worker/test/src/RTC/RTP/TestSharedPacket.cpp | 142 --------------- worker/test/src/RTC/TestConsumer.cpp | 18 +- 15 files changed, 225 insertions(+), 434 deletions(-) delete mode 100644 worker/test/src/RTC/RTP/TestSharedPacket.cpp diff --git a/worker/fuzzer/src/RTC/RTP/FuzzerRetransmissionBuffer.cpp b/worker/fuzzer/src/RTC/RTP/FuzzerRetransmissionBuffer.cpp index 5a871a46c5..d2305c715a 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 (retransmissionBuffer.Insert(packet, nowMs)) + { + retransmissionBuffer.StorePacket(packet->GetSequenceNumber(), 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..3a3718cf44 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 == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED) + { + stream->StoreSharedPacket(packet->GetSequenceNumber(), 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..6a7690e188 100644 --- a/worker/include/RTC/RTP/RetransmissionBuffer.hpp +++ b/worker/include/RTC/RTP/RetransmissionBuffer.hpp @@ -59,7 +59,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 +67,8 @@ 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); + bool Insert(RTP::Packet* packet, int64_t nowMs); + void StorePacket(uint16_t seq, const RTP::SharedPacket& sharedPacket); 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..f479c9af49 100644 --- a/worker/include/RTC/RTP/RtpStreamSend.hpp +++ b/worker/include/RTC/RTP/RtpStreamSend.hpp @@ -152,7 +152,18 @@ 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); + /** + * Store the given cloned packet into the retransmission buffer item with + * the given sequence number. + * + * @remarks + * Must be called right after ReceivePacket() returned + * ACCEPTED_AND_STORED, once the packet has been restored to its original + * state, giving the sequence number that the packet had when it was + * passed to ReceivePacket(). + */ + void StoreSharedPacket(uint16_t seq, const RTP::SharedPacket& sharedPacket); void ReceiveNack(RTC::RTCP::FeedbackRtpNackPacket* nackPacket); diff --git a/worker/include/RTC/RTP/SharedPacket.hpp b/worker/include/RTC/RTP/SharedPacket.hpp index f736a72df5..ebdd3bd434 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* packet); } // namespace RTP } // namespace RTC diff --git a/worker/meson.build b/worker/meson.build index 06963fd019..eccee3a215 100644 --- a/worker/meson.build +++ b/worker/meson.build @@ -539,7 +539,6 @@ test_sources = [ 'test/src/RTC/RTP/TestRetransmissionBuffer.cpp', 'test/src/RTC/RTP/TestRtpStreamSend.cpp', 'test/src/RTC/RTP/TestRtpStreamRecv.cpp', - 'test/src/RTC/RTP/TestSharedPacket.cpp', 'test/src/RTC/RTP/Codecs/TestH264.cpp', 'test/src/RTC/RTP/Codecs/TestVP8.cpp', 'test/src/RTC/RTP/Codecs/TestVP9.cpp', diff --git a/worker/src/RTC/Consumer.cpp b/worker/src/RTC/Consumer.cpp index b9035de992..46c1dd3464 100644 --- a/worker/src/RTC/Consumer.cpp +++ b/worker/src/RTC/Consumer.cpp @@ -1375,8 +1375,7 @@ 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) { @@ -1418,11 +1417,25 @@ 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 == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED) + { + if (!sharedPacket) + { + sharedPacket = RTC::RTP::ClonePacket(packet); + } + + // NOTE: We must give `seq` (the sequence number this stream stored the + // packet with) and not the packet sequence number, already restored to + // its original value above. + rtpStream->StoreSharedPacket(seq, sharedPacket); } // If sent packet was the first packet of a key frame, let's send buffered @@ -1437,7 +1450,7 @@ namespace RTC for (auto& kv : *targetLayerRetransmissionBuffer) { auto& bufferedSharedPacket = kv.second; - auto* bufferedPacket = bufferedSharedPacket.GetPacket(); + auto* bufferedPacket = bufferedSharedPacket.get(); if (bufferedPacket->GetSequenceNumber() > origSeq) { @@ -1882,15 +1895,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..45593761c0 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. The Consumer fills it via + // StorePacket() right after this Insert() call, once the packet has been + // restored to its original state. return new Item{ - .sharedPacket = sharedPacket, .encoder = packet->GetPayloadEncoder(), .ssrc = packet->GetSsrc(), .sequenceNumber = packet->GetSequenceNumber(), @@ -120,8 +113,7 @@ namespace RTC * not properly fit (by ensuring that elements in the buffer are not only * ordered by increasing seq but also that their timestamp are incremental). */ - bool RetransmissionBuffer::Insert( - RTP::Packet* packet, const RTP::SharedPacket& sharedPacket, int64_t nowMs) + bool RetransmissionBuffer::Insert(RTP::Packet* packet, int64_t nowMs) { MS_TRACE(); @@ -136,7 +128,7 @@ namespace RTC { MS_DEBUG_DEV("buffer empty [seq:%" PRIu16 ", timestamp:%" PRIu32 "]", seq, timestamp); - this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, sharedPacket, nowMs)); + this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, nowMs)); return true; } @@ -160,7 +152,7 @@ namespace RTC Clear(); - this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, sharedPacket, nowMs)); + this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, nowMs)); return true; } @@ -187,7 +179,7 @@ namespace RTC seq, timestamp); - this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, sharedPacket, nowMs)); + this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, nowMs)); return true; } @@ -224,7 +216,7 @@ namespace RTC Clear(); - this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, sharedPacket, nowMs)); + this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, nowMs)); return true; } @@ -287,7 +279,7 @@ namespace RTC } // Push the packet, which becomes the newest one in the buffer. - this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, sharedPacket, nowMs)); + this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, nowMs)); } // 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. @@ -351,7 +343,7 @@ namespace RTC } // Insert the packet, which becomes the oldest one in the buffer. - this->buffer.push_front(RetransmissionBuffer::CreateItem(packet, sharedPacket, nowMs)); + this->buffer.push_front(RetransmissionBuffer::CreateItem(packet, nowMs)); } // Otherwise packet must be inserted between oldest and newest stored items // so there is already an allocated slot for it. @@ -443,7 +435,7 @@ namespace RTC } // Store the packet. - this->buffer[idx] = RetransmissionBuffer::CreateItem(packet, sharedPacket, nowMs); + this->buffer[idx] = RetransmissionBuffer::CreateItem(packet, nowMs); } MS_ASSERT( @@ -455,6 +447,25 @@ namespace RTC return true; } + void RetransmissionBuffer::StorePacket(uint16_t seq, const RTP::SharedPacket& sharedPacket) + { + MS_TRACE(); + + MS_ASSERT(sharedPacket, "given sharedPacket is empty"); + + auto* item = Get(seq); + + MS_ASSERT(item, "no item with given seq in the buffer [seq:%" PRIu16 "]", seq); + + MS_ASSERT( + !item->sharedPacket, + "item already holds a packet [seq:%" PRIu16 ", item seq:%" PRIu16 "]", + seq, + item->sequenceNumber); + + item->sharedPacket = sharedPacket; + } + void RetransmissionBuffer::Clear() { MS_TRACE(); diff --git a/worker/src/RTC/RTP/RtpStreamSend.cpp b/worker/src/RTC/RTP/RtpStreamSend.cpp index 368e694209..d7b50a23e2 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(); @@ -135,7 +134,7 @@ namespace RTC return ReceivePacketResult::DISCARDED; } - stored = this->retransmissionBuffer->Insert(packet, sharedPacket, this->shared->GetTimeMs()); + stored = this->retransmissionBuffer->Insert(packet, this->shared->GetTimeMs()); } // Increase transmission counter. @@ -145,6 +144,15 @@ namespace RTC : ReceivePacketResult::ACCEPTED_AND_NOT_STORED; } + void RtpStreamSend::StoreSharedPacket(uint16_t seq, const RTP::SharedPacket& sharedPacket) + { + MS_TRACE(); + + MS_ASSERT(this->retransmissionBuffer, "no retransmission buffer"); + + this->retransmissionBuffer->StorePacket(seq, sharedPacket); + } + void RtpStreamSend::ReceiveNack(RTC::RTCP::FeedbackRtpNackPacket* nackPacket) { MS_TRACE(); @@ -609,14 +617,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..ba7af8f16f 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,148 +38,90 @@ 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* packet) { MS_TRACE(); - const auto* packet = GetPacket(); + const auto* clonedPacket = sharedPacket.get(); - if (!packet && !otherPacket) + if (!clonedPacket && !packet) { return; } - else if (packet && !otherPacket) + else if (clonedPacket && !packet) { - MS_ABORT("there is a packet in sharedPacket but given otherPacket doesn't have value"); + MS_ABORT("there is a packet in sharedPacket but given packet doesn't have value"); } - else if (!packet && otherPacket) + else if (!clonedPacket && packet) { - MS_ABORT("there is no packet in sharedPacket but given otherPacket has value"); + MS_ABORT("there is no packet in sharedPacket but given packet has value"); } else { MS_ASSERT( - packet->GetSsrc() == otherPacket->GetSsrc(), - "SSRC %" PRIu32 " in packet in sharedPacket != SSRC %" PRIu32 " in otherPacket", - packet->GetSsrc(), - otherPacket->GetSsrc()); + clonedPacket->GetSsrc() == packet->GetSsrc(), + "SSRC %" PRIu32 " in packet in sharedPacket != SSRC %" PRIu32 " in given packet", + clonedPacket->GetSsrc(), + packet->GetSsrc()); MS_ASSERT( - packet->GetSequenceNumber() == otherPacket->GetSequenceNumber(), - "seq %" PRIu16 " in packet in sharedPacket != seq %" PRIu16 " in otherPacket", - packet->GetSequenceNumber(), - otherPacket->GetSequenceNumber()); + clonedPacket->GetSequenceNumber() == packet->GetSequenceNumber(), + "seq %" PRIu16 " in packet in sharedPacket != seq %" PRIu16 " in given packet", + clonedPacket->GetSequenceNumber(), + packet->GetSequenceNumber()); MS_ASSERT( - packet->GetTimestamp() == otherPacket->GetTimestamp(), - "timestamp %" PRIu16 " in packet in sharedPacket != timestamp %" PRIu16 " in otherPacket", - packet->GetTimestamp(), - otherPacket->GetTimestamp()); + clonedPacket->GetTimestamp() == packet->GetTimestamp(), + "timestamp %" PRIu32 " in packet in sharedPacket != timestamp %" PRIu32 " in given packet", + clonedPacket->GetTimestamp(), + packet->GetTimestamp()); MS_ASSERT( - packet->GetLength() == otherPacket->GetLength(), - "length %zu of packet in sharedPacket != length %zu of otherPacket", - packet->GetLength(), - otherPacket->GetLength()); + clonedPacket->GetLength() == packet->GetLength(), + "length %zu of packet in sharedPacket != length %zu of given packet", + clonedPacket->GetLength(), + packet->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..a064750804 100644 --- a/worker/test/src/RTC/RTP/TestRetransmissionBuffer.cpp +++ b/worker/test/src/RTC/RTP/TestRetransmissionBuffer.cpp @@ -42,9 +42,10 @@ SCENARIO("RTP RetransmissionBuffer", "[rtp][rtx]") packet->SetSequenceNumber(seq); packet->SetTimestamp(timestamp); - const RTC::RTP::SharedPacket sharedPacket; - - RTC::RTP::RetransmissionBuffer::Insert(packet.get(), sharedPacket, nowMs); + if (RTC::RTP::RetransmissionBuffer::Insert(packet.get(), nowMs)) + { + RTC::RTP::RetransmissionBuffer::StorePacket(seq, 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..3333203753 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 == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED) { - sharedPacket.Assign(packet); + if (!sharedPacket) + { + sharedPacket = RTC::RTP::ClonePacket(packet); + } + + stream->StoreSharedPacket(packet->GetSequenceNumber(), sharedPacket); } } }; @@ -960,16 +963,16 @@ 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 == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED); - sharedPacket.Assign(packet3.get()); + const RTC::RTP::SharedPacket sharedPacket = RTC::RTP::ClonePacket(packet3.get()); + + stream1->StoreSharedPacket(packet3->GetSequenceNumber(), sharedPacket); // Process the third packet with context2 and verify current spatial layers. forwarded = packet3->ProcessPayload(std::addressof(context2), marker); @@ -982,7 +985,10 @@ 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); + if (stream2->ReceivePacket(packet3.get()) == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED) + { + stream2->StoreSharedPacket(packet3->GetSequenceNumber(), sharedPacket); + } // Create a NACK item that requests the third packet. RTC::RTCP::FeedbackRtpNackPacket nackPacket(0, params1.ssrc); @@ -1704,13 +1710,11 @@ 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(), sharedPacket); + auto result = stream->ReceivePacket(packet.get()); REQUIRE(result == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED); - result = stream->ReceivePacket(packet.get(), sharedPacket); + result = stream->ReceivePacket(packet.get()); REQUIRE(result == RTC::RTP::RtpStreamSend::ReceivePacketResult::DISCARDED); } @@ -1735,15 +1739,13 @@ 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); } @@ -1758,13 +1760,13 @@ 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); } @@ -2297,9 +2299,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 +2315,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 +2323,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 deleted file mode 100644 index cbb868f4e8..0000000000 --- a/worker/test/src/RTC/RTP/TestSharedPacket.cpp +++ /dev/null @@ -1,142 +0,0 @@ -#include "common.hpp" -#include "RTC/RTP/Packet.hpp" -#include "RTC/RTP/SharedPacket.hpp" -#include "test/include/RTC/RTP/rtpCommon.hpp" // in worker/test/include/ -#include - -SCENARIO("RTP SharedPacket", "[rtp][sharedpacket]") -{ - const auto compareRtpPackets = [](const RTC::RTP::Packet* packet1, const RTC::RTP::Packet* packet2) - { - REQUIRE(packet1->GetSsrc() == packet2->GetSsrc()); - REQUIRE(packet1->GetSequenceNumber() == packet2->GetSequenceNumber()); - REQUIRE(packet1->GetTimestamp() == packet2->GetTimestamp()); - REQUIRE(packet1->GetLength() == packet2->GetLength()); - }; - - auto* packetA = RTC::RTP::Packet::Factory(rtpCommon::FactoryBuffer, 2000); - - packetA->SetSequenceNumber(1111); - packetA->SetTimestamp(111111); - packetA->SetSsrc(11111111); - - auto* packetB = RTC::RTP::Packet::Factory(rtpCommon::FactoryBuffer + 2000, 2000); - - packetB->SetSequenceNumber(2222); - packetB->SetTimestamp(222222); - packetB->SetSsrc(22222222); - - SECTION("default constructor and assign later") - { - RTC::RTP::SharedPacket sharedPacket; - - REQUIRE(!sharedPacket.HasPacket()); - REQUIRE(sharedPacket.GetPacket() == nullptr); - - sharedPacket.Assign(packetA); - - REQUIRE(sharedPacket.HasPacket()); - compareRtpPackets(sharedPacket.GetPacket(), packetA); - - sharedPacket.Reset(); - - REQUIRE(!sharedPacket.HasPacket()); - REQUIRE(sharedPacket.GetPacket() == nullptr); - - delete packetA; - delete packetB; - } - - SECTION("constructor with packet and copy constructor") - { - // Create sharedPacket1 using constructor with a packet. - RTC::RTP::SharedPacket sharedPacket1(packetA); - - REQUIRE(sharedPacket1.HasPacket()); - compareRtpPackets(sharedPacket1.GetPacket(), packetA); - - // Create sharedPacket2 using copy constructor. - RTC::RTP::SharedPacket 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()); - - 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); - - delete packetA; - delete packetB; - } - - SECTION("copy assignment operator") - { - RTC::RTP::SharedPacket sharedPacket1(packetA); - - REQUIRE(sharedPacket1.HasPacket()); - compareRtpPackets(sharedPacket1.GetPacket(), 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()); - - 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); - - delete packetA; - delete packetB; - } - - SECTION("assign nullptr") - { - RTC::RTP::SharedPacket sharedPacket(packetA); - - REQUIRE(sharedPacket.HasPacket()); - compareRtpPackets(sharedPacket.GetPacket(), packetA); - - sharedPacket.Assign(nullptr); - - REQUIRE(!sharedPacket.HasPacket()); - REQUIRE(sharedPacket.GetPacket() == 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); From f46e442e3b5216310ef816f37ae851ee88a10cde Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Luis=20Mill=C3=A1n?= Date: Fri, 23 Oct 2026 12:00:52 +0200 Subject: [PATCH 2/8] clang-tidy --- worker/include/RTC/RTP/RetransmissionBuffer.hpp | 2 +- worker/src/RTC/RTP/RetransmissionBuffer.cpp | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/worker/include/RTC/RTP/RetransmissionBuffer.hpp b/worker/include/RTC/RTP/RetransmissionBuffer.hpp index 6a7690e188..9736f6f5d9 100644 --- a/worker/include/RTC/RTP/RetransmissionBuffer.hpp +++ b/worker/include/RTC/RTP/RetransmissionBuffer.hpp @@ -68,7 +68,7 @@ namespace RTC void Dump(int indentation = 0) const; Item* Get(uint16_t seq) const; bool Insert(RTP::Packet* packet, int64_t nowMs); - void StorePacket(uint16_t seq, const RTP::SharedPacket& sharedPacket); + void StorePacket(uint16_t seq, const RTP::SharedPacket& sharedPacket) const; void Clear(); /** * Whether the given stored item was stored longer than the maximum diff --git a/worker/src/RTC/RTP/RetransmissionBuffer.cpp b/worker/src/RTC/RTP/RetransmissionBuffer.cpp index 45593761c0..c968845384 100644 --- a/worker/src/RTC/RTP/RetransmissionBuffer.cpp +++ b/worker/src/RTC/RTP/RetransmissionBuffer.cpp @@ -447,7 +447,7 @@ namespace RTC return true; } - void RetransmissionBuffer::StorePacket(uint16_t seq, const RTP::SharedPacket& sharedPacket) + void RetransmissionBuffer::StorePacket(uint16_t seq, const RTP::SharedPacket& sharedPacket) const { MS_TRACE(); From 1aa6043bd5728cad58bec0c84e6cd704cf6a6159 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Luis=20Mill=C3=A1n?= Date: Fri, 23 Oct 2026 12:22:07 +0200 Subject: [PATCH 3/8] CHANGELOG --- CHANGELOG.md | 2 ++ rust/CHANGELOG.md | 2 ++ 2 files changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 45e0e4958c..62c3ca7eef 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ### NEXT +- Worker: Get rid of 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..bdbaa5c12e 100644 --- a/rust/CHANGELOG.md +++ b/rust/CHANGELOG.md @@ -2,6 +2,8 @@ ### NEXT +- Worker: Get rid of 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)). From 22ebe98e2773c35ecc6f8c829d4489dff12bd7d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Luis=20Mill=C3=A1n?= Date: Fri, 23 Oct 2026 16:14:41 +0200 Subject: [PATCH 4/8] CHANGELOG --- CHANGELOG.md | 2 +- rust/CHANGELOG.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 62c3ca7eef..715d72d23c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ ### NEXT -- Worker: Get rid of Worker: Get rid of RTC::RTP::SharedPacket class ([PR #2012](https://github.com/versatica/mediasoup/pull/2012)). +- Worker: Get rid of `RTC::RTP::SharedPacket` class ([PR #2012](https://github.com/versatica/mediasoup/pull/2012)). ### 3.31.0 diff --git a/rust/CHANGELOG.md b/rust/CHANGELOG.md index bdbaa5c12e..29cf179958 100644 --- a/rust/CHANGELOG.md +++ b/rust/CHANGELOG.md @@ -2,7 +2,7 @@ ### NEXT -- Worker: Get rid of Worker: Get rid of RTC::RTP::SharedPacket class ([PR #2012](https://github.com/versatica/mediasoup/pull/2012)). +- Worker: Get rid of `RTC::RTP::SharedPacket` class ([PR #2012](https://github.com/versatica/mediasoup/pull/2012)). ### 0.32.0 From cfbfadaac5c44e690287c05fead317b4043ec59b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Luis=20Mill=C3=A1n?= Date: Fri, 23 Oct 2026 16:25:08 +0200 Subject: [PATCH 5/8] Add TestSharedPacket.cpp --- worker/meson.build | 1 + worker/test/src/RTC/RTP/TestSharedPacket.cpp | 124 +++++++++++++++++++ 2 files changed, 125 insertions(+) create mode 100644 worker/test/src/RTC/RTP/TestSharedPacket.cpp diff --git a/worker/meson.build b/worker/meson.build index eccee3a215..06963fd019 100644 --- a/worker/meson.build +++ b/worker/meson.build @@ -539,6 +539,7 @@ test_sources = [ 'test/src/RTC/RTP/TestRetransmissionBuffer.cpp', 'test/src/RTC/RTP/TestRtpStreamSend.cpp', 'test/src/RTC/RTP/TestRtpStreamRecv.cpp', + 'test/src/RTC/RTP/TestSharedPacket.cpp', 'test/src/RTC/RTP/Codecs/TestH264.cpp', 'test/src/RTC/RTP/Codecs/TestVP8.cpp', 'test/src/RTC/RTP/Codecs/TestVP9.cpp', diff --git a/worker/test/src/RTC/RTP/TestSharedPacket.cpp b/worker/test/src/RTC/RTP/TestSharedPacket.cpp new file mode 100644 index 0000000000..4b57e37499 --- /dev/null +++ b/worker/test/src/RTC/RTP/TestSharedPacket.cpp @@ -0,0 +1,124 @@ +#include "common.hpp" +#include "RTC/RTP/Packet.hpp" +#include "RTC/RTP/SharedPacket.hpp" +#include "test/include/RTC/RTP/rtpCommon.hpp" // in worker/test/include/ +#include + +SCENARIO("RTP SharedPacket", "[rtp][sharedpacket]") +{ + const auto compareRtpPackets = [](const RTC::RTP::Packet* packet1, const RTC::RTP::Packet* packet2) + { + REQUIRE(packet1->GetSsrc() == packet2->GetSsrc()); + REQUIRE(packet1->GetSequenceNumber() == packet2->GetSequenceNumber()); + REQUIRE(packet1->GetTimestamp() == packet2->GetTimestamp()); + REQUIRE(packet1->GetLength() == packet2->GetLength()); + }; + + auto* packetA = RTC::RTP::Packet::Factory(rtpCommon::FactoryBuffer, 2000); + + packetA->SetSequenceNumber(1111); + packetA->SetTimestamp(111111); + packetA->SetSsrc(11111111); + + auto* packetB = RTC::RTP::Packet::Factory(rtpCommon::FactoryBuffer + 2000, 2000); + + packetB->SetSequenceNumber(2222); + packetB->SetTimestamp(222222); + packetB->SetSsrc(22222222); + + SECTION("default constructor and assign later") + { + RTC::RTP::SharedPacket sharedPacket; + + REQUIRE(!sharedPacket); + REQUIRE(sharedPacket.get() == nullptr); + + sharedPacket = RTC::RTP::ClonePacket(packetA); + + REQUIRE(sharedPacket); + compareRtpPackets(sharedPacket.get(), packetA); + + sharedPacket.reset(); + + REQUIRE(!sharedPacket); + REQUIRE(sharedPacket.get() == nullptr); + + delete packetA; + delete packetB; + } + + SECTION("clone packet and copy constructor") + { + // Create sharedPacket1 holding a clone of packetA. + RTC::RTP::SharedPacket sharedPacket1 = RTC::RTP::ClonePacket(packetA); + + REQUIRE(sharedPacket1); + compareRtpPackets(sharedPacket1.get(), packetA); + + // Create sharedPacket2 using copy constructor. + RTC::RTP::SharedPacket sharedPacket2(sharedPacket1); + + REQUIRE(sharedPacket2); + compareRtpPackets(sharedPacket2.get(), packetA); + REQUIRE(sharedPacket1.get() == sharedPacket2.get()); + + 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(); + + REQUIRE(!sharedPacket1); + REQUIRE(sharedPacket1.get() == nullptr); + + delete packetA; + delete packetB; + } + + SECTION("copy assignment operator") + { + RTC::RTP::SharedPacket sharedPacket1 = RTC::RTP::ClonePacket(packetA); + + REQUIRE(sharedPacket1); + compareRtpPackets(sharedPacket1.get(), packetA); + + RTC::RTP::SharedPacket sharedPacket2; + + // Fill sharedPacket2 using copy assignment operator. + sharedPacket2 = sharedPacket1; + + REQUIRE(sharedPacket2); + compareRtpPackets(sharedPacket2.get(), packetA); + REQUIRE(sharedPacket1.get() == sharedPacket2.get()); + + 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(); + + REQUIRE(!sharedPacket1); + REQUIRE(sharedPacket1.get() == nullptr); + + delete packetA; + delete packetB; + } + + SECTION("assign nullptr") + { + RTC::RTP::SharedPacket sharedPacket = RTC::RTP::ClonePacket(packetA); + + REQUIRE(sharedPacket); + compareRtpPackets(sharedPacket.get(), packetA); + + sharedPacket = nullptr; + + REQUIRE(!sharedPacket); + REQUIRE(sharedPacket.get() == nullptr); + + delete packetA; + delete packetB; + } +} From 1be0c649e1cab5f9f67106b4452c95984b9e86a3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Luis=20Mill=C3=A1n?= Date: Fri, 23 Oct 2026 16:48:33 +0200 Subject: [PATCH 6/8] RtpStreamSend::ReceivePacket() returns the retransmission item --- .../RTC/RTP/FuzzerRetransmissionBuffer.cpp | 6 +- .../src/RTC/RTP/FuzzerRtpStreamSend.cpp | 4 +- .../include/RTC/RTP/RetransmissionBuffer.hpp | 18 +++- worker/include/RTC/RTP/RtpStreamSend.hpp | 33 +++---- worker/src/RTC/Consumer.cpp | 16 ++-- worker/src/RTC/RTP/RetransmissionBuffer.cpp | 85 +++++++++---------- worker/src/RTC/RTP/RtpStreamSend.cpp | 21 ++--- .../src/RTC/RTP/TestRetransmissionBuffer.cpp | 6 +- worker/test/src/RTC/RTP/TestRtpStreamSend.cpp | 27 +++--- 9 files changed, 118 insertions(+), 98 deletions(-) diff --git a/worker/fuzzer/src/RTC/RTP/FuzzerRetransmissionBuffer.cpp b/worker/fuzzer/src/RTC/RTP/FuzzerRetransmissionBuffer.cpp index d2305c715a..9decb3790c 100644 --- a/worker/fuzzer/src/RTC/RTP/FuzzerRetransmissionBuffer.cpp +++ b/worker/fuzzer/src/RTC/RTP/FuzzerRetransmissionBuffer.cpp @@ -35,9 +35,11 @@ void FuzzerRtcRtpRetransmissionBuffer::Fuzz(const uint8_t* data, size_t len) packet->SetSequenceNumber(Utils::Byte::Get2Bytes(data, offset)); packet->SetTimestamp(Utils::Byte::Get4Bytes(data, offset)); - if (retransmissionBuffer.Insert(packet, nowMs)) + auto* retransmissionBufferItem = retransmissionBuffer.Insert(packet, nowMs); + + if (retransmissionBufferItem) { - retransmissionBuffer.StorePacket(packet->GetSequenceNumber(), RTC::RTP::ClonePacket(packet)); + retransmissionBufferItem->sharedPacket = RTC::RTP::ClonePacket(packet); } // Let some 'random' time pass and check whether the packet is too old. diff --git a/worker/fuzzer/src/RTC/RTP/FuzzerRtpStreamSend.cpp b/worker/fuzzer/src/RTC/RTP/FuzzerRtpStreamSend.cpp index 3a3718cf44..3590df898a 100644 --- a/worker/fuzzer/src/RTC/RTP/FuzzerRtpStreamSend.cpp +++ b/worker/fuzzer/src/RTC/RTP/FuzzerRtpStreamSend.cpp @@ -54,9 +54,9 @@ void FuzzerRtcRtpStreamSend::Fuzz(const uint8_t* data, size_t len) const auto result = stream->ReceivePacket(packet); - if (result == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED) + if (result.retransmissionBufferItem) { - stream->StoreSharedPacket(packet->GetSequenceNumber(), RTC::RTP::ClonePacket(packet)); + result.retransmissionBufferItem->sharedPacket = RTC::RTP::ClonePacket(packet); } // Feed a 'random' Receiver Report, which is what the loss of the stream is diff --git a/worker/include/RTC/RTP/RetransmissionBuffer.hpp b/worker/include/RTC/RTP/RetransmissionBuffer.hpp index 9736f6f5d9..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 }; /** @@ -67,8 +71,18 @@ namespace RTC void Dump(int indentation = 0) const; Item* Get(uint16_t seq) const; - bool Insert(RTP::Packet* packet, int64_t nowMs); - void StorePacket(uint16_t seq, const RTP::SharedPacket& sharedPacket) const; + /** + * 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 f479c9af49..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: @@ -153,17 +167,6 @@ namespace RTC void SetRtx(uint8_t payloadType, uint32_t ssrc) override; ReceivePacketResult ReceivePacket(RTP::Packet* packet); - /** - * Store the given cloned packet into the retransmission buffer item with - * the given sequence number. - * - * @remarks - * Must be called right after ReceivePacket() returned - * ACCEPTED_AND_STORED, once the packet has been restored to its original - * state, giving the sequence number that the packet had when it was - * passed to ReceivePacket(). - */ - void StoreSharedPacket(uint16_t seq, const RTP::SharedPacket& sharedPacket); void ReceiveNack(RTC::RTCP::FeedbackRtpNackPacket* nackPacket); diff --git a/worker/src/RTC/Consumer.cpp b/worker/src/RTC/Consumer.cpp index 46c1dd3464..197d637d29 100644 --- a/worker/src/RTC/Consumer.cpp +++ b/worker/src/RTC/Consumer.cpp @@ -1377,7 +1377,7 @@ namespace RTC const auto result = rtpStream->ReceivePacket(packet); - if (result != RTC::RTP::RtpStreamSend::ReceivePacketResult::DISCARDED) + if (result.accepted) { if (rtpSeqManager->GetMaxOutput() == packet->GetSequenceNumber()) { @@ -1425,17 +1425,19 @@ namespace RTC // 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 == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED) + if (result.retransmissionBufferItem) { if (!sharedPacket) { sharedPacket = RTC::RTP::ClonePacket(packet); } - // NOTE: We must give `seq` (the sequence number this stream stored the - // packet with) and not the packet sequence number, already restored to - // its original value above. - rtpStream->StoreSharedPacket(seq, sharedPacket); + 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 @@ -1445,7 +1447,7 @@ 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) { diff --git a/worker/src/RTC/RTP/RetransmissionBuffer.cpp b/worker/src/RTC/RTP/RetransmissionBuffer.cpp index c968845384..d282cd1a2c 100644 --- a/worker/src/RTC/RTP/RetransmissionBuffer.cpp +++ b/worker/src/RTC/RTP/RetransmissionBuffer.cpp @@ -18,8 +18,8 @@ namespace RTC // Store some extra info into the item. // - // NOTE: `sharedPacket` is left empty here. The Consumer fills it via - // StorePacket() right after this Insert() call, once the packet has been + // 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{ .encoder = packet->GetPayloadEncoder(), @@ -112,8 +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, int64_t nowMs) + RetransmissionBuffer::Item* RetransmissionBuffer::Insert(RTP::Packet* packet, int64_t nowMs) { MS_TRACE(); @@ -123,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, nowMs)); + item = RetransmissionBuffer::CreateItem(packet, nowMs); + + this->buffer.push_back(item); - return true; + return item; } auto* oldestItem = GetOldest(); @@ -152,9 +159,11 @@ namespace RTC Clear(); - this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, nowMs)); + item = RetransmissionBuffer::CreateItem(packet, nowMs); + + this->buffer.push_back(item); - return true; + return item; } // Clear too old packets in the buffer. @@ -179,9 +188,11 @@ namespace RTC seq, timestamp); - this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, nowMs)); + item = RetransmissionBuffer::CreateItem(packet, nowMs); - return true; + this->buffer.push_back(item); + + return item; } oldestItem = GetOldest(); @@ -216,9 +227,11 @@ namespace RTC Clear(); - this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, nowMs)); + item = RetransmissionBuffer::CreateItem(packet, nowMs); + + this->buffer.push_back(item); - return true; + return item; } MS_WARN_TAG( @@ -229,7 +242,7 @@ namespace RTC seq, timestamp); - return false; + return nullptr; } // Calculate how many blank slots it would be necessary to add when @@ -279,7 +292,9 @@ namespace RTC } // Push the packet, which becomes the newest one in the buffer. - this->buffer.push_back(RetransmissionBuffer::CreateItem(packet, 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. @@ -299,7 +314,7 @@ namespace RTC seq, timestamp); - return false; + return nullptr; } // Ensure that the timestamp of the packet is equal or less than the @@ -314,7 +329,7 @@ namespace RTC seq, timestamp); - return false; + return nullptr; } // Calculate how many blank slots it would be necessary to add when @@ -333,7 +348,7 @@ namespace RTC seq, timestamp); - return false; + return nullptr; } // Push blank slots to the front. @@ -343,7 +358,9 @@ namespace RTC } // Insert the packet, which becomes the oldest one in the buffer. - this->buffer.push_front(RetransmissionBuffer::CreateItem(packet, 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. @@ -357,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. @@ -399,7 +414,7 @@ namespace RTC seq, timestamp); - return false; + return nullptr; } } @@ -430,12 +445,13 @@ namespace RTC seq, timestamp); - return false; + return nullptr; } } // Store the packet. - this->buffer[idx] = RetransmissionBuffer::CreateItem(packet, nowMs); + item = RetransmissionBuffer::CreateItem(packet, nowMs); + this->buffer[idx] = item; } MS_ASSERT( @@ -444,26 +460,9 @@ namespace RTC this->buffer.size(), this->maxItems); - return true; - } - - void RetransmissionBuffer::StorePacket(uint16_t seq, const RTP::SharedPacket& sharedPacket) const - { - MS_TRACE(); - - MS_ASSERT(sharedPacket, "given sharedPacket is empty"); - - auto* item = Get(seq); - - MS_ASSERT(item, "no item with given seq in the buffer [seq:%" PRIu16 "]", seq); - - MS_ASSERT( - !item->sharedPacket, - "item already holds a packet [seq:%" PRIu16 ", item seq:%" PRIu16 "]", - seq, - item->sequenceNumber); + MS_ASSERT(item != nullptr, "inserted item doesn't exist"); - item->sharedPacket = sharedPacket; + return item; } void RetransmissionBuffer::Clear() diff --git a/worker/src/RTC/RTP/RtpStreamSend.cpp b/worker/src/RTC/RTP/RtpStreamSend.cpp index d7b50a23e2..80745b0e31 100644 --- a/worker/src/RTC/RTP/RtpStreamSend.cpp +++ b/worker/src/RTC/RTP/RtpStreamSend.cpp @@ -109,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) @@ -131,26 +131,17 @@ namespace RTC packet->GetSsrc(), packet->GetSequenceNumber()); - return ReceivePacketResult::DISCARDED; + return { .accepted = false }; } - stored = this->retransmissionBuffer->Insert(packet, 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; - } - - void RtpStreamSend::StoreSharedPacket(uint16_t seq, const RTP::SharedPacket& sharedPacket) - { - MS_TRACE(); - - MS_ASSERT(this->retransmissionBuffer, "no retransmission buffer"); - - this->retransmissionBuffer->StorePacket(seq, sharedPacket); + return { .accepted = true, .retransmissionBufferItem = retransmissionBufferItem }; } void RtpStreamSend::ReceiveNack(RTC::RTCP::FeedbackRtpNackPacket* nackPacket) diff --git a/worker/test/src/RTC/RTP/TestRetransmissionBuffer.cpp b/worker/test/src/RTC/RTP/TestRetransmissionBuffer.cpp index a064750804..8219a2ca87 100644 --- a/worker/test/src/RTC/RTP/TestRetransmissionBuffer.cpp +++ b/worker/test/src/RTC/RTP/TestRetransmissionBuffer.cpp @@ -42,9 +42,11 @@ SCENARIO("RTP RetransmissionBuffer", "[rtp][rtx]") packet->SetSequenceNumber(seq); packet->SetTimestamp(timestamp); - if (RTC::RTP::RetransmissionBuffer::Insert(packet.get(), nowMs)) + auto* item = RTC::RTP::RetransmissionBuffer::Insert(packet.get(), nowMs); + + if (item) { - RTC::RTP::RetransmissionBuffer::StorePacket(seq, RTC::RTP::ClonePacket(packet.get())); + item->sharedPacket = RTC::RTP::ClonePacket(packet.get()); } } diff --git a/worker/test/src/RTC/RTP/TestRtpStreamSend.cpp b/worker/test/src/RTC/RTP/TestRtpStreamSend.cpp index 3333203753..34509969f3 100644 --- a/worker/test/src/RTC/RTP/TestRtpStreamSend.cpp +++ b/worker/test/src/RTC/RTP/TestRtpStreamSend.cpp @@ -84,14 +84,14 @@ SCENARIO("RtpStreamSend", "[rtp][rtcp][nack][rtpstream][rtpstreamsend]") // NOTE: Here we must replicate the behaviour of Consumer::sendRtpPacket() // 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 == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED) + if (result.retransmissionBufferItem) { if (!sharedPacket) { sharedPacket = RTC::RTP::ClonePacket(packet); } - stream->StoreSharedPacket(packet->GetSequenceNumber(), sharedPacket); + result.retransmissionBufferItem->sharedPacket = sharedPacket; } } }; @@ -968,11 +968,12 @@ SCENARIO("RtpStreamSend", "[rtp][rtcp][nack][rtpstream][rtpstreamsend]") // 00000001 (S0_T1) active decode target bitmas. const auto result = stream1->ReceivePacket(packet3.get()); - REQUIRE(result == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED); + REQUIRE(result.accepted); + REQUIRE(result.retransmissionBufferItem); const RTC::RTP::SharedPacket sharedPacket = RTC::RTP::ClonePacket(packet3.get()); - stream1->StoreSharedPacket(packet3->GetSequenceNumber(), sharedPacket); + result.retransmissionBufferItem->sharedPacket = sharedPacket; // Process the third packet with context2 and verify current spatial layers. forwarded = packet3->ProcessPayload(std::addressof(context2), marker); @@ -985,9 +986,11 @@ 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. - if (stream2->ReceivePacket(packet3.get()) == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED) + const auto result2 = stream2->ReceivePacket(packet3.get()); + + if (result2.retransmissionBufferItem) { - stream2->StoreSharedPacket(packet3->GetSequenceNumber(), sharedPacket); + result2.retransmissionBufferItem->sharedPacket = sharedPacket; } // Create a NACK item that requests the third packet. @@ -1712,11 +1715,13 @@ SCENARIO("RtpStreamSend", "[rtp][rtcp][nack][rtpstream][rtpstreamsend]") auto result = stream->ReceivePacket(packet.get()); - REQUIRE(result == RTC::RTP::RtpStreamSend::ReceivePacketResult::ACCEPTED_AND_STORED); + REQUIRE(result.accepted); + REQUIRE(result.retransmissionBufferItem); result = stream->ReceivePacket(packet.get()); - 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") @@ -1747,7 +1752,8 @@ SCENARIO("RtpStreamSend", "[rtp][rtcp][nack][rtpstream][rtpstreamsend]") 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 @@ -1768,7 +1774,8 @@ SCENARIO("RtpStreamSend", "[rtp][rtcp][nack][rtpstream][rtpstreamsend]") 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") From 88bda2effd0f9e81de46a3699de06e19ea3984c4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Luis=20Mill=C3=A1n?= Date: Fri, 23 Oct 2026 16:50:25 +0200 Subject: [PATCH 7/8] cosmetic --- worker/fuzzer/src/RTC/RTP/FuzzerRetransmissionBuffer.cpp | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/worker/fuzzer/src/RTC/RTP/FuzzerRetransmissionBuffer.cpp b/worker/fuzzer/src/RTC/RTP/FuzzerRetransmissionBuffer.cpp index 9decb3790c..f87e352983 100644 --- a/worker/fuzzer/src/RTC/RTP/FuzzerRetransmissionBuffer.cpp +++ b/worker/fuzzer/src/RTC/RTP/FuzzerRetransmissionBuffer.cpp @@ -35,11 +35,9 @@ void FuzzerRtcRtpRetransmissionBuffer::Fuzz(const uint8_t* data, size_t len) packet->SetSequenceNumber(Utils::Byte::Get2Bytes(data, offset)); packet->SetTimestamp(Utils::Byte::Get4Bytes(data, offset)); - auto* retransmissionBufferItem = retransmissionBuffer.Insert(packet, nowMs); - - if (retransmissionBufferItem) + if (auto* item = retransmissionBuffer.Insert(packet, nowMs)) { - retransmissionBufferItem->sharedPacket = RTC::RTP::ClonePacket(packet); + item->sharedPacket = RTC::RTP::ClonePacket(packet); } // Let some 'random' time pass and check whether the packet is too old. From e547600443a5ec4b65f8a4279208b40a0fee0337 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Luis=20Mill=C3=A1n?= Date: Fri, 23 Oct 2026 17:13:05 +0200 Subject: [PATCH 8/8] cosmetic --- worker/include/RTC/RTP/SharedPacket.hpp | 2 +- worker/src/RTC/RTP/SharedPacket.cpp | 46 ++++++++++++------------- 2 files changed, 24 insertions(+), 24 deletions(-) diff --git a/worker/include/RTC/RTP/SharedPacket.hpp b/worker/include/RTC/RTP/SharedPacket.hpp index ebdd3bd434..63f83ea7ea 100644 --- a/worker/include/RTC/RTP/SharedPacket.hpp +++ b/worker/include/RTC/RTP/SharedPacket.hpp @@ -32,7 +32,7 @@ namespace RTC * 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* packet); + void AssertSamePacket(const SharedPacket& sharedPacket, const RTP::Packet* otherPacket); } // namespace RTP } // namespace RTC diff --git a/worker/src/RTC/RTP/SharedPacket.cpp b/worker/src/RTC/RTP/SharedPacket.cpp index ba7af8f16f..c61ffe3dbd 100644 --- a/worker/src/RTC/RTP/SharedPacket.cpp +++ b/worker/src/RTC/RTP/SharedPacket.cpp @@ -78,49 +78,49 @@ namespace RTC return SharedPacket{ clonedPacket }; } - void AssertSamePacket(const SharedPacket& sharedPacket, const RTP::Packet* packet) + void AssertSamePacket(const SharedPacket& sharedPacket, const RTP::Packet* otherPacket) { MS_TRACE(); - const auto* clonedPacket = sharedPacket.get(); + const auto* packet = sharedPacket.get(); - if (!clonedPacket && !packet) + if (!packet && !otherPacket) { return; } - else if (clonedPacket && !packet) + else if (packet && !otherPacket) { - MS_ABORT("there is a packet in sharedPacket but given packet doesn't have value"); + MS_ABORT("there is a packet in sharedPacket but given otherPacket doesn't have value"); } - else if (!clonedPacket && packet) + else if (!packet && otherPacket) { - MS_ABORT("there is no packet in sharedPacket but given packet has value"); + MS_ABORT("there is no packet in sharedPacket but given otherPacket has value"); } else { MS_ASSERT( - clonedPacket->GetSsrc() == packet->GetSsrc(), - "SSRC %" PRIu32 " in packet in sharedPacket != SSRC %" PRIu32 " in given packet", - clonedPacket->GetSsrc(), - packet->GetSsrc()); + packet->GetSsrc() == otherPacket->GetSsrc(), + "SSRC %" PRIu32 " in packet in sharedPacket != SSRC %" PRIu32 " in otherPacket", + packet->GetSsrc(), + otherPacket->GetSsrc()); MS_ASSERT( - clonedPacket->GetSequenceNumber() == packet->GetSequenceNumber(), - "seq %" PRIu16 " in packet in sharedPacket != seq %" PRIu16 " in given packet", - clonedPacket->GetSequenceNumber(), - packet->GetSequenceNumber()); + packet->GetSequenceNumber() == otherPacket->GetSequenceNumber(), + "seq %" PRIu16 " in packet in sharedPacket != seq %" PRIu16 " in otherPacket", + packet->GetSequenceNumber(), + otherPacket->GetSequenceNumber()); MS_ASSERT( - clonedPacket->GetTimestamp() == packet->GetTimestamp(), - "timestamp %" PRIu32 " in packet in sharedPacket != timestamp %" PRIu32 " in given packet", - clonedPacket->GetTimestamp(), - packet->GetTimestamp()); + packet->GetTimestamp() == otherPacket->GetTimestamp(), + "timestamp %" PRIu32 " in packet in sharedPacket != timestamp %" PRIu32 " in otherPacket", + packet->GetTimestamp(), + otherPacket->GetTimestamp()); MS_ASSERT( - clonedPacket->GetLength() == packet->GetLength(), - "length %zu of packet in sharedPacket != length %zu of given packet", - clonedPacket->GetLength(), - packet->GetLength()); + packet->GetLength() == otherPacket->GetLength(), + "length %zu of packet in sharedPacket != length %zu of otherPacket", + packet->GetLength(), + otherPacket->GetLength()); } } } // namespace RTP