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