Implements HandleWatcher in terms of MessagePumpMojo

mojo/common/message_pump_mojo.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/message_pump_mojo.h"
 
 #include <algorithm>
 #include <vector>
 
 #include "base/logging.h"
@@ skipping 27 matching lines @@
 };
 
 MessagePumpMojo::MessagePumpMojo() : run_state_(NULL) {
 }
 
 MessagePumpMojo::~MessagePumpMojo() {
 }
 
 void MessagePumpMojo::AddHandler(MessagePumpMojoHandler* handler,
                                  MojoHandle handle,
-                                 MojoWaitFlags wait_flags) {
+                                 MojoWaitFlags wait_flags,
+                                 base::TimeTicks deadline) {
   DCHECK(handler);
   DCHECK_NE(MOJO_HANDLE_INVALID, handle);
+  Handler old_handler;
+  if (handlers_.count(handle))
+    old_handler = handlers_[handle];
   handlers_[handle].handler = handler;

[darin (slow to review), 2013/11/21 18:14:52]

nit: it'd probably be worth optimizing this code to not perform so many lookups
in the map. one approach would be to just construct a new Handler and then do
handlers_[handle] = new_handler;

[sky, 2013/11/22 19:39:44]

Done.

   handlers_[handle].wait_flags = wait_flags;
-
-  SignalControlPipe();
+  handlers_[handle].deadline = deadline;
+  if (old_handler.handler)
+    old_handler.handler->OnHandleError(handle, MOJO_RESULT_CANCELLED);

[darin (slow to review), 2013/11/21 18:14:52]

I'm a little bit worried about calling OnHandleError on some potentially deeply
nested call stack. Normally, calls to the Handler only happen from the run loop.
How important is it to make this callback?

[sky, 2013/11/22 19:39:44]

As discussed I made this DCHECK that there isn't an existing handler.

 }
 
 void MessagePumpMojo::RemoveHandler(MojoHandle handle) {
   handlers_.erase(handle);
-  SignalControlPipe();
 }
 
 void MessagePumpMojo::Run(Delegate* delegate) {
   RunState* old_state = run_state_;
   RunState run_state;
   // TODO: better deal with error handling.
   CHECK_NE(run_state.control_pipe.handle_0(), MOJO_HANDLE_INVALID);
   CHECK_NE(run_state.control_pipe.handle_1(), MOJO_HANDLE_INVALID);
   run_state_ = &run_state;
   bool more_work_is_plausible = true;
@@ skipping 38 matching lines @@
 
 void MessagePumpMojo::ScheduleDelayedWork(
     const base::TimeTicks& delayed_work_time) {
   if (!run_state_)
     return;
   run_state_->delayed_work_time = delayed_work_time;
   SignalControlPipe();
 }
 
 void MessagePumpMojo::DoInternalWork(bool block) {
-  MojoDeadline deadline;
-  if (block && !run_state_->delayed_work_time.is_null()) {
-    const base::TimeDelta delta = run_state_->delayed_work_time -
-        base::TimeTicks::Now();
-    deadline = std::max(static_cast<MojoDeadline>(0),
-                        static_cast<MojoDeadline>(delta.InMicroseconds()));
-  } else {
-    deadline = 0;
-  }
+  const MojoDeadline deadline = block ? GetDeadlineForWait() : 0;
   const WaitState wait_state = GetWaitState();
   const MojoResult result = MojoWaitMany(
       &wait_state.handles.front(),
       &wait_state.wait_flags.front(),
       static_cast<uint32_t>(wait_state.handles.size()),
       deadline);
   if (result == 0) {
     // Control pipe was written to.
     uint32_t num_bytes = 0;
     MojoReadMessage(run_state_->read_handle(), NULL, &num_bytes, NULL, 0,
                     MOJO_READ_MESSAGE_FLAG_MAY_DISCARD);
   } else if (result > 0) {
     const size_t index = static_cast<size_t>(result);
     DCHECK(handlers_.find(wait_state.handles[index]) != handlers_.end());
     handlers_[wait_state.handles[index]].handler->OnHandleReady(
         wait_state.handles[index]);
   } else {
     switch (result) {
       case MOJO_RESULT_INVALID_ARGUMENT:
       case MOJO_RESULT_FAILED_PRECONDITION:
         RemoveFirstInvalidHandle(wait_state);
         break;
       case MOJO_RESULT_DEADLINE_EXCEEDED:
         break;
       default:
         NOTREACHED();
     }
   }
+
+  // Notify and remove any handlers whose time has expired.
+  const base::TimeTicks now(base::TimeTicks::Now());
+  for (HandleToHandler::iterator i = handlers_.begin(); i != handlers_.end();) {
+    if (!i->second.deadline.is_null() && i->second.deadline < now) {
+      const MojoHandle handle = i->first;
+      const Handler handler(i->second);
+      handlers_.erase(i++);
+      handler.handler->OnHandleError(handle, MOJO_RESULT_DEADLINE_EXCEEDED);

[darin (slow to review), 2013/11/21 18:14:52]

OnHandleError can call AddHandler or RemoveHandler which could mess up your
iteration here. You probably want to build a separate list of handlers to
notify, and then notify them all as a separate loop.

[sky, 2013/11/22 19:39:44]

Done.

+    } else {
+      ++i;
+    }
+  }
 }
 
 void MessagePumpMojo::RemoveFirstInvalidHandle(const WaitState& wait_state) {
   // TODO(sky): deal with control pipe going bad.
   for (size_t i = 1; i < wait_state.handles.size(); ++i) {
     const MojoResult result =
         MojoWait(wait_state.handles[i], wait_state.wait_flags[i], 0);
     if (result == MOJO_RESULT_INVALID_ARGUMENT ||
         result == MOJO_RESULT_FAILED_PRECONDITION) {
+      // Remove the handle first, this way if OnHandleError() tries to remove
+      // the handle too our iterator isn't invalidated.

[darin (slow to review), 2013/11/21 18:14:52]

nit: too -> to

[sky, 2013/11/22 19:39:44]

Done.

       DCHECK(handlers_.find(wait_state.handles[i]) != handlers_.end());
       MessagePumpMojoHandler* handler =
           handlers_[wait_state.handles[i]].handler;
       handlers_.erase(wait_state.handles[i]);
       handler->OnHandleError(wait_state.handles[i], result);
       return;
     } else {
       DCHECK_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, result);
     }
   }
@@ skipping 14 matching lines @@
   wait_state.wait_flags.push_back(MOJO_WAIT_FLAG_READABLE);
 
   for (HandleToHandler::const_iterator i = handlers_.begin();
        i != handlers_.end(); ++i) {
     wait_state.handles.push_back(i->first);
     wait_state.wait_flags.push_back(i->second.wait_flags);
   }
   return wait_state;
 }
 
+MojoDeadline MessagePumpMojo::GetDeadlineForWait() const {
+  base::TimeTicks min_time = run_state_->delayed_work_time;
+  for (HandleToHandler::const_iterator i = handlers_.begin();
+       i != handlers_.end(); ++i) {
+    if (min_time.is_null() && i->second.deadline < min_time)
+      min_time = i->second.deadline;
+  }
+  return min_time.is_null() ? MOJO_DEADLINE_INDEFINITE :
+      std::max(static_cast<MojoDeadline>(0),
+               static_cast<MojoDeadline>(
+                   (min_time - base::TimeTicks::Now()).InMicroseconds()));
+}
+
 }  // namespace common
 }  // namespace mojo