Remove synchronization from VideoSendStream construction.

* Make VideoSendStream and VideoSendStreamImpl construction non-blocking.
* Move ownership of the rtp video sender to VideoSendStream.
* Most state is constructed in initializer lists.
* More state is now const (including VideoSendStreamImpl ptr)
* Adding thread checks to classes that appear to have had a race before
  E.g. RtpTransportControllerSend. The change in threading now actually
  fixes an issue we weren't aware of.
* Moved from using weak_ptr to safety flag and made some PostTask calls
  cancellable that could potentially have been problematic. Initalizing
  the flag without thread synchronization is also simpler.

This should speed up renegotiation significantly when there are
multiple channels. A follow-up change will improve SetSend as well
which is another costly step during renegotiation.

Bug: webrtc:12840
Change-Id: If4b28da5a085643ce132c7cfcf80a62cd1a625c5
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/221105
Commit-Queue: Tommi <tommi@webrtc.org>
Reviewed-by: Markus Handell <handellm@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#34224}

call/call.cc

@@ -1108,9 +1108,12 @@ void Call::DestroyVideoSendStream(webrtc::VideoSendStream* send_stream) {
   RTC_DCHECK(send_stream != nullptr);
   RTC_DCHECK_RUN_ON(worker_thread_);
 
-  send_stream->Stop();
+  VideoSendStream* send_stream_impl =
-
+      static_cast<VideoSendStream*>(send_stream);
-  VideoSendStream* send_stream_impl = nullptr;
+  VideoSendStream::RtpStateMap rtp_states;
+  VideoSendStream::RtpPayloadStateMap rtp_payload_states;
+  send_stream_impl->StopPermanentlyAndGetRtpStates(&rtp_states,
+                                                   &rtp_payload_states);
 
   auto it = video_send_ssrcs_.begin();
   while (it != video_send_ssrcs_.end()) {
@@ -1121,6 +1124,7 @@ void Call::DestroyVideoSendStream(webrtc::VideoSendStream* send_stream) {
       ++it;
     }
   }
+
   // Stop forwarding resources to the stream being destroyed.
   for (const auto& resource_forwarder : adaptation_resource_forwarders_) {
     resource_forwarder->OnDestroyVideoSendStream(send_stream_impl);
@@ -1129,12 +1133,6 @@ void Call::DestroyVideoSendStream(webrtc::VideoSendStream* send_stream) {
   if (video_send_streams_.empty())
     video_send_streams_empty_.store(true, std::memory_order_relaxed);
 
-  RTC_CHECK(send_stream_impl != nullptr);
-
-  VideoSendStream::RtpStateMap rtp_states;
-  VideoSendStream::RtpPayloadStateMap rtp_payload_states;
-  send_stream_impl->StopPermanentlyAndGetRtpStates(&rtp_states,
-                                                   &rtp_payload_states);
   for (const auto& kv : rtp_states) {
     suspended_video_send_ssrcs_[kv.first] = kv.second;
   }
@@ -1143,6 +1141,8 @@ void Call::DestroyVideoSendStream(webrtc::VideoSendStream* send_stream) {
   }
 
   UpdateAggregateNetworkState();
+  // TODO(tommi): consider deleting on the same thread as runs
+  // StopPermanentlyAndGetRtpStates.
   delete send_stream_impl;
 }
 

call/rtp_transport_controller_send.cc

@@ -142,6 +142,7 @@ RtpTransportControllerSend::RtpTransportControllerSend(
 }
 
 RtpTransportControllerSend::~RtpTransportControllerSend() {
+  RTC_DCHECK(video_rtp_senders_.empty());
   process_thread_->Stop();
 }
 
@@ -156,6 +157,7 @@ RtpVideoSenderInterface* RtpTransportControllerSend::CreateRtpVideoSender(
     std::unique_ptr<FecController> fec_controller,
     const RtpSenderFrameEncryptionConfig& frame_encryption_config,
     rtc::scoped_refptr<FrameTransformerInterface> frame_transformer) {
+  RTC_DCHECK_RUN_ON(&main_thread_);
   video_rtp_senders_.push_back(std::make_unique<RtpVideoSender>(
       clock_, suspended_ssrcs, states, rtp_config, rtcp_report_interval_ms,
       send_transport, observers,
@@ -169,6 +171,7 @@ RtpVideoSenderInterface* RtpTransportControllerSend::CreateRtpVideoSender(
 
 void RtpTransportControllerSend::DestroyRtpVideoSender(
     RtpVideoSenderInterface* rtp_video_sender) {
+  RTC_DCHECK_RUN_ON(&main_thread_);
   std::vector<std::unique_ptr<RtpVideoSenderInterface>>::iterator it =
       video_rtp_senders_.end();
   for (it = video_rtp_senders_.begin(); it != video_rtp_senders_.end(); ++it) {
@@ -354,6 +357,7 @@ void RtpTransportControllerSend::OnNetworkRouteChanged(
   }
 }
 void RtpTransportControllerSend::OnNetworkAvailability(bool network_available) {
+  RTC_DCHECK_RUN_ON(&main_thread_);
   RTC_LOG(LS_VERBOSE) << "SignalNetworkState "
                       << (network_available ? "Up" : "Down");
   NetworkAvailability msg;
@@ -470,6 +474,7 @@ RtpTransportControllerSend::ApplyOrLiftRelayCap(bool is_relayed) {
 
 void RtpTransportControllerSend::OnTransportOverheadChanged(
     size_t transport_overhead_bytes_per_packet) {
+  RTC_DCHECK_RUN_ON(&main_thread_);
   if (transport_overhead_bytes_per_packet >= kMaxOverheadBytes) {
     RTC_LOG(LS_ERROR) << "Transport overhead exceeds " << kMaxOverheadBytes;
     return;

call/rtp_transport_controller_send.h

@@ -18,6 +18,7 @@
 #include <vector>
 
 #include "api/network_state_predictor.h"
+#include "api/sequence_checker.h"
 #include "api/transport/network_control.h"
 #include "api/units/data_rate.h"
 #include "call/rtp_bitrate_configurator.h"
@@ -62,6 +63,7 @@ class RtpTransportControllerSend final
       const WebRtcKeyValueConfig* trials);
   ~RtpTransportControllerSend() override;
 
+  // TODO(tommi): Change to std::unique_ptr<>.
   RtpVideoSenderInterface* CreateRtpVideoSender(
       std::map<uint32_t, RtpState> suspended_ssrcs,
       const std::map<uint32_t, RtpPayloadState>&
@@ -148,8 +150,10 @@ class RtpTransportControllerSend final
 
   Clock* const clock_;
   RtcEventLog* const event_log_;
+  SequenceChecker main_thread_;
   PacketRouter packet_router_;
-  std::vector<std::unique_ptr<RtpVideoSenderInterface>> video_rtp_senders_;
+  std::vector<std::unique_ptr<RtpVideoSenderInterface>> video_rtp_senders_
+      RTC_GUARDED_BY(&main_thread_);
   RtpBitrateConfigurator bitrate_configurator_;
   std::map<std::string, rtc::NetworkRoute> network_routes_;
   bool pacer_started_;

modules/rtp_rtcp/source/rtp_rtcp_impl2.cc

@@ -651,7 +651,8 @@ void ModuleRtpRtcpImpl2::SetRemoteSSRC(const uint32_t ssrc) {
 }
 
 RtpSendRates ModuleRtpRtcpImpl2::GetSendRates() const {
-  RTC_DCHECK_RUN_ON(worker_queue_);
+  // Typically called on the `rtp_transport_queue_` owned by an
+  // RtpTransportControllerSendInterface instance.
   return rtp_sender_->packet_sender.GetSendRates();
 }
 

video/video_send_stream.cc

@@ -23,7 +23,6 @@
 #include "system_wrappers/include/clock.h"
 #include "system_wrappers/include/field_trial.h"
 #include "video/adaptation/overuse_frame_detector.h"
-#include "video/video_send_stream_impl.h"
 #include "video/video_stream_encoder.h"
 
 namespace webrtc {
@@ -80,6 +79,32 @@ GetBitrateAllocationCallbackType(const VideoSendStream::Config& config) {
       kVideoBitrateAllocationWhenScreenSharing;
 }
 
+RtpSenderFrameEncryptionConfig CreateFrameEncryptionConfig(
+    const VideoSendStream::Config* config) {
+  RtpSenderFrameEncryptionConfig frame_encryption_config;
+  frame_encryption_config.frame_encryptor = config->frame_encryptor;
+  frame_encryption_config.crypto_options = config->crypto_options;
+  return frame_encryption_config;
+}
+
+RtpSenderObservers CreateObservers(RtcpRttStats* call_stats,
+                                   EncoderRtcpFeedback* encoder_feedback,
+                                   SendStatisticsProxy* stats_proxy,
+                                   SendDelayStats* send_delay_stats) {
+  RtpSenderObservers observers;
+  observers.rtcp_rtt_stats = call_stats;
+  observers.intra_frame_callback = encoder_feedback;
+  observers.rtcp_loss_notification_observer = encoder_feedback;
+  observers.report_block_data_observer = stats_proxy;
+  observers.rtp_stats = stats_proxy;
+  observers.bitrate_observer = stats_proxy;
+  observers.frame_count_observer = stats_proxy;
+  observers.rtcp_type_observer = stats_proxy;
+  observers.send_delay_observer = stats_proxy;
+  observers.send_packet_observer = send_delay_stats;
+  return observers;
+}
+
 }  // namespace
 
 namespace internal {
@@ -100,45 +125,64 @@ VideoSendStream::VideoSendStream(
     const std::map<uint32_t, RtpPayloadState>& suspended_payload_states,
     std::unique_ptr<FecController> fec_controller)
     : rtp_transport_queue_(transport->GetWorkerQueue()),
+      transport_(transport),
       stats_proxy_(clock, config, encoder_config.content_type),
       config_(std::move(config)),
-      content_type_(encoder_config.content_type) {
+      content_type_(encoder_config.content_type),
+      video_stream_encoder_(std::make_unique<VideoStreamEncoder>(
+          clock,
+          num_cpu_cores,
+          &stats_proxy_,
+          config_.encoder_settings,
+          std::make_unique<OveruseFrameDetector>(&stats_proxy_),
+          task_queue_factory,
+          GetBitrateAllocationCallbackType(config_))),
+      encoder_feedback_(
+          clock,
+          config_.rtp.ssrcs,
+          video_stream_encoder_.get(),
+          [this](uint32_t ssrc, const std::vector<uint16_t>& seq_nums) {
+            return rtp_video_sender_->GetSentRtpPacketInfos(ssrc, seq_nums);
+          }),
+      rtp_video_sender_(
+          transport->CreateRtpVideoSender(suspended_ssrcs,
+                                          suspended_payload_states,
+                                          config_.rtp,
+                                          config_.rtcp_report_interval_ms,
+                                          config_.send_transport,
+                                          CreateObservers(call_stats,
+                                                          &encoder_feedback_,
+                                                          &stats_proxy_,
+                                                          send_delay_stats),
+                                          event_log,
+                                          std::move(fec_controller),
+                                          CreateFrameEncryptionConfig(&config_),
+                                          config_.frame_transformer)),
+      send_stream_(clock,
+                   &stats_proxy_,
+                   rtp_transport_queue_,
+                   transport,
+                   bitrate_allocator,
+                   video_stream_encoder_.get(),
+                   &config_,
+                   encoder_config.max_bitrate_bps,
+                   encoder_config.bitrate_priority,
+                   encoder_config.content_type,
+                   rtp_video_sender_) {
   RTC_DCHECK(config_.encoder_settings.encoder_factory);
   RTC_DCHECK(config_.encoder_settings.bitrate_allocator_factory);
 
-  video_stream_encoder_ = std::make_unique<VideoStreamEncoder>(
+  video_stream_encoder_->SetFecControllerOverride(rtp_video_sender_);
-      clock, num_cpu_cores, &stats_proxy_, config_.encoder_settings,
-      std::make_unique<OveruseFrameDetector>(&stats_proxy_), task_queue_factory,
-      GetBitrateAllocationCallbackType(config_));
 
-  // TODO(srte): Initialization should not be done posted on a task queue.
+  rtp_video_sender_->RegisterProcessThread(module_process_thread);
-  // Note that the posted task must not outlive this scope since the closure
-  // references local variables.
-  rtp_transport_queue_->PostTask(ToQueuedTask(
-      [this, clock, call_stats, transport, bitrate_allocator, send_delay_stats,
-       event_log, &suspended_ssrcs, &encoder_config, &suspended_payload_states,
-       &fec_controller]() {
-        send_stream_.reset(new VideoSendStreamImpl(
-            clock, &stats_proxy_, rtp_transport_queue_, call_stats, transport,
-            bitrate_allocator, send_delay_stats, video_stream_encoder_.get(),
-            event_log, &config_, encoder_config.max_bitrate_bps,
-            encoder_config.bitrate_priority, suspended_ssrcs,
-            suspended_payload_states, encoder_config.content_type,
-            std::move(fec_controller)));
-      },
-      [this]() { thread_sync_event_.Set(); }));
-
-  // Wait for ConstructionTask to complete so that |send_stream_| can be used.
-  // |module_process_thread| must be registered and deregistered on the thread
-  // it was created on.
-  thread_sync_event_.Wait(rtc::Event::kForever);
-  send_stream_->RegisterProcessThread(module_process_thread);
   ReconfigureVideoEncoder(std::move(encoder_config));
 }
 
 VideoSendStream::~VideoSendStream() {
   RTC_DCHECK_RUN_ON(&thread_checker_);
-  RTC_DCHECK(!send_stream_);
+  RTC_DCHECK(!running_);
+  rtp_video_sender_->DeRegisterProcessThread();
+  transport_->DestroyRtpVideoSender(rtp_video_sender_);
 }
 
 void VideoSendStream::UpdateActiveSimulcastLayers(
@@ -161,9 +205,8 @@ void VideoSendStream::UpdateActiveSimulcastLayers(
   RTC_LOG(LS_INFO) << "UpdateActiveSimulcastLayers: "
                    << active_layers_string.str();
 
-  VideoSendStreamImpl* send_stream = send_stream_.get();
+  rtp_transport_queue_->PostTask([this, active_layers] {
-  rtp_transport_queue_->PostTask([this, send_stream, active_layers] {
+    send_stream_.UpdateActiveSimulcastLayers(active_layers);
-    send_stream->UpdateActiveSimulcastLayers(active_layers);
     thread_sync_event_.Set();
   });
 
@@ -173,9 +216,13 @@ void VideoSendStream::UpdateActiveSimulcastLayers(
 void VideoSendStream::Start() {
   RTC_DCHECK_RUN_ON(&thread_checker_);
   RTC_DLOG(LS_INFO) << "VideoSendStream::Start";
-  VideoSendStreamImpl* send_stream = send_stream_.get();
+  if (running_)
-  rtp_transport_queue_->PostTask([this, send_stream] {
+    return;
-    send_stream->Start();
+
+  running_ = true;
+
+  rtp_transport_queue_->PostTask([this] {
+    send_stream_.Start();
     thread_sync_event_.Set();
   });
 
@@ -187,9 +234,11 @@ void VideoSendStream::Start() {
 
 void VideoSendStream::Stop() {
   RTC_DCHECK_RUN_ON(&thread_checker_);
+  if (!running_)
+    return;
   RTC_DLOG(LS_INFO) << "VideoSendStream::Stop";
-  VideoSendStreamImpl* send_stream = send_stream_.get();
+  running_ = false;
-  rtp_transport_queue_->PostTask([send_stream] { send_stream->Stop(); });
+  send_stream_.Stop();  // Stop() will proceed asynchronously.
 }
 
 void VideoSendStream::AddAdaptationResource(
@@ -227,7 +276,7 @@ VideoSendStream::Stats VideoSendStream::GetStats() {
 }
 
 absl::optional<float> VideoSendStream::GetPacingFactorOverride() const {
-  return send_stream_->configured_pacing_factor();
+  return send_stream_.configured_pacing_factor();
 }
 
 void VideoSendStream::StopPermanentlyAndGetRtpStates(
@@ -235,12 +284,15 @@ void VideoSendStream::StopPermanentlyAndGetRtpStates(
     VideoSendStream::RtpPayloadStateMap* payload_state_map) {
   RTC_DCHECK_RUN_ON(&thread_checker_);
   video_stream_encoder_->Stop();
-  send_stream_->DeRegisterProcessThread();
+
+  running_ = false;
+  // 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_->PostTask([this, rtp_state_map, payload_state_map]() {
-    send_stream_->Stop();
+    send_stream_.Stop();
-    *rtp_state_map = send_stream_->GetRtpStates();
+    *rtp_state_map = send_stream_.GetRtpStates();
-    *payload_state_map = send_stream_->GetRtpPayloadStates();
+    *payload_state_map = send_stream_.GetRtpPayloadStates();
-    send_stream_.reset();
     thread_sync_event_.Set();
   });
   thread_sync_event_.Wait(rtc::Event::kForever);
@@ -248,7 +300,7 @@ void VideoSendStream::StopPermanentlyAndGetRtpStates(
 
 void VideoSendStream::DeliverRtcp(const uint8_t* packet, size_t length) {
   // Called on a network thread.
-  send_stream_->DeliverRtcp(packet, length);
+  send_stream_.DeliverRtcp(packet, length);
 }
 
 }  // namespace internal

video/video_send_stream.h

@@ -22,9 +22,12 @@
 #include "call/video_receive_stream.h"
 #include "call/video_send_stream.h"
 #include "rtc_base/event.h"
+#include "rtc_base/system/no_unique_address.h"
 #include "rtc_base/task_queue.h"
+#include "video/encoder_rtcp_feedback.h"
 #include "video/send_delay_stats.h"
 #include "video/send_statistics_proxy.h"
+#include "video/video_send_stream_impl.h"
 
 namespace webrtc {
 namespace test {
@@ -45,8 +48,7 @@ class VideoSendStreamImpl;
 
 // VideoSendStream implements webrtc::VideoSendStream.
 // Internally, it delegates all public methods to VideoSendStreamImpl and / or
-// VideoStreamEncoder. VideoSendStreamInternal is created and deleted on
+// VideoStreamEncoder.
-// |worker_queue|.
 class VideoSendStream : public webrtc::VideoSendStream {
  public:
   using RtpStateMap = std::map<uint32_t, RtpState>;
@@ -97,15 +99,19 @@ class VideoSendStream : public webrtc::VideoSendStream {
 
   absl::optional<float> GetPacingFactorOverride() const;
 
-  SequenceChecker thread_checker_;
+  RTC_NO_UNIQUE_ADDRESS SequenceChecker thread_checker_;
   rtc::TaskQueue* const rtp_transport_queue_;
+  RtpTransportControllerSendInterface* const transport_;
   rtc::Event thread_sync_event_;
 
   SendStatisticsProxy stats_proxy_;
   const VideoSendStream::Config config_;
   const VideoEncoderConfig::ContentType content_type_;
-  std::unique_ptr<VideoSendStreamImpl> send_stream_;
   std::unique_ptr<VideoStreamEncoderInterface> video_stream_encoder_;
+  EncoderRtcpFeedback encoder_feedback_;
+  RtpVideoSenderInterface* const rtp_video_sender_ = nullptr;
+  VideoSendStreamImpl send_stream_;
+  bool running_ RTC_GUARDED_BY(thread_checker_) = false;
 };
 
 }  // namespace internal

video/video_send_stream_impl.cc

@@ -33,6 +33,7 @@
 #include "rtc_base/experiments/rate_control_settings.h"
 #include "rtc_base/logging.h"
 #include "rtc_base/numerics/safe_conversions.h"
+#include "rtc_base/task_utils/to_queued_task.h"
 #include "rtc_base/trace_event.h"
 #include "system_wrappers/include/clock.h"
 #include "system_wrappers/include/field_trial.h"
@@ -130,32 +131,6 @@ int CalculateMaxPadBitrateBps(const std::vector<VideoStream>& streams,
   return pad_up_to_bitrate_bps;
 }
 
-RtpSenderFrameEncryptionConfig CreateFrameEncryptionConfig(
-    const VideoSendStream::Config* config) {
-  RtpSenderFrameEncryptionConfig frame_encryption_config;
-  frame_encryption_config.frame_encryptor = config->frame_encryptor;
-  frame_encryption_config.crypto_options = config->crypto_options;
-  return frame_encryption_config;
-}
-
-RtpSenderObservers CreateObservers(RtcpRttStats* call_stats,
-                                   EncoderRtcpFeedback* encoder_feedback,
-                                   SendStatisticsProxy* stats_proxy,
-                                   SendDelayStats* send_delay_stats) {
-  RtpSenderObservers observers;
-  observers.rtcp_rtt_stats = call_stats;
-  observers.intra_frame_callback = encoder_feedback;
-  observers.rtcp_loss_notification_observer = encoder_feedback;
-  observers.report_block_data_observer = stats_proxy;
-  observers.rtp_stats = stats_proxy;
-  observers.bitrate_observer = stats_proxy;
-  observers.frame_count_observer = stats_proxy;
-  observers.rtcp_type_observer = stats_proxy;
-  observers.send_delay_observer = stats_proxy;
-  observers.send_packet_observer = send_delay_stats;
-  return observers;
-}
-
 absl::optional<AlrExperimentSettings> GetAlrSettings(
     VideoEncoderConfig::ContentType content_type) {
   if (content_type == VideoEncoderConfig::ContentType::kScreen) {
@@ -231,19 +206,14 @@ VideoSendStreamImpl::VideoSendStreamImpl(
     Clock* clock,
     SendStatisticsProxy* stats_proxy,
     rtc::TaskQueue* rtp_transport_queue,
-    RtcpRttStats* call_stats,
     RtpTransportControllerSendInterface* transport,
     BitrateAllocatorInterface* bitrate_allocator,
-    SendDelayStats* send_delay_stats,
     VideoStreamEncoderInterface* video_stream_encoder,
-    RtcEventLog* event_log,
     const VideoSendStream::Config* config,
     int initial_encoder_max_bitrate,
     double initial_encoder_bitrate_priority,
-    std::map<uint32_t, RtpState> suspended_ssrcs,
-    std::map<uint32_t, RtpPayloadState> suspended_payload_states,
     VideoEncoderConfig::ContentType content_type,
-    std::unique_ptr<FecController> fec_controller)
+    RtpVideoSenderInterface* rtp_video_sender)
     : clock_(clock),
       has_alr_probing_(config->periodic_alr_bandwidth_probing ||
                        GetAlrSettings(content_type)),
@@ -262,49 +232,35 @@ VideoSendStreamImpl::VideoSendStreamImpl(
       encoder_target_rate_bps_(0),
       encoder_bitrate_priority_(initial_encoder_bitrate_priority),
       video_stream_encoder_(video_stream_encoder),
-      encoder_feedback_(
-          clock,
-          config_->rtp.ssrcs,
-          video_stream_encoder,
-          [this](uint32_t ssrc, const std::vector<uint16_t>& seq_nums) {
-            return rtp_video_sender_->GetSentRtpPacketInfos(ssrc, seq_nums);
-          }),
       bandwidth_observer_(transport->GetBandwidthObserver()),
-      rtp_video_sender_(
+      rtp_video_sender_(rtp_video_sender),
-          transport_->CreateRtpVideoSender(suspended_ssrcs,
-                                           suspended_payload_states,
-                                           config_->rtp,
-                                           config_->rtcp_report_interval_ms,
-                                           config_->send_transport,
-                                           CreateObservers(call_stats,
-                                                           &encoder_feedback_,
-                                                           stats_proxy_,
-                                                           send_delay_stats),
-                                           event_log,
-                                           std::move(fec_controller),
-                                           CreateFrameEncryptionConfig(config_),
-                                           config->frame_transformer)),
-      weak_ptr_factory_(this),
       configured_pacing_factor_(
           GetConfiguredPacingFactor(*config_, content_type, pacing_config_)) {
-  RTC_DCHECK_RUN_ON(rtp_transport_queue_);
   RTC_DCHECK_GE(config_->rtp.payload_type, 0);
   RTC_DCHECK_LE(config_->rtp.payload_type, 127);
   RTC_DCHECK(!config_->rtp.ssrcs.empty());
   RTC_DCHECK(transport_);
   RTC_DCHECK_NE(initial_encoder_max_bitrate, 0);
-  RTC_LOG(LS_INFO) << "VideoSendStreamInternal: " << config_->ToString();
+  RTC_LOG(LS_INFO) << "VideoSendStreamImpl: " << config_->ToString();
 
   RTC_CHECK(AlrExperimentSettings::MaxOneFieldTrialEnabled());
 
-  weak_ptr_ = weak_ptr_factory_.GetWeakPtr();
+  // Only request rotation at the source when we positively know that the remote
-  video_stream_encoder->SetFecControllerOverride(rtp_video_sender_);
+  // side doesn't support the rotation extension. This allows us to prepare the
+  // encoder in the expectation that rotation is supported - which is the common
+  // case.
+  bool rotation_applied = absl::c_none_of(
+      config_->rtp.extensions, [](const RtpExtension& extension) {
+        return extension.uri == RtpExtension::kVideoRotationUri;
+      });
+
+  video_stream_encoder_->SetSink(this, rotation_applied);
 
   absl::optional<bool> enable_alr_bw_probing;
 
   // If send-side BWE is enabled, check if we should apply updated probing and
   // pacing settings.
-  if (configured_pacing_factor_.has_value()) {
+  if (configured_pacing_factor_) {
     absl::optional<AlrExperimentSettings> alr_settings =
         GetAlrSettings(content_type);
     int queue_time_limit_ms;
@@ -319,7 +275,6 @@ VideoSendStreamImpl::VideoSendStreamImpl(
     }
 
     transport->SetQueueTimeLimit(queue_time_limit_ms);
-    transport->SetPacingFactor(*configured_pacing_factor_);
   }
 
   if (config_->periodic_alr_bandwidth_probing) {
@@ -330,40 +285,18 @@ VideoSendStreamImpl::VideoSendStreamImpl(
     transport->EnablePeriodicAlrProbing(*enable_alr_bw_probing);
   }
 
+  rtp_transport_queue_->PostTask(ToQueuedTask(transport_queue_safety_, [this] {
+    if (configured_pacing_factor_)
+      transport_->SetPacingFactor(*configured_pacing_factor_);
+
     video_stream_encoder_->SetStartBitrate(
         bitrate_allocator_->GetStartBitrate(this));
+  }));
 }
 
 VideoSendStreamImpl::~VideoSendStreamImpl() {
-  RTC_DCHECK_RUN_ON(rtp_transport_queue_);
+  RTC_DCHECK_RUN_ON(&thread_checker_);
-  RTC_DCHECK(!rtp_video_sender_->IsActive())
+  RTC_LOG(LS_INFO) << "~VideoSendStreamImpl: " << config_->ToString();
-      << "VideoSendStreamImpl::Stop not called";
-  RTC_LOG(LS_INFO) << "~VideoSendStreamInternal: " << config_->ToString();
-  transport_->DestroyRtpVideoSender(rtp_video_sender_);
-}
-
-void VideoSendStreamImpl::RegisterProcessThread(
-    ProcessThread* module_process_thread) {
-  // Called on libjingle's worker thread (not rtp_transport_queue_), as part of
-  // the initialization steps. That's also the correct thread/queue for setting
-  // the state for |video_stream_encoder_|.
-
-  // Only request rotation at the source when we positively know that the remote
-  // side doesn't support the rotation extension. This allows us to prepare the
-  // encoder in the expectation that rotation is supported - which is the common
-  // case.
-  bool rotation_applied = absl::c_none_of(
-      config_->rtp.extensions, [](const RtpExtension& extension) {
-        return extension.uri == RtpExtension::kVideoRotationUri;
-      });
-
-  video_stream_encoder_->SetSink(this, rotation_applied);
-
-  rtp_video_sender_->RegisterProcessThread(module_process_thread);
-}
-
-void VideoSendStreamImpl::DeRegisterProcessThread() {
-  rtp_video_sender_->DeRegisterProcessThread();
 }
 
 void VideoSendStreamImpl::DeliverRtcp(const uint8_t* packet, size_t length) {
@@ -391,6 +324,7 @@ void VideoSendStreamImpl::Start() {
   RTC_LOG(LS_INFO) << "VideoSendStream::Start";
   if (rtp_video_sender_->IsActive())
     return;
+
   TRACE_EVENT_INSTANT0("webrtc", "VideoSendStream::Start");
   rtp_video_sender_->SetActive(true);
   StartupVideoSendStream();
@@ -398,6 +332,9 @@ void VideoSendStreamImpl::Start() {
 
 void VideoSendStreamImpl::StartupVideoSendStream() {
   RTC_DCHECK_RUN_ON(rtp_transport_queue_);
+
+  transport_queue_safety_->SetAlive();
+
   bitrate_allocator_->AddObserver(this, GetAllocationConfig());
   // Start monitoring encoder activity.
   {
@@ -427,21 +364,30 @@ void VideoSendStreamImpl::StartupVideoSendStream() {
 }
 
 void VideoSendStreamImpl::Stop() {
+  if (!rtp_transport_queue_->IsCurrent()) {
+    rtp_transport_queue_->PostTask(
+        ToQueuedTask(transport_queue_safety_, [this] { Stop(); }));
+    return;
+  }
   RTC_DCHECK_RUN_ON(rtp_transport_queue_);
   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_->SetActive(false);
   StopVideoSendStream();
 }
 
+// RTC_RUN_ON(rtp_transport_queue_)
 void VideoSendStreamImpl::StopVideoSendStream() {
   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() {
@@ -458,12 +404,9 @@ void VideoSendStreamImpl::SignalEncoderTimedOut() {
 void VideoSendStreamImpl::OnBitrateAllocationUpdated(
     const VideoBitrateAllocation& allocation) {
   if (!rtp_transport_queue_->IsCurrent()) {
-    auto ptr = weak_ptr_;
+    rtp_transport_queue_->PostTask(ToQueuedTask(transport_queue_safety_, [=] {
-    rtp_transport_queue_->PostTask([=] {
+      OnBitrateAllocationUpdated(allocation);
-      if (!ptr.get())
+    }));
-        return;
-      ptr->OnBitrateAllocationUpdated(allocation);
-    });
     return;
   }
 
@@ -535,14 +478,13 @@ void VideoSendStreamImpl::OnEncoderConfigurationChanged(
     VideoEncoderConfig::ContentType content_type,
     int min_transmit_bitrate_bps) {
   if (!rtp_transport_queue_->IsCurrent()) {
-    rtc::WeakPtr<VideoSendStreamImpl> send_stream = weak_ptr_;
+    rtp_transport_queue_->PostTask(ToQueuedTask(
-    rtp_transport_queue_->PostTask([send_stream, streams, is_svc, content_type,
+        transport_queue_safety_,
+        [this, streams = std::move(streams), is_svc, content_type,
          min_transmit_bitrate_bps]() mutable {
-      if (send_stream) {
+          OnEncoderConfigurationChanged(std::move(streams), is_svc,
-        send_stream->OnEncoderConfigurationChanged(
+                                        content_type, min_transmit_bitrate_bps);
-            std::move(streams), is_svc, content_type, min_transmit_bitrate_bps);
+        }));
-      }
-    });
     return;
   }
 
@@ -618,7 +560,8 @@ EncodedImageCallback::Result VideoSendStreamImpl::OnEncodedImage(
     }
   };
   if (!rtp_transport_queue_->IsCurrent()) {
-    rtp_transport_queue_->PostTask(enable_padding_task);
+    rtp_transport_queue_->PostTask(
+        ToQueuedTask(transport_queue_safety_, std::move(enable_padding_task)));
   } else {
     enable_padding_task();
   }
@@ -628,18 +571,16 @@ EncodedImageCallback::Result VideoSendStreamImpl::OnEncodedImage(
       rtp_video_sender_->OnEncodedImage(encoded_image, codec_specific_info);
   // Check if there's a throttled VideoBitrateAllocation that we should try
   // sending.
-  rtc::WeakPtr<VideoSendStreamImpl> send_stream = weak_ptr_;
+  auto update_task = [this]() {
-  auto update_task = [send_stream]() {
+    RTC_DCHECK_RUN_ON(rtp_transport_queue_);
-    if (send_stream) {
+    auto& context = video_bitrate_allocation_context_;
-      RTC_DCHECK_RUN_ON(send_stream->rtp_transport_queue_);
-      auto& context = send_stream->video_bitrate_allocation_context_;
     if (context && context->throttled_allocation) {
-        send_stream->OnBitrateAllocationUpdated(*context->throttled_allocation);
+      OnBitrateAllocationUpdated(*context->throttled_allocation);
-      }
     }
   };
   if (!rtp_transport_queue_->IsCurrent()) {
-    rtp_transport_queue_->PostTask(update_task);
+    rtp_transport_queue_->PostTask(
+        ToQueuedTask(transport_queue_safety_, std::move(update_task)));
   } else {
     update_task();
   }

video/video_send_stream_impl.h

@@ -19,8 +19,6 @@
 #include <vector>
 
 #include "absl/types/optional.h"
-#include "api/fec_controller.h"
-#include "api/rtc_event_log/rtc_event_log.h"
 #include "api/video/encoded_image.h"
 #include "api/video/video_bitrate_allocation.h"
 #include "api/video/video_bitrate_allocator.h"
@@ -33,18 +31,14 @@
 #include "call/rtp_video_sender_interface.h"
 #include "modules/include/module_common_types.h"
 #include "modules/rtp_rtcp/include/rtp_rtcp_defines.h"
-#include "modules/utility/include/process_thread.h"
 #include "modules/video_coding/include/video_codec_interface.h"
 #include "rtc_base/experiments/field_trial_parser.h"
-#include "rtc_base/synchronization/mutex.h"
+#include "rtc_base/system/no_unique_address.h"
 #include "rtc_base/task_queue.h"
+#include "rtc_base/task_utils/pending_task_safety_flag.h"
 #include "rtc_base/task_utils/repeating_task.h"
 #include "rtc_base/thread_annotations.h"
-#include "rtc_base/weak_ptr.h"
-#include "video/encoder_rtcp_feedback.h"
-#include "video/send_delay_stats.h"
 #include "video/send_statistics_proxy.h"
-#include "video/video_send_stream.h"
 
 namespace webrtc {
 namespace internal {
@@ -69,33 +63,19 @@ struct PacingConfig {
 class VideoSendStreamImpl : public webrtc::BitrateAllocatorObserver,
                             public VideoStreamEncoderInterface::EncoderSink {
  public:
-  VideoSendStreamImpl(
+  VideoSendStreamImpl(Clock* clock,
-      Clock* clock,
                       SendStatisticsProxy* stats_proxy,
                       rtc::TaskQueue* rtp_transport_queue,
-      RtcpRttStats* call_stats,
                       RtpTransportControllerSendInterface* transport,
                       BitrateAllocatorInterface* bitrate_allocator,
-      SendDelayStats* send_delay_stats,
                       VideoStreamEncoderInterface* video_stream_encoder,
-      RtcEventLog* event_log,
                       const VideoSendStream::Config* config,
                       int initial_encoder_max_bitrate,
                       double initial_encoder_bitrate_priority,
-      std::map<uint32_t, RtpState> suspended_ssrcs,
-      std::map<uint32_t, RtpPayloadState> suspended_payload_states,
                       VideoEncoderConfig::ContentType content_type,
-      std::unique_ptr<FecController> fec_controller);
+                      RtpVideoSenderInterface* rtp_video_sender);
   ~VideoSendStreamImpl() override;
 
-  // RegisterProcessThread register |module_process_thread| with those objects
-  // that use it. Registration has to happen on the thread were
-  // |module_process_thread| was created (libjingle's worker thread).
-  // TODO(perkj): Replace the use of |module_process_thread| with a TaskQueue,
-  // maybe |rtp_transport_queue|.
-  void RegisterProcessThread(ProcessThread* module_process_thread);
-  void DeRegisterProcessThread();
-
   void DeliverRtcp(const uint8_t* packet, size_t length);
   void UpdateActiveSimulcastLayers(const std::vector<bool> active_layers);
   void Start();
@@ -148,6 +128,8 @@ class VideoSendStreamImpl : public webrtc::BitrateAllocatorObserver,
   void SignalEncoderActive();
   MediaStreamAllocationConfig GetAllocationConfig() const
       RTC_RUN_ON(rtp_transport_queue_);
+
+  RTC_NO_UNIQUE_ADDRESS SequenceChecker thread_checker_;
   Clock* const clock_;
   const bool has_alr_probing_;
   const PacingConfig pacing_config_;
@@ -166,8 +148,6 @@ class VideoSendStreamImpl : public webrtc::BitrateAllocatorObserver,
   RtpTransportControllerSendInterface* const transport_;
   BitrateAllocatorInterface* const bitrate_allocator_;
 
-  Mutex ivf_writers_mutex_;
-
   bool disable_padding_;
   int max_padding_bitrate_;
   int encoder_min_bitrate_bps_;
@@ -176,18 +156,12 @@ class VideoSendStreamImpl : public webrtc::BitrateAllocatorObserver,
   double encoder_bitrate_priority_;
 
   VideoStreamEncoderInterface* const video_stream_encoder_;
-  EncoderRtcpFeedback encoder_feedback_;
 
   RtcpBandwidthObserver* const bandwidth_observer_;
   RtpVideoSenderInterface* const rtp_video_sender_;
 
-  // |weak_ptr_| to our self. This is used since we can not call
+  rtc::scoped_refptr<PendingTaskSafetyFlag> transport_queue_safety_ =
-  // |weak_ptr_factory_.GetWeakPtr| from multiple sequences but it is ok to copy
+      PendingTaskSafetyFlag::CreateDetached();
-  // an existing WeakPtr.
-  rtc::WeakPtr<VideoSendStreamImpl> weak_ptr_;
-  // |weak_ptr_factory_| must be declared last to make sure all WeakPtr's are
-  // invalidated before any other members are destroyed.
-  rtc::WeakPtrFactory<VideoSendStreamImpl> weak_ptr_factory_;
 
   // Context for the most recent and last sent video bitrate allocation. Used to
   // throttle sending of similar bitrate allocations.

video/video_send_stream_impl_unittest.cc

@@ -31,6 +31,7 @@
 #include "test/mock_transport.h"
 #include "video/call_stats.h"
 #include "video/test/mock_video_stream_encoder.h"
+#include "video/video_send_stream.h"
 
 namespace webrtc {
 
@@ -145,17 +146,24 @@ class VideoSendStreamImplTest : public ::testing::Test {
       int initial_encoder_max_bitrate,
       double initial_encoder_bitrate_priority,
       VideoEncoderConfig::ContentType content_type) {
+    RTC_DCHECK(!test_queue_.IsCurrent());
+
     EXPECT_CALL(bitrate_allocator_, GetStartBitrate(_))
         .WillOnce(Return(123000));
+
     std::map<uint32_t, RtpState> suspended_ssrcs;
     std::map<uint32_t, RtpPayloadState> suspended_payload_states;
-    return std::make_unique<VideoSendStreamImpl>(
+    auto ret = std::make_unique<VideoSendStreamImpl>(
-        &clock_, &stats_proxy_, &test_queue_, &call_stats_,
+        &clock_, &stats_proxy_, &test_queue_, &transport_controller_,
-        &transport_controller_, &bitrate_allocator_, &send_delay_stats_,
+        &bitrate_allocator_, &video_stream_encoder_, &config_,
-        &video_stream_encoder_, &event_log_, &config_,
         initial_encoder_max_bitrate, initial_encoder_bitrate_priority,
-        suspended_ssrcs, suspended_payload_states, content_type,
+        content_type, &rtp_video_sender_);
-        std::make_unique<FecControllerDefault>(&clock_));
+
+    // The call to GetStartBitrate() executes asynchronously on the tq.
+    test_queue_.WaitForPreviouslyPostedTasks();
+    testing::Mock::VerifyAndClearExpectations(&bitrate_allocator_);
+
+    return ret;
   }
 
  protected:
@@ -179,22 +187,22 @@ class VideoSendStreamImplTest : public ::testing::Test {
 };
 
 TEST_F(VideoSendStreamImplTest, RegistersAsBitrateObserverOnStart) {
-  test_queue_.SendTask(
-      [this] {
-        const bool kSuspend = false;
-        config_.suspend_below_min_bitrate = kSuspend;
   auto vss_impl = CreateVideoSendStreamImpl(
       kDefaultInitialBitrateBps, kDefaultBitratePriority,
       VideoEncoderConfig::ContentType::kRealtimeVideo);
+  const bool kSuspend = false;
+  config_.suspend_below_min_bitrate = kSuspend;
   EXPECT_CALL(bitrate_allocator_, AddObserver(vss_impl.get(), _))
-            .WillOnce(Invoke([&](BitrateAllocatorObserver*,
+      .WillOnce(Invoke(
-                                 MediaStreamAllocationConfig config) {
+          [&](BitrateAllocatorObserver*, MediaStreamAllocationConfig config) {
             EXPECT_EQ(config.min_bitrate_bps, 0u);
             EXPECT_EQ(config.max_bitrate_bps, kDefaultInitialBitrateBps);
             EXPECT_EQ(config.pad_up_bitrate_bps, 0u);
             EXPECT_EQ(config.enforce_min_bitrate, !kSuspend);
             EXPECT_EQ(config.bitrate_priority, kDefaultBitratePriority);
           }));
+  test_queue_.SendTask(
+      [&] {
         vss_impl->Start();
         EXPECT_CALL(bitrate_allocator_, RemoveObserver(vss_impl.get()))
             .Times(1);
@@ -204,15 +212,16 @@ TEST_F(VideoSendStreamImplTest, RegistersAsBitrateObserverOnStart) {
 }
 
 TEST_F(VideoSendStreamImplTest, UpdatesObserverOnConfigurationChange) {
-  test_queue_.SendTask(
-      [this] {
   const bool kSuspend = false;
   config_.suspend_below_min_bitrate = kSuspend;
-        config_.rtp.extensions.emplace_back(
+  config_.rtp.extensions.emplace_back(RtpExtension::kTransportSequenceNumberUri,
-            RtpExtension::kTransportSequenceNumberUri, 1);
+                                      1);
   auto vss_impl = CreateVideoSendStreamImpl(
       kDefaultInitialBitrateBps, kDefaultBitratePriority,
       VideoEncoderConfig::ContentType::kRealtimeVideo);
+
+  test_queue_.SendTask(
+      [&] {
         vss_impl->Start();
 
         // QVGA + VGA configuration matching defaults in
@@ -269,16 +278,16 @@ TEST_F(VideoSendStreamImplTest, UpdatesObserverOnConfigurationChange) {
 }
 
 TEST_F(VideoSendStreamImplTest, UpdatesObserverOnConfigurationChangeWithAlr) {
-  test_queue_.SendTask(
-      [this] {
   const bool kSuspend = false;
   config_.suspend_below_min_bitrate = kSuspend;
-        config_.rtp.extensions.emplace_back(
+  config_.rtp.extensions.emplace_back(RtpExtension::kTransportSequenceNumberUri,
-            RtpExtension::kTransportSequenceNumberUri, 1);
+                                      1);
   config_.periodic_alr_bandwidth_probing = true;
   auto vss_impl = CreateVideoSendStreamImpl(
       kDefaultInitialBitrateBps, kDefaultBitratePriority,
       VideoEncoderConfig::ContentType::kScreen);
+  test_queue_.SendTask(
+      [&] {
         vss_impl->Start();
 
         // Simulcast screenshare.
@@ -341,11 +350,12 @@ TEST_F(VideoSendStreamImplTest,
   test::ScopedFieldTrials hysteresis_experiment(
       "WebRTC-VideoRateControl/video_hysteresis:1.25/");
 
-  test_queue_.SendTask(
-      [this] {
   auto vss_impl = CreateVideoSendStreamImpl(
       kDefaultInitialBitrateBps, kDefaultBitratePriority,
       VideoEncoderConfig::ContentType::kRealtimeVideo);
+
+  test_queue_.SendTask(
+      [&] {
         vss_impl->Start();
 
         // 2-layer video simulcast.
@@ -401,17 +411,17 @@ TEST_F(VideoSendStreamImplTest,
 TEST_F(VideoSendStreamImplTest, SetsScreensharePacingFactorWithFeedback) {
   test::ScopedFieldTrials alr_experiment(GetAlrProbingExperimentString());
 
-  test_queue_.SendTask(
-      [this] {
   constexpr int kId = 1;
-        config_.rtp.extensions.emplace_back(
+  config_.rtp.extensions.emplace_back(RtpExtension::kTransportSequenceNumberUri,
-            RtpExtension::kTransportSequenceNumberUri, kId);
+                                      kId);
   EXPECT_CALL(transport_controller_,
               SetPacingFactor(kAlrProbingExperimentPaceMultiplier))
       .Times(1);
   auto vss_impl = CreateVideoSendStreamImpl(
       kDefaultInitialBitrateBps, kDefaultBitratePriority,
       VideoEncoderConfig::ContentType::kScreen);
+  test_queue_.SendTask(
+      [&] {
         vss_impl->Start();
         vss_impl->Stop();
       },
@@ -420,12 +430,12 @@ TEST_F(VideoSendStreamImplTest, SetsScreensharePacingFactorWithFeedback) {
 
 TEST_F(VideoSendStreamImplTest, DoesNotSetPacingFactorWithoutFeedback) {
   test::ScopedFieldTrials alr_experiment(GetAlrProbingExperimentString());
-  test_queue_.SendTask(
-      [this] {
-        EXPECT_CALL(transport_controller_, SetPacingFactor(_)).Times(0);
   auto vss_impl = CreateVideoSendStreamImpl(
       kDefaultInitialBitrateBps, kDefaultBitratePriority,
       VideoEncoderConfig::ContentType::kScreen);
+  test_queue_.SendTask(
+      [&] {
+        EXPECT_CALL(transport_controller_, SetPacingFactor(_)).Times(0);
         vss_impl->Start();
         vss_impl->Stop();
       },
@@ -433,12 +443,12 @@ TEST_F(VideoSendStreamImplTest, DoesNotSetPacingFactorWithoutFeedback) {
 }
 
 TEST_F(VideoSendStreamImplTest, ForwardsVideoBitrateAllocationWhenEnabled) {
-  test_queue_.SendTask(
-      [this] {
-        EXPECT_CALL(transport_controller_, SetPacingFactor(_)).Times(0);
   auto vss_impl = CreateVideoSendStreamImpl(
       kDefaultInitialBitrateBps, kDefaultBitratePriority,
       VideoEncoderConfig::ContentType::kScreen);
+  test_queue_.SendTask(
+      [&] {
+        EXPECT_CALL(transport_controller_, SetPacingFactor(_)).Times(0);
         VideoStreamEncoderInterface::EncoderSink* const sink =
             static_cast<VideoStreamEncoderInterface::EncoderSink*>(
                 vss_impl.get());
@@ -483,11 +493,11 @@ TEST_F(VideoSendStreamImplTest, ForwardsVideoBitrateAllocationWhenEnabled) {
 }
 
 TEST_F(VideoSendStreamImplTest, ThrottlesVideoBitrateAllocationWhenTooSimilar) {
-  test_queue_.SendTask(
-      [this] {
   auto vss_impl = CreateVideoSendStreamImpl(
       kDefaultInitialBitrateBps, kDefaultBitratePriority,
       VideoEncoderConfig::ContentType::kScreen);
+  test_queue_.SendTask(
+      [&] {
         vss_impl->Start();
         // Unpause encoder, to allows allocations to be passed through.
         const uint32_t kBitrateBps = 100000;
@@ -529,8 +539,8 @@ TEST_F(VideoSendStreamImplTest, ThrottlesVideoBitrateAllocationWhenTooSimilar) {
             .Times(1);
         sink->OnBitrateAllocationUpdated(updated_alloc);
 
-        // This is now a decrease compared to last forward allocation, forward
+        // This is now a decrease compared to last forward allocation,
-        // immediately.
+        // forward immediately.
         updated_alloc.SetBitrate(0, 0, base_layer_min_update_bitrate_bps - 1);
         EXPECT_CALL(rtp_video_sender_,
                     OnBitrateAllocationUpdated(updated_alloc))
@@ -543,11 +553,11 @@ TEST_F(VideoSendStreamImplTest, ThrottlesVideoBitrateAllocationWhenTooSimilar) {
 }
 
 TEST_F(VideoSendStreamImplTest, ForwardsVideoBitrateAllocationOnLayerChange) {
-  test_queue_.SendTask(
-      [this] {
   auto vss_impl = CreateVideoSendStreamImpl(
       kDefaultInitialBitrateBps, kDefaultBitratePriority,
       VideoEncoderConfig::ContentType::kScreen);
+  test_queue_.SendTask(
+      [&] {
         vss_impl->Start();
         // Unpause encoder, to allows allocations to be passed through.
         const uint32_t kBitrateBps = 100000;
@@ -572,8 +582,8 @@ TEST_F(VideoSendStreamImplTest, ForwardsVideoBitrateAllocationOnLayerChange) {
             .Times(1);
         sink->OnBitrateAllocationUpdated(alloc);
 
-        // Move some bitrate from one layer to a new one, but keep sum the same.
+        // Move some bitrate from one layer to a new one, but keep sum the
-        // Since layout has changed, immediately trigger forward.
+        // same. Since layout has changed, immediately trigger forward.
         VideoBitrateAllocation updated_alloc = alloc;
         updated_alloc.SetBitrate(2, 0, 10000);
         updated_alloc.SetBitrate(1, 1, alloc.GetBitrate(1, 1) - 10000);
@@ -589,11 +599,11 @@ TEST_F(VideoSendStreamImplTest, ForwardsVideoBitrateAllocationOnLayerChange) {
 }
 
 TEST_F(VideoSendStreamImplTest, ForwardsVideoBitrateAllocationAfterTimeout) {
-  test_queue_.SendTask(
-      [this] {
   auto vss_impl = CreateVideoSendStreamImpl(
       kDefaultInitialBitrateBps, kDefaultBitratePriority,
       VideoEncoderConfig::ContentType::kScreen);
+  test_queue_.SendTask(
+      [&] {
         vss_impl->Start();
         const uint32_t kBitrateBps = 100000;
         // Unpause encoder, to allows allocations to be passed through.
@@ -639,7 +649,8 @@ TEST_F(VideoSendStreamImplTest, ForwardsVideoBitrateAllocationAfterTimeout) {
         clock_.AdvanceTimeMicroseconds(kMaxVbaThrottleTimeMs * 1000);
 
         {
-          // Sending similar allocation again after timeout, should forward.
+          // Sending similar allocation again after timeout, should
+          // forward.
           EXPECT_CALL(rtp_video_sender_, OnBitrateAllocationUpdated(alloc))
               .Times(1);
           sink->OnBitrateAllocationUpdated(alloc);
@@ -661,8 +672,8 @@ TEST_F(VideoSendStreamImplTest, ForwardsVideoBitrateAllocationAfterTimeout) {
         }
 
         {
-          // Advance time and send encoded image, this should wake up and send
+          // Advance time and send encoded image, this should wake up and
-          // cached bitrate allocation.
+          // send cached bitrate allocation.
           clock_.AdvanceTimeMicroseconds(kMaxVbaThrottleTimeMs * 1000);
           EXPECT_CALL(rtp_video_sender_, OnBitrateAllocationUpdated(alloc))
               .Times(1);
@@ -671,8 +682,8 @@ TEST_F(VideoSendStreamImplTest, ForwardsVideoBitrateAllocationAfterTimeout) {
         }
 
         {
-          // Advance time and send encoded image, there should be no cached
+          // Advance time and send encoded image, there should be no
-          // allocation to send.
+          // cached allocation to send.
           clock_.AdvanceTimeMicroseconds(kMaxVbaThrottleTimeMs * 1000);
           EXPECT_CALL(rtp_video_sender_, OnBitrateAllocationUpdated(alloc))
               .Times(0);
@@ -686,15 +697,15 @@ TEST_F(VideoSendStreamImplTest, ForwardsVideoBitrateAllocationAfterTimeout) {
 }
 
 TEST_F(VideoSendStreamImplTest, CallsVideoStreamEncoderOnBitrateUpdate) {
-  test_queue_.SendTask(
-      [this] {
   const bool kSuspend = false;
   config_.suspend_below_min_bitrate = kSuspend;
-        config_.rtp.extensions.emplace_back(
+  config_.rtp.extensions.emplace_back(RtpExtension::kTransportSequenceNumberUri,
-            RtpExtension::kTransportSequenceNumberUri, 1);
+                                      1);
   auto vss_impl = CreateVideoSendStreamImpl(
       kDefaultInitialBitrateBps, kDefaultBitratePriority,
       VideoEncoderConfig::ContentType::kRealtimeVideo);
+  test_queue_.SendTask(
+      [&] {
         vss_impl->Start();
 
         VideoStream qvga_stream;
@@ -733,8 +744,8 @@ TEST_F(VideoSendStreamImplTest, CallsVideoStreamEncoderOnBitrateUpdate) {
         static_cast<BitrateAllocatorObserver*>(vss_impl.get())
             ->OnBitrateUpdated(update);
 
-        // Test allocation where the link allocation is larger than the target,
+        // Test allocation where the link allocation is larger than the
-        // meaning we have some headroom on the link.
+        // target, meaning we have some headroom on the link.
         const DataRate qvga_max_bitrate =
             DataRate::BitsPerSec(qvga_stream.max_bitrate_bps);
         const DataRate headroom = DataRate::BitsPerSec(50000);
@@ -750,8 +761,8 @@ TEST_F(VideoSendStreamImplTest, CallsVideoStreamEncoderOnBitrateUpdate) {
         static_cast<BitrateAllocatorObserver*>(vss_impl.get())
             ->OnBitrateUpdated(update);
 
-        // Add protection bitrate to the mix, this should be subtracted from the
+        // Add protection bitrate to the mix, this should be subtracted
-        // headroom.
+        // from the headroom.
         const uint32_t protection_bitrate_bps = 10000;
         EXPECT_CALL(rtp_video_sender_, GetProtectionBitrateBps())
             .WillOnce(Return(protection_bitrate_bps));
@@ -791,14 +802,11 @@ TEST_F(VideoSendStreamImplTest, CallsVideoStreamEncoderOnBitrateUpdate) {
 
 TEST_F(VideoSendStreamImplTest, DisablesPaddingOnPausedEncoder) {
   int padding_bitrate = 0;
-  std::unique_ptr<VideoSendStreamImpl> vss_impl;
+  std::unique_ptr<VideoSendStreamImpl> vss_impl = CreateVideoSendStreamImpl(
-
-  test_queue_.SendTask(
-      [&] {
-        vss_impl = CreateVideoSendStreamImpl(
       kDefaultInitialBitrateBps, kDefaultBitratePriority,
       VideoEncoderConfig::ContentType::kRealtimeVideo);
-
+  test_queue_.SendTask(
+      [&] {
         // Capture padding bitrate for testing.
         EXPECT_CALL(bitrate_allocator_, AddObserver(vss_impl.get(), _))
             .WillRepeatedly(Invoke([&](BitrateAllocatorObserver*,
@@ -871,7 +879,6 @@ TEST_F(VideoSendStreamImplTest, DisablesPaddingOnPausedEncoder) {
         EXPECT_EQ(0, padding_bitrate);
         testing::Mock::VerifyAndClearExpectations(&bitrate_allocator_);
         vss_impl->Stop();
-        vss_impl.reset();
         done.Set();
       },
       5000);
@@ -881,12 +888,11 @@ TEST_F(VideoSendStreamImplTest, DisablesPaddingOnPausedEncoder) {
 }
 
 TEST_F(VideoSendStreamImplTest, KeepAliveOnDroppedFrame) {
-  std::unique_ptr<VideoSendStreamImpl> vss_impl;
+  std::unique_ptr<VideoSendStreamImpl> vss_impl = CreateVideoSendStreamImpl(
-  test_queue_.SendTask(
-      [&] {
-        vss_impl = CreateVideoSendStreamImpl(
       kDefaultInitialBitrateBps, kDefaultBitratePriority,
       VideoEncoderConfig::ContentType::kRealtimeVideo);
+  test_queue_.SendTask(
+      [&] {
         vss_impl->Start();
         const uint32_t kBitrateBps = 100000;
         EXPECT_CALL(rtp_video_sender_, GetPayloadBitrateBps())
@@ -909,7 +915,6 @@ TEST_F(VideoSendStreamImplTest, KeepAliveOnDroppedFrame) {
       [&] {
         testing::Mock::VerifyAndClearExpectations(&bitrate_allocator_);
         vss_impl->Stop();
-        vss_impl.reset();
         done.Set();
       },
       2000);
@@ -933,8 +938,6 @@ TEST_F(VideoSendStreamImplTest, ConfiguresBitratesForSvc) {
   }
 
   for (const TestConfig& test_config : test_variants) {
-    test_queue_.SendTask(
-        [this, test_config] {
     const bool kSuspend = false;
     config_.suspend_below_min_bitrate = kSuspend;
     config_.rtp.extensions.emplace_back(
@@ -945,6 +948,8 @@ TEST_F(VideoSendStreamImplTest, ConfiguresBitratesForSvc) {
         test_config.screenshare
             ? VideoEncoderConfig::ContentType::kScreen
             : VideoEncoderConfig::ContentType::kRealtimeVideo);
+    test_queue_.SendTask(
+        [&] {
           vss_impl->Start();
 
           // Svc