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..7541475eb347270107ec8dd2eedd39ad2ff5dd9f 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,19 @@ 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.
+  void HandleConnectMessagePipe(ChildBrokerHost* pipe_process,
+                                uint64_t pipe_id);
+  void HandleCancelConnectMessagePipe(uint64_t pipe_id);
 
  private:
   friend struct base::DefaultSingletonTraits<BrokerState>;
@@ -46,20 +58,54 @@ 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.
+  void EnsureProcessesConnected(base::ProcessId pid1, base::ProcessId pid2);
+
+  // Callback when a RoutedRawChannel is destroyed for cleanup.
+  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
+
+  base::Lock lock_;  // Guards access to below.

[yzshen1, 2015/12/03 23:37:50]

If ConnectMessagePipe() and CloseMessagePipe() are only called from the IO
thread, it is possible to access (some of) the following members without
locking?

[jam, 2015/12/04 05:06:47]

The issue is that child_processes_ is accessed in ChildBrokerHostCreated outside
the IO thread. Since that is accessed in other functions, it would be
inconvenient to lock this member but not others (say specifically
child_channels_ in EnsureProcessesConnected).

[yzshen1, 2015/12/04 05:57:05]

I see. Thanks for explanation!

+
+  // For pending connects originiating in this process.
+  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.
+  base::hash_map<MessagePipeDispatcher*, RoutedRawChannel*> connected_pipes_;
+
+  // 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