mojo/common/message_pump_mojo.cc

   1 // Copyright 2013 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "mojo/common/message_pump_mojo.h"
   6
   7 #include <algorithm>
   8 #include <vector>
   9
  10 #include "base/debug/alias.h"
  11 #include "base/logging.h"
  12 #include "base/time/time.h"
  13 #include "mojo/common/message_pump_mojo_handler.h"
  14 #include "mojo/common/time_helper.h"
  15
  16 namespace mojo {
  17 namespace common {
  18 namespace {
  19
  20 MojoDeadline TimeTicksToMojoDeadline(base::TimeTicks time_ticks,
  21                                      base::TimeTicks now) {
  22   // The is_null() check matches that of HandleWatcher as well as how
  23   // |delayed_work_time| is used.
  24   if (time_ticks.is_null())
  25     return MOJO_DEADLINE_INDEFINITE;
  26   const int64_t delta = (time_ticks - now).InMicroseconds();
  27   return delta < 0 ? static_cast<MojoDeadline>(0) :
  28                      static_cast<MojoDeadline>(delta);
  29 }
  30
  31 }  // namespace
  32
  33 // State needed for one iteration of WaitMany. The first handle and flags
  34 // corresponds to that of the control pipe.
  35 struct MessagePumpMojo::WaitState {
  36   std::vector<Handle> handles;
  37   std::vector<MojoHandleSignals> wait_signals;
  38 };
  39
  40 struct MessagePumpMojo::RunState {
  41   RunState() : should_quit(false) {
  42     CreateMessagePipe(NULL, &read_handle, &write_handle);
  43   }
  44
  45   base::TimeTicks delayed_work_time;
  46
  47   // Used to wake up WaitForWork().
  48   ScopedMessagePipeHandle read_handle;
  49   ScopedMessagePipeHandle write_handle;
  50
  51   bool should_quit;
  52 };
  53
  54 MessagePumpMojo::MessagePumpMojo() : run_state_(NULL), next_handler_id_(0) {
  55 }
  56
  57 MessagePumpMojo::~MessagePumpMojo() {
  58 }
  59
  60 // static
  61 scoped_ptr<base::MessagePump> MessagePumpMojo::Create() {
  62   return scoped_ptr<MessagePump>(new MessagePumpMojo());
  63 }
  64
  65 void MessagePumpMojo::AddHandler(MessagePumpMojoHandler* handler,
  66                                  const Handle& handle,
  67                                  MojoHandleSignals wait_signals,
  68                                  base::TimeTicks deadline) {
  69   CHECK(handler);
  70   DCHECK(handle.is_valid());
  71   // Assume it's an error if someone tries to reregister an existing handle.
  72   CHECK_EQ(0u, handlers_.count(handle));
  73   Handler handler_data;
  74   handler_data.handler = handler;
  75   handler_data.wait_signals = wait_signals;
  76   handler_data.deadline = deadline;
  77   handler_data.id = next_handler_id_++;
  78   handlers_[handle] = handler_data;
  79 }
  80
  81 void MessagePumpMojo::RemoveHandler(const Handle& handle) {
  82   handlers_.erase(handle);
  83 }
  84
  85 void MessagePumpMojo::Run(Delegate* delegate) {
  86   RunState run_state;
  87   // TODO: better deal with error handling.
  88   CHECK(run_state.read_handle.is_valid());
  89   CHECK(run_state.write_handle.is_valid());
  90   RunState* old_state = NULL;
  91   {
  92     base::AutoLock auto_lock(run_state_lock_);
  93     old_state = run_state_;
  94     run_state_ = &run_state;
  95   }
  96   DoRunLoop(&run_state, delegate);
  97   {
  98     base::AutoLock auto_lock(run_state_lock_);
  99     run_state_ = old_state;
 100   }
 101 }
 102
 103 void MessagePumpMojo::Quit() {
 104   base::AutoLock auto_lock(run_state_lock_);
 105   if (run_state_)
 106     run_state_->should_quit = true;
 107 }
 108
 109 void MessagePumpMojo::ScheduleWork() {
 110   base::AutoLock auto_lock(run_state_lock_);
 111   if (run_state_)
 112     SignalControlPipe(*run_state_);
 113 }
 114
 115 void MessagePumpMojo::ScheduleDelayedWork(
 116     const base::TimeTicks& delayed_work_time) {
 117   base::AutoLock auto_lock(run_state_lock_);
 118   if (!run_state_)
 119     return;
 120   run_state_->delayed_work_time = delayed_work_time;
 121 }
 122
 123 void MessagePumpMojo::DoRunLoop(RunState* run_state, Delegate* delegate) {
 124   bool more_work_is_plausible = true;
 125   for (;;) {
 126     const bool block = !more_work_is_plausible;
 127     DoInternalWork(*run_state, block);
 128
 129     // There isn't a good way to know if there are more handles ready, we assume
 130     // not.
 131     more_work_is_plausible = false;
 132
 133     if (run_state->should_quit)
 134       break;
 135
 136     more_work_is_plausible |= delegate->DoWork();
 137     if (run_state->should_quit)
 138       break;
 139
 140     more_work_is_plausible |= delegate->DoDelayedWork(
 141         &run_state->delayed_work_time);
 142     if (run_state->should_quit)
 143       break;
 144
 145     if (more_work_is_plausible)
 146       continue;
 147
 148     more_work_is_plausible = delegate->DoIdleWork();
 149     if (run_state->should_quit)
 150       break;
 151   }
 152 }
 153
 154 void MessagePumpMojo::DoInternalWork(const RunState& run_state, bool block) {
 155   const MojoDeadline deadline = block ? GetDeadlineForWait(run_state) : 0;
 156   const WaitState wait_state = GetWaitState(run_state);
 157   const MojoResult result =
 158       WaitMany(wait_state.handles, wait_state.wait_signals, deadline);
 159   if (result == 0) {
 160     // Control pipe was written to.
 161     uint32_t num_bytes = 0;
 162     ReadMessageRaw(run_state.read_handle.get(), NULL, &num_bytes, NULL, NULL,
 163                    MOJO_READ_MESSAGE_FLAG_MAY_DISCARD);
 164   } else if (result > 0) {
 165     const size_t index = static_cast<size_t>(result);
 166     DCHECK(handlers_.find(wait_state.handles[index]) != handlers_.end());
 167     handlers_[wait_state.handles[index]].handler->OnHandleReady(
 168         wait_state.handles[index]);
 169   } else {
 170     switch (result) {
 171       case MOJO_RESULT_CANCELLED:
 172       case MOJO_RESULT_FAILED_PRECONDITION:
 173         RemoveFirstInvalidHandle(wait_state);
 174         break;
 175       case MOJO_RESULT_DEADLINE_EXCEEDED:
 176         break;
 177       default:
 178         base::debug::Alias(&result);
 179         // Unexpected result is likely fatal, crash so we can determine cause.
 180         CHECK(false);
 181     }
 182   }
 183
 184   // Notify and remove any handlers whose time has expired. Make a copy in case
 185   // someone tries to add/remove new handlers from notification.
 186   const HandleToHandler cloned_handlers(handlers_);
 187   const base::TimeTicks now(internal::NowTicks());
 188   for (HandleToHandler::const_iterator i = cloned_handlers.begin();
 189        i != cloned_handlers.end(); ++i) {
 190     // Since we're iterating over a clone of the handlers, verify the handler is
 191     // still valid before notifying.
 192     if (!i->second.deadline.is_null() && i->second.deadline < now &&
 193         handlers_.find(i->first) != handlers_.end() &&
 194         handlers_[i->first].id == i->second.id) {
 195       i->second.handler->OnHandleError(i->first, MOJO_RESULT_DEADLINE_EXCEEDED);
 196     }
 197   }
 198 }
 199
 200 void MessagePumpMojo::RemoveFirstInvalidHandle(const WaitState& wait_state) {
 201   // TODO(sky): deal with control pipe going bad.
 202   for (size_t i = 0; i < wait_state.handles.size(); ++i) {
 203     const MojoResult result =
 204         Wait(wait_state.handles[i], wait_state.wait_signals[i], 0);
 205     if (result == MOJO_RESULT_INVALID_ARGUMENT) {
 206       // We should never have an invalid argument. If we do it indicates
 207       // RemoveHandler() was not invoked and is likely to cause problems else
 208       // where in the stack if we ignore it.
 209       CHECK(false);
 210     } else if (result == MOJO_RESULT_FAILED_PRECONDITION ||
 211                result == MOJO_RESULT_CANCELLED) {
 212       CHECK_NE(i, 0u);  // Indicates the control pipe went bad.
 213
 214       // Remove the handle first, this way if OnHandleError() tries to remove
 215       // the handle our iterator isn't invalidated.
 216       CHECK(handlers_.find(wait_state.handles[i]) != handlers_.end());
 217       MessagePumpMojoHandler* handler =
 218           handlers_[wait_state.handles[i]].handler;
 219       handlers_.erase(wait_state.handles[i]);
 220       handler->OnHandleError(wait_state.handles[i], result);
 221       return;
 222     }
 223   }
 224 }
 225
 226 void MessagePumpMojo::SignalControlPipe(const RunState& run_state) {
 227   const MojoResult result =
 228       WriteMessageRaw(run_state.write_handle.get(), NULL, 0, NULL, 0,
 229                       MOJO_WRITE_MESSAGE_FLAG_NONE);
 230   // If we can't write we likely won't wake up the thread and there is a strong
 231   // chance we'll deadlock.
 232   CHECK_EQ(MOJO_RESULT_OK, result);
 233 }
 234
 235 MessagePumpMojo::WaitState MessagePumpMojo::GetWaitState(
 236     const RunState& run_state) const {
 237   WaitState wait_state;
 238   wait_state.handles.push_back(run_state.read_handle.get());
 239   wait_state.wait_signals.push_back(MOJO_HANDLE_SIGNAL_READABLE);
 240
 241   for (HandleToHandler::const_iterator i = handlers_.begin();
 242        i != handlers_.end(); ++i) {
 243     wait_state.handles.push_back(i->first);
 244     wait_state.wait_signals.push_back(i->second.wait_signals);
 245   }
 246   return wait_state;
 247 }
 248
 249 MojoDeadline MessagePumpMojo::GetDeadlineForWait(
 250     const RunState& run_state) const {
 251   const base::TimeTicks now(internal::NowTicks());
 252   MojoDeadline deadline = TimeTicksToMojoDeadline(run_state.delayed_work_time,
 253                                                   now);
 254   for (HandleToHandler::const_iterator i = handlers_.begin();
 255        i != handlers_.end(); ++i) {
 256     deadline = std::min(
 257         TimeTicksToMojoDeadline(i->second.deadline, now), deadline);
 258   }
 259   return deadline;
 260 }
 261
 262 }  // namespace common
 263 }  // namespace mojo