Use SingleThreadedTaskQueue in DirectTransport

webrtc/video/end_to_end_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>
@@ skipping 21 matching lines @@
 #include "webrtc/modules/rtp_rtcp/source/rtp_format.h"
 #include "webrtc/modules/rtp_rtcp/source/rtp_utility.h"
 #include "webrtc/modules/video_coding/codecs/h264/include/h264.h"
 #include "webrtc/modules/video_coding/codecs/vp8/include/vp8.h"
 #include "webrtc/modules/video_coding/codecs/vp9/include/vp9.h"
 #include "webrtc/modules/video_coding/include/video_coding_defines.h"
 #include "webrtc/rtc_base/checks.h"
 #include "webrtc/rtc_base/event.h"
 #include "webrtc/rtc_base/file.h"
 #include "webrtc/rtc_base/optional.h"
+#include "webrtc/rtc_base/ptr_util.h"
 #include "webrtc/rtc_base/random.h"
 #include "webrtc/rtc_base/rate_limiter.h"
 #include "webrtc/system_wrappers/include/metrics.h"
 #include "webrtc/system_wrappers/include/metrics_default.h"
 #include "webrtc/system_wrappers/include/sleep.h"
 #include "webrtc/test/call_test.h"
 #include "webrtc/test/direct_transport.h"
 #include "webrtc/test/encoder_settings.h"
 #include "webrtc/test/fake_decoder.h"
 #include "webrtc/test/fake_encoder.h"
@@ skipping 158 matching lines @@
     void OnFrame(const VideoFrame& video_frame) override {
       SleepMs(kRenderDelayMs);
       event_.Set();
     }
 
     bool Wait() { return event_.Wait(kDefaultTimeoutMs); }
 
     rtc::Event event_;
   } renderer;
 
-  CreateCalls(Call::Config(event_log_.get()), Call::Config(event_log_.get()));
+  test::FrameForwarder frame_forwarder;
+  std::unique_ptr<test::DirectTransport> sender_transport;
+  std::unique_ptr<test::DirectTransport> receiver_transport;
 
-  test::DirectTransport sender_transport(sender_call_.get(), payload_type_map_);
-  test::DirectTransport receiver_transport(receiver_call_.get(),
-                                           payload_type_map_);
-  sender_transport.SetReceiver(receiver_call_->Receiver());
-  receiver_transport.SetReceiver(sender_call_->Receiver());
+  task_queue_.SendTask([this, &renderer, &frame_forwarder, &sender_transport,
+                        &receiver_transport]() {
+    CreateCalls(Call::Config(event_log_.get()), Call::Config(event_log_.get()));
 
-  CreateSendConfig(1, 0, 0, &sender_transport);
-  CreateMatchingReceiveConfigs(&receiver_transport);
+    sender_transport = rtc::MakeUnique<test::DirectTransport>(

[nisse-webrtc, 2017/08/18 11:08:30]

Looks odd to me to create the transports from a task on the task queue. I takie
it CreateCalls must be on the task queue, but creation of the transports don't
have to, right?

As for the calls to SetReceiver, I don't know. Would perhaps be more natural to
pass the receivers to the DirectTransport constructor?

(This pattern is used in several places, if changed, change them all).

[eladalon, 2017/08/18 12:12:38]

* The way I understand it, the transports also need to happen on the task queue.
* I am aiming to make this CL, which is already big, as small as possible. I
would rather not make more changes than we have to, or make the separation of
on-task-queue and off-task-queue less than trivial. I try to follow a pattern of
"everything on the task queue, except for those things that are the test itself,
waiting for events to happen, events that the task queue is causing."

+        &task_queue_, sender_call_.get(), payload_type_map_);
+    receiver_transport = rtc::MakeUnique<test::DirectTransport>(
+        &task_queue_, receiver_call_.get(), payload_type_map_);
+    sender_transport->SetReceiver(receiver_call_->Receiver());
+    receiver_transport->SetReceiver(sender_call_->Receiver());
 
-  video_receive_configs_[0].renderer = &renderer;
+    CreateSendConfig(1, 0, 0, sender_transport.get());
+    CreateMatchingReceiveConfigs(receiver_transport.get());
 
-  CreateVideoStreams();
-  Start();
+    video_receive_configs_[0].renderer = &renderer;
 
-  // Create frames that are smaller than the send width/height, this is done to
-  // check that the callbacks are done after processing video.
-  std::unique_ptr<test::FrameGenerator> frame_generator(
-      test::FrameGenerator::CreateSquareGenerator(kWidth, kHeight));
-  test::FrameForwarder frame_forwarder;
-  video_send_stream_->SetSource(
-      &frame_forwarder,
-      VideoSendStream::DegradationPreference::kMaintainFramerate);
+    CreateVideoStreams();
+    Start();
 
-  frame_forwarder.IncomingCapturedFrame(*frame_generator->NextFrame());
+    // Create frames that are smaller than the send width/height, this is done
+    // to check that the callbacks are done after processing video.
+    std::unique_ptr<test::FrameGenerator> frame_generator(
+        test::FrameGenerator::CreateSquareGenerator(kWidth, kHeight));
+    video_send_stream_->SetSource(
+        &frame_forwarder,
+        VideoSendStream::DegradationPreference::kMaintainFramerate);
+
+    frame_forwarder.IncomingCapturedFrame(*frame_generator->NextFrame());
+  });
+
   EXPECT_TRUE(renderer.Wait())
       << "Timed out while waiting for the frame to render.";
 
-  Stop();
-
-  sender_transport.StopSending();
-  receiver_transport.StopSending();
-
-  DestroyStreams();
+  task_queue_.SendTask([this, &sender_transport, &receiver_transport]() {
+    Stop();
+    DestroyStreams();
+    sender_transport.reset();
+    receiver_transport.reset();
+    DestroyCalls();
+  });
 }
 
 TEST_F(EndToEndTest, TransmitsFirstFrame) {
   class Renderer : public rtc::VideoSinkInterface<VideoFrame> {
    public:
     Renderer() : event_(false, false) {}
 
     void OnFrame(const VideoFrame& video_frame) override { event_.Set(); }
 
     bool Wait() { return event_.Wait(kDefaultTimeoutMs); }
 
     rtc::Event event_;
   } renderer;
 
-  CreateCalls(Call::Config(event_log_.get()), Call::Config(event_log_.get()));
+  std::unique_ptr<test::FrameGenerator> frame_generator;
+  test::FrameForwarder frame_forwarder;
 
-  test::DirectTransport sender_transport(sender_call_.get(), payload_type_map_);
-  test::DirectTransport receiver_transport(receiver_call_.get(),
-                                           payload_type_map_);
-  sender_transport.SetReceiver(receiver_call_->Receiver());
-  receiver_transport.SetReceiver(sender_call_->Receiver());
+  std::unique_ptr<test::DirectTransport> sender_transport;
+  std::unique_ptr<test::DirectTransport> receiver_transport;
 
-  CreateSendConfig(1, 0, 0, &sender_transport);
-  CreateMatchingReceiveConfigs(&receiver_transport);
-  video_receive_configs_[0].renderer = &renderer;
+  task_queue_.SendTask([this, &renderer, &frame_generator, &frame_forwarder,
+                        &sender_transport, &receiver_transport]() {
+    CreateCalls(Call::Config(event_log_.get()), Call::Config(event_log_.get()));
 
-  CreateVideoStreams();
-  Start();
+    sender_transport = rtc::MakeUnique<test::DirectTransport>(
+        &task_queue_, sender_call_.get(), payload_type_map_);
+    receiver_transport = rtc::MakeUnique<test::DirectTransport>(
+        &task_queue_, receiver_call_.get(), payload_type_map_);
+    sender_transport->SetReceiver(receiver_call_->Receiver());
+    receiver_transport->SetReceiver(sender_call_->Receiver());
 
-  std::unique_ptr<test::FrameGenerator> frame_generator(
-      test::FrameGenerator::CreateSquareGenerator(kDefaultWidth,
-                                                  kDefaultHeight));
-  test::FrameForwarder frame_forwarder;
-  video_send_stream_->SetSource(
-      &frame_forwarder,
-      VideoSendStream::DegradationPreference::kMaintainFramerate);
-  frame_forwarder.IncomingCapturedFrame(*frame_generator->NextFrame());
+    CreateSendConfig(1, 0, 0, sender_transport.get());
+    CreateMatchingReceiveConfigs(receiver_transport.get());
+    video_receive_configs_[0].renderer = &renderer;
+
+    CreateVideoStreams();
+    Start();
+
+    frame_generator = test::FrameGenerator::CreateSquareGenerator(
+        kDefaultWidth, kDefaultHeight);
+    video_send_stream_->SetSource(
+        &frame_forwarder,
+        VideoSendStream::DegradationPreference::kMaintainFramerate);
+    frame_forwarder.IncomingCapturedFrame(*frame_generator->NextFrame());
+  });
 
   EXPECT_TRUE(renderer.Wait())
       << "Timed out while waiting for the frame to render.";
 
-  Stop();
-
-  sender_transport.StopSending();
-  receiver_transport.StopSending();
-
-  DestroyStreams();
+  task_queue_.SendTask([this, &sender_transport, &receiver_transport]() {
+    Stop();
+    DestroyStreams();
+    sender_transport.reset();
+    receiver_transport.reset();
+    DestroyCalls();
+  });
 }
 
 class CodecObserver : public test::EndToEndTest,
                       public rtc::VideoSinkInterface<VideoFrame> {
  public:
   CodecObserver(int no_frames_to_wait_for,
                 VideoRotation rotation_to_test,
                 const std::string& payload_name,
                 webrtc::VideoEncoder* encoder,
                 webrtc::VideoDecoder* decoder)
@@ skipping 221 matching lines @@
     NackObserver()
         : EndToEndTest(kLongTimeoutMs),
           local_ssrc_(0),
           remote_ssrc_(0),
           receive_transport_(nullptr) {}
 
    private:
     size_t GetNumVideoStreams() const override { return 0; }
     size_t GetNumAudioStreams() const override { return 1; }
 
-    test::PacketTransport* CreateReceiveTransport() override {
+    test::PacketTransport* CreateReceiveTransport(
+        test::SingleThreadedTaskQueueForTesting* task_queue) override {
       test::PacketTransport* receive_transport = new test::PacketTransport(
-          nullptr, this, test::PacketTransport::kReceiver, payload_type_map_,
-          FakeNetworkPipe::Config());
+          task_queue, nullptr, this, test::PacketTransport::kReceiver,
+          payload_type_map_, FakeNetworkPipe::Config());
       receive_transport_ = receive_transport;
       return receive_transport;
     }
 
     Action OnSendRtp(const uint8_t* packet, size_t length) override {
       RTPHeader header;
       EXPECT_TRUE(parser_->Parse(packet, length, &header));
 
       if (!sequence_number_to_retransmit_) {
         sequence_number_to_retransmit_ =
@@ skipping 234 matching lines @@
         EXPECT_EQ(test::CallTest::kFlexfecSendSsrc,
                   report_blocks[0].source_ssrc());
         rtc::CritScope lock(&crit_);
         received_flexfec_rtcp_ = true;
       }
     }
 
     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.
     const int kNetworkDelayMs = 100;
     FakeNetworkPipe::Config config;
     config.queue_delay_ms = kNetworkDelayMs;
-    return new test::PacketTransport(sender_call, this,
+    return new test::PacketTransport(task_queue, sender_call, this,
                                      test::PacketTransport::kSender,
                                      test::CallTest::payload_type_map_, config);
   }
 
   void OnFrame(const VideoFrame& video_frame) override {
     rtc::CritScope lock(&crit_);
     // Rendering frame with timestamp of packet that was dropped -> FEC
     // protection worked.
     auto it = dropped_timestamps_.find(video_frame.timestamp());
     if (it != dropped_timestamps_.end()) {
@@ skipping 153 matching lines @@
                               ulpfec_sequence_number_) == nacks.end())
             << "Got nack for ULPFEC packet";
         if (!nacks.empty() &&
             IsNewerSequenceNumber(nacks.back(), ulpfec_sequence_number_)) {
           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.
       // Configure some network delay.
       const int kNetworkDelayMs = 50;
       FakeNetworkPipe::Config config;
       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);
     }
 
     // TODO(holmer): Investigate why we don't send FEC packets when the bitrate
     // is 10 kbps.
     Call::Config GetSenderCallConfig() override {
       Call::Config config(event_log_.get());
       const int kMinBitrateBps = 30000;
       config.bitrate_config.min_bitrate_bps = kMinBitrateBps;
@@ skipping 321 matching lines @@
         EXPECT_EQ(DELIVERY_UNKNOWN_SSRC, delivery_status);
         delivered_packet_.Set();
         return delivery_status;
       }
     }
 
     PacketReceiver* receiver_;
     rtc::Event delivered_packet_;
   };
 
-  CreateCalls(Call::Config(event_log_.get()), Call::Config(event_log_.get()));
+  std::unique_ptr<test::DirectTransport> send_transport;
+  std::unique_ptr<test::DirectTransport> receive_transport;
+  std::unique_ptr<PacketInputObserver> input_observer;
 
-  test::DirectTransport send_transport(sender_call_.get(), payload_type_map_);
-  test::DirectTransport receive_transport(receiver_call_.get(),
-                                          payload_type_map_);
-  PacketInputObserver input_observer(receiver_call_->Receiver());
-  send_transport.SetReceiver(&input_observer);
-  receive_transport.SetReceiver(sender_call_->Receiver());
+  task_queue_.SendTask([this, &send_transport, &receive_transport,
+                        &input_observer]() {
+    CreateCalls(Call::Config(event_log_.get()), Call::Config(event_log_.get()));
 
-  CreateSendConfig(1, 0, 0, &send_transport);
-  CreateMatchingReceiveConfigs(&receive_transport);
+    send_transport = rtc::MakeUnique<test::DirectTransport>(
+        &task_queue_, sender_call_.get(), payload_type_map_);
+    receive_transport = rtc::MakeUnique<test::DirectTransport>(
+        &task_queue_, receiver_call_.get(), payload_type_map_);
+    input_observer =
+        rtc::MakeUnique<PacketInputObserver>(receiver_call_->Receiver());
+    send_transport->SetReceiver(input_observer.get());
+    receive_transport->SetReceiver(sender_call_->Receiver());
 
-  CreateVideoStreams();
-  CreateFrameGeneratorCapturer(kDefaultFramerate, kDefaultWidth,
-                               kDefaultHeight);
-  Start();
+    CreateSendConfig(1, 0, 0, send_transport.get());
+    CreateMatchingReceiveConfigs(receive_transport.get());
 
-  receiver_call_->DestroyVideoReceiveStream(video_receive_streams_[0]);
-  video_receive_streams_.clear();
+    CreateVideoStreams();
+    CreateFrameGeneratorCapturer(kDefaultFramerate, kDefaultWidth,
+                                 kDefaultHeight);
+    Start();
+
+    receiver_call_->DestroyVideoReceiveStream(video_receive_streams_[0]);
+    video_receive_streams_.clear();
+  });
 
   // Wait() waits for a received packet.
-  EXPECT_TRUE(input_observer.Wait());
+  EXPECT_TRUE(input_observer->Wait());
 
-  Stop();
-
-  DestroyStreams();
-
-  send_transport.StopSending();
-  receive_transport.StopSending();
+  task_queue_.SendTask([this, &send_transport, &receive_transport]() {
+    Stop();
+    DestroyStreams();
+    send_transport.reset();
+    receive_transport.reset();
+    DestroyCalls();
+  });
 }
 
 void EndToEndTest::RespectsRtcpMode(RtcpMode rtcp_mode) {
   static const int kNumCompoundRtcpPacketsToObserve = 10;
   class RtcpModeObserver : public test::EndToEndTest {
    public:
     explicit RtcpModeObserver(RtcpMode rtcp_mode)
         : EndToEndTest(kDefaultTimeoutMs),
           rtcp_mode_(rtcp_mode),
           sent_rtp_(0),
@@ skipping 86 matching lines @@
   const uint8_t kVideoPayloadType = 124;
   const std::map<uint8_t, MediaType> payload_type_map_ = {
       {kVideoPayloadType, MediaType::VIDEO}};
 
   struct CodecSettings {
     uint32_t ssrc;
     int width;
     int height;
   } codec_settings[kNumStreams];
 
-  MultiStreamTest() {
+  explicit MultiStreamTest(test::SingleThreadedTaskQueueForTesting* task_queue)
+      : task_queue_(task_queue) {
     // TODO(sprang): Cleanup when msvc supports explicit initializers for array.
     codec_settings[0] = {1, 640, 480};
     codec_settings[1] = {2, 320, 240};
     codec_settings[2] = {3, 240, 160};
   }
 
   virtual ~MultiStreamTest() {}
 
   void RunTest() {
     webrtc::RtcEventLogNullImpl event_log;
     Call::Config config(&event_log);
-    std::unique_ptr<Call> sender_call(Call::Create(config));
-    std::unique_ptr<Call> receiver_call(Call::Create(config));
-    std::unique_ptr<test::DirectTransport> sender_transport(
-        CreateSendTransport(sender_call.get()));
-    std::unique_ptr<test::DirectTransport> receiver_transport(
-        CreateReceiveTransport(receiver_call.get()));
-    sender_transport->SetReceiver(receiver_call->Receiver());
-    receiver_transport->SetReceiver(sender_call->Receiver());
-
-    std::unique_ptr<VideoEncoder> encoders[kNumStreams];
-    for (size_t i = 0; i < kNumStreams; ++i)
-      encoders[i].reset(VP8Encoder::Create());
+    std::unique_ptr<Call> sender_call;
+    std::unique_ptr<Call> receiver_call;
+    std::unique_ptr<test::DirectTransport> sender_transport;
+    std::unique_ptr<test::DirectTransport> receiver_transport;
 
     VideoSendStream* send_streams[kNumStreams];
     VideoReceiveStream* receive_streams[kNumStreams];
-
     test::FrameGeneratorCapturer* frame_generators[kNumStreams];
     std::vector<std::unique_ptr<VideoDecoder>> allocated_decoders;
-    for (size_t i = 0; i < kNumStreams; ++i) {
-      uint32_t ssrc = codec_settings[i].ssrc;
-      int width = codec_settings[i].width;
-      int height = codec_settings[i].height;
+    std::unique_ptr<VideoEncoder> encoders[kNumStreams];
 
-      VideoSendStream::Config send_config(sender_transport.get());
-      send_config.rtp.ssrcs.push_back(ssrc);
-      send_config.encoder_settings.encoder = encoders[i].get();
-      send_config.encoder_settings.payload_name = "VP8";
-      send_config.encoder_settings.payload_type = kVideoPayloadType;
-      VideoEncoderConfig encoder_config;
-      test::FillEncoderConfiguration(1, &encoder_config);
-      encoder_config.max_bitrate_bps = 100000;
+    task_queue_->SendTask([&]() {
+      sender_call = rtc::WrapUnique(Call::Create(config));
+      receiver_call = rtc::WrapUnique(Call::Create(config));
+      sender_transport =
+          rtc::WrapUnique(CreateSendTransport(task_queue_, sender_call.get()));
+      receiver_transport = rtc::WrapUnique(
+          CreateReceiveTransport(task_queue_, receiver_call.get()));
 
-      UpdateSendConfig(i, &send_config, &encoder_config, &frame_generators[i]);
+      sender_transport->SetReceiver(receiver_call->Receiver());
+      receiver_transport->SetReceiver(sender_call->Receiver());
 
-      send_streams[i] = sender_call->CreateVideoSendStream(
-          send_config.Copy(), encoder_config.Copy());
-      send_streams[i]->Start();
+      for (size_t i = 0; i < kNumStreams; ++i)
+        encoders[i].reset(VP8Encoder::Create());
 
-      VideoReceiveStream::Config receive_config(receiver_transport.get());
-      receive_config.rtp.remote_ssrc = ssrc;
-      receive_config.rtp.local_ssrc = test::CallTest::kReceiverLocalVideoSsrc;
-      VideoReceiveStream::Decoder decoder =
-          test::CreateMatchingDecoder(send_config.encoder_settings);
-      allocated_decoders.push_back(
-          std::unique_ptr<VideoDecoder>(decoder.decoder));
-      receive_config.decoders.push_back(decoder);
+      for (size_t i = 0; i < kNumStreams; ++i) {
+        uint32_t ssrc = codec_settings[i].ssrc;
+        int width = codec_settings[i].width;
+        int height = codec_settings[i].height;
 
-      UpdateReceiveConfig(i, &receive_config);
+        VideoSendStream::Config send_config(sender_transport.get());
+        send_config.rtp.ssrcs.push_back(ssrc);
+        send_config.encoder_settings.encoder = encoders[i].get();
+        send_config.encoder_settings.payload_name = "VP8";
+        send_config.encoder_settings.payload_type = kVideoPayloadType;
+        VideoEncoderConfig encoder_config;
+        test::FillEncoderConfiguration(1, &encoder_config);
+        encoder_config.max_bitrate_bps = 100000;
 
-      receive_streams[i] =
-          receiver_call->CreateVideoReceiveStream(std::move(receive_config));
-      receive_streams[i]->Start();
+        UpdateSendConfig(i, &send_config, &encoder_config,
+                         &frame_generators[i]);
 
-      frame_generators[i] = test::FrameGeneratorCapturer::Create(
-          width, height, 30, Clock::GetRealTimeClock());
-      send_streams[i]->SetSource(
-          frame_generators[i],
-          VideoSendStream::DegradationPreference::kMaintainFramerate);
-      frame_generators[i]->Start();
-    }
+        send_streams[i] = sender_call->CreateVideoSendStream(
+            send_config.Copy(), encoder_config.Copy());
+        send_streams[i]->Start();
+
+        VideoReceiveStream::Config receive_config(receiver_transport.get());
+        receive_config.rtp.remote_ssrc = ssrc;
+        receive_config.rtp.local_ssrc = test::CallTest::kReceiverLocalVideoSsrc;
+        VideoReceiveStream::Decoder decoder =
+            test::CreateMatchingDecoder(send_config.encoder_settings);
+        allocated_decoders.push_back(
+            std::unique_ptr<VideoDecoder>(decoder.decoder));
+        receive_config.decoders.push_back(decoder);
+
+        UpdateReceiveConfig(i, &receive_config);
+
+        receive_streams[i] =
+            receiver_call->CreateVideoReceiveStream(std::move(receive_config));
+        receive_streams[i]->Start();
+
+        frame_generators[i] = test::FrameGeneratorCapturer::Create(
+            width, height, 30, Clock::GetRealTimeClock());
+        send_streams[i]->SetSource(
+            frame_generators[i],
+            VideoSendStream::DegradationPreference::kMaintainFramerate);
+        frame_generators[i]->Start();
+      }
+    });
 
     Wait();
 
-    for (size_t i = 0; i < kNumStreams; ++i) {
-      frame_generators[i]->Stop();
-      sender_call->DestroyVideoSendStream(send_streams[i]);
-      receiver_call->DestroyVideoReceiveStream(receive_streams[i]);
-      delete frame_generators[i];
-    }
+    task_queue_->SendTask([&]() {
+      for (size_t i = 0; i < kNumStreams; ++i) {
+        frame_generators[i]->Stop();
+        sender_call->DestroyVideoSendStream(send_streams[i]);
+        receiver_call->DestroyVideoReceiveStream(receive_streams[i]);
+        delete frame_generators[i];
+      }
 
-    sender_transport->StopSending();
-    receiver_transport->StopSending();
+      sender_transport.reset();
+      receiver_transport.reset();
+
+      sender_call.reset();
+      receiver_call.reset();
+    });
   }
 
  protected:
   virtual void Wait() = 0;
   // Note: frame_generator is a point-to-pointer, since the actual instance
   // hasn't been created at the time of this call. Only when packets/frames
   // start flowing should this be dereferenced.
   virtual void UpdateSendConfig(
       size_t stream_index,
       VideoSendStream::Config* send_config,
       VideoEncoderConfig* encoder_config,
       test::FrameGeneratorCapturer** frame_generator) {}
   virtual void UpdateReceiveConfig(size_t stream_index,
                                    VideoReceiveStream::Config* receive_config) {
   }
-  virtual test::DirectTransport* CreateSendTransport(Call* sender_call) {
-    return new test::DirectTransport(sender_call, payload_type_map_);
+  virtual test::DirectTransport* CreateSendTransport(
+      test::SingleThreadedTaskQueueForTesting* task_queue,
+      Call* sender_call) {
+    return new test::DirectTransport(task_queue, sender_call,
+                                     payload_type_map_);
   }
-  virtual test::DirectTransport* CreateReceiveTransport(Call* receiver_call) {
-    return new test::DirectTransport(receiver_call, payload_type_map_);
+  virtual test::DirectTransport* CreateReceiveTransport(
+      test::SingleThreadedTaskQueueForTesting* task_queue,
+      Call* receiver_call) {
+    return new test::DirectTransport(task_queue, receiver_call,
+                                     payload_type_map_);
   }
+
+  test::SingleThreadedTaskQueueForTesting* const task_queue_;
 };
 
 // Each renderer verifies that it receives the expected resolution, and as soon
 // as every renderer has received a frame, the test finishes.
 TEST_F(EndToEndTest, SendsAndReceivesMultipleStreams) {
   class VideoOutputObserver : public rtc::VideoSinkInterface<VideoFrame> {
    public:
     VideoOutputObserver(const MultiStreamTest::CodecSettings& settings,
                         uint32_t ssrc,
                         test::FrameGeneratorCapturer** frame_generator)
@@ skipping 15 matching lines @@
 
    private:
     const MultiStreamTest::CodecSettings& settings_;
     const uint32_t ssrc_;
     test::FrameGeneratorCapturer** const frame_generator_;
     rtc::Event done_;
   };
 
   class Tester : public MultiStreamTest {
    public:
-    Tester() {}
+    explicit Tester(test::SingleThreadedTaskQueueForTesting* task_queue)
+        : MultiStreamTest(task_queue) {}
     virtual ~Tester() {}
 
    protected:
     void Wait() override {
       for (const auto& observer : observers_) {
         EXPECT_TRUE(observer->Wait()) << "Time out waiting for from on ssrc "
                                       << observer->Ssrc();
       }
     }
 
     void UpdateSendConfig(
         size_t stream_index,
         VideoSendStream::Config* send_config,
         VideoEncoderConfig* encoder_config,
         test::FrameGeneratorCapturer** frame_generator) override {
       observers_[stream_index].reset(new VideoOutputObserver(
           codec_settings[stream_index], send_config->rtp.ssrcs.front(),
           frame_generator));
     }
 
     void UpdateReceiveConfig(
         size_t stream_index,
         VideoReceiveStream::Config* receive_config) override {
       receive_config->renderer = observers_[stream_index].get();
     }
 
    private:
     std::unique_ptr<VideoOutputObserver> observers_[kNumStreams];
-  } tester;
+  } tester(&task_queue_);
 
   tester.RunTest();
 }
 
 TEST_F(EndToEndTest, AssignsTransportSequenceNumbers) {
   static const int kExtensionId = 5;
 
   class RtpExtensionHeaderObserver : public test::DirectTransport {
    public:
     RtpExtensionHeaderObserver(
+        test::SingleThreadedTaskQueueForTesting* task_queue,
         Call* sender_call,
         const uint32_t& first_media_ssrc,
         const std::map<uint32_t, uint32_t>& ssrc_map,
         const std::map<uint8_t, MediaType>& payload_type_map)
-        : DirectTransport(sender_call, payload_type_map),
+        : DirectTransport(task_queue, sender_call, payload_type_map),
           done_(false, false),
           parser_(RtpHeaderParser::Create()),
           first_media_ssrc_(first_media_ssrc),
           rtx_to_media_ssrcs_(ssrc_map),
           padding_observed_(false),
           rtx_padding_observed_(false),
           retransmit_observed_(false),
           started_(false) {
       parser_->RegisterRtpHeaderExtension(kRtpExtensionTransportSequenceNumber,
                                           kExtensionId);
@@ skipping 94 matching lines @@
     const uint32_t& first_media_ssrc_;
     const std::map<uint32_t, uint32_t>& rtx_to_media_ssrcs_;
     bool padding_observed_;
     bool rtx_padding_observed_;
     bool retransmit_observed_;
     bool started_;
   };
 
   class TransportSequenceNumberTester : public MultiStreamTest {
    public:
-    TransportSequenceNumberTester()
-        : first_media_ssrc_(0), observer_(nullptr) {}
+    explicit TransportSequenceNumberTester(
+        test::SingleThreadedTaskQueueForTesting* task_queue)
+        : MultiStreamTest(task_queue),
+          first_media_ssrc_(0),
+          observer_(nullptr) {}
     virtual ~TransportSequenceNumberTester() {}
 
    protected:
     void Wait() override {
       RTC_DCHECK(observer_);
       EXPECT_TRUE(observer_->Wait());
     }
 
     void UpdateSendConfig(
         size_t stream_index,
@@ skipping 26 matching lines @@
     void UpdateReceiveConfig(
         size_t stream_index,
         VideoReceiveStream::Config* receive_config) override {
       receive_config->rtp.nack.rtp_history_ms = kNackRtpHistoryMs;
       receive_config->rtp.extensions.clear();
       receive_config->rtp.extensions.push_back(RtpExtension(
           RtpExtension::kTransportSequenceNumberUri, kExtensionId));
       receive_config->renderer = &fake_renderer_;
     }
 
-    test::DirectTransport* CreateSendTransport(Call* sender_call) override {
+    test::DirectTransport* CreateSendTransport(
+        test::SingleThreadedTaskQueueForTesting* task_queue,
+        Call* sender_call) override {
       std::map<uint8_t, MediaType> payload_type_map =
           MultiStreamTest::payload_type_map_;
       RTC_DCHECK(payload_type_map.find(kSendRtxPayloadType) ==
                  payload_type_map.end());
       payload_type_map[kSendRtxPayloadType] = MediaType::VIDEO;
-      observer_ =
-          new RtpExtensionHeaderObserver(sender_call, first_media_ssrc_,
-                                         rtx_to_media_ssrcs_, payload_type_map);
+      observer_ = new RtpExtensionHeaderObserver(
+          task_queue, sender_call, first_media_ssrc_, rtx_to_media_ssrcs_,
+          payload_type_map);
       return observer_;
     }
 
    private:
     test::FakeVideoRenderer fake_renderer_;
     uint32_t first_media_ssrc_;
     std::map<uint32_t, uint32_t> rtx_to_media_ssrcs_;
     RtpExtensionHeaderObserver* observer_;
-  } tester;
+  } tester(&task_queue_);
 
   tester.RunTest();
 }
 
 class TransportFeedbackTester : public test::EndToEndTest {
  public:
   TransportFeedbackTester(bool feedback_enabled,
                           size_t num_video_streams,
                           size_t num_audio_streams)
       : EndToEndTest(::webrtc::EndToEndTest::kDefaultTimeoutMs),
@@ skipping 192 matching lines @@
 
    private:
     std::unique_ptr<uint8_t[]> buffer_;
     size_t length_;
     FrameType frame_type_;
     rtc::Event called_;
   };
 
   EncodedFrameTestObserver post_encode_observer;
   EncodedFrameTestObserver pre_decode_observer;
+  test::FrameForwarder forwarder;
+  std::unique_ptr<test::FrameGenerator> frame_generator;
 
-  CreateCalls(Call::Config(event_log_.get()), Call::Config(event_log_.get()));
+  std::unique_ptr<test::DirectTransport> sender_transport;
+  std::unique_ptr<test::DirectTransport> receiver_transport;
 
-  test::DirectTransport sender_transport(sender_call_.get(), payload_type_map_);
-  test::DirectTransport receiver_transport(receiver_call_.get(),
-                                           payload_type_map_);
-  sender_transport.SetReceiver(receiver_call_->Receiver());
-  receiver_transport.SetReceiver(sender_call_->Receiver());
+  task_queue_.SendTask([&]() {
+    CreateCalls(Call::Config(event_log_.get()), Call::Config(event_log_.get()));
 
-  CreateSendConfig(1, 0, 0, &sender_transport);
-  CreateMatchingReceiveConfigs(&receiver_transport);
-  video_send_config_.post_encode_callback = &post_encode_observer;
-  video_receive_configs_[0].pre_decode_callback = &pre_decode_observer;
+    sender_transport = rtc::MakeUnique<test::DirectTransport>(
+        &task_queue_, sender_call_.get(), payload_type_map_);
+    receiver_transport = rtc::MakeUnique<test::DirectTransport>(
+        &task_queue_, receiver_call_.get(), payload_type_map_);
+    sender_transport->SetReceiver(receiver_call_->Receiver());
+    receiver_transport->SetReceiver(sender_call_->Receiver());
 
-  CreateVideoStreams();
-  Start();
+    CreateSendConfig(1, 0, 0, sender_transport.get());
+    CreateMatchingReceiveConfigs(receiver_transport.get());
+    video_send_config_.post_encode_callback = &post_encode_observer;
+    video_receive_configs_[0].pre_decode_callback = &pre_decode_observer;
 
-  std::unique_ptr<test::FrameGenerator> frame_generator(
-      test::FrameGenerator::CreateSquareGenerator(kDefaultWidth,
-                                                  kDefaultHeight));
-  test::FrameForwarder forwarder;
-  video_send_stream_->SetSource(
-      &forwarder, VideoSendStream::DegradationPreference::kMaintainFramerate);
-  forwarder.IncomingCapturedFrame(*frame_generator->NextFrame());
+    CreateVideoStreams();
+    Start();
+
+    frame_generator = test::FrameGenerator::CreateSquareGenerator(
+        kDefaultWidth, kDefaultHeight);
+    video_send_stream_->SetSource(
+        &forwarder, VideoSendStream::DegradationPreference::kMaintainFramerate);
+    forwarder.IncomingCapturedFrame(*frame_generator->NextFrame());
+  });
 
   EXPECT_TRUE(post_encode_observer.Wait())
       << "Timed out while waiting for send-side encoded-frame callback.";
 
   EXPECT_TRUE(pre_decode_observer.Wait())
       << "Timed out while waiting for pre-decode encoded-frame callback.";
 
   post_encode_observer.ExpectEqualFrames(pre_decode_observer);
 
-  Stop();
-
-  sender_transport.StopSending();
-  receiver_transport.StopSending();
-
-  DestroyStreams();
+  task_queue_.SendTask([this, &sender_transport, &receiver_transport]() {
+    Stop();
+    DestroyStreams();
+    sender_transport.reset();
+    receiver_transport.reset();
+    DestroyCalls();
+  });
 }
 
 TEST_F(EndToEndTest, ReceiveStreamSendsRemb) {
   class RembObserver : public test::EndToEndTest {
    public:
     RembObserver() : EndToEndTest(kDefaultTimeoutMs) {}
 
     void ModifyVideoConfigs(
         VideoSendStream::Config* send_config,
         std::vector<VideoReceiveStream::Config>* receive_configs,
@@ skipping 107 matching lines @@
           receive_transport_(nullptr),
           stop_event_(false, false),
           poller_thread_(&BitrateStatsPollingThread,
                          this,
                          "BitrateStatsPollingThread"),
           state_(kWaitForFirstRampUp),
           retransmission_rate_limiter_(clock_, 1000) {}
 
     ~BweObserver() {}
 
-    test::PacketTransport* CreateReceiveTransport() override {
+    test::PacketTransport* CreateReceiveTransport(
+        test::SingleThreadedTaskQueueForTesting* task_queue) override {
       receive_transport_ = new test::PacketTransport(
-          nullptr, this, test::PacketTransport::kReceiver, payload_type_map_,
-          FakeNetworkPipe::Config());
+          task_queue, nullptr, this, test::PacketTransport::kReceiver,
+          payload_type_map_, FakeNetworkPipe::Config());
       return receive_transport_;
     }
 
     Call::Config GetSenderCallConfig() override {
       Call::Config config(event_log_.get());
       // Set a high start bitrate to reduce the test completion time.
       config.bitrate_config.start_bitrate_bps = remb_bitrate_bps_;
       return config;
     }
 
@@ skipping 86 matching lines @@
     TestState state_;
     RateLimiter retransmission_rate_limiter_;
   } test;
 
   RunBaseTest(&test);
 }
 
 TEST_F(EndToEndTest, StopSendingKeyframeRequestsForInactiveStream) {
   class KeyframeRequestObserver : public test::EndToEndTest {
    public:
-    KeyframeRequestObserver() : clock_(Clock::GetRealTimeClock()) {}
+    explicit KeyframeRequestObserver(
+        test::SingleThreadedTaskQueueForTesting* task_queue)
+        : clock_(Clock::GetRealTimeClock()), task_queue_(task_queue) {}
 
     void OnVideoStreamsCreated(
         VideoSendStream* send_stream,
         const std::vector<VideoReceiveStream*>& receive_streams) override {
       RTC_DCHECK_EQ(1, receive_streams.size());
       send_stream_ = send_stream;
       receive_stream_ = receive_streams[0];
     }
 
     void PerformTest() override {
       bool frame_decoded = false;
       int64_t start_time = clock_->TimeInMilliseconds();
       while (clock_->TimeInMilliseconds() - start_time <= 5000) {
         if (receive_stream_->GetStats().frames_decoded > 0) {
           frame_decoded = true;
           break;
         }
         SleepMs(100);
       }
       ASSERT_TRUE(frame_decoded);
-      send_stream_->Stop();
+      task_queue_->SendTask([this]() { send_stream_->Stop(); });
       SleepMs(10000);
       ASSERT_EQ(
           1U, receive_stream_->GetStats().rtcp_packet_type_counts.pli_packets);
     }
 
    private:
     Clock* clock_;
     VideoSendStream* send_stream_;
     VideoReceiveStream* receive_stream_;
-  } test;
+    test::SingleThreadedTaskQueueForTesting* const task_queue_;
+  } test(&task_queue_);
 
   RunBaseTest(&test);
 }
 
 class ProbingTest : public test::EndToEndTest {
  public:
   explicit ProbingTest(int start_bitrate_bps)
       : clock_(Clock::GetRealTimeClock()),
         start_bitrate_bps_(start_bitrate_bps),
         state_(0),
@@ skipping 64 matching lines @@
 TEST_F(EndToEndTest, DISABLED_TriggerMidCallProbing) {
 // Fails on iOS bots: bugs.webrtc.org/7851
 #elif defined(TARGET_IPHONE_SIMULATOR) && TARGET_IPHONE_SIMULATOR
 TEST_F(EndToEndTest, DISABLED_TriggerMidCallProbing) {
 #else
 TEST_F(EndToEndTest, TriggerMidCallProbing) {
 #endif
 
   class TriggerMidCallProbingTest : public ProbingTest {
    public:
-    explicit TriggerMidCallProbingTest(bool* success)
-        : ProbingTest(300000), success_(success) {}
+    TriggerMidCallProbingTest(
+        test::SingleThreadedTaskQueueForTesting* task_queue,
+        bool* success)
+        : ProbingTest(300000), success_(success), task_queue_(task_queue) {}
 
     void PerformTest() override {
       *success_ = false;
       int64_t start_time_ms = clock_->TimeInMilliseconds();
       do {
         if (clock_->TimeInMilliseconds() - start_time_ms > kTimeoutMs)
           break;
 
         Call::Stats stats = sender_call_->GetStats();
 
         switch (state_) {
           case 0:
             if (stats.send_bandwidth_bps > 5 * 300000) {
               Call::Config::BitrateConfig bitrate_config;
               bitrate_config.max_bitrate_bps = 100000;
-              sender_call_->SetBitrateConfig(bitrate_config);
+              task_queue_->SendTask([this, &bitrate_config]() {
+                sender_call_->SetBitrateConfig(bitrate_config);
+              });
               ++state_;
             }
             break;
           case 1:
             if (stats.send_bandwidth_bps < 110000) {
               Call::Config::BitrateConfig bitrate_config;
               bitrate_config.max_bitrate_bps = 2500000;
-              sender_call_->SetBitrateConfig(bitrate_config);
+              task_queue_->SendTask([this, &bitrate_config]() {
+                sender_call_->SetBitrateConfig(bitrate_config);
+              });
               ++state_;
             }
             break;
           case 2:
             // During high cpu load the pacer will not be able to pace packets
             // at the correct speed, but if we go from 110 to 1250 kbps
             // in 5 seconds then it is due to probing.
             if (stats.send_bandwidth_bps > 1250000) {
               *success_ = true;
               observation_complete_.Set();
             }
             break;
         }
       } while (!observation_complete_.Wait(20));
     }
 
    private:
     const int kTimeoutMs = 5000;
     bool* const success_;
+    test::SingleThreadedTaskQueueForTesting* const task_queue_;
   };
 
   bool success = false;
   const int kMaxAttempts = 3;
   for (int i = 0; i < kMaxAttempts; ++i) {
-    TriggerMidCallProbingTest test(&success);
+    TriggerMidCallProbingTest test(&task_queue_, &success);
     RunBaseTest(&test);
     if (success)
       return;
   }
   EXPECT_TRUE(success) << "Failed to perform mid call probing (" << kMaxAttempts
                        << " attempts).";
 }
 
 TEST_F(EndToEndTest, VerifyNackStats) {
   static const int kPacketNumberToDrop = 200;
@@ skipping 211 matching lines @@
     const std::unique_ptr<VideoEncoder> vp8_encoder_;
     Call* sender_call_;
     Call* receiver_call_;
     int64_t start_runtime_ms_;
     int num_frames_received_ GUARDED_BY(&crit_);
   } test(use_rtx, use_red, screenshare);
 
   metrics::Reset();
   RunBaseTest(&test);
 
-  // Delete the call for Call stats to be reported.
-  sender_call_.reset();
-  receiver_call_.reset();
-
   std::string video_prefix =
       screenshare ? "WebRTC.Video.Screenshare." : "WebRTC.Video.";
 
   // Verify that stats have been updated once.
   EXPECT_EQ(2, metrics::NumSamples("WebRTC.Call.LifetimeInSeconds"));
   EXPECT_EQ(1, metrics::NumSamples(
                    "WebRTC.Call.TimeReceivingVideoRtpPacketsInSeconds"));
   EXPECT_EQ(1, metrics::NumSamples("WebRTC.Call.VideoBitrateReceivedInKbps"));
   EXPECT_EQ(1, metrics::NumSamples("WebRTC.Call.RtcpBitrateReceivedInBps"));
   EXPECT_EQ(1, metrics::NumSamples("WebRTC.Call.BitrateReceivedInKbps"));
@@ skipping 146 matching lines @@
 
     rtc::CriticalSection crit_;
     int num_frames_received_ GUARDED_BY(&crit_);
   } test;
 
   test::ScopedFieldTrials override_field_trials(
       "WebRTC-VideoContentTypeExtension/Enabled/");
   metrics::Reset();
 
   Call::Config send_config(test.GetSenderCallConfig());
-  CreateSenderCall(send_config);
   Call::Config recv_config(test.GetReceiverCallConfig());
-  CreateReceiverCall(recv_config);
-  receive_transport_.reset(test.CreateReceiveTransport());
-  send_transport_.reset(test.CreateSendTransport(sender_call_.get()));
-  send_transport_->SetReceiver(receiver_call_->Receiver());
-  receive_transport_->SetReceiver(sender_call_->Receiver());
-  receiver_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkUp);
-  CreateSendConfig(1, 0, 0, send_transport_.get());
-  CreateMatchingReceiveConfigs(receive_transport_.get());
+  VideoEncoderConfig encoder_config_with_screenshare;
 
-  // Modify send and receive configs.
-  video_send_config_.rtp.nack.rtp_history_ms = kNackRtpHistoryMs;
-  video_receive_configs_[0].rtp.nack.rtp_history_ms = kNackRtpHistoryMs;
-  video_receive_configs_[0].renderer = &test;
-  // RTT needed for RemoteNtpTimeEstimator for the receive stream.
-  video_receive_configs_[0].rtp.rtcp_xr.receiver_reference_time_report = true;
-  // Start with realtime video.
-  video_encoder_config_.content_type =
-      VideoEncoderConfig::ContentType::kRealtimeVideo;
-  // Second encoder config for the second part of the test uses screenshare
-  VideoEncoderConfig encoder_config_with_screenshare_ =
-      video_encoder_config_.Copy();
-  encoder_config_with_screenshare_.content_type =
-      VideoEncoderConfig::ContentType::kScreen;
+  task_queue_.SendTask([this, &test, &override_field_trials, &send_config,
+                        &recv_config, &encoder_config_with_screenshare]() {
+    CreateSenderCall(send_config);
+    CreateReceiverCall(recv_config);
 
-  CreateVideoStreams();
-  CreateFrameGeneratorCapturer(kDefaultFramerate, kDefaultWidth,
-                               kDefaultHeight);
-  Start();
+    receive_transport_.reset(test.CreateReceiveTransport(&task_queue_));
+    send_transport_.reset(
+        test.CreateSendTransport(&task_queue_, sender_call_.get()));
+    send_transport_->SetReceiver(receiver_call_->Receiver());
+    receive_transport_->SetReceiver(sender_call_->Receiver());
+
+    receiver_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkUp);
+    CreateSendConfig(1, 0, 0, send_transport_.get());
+    CreateMatchingReceiveConfigs(receive_transport_.get());
+
+    // Modify send and receive configs.
+    video_send_config_.rtp.nack.rtp_history_ms = kNackRtpHistoryMs;
+    video_receive_configs_[0].rtp.nack.rtp_history_ms = kNackRtpHistoryMs;
+    video_receive_configs_[0].renderer = &test;
+    // RTT needed for RemoteNtpTimeEstimator for the receive stream.
+    video_receive_configs_[0].rtp.rtcp_xr.receiver_reference_time_report = true;
+    // Start with realtime video.
+    video_encoder_config_.content_type =
+        VideoEncoderConfig::ContentType::kRealtimeVideo;
+    // Second encoder config for the second part of the test uses screenshare
+    encoder_config_with_screenshare = video_encoder_config_.Copy();
+    encoder_config_with_screenshare.content_type =
+        VideoEncoderConfig::ContentType::kScreen;
+
+    CreateVideoStreams();
+    CreateFrameGeneratorCapturer(kDefaultFramerate, kDefaultWidth,
+                                 kDefaultHeight);
+    Start();
+  });
 
   test.PerformTest();
 
   // Replace old send stream.
-  sender_call_->DestroyVideoSendStream(video_send_stream_);
-  video_send_stream_ = sender_call_->CreateVideoSendStream(
-      video_send_config_.Copy(), encoder_config_with_screenshare_.Copy());
-  video_send_stream_->SetSource(
-      frame_generator_capturer_.get(),
-      VideoSendStream::DegradationPreference::kBalanced);
-  video_send_stream_->Start();
+  task_queue_.SendTask([this, &encoder_config_with_screenshare]() {
+    sender_call_->DestroyVideoSendStream(video_send_stream_);
+    video_send_stream_ = sender_call_->CreateVideoSendStream(
+        video_send_config_.Copy(), encoder_config_with_screenshare.Copy());
+    video_send_stream_->SetSource(
+        frame_generator_capturer_.get(),
+        VideoSendStream::DegradationPreference::kBalanced);
+    video_send_stream_->Start();
+  });
 
   // Continue to run test but now with screenshare.
   test.PerformTest();
 
-  send_transport_->StopSending();
-  receive_transport_->StopSending();
-  Stop();
-  DestroyStreams();
-  DestroyCalls();
-  // Delete the call for Call stats to be reported.
-  sender_call_.reset();
-  receiver_call_.reset();
+  task_queue_.SendTask([this]() {
+    Stop();
+    DestroyStreams();
+    send_transport_.reset();
+    receive_transport_.reset();
+    DestroyCalls();
+  });
 
   // Verify that stats have been updated for both screenshare and video.
   EXPECT_EQ(1, metrics::NumSamples("WebRTC.Video.EndToEndDelayInMs"));
   EXPECT_EQ(1,
             metrics::NumSamples("WebRTC.Video.Screenshare.EndToEndDelayInMs"));
   EXPECT_EQ(1, metrics::NumSamples("WebRTC.Video.EndToEndDelayMaxInMs"));
   EXPECT_EQ(
       1, metrics::NumSamples("WebRTC.Video.Screenshare.EndToEndDelayMaxInMs"));
   EXPECT_EQ(1, metrics::NumSamples("WebRTC.Video.InterframeDelayInMs"));
   EXPECT_EQ(1,
@@ skipping 143 matching lines @@
     VideoEncoderConfig video_encoder_config_all_streams_;
   } test(kVideoSendSsrcs, num_ssrcs, send_single_ssrc_first);
 
   RunBaseTest(&test);
 }
 
 TEST_F(EndToEndTest, ReportsSetEncoderRates) {
   class EncoderRateStatsTest : public test::EndToEndTest,
                                public test::FakeEncoder {
    public:
-    EncoderRateStatsTest()
+    explicit EncoderRateStatsTest(
+        test::SingleThreadedTaskQueueForTesting* task_queue)
         : EndToEndTest(kDefaultTimeoutMs),
           FakeEncoder(Clock::GetRealTimeClock()),
+          task_queue_(task_queue),
           send_stream_(nullptr),
           bitrate_kbps_(0) {}
 
     void OnVideoStreamsCreated(
         VideoSendStream* send_stream,
         const std::vector<VideoReceiveStream*>& receive_streams) override {
       send_stream_ = send_stream;
     }
 
     void ModifyVideoConfigs(
@@ skipping 11 matching lines @@
         return 0;
       rtc::CritScope lock(&crit_);
       bitrate_kbps_ = rate_allocation.get_sum_kbps();
       observation_complete_.Set();
       return 0;
     }
 
     void PerformTest() override {
       ASSERT_TRUE(Wait())
           << "Timed out while waiting for encoder SetRates() call.";
-      WaitForEncoderTargetBitrateMatchStats();
-      send_stream_->Stop();
-      WaitForStatsReportZeroTargetBitrate();
-      send_stream_->Start();
-      WaitForEncoderTargetBitrateMatchStats();
+
+      task_queue_->SendTask([this]() {
+        WaitForEncoderTargetBitrateMatchStats();
+        send_stream_->Stop();
+        WaitForStatsReportZeroTargetBitrate();
+        send_stream_->Start();
+        WaitForEncoderTargetBitrateMatchStats();
+      });
     }
 
     void WaitForEncoderTargetBitrateMatchStats() {
       for (int i = 0; i < kDefaultTimeoutMs; ++i) {
         VideoSendStream::Stats stats = send_stream_->GetStats();
         {
           rtc::CritScope lock(&crit_);
           if ((stats.target_media_bitrate_bps + 500) / 1000 ==
               static_cast<int>(bitrate_kbps_)) {
             return;
           }
         }
         SleepMs(1);
       }
       FAIL()
           << "Timed out waiting for stats reporting the currently set bitrate.";
     }
 
     void WaitForStatsReportZeroTargetBitrate() {
       for (int i = 0; i < kDefaultTimeoutMs; ++i) {
         if (send_stream_->GetStats().target_media_bitrate_bps == 0) {
           return;
         }
         SleepMs(1);
       }
       FAIL() << "Timed out waiting for stats reporting zero bitrate.";
     }
 
    private:
+    test::SingleThreadedTaskQueueForTesting* const task_queue_;
     rtc::CriticalSection crit_;
     VideoSendStream* send_stream_;
     uint32_t bitrate_kbps_ GUARDED_BY(crit_);
-  } test;
+  } test(&task_queue_);
 
   RunBaseTest(&test);
 }
 
 TEST_F(EndToEndTest, GetStats) {
   static const int kStartBitrateBps = 3000000;
   static const int kExpectedRenderDelayMs = 20;
 
   class ReceiveStreamRenderer : public rtc::VideoSinkInterface<VideoFrame> {
    public:
@@ skipping 186 matching lines @@
     }
 
     bool AllStatsFilled(const std::map<std::string, bool>& stats_map) {
       for (const auto& stat : stats_map) {
         if (!stat.second)
           return false;
       }
       return true;
     }
 
-    test::PacketTransport* CreateSendTransport(Call* sender_call) override {
+    test::PacketTransport* CreateSendTransport(
+        test::SingleThreadedTaskQueueForTesting* task_queue,
+        Call* sender_call) override {
       FakeNetworkPipe::Config network_config;
       network_config.loss_percent = 5;
-      return new test::PacketTransport(sender_call, this,
+      return new test::PacketTransport(task_queue, sender_call, this,
                                        test::PacketTransport::kSender,
                                        payload_type_map_, network_config);
     }
 
     Call::Config GetSenderCallConfig() override {
       Call::Config config = EndToEndTest::GetSenderCallConfig();
       config.bitrate_config.start_bitrate_bps = kStartBitrateBps;
       return config;
     }
 
@@ skipping 84 matching lines @@
 
         int64_t time_until_timout_ = stop_time - now;
         if (time_until_timout_ > 0)
           check_stats_event_.Wait(time_until_timout_);
         now = clock->TimeInMilliseconds();
       }
 
       ADD_FAILURE() << "Timed out waiting for filled stats.";
       for (std::map<std::string, bool>::const_iterator it =
                receive_stats_filled_.begin();
-           it != receive_stats_filled_.end();
-           ++it) {
+           it != receive_stats_filled_.end(); ++it) {
         if (!it->second) {
           ADD_FAILURE() << "Missing receive stats: " << it->first;
         }
       }
 
       for (std::map<std::string, bool>::const_iterator it =
                send_stats_filled_.begin();
-           it != send_stats_filled_.end();
-           ++it) {
+           it != send_stats_filled_.end(); ++it) {
         if (!it->second) {
           ADD_FAILURE() << "Missing send stats: " << it->first;
         }
       }
     }
 
     test::DelayedEncoder encoder_;
     std::vector<VideoReceiveStream*> receive_streams_;
     std::map<std::string, bool> receive_stats_filled_;
 
@@ skipping 468 matching lines @@
     SequenceNumberUnwrapper seq_numbers_unwrapper_;
     std::map<uint32_t, std::list<int64_t>> last_observed_seq_numbers_;
     std::map<uint32_t, uint32_t> last_observed_timestamp_;
     std::map<uint32_t, bool> ssrc_is_rtx_;
 
     rtc::CriticalSection crit_;
     size_t ssrcs_to_observe_ GUARDED_BY(crit_);
     std::map<uint32_t, bool> ssrc_observed_ GUARDED_BY(crit_);
   } observer(use_rtx);
 
+  std::unique_ptr<test::PacketTransport> send_transport;
+  std::unique_ptr<test::PacketTransport> receive_transport;
+
   Call::Config config(event_log_.get());
-  CreateCalls(config, config);
+  VideoEncoderConfig one_stream;
 
-  test::PacketTransport send_transport(
-      sender_call_.get(), &observer, test::PacketTransport::kSender,
-      payload_type_map_, FakeNetworkPipe::Config());
-  test::PacketTransport receive_transport(
-      nullptr, &observer, test::PacketTransport::kReceiver, payload_type_map_,
-      FakeNetworkPipe::Config());
-  send_transport.SetReceiver(receiver_call_->Receiver());
-  receive_transport.SetReceiver(sender_call_->Receiver());
+  task_queue_.SendTask([this, &observer, &send_transport, &receive_transport,
+                        &config, &one_stream, use_rtx]() {
+    CreateCalls(config, config);
 
-  CreateSendConfig(kNumSsrcs, 0, 0, &send_transport);
+    send_transport = rtc::MakeUnique<test::PacketTransport>(
+        &task_queue_, sender_call_.get(), &observer,
+        test::PacketTransport::kSender, payload_type_map_,
+        FakeNetworkPipe::Config());
+    receive_transport = rtc::MakeUnique<test::PacketTransport>(
+        &task_queue_, nullptr, &observer, test::PacketTransport::kReceiver,
+        payload_type_map_, FakeNetworkPipe::Config());
+    send_transport->SetReceiver(receiver_call_->Receiver());
+    receive_transport->SetReceiver(sender_call_->Receiver());
 
-  if (use_rtx) {
-    for (size_t i = 0; i < kNumSsrcs; ++i) {
-      video_send_config_.rtp.rtx.ssrcs.push_back(kSendRtxSsrcs[i]);
+    CreateSendConfig(kNumSsrcs, 0, 0, send_transport.get());
+
+    if (use_rtx) {
+      for (size_t i = 0; i < kNumSsrcs; ++i) {
+        video_send_config_.rtp.rtx.ssrcs.push_back(kSendRtxSsrcs[i]);
+      }
+      video_send_config_.rtp.rtx.payload_type = kSendRtxPayloadType;
     }
-    video_send_config_.rtp.rtx.payload_type = kSendRtxPayloadType;
-  }
 
-  video_encoder_config_.video_stream_factory =
-      new rtc::RefCountedObject<VideoStreamFactory>();
-  // Use the same total bitrates when sending a single stream to avoid lowering
-  // the bitrate estimate and requiring a subsequent rampup.
-  VideoEncoderConfig one_stream = video_encoder_config_.Copy();
-  // one_stream.streams.resize(1);
-  one_stream.number_of_streams = 1;
-  CreateMatchingReceiveConfigs(&receive_transport);
+    video_encoder_config_.video_stream_factory =
+        new rtc::RefCountedObject<VideoStreamFactory>();
+    // Use the same total bitrates when sending a single stream to avoid
+    // lowering the bitrate estimate and requiring a subsequent rampup.
+    one_stream = video_encoder_config_.Copy();
+    // one_stream.streams.resize(1);
+    one_stream.number_of_streams = 1;
+    CreateMatchingReceiveConfigs(receive_transport.get());
 
-  CreateVideoStreams();
-  CreateFrameGeneratorCapturer(30, 1280, 720);
+    CreateVideoStreams();
+    CreateFrameGeneratorCapturer(30, 1280, 720);
 
-  Start();
+    Start();
+  });
+
   EXPECT_TRUE(observer.Wait())
       << "Timed out waiting for all SSRCs to send packets.";
 
   // Test stream resetting more than once to make sure that the state doesn't
   // get set once (this could be due to using std::map::insert for instance).
   for (size_t i = 0; i < 3; ++i) {
-    frame_generator_capturer_->Stop();
-    sender_call_->DestroyVideoSendStream(video_send_stream_);
+    task_queue_.SendTask([&]() {
+      frame_generator_capturer_->Stop();
+      sender_call_->DestroyVideoSendStream(video_send_stream_);
 
-    // Re-create VideoSendStream with only one stream.
-    video_send_stream_ = sender_call_->CreateVideoSendStream(
-        video_send_config_.Copy(), one_stream.Copy());
-    video_send_stream_->Start();
-    if (provoke_rtcpsr_before_rtp) {
-      // Rapid Resync Request forces sending RTCP Sender Report back.
-      // Using this request speeds up this test because then there is no need
-      // to wait for a second for periodic Sender Report.
-      rtcp::RapidResyncRequest force_send_sr_back_request;
-      rtc::Buffer packet = force_send_sr_back_request.Build();
-      static_cast<webrtc::test::DirectTransport&>(receive_transport)
-          .SendRtcp(packet.data(), packet.size());
-    }
-    CreateFrameGeneratorCapturer(30, 1280, 720);
-    frame_generator_capturer_->Start();
+      // Re-create VideoSendStream with only one stream.
+      video_send_stream_ = sender_call_->CreateVideoSendStream(
+          video_send_config_.Copy(), one_stream.Copy());
+      video_send_stream_->Start();
+      if (provoke_rtcpsr_before_rtp) {
+        // Rapid Resync Request forces sending RTCP Sender Report back.
+        // Using this request speeds up this test because then there is no need
+        // to wait for a second for periodic Sender Report.
+        rtcp::RapidResyncRequest force_send_sr_back_request;
+        rtc::Buffer packet = force_send_sr_back_request.Build();
+        static_cast<webrtc::test::DirectTransport*>(receive_transport.get())
+            ->SendRtcp(packet.data(), packet.size());
+      }
+      CreateFrameGeneratorCapturer(30, 1280, 720);
+      frame_generator_capturer_->Start();
+    });
 
     observer.ResetExpectedSsrcs(1);
     EXPECT_TRUE(observer.Wait()) << "Timed out waiting for single RTP packet.";
 
     // Reconfigure back to use all streams.
-    video_send_stream_->ReconfigureVideoEncoder(video_encoder_config_.Copy());
+    task_queue_.SendTask([this]() {
+      video_send_stream_->ReconfigureVideoEncoder(video_encoder_config_.Copy());
+    });
     observer.ResetExpectedSsrcs(kNumSsrcs);
     EXPECT_TRUE(observer.Wait())
         << "Timed out waiting for all SSRCs to send packets.";
 
     // Reconfigure down to one stream.
-    video_send_stream_->ReconfigureVideoEncoder(one_stream.Copy());
+    task_queue_.SendTask([this, &one_stream]() {
+      video_send_stream_->ReconfigureVideoEncoder(one_stream.Copy());
+    });
     observer.ResetExpectedSsrcs(1);
     EXPECT_TRUE(observer.Wait()) << "Timed out waiting for single RTP packet.";
 
     // Reconfigure back to use all streams.
-    video_send_stream_->ReconfigureVideoEncoder(video_encoder_config_.Copy());
+    task_queue_.SendTask([this]() {
+      video_send_stream_->ReconfigureVideoEncoder(video_encoder_config_.Copy());
+    });
     observer.ResetExpectedSsrcs(kNumSsrcs);
     EXPECT_TRUE(observer.Wait())
         << "Timed out waiting for all SSRCs to send packets.";
   }
 
-  send_transport.StopSending();
-  receive_transport.StopSending();
-
-  Stop();
-  DestroyStreams();
+  task_queue_.SendTask([this, &send_transport, &receive_transport]() {
+    Stop();
+    DestroyStreams();
+    send_transport.reset();
+    receive_transport.reset();
+    DestroyCalls();
+  });
 }
 
 TEST_F(EndToEndTest, RestartingSendStreamPreservesRtpState) {
   TestRtpStatePreservation(false, false);
 }
 
 TEST_F(EndToEndTest, RestartingSendStreamPreservesRtpStatesWithRtx) {
   TestRtpStatePreservation(true, false);
 }
 
@@ skipping 70 matching lines @@
 
       return SEND_PACKET;
     }
 
     rtc::Optional<uint16_t> last_observed_sequence_number_ GUARDED_BY(crit_);
     rtc::Optional<uint32_t> last_observed_timestamp_ GUARDED_BY(crit_);
     size_t num_flexfec_packets_sent_ GUARDED_BY(crit_);
     rtc::CriticalSection crit_;
   } observer;
 
+  constexpr int kFrameMaxWidth = 320;
+  constexpr int kFrameMaxHeight = 180;
+  constexpr int kFrameRate = 15;
+
   Call::Config config(event_log_.get());
-  CreateCalls(config, config);
 
-  FakeNetworkPipe::Config lossy_delayed_link;
-  lossy_delayed_link.loss_percent = 2;
-  lossy_delayed_link.queue_delay_ms = 50;
-  test::PacketTransport send_transport(sender_call_.get(), &observer,
-                                       test::PacketTransport::kSender,
-                                       payload_type_map_, lossy_delayed_link);
-  send_transport.SetReceiver(receiver_call_->Receiver());
+  std::unique_ptr<test::PacketTransport> send_transport;
+  std::unique_ptr<test::PacketTransport> receive_transport;
+  std::unique_ptr<VideoEncoder> encoder;
 
-  FakeNetworkPipe::Config flawless_link;
-  test::PacketTransport receive_transport(nullptr, &observer,
-                                          test::PacketTransport::kReceiver,
-                                          payload_type_map_, flawless_link);
-  receive_transport.SetReceiver(sender_call_->Receiver());
+  task_queue_.SendTask([&]() {
+    CreateCalls(config, config);
 
-  // For reduced flakyness, we use a real VP8 encoder together with NACK
-  // and RTX.
-  const int kNumVideoStreams = 1;
-  const int kNumFlexfecStreams = 1;
-  CreateSendConfig(kNumVideoStreams, 0, kNumFlexfecStreams, &send_transport);
-  std::unique_ptr<VideoEncoder> encoder(VP8Encoder::Create());
-  video_send_config_.encoder_settings.encoder = encoder.get();
-  video_send_config_.encoder_settings.payload_name = "VP8";
-  video_send_config_.encoder_settings.payload_type = kVideoSendPayloadType;
-  video_send_config_.rtp.nack.rtp_history_ms = kNackRtpHistoryMs;
-  video_send_config_.rtp.rtx.ssrcs.push_back(kSendRtxSsrcs[0]);
-  video_send_config_.rtp.rtx.payload_type = kSendRtxPayloadType;
+    FakeNetworkPipe::Config lossy_delayed_link;
+    lossy_delayed_link.loss_percent = 2;
+    lossy_delayed_link.queue_delay_ms = 50;
 
-  CreateMatchingReceiveConfigs(&receive_transport);
-  video_receive_configs_[0].rtp.nack.rtp_history_ms = kNackRtpHistoryMs;
-  video_receive_configs_[0].rtp.rtx_ssrc = kSendRtxSsrcs[0];
-  video_receive_configs_[0].rtp.rtx_payload_types[kVideoSendPayloadType] =
-      kSendRtxPayloadType;
+    send_transport = rtc::MakeUnique<test::PacketTransport>(
+        &task_queue_, sender_call_.get(), &observer,
+        test::PacketTransport::kSender, payload_type_map_, lossy_delayed_link);
+    send_transport->SetReceiver(receiver_call_->Receiver());
 
-  // The matching FlexFEC receive config is not created by
-  // CreateMatchingReceiveConfigs since this is not a test::BaseTest.
-  // Set up the receive config manually instead.
-  FlexfecReceiveStream::Config flexfec_receive_config(&receive_transport);
-  flexfec_receive_config.payload_type =
-      video_send_config_.rtp.flexfec.payload_type;
-  flexfec_receive_config.remote_ssrc = video_send_config_.rtp.flexfec.ssrc;
-  flexfec_receive_config.protected_media_ssrcs =
-      video_send_config_.rtp.flexfec.protected_media_ssrcs;
-  flexfec_receive_config.local_ssrc = kReceiverLocalVideoSsrc;
-  flexfec_receive_config.transport_cc = true;
-  flexfec_receive_config.rtp_header_extensions.emplace_back(
-      RtpExtension::kTransportSequenceNumberUri,
-      test::kTransportSequenceNumberExtensionId);
-  flexfec_receive_configs_.push_back(flexfec_receive_config);
+    FakeNetworkPipe::Config flawless_link;
+    receive_transport = rtc::MakeUnique<test::PacketTransport>(
+        &task_queue_, nullptr, &observer, test::PacketTransport::kReceiver,
+        payload_type_map_, flawless_link);
+    receive_transport->SetReceiver(sender_call_->Receiver());
 
-  CreateFlexfecStreams();
-  CreateVideoStreams();
+    // For reduced flakyness, we use a real VP8 encoder together with NACK
+    // and RTX.
+    const int kNumVideoStreams = 1;
+    const int kNumFlexfecStreams = 1;
+    CreateSendConfig(kNumVideoStreams, 0, kNumFlexfecStreams,
+                     send_transport.get());
+    encoder = rtc::WrapUnique(VP8Encoder::Create());
+    video_send_config_.encoder_settings.encoder = encoder.get();
+    video_send_config_.encoder_settings.payload_name = "VP8";
+    video_send_config_.encoder_settings.payload_type = kVideoSendPayloadType;
+    video_send_config_.rtp.nack.rtp_history_ms = kNackRtpHistoryMs;
+    video_send_config_.rtp.rtx.ssrcs.push_back(kSendRtxSsrcs[0]);
+    video_send_config_.rtp.rtx.payload_type = kSendRtxPayloadType;
 
-  // RTCP might be disabled if the network is "down".
-  sender_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkUp);
-  receiver_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkUp);
+    CreateMatchingReceiveConfigs(receive_transport.get());
+    video_receive_configs_[0].rtp.nack.rtp_history_ms = kNackRtpHistoryMs;
+    video_receive_configs_[0].rtp.rtx_ssrc = kSendRtxSsrcs[0];
+    video_receive_configs_[0].rtp.rtx_payload_types[kVideoSendPayloadType] =
+        kSendRtxPayloadType;
 
-  const int kFrameMaxWidth = 320;
-  const int kFrameMaxHeight = 180;
-  const int kFrameRate = 15;
-  CreateFrameGeneratorCapturer(kFrameRate, kFrameMaxWidth, kFrameMaxHeight);
+    // The matching FlexFEC receive config is not created by
+    // CreateMatchingReceiveConfigs since this is not a test::BaseTest.
+    // Set up the receive config manually instead.
+    FlexfecReceiveStream::Config flexfec_receive_config(
+        receive_transport.get());
+    flexfec_receive_config.payload_type =
+        video_send_config_.rtp.flexfec.payload_type;
+    flexfec_receive_config.remote_ssrc = video_send_config_.rtp.flexfec.ssrc;
+    flexfec_receive_config.protected_media_ssrcs =
+        video_send_config_.rtp.flexfec.protected_media_ssrcs;
+    flexfec_receive_config.local_ssrc = kReceiverLocalVideoSsrc;
+    flexfec_receive_config.transport_cc = true;
+    flexfec_receive_config.rtp_header_extensions.emplace_back(
+        RtpExtension::kTransportSequenceNumberUri,
+        test::kTransportSequenceNumberExtensionId);
+    flexfec_receive_configs_.push_back(flexfec_receive_config);
+
+    CreateFlexfecStreams();
+    CreateVideoStreams();
+
+    // RTCP might be disabled if the network is "down".
+    sender_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkUp);
+    receiver_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkUp);
+
+    CreateFrameGeneratorCapturer(kFrameRate, kFrameMaxWidth, kFrameMaxHeight);
+
+    Start();
+  });
 
   // Initial test.
-  Start();
   EXPECT_TRUE(observer.Wait()) << "Timed out waiting for packets.";
 
-  // Ensure monotonicity when the VideoSendStream is restarted.
-  Stop();
-  observer.ResetPacketCount();
-  Start();
+  task_queue_.SendTask([this, &observer]() {
+    // Ensure monotonicity when the VideoSendStream is restarted.
+    Stop();
+    observer.ResetPacketCount();
+    Start();
+  });
+
   EXPECT_TRUE(observer.Wait()) << "Timed out waiting for packets.";
 
-  // Ensure monotonicity when the VideoSendStream is recreated.
-  frame_generator_capturer_->Stop();
-  sender_call_->DestroyVideoSendStream(video_send_stream_);
-  observer.ResetPacketCount();
-  video_send_stream_ = sender_call_->CreateVideoSendStream(
-      video_send_config_.Copy(), video_encoder_config_.Copy());
-  video_send_stream_->Start();
-  CreateFrameGeneratorCapturer(kFrameRate, kFrameMaxWidth, kFrameMaxHeight);
-  frame_generator_capturer_->Start();
+  task_queue_.SendTask([this, &observer]() {
+    // Ensure monotonicity when the VideoSendStream is recreated.
+    frame_generator_capturer_->Stop();
+    sender_call_->DestroyVideoSendStream(video_send_stream_);
+    observer.ResetPacketCount();
+    video_send_stream_ = sender_call_->CreateVideoSendStream(
+        video_send_config_.Copy(), video_encoder_config_.Copy());
+    video_send_stream_->Start();
+    CreateFrameGeneratorCapturer(kFrameRate, kFrameMaxWidth, kFrameMaxHeight);
+    frame_generator_capturer_->Start();
+  });
+
   EXPECT_TRUE(observer.Wait()) << "Timed out waiting for packets.";
 
   // Cleanup.
-  send_transport.StopSending();
-  receive_transport.StopSending();
-  Stop();
-  DestroyStreams();
+  task_queue_.SendTask([this, &send_transport, &receive_transport]() {
+    Stop();
+    DestroyStreams();
+    send_transport.reset();
+    receive_transport.reset();
+    DestroyCalls();
+  });
 }
 
 TEST_F(EndToEndTest, RespectsNetworkState) {
   // TODO(pbos): Remove accepted downtime packets etc. when signaling network
   // down blocks until no more packets will be sent.
 
   // Pacer will send from its packet list and then send required padding before
   // checking paused_ again. This should be enough for one round of pacing,
   // otherwise increase.
   static const int kNumAcceptedDowntimeRtp = 5;
   // A single RTCP may be in the pipeline.
   static const int kNumAcceptedDowntimeRtcp = 1;
   class NetworkStateTest : public test::EndToEndTest, public test::FakeEncoder {
    public:
-    NetworkStateTest()
+    explicit NetworkStateTest(
+        test::SingleThreadedTaskQueueForTesting* task_queue)
         : EndToEndTest(kDefaultTimeoutMs),
           FakeEncoder(Clock::GetRealTimeClock()),
+          task_queue_(task_queue),
           encoded_frames_(false, false),
           packet_event_(false, false),
           sender_call_(nullptr),
           receiver_call_(nullptr),
           sender_state_(kNetworkUp),
           sender_rtp_(0),
           sender_padding_(0),
           sender_rtcp_(0),
           receiver_rtcp_(0),
           down_frames_(0) {}
@@ skipping 36 matching lines @@
     void ModifyVideoConfigs(
         VideoSendStream::Config* send_config,
         std::vector<VideoReceiveStream::Config>* receive_configs,
         VideoEncoderConfig* encoder_config) override {
       send_config->encoder_settings.encoder = this;
     }
 
     void PerformTest() override {
       EXPECT_TRUE(encoded_frames_.Wait(kDefaultTimeoutMs))
           << "No frames received by the encoder.";
-      // Wait for packets from both sender/receiver.
-      WaitForPacketsOrSilence(false, false);
 
-      // Sender-side network down for audio; there should be no effect on video
-      sender_call_->SignalChannelNetworkState(MediaType::AUDIO, kNetworkDown);
-      WaitForPacketsOrSilence(false, false);
+      task_queue_->SendTask([this]() {
+        // Wait for packets from both sender/receiver.
+        WaitForPacketsOrSilence(false, false);
 
-      // Receiver-side network down for audio; no change expected
-      receiver_call_->SignalChannelNetworkState(MediaType::AUDIO, kNetworkDown);
-      WaitForPacketsOrSilence(false, false);
+        // Sender-side network down for audio; there should be no effect on
+        // video
+        sender_call_->SignalChannelNetworkState(MediaType::AUDIO, kNetworkDown);
+        WaitForPacketsOrSilence(false, false);
 
-      // Sender-side network down.
-      sender_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkDown);
-      {
-        rtc::CritScope lock(&test_crit_);
-        // After network goes down we shouldn't be encoding more frames.
-        sender_state_ = kNetworkDown;
-      }
-      // Wait for receiver-packets and no sender packets.
-      WaitForPacketsOrSilence(true, false);
+        // Receiver-side network down for audio; no change expected
+        receiver_call_->SignalChannelNetworkState(MediaType::AUDIO,
+                                                  kNetworkDown);
+        WaitForPacketsOrSilence(false, false);
 
-      // Receiver-side network down.
-      receiver_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkDown);
-      WaitForPacketsOrSilence(true, true);
+        // Sender-side network down.
+        sender_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkDown);
+        {
+          rtc::CritScope lock(&test_crit_);
+          // After network goes down we shouldn't be encoding more frames.
+          sender_state_ = kNetworkDown;
+        }
+        // Wait for receiver-packets and no sender packets.
+        WaitForPacketsOrSilence(true, false);
 
-      // Network up for audio for both sides; video is still not expected to
-      // start
-      sender_call_->SignalChannelNetworkState(MediaType::AUDIO, kNetworkUp);
-      receiver_call_->SignalChannelNetworkState(MediaType::AUDIO, kNetworkUp);
-      WaitForPacketsOrSilence(true, true);
+        // Receiver-side network down.
+        receiver_call_->SignalChannelNetworkState(MediaType::VIDEO,
+                                                  kNetworkDown);
+        WaitForPacketsOrSilence(true, true);
 
-      // Network back up again for both.
-      {
-        rtc::CritScope lock(&test_crit_);
-        // It's OK to encode frames again, as we're about to bring up the
-        // network.
-        sender_state_ = kNetworkUp;
-      }
-      sender_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkUp);
-      receiver_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkUp);
-      WaitForPacketsOrSilence(false, false);
+        // Network up for audio for both sides; video is still not expected to
+        // start
+        sender_call_->SignalChannelNetworkState(MediaType::AUDIO, kNetworkUp);
+        receiver_call_->SignalChannelNetworkState(MediaType::AUDIO, kNetworkUp);
+        WaitForPacketsOrSilence(true, true);
 
-      // TODO(skvlad): add tests to verify that the audio streams are stopped
-      // when the network goes down for audio once the workaround in
-      // paced_sender.cc is removed.
+        // Network back up again for both.
+        {
+          rtc::CritScope lock(&test_crit_);
+          // It's OK to encode frames again, as we're about to bring up the
+          // network.
+          sender_state_ = kNetworkUp;
+        }
+        sender_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkUp);
+        receiver_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkUp);
+        WaitForPacketsOrSilence(false, false);
+
+        // TODO(skvlad): add tests to verify that the audio streams are stopped
+        // when the network goes down for audio once the workaround in
+        // paced_sender.cc is removed.
+      });
     }
 
     int32_t Encode(const VideoFrame& input_image,
                    const CodecSpecificInfo* codec_specific_info,
                    const std::vector<FrameType>* frame_types) override {
       {
         rtc::CritScope lock(&test_crit_);
         if (sender_state_ == kNetworkDown) {
           ++down_frames_;
           EXPECT_LE(down_frames_, 1)
@@ skipping 49 matching lines @@
               static_cast<int64_t>(kSilenceTimeoutMs)) {
             receiver_done = true;
           }
         } else {
           if (receiver_rtcp_ > initial_receiver_rtcp + kNumAcceptedDowntimeRtcp)
             receiver_done = true;
         }
       }
     }
 
+    test::SingleThreadedTaskQueueForTesting* const task_queue_;
     rtc::CriticalSection test_crit_;
     rtc::Event encoded_frames_;
     rtc::Event packet_event_;
     Call* sender_call_;
     Call* receiver_call_;
     NetworkState sender_state_ GUARDED_BY(test_crit_);
     int sender_rtp_ GUARDED_BY(test_crit_);
     int sender_padding_ GUARDED_BY(test_crit_);
     int sender_rtcp_ GUARDED_BY(test_crit_);
     int receiver_rtcp_ GUARDED_BY(test_crit_);
     int down_frames_ GUARDED_BY(test_crit_);
-  } test;
+  } test(&task_queue_);
 
   RunBaseTest(&test);
 }
 
 TEST_F(EndToEndTest, CallReportsRttForSender) {
   static const int kSendDelayMs = 30;
   static const int kReceiveDelayMs = 70;
-  CreateCalls(Call::Config(event_log_.get()), Call::Config(event_log_.get()));
 
-  FakeNetworkPipe::Config config;
-  config.queue_delay_ms = kSendDelayMs;
-  test::DirectTransport sender_transport(config, sender_call_.get(),
-                                         payload_type_map_);
-  config.queue_delay_ms = kReceiveDelayMs;
-  test::DirectTransport receiver_transport(config, receiver_call_.get(),
-                                           payload_type_map_);
-  sender_transport.SetReceiver(receiver_call_->Receiver());
-  receiver_transport.SetReceiver(sender_call_->Receiver());
+  std::unique_ptr<test::DirectTransport> sender_transport;
+  std::unique_ptr<test::DirectTransport> receiver_transport;
 
-  CreateSendConfig(1, 0, 0, &sender_transport);
-  CreateMatchingReceiveConfigs(&receiver_transport);
+  task_queue_.SendTask([this, &sender_transport, &receiver_transport]() {
+    FakeNetworkPipe::Config config;
+    config.queue_delay_ms = kSendDelayMs;
+    CreateCalls(Call::Config(event_log_.get()), Call::Config(event_log_.get()));
+    sender_transport = rtc::MakeUnique<test::DirectTransport>(
+        &task_queue_, config, sender_call_.get(), payload_type_map_);
+    config.queue_delay_ms = kReceiveDelayMs;
+    receiver_transport = rtc::MakeUnique<test::DirectTransport>(
+        &task_queue_, config, receiver_call_.get(), payload_type_map_);
+    sender_transport->SetReceiver(receiver_call_->Receiver());
+    receiver_transport->SetReceiver(sender_call_->Receiver());
 
-  CreateVideoStreams();
-  CreateFrameGeneratorCapturer(kDefaultFramerate, kDefaultWidth,
-                               kDefaultHeight);
-  Start();
+    CreateSendConfig(1, 0, 0, sender_transport.get());
+    CreateMatchingReceiveConfigs(receiver_transport.get());
+
+    CreateVideoStreams();
+    CreateFrameGeneratorCapturer(kDefaultFramerate, kDefaultWidth,
+                                 kDefaultHeight);
+    Start();
+  });
 
   int64_t start_time_ms = clock_->TimeInMilliseconds();
   while (true) {
     Call::Stats stats = sender_call_->GetStats();
     ASSERT_GE(start_time_ms + kDefaultTimeoutMs,
               clock_->TimeInMilliseconds())
         << "No RTT stats before timeout!";
     if (stats.rtt_ms != -1) {
       // To avoid failures caused by rounding or minor ntp clock adjustments,
       // relax expectation by 1ms.
       constexpr int kAllowedErrorMs = 1;
       EXPECT_GE(stats.rtt_ms, kSendDelayMs + kReceiveDelayMs - kAllowedErrorMs);
       break;
     }
     SleepMs(10);
   }
 
-  sender_transport.StopSending();
-  receiver_transport.StopSending();
-  Stop();
-  DestroyStreams();
-  DestroyCalls();
+  task_queue_.SendTask([this, &sender_transport, &receiver_transport]() {
+    Stop();
+    DestroyStreams();
+    sender_transport.reset();
+    receiver_transport.reset();
+    DestroyCalls();
+  });
 }
 
 void EndToEndTest::VerifyNewVideoSendStreamsRespectNetworkState(
     MediaType network_to_bring_up,
     VideoEncoder* encoder,
     Transport* transport) {
-  CreateSenderCall(Call::Config(event_log_.get()));
-  sender_call_->SignalChannelNetworkState(network_to_bring_up, kNetworkUp);
+  task_queue_.SendTask([this, network_to_bring_up, encoder, transport]() {
+    CreateSenderCall(Call::Config(event_log_.get()));
+    sender_call_->SignalChannelNetworkState(network_to_bring_up, kNetworkUp);
 
-  CreateSendConfig(1, 0, 0, transport);
-  video_send_config_.encoder_settings.encoder = encoder;
-  CreateVideoStreams();
-  CreateFrameGeneratorCapturer(kDefaultFramerate, kDefaultWidth,
-                               kDefaultHeight);
+    CreateSendConfig(1, 0, 0, transport);
+    video_send_config_.encoder_settings.encoder = encoder;
+    CreateVideoStreams();
+    CreateFrameGeneratorCapturer(kDefaultFramerate, kDefaultWidth,
+                                 kDefaultHeight);
 
-  Start();
+    Start();
+  });
+
   SleepMs(kSilenceTimeoutMs);
-  Stop();
 
-  DestroyStreams();
+  task_queue_.SendTask([this]() {
+    Stop();
+    DestroyStreams();
+    DestroyCalls();
+  });
 }
 
 void EndToEndTest::VerifyNewVideoReceiveStreamsRespectNetworkState(
     MediaType network_to_bring_up,
     Transport* transport) {
-  Call::Config config(event_log_.get());
-  CreateCalls(config, config);
-  receiver_call_->SignalChannelNetworkState(network_to_bring_up, kNetworkUp);
+  std::unique_ptr<test::DirectTransport> sender_transport;
 
-  test::DirectTransport sender_transport(sender_call_.get(), payload_type_map_);
-  sender_transport.SetReceiver(receiver_call_->Receiver());
-  CreateSendConfig(1, 0, 0, &sender_transport);
-  CreateMatchingReceiveConfigs(transport);
-  CreateVideoStreams();
-  CreateFrameGeneratorCapturer(kDefaultFramerate, kDefaultWidth,
-                               kDefaultHeight);
+  task_queue_.SendTask([this, &sender_transport, network_to_bring_up,
+                        transport]() {
+    Call::Config config(event_log_.get());
+    CreateCalls(config, config);
+    receiver_call_->SignalChannelNetworkState(network_to_bring_up, kNetworkUp);
+    sender_transport = rtc::MakeUnique<test::DirectTransport>(
+        &task_queue_, sender_call_.get(), payload_type_map_);
+    sender_transport->SetReceiver(receiver_call_->Receiver());
+    CreateSendConfig(1, 0, 0, sender_transport.get());
+    CreateMatchingReceiveConfigs(transport);
+    CreateVideoStreams();
+    CreateFrameGeneratorCapturer(kDefaultFramerate, kDefaultWidth,
+                                 kDefaultHeight);
+    Start();
+  });
 
-  Start();
   SleepMs(kSilenceTimeoutMs);
-  Stop();
 
-  sender_transport.StopSending();
-
-  DestroyStreams();
+  task_queue_.SendTask([this, &sender_transport]() {
+    Stop();
+    DestroyStreams();
+    sender_transport.reset();
+    DestroyCalls();
+  });
 }
 
 TEST_F(EndToEndTest, NewVideoSendStreamsRespectVideoNetworkDown) {
   class UnusedEncoder : public test::FakeEncoder {
    public:
     UnusedEncoder() : FakeEncoder(Clock::GetRealTimeClock()) {}
 
     int32_t InitEncode(const VideoCodec* config,
                        int32_t number_of_cores,
                        size_t max_payload_size) override {
@@ skipping 289 matching lines @@
     std::unique_ptr<VideoEncoder> encoder_;
     std::unique_ptr<VideoDecoder> decoder_;
     rtc::CriticalSection crit_;
     int recorded_frames_ GUARDED_BY(crit_);
   } test(this);
 
   RunBaseTest(&test);
 }
 
 }  // namespace webrtc