content/browser/message_port_service.cc

   1 // Copyright 2013 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 "content/browser/message_port_service.h"
   6
   7 #include "content/common/message_port_messages.h"
   8 #include "content/public/browser/browser_thread.h"
   9 #include "content/public/browser/message_port_delegate.h"
  10
  11 namespace content {
  12
  13 struct MessagePortService::MessagePort {
  14   // |delegate| and |route_id| are what we need to send messages to the port.
  15   // |delegate| is just a raw pointer since it notifies us by calling
  16   // OnMessagePortDelegateClosing before it gets destroyed.
  17   MessagePortDelegate* delegate;
  18   int route_id;
  19   // A globally unique id for this message port.
  20   int message_port_id;
  21   // The globally unique id of the entangled message port.
  22   int entangled_message_port_id;
  23   // If true, all messages to this message port are queued and not delivered.
  24   // This is needed so that when a message port is sent between processes all
  25   // pending message get transferred. There are two possibilities for pending
  26   // messages: either they are already received by the child process, or they're
  27   // in-flight. This flag ensures that the latter type get flushed through the
  28   // system.
  29   // This flag should only be set to true in response to
  30   // MessagePortHostMsg_QueueMessages.
  31   bool queue_for_inflight_messages;
  32   // If true, all messages to this message port are queued and not delivered.
  33   // This is needed so that when a message port is sent to a new process all
  34   // messages are held in the browser process until the destination process is
  35   // ready to receive messages. This flag is set true when a message port is
  36   // transferred to a different process but there isn't immediately a
  37   // MessagePortDelegate available for that new process. Once the
  38   // destination process is ready to receive messages it sends
  39   // MessagePortHostMsg_ReleaseMessages to set this flag to false.
  40   bool hold_messages_for_destination;
  41   // Returns true if messages should be queued for either reason.
  42   bool queue_messages() const {
  43     return queue_for_inflight_messages || hold_messages_for_destination;
  44   }
  45   // If true, the message port should be destroyed but was currently still
  46   // waiting for a SendQueuedMessages message from a renderer. As soon as that
  47   // message is received the port will actually be destroyed.
  48   bool should_be_destroyed;
  49   QueuedMessages queued_messages;
  50 };
  51
  52 MessagePortService* MessagePortService::GetInstance() {
  53   return Singleton<MessagePortService>::get();
  54 }
  55
  56 MessagePortService::MessagePortService()
  57     : next_message_port_id_(0) {
  58 }
  59
  60 MessagePortService::~MessagePortService() {
  61 }
  62
  63 void MessagePortService::UpdateMessagePort(int message_port_id,
  64                                            MessagePortDelegate* delegate,
  65                                            int routing_id) {
  66   DCHECK_CURRENTLY_ON(BrowserThread::IO);
  67   if (!message_ports_.count(message_port_id)) {
  68     NOTREACHED();
  69     return;
  70   }
  71
  72   MessagePort& port = message_ports_[message_port_id];
  73   port.delegate = delegate;
  74   port.route_id = routing_id;
  75 }
  76
  77 void MessagePortService::GetMessagePortInfo(int message_port_id,
  78                                             MessagePortDelegate** delegate,
  79                                             int* routing_id) {
  80   DCHECK_CURRENTLY_ON(BrowserThread::IO);
  81   if (!message_ports_.count(message_port_id)) {
  82     NOTREACHED();
  83     return;
  84   }
  85
  86   const MessagePort& port = message_ports_[message_port_id];
  87   if (delegate)
  88     *delegate = port.delegate;
  89   if (routing_id)
  90     *routing_id = port.route_id;
  91 }
  92
  93 void MessagePortService::OnMessagePortDelegateClosing(
  94     MessagePortDelegate* delegate) {
  95   DCHECK_CURRENTLY_ON(BrowserThread::IO);
  96   // Check if the (possibly) crashed process had any message ports.
  97   for (MessagePorts::iterator iter = message_ports_.begin();
  98        iter != message_ports_.end();) {
  99     MessagePorts::iterator cur_item = iter++;
 100     if (cur_item->second.delegate == delegate) {
 101       Erase(cur_item->first);
 102     }
 103   }
 104 }
 105
 106 void MessagePortService::Create(int route_id,
 107                                 MessagePortDelegate* delegate,
 108                                 int* message_port_id) {
 109   DCHECK_CURRENTLY_ON(BrowserThread::IO);
 110   *message_port_id = ++next_message_port_id_;
 111
 112   MessagePort port;
 113   port.delegate = delegate;
 114   port.route_id = route_id;
 115   port.message_port_id = *message_port_id;
 116   port.entangled_message_port_id = MSG_ROUTING_NONE;
 117   port.queue_for_inflight_messages = false;
 118   port.hold_messages_for_destination = false;
 119   port.should_be_destroyed = false;
 120   message_ports_[*message_port_id] = port;
 121 }
 122
 123 void MessagePortService::Destroy(int message_port_id) {
 124   DCHECK_CURRENTLY_ON(BrowserThread::IO);
 125   if (!message_ports_.count(message_port_id)) {
 126     NOTREACHED();
 127     return;
 128   }
 129
 130   DCHECK(message_ports_[message_port_id].queued_messages.empty());
 131
 132   Erase(message_port_id);
 133 }
 134
 135 void MessagePortService::Entangle(int local_message_port_id,
 136                                   int remote_message_port_id) {
 137   DCHECK_CURRENTLY_ON(BrowserThread::IO);
 138   if (!message_ports_.count(local_message_port_id) ||
 139       !message_ports_.count(remote_message_port_id)) {
 140     NOTREACHED();
 141     return;
 142   }
 143
 144   DCHECK(message_ports_[remote_message_port_id].entangled_message_port_id ==
 145       MSG_ROUTING_NONE);
 146   message_ports_[remote_message_port_id].entangled_message_port_id =
 147       local_message_port_id;
 148 }
 149
 150 void MessagePortService::PostMessage(
 151     int sender_message_port_id,
 152     const MessagePortMessage& message,
 153     const std::vector<TransferredMessagePort>& sent_message_ports) {
 154   DCHECK_CURRENTLY_ON(BrowserThread::IO);
 155   if (!message_ports_.count(sender_message_port_id)) {
 156     NOTREACHED();
 157     return;
 158   }
 159
 160   int entangled_message_port_id =
 161       message_ports_[sender_message_port_id].entangled_message_port_id;
 162   if (entangled_message_port_id == MSG_ROUTING_NONE)
 163     return;  // Process could have crashed.
 164
 165   if (!message_ports_.count(entangled_message_port_id)) {
 166     NOTREACHED();
 167     return;
 168   }
 169
 170   PostMessageTo(entangled_message_port_id, message, sent_message_ports);
 171 }
 172
 173 void MessagePortService::PostMessageTo(
 174     int message_port_id,
 175     const MessagePortMessage& message,
 176     const std::vector<TransferredMessagePort>& sent_message_ports) {
 177   if (!message_ports_.count(message_port_id)) {
 178     NOTREACHED();
 179     return;
 180   }
 181   for (size_t i = 0; i < sent_message_ports.size(); ++i) {
 182     if (!message_ports_.count(sent_message_ports[i].id)) {
 183       NOTREACHED();
 184       return;
 185     }
 186   }
 187
 188   MessagePort& entangled_port = message_ports_[message_port_id];
 189   if (entangled_port.queue_messages()) {
 190     // If the target port is currently holding messages because the destination
 191     // renderer isn't available yet, all message ports being sent should also be
 192     // put in this state.
 193     if (entangled_port.hold_messages_for_destination) {
 194       for (const auto& port : sent_message_ports)
 195         HoldMessages(port.id);
 196     }
 197     entangled_port.queued_messages.push_back(
 198         std::make_pair(message, sent_message_ports));
 199
 200     if (entangled_port.delegate)
 201       entangled_port.delegate->MessageWasHeld(entangled_port.route_id);
 202
 203     return;
 204   }
 205
 206   if (!entangled_port.delegate) {
 207     NOTREACHED();
 208     return;
 209   }
 210
 211   // Now send the message to the entangled port.
 212   entangled_port.delegate->SendMessage(entangled_port.route_id, message,
 213                                        sent_message_ports);
 214 }
 215
 216 void MessagePortService::QueueMessages(int message_port_id) {
 217   DCHECK_CURRENTLY_ON(BrowserThread::IO);
 218   if (!message_ports_.count(message_port_id)) {
 219     NOTREACHED();
 220     return;
 221   }
 222
 223   MessagePort& port = message_ports_[message_port_id];
 224   if (port.delegate) {
 225     port.delegate->SendMessagesAreQueued(port.route_id);
 226     port.queue_for_inflight_messages = true;
 227     port.delegate = NULL;
 228   }
 229 }
 230
 231 void MessagePortService::SendQueuedMessages(
 232     int message_port_id,
 233     const QueuedMessages& queued_messages) {
 234   DCHECK_CURRENTLY_ON(BrowserThread::IO);
 235   if (!message_ports_.count(message_port_id)) {
 236     NOTREACHED();
 237     return;
 238   }
 239
 240   // Send the queued messages to the port again.  This time they'll reach the
 241   // new location.
 242   MessagePort& port = message_ports_[message_port_id];
 243   port.queue_for_inflight_messages = false;
 244
 245   // If the port is currently holding messages waiting for the target renderer,
 246   // all ports in messages being sent to the port should also be put on hold.
 247   if (port.hold_messages_for_destination) {
 248     for (const auto& message : queued_messages)
 249       for (const TransferredMessagePort& sent_port : message.second)
 250         HoldMessages(sent_port.id);
 251   }
 252
 253   port.queued_messages.insert(port.queued_messages.begin(),
 254                               queued_messages.begin(),
 255                               queued_messages.end());
 256
 257   if (port.should_be_destroyed)
 258     ClosePort(message_port_id);
 259   else
 260     SendQueuedMessagesIfPossible(message_port_id);
 261 }
 262
 263 void MessagePortService::SendQueuedMessagesIfPossible(int message_port_id) {
 264   DCHECK_CURRENTLY_ON(BrowserThread::IO);
 265   if (!message_ports_.count(message_port_id)) {
 266     NOTREACHED();
 267     return;
 268   }
 269
 270   MessagePort& port = message_ports_[message_port_id];
 271   if (port.queue_messages() || !port.delegate)
 272     return;
 273
 274   for (QueuedMessages::iterator iter = port.queued_messages.begin();
 275        iter != port.queued_messages.end(); ++iter) {
 276     PostMessageTo(message_port_id, iter->first, iter->second);
 277   }
 278   port.queued_messages.clear();
 279 }
 280
 281 void MessagePortService::HoldMessages(int message_port_id) {
 282   DCHECK_CURRENTLY_ON(BrowserThread::IO);
 283   if (!message_ports_.count(message_port_id)) {
 284     NOTREACHED();
 285     return;
 286   }
 287
 288   // Any ports in messages currently in the queue should also be put on hold.
 289   for (const auto& message : message_ports_[message_port_id].queued_messages)
 290     for (const TransferredMessagePort& sent_port : message.second)
 291       HoldMessages(sent_port.id);
 292
 293   message_ports_[message_port_id].hold_messages_for_destination = true;
 294 }
 295
 296 void MessagePortService::ClosePort(int message_port_id) {
 297   DCHECK_CURRENTLY_ON(BrowserThread::IO);
 298   if (!message_ports_.count(message_port_id)) {
 299     NOTREACHED();
 300     return;
 301   }
 302
 303   if (message_ports_[message_port_id].queue_for_inflight_messages) {
 304     message_ports_[message_port_id].should_be_destroyed = true;
 305     return;
 306   }
 307
 308   // First close any message ports in the queue for this message port.
 309   for (const auto& message : message_ports_[message_port_id].queued_messages)
 310     for (const TransferredMessagePort& sent_port : message.second)
 311       ClosePort(sent_port.id);
 312
 313   Erase(message_port_id);
 314 }
 315
 316 void MessagePortService::ReleaseMessages(int message_port_id) {
 317   DCHECK_CURRENTLY_ON(BrowserThread::IO);
 318   if (!message_ports_.count(message_port_id)) {
 319     NOTREACHED();
 320     return;
 321   }
 322
 323   message_ports_[message_port_id].hold_messages_for_destination = false;
 324   SendQueuedMessagesIfPossible(message_port_id);
 325 }
 326
 327 void MessagePortService::Erase(int message_port_id) {
 328   MessagePorts::iterator erase_item = message_ports_.find(message_port_id);
 329   DCHECK(erase_item != message_ports_.end());
 330
 331   int entangled_id = erase_item->second.entangled_message_port_id;
 332   if (entangled_id != MSG_ROUTING_NONE) {
 333     // Do the disentanglement (and be paranoid about the other side existing
 334     // just in case something unusual happened during entanglement).
 335     if (message_ports_.count(entangled_id)) {
 336       message_ports_[entangled_id].entangled_message_port_id = MSG_ROUTING_NONE;
 337     }
 338   }
 339   message_ports_.erase(erase_item);
 340 }
 341
 342 }  // namespace content