Fix race condition with message ports being dropped if sent before a worker starts.

BUG=22315
TEST=covered by worker-messageport-gc.html
Review URL: http://codereview.chromium.org/208045

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@26749 0039d316-1c4b-4281-b951-d872f2087c98

3 files changed, 45 insertions, 24 deletions

diff --git a/chrome/browser/worker_host/message_port_dispatcher.cc b/chrome/browser/worker_host/message_port_dispatcher.cc
index 41ebc37..c6e992d 100644
--- a/chrome/browser/worker_host/message_port_dispatcher.cc
+++ b/chrome/browser/worker_host/message_port_dispatcher.cc
@@ -193,8 +193,9 @@ void MessagePortDispatcher::OnQueueMessages(int message_port_id) {
   }
 
   MessagePort& port = message_ports_[message_port_id];
-  port.queue_messages = true;
   port.sender->Send(new WorkerProcessMsg_MessagesQueued(port.route_id));
+  port.queue_messages = true;
+  port.sender = NULL;
 }
 
 void MessagePortDispatcher::OnSendQueuedMessages(
@@ -212,6 +213,14 @@ void MessagePortDispatcher::OnSendQueuedMessages(
   port.queued_messages.insert(port.queued_messages.begin(),
                               queued_messages.begin(),
                               queued_messages.end());
+  SendQueuedMessagesIfPossible(message_port_id);
+}
+
+void MessagePortDispatcher::SendQueuedMessagesIfPossible(int message_port_id) {
+  MessagePort& port = message_ports_[message_port_id];
+  if (port.queue_messages || !port.sender)
+    return;
+
   for (QueuedMessages::iterator iter = port.queued_messages.begin();
        iter != port.queued_messages.end(); ++iter) {
     PostMessageTo(message_port_id, iter->first, iter->second);
diff --git a/chrome/browser/worker_host/message_port_dispatcher.h b/chrome/browser/worker_host/message_port_dispatcher.h
index c728f38..21be0c6 100644
--- a/chrome/browser/worker_host/message_port_dispatcher.h
+++ b/chrome/browser/worker_host/message_port_dispatcher.h
@@ -28,12 +28,17 @@ class MessagePortDispatcher : public NotificationObserver {
                          CallbackWithReturnValue<int>::Type* next_routing_id,
                          bool* message_was_ok);
 
+  // Updates the information needed to reach a message port when it's sent to a
+  // (possibly different) process.
   void UpdateMessagePort(
       int message_port_id,
       IPC::Message::Sender* sender,
       int routing_id,
       CallbackWithReturnValue<int>::Type* next_routing_id);
 
+  // Attempts to send the queued messages for a message port.
+  void SendQueuedMessagesIfPossible(int message_port_id);
+
   bool Send(IPC::Message* message);
 
  private:
diff --git a/chrome/browser/worker_host/worker_process_host.cc b/chrome/browser/worker_host/worker_process_host.cc
index 9570743..b3a5b1c 100644
--- a/chrome/browser/worker_host/worker_process_host.cc
+++ b/chrome/browser/worker_host/worker_process_host.cc
@@ -228,33 +228,40 @@ void WorkerProcessHost::RelayMessage(
     IPC::Message::Sender* sender,
     int route_id,
     CallbackWithReturnValue<int>::Type* next_route_id) {
-  IPC::Message* new_message;
-  if (message.type() == WorkerMsg_PostMessage::ID) {
-    // We want to send the receiver a routing id for the new channel, so
-    // crack the message first.
-    string16 msg;
-    std::vector<int> sent_message_port_ids;
-    std::vector<int> new_routing_ids;
-    if (!WorkerMsg_PostMessage::Read(
-            &message, &msg, &sent_message_port_ids, &new_routing_ids)) {
-      return;
-    }
-    DCHECK(sent_message_port_ids.size() == new_routing_ids.size());
+  if (message.type() != WorkerMsg_PostMessage::ID) {
+    IPC::Message* new_message = new IPC::Message(message);
+    new_message->set_routing_id(route_id);
+    sender->Send(new_message);
+    return;
+  }
 
-    for (size_t i = 0; i < sent_message_port_ids.size(); ++i) {
-      new_routing_ids[i] = next_route_id->Run();
-      MessagePortDispatcher::GetInstance()->UpdateMessagePort(
-          sent_message_port_ids[i], sender, new_routing_ids[i], next_route_id);
-    }
+  // We want to send the receiver a routing id for the new channel, so
+  // crack the message first.
+  string16 msg;
+  std::vector<int> sent_message_port_ids;
+  std::vector<int> new_routing_ids;
+  if (!WorkerMsg_PostMessage::Read(
+          &message, &msg, &sent_message_port_ids, &new_routing_ids)) {
+    return;
+  }
+  DCHECK(sent_message_port_ids.size() == new_routing_ids.size());
 
-    new_message = new WorkerMsg_PostMessage(
-        route_id, msg, sent_message_port_ids, new_routing_ids);
-  } else {
-    new_message = new IPC::Message(message);
-    new_message->set_routing_id(route_id);
+  for (size_t i = 0; i < sent_message_port_ids.size(); ++i) {
+    new_routing_ids[i] = next_route_id->Run();
+    MessagePortDispatcher::GetInstance()->UpdateMessagePort(
+        sent_message_port_ids[i], sender, new_routing_ids[i], next_route_id);
   }
 
-  sender->Send(new_message);
+  sender->Send(new WorkerMsg_PostMessage(
+      route_id, msg, sent_message_port_ids, new_routing_ids));
+
+  // Send any queued messages to the sent message ports.  We can only do this
+  // after sending the above message, since it's the one that sets up the
+  // message port route which the queued messages are sent to.
+  for (size_t i = 0; i < sent_message_port_ids.size(); ++i) {
+    MessagePortDispatcher::GetInstance()->
+        SendQueuedMessagesIfPossible(sent_message_port_ids[i]);
+  }
 }
 
 void WorkerProcessHost::SenderShutdown(IPC::Message::Sender* sender) {