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 schedule request to start a watch.

[darin (slow to review), 2014/08/21 23:55:12]

nit: schedule -> scheduled

[sky, 2014/08/22 00:03:51]

Done.

+  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 (!got_ready_)

[darin (slow to review), 2014/08/21 23:55:12]

isn't there a race condition here? what if StartWatching got far enough into
doing its work but not quite to the point of running the callback?

also, it seems like it would be cleaner if every call to StartWatching had to be
paired with a call to StopWatching, rather than trying to observe cases where
StopWatching may not be necessary.

[sky, 2014/08/22 00:03:51]

Then got_ready_ would be false, right?
That's effectively what got_ready_ is doing. I don't think it makes sense to
track that in WatcherBackend as it would mean an additional thread hop is
necessary every time State is destroyed. I think it's also nice that
WatcherBackend (like HandleWatcher) can assume once it's notified the caller
that it removes internal state.

+      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