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_EVENT1("ipc,toplevel", "ChannelReader::DispatchInputData", "name",
 159                name);
 160 #else
 161   TRACE_EVENT2("ipc,toplevel", "ChannelReader::DispatchInputData", "class",
 162                IPC_MESSAGE_ID_CLASS(m->type()), "line",
 163                IPC_MESSAGE_ID_LINE(m->type()));
 164 #endif
 165   m->TraceMessageEnd();
 166
 167   bool handled = false;
 168   if (IsInternalMessage(*m)) {
 169     HandleInternalMessage(*m);
 170     handled = true;
 171   }
 172 #if USE_ATTACHMENT_BROKER
 173   if (!handled && IsAttachmentBrokerEndpoint() && GetAttachmentBroker()) {
 174     handled = GetAttachmentBroker()->OnMessageReceived(*m);
 175   }
 176 #endif  // USE_ATTACHMENT_BROKER
 177   if (!handled)
 178     listener_->OnMessageReceived(*m);
 179   if (m->dispatch_error())
 180     listener_->OnBadMessageReceived(*m);
 181 }
 182
 183 ChannelReader::AttachmentIdSet ChannelReader::GetBrokeredAttachments(
 184     Message* msg) {
 185   std::set<BrokerableAttachment::AttachmentId> blocked_ids;
 186
 187 #if USE_ATTACHMENT_BROKER
 188   MessageAttachmentSet* set = msg->attachment_set();
 189   for (const scoped_refptr<BrokerableAttachment>& attachment :
 190        set->GetBrokerableAttachmentsForUpdating()) {
 191     if (attachment->NeedsBrokering()) {
 192       AttachmentBroker* broker = GetAttachmentBroker();
 193       scoped_refptr<BrokerableAttachment> brokered_attachment;
 194       bool result = broker->GetAttachmentWithId(attachment->GetIdentifier(),
 195                                                 &brokered_attachment);
 196       if (!result) {
 197         blocked_ids.insert(attachment->GetIdentifier());
 198         continue;
 199       }
 200
 201       attachment->PopulateWithAttachment(brokered_attachment.get());
 202     }
 203   }
 204 #endif  // USE_ATTACHMENT_BROKER
 205
 206   return blocked_ids;
 207 }
 208
 209 void ChannelReader::ReceivedBrokerableAttachmentWithId(
 210     const BrokerableAttachment::AttachmentId& id) {
 211   if (blocked_ids_.empty())
 212     return;
 213
 214   auto it = find(blocked_ids_.begin(), blocked_ids_.end(), id);
 215   if (it != blocked_ids_.end())
 216     blocked_ids_.erase(it);
 217
 218   if (blocked_ids_.empty()) {
 219     StopObservingAttachmentBroker();
 220     DispatchMessages();
 221   }
 222 }
 223
 224 void ChannelReader::StartObservingAttachmentBroker() {
 225 #if USE_ATTACHMENT_BROKER
 226   GetAttachmentBroker()->AddObserver(this);
 227 #endif  // USE_ATTACHMENT_BROKER
 228 }
 229
 230 void ChannelReader::StopObservingAttachmentBroker() {
 231 #if USE_ATTACHMENT_BROKER
 232   GetAttachmentBroker()->RemoveObserver(this);
 233 #endif  // USE_ATTACHMENT_BROKER
 234 }
 235
 236 }  // namespace internal
 237 }  // namespace IPC