| /* |
| * Copyright (c) 2015 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/congestion_controller/rtp/transport_feedback_adapter.h" |
| |
| #include <algorithm> |
| #include <cstdint> |
| #include <cstdlib> |
| #include <optional> |
| #include <utility> |
| #include <vector> |
| |
| #include "absl/algorithm/container.h" |
| #include "api/transport/ecn_marking.h" |
| #include "api/transport/network_types.h" |
| #include "api/units/data_size.h" |
| #include "api/units/time_delta.h" |
| #include "api/units/timestamp.h" |
| #include "modules/rtp_rtcp/include/rtp_rtcp_defines.h" |
| #include "modules/rtp_rtcp/source/ntp_time_util.h" |
| #include "modules/rtp_rtcp/source/rtcp_packet/congestion_control_feedback.h" |
| #include "modules/rtp_rtcp/source/rtcp_packet/transport_feedback.h" |
| #include "modules/rtp_rtcp/source/rtp_packet_to_send.h" |
| #include "rtc_base/checks.h" |
| #include "rtc_base/logging.h" |
| #include "rtc_base/network/sent_packet.h" |
| #include "rtc_base/network_route.h" |
| |
| namespace webrtc { |
| namespace { |
| |
| constexpr TimeDelta kSendTimeHistoryWindow = TimeDelta::Seconds(60); |
| |
| class MinMax { |
| public: |
| explicit MinMax(int64_t value) : minimum_(value), maximum_(value) {} |
| |
| void Update(int64_t value) { |
| minimum_ = std::min(minimum_, value); |
| maximum_ = std::max(maximum_, value); |
| } |
| |
| int64_t minimum() const { return minimum_; } |
| int64_t maximum() const { return maximum_; } |
| |
| private: |
| int64_t minimum_; |
| int64_t maximum_; |
| }; |
| |
| } // namespace |
| |
| void InFlightBytesTracker::AddInFlightPacketBytes( |
| const PacketFeedback& packet) { |
| RTC_DCHECK(packet.sent.send_time.IsFinite()); |
| auto it = in_flight_data_.find(packet.network_route); |
| if (it != in_flight_data_.end()) { |
| it->second += packet.sent.size; |
| } else { |
| in_flight_data_.insert({packet.network_route, packet.sent.size}); |
| } |
| } |
| |
| void InFlightBytesTracker::RemoveInFlightPacketBytes( |
| const PacketFeedback& packet) { |
| if (packet.sent.send_time.IsInfinite()) |
| return; |
| auto it = in_flight_data_.find(packet.network_route); |
| if (it != in_flight_data_.end()) { |
| RTC_DCHECK_GE(it->second, packet.sent.size); |
| it->second -= packet.sent.size; |
| if (it->second.IsZero()) |
| in_flight_data_.erase(it); |
| } |
| } |
| |
| DataSize InFlightBytesTracker::GetOutstandingData( |
| const NetworkRoute& network_route) const { |
| auto it = in_flight_data_.find(network_route); |
| if (it != in_flight_data_.end()) { |
| return it->second; |
| } else { |
| return DataSize::Zero(); |
| } |
| } |
| |
| // Comparator for consistent map with NetworkRoute as key. |
| bool InFlightBytesTracker::NetworkRouteComparator::operator()( |
| const NetworkRoute& a, |
| const NetworkRoute& b) const { |
| if (a.local.network_id() != b.local.network_id()) |
| return a.local.network_id() < b.local.network_id(); |
| if (a.remote.network_id() != b.remote.network_id()) |
| return a.remote.network_id() < b.remote.network_id(); |
| |
| if (a.local.adapter_id() != b.local.adapter_id()) |
| return a.local.adapter_id() < b.local.adapter_id(); |
| if (a.remote.adapter_id() != b.remote.adapter_id()) |
| return a.remote.adapter_id() < b.remote.adapter_id(); |
| |
| if (a.local.uses_turn() != b.local.uses_turn()) |
| return a.local.uses_turn() < b.local.uses_turn(); |
| if (a.remote.uses_turn() != b.remote.uses_turn()) |
| return a.remote.uses_turn() < b.remote.uses_turn(); |
| |
| return a.connected < b.connected; |
| } |
| |
| TransportFeedbackAdapter::TransportFeedbackAdapter() = default; |
| |
| void TransportFeedbackAdapter::AddPacket(const RtpPacketToSend& packet_to_send, |
| const PacedPacketInfo& pacing_info, |
| size_t overhead_bytes, |
| Timestamp creation_time) { |
| RTC_DCHECK(packet_to_send.transport_sequence_number()); |
| PacketFeedback feedback; |
| |
| feedback.creation_time = creation_time; |
| // Note, if transport sequence number header extension is used, transport |
| // sequence numbers are wrapped to 16 bit. See |
| // RtpSenderEgress::CompleteSendPacket. |
| feedback.sent.sequence_number = seq_num_unwrapper_.Unwrap( |
| packet_to_send.transport_sequence_number().value_or(0)); |
| feedback.sent.size = DataSize::Bytes(packet_to_send.size() + overhead_bytes); |
| feedback.sent.audio = |
| packet_to_send.packet_type() == RtpPacketMediaType::kAudio; |
| feedback.network_route = network_route_; |
| feedback.sent.pacing_info = pacing_info; |
| feedback.ssrc = packet_to_send.Ssrc(); |
| feedback.rtp_sequence_number = packet_to_send.SequenceNumber(); |
| feedback.is_retransmission = |
| packet_to_send.packet_type() == RtpPacketMediaType::kRetransmission; |
| feedback.sent_with_ect1 = packet_to_send.send_as_ect1(); |
| |
| while (!history_.empty() && |
| creation_time - history_.begin()->second.creation_time > |
| kSendTimeHistoryWindow) { |
| // TODO(sprang): Warn if erasing (too many) old items? |
| if (history_.begin()->second.sent.sequence_number > last_ack_seq_num_) |
| in_flight_.RemoveInFlightPacketBytes(history_.begin()->second); |
| |
| const PacketFeedback& packet = history_.begin()->second; |
| rtp_to_transport_sequence_number_.erase( |
| {.ssrc = packet.ssrc, |
| .rtp_sequence_number = packet.rtp_sequence_number}); |
| history_.erase(history_.begin()); |
| } |
| // Note that it can happen that the same SSRC and sequence number is sent |
| // again, e.g., audio retransmission. Delay metrics from such packets are |
| // ambiguous, since it is not specified which of the two packets was |
| // received. |
| auto [it, inserted] = rtp_to_transport_sequence_number_.emplace( |
| SsrcAndRtpSequencenumber( |
| {.ssrc = feedback.ssrc, |
| .rtp_sequence_number = feedback.rtp_sequence_number}), |
| feedback.sent.sequence_number); |
| if (!inserted) { |
| feedback.ambiguous_receive_time = true; |
| auto prev_it = history_.find(it->second); |
| if (prev_it != history_.end()) { |
| prev_it->second.ambiguous_receive_time = true; |
| } |
| } else if (feedback.is_retransmission && |
| (!packet_to_send.original_ssrc().has_value() || |
| packet_to_send.original_ssrc() == packet_to_send.Ssrc())) { |
| // A retransmission without RTX (same SSRC and sequence number) whose |
| // original packet is no longer in `rtp_to_transport_sequence_number_` |
| // (e.g., the original packet was already acknowledged and removed from |
| // history). |
| feedback.ambiguous_receive_time = true; |
| } |
| history_.emplace(feedback.sent.sequence_number, feedback); |
| } |
| |
| std::optional<SentPacket> TransportFeedbackAdapter::ProcessSentPacket( |
| const SentPacketInfo& sent_packet) { |
| auto send_time = Timestamp::Millis(sent_packet.send_time_ms); |
| // TODO(srte): Only use one way to indicate that packet feedback is used. |
| if (sent_packet.info.included_in_feedback || sent_packet.packet_id != -1) { |
| int64_t unwrapped_seq_num = |
| seq_num_unwrapper_.Unwrap(sent_packet.packet_id); |
| auto it = history_.find(unwrapped_seq_num); |
| if (it != history_.end()) { |
| bool packet_retransmit = it->second.sent.send_time.IsFinite(); |
| it->second.sent.send_time = send_time; |
| last_send_time_ = std::max(last_send_time_, send_time); |
| // TODO(srte): Don't do this on retransmit. |
| if (!pending_untracked_size_.IsZero()) { |
| if (send_time < last_untracked_send_time_) |
| RTC_LOG(LS_WARNING) |
| << "appending acknowledged data for out of order packet. (Diff: " |
| << ToString(last_untracked_send_time_ - send_time) << " ms.)"; |
| it->second.sent.prior_unacked_data += pending_untracked_size_; |
| pending_untracked_size_ = DataSize::Zero(); |
| } |
| if (!packet_retransmit) { |
| if (it->second.sent.sequence_number > last_ack_seq_num_) |
| in_flight_.AddInFlightPacketBytes(it->second); |
| it->second.sent.data_in_flight = GetOutstandingData(); |
| return it->second.sent; |
| } |
| } |
| } else if (sent_packet.info.included_in_allocation) { |
| if (send_time < last_send_time_) { |
| RTC_LOG(LS_WARNING) << "ignoring untracked data for out of order packet."; |
| } |
| pending_untracked_size_ += |
| DataSize::Bytes(sent_packet.info.packet_size_bytes); |
| last_untracked_send_time_ = std::max(last_untracked_send_time_, send_time); |
| } |
| return std::nullopt; |
| } |
| |
| std::optional<TransportPacketsFeedback> |
| TransportFeedbackAdapter::ProcessTransportFeedback( |
| const rtcp::TransportFeedback& feedback, |
| Timestamp feedback_receive_time) { |
| if (feedback.GetPacketStatusCount() == 0) { |
| RTC_LOG(LS_INFO) << "Empty transport feedback packet received."; |
| return std::nullopt; |
| } |
| |
| // Add timestamp deltas to a local time base selected on first packet arrival. |
| // This won't be the true time base, but makes it easier to manually inspect |
| // time stamps. |
| if (last_transport_feedback_base_time_.IsInfinite()) { |
| current_offset_ = feedback_receive_time; |
| } else { |
| // TODO(srte): We shouldn't need to do rounding here. |
| const TimeDelta delta = |
| feedback.GetBaseDelta(last_transport_feedback_base_time_) |
| .RoundDownTo(TimeDelta::Millis(1)); |
| // Protect against assigning current_offset_ negative value. |
| if (delta < Timestamp::Zero() - current_offset_) { |
| RTC_LOG(LS_WARNING) << "Unexpected feedback timestamp received."; |
| current_offset_ = feedback_receive_time; |
| } else { |
| current_offset_ += delta; |
| } |
| } |
| last_transport_feedback_base_time_ = feedback.BaseTime(); |
| |
| std::vector<PacketResult> packet_result_vector; |
| packet_result_vector.reserve(feedback.GetPacketStatusCount()); |
| |
| size_t failed_lookups = 0; |
| size_t ignored = 0; |
| |
| feedback.ForAllPackets([&](uint16_t sequence_number, |
| TimeDelta delta_since_base) { |
| int64_t seq_num = seq_num_unwrapper_.Unwrap(sequence_number); |
| std::optional<PacketFeedback> packet_feedback = RetrievePacketFeedback( |
| seq_num, /*received=*/delta_since_base.IsFinite()); |
| if (!packet_feedback) { |
| ++failed_lookups; |
| return; |
| } |
| |
| if (packet_feedback->network_route == network_route_) { |
| PacketResult result; |
| result.sent_packet = packet_feedback->sent; |
| if (delta_since_base.IsFinite()) { |
| result.receive_time = current_offset_ + delta_since_base.RoundDownTo( |
| TimeDelta::Millis(1)); |
| } |
| result.rtp_packet_info = { |
| .ssrc = packet_feedback->ssrc, |
| .rtp_sequence_number = packet_feedback->rtp_sequence_number, |
| .is_retransmission = packet_feedback->is_retransmission}; |
| result.ambiguous_receive_time = packet_feedback->ambiguous_receive_time; |
| packet_result_vector.push_back(result); |
| } else { |
| ++ignored; |
| } |
| }); |
| |
| if (failed_lookups > 0) { |
| RTC_LOG(LS_WARNING) |
| << "Failed to lookup send time for " << failed_lookups << " packet" |
| << (failed_lookups > 1 ? "s" : "") |
| << ". Packets reordered or send time history too small?"; |
| } |
| if (ignored > 0) { |
| RTC_LOG(LS_INFO) << "Ignoring " << ignored |
| << " packets because they were sent on a different route."; |
| } |
| return ToTransportFeedback(std::move(packet_result_vector), |
| feedback_receive_time, /*supports_ecn=*/false); |
| } |
| |
| std::optional<TransportPacketsFeedback> |
| TransportFeedbackAdapter::ProcessCongestionControlFeedback( |
| const rtcp::CongestionControlFeedback& feedback, |
| Timestamp feedback_receive_time) { |
| if (feedback.packets().empty()) { |
| RTC_LOG(LS_INFO) << "Empty congestion control feedback packet received."; |
| return std::nullopt; |
| } |
| if (current_offset_.IsInfinite()) { |
| current_offset_ = feedback_receive_time; |
| } |
| TimeDelta feedback_delta = last_feedback_compact_ntp_time_ |
| ? CompactNtpIntervalToTimeDelta( |
| feedback.report_timestamp_compact_ntp() - |
| *last_feedback_compact_ntp_time_) |
| : TimeDelta::Zero(); |
| last_feedback_compact_ntp_time_ = feedback.report_timestamp_compact_ntp(); |
| |
| if (feedback_delta < TimeDelta::Seconds(-5)) { |
| RTC_LOG(LS_WARNING) << "Unexpected feedback ntp time delta " |
| << feedback_delta << "."; |
| current_offset_ = feedback_receive_time; |
| } else { |
| current_offset_ += feedback_delta; |
| } |
| |
| int ignored_packets = 0; |
| int failed_lookups = 0; |
| bool unexpected_arrival_time_offset = false; |
| bool supports_ecn = true; |
| std::vector<PacketResult> packet_result_vector; |
| std::optional<MinMax> sequence_number_in_report; |
| for (const rtcp::CongestionControlFeedback::PacketInfo& packet_info : |
| feedback.packets()) { |
| std::optional<PacketFeedback> packet_feedback = RetrievePacketFeedback( |
| {.ssrc = packet_info.ssrc, |
| .rtp_sequence_number = packet_info.sequence_number}, |
| /*received=*/packet_info.arrival_time_offset.IsFinite()); |
| if (!packet_feedback) { |
| ++failed_lookups; |
| continue; |
| } |
| if (packet_feedback->network_route != network_route_) { |
| ++ignored_packets; |
| continue; |
| } |
| if (!sequence_number_in_report.has_value()) { |
| sequence_number_in_report.emplace(packet_feedback->sent.sequence_number); |
| } else { |
| sequence_number_in_report->Update(packet_feedback->sent.sequence_number); |
| } |
| PacketResult result; |
| result.sent_packet = packet_feedback->sent; |
| if (packet_info.arrival_time_offset.IsFinite()) { |
| // Arrival time offset (ATO) is expected to be bounded by: |
| // 1. The interval between this feedback and the last received feedback |
| // (using the receiver's NTP report timestamps). Under RFC 8888, |
| // received packets must be reported. Since previously acknowledged |
| // packets are erased from `history_`, any packet newly acknowledged |
| // here must have arrived after the previous received feedback was |
| // generated. |
| // 2. The round-trip elapsed time between packet send and feedback |
| // receive. |
| // A packet cannot arrive before it was sent, so the time the receiver |
| // held the packet (ATO) cannot exceed the total round-trip duration. |
| TimeDelta time_between_two_last_feedback = |
| feedback_delta > TimeDelta::Zero() |
| ? feedback_delta + TimeDelta::Millis(1) |
| : TimeDelta::PlusInfinity(); |
| TimeDelta feedback_send_time_delta = |
| std::max(feedback_receive_time - packet_feedback->sent.send_time, |
| TimeDelta::Millis(0)); |
| TimeDelta max_expected_ato = |
| std::min(time_between_two_last_feedback, feedback_send_time_delta); |
| TimeDelta arrival_time_offset = packet_info.arrival_time_offset; |
| if (arrival_time_offset > max_expected_ato) { |
| RTC_LOG_IF(LS_WARNING, !unexpected_arrival_time_offset) |
| << "Arrival time offset " << packet_info.arrival_time_offset |
| << " exceeds max expected ATO " << max_expected_ato |
| << " (feedback_send_time_delta: " << feedback_send_time_delta |
| << ", time_between_two_last_feedback: " |
| << time_between_two_last_feedback << ")"; |
| unexpected_arrival_time_offset = true; |
| arrival_time_offset = max_expected_ato; |
| } |
| result.receive_time = current_offset_ - arrival_time_offset; |
| // Note that this deliberately does not look at how the packet was sent, |
| // in order to keep the deprecated `transport_supports_ecn` bit for bit |
| // compatible. Use `HasPacketWithBleachedEct1()` instead. |
| supports_ecn &= packet_info.ecn != EcnMarking::kNotEct; |
| result.arrival_time_offset = arrival_time_offset; |
| } |
| result.ambiguous_receive_time = packet_feedback->ambiguous_receive_time; |
| result.ecn = packet_info.ecn; |
| result.sent_with_ect1 = packet_feedback->sent_with_ect1; |
| result.reported_recovered_for_the_first_time = |
| packet_info.received() && packet_feedback->previously_reported_lost; |
| result.reported_lost_for_the_first_time = |
| !packet_info.received() && !packet_feedback->previously_reported_lost; |
| |
| result.rtp_packet_info = { |
| .ssrc = packet_feedback->ssrc, |
| .rtp_sequence_number = packet_feedback->rtp_sequence_number, |
| .is_retransmission = packet_feedback->is_retransmission}; |
| packet_result_vector.push_back(result); |
| } |
| |
| // Report packets as lost that are in the same transport sequence number range |
| // as reported packets, but that are not mentioned in the report. |
| // Code below relies on the fact that received packets are already erased from |
| // the `history_`. |
| if (sequence_number_in_report.has_value()) { |
| auto begin = history_.lower_bound(sequence_number_in_report->minimum()); |
| auto end = history_.upper_bound(sequence_number_in_report->maximum()); |
| for (auto it = begin; it != end; ++it) { |
| PacketFeedback& packet_feedback = it->second; |
| if (it->second.previously_reported_lost) { |
| continue; |
| } |
| it->second.previously_reported_lost = true; |
| PacketResult result; |
| result.sent_packet = packet_feedback.sent; |
| result.sent_with_ect1 = packet_feedback.sent_with_ect1; |
| result.reported_lost_for_the_first_time = true; |
| result.rtp_packet_info = { |
| .ssrc = packet_feedback.ssrc, |
| .rtp_sequence_number = packet_feedback.rtp_sequence_number, |
| .is_retransmission = packet_feedback.is_retransmission}; |
| packet_result_vector.push_back(result); |
| } |
| } |
| |
| if (failed_lookups > 0) { |
| RTC_LOG(LS_INFO) << "Failed to lookup send time for " << failed_lookups |
| << " packet" << (failed_lookups > 1 ? "s" : "") |
| << ". Packets reordered or send time history too " |
| "small, or packet reported received more than once."; |
| } |
| if (ignored_packets > 0) { |
| RTC_LOG(LS_INFO) << "Ignoring " << ignored_packets |
| << " packets because they were sent on a different route."; |
| } |
| |
| // Feedback is expected to be sorted in send order. |
| absl::c_sort(packet_result_vector, [](const PacketResult& lhs, |
| const PacketResult& rhs) { |
| return lhs.sent_packet.sequence_number < rhs.sent_packet.sequence_number; |
| }); |
| return ToTransportFeedback(std::move(packet_result_vector), |
| feedback_receive_time, supports_ecn); |
| } |
| |
| std::optional<TransportPacketsFeedback> |
| TransportFeedbackAdapter::ToTransportFeedback( |
| std::vector<PacketResult> packet_results, |
| Timestamp feedback_receive_time, |
| bool supports_ecn) { |
| TransportPacketsFeedback msg; |
| msg.feedback_time = feedback_receive_time; |
| if (packet_results.empty()) { |
| return std::nullopt; |
| } |
| msg.packet_feedbacks = std::move(packet_results); |
| msg.data_in_flight = in_flight_.GetOutstandingData(network_route_); |
| #pragma clang diagnostic push |
| #pragma clang diagnostic ignored "-Wdeprecated-declarations" |
| msg.transport_supports_ecn = supports_ecn; |
| #pragma clang diagnostic pop |
| |
| return msg; |
| } |
| |
| void TransportFeedbackAdapter::SetNetworkRoute( |
| const NetworkRoute& network_route) { |
| network_route_ = network_route; |
| } |
| |
| DataSize TransportFeedbackAdapter::GetOutstandingData() const { |
| return in_flight_.GetOutstandingData(network_route_); |
| } |
| |
| std::optional<PacketFeedback> TransportFeedbackAdapter::RetrievePacketFeedback( |
| const SsrcAndRtpSequencenumber& key, |
| bool received) { |
| auto it = rtp_to_transport_sequence_number_.find(key); |
| if (it == rtp_to_transport_sequence_number_.end()) { |
| return std::nullopt; |
| } |
| return RetrievePacketFeedback(it->second, received); |
| } |
| |
| std::optional<PacketFeedback> TransportFeedbackAdapter::RetrievePacketFeedback( |
| int64_t transport_seq_num, |
| bool received) { |
| if (transport_seq_num > last_ack_seq_num_) { |
| // Starts at history_.begin() if last_ack_seq_num_ < 0, since any |
| // valid sequence number is >= 0. |
| for (auto it = history_.upper_bound(last_ack_seq_num_); |
| it != history_.upper_bound(transport_seq_num); ++it) { |
| in_flight_.RemoveInFlightPacketBytes(it->second); |
| } |
| last_ack_seq_num_ = transport_seq_num; |
| } |
| |
| auto it = history_.find(transport_seq_num); |
| if (it == history_.end()) { |
| RTC_LOG(LS_INFO) << "Failed to lookup send time for packet with " |
| << transport_seq_num |
| << ". Send time history too small or packet was " |
| "reported as received twice."; |
| return std::nullopt; |
| } |
| |
| if (it->second.sent.send_time.IsInfinite()) { |
| // TODO(srte): Fix the tests that makes this happen and make this a |
| // DCHECK. |
| RTC_DLOG(LS_ERROR) |
| << "Received feedback before packet was indicated as sent"; |
| return std::nullopt; |
| } |
| |
| PacketFeedback packet_feedback = it->second; |
| if (received) { |
| // Note: Lost packets are not removed from history because they might |
| // be reported as received by a later feedback. |
| rtp_to_transport_sequence_number_.erase( |
| {.ssrc = packet_feedback.ssrc, |
| .rtp_sequence_number = packet_feedback.rtp_sequence_number}); |
| history_.erase(it); |
| } else { |
| // Update `previously_reported_lost` after taking copy of the |
| // `packet_feedback'. Thus the new value would be returned next time caller |
| // retrieves the feedback. This way caller can distinguish when packet was |
| // reported lost for the first time and thus account such loss only once. |
| it->second.previously_reported_lost = true; |
| } |
| return packet_feedback; |
| } |
| |
| } // namespace webrtc |
