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