NackModule2: coalesce repeating tasks.
NackModule2 creates repeating tasks, but as there are
many modules (one per receiver) these tasks execute out
of phase with each other, multipliying the amount of wakeups
caused.
Fix this by creating a single wakeup source that serves all
NackModule2 instances in a call.
Bug: webrtc:12989
Change-Id: Ia9c84307eb57349679e42b673474feb2cb43f08e
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/226464
Reviewed-by: Ilya Nikolaevskiy <ilnik@webrtc.org>
Reviewed-by: Harald Alvestrand <hta@webrtc.org>
Commit-Queue: Markus Handell <handellm@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#34527}
diff --git a/call/call.cc b/call/call.cc
index fb1d7cd..a0c33ff 100644
--- a/call/call.cc
+++ b/call/call.cc
@@ -376,6 +376,9 @@
// network thread.
bool aggregate_network_up_ RTC_GUARDED_BY(worker_thread_);
+ // Schedules nack periodic processing on behalf of all streams.
+ NackPeriodicProcessor nack_periodic_processor_;
+
// Audio, Video, and FlexFEC receive streams are owned by the client that
// creates them.
// TODO(bugs.webrtc.org/11993): Move audio_receive_streams_,
@@ -1124,7 +1127,8 @@
VideoReceiveStream2* receive_stream = new VideoReceiveStream2(
task_queue_factory_, this, num_cpu_cores_,
transport_send_->packet_router(), std::move(configuration),
- call_stats_.get(), clock_, new VCMTiming(clock_));
+ call_stats_.get(), clock_, new VCMTiming(clock_),
+ &nack_periodic_processor_);
// TODO(bugs.webrtc.org/11993): Set this up asynchronously on the network
// thread.
receive_stream->RegisterWithTransport(&video_receiver_controller_);
diff --git a/modules/video_coding/BUILD.gn b/modules/video_coding/BUILD.gn
index 50f2e8d..d88ce42 100644
--- a/modules/video_coding/BUILD.gn
+++ b/modules/video_coding/BUILD.gn
@@ -82,6 +82,7 @@
deps = [
"..:module_api",
"../../api:sequence_checker",
+ "../../api/task_queue",
"../../api/units:time_delta",
"../../api/units:timestamp",
"../../rtc_base:checks",
diff --git a/modules/video_coding/nack_module2.cc b/modules/video_coding/nack_module2.cc
index 8a3a731..41a5b31 100644
--- a/modules/video_coding/nack_module2.cc
+++ b/modules/video_coding/nack_module2.cc
@@ -13,6 +13,8 @@
#include <algorithm>
#include <limits>
+#include "api/sequence_checker.h"
+#include "api/task_queue/task_queue_base.h"
#include "api/units/timestamp.h"
#include "rtc_base/checks.h"
#include "rtc_base/experiments/field_trial_parser.h"
@@ -43,7 +45,52 @@
}
} // namespace
-constexpr TimeDelta NackModule2::kUpdateInterval;
+constexpr TimeDelta NackPeriodicProcessor::kUpdateInterval;
+
+NackPeriodicProcessor::NackPeriodicProcessor(TimeDelta update_interval)
+ : update_interval_(update_interval) {}
+
+NackPeriodicProcessor::~NackPeriodicProcessor() {}
+
+void NackPeriodicProcessor::RegisterNackModule(NackModuleBase* module) {
+ RTC_DCHECK_RUN_ON(&sequence_);
+ modules_.push_back(module);
+ if (modules_.size() != 1)
+ return;
+ repeating_task_ = RepeatingTaskHandle::DelayedStart(
+ TaskQueueBase::Current(), update_interval_, [this] {
+ RTC_DCHECK_RUN_ON(&sequence_);
+ ProcessNackModules();
+ return update_interval_;
+ });
+}
+
+void NackPeriodicProcessor::UnregisterNackModule(NackModuleBase* module) {
+ RTC_DCHECK_RUN_ON(&sequence_);
+ auto it = std::find(modules_.begin(), modules_.end(), module);
+ RTC_DCHECK(it != modules_.end());
+ modules_.erase(it);
+ if (modules_.empty())
+ repeating_task_.Stop();
+}
+
+// RTC_RUN_ON(sequence_)
+void NackPeriodicProcessor::ProcessNackModules() {
+ for (NackModuleBase* module : modules_)
+ module->ProcessNacks();
+}
+
+ScopedNackPeriodicProcessorRegistration::
+ ScopedNackPeriodicProcessorRegistration(NackModuleBase* module,
+ NackPeriodicProcessor* processor)
+ : module_(module), processor_(processor) {
+ processor_->RegisterNackModule(module_);
+}
+
+ScopedNackPeriodicProcessorRegistration::
+ ~ScopedNackPeriodicProcessorRegistration() {
+ processor_->UnregisterNackModule(module_);
+}
NackModule2::NackInfo::NackInfo()
: seq_num(0), send_at_seq_num(0), sent_at_time(-1), retries(0) {}
@@ -89,12 +136,11 @@
}
NackModule2::NackModule2(TaskQueueBase* current_queue,
+ NackPeriodicProcessor* periodic_processor,
Clock* clock,
NackSender* nack_sender,
- KeyFrameRequestSender* keyframe_request_sender,
- TimeDelta update_interval /*= kUpdateInterval*/)
+ KeyFrameRequestSender* keyframe_request_sender)
: worker_thread_(current_queue),
- update_interval_(update_interval),
clock_(clock),
nack_sender_(nack_sender),
keyframe_request_sender_(keyframe_request_sender),
@@ -103,32 +149,27 @@
rtt_ms_(kDefaultRttMs),
newest_seq_num_(0),
send_nack_delay_ms_(GetSendNackDelay()),
- backoff_settings_(BackoffSettings::ParseFromFieldTrials()) {
+ backoff_settings_(BackoffSettings::ParseFromFieldTrials()),
+ processor_registration_(this, periodic_processor) {
RTC_DCHECK(clock_);
RTC_DCHECK(nack_sender_);
RTC_DCHECK(keyframe_request_sender_);
- RTC_DCHECK_GT(update_interval.ms(), 0);
RTC_DCHECK(worker_thread_);
RTC_DCHECK(worker_thread_->IsCurrent());
-
- repeating_task_ = RepeatingTaskHandle::DelayedStart(
- TaskQueueBase::Current(), update_interval_,
- [this]() {
- RTC_DCHECK_RUN_ON(worker_thread_);
- std::vector<uint16_t> nack_batch = GetNackBatch(kTimeOnly);
- if (!nack_batch.empty()) {
- // This batch of NACKs is triggered externally; there is no external
- // initiator who can batch them with other feedback messages.
- nack_sender_->SendNack(nack_batch, /*buffering_allowed=*/false);
- }
- return update_interval_;
- },
- clock_);
}
NackModule2::~NackModule2() {
RTC_DCHECK_RUN_ON(worker_thread_);
- repeating_task_.Stop();
+}
+
+void NackModule2::ProcessNacks() {
+ RTC_DCHECK_RUN_ON(worker_thread_);
+ std::vector<uint16_t> nack_batch = GetNackBatch(kTimeOnly);
+ if (!nack_batch.empty()) {
+ // This batch of NACKs is triggered externally; there is no external
+ // initiator who can batch them with other feedback messages.
+ nack_sender_->SendNack(nack_batch, /*buffering_allowed=*/false);
+ }
}
int NackModule2::OnReceivedPacket(uint16_t seq_num, bool is_keyframe) {
diff --git a/modules/video_coding/nack_module2.h b/modules/video_coding/nack_module2.h
index f58f886..f1f212c 100644
--- a/modules/video_coding/nack_module2.h
+++ b/modules/video_coding/nack_module2.h
@@ -30,20 +30,54 @@
namespace webrtc {
+class NackModuleBase {
+ public:
+ virtual ~NackModuleBase() = default;
+ virtual void ProcessNacks() = 0;
+};
+
+class NackPeriodicProcessor {
+ public:
+ static constexpr TimeDelta kUpdateInterval = TimeDelta::Millis(20);
+ explicit NackPeriodicProcessor(TimeDelta update_interval = kUpdateInterval);
+ ~NackPeriodicProcessor();
+ void RegisterNackModule(NackModuleBase* module);
+ void UnregisterNackModule(NackModuleBase* module);
+
+ private:
+ void ProcessNackModules() RTC_RUN_ON(sequence_);
+
+ const TimeDelta update_interval_;
+ RepeatingTaskHandle repeating_task_ RTC_GUARDED_BY(sequence_);
+ std::vector<NackModuleBase*> modules_ RTC_GUARDED_BY(sequence_);
+ RTC_NO_UNIQUE_ADDRESS SequenceChecker sequence_;
+};
+
+class ScopedNackPeriodicProcessorRegistration {
+ public:
+ ScopedNackPeriodicProcessorRegistration(NackModuleBase* module,
+ NackPeriodicProcessor* processor);
+ ~ScopedNackPeriodicProcessorRegistration();
+
+ private:
+ NackModuleBase* const module_;
+ NackPeriodicProcessor* const processor_;
+};
+
// TODO(bugs.webrtc.org/11594): This class no longer implements the Module
// interface and therefore "NackModule" may not be a descriptive name anymore.
// Consider renaming to e.g. NackTracker or NackRequester.
-class NackModule2 final {
+class NackModule2 final : public NackModuleBase {
public:
- static constexpr TimeDelta kUpdateInterval = TimeDelta::Millis(20);
-
NackModule2(TaskQueueBase* current_queue,
+ NackPeriodicProcessor* periodic_processor,
Clock* clock,
NackSender* nack_sender,
- KeyFrameRequestSender* keyframe_request_sender,
- TimeDelta update_interval = kUpdateInterval);
+ KeyFrameRequestSender* keyframe_request_sender);
~NackModule2();
+ void ProcessNacks() override;
+
int OnReceivedPacket(uint16_t seq_num, bool is_keyframe);
int OnReceivedPacket(uint16_t seq_num, bool is_keyframe, bool is_recovered);
@@ -103,11 +137,6 @@
RTC_EXCLUSIVE_LOCKS_REQUIRED(worker_thread_);
TaskQueueBase* const worker_thread_;
-
- // Used to regularly call SendNack if needed.
- RepeatingTaskHandle repeating_task_ RTC_GUARDED_BY(worker_thread_);
- const TimeDelta update_interval_;
-
Clock* const clock_;
NackSender* const nack_sender_;
KeyFrameRequestSender* const keyframe_request_sender_;
@@ -131,6 +160,8 @@
const absl::optional<BackoffSettings> backoff_settings_;
+ ScopedNackPeriodicProcessorRegistration processor_registration_;
+
// Used to signal destruction to potentially pending tasks.
ScopedTaskSafety task_safety_;
};
diff --git a/modules/video_coding/nack_module2_unittest.cc b/modules/video_coding/nack_module2_unittest.cc
index acd1eea..7ffcc12 100644
--- a/modules/video_coding/nack_module2_unittest.cc
+++ b/modules/video_coding/nack_module2_unittest.cc
@@ -80,10 +80,13 @@
}
NackModule2& CreateNackModule(
- TimeDelta interval = NackModule2::kUpdateInterval) {
+ TimeDelta interval = NackPeriodicProcessor::kUpdateInterval) {
RTC_DCHECK(!nack_module_.get());
- nack_module_ = std::make_unique<NackModule2>(
- TaskQueueBase::Current(), clock_.get(), this, this, interval);
+ nack_periodic_processor_ =
+ std::make_unique<NackPeriodicProcessor>(interval);
+ nack_module_ = std::make_unique<NackModule2>(TaskQueueBase::Current(),
+ nack_periodic_processor_.get(),
+ clock_.get(), this, this);
nack_module_->UpdateRtt(kDefaultRttMs);
return *nack_module_.get();
}
@@ -92,6 +95,7 @@
test::RunLoop loop_;
std::unique_ptr<SimulatedClock> clock_;
test::ScopedFieldTrials field_trial_;
+ std::unique_ptr<NackPeriodicProcessor> nack_periodic_processor_;
std::unique_ptr<NackModule2> nack_module_;
std::vector<uint16_t> sent_nacks_;
int keyframes_requested_;
@@ -379,7 +383,11 @@
TestNackModule2WithFieldTrial()
: nack_delay_field_trial_("WebRTC-SendNackDelayMs/10/"),
clock_(new SimulatedClock(0)),
- nack_module_(TaskQueueBase::Current(), clock_.get(), this, this),
+ nack_module_(TaskQueueBase::Current(),
+ &nack_periodic_processor_,
+ clock_.get(),
+ this,
+ this),
keyframes_requested_(0) {}
void SendNack(const std::vector<uint16_t>& sequence_numbers,
@@ -392,6 +400,7 @@
test::ScopedFieldTrials nack_delay_field_trial_;
std::unique_ptr<SimulatedClock> clock_;
+ NackPeriodicProcessor nack_periodic_processor_;
NackModule2 nack_module_;
std::vector<uint16_t> sent_nacks_;
int keyframes_requested_;
diff --git a/video/rtp_video_stream_receiver2.cc b/video/rtp_video_stream_receiver2.cc
index 4b43247..8929ab4 100644
--- a/video/rtp_video_stream_receiver2.cc
+++ b/video/rtp_video_stream_receiver2.cc
@@ -105,6 +105,7 @@
std::unique_ptr<NackModule2> MaybeConstructNackModule(
TaskQueueBase* current_queue,
+ NackPeriodicProcessor* nack_periodic_processor,
const VideoReceiveStream::Config& config,
Clock* clock,
NackSender* nack_sender,
@@ -113,7 +114,8 @@
return nullptr;
// TODO(bugs.webrtc.org/12420): pass rtp_history_ms to the nack module.
- return std::make_unique<NackModule2>(current_queue, clock, nack_sender,
+ return std::make_unique<NackModule2>(current_queue, nack_periodic_processor,
+ clock, nack_sender,
keyframe_request_sender);
}
@@ -210,6 +212,7 @@
ReceiveStatistics* rtp_receive_statistics,
RtcpPacketTypeCounterObserver* rtcp_packet_type_counter_observer,
RtcpCnameCallback* rtcp_cname_callback,
+ NackPeriodicProcessor* nack_periodic_processor,
NackSender* nack_sender,
KeyFrameRequestSender* keyframe_request_sender,
OnCompleteFrameCallback* complete_frame_callback,
@@ -243,6 +246,7 @@
// directly with |rtp_rtcp_|.
rtcp_feedback_buffer_(this, nack_sender, this),
nack_module_(MaybeConstructNackModule(current_queue,
+ nack_periodic_processor,
config_,
clock_,
&rtcp_feedback_buffer_,
diff --git a/video/rtp_video_stream_receiver2.h b/video/rtp_video_stream_receiver2.h
index ddff26b..d97f521 100644
--- a/video/rtp_video_stream_receiver2.h
+++ b/video/rtp_video_stream_receiver2.h
@@ -39,6 +39,7 @@
#include "modules/rtp_rtcp/source/video_rtp_depacketizer.h"
#include "modules/video_coding/h264_sps_pps_tracker.h"
#include "modules/video_coding/loss_notification_controller.h"
+#include "modules/video_coding/nack_module2.h"
#include "modules/video_coding/packet_buffer.h"
#include "modules/video_coding/rtp_frame_reference_finder.h"
#include "modules/video_coding/unique_timestamp_counter.h"
@@ -89,6 +90,7 @@
ReceiveStatistics* rtp_receive_statistics,
RtcpPacketTypeCounterObserver* rtcp_packet_type_counter_observer,
RtcpCnameCallback* rtcp_cname_callback,
+ NackPeriodicProcessor* nack_periodic_processor,
NackSender* nack_sender,
// The KeyFrameRequestSender is optional; if not provided, key frame
// requests are sent via the internal RtpRtcp module.
diff --git a/video/rtp_video_stream_receiver2_unittest.cc b/video/rtp_video_stream_receiver2_unittest.cc
index 7ccf0a5..f97ccb6 100644
--- a/video/rtp_video_stream_receiver2_unittest.cc
+++ b/video/rtp_video_stream_receiver2_unittest.cc
@@ -172,8 +172,9 @@
rtp_video_stream_receiver_ = std::make_unique<RtpVideoStreamReceiver2>(
TaskQueueBase::Current(), Clock::GetRealTimeClock(), &mock_transport_,
nullptr, nullptr, &config_, rtp_receive_statistics_.get(), nullptr,
- nullptr, &mock_nack_sender_, &mock_key_frame_request_sender_,
- &mock_on_complete_frame_callback_, nullptr, nullptr);
+ nullptr, &nack_periodic_processor_, &mock_nack_sender_,
+ &mock_key_frame_request_sender_, &mock_on_complete_frame_callback_,
+ nullptr, nullptr);
VideoCodec codec;
codec.codecType = kVideoCodecGeneric;
rtp_video_stream_receiver_->AddReceiveCodec(kPayloadType, codec, {},
@@ -244,6 +245,7 @@
const webrtc::test::ScopedFieldTrials override_field_trials_;
VideoReceiveStream::Config config_;
+ NackPeriodicProcessor nack_periodic_processor_;
MockNackSender mock_nack_sender_;
MockKeyFrameRequestSender mock_key_frame_request_sender_;
MockTransport mock_transport_;
@@ -1132,8 +1134,8 @@
auto receiver = std::make_unique<RtpVideoStreamReceiver2>(
TaskQueueBase::Current(), Clock::GetRealTimeClock(), &mock_transport_,
nullptr, nullptr, &config_, rtp_receive_statistics_.get(), nullptr,
- nullptr, &mock_nack_sender_, nullptr, &mock_on_complete_frame_callback_,
- nullptr, mock_frame_transformer);
+ nullptr, &nack_periodic_processor_, &mock_nack_sender_, nullptr,
+ &mock_on_complete_frame_callback_, nullptr, mock_frame_transformer);
VideoCodec video_codec;
video_codec.codecType = kVideoCodecGeneric;
receiver->AddReceiveCodec(kPayloadType, video_codec, {},
diff --git a/video/video_receive_stream2.cc b/video/video_receive_stream2.cc
index 72257f0..0bb2307 100644
--- a/video/video_receive_stream2.cc
+++ b/video/video_receive_stream2.cc
@@ -211,14 +211,16 @@
: kMaxWaitForFrameMs;
}
-VideoReceiveStream2::VideoReceiveStream2(TaskQueueFactory* task_queue_factory,
- Call* call,
- int num_cpu_cores,
- PacketRouter* packet_router,
- VideoReceiveStream::Config config,
- CallStats* call_stats,
- Clock* clock,
- VCMTiming* timing)
+VideoReceiveStream2::VideoReceiveStream2(
+ TaskQueueFactory* task_queue_factory,
+ Call* call,
+ int num_cpu_cores,
+ PacketRouter* packet_router,
+ VideoReceiveStream::Config config,
+ CallStats* call_stats,
+ Clock* clock,
+ VCMTiming* timing,
+ NackPeriodicProcessor* nack_periodic_processor)
: task_queue_factory_(task_queue_factory),
transport_adapter_(config.rtcp_send_transport),
config_(std::move(config)),
@@ -240,6 +242,7 @@
rtp_receive_statistics_.get(),
&stats_proxy_,
&stats_proxy_,
+ nack_periodic_processor,
this, // NackSender
nullptr, // Use default KeyFrameRequestSender
this, // OnCompleteFrameCallback
diff --git a/video/video_receive_stream2.h b/video/video_receive_stream2.h
index 9557044..9ed5942 100644
--- a/video/video_receive_stream2.h
+++ b/video/video_receive_stream2.h
@@ -25,6 +25,7 @@
#include "modules/rtp_rtcp/include/flexfec_receiver.h"
#include "modules/rtp_rtcp/source/source_tracker.h"
#include "modules/video_coding/frame_buffer2.h"
+#include "modules/video_coding/nack_module2.h"
#include "modules/video_coding/video_receiver2.h"
#include "rtc_base/system/no_unique_address.h"
#include "rtc_base/task_queue.h"
@@ -97,7 +98,8 @@
VideoReceiveStream::Config config,
CallStats* call_stats,
Clock* clock,
- VCMTiming* timing);
+ VCMTiming* timing,
+ NackPeriodicProcessor* nack_periodic_processor);
// Destruction happens on the worker thread. Prior to destruction the caller
// must ensure that a registration with the transport has been cleared. See
// `RegisterWithTransport` for details.
diff --git a/video/video_receive_stream2_unittest.cc b/video/video_receive_stream2_unittest.cc
index 850fd0d..0b0ea35 100644
--- a/video/video_receive_stream2_unittest.cc
+++ b/video/video_receive_stream2_unittest.cc
@@ -137,7 +137,8 @@
video_receive_stream_ =
std::make_unique<webrtc::internal::VideoReceiveStream2>(
task_queue_factory_.get(), &fake_call_, kDefaultNumCpuCores,
- &packet_router_, config_.Copy(), &call_stats_, clock_, timing_);
+ &packet_router_, config_.Copy(), &call_stats_, clock_, timing_,
+ &nack_periodic_processor_);
video_receive_stream_->RegisterWithTransport(
&rtp_stream_receiver_controller_);
}
@@ -146,6 +147,7 @@
test::RunLoop loop_;
const std::unique_ptr<TaskQueueFactory> task_queue_factory_;
test::VideoDecoderProxyFactory h264_decoder_factory_;
+ NackPeriodicProcessor nack_periodic_processor_;
VideoReceiveStream::Config config_;
internal::CallStats call_stats_;
MockVideoDecoder mock_h264_video_decoder_;
@@ -316,7 +318,8 @@
timing_ = new VCMTiming(clock_);
video_receive_stream_.reset(new webrtc::internal::VideoReceiveStream2(
task_queue_factory_.get(), &fake_call_, kDefaultNumCpuCores,
- &packet_router_, config_.Copy(), &call_stats_, clock_, timing_));
+ &packet_router_, config_.Copy(), &call_stats_, clock_, timing_,
+ &nack_periodic_processor_));
video_receive_stream_->RegisterWithTransport(
&rtp_stream_receiver_controller_);
video_receive_stream_->SetAndGetRecordingState(std::move(state), false);
@@ -326,6 +329,7 @@
test::RunLoop loop_;
test::FunctionVideoDecoderFactory fake_decoder_factory_;
const std::unique_ptr<TaskQueueFactory> task_queue_factory_;
+ NackPeriodicProcessor nack_periodic_processor_;
VideoReceiveStream::Config config_;
internal::CallStats call_stats_;
cricket::FakeVideoRenderer fake_renderer_;
@@ -581,7 +585,8 @@
config_.Copy(),
&call_stats_,
time_controller_.GetClock(),
- new VCMTiming(time_controller_.GetClock())) {
+ new VCMTiming(time_controller_.GetClock()),
+ &nack_periodic_processor_) {
video_receive_stream_.RegisterWithTransport(
&rtp_stream_receiver_controller_);
video_receive_stream_.Start();
@@ -608,6 +613,7 @@
MockTransport mock_transport_;
FakeRenderer fake_renderer_;
cricket::FakeCall fake_call_;
+ NackPeriodicProcessor nack_periodic_processor_;
VideoReceiveStream::Config config_;
internal::CallStats call_stats_;
PacketRouter packet_router_;
@@ -749,7 +755,8 @@
video_receive_stream_ =
std::make_unique<webrtc::internal::VideoReceiveStream2>(
task_queue_factory_.get(), &fake_call_, kDefaultNumCpuCores,
- &packet_router_, config_.Copy(), &call_stats_, clock_, timing_);
+ &packet_router_, config_.Copy(), &call_stats_, clock_, timing_,
+ &nack_periodic_processor_);
video_receive_stream_->RegisterWithTransport(
&rtp_stream_receiver_controller_);
}
@@ -758,6 +765,7 @@
test::RunLoop loop_;
const std::unique_ptr<TaskQueueFactory> task_queue_factory_;
MockVideoDecoderFactory mock_h264_decoder_factory_;
+ NackPeriodicProcessor nack_periodic_processor_;
VideoReceiveStream::Config config_;
internal::CallStats call_stats_;
MockVideoDecoder mock_h264_video_decoder_;
