Fixing race between ~AsyncInvoker and ~AsyncClosure.

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() : invocation_complete_(false, false) {}
+AsyncInvoker::AsyncInvoker() {}
 
 AsyncInvoker::~AsyncInvoker() {
   destroying_ = true;
   // 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.
+    // Release time slice to allow other async-invoked tasks to make progress.
+    rtc::Thread::Current()->SleepMs(0);

[nisse-webrtc, 2017/05/19 07:16:59]

I think it's better with some small but non-zero value, say 10ms, so we don't
hog the cpu if the pending tasks is sleeping.

+    // One of the async-invoked tasks in progress could have posted an
+    // additional task to the current thread. So we need to keep calling Clear
+    // to discard these posts.
     MessageQueueManager::Clear(this);

[nisse-webrtc, 2017/05/19 07:16:59]

According to the comment, we only need to clear the queue for the current
thread. Then

  rtc::Thread::Current()->Clear(this);

should be enough, right? The MessageQueueManager method iterates over all
MessageQueues it knows about.

-    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 14 matching lines @@
   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_) {

[nisse-webrtc, 2017/05/19 07:16:59]

I'm still a bit confused by this flag. My current understanding is that it's set
in the destructor, and we want it to "eventually" propagate to other threads
that might be posting tasks to the thread that is blocking in the destructor.
Maybe a volatile declaration is good enough for that purpose.

But since it is accessed from multiple threads, tsan is likely to warn about it,
and it would be appropriate using atomic access, and a comment on the weak
synchronization intended.

It's a bit silly, in that I think we already have all the locks needed to do
hard synchronization, but they're private to other classes. What we'd need is a
a Post method with an additional std::function<bool()>  argument, with the
semantics that MessageQueue::Post will call the function *while holding the
message queue lock*, and push the message only if the function returns true.

And a similar callback for the Clear function. Then we could get setting and
checking of the |destroying_| flag to be atomic with respect to the queue
modifications.

I don't have that much experience with thread programming. Are there any known
patterns with that kind of callbacks, to piggy back on another class' atomic
operations? I imagine there are some pitfalls.

     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(AsyncInvoker* invoker) : invoker_(invoker) {
+  AtomicOps::Increment(&invoker_->pending_invocations_);
+}
+
 AsyncClosure::~AsyncClosure() {
   AtomicOps::Decrement(&invoker_->pending_invocations_);
-  invoker_->invocation_complete_.Set();
 }
 
 NotifyingAsyncClosureBase::NotifyingAsyncClosureBase(
     AsyncInvoker* invoker,
     const Location& callback_posted_from,
     Thread* calling_thread)
     : AsyncClosure(invoker),
       callback_posted_from_(callback_posted_from),
       calling_thread_(calling_thread) {
   calling_thread->SignalQueueDestroyed.connect(
@@ skipping 18 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_ == nullptr means do not trigger the callback.
   calling_thread_ = nullptr;
 }
 
 }  // namespace rtc