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

webrtc/rtc_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/rtc_base/asyncinvoker.h"
 
 #include "webrtc/rtc_base/atomicops.h"
 #include "webrtc/rtc_base/checks.h"
 #include "webrtc/rtc_base/logging.h"
 
 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
   // 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);
+    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);
   }
 }
 
 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;
 }
 
 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) {
     thread->Invoke<void>(RTC_FROM_HERE,
                          Bind(&AsyncInvoker::Flush, this, thread, id));
     return;
   }
 
   MessageList removed;
   thread->Clear(this, id, &removed);
   for (MessageList::iterator it = removed.begin(); it != removed.end(); ++it) {
     // This message was pending on this thread, so run it now.
     thread->Send(it->posted_from, it->phandler, it->message_id, it->pdata);
   }
 }
 
 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)));
 }
 
 void AsyncInvoker::DoInvokeDelayed(const Location& posted_from,
                                    Thread* thread,
                                    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)));
 }
 
 GuardedAsyncInvoker::GuardedAsyncInvoker() : thread_(Thread::Current()) {
   thread_->SignalQueueDestroyed.connect(this,
                                         &GuardedAsyncInvoker::ThreadDestroyed);
 }
 
 GuardedAsyncInvoker::~GuardedAsyncInvoker() {
 }
 
 bool GuardedAsyncInvoker::Flush(uint32_t id) {
-  rtc::CritScope cs(&crit_);
+  CritScope cs(&crit_);
   if (thread_ == nullptr)
     return false;
   invoker_.Flush(thread_, id);
   return true;
 }
 
 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