Performance tuning of HandleWatcher

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"
@@ skipping 53 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);
 
-  // Cancels a previously schedule request to start a watch. When done signals
-  // |event|.
-  void StopWatching(WatcherID watcher_id, base::WaitableEvent* event);
+  // Cancels a previously scheduled request to start a watch.
+  void StopWatching(WatcherID watcher_id);
 
  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,
-                                  base::WaitableEvent* event) {
+void WatcherBackend::StopWatching(WatcherID watcher_id) {
   // 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)) {
     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 40 matching lines @@
   WatcherID StartWatching(const Handle& handle,
                           MojoHandleSignals handle_signals,
                           base::TimeTicks deadline,
                           const base::Callback<void(MojoResult)>& callback);
 
   // Stops watching a handle.
   // This may be invoked on any thread.
   void StopWatching(WatcherID watcher_id);
 
  private:
+  enum RequestType {
+    REQUEST_START,
+    REQUEST_STOP,
+  };
+
+  // See description of |requests_| for details.
+  struct RequestData {
+    RequestData() : type(REQUEST_START), stop_id(0), stop_event(NULL) {}
+
+    RequestType type;
+    WatchData start_data;
+    WatcherID stop_id;
+    base::WaitableEvent* stop_event;
+  };
+
+  typedef std::vector<RequestData> Requests;
+
   friend struct DefaultSingletonTraits<WatcherThreadManager>;
+
   WatcherThreadManager();
 
+  // Schedules a request on the background thread. See |requests_| for details.
+  void AddRequest(const RequestData& data);
+
+  // Processes requests added to |requests_|. This is invoked on the backend
+  // thread.
+  void ProcessRequestsOnBackendThread();
+
   base::Thread thread_;
 
   base::AtomicSequenceNumber watcher_id_generator_;
 
   WatcherBackend backend_;
 
+  // Protects |requests_|.
+  base::Lock lock_;
+
+  // Start/Stop result in adding a RequestData to |requests_| (protected by
+  // |lock_|). When the background thread wakes up it processes the requests.
+  Requests requests_;
+
   DISALLOW_COPY_AND_ASSIGN(WatcherThreadManager);
 };
 
 WatcherThreadManager::~WatcherThreadManager() {
   thread_.Stop();
 }
 
 WatcherThreadManager* WatcherThreadManager::GetInstance() {
   return Singleton<WatcherThreadManager>::get();
 }
 
 WatcherID WatcherThreadManager::StartWatching(
     const Handle& handle,
     MojoHandleSignals handle_signals,
     base::TimeTicks deadline,
     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();
+  RequestData request_data;
+  request_data.type = REQUEST_START;
+  request_data.start_data.id = watcher_id_generator_.GetNext();
+  request_data.start_data.handle = handle;
+  request_data.start_data.callback = callback;
+  request_data.start_data.handle_signals = handle_signals;
+  request_data.start_data.deadline = deadline;
+  request_data.start_data.message_loop = base::MessageLoopProxy::current();
   DCHECK_NE(static_cast<base::MessageLoopProxy*>(NULL),
-            data.message_loop.get());
-  // 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;
+            request_data.start_data.message_loop.get());
+  AddRequest(request_data);
+  return request_data.start_data.id;
 }
 
 void WatcherThreadManager::StopWatching(WatcherID watcher_id) {
+  // Handle the case of StartWatching() followed by StopWatching() before
+  // |thread_| woke up.
+  {
+    base::AutoLock auto_lock(lock_);
+    for (Requests::iterator i = requests_.begin(); i != requests_.end(); ++i) {
+      if (i->type == REQUEST_START && i->start_data.id == watcher_id) {
+        // Watcher ids are not reused, so if we find it we can stop.
+        requests_.erase(i);
+        return;
+      }
+    }
+  }
+
   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,
-                 &event));
+  RequestData request_data;
+  request_data.type = REQUEST_STOP;
+  request_data.stop_id = watcher_id;
+  request_data.stop_event = &event;
+  AddRequest(request_data);
 
   // We need to block until the handle is actually removed.
   event.Wait();
 }
 
+void WatcherThreadManager::AddRequest(const RequestData& data) {
+  {
+    base::AutoLock auto_lock(lock_);
+    const bool was_empty = requests_.empty();
+    requests_.push_back(data);
+    if (!was_empty)
+      return;
+  }
+  // We own |thread_|, so it's safe to use Unretained() here.
+  thread_.message_loop()->PostTask(
+      FROM_HERE,
+      base::Bind(&WatcherThreadManager::ProcessRequestsOnBackendThread,
+                 base::Unretained(this)));
+}
+
+void WatcherThreadManager::ProcessRequestsOnBackendThread() {
+  DCHECK_EQ(thread_.message_loop(), base::MessageLoop::current());
+
+  Requests requests;
+  {
+    base::AutoLock auto_lock(lock_);
+    requests_.swap(requests);
+  }
+  for (size_t i = 0; i < requests.size(); ++i) {
+    if (requests[i].type == REQUEST_START) {
+      backend_.StartWatching(requests[i].start_data);
+    } else {
+      backend_.StopWatching(requests[i].stop_id);
+      requests[i].stop_event->Signal();
+    }
+  }
+}
+
 WatcherThreadManager::WatcherThreadManager()
     : thread_(kWatcherThreadName) {
   base::Thread::Options thread_options;
   thread_options.message_pump_factory = base::Bind(&CreateMessagePumpMojo);
   thread_.StartWithOptions(thread_options);
 }
 
 // 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,
         MojoDeadline deadline,
         const base::Callback<void(MojoResult)>& callback)
       : watcher_(watcher),
         callback_(callback),
+        got_ready_(false),
         weak_factory_(this) {
     base::MessageLoop::current()->AddDestructionObserver(this);
 
     watcher_id_ = WatcherThreadManager::GetInstance()->StartWatching(
         handle,
         handle_signals,
         MojoDeadlineToTimeTicks(deadline),
         base::Bind(&State::OnHandleReady, weak_factory_.GetWeakPtr()));
   }
 
   virtual ~State() {
     base::MessageLoop::current()->RemoveDestructionObserver(this);
 
-    WatcherThreadManager::GetInstance()->StopWatching(watcher_id_);
+    // If we've been notified the handle is ready (|got_ready_| is true) then
+    // the watch has been implicitly removed by
+    // WatcherThreadManager/MessagePumpMojo and we don't have to call
+    // StopWatching(). To do so would needlessly entail posting a task and
+    // blocking until the background thread services it.
+    if (!got_ready_)
+      WatcherThreadManager::GetInstance()->StopWatching(watcher_id_);
   }
 
  private:
   virtual void WillDestroyCurrentMessageLoop() OVERRIDE {
     // The current thread is exiting. Simulate a watch error.
-    OnHandleReady(MOJO_RESULT_ABORTED);
+    NotifyAndDestroy(MOJO_RESULT_ABORTED);
   }
 
   void OnHandleReady(MojoResult result) {
+    got_ready_ = true;
+    NotifyAndDestroy(result);
+  }
+
+  void NotifyAndDestroy(MojoResult result) {
     base::Callback<void(MojoResult)> callback = callback_;
     watcher_->Stop();  // Destroys |this|.
 
     callback.Run(result);
   }
 
   HandleWatcher* watcher_;
   WatcherID watcher_id_;
   base::Callback<void(MojoResult)> callback_;
 
+  // Have we been notified that the handle is ready?
+  bool got_ready_;
+
   // Used to weakly bind |this| to the WatcherThreadManager.
   base::WeakPtrFactory<State> weak_factory_;
 };
 
 // HandleWatcher ---------------------------------------------------------------
 
 HandleWatcher::HandleWatcher() {
 }
 
 HandleWatcher::~HandleWatcher() {
 }
 
 void HandleWatcher::Start(const Handle& handle,
                           MojoHandleSignals handle_signals,
                           MojoDeadline deadline,
                           const base::Callback<void(MojoResult)>& callback) {
   DCHECK(handle.is_valid());
   DCHECK_NE(MOJO_HANDLE_SIGNAL_NONE, handle_signals);
 
   state_.reset(new State(this, handle, handle_signals, deadline, callback));
 }
 
 void HandleWatcher::Stop() {
   state_.reset();
 }
 
 }  // namespace common
 }  // namespace mojo