Add handle waiting to StashBackend.

This change adds support for monitoring stashed handles for being
readable. This will be used to relaunch suspended extension processes
when a persistent service sends a message or data to its client in the
suspended extension process.

BUG=389016

Review URL: https://codereview.chromium.org/648853007

Cr-Commit-Position: refs/heads/master@{#314989}

extensions/BUILD.gn

@@ -292,6 +292,7 @@ if (false) {
       "//third_party/mojo/src/mojo/edk/js",
       "//third_party/mojo/src/mojo/edk/system",
       "//third_party/mojo/src/mojo/public/cpp/bindings",
+      "//third_party/mojo/src/mojo/public/interfaces/application",
     ]
 
     if (is_win) {

extensions/browser/mojo/stash_backend.cc

@@ -4,27 +4,94 @@
 
 #include "extensions/browser/mojo/stash_backend.h"
 
+#include "base/bind.h"
+#include "third_party/mojo/src/mojo/public/cpp/bindings/strong_binding.h"
+#include "third_party/mojo/src/mojo/public/cpp/environment/async_waiter.h"
+
 namespace extensions {
+namespace {
 
 // An implementation of StashService that forwards calls to a StashBackend.
-class StashServiceImpl : public mojo::InterfaceImpl<StashService> {
+class StashServiceImpl : public StashService {
  public:
-  explicit StashServiceImpl(base::WeakPtr<StashBackend> backend);
+  StashServiceImpl(mojo::InterfaceRequest<StashService> request,
+                   base::WeakPtr<StashBackend> backend);
   ~StashServiceImpl() override;
 
-  // mojo::InterfaceImpl<StashService> overrides.
+  // StashService overrides.
   void AddToStash(mojo::Array<StashedObjectPtr> stash) override;
   void RetrieveStash(
       const mojo::Callback<void(mojo::Array<StashedObjectPtr> stash)>& callback)
       override;
 
  private:
+  mojo::StrongBinding<StashService> binding_;
   base::WeakPtr<StashBackend> backend_;
 
   DISALLOW_COPY_AND_ASSIGN(StashServiceImpl);
 };
 
-StashBackend::StashBackend() : weak_factory_(this) {
+StashServiceImpl::StashServiceImpl(mojo::InterfaceRequest<StashService> request,
+                                   base::WeakPtr<StashBackend> backend)
+    : binding_(this, request.Pass()), backend_(backend) {
+}
+
+StashServiceImpl::~StashServiceImpl() {
+}
+
+void StashServiceImpl::AddToStash(
+    mojo::Array<StashedObjectPtr> stashed_objects) {
+  if (!backend_)
+    return;
+  backend_->AddToStash(stashed_objects.Pass());
+}
+
+void StashServiceImpl::RetrieveStash(
+    const mojo::Callback<void(mojo::Array<StashedObjectPtr>)>& callback) {
+  if (!backend_) {
+    callback.Run(mojo::Array<StashedObjectPtr>(0));
+    return;
+  }
+  callback.Run(backend_->RetrieveStash());
+}
+
+}  // namespace
+
+// A stash entry for a stashed object. This handles notifications if a handle
+// within the stashed object is readable.
+class StashBackend::StashEntry {
+ public:
+  // Construct a StashEntry for |stashed_object|. If |on_handle_readable| is
+  // non-null, it will be invoked when any handle on |stashed_object| is
+  // readable.
+  StashEntry(StashedObjectPtr stashed_object,
+             const base::Closure& on_handle_readable);
+  ~StashEntry();
+
+  // Returns the stashed object.
+  StashedObjectPtr Release();
+
+  // Cancels notifications for handles becoming readable.
+  void CancelHandleNotifications();
+
+ private:
+  // Invoked when a handle within |stashed_object_| is readable.
+  void OnHandleReady(MojoResult result);
+
+  // The waiters that are waiting for handles to be readable.
+  ScopedVector<mojo::AsyncWaiter> waiters_;
+
+  StashedObjectPtr stashed_object_;
+
+  // If non-null, a callback to call when a handle contained within
+  // |stashed_object_| is readable.
+  const base::Closure on_handle_readable_;
+};
+
+StashBackend::StashBackend(const base::Closure& on_handle_readable)
+    : on_handle_readable_(on_handle_readable),
+      has_notified_(false),
+      weak_factory_(this) {
 }
 
 StashBackend::~StashBackend() {
@@ -32,42 +99,69 @@ StashBackend::~StashBackend() {
 
 void StashBackend::AddToStash(mojo::Array<StashedObjectPtr> stashed_objects) {
   for (size_t i = 0; i < stashed_objects.size(); i++) {
-    stashed_objects_.push_back(stashed_objects[i].Pass());
+    stashed_objects_.push_back(
+        new StashEntry(stashed_objects[i].Pass(),
+                       has_notified_ ? base::Closure()
+                                     : base::Bind(&StashBackend::OnHandleReady,
+                                                  weak_factory_.GetWeakPtr())));
   }
 }
 
 mojo::Array<StashedObjectPtr> StashBackend::RetrieveStash() {
-  if (stashed_objects_.is_null())
-    stashed_objects_.resize(0);
-  return stashed_objects_.Pass();
+  has_notified_ = false;
+  mojo::Array<StashedObjectPtr> result(0);
+  for (auto& entry : stashed_objects_) {
+    result.push_back(entry->Release());
+  }
+  stashed_objects_.clear();
+  return result.Pass();
 }
 
 void StashBackend::BindToRequest(mojo::InterfaceRequest<StashService> request) {
-  mojo::BindToRequest(new StashServiceImpl(weak_factory_.GetWeakPtr()),
-                      &request);
+  new StashServiceImpl(request.Pass(), weak_factory_.GetWeakPtr());
 }
 
-StashServiceImpl::StashServiceImpl(base::WeakPtr<StashBackend> backend)
-    : backend_(backend) {
+void StashBackend::OnHandleReady() {
+  DCHECK(!has_notified_);
+  has_notified_ = true;
+  for (auto& entry : stashed_objects_) {
+    entry->CancelHandleNotifications();
+  }
+  if (!on_handle_readable_.is_null())
+    on_handle_readable_.Run();
 }
 
-StashServiceImpl::~StashServiceImpl() {
+StashBackend::StashEntry::StashEntry(StashedObjectPtr stashed_object,
+                                     const base::Closure& on_handle_readable)
+    : stashed_object_(stashed_object.Pass()),
+      on_handle_readable_(on_handle_readable) {
+  if (on_handle_readable_.is_null() || !stashed_object_->monitor_handles)
+    return;
+
+  for (size_t i = 0; i < stashed_object_->stashed_handles.size(); i++) {
+    waiters_.push_back(new mojo::AsyncWaiter(
+        stashed_object_->stashed_handles[i].get(), MOJO_HANDLE_SIGNAL_READABLE,
+        base::Bind(&StashBackend::StashEntry::OnHandleReady,
+                   base::Unretained(this))));
+  }
 }
 
-void StashServiceImpl::AddToStash(
-    mojo::Array<StashedObjectPtr> stashed_objects) {
-  if (!backend_)
-    return;
-  backend_->AddToStash(stashed_objects.Pass());
+StashBackend::StashEntry::~StashEntry() {
 }
 
-void StashServiceImpl::RetrieveStash(
-    const mojo::Callback<void(mojo::Array<StashedObjectPtr>)>& callback) {
-  if (!backend_) {
-    callback.Run(mojo::Array<StashedObjectPtr>(0));
+StashedObjectPtr StashBackend::StashEntry::Release() {
+  waiters_.clear();
+  return stashed_object_.Pass();
+}
+
+void StashBackend::StashEntry::CancelHandleNotifications() {
+  waiters_.clear();
+}
+
+void StashBackend::StashEntry::OnHandleReady(MojoResult result) {
+  if (result != MOJO_RESULT_OK)
     return;
-  }
-  callback.Run(backend_->RetrieveStash());
+  on_handle_readable_.Run();
 }
 
 }  // namespace extensions

extensions/browser/mojo/stash_backend.h

@@ -5,9 +5,8 @@
 #ifndef EXTENSIONS_BROWSER_MOJO_STASH_BACKEND_H_
 #define EXTENSIONS_BROWSER_MOJO_STASH_BACKEND_H_
 
-#include <vector>
-
-#include "base/memory/linked_ptr.h"
+#include "base/callback.h"
+#include "base/memory/scoped_vector.h"
 #include "base/memory/weak_ptr.h"
 #include "extensions/common/mojo/stash.mojom.h"
 #include "third_party/mojo/src/mojo/public/cpp/bindings/interface_request.h"
@@ -17,7 +16,7 @@ namespace extensions {
 // A backend that provides access to StashService for a single extension.
 class StashBackend {
  public:
-  StashBackend();
+  explicit StashBackend(const base::Closure& on_handle_readable);
   ~StashBackend();
 
   // Creates a StashService that forwards calls to this StashBackend and bind it
@@ -32,8 +31,19 @@ class StashBackend {
   mojo::Array<StashedObjectPtr> RetrieveStash();
 
  private:
+  class StashEntry;
+
+  // Invoked when a handle is readable.
+  void OnHandleReady();
+
   // The objects that have been stashed.
-  mojo::Array<StashedObjectPtr> stashed_objects_;
+  ScopedVector<StashEntry> stashed_objects_;
+
+  // The callback to call when a handle is readable.
+  const base::Closure on_handle_readable_;
+
+  // Whether a handle has become readable since the last RetrieveStash() call.
+  bool has_notified_;
 
   base::WeakPtrFactory<StashBackend> weak_factory_;
 

extensions/browser/mojo/stash_backend_unittest.cc

@@ -7,14 +7,38 @@
 #include "base/run_loop.h"
 #include "extensions/browser/mojo/stash_backend.h"
 #include "testing/gtest/include/gtest/gtest.h"
+#include "third_party/mojo/src/mojo/public/interfaces/application/service_provider.mojom.h"
 
 namespace extensions {
+namespace {
+
+// Create a data pipe, write some data to the producer handle and return the
+// consumer handle.
+mojo::ScopedHandle CreateReadableHandle() {
+  mojo::ScopedDataPipeConsumerHandle consumer_handle;
+  mojo::ScopedDataPipeProducerHandle producer_handle;
+  MojoCreateDataPipeOptions options = {
+      sizeof(options), MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE, 1, 1,
+  };
+  MojoResult result =
+      mojo::CreateDataPipe(&options, &producer_handle, &consumer_handle);
+  EXPECT_EQ(MOJO_RESULT_OK, result);
+  uint32_t num_bytes = 1;
+  result = mojo::WriteDataRaw(producer_handle.get(), "a", &num_bytes,
+                              MOJO_WRITE_DATA_FLAG_NONE);
+  EXPECT_EQ(MOJO_RESULT_OK, result);
+  EXPECT_EQ(1u, num_bytes);
+  return mojo::ScopedHandle::From(consumer_handle.Pass());
+}
+
+}  // namespace
 
 class StashServiceTest : public testing::Test, public mojo::ErrorHandler {
  public:
   enum Event {
     EVENT_NONE,
     EVENT_STASH_RETRIEVED,
+    EVENT_HANDLE_READY,
   };
 
   StashServiceTest() {}
@@ -22,9 +46,11 @@ class StashServiceTest : public testing::Test, public mojo::ErrorHandler {
   void SetUp() override {
     expecting_error_ = false;
     expected_event_ = EVENT_NONE;
-    stash_backend_.reset(new StashBackend);
+    stash_backend_.reset(new StashBackend(base::Bind(
+        &StashServiceTest::OnHandleReadyToRead, base::Unretained(this))));
     stash_backend_->BindToRequest(mojo::GetProxy(&stash_service_));
     stash_service_.set_error_handler(this);
+    handles_ready_ = 0;
   }
 
   void OnConnectionError() override { FAIL() << "Unexpected connection error"; }
@@ -55,6 +81,11 @@ class StashServiceTest : public testing::Test, public mojo::ErrorHandler {
       stop_run_loop_.Run();
   }
 
+  void OnHandleReadyToRead() {
+    handles_ready_++;
+    EventReceived(EVENT_HANDLE_READY);
+  }
+
  protected:
   base::MessageLoop message_loop_;
   base::Closure stop_run_loop_;
@@ -62,6 +93,7 @@ class StashServiceTest : public testing::Test, public mojo::ErrorHandler {
   Event expected_event_;
   bool expecting_error_;
   mojo::InterfacePtr<StashService> stash_service_;
+  int handles_ready_;
 
  private:
   DISALLOW_COPY_AND_ASSIGN(StashServiceTest);
@@ -147,6 +179,94 @@ TEST_F(StashServiceTest, RetrieveWithoutStashing) {
   EXPECT_EQ(0u, stashed_objects.size());
 }
 
+TEST_F(StashServiceTest, NotifyOnReadableHandle) {
+  mojo::Array<StashedObjectPtr> stash_entries;
+  StashedObjectPtr stashed_object(StashedObject::New());
+  stashed_object->id = "test type";
+  stashed_object->data.push_back(0);
+  stashed_object->monitor_handles = true;
+  mojo::ServiceProviderPtr service_provider;
+
+  // Stash the ServiceProvider request. When we make a call on
+  // |service_provider|, the stashed handle will become readable.
+  stashed_object->stashed_handles.push_back(mojo::ScopedHandle::From(
+      mojo::GetProxy(&service_provider).PassMessagePipe()));
+
+  stash_entries.push_back(stashed_object.Pass());
+  stash_service_->AddToStash(stash_entries.Pass());
+
+  mojo::MessagePipe pipe;
+  service_provider->ConnectToService("", pipe.handle0.Pass());
+
+  WaitForEvent(EVENT_HANDLE_READY);
+  EXPECT_EQ(1, handles_ready_);
+}
+
+TEST_F(StashServiceTest, NotifyOnReadableDataPipeHandle) {
+  mojo::Array<StashedObjectPtr> stash_entries;
+  StashedObjectPtr stashed_object(StashedObject::New());
+  stashed_object->id = "test type";
+  stashed_object->monitor_handles = true;
+
+  MojoCreateDataPipeOptions options = {
+      sizeof(options), MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE, 1, 1,
+  };
+  mojo::ScopedDataPipeConsumerHandle consumer_handle;
+  mojo::ScopedDataPipeProducerHandle producer_handle;
+  uint32_t num_bytes = 1;
+  MojoResult result =
+      mojo::CreateDataPipe(&options, &producer_handle, &consumer_handle);
+  ASSERT_EQ(MOJO_RESULT_OK, result);
+  result = mojo::WriteDataRaw(producer_handle.get(), "a", &num_bytes,
+                              MOJO_WRITE_DATA_FLAG_NONE);
+  ASSERT_EQ(MOJO_RESULT_OK, result);
+  ASSERT_EQ(1u, num_bytes);
+  stashed_object->stashed_handles.push_back(
+      mojo::ScopedHandle::From(producer_handle.Pass()));
+  stashed_object->stashed_handles.push_back(
+      mojo::ScopedHandle::From(consumer_handle.Pass()));
+  stashed_object->data.push_back(1);
+
+  stash_entries.push_back(stashed_object.Pass());
+  stash_service_->AddToStash(stash_entries.Pass());
+  WaitForEvent(EVENT_HANDLE_READY);
+  EXPECT_EQ(1, handles_ready_);
+}
+
+TEST_F(StashServiceTest, NotifyOncePerStashOnReadableHandles) {
+  mojo::Array<StashedObjectPtr> stash_entries;
+  StashedObjectPtr stashed_object(StashedObject::New());
+  stashed_object->id = "test type";
+  stashed_object->data.push_back(1);
+  stashed_object->monitor_handles = true;
+  stashed_object->stashed_handles.push_back(CreateReadableHandle());
+  stashed_object->stashed_handles.push_back(CreateReadableHandle());
+  stash_entries.push_back(stashed_object.Pass());
+  stashed_object = StashedObject::New();
+  stashed_object->id = "another test type";
+  stashed_object->data.push_back(2);
+  stashed_object->monitor_handles = true;
+  stashed_object->stashed_handles.push_back(CreateReadableHandle());
+  stashed_object->stashed_handles.push_back(CreateReadableHandle());
+  stash_entries.push_back(stashed_object.Pass());
+  stash_service_->AddToStash(stash_entries.Pass());
+  WaitForEvent(EVENT_HANDLE_READY);
+  EXPECT_EQ(1, handles_ready_);
+
+  stashed_object = StashedObject::New();
+  stashed_object->id = "yet another test type";
+  stashed_object->data.push_back(3);
+  stashed_object->monitor_handles = true;
+  stashed_object->stashed_handles.push_back(CreateReadableHandle());
+  stashed_object->stashed_handles.push_back(CreateReadableHandle());
+  stash_entries.push_back(stashed_object.Pass());
+  stash_service_->AddToStash(stash_entries.Pass());
+
+  stash_service_->AddToStash(RetrieveStash());
+  WaitForEvent(EVENT_HANDLE_READY);
+  EXPECT_EQ(2, handles_ready_);
+}
+
 // Test that a stash service discards stashed objects when the backend no longer
 // exists.
 TEST_F(StashServiceTest, ServiceWithDeletedBackend) {

extensions/common/mojo/stash.mojom

@@ -17,6 +17,10 @@ struct StashedObject {
 
   // The handles contained within the serialization struct.
   array<handle> stashed_handles;
+
+  // Whether to monitor |stashed_handles| and relaunch the client when a handle
+  // becomes readable.
+  bool monitor_handles = false;
 };
 
 interface StashService {

extensions/extensions_tests.gyp

@@ -27,6 +27,7 @@
         '../third_party/mojo/mojo_edk.gyp:mojo_js_lib',
         '../third_party/mojo/mojo_edk.gyp:mojo_system_impl',
         '../third_party/mojo/mojo_public.gyp:mojo_cpp_bindings',
+        '../third_party/mojo/mojo_public.gyp:mojo_application_bindings',
         'common/api/api.gyp:cast_channel_proto',
         'extensions.gyp:extensions_common',
         'extensions.gyp:extensions_renderer',