Makes HandleWatcher block until no longer waiting on pipe

mojo/common/handle_watcher.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/common/handle_watcher.h"
 
 #include <map>
 
 #include "base/atomic_sequence_num.h"
 #include "base/bind.h"
 #include "base/lazy_instance.h"
 #include "base/memory/singleton.h"
 #include "base/memory/weak_ptr.h"
 #include "base/message_loop/message_loop.h"
 #include "base/message_loop/message_loop_proxy.h"
+#include "base/synchronization/condition_variable.h"
 #include "base/synchronization/lock.h"
 #include "base/threading/thread.h"
 #include "base/time/time.h"
 #include "mojo/common/message_pump_mojo.h"
 #include "mojo/common/message_pump_mojo_handler.h"
 #include "mojo/common/time_helper.h"
 
 namespace mojo {
 namespace common {
 
@@ skipping 24 matching lines @@
         message_loop(NULL) {}
 
   WatcherID id;
   Handle handle;
   MojoHandleSignals handle_signals;
   base::TimeTicks deadline;
   base::Callback<void(MojoResult)> callback;
   scoped_refptr<base::MessageLoopProxy> message_loop;
 };
 
+// WaitState -------------------------------------------------------------------
+
+// Used when removing a handle. The main thread blocks until the background
+// thread successfully removes the handle from the set of handles being watched.
+// When the background thread removes the handle it sets |removed| and signals
+// |condition|.
+struct WaitState {

[darin (slow to review), 2014/07/22 17:42:56]

nit: You could also just use base::WaitableEvent here as that is essentially
just a lock+cvar+bool. It would simplify this CL.

[sky, 2014/07/22 18:20:50]

I thought there was something like that. Done!

+  WaitState() : condition(&lock), removed(false) {}
+
+  base::Lock lock;
+  base::ConditionVariable condition;
+  bool removed;
+};
+
 // WatcherBackend --------------------------------------------------------------
 
 // WatcherBackend is responsible for managing the requests and interacting with
 // MessagePumpMojo. All access (outside of creation/destruction) is done on the
 // thread WatcherThreadManager creates.
 class WatcherBackend : public MessagePumpMojoHandler {
  public:
   WatcherBackend();
   virtual ~WatcherBackend();
 
   void StartWatching(const WatchData& data);
-  void StopWatching(WatcherID watcher_id);
+
+  // Cancels a previously schedule request to start a watch. When done sets
+  // |wait_state->removed| and signals the condition.
+  void StopWatching(WatcherID watcher_id, WaitState* wait_state);
 
  private:
   typedef std::map<Handle, WatchData> HandleToWatchDataMap;
 
   // Invoked when a handle needs to be removed and notified.
   void RemoveAndNotify(const Handle& handle, MojoResult result);
 
   // Searches through |handle_to_data_| for |watcher_id|. Returns true if found
   // and sets |handle| to the Handle. Returns false if not a known id.
   bool GetMojoHandleByWatcherID(WatcherID watcher_id, Handle* handle) const;
@@ skipping 18 matching lines @@
   RemoveAndNotify(data.handle, MOJO_RESULT_CANCELLED);
 
   DCHECK_EQ(0u, handle_to_data_.count(data.handle));
 
   handle_to_data_[data.handle] = data;
   message_pump_mojo->AddHandler(this, data.handle,
                                 data.handle_signals,
                                 data.deadline);
 }
 
-void WatcherBackend::StopWatching(WatcherID watcher_id) {
+void WatcherBackend::StopWatching(WatcherID watcher_id, WaitState* wait_state) {
   // Because of the thread hop it is entirely possible to get here and not
   // have a valid handle registered for |watcher_id|.
   Handle handle;
-  if (!GetMojoHandleByWatcherID(watcher_id, &handle))
-    return;
-
-  handle_to_data_.erase(handle);
-  message_pump_mojo->RemoveHandler(handle);
+  if (GetMojoHandleByWatcherID(watcher_id, &handle)) {
+    handle_to_data_.erase(handle);
+    message_pump_mojo->RemoveHandler(handle);
+  }
+  {
+    base::AutoLock auto_lock(wait_state->lock);
+    wait_state->removed = true;
+    wait_state->condition.Signal();
+  }
 }
 
 void WatcherBackend::RemoveAndNotify(const Handle& handle,
                                      MojoResult result) {
   if (handle_to_data_.count(handle) == 0)
     return;
 
   const WatchData data(handle_to_data_[handle]);
   handle_to_data_.erase(handle);
   message_pump_mojo->RemoveHandler(handle);
@@ skipping 72 matching lines @@
     const base::Callback<void(MojoResult)>& callback) {
   WatchData data;
   data.id = watcher_id_generator_.GetNext();
   data.handle = handle;
   data.callback = callback;
   data.handle_signals = handle_signals;
   data.deadline = deadline;
   data.message_loop = base::MessageLoopProxy::current();
   DCHECK_NE(static_cast<base::MessageLoopProxy*>(NULL),
             data.message_loop.get());
-  // We outlive |thread_|, so it's safe to use Unretained() here.
+  // We own |thread_|, so it's safe to use Unretained() here.
   thread_.message_loop()->PostTask(
       FROM_HERE,
       base::Bind(&WatcherBackend::StartWatching,
                  base::Unretained(&backend_),
                  data));
   return data.id;
 }
 
 void WatcherThreadManager::StopWatching(WatcherID watcher_id) {
-  // We outlive |thread_|, so it's safe to use Unretained() here.
+  // We own |thread_|, so it's safe to use Unretained() here.
+  WaitState wait_state;
   thread_.message_loop()->PostTask(
       FROM_HERE,
       base::Bind(&WatcherBackend::StopWatching,
                  base::Unretained(&backend_),
-                 watcher_id));
+                 watcher_id,
+                 &wait_state));
+
+  // We need to block until the handle is actually removed.
+  {
+    base::AutoLock auto_lock(wait_state.lock);
+    while (!wait_state.removed)
+      wait_state.condition.Wait();
+  }
 }
 
 WatcherThreadManager::WatcherThreadManager()
     : thread_(kWatcherThreadName) {
   base::Thread::Options thread_options;
   thread_options.message_pump_factory = base::Bind(&CreateMessagePumpMojo);
   thread_.StartWithOptions(thread_options);
 }
 
 }  // namespace
@@ skipping 66 matching lines @@
 
   state_.reset(new State(this, handle, handle_signals, deadline, callback));
 }
 
 void HandleWatcher::Stop() {
   state_.reset();
 }
 
 }  // namespace common
 }  // namespace mojo