Use SingleThreadedTaskQueue in DirectTransport

webrtc/test/direct_transport.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 "webrtc/test/direct_transport.h"
 
 #include "webrtc/call/call.h"
 #include "webrtc/system_wrappers/include/clock.h"
+#include "webrtc/test/single_threaded_task_queue.h"
 
 namespace webrtc {
 namespace test {
 
 DirectTransport::DirectTransport(
+    SingleThreadedTaskQueue* task_queue,
     Call* send_call,
     const std::map<uint8_t, MediaType>& payload_type_map)
-    : DirectTransport(FakeNetworkPipe::Config(), send_call, payload_type_map) {}
+    : DirectTransport(task_queue,
+                      FakeNetworkPipe::Config(),
+                      send_call,
+                      payload_type_map) {
+  RTC_DCHECK(task_queue);
+}
 
 DirectTransport::DirectTransport(
+    SingleThreadedTaskQueue* task_queue,
     const FakeNetworkPipe::Config& config,
     Call* send_call,
     const std::map<uint8_t, MediaType>& payload_type_map)
     : DirectTransport(
+          task_queue,
           config,
           send_call,
-          std::unique_ptr<Demuxer>(new DemuxerImpl(payload_type_map))) {}
+          std::unique_ptr<Demuxer>(new DemuxerImpl(payload_type_map))) {
+  RTC_DCHECK(task_queue);
+}
 
-DirectTransport::DirectTransport(const FakeNetworkPipe::Config& config,
+DirectTransport::DirectTransport(SingleThreadedTaskQueue* task_queue,
+                                 const FakeNetworkPipe::Config& config,
                                  Call* send_call,
                                  std::unique_ptr<Demuxer> demuxer)
     : send_call_(send_call),
-      packet_event_(false, false),
-      thread_(NetworkProcess, this, "NetworkProcess"),
       clock_(Clock::GetRealTimeClock()),
-      shutting_down_(false),
+      task_queue_(task_queue),
       fake_network_(clock_, config, std::move(demuxer)) {
-  thread_.Start();
   if (send_call_) {
     send_call_->SignalChannelNetworkState(MediaType::AUDIO, kNetworkUp);
     send_call_->SignalChannelNetworkState(MediaType::VIDEO, kNetworkUp);
   }
+  SendPackets();
 }
 
-DirectTransport::~DirectTransport() { StopSending(); }
+DirectTransport::~DirectTransport() {
+  RTC_DCHECK_RUN_ON(&thread_checker_);
+  // Constructor updates |next_task_|, so it's guaranteed to be initialized.
+  task_queue_->CancelTask(next_task_);
+}
 
 void DirectTransport::SetConfig(const FakeNetworkPipe::Config& config) {
   fake_network_.SetConfig(config);
 }
 
-void DirectTransport::StopSending() {
-  {
-    rtc::CritScope crit(&lock_);
-    shutting_down_ = true;
-  }
-
-  packet_event_.Set();
-  thread_.Stop();
-}
-
 void DirectTransport::SetReceiver(PacketReceiver* receiver) {
   fake_network_.SetReceiver(receiver);
 }
 
 bool DirectTransport::SendRtp(const uint8_t* data,
                               size_t length,
                               const PacketOptions& options) {
+  // TODO(eladalon): !!! Protect this, too?
   if (send_call_) {
     rtc::SentPacket sent_packet(options.packet_id,
                                 clock_->TimeInMilliseconds());
     send_call_->OnSentPacket(sent_packet);
   }
   fake_network_.SendPacket(data, length);
-  packet_event_.Set();
   return true;
 }
 
 bool DirectTransport::SendRtcp(const uint8_t* data, size_t length) {
   fake_network_.SendPacket(data, length);
-  packet_event_.Set();
   return true;
 }
 
 int DirectTransport::GetAverageDelayMs() {
   return fake_network_.AverageDelay();
 }
 
 DirectTransport::ForceDemuxer::ForceDemuxer(MediaType media_type)
     : media_type_(media_type) {}
 
 void DirectTransport::ForceDemuxer::SetReceiver(PacketReceiver* receiver) {
   packet_receiver_ = receiver;
 }
 
 void DirectTransport::ForceDemuxer::DeliverPacket(
     const NetworkPacket* packet,
     const PacketTime& packet_time) {
   if (!packet_receiver_)
     return;
   packet_receiver_->DeliverPacket(media_type_, packet->data(),
                                   packet->data_length(), packet_time);
 }
 
-bool DirectTransport::NetworkProcess(void* transport) {
-  return static_cast<DirectTransport*>(transport)->SendPackets();
-}
+void DirectTransport::SendPackets() {
+  RTC_DCHECK_RUN_ON(&thread_checker_);
 
-bool DirectTransport::SendPackets() {
-  fake_network_.Process();
-  int64_t wait_time_ms = fake_network_.TimeUntilNextProcess();
-  if (wait_time_ms > 0) {
-    packet_event_.Wait(static_cast<int>(wait_time_ms));
-  }
-  rtc::CritScope crit(&lock_);
-  return shutting_down_ ? false : true;
+  fake_network_.Process();  // Actual work.
+
+  next_task_ = task_queue_->PostTask([this]() {  // Schedule next call.
+    SendPackets();

[stefan-webrtc, 2017/08/14 11:34:54]

This will busy loop, won't it?

[eladalon, 2017/08/14 15:38:00]

It doesn't execute SendPackets(), but rather schedules it for a next call. Other
things might get scheduled in between, or the thread might decide to rest in
between to make sure it doesn't hog the CPU.

(From an empirical stand-point, I can tell you that the this-thread-is-too-busy
error is not scheduled, and that I can see several things happening on the same
queue at the same time without interruption. The video-loopback works fine, for
example.)

[stefan-webrtc, 2017/08/14 16:30:21]

But can't we delay the task until it's time to actually send the next packet?

[eladalon, 2017/08/14 17:08:26]

More things might be working on this task than this one DirectTransport
instance.

[eladalon, 2017/08/14 17:17:54]

I mean to say, we have two options here:
1. Wait() from here, as before. This is not a real option, because more things
might be working on this SingleThreadedTaskQueue than just this one
DirectTransport.
2. Introduce PostDelayedTask(). This is possible, but would increase the
complexity of SingleThreadedTaskQueue. Do you think we should do it? If so, in
this CL, or a separate additional one?

+  });
 }
 }  // namespace test
 }  // namespace webrtc