Making AsyncInvoker destructor thread-safe.

webrtc/base/asyncinvoker.cc

 /*
  *  Copyright 2014 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/asyncinvoker.h"
 
+#include "webrtc/base/atomicops.h"
 #include "webrtc/base/checks.h"
 #include "webrtc/base/logging.h"
 
 namespace rtc {
 
-AsyncInvoker::AsyncInvoker() : destroying_(false) {}
+AsyncInvoker::AsyncInvoker() : invocation_complete_(false, false) {}
 
 AsyncInvoker::~AsyncInvoker() {
   destroying_ = true;
   SignalInvokerDestroyed();
   // Messages for this need to be cleared *before* our destructor is complete.
   MessageQueueManager::Clear(this);
+  // And we need to wait for any invocations that are still in progress on
+  // other threads.
+  while (AtomicOps::AcquireLoad(&pending_invocations_)) {
+    // If the destructor was called while AsyncInvoke was being called by
+    // another thread, WITHIN an AsyncInvoked functor, it may do another
+    // Thread::Post even after we called MessageQueueManager::Clear(this). So
+    // we need to keep calling Clear to discard these posts.
+    MessageQueueManager::Clear(this);
+    invocation_complete_.Wait(Event::kForever);
+  }
 }
 
 void AsyncInvoker::OnMessage(Message* msg) {
   // Get the AsyncClosure shared ptr from this message's data.
   ScopedMessageData<AsyncClosure>* data =
       static_cast<ScopedMessageData<AsyncClosure>*>(msg->pdata);
   // Execute the closure and trigger the return message if needed.
   data->inner_data().Execute();
   delete data;
 }
@@ skipping 17 matching lines @@
 }
 
 void AsyncInvoker::DoInvoke(const Location& posted_from,
                             Thread* thread,
                             std::unique_ptr<AsyncClosure> closure,
                             uint32_t id) {
   if (destroying_) {
     LOG(LS_WARNING) << "Tried to invoke while destroying the invoker.";
     return;
   }
+  AtomicOps::Increment(&pending_invocations_);
   thread->Post(posted_from, this, id,
                new ScopedMessageData<AsyncClosure>(std::move(closure)));
 }
 
 void AsyncInvoker::DoInvokeDelayed(const Location& posted_from,
                                    Thread* thread,
                                    std::unique_ptr<AsyncClosure> closure,
                                    uint32_t delay_ms,
                                    uint32_t id) {
   if (destroying_) {
     LOG(LS_WARNING) << "Tried to invoke while destroying the invoker.";
     return;
   }
+  AtomicOps::Increment(&pending_invocations_);
   thread->PostDelayed(posted_from, delay_ms, this, id,
                       new ScopedMessageData<AsyncClosure>(std::move(closure)));
 }
 
 GuardedAsyncInvoker::GuardedAsyncInvoker() : thread_(Thread::Current()) {
   thread_->SignalQueueDestroyed.connect(this,
                                         &GuardedAsyncInvoker::ThreadDestroyed);
 }
 
 GuardedAsyncInvoker::~GuardedAsyncInvoker() {
 }
 
 bool GuardedAsyncInvoker::Flush(uint32_t id) {
   rtc::CritScope cs(&crit_);
   if (thread_ == nullptr)
     return false;
   invoker_.Flush(thread_, id);
   return true;
 }
 
 void GuardedAsyncInvoker::ThreadDestroyed() {
   rtc::CritScope cs(&crit_);
   // We should never get more than one notification about the thread dying.
   RTC_DCHECK(thread_ != nullptr);
   thread_ = nullptr;
 }
 
+AsyncClosure::~AsyncClosure() {
+  AtomicOps::Decrement(&invoker_->pending_invocations_);
+  invoker_->invocation_complete_.Set();
+}
+
 NotifyingAsyncClosureBase::NotifyingAsyncClosureBase(
     AsyncInvoker* invoker,
     const Location& callback_posted_from,
     Thread* calling_thread)
-    : invoker_(invoker),
+    : AsyncClosure(invoker),
       callback_posted_from_(callback_posted_from),
       calling_thread_(calling_thread) {
   calling_thread->SignalQueueDestroyed.connect(
       this, &NotifyingAsyncClosureBase::CancelCallback);
   invoker->SignalInvokerDestroyed.connect(
       this, &NotifyingAsyncClosureBase::CancelCallback);
 }
 
 NotifyingAsyncClosureBase::~NotifyingAsyncClosureBase() {
   disconnect_all();
@@ skipping 10 matching lines @@
 void NotifyingAsyncClosureBase::CancelCallback() {
   // If the callback is triggering when this is called, block the
   // destructor of the dying object here by waiting until the callback
   // is done triggering.
   CritScope cs(&crit_);
   // calling_thread_ == NULL means do not trigger the callback.
   calling_thread_ = NULL;
 }
 
 }  // namespace rtc