Use SingleThreadedTaskQueue in DirectTransport

webrtc/call/call_perf_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>
 #include <limits>
 #include <memory>
 #include <string>
 
 #include "webrtc/api/audio_codecs/builtin_audio_encoder_factory.h"
 #include "webrtc/call/call.h"
 #include "webrtc/config.h"
 #include "webrtc/logging/rtc_event_log/rtc_event_log.h"
 #include "webrtc/modules/audio_coding/include/audio_coding_module.h"
 #include "webrtc/modules/audio_mixer/audio_mixer_impl.h"
 #include "webrtc/modules/rtp_rtcp/include/rtp_header_parser.h"
 #include "webrtc/rtc_base/checks.h"
-#include "webrtc/rtc_base/constructormagic.h"
+#include "webrtc/rtc_base/ptr_util.h"
 #include "webrtc/rtc_base/thread_annotations.h"
 #include "webrtc/system_wrappers/include/metrics_default.h"
 #include "webrtc/test/call_test.h"
 #include "webrtc/test/direct_transport.h"
 #include "webrtc/test/drifting_clock.h"
 #include "webrtc/test/encoder_settings.h"
 #include "webrtc/test/fake_audio_device.h"
 #include "webrtc/test/fake_encoder.h"
 #include "webrtc/test/field_trial.h"
 #include "webrtc/test/frame_generator.h"
 #include "webrtc/test/frame_generator_capturer.h"
 #include "webrtc/test/gtest.h"
 #include "webrtc/test/rtp_rtcp_observer.h"
+#include "webrtc/test/single_threaded_task_queue.h"
 #include "webrtc/test/testsupport/fileutils.h"
 #include "webrtc/test/testsupport/perf_test.h"
 #include "webrtc/video/transport_adapter.h"
 #include "webrtc/voice_engine/include/voe_base.h"
 
 using webrtc::test::DriftingClock;
 using webrtc::test::FakeAudioDevice;
 
 namespace webrtc {
 
@@ skipping 88 matching lines @@
 
 void CallPerfTest::TestAudioVideoSync(FecMode fec,
                                       CreateOrder create_first,
                                       float video_ntp_speed,
                                       float video_rtp_speed,
                                       float audio_rtp_speed) {
   const char* kSyncGroup = "av_sync";
   const uint32_t kAudioSendSsrc = 1234;
   const uint32_t kAudioRecvSsrc = 5678;
 
-  metrics::Reset();
-  rtc::scoped_refptr<AudioProcessing> audio_processing =
-      AudioProcessing::Create();
-  VoiceEngine* voice_engine = VoiceEngine::Create();
-  VoEBase* voe_base = VoEBase::GetInterface(voice_engine);
-  FakeAudioDevice fake_audio_device(
-      FakeAudioDevice::CreatePulsedNoiseCapturer(256, 48000),
-      FakeAudioDevice::CreateDiscardRenderer(48000), audio_rtp_speed);
-  EXPECT_EQ(0, voe_base->Init(&fake_audio_device, audio_processing.get(),
-                              decoder_factory_));
-  VoEBase::ChannelConfig config;
-  config.enable_voice_pacing = true;
-  int send_channel_id = voe_base->CreateChannel(config);
-  int recv_channel_id = voe_base->CreateChannel();
-
-  AudioState::Config send_audio_state_config;
-  send_audio_state_config.voice_engine = voice_engine;
-  send_audio_state_config.audio_mixer = AudioMixerImpl::Create();
-  send_audio_state_config.audio_processing = audio_processing;
-  Call::Config sender_config(event_log_.get());
-
-  sender_config.audio_state = AudioState::Create(send_audio_state_config);
-  Call::Config receiver_config(event_log_.get());
-  receiver_config.audio_state = sender_config.audio_state;
-  CreateCalls(sender_config, receiver_config);
-
-
-  VideoRtcpAndSyncObserver observer(Clock::GetRealTimeClock());
+  int send_channel_id;
+  int recv_channel_id;
 
   FakeNetworkPipe::Config audio_net_config;
   audio_net_config.queue_delay_ms = 500;
   audio_net_config.loss_percent = 5;
 
+  rtc::scoped_refptr<AudioProcessing> audio_processing;
+  VoiceEngine* voice_engine;
+  VoEBase* voe_base;
+  std::unique_ptr<FakeAudioDevice> fake_audio_device;
+  VideoRtcpAndSyncObserver observer(Clock::GetRealTimeClock());
+
   std::map<uint8_t, MediaType> audio_pt_map;
   std::map<uint8_t, MediaType> video_pt_map;
-  std::copy_if(std::begin(payload_type_map_), std::end(payload_type_map_),
-               std::inserter(audio_pt_map, audio_pt_map.end()),
-               [](const std::pair<const uint8_t, MediaType>& pair) {
-                 return pair.second == MediaType::AUDIO;
-               });
-  std::copy_if(std::begin(payload_type_map_), std::end(payload_type_map_),
-               std::inserter(video_pt_map, video_pt_map.end()),
-               [](const std::pair<const uint8_t, MediaType>& pair) {
-                 return pair.second == MediaType::VIDEO;
-               });
 
-  test::PacketTransport audio_send_transport(sender_call_.get(), &observer,
-                                             test::PacketTransport::kSender,
-                                             audio_pt_map, audio_net_config);
-  audio_send_transport.SetReceiver(receiver_call_->Receiver());
+  std::unique_ptr<test::PacketTransport> audio_send_transport;
+  std::unique_ptr<test::PacketTransport> video_send_transport;
+  std::unique_ptr<test::PacketTransport> receive_transport;
 
-  test::PacketTransport video_send_transport(
-      sender_call_.get(), &observer, test::PacketTransport::kSender,
-      video_pt_map, FakeNetworkPipe::Config());
-  video_send_transport.SetReceiver(receiver_call_->Receiver());
+  AudioSendStream* audio_send_stream;
+  AudioReceiveStream* audio_receive_stream;
+  std::unique_ptr<DriftingClock> drifting_clock;
 
-  test::PacketTransport receive_transport(
-      receiver_call_.get(), &observer, test::PacketTransport::kReceiver,
-      payload_type_map_, FakeNetworkPipe::Config());
-  receive_transport.SetReceiver(sender_call_->Receiver());
+  task_queue_.SendTask([&]() {
+    metrics::Reset();
+    audio_processing = AudioProcessing::Create();
+    voice_engine = VoiceEngine::Create();
+    voe_base = VoEBase::GetInterface(voice_engine);
+    fake_audio_device = rtc::MakeUnique<FakeAudioDevice>(
+        FakeAudioDevice::CreatePulsedNoiseCapturer(256, 48000),
+        FakeAudioDevice::CreateDiscardRenderer(48000), audio_rtp_speed);
+    EXPECT_EQ(0, voe_base->Init(fake_audio_device.get(), audio_processing.get(),
+                                decoder_factory_));
+    VoEBase::ChannelConfig config;
+    config.enable_voice_pacing = true;
+    send_channel_id = voe_base->CreateChannel(config);
+    recv_channel_id = voe_base->CreateChannel();
 
-  CreateSendConfig(1, 0, 0, &video_send_transport);
-  CreateMatchingReceiveConfigs(&receive_transport);
+    AudioState::Config send_audio_state_config;
+    send_audio_state_config.voice_engine = voice_engine;
+    send_audio_state_config.audio_mixer = AudioMixerImpl::Create();
+    send_audio_state_config.audio_processing = audio_processing;
+    Call::Config sender_config(event_log_.get());
 
-  AudioSendStream::Config audio_send_config(&audio_send_transport);
-  audio_send_config.voe_channel_id = send_channel_id;
-  audio_send_config.rtp.ssrc = kAudioSendSsrc;
-  audio_send_config.send_codec_spec =
-      rtc::Optional<AudioSendStream::Config::SendCodecSpec>(
-          {kAudioSendPayloadType, {"ISAC", 16000, 1}});
-  audio_send_config.encoder_factory = CreateBuiltinAudioEncoderFactory();
-  AudioSendStream* audio_send_stream =
-      sender_call_->CreateAudioSendStream(audio_send_config);
+    sender_config.audio_state = AudioState::Create(send_audio_state_config);
+    Call::Config receiver_config(event_log_.get());
+    receiver_config.audio_state = sender_config.audio_state;
+    CreateCalls(sender_config, receiver_config);
 
-  video_send_config_.rtp.nack.rtp_history_ms = kNackRtpHistoryMs;
-  if (fec == FecMode::kOn) {
-    video_send_config_.rtp.ulpfec.red_payload_type = kRedPayloadType;
-    video_send_config_.rtp.ulpfec.ulpfec_payload_type = kUlpfecPayloadType;
-    video_receive_configs_[0].rtp.ulpfec.red_payload_type = kRedPayloadType;
-    video_receive_configs_[0].rtp.ulpfec.ulpfec_payload_type =
-        kUlpfecPayloadType;
-  }
-  video_receive_configs_[0].rtp.nack.rtp_history_ms = 1000;
-  video_receive_configs_[0].renderer = &observer;
-  video_receive_configs_[0].sync_group = kSyncGroup;
+    std::copy_if(std::begin(payload_type_map_), std::end(payload_type_map_),
+                 std::inserter(audio_pt_map, audio_pt_map.end()),
+                 [](const std::pair<const uint8_t, MediaType>& pair) {
+                   return pair.second == MediaType::AUDIO;
+                 });
+    std::copy_if(std::begin(payload_type_map_), std::end(payload_type_map_),
+                 std::inserter(video_pt_map, video_pt_map.end()),
+                 [](const std::pair<const uint8_t, MediaType>& pair) {
+                   return pair.second == MediaType::VIDEO;
+                 });
 
-  AudioReceiveStream::Config audio_recv_config;
-  audio_recv_config.rtp.remote_ssrc = kAudioSendSsrc;
-  audio_recv_config.rtp.local_ssrc = kAudioRecvSsrc;
-  audio_recv_config.voe_channel_id = recv_channel_id;
-  audio_recv_config.sync_group = kSyncGroup;
-  audio_recv_config.decoder_factory = decoder_factory_;
-  audio_recv_config.decoder_map = {{kAudioSendPayloadType, {"ISAC", 16000, 1}}};
+    audio_send_transport = rtc::MakeUnique<test::PacketTransport>(
+        &task_queue_, sender_call_.get(), &observer,
+        test::PacketTransport::kSender, audio_pt_map, audio_net_config);
+    audio_send_transport->SetReceiver(receiver_call_->Receiver());
 
-  AudioReceiveStream* audio_receive_stream;
+    video_send_transport = rtc::MakeUnique<test::PacketTransport>(
+        &task_queue_, sender_call_.get(), &observer,
+        test::PacketTransport::kSender, video_pt_map,
+        FakeNetworkPipe::Config());
+    video_send_transport->SetReceiver(receiver_call_->Receiver());
 
-  if (create_first == CreateOrder::kAudioFirst) {
-    audio_receive_stream =
-        receiver_call_->CreateAudioReceiveStream(audio_recv_config);
-    CreateVideoStreams();
-  } else {
-    CreateVideoStreams();
-    audio_receive_stream =
-        receiver_call_->CreateAudioReceiveStream(audio_recv_config);
-  }
-  EXPECT_EQ(1u, video_receive_streams_.size());
-  observer.set_receive_stream(video_receive_streams_[0]);
-  DriftingClock drifting_clock(clock_, video_ntp_speed);
-  CreateFrameGeneratorCapturerWithDrift(&drifting_clock, video_rtp_speed,
-                                        kDefaultFramerate, kDefaultWidth,
-                                        kDefaultHeight);
+    receive_transport = rtc::MakeUnique<test::PacketTransport>(
+        &task_queue_, receiver_call_.get(), &observer,
+        test::PacketTransport::kReceiver, payload_type_map_,
+        FakeNetworkPipe::Config());
+    receive_transport->SetReceiver(sender_call_->Receiver());
 
-  Start();
+    CreateSendConfig(1, 0, 0, video_send_transport.get());
+    CreateMatchingReceiveConfigs(receive_transport.get());
 
-  audio_send_stream->Start();
-  audio_receive_stream->Start();
+    AudioSendStream::Config audio_send_config(audio_send_transport.get());
+    audio_send_config.voe_channel_id = send_channel_id;
+    audio_send_config.rtp.ssrc = kAudioSendSsrc;
+    audio_send_config.send_codec_spec =
+        rtc::Optional<AudioSendStream::Config::SendCodecSpec>(
+            {kAudioSendPayloadType, {"ISAC", 16000, 1}});
+    audio_send_config.encoder_factory = CreateBuiltinAudioEncoderFactory();
+    audio_send_stream = sender_call_->CreateAudioSendStream(audio_send_config);
+
+    video_send_config_.rtp.nack.rtp_history_ms = kNackRtpHistoryMs;
+    if (fec == FecMode::kOn) {
+      video_send_config_.rtp.ulpfec.red_payload_type = kRedPayloadType;
+      video_send_config_.rtp.ulpfec.ulpfec_payload_type = kUlpfecPayloadType;
+      video_receive_configs_[0].rtp.ulpfec.red_payload_type = kRedPayloadType;
+      video_receive_configs_[0].rtp.ulpfec.ulpfec_payload_type =
+          kUlpfecPayloadType;
+    }
+    video_receive_configs_[0].rtp.nack.rtp_history_ms = 1000;
+    video_receive_configs_[0].renderer = &observer;
+    video_receive_configs_[0].sync_group = kSyncGroup;
+
+    AudioReceiveStream::Config audio_recv_config;
+    audio_recv_config.rtp.remote_ssrc = kAudioSendSsrc;
+    audio_recv_config.rtp.local_ssrc = kAudioRecvSsrc;
+    audio_recv_config.voe_channel_id = recv_channel_id;
+    audio_recv_config.sync_group = kSyncGroup;
+    audio_recv_config.decoder_factory = decoder_factory_;
+    audio_recv_config.decoder_map = {
+        {kAudioSendPayloadType, {"ISAC", 16000, 1}}};
+
+    if (create_first == CreateOrder::kAudioFirst) {
+      audio_receive_stream =
+          receiver_call_->CreateAudioReceiveStream(audio_recv_config);
+      CreateVideoStreams();
+    } else {
+      CreateVideoStreams();
+      audio_receive_stream =
+          receiver_call_->CreateAudioReceiveStream(audio_recv_config);
+    }
+    EXPECT_EQ(1u, video_receive_streams_.size());
+    observer.set_receive_stream(video_receive_streams_[0]);
+    drifting_clock = rtc::MakeUnique<DriftingClock>(clock_, video_ntp_speed);
+    CreateFrameGeneratorCapturerWithDrift(drifting_clock.get(), video_rtp_speed,
+                                          kDefaultFramerate, kDefaultWidth,
+                                          kDefaultHeight);
+
+    Start();
+
+    audio_send_stream->Start();
+    audio_receive_stream->Start();
+  });
 
   EXPECT_TRUE(observer.Wait())
       << "Timed out while waiting for audio and video to be synchronized.";
 
-  audio_send_stream->Stop();
-  audio_receive_stream->Stop();
+  task_queue_.SendTask([&]() {
+    audio_send_stream->Stop();
+    audio_receive_stream->Stop();
 
-  Stop();
-  video_send_transport.StopSending();
-  audio_send_transport.StopSending();
-  receive_transport.StopSending();
+    Stop();
 
-  DestroyStreams();
+    DestroyStreams();
 
-  sender_call_->DestroyAudioSendStream(audio_send_stream);
-  receiver_call_->DestroyAudioReceiveStream(audio_receive_stream);
+    video_send_transport.reset();
+    audio_send_transport.reset();
+    receive_transport.reset();
 
-  voe_base->DeleteChannel(send_channel_id);
-  voe_base->DeleteChannel(recv_channel_id);
-  voe_base->Release();
+    sender_call_->DestroyAudioSendStream(audio_send_stream);
+    receiver_call_->DestroyAudioReceiveStream(audio_receive_stream);
 
-  DestroyCalls();
+    voe_base->DeleteChannel(send_channel_id);
+    voe_base->DeleteChannel(recv_channel_id);
+    voe_base->Release();
 
-  VoiceEngine::Delete(voice_engine);
+    DestroyCalls();
+
+    VoiceEngine::Delete(voice_engine);
+
+    fake_audio_device.reset();
+  });
 
   observer.PrintResults();
 
   // In quick test synchronization may not be achieved in time.
   if (!field_trial::IsEnabled("WebRTC-QuickPerfTest")) {
     EXPECT_EQ(1, metrics::NumSamples("WebRTC.Video.AVSyncOffsetInMs"));
   }
 }
 
 TEST_F(CallPerfTest, PlaysOutAudioAndVideoInSyncWithVideoNtpDrift) {
@@ skipping 32 matching lines @@
           clock_(Clock::GetRealTimeClock()),
           threshold_ms_(threshold_ms),
           start_time_ms_(start_time_ms),
           run_time_ms_(run_time_ms),
           creation_time_ms_(clock_->TimeInMilliseconds()),
           capturer_(nullptr),
           rtp_start_timestamp_set_(false),
           rtp_start_timestamp_(0) {}
 
    private:
-    test::PacketTransport* CreateSendTransport(Call* sender_call) override {
-      return new test::PacketTransport(sender_call, this,
+    test::PacketTransport* CreateSendTransport(
+        test::SingleThreadedTaskQueueForTesting* task_queue,
+        Call* sender_call) override {
+      return new test::PacketTransport(task_queue, sender_call, this,
                                        test::PacketTransport::kSender,
                                        payload_type_map_, net_config_);
     }
 
-    test::PacketTransport* CreateReceiveTransport() override {
-      return new test::PacketTransport(nullptr, this,
+    test::PacketTransport* CreateReceiveTransport(
+        test::SingleThreadedTaskQueueForTesting* task_queue) override {
+      return new test::PacketTransport(task_queue, nullptr, this,
                                        test::PacketTransport::kReceiver,
                                        payload_type_map_, net_config_);
     }
 
     void OnFrame(const VideoFrame& video_frame) override {
       rtc::CritScope lock(&crit_);
       if (video_frame.ntp_time_ms() <= 0) {
         // Haven't got enough RTCP SR in order to calculate the capture ntp
         // time.
         return;
@@ skipping 398 matching lines @@
     uint32_t last_set_bitrate_kbps_;
     VideoSendStream* send_stream_;
     test::FrameGeneratorCapturer* frame_generator_;
     VideoEncoderConfig encoder_config_;
   } test;
 
   RunBaseTest(&test);
 }
 
 }  // namespace webrtc