admin/webrtc_m130
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[WebRTC-SendPacketsOnWorkerThread] Cleanup VideoSendStream(Impl)

Browse Source 
Cleanup and remove usage of MaybeWorkerThread from VideoSendStream.
VideoSendStream is now created and lives on the worker thread.

Bug: webrtc:14502
Change-Id: I81ccf6b9fc6e8889db81b09bd4a75a3831a003e2
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/300842
Reviewed-by: Erik Språng <sprang@webrtc.org>
Commit-Queue: Per Kjellander <perkj@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#39814}
This commit is contained in:
Per K 2023-04-11 11:37:38 +02:00 committed by WebRTC LUCI CQ
parent 401c14aaf6
commit 77c47947ad
4 changed files with 50 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
video/video_send_stream.cc
View File

@ -149,8 +149,7 @@ VideoSendStream::VideoSendStream(
const std::map<uint32_t, RtpPayloadState>& suspended_payload_states,
std::unique_ptr<FecController> fec_controller,
const FieldTrialsView& field_trials)
: rtp_transport_queue_(transport->GetWorkerQueue()),
transport_(transport),
: transport_(transport),
stats_proxy_(clock, config, encoder_config.content_type, field_trials),
config_(std::move(config)),
content_type_(encoder_config.content_type),
@ -237,12 +236,7 @@ void VideoSendStream::StartPerRtpStream(const std::vector<bool> active_layers) {
}
active_layers_string << "}";
RTC_LOG(LS_INFO) << "StartPerRtpStream: " << active_layers_string.str();
rtp_transport_queue_->RunOrPost(
SafeTask(transport_queue_safety_, [this, active_layers] {
send_stream_.StartPerRtpStream(active_layers);
}));
send_stream_.StartPerRtpStream(active_layers);
running_ = running;
}
@ -252,13 +246,7 @@ void VideoSendStream::Stop() {
return;
RTC_DLOG(LS_INFO) << "VideoSendStream::Stop";
running_ = false;
rtp_transport_queue_->RunOrPost(SafeTask(transport_queue_safety_, [this] {
// As the stream can get re-used and implicitly restarted via changing
// the state of the active layers, we do not mark the
// `transport_queue_safety_` flag with `SetNotAlive()` here. That's only
// done when we stop permanently via `StopPermanentlyAndGetRtpStates()`.
send_stream_.Stop();
}));
send_stream_.Stop();
}
bool VideoSendStream::started() {
@ -300,9 +288,7 @@ void VideoSendStream::ReconfigureVideoEncoder(VideoEncoderConfig config,
}
VideoSendStream::Stats VideoSendStream::GetStats() {
// TODO(perkj, solenberg): Some test cases in EndToEndTest call GetStats from
// a network thread. See comment in Call::GetStats().
// RTC_DCHECK_RUN_ON(&thread_checker_);
RTC_DCHECK_RUN_ON(&thread_checker_);
return stats_proxy_.GetStats();
}
@ -320,13 +306,9 @@ void VideoSendStream::StopPermanentlyAndGetRtpStates(
// Always run these cleanup steps regardless of whether running_ was set
// or not. This will unregister callbacks before destruction.
// See `VideoSendStreamImpl::StopVideoSendStream` for more.
rtp_transport_queue_->RunSynchronous(
[this, rtp_state_map, payload_state_map]() {
transport_queue_safety_->SetNotAlive();
send_stream_.Stop();
*rtp_state_map = send_stream_.GetRtpStates();
*payload_state_map = send_stream_.GetRtpPayloadStates();
});
send_stream_.Stop();
*rtp_state_map = send_stream_.GetRtpStates();
*payload_state_map = send_stream_.GetRtpPayloadStates();
}
void VideoSendStream::DeliverRtcp(const uint8_t* packet, size_t length) {
@ -335,6 +317,7 @@ void VideoSendStream::DeliverRtcp(const uint8_t* packet, size_t length) {
}
void VideoSendStream::GenerateKeyFrame(const std::vector<std::string>& rids) {
RTC_DCHECK_RUN_ON(&thread_checker_);
// Map rids to layers. If rids is empty, generate a keyframe for all layers.
std::vector<VideoFrameType> next_frames(config_.rtp.ssrcs.size(),
VideoFrameType::kVideoFrameKey);

4
video/video_send_stream.h
View File

@ -104,11 +104,7 @@ class VideoSendStream : public webrtc::VideoSendStream {
absl::optional<float> GetPacingFactorOverride() const;
RTC_NO_UNIQUE_ADDRESS SequenceChecker thread_checker_;
MaybeWorkerThread* const rtp_transport_queue_;
RtpTransportControllerSendInterface* const transport_;
rtc::Event thread_sync_event_;
rtc::scoped_refptr<PendingTaskSafetyFlag> transport_queue_safety_ =
PendingTaskSafetyFlag::CreateDetached();
SendStatisticsProxy stats_proxy_;
const VideoSendStream::Config config_;

69
video/video_send_stream_impl.cc
View File

@ -233,7 +233,7 @@ VideoSendStreamImpl::VideoSendStreamImpl(
pacing_config_(PacingConfig(field_trials)),
stats_proxy_(stats_proxy),
config_(config),
rtp_transport_queue_(transport->GetWorkerQueue()),
worker_queue_(TaskQueueBase::Current()),
timed_out_(false),
transport_(transport),
bitrate_allocator_(bitrate_allocator),
@ -298,33 +298,26 @@ VideoSendStreamImpl::VideoSendStreamImpl(
transport->EnablePeriodicAlrProbing(*enable_alr_bw_probing);
}
rtp_transport_queue_->RunOrPost(SafeTask(transport_queue_safety_, [this] {
if (configured_pacing_factor_)
transport_->SetPacingFactor(*configured_pacing_factor_);
if (configured_pacing_factor_)
transport_->SetPacingFactor(*configured_pacing_factor_);
video_stream_encoder_->SetStartBitrate(
bitrate_allocator_->GetStartBitrate(this));
}));
video_stream_encoder_->SetStartBitrate(
bitrate_allocator_->GetStartBitrate(this));
}
VideoSendStreamImpl::~VideoSendStreamImpl() {
RTC_DCHECK_RUN_ON(&thread_checker_);
RTC_LOG(LS_INFO) << "~VideoSendStreamImpl: " << config_->ToString();
// TODO(webrtc:14502): Change `transport_queue_safety_` to be of type
// ScopedTaskSafety if experiment WebRTC-SendPacketsOnWorkerThread succeed.
if (rtp_transport_queue_->IsCurrent()) {
transport_queue_safety_->SetNotAlive();
}
}
void VideoSendStreamImpl::DeliverRtcp(const uint8_t* packet, size_t length) {
// Runs on a worker thread.
RTC_DCHECK_RUN_ON(&thread_checker_);
rtp_video_sender_->DeliverRtcp(packet, length);
}
void VideoSendStreamImpl::StartPerRtpStream(
const std::vector<bool> active_layers) {
RTC_DCHECK_RUN_ON(rtp_transport_queue_);
RTC_DCHECK_RUN_ON(&thread_checker_);
bool previously_active = rtp_video_sender_->IsActive();
rtp_video_sender_->SetActiveModules(active_layers);
if (!rtp_video_sender_->IsActive() && previously_active) {
@ -335,8 +328,7 @@ void VideoSendStreamImpl::StartPerRtpStream(
}
void VideoSendStreamImpl::StartupVideoSendStream() {
RTC_DCHECK_RUN_ON(rtp_transport_queue_);
transport_queue_safety_->SetAlive();
RTC_DCHECK_RUN_ON(&thread_checker_);
bitrate_allocator_->AddObserver(this, GetAllocationConfig());
// Start monitoring encoder activity.
@ -346,9 +338,8 @@ void VideoSendStreamImpl::StartupVideoSendStream() {
activity_ = false;
timed_out_ = false;
check_encoder_activity_task_ = RepeatingTaskHandle::DelayedStart(
rtp_transport_queue_->TaskQueueForDelayedTasks(), kEncoderTimeOut,
[this] {
RTC_DCHECK_RUN_ON(rtp_transport_queue_);
worker_queue_, kEncoderTimeOut, [this] {
RTC_DCHECK_RUN_ON(&thread_checker_);
if (!activity_) {
if (!timed_out_) {
SignalEncoderTimedOut();
@ -368,29 +359,27 @@ void VideoSendStreamImpl::StartupVideoSendStream() {
}
void VideoSendStreamImpl::Stop() {
RTC_DCHECK_RUN_ON(rtp_transport_queue_);
RTC_DCHECK_RUN_ON(&thread_checker_);
RTC_LOG(LS_INFO) << "VideoSendStreamImpl::Stop";
if (!rtp_video_sender_->IsActive())
return;
RTC_DCHECK(transport_queue_safety_->alive());
TRACE_EVENT_INSTANT0("webrtc", "VideoSendStream::Stop");
rtp_video_sender_->Stop();
StopVideoSendStream();
}
void VideoSendStreamImpl::StopVideoSendStream() {
RTC_DCHECK_RUN_ON(rtp_transport_queue_);
RTC_DCHECK_RUN_ON(&thread_checker_);
bitrate_allocator_->RemoveObserver(this);
check_encoder_activity_task_.Stop();
video_stream_encoder_->OnBitrateUpdated(DataRate::Zero(), DataRate::Zero(),
DataRate::Zero(), 0, 0, 0);
stats_proxy_->OnSetEncoderTargetRate(0);
transport_queue_safety_->SetNotAlive();
}
void VideoSendStreamImpl::SignalEncoderTimedOut() {
RTC_DCHECK_RUN_ON(rtp_transport_queue_);
RTC_DCHECK_RUN_ON(&thread_checker_);
// If the encoder has not produced anything the last kEncoderTimeOut and it
// is supposed to, deregister as BitrateAllocatorObserver. This can happen
// if a camera stops producing frames.
@ -403,9 +392,9 @@ void VideoSendStreamImpl::SignalEncoderTimedOut() {
void VideoSendStreamImpl::OnBitrateAllocationUpdated(
const VideoBitrateAllocation& allocation) {
// OnBitrateAllocationUpdated is invoked from the encoder task queue or
// the rtp_transport_queue_.
auto task = [=] {
RTC_DCHECK_RUN_ON(rtp_transport_queue_);
// the worker_queue_.
auto task = [this, allocation] {
RTC_DCHECK_RUN_ON(&thread_checker_);
if (encoder_target_rate_bps_ == 0) {
return;
}
@ -441,9 +430,9 @@ void VideoSendStreamImpl::OnBitrateAllocationUpdated(
// Send bitrate allocation metadata only if encoder is not paused.
rtp_video_sender_->OnBitrateAllocationUpdated(allocation);
};
if (!rtp_transport_queue_->IsCurrent()) {
rtp_transport_queue_->TaskQueueForPost()->PostTask(
SafeTask(transport_queue_safety_, std::move(task)));
if (!worker_queue_->IsCurrent()) {
worker_queue_->PostTask(
SafeTask(worker_queue_safety_.flag(), std::move(task)));
} else {
task();
}
@ -457,7 +446,7 @@ void VideoSendStreamImpl::OnVideoLayersAllocationUpdated(
}
void VideoSendStreamImpl::SignalEncoderActive() {
RTC_DCHECK_RUN_ON(rtp_transport_queue_);
RTC_DCHECK_RUN_ON(&thread_checker_);
if (rtp_video_sender_->IsActive()) {
RTC_LOG(LS_INFO) << "SignalEncoderActive, Encoder is active.";
bitrate_allocator_->AddObserver(this, GetAllocationConfig());
@ -480,12 +469,12 @@ void VideoSendStreamImpl::OnEncoderConfigurationChanged(
VideoEncoderConfig::ContentType content_type,
int min_transmit_bitrate_bps) {
// Currently called on the encoder TQ
RTC_DCHECK(!rtp_transport_queue_->IsCurrent());
RTC_DCHECK(!worker_queue_->IsCurrent());
auto closure = [this, streams = std::move(streams), is_svc, content_type,
min_transmit_bitrate_bps]() mutable {
RTC_DCHECK_GE(config_->rtp.ssrcs.size(), streams.size());
TRACE_EVENT0("webrtc", "VideoSendStream::OnEncoderConfigurationChanged");
RTC_DCHECK_RUN_ON(rtp_transport_queue_);
RTC_DCHECK_RUN_ON(&thread_checker_);
const VideoCodecType codec_type =
PayloadStringToCodecType(config_->rtp.payload_name);
@ -537,8 +526,8 @@ void VideoSendStreamImpl::OnEncoderConfigurationChanged(
}
};
rtp_transport_queue_->TaskQueueForPost()->PostTask(
SafeTask(transport_queue_safety_, std::move(closure)));
worker_queue_->PostTask(
SafeTask(worker_queue_safety_.flag(), std::move(closure)));
}
EncodedImageCallback::Result VideoSendStreamImpl::OnEncodedImage(
@ -550,10 +539,10 @@ EncodedImageCallback::Result VideoSendStreamImpl::OnEncodedImage(
// Indicate that there still is activity going on.
activity_ = true;
RTC_DCHECK(!rtp_transport_queue_->IsCurrent());
RTC_DCHECK(!worker_queue_->IsCurrent());
auto task_to_run_on_worker = [this]() {
RTC_DCHECK_RUN_ON(rtp_transport_queue_);
RTC_DCHECK_RUN_ON(&thread_checker_);
if (disable_padding_) {
disable_padding_ = false;
// To ensure that padding bitrate is propagated to the bitrate allocator.
@ -566,8 +555,8 @@ EncodedImageCallback::Result VideoSendStreamImpl::OnEncodedImage(
OnBitrateAllocationUpdated(*context->throttled_allocation);
}
};
rtp_transport_queue_->TaskQueueForPost()->PostTask(
SafeTask(transport_queue_safety_, std::move(task_to_run_on_worker)));
worker_queue_->PostTask(
SafeTask(worker_queue_safety_.flag(), std::move(task_to_run_on_worker)));
return rtp_video_sender_->OnEncodedImage(encoded_image, codec_specific_info);
}
@ -587,7 +576,7 @@ std::map<uint32_t, RtpPayloadState> VideoSendStreamImpl::GetRtpPayloadStates()
}
uint32_t VideoSendStreamImpl::OnBitrateUpdated(BitrateAllocationUpdate update) {
RTC_DCHECK_RUN_ON(rtp_transport_queue_);
RTC_DCHECK_RUN_ON(&thread_checker_);
RTC_DCHECK(rtp_video_sender_->IsActive())
<< "VideoSendStream::Start has not been called.";

27
video/video_send_stream_impl.h
View File

@ -121,14 +121,14 @@ class VideoSendStreamImpl : public webrtc::BitrateAllocatorObserver,
void StartupVideoSendStream();
// Removes the bitrate observer, stops monitoring and notifies the video
// encoder of the bitrate update.
void StopVideoSendStream() RTC_RUN_ON(rtp_transport_queue_);
void StopVideoSendStream() RTC_RUN_ON(thread_checker_);
void ConfigureProtection();
void ConfigureSsrcs();
void SignalEncoderTimedOut();
void SignalEncoderActive();
MediaStreamAllocationConfig GetAllocationConfig() const
RTC_RUN_ON(rtp_transport_queue_);
RTC_RUN_ON(thread_checker_);
RTC_NO_UNIQUE_ADDRESS SequenceChecker thread_checker_;
Clock* const clock_;
@ -138,31 +138,30 @@ class VideoSendStreamImpl : public webrtc::BitrateAllocatorObserver,
SendStatisticsProxy* const stats_proxy_;
const VideoSendStream::Config* const config_;
MaybeWorkerThread* const rtp_transport_queue_;
TaskQueueBase* const worker_queue_;
RepeatingTaskHandle check_encoder_activity_task_
RTC_GUARDED_BY(rtp_transport_queue_);
RTC_GUARDED_BY(thread_checker_);
std::atomic_bool activity_;
bool timed_out_ RTC_GUARDED_BY(rtp_transport_queue_);
bool timed_out_ RTC_GUARDED_BY(thread_checker_);
RtpTransportControllerSendInterface* const transport_;
BitrateAllocatorInterface* const bitrate_allocator_;
bool disable_padding_;
int max_padding_bitrate_;
int encoder_min_bitrate_bps_;
uint32_t encoder_max_bitrate_bps_;
uint32_t encoder_target_rate_bps_;
double encoder_bitrate_priority_;
bool disable_padding_ RTC_GUARDED_BY(thread_checker_);
int max_padding_bitrate_ RTC_GUARDED_BY(thread_checker_);
int encoder_min_bitrate_bps_ RTC_GUARDED_BY(thread_checker_);
uint32_t encoder_max_bitrate_bps_ RTC_GUARDED_BY(thread_checker_);
uint32_t encoder_target_rate_bps_ RTC_GUARDED_BY(thread_checker_);
double encoder_bitrate_priority_ RTC_GUARDED_BY(thread_checker_);
VideoStreamEncoderInterface* const video_stream_encoder_;
RtcpBandwidthObserver* const bandwidth_observer_;
RtpVideoSenderInterface* const rtp_video_sender_;
rtc::scoped_refptr<PendingTaskSafetyFlag> transport_queue_safety_ =
PendingTaskSafetyFlag::CreateDetached();
ScopedTaskSafety worker_queue_safety_;
// Context for the most recent and last sent video bitrate allocation. Used to
// throttle sending of similar bitrate allocations.
@ -172,7 +171,7 @@ class VideoSendStreamImpl : public webrtc::BitrateAllocatorObserver,
int64_t last_send_time_ms;
};
absl::optional<VbaSendContext> video_bitrate_allocation_context_
RTC_GUARDED_BY(rtp_transport_queue_);
RTC_GUARDED_BY(thread_checker_);
const absl::optional<float> configured_pacing_factor_;
};
} // namespace internal