New task queueing primitive for async tasks: TaskQueue.

webrtc/base/task_queue_unittest.cc

Index: webrtc/base/task_queue_unittest.cc
diff --git a/webrtc/base/task_queue_unittest.cc b/webrtc/base/task_queue_unittest.cc
new file mode 100644
index 0000000000000000000000000000000000000000..48e5b7a06104716dbab3ca6924980f373bfc073d
--- /dev/null
+++ b/webrtc/base/task_queue_unittest.cc
@@ -0,0 +1,245 @@
+/*
+ *  Copyright 2016 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 <memory>
+#include <vector>
+
+#include "webrtc/base/bind.h"
+#include "webrtc/base/event.h"
+#include "webrtc/base/gunit.h"
+#include "webrtc/base/task_queue.h"
+#include "webrtc/base/timeutils.h"
+
+namespace rtc {
+
+namespace {
+void CheckCurrent(const char* expected_queue, Event* signal, TaskQueue* queue) {
+  EXPECT_TRUE(TaskQueue::IsCurrent(expected_queue));
+  EXPECT_TRUE(queue->IsCurrent());
+  if (signal)
+    signal->Set();
+}
+
+}  // namespace
+
+TEST(TaskQueueTest, Construct) {
+  static const char kQueueName[] = "Construct";
+  TaskQueue queue(kQueueName);
+  EXPECT_FALSE(queue.IsCurrent());
+}
+
+TEST(TaskQueueTest, PostAndCheckCurrent) {
+  static const char kQueueName[] = "PostAndCheckCurrent";
+  TaskQueue queue(kQueueName);
+
+  // We're not running a task, so there shouldn't be a current queue.
+  EXPECT_FALSE(queue.IsCurrent());
+  EXPECT_FALSE(TaskQueue::Current());
+
+  Event event(false, false);
+  queue.PostTask(Bind(&CheckCurrent, kQueueName, &event, &queue));
+  EXPECT_TRUE(event.Wait(1000));
+}
+
+TEST(TaskQueueTest, PostCustomTask) {
+  static const char kQueueName[] = "PostCustomImplementation";
+  TaskQueue queue(kQueueName);
+
+  Event event(false, false);
+
+  class CustomTask : public QueuedTask {
+   public:
+    CustomTask(Event* event) : event_(event) {}
+
+   private:
+    bool Run() override {
+      event_->Set();
+      return false;  // Never allows the task to be deleted by the queue.
+    }
+
+    Event* const event_;
+  };
+
+  queue.PostTask(std::unique_ptr<QueuedTask>(new CustomTask(&event)));

[perkj_webrtc, 2016/04/28 14:40:18]

leaks CustomTask since Run returns false :-> See bots. Move
std::unique_ptr<QueuedTask>(new CustomTask(&event))) to outside PostTask.

[tommi, 2016/04/28 15:30:33]

Ah, thanks. I was debating allocating the task in a certain way (on the stack)
but decided to half-way not do it.  I'll change it back to doing that even
though we shouldn't do that in production code.

+  EXPECT_TRUE(event.Wait(1000));
+}
+
+TEST(TaskQueueTest, PostLambda) {
+  static const char kQueueName[] = "PostLambda";
+  TaskQueue queue(kQueueName);
+
+  Event event(false, false);
+  queue.PostTask([&event]() { event.Set(); });
+  EXPECT_TRUE(event.Wait(1000));
+}
+
+TEST(TaskQueueTest, PostFromQueue) {
+  static const char kQueueName[] = "PostFromQueue";
+  TaskQueue queue(kQueueName);
+
+  Event event(false, false);
+  queue.PostTask(
+      [&event, &queue]() { queue.PostTask([&event]() { event.Set(); }); });
+  EXPECT_TRUE(event.Wait(1000));
+}
+
+TEST(TaskQueueTest, PostDelayed) {
+  static const char kQueueName[] = "PostDelayed";
+  TaskQueue queue(kQueueName);
+
+  Event event(false, false);
+  uint32_t start = Time();
+  queue.PostDelayedTask(Bind(&CheckCurrent, kQueueName, &event, &queue), 100);
+  EXPECT_TRUE(event.Wait(1000));
+  uint32_t end = Time();
+  EXPECT_GE(end - start, 100u);
+  EXPECT_NEAR(end - start, 200u, 100u);  // Accept 100-300.

[perkj_webrtc, 2016/04/28 14:40:18]

200?

[tommi, 2016/04/28 15:30:33]

Yes... I figured that because some of our trybots are *very* slow, that an ample
acceptable range would be appropriate to avoid making the test flaky.  So, to
explain 200 (since that appears to be the question):

acceptable range is: 100-300.
(100+300) / 2 == 200

so, 200 +/- 100 is allowed. :)

+}
+
+TEST(TaskQueueTest, PostMultipleDelayed) {
+  static const char kQueueName[] = "PostMultipleDelayed";
+  TaskQueue queue(kQueueName);
+
+  std::vector<std::unique_ptr<Event>> events;
+  for (int i = 0; i < 10; ++i) {
+    events.push_back(std::unique_ptr<Event>(new Event(false, false)));
+    queue.PostDelayedTask(
+        Bind(&CheckCurrent, kQueueName, events.back().get(), &queue), 10);
+  }
+
+  for (const auto& e : events)
+    EXPECT_TRUE(e->Wait(100));
+}
+
+TEST(TaskQueueTest, PostDelayedAfterDestruct) {
+  static const char kQueueName[] = "PostDelayedAfterDestruct";
+  Event event(false, false);
+  {
+    TaskQueue queue(kQueueName);
+    queue.PostDelayedTask(Bind(&CheckCurrent, kQueueName, &event, &queue), 100);
+  }
+  EXPECT_FALSE(event.Wait(200));  // Task should not run.
+}
+
+TEST(TaskQueueTest, PostAndReply) {
+  static const char kPostQueue[] = "PostQueue";
+  static const char kReplyQueue[] = "ReplyQueue";
+  TaskQueue post_queue(kPostQueue);
+  TaskQueue reply_queue(kReplyQueue);
+
+  Event event(false, false);
+  post_queue.PostTaskAndReply(
+      Bind(&CheckCurrent, kPostQueue, nullptr, &post_queue),
+      Bind(&CheckCurrent, kReplyQueue, &event, &reply_queue), &reply_queue);

[perkj_webrtc, 2016/04/28 14:40:17]

So you made PostTaskAndReply to another queue public.

[tommi, 2016/04/28 15:30:33]

It was already public.  I had forgotten about this test :)

+  EXPECT_TRUE(event.Wait(1000));
+}
+
+TEST(TaskQueueTest, PostAndReuse) {
+  static const char kPostQueue[] = "PostQueue";
+  static const char kReplyQueue[] = "ReplyQueue";
+  TaskQueue post_queue(kPostQueue);
+  TaskQueue reply_queue(kReplyQueue);
+
+  int call_count = 0;
+
+  class ReusedTask : public QueuedTask {
+   public:
+    ReusedTask(int* counter, TaskQueue* reply_queue, Event* event)
+        : counter_(counter), reply_queue_(reply_queue), event_(event) {
+      EXPECT_EQ(0, *counter_);
+    }
+
+   private:
+    bool Run() override {
+      if (++(*counter_) == 1) {
+        std::unique_ptr<QueuedTask> myself(this);
+        reply_queue_->PostTask(std::move(myself));
+        // At this point, the object is owned by reply_queue_ and it's
+        // theoratically possible that the object has been deleted (e.g. if
+        // posting wasn't possible).  So, don't touch any member variables here.
+
+        // Indicate to the current queue that ownership has been transferred.
+        return false;
+      } else {
+        EXPECT_EQ(2, *counter_);
+        EXPECT_TRUE(reply_queue_->IsCurrent());
+        event_->Set();
+        return true;  // Indicate that the object should be deleted.
+      }
+    }
+
+    int* const counter_;
+    TaskQueue* const reply_queue_;
+    Event* const event_;
+  };
+
+  Event event(false, false);
+  std::unique_ptr<QueuedTask> task(
+      new ReusedTask(&call_count, &reply_queue, &event));
+
+  post_queue.PostTask(std::move(task));
+  EXPECT_TRUE(event.Wait(1000));
+}
+
+TEST(TaskQueueTest, PostAndReplyLambda) {
+  static const char kPostQueue[] = "PostQueue";
+  static const char kReplyQueue[] = "ReplyQueue";
+  TaskQueue post_queue(kPostQueue);
+  TaskQueue reply_queue(kReplyQueue);
+
+  Event event(false, false);
+  bool my_flag = false;
+  post_queue.PostTaskAndReply([&my_flag]() { my_flag = true; },
+                              [&event]() { event.Set(); }, &reply_queue);
+  EXPECT_TRUE(event.Wait(1000));
+  EXPECT_TRUE(my_flag);
+}
+
+void TestPostTaskAndReply(TaskQueue* work_queue,
+                          const char* work_queue_name,
+                          Event* event) {
+  ASSERT_FALSE(work_queue->IsCurrent());
+  work_queue->PostTaskAndReply(
+      Bind(&CheckCurrent, work_queue_name, nullptr, work_queue),
+      NewClosure([event]() { event->Set(); }));
+}
+
+// Does a PostTaskAndReply from within a task to post and reply to the current
+// queue.  All in all there will be 3 tasks posted and run.
+TEST(TaskQueueTest, PostAndReply2) {
+  static const char kQueueName[] = "PostAndReply2";
+  static const char kWorkQueueName[] = "PostAndReply2_Worker";
+  TaskQueue queue(kQueueName);
+  TaskQueue work_queue(kWorkQueueName);
+
+  Event event(false, false);
+  queue.PostTask(
+      Bind(&TestPostTaskAndReply, &work_queue, kWorkQueueName, &event));
+  EXPECT_TRUE(event.Wait(1000));
+}
+
+TEST(TaskQueueTest, PostALot) {

[perkj_webrtc, 2016/04/28 14:40:18]

please add a better description of what this tests. Deletes pending tasks while
being destructed.

[tommi, 2016/04/28 15:30:33]

Done.

I also modified the test a bit as well as supporting specifying lambdas for
doing cleanup if that's required.  This test can then use that ability to detect
if a task was dropped or processed.  On Mac, no tasks will actually get dropped
in this test (since queuing up >0xFFFF tasks, is fine).

+  // To destruct the event after the queue has gone out of scope.
+  Event event(false, false);
+
+  static const char kQueueName[] = "PostALot";

[perkj_webrtc, 2016/04/28 14:40:18]

why static?

[tommi, 2016/04/28 15:30:33]

since this is supposed to be a compile-time constant and not runtime.
An alternative would be:

const char queue_name[] = "PostALot";

but that's actual code that runs every time (i.e. not compile time and that's
also why the name would not be kQueueName).

+  TaskQueue queue(kQueueName);
+
+  // On linux, the limit of pending bytes in the pipe buffer is 0xffff.
+  // So here we post a total of 0xffff+1 messages, which triggers a failure case
+  // inside of the libevent queue implementation.
+
+  queue.PostTask([&event]() { event.Wait(Event::kForever); });
+  for (int i = 0; i < 0xffff; ++i)
+    queue.PostTask([]() {});
+  event.Set();
+}
+
+}  // namespace rtc