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