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_)) {

[tommi, 2017/02/22 11:19:53]

is there a chance that we could get 0 returned here, immediately run out of
quanta on the current thread, another thread increments pending_invocations_ (it
would have to have checked |destroying_| first, but I think that's a
possibility) and later we continue with destruction on this thread with a
non-zero pending_invocations_?

[Taylor Brandstetter, 2017/02/23 01:19:00]

That would mean that the invoker is being destroyed at the same time as
AsyncInvoke is being called on another thread, and *not* from within an
AsyncClosure. I don't think there's any way to guarantee that that's
thread-safe. The destructor could finish, and then the other thread could access
destroying_ and segfault.

The thing the "destroying_" flag does protect against is this kind of thing:

invoker.AsyncInvoke(thread_a, [&invoker]{
  Foo();
  invoker.AsyncInvoke(thread_b, []{
    Bar();
  });
});

Suppose after I call this, I destroy the invoker, on the same thread that the
initial AsyncInvoke was called on.

If "Foo()" is in progress, then the destructor will wait for it to complete, the
inner AsyncInvoke will return early due to "destroying_", and we'll be ok.

And even if the destructor is called *right* after the destroying_ flag is
checked, pending_invocations_ will still be 1, because the outer AsyncClosure
hasn't yet finished. So pending_invocations_ will be incremented to 2, then
decremented to 1, and the destructor will either end up waiting for Bar() to
finish, or MessageQueueManager::Clear() will clear the Bar task.

+    // 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.
   ScopedRefMessageData<AsyncClosure>* data =
       static_cast<ScopedRefMessageData<AsyncClosure>*>(msg->pdata);
   scoped_refptr<AsyncClosure> closure = data->data();
   delete msg->pdata;
   msg->pdata = NULL;
 
@@ skipping 20 matching lines @@
 }
 
 void AsyncInvoker::DoInvoke(const Location& posted_from,
                             Thread* thread,
                             const scoped_refptr<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 ScopedRefMessageData<AsyncClosure>(closure));
 }
 
 void AsyncInvoker::DoInvokeDelayed(const Location& posted_from,
                                    Thread* thread,
                                    const scoped_refptr<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_);

[pthatcher1, 2017/02/17 18:19:38]

Do we have any sense of what perf impact these additional atomic operations and
events firing will have?  We these methods are called a lot in our code (maybe
several times per packet?).

[Taylor Brandstetter, 2017/02/17 19:00:46]

Assuming the CPU architecture supports atomic operations, should be pretty low;
much lower than the cost of a mutex. See
https://www.arangodb.com/2015/02/comparing-atomic-mutex-rwlocks/ and
http://stackoverflow.com/questions/2538070/atomic-operation-cost

   thread->PostDelayed(posted_from, delay_ms, this, id,
                       new ScopedRefMessageData<AsyncClosure>(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();

[pthatcher1, 2017/02/17 18:19:38]

Why does it need the friend class to touch a private member?  Why not just have
a public method that can be called?

[Taylor Brandstetter, 2017/02/17 19:00:46]

Because only AsyncClosure should be able to do this, which is exactly the
purpose of friend classes. This is an acceptable use of "friend" since
AsyncClosure is an implementation detail of AsyncInvoker which nothing outside
it will use directly.

+}
+
 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