Add multiplexing of message pipes in the new EDK.

mojo/edk/system/broker_state.h

Index: mojo/edk/system/broker_state.h
diff --git a/mojo/edk/system/broker_state.h b/mojo/edk/system/broker_state.h
index cf982c7b22cae2e36f732243fd5de5bf6d6fe4ed..29f3ffe35c39faa3e6b1cbf562c13acf840a706d 100644
--- a/mojo/edk/system/broker_state.h
+++ b/mojo/edk/system/broker_state.h
@@ -7,15 +7,18 @@
 
 #include "base/compiler_specific.h"
 #include "base/containers/hash_tables.h"
+#include "base/macros.h"
 #include "base/memory/singleton.h"
+#include "base/process/process_handle.h"
 #include "base/synchronization/lock.h"
-#include "base/threading/thread.h"
 #include "mojo/edk/embedder/scoped_platform_handle.h"
 #include "mojo/edk/system/broker.h"
 #include "mojo/edk/system/system_impl_export.h"
 
 namespace mojo {
 namespace edk {
+class ChildBrokerHost;
+class RoutedRawChannel;
 
 // Common broker state that has to live in a parent process. There is one
 // instance of this class in the parent process. This class implements the
@@ -24,7 +27,7 @@ class MOJO_SYSTEM_IMPL_EXPORT BrokerState : NON_EXPORTED_BASE(public Broker) {
  public:
   static BrokerState* GetInstance();
 
-  // Broker implementation.
+  // Broker implementation:
 #if defined(OS_WIN)
   void CreatePlatformChannelPair(ScopedPlatformHandle* server,
                                  ScopedPlatformHandle* client) override;
@@ -35,10 +38,20 @@ class MOJO_SYSTEM_IMPL_EXPORT BrokerState : NON_EXPORTED_BASE(public Broker) {
                      size_t count,
                      PlatformHandle* handles) override;
 #endif
+  void ConnectMessagePipe(uint64_t pipe_id,
+                          MessagePipeDispatcher* message_pipe) override;
+  void CloseMessagePipe(uint64_t pipe_id,
+                        MessagePipeDispatcher* message_pipe) override;
 
-  scoped_refptr<base::TaskRunner> broker_thread() {
-    return broker_thread_.task_runner();
-  }
+  // Called by ChildBrokerHost on construction and destruction.
+  void ChildBrokerHostCreated(ChildBrokerHost* child_broker_host);
+  void ChildBrokerHostDestructed(ChildBrokerHost* child_broker_host);
+
+  // These are called by ChildBrokerHost as they dispatch IPCs from ChildBroker.
+  // They are called on the IO thread.
+  void HandleConnectMessagePipe(ChildBrokerHost* pipe_process,
+                                uint64_t pipe_id);
+  void HandleCancelConnectMessagePipe(uint64_t pipe_id);
 
  private:
   friend struct base::DefaultSingletonTraits<BrokerState>;
@@ -46,20 +59,58 @@ class MOJO_SYSTEM_IMPL_EXPORT BrokerState : NON_EXPORTED_BASE(public Broker) {
   BrokerState();
   ~BrokerState() override;
 
-  // A separate thread to handle sync IPCs from child processes for exchanging
-  // platform handles with tokens. We use a separate thread because latency is
-  // very sensitive (since any time a pipe is created or sent, a child process
-  // makes a sync call to this class).
-  base::Thread broker_thread_;
+  // Checks if there's a direct channel between the two processes, and if not
+  // creates one and tells them about it.
+  // If one of the processes is the current one, it should be pid1.
+  // Called on the IO thread.
+  void EnsureProcessesConnected(base::ProcessId pid1, base::ProcessId pid2);
+
+  // Callback when a RoutedRawChannel is destroyed for cleanup.
+  // Called on the IO thread.
+  void ChannelDestructed(RoutedRawChannel* channel);
 
 #if defined(OS_WIN)
   // Used in the parent (unsandboxed) process to hold a mapping between HANDLES
   // and tokens. When a child process wants to send a HANDLE to another process,
   // it exchanges it to a token and then the other process exchanges that token
   // back to a HANDLE.
-  base::Lock lock_;  // Guards access to below.
+  base::Lock token_map_lock_;
   base::hash_map<uint64_t, HANDLE> token_map_;
 #endif
+
+  // For pending connects originiating in this process.
+  // Only accessed on the IO thread.
+  base::hash_map<uint64_t, MessagePipeDispatcher*> pending_connects_;
+
+  // For connected message pipes in this process. This is needed so that when a
+  // MessagePipeDispatcher is closed we can remove the route for the
+  // corresponding RoutedRawChannel.
+  // Only accessed on the IO thread.
+  base::hash_map<MessagePipeDispatcher*, RoutedRawChannel*> connected_pipes_;
+
+  base::Lock lock_;  // Guards access to below.
+
+  // Holds a map of all the RoutedRawChannel that connect this parent process to
+  // a child process. The key is the child process'd pid.
+  base::hash_map<base::ProcessId, RoutedRawChannel*> child_channels_;
+
+  base::hash_map<uint64_t, ChildBrokerHost*> pending_child_connects_;
+
+  // Each entry is an std::pair of ints of processe IDs that have
+  // RoutedRawChannel objects between them. The pair always has the smaller
+  // process id value first.
+  // For now, we don't reap connections if there are no more routes between two
+  // processes.
+  base::hash_set<std::pair<base::ProcessId, base::ProcessId>>
+      connected_processes_;
+
+  base::hash_map<base::ProcessId, ChildBrokerHost*> child_processes_;
+
+  // Used for message pipes in the same process.
+  RoutedRawChannel* in_process_pipes_channel1_;
+  RoutedRawChannel* in_process_pipes_channel2_;
+
+  DISALLOW_COPY_AND_ASSIGN(BrokerState);
 };
 
 }  // namespace edk