Making AsyncInvoker destructor thread-safe.

webrtc/base/asyncinvoker.cc

Index: webrtc/base/asyncinvoker.cc
diff --git a/webrtc/base/asyncinvoker.cc b/webrtc/base/asyncinvoker.cc
index 83a873811e7258a16a982c39609598114ca78055..8f141e05ec36a5bed9d2b3d49dce861c27c31bd6 100644
--- a/webrtc/base/asyncinvoker.cc
+++ b/webrtc/base/asyncinvoker.cc
@@ -10,18 +10,29 @@
 
 #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) {
@@ -62,6 +73,7 @@ void AsyncInvoker::DoInvoke(const Location& posted_from,
     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));
 }
@@ -75,6 +87,7 @@ void AsyncInvoker::DoInvokeDelayed(const Location& posted_from,
     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));
 }
@@ -102,11 +115,16 @@ void GuardedAsyncInvoker::ThreadDestroyed() {
   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(