| /* |
| * Copyright (c) 2012 The WebRTC project authors. All Rights Reserved. |
| * |
| * Use of this source code is governed by a BSD-style license |
| * that can be found in the LICENSE file in the root of the source |
| * tree. An additional intellectual property rights grant can be found |
| * in the file PATENTS. All contributing project authors may |
| * be found in the AUTHORS file in the root of the source tree. |
| */ |
| |
| #include "modules/rtp_rtcp/source/rtp_packet_history.h" |
| |
| #include <algorithm> |
| #include <limits> |
| #include <memory> |
| #include <utility> |
| |
| #include "modules/include/module_common_types_public.h" |
| #include "modules/rtp_rtcp/source/rtp_packet_to_send.h" |
| #include "rtc_base/checks.h" |
| #include "rtc_base/logging.h" |
| #include "system_wrappers/include/clock.h" |
| |
| namespace webrtc { |
| |
| constexpr size_t RtpPacketHistory::kMaxCapacity; |
| constexpr size_t RtpPacketHistory::kMaxPaddingtHistory; |
| constexpr int64_t RtpPacketHistory::kMinPacketDurationMs; |
| constexpr int RtpPacketHistory::kMinPacketDurationRtt; |
| constexpr int RtpPacketHistory::kPacketCullingDelayFactor; |
| |
| RtpPacketHistory::PacketState::PacketState() = default; |
| RtpPacketHistory::PacketState::PacketState(const PacketState&) = default; |
| RtpPacketHistory::PacketState::~PacketState() = default; |
| |
| RtpPacketHistory::StoredPacket::StoredPacket( |
| std::unique_ptr<RtpPacketToSend> packet, |
| absl::optional<int64_t> send_time_ms, |
| uint64_t insert_order) |
| : send_time_ms_(send_time_ms), |
| packet_(std::move(packet)), |
| // No send time indicates packet is not sent immediately, but instead will |
| // be put in the pacer queue and later retrieved via |
| // GetPacketAndSetSendTime(). |
| pending_transmission_(!send_time_ms.has_value()), |
| insert_order_(insert_order), |
| times_retransmitted_(0) {} |
| |
| RtpPacketHistory::StoredPacket::StoredPacket(StoredPacket&&) = default; |
| RtpPacketHistory::StoredPacket& RtpPacketHistory::StoredPacket::operator=( |
| RtpPacketHistory::StoredPacket&&) = default; |
| RtpPacketHistory::StoredPacket::~StoredPacket() = default; |
| |
| void RtpPacketHistory::StoredPacket::IncrementTimesRetransmitted( |
| PacketPrioritySet* priority_set) { |
| // Check if this StoredPacket is in the priority set. If so, we need to remove |
| // it before updating |times_retransmitted_| since that is used in sorting, |
| // and then add it back. |
| const bool in_priority_set = priority_set->erase(this) > 0; |
| ++times_retransmitted_; |
| if (in_priority_set) { |
| auto it = priority_set->insert(this); |
| RTC_DCHECK(it.second) |
| << "ERROR: Priority set already contains matching packet! In set: " |
| "insert order = " |
| << (*it.first)->insert_order_ |
| << ", times retransmitted = " << (*it.first)->times_retransmitted_ |
| << ". Trying to add: insert order = " << insert_order_ |
| << ", times retransmitted = " << times_retransmitted_; |
| } |
| } |
| |
| bool RtpPacketHistory::MoreUseful::operator()(StoredPacket* lhs, |
| StoredPacket* rhs) const { |
| // Prefer to send packets we haven't already sent as padding. |
| if (lhs->times_retransmitted() != rhs->times_retransmitted()) { |
| return lhs->times_retransmitted() < rhs->times_retransmitted(); |
| } |
| // All else being equal, prefer newer packets. |
| return lhs->insert_order() > rhs->insert_order(); |
| } |
| |
| RtpPacketHistory::RtpPacketHistory(Clock* clock) |
| : clock_(clock), |
| number_to_store_(0), |
| mode_(StorageMode::kDisabled), |
| rtt_ms_(-1), |
| packets_inserted_(0) {} |
| |
| RtpPacketHistory::~RtpPacketHistory() {} |
| |
| void RtpPacketHistory::SetStorePacketsStatus(StorageMode mode, |
| size_t number_to_store) { |
| RTC_DCHECK_LE(number_to_store, kMaxCapacity); |
| rtc::CritScope cs(&lock_); |
| if (mode != StorageMode::kDisabled && mode_ != StorageMode::kDisabled) { |
| RTC_LOG(LS_WARNING) << "Purging packet history in order to re-set status."; |
| } |
| Reset(); |
| mode_ = mode; |
| number_to_store_ = std::min(kMaxCapacity, number_to_store); |
| } |
| |
| RtpPacketHistory::StorageMode RtpPacketHistory::GetStorageMode() const { |
| rtc::CritScope cs(&lock_); |
| return mode_; |
| } |
| |
| void RtpPacketHistory::SetRtt(int64_t rtt_ms) { |
| rtc::CritScope cs(&lock_); |
| RTC_DCHECK_GE(rtt_ms, 0); |
| rtt_ms_ = rtt_ms; |
| // If storage is not disabled, packets will be removed after a timeout |
| // that depends on the RTT. Changing the RTT may thus cause some packets |
| // become "old" and subject to removal. |
| if (mode_ != StorageMode::kDisabled) { |
| CullOldPackets(clock_->TimeInMilliseconds()); |
| } |
| } |
| |
| void RtpPacketHistory::PutRtpPacket(std::unique_ptr<RtpPacketToSend> packet, |
| absl::optional<int64_t> send_time_ms) { |
| RTC_DCHECK(packet); |
| rtc::CritScope cs(&lock_); |
| int64_t now_ms = clock_->TimeInMilliseconds(); |
| if (mode_ == StorageMode::kDisabled) { |
| return; |
| } |
| |
| RTC_DCHECK(packet->allow_retransmission()); |
| CullOldPackets(now_ms); |
| |
| // Store packet. |
| const uint16_t rtp_seq_no = packet->SequenceNumber(); |
| int packet_index = GetPacketIndex(rtp_seq_no); |
| if (packet_index >= 0u && |
| static_cast<size_t>(packet_index) < packet_history_.size() && |
| packet_history_[packet_index].packet_ != nullptr) { |
| RTC_LOG(LS_WARNING) << "Duplicate packet inserted: " << rtp_seq_no; |
| // Remove previous packet to avoid inconsistent state. |
| RemovePacket(packet_index); |
| packet_index = GetPacketIndex(rtp_seq_no); |
| } |
| |
| // Packet to be inserted ahead of first packet, expand front. |
| for (; packet_index < 0; ++packet_index) { |
| packet_history_.emplace_front(nullptr, absl::nullopt, 0); |
| } |
| // Packet to be inserted behind last packet, expand back. |
| while (static_cast<int>(packet_history_.size()) <= packet_index) { |
| packet_history_.emplace_back(nullptr, absl::nullopt, 0); |
| } |
| |
| RTC_DCHECK_GE(packet_index, 0); |
| RTC_DCHECK_LT(packet_index, packet_history_.size()); |
| RTC_DCHECK(packet_history_[packet_index].packet_ == nullptr); |
| |
| packet_history_[packet_index] = |
| StoredPacket(std::move(packet), send_time_ms, packets_inserted_++); |
| |
| if (padding_priority_.size() >= kMaxPaddingtHistory - 1) { |
| padding_priority_.erase(std::prev(padding_priority_.end())); |
| } |
| auto prio_it = padding_priority_.insert(&packet_history_[packet_index]); |
| RTC_DCHECK(prio_it.second) << "Failed to insert packet into prio set."; |
| } |
| |
| std::unique_ptr<RtpPacketToSend> RtpPacketHistory::GetPacketAndSetSendTime( |
| uint16_t sequence_number) { |
| rtc::CritScope cs(&lock_); |
| if (mode_ == StorageMode::kDisabled) { |
| return nullptr; |
| } |
| |
| StoredPacket* packet = GetStoredPacket(sequence_number); |
| if (packet == nullptr) { |
| return nullptr; |
| } |
| |
| int64_t now_ms = clock_->TimeInMilliseconds(); |
| if (!VerifyRtt(*packet, now_ms)) { |
| return nullptr; |
| } |
| |
| if (packet->send_time_ms_) { |
| packet->IncrementTimesRetransmitted(&padding_priority_); |
| } |
| |
| // Update send-time and mark as no long in pacer queue. |
| packet->send_time_ms_ = now_ms; |
| packet->pending_transmission_ = false; |
| |
| // Return copy of packet instance since it may need to be retransmitted. |
| return std::make_unique<RtpPacketToSend>(*packet->packet_); |
| } |
| |
| std::unique_ptr<RtpPacketToSend> RtpPacketHistory::GetPacketAndMarkAsPending( |
| uint16_t sequence_number) { |
| return GetPacketAndMarkAsPending( |
| sequence_number, [](const RtpPacketToSend& packet) { |
| return std::make_unique<RtpPacketToSend>(packet); |
| }); |
| } |
| |
| std::unique_ptr<RtpPacketToSend> RtpPacketHistory::GetPacketAndMarkAsPending( |
| uint16_t sequence_number, |
| rtc::FunctionView<std::unique_ptr<RtpPacketToSend>(const RtpPacketToSend&)> |
| encapsulate) { |
| rtc::CritScope cs(&lock_); |
| if (mode_ == StorageMode::kDisabled) { |
| return nullptr; |
| } |
| |
| StoredPacket* packet = GetStoredPacket(sequence_number); |
| if (packet == nullptr) { |
| return nullptr; |
| } |
| |
| if (packet->pending_transmission_) { |
| // Packet already in pacer queue, ignore this request. |
| return nullptr; |
| } |
| |
| if (!VerifyRtt(*packet, clock_->TimeInMilliseconds())) { |
| // Packet already resent within too short a time window, ignore. |
| return nullptr; |
| } |
| |
| // Copy and/or encapsulate packet. |
| std::unique_ptr<RtpPacketToSend> encapsulated_packet = |
| encapsulate(*packet->packet_); |
| if (encapsulated_packet) { |
| packet->pending_transmission_ = true; |
| } |
| |
| return encapsulated_packet; |
| } |
| |
| void RtpPacketHistory::MarkPacketAsSent(uint16_t sequence_number) { |
| rtc::CritScope cs(&lock_); |
| if (mode_ == StorageMode::kDisabled) { |
| return; |
| } |
| |
| StoredPacket* packet = GetStoredPacket(sequence_number); |
| if (packet == nullptr) { |
| return; |
| } |
| |
| RTC_DCHECK(packet->send_time_ms_); |
| |
| // Update send-time, mark as no longer in pacer queue, and increment |
| // transmission count. |
| packet->send_time_ms_ = clock_->TimeInMilliseconds(); |
| packet->pending_transmission_ = false; |
| packet->IncrementTimesRetransmitted(&padding_priority_); |
| } |
| |
| absl::optional<RtpPacketHistory::PacketState> RtpPacketHistory::GetPacketState( |
| uint16_t sequence_number) const { |
| rtc::CritScope cs(&lock_); |
| if (mode_ == StorageMode::kDisabled) { |
| return absl::nullopt; |
| } |
| |
| int packet_index = GetPacketIndex(sequence_number); |
| if (packet_index < 0 || |
| static_cast<size_t>(packet_index) >= packet_history_.size()) { |
| return absl::nullopt; |
| } |
| const StoredPacket& packet = packet_history_[packet_index]; |
| if (packet.packet_ == nullptr) { |
| return absl::nullopt; |
| } |
| |
| if (!VerifyRtt(packet, clock_->TimeInMilliseconds())) { |
| return absl::nullopt; |
| } |
| |
| return StoredPacketToPacketState(packet); |
| } |
| |
| bool RtpPacketHistory::VerifyRtt(const RtpPacketHistory::StoredPacket& packet, |
| int64_t now_ms) const { |
| if (packet.send_time_ms_) { |
| // Send-time already set, this check must be for a retransmission. |
| if (packet.times_retransmitted() > 0 && |
| now_ms < *packet.send_time_ms_ + rtt_ms_) { |
| // This packet has already been retransmitted once, and the time since |
| // that even is lower than on RTT. Ignore request as this packet is |
| // likely already in the network pipe. |
| return false; |
| } |
| } |
| |
| return true; |
| } |
| |
| std::unique_ptr<RtpPacketToSend> RtpPacketHistory::GetPayloadPaddingPacket() { |
| // Default implementation always just returns a copy of the packet. |
| return GetPayloadPaddingPacket([](const RtpPacketToSend& packet) { |
| return std::make_unique<RtpPacketToSend>(packet); |
| }); |
| } |
| |
| std::unique_ptr<RtpPacketToSend> RtpPacketHistory::GetPayloadPaddingPacket( |
| rtc::FunctionView<std::unique_ptr<RtpPacketToSend>(const RtpPacketToSend&)> |
| encapsulate) { |
| rtc::CritScope cs(&lock_); |
| if (mode_ == StorageMode::kDisabled || padding_priority_.empty()) { |
| return nullptr; |
| } |
| |
| auto best_packet_it = padding_priority_.begin(); |
| StoredPacket* best_packet = *best_packet_it; |
| if (best_packet->pending_transmission_) { |
| // Because PacedSender releases it's lock when it calls |
| // GeneratePadding() there is the potential for a race where a new |
| // packet ends up here instead of the regular transmit path. In such a |
| // case, just return empty and it will be picked up on the next |
| // Process() call. |
| return nullptr; |
| } |
| |
| auto padding_packet = encapsulate(*best_packet->packet_); |
| if (!padding_packet) { |
| return nullptr; |
| } |
| |
| best_packet->send_time_ms_ = clock_->TimeInMilliseconds(); |
| best_packet->IncrementTimesRetransmitted(&padding_priority_); |
| |
| return padding_packet; |
| } |
| |
| void RtpPacketHistory::CullAcknowledgedPackets( |
| rtc::ArrayView<const uint16_t> sequence_numbers) { |
| rtc::CritScope cs(&lock_); |
| for (uint16_t sequence_number : sequence_numbers) { |
| int packet_index = GetPacketIndex(sequence_number); |
| if (packet_index < 0 || |
| static_cast<size_t>(packet_index) >= packet_history_.size()) { |
| continue; |
| } |
| RemovePacket(packet_index); |
| } |
| } |
| |
| bool RtpPacketHistory::SetPendingTransmission(uint16_t sequence_number) { |
| rtc::CritScope cs(&lock_); |
| if (mode_ == StorageMode::kDisabled) { |
| return false; |
| } |
| |
| StoredPacket* packet = GetStoredPacket(sequence_number); |
| if (packet == nullptr) { |
| return false; |
| } |
| |
| packet->pending_transmission_ = true; |
| return true; |
| } |
| |
| void RtpPacketHistory::Clear() { |
| rtc::CritScope cs(&lock_); |
| Reset(); |
| } |
| |
| void RtpPacketHistory::Reset() { |
| packet_history_.clear(); |
| padding_priority_.clear(); |
| } |
| |
| void RtpPacketHistory::CullOldPackets(int64_t now_ms) { |
| int64_t packet_duration_ms = |
| std::max(kMinPacketDurationRtt * rtt_ms_, kMinPacketDurationMs); |
| while (!packet_history_.empty()) { |
| if (packet_history_.size() >= kMaxCapacity) { |
| // We have reached the absolute max capacity, remove one packet |
| // unconditionally. |
| RemovePacket(0); |
| continue; |
| } |
| |
| const StoredPacket& stored_packet = packet_history_.front(); |
| if (stored_packet.pending_transmission_) { |
| // Don't remove packets in the pacer queue, pending tranmission. |
| return; |
| } |
| |
| if (*stored_packet.send_time_ms_ + packet_duration_ms > now_ms) { |
| // Don't cull packets too early to avoid failed retransmission requests. |
| return; |
| } |
| |
| if (packet_history_.size() >= number_to_store_ || |
| *stored_packet.send_time_ms_ + |
| (packet_duration_ms * kPacketCullingDelayFactor) <= |
| now_ms) { |
| // Too many packets in history, or this packet has timed out. Remove it |
| // and continue. |
| RemovePacket(0); |
| } else { |
| // No more packets can be removed right now. |
| return; |
| } |
| } |
| } |
| |
| std::unique_ptr<RtpPacketToSend> RtpPacketHistory::RemovePacket( |
| int packet_index) { |
| // Move the packet out from the StoredPacket container. |
| std::unique_ptr<RtpPacketToSend> rtp_packet = |
| std::move(packet_history_[packet_index].packet_); |
| |
| // Erase from padding priority set, if eligible. |
| padding_priority_.erase(&packet_history_[packet_index]); |
| |
| if (packet_index == 0) { |
| while (!packet_history_.empty() && |
| packet_history_.front().packet_ == nullptr) { |
| packet_history_.pop_front(); |
| } |
| } |
| |
| return rtp_packet; |
| } |
| |
| int RtpPacketHistory::GetPacketIndex(uint16_t sequence_number) const { |
| if (packet_history_.empty()) { |
| return 0; |
| } |
| |
| RTC_DCHECK(packet_history_.front().packet_ != nullptr); |
| int first_seq = packet_history_.front().packet_->SequenceNumber(); |
| if (first_seq == sequence_number) { |
| return 0; |
| } |
| |
| int packet_index = sequence_number - first_seq; |
| constexpr int kSeqNumSpan = std::numeric_limits<uint16_t>::max() + 1; |
| |
| if (IsNewerSequenceNumber(sequence_number, first_seq)) { |
| if (sequence_number < first_seq) { |
| // Forward wrap. |
| packet_index += kSeqNumSpan; |
| } |
| } else if (sequence_number > first_seq) { |
| // Backwards wrap. |
| packet_index -= kSeqNumSpan; |
| } |
| |
| return packet_index; |
| } |
| |
| RtpPacketHistory::StoredPacket* RtpPacketHistory::GetStoredPacket( |
| uint16_t sequence_number) { |
| int index = GetPacketIndex(sequence_number); |
| if (index < 0 || static_cast<size_t>(index) >= packet_history_.size()) { |
| return nullptr; |
| } |
| return &packet_history_[index]; |
| } |
| |
| RtpPacketHistory::PacketState RtpPacketHistory::StoredPacketToPacketState( |
| const RtpPacketHistory::StoredPacket& stored_packet) { |
| RtpPacketHistory::PacketState state; |
| state.rtp_sequence_number = stored_packet.packet_->SequenceNumber(); |
| state.send_time_ms = stored_packet.send_time_ms_; |
| state.capture_time_ms = stored_packet.packet_->capture_time_ms(); |
| state.ssrc = stored_packet.packet_->Ssrc(); |
| state.packet_size = stored_packet.packet_->size(); |
| state.times_retransmitted = stored_packet.times_retransmitted(); |
| state.pending_transmission = stored_packet.pending_transmission_; |
| return state; |
| } |
| |
| } // namespace webrtc |