Fix network emulation behavior when changing bandwidth.
Calculate packet exit times "just in time" rather than at send time.
This allows changing bandwidth with packets in the queue being reflected
correctly.
Bug: webrtc:10265
Change-Id: I5a38663def4d2bfee64164f9ae62bc61277064bb
Reviewed-on: https://webrtc-review.googlesource.com/c/120403
Commit-Queue: Christoffer Rodbro <crodbro@webrtc.org>
Reviewed-by: Sebastian Jansson <srte@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#26487}diff --git a/call/simulated_network.cc b/call/simulated_network.cc
index 01a74e5..92e945b 100644
--- a/call/simulated_network.cc
+++ b/call/simulated_network.cc
@@ -31,9 +31,6 @@
void SimulatedNetwork::SetConfig(const SimulatedNetwork::Config& config) {
rtc::CritScope crit(&config_lock_);
- if (config_.link_capacity_kbps != config.link_capacity_kbps) {
- reset_capacity_delay_error_ = true;
- }
config_ = config; // Shallow copy of the struct.
double prob_loss = config.loss_percent / 100.0;
if (config_.avg_burst_loss_length == -1) {
@@ -66,6 +63,9 @@
rtc::CritScope crit(&config_lock_);
config = config_;
}
+
+ UpdateCapacityQueue(packet.send_time_us);
+
packet.size += config.packet_overhead;
rtc::CritScope crit(&process_lock_);
if (config.queue_length_packets > 0 &&
@@ -73,40 +73,12 @@
// Too many packet on the link, drop this one.
return false;
}
- int64_t network_start_time_us = packet.send_time_us;
- {
- rtc::CritScope crit(&config_lock_);
- if (reset_capacity_delay_error_) {
- capacity_delay_error_bytes_ = 0;
- reset_capacity_delay_error_ = false;
- }
- if (pause_transmission_until_us_) {
- network_start_time_us =
- std::max(network_start_time_us, *pause_transmission_until_us_);
- pause_transmission_until_us_.reset();
- }
- }
- // Delay introduced by the link capacity.
- TimeDelta capacity_delay = TimeDelta::Zero();
- if (config.link_capacity_kbps > 0) {
- const DataRate link_capacity = DataRate::kbps(config.link_capacity_kbps);
- int64_t compensated_size =
- static_cast<int64_t>(packet.size) + capacity_delay_error_bytes_;
- capacity_delay = DataSize::bytes(compensated_size) / link_capacity;
+ // Set arrival time = send time for now; actual arrival time will be
+ // calculated in UpdateCapacityQueue.
+ queue_size_bytes_ += packet.size;
+ capacity_link_.push({packet, packet.send_time_us});
- capacity_delay_error_bytes_ +=
- packet.size - (capacity_delay * link_capacity).bytes();
- }
-
- // Check if there already are packets on the link and change network start
- // time forward if there is.
- if (!capacity_link_.empty() &&
- network_start_time_us < capacity_link_.back().arrival_time_us)
- network_start_time_us = capacity_link_.back().arrival_time_us;
-
- int64_t arrival_time_us = network_start_time_us + capacity_delay.us();
- capacity_link_.push({packet, arrival_time_us});
return true;
}
@@ -116,82 +88,128 @@
return delay_link_.begin()->arrival_time_us;
return absl::nullopt;
}
-std::vector<PacketDeliveryInfo> SimulatedNetwork::DequeueDeliverablePackets(
- int64_t receive_time_us) {
- int64_t time_now_us = receive_time_us;
+
+void SimulatedNetwork::UpdateCapacityQueue(int64_t time_now_us) {
Config config;
double prob_loss_bursting;
double prob_start_bursting;
+ int64_t pause_transmission_until_us;
{
rtc::CritScope crit(&config_lock_);
config = config_;
prob_loss_bursting = prob_loss_bursting_;
prob_start_bursting = prob_start_bursting_;
+ pause_transmission_until_us = pause_transmission_until_us_.value_or(0);
}
{
rtc::CritScope crit(&process_lock_);
+ bool needs_sort = false;
+
+ // Catch for thread races.
+ if (time_now_us < last_capacity_link_visit_us_.value_or(time_now_us))
+ return;
+
+ int64_t time_us = last_capacity_link_visit_us_.value_or(time_now_us);
// Check the capacity link first.
- if (!capacity_link_.empty()) {
- int64_t last_arrival_time_us =
- delay_link_.empty() ? -1 : delay_link_.back().arrival_time_us;
- bool needs_sort = false;
- while (!capacity_link_.empty() &&
- time_now_us >= capacity_link_.front().arrival_time_us) {
- // Time to get this packet.
- PacketInfo packet = std::move(capacity_link_.front());
- capacity_link_.pop();
+ while (!capacity_link_.empty()) {
+ int64_t time_until_front_exits_us = 0;
+ if (config.link_capacity_kbps > 0) {
+ int64_t remaining_bits =
+ capacity_link_.front().packet.size * 8 - pending_drain_bits_;
+ RTC_DCHECK(remaining_bits > 0);
+ // Division rounded up - packet not delivered until its last bit is.
+ time_until_front_exits_us =
+ (1000 * remaining_bits + config.link_capacity_kbps - 1) /
+ config.link_capacity_kbps;
+ }
- // Drop packets at an average rate of |config_.loss_percent| with
- // and average loss burst length of |config_.avg_burst_loss_length|.
- if ((bursting_ && random_.Rand<double>() < prob_loss_bursting) ||
- (!bursting_ && random_.Rand<double>() < prob_start_bursting)) {
- bursting_ = true;
- packet.arrival_time_us = PacketDeliveryInfo::kNotReceived;
- } else {
- bursting_ = false;
- int64_t arrival_time_jitter_us = std::max(
- random_.Gaussian(config.queue_delay_ms * 1000,
- config.delay_standard_deviation_ms * 1000),
- 0.0);
+ if (time_us + time_until_front_exits_us > time_now_us) {
+ // Packet at front will not exit yet. Will not enter here on infinite
+ // capacity(=0) so no special handling needed.
+ pending_drain_bits_ +=
+ ((time_now_us - time_us) * config.link_capacity_kbps) / 1000;
+ break;
+ }
+ if (config.link_capacity_kbps > 0) {
+ pending_drain_bits_ +=
+ (time_until_front_exits_us * config.link_capacity_kbps) / 1000;
+ } else {
+ // Enough to drain the whole queue.
+ pending_drain_bits_ = queue_size_bytes_ * 8;
+ }
- // If reordering is not allowed then adjust arrival_time_jitter
- // to make sure all packets are sent in order.
- if (!config.allow_reordering && !delay_link_.empty() &&
- packet.arrival_time_us + arrival_time_jitter_us <
- last_arrival_time_us) {
- arrival_time_jitter_us =
- last_arrival_time_us - packet.arrival_time_us;
- }
- packet.arrival_time_us += arrival_time_jitter_us;
- if (packet.arrival_time_us >= last_arrival_time_us) {
- last_arrival_time_us = packet.arrival_time_us;
- } else {
- needs_sort = true;
- }
+ // Time to get this packet.
+ PacketInfo packet = std::move(capacity_link_.front());
+ capacity_link_.pop();
+
+ time_us += time_until_front_exits_us;
+ RTC_DCHECK(time_us >= packet.packet.send_time_us);
+ packet.arrival_time_us = std::max(pause_transmission_until_us, time_us);
+ queue_size_bytes_ -= packet.packet.size;
+ pending_drain_bits_ -= packet.packet.size * 8;
+ RTC_DCHECK(pending_drain_bits_ >= 0);
+
+ // Drop packets at an average rate of |config_.loss_percent| with
+ // and average loss burst length of |config_.avg_burst_loss_length|.
+ if ((bursting_ && random_.Rand<double>() < prob_loss_bursting) ||
+ (!bursting_ && random_.Rand<double>() < prob_start_bursting)) {
+ bursting_ = true;
+ packet.arrival_time_us = PacketDeliveryInfo::kNotReceived;
+ } else {
+ bursting_ = false;
+ int64_t arrival_time_jitter_us = std::max(
+ random_.Gaussian(config.queue_delay_ms * 1000,
+ config.delay_standard_deviation_ms * 1000),
+ 0.0);
+
+ // If reordering is not allowed then adjust arrival_time_jitter
+ // to make sure all packets are sent in order.
+ int64_t last_arrival_time_us =
+ delay_link_.empty() ? -1 : delay_link_.back().arrival_time_us;
+ if (!config.allow_reordering && !delay_link_.empty() &&
+ packet.arrival_time_us + arrival_time_jitter_us <
+ last_arrival_time_us) {
+ arrival_time_jitter_us =
+ last_arrival_time_us - packet.arrival_time_us;
}
- delay_link_.emplace_back(std::move(packet));
+ packet.arrival_time_us += arrival_time_jitter_us;
+ if (packet.arrival_time_us >= last_arrival_time_us) {
+ last_arrival_time_us = packet.arrival_time_us;
+ } else {
+ needs_sort = true;
+ }
}
-
- if (needs_sort) {
- // Packet(s) arrived out of order, make sure list is sorted.
- std::sort(delay_link_.begin(), delay_link_.end(),
- [](const PacketInfo& p1, const PacketInfo& p2) {
- return p1.arrival_time_us < p2.arrival_time_us;
- });
- }
+ delay_link_.emplace_back(std::move(packet));
}
+ last_capacity_link_visit_us_ = time_now_us;
+ // Cannot save unused capacity for later.
+ pending_drain_bits_ = std::min(pending_drain_bits_, queue_size_bytes_ * 8);
- std::vector<PacketDeliveryInfo> packets_to_deliver;
- // Check the extra delay queue.
- while (!delay_link_.empty() &&
- time_now_us >= delay_link_.front().arrival_time_us) {
- PacketInfo packet_info = delay_link_.front();
- packets_to_deliver.emplace_back(
- PacketDeliveryInfo(packet_info.packet, packet_info.arrival_time_us));
- delay_link_.pop_front();
+ if (needs_sort) {
+ // Packet(s) arrived out of order, make sure list is sorted.
+ std::sort(delay_link_.begin(), delay_link_.end(),
+ [](const PacketInfo& p1, const PacketInfo& p2) {
+ return p1.arrival_time_us < p2.arrival_time_us;
+ });
}
- return packets_to_deliver;
}
}
+std::vector<PacketDeliveryInfo> SimulatedNetwork::DequeueDeliverablePackets(
+ int64_t receive_time_us) {
+ UpdateCapacityQueue(receive_time_us);
+
+ rtc::CritScope crit(&process_lock_);
+ std::vector<PacketDeliveryInfo> packets_to_deliver;
+ // Check the extra delay queue.
+ while (!delay_link_.empty() &&
+ receive_time_us >= delay_link_.front().arrival_time_us) {
+ PacketInfo packet_info = delay_link_.front();
+ packets_to_deliver.emplace_back(
+ PacketDeliveryInfo(packet_info.packet, packet_info.arrival_time_us));
+ delay_link_.pop_front();
+ }
+ return packets_to_deliver;
+}
+
} // namespace webrtc
diff --git a/call/simulated_network.h b/call/simulated_network.h
index 0336bfc..4085600 100644
--- a/call/simulated_network.h
+++ b/call/simulated_network.h
@@ -48,8 +48,11 @@
PacketInFlightInfo packet;
int64_t arrival_time_us;
};
+
+ // Moves packets from capacity- to delay link.
+ void UpdateCapacityQueue(int64_t time_now_us);
+
rtc::CriticalSection config_lock_;
- bool reset_capacity_delay_error_ RTC_GUARDED_BY(config_lock_) = false;
// |process_lock| guards the data structures involved in delay and loss
// processes, such as the packet queues.
@@ -73,7 +76,11 @@
// The probability to drop a burst of packets.
double prob_start_bursting_ RTC_GUARDED_BY(config_lock_);
- int64_t capacity_delay_error_bytes_ = 0;
+
+ int64_t queue_size_bytes_ RTC_GUARDED_BY(process_lock_) = 0;
+ int64_t pending_drain_bits_ RTC_GUARDED_BY(process_lock_) = 0;
+ absl::optional<int64_t> last_capacity_link_visit_us_
+ RTC_GUARDED_BY(process_lock_);
};
} // namespace webrtc
diff --git a/call/test/fake_network_pipe_unittest.cc b/call/test/fake_network_pipe_unittest.cc
index 491b8fe..9f2a663 100644
--- a/call/test/fake_network_pipe_unittest.cc
+++ b/call/test/fake_network_pipe_unittest.cc
@@ -276,45 +276,37 @@
std::unique_ptr<FakeNetworkPipe> pipe(
new FakeNetworkPipe(&fake_clock_, std::move(network), &receiver));
- // Add 10 packets of 1000 bytes, = 80 kb.
- const int kNumPackets = 10;
+ // Add 20 packets of 1000 bytes, = 80 kb.
+ const int kNumPackets = 20;
const int kPacketSize = 1000;
SendPackets(pipe.get(), kNumPackets, kPacketSize);
- // Time to get one packet through the link at the initial speed.
- int packet_time_1_ms = PacketTimeMs(config.link_capacity_kbps, kPacketSize);
-
- // Change the capacity.
- config.link_capacity_kbps *= 2; // Double the capacity.
- simulated_network->SetConfig(config);
-
- // Add another 10 packets of 1000 bytes, = 80 kb, and verify it takes two
- // seconds to get them through the pipe.
- SendPackets(pipe.get(), kNumPackets, kPacketSize);
-
- // Time to get one packet through the link at the new capacity.
- int packet_time_2_ms = PacketTimeMs(config.link_capacity_kbps, kPacketSize);
-
// Time hasn't increased yet, so we souldn't get any packets.
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(0);
pipe->Process();
- // Advance time in steps to release one packet at a time.
- for (int i = 0; i < kNumPackets; ++i) {
- fake_clock_.AdvanceTimeMilliseconds(packet_time_1_ms);
+ // Advance time in steps to release half of the packets one at a time.
+ int step_ms = PacketTimeMs(config.link_capacity_kbps, kPacketSize);
+ for (int i = 0; i < kNumPackets / 2; ++i) {
+ fake_clock_.AdvanceTimeMilliseconds(step_ms);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(1);
pipe->Process();
}
- // Advance time in steps to release one packet at a time.
- for (int i = 0; i < kNumPackets; ++i) {
- fake_clock_.AdvanceTimeMilliseconds(packet_time_2_ms);
+ // Change the capacity.
+ config.link_capacity_kbps *= 2; // Double the capacity.
+ simulated_network->SetConfig(config);
+
+ // Advance time in steps to release remaining packets one at a time.
+ step_ms = PacketTimeMs(config.link_capacity_kbps, kPacketSize);
+ for (int i = 0; i < kNumPackets / 2; ++i) {
+ fake_clock_.AdvanceTimeMilliseconds(step_ms);
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(1);
pipe->Process();
}
// Check that all the packets were sent.
- EXPECT_EQ(static_cast<size_t>(2 * kNumPackets), pipe->SentPackets());
+ EXPECT_EQ(static_cast<size_t>(kNumPackets), pipe->SentPackets());
fake_clock_.AdvanceTimeMilliseconds(pipe->TimeUntilNextProcess());
EXPECT_CALL(receiver, DeliverPacket(_, _, _)).Times(0);
pipe->Process();
diff --git a/modules/congestion_controller/pcc/pcc_network_controller_unittest.cc b/modules/congestion_controller/pcc/pcc_network_controller_unittest.cc
index bcdc75f..cb4e680 100644
--- a/modules/congestion_controller/pcc/pcc_network_controller_unittest.cc
+++ b/modules/congestion_controller/pcc/pcc_network_controller_unittest.cc
@@ -114,7 +114,7 @@
ret_net->UpdateConfig(
[](NetworkNodeConfig* c) { c->simulation.delay = TimeDelta::ms(200); });
- s.RunFor(TimeDelta::seconds(10));
+ s.RunFor(TimeDelta::seconds(20));
EXPECT_NEAR(client->target_rate_kbps(), 200, 40);
}
