Mojo: cancel pending AsyncWait calls when the callers thread exits

This should resolve some memory leak issues.

BUG=372487

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

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@270639 0039d316-1c4b-4281-b951-d872f2087c98

4 files changed, 81 insertions, 57 deletions

diff --git a/mojo/apps/js/test/js_to_cpp_unittest.cc b/mojo/apps/js/test/js_to_cpp_unittest.cc
index 90a9275..d9e4cbd 100644
--- a/mojo/apps/js/test/js_to_cpp_unittest.cc
+++ b/mojo/apps/js/test/js_to_cpp_unittest.cc
@@ -179,7 +179,6 @@ class CppSideConnection : public js_to_cpp::CppSide {
   js_to_cpp::JsSide* js_side_;
 
  private:
-  Environment environment;
   DISALLOW_COPY_AND_ASSIGN(CppSideConnection);
 };
 
@@ -314,6 +313,7 @@ class JsToCppTest : public testing::Test {
   }
 
  private:
+  Environment environment;
   base::MessageLoop loop;
   base::RunLoop run_loop_;
 
diff --git a/mojo/common/handle_watcher.cc b/mojo/common/handle_watcher.cc
index a95669c..1affa90 100644
--- a/mojo/common/handle_watcher.cc
+++ b/mojo/common/handle_watcher.cc
@@ -253,29 +253,55 @@ WatcherThreadManager::WatcherThreadManager()
 
 }  // namespace
 
-// HandleWatcher::StartState ---------------------------------------------------
+// HandleWatcher::State --------------------------------------------------------
 
-// Contains the information passed to Start().
-struct HandleWatcher::StartState {
-  explicit StartState(HandleWatcher* watcher) : weak_factory(watcher) {
+// Represents the state of the HandleWatcher. Owns the user's callback and
+// monitors the current thread's MessageLoop to know when to force the callback
+// to run (with an error) even though the pipe hasn't been signaled yet.
+class HandleWatcher::State : public base::MessageLoop::DestructionObserver {
+ public:
+  State(HandleWatcher* watcher,
+        const Handle& handle,
+        MojoWaitFlags wait_flags,
+        MojoDeadline deadline,
+        const base::Callback<void(MojoResult)>& callback)
+      : watcher_(watcher),
+        callback_(callback),
+        weak_factory_(this) {
+    base::MessageLoop::current()->AddDestructionObserver(this);
+
+    watcher_id_ = WatcherThreadManager::GetInstance()->StartWatching(
+        handle,
+        wait_flags,
+        MojoDeadlineToTimeTicks(deadline),
+        base::Bind(&State::OnHandleReady, weak_factory_.GetWeakPtr()));
   }
 
-  ~StartState() {
+  virtual ~State() {
+    base::MessageLoop::current()->RemoveDestructionObserver(this);
+
+    WatcherThreadManager::GetInstance()->StopWatching(watcher_id_);
   }
 
-  // ID assigned by WatcherThreadManager.
-  WatcherID watcher_id;
+ private:
+  virtual void WillDestroyCurrentMessageLoop() OVERRIDE {
+    // The current thread is exiting. Simulate a watch error.
+    OnHandleReady(MOJO_RESULT_ABORTED);
+  }
 
-  // Callback to notify when done.
-  base::Callback<void(MojoResult)> callback;
+  void OnHandleReady(MojoResult result) {
+    base::Callback<void(MojoResult)> callback = callback_;
+    watcher_->Stop();  // Destroys |this|.
+
+    callback.Run(result);
+  }
 
-  // When Start() is invoked a callback is passed to WatcherThreadManager
-  // using a WeakRef from |weak_refactory_|. The callback invokes
-  // OnHandleReady() (on the thread Start() is invoked from) which in turn
-  // notifies |callback_|. Doing this allows us to reset state when the handle
-  // is ready, and then notify the callback. Doing this also means Stop()
-  // cancels any pending callbacks that may be inflight.
-  base::WeakPtrFactory<HandleWatcher> weak_factory;
+  HandleWatcher* watcher_;
+  WatcherID watcher_id_;
+  base::Callback<void(MojoResult)> callback_;
+
+  // Used to weakly bind |this| to the WatcherThreadManager.
+  base::WeakPtrFactory<State> weak_factory_;
 };
 
 // HandleWatcher ---------------------------------------------------------------
@@ -284,7 +310,6 @@ HandleWatcher::HandleWatcher() {
 }
 
 HandleWatcher::~HandleWatcher() {
-  Stop();
 }
 
 void HandleWatcher::Start(const Handle& handle,
@@ -294,33 +319,11 @@ void HandleWatcher::Start(const Handle& handle,
   DCHECK(handle.is_valid());
   DCHECK_NE(MOJO_WAIT_FLAG_NONE, wait_flags);
 
-  Stop();
-
-  start_state_.reset(new StartState(this));
-  start_state_->callback = callback;
-  start_state_->watcher_id =
-      WatcherThreadManager::GetInstance()->StartWatching(
-          handle,
-          wait_flags,
-          MojoDeadlineToTimeTicks(deadline),
-          base::Bind(&HandleWatcher::OnHandleReady,
-                     start_state_->weak_factory.GetWeakPtr()));
+  state_.reset(new State(this, handle, wait_flags, deadline, callback));
 }
 
 void HandleWatcher::Stop() {
-  if (!start_state_.get())
-    return;
-
-  scoped_ptr<StartState> old_state(start_state_.Pass());
-  WatcherThreadManager::GetInstance()->StopWatching(old_state->watcher_id);
-}
-
-void HandleWatcher::OnHandleReady(MojoResult result) {
-  DCHECK(start_state_.get());
-  scoped_ptr<StartState> old_state(start_state_.Pass());
-  old_state->callback.Run(result);
-
-  // NOTE: We may have been deleted during callback execution.
+  state_.reset();
 }
 
 }  // namespace common
diff --git a/mojo/common/handle_watcher.h b/mojo/common/handle_watcher.h
index aa151dc..0660abb 100644
--- a/mojo/common/handle_watcher.h
+++ b/mojo/common/handle_watcher.h
@@ -33,7 +33,9 @@ class MOJO_COMMON_EXPORT HandleWatcher {
   // words, Start() performs one asynchronous watch at a time. It is ok to call
   // Start() multiple times, but it cancels any existing watches. |callback| is
   // notified when the handle is ready, invalid or deadline has passed and is
-  // notified on the thread Start() was invoked on.
+  // notified on the thread Start() was invoked on. If the current thread exits
+  // before the handle is ready, then |callback| is invoked with a result of
+  // MOJO_RESULT_ABORTED.
   void Start(const Handle& handle,
              MojoWaitFlags wait_flags,
              MojoDeadline deadline,
@@ -43,13 +45,10 @@ class MOJO_COMMON_EXPORT HandleWatcher {
   void Stop();
 
  private:
-  struct StartState;
-
-  // See description of |StartState::weak_factory| for details.
-  void OnHandleReady(MojoResult result);
+  class State;
 
   // If non-NULL Start() has been invoked.
-  scoped_ptr<StartState> start_state_;
+  scoped_ptr<State> state_;
 
   DISALLOW_COPY_AND_ASSIGN(HandleWatcher);
 };
diff --git a/mojo/common/handle_watcher_unittest.cc b/mojo/common/handle_watcher_unittest.cc
index 5cc8cfa..37fde23 100644
--- a/mojo/common/handle_watcher_unittest.cc
+++ b/mojo/common/handle_watcher_unittest.cc
@@ -20,6 +20,13 @@ namespace mojo {
 namespace common {
 namespace test {
 
+void ObserveCallback(bool* was_signaled,
+                     MojoResult* result_observed,
+                     MojoResult result) {
+  *was_signaled = true;
+  *result_observed = result;
+}
+
 void RunUntilIdle() {
   base::RunLoop run_loop;
   run_loop.RunUntilIdle();
@@ -101,14 +108,6 @@ class HandleWatcherTest : public testing::Test {
     test::SetTickClockForTest(NULL);
   }
 
-  virtual void SetUp() OVERRIDE {
-    environment_.reset(new Environment);
-  }
-
-  virtual void TearDown() OVERRIDE {
-    environment_.reset();
-  }
-
  protected:
   void InstallTickClock() {
     test::SetTickClockForTest(&tick_clock_);
@@ -117,8 +116,8 @@ class HandleWatcherTest : public testing::Test {
   base::SimpleTestTickClock tick_clock_;
 
  private:
+  Environment environment_;
   base::MessageLoop message_loop_;
-  scoped_ptr<Environment> environment_;
 
   DISALLOW_COPY_AND_ASSIGN(HandleWatcherTest);
 };
@@ -315,6 +314,29 @@ TEST_F(HandleWatcherTest, DeleteInCallback) {
   EXPECT_TRUE(callback_helper.got_callback());
 }
 
+TEST(HandleWatcherCleanEnvironmentTest, AbortedOnMessageLoopDestruction) {
+  bool was_signaled = false;
+  MojoResult result = MOJO_RESULT_OK;
+
+  Environment env;
+
+  MessagePipe pipe;
+  HandleWatcher watcher;
+  {
+    base::MessageLoop loop;
+
+    watcher.Start(pipe.handle0.get(),
+                  MOJO_WAIT_FLAG_READABLE,
+                  MOJO_DEADLINE_INDEFINITE,
+                  base::Bind(&ObserveCallback, &was_signaled, &result));
+
+    // Now, let the MessageLoop get torn down. We expect our callback to run.
+  }
+
+  EXPECT_TRUE(was_signaled);
+  EXPECT_EQ(MOJO_RESULT_ABORTED, result);
+}
+
 }  // namespace test
 }  // namespace common
 }  // namespace mojo