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/lock.h"
+#include "base/synchronization/waitable_event.h"
 #include "base/threading/thread.h"
+#include "base/threading/thread_restrictions.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 {
 
 typedef int WatcherID;
 
@@ skipping 33 matching lines @@
 
 // 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 signals
+  // |event|.
+  void StopWatching(WatcherID watcher_id, base::WaitableEvent* event);
 
  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,
+                                  base::WaitableEvent* event) {
   // 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);
+  }
+  event->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 17 matching lines @@
 }
 
 void WatcherBackend::OnHandleError(const Handle& handle, MojoResult result) {
   RemoveAndNotify(handle, result);
 }
 
 // WatcherThreadManager --------------------------------------------------------
 
 // WatcherThreadManager manages the background thread that listens for handles
 // to be ready. All requests are handled by WatcherBackend.
+}  // namespace
+
 class WatcherThreadManager {
  public:
   ~WatcherThreadManager();
 
   // Returns the shared instance.
   static WatcherThreadManager* GetInstance();
 
   // Starts watching the requested handle. Returns a unique ID that is used to
   // stop watching the handle. When the handle is ready |callback| is notified
   // on the thread StartWatching() was invoked on.
@@ skipping 35 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.
+  base::ThreadRestrictions::ScopedAllowWait allow_wait;
+  base::WaitableEvent event(true, false);
+  // We own |thread_|, so it's safe to use Unretained() here.
   thread_.message_loop()->PostTask(
       FROM_HERE,
       base::Bind(&WatcherBackend::StopWatching,
                  base::Unretained(&backend_),
-                 watcher_id));
+                 watcher_id,
+                 &event));
+
+  // We need to block until the handle is actually removed.
+  event.Wait();
 }
 
 WatcherThreadManager::WatcherThreadManager()
     : thread_(kWatcherThreadName) {
   base::Thread::Options thread_options;
   thread_options.message_pump_factory = base::Bind(&CreateMessagePumpMojo);
   thread_.StartWithOptions(thread_options);
 }
 
-}  // namespace
-
 // HandleWatcher::State --------------------------------------------------------
 
 // 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,
         MojoHandleSignals handle_signals,
@@ skipping 55 matching lines @@
 
   state_.reset(new State(this, handle, handle_signals, deadline, callback));
 }
 
 void HandleWatcher::Stop() {
   state_.reset();
 }
 
 }  // namespace common
 }  // namespace mojo