Fix race on mojo message pump shutdown.

mojo/message_pump/message_pump_mojo.cc

 // Copyright 2013 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "mojo/message_pump/message_pump_mojo.h"
 
 #include <stdint.h>
 
 #include <algorithm>
 #include <map>
 #include <vector>
 
 #include "base/containers/small_map.h"
 #include "base/debug/alias.h"
 #include "base/lazy_instance.h"
 #include "base/logging.h"
 #include "base/threading/thread_local.h"
+#include "base/threading/thread_restrictions.h"
 #include "base/time/time.h"
 #include "mojo/message_pump/message_pump_mojo_handler.h"
 #include "mojo/message_pump/time_helper.h"
 #include "mojo/public/c/system/wait_set.h"
 
 namespace mojo {
 namespace common {
 namespace {
 
 base::LazyInstance<base::ThreadLocalPointer<MessagePumpMojo> >::Leaky
@@ skipping 13 matching lines @@
 }  // namespace
 
 struct MessagePumpMojo::RunState {
   RunState() : should_quit(false) {}
 
   base::TimeTicks delayed_work_time;
 
   bool should_quit;
 };
 
-MessagePumpMojo::MessagePumpMojo() : run_state_(NULL), next_handler_id_(0) {
+MessagePumpMojo::MessagePumpMojo()
+    : run_state_(NULL), next_handler_id_(0), event_(false, false) {
   DCHECK(!current())
       << "There is already a MessagePumpMojo instance on this thread.";
   g_tls_current_pump.Pointer()->Set(this);
 
   MojoResult result = CreateMessagePipe(nullptr, &read_handle_, &write_handle_);
   CHECK_EQ(result, MOJO_RESULT_OK);
   CHECK(read_handle_.is_valid());
   CHECK(write_handle_.is_valid());
 
   MojoHandle handle;
@@ skipping 101 matching lines @@
   base::AutoLock auto_lock(run_state_lock_);
   if (!run_state_)
     return;
   run_state_->delayed_work_time = delayed_work_time;
 }
 
 void MessagePumpMojo::DoRunLoop(RunState* run_state, Delegate* delegate) {
   bool more_work_is_plausible = true;
   for (;;) {
     const bool block = !more_work_is_plausible;
-    more_work_is_plausible = DoInternalWork(*run_state, block);
+    if (read_handle_.is_valid()) {
+      more_work_is_plausible = DoInternalWork(*run_state, block);
+    } else {
+      more_work_is_plausible = DoNonMojoWork(*run_state, block);
+    }
 
     if (run_state->should_quit)
       break;
 
     more_work_is_plausible |= delegate->DoWork();
     if (run_state->should_quit)
       break;
 
     more_work_is_plausible |= delegate->DoDelayedWork(
         &run_state->delayed_work_time);
@@ skipping 18 matching lines @@
     // unnecessary work.
     did_work = WaitForReadyHandles(run_state);
   }
 
   did_work |= ProcessReadyHandles();
   did_work |= RemoveExpiredHandles();
 
   return did_work;
 }
 
+bool MessagePumpMojo::DoNonMojoWork(const RunState& run_state, bool block) {
+  bool did_work = block;
+  if (block) {
+    const MojoDeadline deadline = GetDeadlineForWait(run_state);
+    // Stolen from base/message_loop/message_pump_default.cc
+    base::ThreadRestrictions::ScopedAllowWait allow_wait;
+    if (deadline == MOJO_DEADLINE_INDEFINITE) {
+      event_.Wait();
+    } else {
+      if (deadline > 0) {
+        event_.TimedWait(base::TimeDelta::FromMicroseconds(deadline));
+      } else {
+        did_work = false;
+      }
+    }
+    // Since event_ is auto-reset, we don't need to do anything special here
+    // other than service each delegate method.
+  }
+
+  did_work |= RemoveExpiredHandles();
+
+  return did_work;
+}
+
 bool MessagePumpMojo::WaitForReadyHandles(const RunState& run_state) const {
   const MojoDeadline deadline = GetDeadlineForWait(run_state);
   const MojoResult wait_result = Wait(
       wait_set_handle_.get(), MOJO_HANDLE_SIGNAL_READABLE, deadline, nullptr);
   if (wait_result == MOJO_RESULT_OK) {
     // Handles may be ready. Or not since wake-ups can be spurious in certain
     // circumstances.
     return true;
   } else if (wait_result == MOJO_RESULT_DEADLINE_EXCEEDED) {
     return false;
@@ skipping 61 matching lines @@
         (it == handlers_.end() || it->second.id != ready_handles[handle])) {
       continue;
     }
 
     switch (handle_results[i]) {
       case MOJO_RESULT_CANCELLED:
       case MOJO_RESULT_FAILED_PRECONDITION:
         DVLOG(1) << "Error: " << handle_results[i]
                  << " handle: " << handle.value();
         if (handle.value() == read_handle_.get().value()) {
-          // The Mojo EDK is shutting down. The ThreadQuitHelper task in
-          // base::Thread won't get run since the control pipe depends on the
-          // EDK staying alive. So quit manually to avoid this thread hanging.
-          Quit();
+          // The Mojo EDK is shutting down. We can't just quit the message pump
+          // because that may cause the thread to quit, which causes the
+          // thread's MessageLoop to be destroyed, which races with any use of
+          // |Thread::task_runner()|. So instead, we enter a "dumb" mode which
+          // bypasses Mojo and just acts like a trivial message pump. That way,
+          // we can wait for the usual thread exiting mechanism to happen.
+          // The dumb mode is indicated by releasing the control pipe's read
+          // handle.
+          read_handle_.reset();
         } else {
           RemoveInvalidHandle(handle_results[i], handle);
         }
         break;
       case MOJO_RESULT_OK:
         if (handle.value() == read_handle_.get().value()) {
           DVLOG(1) << "Signaled control pipe";
           // Control pipe was written to.
           ReadMessageRaw(read_handle_.get(), nullptr, nullptr, nullptr, nullptr,
                          MOJO_READ_MESSAGE_FLAG_MAY_DISCARD);
@@ skipping 61 matching lines @@
   }
   return removed;
 }
 
 void MessagePumpMojo::SignalControlPipe() {
   const MojoResult result =
       WriteMessageRaw(write_handle_.get(), NULL, 0, NULL, 0,
                       MOJO_WRITE_MESSAGE_FLAG_NONE);
   if (result == MOJO_RESULT_FAILED_PRECONDITION) {
     // Mojo EDK is shutting down.
+    event_.Signal();
     return;
   }
 
   // If we can't write we likely won't wake up the thread and there is a strong
   // chance we'll deadlock.
   CHECK_EQ(MOJO_RESULT_OK, result);
 }
 
 MojoDeadline MessagePumpMojo::GetDeadlineForWait(
     const RunState& run_state) const {
@@ skipping 19 matching lines @@
 void MessagePumpMojo::WillSignalHandler() {
   FOR_EACH_OBSERVER(Observer, observers_, WillSignalHandler());
 }
 
 void MessagePumpMojo::DidSignalHandler() {
   FOR_EACH_OBSERVER(Observer, observers_, DidSignalHandler());
 }
 
 }  // namespace common
 }  // namespace mojo