Use SingleThreadedTaskQueue in DirectTransport

webrtc/test/single_threaded_task_queue_unittest.cc

Index: webrtc/test/single_threaded_task_queue_unittest.cc
diff --git a/webrtc/test/single_threaded_task_queue_unittest.cc b/webrtc/test/single_threaded_task_queue_unittest.cc
new file mode 100644
index 0000000000000000000000000000000000000000..9ccb8202ecce6f164bde08eff0e6cb3326c6c2ba
--- /dev/null
+++ b/webrtc/test/single_threaded_task_queue_unittest.cc
@@ -0,0 +1,362 @@
+/*
+ *  Copyright (c) 2017 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/single_threaded_task_queue.h"
+
+#include <atomic>
+#include <memory>
+#include <vector>
+
+#include "webrtc/rtc_base/event.h"
+#include "webrtc/rtc_base/ptr_util.h"
+#include "webrtc/test/gtest.h"
+
+namespace webrtc {
+namespace test {
+
+namespace {
+
+using TaskId = SingleThreadedTaskQueueForTesting::TaskId;
+
+// Test should not rely on the object under test not being faulty. If the task
+// queue ever blocks forever, we want the tests to fail, rather than hang.
+constexpr int kMaxWaitTimeMs = 10000;
+
+TEST(SingleThreadedTaskQueueForTestingTest, SanityConstructionDestruction) {
+  SingleThreadedTaskQueueForTesting task_queue("task_queue");
+}
+
+TEST(SingleThreadedTaskQueueForTestingTest, ExecutesPostedTasks) {
+  SingleThreadedTaskQueueForTesting task_queue("task_queue");
+
+  std::atomic<bool> executed(false);
+  rtc::Event done(true, false);
+
+  task_queue.PostTask([&]() {

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

Prefer explicit capture list, for this and other lambdas.

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

Done.

+    executed.store(true);
+    done.Set();
+  });
+  ASSERT_TRUE(done.Wait(kMaxWaitTimeMs));
+
+  EXPECT_TRUE(executed.load());
+}
+
+TEST(SingleThreadedTaskQueueForTestingTest,
+     PostMultipleTasksFromSameExternalThread) {
+  SingleThreadedTaskQueueForTesting task_queue("task_queue");
+
+  std::vector<std::unique_ptr<rtc::Event>> done_events;
+  for (size_t i = 0; i < 3; i++) {
+    done_events.emplace_back(rtc::MakeUnique<rtc::Event>(false, false));
+  }
+  std::atomic<bool> executed[done_events.size()];
+  for (std::atomic<bool>& exec : executed) {
+    exec.store(false);
+  }
+
+  // To avoid the tasks which comprise the actual test from running before they
+  // have all be posted, which could result in only one task ever being in the
+  // queue at any given time, post one waiting task that would block the
+  // task-queue, and unblock only after all tasks have been posted.
+  rtc::Event rendezvous(true, false);
+  task_queue.PostTask([&]() {
+    ASSERT_TRUE(rendezvous.Wait(kMaxWaitTimeMs));
+  });
+
+  // Post the tasks which comprise the test.
+  for (size_t i = 0; i < done_events.size(); i++) {
+    task_queue.PostTask([&executed, &done_events, i]() {  // |i| by value.
+      executed[i].store(true);
+      done_events[i]->Set();
+    });
+  }
+
+  rendezvous.Set();  // Release the task-queue.
+
+  // Wait until the task queue has executed all the tasks.
+  for (size_t i = 0; i < done_events.size(); i++) {
+    ASSERT_TRUE(done_events[i]->Wait(kMaxWaitTimeMs));
+  }
+
+  for (size_t i = 0; i < done_events.size(); i++) {
+    EXPECT_TRUE(executed[i].load());
+  }
+}
+
+TEST(SingleThreadedTaskQueueForTestingTest, PostToTaskQueueFromOwnThread) {
+  SingleThreadedTaskQueueForTesting task_queue("task_queue");
+
+  std::atomic<bool> executed(false);
+  rtc::Event done(true, false);
+
+  auto internally_posted_task = [&executed, &done]() {
+    executed.store(true);
+    done.Set();
+  };
+
+  auto externally_posted_task = [&task_queue, &internally_posted_task]() {
+    task_queue.PostTask(internally_posted_task);
+  };
+
+  task_queue.PostTask(externally_posted_task);
+
+  ASSERT_TRUE(done.Wait(kMaxWaitTimeMs));
+  EXPECT_TRUE(executed.load());
+}
+
+TEST(SingleThreadedTaskQueueForTestingTest, TasksExecutedInSequence) {
+  SingleThreadedTaskQueueForTesting task_queue("task_queue");
+
+  // The first task would perform:
+  // accumulator = 10 * accumulator + i
+  // Where |i| is 1, 2 and 3 for the 1st, 2nd and 3rd tasks, respectively.
+  // The result would be 123 if and only iff the tasks were executed in order.
+  size_t accumulator = 0;
+  size_t expected_value = 0;  // Updates to the correct value.
+
+  // Prevent the chain from being set in motion before we've had time to
+  // schedule it all, lest the queue only contain one task at a time.
+  rtc::Event rendezvous(true, false);
+  task_queue.PostTask([&rendezvous]() {
+    ASSERT_TRUE(rendezvous.Wait(kMaxWaitTimeMs));
+  });
+
+  for (size_t i = 0; i < 3; i++) {
+    task_queue.PostTask([&accumulator, i]() {  // |i| passed by value.
+      accumulator = 10 * accumulator + i;
+    });
+    expected_value = 10 * expected_value + i;
+  }
+
+  // The test will wait for the task-queue to finish.
+  rtc::Event done(true, false);
+  task_queue.PostTask([&done]() {
+    done.Set();
+  });
+
+  rendezvous.Set();  // Set the chain in motion.
+
+  ASSERT_TRUE(done.Wait(kMaxWaitTimeMs));
+
+  EXPECT_EQ(accumulator, expected_value);
+}
+
+TEST(SingleThreadedTaskQueueForTestingTest, ExecutesPostedDelayedTask) {
+  SingleThreadedTaskQueueForTesting task_queue("task_queue");
+
+  std::atomic<bool> executed(false);
+  rtc::Event done(true, false);
+
+  constexpr int64_t delay_ms = 20;
+  static_assert(delay_ms < kMaxWaitTimeMs / 2, "Delay too long for tests.");
+
+  task_queue.PostDelayedTask([&]() {
+    executed.store(true);
+    done.Set();
+  }, delay_ms);
+  ASSERT_TRUE(done.Wait(kMaxWaitTimeMs));
+
+  EXPECT_TRUE(executed.load());
+}
+
+TEST(SingleThreadedTaskQueueForTestingTest, DoesNotExecuteDelayedTaskTooSoon) {
+  SingleThreadedTaskQueueForTesting task_queue("task_queue");
+
+  std::atomic<bool> executed(false);
+
+  constexpr int64_t delay_ms = 2000;
+  static_assert(delay_ms < kMaxWaitTimeMs / 2, "Delay too long for tests.");
+
+  task_queue.PostDelayedTask([&]() {
+    executed.store(true);
+  }, delay_ms);
+
+  // Wait less than is enough, make sure the task was not yet executed.

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

Tests like this easily become flaky. Ideally, use a fake clock
(rtc_base/fakeclock.h), or if that doesn't work, you really need huge margins.

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

I had a TODO for using a fake-clock here, but I removed it when I realized that
after the fake-clock was advanced, I'd still need to wait some time in
real-clock-space, because the CPU would still need to have time to execute the
scheduled tasks.

Perhaps you would like to suggest different values for |delay_ms|, etc., to
reduce the chances of flakiness? If so, please bear in mind that many small
delays add up to long run times for the entire test suite.

+  rtc::Event not_done(true, false);
+  ASSERT_FALSE(not_done.Wait(delay_ms / 2));
+  EXPECT_FALSE(executed.load());
+}
+
+TEST(SingleThreadedTaskQueueForTestingTest,
+     TaskWithLesserDelayPostedAfterFirstDelayedTaskExectuedBeforeFirst) {

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

This is also potentially flaky.

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

Same answer.

+  SingleThreadedTaskQueueForTesting task_queue("task_queue");
+
+  std::atomic<bool> earlier_executed(false);
+  constexpr int64_t earlier_delay_ms = 500;
+
+  std::atomic<bool> later_executed(false);
+  constexpr int64_t later_delay_ms = 1000;
+
+  static_assert(earlier_delay_ms + later_delay_ms < kMaxWaitTimeMs / 2,
+                "Delay too long for tests.");
+
+  rtc::Event done(true, false);
+
+  auto earlier_task = [&]() {
+    EXPECT_FALSE(later_executed.load());
+    earlier_executed.store(true);
+  };
+
+  auto later_task = [&]() {
+    EXPECT_TRUE(earlier_executed.load());
+    later_executed.store(true);
+    done.Set();
+  };
+
+  task_queue.PostDelayedTask(later_task, later_delay_ms);
+  task_queue.PostDelayedTask(earlier_task, earlier_delay_ms);
+
+  ASSERT_TRUE(done.Wait(kMaxWaitTimeMs));
+  ASSERT_TRUE(earlier_executed);
+  ASSERT_TRUE(later_executed);
+}
+
+TEST(SingleThreadedTaskQueueForTestingTest,
+     TaskWithGreaterDelayPostedAfterFirstDelayedTaskExectuedAfterFirst) {
+  SingleThreadedTaskQueueForTesting task_queue("task_queue");
+
+  std::atomic<bool> earlier_executed(false);
+  constexpr int64_t earlier_delay_ms = 500;
+
+  std::atomic<bool> later_executed(false);
+  constexpr int64_t later_delay_ms = 1000;
+
+  static_assert(earlier_delay_ms + later_delay_ms < kMaxWaitTimeMs / 2,
+                "Delay too long for tests.");
+
+  rtc::Event done(true, false);
+
+  auto earlier_task = [&]() {
+    EXPECT_FALSE(later_executed.load());
+    earlier_executed.store(true);
+  };
+
+  auto later_task = [&]() {
+    EXPECT_TRUE(earlier_executed.load());
+    later_executed.store(true);
+    done.Set();
+  };
+
+  task_queue.PostDelayedTask(earlier_task, earlier_delay_ms);
+  task_queue.PostDelayedTask(later_task, later_delay_ms);
+
+  ASSERT_TRUE(done.Wait(kMaxWaitTimeMs));
+  ASSERT_TRUE(earlier_executed);
+  ASSERT_TRUE(later_executed);
+}
+
+TEST(SingleThreadedTaskQueueForTestingTest, ExternalThreadCancelsTask) {
+  SingleThreadedTaskQueueForTesting task_queue("task_queue");
+
+  rtc::Event done(true, false);
+
+  // Prevent the to-be-cancelled task from being executed before we've had
+  // time to cancel it.
+  rtc::Event rendezvous(true, false);
+  task_queue.PostTask([&]() {
+    ASSERT_TRUE(rendezvous.Wait(kMaxWaitTimeMs));
+  });
+
+  TaskId cancelled_task_id = task_queue.PostTask([]() {
+    EXPECT_TRUE(false);
+  });
+  task_queue.PostTask([&done]() {
+    done.Set();
+  });
+
+  task_queue.CancelTask(cancelled_task_id);
+
+  // Set the tasks in motion; the cancelled task does not run (otherwise the
+  // test would fail). The last task ends the test, showing that the queue
+  // progressed beyond the cancelled task.
+  rendezvous.Set();
+  ASSERT_TRUE(done.Wait(kMaxWaitTimeMs));
+}
+
+// In this test, we'll set off a chain where the first task cancels the second
+// task, then a third task runs (showing that we really cancelled the task,
+// rather than just halted the task-queue).
+TEST(SingleThreadedTaskQueueForTestingTest, InternalThreadCancelsTask) {
+  SingleThreadedTaskQueueForTesting task_queue("task_queue");
+
+  rtc::Event done(true, false);
+
+  // Prevent the chain from being set-off before we've set everything up.
+  rtc::Event rendezvous(true, false);
+  task_queue.PostTask([&]() {
+    ASSERT_TRUE(rendezvous.Wait(kMaxWaitTimeMs));
+  });
+
+  // This is the canceller-task. It takes cancelled_task_id by reference,
+  // because the ID will only become known after the cancelled task is
+  // scheduled.
+  TaskId cancelled_task_id;
+  auto canceller_task = [&task_queue, &cancelled_task_id]() {
+    task_queue.CancelTask(cancelled_task_id);
+  };
+  task_queue.PostTask(canceller_task);
+
+  // This task will be cancelled by the task before it.
+  auto cancelled_task = []() {
+    EXPECT_TRUE(false);
+  };
+  cancelled_task_id = task_queue.PostTask(cancelled_task);
+
+  // When this task runs, it will allow the test to be finished.
+  auto completion_marker_task = [&done]() {
+    done.Set();
+  };
+  task_queue.PostTask(completion_marker_task);
+
+  rendezvous.Set();  // Set the chain in motion.
+
+  ASSERT_TRUE(done.Wait(kMaxWaitTimeMs));
+}
+
+TEST(SingleThreadedTaskQueueForTestingTest, SendTask) {
+  SingleThreadedTaskQueueForTesting task_queue("task_queue");
+
+  std::atomic<bool> executed(false);
+
+  task_queue.SendTask([&executed]() {
+    // Intentionally delay, so that if SendTask didn't block, the sender thread
+    // would have time to read |executed|.
+    rtc::Event delay(true, false);
+    ASSERT_FALSE(delay.Wait(1000));

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

When delay is all you want, prefer the sleep function.

Not sure what's the preferred sleep function is, but perhaps Thread::SleepMs.

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

Thread::SleepMs() is only used in six places in our codebase, so I'm wary of
using it. Perhaps something else?

+    executed.store(true);
+  });
+
+  EXPECT_TRUE(executed);
+}
+
+TEST(SingleThreadedTaskQueueForTestingTest,
+     DestructTaskQueueWhileTasksPending) {
+  auto task_queue =
+      rtc::MakeUnique<SingleThreadedTaskQueueForTesting>("task_queue");
+
+  std::atomic<size_t> counter(0);
+
+  constexpr size_t tasks = 10;
+  for (size_t i = 0; i < tasks; i++) {
+    task_queue->PostTask([&counter]() {
+      std::atomic_fetch_add(&counter, static_cast<size_t>(1));
+      rtc::Event delay(true, false);
+      ASSERT_FALSE(delay.Wait(500));
+    });
+  }
+
+  task_queue.reset();
+
+  EXPECT_LT(counter, tasks);
+}
+
+}  // namespace
+}  // namespace test
+}  // namespace webrtc