diff options
Diffstat (limited to 'modules/remote_bitrate_estimator')
13 files changed, 870 insertions, 454 deletions
diff --git a/modules/remote_bitrate_estimator/BUILD.gn b/modules/remote_bitrate_estimator/BUILD.gn index 81aa1efdda..923f00a74c 100644 --- a/modules/remote_bitrate_estimator/BUILD.gn +++ b/modules/remote_bitrate_estimator/BUILD.gn @@ -21,6 +21,8 @@ rtc_library("remote_bitrate_estimator") { "overuse_detector.h", "overuse_estimator.cc", "overuse_estimator.h", + "packet_arrival_map.cc", + "packet_arrival_map.h", "remote_bitrate_estimator_abs_send_time.cc", "remote_bitrate_estimator_abs_send_time.h", "remote_bitrate_estimator_single_stream.cc", @@ -45,6 +47,8 @@ rtc_library("remote_bitrate_estimator") { "../../api/transport:network_control", "../../api/transport:webrtc_key_value_config", "../../api/units:data_rate", + "../../api/units:data_size", + "../../api/units:time_delta", "../../api/units:timestamp", "../../modules:module_api", "../../modules:module_api_public", @@ -74,10 +78,9 @@ if (!build_with_chromium) { "tools/bwe_rtp.h", ] deps = [ - ":remote_bitrate_estimator", "../../rtc_base:rtc_base_approved", "../../test:rtp_test_utils", - "../rtp_rtcp", + "../rtp_rtcp:rtp_rtcp_format", ] absl_deps = [ "//third_party/abseil-cpp/absl/flags:flag", @@ -90,10 +93,10 @@ if (!build_with_chromium) { sources = [ "tools/rtp_to_text.cc" ] deps = [ ":bwe_rtp", - "../../modules/rtp_rtcp", "../../rtc_base:macromagic", "../../rtc_base:stringutils", "../../test:rtp_test_utils", + "../rtp_rtcp:rtp_rtcp_format", ] } } @@ -106,6 +109,7 @@ if (rtc_include_tests) { "aimd_rate_control_unittest.cc", "inter_arrival_unittest.cc", "overuse_detector_unittest.cc", + "packet_arrival_map_test.cc", "remote_bitrate_estimator_abs_send_time_unittest.cc", "remote_bitrate_estimator_single_stream_unittest.cc", "remote_bitrate_estimator_unittest_helper.cc", diff --git a/modules/remote_bitrate_estimator/include/remote_bitrate_estimator.h b/modules/remote_bitrate_estimator/include/remote_bitrate_estimator.h index c60c030e8d..ac937bbfe0 100644 --- a/modules/remote_bitrate_estimator/include/remote_bitrate_estimator.h +++ b/modules/remote_bitrate_estimator/include/remote_bitrate_estimator.h @@ -38,14 +38,6 @@ class RemoteBitrateObserver { virtual ~RemoteBitrateObserver() {} }; -class TransportFeedbackSenderInterface { - public: - virtual ~TransportFeedbackSenderInterface() = default; - - virtual bool SendCombinedRtcpPacket( - std::vector<std::unique_ptr<rtcp::RtcpPacket>> packets) = 0; -}; - // TODO(holmer): Remove when all implementations have been updated. struct ReceiveBandwidthEstimatorStats {}; diff --git a/modules/remote_bitrate_estimator/packet_arrival_map.cc b/modules/remote_bitrate_estimator/packet_arrival_map.cc new file mode 100644 index 0000000000..72696f6c80 --- /dev/null +++ b/modules/remote_bitrate_estimator/packet_arrival_map.cc @@ -0,0 +1,123 @@ +/* + * Copyright (c) 2021 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/remote_bitrate_estimator/packet_arrival_map.h" + +#include <algorithm> + +#include "rtc_base/numerics/safe_minmax.h" + +namespace webrtc { + +constexpr size_t PacketArrivalTimeMap::kMaxNumberOfPackets; + +void PacketArrivalTimeMap::AddPacket(int64_t sequence_number, + int64_t arrival_time_ms) { + if (!has_seen_packet_) { + // First packet. + has_seen_packet_ = true; + begin_sequence_number_ = sequence_number; + arrival_times.push_back(arrival_time_ms); + return; + } + + int64_t pos = sequence_number - begin_sequence_number_; + if (pos >= 0 && pos < static_cast<int64_t>(arrival_times.size())) { + // The packet is within the buffer - no need to expand it. + arrival_times[pos] = arrival_time_ms; + return; + } + + if (pos < 0) { + // The packet goes before the current buffer. Expand to add packet, but only + // if it fits within kMaxNumberOfPackets. + size_t missing_packets = -pos; + if (missing_packets + arrival_times.size() > kMaxNumberOfPackets) { + // Don't expand the buffer further, as that would remove newly received + // packets. + return; + } + + arrival_times.insert(arrival_times.begin(), missing_packets, 0); + arrival_times[0] = arrival_time_ms; + begin_sequence_number_ = sequence_number; + return; + } + + // The packet goes after the buffer. + + if (static_cast<size_t>(pos) >= kMaxNumberOfPackets) { + // The buffer grows too large - old packets have to be removed. + size_t packets_to_remove = pos - kMaxNumberOfPackets + 1; + if (packets_to_remove >= arrival_times.size()) { + arrival_times.clear(); + begin_sequence_number_ = sequence_number; + pos = 0; + } else { + // Also trim the buffer to remove leading non-received packets, to + // ensure that the buffer only spans received packets. + while (packets_to_remove < arrival_times.size() && + arrival_times[packets_to_remove] == 0) { + ++packets_to_remove; + } + + arrival_times.erase(arrival_times.begin(), + arrival_times.begin() + packets_to_remove); + begin_sequence_number_ += packets_to_remove; + pos -= packets_to_remove; + RTC_DCHECK_GE(pos, 0); + } + } + + // Packets can be received out-of-order. If this isn't the next expected + // packet, add enough placeholders to fill the gap. + size_t missing_gap_packets = pos - arrival_times.size(); + if (missing_gap_packets > 0) { + arrival_times.insert(arrival_times.end(), missing_gap_packets, 0); + } + RTC_DCHECK_EQ(arrival_times.size(), pos); + arrival_times.push_back(arrival_time_ms); + RTC_DCHECK_LE(arrival_times.size(), kMaxNumberOfPackets); +} + +void PacketArrivalTimeMap::RemoveOldPackets(int64_t sequence_number, + int64_t arrival_time_limit) { + while (!arrival_times.empty() && begin_sequence_number_ < sequence_number && + arrival_times.front() <= arrival_time_limit) { + arrival_times.pop_front(); + ++begin_sequence_number_; + } +} + +bool PacketArrivalTimeMap::has_received(int64_t sequence_number) const { + int64_t pos = sequence_number - begin_sequence_number_; + if (pos >= 0 && pos < static_cast<int64_t>(arrival_times.size()) && + arrival_times[pos] != 0) { + return true; + } + return false; +} + +void PacketArrivalTimeMap::EraseTo(int64_t sequence_number) { + if (sequence_number > begin_sequence_number_) { + size_t count = + std::min(static_cast<size_t>(sequence_number - begin_sequence_number_), + arrival_times.size()); + + arrival_times.erase(arrival_times.begin(), arrival_times.begin() + count); + begin_sequence_number_ += count; + } +} + +int64_t PacketArrivalTimeMap::clamp(int64_t sequence_number) const { + return rtc::SafeClamp(sequence_number, begin_sequence_number(), + end_sequence_number()); +} + +} // namespace webrtc diff --git a/modules/remote_bitrate_estimator/packet_arrival_map.h b/modules/remote_bitrate_estimator/packet_arrival_map.h new file mode 100644 index 0000000000..10659e0f65 --- /dev/null +++ b/modules/remote_bitrate_estimator/packet_arrival_map.h @@ -0,0 +1,88 @@ +/* + * Copyright (c) 2021 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. + */ +#ifndef MODULES_REMOTE_BITRATE_ESTIMATOR_PACKET_ARRIVAL_MAP_H_ +#define MODULES_REMOTE_BITRATE_ESTIMATOR_PACKET_ARRIVAL_MAP_H_ + +#include <cstddef> +#include <cstdint> +#include <deque> + +#include "rtc_base/checks.h" + +namespace webrtc { + +// PacketArrivalTimeMap is an optimized map of packet sequence number to arrival +// time, limited in size to never exceed `kMaxNumberOfPackets`. It will grow as +// needed, and remove old packets, and will expand to allow earlier packets to +// be added (out-of-order). +// +// Not yet received packets have the arrival time zero. The queue will not span +// larger than necessary and the last packet should always be received. The +// first packet in the queue doesn't have to be received in case of receiving +// packets out-of-order. +class PacketArrivalTimeMap { + public: + // Impossible to request feedback older than what can be represented by 15 + // bits. + static constexpr size_t kMaxNumberOfPackets = (1 << 15); + + // Indicates if the packet with `sequence_number` has already been received. + bool has_received(int64_t sequence_number) const; + + // Returns the sequence number of the first entry in the map, i.e. the + // sequence number that a `begin()` iterator would represent. + int64_t begin_sequence_number() const { return begin_sequence_number_; } + + // Returns the sequence number of the element just after the map, i.e. the + // sequence number that an `end()` iterator would represent. + int64_t end_sequence_number() const { + return begin_sequence_number_ + arrival_times.size(); + } + + // Returns an element by `sequence_number`, which must be valid, i.e. + // between [begin_sequence_number, end_sequence_number). + int64_t get(int64_t sequence_number) { + int64_t pos = sequence_number - begin_sequence_number_; + RTC_DCHECK(pos >= 0 && pos < static_cast<int64_t>(arrival_times.size())); + return arrival_times[pos]; + } + + // Clamps `sequence_number` between [begin_sequence_number, + // end_sequence_number]. + int64_t clamp(int64_t sequence_number) const; + + // Erases all elements from the beginning of the map until `sequence_number`. + void EraseTo(int64_t sequence_number); + + // Records the fact that a packet with `sequence_number` arrived at + // `arrival_time_ms`. + void AddPacket(int64_t sequence_number, int64_t arrival_time_ms); + + // Removes packets from the beginning of the map as long as they are received + // before `sequence_number` and with an age older than `arrival_time_limit` + void RemoveOldPackets(int64_t sequence_number, int64_t arrival_time_limit); + + private: + // Deque representing unwrapped sequence number -> time, where the index + + // `begin_sequence_number_` represents the packet's sequence number. + std::deque<int64_t> arrival_times; + + // The unwrapped sequence number for the first element in + // `arrival_times`. + int64_t begin_sequence_number_ = 0; + + // Indicates if this map has had any packet added to it. The first packet + // decides the initial sequence number. + bool has_seen_packet_ = false; +}; + +} // namespace webrtc + +#endif // MODULES_REMOTE_BITRATE_ESTIMATOR_PACKET_ARRIVAL_MAP_H_ diff --git a/modules/remote_bitrate_estimator/packet_arrival_map_test.cc b/modules/remote_bitrate_estimator/packet_arrival_map_test.cc new file mode 100644 index 0000000000..afc7038832 --- /dev/null +++ b/modules/remote_bitrate_estimator/packet_arrival_map_test.cc @@ -0,0 +1,252 @@ +/* + * Copyright (c) 2021 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/remote_bitrate_estimator/packet_arrival_map.h" + +#include "test/gmock.h" +#include "test/gtest.h" + +namespace webrtc { +namespace { + +TEST(PacketArrivalMapTest, IsConsistentWhenEmpty) { + PacketArrivalTimeMap map; + + EXPECT_EQ(map.begin_sequence_number(), map.end_sequence_number()); + EXPECT_FALSE(map.has_received(0)); + EXPECT_EQ(map.clamp(-5), 0); + EXPECT_EQ(map.clamp(5), 0); +} + +TEST(PacketArrivalMapTest, InsertsFirstItemIntoMap) { + PacketArrivalTimeMap map; + + map.AddPacket(42, 10); + EXPECT_EQ(map.begin_sequence_number(), 42); + EXPECT_EQ(map.end_sequence_number(), 43); + + EXPECT_FALSE(map.has_received(41)); + EXPECT_TRUE(map.has_received(42)); + EXPECT_FALSE(map.has_received(44)); + + EXPECT_EQ(map.clamp(-100), 42); + EXPECT_EQ(map.clamp(42), 42); + EXPECT_EQ(map.clamp(100), 43); +} + +TEST(PacketArrivalMapTest, InsertsWithGaps) { + PacketArrivalTimeMap map; + + map.AddPacket(42, 10); + map.AddPacket(45, 11); + EXPECT_EQ(map.begin_sequence_number(), 42); + EXPECT_EQ(map.end_sequence_number(), 46); + + EXPECT_FALSE(map.has_received(41)); + EXPECT_TRUE(map.has_received(42)); + EXPECT_FALSE(map.has_received(43)); + EXPECT_FALSE(map.has_received(44)); + EXPECT_TRUE(map.has_received(45)); + EXPECT_FALSE(map.has_received(46)); + + EXPECT_EQ(map.get(42), 10); + EXPECT_EQ(map.get(43), 0); + EXPECT_EQ(map.get(44), 0); + EXPECT_EQ(map.get(45), 11); + + EXPECT_EQ(map.clamp(-100), 42); + EXPECT_EQ(map.clamp(44), 44); + EXPECT_EQ(map.clamp(100), 46); +} + +TEST(PacketArrivalMapTest, InsertsWithinBuffer) { + PacketArrivalTimeMap map; + + map.AddPacket(42, 10); + map.AddPacket(45, 11); + + map.AddPacket(43, 12); + map.AddPacket(44, 13); + + EXPECT_EQ(map.begin_sequence_number(), 42); + EXPECT_EQ(map.end_sequence_number(), 46); + + EXPECT_FALSE(map.has_received(41)); + EXPECT_TRUE(map.has_received(42)); + EXPECT_TRUE(map.has_received(43)); + EXPECT_TRUE(map.has_received(44)); + EXPECT_TRUE(map.has_received(45)); + EXPECT_FALSE(map.has_received(46)); + + EXPECT_EQ(map.get(42), 10); + EXPECT_EQ(map.get(43), 12); + EXPECT_EQ(map.get(44), 13); + EXPECT_EQ(map.get(45), 11); +} + +TEST(PacketArrivalMapTest, GrowsBufferAndRemoveOld) { + PacketArrivalTimeMap map; + + constexpr int64_t kLargeSeq = 42 + PacketArrivalTimeMap::kMaxNumberOfPackets; + map.AddPacket(42, 10); + map.AddPacket(43, 11); + map.AddPacket(44, 12); + map.AddPacket(45, 13); + map.AddPacket(kLargeSeq, 12); + + EXPECT_EQ(map.begin_sequence_number(), 43); + EXPECT_EQ(map.end_sequence_number(), kLargeSeq + 1); + EXPECT_EQ(static_cast<size_t>(map.end_sequence_number() - + map.begin_sequence_number()), + PacketArrivalTimeMap::kMaxNumberOfPackets); + + EXPECT_FALSE(map.has_received(41)); + EXPECT_FALSE(map.has_received(42)); + EXPECT_TRUE(map.has_received(43)); + EXPECT_TRUE(map.has_received(44)); + EXPECT_TRUE(map.has_received(45)); + EXPECT_FALSE(map.has_received(46)); + EXPECT_TRUE(map.has_received(kLargeSeq)); + EXPECT_FALSE(map.has_received(kLargeSeq + 1)); +} + +TEST(PacketArrivalMapTest, GrowsBufferAndRemoveOldTrimsBeginning) { + PacketArrivalTimeMap map; + + constexpr int64_t kLargeSeq = 42 + PacketArrivalTimeMap::kMaxNumberOfPackets; + map.AddPacket(42, 10); + // Missing: 43, 44 + map.AddPacket(45, 13); + map.AddPacket(kLargeSeq, 12); + + EXPECT_EQ(map.begin_sequence_number(), 45); + EXPECT_EQ(map.end_sequence_number(), kLargeSeq + 1); + + EXPECT_FALSE(map.has_received(44)); + EXPECT_TRUE(map.has_received(45)); + EXPECT_FALSE(map.has_received(46)); + EXPECT_TRUE(map.has_received(kLargeSeq)); + EXPECT_FALSE(map.has_received(kLargeSeq + 1)); +} + +TEST(PacketArrivalMapTest, SequenceNumberJumpsDeletesAll) { + PacketArrivalTimeMap map; + + constexpr int64_t kLargeSeq = + 42 + 2 * PacketArrivalTimeMap::kMaxNumberOfPackets; + map.AddPacket(42, 10); + map.AddPacket(kLargeSeq, 12); + + EXPECT_EQ(map.begin_sequence_number(), kLargeSeq); + EXPECT_EQ(map.end_sequence_number(), kLargeSeq + 1); + + EXPECT_FALSE(map.has_received(42)); + EXPECT_TRUE(map.has_received(kLargeSeq)); + EXPECT_FALSE(map.has_received(kLargeSeq + 1)); +} + +TEST(PacketArrivalMapTest, ExpandsBeforeBeginning) { + PacketArrivalTimeMap map; + + map.AddPacket(42, 10); + map.AddPacket(-1000, 13); + + EXPECT_EQ(map.begin_sequence_number(), -1000); + EXPECT_EQ(map.end_sequence_number(), 43); + + EXPECT_FALSE(map.has_received(-1001)); + EXPECT_TRUE(map.has_received(-1000)); + EXPECT_FALSE(map.has_received(-999)); + EXPECT_TRUE(map.has_received(42)); + EXPECT_FALSE(map.has_received(43)); +} + +TEST(PacketArrivalMapTest, ExpandingBeforeBeginningKeepsReceived) { + PacketArrivalTimeMap map; + + map.AddPacket(42, 10); + constexpr int64_t kSmallSeq = + static_cast<int64_t>(42) - 2 * PacketArrivalTimeMap::kMaxNumberOfPackets; + map.AddPacket(kSmallSeq, 13); + + EXPECT_EQ(map.begin_sequence_number(), 42); + EXPECT_EQ(map.end_sequence_number(), 43); +} + +TEST(PacketArrivalMapTest, ErasesToRemoveElements) { + PacketArrivalTimeMap map; + + map.AddPacket(42, 10); + map.AddPacket(43, 11); + map.AddPacket(44, 12); + map.AddPacket(45, 13); + + map.EraseTo(44); + + EXPECT_EQ(map.begin_sequence_number(), 44); + EXPECT_EQ(map.end_sequence_number(), 46); + + EXPECT_FALSE(map.has_received(43)); + EXPECT_TRUE(map.has_received(44)); + EXPECT_TRUE(map.has_received(45)); + EXPECT_FALSE(map.has_received(46)); +} + +TEST(PacketArrivalMapTest, ErasesInEmptyMap) { + PacketArrivalTimeMap map; + + EXPECT_EQ(map.begin_sequence_number(), map.end_sequence_number()); + + map.EraseTo(map.end_sequence_number()); + EXPECT_EQ(map.begin_sequence_number(), map.end_sequence_number()); +} + +TEST(PacketArrivalMapTest, IsTolerantToWrongArgumentsForErase) { + PacketArrivalTimeMap map; + + map.AddPacket(42, 10); + map.AddPacket(43, 11); + + map.EraseTo(1); + + EXPECT_EQ(map.begin_sequence_number(), 42); + EXPECT_EQ(map.end_sequence_number(), 44); + + map.EraseTo(100); + + EXPECT_EQ(map.begin_sequence_number(), 44); + EXPECT_EQ(map.end_sequence_number(), 44); +} + +TEST(PacketArrivalMapTest, EraseAllRemembersBeginningSeqNbr) { + PacketArrivalTimeMap map; + + map.AddPacket(42, 10); + map.AddPacket(43, 11); + map.AddPacket(44, 12); + map.AddPacket(45, 13); + + map.EraseTo(46); + + map.AddPacket(50, 10); + + EXPECT_EQ(map.begin_sequence_number(), 46); + EXPECT_EQ(map.end_sequence_number(), 51); + + EXPECT_FALSE(map.has_received(45)); + EXPECT_FALSE(map.has_received(46)); + EXPECT_FALSE(map.has_received(47)); + EXPECT_FALSE(map.has_received(48)); + EXPECT_FALSE(map.has_received(49)); + EXPECT_TRUE(map.has_received(50)); + EXPECT_FALSE(map.has_received(51)); +} + +} // namespace +} // namespace webrtc diff --git a/modules/remote_bitrate_estimator/remote_bitrate_estimator_abs_send_time.cc b/modules/remote_bitrate_estimator/remote_bitrate_estimator_abs_send_time.cc index 4196f6dc57..ae960ab960 100644 --- a/modules/remote_bitrate_estimator/remote_bitrate_estimator_abs_send_time.cc +++ b/modules/remote_bitrate_estimator/remote_bitrate_estimator_abs_send_time.cc @@ -13,18 +13,36 @@ #include <math.h> #include <algorithm> +#include <memory> +#include <utility> #include "api/transport/field_trial_based_config.h" +#include "api/units/data_rate.h" +#include "api/units/data_size.h" +#include "api/units/time_delta.h" +#include "api/units/timestamp.h" #include "modules/remote_bitrate_estimator/include/bwe_defines.h" #include "modules/remote_bitrate_estimator/include/remote_bitrate_estimator.h" #include "rtc_base/checks.h" -#include "rtc_base/constructor_magic.h" #include "rtc_base/logging.h" #include "rtc_base/thread_annotations.h" #include "system_wrappers/include/metrics.h" namespace webrtc { namespace { + +constexpr TimeDelta kMinClusterDelta = TimeDelta::Millis(1); +constexpr TimeDelta kInitialProbingInterval = TimeDelta::Seconds(2); +constexpr int kTimestampGroupLengthMs = 5; +constexpr int kAbsSendTimeInterArrivalUpshift = 8; +constexpr int kInterArrivalShift = + RTPHeaderExtension::kAbsSendTimeFraction + kAbsSendTimeInterArrivalUpshift; +constexpr int kMinClusterSize = 4; +constexpr int kMaxProbePackets = 15; +constexpr int kExpectedNumberOfProbes = 3; +constexpr double kTimestampToMs = + 1000.0 / static_cast<double>(1 << kInterArrivalShift); + absl::optional<DataRate> OptionalRateFromOptionalBps( absl::optional<int> bitrate_bps) { if (bitrate_bps) { @@ -33,62 +51,48 @@ absl::optional<DataRate> OptionalRateFromOptionalBps( return absl::nullopt; } } -} // namespace - -enum { - kTimestampGroupLengthMs = 5, - kAbsSendTimeInterArrivalUpshift = 8, - kInterArrivalShift = RTPHeaderExtension::kAbsSendTimeFraction + - kAbsSendTimeInterArrivalUpshift, - kInitialProbingIntervalMs = 2000, - kMinClusterSize = 4, - kMaxProbePackets = 15, - kExpectedNumberOfProbes = 3 -}; - -static const double kTimestampToMs = - 1000.0 / static_cast<double>(1 << kInterArrivalShift); template <typename K, typename V> std::vector<K> Keys(const std::map<K, V>& map) { std::vector<K> keys; keys.reserve(map.size()); - for (typename std::map<K, V>::const_iterator it = map.begin(); - it != map.end(); ++it) { - keys.push_back(it->first); + for (const auto& kv_pair : map) { + keys.push_back(kv_pair.first); } return keys; } -uint32_t ConvertMsTo24Bits(int64_t time_ms) { - uint32_t time_24_bits = - static_cast<uint32_t>(((static_cast<uint64_t>(time_ms) - << RTPHeaderExtension::kAbsSendTimeFraction) + - 500) / - 1000) & - 0x00FFFFFF; - return time_24_bits; -} +} // namespace RemoteBitrateEstimatorAbsSendTime::~RemoteBitrateEstimatorAbsSendTime() = default; bool RemoteBitrateEstimatorAbsSendTime::IsWithinClusterBounds( - int send_delta_ms, + TimeDelta send_delta, const Cluster& cluster_aggregate) { if (cluster_aggregate.count == 0) return true; - float cluster_mean = cluster_aggregate.send_mean_ms / - static_cast<float>(cluster_aggregate.count); - return fabs(static_cast<float>(send_delta_ms) - cluster_mean) < 2.5f; + TimeDelta cluster_mean = + cluster_aggregate.send_mean / cluster_aggregate.count; + return (send_delta - cluster_mean).Abs() < TimeDelta::Micros(2'500); } -void RemoteBitrateEstimatorAbsSendTime::AddCluster(std::list<Cluster>* clusters, - Cluster* cluster) { - cluster->send_mean_ms /= static_cast<float>(cluster->count); - cluster->recv_mean_ms /= static_cast<float>(cluster->count); - cluster->mean_size /= cluster->count; - clusters->push_back(*cluster); +void RemoteBitrateEstimatorAbsSendTime::MaybeAddCluster( + const Cluster& cluster_aggregate, + std::list<Cluster>& clusters) { + if (cluster_aggregate.count < kMinClusterSize || + cluster_aggregate.send_mean <= TimeDelta::Zero() || + cluster_aggregate.recv_mean <= TimeDelta::Zero()) { + return; + } + + Cluster cluster; + cluster.send_mean = cluster_aggregate.send_mean / cluster_aggregate.count; + cluster.recv_mean = cluster_aggregate.recv_mean / cluster_aggregate.count; + cluster.mean_size = cluster_aggregate.mean_size / cluster_aggregate.count; + cluster.count = cluster_aggregate.count; + cluster.num_above_min_delta = cluster_aggregate.num_above_min_delta; + clusters.push_back(cluster); } RemoteBitrateEstimatorAbsSendTime::RemoteBitrateEstimatorAbsSendTime( @@ -96,91 +100,77 @@ RemoteBitrateEstimatorAbsSendTime::RemoteBitrateEstimatorAbsSendTime( Clock* clock) : clock_(clock), observer_(observer), - inter_arrival_(), - estimator_(), detector_(&field_trials_), - incoming_bitrate_(kBitrateWindowMs, 8000), - incoming_bitrate_initialized_(false), - total_probes_received_(0), - first_packet_time_ms_(-1), - last_update_ms_(-1), - uma_recorded_(false), remote_rate_(&field_trials_) { RTC_DCHECK(clock_); RTC_DCHECK(observer_); RTC_LOG(LS_INFO) << "RemoteBitrateEstimatorAbsSendTime: Instantiating."; } -void RemoteBitrateEstimatorAbsSendTime::ComputeClusters( - std::list<Cluster>* clusters) const { - Cluster current; - int64_t prev_send_time = -1; - int64_t prev_recv_time = -1; - for (std::list<Probe>::const_iterator it = probes_.begin(); - it != probes_.end(); ++it) { - if (prev_send_time >= 0) { - int send_delta_ms = it->send_time_ms - prev_send_time; - int recv_delta_ms = it->recv_time_ms - prev_recv_time; - if (send_delta_ms >= 1 && recv_delta_ms >= 1) { - ++current.num_above_min_delta; +std::list<RemoteBitrateEstimatorAbsSendTime::Cluster> +RemoteBitrateEstimatorAbsSendTime::ComputeClusters() const { + std::list<Cluster> clusters; + Cluster cluster_aggregate; + Timestamp prev_send_time = Timestamp::MinusInfinity(); + Timestamp prev_recv_time = Timestamp::MinusInfinity(); + for (const Probe& probe : probes_) { + if (prev_send_time.IsFinite()) { + TimeDelta send_delta = probe.send_time - prev_send_time; + TimeDelta recv_delta = probe.recv_time - prev_recv_time; + if (send_delta >= kMinClusterDelta && recv_delta >= kMinClusterDelta) { + ++cluster_aggregate.num_above_min_delta; } - if (!IsWithinClusterBounds(send_delta_ms, current)) { - if (current.count >= kMinClusterSize && current.send_mean_ms > 0.0f && - current.recv_mean_ms > 0.0f) { - AddCluster(clusters, ¤t); - } - current = Cluster(); + if (!IsWithinClusterBounds(send_delta, cluster_aggregate)) { + MaybeAddCluster(cluster_aggregate, clusters); + cluster_aggregate = Cluster(); } - current.send_mean_ms += send_delta_ms; - current.recv_mean_ms += recv_delta_ms; - current.mean_size += it->payload_size; - ++current.count; + cluster_aggregate.send_mean += send_delta; + cluster_aggregate.recv_mean += recv_delta; + cluster_aggregate.mean_size += probe.payload_size; + ++cluster_aggregate.count; } - prev_send_time = it->send_time_ms; - prev_recv_time = it->recv_time_ms; - } - if (current.count >= kMinClusterSize && current.send_mean_ms > 0.0f && - current.recv_mean_ms > 0.0f) { - AddCluster(clusters, ¤t); + prev_send_time = probe.send_time; + prev_recv_time = probe.recv_time; } + MaybeAddCluster(cluster_aggregate, clusters); + return clusters; } -std::list<Cluster>::const_iterator +const RemoteBitrateEstimatorAbsSendTime::Cluster* RemoteBitrateEstimatorAbsSendTime::FindBestProbe( const std::list<Cluster>& clusters) const { - int highest_probe_bitrate_bps = 0; - std::list<Cluster>::const_iterator best_it = clusters.end(); - for (std::list<Cluster>::const_iterator it = clusters.begin(); - it != clusters.end(); ++it) { - if (it->send_mean_ms == 0 || it->recv_mean_ms == 0) + DataRate highest_probe_bitrate = DataRate::Zero(); + const Cluster* best = nullptr; + for (const auto& cluster : clusters) { + if (cluster.send_mean == TimeDelta::Zero() || + cluster.recv_mean == TimeDelta::Zero()) { continue; - if (it->num_above_min_delta > it->count / 2 && - (it->recv_mean_ms - it->send_mean_ms <= 2.0f && - it->send_mean_ms - it->recv_mean_ms <= 5.0f)) { - int probe_bitrate_bps = - std::min(it->GetSendBitrateBps(), it->GetRecvBitrateBps()); - if (probe_bitrate_bps > highest_probe_bitrate_bps) { - highest_probe_bitrate_bps = probe_bitrate_bps; - best_it = it; + } + if (cluster.num_above_min_delta > cluster.count / 2 && + (cluster.recv_mean - cluster.send_mean <= TimeDelta::Millis(2) && + cluster.send_mean - cluster.recv_mean <= TimeDelta::Millis(5))) { + DataRate probe_bitrate = + std::min(cluster.SendBitrate(), cluster.RecvBitrate()); + if (probe_bitrate > highest_probe_bitrate) { + highest_probe_bitrate = probe_bitrate; + best = &cluster; } } else { - int send_bitrate_bps = it->mean_size * 8 * 1000 / it->send_mean_ms; - int recv_bitrate_bps = it->mean_size * 8 * 1000 / it->recv_mean_ms; - RTC_LOG(LS_INFO) << "Probe failed, sent at " << send_bitrate_bps - << " bps, received at " << recv_bitrate_bps - << " bps. Mean send delta: " << it->send_mean_ms - << " ms, mean recv delta: " << it->recv_mean_ms - << " ms, num probes: " << it->count; + RTC_LOG(LS_INFO) << "Probe failed, sent at " + << cluster.SendBitrate().bps() << " bps, received at " + << cluster.RecvBitrate().bps() + << " bps. Mean send delta: " << cluster.send_mean.ms() + << " ms, mean recv delta: " << cluster.recv_mean.ms() + << " ms, num probes: " << cluster.count; break; } } - return best_it; + return best; } RemoteBitrateEstimatorAbsSendTime::ProbeResult -RemoteBitrateEstimatorAbsSendTime::ProcessClusters(int64_t now_ms) { - std::list<Cluster> clusters; - ComputeClusters(&clusters); +RemoteBitrateEstimatorAbsSendTime::ProcessClusters(Timestamp now) { + std::list<Cluster> clusters = ComputeClusters(); if (clusters.empty()) { // If we reach the max number of probe packets and still have no clusters, // we will remove the oldest one. @@ -189,21 +179,18 @@ RemoteBitrateEstimatorAbsSendTime::ProcessClusters(int64_t now_ms) { return ProbeResult::kNoUpdate; } - std::list<Cluster>::const_iterator best_it = FindBestProbe(clusters); - if (best_it != clusters.end()) { - int probe_bitrate_bps = - std::min(best_it->GetSendBitrateBps(), best_it->GetRecvBitrateBps()); + if (const Cluster* best = FindBestProbe(clusters)) { + DataRate probe_bitrate = std::min(best->SendBitrate(), best->RecvBitrate()); // Make sure that a probe sent on a lower bitrate than our estimate can't // reduce the estimate. - if (IsBitrateImproving(probe_bitrate_bps)) { + if (IsBitrateImproving(probe_bitrate)) { RTC_LOG(LS_INFO) << "Probe successful, sent at " - << best_it->GetSendBitrateBps() << " bps, received at " - << best_it->GetRecvBitrateBps() - << " bps. Mean send delta: " << best_it->send_mean_ms - << " ms, mean recv delta: " << best_it->recv_mean_ms - << " ms, num probes: " << best_it->count; - remote_rate_.SetEstimate(DataRate::BitsPerSec(probe_bitrate_bps), - Timestamp::Millis(now_ms)); + << best->SendBitrate().bps() << " bps, received at " + << best->RecvBitrate().bps() + << " bps. Mean send delta: " << best->send_mean.ms() + << " ms, mean recv delta: " << best->recv_mean.ms() + << " ms, num probes: " << best->count; + remote_rate_.SetEstimate(probe_bitrate, now); return ProbeResult::kBitrateUpdated; } } @@ -216,11 +203,11 @@ RemoteBitrateEstimatorAbsSendTime::ProcessClusters(int64_t now_ms) { } bool RemoteBitrateEstimatorAbsSendTime::IsBitrateImproving( - int new_bitrate_bps) const { - bool initial_probe = !remote_rate_.ValidEstimate() && new_bitrate_bps > 0; - bool bitrate_above_estimate = - remote_rate_.ValidEstimate() && - new_bitrate_bps > remote_rate_.LatestEstimate().bps<int>(); + DataRate probe_bitrate) const { + bool initial_probe = + !remote_rate_.ValidEstimate() && probe_bitrate > DataRate::Zero(); + bool bitrate_above_estimate = remote_rate_.ValidEstimate() && + probe_bitrate > remote_rate_.LatestEstimate(); return initial_probe || bitrate_above_estimate; } @@ -235,14 +222,15 @@ void RemoteBitrateEstimatorAbsSendTime::IncomingPacket( "is missing absolute send time extension!"; return; } - IncomingPacketInfo(arrival_time_ms, header.extension.absoluteSendTime, - payload_size, header.ssrc); + IncomingPacketInfo(Timestamp::Millis(arrival_time_ms), + header.extension.absoluteSendTime, + DataSize::Bytes(payload_size), header.ssrc); } void RemoteBitrateEstimatorAbsSendTime::IncomingPacketInfo( - int64_t arrival_time_ms, + Timestamp arrival_time, uint32_t send_time_24bits, - size_t payload_size, + DataSize payload_size, uint32_t ssrc) { RTC_CHECK(send_time_24bits < (1ul << 24)); if (!uma_recorded_) { @@ -253,15 +241,16 @@ void RemoteBitrateEstimatorAbsSendTime::IncomingPacketInfo( // Shift up send time to use the full 32 bits that inter_arrival works with, // so wrapping works properly. uint32_t timestamp = send_time_24bits << kAbsSendTimeInterArrivalUpshift; - int64_t send_time_ms = static_cast<int64_t>(timestamp) * kTimestampToMs; + Timestamp send_time = + Timestamp::Millis(static_cast<int64_t>(timestamp) * kTimestampToMs); - int64_t now_ms = clock_->TimeInMilliseconds(); + Timestamp now = clock_->CurrentTime(); // TODO(holmer): SSRCs are only needed for REMB, should be broken out from // here. // Check if incoming bitrate estimate is valid, and if it needs to be reset. absl::optional<uint32_t> incoming_bitrate = - incoming_bitrate_.Rate(arrival_time_ms); + incoming_bitrate_.Rate(arrival_time.ms()); if (incoming_bitrate) { incoming_bitrate_initialized_ = true; } else if (incoming_bitrate_initialized_) { @@ -271,74 +260,82 @@ void RemoteBitrateEstimatorAbsSendTime::IncomingPacketInfo( incoming_bitrate_.Reset(); incoming_bitrate_initialized_ = false; } - incoming_bitrate_.Update(payload_size, arrival_time_ms); + incoming_bitrate_.Update(payload_size.bytes(), arrival_time.ms()); - if (first_packet_time_ms_ == -1) - first_packet_time_ms_ = now_ms; + if (first_packet_time_.IsInfinite()) { + first_packet_time_ = now; + } uint32_t ts_delta = 0; int64_t t_delta = 0; int size_delta = 0; bool update_estimate = false; - uint32_t target_bitrate_bps = 0; + DataRate target_bitrate = DataRate::Zero(); std::vector<uint32_t> ssrcs; { MutexLock lock(&mutex_); - TimeoutStreams(now_ms); - RTC_DCHECK(inter_arrival_.get()); - RTC_DCHECK(estimator_.get()); - ssrcs_[ssrc] = now_ms; + TimeoutStreams(now); + RTC_DCHECK(inter_arrival_); + RTC_DCHECK(estimator_); + // TODO(danilchap): Replace 5 lines below with insert_or_assign when that + // c++17 function is available. + auto inserted = ssrcs_.insert(std::make_pair(ssrc, now)); + if (!inserted.second) { + // Already inserted, update. + inserted.first->second = now; + } // For now only try to detect probes while we don't have a valid estimate. // We currently assume that only packets larger than 200 bytes are paced by // the sender. - const size_t kMinProbePacketSize = 200; + static constexpr DataSize kMinProbePacketSize = DataSize::Bytes(200); if (payload_size > kMinProbePacketSize && (!remote_rate_.ValidEstimate() || - now_ms - first_packet_time_ms_ < kInitialProbingIntervalMs)) { + now - first_packet_time_ < kInitialProbingInterval)) { // TODO(holmer): Use a map instead to get correct order? if (total_probes_received_ < kMaxProbePackets) { - int send_delta_ms = -1; - int recv_delta_ms = -1; + TimeDelta send_delta = TimeDelta::Millis(-1); + TimeDelta recv_delta = TimeDelta::Millis(-1); if (!probes_.empty()) { - send_delta_ms = send_time_ms - probes_.back().send_time_ms; - recv_delta_ms = arrival_time_ms - probes_.back().recv_time_ms; + send_delta = send_time - probes_.back().send_time; + recv_delta = arrival_time - probes_.back().recv_time; } - RTC_LOG(LS_INFO) << "Probe packet received: send time=" << send_time_ms - << " ms, recv time=" << arrival_time_ms - << " ms, send delta=" << send_delta_ms - << " ms, recv delta=" << recv_delta_ms << " ms."; + RTC_LOG(LS_INFO) << "Probe packet received: send time=" + << send_time.ms() + << " ms, recv time=" << arrival_time.ms() + << " ms, send delta=" << send_delta.ms() + << " ms, recv delta=" << recv_delta.ms() << " ms."; } - probes_.push_back(Probe(send_time_ms, arrival_time_ms, payload_size)); + probes_.emplace_back(send_time, arrival_time, payload_size); ++total_probes_received_; // Make sure that a probe which updated the bitrate immediately has an // effect by calling the OnReceiveBitrateChanged callback. - if (ProcessClusters(now_ms) == ProbeResult::kBitrateUpdated) + if (ProcessClusters(now) == ProbeResult::kBitrateUpdated) update_estimate = true; } - if (inter_arrival_->ComputeDeltas(timestamp, arrival_time_ms, now_ms, - payload_size, &ts_delta, &t_delta, + if (inter_arrival_->ComputeDeltas(timestamp, arrival_time.ms(), now.ms(), + payload_size.bytes(), &ts_delta, &t_delta, &size_delta)) { double ts_delta_ms = (1000.0 * ts_delta) / (1 << kInterArrivalShift); estimator_->Update(t_delta, ts_delta_ms, size_delta, detector_.State(), - arrival_time_ms); + arrival_time.ms()); detector_.Detect(estimator_->offset(), ts_delta_ms, - estimator_->num_of_deltas(), arrival_time_ms); + estimator_->num_of_deltas(), arrival_time.ms()); } if (!update_estimate) { // Check if it's time for a periodic update or if we should update because // of an over-use. - if (last_update_ms_ == -1 || - now_ms - last_update_ms_ > remote_rate_.GetFeedbackInterval().ms()) { + if (last_update_.IsInfinite() || + now.ms() - last_update_.ms() > + remote_rate_.GetFeedbackInterval().ms()) { update_estimate = true; } else if (detector_.State() == BandwidthUsage::kBwOverusing) { absl::optional<uint32_t> incoming_rate = - incoming_bitrate_.Rate(arrival_time_ms); + incoming_bitrate_.Rate(arrival_time.ms()); if (incoming_rate && remote_rate_.TimeToReduceFurther( - Timestamp::Millis(now_ms), - DataRate::BitsPerSec(*incoming_rate))) { + now, DataRate::BitsPerSec(*incoming_rate))) { update_estimate = true; } } @@ -349,18 +346,16 @@ void RemoteBitrateEstimatorAbsSendTime::IncomingPacketInfo( // We also have to update the estimate immediately if we are overusing // and the target bitrate is too high compared to what we are receiving. const RateControlInput input( - detector_.State(), - OptionalRateFromOptionalBps(incoming_bitrate_.Rate(arrival_time_ms))); - target_bitrate_bps = - remote_rate_.Update(&input, Timestamp::Millis(now_ms)) - .bps<uint32_t>(); + detector_.State(), OptionalRateFromOptionalBps( + incoming_bitrate_.Rate(arrival_time.ms()))); + target_bitrate = remote_rate_.Update(&input, now); update_estimate = remote_rate_.ValidEstimate(); ssrcs = Keys(ssrcs_); } } if (update_estimate) { - last_update_ms_ = now_ms; - observer_->OnReceiveBitrateChanged(ssrcs, target_bitrate_bps); + last_update_ = now; + observer_->OnReceiveBitrateChanged(ssrcs, target_bitrate.bps<uint32_t>()); } } @@ -371,9 +366,9 @@ int64_t RemoteBitrateEstimatorAbsSendTime::TimeUntilNextProcess() { return kDisabledModuleTime; } -void RemoteBitrateEstimatorAbsSendTime::TimeoutStreams(int64_t now_ms) { - for (Ssrcs::iterator it = ssrcs_.begin(); it != ssrcs_.end();) { - if ((now_ms - it->second) > kStreamTimeOutMs) { +void RemoteBitrateEstimatorAbsSendTime::TimeoutStreams(Timestamp now) { + for (auto it = ssrcs_.begin(); it != ssrcs_.end();) { + if (now - it->second > TimeDelta::Millis(kStreamTimeOutMs)) { ssrcs_.erase(it++); } else { ++it; @@ -381,17 +376,17 @@ void RemoteBitrateEstimatorAbsSendTime::TimeoutStreams(int64_t now_ms) { } if (ssrcs_.empty()) { // We can't update the estimate if we don't have any active streams. - inter_arrival_.reset( - new InterArrival((kTimestampGroupLengthMs << kInterArrivalShift) / 1000, - kTimestampToMs, true)); - estimator_.reset(new OveruseEstimator(OverUseDetectorOptions())); + inter_arrival_ = std::make_unique<InterArrival>( + (kTimestampGroupLengthMs << kInterArrivalShift) / 1000, kTimestampToMs, + true); + estimator_ = std::make_unique<OveruseEstimator>(OverUseDetectorOptions()); // We deliberately don't reset the first_packet_time_ms_ here for now since // we only probe for bandwidth in the beginning of a call right now. } } void RemoteBitrateEstimatorAbsSendTime::OnRttUpdate(int64_t avg_rtt_ms, - int64_t max_rtt_ms) { + int64_t /*max_rtt_ms*/) { MutexLock lock(&mutex_); remote_rate_.SetRtt(TimeDelta::Millis(avg_rtt_ms)); } diff --git a/modules/remote_bitrate_estimator/remote_bitrate_estimator_abs_send_time.h b/modules/remote_bitrate_estimator/remote_bitrate_estimator_abs_send_time.h index f42a28f8c8..4117382577 100644 --- a/modules/remote_bitrate_estimator/remote_bitrate_estimator_abs_send_time.h +++ b/modules/remote_bitrate_estimator/remote_bitrate_estimator_abs_send_time.h @@ -21,6 +21,10 @@ #include "api/rtp_headers.h" #include "api/transport/field_trial_based_config.h" +#include "api/units/data_rate.h" +#include "api/units/data_size.h" +#include "api/units/time_delta.h" +#include "api/units/timestamp.h" #include "modules/remote_bitrate_estimator/aimd_rate_control.h" #include "modules/remote_bitrate_estimator/include/remote_bitrate_estimator.h" #include "modules/remote_bitrate_estimator/inter_arrival.h" @@ -35,42 +39,6 @@ namespace webrtc { -struct Probe { - Probe(int64_t send_time_ms, int64_t recv_time_ms, size_t payload_size) - : send_time_ms(send_time_ms), - recv_time_ms(recv_time_ms), - payload_size(payload_size) {} - int64_t send_time_ms; - int64_t recv_time_ms; - size_t payload_size; -}; - -struct Cluster { - Cluster() - : send_mean_ms(0.0f), - recv_mean_ms(0.0f), - mean_size(0), - count(0), - num_above_min_delta(0) {} - - int GetSendBitrateBps() const { - RTC_CHECK_GT(send_mean_ms, 0.0f); - return mean_size * 8 * 1000 / send_mean_ms; - } - - int GetRecvBitrateBps() const { - RTC_CHECK_GT(recv_mean_ms, 0.0f); - return mean_size * 8 * 1000 / recv_mean_ms; - } - - float send_mean_ms; - float recv_mean_ms; - // TODO(holmer): Add some variance metric as well? - size_t mean_size; - int count; - int num_above_min_delta; -}; - class RemoteBitrateEstimatorAbsSendTime : public RemoteBitrateEstimator { public: RemoteBitrateEstimatorAbsSendTime(RemoteBitrateObserver* observer, @@ -100,32 +68,54 @@ class RemoteBitrateEstimatorAbsSendTime : public RemoteBitrateEstimator { void SetMinBitrate(int min_bitrate_bps) override; private: - typedef std::map<uint32_t, int64_t> Ssrcs; + struct Probe { + Probe(Timestamp send_time, Timestamp recv_time, DataSize payload_size) + : send_time(send_time), + recv_time(recv_time), + payload_size(payload_size) {} + + Timestamp send_time; + Timestamp recv_time; + DataSize payload_size; + }; + + struct Cluster { + DataRate SendBitrate() const { return mean_size / send_mean; } + DataRate RecvBitrate() const { return mean_size / recv_mean; } + + TimeDelta send_mean = TimeDelta::Zero(); + TimeDelta recv_mean = TimeDelta::Zero(); + // TODO(holmer): Add some variance metric as well? + DataSize mean_size = DataSize::Zero(); + int count = 0; + int num_above_min_delta = 0; + }; + enum class ProbeResult { kBitrateUpdated, kNoUpdate }; - static bool IsWithinClusterBounds(int send_delta_ms, + static bool IsWithinClusterBounds(TimeDelta send_delta, const Cluster& cluster_aggregate); - static void AddCluster(std::list<Cluster>* clusters, Cluster* cluster); + static void MaybeAddCluster(const Cluster& cluster_aggregate, + std::list<Cluster>& clusters); - void IncomingPacketInfo(int64_t arrival_time_ms, + void IncomingPacketInfo(Timestamp arrival_time, uint32_t send_time_24bits, - size_t payload_size, + DataSize payload_size, uint32_t ssrc); - void ComputeClusters(std::list<Cluster>* clusters) const; + std::list<Cluster> ComputeClusters() const; - std::list<Cluster>::const_iterator FindBestProbe( - const std::list<Cluster>& clusters) const; + const Cluster* FindBestProbe(const std::list<Cluster>& clusters) const; // Returns true if a probe which changed the estimate was detected. - ProbeResult ProcessClusters(int64_t now_ms) + ProbeResult ProcessClusters(Timestamp now) RTC_EXCLUSIVE_LOCKS_REQUIRED(&mutex_); - bool IsBitrateImproving(int probe_bitrate_bps) const + bool IsBitrateImproving(DataRate probe_bitrate) const RTC_EXCLUSIVE_LOCKS_REQUIRED(&mutex_); - void TimeoutStreams(int64_t now_ms) RTC_EXCLUSIVE_LOCKS_REQUIRED(&mutex_); + void TimeoutStreams(Timestamp now) RTC_EXCLUSIVE_LOCKS_REQUIRED(&mutex_); rtc::RaceChecker network_race_; Clock* const clock_; @@ -134,18 +124,16 @@ class RemoteBitrateEstimatorAbsSendTime : public RemoteBitrateEstimator { std::unique_ptr<InterArrival> inter_arrival_; std::unique_ptr<OveruseEstimator> estimator_; OveruseDetector detector_; - RateStatistics incoming_bitrate_; - bool incoming_bitrate_initialized_; - std::vector<int> recent_propagation_delta_ms_; - std::vector<int64_t> recent_update_time_ms_; + RateStatistics incoming_bitrate_{kBitrateWindowMs, 8000}; + bool incoming_bitrate_initialized_ = false; std::list<Probe> probes_; - size_t total_probes_received_; - int64_t first_packet_time_ms_; - int64_t last_update_ms_; - bool uma_recorded_; + size_t total_probes_received_ = 0; + Timestamp first_packet_time_ = Timestamp::MinusInfinity(); + Timestamp last_update_ = Timestamp::MinusInfinity(); + bool uma_recorded_ = false; mutable Mutex mutex_; - Ssrcs ssrcs_ RTC_GUARDED_BY(&mutex_); + std::map<uint32_t, Timestamp> ssrcs_ RTC_GUARDED_BY(&mutex_); AimdRateControl remote_rate_ RTC_GUARDED_BY(&mutex_); }; diff --git a/modules/remote_bitrate_estimator/remote_estimator_proxy.cc b/modules/remote_bitrate_estimator/remote_estimator_proxy.cc index a9cc170a35..7764e60ef2 100644 --- a/modules/remote_bitrate_estimator/remote_estimator_proxy.cc +++ b/modules/remote_bitrate_estimator/remote_estimator_proxy.cc @@ -23,9 +23,6 @@ namespace webrtc { -// Impossible to request feedback older than what can be represented by 15 bits. -const int RemoteEstimatorProxy::kMaxNumberOfPackets = (1 << 15); - // The maximum allowed value for a timestamp in milliseconds. This is lower // than the numerical limit since we often convert to microseconds. static constexpr int64_t kMaxTimeMs = @@ -33,11 +30,11 @@ static constexpr int64_t kMaxTimeMs = RemoteEstimatorProxy::RemoteEstimatorProxy( Clock* clock, - TransportFeedbackSenderInterface* feedback_sender, + TransportFeedbackSender feedback_sender, const WebRtcKeyValueConfig* key_value_config, NetworkStateEstimator* network_state_estimator) : clock_(clock), - feedback_sender_(feedback_sender), + feedback_sender_(std::move(feedback_sender)), send_config_(key_value_config), last_process_time_ms_(-1), network_state_estimator_(network_state_estimator), @@ -54,6 +51,18 @@ RemoteEstimatorProxy::RemoteEstimatorProxy( RemoteEstimatorProxy::~RemoteEstimatorProxy() {} +void RemoteEstimatorProxy::MaybeCullOldPackets(int64_t sequence_number, + int64_t arrival_time_ms) { + if (periodic_window_start_seq_.has_value()) { + if (*periodic_window_start_seq_ >= + packet_arrival_times_.end_sequence_number()) { + // Start new feedback packet, cull old packets. + packet_arrival_times_.RemoveOldPackets( + sequence_number, arrival_time_ms - send_config_.back_window->ms()); + } + } +} + void RemoteEstimatorProxy::IncomingPacket(int64_t arrival_time_ms, size_t payload_size, const RTPHeader& header) { @@ -69,39 +78,26 @@ void RemoteEstimatorProxy::IncomingPacket(int64_t arrival_time_ms, seq = unwrapper_.Unwrap(header.extension.transportSequenceNumber); if (send_periodic_feedback_) { - if (periodic_window_start_seq_ && - packet_arrival_times_.lower_bound(*periodic_window_start_seq_) == - packet_arrival_times_.end()) { - // Start new feedback packet, cull old packets. - for (auto it = packet_arrival_times_.begin(); - it != packet_arrival_times_.end() && it->first < seq && - arrival_time_ms - it->second >= send_config_.back_window->ms();) { - it = packet_arrival_times_.erase(it); - } - } + MaybeCullOldPackets(seq, arrival_time_ms); + if (!periodic_window_start_seq_ || seq < *periodic_window_start_seq_) { periodic_window_start_seq_ = seq; } } // We are only interested in the first time a packet is received. - if (packet_arrival_times_.find(seq) != packet_arrival_times_.end()) + if (packet_arrival_times_.has_received(seq)) { return; + } - packet_arrival_times_[seq] = arrival_time_ms; + packet_arrival_times_.AddPacket(seq, arrival_time_ms); // Limit the range of sequence numbers to send feedback for. - auto first_arrival_time_to_keep = packet_arrival_times_.lower_bound( - packet_arrival_times_.rbegin()->first - kMaxNumberOfPackets); - if (first_arrival_time_to_keep != packet_arrival_times_.begin()) { - packet_arrival_times_.erase(packet_arrival_times_.begin(), - first_arrival_time_to_keep); - if (send_periodic_feedback_) { - // |packet_arrival_times_| cannot be empty since we just added one - // element and the last element is not deleted. - RTC_DCHECK(!packet_arrival_times_.empty()); - periodic_window_start_seq_ = packet_arrival_times_.begin()->first; - } + if (!periodic_window_start_seq_.has_value() || + periodic_window_start_seq_.value() < + packet_arrival_times_.begin_sequence_number()) { + periodic_window_start_seq_ = + packet_arrival_times_.begin_sequence_number(); } if (header.extension.feedback_request) { @@ -113,8 +109,8 @@ void RemoteEstimatorProxy::IncomingPacket(int64_t arrival_time_ms, if (network_state_estimator_ && header.extension.hasAbsoluteSendTime) { PacketResult packet_result; packet_result.receive_time = Timestamp::Millis(arrival_time_ms); - // Ignore reordering of packets and assume they have approximately the same - // send time. + // Ignore reordering of packets and assume they have approximately the + // same send time. abs_send_timestamp_ += std::max( header.extension.GetAbsoluteSendTimeDelta(previous_abs_send_time_), TimeDelta::Millis(0)); @@ -183,9 +179,9 @@ void RemoteEstimatorProxy::SetSendPeriodicFeedback( } void RemoteEstimatorProxy::SendPeriodicFeedbacks() { - // |periodic_window_start_seq_| is the first sequence number to include in the - // current feedback packet. Some older may still be in the map, in case a - // reordering happens and we need to retransmit them. + // |periodic_window_start_seq_| is the first sequence number to include in + // the current feedback packet. Some older may still be in the map, in case + // a reordering happens and we need to retransmit them. if (!periodic_window_start_seq_) return; @@ -199,15 +195,17 @@ void RemoteEstimatorProxy::SendPeriodicFeedbacks() { } } - for (auto begin_iterator = - packet_arrival_times_.lower_bound(*periodic_window_start_seq_); - begin_iterator != packet_arrival_times_.cend(); - begin_iterator = - packet_arrival_times_.lower_bound(*periodic_window_start_seq_)) { - auto feedback_packet = std::make_unique<rtcp::TransportFeedback>(); - periodic_window_start_seq_ = BuildFeedbackPacket( - feedback_packet_count_++, media_ssrc_, *periodic_window_start_seq_, - begin_iterator, packet_arrival_times_.cend(), feedback_packet.get()); + int64_t packet_arrival_times_end_seq = + packet_arrival_times_.end_sequence_number(); + while (periodic_window_start_seq_ < packet_arrival_times_end_seq) { + auto feedback_packet = MaybeBuildFeedbackPacket( + /*include_timestamps=*/true, periodic_window_start_seq_.value(), + packet_arrival_times_end_seq, + /*is_periodic_update=*/true); + + if (feedback_packet == nullptr) { + break; + } RTC_DCHECK(feedback_sender_ != nullptr); @@ -217,10 +215,10 @@ void RemoteEstimatorProxy::SendPeriodicFeedbacks() { } packets.push_back(std::move(feedback_packet)); - feedback_sender_->SendCombinedRtcpPacket(std::move(packets)); - // Note: Don't erase items from packet_arrival_times_ after sending, in case - // they need to be re-sent after a reordering. Removal will be handled - // by OnPacketArrival once packets are too old. + feedback_sender_(std::move(packets)); + // Note: Don't erase items from packet_arrival_times_ after sending, in + // case they need to be re-sent after a reordering. Removal will be + // handled by OnPacketArrival once packets are too old. } } @@ -231,61 +229,79 @@ void RemoteEstimatorProxy::SendFeedbackOnRequest( return; } - auto feedback_packet = std::make_unique<rtcp::TransportFeedback>( - feedback_request.include_timestamps); - int64_t first_sequence_number = sequence_number - feedback_request.sequence_count + 1; - auto begin_iterator = - packet_arrival_times_.lower_bound(first_sequence_number); - auto end_iterator = packet_arrival_times_.upper_bound(sequence_number); - BuildFeedbackPacket(feedback_packet_count_++, media_ssrc_, - first_sequence_number, begin_iterator, end_iterator, - feedback_packet.get()); + auto feedback_packet = MaybeBuildFeedbackPacket( + feedback_request.include_timestamps, first_sequence_number, + sequence_number + 1, /*is_periodic_update=*/false); + + // This is called when a packet has just been added. + RTC_DCHECK(feedback_packet != nullptr); // Clear up to the first packet that is included in this feedback packet. - packet_arrival_times_.erase(packet_arrival_times_.begin(), begin_iterator); + packet_arrival_times_.EraseTo(first_sequence_number); RTC_DCHECK(feedback_sender_ != nullptr); std::vector<std::unique_ptr<rtcp::RtcpPacket>> packets; packets.push_back(std::move(feedback_packet)); - feedback_sender_->SendCombinedRtcpPacket(std::move(packets)); + feedback_sender_(std::move(packets)); } -int64_t RemoteEstimatorProxy::BuildFeedbackPacket( - uint8_t feedback_packet_count, - uint32_t media_ssrc, - int64_t base_sequence_number, - std::map<int64_t, int64_t>::const_iterator begin_iterator, - std::map<int64_t, int64_t>::const_iterator end_iterator, - rtcp::TransportFeedback* feedback_packet) { - RTC_DCHECK(begin_iterator != end_iterator); - - // TODO(sprang): Measure receive times in microseconds and remove the - // conversions below. - feedback_packet->SetMediaSsrc(media_ssrc); - // Base sequence number is the expected first sequence number. This is known, - // but we might not have actually received it, so the base time shall be the - // time of the first received packet in the feedback. - feedback_packet->SetBase(static_cast<uint16_t>(base_sequence_number & 0xFFFF), - begin_iterator->second * 1000); - feedback_packet->SetFeedbackSequenceNumber(feedback_packet_count); - int64_t next_sequence_number = base_sequence_number; - for (auto it = begin_iterator; it != end_iterator; ++it) { - if (!feedback_packet->AddReceivedPacket( - static_cast<uint16_t>(it->first & 0xFFFF), it->second * 1000)) { - // If we can't even add the first seq to the feedback packet, we won't be - // able to build it at all. - RTC_CHECK(begin_iterator != it); +std::unique_ptr<rtcp::TransportFeedback> +RemoteEstimatorProxy::MaybeBuildFeedbackPacket( + bool include_timestamps, + int64_t begin_sequence_number_inclusive, + int64_t end_sequence_number_exclusive, + bool is_periodic_update) { + RTC_DCHECK_LT(begin_sequence_number_inclusive, end_sequence_number_exclusive); + + int64_t start_seq = + packet_arrival_times_.clamp(begin_sequence_number_inclusive); + + int64_t end_seq = packet_arrival_times_.clamp(end_sequence_number_exclusive); + + // Create the packet on demand, as it's not certain that there are packets + // in the range that have been received. + std::unique_ptr<rtcp::TransportFeedback> feedback_packet = nullptr; + + int64_t next_sequence_number = begin_sequence_number_inclusive; + for (int64_t seq = start_seq; seq < end_seq; ++seq) { + int64_t arrival_time_ms = packet_arrival_times_.get(seq); + if (arrival_time_ms == 0) { + // Packet not received. + continue; + } + + if (feedback_packet == nullptr) { + feedback_packet = + std::make_unique<rtcp::TransportFeedback>(include_timestamps); + // TODO(sprang): Measure receive times in microseconds and remove the + // conversions below. + feedback_packet->SetMediaSsrc(media_ssrc_); + // Base sequence number is the expected first sequence number. This is + // known, but we might not have actually received it, so the base time + // shall be the time of the first received packet in the feedback. + feedback_packet->SetBase( + static_cast<uint16_t>(begin_sequence_number_inclusive & 0xFFFF), + arrival_time_ms * 1000); + feedback_packet->SetFeedbackSequenceNumber(feedback_packet_count_++); + } + + if (!feedback_packet->AddReceivedPacket(static_cast<uint16_t>(seq & 0xFFFF), + arrival_time_ms * 1000)) { // Could not add timestamp, feedback packet might be full. Return and // try again with a fresh packet. break; } - next_sequence_number = it->first + 1; + + next_sequence_number = seq + 1; + } + if (is_periodic_update) { + periodic_window_start_seq_ = next_sequence_number; } - return next_sequence_number; + return feedback_packet; } } // namespace webrtc diff --git a/modules/remote_bitrate_estimator/remote_estimator_proxy.h b/modules/remote_bitrate_estimator/remote_estimator_proxy.h index a4adefc5ee..4f89409995 100644 --- a/modules/remote_bitrate_estimator/remote_estimator_proxy.h +++ b/modules/remote_bitrate_estimator/remote_estimator_proxy.h @@ -11,12 +11,15 @@ #ifndef MODULES_REMOTE_BITRATE_ESTIMATOR_REMOTE_ESTIMATOR_PROXY_H_ #define MODULES_REMOTE_BITRATE_ESTIMATOR_REMOTE_ESTIMATOR_PROXY_H_ -#include <map> +#include <deque> +#include <functional> +#include <memory> #include <vector> #include "api/transport/network_control.h" #include "api/transport/webrtc_key_value_config.h" #include "modules/remote_bitrate_estimator/include/remote_bitrate_estimator.h" +#include "modules/remote_bitrate_estimator/packet_arrival_map.h" #include "rtc_base/experiments/field_trial_parser.h" #include "rtc_base/numerics/sequence_number_util.h" #include "rtc_base/synchronization/mutex.h" @@ -24,7 +27,6 @@ namespace webrtc { class Clock; -class PacketRouter; namespace rtcp { class TransportFeedback; } @@ -32,11 +34,14 @@ class TransportFeedback; // Class used when send-side BWE is enabled: This proxy is instantiated on the // receive side. It buffers a number of receive timestamps and then sends // transport feedback messages back too the send side. - class RemoteEstimatorProxy : public RemoteBitrateEstimator { public: + // Used for sending transport feedback messages when send side + // BWE is used. + using TransportFeedbackSender = std::function<void( + std::vector<std::unique_ptr<rtcp::RtcpPacket>> packets)>; RemoteEstimatorProxy(Clock* clock, - TransportFeedbackSenderInterface* feedback_sender, + TransportFeedbackSender feedback_sender, const WebRtcKeyValueConfig* key_value_config, NetworkStateEstimator* network_state_estimator); ~RemoteEstimatorProxy() override; @@ -71,24 +76,33 @@ class RemoteEstimatorProxy : public RemoteBitrateEstimator { } }; - static const int kMaxNumberOfPackets; - + void MaybeCullOldPackets(int64_t sequence_number, int64_t arrival_time_ms) + RTC_EXCLUSIVE_LOCKS_REQUIRED(&lock_); void SendPeriodicFeedbacks() RTC_EXCLUSIVE_LOCKS_REQUIRED(&lock_); void SendFeedbackOnRequest(int64_t sequence_number, const FeedbackRequest& feedback_request) RTC_EXCLUSIVE_LOCKS_REQUIRED(&lock_); - static int64_t BuildFeedbackPacket( - uint8_t feedback_packet_count, - uint32_t media_ssrc, - int64_t base_sequence_number, - std::map<int64_t, int64_t>::const_iterator - begin_iterator, // |begin_iterator| is inclusive. - std::map<int64_t, int64_t>::const_iterator - end_iterator, // |end_iterator| is exclusive. - rtcp::TransportFeedback* feedback_packet); + + // Returns a Transport Feedback packet with information about as many packets + // that has been received between [`begin_sequence_number_incl`, + // `end_sequence_number_excl`) that can fit in it. If `is_periodic_update`, + // this represents sending a periodic feedback message, which will make it + // update the `periodic_window_start_seq_` variable with the first packet that + // was not included in the feedback packet, so that the next update can + // continue from that sequence number. + // + // If no incoming packets were added, nullptr is returned. + // + // `include_timestamps` decide if the returned TransportFeedback should + // include timestamps. + std::unique_ptr<rtcp::TransportFeedback> MaybeBuildFeedbackPacket( + bool include_timestamps, + int64_t begin_sequence_number_inclusive, + int64_t end_sequence_number_exclusive, + bool is_periodic_update) RTC_EXCLUSIVE_LOCKS_REQUIRED(&lock_); Clock* const clock_; - TransportFeedbackSenderInterface* const feedback_sender_; + const TransportFeedbackSender feedback_sender_; const TransportWideFeedbackConfig send_config_; int64_t last_process_time_ms_; @@ -99,9 +113,14 @@ class RemoteEstimatorProxy : public RemoteBitrateEstimator { uint32_t media_ssrc_ RTC_GUARDED_BY(&lock_); uint8_t feedback_packet_count_ RTC_GUARDED_BY(&lock_); SeqNumUnwrapper<uint16_t> unwrapper_ RTC_GUARDED_BY(&lock_); + + // The next sequence number that should be the start sequence number during + // periodic reporting. Will be absl::nullopt before the first seen packet. absl::optional<int64_t> periodic_window_start_seq_ RTC_GUARDED_BY(&lock_); - // Map unwrapped seq -> time. - std::map<int64_t, int64_t> packet_arrival_times_ RTC_GUARDED_BY(&lock_); + + // Packet arrival times, by sequence number. + PacketArrivalTimeMap packet_arrival_times_ RTC_GUARDED_BY(&lock_); + int64_t send_interval_ms_ RTC_GUARDED_BY(&lock_); bool send_periodic_feedback_ RTC_GUARDED_BY(&lock_); diff --git a/modules/remote_bitrate_estimator/remote_estimator_proxy_unittest.cc b/modules/remote_bitrate_estimator/remote_estimator_proxy_unittest.cc index da995922d9..296724fa71 100644 --- a/modules/remote_bitrate_estimator/remote_estimator_proxy_unittest.cc +++ b/modules/remote_bitrate_estimator/remote_estimator_proxy_unittest.cc @@ -16,8 +16,8 @@ #include "api/transport/field_trial_based_config.h" #include "api/transport/network_types.h" #include "api/transport/test/mock_network_control.h" -#include "modules/pacing/packet_router.h" #include "modules/rtp_rtcp/source/rtcp_packet/transport_feedback.h" +#include "modules/rtp_rtcp/source/rtp_header_extensions.h" #include "system_wrappers/include/clock.h" #include "test/gmock.h" #include "test/gtest.h" @@ -25,6 +25,7 @@ using ::testing::_; using ::testing::ElementsAre; using ::testing::Invoke; +using ::testing::MockFunction; using ::testing::Return; using ::testing::SizeIs; @@ -63,20 +64,12 @@ std::vector<int64_t> TimestampsMs( return timestamps; } -class MockTransportFeedbackSender : public TransportFeedbackSenderInterface { - public: - MOCK_METHOD(bool, - SendCombinedRtcpPacket, - (std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets), - (override)); -}; - class RemoteEstimatorProxyTest : public ::testing::Test { public: RemoteEstimatorProxyTest() : clock_(0), proxy_(&clock_, - &router_, + feedback_sender_.AsStdFunction(), &field_trial_config_, &network_state_estimator_) {} @@ -113,7 +106,8 @@ class RemoteEstimatorProxyTest : public ::testing::Test { FieldTrialBasedConfig field_trial_config_; SimulatedClock clock_; - ::testing::StrictMock<MockTransportFeedbackSender> router_; + MockFunction<void(std::vector<std::unique_ptr<rtcp::RtcpPacket>>)> + feedback_sender_; ::testing::NiceMock<MockNetworkStateEstimator> network_state_estimator_; RemoteEstimatorProxy proxy_; }; @@ -121,7 +115,7 @@ class RemoteEstimatorProxyTest : public ::testing::Test { TEST_F(RemoteEstimatorProxyTest, SendsSinglePacketFeedback) { IncomingPacket(kBaseSeq, kBaseTimeMs); - EXPECT_CALL(router_, SendCombinedRtcpPacket) + EXPECT_CALL(feedback_sender_, Call) .WillOnce(Invoke( [](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { rtcp::TransportFeedback* feedback_packet = @@ -134,7 +128,6 @@ TEST_F(RemoteEstimatorProxyTest, SendsSinglePacketFeedback) { ElementsAre(kBaseSeq)); EXPECT_THAT(TimestampsMs(*feedback_packet), ElementsAre(kBaseTimeMs)); - return true; })); Process(); @@ -144,7 +137,7 @@ TEST_F(RemoteEstimatorProxyTest, DuplicatedPackets) { IncomingPacket(kBaseSeq, kBaseTimeMs); IncomingPacket(kBaseSeq, kBaseTimeMs + 1000); - EXPECT_CALL(router_, SendCombinedRtcpPacket) + EXPECT_CALL(feedback_sender_, Call) .WillOnce(Invoke( [](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { rtcp::TransportFeedback* feedback_packet = @@ -167,13 +160,13 @@ TEST_F(RemoteEstimatorProxyTest, FeedbackWithMissingStart) { // First feedback. IncomingPacket(kBaseSeq, kBaseTimeMs); IncomingPacket(kBaseSeq + 1, kBaseTimeMs + 1000); - EXPECT_CALL(router_, SendCombinedRtcpPacket).WillOnce(Return(true)); + EXPECT_CALL(feedback_sender_, Call); Process(); // Second feedback starts with a missing packet (DROP kBaseSeq + 2). IncomingPacket(kBaseSeq + 3, kBaseTimeMs + 3000); - EXPECT_CALL(router_, SendCombinedRtcpPacket) + EXPECT_CALL(feedback_sender_, Call) .WillOnce(Invoke( [](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { rtcp::TransportFeedback* feedback_packet = @@ -186,7 +179,6 @@ TEST_F(RemoteEstimatorProxyTest, FeedbackWithMissingStart) { ElementsAre(kBaseSeq + 3)); EXPECT_THAT(TimestampsMs(*feedback_packet), ElementsAre(kBaseTimeMs + 3000)); - return true; })); Process(); @@ -197,7 +189,7 @@ TEST_F(RemoteEstimatorProxyTest, SendsFeedbackWithVaryingDeltas) { IncomingPacket(kBaseSeq + 1, kBaseTimeMs + kMaxSmallDeltaMs); IncomingPacket(kBaseSeq + 2, kBaseTimeMs + (2 * kMaxSmallDeltaMs) + 1); - EXPECT_CALL(router_, SendCombinedRtcpPacket) + EXPECT_CALL(feedback_sender_, Call) .WillOnce(Invoke( [](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { rtcp::TransportFeedback* feedback_packet = @@ -211,7 +203,6 @@ TEST_F(RemoteEstimatorProxyTest, SendsFeedbackWithVaryingDeltas) { EXPECT_THAT(TimestampsMs(*feedback_packet), ElementsAre(kBaseTimeMs, kBaseTimeMs + kMaxSmallDeltaMs, kBaseTimeMs + (2 * kMaxSmallDeltaMs) + 1)); - return true; })); Process(); @@ -224,7 +215,7 @@ TEST_F(RemoteEstimatorProxyTest, SendsFragmentedFeedback) { IncomingPacket(kBaseSeq, kBaseTimeMs); IncomingPacket(kBaseSeq + 1, kBaseTimeMs + kTooLargeDelta); - EXPECT_CALL(router_, SendCombinedRtcpPacket) + EXPECT_CALL(feedback_sender_, Call) .WillOnce(Invoke( [](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { rtcp::TransportFeedback* feedback_packet = @@ -237,7 +228,6 @@ TEST_F(RemoteEstimatorProxyTest, SendsFragmentedFeedback) { ElementsAre(kBaseSeq)); EXPECT_THAT(TimestampsMs(*feedback_packet), ElementsAre(kBaseTimeMs)); - return true; })) .WillOnce(Invoke( [](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { @@ -251,7 +241,6 @@ TEST_F(RemoteEstimatorProxyTest, SendsFragmentedFeedback) { ElementsAre(kBaseSeq + 1)); EXPECT_THAT(TimestampsMs(*feedback_packet), ElementsAre(kBaseTimeMs + kTooLargeDelta)); - return true; })); Process(); @@ -263,7 +252,7 @@ TEST_F(RemoteEstimatorProxyTest, HandlesReorderingAndWrap) { IncomingPacket(kBaseSeq, kBaseTimeMs); IncomingPacket(kLargeSeq, kBaseTimeMs + kDeltaMs); - EXPECT_CALL(router_, SendCombinedRtcpPacket) + EXPECT_CALL(feedback_sender_, Call) .WillOnce(Invoke( [&](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { rtcp::TransportFeedback* feedback_packet = @@ -274,7 +263,6 @@ TEST_F(RemoteEstimatorProxyTest, HandlesReorderingAndWrap) { EXPECT_THAT(TimestampsMs(*feedback_packet), ElementsAre(kBaseTimeMs + kDeltaMs, kBaseTimeMs)); - return true; })); Process(); @@ -293,7 +281,7 @@ TEST_F(RemoteEstimatorProxyTest, HandlesMalformedSequenceNumbers) { } // Only expect feedback for the last two packets. - EXPECT_CALL(router_, SendCombinedRtcpPacket) + EXPECT_CALL(feedback_sender_, Call) .WillOnce(Invoke( [&](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { rtcp::TransportFeedback* feedback_packet = @@ -306,7 +294,6 @@ TEST_F(RemoteEstimatorProxyTest, HandlesMalformedSequenceNumbers) { EXPECT_THAT(TimestampsMs(*feedback_packet), ElementsAre(kBaseTimeMs + 28 * kDeltaMs, kBaseTimeMs + 29 * kDeltaMs)); - return true; })); Process(); @@ -324,7 +311,7 @@ TEST_F(RemoteEstimatorProxyTest, HandlesBackwardsWrappingSequenceNumbers) { } // Only expect feedback for the first two packets. - EXPECT_CALL(router_, SendCombinedRtcpPacket) + EXPECT_CALL(feedback_sender_, Call) .WillOnce(Invoke( [&](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { rtcp::TransportFeedback* feedback_packet = @@ -336,7 +323,6 @@ TEST_F(RemoteEstimatorProxyTest, HandlesBackwardsWrappingSequenceNumbers) { ElementsAre(kBaseSeq + 40000, kBaseSeq)); EXPECT_THAT(TimestampsMs(*feedback_packet), ElementsAre(kBaseTimeMs + kDeltaMs, kBaseTimeMs)); - return true; })); Process(); @@ -346,7 +332,7 @@ TEST_F(RemoteEstimatorProxyTest, ResendsTimestampsOnReordering) { IncomingPacket(kBaseSeq, kBaseTimeMs); IncomingPacket(kBaseSeq + 2, kBaseTimeMs + 2); - EXPECT_CALL(router_, SendCombinedRtcpPacket) + EXPECT_CALL(feedback_sender_, Call) .WillOnce(Invoke( [](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { rtcp::TransportFeedback* feedback_packet = @@ -359,14 +345,13 @@ TEST_F(RemoteEstimatorProxyTest, ResendsTimestampsOnReordering) { ElementsAre(kBaseSeq, kBaseSeq + 2)); EXPECT_THAT(TimestampsMs(*feedback_packet), ElementsAre(kBaseTimeMs, kBaseTimeMs + 2)); - return true; })); Process(); IncomingPacket(kBaseSeq + 1, kBaseTimeMs + 1); - EXPECT_CALL(router_, SendCombinedRtcpPacket) + EXPECT_CALL(feedback_sender_, Call) .WillOnce(Invoke( [](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { rtcp::TransportFeedback* feedback_packet = @@ -379,7 +364,6 @@ TEST_F(RemoteEstimatorProxyTest, ResendsTimestampsOnReordering) { ElementsAre(kBaseSeq + 1, kBaseSeq + 2)); EXPECT_THAT(TimestampsMs(*feedback_packet), ElementsAre(kBaseTimeMs + 1, kBaseTimeMs + 2)); - return true; })); Process(); @@ -390,7 +374,7 @@ TEST_F(RemoteEstimatorProxyTest, RemovesTimestampsOutOfScope) { IncomingPacket(kBaseSeq + 2, kBaseTimeMs); - EXPECT_CALL(router_, SendCombinedRtcpPacket) + EXPECT_CALL(feedback_sender_, Call) .WillOnce(Invoke( [](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { rtcp::TransportFeedback* feedback_packet = @@ -400,14 +384,13 @@ TEST_F(RemoteEstimatorProxyTest, RemovesTimestampsOutOfScope) { EXPECT_THAT(TimestampsMs(*feedback_packet), ElementsAre(kBaseTimeMs)); - return true; })); Process(); IncomingPacket(kBaseSeq + 3, kTimeoutTimeMs); // kBaseSeq + 2 times out here. - EXPECT_CALL(router_, SendCombinedRtcpPacket) + EXPECT_CALL(feedback_sender_, Call) .WillOnce(Invoke( [&](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { rtcp::TransportFeedback* feedback_packet = @@ -417,7 +400,6 @@ TEST_F(RemoteEstimatorProxyTest, RemovesTimestampsOutOfScope) { EXPECT_THAT(TimestampsMs(*feedback_packet), ElementsAre(kTimeoutTimeMs)); - return true; })); Process(); @@ -427,7 +409,7 @@ TEST_F(RemoteEstimatorProxyTest, RemovesTimestampsOutOfScope) { IncomingPacket(kBaseSeq, kBaseTimeMs - 1); IncomingPacket(kBaseSeq + 1, kTimeoutTimeMs - 1); - EXPECT_CALL(router_, SendCombinedRtcpPacket) + EXPECT_CALL(feedback_sender_, Call) .WillOnce(Invoke( [&](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { rtcp::TransportFeedback* feedback_packet = @@ -440,7 +422,6 @@ TEST_F(RemoteEstimatorProxyTest, RemovesTimestampsOutOfScope) { EXPECT_THAT(TimestampsMs(*feedback_packet), ElementsAre(kBaseTimeMs - 1, kTimeoutTimeMs - 1, kTimeoutTimeMs)); - return true; })); Process(); @@ -496,7 +477,7 @@ TEST_F(RemoteEstimatorProxyOnRequestTest, TimeUntilNextProcessIsHigh) { TEST_F(RemoteEstimatorProxyOnRequestTest, ProcessDoesNotSendFeedback) { proxy_.SetSendPeriodicFeedback(false); IncomingPacket(kBaseSeq, kBaseTimeMs); - EXPECT_CALL(router_, SendCombinedRtcpPacket).Times(0); + EXPECT_CALL(feedback_sender_, Call).Times(0); Process(); } @@ -506,7 +487,7 @@ TEST_F(RemoteEstimatorProxyOnRequestTest, RequestSinglePacketFeedback) { IncomingPacket(kBaseSeq + 1, kBaseTimeMs + kMaxSmallDeltaMs); IncomingPacket(kBaseSeq + 2, kBaseTimeMs + 2 * kMaxSmallDeltaMs); - EXPECT_CALL(router_, SendCombinedRtcpPacket) + EXPECT_CALL(feedback_sender_, Call) .WillOnce(Invoke( [](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { rtcp::TransportFeedback* feedback_packet = @@ -519,7 +500,6 @@ TEST_F(RemoteEstimatorProxyOnRequestTest, RequestSinglePacketFeedback) { ElementsAre(kBaseSeq + 3)); EXPECT_THAT(TimestampsMs(*feedback_packet), ElementsAre(kBaseTimeMs + 3 * kMaxSmallDeltaMs)); - return true; })); constexpr FeedbackRequest kSinglePacketFeedbackRequest = { @@ -535,7 +515,7 @@ TEST_F(RemoteEstimatorProxyOnRequestTest, RequestLastFivePacketFeedback) { IncomingPacket(kBaseSeq + i, kBaseTimeMs + i * kMaxSmallDeltaMs); } - EXPECT_CALL(router_, SendCombinedRtcpPacket) + EXPECT_CALL(feedback_sender_, Call) .WillOnce(Invoke( [](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { rtcp::TransportFeedback* feedback_packet = @@ -553,7 +533,6 @@ TEST_F(RemoteEstimatorProxyOnRequestTest, RequestLastFivePacketFeedback) { kBaseTimeMs + 8 * kMaxSmallDeltaMs, kBaseTimeMs + 9 * kMaxSmallDeltaMs, kBaseTimeMs + 10 * kMaxSmallDeltaMs)); - return true; })); constexpr FeedbackRequest kFivePacketsFeedbackRequest = { @@ -571,7 +550,7 @@ TEST_F(RemoteEstimatorProxyOnRequestTest, IncomingPacket(kBaseSeq + i, kBaseTimeMs + i * kMaxSmallDeltaMs); } - EXPECT_CALL(router_, SendCombinedRtcpPacket) + EXPECT_CALL(feedback_sender_, Call) .WillOnce(Invoke( [](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { rtcp::TransportFeedback* feedback_packet = @@ -586,7 +565,6 @@ TEST_F(RemoteEstimatorProxyOnRequestTest, ElementsAre(kBaseTimeMs + 6 * kMaxSmallDeltaMs, kBaseTimeMs + 8 * kMaxSmallDeltaMs, kBaseTimeMs + 10 * kMaxSmallDeltaMs)); - return true; })); constexpr FeedbackRequest kFivePacketsFeedbackRequest = { @@ -658,13 +636,7 @@ TEST_F(RemoteEstimatorProxyTest, SendTransportFeedbackAndNetworkStateUpdate) { AbsoluteSendTime::MsTo24Bits(kBaseTimeMs - 1))); EXPECT_CALL(network_state_estimator_, GetCurrentEstimate()) .WillOnce(Return(NetworkStateEstimate())); - EXPECT_CALL(router_, SendCombinedRtcpPacket) - .WillOnce( - [](std::vector<std::unique_ptr<rtcp::RtcpPacket>> feedback_packets) { - EXPECT_THAT(feedback_packets, SizeIs(2)); - return true; - }); - + EXPECT_CALL(feedback_sender_, Call(SizeIs(2))); Process(); } diff --git a/modules/remote_bitrate_estimator/tools/bwe_rtp.cc b/modules/remote_bitrate_estimator/tools/bwe_rtp.cc index c0b3a37ba5..403f81fd03 100644 --- a/modules/remote_bitrate_estimator/tools/bwe_rtp.cc +++ b/modules/remote_bitrate_estimator/tools/bwe_rtp.cc @@ -18,10 +18,8 @@ #include "absl/flags/flag.h" #include "absl/flags/parse.h" -#include "modules/remote_bitrate_estimator/remote_bitrate_estimator_abs_send_time.h" -#include "modules/remote_bitrate_estimator/remote_bitrate_estimator_single_stream.h" +#include "modules/rtp_rtcp/include/rtp_header_extension_map.h" #include "test/rtp_file_reader.h" -#include "test/rtp_header_parser.h" ABSL_FLAG(std::string, extension_type, @@ -65,14 +63,11 @@ std::set<uint32_t> SsrcFilter() { return ssrcs; } -std::unique_ptr<webrtc::RtpHeaderParser> ParseArgsAndSetupEstimator( +bool ParseArgsAndSetupRtpReader( int argc, char** argv, - webrtc::Clock* clock, - webrtc::RemoteBitrateObserver* observer, - std::unique_ptr<webrtc::test::RtpFileReader>* rtp_reader, - std::unique_ptr<webrtc::RemoteBitrateEstimator>* estimator, - std::string* estimator_used) { + std::unique_ptr<webrtc::test::RtpFileReader>& rtp_reader, + webrtc::RtpHeaderExtensionMap& rtp_header_extensions) { absl::ParseCommandLine(argc, argv); std::string filename = InputFile(); @@ -84,16 +79,16 @@ std::unique_ptr<webrtc::RtpHeaderParser> ParseArgsAndSetupEstimator( fprintf(stderr, "\n"); if (filename.substr(filename.find_last_of('.')) == ".pcap") { fprintf(stderr, "Opening as pcap\n"); - rtp_reader->reset(webrtc::test::RtpFileReader::Create( + rtp_reader.reset(webrtc::test::RtpFileReader::Create( webrtc::test::RtpFileReader::kPcap, filename.c_str(), SsrcFilter())); } else { fprintf(stderr, "Opening as rtp\n"); - rtp_reader->reset(webrtc::test::RtpFileReader::Create( + rtp_reader.reset(webrtc::test::RtpFileReader::Create( webrtc::test::RtpFileReader::kRtpDump, filename.c_str())); } - if (!*rtp_reader) { + if (!rtp_reader) { fprintf(stderr, "Cannot open input file %s\n", filename.c_str()); - return nullptr; + return false; } fprintf(stderr, "Input file: %s\n\n", filename.c_str()); @@ -105,31 +100,10 @@ std::unique_ptr<webrtc::RtpHeaderParser> ParseArgsAndSetupEstimator( fprintf(stderr, "Extension: abs\n"); } else { fprintf(stderr, "Unknown extension type\n"); - return nullptr; + return false; } - // Setup the RTP header parser and the bitrate estimator. - auto parser = webrtc::RtpHeaderParser::CreateForTest(); - parser->RegisterRtpHeaderExtension(extension, ExtensionId()); - if (estimator) { - switch (extension) { - case webrtc::kRtpExtensionAbsoluteSendTime: { - estimator->reset( - new webrtc::RemoteBitrateEstimatorAbsSendTime(observer, clock)); - *estimator_used = "AbsoluteSendTimeRemoteBitrateEstimator"; - break; - } - case webrtc::kRtpExtensionTransmissionTimeOffset: { - estimator->reset( - new webrtc::RemoteBitrateEstimatorSingleStream(observer, clock)); - *estimator_used = "RemoteBitrateEstimator"; - break; - } - default: - assert(false); - return nullptr; - } - } + rtp_header_extensions.RegisterByType(ExtensionId(), extension); - return parser; + return true; } diff --git a/modules/remote_bitrate_estimator/tools/bwe_rtp.h b/modules/remote_bitrate_estimator/tools/bwe_rtp.h index 4285f926b5..3b161db37b 100644 --- a/modules/remote_bitrate_estimator/tools/bwe_rtp.h +++ b/modules/remote_bitrate_estimator/tools/bwe_rtp.h @@ -12,25 +12,14 @@ #define MODULES_REMOTE_BITRATE_ESTIMATOR_TOOLS_BWE_RTP_H_ #include <memory> -#include <string> -namespace webrtc { -class Clock; -class RemoteBitrateEstimator; -class RemoteBitrateObserver; -class RtpHeaderParser; -namespace test { -class RtpFileReader; -} -} // namespace webrtc +#include "modules/rtp_rtcp/include/rtp_header_extension_map.h" +#include "test/rtp_file_reader.h" -std::unique_ptr<webrtc::RtpHeaderParser> ParseArgsAndSetupEstimator( +bool ParseArgsAndSetupRtpReader( int argc, char** argv, - webrtc::Clock* clock, - webrtc::RemoteBitrateObserver* observer, - std::unique_ptr<webrtc::test::RtpFileReader>* rtp_reader, - std::unique_ptr<webrtc::RemoteBitrateEstimator>* estimator, - std::string* estimator_used); + std::unique_ptr<webrtc::test::RtpFileReader>& rtp_reader, + webrtc::RtpHeaderExtensionMap& rtp_header_extensions); #endif // MODULES_REMOTE_BITRATE_ESTIMATOR_TOOLS_BWE_RTP_H_ diff --git a/modules/remote_bitrate_estimator/tools/rtp_to_text.cc b/modules/remote_bitrate_estimator/tools/rtp_to_text.cc index 7f1e009793..98f502a42e 100644 --- a/modules/remote_bitrate_estimator/tools/rtp_to_text.cc +++ b/modules/remote_bitrate_estimator/tools/rtp_to_text.cc @@ -13,17 +13,19 @@ #include <memory> #include "modules/remote_bitrate_estimator/tools/bwe_rtp.h" +#include "modules/rtp_rtcp/include/rtp_header_extension_map.h" +#include "modules/rtp_rtcp/source/rtp_header_extensions.h" +#include "modules/rtp_rtcp/source/rtp_packet.h" #include "rtc_base/format_macros.h" #include "rtc_base/strings/string_builder.h" #include "test/rtp_file_reader.h" -#include "test/rtp_header_parser.h" int main(int argc, char* argv[]) { std::unique_ptr<webrtc::test::RtpFileReader> reader; - std::unique_ptr<webrtc::RtpHeaderParser> parser(ParseArgsAndSetupEstimator( - argc, argv, nullptr, nullptr, &reader, nullptr, nullptr)); - if (!parser) + webrtc::RtpHeaderExtensionMap rtp_header_extensions; + if (!ParseArgsAndSetupRtpReader(argc, argv, reader, rtp_header_extensions)) { return -1; + } bool arrival_time_only = (argc >= 5 && strncmp(argv[4], "-t", 2) == 0); @@ -35,11 +37,15 @@ int main(int argc, char* argv[]) { int non_zero_ts_offsets = 0; webrtc::test::RtpPacket packet; while (reader->NextPacket(&packet)) { - webrtc::RTPHeader header; - parser->Parse(packet.data, packet.length, &header); - if (header.extension.absoluteSendTime != 0) + webrtc::RtpPacket header(&rtp_header_extensions); + header.Parse(packet.data, packet.length); + uint32_t abs_send_time = 0; + if (header.GetExtension<webrtc::AbsoluteSendTime>(&abs_send_time) && + abs_send_time != 0) ++non_zero_abs_send_time; - if (header.extension.transmissionTimeOffset != 0) + int32_t toffset = 0; + if (header.GetExtension<webrtc::TransmissionOffset>(&toffset) && + toffset != 0) ++non_zero_ts_offsets; if (arrival_time_only) { rtc::StringBuilder ss; @@ -47,11 +53,9 @@ int main(int argc, char* argv[]) { fprintf(stdout, "%s\n", ss.str().c_str()); } else { fprintf(stdout, "%u %u %d %u %u %d %u %" RTC_PRIuS " %" RTC_PRIuS "\n", - header.sequenceNumber, header.timestamp, - header.extension.transmissionTimeOffset, - header.extension.absoluteSendTime, packet.time_ms, - header.markerBit, header.ssrc, packet.length, - packet.original_length); + header.SequenceNumber(), header.Timestamp(), toffset, + abs_send_time, packet.time_ms, header.Marker(), header.Ssrc(), + packet.length, packet.original_length); } ++packet_counter; } |