mollyim/webrtc
1
0
Fork 0
mirror of https://github.com/mollyim/webrtc.git synced 2025-05-13 13:50:40 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Migrate call/ to absl::AnyInvocable based TaskQueueBase interface

Browse source 
Bug: webrtc:14245
Change-Id: Ifcdcd343fcba1d850e40813bc08862c42647b0c5
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/268002
Commit-Queue: Danil Chapovalov <danilchap@webrtc.org>
Reviewed-by: Erik Språng <sprang@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#37477}
This commit is contained in:
Danil Chapovalov 2022-07-06 18:35:01 +02:00 committed by WebRTC LUCI CQ
parent 3b526d4501
commit b7128ed172
7 changed files with 40 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
call/BUILD.gn
View file

@ -526,6 +526,7 @@ if (rtc_include_tests) {
] ]
absl_deps = [ absl_deps = [
"//third_party/abseil-cpp/absl/container:inlined_vector", "//third_party/abseil-cpp/absl/container:inlined_vector",
"//third_party/abseil-cpp/absl/functional:any_invocable",
"//third_party/abseil-cpp/absl/memory", "//third_party/abseil-cpp/absl/memory",
"//third_party/abseil-cpp/absl/strings", "//third_party/abseil-cpp/absl/strings",
"//third_party/abseil-cpp/absl/types:optional", "//third_party/abseil-cpp/absl/types:optional",

8
call/adaptation/resource_adaptation_processor.cc
View file

@ -41,11 +41,11 @@ void ResourceAdaptationProcessor::ResourceListenerDelegate::
OnResourceUsageStateMeasured(rtc::scoped_refptr<Resource> resource, OnResourceUsageStateMeasured(rtc::scoped_refptr<Resource> resource,
ResourceUsageState usage_state) { ResourceUsageState usage_state) {
if (!task_queue_->IsCurrent()) { if (!task_queue_->IsCurrent()) {
task_queue_->PostTask(ToQueuedTask( task_queue_->PostTask(
[this_ref = rtc::scoped_refptr<ResourceListenerDelegate>(this), [this_ref = rtc::scoped_refptr<ResourceListenerDelegate>(this),
resource, usage_state] { resource, usage_state] {
this_ref->OnResourceUsageStateMeasured(resource, usage_state); this_ref->OnResourceUsageStateMeasured(resource, usage_state);
})); });
return; return;
} }
RTC_DCHECK_RUN_ON(task_queue_); RTC_DCHECK_RUN_ON(task_queue_);
@ -142,8 +142,8 @@ void ResourceAdaptationProcessor::RemoveResource(
void ResourceAdaptationProcessor::RemoveLimitationsImposedByResource( void ResourceAdaptationProcessor::RemoveLimitationsImposedByResource(
rtc::scoped_refptr<Resource> resource) { rtc::scoped_refptr<Resource> resource) {
if (!task_queue_->IsCurrent()) { if (!task_queue_->IsCurrent()) {
task_queue_->PostTask(ToQueuedTask( task_queue_->PostTask(
[this, resource]() { RemoveLimitationsImposedByResource(resource); })); [this, resource]() { RemoveLimitationsImposedByResource(resource); });
return; return;
} }
RTC_DCHECK_RUN_ON(task_queue_); RTC_DCHECK_RUN_ON(task_queue_);

12
call/adaptation/resource_adaptation_processor_unittest.cc
View file

@ -430,8 +430,8 @@ TEST_F(ResourceAdaptationProcessorTest,
SetInputStates(true, kDefaultFrameRate, kDefaultFrameSize); SetInputStates(true, kDefaultFrameRate, kDefaultFrameSize);
TaskQueueForTest resource_task_queue("ResourceTaskQueue"); TaskQueueForTest resource_task_queue("ResourceTaskQueue");
resource_task_queue.PostTask(ToQueuedTask( resource_task_queue.PostTask(
[&]() { resource_->SetUsageState(ResourceUsageState::kOveruse); })); [&]() { resource_->SetUsageState(ResourceUsageState::kOveruse); });
EXPECT_EQ_WAIT(1u, restrictions_listener_.restrictions_updated_count(), EXPECT_EQ_WAIT(1u, restrictions_listener_.restrictions_updated_count(),
kDefaultTimeoutMs); kDefaultTimeoutMs);
@ -447,10 +447,10 @@ TEST_F(ResourceAdaptationProcessorTest,
// has passed it on to the processor's task queue. // has passed it on to the processor's task queue.
rtc::Event resource_event; rtc::Event resource_event;
TaskQueueForTest resource_task_queue("ResourceTaskQueue"); TaskQueueForTest resource_task_queue("ResourceTaskQueue");
resource_task_queue.PostTask(ToQueuedTask([&]() { resource_task_queue.PostTask([&]() {
resource_->SetUsageState(ResourceUsageState::kOveruse); resource_->SetUsageState(ResourceUsageState::kOveruse);
resource_event.Set(); resource_event.Set();
})); });
EXPECT_TRUE(resource_event.Wait(kDefaultTimeoutMs)); EXPECT_TRUE(resource_event.Wait(kDefaultTimeoutMs));
// Now destroy the processor while handling the overuse is in flight. // Now destroy the processor while handling the overuse is in flight.
@ -470,10 +470,10 @@ TEST_F(ResourceAdaptationProcessorTest,
rtc::Event overuse_event; rtc::Event overuse_event;
TaskQueueForTest resource_task_queue("ResourceTaskQueue"); TaskQueueForTest resource_task_queue("ResourceTaskQueue");
// Queues task for `resource_` overuse while `processor_` is still listening. // Queues task for `resource_` overuse while `processor_` is still listening.
resource_task_queue.PostTask(ToQueuedTask([&]() { resource_task_queue.PostTask([&]() {
resource_->SetUsageState(ResourceUsageState::kOveruse); resource_->SetUsageState(ResourceUsageState::kOveruse);
overuse_event.Set(); overuse_event.Set();
})); });
EXPECT_TRUE(overuse_event.Wait(kDefaultTimeoutMs)); EXPECT_TRUE(overuse_event.Wait(kDefaultTimeoutMs));
// Once we know the overuse task is queued, remove `resource_` so that // Once we know the overuse task is queued, remove `resource_` so that
// `processor_` is not listening to it. // `processor_` is not listening to it.

6
call/call.cc
View file

@ -1209,14 +1209,14 @@ void Call::SignalChannelNetworkState(MediaType media, NetworkState state) {
} else { } else {
// TODO(bugs.webrtc.org/11993): Remove workaround when we no longer need to // TODO(bugs.webrtc.org/11993): Remove workaround when we no longer need to
// post to the worker thread. // post to the worker thread.
worker_thread_->PostTask(ToQueuedTask(task_safety_, std::move(closure))); worker_thread_->PostTask(SafeTask(task_safety_.flag(), std::move(closure)));
} }
} }
void Call::OnAudioTransportOverheadChanged(int transport_overhead_per_packet) { void Call::OnAudioTransportOverheadChanged(int transport_overhead_per_packet) {
RTC_DCHECK_RUN_ON(network_thread_); RTC_DCHECK_RUN_ON(network_thread_);
worker_thread_->PostTask( worker_thread_->PostTask(
ToQueuedTask(task_safety_, [this, transport_overhead_per_packet]() { SafeTask(task_safety_.flag(), [this, transport_overhead_per_packet]() {
// TODO(bugs.webrtc.org/11993): Move this over to the network thread. // TODO(bugs.webrtc.org/11993): Move this over to the network thread.
RTC_DCHECK_RUN_ON(worker_thread_); RTC_DCHECK_RUN_ON(worker_thread_);
for (auto& kv : audio_send_ssrcs_) { for (auto& kv : audio_send_ssrcs_) {
@ -1408,7 +1408,7 @@ void Call::DeliverRtcp(MediaType media_type, rtc::CopyOnWriteBuffer packet) {
// TODO(bugs.webrtc.org/11993): This should execute directly on the network // TODO(bugs.webrtc.org/11993): This should execute directly on the network
// thread. // thread.
worker_thread_->PostTask( worker_thread_->PostTask(
ToQueuedTask(task_safety_, [this, packet = std::move(packet)]() { SafeTask(task_safety_.flag(), [this, packet = std::move(packet)]() {
RTC_DCHECK_RUN_ON(worker_thread_); RTC_DCHECK_RUN_ON(worker_thread_);
receive_stats_.AddReceivedRtcpBytes(static_cast<int>(packet.size())); receive_stats_.AddReceivedRtcpBytes(static_cast<int>(packet.size()));

10
call/call_perf_tests.cc
View file

@ -113,7 +113,7 @@ class VideoRtcpAndSyncObserver : public test::RtpRtcpObserver,
task_queue_(task_queue) {} task_queue_(task_queue) {}
void OnFrame(const VideoFrame& video_frame) override { void OnFrame(const VideoFrame& video_frame) override {
task_queue_->PostTask(ToQueuedTask([this]() { CheckStats(); })); task_queue_->PostTask([this]() { CheckStats(); });
} }
void CheckStats() { void CheckStats() {
@ -343,7 +343,7 @@ void CallPerfTest::TestAudioVideoSync(FecMode fec,
} }
task_queue()->PostTask( task_queue()->PostTask(
ToQueuedTask([to_delete = observer.release()]() { delete to_delete; })); [to_delete = observer.release()]() { delete to_delete; });
} }
TEST_F(CallPerfTest, Synchronization_PlaysOutAudioAndVideoWithoutClockDrift) { TEST_F(CallPerfTest, Synchronization_PlaysOutAudioAndVideoWithoutClockDrift) {
@ -680,7 +680,7 @@ void CallPerfTest::TestMinTransmitBitrate(bool pad_to_min_bitrate) {
private: private:
// TODO(holmer): Run this with a timer instead of once per packet. // TODO(holmer): Run this with a timer instead of once per packet.
Action OnSendRtp(const uint8_t* packet, size_t length) override { Action OnSendRtp(const uint8_t* packet, size_t length) override {
task_queue_->PostTask(ToQueuedTask(task_safety_flag_, [this]() { task_queue_->PostTask(SafeTask(task_safety_flag_, [this]() {
VideoSendStream::Stats stats = send_stream_->GetStats(); VideoSendStream::Stats stats = send_stream_->GetStats();
if (!stats.substreams.empty()) { if (!stats.substreams.empty()) {
@ -1146,7 +1146,7 @@ void CallPerfTest::TestEncodeFramerate(VideoEncoderFactory* encoder_factory,
const Timestamp now = clock_->CurrentTime(); const Timestamp now = clock_->CurrentTime();
if (now - last_getstats_time_ > kMinGetStatsInterval) { if (now - last_getstats_time_ > kMinGetStatsInterval) {
last_getstats_time_ = now; last_getstats_time_ = now;
task_queue_->PostTask(ToQueuedTask([this, now]() { task_queue_->PostTask([this, now]() {
VideoSendStream::Stats stats = send_stream_->GetStats(); VideoSendStream::Stats stats = send_stream_->GetStats();
for (const auto& stat : stats.substreams) { for (const auto& stat : stats.substreams) {
encode_frame_rate_lists_[stat.first].push_back( encode_frame_rate_lists_[stat.first].push_back(
@ -1156,7 +1156,7 @@ void CallPerfTest::TestEncodeFramerate(VideoEncoderFactory* encoder_factory,
VerifyStats(); VerifyStats();
observation_complete_.Set(); observation_complete_.Set();
} }
})); });
} }
return SEND_PACKET; return SEND_PACKET;
} }

35
call/degraded_call.cc
View file

@ -62,20 +62,20 @@ bool DegradedCall::FakeNetworkPipeOnTaskQueue::Process() {
return false; return false;
} }
task_queue_->PostTask(ToQueuedTask(task_safety_, [this, time_to_next] { task_queue_->PostTask(SafeTask(task_safety_.flag(), [this, time_to_next] {
RTC_DCHECK_RUN_ON(task_queue_); RTC_DCHECK_RUN_ON(task_queue_);
int64_t next_process_time = *time_to_next + clock_->TimeInMilliseconds(); int64_t next_process_time = *time_to_next + clock_->TimeInMilliseconds();
if (!next_process_ms_ || next_process_time < *next_process_ms_) { if (!next_process_ms_ || next_process_time < *next_process_ms_) {
next_process_ms_ = next_process_time; next_process_ms_ = next_process_time;
task_queue_->PostDelayedHighPrecisionTask( task_queue_->PostDelayedHighPrecisionTask(
ToQueuedTask(task_safety_, SafeTask(task_safety_.flag(),
[this] { [this] {
RTC_DCHECK_RUN_ON(task_queue_); RTC_DCHECK_RUN_ON(task_queue_);
if (!Process()) { if (!Process()) {
next_process_ms_.reset(); next_process_ms_.reset();
} }
}), }),
*time_to_next); TimeDelta::Millis(*time_to_next));
} }
})); }));
@ -146,8 +146,9 @@ DegradedCall::DegradedCall(
receive_pipe_->SetReceiver(call_->Receiver()); receive_pipe_->SetReceiver(call_->Receiver());
if (receive_configs_.size() > 1) { if (receive_configs_.size() > 1) {
call_->network_thread()->PostDelayedTask( call_->network_thread()->PostDelayedTask(
ToQueuedTask(task_safety_, [this] { UpdateReceiveNetworkConfig(); }), SafeTask(task_safety_.flag(),
receive_configs_[0].duration.ms()); [this] { UpdateReceiveNetworkConfig(); }),
receive_configs_[0].duration);
} }
} }
if (!send_configs_.empty()) { if (!send_configs_.empty()) {
@ -157,8 +158,8 @@ DegradedCall::DegradedCall(
call_->network_thread(), task_safety_, clock_, std::move(network)); call_->network_thread(), task_safety_, clock_, std::move(network));
if (send_configs_.size() > 1) { if (send_configs_.size() > 1) {
call_->network_thread()->PostDelayedTask( call_->network_thread()->PostDelayedTask(
ToQueuedTask(task_safety_, [this] { UpdateSendNetworkConfig(); }), SafeTask(task_safety_.flag(), [this] { UpdateSendNetworkConfig(); }),
send_configs_[0].duration.ms()); send_configs_[0].duration);
} }
} }
} }
@ -352,8 +353,8 @@ void DegradedCall::UpdateSendNetworkConfig() {
send_config_index_ = (send_config_index_ + 1) % send_configs_.size(); send_config_index_ = (send_config_index_ + 1) % send_configs_.size();
send_simulated_network_->SetConfig(send_configs_[send_config_index_]); send_simulated_network_->SetConfig(send_configs_[send_config_index_]);
call_->network_thread()->PostDelayedTask( call_->network_thread()->PostDelayedTask(
ToQueuedTask(task_safety_, [this] { UpdateSendNetworkConfig(); }), SafeTask(task_safety_.flag(), [this] { UpdateSendNetworkConfig(); }),
send_configs_[send_config_index_].duration.ms()); send_configs_[send_config_index_].duration);
} }
void DegradedCall::UpdateReceiveNetworkConfig() { void DegradedCall::UpdateReceiveNetworkConfig() {
@ -361,7 +362,7 @@ void DegradedCall::UpdateReceiveNetworkConfig() {
receive_simulated_network_->SetConfig( receive_simulated_network_->SetConfig(
receive_configs_[receive_config_index_]); receive_configs_[receive_config_index_]);
call_->network_thread()->PostDelayedTask( call_->network_thread()->PostDelayedTask(
ToQueuedTask(task_safety_, [this] { UpdateReceiveNetworkConfig(); }), SafeTask(task_safety_.flag(), [this] { UpdateReceiveNetworkConfig(); }),
receive_configs_[receive_config_index_].duration.ms()); receive_configs_[receive_config_index_].duration);
} }
} // namespace webrtc } // namespace webrtc

6
call/rtp_video_sender_unittest.cc
View file

@ -15,6 +15,7 @@
#include <string> #include <string>
#include <utility> #include <utility>
#include "absl/functional/any_invocable.h"
#include "call/rtp_transport_controller_send.h" #include "call/rtp_transport_controller_send.h"
#include "modules/rtp_rtcp/include/rtp_rtcp_defines.h" #include "modules/rtp_rtcp/include/rtp_rtcp_defines.h"
#include "modules/rtp_rtcp/source/byte_io.h" #include "modules/rtp_rtcp/source/byte_io.h"
@ -203,10 +204,9 @@ class RtpVideoSenderTestFixture {
// default thread as the transport queue, explicit checks for the transport // default thread as the transport queue, explicit checks for the transport
// queue (not just using a SequenceChecker) aren't possible unless such a // queue (not just using a SequenceChecker) aren't possible unless such a
// queue is actually active. So RunOnTransportQueue is a convenience function // queue is actually active. So RunOnTransportQueue is a convenience function
// that allow for running a closure on the transport queue, similar to // that allow for running a `task` on the transport queue, similar to
// SendTask(). // SendTask().
template <typename Closure> void RunOnTransportQueue(absl::AnyInvocable<void() &&> task) {
void RunOnTransportQueue(Closure&& task) {
transport_controller_.GetWorkerQueue()->PostTask(std::move(task)); transport_controller_.GetWorkerQueue()->PostTask(std::move(task));
AdvanceTime(TimeDelta::Millis(0)); AdvanceTime(TimeDelta::Millis(0));
} }