[mojo] Add IPC support to core shared library

Updates the mojo_core shared library so that it is actually suitable
for production IPC use. Namely:

- MojoInitialize now spins up a background thread for I/O and gives
  it to the core impl.
- A flag is added to Initialize(), allowing shared library consumers
  to initialize themselves as a broker process.
- The thunks helper correctly loads the mojo_core library with
  RTLD_DEEPBIND set, avoiding heap crossover within the core impl
  due to e.g. malloc and free lazy-binding to different heaps
- Fixes the thunks header to compile as C once again, as it was
  missing some struct keywords.
- Adds MojoShutdown to the core ABI as a necessary call for shared
  library consumers who want clean shutdown.

This also adds a new multiprocess test consuming the shared library
to tie together and validate all of the above changes.

Bug: 809320
Change-Id: Ic1c56d99c86fd4a2dc7f812ee152994ced35ece6
Reviewed-on: https://chromium-review.googlesource.com/c/1306347
Commit-Queue: Ken Rockot <rockot@google.com>
Reviewed-by: Reilly Grant <reillyg@chromium.org>
Cr-Commit-Position: refs/heads/master@{#605109}

mojo/core/BUILD.gn

@@ -233,6 +233,9 @@ if (is_chromeos || is_linux || is_android || is_win) {
   }
 
   config("export_only_thunks_api") {
+    inputs = [
+      "export_only_thunks_api.lst",
+    ]
     ldflags = [ "-Wl,--version-script=" +
                 rebase_path("//mojo/core/export_only_thunks_api.lst",
                             root_build_dir) ]
@@ -249,6 +252,8 @@ if (is_chromeos || is_linux || is_android || is_win) {
         "//base",
         "//base/test:test_support",
         "//mojo/public/c/system",
+        "//mojo/public/cpp/platform",
+        "//mojo/public/cpp/system",
         "//testing/gtest",
       ]
 

mojo/core/entrypoints.cc

@@ -21,8 +21,8 @@ mojo::core::Core* g_core;
 extern "C" {
 
 MojoResult MojoInitializeImpl(const struct MojoInitializeOptions* options) {
-  NOTREACHED() << "Do not call MojoInitialize() as an EDK embedder!";
-  return MOJO_RESULT_OK;
+  NOTREACHED() << "Do not call MojoInitialize() as a Mojo Core embedder!";
+  return MOJO_RESULT_UNIMPLEMENTED;
 }
 
 MojoTimeTicks MojoGetTimeTicksNowImpl() {
@@ -345,6 +345,11 @@ MojoResult MojoQueryQuotaImpl(MojoHandle handle,
                             current_usage);
 }
 
+MojoResult MojoShutdownImpl(const MojoShutdownOptions* options) {
+  NOTREACHED() << "Do not call MojoShutdown() as a Mojo Core embedder!";
+  return MOJO_RESULT_UNIMPLEMENTED;
+}
+
 }  // extern "C"
 
 MojoSystemThunks g_thunks = {sizeof(MojoSystemThunks),
@@ -390,7 +395,8 @@ MojoSystemThunks g_thunks = {sizeof(MojoSystemThunks),
                              MojoSendInvitationImpl,
                              MojoAcceptInvitationImpl,
                              MojoSetQuotaImpl,
-                             MojoQueryQuotaImpl};
+                             MojoQueryQuotaImpl,
+                             MojoShutdownImpl};
 
 }  // namespace
 

mojo/core/mojo_core.cc

@@ -2,18 +2,85 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+#include <stddef.h>
+
+#include "base/at_exit.h"
+#include "base/bind.h"
+#include "base/macros.h"
+#include "base/no_destructor.h"
+#include "base/synchronization/waitable_event.h"
+#include "base/threading/thread.h"
 #include "base/time/time.h"
+#include "mojo/core/configuration.h"
 #include "mojo/core/core.h"
 #include "mojo/core/entrypoints.h"
 #include "mojo/public/c/system/core.h"
 #include "mojo/public/c/system/thunks.h"
 
+namespace {
+
+class IPCSupport {
+ public:
+  IPCSupport() : ipc_thread_("Mojo IPC") {
+    base::Thread::Options options(base::MessageLoop::TYPE_IO, 0);
+    ipc_thread_.StartWithOptions(options);
+    mojo::core::Core::Get()->SetIOTaskRunner(ipc_thread_.task_runner());
+  }
+
+  ~IPCSupport() {
+    base::WaitableEvent wait(base::WaitableEvent::ResetPolicy::MANUAL,
+                             base::WaitableEvent::InitialState::NOT_SIGNALED);
+    mojo::core::Core::Get()->RequestShutdown(base::BindRepeating(
+        &base::WaitableEvent::Signal, base::Unretained(&wait)));
+    wait.Wait();
+  }
+
+ private:
+#if !defined(COMPONENT_BUILD)
+  // NOTE: For component builds, we assume the consumer is always a target in
+  // the Chromium tree which already depends on base initialization stuff and
+  // therefore already has an AtExitManager. For non-component builds, use of
+  // this AtExitManager is strictly isolated to Mojo Core internals, so running
+  // hooks on |MojoShutdown()| (where |this| is destroyed) makes sense.
+  base::AtExitManager at_exit_manager_;
+#endif  // !defined(COMPONENT_BUILD)
+
+  base::Thread ipc_thread_;
+
+  DISALLOW_COPY_AND_ASSIGN(IPCSupport);
+};
+
+std::unique_ptr<IPCSupport>& GetIPCSupport() {
+  static base::NoDestructor<std::unique_ptr<IPCSupport>> state;
+  return *state;
+}
+
+}  // namespace
+
 extern "C" {
 
 namespace {
 
 MojoResult InitializeImpl(const struct MojoInitializeOptions* options) {
+  mojo::core::Configuration config;
+  config.is_broker_process =
+      options && options->flags & MOJO_INITIALIZE_FLAG_AS_BROKER;
+  mojo::core::internal::g_configuration = config;
+
   mojo::core::InitializeCore();
+  GetIPCSupport() = std::make_unique<IPCSupport>();
+  return MOJO_RESULT_OK;
+}
+
+MojoResult ShutdownImpl(const struct MojoShutdownOptions* options) {
+  if (options && options->struct_size < sizeof(*options))
+    return MOJO_RESULT_INVALID_ARGUMENT;
+
+  std::unique_ptr<IPCSupport>& ipc_support = GetIPCSupport();
+  if (!ipc_support)
+    return MOJO_RESULT_FAILED_PRECONDITION;
+
+  ipc_support.reset();
   return MOJO_RESULT_OK;
 }
 
@@ -31,12 +98,19 @@ EXPORT_FROM_MOJO_CORE void MojoGetSystemThunks(MojoSystemThunks* thunks) {
   if (!g_thunks.size) {
     g_thunks = mojo::core::GetSystemThunks();
     g_thunks.Initialize = InitializeImpl;
+    g_thunks.Shutdown = ShutdownImpl;
   }
 
-  // Since this is the first version of the library, no valid system API
-  // implementation can request fewer thunks than we have available.
-  CHECK_GE(thunks->size, g_thunks.size);
-  memcpy(thunks, &g_thunks, g_thunks.size);
+  // Caller must provide a thunk structure at least large enough to hold Core
+  // ABI version 0. SetQuota is the first function introduced in ABI version 1.
+  CHECK_GE(thunks->size, offsetof(MojoSystemThunks, SetQuota));
+
+  // NOTE: This also overrites |thunks->size| with the actual size of our own
+  // thunks if smaller than the caller's. This informs the caller that we
+  // implement an older version of the ABI.
+  if (thunks->size > g_thunks.size)
+    thunks->size = g_thunks.size;
+  memcpy(thunks, &g_thunks, thunks->size);
 }
 
 }  // extern "C"

mojo/core/mojo_core_unittest.cc

@@ -2,11 +2,36 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+#include <stdint.h>
+
+#include <vector>
+
+#include "base/command_line.h"
+#include "base/process/launch.h"
+#include "base/test/multiprocess_test.h"
+#include "base/test/test_timeouts.h"
 #include "mojo/public/c/system/core.h"
+#include "mojo/public/cpp/platform/platform_channel.h"
+#include "mojo/public/cpp/platform/platform_handle.h"
+#include "mojo/public/cpp/system/invitation.h"
+#include "mojo/public/cpp/system/message_pipe.h"
+#include "mojo/public/cpp/system/wait.h"
 #include "testing/gtest/include/gtest/gtest.h"
+#include "testing/multiprocess_func_list.h"
 
 namespace {
 
+uint64_t kTestPipeName = 0;
+const char kTestMessage[] = "hai";
+const char kTestReply[] = "bai";
+
+std::string ReadMessageAsString(mojo::MessagePipeHandle handle) {
+  std::vector<uint8_t> data;
+  CHECK_EQ(MOJO_RESULT_OK, mojo::ReadMessageRaw(handle, &data, nullptr,
+                                                MOJO_READ_MESSAGE_FLAG_NONE));
+  return std::string(data.begin(), data.end());
+}
+
 TEST(MojoCoreTest, SanityCheck) {
   // Exercises some APIs against the mojo_core library and expects them to work
   // as intended.
@@ -36,4 +61,47 @@ TEST(MojoCoreTest, SanityCheck) {
   EXPECT_EQ(MOJO_RESULT_OK, MojoClose(a));
 }
 
+TEST(MojoCoreTest, BasicMultiprocess) {
+  base::CommandLine child_cmd(base::GetMultiProcessTestChildBaseCommandLine());
+  base::LaunchOptions options;
+
+  mojo::PlatformChannel channel;
+  channel.PrepareToPassRemoteEndpoint(&options, &child_cmd);
+  base::Process child_process = base::SpawnMultiProcessTestChild(
+      "BasicMultiprocessClientMain", child_cmd, options);
+  channel.RemoteProcessLaunchAttempted();
+
+  mojo::OutgoingInvitation invitation;
+  auto child_pipe = invitation.AttachMessagePipe(kTestPipeName);
+  mojo::OutgoingInvitation::Send(std::move(invitation), child_process.Handle(),
+                                 channel.TakeLocalEndpoint());
+
+  mojo::Wait(child_pipe.get(), MOJO_HANDLE_SIGNAL_READABLE);
+  EXPECT_EQ(kTestMessage, ReadMessageAsString(child_pipe.get()));
+
+  mojo::WriteMessageRaw(child_pipe.get(), kTestReply, sizeof(kTestReply) - 1,
+                        nullptr, 0, MOJO_WRITE_MESSAGE_FLAG_NONE);
+
+  int rv = -1;
+  ASSERT_TRUE(base::WaitForMultiprocessTestChildExit(
+      child_process, TestTimeouts::action_timeout(), &rv));
+  EXPECT_EQ(0, rv);
+}
+
+MULTIPROCESS_TEST_MAIN(BasicMultiprocessClientMain) {
+  auto endpoint = mojo::PlatformChannel::RecoverPassedEndpointFromCommandLine(
+      *base::CommandLine::ForCurrentProcess());
+  auto invitation = mojo::IncomingInvitation::Accept(std::move(endpoint));
+  auto parent_pipe = invitation.ExtractMessagePipe(kTestPipeName);
+
+  mojo::WriteMessageRaw(parent_pipe.get(), kTestMessage,
+                        sizeof(kTestMessage) - 1, nullptr, 0,
+                        MOJO_WRITE_MESSAGE_FLAG_NONE);
+
+  mojo::Wait(parent_pipe.get(), MOJO_HANDLE_SIGNAL_READABLE);
+  EXPECT_EQ(kTestReply, ReadMessageAsString(parent_pipe.get()));
+
+  return 0;
+}
+
 }  // namespace

mojo/core/run_all_core_unittests.cc

@@ -2,6 +2,7 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+#include "base/base_switches.h"
 #include "base/bind.h"
 #include "base/test/launcher/unit_test_launcher.h"
 #include "base/test/test_suite.h"
@@ -11,8 +12,21 @@
 int main(int argc, char** argv) {
   base::TestSuite test_suite(argc, argv);
 
-  CHECK_EQ(MOJO_RESULT_OK, MojoInitialize(nullptr));
-  return base::LaunchUnitTests(
+  MojoInitializeOptions options;
+  options.struct_size = sizeof(options);
+  options.flags = MOJO_INITIALIZE_FLAG_NONE;
+  options.mojo_core_path = NULL;
+  options.mojo_core_path_length = 0;
+  if (!base::CommandLine::ForCurrentProcess()->HasSwitch(
+          switches::kTestChildProcess)) {
+    options.flags = MOJO_INITIALIZE_FLAG_AS_BROKER;
+  }
+
+  CHECK_EQ(MOJO_RESULT_OK, MojoInitialize(&options));
+  int result = base::LaunchUnitTests(
       argc, argv,
       base::BindOnce(&base::TestSuite::Run, base::Unretained(&test_suite)));
+
+  CHECK_EQ(MOJO_RESULT_OK, MojoShutdown(nullptr));
+  return result;
 }

mojo/public/c/system/functions.h

@@ -27,14 +27,34 @@ extern "C" {
 // |options| may be null.
 //
 // Returns:
-//   |MOJO_RESULT_OK| if Mojo intiailization was successful.
-//   |MOJO_RESULT_INVALID_ARGUMENT| if |options| was null or invalid.
+//   |MOJO_RESULT_OK| if Mojo initialization was successful.
+//   |MOJO_RESULT_INVALID_ARGUMENT| if |options| was non-null and invalid.
 //   |MOJO_RESULT_FAILED_PRECONDITION| if |MojoInitialize()| was already called
 //       once or if the application already explicitly initialized a Mojo Core
 //       environment as an embedder.
 MOJO_SYSTEM_EXPORT MojoResult
 MojoInitialize(const struct MojoInitializeOptions* options);
 
+// Shuts down Mojo in the calling application.
+//
+// This should only be called if |MojoInitialize()| was also called at some
+// point in the calling process. It therefore only applies to consumers of Mojo
+// as a shared library, not Mojo Core embedders.
+//
+// |options| may be null.
+//
+// NOTE: It is NOT safe to attempt to call |MojoInitialize()| again (or any
+// other Mojo APIs, for that matter) after calling |MojoShutdown()|.
+//
+// Returns:
+//   |MOJO_RESULT_OK| if shutdown was successful.
+//   |MOJO_RESULT_INVALID_ARGUMENT| if |options| was non-null and invalid.
+//   |MOJO_RESULT_FAILED_PRECONDITION| if |MojoInitialize()| was never called.
+//   |MOJO_RESULT_UNIMPLEMENTED| if the caller is a Mojo Core embedder and is
+//       therefore not loading Mojo Core as a shared library.
+MOJO_SYSTEM_EXPORT MojoResult
+MojoShutdown(const struct MojoShutdownOptions* options);
+
 // Returns the time, in microseconds, since some undefined point in the past.
 // The values are only meaningful relative to other values that were obtained
 // from the same device without an intervening system restart. Such values are

mojo/public/c/system/tests/core_unittest_pure_c.c

@@ -14,6 +14,7 @@
 // core.h, since it's the most important one.
 #include "mojo/public/c/system/core.h"
 #include "mojo/public/c/system/macros.h"
+#include "mojo/public/c/system/thunks.h"
 
 // The joys of the C preprocessor....
 #define STRINGIFY(x) #x

mojo/public/c/system/thunks.cc

@@ -92,8 +92,21 @@ class CoreLibraryInitializer {
       library_path.emplace(kDefaultLibraryPathValue);
     }
 
+    // NOTE: |prefer_own_symbols| on POSIX implies that the library is loaded
+    // with RTLD_DEEPBIND, which is critical given that libmojo_core.so links
+    // against base's allocator shim. Essentially, this ensures that mojo_core
+    // internals get their own heap, and this is OK since heap pointer ownership
+    // is never passed across the ABI boundary.
     base::ScopedAllowBlocking allow_blocking;
-    library_.emplace(*library_path);
+    base::NativeLibraryOptions library_options;
+#if !defined(ADDRESS_SANITIZER) && !defined(THREAD_SANITIZER) && \
+    !defined(MEMORY_SANITIZER) && !defined(LEAK_SANITIZER)
+    // Sanitizer builds cannnot support RTLD_DEEPBIND, but they also disable
+    // allocator shims, so it's unnecessary there.
+    library_options.prefer_own_symbols = true;
+#endif
+    library_.emplace(base::LoadNativeLibraryWithOptions(
+        *library_path, library_options, nullptr));
     if (!application_provided_path) {
       CHECK(library_->is_valid())
           << "Unable to load the mojo_core library. Make sure the library is "
@@ -468,6 +481,10 @@ MojoResult MojoQueryQuota(MojoHandle handle,
   return INVOKE_THUNK(QueryQuota, handle, type, options, limit, usage);
 }
 
+MojoResult MojoShutdown(const MojoShutdownOptions* options) {
+  return INVOKE_THUNK(Shutdown, options);
+}
+
 }  // extern "C"
 
 void MojoEmbedderSetSystemThunks(const MojoSystemThunks* thunks) {

mojo/public/c/system/thunks.h

@@ -166,7 +166,7 @@ struct MojoSystemThunks {
       MojoHandle* mojo_handle);
   MojoResult (*UnwrapPlatformHandle)(
       MojoHandle mojo_handle,
-      const MojoUnwrapPlatformHandleOptions* options,
+      const struct MojoUnwrapPlatformHandleOptions* options,
       struct MojoPlatformHandle* platform_handle);
   MojoResult (*WrapPlatformSharedMemoryRegion)(
       const struct MojoPlatformHandle* platform_handles,
@@ -193,13 +193,13 @@ struct MojoSystemThunks {
       MojoHandle invitation_handle,
       const void* name,
       uint32_t name_num_bytes,
-      const MojoAttachMessagePipeToInvitationOptions* options,
+      const struct MojoAttachMessagePipeToInvitationOptions* options,
       MojoHandle* message_pipe_handle);
   MojoResult (*ExtractMessagePipeFromInvitation)(
       MojoHandle invitation_handle,
       const void* name,
       uint32_t name_num_bytes,
-      const MojoExtractMessagePipeFromInvitationOptions* options,
+      const struct MojoExtractMessagePipeFromInvitationOptions* options,
       MojoHandle* message_pipe_handle);
   MojoResult (*SendInvitation)(
       MojoHandle invitation_handle,
@@ -212,6 +212,9 @@ struct MojoSystemThunks {
       const struct MojoInvitationTransportEndpoint* transport_endpoint,
       const struct MojoAcceptInvitationOptions* options,
       MojoHandle* invitation_handle);
+
+  // Core ABI version 1 additions begin here.
+
   MojoResult (*SetQuota)(MojoHandle handle,
                          MojoQuotaType type,
                          uint64_t limit,
@@ -221,6 +224,10 @@ struct MojoSystemThunks {
                            const struct MojoQueryQuotaOptions* options,
                            uint64_t* limit,
                            uint64_t* usage);
+
+  // Core ABI version 2 additions begin here.
+
+  MojoResult (*Shutdown)(const struct MojoShutdownOptions* options);
 };
 #pragma pack(pop)
 

mojo/public/c/system/types.h

@@ -142,6 +142,13 @@ typedef uint32_t MojoInitializeFlags;
 // No flags.
 #define MOJO_INITIALIZE_FLAG_NONE ((MojoInitializeFlags)0)
 
+// The calling process will be initialized as the broker process for its IPC
+// network. Any connected graph of Mojo consumers must have exactly one broker
+// process. That process is always the first member of the network and it should
+// set this flag during initialization. Attempts to invite a broker process into
+// an existing network will always fail.
+#define MOJO_INITIALIZE_FLAG_AS_BROKER ((MojoInitializeFlags)1)
+
 // Options passed to |MojoInitialize()|.
 struct MOJO_ALIGNAS(8) MojoInitializeOptions {
   // The size of this structure, used for versioning.
@@ -160,6 +167,23 @@ struct MOJO_ALIGNAS(8) MojoInitializeOptions {
 MOJO_STATIC_ASSERT(sizeof(MojoInitializeOptions) == 24,
                    "MojoInitializeOptions has wrong size");
 
+// Flags passed to |MojoShutdown()| via |MojoShutdownOptions|.
+typedef uint32_t MojoShutdownFlags;
+
+// No flags.
+#define MOJO_SHUTDOWN_FLAG_NONE ((MojoShutdownFlags)0)
+
+// Options passed to |MojoShutdown()|.
+struct MOJO_ALIGNAS(8) MojoShutdownOptions {
+  // The size of this structure, used for versioning.
+  uint32_t struct_size;
+
+  // See |MojoShutdownFlags|.
+  MojoShutdownFlags flags;
+};
+MOJO_STATIC_ASSERT(sizeof(MojoShutdownOptions) == 8,
+                   "MojoShutdownOptions has wrong size");
+
 // |MojoHandleSignals|: Used to specify signals that can be watched for on a
 // handle (and which can be triggered), e.g., the ability to read or write to
 // the handle.