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