diff --git a/modules/rtp_rtcp/source/flexfec_receiver.cc b/modules/rtp_rtcp/source/flexfec_receiver.cc index ba09db9e1c..28c8b26834 100644 --- a/modules/rtp_rtcp/source/flexfec_receiver.cc +++ b/modules/rtp_rtcp/source/flexfec_receiver.cc @@ -109,10 +109,8 @@ FlexfecReceiver::AddReceivedPacket(const RtpPacketReceived& packet) { // Insert packet payload into erasure code. received_packet->pkt = rtc::scoped_refptr( new ForwardErrorCorrection::Packet()); - // TODO(ilnik): after slice capability is added to COW, use it here instead - // of initializing COW buffer with ArrayView. - auto payload = packet.payload(); - received_packet->pkt->data.SetData(payload.data(), payload.size()); + received_packet->pkt->data = + packet.Buffer().Slice(packet.headers_size(), packet.payload_size()); } else { // This is a media packet, or a FlexFEC packet belonging to some // other FlexFEC stream. diff --git a/modules/rtp_rtcp/source/ulpfec_receiver_impl.cc b/modules/rtp_rtcp/source/ulpfec_receiver_impl.cc index 43ce2b0245..24348f3c2b 100644 --- a/modules/rtp_rtcp/source/ulpfec_receiver_impl.cc +++ b/modules/rtp_rtcp/source/ulpfec_receiver_impl.cc @@ -119,13 +119,14 @@ bool UlpfecReceiverImpl::AddReceivedRedPacket(const RtpPacket& rtp_packet, packet_counter_.first_packet_time_ms = rtc::TimeMillis(); } - auto red_payload = rtp_packet.payload().subview(kRedHeaderLength); if (received_packet->is_fec) { ++packet_counter_.num_fec_packets; - // everything behind the RED header - received_packet->pkt->data.SetData(red_payload.data(), red_payload.size()); + received_packet->pkt->data = + rtp_packet.Buffer().Slice(rtp_packet.headers_size() + kRedHeaderLength, + rtp_packet.payload_size() - kRedHeaderLength); } else { + auto red_payload = rtp_packet.payload().subview(kRedHeaderLength); received_packet->pkt->data.EnsureCapacity(rtp_packet.headers_size() + red_payload.size()); // Copy RTP header. @@ -170,6 +171,7 @@ int32_t UlpfecReceiverImpl::ProcessReceivedFec() { crit_sect_.Enter(); // Create a packet with the buffer to modify it. RtpPacketReceived rtp_packet; + const uint8_t* const original_data = packet->data.cdata(); rtp_packet.Parse(packet->data); rtp_packet.IdentifyExtensions(extensions_); // Reset buffer reference, so zeroing would work on a buffer with a @@ -177,6 +179,8 @@ int32_t UlpfecReceiverImpl::ProcessReceivedFec() { packet->data = rtc::CopyOnWriteBuffer(0); rtp_packet.ZeroMutableExtensions(); packet->data = rtp_packet.Buffer(); + // Ensure that zeroing of extensions was done in place. + RTC_DCHECK_EQ(packet->data.cdata(), original_data); } fec_->DecodeFec(*received_packet, &recovered_packets_); } diff --git a/rtc_base/copy_on_write_buffer.cc b/rtc_base/copy_on_write_buffer.cc index de003f2d8e..73182a12b1 100644 --- a/rtc_base/copy_on_write_buffer.cc +++ b/rtc_base/copy_on_write_buffer.cc @@ -14,40 +14,47 @@ namespace rtc { -CopyOnWriteBuffer::CopyOnWriteBuffer() { +CopyOnWriteBuffer::CopyOnWriteBuffer() : offset_(0), size_(0) { RTC_DCHECK(IsConsistent()); } CopyOnWriteBuffer::CopyOnWriteBuffer(const CopyOnWriteBuffer& buf) - : buffer_(buf.buffer_) {} + : buffer_(buf.buffer_), offset_(buf.offset_), size_(buf.size_) {} CopyOnWriteBuffer::CopyOnWriteBuffer(CopyOnWriteBuffer&& buf) - : buffer_(std::move(buf.buffer_)) {} + : buffer_(std::move(buf.buffer_)), offset_(buf.offset_), size_(buf.size_) { + buf.offset_ = 0; + buf.size_ = 0; + RTC_DCHECK(IsConsistent()); +} CopyOnWriteBuffer::CopyOnWriteBuffer(const std::string& s) : CopyOnWriteBuffer(s.data(), s.length()) {} CopyOnWriteBuffer::CopyOnWriteBuffer(size_t size) - : buffer_(size > 0 ? new RefCountedObject(size) : nullptr) { + : buffer_(size > 0 ? new RefCountedObject(size) : nullptr), + offset_(0), + size_(size) { RTC_DCHECK(IsConsistent()); } CopyOnWriteBuffer::CopyOnWriteBuffer(size_t size, size_t capacity) : buffer_(size > 0 || capacity > 0 ? new RefCountedObject(size, capacity) - : nullptr) { + : nullptr), + offset_(0), + size_(size) { RTC_DCHECK(IsConsistent()); } CopyOnWriteBuffer::~CopyOnWriteBuffer() = default; bool CopyOnWriteBuffer::operator==(const CopyOnWriteBuffer& buf) const { - // Must either use the same buffer internally or have the same contents. + // Must either be the same view of the same buffer or have the same contents. RTC_DCHECK(IsConsistent()); RTC_DCHECK(buf.IsConsistent()); - return buffer_.get() == buf.buffer_.get() || - (buffer_.get() && buf.buffer_.get() && - *buffer_.get() == *buf.buffer_.get()); + return size_ == buf.size_ && + (cdata() == buf.cdata() || memcmp(cdata(), buf.cdata(), size_) == 0); } void CopyOnWriteBuffer::SetSize(size_t size) { @@ -55,35 +62,39 @@ void CopyOnWriteBuffer::SetSize(size_t size) { if (!buffer_) { if (size > 0) { buffer_ = new RefCountedObject(size); + offset_ = 0; + size_ = size; } RTC_DCHECK(IsConsistent()); return; } - // Clone data if referenced. - if (!buffer_->HasOneRef()) { - buffer_ = new RefCountedObject(buffer_->data(), - std::min(buffer_->size(), size), - std::max(buffer_->capacity(), size)); + if (size <= size_) { + size_ = size; + return; } - buffer_->SetSize(size); + + UnshareAndEnsureCapacity(std::max(capacity(), size)); + buffer_->SetSize(size + offset_); + size_ = size; RTC_DCHECK(IsConsistent()); } -void CopyOnWriteBuffer::EnsureCapacity(size_t capacity) { +void CopyOnWriteBuffer::EnsureCapacity(size_t new_capacity) { RTC_DCHECK(IsConsistent()); if (!buffer_) { - if (capacity > 0) { - buffer_ = new RefCountedObject(0, capacity); + if (new_capacity > 0) { + buffer_ = new RefCountedObject(0, new_capacity); + offset_ = 0; + size_ = 0; } RTC_DCHECK(IsConsistent()); return; - } else if (capacity <= buffer_->capacity()) { + } else if (new_capacity <= capacity()) { return; } - CloneDataIfReferenced(std::max(buffer_->capacity(), capacity)); - buffer_->EnsureCapacity(capacity); + UnshareAndEnsureCapacity(new_capacity); RTC_DCHECK(IsConsistent()); } @@ -94,18 +105,21 @@ void CopyOnWriteBuffer::Clear() { if (buffer_->HasOneRef()) { buffer_->Clear(); } else { - buffer_ = new RefCountedObject(0, buffer_->capacity()); + buffer_ = new RefCountedObject(0, capacity()); } + offset_ = 0; + size_ = 0; RTC_DCHECK(IsConsistent()); } -void CopyOnWriteBuffer::CloneDataIfReferenced(size_t new_capacity) { - if (buffer_->HasOneRef()) { +void CopyOnWriteBuffer::UnshareAndEnsureCapacity(size_t new_capacity) { + if (buffer_->HasOneRef() && new_capacity <= capacity()) { return; } - buffer_ = new RefCountedObject(buffer_->data(), buffer_->size(), + buffer_ = new RefCountedObject(buffer_->data() + offset_, size_, new_capacity); + offset_ = 0; RTC_DCHECK(IsConsistent()); } diff --git a/rtc_base/copy_on_write_buffer.h b/rtc_base/copy_on_write_buffer.h index c60e78be1d..ea4868fc14 100644 --- a/rtc_base/copy_on_write_buffer.h +++ b/rtc_base/copy_on_write_buffer.h @@ -56,6 +56,8 @@ class CopyOnWriteBuffer { : CopyOnWriteBuffer(size, capacity) { if (buffer_) { std::memcpy(buffer_->data(), data, size); + offset_ = 0; + size_ = size; } } @@ -88,8 +90,8 @@ class CopyOnWriteBuffer { if (!buffer_) { return nullptr; } - CloneDataIfReferenced(buffer_->capacity()); - return buffer_->data(); + UnshareAndEnsureCapacity(capacity()); + return buffer_->data() + offset_; } // Get const pointer to the data. This will not create a copy of the @@ -102,17 +104,17 @@ class CopyOnWriteBuffer { if (!buffer_) { return nullptr; } - return buffer_->data(); + return buffer_->data() + offset_; } size_t size() const { RTC_DCHECK(IsConsistent()); - return buffer_ ? buffer_->size() : 0; + return size_; } size_t capacity() const { RTC_DCHECK(IsConsistent()); - return buffer_ ? buffer_->capacity() : 0; + return buffer_ ? buffer_->capacity() - offset_ : 0; } CopyOnWriteBuffer& operator=(const CopyOnWriteBuffer& buf) { @@ -120,6 +122,8 @@ class CopyOnWriteBuffer { RTC_DCHECK(buf.IsConsistent()); if (&buf != this) { buffer_ = buf.buffer_; + offset_ = buf.offset_; + size_ = buf.size_; } return *this; } @@ -128,6 +132,10 @@ class CopyOnWriteBuffer { RTC_DCHECK(IsConsistent()); RTC_DCHECK(buf.IsConsistent()); buffer_ = std::move(buf.buffer_); + offset_ = buf.offset_; + size_ = buf.size_; + buf.offset_ = 0; + buf.size_ = 0; return *this; } @@ -157,10 +165,13 @@ class CopyOnWriteBuffer { if (!buffer_) { buffer_ = size > 0 ? new RefCountedObject(data, size) : nullptr; } else if (!buffer_->HasOneRef()) { - buffer_ = new RefCountedObject(data, size, buffer_->capacity()); + buffer_ = new RefCountedObject(data, size, capacity()); } else { buffer_->SetData(data, size); } + offset_ = 0; + size_ = size; + RTC_DCHECK(IsConsistent()); } @@ -177,6 +188,8 @@ class CopyOnWriteBuffer { RTC_DCHECK(buf.IsConsistent()); if (&buf != this) { buffer_ = buf.buffer_; + offset_ = buf.offset_; + size_ = buf.size_; } } @@ -188,13 +201,19 @@ class CopyOnWriteBuffer { RTC_DCHECK(IsConsistent()); if (!buffer_) { buffer_ = new RefCountedObject(data, size); + offset_ = 0; + size_ = size; RTC_DCHECK(IsConsistent()); return; } - CloneDataIfReferenced( - std::max(buffer_->capacity(), buffer_->size() + size)); + UnshareAndEnsureCapacity(std::max(capacity(), size_ + size)); + + buffer_->SetSize(offset_ + + size_); // Remove data to the right of the slice. buffer_->AppendData(data, size); + size_ += size; + RTC_DCHECK(IsConsistent()); } @@ -228,18 +247,41 @@ class CopyOnWriteBuffer { // Swaps two buffers. friend void swap(CopyOnWriteBuffer& a, CopyOnWriteBuffer& b) { std::swap(a.buffer_, b.buffer_); + std::swap(a.offset_, b.offset_); + std::swap(a.size_, b.size_); + } + + CopyOnWriteBuffer Slice(size_t offset, size_t length) const { + CopyOnWriteBuffer slice(*this); + RTC_DCHECK_LE(offset, size_); + RTC_DCHECK_LE(length + offset, size_); + slice.offset_ += offset; + slice.size_ = length; + return slice; } private: // Create a copy of the underlying data if it is referenced from other Buffer - // objects. - void CloneDataIfReferenced(size_t new_capacity); + // objects or there is not enough capacity. + void UnshareAndEnsureCapacity(size_t new_capacity); // Pre- and postcondition of all methods. - bool IsConsistent() const { return (!buffer_ || buffer_->capacity() > 0); } + bool IsConsistent() const { + if (buffer_) { + return buffer_->capacity() > 0 && offset_ <= buffer_->size() && + offset_ + size_ <= buffer_->size(); + } else { + return size_ == 0 && offset_ == 0; + } + } // buffer_ is either null, or points to an rtc::Buffer with capacity > 0. scoped_refptr> buffer_; + // This buffer may represent a slice of a original data. + size_t offset_; // Offset of a current slice in the original data in buffer_. + // Should be 0 if the buffer_ is empty. + size_t size_; // Size of a current slice in the original data in buffer_. + // Should be 0 if the buffer_ is empty. }; } // namespace rtc diff --git a/rtc_base/copy_on_write_buffer_unittest.cc b/rtc_base/copy_on_write_buffer_unittest.cc index fc569bdd62..b35cd79454 100644 --- a/rtc_base/copy_on_write_buffer_unittest.cc +++ b/rtc_base/copy_on_write_buffer_unittest.cc @@ -319,4 +319,50 @@ TEST(CopyOnWriteBufferTest, TestBacketWrite) { EXPECT_EQ(0, memcmp(buf2.cdata(), kTestData, 3)); } +TEST(CopyOnWriteBufferTest, CreateSlice) { + CopyOnWriteBuffer buf(kTestData, 10, 10); + CopyOnWriteBuffer slice = buf.Slice(3, 4); + EXPECT_EQ(slice.size(), 4u); + EXPECT_EQ(0, memcmp(buf.cdata() + 3, slice.cdata(), 4)); +} + +TEST(CopyOnWriteBufferTest, NoCopyDataOnSlice) { + CopyOnWriteBuffer buf(kTestData, 10, 10); + CopyOnWriteBuffer slice = buf.Slice(3, 4); + EXPECT_EQ(buf.cdata() + 3, slice.cdata()); +} + +TEST(CopyOnWriteBufferTest, WritingCopiesData) { + CopyOnWriteBuffer buf(kTestData, 10, 10); + CopyOnWriteBuffer slice = buf.Slice(3, 4); + slice[0] = 0xaa; + EXPECT_NE(buf.cdata() + 3, slice.cdata()); + EXPECT_EQ(0, memcmp(buf.cdata(), kTestData, 10)); +} + +TEST(CopyOnWriteBufferTest, WritingToBufferDoesntAffectsSlice) { + CopyOnWriteBuffer buf(kTestData, 10, 10); + CopyOnWriteBuffer slice = buf.Slice(3, 4); + buf[0] = 0xaa; + EXPECT_NE(buf.cdata() + 3, slice.cdata()); + EXPECT_EQ(0, memcmp(slice.cdata(), kTestData + 3, 4)); +} + +TEST(CopyOnWriteBufferTest, SliceOfASlice) { + CopyOnWriteBuffer buf(kTestData, 10, 10); + CopyOnWriteBuffer slice = buf.Slice(3, 7); + CopyOnWriteBuffer slice2 = slice.Slice(2, 3); + EXPECT_EQ(slice2.size(), 3u); + EXPECT_EQ(slice.cdata() + 2, slice2.cdata()); + EXPECT_EQ(buf.cdata() + 5, slice2.cdata()); +} + +TEST(CopyOnWriteBufferTest, SlicesAreIndependent) { + CopyOnWriteBuffer buf(kTestData, 10, 10); + CopyOnWriteBuffer slice = buf.Slice(3, 7); + CopyOnWriteBuffer slice2 = buf.Slice(3, 7); + slice2[0] = 0xaa; + EXPECT_EQ(buf.cdata() + 3, slice.cdata()); +} + } // namespace rtc