Make AsyncHandleWaiter reenterancy-safe

On Windows, message pump can re-enter. AsyncHandleWaiter should
be prepared for that.

R=viettrungluu@chromium.org
BUG=none
TEST=ExtensionApiTest.SharedModuleWhitelist (this hits the DCHECK)

Review URL: https://codereview.chromium.org/973213002

Cr-Commit-Position: refs/heads/master@{#319202}

ipc/mojo/async_handle_waiter.cc

@@ -9,7 +9,6 @@
 #include "base/bind_helpers.h"
 #include "base/location.h"
 #include "base/logging.h"
-#include "base/message_loop/message_loop.h"
 #include "third_party/mojo/src/mojo/edk/embedder/embedder.h"
 
 namespace IPC {
@@ -32,7 +31,7 @@ class AsyncHandleWaiter::Context
       : io_runner_(base::MessageLoopForIO::current()->task_runner()),
         waiter_(waiter),
         last_result_(MOJO_RESULT_INTERNAL),
-        processing_(false),
+        io_loop_level_(0),
         should_invoke_callback_(false) {
     base::MessageLoopForIO::current()->AddIOObserver(this);
   }
@@ -67,7 +66,7 @@ class AsyncHandleWaiter::Context
       return false;
     if (loop->task_runner() != io_runner_)
       return false;
-    return processing_;
+    return io_loop_level_ > 0;
   }
 
   // Called from |io_runner_| thus safe to touch |waiter_|.
@@ -81,19 +80,25 @@ class AsyncHandleWaiter::Context
   // IOObserver implementation:
 
   void WillProcessIOEvent() override {
-    DCHECK(!should_invoke_callback_);
-    DCHECK(!processing_);
-    processing_ = true;
+    DCHECK(io_loop_level_ != 0 || !should_invoke_callback_);
+    DCHECK_GE(io_loop_level_, 0);
+    io_loop_level_++;
   }
 
   void DidProcessIOEvent() override {
-    DCHECK(processing_);
+    DCHECK_GE(io_loop_level_, 1);
+
+    // Leaving a nested loop.
+    if (io_loop_level_ > 1) {
+      io_loop_level_--;
+      return;
+    }
 
     // The zero |waiter_| indicates that |this| have lost the owner and can be
     // under destruction. So we cannot wrap it with a |scoped_refptr| anymore.
     if (!waiter_) {
       should_invoke_callback_ = false;
-      processing_ = false;
+      io_loop_level_--;
       return;
     }
 
@@ -105,7 +110,7 @@ class AsyncHandleWaiter::Context
       InvokeWaiterCallback();
     }
 
-    processing_ = false;
+    io_loop_level_--;
   }
 
   // Only |io_runner_| is accessed from arbitrary threads.  Others are touched
@@ -114,7 +119,7 @@ class AsyncHandleWaiter::Context
 
   const base::WeakPtr<AsyncHandleWaiter> waiter_;
   MojoResult last_result_;
-  bool processing_;
+  int io_loop_level_;
   bool should_invoke_callback_;
 
   DISALLOW_COPY_AND_ASSIGN(Context);
@@ -139,6 +144,14 @@ void AsyncHandleWaiter::InvokeCallback(MojoResult result) {
   callback_.Run(result);
 }
 
+base::MessageLoopForIO::IOObserver* AsyncHandleWaiter::GetIOObserverForTest() {
+  return context_.get();
+}
+
+base::Callback<void(MojoResult)> AsyncHandleWaiter::GetWaitCallbackForTest() {
+  return base::Bind(&Context::HandleIsReady, context_);
+}
+
 // static
 void AsyncHandleWaiterContextTraits::Destruct(
     const AsyncHandleWaiter::Context* context) {

ipc/mojo/async_handle_waiter.h

@@ -8,6 +8,7 @@
 #include "base/callback.h"
 #include "base/memory/ref_counted.h"
 #include "base/memory/weak_ptr.h"
+#include "base/message_loop/message_loop.h"
 #include "ipc/ipc_export.h"
 #include "third_party/mojo/src/mojo/public/c/system/types.h"
 
@@ -32,6 +33,9 @@ class IPC_MOJO_EXPORT AsyncHandleWaiter {
 
   MojoResult Wait(MojoHandle handle, MojoHandleSignals signals);
 
+  base::MessageLoopForIO::IOObserver* GetIOObserverForTest();
+  base::Callback<void(MojoResult)> GetWaitCallbackForTest();
+
  private:
   void InvokeCallback(MojoResult result);
 

ipc/mojo/async_handle_waiter_unittest.cc

@@ -204,6 +204,53 @@ TEST_F(AsyncHandleWaiterTest, RestartWaitingWhileSignaled) {
   ignore_result(pipe_to_read_.release());
 }
 
+class AsyncHandleWaiterIOObserverTest : public testing::Test {
+ public:
+  void SetUp() override {
+    message_loop_.reset(new base::MessageLoopForIO());
+    target_.reset(new AsyncHandleWaiter(
+        base::Bind(&AsyncHandleWaiterIOObserverTest::HandleIsReady,
+                   base::Unretained(this))));
+    invocation_count_ = 0;
+  }
+
+  void HandleIsReady(MojoResult result) { invocation_count_++; }
+
+  scoped_ptr<base::MessageLoop> message_loop_;
+  scoped_ptr<AsyncHandleWaiter> target_;
+  size_t invocation_count_;
+};
+
+TEST_F(AsyncHandleWaiterIOObserverTest, OutsideIOEvnet) {
+  target_->GetWaitCallbackForTest().Run(MOJO_RESULT_OK);
+  EXPECT_EQ(0U, invocation_count_);
+  message_loop_->RunUntilIdle();
+  EXPECT_EQ(1U, invocation_count_);
+}
+
+TEST_F(AsyncHandleWaiterIOObserverTest, InsideIOEvnet) {
+  target_->GetIOObserverForTest()->WillProcessIOEvent();
+  target_->GetWaitCallbackForTest().Run(MOJO_RESULT_OK);
+  EXPECT_EQ(0U, invocation_count_);
+  target_->GetIOObserverForTest()->DidProcessIOEvent();
+  EXPECT_EQ(1U, invocation_count_);
+}
+
+TEST_F(AsyncHandleWaiterIOObserverTest, Reenter) {
+  target_->GetIOObserverForTest()->WillProcessIOEvent();
+  target_->GetWaitCallbackForTest().Run(MOJO_RESULT_OK);
+  EXPECT_EQ(0U, invocation_count_);
+
+  // As if some other io handler start nested loop.
+  target_->GetIOObserverForTest()->WillProcessIOEvent();
+  target_->GetWaitCallbackForTest().Run(MOJO_RESULT_OK);
+  target_->GetIOObserverForTest()->DidProcessIOEvent();
+  EXPECT_EQ(0U, invocation_count_);
+
+  target_->GetIOObserverForTest()->DidProcessIOEvent();
+  EXPECT_EQ(1U, invocation_count_);
+}
+
 }  // namespace
 }  // namespace internal
 }  // namespace IPC