Don't leak the WatcherThreadManager background thread.

Leaking it triggered some canary/dev build checks on Windows that ensure we don't leak handles. In this case, the leak is ok, but the simplest thing for now is to stop leaking it. A side-benefit is we ensure that Mojo message loops shut down correctly at process shutdown.

BUG=559238
R=sky@chromium.org

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

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

base/threading/thread.cc

@@ -262,8 +262,12 @@ void Thread::ThreadMain() {
   com_initializer.reset();
 #endif
 
-  // Assert that MessageLoop::QuitWhenIdle was called by ThreadQuitHelper.
-  DCHECK(GetThreadWasQuitProperly());
+  if (message_loop->type() != MessageLoop::TYPE_CUSTOM) {
+    // Assert that MessageLoop::QuitWhenIdle was called by ThreadQuitHelper.
+    // Don't check for custom message pumps, because their shutdown might not
+    // allow this.
+    DCHECK(GetThreadWasQuitProperly());
+  }
 
   // We can't receive messages anymore.
   // (The message loop is destructed at the end of this block)

mojo/edk/system/raw_channel.cc

@@ -222,6 +222,7 @@ void RawChannel::LazyInitialize() {
     return;
   initialized_ = true;
   internal::ChannelStarted();
+  base::MessageLoop::current()->AddDestructionObserver(this);
 
   OnInit();
 
@@ -286,8 +287,10 @@ void RawChannel::Shutdown() {
       OnShutdownNoLock(read_buffer_.Pass(), write_buffer_.Pass());
     }
 
-    if (initialized_)
+    if (initialized_) {
       internal::ChannelShutdown();
+      base::MessageLoop::current()->RemoveDestructionObserver(this);
+    }
     delete this;
     return;
   }
@@ -715,9 +718,14 @@ void RawChannel::UpdateWriteBuffer(size_t platform_handles_written,
 }
 
 void RawChannel::CallOnReadCompleted(IOResult io_result, size_t bytes_read) {
-  base::AutoLock locker(read_lock());
+  base::AutoLock locker(read_lock_);
   OnReadCompletedNoLock(io_result, bytes_read);
 }
 
+void RawChannel::WillDestroyCurrentMessageLoop() {
+  base::AutoLock locker(read_lock_);
+  OnReadCompletedNoLock(IO_FAILED_SHUTDOWN, 0);
+}
+
 }  // namespace edk
 }  // namespace mojo

mojo/edk/system/raw_channel.h

@@ -9,6 +9,7 @@
 
 #include "base/memory/scoped_ptr.h"
 #include "base/memory/weak_ptr.h"
+#include "base/message_loop/message_loop.h"
 #include "base/synchronization/lock.h"
 #include "mojo/edk/embedder/platform_handle_vector.h"
 #include "mojo/edk/embedder/scoped_platform_handle.h"
@@ -35,7 +36,8 @@ namespace edk {
 // With the exception of |WriteMessage()|, this class is thread-unsafe (and in
 // general its methods should only be used on the I/O thread, i.e., the thread
 // on which |Init()| is called).
-class MOJO_SYSTEM_IMPL_EXPORT RawChannel {
+class MOJO_SYSTEM_IMPL_EXPORT RawChannel :
+    public base::MessageLoop::DestructionObserver {
  public:
 
   // The |Delegate| is only accessed on the same thread as the message loop
@@ -239,11 +241,11 @@ class MOJO_SYSTEM_IMPL_EXPORT RawChannel {
   // Shutdown must be called on the IO thread. This object deletes itself once
   // it's flushed all pending writes and insured that the other side of the pipe
   // read them.
-  virtual ~RawChannel();
+  ~RawChannel() override;
 
   // |result| must not be |IO_PENDING|. Must be called on the I/O thread WITHOUT
   // |write_lock_| held. This object may be destroyed by this call. This
-  // acquires |read_lock_| inside of it. The caller needs to acquire read_lock_
+  // acquires |write_lock_| inside of it. The caller needs to acquire read_lock_
   // first.
   void OnReadCompletedNoLock(IOResult io_result, size_t bytes_read);
   // |result| must not be |IO_PENDING|. Must be called on the I/O thread WITHOUT
@@ -414,6 +416,9 @@ class MOJO_SYSTEM_IMPL_EXPORT RawChannel {
   // Connects to the OS pipe.
   void LazyInitialize();
 
+  // base::MessageLoop::DestructionObserver:
+  void WillDestroyCurrentMessageLoop() override;
+
 
 
 

mojo/message_pump/handle_watcher.cc

@@ -223,13 +223,7 @@ WatcherThreadManager::~WatcherThreadManager() {
 }
 
 WatcherThreadManager* WatcherThreadManager::GetInstance() {
-  // We need to leak this because otherwise when the process dies, AtExitManager
-  // waits for destruction which waits till the handle watcher thread is joined.
-  // But that can't happen since the pump uses mojo message pipes to wake up the
-  // pump. Since mojo EDK has been shutdown already, this never completes.
-  return base::Singleton<WatcherThreadManager,
-                         base::LeakySingletonTraits<WatcherThreadManager>>::
-      get();
+  return base::Singleton<WatcherThreadManager>::get();
 }
 
 WatcherID WatcherThreadManager::StartWatching(

mojo/message_pump/message_pump_mojo.cc

@@ -168,15 +168,23 @@ bool MessagePumpMojo::DoInternalWork(const RunState& run_state, bool block) {
   const MojoDeadline deadline = block ? GetDeadlineForWait(run_state) : 0;
   const WaitState wait_state = GetWaitState();
 
+  std::vector<MojoHandleSignalsState> states(wait_state.handles.size());
   const WaitManyResult wait_many_result =
-      WaitMany(wait_state.handles, wait_state.wait_signals, deadline, nullptr);
+      WaitMany(wait_state.handles, wait_state.wait_signals, deadline, &states);
   const MojoResult result = wait_many_result.result;
   bool did_work = true;
   if (result == MOJO_RESULT_OK) {
     if (wait_many_result.index == 0) {
-      // Control pipe was written to.
-      ReadMessageRaw(read_handle_.get(), NULL, NULL, NULL, NULL,
-                     MOJO_READ_MESSAGE_FLAG_MAY_DISCARD);
+      if (states[0].satisfied_signals & MOJO_HANDLE_SIGNAL_PEER_CLOSED) {
+        // 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();
+      } else {
+        // Control pipe was written to.
+        ReadMessageRaw(read_handle_.get(), NULL, NULL, NULL, NULL,
+                       MOJO_READ_MESSAGE_FLAG_MAY_DISCARD);
+      }
     } else {
       DCHECK(handlers_.find(wait_state.handles[wait_many_result.index]) !=
              handlers_.end());
@@ -245,6 +253,11 @@ 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.
+    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);
@@ -253,7 +266,8 @@ void MessagePumpMojo::SignalControlPipe() {
 MessagePumpMojo::WaitState MessagePumpMojo::GetWaitState() const {
   WaitState wait_state;
   wait_state.handles.push_back(read_handle_.get());
-  wait_state.wait_signals.push_back(MOJO_HANDLE_SIGNAL_READABLE);
+  wait_state.wait_signals.push_back(
+      MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED);
 
   for (HandleToHandler::const_iterator i = handlers_.begin();
        i != handlers_.end(); ++i) {