Introduce a RunLoop class that supports the TaskQueue interface

on the current thread.

This simplifies writing async tests that use TaskQueue and doesn't
require spinning up a new thread for simple things. The implementation
is currently based on rtc::Thread, which could also be useful in
some circumstances while migrating code over to TQ.

Remove PressEnterToContinue from the test_common files since
it's very specific and only used from one file.

Bug: none
Change-Id: I8b2c6c40809271a109ec17cf7e1120847645d58a
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/174260
Reviewed-by: Mirko Bonadei <mbonadei@webrtc.org>
Commit-Queue: Tommi <tommi@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#31160}

test/BUILD.gn

@@ -458,6 +458,7 @@ if (rtc_include_tests) {
       ":perf_test",
       ":rtc_expect_death",
       ":rtp_test_utils",
+      ":test_common",
       ":test_main",
       ":test_support",
       ":test_support_test_artifacts",
@@ -487,7 +488,9 @@ if (rtc_include_tests) {
       "../modules/video_coding:webrtc_vp9",
       "../rtc_base:criticalsection",
       "../rtc_base:rtc_event",
+      "../rtc_base:rtc_task_queue",
       "../rtc_base/system:file_wrapper",
+      "../rtc_base/task_utils:to_queued_task",
       "pc/e2e:e2e_unittests",
       "peer_scenario/tests",
       "scenario:scenario_unittests",
@@ -505,6 +508,7 @@ if (rtc_include_tests) {
       "frame_generator_unittest.cc",
       "rtp_file_reader_unittest.cc",
       "rtp_file_writer_unittest.cc",
+      "run_loop_unittest.cc",
       "testsupport/ivf_video_frame_generator_unittest.cc",
       "testsupport/perf_test_unittest.cc",
       "testsupport/test_artifacts_unittest.cc",
@@ -784,22 +788,11 @@ rtc_library("test_common") {
     "layer_filtering_transport.cc",
     "layer_filtering_transport.h",
     "rtp_rtcp_observer.h",
+    "run_loop.cc",
+    "run_loop.h",
     "video_decoder_proxy_factory.h",
     "video_encoder_proxy_factory.h",
   ]
-  if (current_os != "winuwp") {
-    # The filtering of *_win.cc is not done for WinUWP (intentionally) as
-    # most _win.cc files are compatible with WinUWP. However, the
-    # peek/dispatch Win32 runloops are entirely WinUWP incompatible thus
-    # WinUWP uses the generic runloop as defined for non-Windows targets.
-    sources += [ "win/run_loop_win.cc" ]
-  }
-  if (!is_win || current_os == "winuwp") {
-    sources += [
-      "run_loop.cc",
-      "run_loop.h",
-    ]
-  }
 
   deps = [
     ":direct_transport",
@@ -840,8 +833,10 @@ rtc_library("test_common") {
     "../modules/video_coding:codec_globals_headers",
     "../rtc_base:checks",
     "../rtc_base:criticalsection",
+    "../rtc_base:rtc_base",
     "../rtc_base:rtc_event",
     "../rtc_base:task_queue_for_test",
+    "../rtc_base/task_utils:to_queued_task",
     "../system_wrappers",
     "../system_wrappers:field_trial",
     "//third_party/abseil-cpp/absl/types:optional",

test/run_loop.cc

@@ -9,15 +9,65 @@
  */
 #include "test/run_loop.h"
 
-#include <stdio.h>
+#include "rtc_base/task_utils/to_queued_task.h"
 
 namespace webrtc {
 namespace test {
 
-void PressEnterToContinue(TaskQueueBase* /*task_queue*/) {
+RunLoop::RunLoop() {
-  puts(">> Press ENTER to continue...");
+  worker_thread_.WrapCurrent();
-  while (getc(stdin) != '\n' && !feof(stdin))
-    ;
 }
+
+RunLoop::~RunLoop() {
+  worker_thread_.UnwrapCurrent();
+}
+
+TaskQueueBase* RunLoop::task_queue() {
+  return &worker_thread_;
+}
+
+void RunLoop::Run() {
+  worker_thread_.ProcessMessages(WorkerThread::kForever);
+}
+
+void RunLoop::Quit() {
+  socket_server_.FailNextWait();
+}
+
+void RunLoop::Flush() {
+  worker_thread_.PostTask(
+      ToQueuedTask([this]() { socket_server_.FailNextWait(); }));
+  worker_thread_.ProcessMessages(1000);
+}
+
+RunLoop::FakeSocketServer::FakeSocketServer() = default;
+RunLoop::FakeSocketServer::~FakeSocketServer() = default;
+
+void RunLoop::FakeSocketServer::FailNextWait() {
+  fail_next_wait_ = true;
+}
+
+bool RunLoop::FakeSocketServer::Wait(int cms, bool process_io) {
+  if (fail_next_wait_) {
+    fail_next_wait_ = false;
+    return false;
+  }
+  return true;
+}
+
+void RunLoop::FakeSocketServer::WakeUp() {}
+
+rtc::Socket* RunLoop::FakeSocketServer::CreateSocket(int family, int type) {
+  return nullptr;
+}
+
+rtc::AsyncSocket* RunLoop::FakeSocketServer::CreateAsyncSocket(int family,
+                                                               int type) {
+  return nullptr;
+}
+
+RunLoop::WorkerThread::WorkerThread(rtc::SocketServer* ss)
+    : rtc::Thread(ss), tq_setter_(this) {}
+
 }  // namespace test
 }  // namespace webrtc

test/run_loop.h

@@ -10,13 +10,69 @@
 #ifndef TEST_RUN_LOOP_H_
 #define TEST_RUN_LOOP_H_
 
-#include "api/task_queue/task_queue_base.h"
+#include "rtc_base/task_utils/to_queued_task.h"
+#include "rtc_base/thread.h"
 
 namespace webrtc {
 namespace test {
 
-// Blocks until the user presses enter.
+// This utility class allows you to run a TaskQueue supported interface on the
-void PressEnterToContinue(TaskQueueBase* task_queue);
+// main test thread, call Run() while doing things asynchonously and break
+// the loop (from the same thread) from a callback by calling Quit().
+class RunLoop {
+ public:
+  RunLoop();
+  ~RunLoop();
+
+  TaskQueueBase* task_queue();
+
+  void Run();
+  void Quit();
+
+  void Flush();
+
+  // Convenience methods since TaskQueueBase doesn't support this sort of magic.
+  template <typename Closure>
+  void PostTask(Closure&& task) {
+    task_queue()->PostTask(ToQueuedTask(std::forward<Closure>(task)));
+  }
+
+  template <typename Closure>
+  void PostDelayedTask(Closure&& task, uint32_t milliseconds) {
+    task_queue()->PostDelayedTask(ToQueuedTask(std::forward<Closure>(task)),
+                                  milliseconds);
+  }
+
+ private:
+  class FakeSocketServer : public rtc::SocketServer {
+   public:
+    FakeSocketServer();
+    ~FakeSocketServer();
+
+    void FailNextWait();
+
+   private:
+    bool Wait(int cms, bool process_io) override;
+    void WakeUp() override;
+
+    rtc::Socket* CreateSocket(int family, int type) override;
+    rtc::AsyncSocket* CreateAsyncSocket(int family, int type) override;
+
+   private:
+    bool fail_next_wait_ = false;
+  };
+
+  class WorkerThread : public rtc::Thread {
+   public:
+    explicit WorkerThread(rtc::SocketServer* ss);
+
+   private:
+    CurrentTaskQueueSetter tq_setter_;
+  };
+
+  FakeSocketServer socket_server_;
+  WorkerThread worker_thread_{&socket_server_};
+};
 
 }  // namespace test
 }  // namespace webrtc

test/run_loop_unittest.cc

@@ -0,0 +1,61 @@
+/*
+ *  Copyright (c) 2020 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 "test/run_loop.h"
+
+#include "rtc_base/task_queue.h"
+#include "rtc_base/task_utils/to_queued_task.h"
+#include "test/gtest.h"
+
+namespace webrtc {
+
+TEST(RunLoopTest, TaskQueueOnThread) {
+  EXPECT_EQ(TaskQueueBase::Current(), nullptr);
+  test::RunLoop loop;
+  EXPECT_EQ(TaskQueueBase::Current(), loop.task_queue());
+  EXPECT_TRUE(loop.task_queue()->IsCurrent());
+}
+
+TEST(RunLoopTest, Flush) {
+  test::RunLoop loop;
+  int counter = 0;
+  loop.PostTask([&counter]() { ++counter; });
+  EXPECT_EQ(counter, 0);
+  loop.Flush();
+  EXPECT_EQ(counter, 1);
+}
+
+TEST(RunLoopTest, Delayed) {
+  test::RunLoop loop;
+  bool ran = false;
+  loop.PostDelayedTask(
+      [&ran, &loop]() {
+        ran = true;
+        loop.Quit();
+      },
+      100);
+  loop.Flush();
+  EXPECT_FALSE(ran);
+  loop.Run();
+  EXPECT_TRUE(ran);
+}
+
+TEST(RunLoopTest, PostAndQuit) {
+  test::RunLoop loop;
+  bool ran = false;
+  loop.PostTask([&ran, &loop]() {
+    ran = true;
+    loop.Quit();
+  });
+  loop.Run();
+  EXPECT_TRUE(ran);
+}
+
+}  // namespace webrtc

test/win/run_loop_win.cc

@@ -1,36 +0,0 @@
-/*
- *  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 <Windows.h>
-#include <assert.h>
-#include <conio.h>
-#include <stdio.h>
-
-#include "rtc_base/task_queue_for_test.h"
-#include "test/run_loop.h"
-
-namespace webrtc {
-namespace test {
-
-void PressEnterToContinue(TaskQueueBase* task_queue) {
-  puts(">> Press ENTER to continue...");
-
-  while (!_kbhit() || _getch() != '\r') {
-    // Drive the message loop for the thread running the task_queue
-    SendTask(RTC_FROM_HERE, task_queue, [&]() {
-      MSG msg;
-      if (PeekMessage(&msg, NULL, 0, 0, PM_REMOVE)) {
-        TranslateMessage(&msg);
-        DispatchMessage(&msg);
-      }
-    });
-  }
-}
-}  // namespace test
-}  // namespace webrtc

video/video_quality_test.cc

@@ -11,6 +11,10 @@
 
 #include <stdio.h>
 
+#if defined(WEBRTC_WIN)
+#include <conio.h>
+#endif
+
 #include <algorithm>
 #include <deque>
 #include <map>
@@ -43,7 +47,6 @@
 #include "rtc_base/strings/string_builder.h"
 #include "rtc_base/task_queue_for_test.h"
 #include "test/platform_video_capturer.h"
-#include "test/run_loop.h"
 #include "test/testsupport/file_utils.h"
 #include "test/video_renderer.h"
 #include "video/frame_dumping_decoder.h"
@@ -270,6 +273,29 @@ class QualityTestVideoEncoder : public VideoEncoder,
   VideoCodec codec_settings_;
 };
 
+#if defined(WEBRTC_WIN) && !defined(WINUWP)
+void PressEnterToContinue(TaskQueueBase* task_queue) {
+  puts(">> Press ENTER to continue...");
+
+  while (!_kbhit() || _getch() != '\r') {
+    // Drive the message loop for the thread running the task_queue
+    SendTask(RTC_FROM_HERE, task_queue, [&]() {
+      MSG msg;
+      if (PeekMessage(&msg, NULL, 0, 0, PM_REMOVE)) {
+        TranslateMessage(&msg);
+        DispatchMessage(&msg);
+      }
+    });
+  }
+}
+#else
+void PressEnterToContinue(TaskQueueBase* /*task_queue*/) {
+  puts(">> Press ENTER to continue...");
+  while (getc(stdin) != '\n' && !feof(stdin))
+    ;  // NOLINT
+}
+#endif
+
 }  // namespace
 
 std::unique_ptr<VideoDecoder> VideoQualityTest::CreateVideoDecoder(
@@ -1570,7 +1596,7 @@ void VideoQualityTest::RunWithRenderers(const Params& params) {
     Start();
   });
 
-  test::PressEnterToContinue(task_queue());
+  PressEnterToContinue(task_queue());
 
   SendTask(RTC_FROM_HERE, task_queue(), [&]() {
     Stop();