src/mojo/public/cpp/bindings/lib/connector.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/public/cpp/bindings/lib/connector.h"
   6
   7 #include "mojo/public/cpp/bindings/error_handler.h"
   8 #include "mojo/public/cpp/environment/logging.h"
   9
  10 namespace mojo {
  11 namespace internal {
  12
  13 // ----------------------------------------------------------------------------
  14
  15 Connector::Connector(ScopedMessagePipeHandle message_pipe,
  16                      const MojoAsyncWaiter* waiter)
  17     : error_handler_(nullptr),
  18       waiter_(waiter),
  19       message_pipe_(message_pipe.Pass()),
  20       incoming_receiver_(nullptr),
  21       async_wait_id_(0),
  22       error_(false),
  23       drop_writes_(false),
  24       enforce_errors_from_incoming_receiver_(true),
  25       destroyed_flag_(nullptr) {
  26   // Even though we don't have an incoming receiver, we still want to monitor
  27   // the message pipe to know if is closed or encounters an error.
  28   WaitToReadMore();
  29 }
  30
  31 Connector::~Connector() {
  32   if (destroyed_flag_)
  33     *destroyed_flag_ = true;
  34
  35   CancelWait();
  36 }
  37
  38 void Connector::CloseMessagePipe() {
  39   CancelWait();
  40   Close(message_pipe_.Pass());
  41 }
  42
  43 ScopedMessagePipeHandle Connector::PassMessagePipe() {
  44   CancelWait();
  45   return message_pipe_.Pass();
  46 }
  47
  48 bool Connector::WaitForIncomingMessage() {
  49   if (error_)
  50     return false;
  51
  52   MojoResult rv = Wait(message_pipe_.get(),
  53                        MOJO_HANDLE_SIGNAL_READABLE,
  54                        MOJO_DEADLINE_INDEFINITE);
  55   if (rv != MOJO_RESULT_OK) {
  56     NotifyError();
  57     return false;
  58   }
  59   mojo_ignore_result(ReadSingleMessage(&rv));
  60   return (rv == MOJO_RESULT_OK);
  61 }
  62
  63 bool Connector::Accept(Message* message) {
  64   MOJO_CHECK(message_pipe_.is_valid());
  65
  66   if (error_)
  67     return false;
  68
  69   if (drop_writes_)
  70     return true;
  71
  72   MojoResult rv =
  73       WriteMessageRaw(message_pipe_.get(),
  74                       message->data(),
  75                       message->data_num_bytes(),
  76                       message->mutable_handles()->empty()
  77                           ? nullptr
  78                           : reinterpret_cast<const MojoHandle*>(
  79                                 &message->mutable_handles()->front()),
  80                       static_cast<uint32_t>(message->mutable_handles()->size()),
  81                       MOJO_WRITE_MESSAGE_FLAG_NONE);
  82
  83   switch (rv) {
  84     case MOJO_RESULT_OK:
  85       // The handles were successfully transferred, so we don't need the message
  86       // to track their lifetime any longer.
  87       message->mutable_handles()->clear();
  88       break;
  89     case MOJO_RESULT_FAILED_PRECONDITION:
  90       // There's no point in continuing to write to this pipe since the other
  91       // end is gone. Avoid writing any future messages. Hide write failures
  92       // from the caller since we'd like them to continue consuming any backlog
  93       // of incoming messages before regarding the message pipe as closed.
  94       drop_writes_ = true;
  95       break;
  96     case MOJO_RESULT_BUSY:
  97       // We'd get a "busy" result if one of the message's handles is:
  98       //   - |message_pipe_|'s own handle;
  99       //   - simultaneously being used on another thread; or
 100       //   - in a "busy" state that prohibits it from being transferred (e.g.,
 101       //     a data pipe handle in the middle of a two-phase read/write,
 102       //     regardless of which thread that two-phase read/write is happening
 103       //     on).
 104       // TODO(vtl): I wonder if this should be a |MOJO_DCHECK()|. (But, until
 105       // crbug.com/389666, etc. are resolved, this will make tests fail quickly
 106       // rather than hanging.)
 107       MOJO_CHECK(false) << "Race condition or other bug detected";
 108       return false;
 109     default:
 110       // This particular write was rejected, presumably because of bad input.
 111       // The pipe is not necessarily in a bad state.
 112       return false;
 113   }
 114   return true;
 115 }
 116
 117 // static
 118 void Connector::CallOnHandleReady(void* closure, MojoResult result) {
 119   Connector* self = static_cast<Connector*>(closure);
 120   self->OnHandleReady(result);
 121 }
 122
 123 void Connector::OnHandleReady(MojoResult result) {
 124   MOJO_CHECK(async_wait_id_ != 0);
 125   async_wait_id_ = 0;
 126   if (result != MOJO_RESULT_OK) {
 127     NotifyError();
 128     return;
 129   }
 130   ReadAllAvailableMessages();
 131   // At this point, this object might have been deleted. Return.
 132 }
 133
 134 void Connector::WaitToReadMore() {
 135   MOJO_CHECK(!async_wait_id_);
 136   async_wait_id_ = waiter_->AsyncWait(message_pipe_.get().value(),
 137                                       MOJO_HANDLE_SIGNAL_READABLE,
 138                                       MOJO_DEADLINE_INDEFINITE,
 139                                       &Connector::CallOnHandleReady,
 140                                       this);
 141 }
 142
 143 bool Connector::ReadSingleMessage(MojoResult* read_result) {
 144   bool receiver_result = false;
 145
 146   // Detect if |this| was destroyed during message dispatch. Allow for the
 147   // possibility of re-entering ReadMore() through message dispatch.
 148   bool was_destroyed_during_dispatch = false;
 149   bool* previous_destroyed_flag = destroyed_flag_;
 150   destroyed_flag_ = &was_destroyed_during_dispatch;
 151
 152   MojoResult rv = ReadAndDispatchMessage(
 153       message_pipe_.get(), incoming_receiver_, &receiver_result);
 154   if (read_result)
 155     *read_result = rv;
 156
 157   if (was_destroyed_during_dispatch) {
 158     if (previous_destroyed_flag)
 159       *previous_destroyed_flag = true;  // Propagate flag.
 160     return false;
 161   }
 162   destroyed_flag_ = previous_destroyed_flag;
 163
 164   if (rv == MOJO_RESULT_SHOULD_WAIT)
 165     return true;
 166
 167   if (rv != MOJO_RESULT_OK ||
 168       (enforce_errors_from_incoming_receiver_ && !receiver_result)) {
 169     NotifyError();
 170     return false;
 171   }
 172   return true;
 173 }
 174
 175 void Connector::ReadAllAvailableMessages() {
 176   while (!error_) {
 177     MojoResult rv;
 178
 179     // Return immediately if |this| was destroyed. Do not touch any members!
 180     if (!ReadSingleMessage(&rv))
 181       return;
 182
 183     if (rv == MOJO_RESULT_SHOULD_WAIT) {
 184       WaitToReadMore();
 185       break;
 186     }
 187   }
 188 }
 189
 190 void Connector::CancelWait() {
 191   if (!async_wait_id_)
 192     return;
 193
 194   waiter_->CancelWait(async_wait_id_);
 195   async_wait_id_ = 0;
 196 }
 197
 198 void Connector::NotifyError() {
 199   error_ = true;
 200   CancelWait();
 201   if (error_handler_)
 202     error_handler_->OnConnectionError();
 203 }
 204
 205 }  // namespace internal
 206 }  // namespace mojo