ipc/ipc_channel_reader.cc

   1 // Copyright (c) 2012 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/ipc_channel_reader.h"
   6
   7 #include <algorithm>
   8
   9 #include "ipc/ipc_listener.h"
  10 #include "ipc/ipc_logging.h"
  11 #include "ipc/ipc_message.h"
  12 #include "ipc/ipc_message_attachment_set.h"
  13 #include "ipc/ipc_message_macros.h"
  14
  15 namespace IPC {
  16 namespace internal {
  17
  18 ChannelReader::ChannelReader(Listener* listener) : listener_(listener) {
  19   memset(input_buf_, 0, sizeof(input_buf_));
  20 }
  21
  22 ChannelReader::~ChannelReader() {
  23   if (!blocked_ids_.empty())
  24     StopObservingAttachmentBroker();
  25 }
  26
  27 ChannelReader::DispatchState ChannelReader::ProcessIncomingMessages() {
  28   while (true) {
  29     int bytes_read = 0;
  30     ReadState read_state = ReadData(input_buf_, Channel::kReadBufferSize,
  31                                     &bytes_read);
  32     if (read_state == READ_FAILED)
  33       return DISPATCH_ERROR;
  34     if (read_state == READ_PENDING)
  35       return DISPATCH_FINISHED;
  36
  37     DCHECK(bytes_read > 0);
  38     if (!TranslateInputData(input_buf_, bytes_read))
  39       return DISPATCH_ERROR;
  40
  41     DispatchState state = DispatchMessages();
  42     if (state != DISPATCH_FINISHED)
  43       return state;
  44   }
  45 }
  46
  47 ChannelReader::DispatchState ChannelReader::AsyncReadComplete(int bytes_read) {
  48   if (!TranslateInputData(input_buf_, bytes_read))
  49     return DISPATCH_ERROR;
  50
  51   return DispatchMessages();
  52 }
  53
  54 bool ChannelReader::IsInternalMessage(const Message& m) {
  55   return m.routing_id() == MSG_ROUTING_NONE &&
  56       m.type() >= Channel::CLOSE_FD_MESSAGE_TYPE &&
  57       m.type() <= Channel::HELLO_MESSAGE_TYPE;
  58 }
  59
  60 bool ChannelReader::IsHelloMessage(const Message& m) {
  61   return m.routing_id() == MSG_ROUTING_NONE &&
  62       m.type() == Channel::HELLO_MESSAGE_TYPE;
  63 }
  64
  65 bool ChannelReader::TranslateInputData(const char* input_data,
  66                                        int input_data_len) {
  67   const char* p;
  68   const char* end;
  69
  70   // Possibly combine with the overflow buffer to make a larger buffer.
  71   if (input_overflow_buf_.empty()) {
  72     p = input_data;
  73     end = input_data + input_data_len;
  74   } else {
  75     if (input_overflow_buf_.size() + input_data_len >
  76         Channel::kMaximumMessageSize) {
  77       input_overflow_buf_.clear();
  78       LOG(ERROR) << "IPC message is too big";
  79       return false;
  80     }
  81     input_overflow_buf_.append(input_data, input_data_len);
  82     p = input_overflow_buf_.data();
  83     end = p + input_overflow_buf_.size();
  84   }
  85
  86   // Dispatch all complete messages in the data buffer.
  87   while (p < end) {
  88     const char* message_tail = Message::FindNext(p, end);
  89     if (message_tail) {
  90       int len = static_cast<int>(message_tail - p);
  91
  92       Message translated_message(p, len);
  93       if (!GetNonBrokeredAttachments(&translated_message))
  94         return false;
  95
  96       // If there are no queued messages, attempt to immediately dispatch the
  97       // newly translated message.
  98       if (queued_messages_.empty()) {
  99         DCHECK(blocked_ids_.empty());
 100         AttachmentIdSet blocked_ids =
 101             GetBrokeredAttachments(&translated_message);
 102
 103         if (blocked_ids.empty()) {
 104           // Dispatch the message and continue the loop.
 105           DispatchMessage(&translated_message);
 106           p = message_tail;
 107           continue;
 108         }
 109
 110         blocked_ids_.swap(blocked_ids);
 111         StartObservingAttachmentBroker();
 112       }
 113
 114       // Make a deep copy of |translated_message| to add to the queue.
 115       scoped_ptr<Message> m(new Message(translated_message));
 116       queued_messages_.push_back(m.release());
 117       p = message_tail;
 118     } else {
 119       // Last message is partial.
 120       break;
 121     }
 122   }
 123
 124   // Save any partial data in the overflow buffer.
 125   input_overflow_buf_.assign(p, end - p);
 126
 127   if (input_overflow_buf_.empty() && !DidEmptyInputBuffers())
 128     return false;
 129   return true;
 130 }
 131
 132 ChannelReader::DispatchState ChannelReader::DispatchMessages() {
 133   while (!queued_messages_.empty()) {
 134     if (!blocked_ids_.empty())
 135       return DISPATCH_WAITING_ON_BROKER;
 136
 137     Message* m = queued_messages_.front();
 138
 139     AttachmentIdSet blocked_ids = GetBrokeredAttachments(m);
 140     if (!blocked_ids.empty()) {
 141       blocked_ids_.swap(blocked_ids);
 142       StartObservingAttachmentBroker();
 143       return DISPATCH_WAITING_ON_BROKER;
 144     }
 145
 146     DispatchMessage(m);
 147     queued_messages_.erase(queued_messages_.begin());
 148   }
 149   return DISPATCH_FINISHED;
 150 }
 151
 152 void ChannelReader::DispatchMessage(Message* m) {
 153   m->set_sender_pid(GetSenderPID());
 154
 155 #ifdef IPC_MESSAGE_LOG_ENABLED
 156   std::string name;
 157   Logging::GetInstance()->GetMessageText(m->type(), &name, m, NULL);
 158   TRACE_EVENT_WITH_FLOW1("ipc,toplevel",
 159                          "ChannelReader::DispatchInputData",
 160                          m->flags(),
 161                          TRACE_EVENT_FLAG_FLOW_IN,
 162                          "name", name);
 163 #else
 164   TRACE_EVENT_WITH_FLOW2("ipc,toplevel",
 165                          "ChannelReader::DispatchInputData",
 166                          m->flags(),
 167                          TRACE_EVENT_FLAG_FLOW_IN,
 168                          "class", IPC_MESSAGE_ID_CLASS(m->type()),
 169                          "line", IPC_MESSAGE_ID_LINE(m->type()));
 170 #endif
 171
 172   bool handled = false;
 173   if (IsInternalMessage(*m)) {
 174     HandleInternalMessage(*m);
 175     handled = true;
 176   }
 177 #if USE_ATTACHMENT_BROKER
 178   if (!handled && IsAttachmentBrokerEndpoint() && GetAttachmentBroker()) {
 179     handled = GetAttachmentBroker()->OnMessageReceived(*m);
 180   }
 181 #endif  // USE_ATTACHMENT_BROKER
 182   if (!handled)
 183     listener_->OnMessageReceived(*m);
 184   if (m->dispatch_error())
 185     listener_->OnBadMessageReceived(*m);
 186 }
 187
 188 ChannelReader::AttachmentIdSet ChannelReader::GetBrokeredAttachments(
 189     Message* msg) {
 190   std::set<BrokerableAttachment::AttachmentId> blocked_ids;
 191
 192 #if USE_ATTACHMENT_BROKER
 193   MessageAttachmentSet* set = msg->attachment_set();
 194   for (const scoped_refptr<BrokerableAttachment>& attachment :
 195        set->GetBrokerableAttachmentsForUpdating()) {
 196     if (attachment->NeedsBrokering()) {
 197       AttachmentBroker* broker = GetAttachmentBroker();
 198       scoped_refptr<BrokerableAttachment> brokered_attachment;
 199       bool result = broker->GetAttachmentWithId(attachment->GetIdentifier(),
 200                                                 &brokered_attachment);
 201       if (!result) {
 202         blocked_ids.insert(attachment->GetIdentifier());
 203         continue;
 204       }
 205
 206       attachment->PopulateWithAttachment(brokered_attachment.get());
 207     }
 208   }
 209 #endif  // USE_ATTACHMENT_BROKER
 210
 211   return blocked_ids;
 212 }
 213
 214 void ChannelReader::ReceivedBrokerableAttachmentWithId(
 215     const BrokerableAttachment::AttachmentId& id) {
 216   if (blocked_ids_.empty())
 217     return;
 218
 219   auto it = find(blocked_ids_.begin(), blocked_ids_.end(), id);
 220   if (it != blocked_ids_.end())
 221     blocked_ids_.erase(it);
 222
 223   if (blocked_ids_.empty()) {
 224     StopObservingAttachmentBroker();
 225     DispatchMessages();
 226   }
 227 }
 228
 229 void ChannelReader::StartObservingAttachmentBroker() {
 230 #if USE_ATTACHMENT_BROKER
 231   GetAttachmentBroker()->AddObserver(this);
 232 #endif  // USE_ATTACHMENT_BROKER
 233 }
 234
 235 void ChannelReader::StopObservingAttachmentBroker() {
 236 #if USE_ATTACHMENT_BROKER
 237   GetAttachmentBroker()->RemoveObserver(this);
 238 #endif  // USE_ATTACHMENT_BROKER
 239 }
 240
 241 }  // namespace internal
 242 }  // namespace IPC