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);
-  handlers_[handle].handler = handler;
-  handlers_[handle].wait_flags = wait_flags;
-
-  SignalControlPipe();
+  // Assume it's an error if someone tries to reregister an existing handle.
+  DCHECK_EQ(0u, handlers_.count(handle));
+  Handler handler_data;
+  handler_data.handler = handler;
+  handler_data.wait_flags = wait_flags;
+  handler_data.deadline = deadline;
+  handlers_[handle] = handler_data;
 }
 
 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. Make a copy in case
+  // someone tries to add/remove new handlers from notification.
+  const HandleToHandler cloned_handlers(handlers_);
+  const base::TimeTicks now(base::TimeTicks::Now());
+  for (HandleToHandler::const_iterator i = cloned_handlers.begin();
+       i != cloned_handlers.end(); ++i) {
+    // Since we're iterating over a clone of the handlers, verify the handler is
+    // still valid before notifying.
+    if (!i->second.deadline.is_null() && i->second.deadline < now &&
+        handlers_.find(i->first) != handlers_.end() &&

[darin (slow to review), 2013/11/22 23:03:32]

I think there could be subtle bugs here with the handler being removed and
re-added with a different deadline not equal to i->second.deadline.

Too bad we don't have a map equivalent to base/observer_list.h.
base/id_map.{h,cc} implements mutation safe iteration.

Maybe you could have a map to entries in the ObserverList? That way when you
need fast lookup, you go through the map, and when you need to iterate over all
entries, you just use the ObserverList.

[sky, 2013/11/22 23:25:25]

I went with an id for equality.

+        i->second.handler == handlers_[i->first].handler) {
+      i->second.handler->OnHandleError(i->first, MOJO_RESULT_DEADLINE_EXCEEDED);
+    }
+  }
 }
 
 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 our iterator isn't invalidated.
       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