[mojo-edk] Broadcast surprise port disruptions

mojo/edk/system/node_controller.cc

 // Copyright 2016 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "mojo/edk/system/node_controller.h"
 
 #include <algorithm>
 #include <limits>
 
 #include "base/bind.h"
@@ skipping 55 matching lines @@
   DCHECK_LE(count, static_cast<size_t>(std::numeric_limits<int32_t>::max()));
 
   // 8k is the maximum number of file descriptors allowed in Chrome.
   UMA_HISTOGRAM_CUSTOM_COUNTS("Mojo.System.Node.PendingChildren",
                               static_cast<int32_t>(count),
                               0 /* min */,
                               8000 /* max */,
                               50 /* bucket count */);
 }
 
+bool ParsePortsMessage(Channel::Message* message,
+                       void** data,
+                       size_t* num_data_bytes,
+                       size_t* num_header_bytes,
+                       size_t* num_payload_bytes,
+                       size_t* num_ports_bytes) {
+  DCHECK(data && num_data_bytes && num_header_bytes && num_payload_bytes &&
+         num_ports_bytes);
+
+  NodeChannel::GetPortsMessageData(message, data, num_data_bytes);
+  if (!*num_data_bytes)
+    return false;
+
+  if (!ports::Message::Parse(*data, *num_data_bytes, num_header_bytes,
+                             num_payload_bytes, num_ports_bytes)) {
+    return false;
+  }
+
+  return true;
+}
+
 // Used by NodeController to watch for shutdown. Since no IO can happen once
 // the IO thread is killed, the NodeController can cleanly drop all its peers
 // at that time.
 class ThreadDestructionObserver :
     public base::MessageLoop::DestructionObserver {
  public:
   static void Create(scoped_refptr<base::TaskRunner> task_runner,
                      const base::Closure& callback) {
     if (task_runner->RunsTasksOnCurrentThread()) {
       // Owns itself.
@@ skipping 579 matching lines @@
     // (synchronously) in response to Node's ClosePort, SendMessage, or
     // AcceptMessage, we flush the queue after calling any of those methods.
     base::AutoLock lock(messages_lock_);
     incoming_messages_.emplace(std::move(message));
     incoming_messages_flag_.Set(true);
   } else {
     SendPeerMessage(node, std::move(message));
   }
 }
 
+void NodeController::BroadcastMessage(ports::ScopedMessage message) {
+  CHECK_EQ(message->num_ports(), 0u);
+  Channel::MessagePtr channel_message =
+      static_cast<PortsMessage*>(message.get())->TakeChannelMessage();
+  CHECK(!channel_message->has_handles());
+
+  scoped_refptr<NodeChannel> broker = GetBrokerChannel();
+  if (broker)
+    broker->Broadcast(std::move(channel_message));
+  else
+    OnBroadcast(name_, std::move(channel_message));
+}
+
 void NodeController::PortStatusChanged(const ports::PortRef& port) {
   scoped_refptr<ports::UserData> user_data;
   node_->GetUserData(port, &user_data);
 
   PortObserver* observer = static_cast<PortObserver*>(user_data.get());
   if (observer) {
     observer->OnPortStatusChanged();
   } else {
     DVLOG(2) << "Ignoring status change for " << port.name() << " because it "
              << "doesn't have an observer.";
@@ skipping 214 matching lines @@
 #endif
 
   DVLOG(1) << "Child " << name_ << " accepted by broker " << broker_name;
 }
 
 void NodeController::OnPortsMessage(const ports::NodeName& from_node,
                                     Channel::MessagePtr channel_message) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
   void* data;
-  size_t num_data_bytes;
-  NodeChannel::GetPortsMessageData(
-      channel_message.get(), &data, &num_data_bytes);
-  if (!num_data_bytes) {
+  size_t num_data_bytes, num_header_bytes, num_payload_bytes, num_ports_bytes;
+  if (!ParsePortsMessage(channel_message.get(), &data, &num_data_bytes,
+                         &num_header_bytes, &num_payload_bytes,
+                         &num_ports_bytes)) {
     DropPeer(from_node);
     return;
   }
-
-  size_t num_header_bytes, num_payload_bytes, num_ports_bytes;
-  if (!ports::Message::Parse(data,
-                             num_data_bytes,
-                             &num_header_bytes,
-                             &num_payload_bytes,
-                             &num_ports_bytes)) {
-    DropPeer(from_node);
-    return;
-  }
 
   CHECK(channel_message);
   std::unique_ptr<PortsMessage> ports_message(
       new PortsMessage(num_header_bytes,
                        num_payload_bytes,
                        num_ports_bytes,
                        std::move(channel_message)));
   ports_message->set_source_node(from_node);
   node_->AcceptMessage(ports::ScopedMessage(ports_message.release()));
   AcceptIncomingMessages();
@@ skipping 63 matching lines @@
   }
 
   scoped_refptr<NodeChannel> channel =
       NodeChannel::Create(this, std::move(channel_handle), io_task_runner_,
                           ProcessErrorCallback());
 
   DVLOG(1) << "Adding new peer " << name << " via parent introduction.";
   AddPeer(name, channel, true /* start_channel */);
 }
 
+void NodeController::OnBroadcast(const ports::NodeName& from_node,
+                                 Channel::MessagePtr message) {
+  DCHECK(!message->has_handles());
+
+  void* data;
+  size_t num_data_bytes, num_header_bytes, num_payload_bytes, num_ports_bytes;
+  if (!ParsePortsMessage(message.get(), &data, &num_data_bytes,
+                         &num_header_bytes, &num_payload_bytes,
+                         &num_ports_bytes)) {
+    DropPeer(from_node);
+    return;
+  }
+
+  // Broadcast messages must not contain ports.
+  if (num_ports_bytes > 0) {
+    DropPeer(from_node);
+    return;
+  }
+
+  base::AutoLock lock(peers_lock_);
+  for (auto& iter : peers_) {
+    // Copy and send the message to each known peer.
+    Channel::MessagePtr peer_message(
+        new Channel::Message(message->payload_size(), 0));
+    memcpy(peer_message->mutable_payload(), message->payload(),
+           message->payload_size());
+    iter.second->PortsMessage(std::move(peer_message));
+  }
+}
+
 #if defined(OS_WIN) || (defined(OS_MACOSX) && !defined(OS_IOS))
 void NodeController::OnRelayPortsMessage(const ports::NodeName& from_node,
                                          base::ProcessHandle from_process,
                                          const ports::NodeName& destination,
                                          Channel::MessagePtr message) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
   if (GetBrokerChannel()) {
     // Only the broker should be asked to relay a message.
     LOG(ERROR) << "Non-broker refusing to relay message.";
@@ skipping 117 matching lines @@
     shutdown_callback_flag_.Set(false);
   }
 
   DCHECK(!callback.is_null());
 
   callback.Run();
 }
 
 }  // namespace edk
 }  // namespace mojo