Showing alternative approach to fixing TaskQueue deadlock.

webrtc/base/task_queue_libevent.cc

 /*
  *  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"
 
 #include <fcntl.h>
 #include <string.h>
 #include <unistd.h>
+#include <time.h>
 
 #include "base/third_party/libevent/event.h"
 #include "webrtc/base/checks.h"
 #include "webrtc/base/logging.h"
 #include "webrtc/base/task_queue_posix.h"
 #include "webrtc/base/timeutils.h"
 
 namespace rtc {
 using internal::GetQueuePtrTls;
 using internal::AutoSetCurrentQueuePtr;
@@ skipping 38 matching lines @@
 }  // namespace
 
 struct TaskQueue::QueueContext {
   explicit QueueContext(TaskQueue* q) : queue(q), is_active(true) {}
   TaskQueue* queue;
   bool is_active;
   // Holds a list of events pending timers for cleanup when the loop exits.
   std::list<TimerEvent*> pending_timers_;
 };
 
-class TaskQueue::PostAndReplyTask : public QueuedTask {
+class TaskQueue::PostAndReplyTask {
  public:
   PostAndReplyTask(std::unique_ptr<QueuedTask> task,
                    std::unique_ptr<QueuedTask> reply,
                    TaskQueue* reply_queue)
       : task_(std::move(task)),
         reply_(std::move(reply)),
         reply_queue_(reply_queue) {
-    reply_queue->PrepareReplyTask(this);
-  }
-
-  ~PostAndReplyTask() override {
-    CritScope lock(&lock_);
-    if (reply_queue_)
-      reply_queue_->ReplyTaskDone(this);
   }
 
   void OnReplyQueueGone() {
     CritScope lock(&lock_);
     reply_queue_ = nullptr;
   }
 
  private:
-  bool Run() override {
+  bool Run() {
     if (!task_->Run())
       task_.release();
 
     CritScope lock(&lock_);
-    if (reply_queue_)
+    if (reply_queue_) {
       reply_queue_->PostTask(std::move(reply_));
+      reply_queue_->PostAndReplyTaskDone(this);
+    }
     return true;
   }
 
   CriticalSection lock_;
   std::unique_ptr<QueuedTask> task_;
   std::unique_ptr<QueuedTask> reply_;
   TaskQueue* reply_queue_ GUARDED_BY(lock_);
+  friend class PostAndReplyTaskRefOwner;
+};
+
+// Inherits from QueuedTask and owns a ref to PostAndReplyTaskRef.
+class TaskQueue::PostAndReplyTaskRefOwner : public QueuedTask {
+ public:
+  PostAndReplyTaskRefOwner(scoped_refptr<PostAndReplyTaskRef> task)
+      : task_(task) {}
+
+ private:
+  bool Run() override { return task_->Run(); }
+
+  scoped_refptr<PostAndReplyTaskRef> task_;
 };
 
 class TaskQueue::SetTimerTask : public QueuedTask {
  public:
   SetTimerTask(std::unique_ptr<QueuedTask> task, uint32_t milliseconds)
       : task_(std::move(task)),
         milliseconds_(milliseconds),
         posted_(Time32()) {}
 
  private:
@@ skipping 24 matching lines @@
   wakeup_pipe_out_ = fds[0];
   wakeup_pipe_in_ = fds[1];
   EventAssign(wakeup_event_.get(), event_base_, wakeup_pipe_out_,
               EV_READ | EV_PERSIST, OnWakeup, this);
   event_add(wakeup_event_.get(), 0);
   thread_.Start();
 }
 
 TaskQueue::~TaskQueue() {
   RTC_DCHECK(!IsCurrent());
+  destroying_ = true;
   struct timespec ts;
   char message = kQuit;
   while (write(wakeup_pipe_in_, &message, sizeof(message)) != sizeof(message)) {
     // The queue is full, so we have no choice but to wait and retry.
     RTC_CHECK_EQ(EAGAIN, errno);
     ts.tv_sec = 0;
     ts.tv_nsec = 1000000;
     nanosleep(&ts, nullptr);
   }
 
   thread_.Stop();
 
   event_del(wakeup_event_.get());
   close(wakeup_pipe_in_);
   close(wakeup_pipe_out_);
   wakeup_pipe_in_ = -1;
   wakeup_pipe_out_ = -1;
 
+  std::list<scoped_refptr<PostAndReplyTaskRef>> pending_replies;
   {
     // Synchronize against any pending reply tasks that might be running on
-    // other queues.
+    // other queues. Note that, since |destroying_| was set to true above, it
+    // will be impossible for |pending_replies_| to grow after this.
     CritScope lock(&pending_lock_);
-    for (auto* reply : pending_replies_)
-      reply->OnReplyQueueGone();
-    pending_replies_.clear();
+    pending_replies.swap(pending_replies_);
   }
+  for (auto reply : pending_replies) {
+    reply->OnReplyQueueGone();
+  }
+  pending_replies.clear();
 
   event_base_free(event_base_);
 }
 
 // static
 TaskQueue* TaskQueue::Current() {
   QueueContext* ctx =
       static_cast<QueueContext*>(pthread_getspecific(GetQueuePtrTls()));
   return ctx ? ctx->queue : nullptr;
 }
@@ skipping 47 matching lines @@
     event_add(&timer->ev, &tv);
   } else {
     PostTask(std::unique_ptr<QueuedTask>(
         new SetTimerTask(std::move(task), milliseconds)));
   }
 }
 
 void TaskQueue::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
                                  std::unique_ptr<QueuedTask> reply,
                                  TaskQueue* reply_queue) {
-  std::unique_ptr<QueuedTask> wrapper_task(
-      new PostAndReplyTask(std::move(task), std::move(reply), reply_queue));
-  PostTask(std::move(wrapper_task));
+  scoped_refptr<PostAndReplyTaskRef> ref(
+      new PostAndReplyTaskRef(std::move(task), std::move(reply), reply_queue));
+  reply_queue->PrepareReplyTask(ref);
+  PostTask(std::unique_ptr<QueuedTask>(new PostAndReplyTaskRefOwner(ref)));
 }
 
 void TaskQueue::PostTaskAndReply(std::unique_ptr<QueuedTask> task,
                                  std::unique_ptr<QueuedTask> reply) {
   return PostTaskAndReply(std::move(task), std::move(reply), Current());
 }
 
 // static
 bool TaskQueue::ThreadMain(void* context) {
   TaskQueue* me = static_cast<TaskQueue*>(context);
@@ skipping 54 matching lines @@
 void TaskQueue::RunTimer(int fd, short flags, void* context) {  // NOLINT
   TimerEvent* timer = static_cast<TimerEvent*>(context);
   if (!timer->task->Run())
     timer->task.release();
   QueueContext* ctx =
       static_cast<QueueContext*>(pthread_getspecific(GetQueuePtrTls()));
   ctx->pending_timers_.remove(timer);
   delete timer;
 }
 
-void TaskQueue::PrepareReplyTask(PostAndReplyTask* reply_task) {
+void TaskQueue::PrepareReplyTask(
+    scoped_refptr<PostAndReplyTaskRef> reply_task) {
   RTC_DCHECK(reply_task);
   CritScope lock(&pending_lock_);
+  if (destroying_) {
+    return;
+  }
   pending_replies_.push_back(reply_task);
 }
 
-void TaskQueue::ReplyTaskDone(PostAndReplyTask* reply_task) {
+void TaskQueue::PostAndReplyTaskDone(PostAndReplyTask* reply_task) {
   CritScope lock(&pending_lock_);
-  pending_replies_.remove(reply_task);
+  // Note: If the destructor is currently being run, and "pending_replies_" is
+  // already swapped out, this may fail to find |reply_task|, which is ok.
+  pending_replies_.remove_if(
+      [reply_task](scoped_refptr<PostAndReplyTaskRef> task) {
+        return task.get() == reply_task;
+      });
 }
 
 }  // namespace rtc