zqz/platform-external-webrtc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Migrate rtp_rtcp to absl::AnyInvocable based TaskQueueBase interface

Browse Source 
Bug: webrtc:14245
Change-Id: I037f964130648caf0bd1de86611f8681d475b078
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/268146
Reviewed-by: Erik Språng <sprang@webrtc.org>
Commit-Queue: Danil Chapovalov <danilchap@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#37481}
This commit is contained in:
Danil Chapovalov
2022-07-07 13:34:25 +02:00
committed by WebRTC LUCI CQ
parent c52e627c83
commit 677c1ddde5
10 changed files with 78 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
modules/rtp_rtcp/BUILD.gn
View File

@ -275,7 +275,6 @@ rtc_library("rtp_rtcp") {
"../../api/rtc_event_log",
"../../api/task_queue:pending_task_safety_flag",
"../../api/task_queue:task_queue",
"../../api/task_queue:to_queued_task",
"../../api/transport:field_trial_based_config",
"../../api/transport/rtp:dependency_descriptor",
"../../api/transport/rtp:rtp_source",
@ -394,7 +393,6 @@ rtc_library("rtcp_transceiver") {
"../../api:rtp_headers",
"../../api:transport_api",
"../../api/task_queue",
"../../api/task_queue:to_queued_task",
"../../api/units:data_rate",
"../../api/units:time_delta",
"../../api/units:timestamp",
@ -412,6 +410,8 @@ rtc_library("rtcp_transceiver") {
absl_deps = [
"//third_party/abseil-cpp/absl/algorithm:container",
"//third_party/abseil-cpp/absl/base:core_headers",
"//third_party/abseil-cpp/absl/cleanup",
"//third_party/abseil-cpp/absl/functional:any_invocable",
"//third_party/abseil-cpp/absl/memory",
"//third_party/abseil-cpp/absl/types:optional",
]
@ -616,7 +616,6 @@ if (rtc_include_tests) {
"../../api:time_controller",
"../../api:transport_api",
"../../api/rtc_event_log",
"../../api/task_queue:to_queued_task",
"../../api/transport:field_trial_based_config",
"../../api/transport/rtp:dependency_descriptor",
"../../api/units:data_rate",

61
modules/rtp_rtcp/source/rtcp_transceiver.cc
View File

@ -14,7 +14,7 @@
#include <utility>
#include <vector>
#include "api/task_queue/to_queued_task.h"
#include "absl/cleanup/cleanup.h"
#include "api/units/timestamp.h"
#include "modules/rtp_rtcp/source/rtcp_packet/transport_feedback.h"
#include "rtc_base/checks.h"
@ -34,21 +34,19 @@ RtcpTransceiver::~RtcpTransceiver() {
if (!rtcp_transceiver_)
return;
auto rtcp_transceiver = std::move(rtcp_transceiver_);
task_queue_->PostTask(
ToQueuedTask([rtcp_transceiver = std::move(rtcp_transceiver)] {
task_queue_->PostTask([rtcp_transceiver = std::move(rtcp_transceiver)] {
rtcp_transceiver->StopPeriodicTask();
}));
});
RTC_DCHECK(!rtcp_transceiver_);
}
void RtcpTransceiver::Stop(std::function<void()> on_destroyed) {
void RtcpTransceiver::Stop(absl::AnyInvocable<void() &&> on_destroyed) {
RTC_DCHECK(rtcp_transceiver_);
auto rtcp_transceiver = std::move(rtcp_transceiver_);
task_queue_->PostTask(ToQueuedTask(
[rtcp_transceiver = std::move(rtcp_transceiver)] {
rtcp_transceiver->StopPeriodicTask();
},
std::move(on_destroyed)));
absl::Cleanup cleanup = std::move(on_destroyed);
task_queue_->PostTask(
[rtcp_transceiver = std::move(rtcp_transceiver),
cleanup = std::move(cleanup)] { rtcp_transceiver->StopPeriodicTask(); });
RTC_DCHECK(!rtcp_transceiver_);
}
@ -57,28 +55,28 @@ void RtcpTransceiver::AddMediaReceiverRtcpObserver(
MediaReceiverRtcpObserver* observer) {
RTC_CHECK(rtcp_transceiver_);
RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
task_queue_->PostTask(ToQueuedTask([ptr, remote_ssrc, observer] {
task_queue_->PostTask([ptr, remote_ssrc, observer] {
ptr->AddMediaReceiverRtcpObserver(remote_ssrc, observer);
}));
});
}
void RtcpTransceiver::RemoveMediaReceiverRtcpObserver(
uint32_t remote_ssrc,
MediaReceiverRtcpObserver* observer,
std::function<void()> on_removed) {
absl::AnyInvocable<void() &&> on_removed) {
RTC_CHECK(rtcp_transceiver_);
RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
auto remove = [ptr, remote_ssrc, observer] {
absl::Cleanup cleanup = std::move(on_removed);
task_queue_->PostTask(
[ptr, remote_ssrc, observer, cleanup = std::move(cleanup)] {
ptr->RemoveMediaReceiverRtcpObserver(remote_ssrc, observer);
};
task_queue_->PostTask(ToQueuedTask(std::move(remove), std::move(on_removed)));
});
}
void RtcpTransceiver::SetReadyToSend(bool ready) {
RTC_CHECK(rtcp_transceiver_);
RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
task_queue_->PostTask(
ToQueuedTask([ptr, ready] { ptr->SetReadyToSend(ready); }));
task_queue_->PostTask([ptr, ready] { ptr->SetReadyToSend(ready); });
}
void RtcpTransceiver::ReceivePacket(rtc::CopyOnWriteBuffer packet) {
@ -86,29 +84,28 @@ void RtcpTransceiver::ReceivePacket(rtc::CopyOnWriteBuffer packet) {
RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
Timestamp now = clock_->CurrentTime();
task_queue_->PostTask(
ToQueuedTask([ptr, packet, now] { ptr->ReceivePacket(packet, now); }));
[ptr, packet, now] { ptr->ReceivePacket(packet, now); });
}
void RtcpTransceiver::SendCompoundPacket() {
RTC_CHECK(rtcp_transceiver_);
RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
task_queue_->PostTask(ToQueuedTask([ptr] { ptr->SendCompoundPacket(); }));
task_queue_->PostTask([ptr] { ptr->SendCompoundPacket(); });
}
void RtcpTransceiver::SetRemb(int64_t bitrate_bps,
std::vector<uint32_t> ssrcs) {
RTC_CHECK(rtcp_transceiver_);
RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
task_queue_->PostTask(
ToQueuedTask([ptr, bitrate_bps, ssrcs = std::move(ssrcs)]() mutable {
task_queue_->PostTask([ptr, bitrate_bps, ssrcs = std::move(ssrcs)]() mutable {
ptr->SetRemb(bitrate_bps, std::move(ssrcs));
}));
});
}
void RtcpTransceiver::UnsetRemb() {
RTC_CHECK(rtcp_transceiver_);
RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
task_queue_->PostTask(ToQueuedTask([ptr] { ptr->UnsetRemb(); }));
task_queue_->PostTask([ptr] { ptr->UnsetRemb(); });
}
void RtcpTransceiver::SendCombinedRtcpPacket(
@ -116,26 +113,25 @@ void RtcpTransceiver::SendCombinedRtcpPacket(
RTC_CHECK(rtcp_transceiver_);
RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
task_queue_->PostTask(
ToQueuedTask([ptr, rtcp_packets = std::move(rtcp_packets)]() mutable {
[ptr, rtcp_packets = std::move(rtcp_packets)]() mutable {
ptr->SendCombinedRtcpPacket(std::move(rtcp_packets));
}));
});
}
void RtcpTransceiver::SendNack(uint32_t ssrc,
std::vector<uint16_t> sequence_numbers) {
RTC_CHECK(rtcp_transceiver_);
RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
task_queue_->PostTask(ToQueuedTask(
task_queue_->PostTask(
[ptr, ssrc, sequence_numbers = std::move(sequence_numbers)]() mutable {
ptr->SendNack(ssrc, std::move(sequence_numbers));
}));
});
}
void RtcpTransceiver::SendPictureLossIndication(uint32_t ssrc) {
RTC_CHECK(rtcp_transceiver_);
RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
task_queue_->PostTask(
ToQueuedTask([ptr, ssrc] { ptr->SendPictureLossIndication(ssrc); }));
task_queue_->PostTask([ptr, ssrc] { ptr->SendPictureLossIndication(ssrc); });
}
void RtcpTransceiver::SendFullIntraRequest(std::vector<uint32_t> ssrcs) {
@ -146,10 +142,9 @@ void RtcpTransceiver::SendFullIntraRequest(std::vector<uint32_t> ssrcs,
bool new_request) {
RTC_CHECK(rtcp_transceiver_);
RtcpTransceiverImpl* ptr = rtcp_transceiver_.get();
task_queue_->PostTask(
ToQueuedTask([ptr, ssrcs = std::move(ssrcs), new_request] {
task_queue_->PostTask([ptr, ssrcs = std::move(ssrcs), new_request] {
ptr->SendFullIntraRequest(ssrcs, new_request);
}));
});
}
} // namespace webrtc

9
modules/rtp_rtcp/source/rtcp_transceiver.h
View File

@ -11,11 +11,11 @@
#ifndef MODULES_RTP_RTCP_SOURCE_RTCP_TRANSCEIVER_H_
#define MODULES_RTP_RTCP_SOURCE_RTCP_TRANSCEIVER_H_
#include <functional>
#include <memory>
#include <string>
#include <vector>
#include "absl/functional/any_invocable.h"
#include "api/task_queue/task_queue_base.h"
#include "modules/rtp_rtcp/source/rtcp_transceiver_config.h"
#include "modules/rtp_rtcp/source/rtcp_transceiver_impl.h"
@ -44,7 +44,7 @@ class RtcpTransceiver : public RtcpFeedbackSenderInterface {
// Note that interfaces provided in constructor or registered with AddObserver
// still might be used by the transceiver on the task queue
// until `on_destroyed` runs.
void Stop(std::function<void()> on_destroyed);
void Stop(absl::AnyInvocable<void() &&> on_destroyed);
// Registers observer to be notified about incoming rtcp packets.
// Calls to observer will be done on the `config.task_queue`.
@ -52,9 +52,10 @@ class RtcpTransceiver : public RtcpFeedbackSenderInterface {
MediaReceiverRtcpObserver* observer);
// Deregisters the observer. Might return before observer is deregistered.
// Runs `on_removed` when observer is deregistered.
void RemoveMediaReceiverRtcpObserver(uint32_t remote_ssrc,
void RemoveMediaReceiverRtcpObserver(
uint32_t remote_ssrc,
MediaReceiverRtcpObserver* observer,
std::function<void()> on_removed);
absl::AnyInvocable<void() &&> on_removed);
// Enables/disables sending rtcp packets eventually.
// Packets may be sent after the SetReadyToSend(false) returns, but no new

1
modules/rtp_rtcp/source/rtcp_transceiver_impl.cc
View File

@ -16,7 +16,6 @@
#include "absl/algorithm/container.h"
#include "absl/memory/memory.h"
#include "api/call/transport.h"
#include "api/task_queue/to_queued_task.h"
#include "api/video/video_bitrate_allocation.h"
#include "modules/rtp_rtcp/include/receive_statistics.h"
#include "modules/rtp_rtcp/include/rtp_rtcp_defines.h"

43
modules/rtp_rtcp/source/rtcp_transceiver_impl_unittest.cc
View File

@ -16,7 +16,6 @@
#include "absl/memory/memory.h"
#include "api/rtp_headers.h"
#include "api/task_queue/to_queued_task.h"
#include "api/test/create_time_controller.h"
#include "api/test/time_controller.h"
#include "api/units/data_rate.h"
@ -216,11 +215,11 @@ TEST_F(RtcpTransceiverImplTest, NeedToStopPeriodicTaskToDestroyOnTaskQueue) {
EXPECT_TRUE(transport.WaitPacket());
bool done = false;
queue->PostTask(ToQueuedTask([rtcp_transceiver, &done] {
queue->PostTask([rtcp_transceiver, &done] {
rtcp_transceiver->StopPeriodicTask();
delete rtcp_transceiver;
done = true;
}));
});
ASSERT_TRUE(time_controller().Wait([&] { return done; }, kAlmostForever));
}
@ -233,11 +232,11 @@ TEST_F(RtcpTransceiverImplTest, CanBeDestroyedRightAfterCreation) {
config.outgoing_transport = &transport;
bool done = false;
queue->PostTask(ToQueuedTask([&] {
queue->PostTask([&] {
RtcpTransceiverImpl rtcp_transceiver(config);
rtcp_transceiver.StopPeriodicTask();
done = true;
}));
});
ASSERT_TRUE(time_controller().Wait([&] { return done; }, kAlmostForever));
}
@ -268,18 +267,18 @@ TEST_F(RtcpTransceiverImplTest, DelaysSendingFirstCompondPacket) {
absl::optional<RtcpTransceiverImpl> rtcp_transceiver;
Timestamp started = CurrentTime();
queue->PostTask(ToQueuedTask([&] { rtcp_transceiver.emplace(config); }));
queue->PostTask([&] { rtcp_transceiver.emplace(config); });
EXPECT_TRUE(transport.WaitPacket());
EXPECT_GE(CurrentTime() - started, config.initial_report_delay);
// Cleanup.
bool done = false;
queue->PostTask(ToQueuedTask([&] {
queue->PostTask([&] {
rtcp_transceiver->StopPeriodicTask();
rtcp_transceiver.reset();
done = true;
}));
});
ASSERT_TRUE(time_controller().Wait([&] { return done; }, kAlmostForever));
}
@ -294,12 +293,12 @@ TEST_F(RtcpTransceiverImplTest, PeriodicallySendsPackets) {
config.task_queue = queue.get();
absl::optional<RtcpTransceiverImpl> rtcp_transceiver;
Timestamp time_just_before_1st_packet = Timestamp::MinusInfinity();
queue->PostTask(ToQueuedTask([&] {
queue->PostTask([&] {
// Because initial_report_delay_ms is set to 0, time_just_before_the_packet
// should be very close to the time_of_the_packet.
time_just_before_1st_packet = CurrentTime();
rtcp_transceiver.emplace(config);
}));
});
EXPECT_TRUE(transport.WaitPacket());
EXPECT_TRUE(transport.WaitPacket());
@ -310,11 +309,11 @@ TEST_F(RtcpTransceiverImplTest, PeriodicallySendsPackets) {
// Cleanup.
bool done = false;
queue->PostTask(ToQueuedTask([&] {
queue->PostTask([&] {
rtcp_transceiver->StopPeriodicTask();
rtcp_transceiver.reset();
done = true;
}));
});
ASSERT_TRUE(time_controller().Wait([&] { return done; }, kAlmostForever));
}
@ -328,19 +327,20 @@ TEST_F(RtcpTransceiverImplTest, SendCompoundPacketDelaysPeriodicSendPackets) {
config.report_period = kReportPeriod;
config.task_queue = queue.get();
absl::optional<RtcpTransceiverImpl> rtcp_transceiver;
queue->PostTask(ToQueuedTask([&] { rtcp_transceiver.emplace(config); }));
queue->PostTask([&] { rtcp_transceiver.emplace(config); });
// Wait for the first packet.
EXPECT_TRUE(transport.WaitPacket());
// Send non periodic one after half period.
bool non_periodic = false;
Timestamp time_of_non_periodic_packet = Timestamp::MinusInfinity();
queue->PostDelayedTask(ToQueuedTask([&] {
queue->PostDelayedTask(
[&] {
time_of_non_periodic_packet = CurrentTime();
rtcp_transceiver->SendCompoundPacket();
non_periodic = true;
}),
(config.report_period / 2).ms());
},
config.report_period / 2);
// Though non-periodic packet is scheduled just in between periodic, due to
// small period and task queue flakiness it migth end-up 1ms after next
// periodic packet. To be sure duration after non-periodic packet is tested
@ -356,11 +356,11 @@ TEST_F(RtcpTransceiverImplTest, SendCompoundPacketDelaysPeriodicSendPackets) {
// Cleanup.
bool done = false;
queue->PostTask(ToQueuedTask([&] {
queue->PostTask([&] {
rtcp_transceiver->StopPeriodicTask();
rtcp_transceiver.reset();
done = true;
}));
});
ASSERT_TRUE(time_controller().Wait([&] { return done; }, kAlmostForever));
}
@ -415,18 +415,17 @@ TEST_F(RtcpTransceiverImplTest, SendsPeriodicRtcpWhenNetworkStateIsUp) {
absl::optional<RtcpTransceiverImpl> rtcp_transceiver;
rtcp_transceiver.emplace(config);
queue->PostTask(
ToQueuedTask([&] { rtcp_transceiver->SetReadyToSend(true); }));
queue->PostTask([&] { rtcp_transceiver->SetReadyToSend(true); });
EXPECT_TRUE(transport.WaitPacket());
// Cleanup.
bool done = false;
queue->PostTask(ToQueuedTask([&] {
queue->PostTask([&] {
rtcp_transceiver->StopPeriodicTask();
rtcp_transceiver.reset();
done = true;
}));
});
ASSERT_TRUE(time_controller().Wait([&] { return done; }, kAlmostForever));
}

14
modules/rtp_rtcp/source/rtp_rtcp_impl2.cc
View File

@ -22,7 +22,6 @@
#include "absl/strings/string_view.h"
#include "absl/types/optional.h"
#include "api/sequence_checker.h"
#include "api/task_queue/to_queued_task.h"
#include "api/transport/field_trial_based_config.h"
#include "api/units/time_delta.h"
#include "api/units/timestamp.h"
@ -52,13 +51,6 @@ RTCPSender::Configuration AddRtcpSendEvaluationCallback(
return config;
}
int DelayMillisForDuration(TimeDelta duration) {
// TimeDelta::ms() rounds downwards sometimes which leads to too little time
// slept. Account for this, unless `duration` is exactly representable in
// millisecs.
return (duration.us() + rtc::kNumMillisecsPerSec - 1) /
rtc::kNumMicrosecsPerMillisec;
}
} // namespace
ModuleRtpRtcpImpl2::RtpSenderContext::RtpSenderContext(
@ -805,7 +797,7 @@ void ModuleRtpRtcpImpl2::ScheduleRtcpSendEvaluation(TimeDelta duration) {
// than the worker queue on which it's created on implies that external
// synchronization is present and removes this activity before destruction.
if (duration.IsZero()) {
worker_queue_->PostTask(ToQueuedTask(task_safety_, [this] {
worker_queue_->PostTask(SafeTask(task_safety_.flag(), [this] {
RTC_DCHECK_RUN_ON(worker_queue_);
MaybeSendRtcp();
}));
@ -823,12 +815,12 @@ void ModuleRtpRtcpImpl2::ScheduleMaybeSendRtcpAtOrAfterTimestamp(
// See note in ScheduleRtcpSendEvaluation about why `worker_queue_` can be
// accessed.
worker_queue_->PostDelayedTask(
ToQueuedTask(task_safety_,
SafeTask(task_safety_.flag(),
[this, execution_time] {
RTC_DCHECK_RUN_ON(worker_queue_);
MaybeSendRtcpAtOrAfterTimestamp(execution_time);
}),
DelayMillisForDuration(duration));
duration.RoundUpTo(TimeDelta::Millis(1)));
}
} // namespace webrtc

1
modules/rtp_rtcp/source/rtp_rtcp_impl2.h
View File

@ -25,7 +25,6 @@
#include "api/sequence_checker.h"
#include "api/task_queue/pending_task_safety_flag.h"
#include "api/task_queue/task_queue_base.h"
#include "api/task_queue/to_queued_task.h"
#include "api/units/time_delta.h"
#include "api/video/video_bitrate_allocation.h"
#include "modules/include/module_fec_types.h"

9
modules/rtp_rtcp/source/rtp_sender_egress.cc
View File

@ -16,7 +16,6 @@
#include <utility>
#include "absl/strings/match.h"
#include "api/task_queue/to_queued_task.h"
#include "api/transport/field_trial_based_config.h"
#include "logging/rtc_event_log/events/rtc_event_rtp_packet_outgoing.h"
#include "rtc_base/logging.h"
@ -159,15 +158,15 @@ void RtpSenderEgress::SendPacket(RtpPacketToSend* packet,
#if BWE_TEST_LOGGING_COMPILE_TIME_ENABLE
worker_queue_->PostTask(
ToQueuedTask(task_safety_, [this, now, packet_ssrc]() {
SafeTask(task_safety_.flag(), [this, now, packet_ssrc]() {
BweTestLoggingPlot(now.ms(), packet_ssrc);
}));
#endif
if (need_rtp_packet_infos_ &&
packet->packet_type() == RtpPacketToSend::Type::kVideo) {
worker_queue_->PostTask(ToQueuedTask(
task_safety_,
worker_queue_->PostTask(SafeTask(
task_safety_.flag(),
[this, packet_timestamp = packet->Timestamp(),
is_first_packet_of_frame = packet->is_first_packet_of_frame(),
is_last_packet_of_frame = packet->Marker(),
@ -294,7 +293,7 @@ void RtpSenderEgress::SendPacket(RtpPacketToSend* packet,
RtpPacketCounter counter(*packet);
size_t size = packet->size();
worker_queue_->PostTask(
ToQueuedTask(task_safety_, [this, now, packet_ssrc, packet_type,
SafeTask(task_safety_.flag(), [this, now, packet_ssrc, packet_type,
counter = std::move(counter), size]() {
RTC_DCHECK_RUN_ON(worker_queue_);
UpdateRtpStats(now.ms(), packet_ssrc, packet_type, std::move(counter),

1
modules/rtp_rtcp/source/rtp_sender_unittest.cc
View File

@ -15,7 +15,6 @@
#include "absl/strings/string_view.h"
#include "api/rtc_event_log/rtc_event.h"
#include "api/task_queue/to_queued_task.h"
#include "api/transport/field_trial_based_config.h"
#include "api/video/video_codec_constants.h"
#include "api/video/video_timing.h"

5
modules/rtp_rtcp/source/rtp_sender_video_frame_transformer_delegate.cc
View File

@ -14,7 +14,6 @@
#include <vector>
#include "absl/memory/memory.h"
#include "api/task_queue/to_queued_task.h"
#include "modules/rtp_rtcp/source/rtp_descriptor_authentication.h"
#include "modules/rtp_rtcp/source/rtp_sender_video.h"
@ -140,10 +139,10 @@ void RTPSenderVideoFrameTransformerDelegate::OnTransformedFrame(
if (!sender_ || !encoder_queue_)
return;
rtc::scoped_refptr<RTPSenderVideoFrameTransformerDelegate> delegate(this);
encoder_queue_->PostTask(ToQueuedTask(
encoder_queue_->PostTask(
[delegate = std::move(delegate), frame = std::move(frame)]() mutable {
delegate->SendVideo(std::move(frame));
}));
});
}
void RTPSenderVideoFrameTransformerDelegate::SendVideo(