src/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/logging.h"
  11 #include "base/time/time.h"
  12 #include "mojo/common/message_pump_mojo_handler.h"
  13 #include "mojo/common/time_helper.h"
  14
  15 namespace mojo {
  16 namespace common {
  17
  18 // State needed for one iteration of WaitMany. The first handle and flags
  19 // corresponds to that of the control pipe.
  20 struct MessagePumpMojo::WaitState {
  21   std::vector<Handle> handles;
  22   std::vector<MojoWaitFlags> wait_flags;
  23 };
  24
  25 struct MessagePumpMojo::RunState {
  26   RunState() : should_quit(false) {
  27     CreateMessagePipe(&read_handle, &write_handle);
  28   }
  29
  30   base::TimeTicks delayed_work_time;
  31
  32   // Used to wake up WaitForWork().
  33   ScopedMessagePipeHandle read_handle;
  34   ScopedMessagePipeHandle write_handle;
  35
  36   bool should_quit;
  37 };
  38
  39 MessagePumpMojo::MessagePumpMojo() : run_state_(NULL), next_handler_id_(0) {
  40 }
  41
  42 MessagePumpMojo::~MessagePumpMojo() {
  43 }
  44
  45 // static
  46 scoped_ptr<base::MessagePump> MessagePumpMojo::Create() {
  47   return scoped_ptr<MessagePump>(new MessagePumpMojo());
  48 }
  49
  50 void MessagePumpMojo::AddHandler(MessagePumpMojoHandler* handler,
  51                                  const Handle& handle,
  52                                  MojoWaitFlags wait_flags,
  53                                  base::TimeTicks deadline) {
  54   DCHECK(handler);
  55   DCHECK(handle.is_valid());
  56   // Assume it's an error if someone tries to reregister an existing handle.
  57   DCHECK_EQ(0u, handlers_.count(handle));
  58   Handler handler_data;
  59   handler_data.handler = handler;
  60   handler_data.wait_flags = wait_flags;
  61   handler_data.deadline = deadline;
  62   handler_data.id = next_handler_id_++;
  63   handlers_[handle] = handler_data;
  64 }
  65
  66 void MessagePumpMojo::RemoveHandler(const Handle& handle) {
  67   handlers_.erase(handle);
  68 }
  69
  70 void MessagePumpMojo::Run(Delegate* delegate) {
  71   RunState run_state;
  72   // TODO: better deal with error handling.
  73   CHECK(run_state.read_handle.is_valid());
  74   CHECK(run_state.write_handle.is_valid());
  75   RunState* old_state = NULL;
  76   {
  77     base::AutoLock auto_lock(run_state_lock_);
  78     old_state = run_state_;
  79     run_state_ = &run_state;
  80   }
  81   DoRunLoop(&run_state, delegate);
  82   {
  83     base::AutoLock auto_lock(run_state_lock_);
  84     run_state_ = old_state;
  85   }
  86 }
  87
  88 void MessagePumpMojo::Quit() {
  89   base::AutoLock auto_lock(run_state_lock_);
  90   if (run_state_)
  91     run_state_->should_quit = true;
  92 }
  93
  94 void MessagePumpMojo::ScheduleWork() {
  95   base::AutoLock auto_lock(run_state_lock_);
  96   if (run_state_)
  97     SignalControlPipe(*run_state_);
  98 }
  99
 100 void MessagePumpMojo::ScheduleDelayedWork(
 101     const base::TimeTicks& delayed_work_time) {
 102   base::AutoLock auto_lock(run_state_lock_);
 103   if (!run_state_)
 104     return;
 105   run_state_->delayed_work_time = delayed_work_time;
 106   SignalControlPipe(*run_state_);
 107 }
 108
 109 void MessagePumpMojo::DoRunLoop(RunState* run_state, Delegate* delegate) {
 110   bool more_work_is_plausible = true;
 111   for (;;) {
 112     const bool block = !more_work_is_plausible;
 113     DoInternalWork(*run_state, block);
 114
 115     // There isn't a good way to know if there are more handles ready, we assume
 116     // not.
 117     more_work_is_plausible = false;
 118
 119     if (run_state->should_quit)
 120       break;
 121
 122     more_work_is_plausible |= delegate->DoWork();
 123     if (run_state->should_quit)
 124       break;
 125
 126     more_work_is_plausible |= delegate->DoDelayedWork(
 127         &run_state->delayed_work_time);
 128     if (run_state->should_quit)
 129       break;
 130
 131     if (more_work_is_plausible)
 132       continue;
 133
 134     more_work_is_plausible = delegate->DoIdleWork();
 135     if (run_state->should_quit)
 136       break;
 137   }
 138 }
 139
 140 void MessagePumpMojo::DoInternalWork(const RunState& run_state, bool block) {
 141   const MojoDeadline deadline = block ? GetDeadlineForWait(run_state) : 0;
 142   const WaitState wait_state = GetWaitState(run_state);
 143   const MojoResult result =
 144       WaitMany(wait_state.handles, wait_state.wait_flags, deadline);
 145   if (result == 0) {
 146     // Control pipe was written to.
 147     uint32_t num_bytes = 0;
 148     ReadMessageRaw(run_state.read_handle.get(), NULL, &num_bytes, NULL, NULL,
 149                    MOJO_READ_MESSAGE_FLAG_MAY_DISCARD);
 150   } else if (result > 0) {
 151     const size_t index = static_cast<size_t>(result);
 152     DCHECK(handlers_.find(wait_state.handles[index]) != handlers_.end());
 153     handlers_[wait_state.handles[index]].handler->OnHandleReady(
 154         wait_state.handles[index]);
 155   } else {
 156     switch (result) {
 157       case MOJO_RESULT_INVALID_ARGUMENT:
 158       case MOJO_RESULT_FAILED_PRECONDITION:
 159         RemoveFirstInvalidHandle(wait_state);
 160         break;
 161       case MOJO_RESULT_DEADLINE_EXCEEDED:
 162         break;
 163       default:
 164         NOTREACHED();
 165     }
 166   }
 167
 168   // Notify and remove any handlers whose time has expired. Make a copy in case
 169   // someone tries to add/remove new handlers from notification.
 170   const HandleToHandler cloned_handlers(handlers_);
 171   const base::TimeTicks now(internal::NowTicks());
 172   for (HandleToHandler::const_iterator i = cloned_handlers.begin();
 173        i != cloned_handlers.end(); ++i) {
 174     // Since we're iterating over a clone of the handlers, verify the handler is
 175     // still valid before notifying.
 176     if (!i->second.deadline.is_null() && i->second.deadline < now &&
 177         handlers_.find(i->first) != handlers_.end() &&
 178         handlers_[i->first].id == i->second.id) {
 179       i->second.handler->OnHandleError(i->first, MOJO_RESULT_DEADLINE_EXCEEDED);
 180     }
 181   }
 182 }
 183
 184 void MessagePumpMojo::RemoveFirstInvalidHandle(const WaitState& wait_state) {
 185   // TODO(sky): deal with control pipe going bad.
 186   for (size_t i = 1; i < wait_state.handles.size(); ++i) {
 187     const MojoResult result =
 188         Wait(wait_state.handles[i], wait_state.wait_flags[i], 0);
 189     if (result == MOJO_RESULT_INVALID_ARGUMENT ||
 190         result == MOJO_RESULT_FAILED_PRECONDITION) {
 191       // Remove the handle first, this way if OnHandleError() tries to remove
 192       // the handle our iterator isn't invalidated.
 193       DCHECK(handlers_.find(wait_state.handles[i]) != handlers_.end());
 194       MessagePumpMojoHandler* handler =
 195           handlers_[wait_state.handles[i]].handler;
 196       handlers_.erase(wait_state.handles[i]);
 197       handler->OnHandleError(wait_state.handles[i], result);
 198       return;
 199     }
 200   }
 201 }
 202
 203 void MessagePumpMojo::SignalControlPipe(const RunState& run_state) {
 204   // TODO(sky): deal with error?
 205   WriteMessageRaw(run_state.write_handle.get(), NULL, 0, NULL, 0,
 206                   MOJO_WRITE_MESSAGE_FLAG_NONE);
 207 }
 208
 209 MessagePumpMojo::WaitState MessagePumpMojo::GetWaitState(
 210     const RunState& run_state) const {
 211   WaitState wait_state;
 212   wait_state.handles.push_back(run_state.read_handle.get());
 213   wait_state.wait_flags.push_back(MOJO_WAIT_FLAG_READABLE);
 214
 215   for (HandleToHandler::const_iterator i = handlers_.begin();
 216        i != handlers_.end(); ++i) {
 217     wait_state.handles.push_back(i->first);
 218     wait_state.wait_flags.push_back(i->second.wait_flags);
 219   }
 220   return wait_state;
 221 }
 222
 223 MojoDeadline MessagePumpMojo::GetDeadlineForWait(
 224     const RunState& run_state) const {
 225   base::TimeTicks min_time = run_state.delayed_work_time;
 226   for (HandleToHandler::const_iterator i = handlers_.begin();
 227        i != handlers_.end(); ++i) {
 228     if (min_time.is_null() && i->second.deadline < min_time)
 229       min_time = i->second.deadline;
 230   }
 231   return min_time.is_null() ? MOJO_DEADLINE_INDEFINITE :
 232       std::max(static_cast<MojoDeadline>(0),
 233                static_cast<MojoDeadline>(
 234                    (min_time - internal::NowTicks()).InMicroseconds()));
 235 }
 236
 237 }  // namespace common
 238 }  // namespace mojo