Use SingleThreadedTaskQueue in DirectTransport

webrtc/video/video_send_stream_tests.cc

 /*
  *  Copyright (c) 2013 The WebRTC project authors. All Rights Reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */
 #include <algorithm>  // max
@@ skipping 49 matching lines @@
                               uint8_t retransmit_payload_type);
   void TestPacketFragmentationSize(VideoFormat format, bool with_fec);
 
   void TestVp9NonFlexMode(uint8_t num_temporal_layers,
                           uint8_t num_spatial_layers);
 
   void TestRequestSourceRotateVideo(bool support_orientation_ext);
 };
 
 TEST_F(VideoSendStreamTest, CanStartStartedStream) {
-  CreateSenderCall(Call::Config(event_log_.get()));
+  task_queue_.SendTask([this]() {
+    CreateSenderCall(Call::Config(event_log_.get()));
 
-  test::NullTransport transport;
-  CreateSendConfig(1, 0, 0, &transport);
-  CreateVideoStreams();
-  video_send_stream_->Start();
-  video_send_stream_->Start();
-  DestroyStreams();
+    test::NullTransport transport;
+    CreateSendConfig(1, 0, 0, &transport);
+    CreateVideoStreams();
+    video_send_stream_->Start();
+    video_send_stream_->Start();
+    DestroyStreams();
+    DestroyCalls();
+  });
 }
 
 TEST_F(VideoSendStreamTest, CanStopStoppedStream) {
-  CreateSenderCall(Call::Config(event_log_.get()));
+  task_queue_.SendTask([this]() {
+    CreateSenderCall(Call::Config(event_log_.get()));
 
-  test::NullTransport transport;
-  CreateSendConfig(1, 0, 0, &transport);
-  CreateVideoStreams();
-  video_send_stream_->Stop();
-  video_send_stream_->Stop();
-  DestroyStreams();
+    test::NullTransport transport;
+    CreateSendConfig(1, 0, 0, &transport);
+    CreateVideoStreams();
+    video_send_stream_->Stop();
+    video_send_stream_->Stop();
+    DestroyStreams();
+    DestroyCalls();
+  });
 }
 
 TEST_F(VideoSendStreamTest, SupportsCName) {
   static std::string kCName = "PjQatC14dGfbVwGPUOA9IH7RlsFDbWl4AhXEiDsBizo=";
   class CNameObserver : public test::SendTest {
    public:
     CNameObserver() : SendTest(kDefaultTimeoutMs) {}
 
    private:
     Action OnSendRtcp(const uint8_t* packet, size_t length) override {
@@ skipping 374 matching lines @@
 
     if (sent_media_ && sent_ulpfec_) {
       observation_complete_.Set();
     }
 
     prev_header_ = header;
 
     return SEND_PACKET;
   }
 
-  test::PacketTransport* CreateSendTransport(Call* sender_call) override {
+  test::PacketTransport* CreateSendTransport(
+      test::SingleThreadedTaskQueueForTesting* task_queue,
+      Call* sender_call) override {
     // At low RTT (< kLowRttNackMs) -> NACK only, no FEC.
     // Configure some network delay.
     const int kNetworkDelayMs = 100;
     FakeNetworkPipe::Config config;
     config.loss_percent = 5;
     config.queue_delay_ms = kNetworkDelayMs;
     return new test::PacketTransport(
-        sender_call, this, test::PacketTransport::kSender,
+        task_queue, sender_call, this, test::PacketTransport::kSender,
         VideoSendStreamTest::payload_type_map_, config);
   }
 
   void ModifyVideoConfigs(
       VideoSendStream::Config* send_config,
       std::vector<VideoReceiveStream::Config>* receive_configs,
       VideoEncoderConfig* encoder_config) override {
     if (use_nack_) {
       send_config->rtp.nack.rtp_history_ms =
           (*receive_configs)[0].rtp.nack.rtp_history_ms =
@@ skipping 127 matching lines @@
       EXPECT_TRUE(header.extension.hasTransportSequenceNumber);
     }
 
     if (sent_media_ && sent_flexfec_) {
       observation_complete_.Set();
     }
 
     return SEND_PACKET;
   }
 
-  test::PacketTransport* CreateSendTransport(Call* sender_call) override {
+  test::PacketTransport* CreateSendTransport(
+      test::SingleThreadedTaskQueueForTesting* task_queue,
+      Call* sender_call) override {
     // At low RTT (< kLowRttNackMs) -> NACK only, no FEC.
     // Therefore we need some network delay.
     const int kNetworkDelayMs = 100;
     FakeNetworkPipe::Config config;
     config.loss_percent = 5;
     config.queue_delay_ms = kNetworkDelayMs;
     return new test::PacketTransport(
-        sender_call, this, test::PacketTransport::kSender,
+        task_queue, sender_call, this, test::PacketTransport::kSender,
         VideoSendStreamTest::payload_type_map_, config);
   }
 
   void ModifyVideoConfigs(
       VideoSendStream::Config* send_config,
       std::vector<VideoReceiveStream::Config>* receive_configs,
       VideoEncoderConfig* encoder_config) override {
     if (use_nack_) {
       send_config->rtp.nack.rtp_history_ms =
           (*receive_configs)[0].rtp.nack.rtp_history_ms =
@@ skipping 672 matching lines @@
     Action OnSendRtp(const uint8_t* packet, size_t length) override {
       rtc::CritScope lock(&crit_);
 
       RTPHeader header;
       parser_->Parse(packet, length, &header);
       padding_length_ += header.paddingLength;
       total_length_ += length;
       return SEND_PACKET;
     }
 
-    test::PacketTransport* CreateSendTransport(Call* sender_call) override {
+    test::PacketTransport* CreateSendTransport(
+        test::SingleThreadedTaskQueueForTesting* task_queue,
+        Call* sender_call) override {
       const int kNetworkDelayMs = 50;
       FakeNetworkPipe::Config config;
       config.loss_percent = 10;
       config.link_capacity_kbps = kCapacityKbps;
       config.queue_delay_ms = kNetworkDelayMs;
-      return new test::PacketTransport(sender_call, this,
+      return new test::PacketTransport(task_queue, sender_call, this,
                                        test::PacketTransport::kSender,
                                        payload_type_map_, config);
     }
 
     void ModifyVideoConfigs(
         VideoSendStream::Config* send_config,
         std::vector<VideoReceiveStream::Config>* receive_configs,
         VideoEncoderConfig* encoder_config) override {
       // Turn on RTX.
       send_config->rtp.rtx.payload_type = kFakeVideoSendPayloadType;
@@ skipping 112 matching lines @@
 
   RunBaseTest(&test);
 }
 
 TEST_F(VideoSendStreamTest, ChangingNetworkRoute) {
   static const int kStartBitrateBps = 300000;
   static const int kNewMaxBitrateBps = 1234567;
   static const uint8_t kExtensionId = test::kTransportSequenceNumberExtensionId;
   class ChangingNetworkRouteTest : public test::EndToEndTest {
    public:
-    ChangingNetworkRouteTest()
-        : EndToEndTest(test::CallTest::kDefaultTimeoutMs), call_(nullptr) {
+    explicit ChangingNetworkRouteTest(
+        test::SingleThreadedTaskQueueForTesting* task_queue)
+        : EndToEndTest(test::CallTest::kDefaultTimeoutMs),
+          task_queue_(task_queue),
+          call_(nullptr) {
       EXPECT_TRUE(parser_->RegisterRtpHeaderExtension(
           kRtpExtensionTransportSequenceNumber, kExtensionId));
     }
 
     void OnCallsCreated(Call* sender_call, Call* receiver_call) override {
       call_ = sender_call;
     }
 
     void ModifyVideoConfigs(
         VideoSendStream::Config* send_config,
@@ skipping 20 matching lines @@
     Action OnSendRtp(const uint8_t* packet, size_t length) override {
       if (call_->GetStats().send_bandwidth_bps > kStartBitrateBps) {
         observation_complete_.Set();
       }
 
       return SEND_PACKET;
     }
 
     void PerformTest() override {
       rtc::NetworkRoute new_route(true, 10, 20, -1);
-      call_->OnNetworkRouteChanged("transport", new_route);
       Call::Config::BitrateConfig bitrate_config;
-      bitrate_config.start_bitrate_bps = kStartBitrateBps;
-      call_->SetBitrateConfig(bitrate_config);
+
+      task_queue_->SendTask([this, &new_route, &bitrate_config]() {
+        call_->OnNetworkRouteChanged("transport", new_route);
+        bitrate_config.start_bitrate_bps = kStartBitrateBps;
+        call_->SetBitrateConfig(bitrate_config);
+      });
+
       EXPECT_TRUE(Wait())
           << "Timed out while waiting for start bitrate to be exceeded.";
 
-      bitrate_config.start_bitrate_bps = -1;
-      bitrate_config.max_bitrate_bps = kNewMaxBitrateBps;
-      call_->SetBitrateConfig(bitrate_config);
-      // TODO(holmer): We should set the last sent packet id here and verify
-      // that we correctly ignore any packet loss reported prior to that id.
-      ++new_route.local_network_id;
-      call_->OnNetworkRouteChanged("transport", new_route);
-      EXPECT_GE(call_->GetStats().send_bandwidth_bps, kStartBitrateBps);
+      task_queue_->SendTask([this, &new_route, &bitrate_config]() {
+        bitrate_config.start_bitrate_bps = -1;
+        bitrate_config.max_bitrate_bps = kNewMaxBitrateBps;
+        call_->SetBitrateConfig(bitrate_config);
+        // TODO(holmer): We should set the last sent packet id here and verify
+        // that we correctly ignore any packet loss reported prior to that id.
+        ++new_route.local_network_id;
+        call_->OnNetworkRouteChanged("transport", new_route);
+        EXPECT_GE(call_->GetStats().send_bandwidth_bps, kStartBitrateBps);
+      });
     }
 
    private:
+    test::SingleThreadedTaskQueueForTesting* const task_queue_;
     Call* call_;
-  } test;
+  } test(&task_queue_);
 
   RunBaseTest(&test);
 }
 
 TEST_F(VideoSendStreamTest, ChangingTransportOverhead) {
   class ChangingTransportOverheadTest : public test::EndToEndTest {
    public:
-    ChangingTransportOverheadTest()
+    explicit ChangingTransportOverheadTest(
+        test::SingleThreadedTaskQueueForTesting* task_queue)
         : EndToEndTest(test::CallTest::kDefaultTimeoutMs),
+          task_queue_(task_queue),
           call_(nullptr),
           packets_sent_(0),
           transport_overhead_(0) {}
 
     void OnCallsCreated(Call* sender_call, Call* receiver_call) override {
       call_ = sender_call;
     }
 
     Action OnSendRtp(const uint8_t* packet, size_t length) override {
       EXPECT_LE(length, kMaxRtpPacketSize);
       rtc::CritScope cs(&lock_);
       if (++packets_sent_ < 100)
         return SEND_PACKET;
       observation_complete_.Set();
       return SEND_PACKET;
     }
 
     void ModifyVideoConfigs(
         VideoSendStream::Config* send_config,
         std::vector<VideoReceiveStream::Config>* receive_configs,
         VideoEncoderConfig* encoder_config) override {
       send_config->rtp.max_packet_size = kMaxRtpPacketSize;
     }
 
     void PerformTest() override {
-      transport_overhead_ = 100;
-      call_->OnTransportOverheadChanged(webrtc::MediaType::VIDEO,
-                                        transport_overhead_);
+      task_queue_->SendTask([this]() {
+        transport_overhead_ = 100;
+        call_->OnTransportOverheadChanged(webrtc::MediaType::VIDEO,
+                                          transport_overhead_);
+      });
+
       EXPECT_TRUE(Wait());
+
       {
         rtc::CritScope cs(&lock_);
         packets_sent_ = 0;
       }
-      transport_overhead_ = 500;
-      call_->OnTransportOverheadChanged(webrtc::MediaType::VIDEO,
-                                        transport_overhead_);
+
+      task_queue_->SendTask([this]() {
+        transport_overhead_ = 500;
+        call_->OnTransportOverheadChanged(webrtc::MediaType::VIDEO,
+                                          transport_overhead_);
+      });
+
       EXPECT_TRUE(Wait());
     }
 
    private:
+    test::SingleThreadedTaskQueueForTesting* const task_queue_;
     Call* call_;
     rtc::CriticalSection lock_;
     int packets_sent_ GUARDED_BY(lock_);
     int transport_overhead_;
     const size_t kMaxRtpPacketSize = 1000;
-  } test;
+  } test(&task_queue_);
 
   RunBaseTest(&test);
 }
 
 // Test class takes takes as argument a switch selecting if type switch should
 // occur and a function pointer to reset the send stream. This is necessary
 // since you cannot change the content type of a VideoSendStream, you need to
 // recreate it. Stopping and recreating the stream can only be done on the main
 // thread and in the context of VideoSendStreamTest (not BaseTest).
 template <typename T>
@@ skipping 103 matching lines @@
   auto reset_fun = [](const VideoSendStream::Config& send_stream_config,
                       const VideoEncoderConfig& encoder_config) {};
   MaxPaddingSetTest<decltype(reset_fun)> test(false, &reset_fun);
   RunBaseTest(&test);
 }
 
 TEST_F(VideoSendStreamTest, RespectsMinTransmitBitrateAfterContentSwitch) {
   // Function for removing and recreating the send stream with a new config.
   auto reset_fun = [this](const VideoSendStream::Config& send_stream_config,
                           const VideoEncoderConfig& encoder_config) {
-    Stop();
-    sender_call_->DestroyVideoSendStream(video_send_stream_);
-    video_send_config_ = send_stream_config.Copy();
-    video_encoder_config_ = encoder_config.Copy();
-    video_send_stream_ = sender_call_->CreateVideoSendStream(
-        video_send_config_.Copy(), video_encoder_config_.Copy());
-    video_send_stream_->SetSource(
-        frame_generator_capturer_.get(),
-        VideoSendStream::DegradationPreference::kMaintainResolution);
-    Start();
+    task_queue_.SendTask([this, &send_stream_config, &encoder_config]() {
+      Stop();
+      sender_call_->DestroyVideoSendStream(video_send_stream_);
+      video_send_config_ = send_stream_config.Copy();
+      video_encoder_config_ = encoder_config.Copy();
+      video_send_stream_ = sender_call_->CreateVideoSendStream(
+          video_send_config_.Copy(), video_encoder_config_.Copy());
+      video_send_stream_->SetSource(
+          frame_generator_capturer_.get(),
+          VideoSendStream::DegradationPreference::kMaintainResolution);
+      Start();
+    });
   };
   MaxPaddingSetTest<decltype(reset_fun)> test(true, &reset_fun);
   RunBaseTest(&test);
 }
 
 // This test verifies that new frame sizes reconfigures encoders even though not
 // (yet) sending. The purpose of this is to permit encoding as quickly as
 // possible once we start sending. Likely the frames being input are from the
 // same source that will be sent later, which just means that we're ready
 // earlier.
@@ skipping 45 matching lines @@
       return 0;
     }
 
     rtc::CriticalSection crit_;
     rtc::Event init_encode_called_;
     size_t number_of_initializations_ GUARDED_BY(&crit_);
     int last_initialized_frame_width_ GUARDED_BY(&crit_);
     int last_initialized_frame_height_ GUARDED_BY(&crit_);
   };
 
-  CreateSenderCall(Call::Config(event_log_.get()));
   test::NullTransport transport;
-  CreateSendConfig(1, 0, 0, &transport);
   EncoderObserver encoder;
-  video_send_config_.encoder_settings.encoder = &encoder;
-  CreateVideoStreams();
-  CreateFrameGeneratorCapturer(kDefaultFramerate, kDefaultWidth,
-                               kDefaultHeight);
-  frame_generator_capturer_->Start();
+
+  task_queue_.SendTask([this, &transport, &encoder]() {
+    CreateSenderCall(Call::Config(event_log_.get()));
+    CreateSendConfig(1, 0, 0, &transport);
+    video_send_config_.encoder_settings.encoder = &encoder;
+    CreateVideoStreams();
+    CreateFrameGeneratorCapturer(kDefaultFramerate, kDefaultWidth,
+                                 kDefaultHeight);
+    frame_generator_capturer_->Start();
+  });
 
   encoder.WaitForResolution(kDefaultWidth, kDefaultHeight);
-  frame_generator_capturer_->ChangeResolution(kDefaultWidth * 2,
-                                              kDefaultHeight * 2);
+
+  task_queue_.SendTask([this]() {
+    frame_generator_capturer_->ChangeResolution(kDefaultWidth * 2,
+                                                kDefaultHeight * 2);
+  });
+
   encoder.WaitForResolution(kDefaultWidth * 2, kDefaultHeight * 2);
-  DestroyStreams();
+
+  task_queue_.SendTask([this]() {
+    DestroyStreams();
+    DestroyCalls();
+  });
 }
 
 TEST_F(VideoSendStreamTest, CanReconfigureToUseStartBitrateAbovePreviousMax) {
   class StartBitrateObserver : public test::FakeEncoder {
    public:
     StartBitrateObserver()
         : FakeEncoder(Clock::GetRealTimeClock()),
           start_bitrate_changed_(false, false),
           start_bitrate_kbps_(0) {}
     int32_t InitEncode(const VideoCodec* config,
@@ skipping 111 matching lines @@
       return false;
     }
 
    private:
     rtc::CriticalSection crit_;
     rtc::Event encoder_init_;
     rtc::Event bitrate_changed_;
     rtc::Optional<int> bitrate_kbps_ GUARDED_BY(crit_);
   };
 
-  CreateSenderCall(Call::Config(event_log_.get()));
+  test::NullTransport transport;
+  StartStopBitrateObserver encoder;
 
-  test::NullTransport transport;
-  CreateSendConfig(1, 0, 0, &transport);
+  task_queue_.SendTask([this, &transport, &encoder]() {
+    CreateSenderCall(Call::Config(event_log_.get()));
+    CreateSendConfig(1, 0, 0, &transport);
 
-  sender_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkUp);
+    sender_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkUp);
 
-  StartStopBitrateObserver encoder;
-  video_send_config_.encoder_settings.encoder = &encoder;
-  video_send_config_.encoder_settings.internal_source = true;
+    video_send_config_.encoder_settings.encoder = &encoder;
+    video_send_config_.encoder_settings.internal_source = true;
 
-  CreateVideoStreams();
+    CreateVideoStreams();
+  });
 
   EXPECT_TRUE(encoder.WaitForEncoderInit());
 
-  video_send_stream_->Start();
+  task_queue_.SendTask([this]() {
+    video_send_stream_->Start();
+  });
   EXPECT_TRUE(encoder.WaitBitrateChanged(true));
 
-  video_send_stream_->Stop();
+  task_queue_.SendTask([this]() {
+    video_send_stream_->Stop();
+  });
   EXPECT_TRUE(encoder.WaitBitrateChanged(false));
 
-  video_send_stream_->Start();
+  task_queue_.SendTask([this]() {
+    video_send_stream_->Start();
+  });
   EXPECT_TRUE(encoder.WaitBitrateChanged(true));
 
-  DestroyStreams();
+  task_queue_.SendTask([this]() {
+    DestroyStreams();
+    DestroyCalls();
+  });
 }
 
 TEST_F(VideoSendStreamTest, CapturesTextureAndVideoFrames) {
   class FrameObserver : public rtc::VideoSinkInterface<VideoFrame> {
    public:
     FrameObserver() : output_frame_event_(false, false) {}
 
     void OnFrame(const VideoFrame& video_frame) override {
       output_frames_.push_back(video_frame);
       output_frame_event_.Set();
@@ skipping 10 matching lines @@
     }
 
    private:
     // Delivered output frames.
     std::vector<VideoFrame> output_frames_;
 
     // Indicate an output frame has arrived.
     rtc::Event output_frame_event_;
   };
 
-  // Initialize send stream.
-  CreateSenderCall(Call::Config(event_log_.get()));
+  test::NullTransport transport;
+  FrameObserver observer;
+  std::vector<VideoFrame> input_frames;
 
-  test::NullTransport transport;
-  CreateSendConfig(1, 0, 0, &transport);
-  FrameObserver observer;
-  video_send_config_.pre_encode_callback = &observer;
-  CreateVideoStreams();
+  task_queue_.SendTask([this, &transport, &observer, &input_frames]() {
+    // Initialize send stream.
+    CreateSenderCall(Call::Config(event_log_.get()));
 
-  // Prepare five input frames. Send ordinary VideoFrame and texture frames
-  // alternatively.
-  std::vector<VideoFrame> input_frames;
-  int width = 168;
-  int height = 132;
+    CreateSendConfig(1, 0, 0, &transport);
+    video_send_config_.pre_encode_callback = &observer;
+    CreateVideoStreams();
 
-  input_frames.push_back(test::FakeNativeBuffer::CreateFrame(
-      width, height, 1, 1, kVideoRotation_0));
-  input_frames.push_back(test::FakeNativeBuffer::CreateFrame(
-      width, height, 2, 2, kVideoRotation_0));
-  input_frames.push_back(CreateVideoFrame(width, height, 3));
-  input_frames.push_back(CreateVideoFrame(width, height, 4));
-  input_frames.push_back(test::FakeNativeBuffer::CreateFrame(
-      width, height, 5, 5, kVideoRotation_0));
+    // Prepare five input frames. Send ordinary VideoFrame and texture frames
+    // alternatively.
+    int width = 168;
+    int height = 132;
 
-  video_send_stream_->Start();
-  test::FrameForwarder forwarder;
-  video_send_stream_->SetSource(
-      &forwarder, VideoSendStream::DegradationPreference::kMaintainFramerate);
-  for (size_t i = 0; i < input_frames.size(); i++) {
-    forwarder.IncomingCapturedFrame(input_frames[i]);
-    // Wait until the output frame is received before sending the next input
-    // frame. Or the previous input frame may be replaced without delivering.
-    observer.WaitOutputFrame();
-  }
-  video_send_stream_->Stop();
-  video_send_stream_->SetSource(
-      nullptr, VideoSendStream::DegradationPreference::kMaintainFramerate);
+    input_frames.push_back(test::FakeNativeBuffer::CreateFrame(
+        width, height, 1, 1, kVideoRotation_0));
+    input_frames.push_back(test::FakeNativeBuffer::CreateFrame(
+        width, height, 2, 2, kVideoRotation_0));
+    input_frames.push_back(CreateVideoFrame(width, height, 3));
+    input_frames.push_back(CreateVideoFrame(width, height, 4));
+    input_frames.push_back(test::FakeNativeBuffer::CreateFrame(
+        width, height, 5, 5, kVideoRotation_0));
+
+    video_send_stream_->Start();
+    test::FrameForwarder forwarder;
+    video_send_stream_->SetSource(
+        &forwarder, VideoSendStream::DegradationPreference::kMaintainFramerate);
+    for (size_t i = 0; i < input_frames.size(); i++) {
+      forwarder.IncomingCapturedFrame(input_frames[i]);
+      // Wait until the output frame is received before sending the next input
+      // frame. Or the previous input frame may be replaced without delivering.
+      observer.WaitOutputFrame();
+    }
+    video_send_stream_->Stop();
+    video_send_stream_->SetSource(
+        nullptr, VideoSendStream::DegradationPreference::kMaintainFramerate);
+  });
 
   // Test if the input and output frames are the same. render_time_ms and
   // timestamp are not compared because capturer sets those values.
   ExpectEqualFramesVector(input_frames, observer.output_frames());
 
-  DestroyStreams();
+  task_queue_.SendTask([this]() {
+    DestroyStreams();
+    DestroyCalls();
+  });
 }
 
 void ExpectEqualFramesVector(const std::vector<VideoFrame>& frames1,
                              const std::vector<VideoFrame>& frames2) {
   EXPECT_EQ(frames1.size(), frames2.size());
   for (size_t i = 0; i < std::min(frames1.size(), frames2.size()); ++i)
     // Compare frame buffers, since we don't care about differing timestamps.
     EXPECT_TRUE(test::FrameBufsEqual(frames1[i].video_frame_buffer(),
                                      frames2[i].video_frame_buffer()));
 }
 
 VideoFrame CreateVideoFrame(int width, int height, uint8_t data) {
   const int kSizeY = width * height * 2;
   std::unique_ptr<uint8_t[]> buffer(new uint8_t[kSizeY]);
   memset(buffer.get(), data, kSizeY);
   VideoFrame frame(I420Buffer::Create(width, height), kVideoRotation_0, data);
   frame.set_timestamp(data);
   // Use data as a ms timestamp.
   frame.set_timestamp_us(data * rtc::kNumMicrosecsPerMillisec);
   return frame;
 }
 
 TEST_F(VideoSendStreamTest, EncoderIsProperlyInitializedAndDestroyed) {
   class EncoderStateObserver : public test::SendTest, public VideoEncoder {
    public:
-    EncoderStateObserver()
+    explicit EncoderStateObserver(
+        test::SingleThreadedTaskQueueForTesting* task_queue)
         : SendTest(kDefaultTimeoutMs),
+          task_queue_(task_queue),
           stream_(nullptr),
           initialized_(false),
           callback_registered_(false),
           num_releases_(0),
           released_(false) {}
 
     bool IsReleased() {
       rtc::CritScope lock(&crit_);
       return released_;
     }
@@ skipping 66 matching lines @@
     void ModifyVideoConfigs(
         VideoSendStream::Config* send_config,
         std::vector<VideoReceiveStream::Config>* receive_configs,
         VideoEncoderConfig* encoder_config) override {
       send_config->encoder_settings.encoder = this;
       encoder_config_ = encoder_config->Copy();
     }
 
     void PerformTest() override {
       EXPECT_TRUE(Wait()) << "Timed out while waiting for Encode.";
-      EXPECT_EQ(0u, num_releases());
-      stream_->ReconfigureVideoEncoder(std::move(encoder_config_));
-      EXPECT_EQ(0u, num_releases());
-      stream_->Stop();
-      // Encoder should not be released before destroying the VideoSendStream.
-      EXPECT_FALSE(IsReleased());
-      EXPECT_TRUE(IsReadyForEncode());
-      stream_->Start();
+
+      task_queue_->SendTask([this]() {
+        EXPECT_EQ(0u, num_releases());
+        stream_->ReconfigureVideoEncoder(std::move(encoder_config_));
+        EXPECT_EQ(0u, num_releases());
+        stream_->Stop();
+        // Encoder should not be released before destroying the VideoSendStream.
+        EXPECT_FALSE(IsReleased());
+        EXPECT_TRUE(IsReadyForEncode());
+        stream_->Start();
+      });
+
       // Sanity check, make sure we still encode frames with this encoder.
       EXPECT_TRUE(Wait()) << "Timed out while waiting for Encode.";
     }
 
+    test::SingleThreadedTaskQueueForTesting* const task_queue_;
     rtc::CriticalSection crit_;
     VideoSendStream* stream_;
     bool initialized_ GUARDED_BY(crit_);
     bool callback_registered_ GUARDED_BY(crit_);
     size_t num_releases_ GUARDED_BY(crit_);
     bool released_ GUARDED_BY(crit_);
     VideoEncoderConfig encoder_config_;
-  } test_encoder;
+  } test_encoder(&task_queue_);
 
   RunBaseTest(&test_encoder);
 
   EXPECT_TRUE(test_encoder.IsReleased());
   EXPECT_EQ(1u, test_encoder.num_releases());
 }
 
 TEST_F(VideoSendStreamTest, EncoderSetupPropagatesCommonEncoderConfigValues) {
   class VideoCodecConfigObserver : public test::SendTest,
                                    public test::FakeEncoder {
@@ skipping 365 matching lines @@
   // default values.
   static const int kMinBitrateKbps = 137;
   static const int kStartBitrateKbps = 345;
   static const int kLowerMaxBitrateKbps = 312;
   static const int kMaxBitrateKbps = 413;
   static const int kIncreasedStartBitrateKbps = 451;
   static const int kIncreasedMaxBitrateKbps = 597;
   class EncoderBitrateThresholdObserver : public test::SendTest,
                                           public test::FakeEncoder {
    public:
-    EncoderBitrateThresholdObserver()
+    explicit EncoderBitrateThresholdObserver(
+        test::SingleThreadedTaskQueueForTesting* task_queue)
         : SendTest(kDefaultTimeoutMs),
           FakeEncoder(Clock::GetRealTimeClock()),
+          task_queue_(task_queue),
           init_encode_event_(false, false),
           bitrate_changed_event_(false, false),
           target_bitrate_(0),
           num_initializations_(0),
           call_(nullptr),
           send_stream_(nullptr) {}
 
    private:
     int32_t InitEncode(const VideoCodec* codecSettings,
                        int32_t numberOfCores,
@@ skipping 104 matching lines @@
     }
 
     void PerformTest() override {
       ASSERT_TRUE(
           init_encode_event_.Wait(VideoSendStreamTest::kDefaultTimeoutMs))
           << "Timed out while waiting for encoder to be configured.";
       WaitForSetRates(kStartBitrateKbps);
       Call::Config::BitrateConfig bitrate_config;
       bitrate_config.start_bitrate_bps = kIncreasedStartBitrateKbps * 1000;
       bitrate_config.max_bitrate_bps = kIncreasedMaxBitrateKbps * 1000;
-      call_->SetBitrateConfig(bitrate_config);
+      task_queue_->SendTask([this, &bitrate_config]() {
+        call_->SetBitrateConfig(bitrate_config);
+      });
       // Encoder rate is capped by EncoderConfig max_bitrate_bps.
       WaitForSetRates(kMaxBitrateKbps);
       encoder_config_.max_bitrate_bps = kLowerMaxBitrateKbps * 1000;
       send_stream_->ReconfigureVideoEncoder(encoder_config_.Copy());
       ASSERT_TRUE(
           init_encode_event_.Wait(VideoSendStreamTest::kDefaultTimeoutMs));
       EXPECT_EQ(2, num_initializations_)
           << "Encoder should have been reconfigured with the new value.";
       WaitForSetRates(kLowerMaxBitrateKbps);
 
       encoder_config_.max_bitrate_bps = kIncreasedMaxBitrateKbps * 1000;
       send_stream_->ReconfigureVideoEncoder(encoder_config_.Copy());
       ASSERT_TRUE(
           init_encode_event_.Wait(VideoSendStreamTest::kDefaultTimeoutMs));
       EXPECT_EQ(3, num_initializations_)
           << "Encoder should have been reconfigured with the new value.";
       // Expected target bitrate is the start bitrate set in the call to
       // call_->SetBitrateConfig.
       WaitForSetRates(kIncreasedStartBitrateKbps);
     }
 
+    test::SingleThreadedTaskQueueForTesting* const task_queue_;
     rtc::Event init_encode_event_;
     rtc::Event bitrate_changed_event_;
     rtc::CriticalSection crit_;
     uint32_t target_bitrate_ GUARDED_BY(&crit_);
 
     int num_initializations_;
     webrtc::Call* call_;
     webrtc::VideoSendStream* send_stream_;
     webrtc::VideoEncoderConfig encoder_config_;
-  } test;
+  } test(&task_queue_);
 
   RunBaseTest(&test);
 }
 
 TEST_F(VideoSendStreamTest, ReportsSentResolution) {
   static const size_t kNumStreams = 3;
   // Unusual resolutions to make sure that they are the ones being reported.
   static const struct {
     int width;
     int height;
@@ skipping 570 matching lines @@
 
 // This test verifies that overhead is removed from the bandwidth estimate by
 // testing that the maximum possible target payload rate is smaller than the
 // maximum bandwidth estimate by the overhead rate.
 TEST_F(VideoSendStreamTest, RemoveOverheadFromBandwidth) {
   test::ScopedFieldTrials override_field_trials(
       "WebRTC-SendSideBwe-WithOverhead/Enabled/");
   class RemoveOverheadFromBandwidthTest : public test::EndToEndTest,
                                           public test::FakeEncoder {
    public:
-    RemoveOverheadFromBandwidthTest()
+    explicit RemoveOverheadFromBandwidthTest(
+        test::SingleThreadedTaskQueueForTesting* task_queue)
         : EndToEndTest(test::CallTest::kDefaultTimeoutMs),
           FakeEncoder(Clock::GetRealTimeClock()),
+          task_queue_(task_queue),
           call_(nullptr),
           max_bitrate_bps_(0),
           first_packet_sent_(false),
           bitrate_changed_event_(false, false) {}
 
     int32_t SetRateAllocation(const BitrateAllocation& bitrate,
                               uint32_t frameRate) override {
       rtc::CritScope lock(&crit_);
       // Wait for the first sent packet so that videosendstream knows
       // rtp_overhead.
@@ skipping 24 matching lines @@
     }
 
     void PerformTest() override {
       Call::Config::BitrateConfig bitrate_config;
       constexpr int kStartBitrateBps = 60000;
       constexpr int kMaxBitrateBps = 60000;
       constexpr int kMinBitrateBps = 10000;
       bitrate_config.start_bitrate_bps = kStartBitrateBps;
       bitrate_config.max_bitrate_bps = kMaxBitrateBps;
       bitrate_config.min_bitrate_bps = kMinBitrateBps;
-      call_->SetBitrateConfig(bitrate_config);
-      call_->OnTransportOverheadChanged(webrtc::MediaType::VIDEO, 40);
+      task_queue_->SendTask([this, &bitrate_config]() {
+        call_->SetBitrateConfig(bitrate_config);
+        call_->OnTransportOverheadChanged(webrtc::MediaType::VIDEO, 40);
+      });
 
       // At a bitrate of 60kbps with a packet size of 1200B video and an
       // overhead of 40B per packet video produces 2240bps overhead.
       // So the encoder BW should be set to 57760bps.
       bitrate_changed_event_.Wait(VideoSendStreamTest::kDefaultTimeoutMs);
       {
         rtc::CritScope lock(&crit_);
         EXPECT_LE(max_bitrate_bps_, 57760u);
       }
     }
 
    private:
+    test::SingleThreadedTaskQueueForTesting* const task_queue_;
     Call* call_;
     rtc::CriticalSection crit_;
     uint32_t max_bitrate_bps_ GUARDED_BY(&crit_);
     bool first_packet_sent_ GUARDED_BY(&crit_);
     rtc::Event bitrate_changed_event_;
-  } test;
-
+  } test(&task_queue_);
   RunBaseTest(&test);
 }
 
 TEST_F(VideoSendStreamTest, SendsKeepAlive) {
   const int kTimeoutMs = 50;  // Really short timeout for testing.
 
   class KeepaliveObserver : public test::SendTest {
    public:
     KeepaliveObserver() : SendTest(kDefaultTimeoutMs) {}
 
@@ skipping 30 matching lines @@
       capturer_ = frame_generator_capturer;
     }
 
     test::FrameGeneratorCapturer* capturer_ = nullptr;
   } test;
 
   RunBaseTest(&test);
 }
 
 }  // namespace webrtc