Fixing race between ~AsyncInvoker and ~AsyncClosure, using ref-counting.

webrtc/rtc_base/asyncinvoker.cc

Index: webrtc/rtc_base/asyncinvoker.cc
diff --git a/webrtc/rtc_base/asyncinvoker.cc b/webrtc/rtc_base/asyncinvoker.cc
index 94abfd5d087a59c66cb121270107b9be973282cd..d1fabe0bc8b5e41840f386e215e57a3dc5a5c672 100644
--- a/webrtc/rtc_base/asyncinvoker.cc
+++ b/webrtc/rtc_base/asyncinvoker.cc
@@ -16,10 +16,11 @@
 
 namespace rtc {
 
-AsyncInvoker::AsyncInvoker() : invocation_complete_(false, false) {}
+AsyncInvoker::AsyncInvoker()
+    : invocation_complete_(new RefCountedObject<Event>(false, false)) {}
 
 AsyncInvoker::~AsyncInvoker() {
-  destroying_ = true;
+  AtomicOps::Increment(&destroying_);
   // 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
@@ -29,8 +30,8 @@ AsyncInvoker::~AsyncInvoker() {
     // 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);
+    Thread::Current()->Clear(this);

[kwiberg-webrtc, 2017/08/04 09:28:09]

What does the change on this line do?

[Taylor Brandstetter, 2017/08/04 19:14:22]

It only clears events on the current thread, which is all that's necessary (see
comment above). This is something nisse@ pointed out in a previous review:
https://codereview.webrtc.org/2876273002/diff/1/webrtc/base/asyncinvoker.cc#n...

+    invocation_complete_->Wait(Event::kForever);
   }
 }
 
@@ -44,7 +45,8 @@ void AsyncInvoker::OnMessage(Message* msg) {
 }
 
 void AsyncInvoker::Flush(Thread* thread, uint32_t id /*= MQID_ANY*/) {
-  if (destroying_) return;
+  if (AtomicOps::AcquireLoad(&destroying_))
+    return;

[kwiberg-webrtc, 2017/08/04 09:28:09]

A comment explaining why this is a no-op if we're being destroyed?

[Taylor Brandstetter, 2017/08/04 19:14:22]

I don't know exactly. I went back to the CL where this was added, in 2014, but
didn't find a clear explanation.

Though it makes sense that if you're destroying, you don't want to execute even
more tasks. The goal is to finish all tasks in progress, not start executing new
ones. I'll add a comment.

 
   // Run this on |thread| to reduce the number of context switches.
   if (Thread::Current() != thread) {
@@ -65,11 +67,14 @@ void AsyncInvoker::DoInvoke(const Location& posted_from,
                             Thread* thread,
                             std::unique_ptr<AsyncClosure> closure,
                             uint32_t id) {
-  if (destroying_) {
+  if (AtomicOps::AcquireLoad(&destroying_)) {
+    // Note that this may be expected, if the application is AsyncInvoking
+    // tasks that AsyncInvoke other tasks. But otherwise it indicates a race
+    // between a thread destroying the AsyncInvoker and a thread still trying
+    // to use it.

[kwiberg-webrtc, 2017/08/04 09:28:09]

Doesn't that mean that the warning message is flaky? If we don't know if a
situation is an error or not, should we warn about it?

[Taylor Brandstetter, 2017/08/04 19:14:22]

My thinking was "I'd rather have a flaky warning that directs people to this
comment than have no warning at all." Our guidelines say "warning" log levels
should be used "for undesired but relatively expected events, which *may*
indicate a problem."

Though, are there any chromium-specific log level guidelines? I couldn't find
any.

[kwiberg-webrtc, 2017/08/06 04:17:12]

Hmm, OK.
Not that I'm aware of.

     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)));
 }
@@ -79,11 +84,11 @@ void AsyncInvoker::DoInvokeDelayed(const Location& posted_from,
                                    std::unique_ptr<AsyncClosure> closure,
                                    uint32_t delay_ms,
                                    uint32_t id) {
-  if (destroying_) {
+  if (AtomicOps::AcquireLoad(&destroying_)) {
+    // See above comment.
     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)));
 }
@@ -97,7 +102,7 @@ GuardedAsyncInvoker::~GuardedAsyncInvoker() {
 }
 
 bool GuardedAsyncInvoker::Flush(uint32_t id) {
-  rtc::CritScope cs(&crit_);
+  CritScope cs(&crit_);
   if (thread_ == nullptr)
     return false;
   invoker_.Flush(thread_, id);
@@ -105,15 +110,27 @@ bool GuardedAsyncInvoker::Flush(uint32_t id) {
 }
 
 void GuardedAsyncInvoker::ThreadDestroyed() {
-  rtc::CritScope cs(&crit_);
+  CritScope cs(&crit_);
   // We should never get more than one notification about the thread dying.
   RTC_DCHECK(thread_ != nullptr);
   thread_ = nullptr;
 }
 
+AsyncClosure::AsyncClosure(AsyncInvoker* invoker)
+    : invoker_(invoker), invocation_complete_(invoker_->invocation_complete_) {
+  AtomicOps::Increment(&invoker_->pending_invocations_);
+}
+
 AsyncClosure::~AsyncClosure() {
   AtomicOps::Decrement(&invoker_->pending_invocations_);
-  invoker_->invocation_complete_.Set();
+  // After |pending_invocations_| is decremented, we may need to signal
+  // |invocation_complete_| in case the AsyncInvoker is being destroyed and
+  // waiting for pending tasks to complete.
+  //
+  // It's also possible that the destructor finishes before "Set()" is called,
+  // which is safe because the event is reference counted (and in a thread-safe
+  // way).
+  invocation_complete_->Set();
 }
 
 }  // namespace rtc