From 03bce3f49d84f4f88d8eca3450837cb26a272dd1 Mon Sep 17 00:00:00 2001 From: Etienne Pierre-doray Date: Mon, 29 Mar 2021 17:36:15 +0000 Subject: [PATCH] Reland "[Battery]: Delay start of TaskQueuePacedSender." Take 3 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This is a reland of 89cb65ed663a9000b9f7c90a78039bd85731e9ae ... and f28aade91dcc2cb8f590dc1379ac7ab5c1981909 ... and 2072b87261a6505a88561bdeab3e7405d7038eaa Reason for revert: Failing DuoGroupsMediaQualityTest due to missing TaskQueuePacedSender::EnsureStarted() in google3. Fix: This CL adds the logic behind TaskQueuePacedSender::EnsureStarted, but initializes with |is_started| = true. Once the caller in google3 is updated, |is_started| can be switched to false by default. > Original change's description: > Reason for revert: crashes due to uninitialized pacing_bitrate_ > crbug.com/1190547 > Apparently pacer() is sometimes being used before EnsureStarted() > Fix: Instead of delaying first call to SetPacingRates(), > this CL no-ops MaybeProcessPackets() until EnsureStarted() > is called for the first time. > Original change's description: > > [Battery]: Delay start of TaskQueuePacedSender. > > > > To avoid unnecessary repeating tasks, TaskQueuePacedSender is started > > only upon RtpTransportControllerSend::EnsureStarted(). > > > > More specifically, the repeating task happens in > > TaskQueuePacedSender::MaybeProcessPackets() every 500ms, using a self > > task_queue_.PostDelayedTask(). > > > > Bug: chromium:1152887 > > Change-Id: I72c96d2c4b491d5edb45a30b210b3797165cbf48 > > Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/208560 > > Commit-Queue: Etienne Pierre-Doray > > Reviewed-by: Henrik Boström > > Reviewed-by: Erik Språng > > Cr-Commit-Position: refs/heads/master@{#33421} > > Bug: chromium:1152887 > Change-Id: I9aba4882a64bbee7d97ace9059dea8a24c144f93 > Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/212880 > Reviewed-by: Erik Språng > Reviewed-by: Henrik Boström > Commit-Queue: Etienne Pierre-Doray > Cr-Commit-Position: refs/heads/master@{#33554} Bug: chromium:1152887 Change-Id: Ie365562bd83aefdb2757a65e20a4cf3eece678b9 Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/213000 Reviewed-by: Henrik Boström Reviewed-by: Erik Språng Commit-Queue: Etienne Pierre-Doray Cr-Commit-Position: refs/heads/master@{#33629} --- call/rtp_transport_controller_send.cc | 12 ++++-- call/rtp_transport_controller_send.h | 2 +- modules/pacing/task_queue_paced_sender.cc | 10 ++++- modules/pacing/task_queue_paced_sender.h | 9 +++++ .../task_queue_paced_sender_unittest.cc | 38 +++++++++++++++++++ 5 files changed, 65 insertions(+), 6 deletions(-) diff --git a/call/rtp_transport_controller_send.cc b/call/rtp_transport_controller_send.cc index f5adae68ae..d743a0bf43 100644 --- a/call/rtp_transport_controller_send.cc +++ b/call/rtp_transport_controller_send.cc @@ -87,7 +87,7 @@ RtpTransportControllerSend::RtpTransportControllerSend( : clock_(clock), event_log_(event_log), bitrate_configurator_(bitrate_config), - process_thread_started_(false), + pacer_started_(false), process_thread_(std::move(process_thread)), use_task_queue_pacer_(IsEnabled(trials, "WebRTC-TaskQueuePacer")), process_thread_pacer_(use_task_queue_pacer_ @@ -496,9 +496,13 @@ void RtpTransportControllerSend::IncludeOverheadInPacedSender() { } void RtpTransportControllerSend::EnsureStarted() { - if (!use_task_queue_pacer_ && !process_thread_started_) { - process_thread_started_ = true; - process_thread_->Start(); + if (!pacer_started_) { + pacer_started_ = true; + if (use_task_queue_pacer_) { + task_queue_pacer_->EnsureStarted(); + } else { + process_thread_->Start(); + } } } diff --git a/call/rtp_transport_controller_send.h b/call/rtp_transport_controller_send.h index 7025b03312..f0f74c9f2a 100644 --- a/call/rtp_transport_controller_send.h +++ b/call/rtp_transport_controller_send.h @@ -152,7 +152,7 @@ class RtpTransportControllerSend final std::vector> video_rtp_senders_; RtpBitrateConfigurator bitrate_configurator_; std::map network_routes_; - bool process_thread_started_; + bool pacer_started_; const std::unique_ptr process_thread_; const bool use_task_queue_pacer_; std::unique_ptr process_thread_pacer_; diff --git a/modules/pacing/task_queue_paced_sender.cc b/modules/pacing/task_queue_paced_sender.cc index 8ba49770d4..0944741810 100644 --- a/modules/pacing/task_queue_paced_sender.cc +++ b/modules/pacing/task_queue_paced_sender.cc @@ -62,6 +62,14 @@ TaskQueuePacedSender::~TaskQueuePacedSender() { }); } +void TaskQueuePacedSender::EnsureStarted() { + task_queue_.PostTask([this]() { + RTC_DCHECK_RUN_ON(&task_queue_); + is_started_ = true; + MaybeProcessPackets(Timestamp::MinusInfinity()); + }); +} + void TaskQueuePacedSender::CreateProbeCluster(DataRate bitrate, int cluster_id) { task_queue_.PostTask([this, bitrate, cluster_id]() { @@ -197,7 +205,7 @@ void TaskQueuePacedSender::MaybeProcessPackets( Timestamp scheduled_process_time) { RTC_DCHECK_RUN_ON(&task_queue_); - if (is_shutdown_) { + if (is_shutdown_ || !is_started_) { return; } diff --git a/modules/pacing/task_queue_paced_sender.h b/modules/pacing/task_queue_paced_sender.h index dc4c124ab7..a593069b7d 100644 --- a/modules/pacing/task_queue_paced_sender.h +++ b/modules/pacing/task_queue_paced_sender.h @@ -55,6 +55,9 @@ class TaskQueuePacedSender : public RtpPacketPacer, public RtpPacketSender { ~TaskQueuePacedSender() override; + // Ensure that necessary delayed tasks are scheduled. + void EnsureStarted(); + // Methods implementing RtpPacketSender. // Adds the packet to the queue and calls PacketRouter::SendPacket() when @@ -150,6 +153,12 @@ class TaskQueuePacedSender : public RtpPacketPacer, public RtpPacketSender { // Last time stats were updated. Timestamp last_stats_time_ RTC_GUARDED_BY(task_queue_); + // Indicates if this task queue is started. If not, don't allow + // posting delayed tasks yet. + // TODO(crbug.com/1152887): Initialize to false once all users call + // EnsureStarted(). + bool is_started_ RTC_GUARDED_BY(task_queue_) = true; + // Indicates if this task queue is shutting down. If so, don't allow // posting any more delayed tasks as that can cause the task queue to // never drain. diff --git a/modules/pacing/task_queue_paced_sender_unittest.cc b/modules/pacing/task_queue_paced_sender_unittest.cc index d389e271f7..ce26651824 100644 --- a/modules/pacing/task_queue_paced_sender_unittest.cc +++ b/modules/pacing/task_queue_paced_sender_unittest.cc @@ -157,6 +157,7 @@ namespace test { pacer.SetPacingRates( DataRate::BitsPerSec(kDefaultPacketSize * 8 * kPacketsToSend), DataRate::Zero()); + pacer.EnsureStarted(); pacer.EnqueuePackets( GeneratePackets(RtpPacketMediaType::kVideo, kPacketsToSend)); @@ -196,6 +197,7 @@ namespace test { const DataRate kPacingRate = DataRate::BitsPerSec(kDefaultPacketSize * 8 * kPacketsPerSecond); pacer.SetPacingRates(kPacingRate, DataRate::Zero()); + pacer.EnsureStarted(); // Send some initial packets to be rid of any probes. EXPECT_CALL(packet_router, SendPacket).Times(kPacketsPerSecond); @@ -247,6 +249,7 @@ namespace test { const TimeDelta kPacketPacingTime = kPacketSize / kPacingDataRate; pacer.SetPacingRates(kPacingDataRate, DataRate::Zero()); + pacer.EnsureStarted(); // Add some initial video packets, only one should be sent. EXPECT_CALL(packet_router, SendPacket); @@ -280,6 +283,7 @@ namespace test { const DataRate kPacingDataRate = kPacketSize / kPacketPacingTime; pacer.SetPacingRates(kPacingDataRate, DataRate::Zero()); + pacer.EnsureStarted(); // Add 10 packets. The first should be sent immediately since the buffers // are clear. @@ -316,6 +320,7 @@ namespace test { const DataRate kPacingDataRate = kPacketSize / kPacketPacingTime; pacer.SetPacingRates(kPacingDataRate, DataRate::Zero()); + pacer.EnsureStarted(); // Add 10 packets. The first should be sent immediately since the buffers // are clear. This will also trigger the probe to start. @@ -342,6 +347,7 @@ namespace test { kCoalescingWindow); const DataRate kPacingDataRate = DataRate::KilobitsPerSec(300); pacer.SetPacingRates(kPacingDataRate, DataRate::Zero()); + pacer.EnsureStarted(); const TimeDelta kMinTimeBetweenStatsUpdates = TimeDelta::Millis(1); @@ -388,6 +394,7 @@ namespace test { size_t num_expected_stats_updates = 0; EXPECT_EQ(pacer.num_stats_updates_, num_expected_stats_updates); pacer.SetPacingRates(kPacingDataRate, DataRate::Zero()); + pacer.EnsureStarted(); time_controller.AdvanceTime(kMinTimeBetweenStatsUpdates); // Updating pacing rates refreshes stats. EXPECT_EQ(pacer.num_stats_updates_, ++num_expected_stats_updates); @@ -443,6 +450,7 @@ namespace test { const TimeDelta kPacketPacingTime = TimeDelta::Millis(4); const DataRate kPacingDataRate = kPacketSize / kPacketPacingTime; pacer.SetPacingRates(kPacingDataRate, /*padding_rate=*/DataRate::Zero()); + pacer.EnsureStarted(); EXPECT_CALL(packet_router, FetchFec).WillRepeatedly([]() { return std::vector>(); }); @@ -514,6 +522,7 @@ namespace test { const TimeDelta kPacketPacingTime = TimeDelta::Millis(4); const DataRate kPacingDataRate = kPacketSize / kPacketPacingTime; pacer.SetPacingRates(kPacingDataRate, /*padding_rate=*/DataRate::Zero()); + pacer.EnsureStarted(); EXPECT_CALL(packet_router, FetchFec).WillRepeatedly([]() { return std::vector>(); }); @@ -552,5 +561,34 @@ namespace test { EXPECT_EQ(data_sent, kProbingRate * TimeDelta::Millis(1) + DataSize::Bytes(1)); } + + // TODO(crbug.com/1152887): Enable once pacer no longer auto-starts. + TEST(TaskQueuePacedSenderTest, DISABLED_NoStatsUpdatesBeforeStart) { + const TimeDelta kCoalescingWindow = TimeDelta::Millis(5); + GlobalSimulatedTimeController time_controller(Timestamp::Millis(1234)); + MockPacketRouter packet_router; + TaskQueuePacedSenderForTest pacer( + time_controller.GetClock(), &packet_router, + /*event_log=*/nullptr, + /*field_trials=*/nullptr, time_controller.GetTaskQueueFactory(), + kCoalescingWindow); + const DataRate kPacingDataRate = DataRate::KilobitsPerSec(300); + pacer.SetPacingRates(kPacingDataRate, DataRate::Zero()); + + const TimeDelta kMinTimeBetweenStatsUpdates = TimeDelta::Millis(1); + + // Nothing inserted, no stats updates yet. + EXPECT_EQ(pacer.num_stats_updates_, 0u); + + // Insert one packet, stats should not be updated. + pacer.EnqueuePackets(GeneratePackets(RtpPacketMediaType::kVideo, 1)); + time_controller.AdvanceTime(TimeDelta::Zero()); + EXPECT_EQ(pacer.num_stats_updates_, 0u); + + // Advance time of the min stats update interval, and trigger a + // refresh - stats should not be updated still. + time_controller.AdvanceTime(kMinTimeBetweenStatsUpdates); + EXPECT_EQ(pacer.num_stats_updates_, 0u); + } } // namespace test } // namespace webrtc