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([&]() {

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

Check what coding style says about capture-everything, I think it's discouraged
even if I guess the alternative wouldn't look very nice here.

Also, I think I'd prefer to only put work that must be done on the right thread
inside the task (and comment on why it has to be there), and keep simple local
setup, like the copy_if blocks, outside of the task.

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

* When there weren't too many objects to capture, I listed them explicitly. This
is one case where this would have been excessive.
* About the lambdas - the coding style appears to allow that in cases such as
this, where the lambda cannot outlive the captured object.
* The copy_if is in the lambda because I am aiming to minimize the amount of
changes this CL introduces. Moving it outside of the capture block would have
required either rearranging the function, which would have made the review more
difficult, or breaking things into more lambdas than is strictly necessary,
which I think would have been sub-optimal.
* Please also note that when creating a CL with so many changes, I could not
delve into all of the details of each test I was modifying. I had to work
somewhat mechanically (outside of the direct_transport and
single_thread_task_queue files), leaving things as well/badly-written as they
were before.

+    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