New task queueing primitive for async tasks: TaskQueue.

webrtc/base/task_queue_gcd.cc

Index: webrtc/base/task_queue_gcd.cc
diff --git a/webrtc/base/task_queue_gcd.cc b/webrtc/base/task_queue_gcd.cc
new file mode 100644
index 0000000000000000000000000000000000000000..b4235c4fea739dd6eecc18e75ac2be80530e6a1d
--- /dev/null
+++ b/webrtc/base/task_queue_gcd.cc
@@ -0,0 +1,151 @@
+/*
+ *  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 "webrtc/base/task_queue.h"

[perkj_webrtc, 2016/04/26 14:30:37]

Can you add a note what this implement?- ie -what is gcd?

[tommi, 2016/04/28 12:04:01]

Done.

+
+#include <string.h>
+
+#include "webrtc/base/checks.h"
+#include "webrtc/base/logging.h"
+#include "webrtc/base/task_queue_posix.h"
+
+namespace rtc {
+using internal::GetQueuePtrTls;
+using internal::AutoSetCurrentQueuePtr;
+
+struct TaskQueue::QueueContext {
+  explicit QueueContext(TaskQueue* q) : queue(q), is_active(true) {}
+
+  static void DeleteContext(void* context) {
+    QueueContext* qc = reinterpret_cast<QueueContext*>(context);
+    delete qc;
+  }
+
+  TaskQueue* const queue;
+  bool is_active;
+};
+
+TaskQueue::TaskQueue(const char* queue_name)
+    : queue_(dispatch_queue_create(queue_name, DISPATCH_QUEUE_SERIAL)),
+      context_(new QueueContext(this)) {
+  RTC_DCHECK(queue_name);
+  queue_ = dispatch_queue_create(queue_name, DISPATCH_QUEUE_SERIAL);

[perkj_webrtc, 2016/04/26 14:30:38]

already created at line 36.

[tommi, 2016/04/28 12:04:01]

Doh! artefact of moving code around. Thanks for catching.

+  RTC_CHECK(queue_);
+  dispatch_set_context(queue_, context_);
+  // Assign a finalizer that will delete the context when the last reference
+  // to the queue is released.  This may run after the TaskQueue object has
+  // been deleted.
+  dispatch_set_finalizer_f(queue_, &QueueContext::DeleteContext);
+}
+
+TaskQueue::~TaskQueue() {
+  RTC_DCHECK(!IsCurrent());
+  // Implementation/behavioral note:
+  // Dispatch queues are reference counted via calls to dispatch_retain and
+  // dispatch_release. Pending blocks submitted to a queue also hold a
+  // reference to the queue until they have finished. Once all references to a
+  // queue have been released, the queue will be deallocated by the system.
+  // This is why we check the context before running tasks.
+
+  // Use dispatch_sync to set the context to null to guarantee that there's not
+  // a race between checking the context and using it from a task.
+  dispatch_sync(queue_, ^{
+    context_->is_active = false;
+  });
+  dispatch_release(queue_);
+}
+
+// static
+TaskQueue* TaskQueue::Current() {

[perkj_webrtc, 2016/04/26 14:30:37]

Can static TaskQueue::Current be made public? 

[tommi, 2016/04/28 12:04:01]

Done.

+  return static_cast<TaskQueue*>(pthread_getspecific(GetQueuePtrTls()));
+}
+
+// static
+bool TaskQueue::IsCurrent(const char* queue_name) {
+  TaskQueue* current = Current();
+  return current &&
+         strcmp(queue_name, dispatch_queue_get_label(current->queue_)) == 0;
+}
+
+bool TaskQueue::IsCurrent() const {
+  RTC_DCHECK(queue_);
+  return this == Current();
+}
+
+void TaskQueue::PostTask(std::unique_ptr<QueuedTask> task) {
+  QueuedTask* task_ptr = task.release();
+  QueueContext* context = context_;
+  dispatch_async(queue_, ^{

[perkj_webrtc, 2016/04/26 14:30:37]

I can read this but I have never seen ^{ before- you might want to get someone
who knows objective c to take a look. zeke? 

[tommi, 2016/04/28 12:04:01]

Good idea, however I don't think it's needed now after switching to the _f
variants of the API.

+    QueuedTask* t = task_ptr;  // since task_ptr is unassignable.

[perkj_webrtc, 2016/04/26 14:30:37]

Are ^{ lambdas with similar rules? 
Is task_ptr and context valid here? ie- captured by val at creation of "lambda"
? 

[tommi, 2016/04/28 12:04:01]

Yes, the rules are similar.  However, now I've removed all use of blocks in this
file.

+    if (context->is_active) {

[perkj_webrtc, 2016/04/26 14:30:37]

Does context->is_active mean that it is safe to do 

TaskQueue* q = new TaskQueue(...);
q->PostTask(...);
delete q;

[tommi, 2016/04/28 12:04:01]

I think we discussed this offline today but let me know if there's a further
action I should take.

+      AutoSetCurrentQueuePtr set_current(context->queue);
+      if (!t->Run())
+        t = nullptr;
+    }
+    if (t)
+      delete t;
+  });
+}
+
+void TaskQueue::PostDelayedTask(std::unique_ptr<QueuedTask> task,
+                                uint32_t milliseconds) {
+  QueuedTask* task_ptr = task.release();
+  QueueContext* context = context_;
+  dispatch_after(dispatch_time(DISPATCH_TIME_NOW, milliseconds * NSEC_PER_MSEC),
+                 queue_, ^{
+                   QueuedTask* t = task_ptr;

[perkj_webrtc, 2016/04/26 14:30:37]

Humm- this is code duplication. Is there a way to avoid that? 

[perkj_webrtc, 2016/04/26 14:30:38]

indentation looks weird.

[tommi, 2016/04/28 12:04:01]

git cl format.  However, I think that this won't be an issue after switching to
dispatch_after_f().

+                   if (context->is_active) {
+                     AutoSetCurrentQueuePtr set_current(context->queue);
+                     if (!t->Run())
+                       t = nullptr;
+                   }
+                   if (t)
+                     delete t;
+                 });
+}
+
+void TaskQueue::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
+                                 std::unique_ptr<QueuedTask> reply,
+                                 TaskQueue* reply_queue) {
+  QueuedTask* task_ptr = task.release();
+  QueuedTask* reply_task_ptr = reply.release();
+  dispatch_queue_t target_queue = reply_queue->queue_;
+  dispatch_retain(target_queue);
+  QueueContext* context = context_;
+  QueueContext* reply_context = reply_queue->context_;
+  dispatch_async(queue_, ^{
+    QueuedTask* t = task_ptr;
+    if (context->is_active) {
+      AutoSetCurrentQueuePtr set_current(context->queue);
+      if (!t->Run())
+        t = nullptr;
+    }
+    if (t)
+      delete t;
+    dispatch_async(target_queue, ^{
+      QueuedTask* r = reply_task_ptr;
+      if (reply_context->is_active) {
+        AutoSetCurrentQueuePtr set_current(reply_context->queue);
+        if (!r->Run())
+          r = nullptr;
+      }
+      if (r)
+        delete r;
+    });
+    dispatch_release(target_queue);
+  });
+}
+
+void TaskQueue::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
+                                 std::unique_ptr<QueuedTask> reply) {
+  return PostTaskAndReply(std::move(task), std::move(reply), Current());
+}
+
+}  // namespace rtc