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 48 matching lines @@
   void TestNackRetransmission(uint32_t retransmit_ssrc,
                               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);
 };
 
+// TODO(eladalon): !!! Doesn't look like this file should really work...
+
 TEST_F(VideoSendStreamTest, CanStartStartedStream) {
   CreateSenderCall(Call::Config(event_log_.get()));
 
   test::NullTransport transport;
   CreateSendConfig(1, 0, 0, &transport);
   CreateVideoStreams();
   video_send_stream_->Start();
   video_send_stream_->Start();
   DestroyStreams();
 }
@@ skipping 394 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::SingleThreadedTaskQueue* 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::SingleThreadedTaskQueue* 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::SingleThreadedTaskQueue* 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::SingleThreadedTaskQueue* 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::SingleThreadedTaskQueue* 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::SingleThreadedTaskQueue* 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::SingleThreadedTaskQueue* 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 314 matching lines @@
   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::SingleThreadedTaskQueue* 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::SingleThreadedTaskQueue* 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::SingleThreadedTaskQueue* 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::SingleThreadedTaskQueue* 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::SingleThreadedTaskQueue* 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::SingleThreadedTaskQueue* 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