ipc/mojo/ipc_message_pipe_reader.cc

   1 // Copyright 2014 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 "ipc/mojo/ipc_message_pipe_reader.h"
   6
   7 #include "base/bind.h"
   8 #include "base/bind_helpers.h"
   9 #include "base/location.h"
  10 #include "base/logging.h"
  11 #include "base/message_loop/message_loop_proxy.h"
  12 #include "ipc/mojo/async_handle_waiter.h"
  13 #include "ipc/mojo/ipc_channel_mojo.h"
  14
  15 namespace IPC {
  16 namespace internal {
  17
  18 MessagePipeReader::MessagePipeReader(mojo::ScopedMessagePipeHandle handle,
  19                                      MessagePipeReader::Delegate* delegate)
  20     : pipe_(handle.Pass()),
  21       delegate_(delegate),
  22       async_waiter_(
  23           new AsyncHandleWaiter(base::Bind(&MessagePipeReader::PipeIsReady,
  24                                            base::Unretained(this)))) {
  25 }
  26
  27 MessagePipeReader::~MessagePipeReader() {
  28   CHECK(!IsValid());
  29 }
  30
  31 void MessagePipeReader::Close() {
  32   async_waiter_.reset();
  33   pipe_.reset();
  34   OnPipeClosed();
  35 }
  36
  37 void MessagePipeReader::CloseWithError(MojoResult error) {
  38   OnPipeError(error);
  39   Close();
  40 }
  41
  42 bool MessagePipeReader::Send(scoped_ptr<Message> message) {
  43   DCHECK(IsValid());
  44
  45   message->TraceMessageBegin();
  46   std::vector<MojoHandle> handles;
  47   MojoResult result = MOJO_RESULT_OK;
  48   result = ChannelMojo::ReadFromMessageAttachmentSet(message.get(), &handles);
  49   if (result == MOJO_RESULT_OK) {
  50     result = MojoWriteMessage(handle(),
  51                               message->data(),
  52                               static_cast<uint32>(message->size()),
  53                               handles.empty() ? nullptr : &handles[0],
  54                               static_cast<uint32>(handles.size()),
  55                               MOJO_WRITE_MESSAGE_FLAG_NONE);
  56   }
  57
  58   if (result != MOJO_RESULT_OK) {
  59     std::for_each(handles.begin(), handles.end(), &MojoClose);
  60     CloseWithError(result);
  61     return false;
  62   }
  63
  64   return true;
  65 }
  66
  67 void MessagePipeReader::OnMessageReceived() {
  68   Message message(data_buffer().empty() ? "" : &data_buffer()[0],
  69                   static_cast<uint32>(data_buffer().size()));
  70
  71   std::vector<MojoHandle> handle_buffer;
  72   TakeHandleBuffer(&handle_buffer);
  73   MojoResult write_result =
  74       ChannelMojo::WriteToMessageAttachmentSet(handle_buffer, &message);
  75   if (write_result != MOJO_RESULT_OK) {
  76     CloseWithError(write_result);
  77     return;
  78   }
  79
  80   message.TraceMessageEnd();
  81   delegate_->OnMessageReceived(message);
  82 }
  83
  84 void MessagePipeReader::OnPipeClosed() {
  85   if (!delegate_)
  86     return;
  87   delegate_->OnPipeClosed(this);
  88   delegate_ = nullptr;
  89 }
  90
  91 void MessagePipeReader::OnPipeError(MojoResult error) {
  92   if (!delegate_)
  93     return;
  94   delegate_->OnPipeError(this);
  95 }
  96
  97 MojoResult MessagePipeReader::ReadMessageBytes() {
  98   DCHECK(handle_buffer_.empty());
  99
 100   uint32_t num_bytes = static_cast<uint32_t>(data_buffer_.size());
 101   uint32_t num_handles = 0;
 102   MojoResult result = MojoReadMessage(pipe_.get().value(),
 103                                       num_bytes ? &data_buffer_[0] : nullptr,
 104                                       &num_bytes,
 105                                       nullptr,
 106                                       &num_handles,
 107                                       MOJO_READ_MESSAGE_FLAG_NONE);
 108   data_buffer_.resize(num_bytes);
 109   handle_buffer_.resize(num_handles);
 110   if (result == MOJO_RESULT_RESOURCE_EXHAUSTED) {
 111     // MOJO_RESULT_RESOURCE_EXHAUSTED was asking the caller that
 112     // it needs more bufer. So we re-read it with resized buffers.
 113     result = MojoReadMessage(pipe_.get().value(),
 114                              num_bytes ? &data_buffer_[0] : nullptr,
 115                              &num_bytes,
 116                              num_handles ? &handle_buffer_[0] : nullptr,
 117                              &num_handles,
 118                              MOJO_READ_MESSAGE_FLAG_NONE);
 119   }
 120
 121   DCHECK(0 == num_bytes || data_buffer_.size() == num_bytes);
 122   DCHECK(0 == num_handles || handle_buffer_.size() == num_handles);
 123   return result;
 124 }
 125
 126 void MessagePipeReader::ReadAvailableMessages() {
 127   while (pipe_.is_valid()) {
 128     MojoResult read_result = ReadMessageBytes();
 129     if (read_result == MOJO_RESULT_SHOULD_WAIT)
 130       break;
 131     if (read_result != MOJO_RESULT_OK) {
 132       // FAILED_PRECONDITION means that all the received messages
 133       // got consumed and the peer is already closed.
 134       if (read_result != MOJO_RESULT_FAILED_PRECONDITION) {
 135         DLOG(WARNING)
 136             << "Pipe got error from ReadMessage(). Closing: " << read_result;
 137         OnPipeError(read_result);
 138       }
 139
 140       Close();
 141       break;
 142     }
 143
 144     OnMessageReceived();
 145   }
 146
 147 }
 148
 149 void MessagePipeReader::ReadMessagesThenWait() {
 150   while (true) {
 151     ReadAvailableMessages();
 152     if (!pipe_.is_valid())
 153       break;
 154     // |Wait()| is safe to call only after all messages are read.
 155     // If can fail with |MOJO_RESULT_ALREADY_EXISTS| otherwise.
 156     // Also, we don't use MOJO_HANDLE_SIGNAL_WRITABLE here, expecting buffer in
 157     // MessagePipe.
 158     MojoResult result =
 159         async_waiter_->Wait(pipe_.get().value(), MOJO_HANDLE_SIGNAL_READABLE);
 160     // If the result is |MOJO_RESULT_ALREADY_EXISTS|, there could be messages
 161     // that have been arrived after the last |ReadAvailableMessages()|.
 162     // We have to consume then and retry in that case.
 163     if (result != MOJO_RESULT_ALREADY_EXISTS) {
 164       if (result != MOJO_RESULT_OK) {
 165         LOG(ERROR) << "Failed to wait on the pipe. Result is " << result;
 166         OnPipeError(result);
 167         Close();
 168       }
 169
 170       break;
 171     }
 172   }
 173 }
 174
 175 void MessagePipeReader::PipeIsReady(MojoResult wait_result) {
 176   if (wait_result != MOJO_RESULT_OK) {
 177     if (wait_result != MOJO_RESULT_ABORTED) {
 178       // FAILED_PRECONDITION happens every time the peer is dead so
 179       // it isn't worth polluting the log message.
 180       LOG_IF(WARNING, wait_result != MOJO_RESULT_FAILED_PRECONDITION)
 181           << "Pipe got error from the waiter. Closing: " << wait_result;
 182       OnPipeError(wait_result);
 183     }
 184
 185     Close();
 186     return;
 187   }
 188
 189   ReadMessagesThenWait();
 190 }
 191
 192 void MessagePipeReader::DelayedDeleter::operator()(
 193     MessagePipeReader* ptr) const {
 194   ptr->Close();
 195   base::MessageLoopProxy::current()->PostTask(
 196       FROM_HERE, base::Bind(&DeleteNow, ptr));
 197 }
 198
 199 }  // namespace internal
 200 }  // namespace IPC