From 77c47947ad098e4182a6244cb998e4fa8c7bd37e Mon Sep 17 00:00:00 2001 From: Per K Date: Tue, 11 Apr 2023 11:37:38 +0200 Subject: [PATCH] [WebRTC-SendPacketsOnWorkerThread] Cleanup VideoSendStream(Impl) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Commit-Queue: Per Kjellander Cr-Commit-Position: refs/heads/main@{#39814} --- video/video_send_stream.cc | 33 ++++------------ video/video_send_stream.h | 4 -- video/video_send_stream_impl.cc | 69 ++++++++++++++------------------- video/video_send_stream_impl.h | 27 +++++++------ 4 files changed, 50 insertions(+), 83 deletions(-) diff --git a/video/video_send_stream.cc b/video/video_send_stream.cc index e5545e761c..e95ca260a4 100644 --- a/video/video_send_stream.cc +++ b/video/video_send_stream.cc @@ -149,8 +149,7 @@ VideoSendStream::VideoSendStream( const std::map& suspended_payload_states, std::unique_ptr 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 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& rids) { + RTC_DCHECK_RUN_ON(&thread_checker_); // Map rids to layers. If rids is empty, generate a keyframe for all layers. std::vector next_frames(config_.rtp.ssrcs.size(), VideoFrameType::kVideoFrameKey); diff --git a/video/video_send_stream.h b/video/video_send_stream.h index a7ce112b21..7a3454a3a0 100644 --- a/video/video_send_stream.h +++ b/video/video_send_stream.h @@ -104,11 +104,7 @@ class VideoSendStream : public webrtc::VideoSendStream { absl::optional 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 transport_queue_safety_ = - PendingTaskSafetyFlag::CreateDetached(); SendStatisticsProxy stats_proxy_; const VideoSendStream::Config config_; diff --git a/video/video_send_stream_impl.cc b/video/video_send_stream_impl.cc index 5fa2af398d..597bf39198 100644 --- a/video/video_send_stream_impl.cc +++ b/video/video_send_stream_impl.cc @@ -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 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 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."; diff --git a/video/video_send_stream_impl.h b/video/video_send_stream_impl.h index f145450655..dc1bf89508 100644 --- a/video/video_send_stream_impl.h +++ b/video/video_send_stream_impl.h @@ -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 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 video_bitrate_allocation_context_ - RTC_GUARDED_BY(rtp_transport_queue_); + RTC_GUARDED_BY(thread_checker_); const absl::optional configured_pacing_factor_; }; } // namespace internal