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