Fixing potential AsyncInvoker deadlock that occurs for "reentrant" invocations.

webrtc/base/thread_unittest.cc

 /*
  *  Copyright 2004 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.
  */
 
@@ skipping 86 matching lines @@
     TestMessage* msg = static_cast<TestMessage*>(pmsg->pdata);
     int result = Next(msg->value);
     EXPECT_GE(socket_->Send(&result, sizeof(result)), 0);
     delete msg;
   }
 
  private:
   Socket* socket_;
 };
 
-class CustomThread : public rtc::Thread {
+class CustomThread : public Thread {
  public:
   CustomThread() {}
   virtual ~CustomThread() { Stop(); }
   bool Start() { return false; }
 
   bool WrapCurrent() {
     return Thread::WrapCurrent();
   }
   void UnwrapCurrent() {
     Thread::UnwrapCurrent();
@@ skipping 25 matching lines @@
 // A bool wrapped in a mutex, to avoid data races. Using a volatile
 // bool should be sufficient for correct code ("eventual consistency"
 // between caches is sufficient), but we can't tell the compiler about
 // that, and then tsan complains about a data race.
 
 // See also discussion at
 // http://stackoverflow.com/questions/7223164/is-mutex-needed-to-synchronize-a-simple-flag-between-pthreads
 
 // Using std::atomic<bool> or std::atomic_flag in C++11 is probably
 // the right thing to do, but those features are not yet allowed. Or
-// rtc::AtomicInt, if/when that is added. Since the use isn't
+// AtomicInt, if/when that is added. Since the use isn't
 // performance critical, use a plain critical section for the time
 // being.
 
 class AtomicBool {
  public:
   explicit AtomicBool(bool value = false) : flag_(value) {}
   AtomicBool& operator=(bool value) {
     CritScope scoped_lock(&cs_);
     flag_ = value;
     return *this;
@@ skipping 280 matching lines @@
   AtomicBool called;
   invoker.AsyncInvoke<void>(RTC_FROM_HERE, &thread, FunctorB(&called));
   EXPECT_TRUE_WAIT(called.get(), kWaitTimeout);
 }
 
 TEST_F(AsyncInvokeTest, KillInvokerDuringExecute) {
   // Use these events to get in a state where the functor is in the middle of
   // executing, and then to wait for it to finish, ensuring the "EXPECT_FALSE"
   // is run.
   Event functor_started(false, false);
-  Event functor_continue(false, false);
   Event functor_finished(false, false);
 
   Thread thread;
   thread.Start();
   volatile bool invoker_destroyed = false;
   {
+    auto functor = [&functor_started, &functor_finished, &invoker_destroyed] {
+      functor_started.Set();
+      Thread::Current()->SleepMs(kWaitTimeout);
+      EXPECT_FALSE(invoker_destroyed);
+      functor_finished.Set();
+    };
     AsyncInvoker invoker;
-    invoker.AsyncInvoke<void>(RTC_FROM_HERE, &thread,
-                              [&functor_started, &functor_continue,
-                               &functor_finished, &invoker_destroyed] {
-                                functor_started.Set();
-                                functor_continue.Wait(Event::kForever);
-                                rtc::Thread::Current()->SleepMs(kWaitTimeout);
-                                EXPECT_FALSE(invoker_destroyed);
-                                functor_finished.Set();
-                              });
+    invoker.AsyncInvoke<void>(RTC_FROM_HERE, &thread, functor);
     functor_started.Wait(Event::kForever);
-
-    // Allow the functor to continue and immediately destroy the invoker.
-    functor_continue.Set();
+    // Destroy the invoker while the functor is still executing (doing
+    // SleepMs).
   }
 
   // If the destructor DIDN'T wait for the functor to finish executing, it will
   // hit the EXPECT_FALSE(invoker_destroyed) after it finishes sleeping for a
   // second.
   invoker_destroyed = true;
   functor_finished.Wait(Event::kForever);
 }
 
+// Variant of the above test where the async-invoked task calls AsyncInvoke
+// again, for the thread on which the AsyncInvoker is currently being
+// destroyed. This shouldn't deadlock or crash; this second invocation should
+// just be ignored.
+TEST_F(AsyncInvokeTest, KillInvokerDuringExecuteWithReentrantInvoke) {
+  Event functor_started(false, false);
+  bool reentrant_functor_run = false;
+
+  Thread* main = Thread::Current();
+  Thread thread;
+  thread.Start();
+  {
+    AsyncInvoker invoker;
+    auto reentrant_functor = [&reentrant_functor_run] {
+      reentrant_functor_run = true;
+    };
+    auto functor = [&functor_started, &invoker, main, reentrant_functor] {
+      functor_started.Set();
+      Thread::Current()->SleepMs(kWaitTimeout);
+      invoker.AsyncInvoke<void>(RTC_FROM_HERE, main, reentrant_functor);
+    };
+    // This queues a task on |thread| to sleep for |kWaitTimeout| then queue a
+    // task on |main|. But this second queued task should never run.
+    invoker.AsyncInvoke<void>(RTC_FROM_HERE, &thread, functor);
+    functor_started.Wait(Event::kForever);
+  }

[nisse-webrtc, 2017/05/18 08:04:55]

I'm trying to understand what happens here... Scope ends, so AsyncInvoker is
destroyed. And its destructor is going to block until |functor| has returned,
which takes about 1s?

That |reentrant_functor| never runs, is that due to the |destroying_| flag
(seems a bit scary, since it's set by the destructor running on main thread, and
read by the AsyncInvoke call on the other thread, with no synchronization), or
due to the MessageQueueManager::Clear call?

To me, it seems the synchronization primitive needed to make things easy is to
atomically {check that destroying_ is false and if so, increment the reference
count}. But seems difficult to implement without locks.

+  EXPECT_FALSE(reentrant_functor_run);
+}
+
 TEST_F(AsyncInvokeTest, Flush) {
   AsyncInvoker invoker;
   AtomicBool flag1;
   AtomicBool flag2;
   // Queue two async calls to the current thread.
   invoker.AsyncInvoke<void>(RTC_FROM_HERE, Thread::Current(), FunctorB(&flag1));
   invoker.AsyncInvoke<void>(RTC_FROM_HERE, Thread::Current(), FunctorB(&flag2));
   // Because we haven't pumped messages, these should not have run yet.
   EXPECT_FALSE(flag1.get());
   EXPECT_FALSE(flag2.get());
@@ skipping 110 matching lines @@
   // Execute pending calls with id == 5.
   EXPECT_TRUE(invoker.Flush(5));
   EXPECT_TRUE(flag1.get());
   EXPECT_FALSE(flag2.get());
   flag1 = false;
   // Execute all pending calls. The id == 5 call should not execute again.
   EXPECT_TRUE(invoker.Flush());
   EXPECT_FALSE(flag1.get());
   EXPECT_TRUE(flag2.get());
 }