Use SingleThreadedTaskQueue in DirectTransport

webrtc/test/direct_transport.cc

Index: webrtc/test/direct_transport.cc
diff --git a/webrtc/test/direct_transport.cc b/webrtc/test/direct_transport.cc
index 370425c5374930f7562e3f23f71277444a43d2e5..56bbb943e7ccfa79128c36110d9da4e4317aecd0 100644
--- a/webrtc/test/direct_transport.cc
+++ b/webrtc/test/direct_transport.cc
@@ -11,56 +11,60 @@
 
 #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);
 }
@@ -68,19 +72,18 @@ void DirectTransport::SetReceiver(PacketReceiver* 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;
 }
 
@@ -104,18 +107,14 @@ void DirectTransport::ForceDemuxer::DeliverPacket(
                                   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