[mojo-edk] Expose notification source to MojoWatch callbacks

This adds a flags argument to watch callbacks and exposes a flag
to indicate that the callback was invoked as a result of some
external process event (i.e. an incoming EDK IPC message). The
public C++ Watcher implementation is updated to take advantage
of this, effectively allowing us to dispatch to Mojo bindings
synchronously when they live on the IO thread and are servicing
messages from out-of-process.

BUG=590495

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

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

15 files changed, 113 insertions, 42 deletions

diff --git a/mojo/edk/system/core.cc b/mojo/edk/system/core.cc
index 49cb242..ca6105d 100644
--- a/mojo/edk/system/core.cc
+++ b/mojo/edk/system/core.cc
@@ -51,9 +51,10 @@ const uint64_t kUnknownPipeIdForDebug = 0x7f7f7f7f7f7f7f7fUL;
 void CallWatchCallback(MojoWatchCallback callback,
                        uintptr_t context,
                        MojoResult result,
-                       const HandleSignalsState& signals_state) {
-  callback(context, result,
-           static_cast<MojoHandleSignalsState>(signals_state));
+                       const HandleSignalsState& signals_state,
+                       MojoWatchNotificationFlags flags) {
+  callback(context, result, static_cast<MojoHandleSignalsState>(signals_state),
+      flags);
 }
 
 }  // namespace
diff --git a/mojo/edk/system/data_pipe_consumer_dispatcher.cc b/mojo/edk/system/data_pipe_consumer_dispatcher.cc
index 10a8069..f978f94 100644
--- a/mojo/edk/system/data_pipe_consumer_dispatcher.cc
+++ b/mojo/edk/system/data_pipe_consumer_dispatcher.cc
@@ -495,9 +495,7 @@ void DataPipeConsumerDispatcher::NotifyRead(uint32_t num_bytes) {
 }
 
 void DataPipeConsumerDispatcher::OnPortStatusChanged() {
-  // This has to be outside |lock_| because the watch callback can call data
-  // pipe functions which then try to acquire |lock_|.
-  RequestContext request_context;
+  DCHECK(RequestContext::current());
 
   base::AutoLock lock(lock_);
 
diff --git a/mojo/edk/system/data_pipe_producer_dispatcher.cc b/mojo/edk/system/data_pipe_producer_dispatcher.cc
index 9f6a72e..d8d8abf 100644
--- a/mojo/edk/system/data_pipe_producer_dispatcher.cc
+++ b/mojo/edk/system/data_pipe_producer_dispatcher.cc
@@ -473,9 +473,8 @@ void DataPipeProducerDispatcher::NotifyWrite(uint32_t num_bytes) {
 }
 
 void DataPipeProducerDispatcher::OnPortStatusChanged() {
-  // This has to be outside |lock_| because the watch callback can call data
-  // pipe functions which then try to acquire |lock_|.
-  RequestContext request_context;
+  DCHECK(RequestContext::current());
+
   base::AutoLock lock(lock_);
 
   // We stop observing the control port as soon it's transferred, but this can
diff --git a/mojo/edk/system/message_pipe_dispatcher.cc b/mojo/edk/system/message_pipe_dispatcher.cc
index 1722567..187176d 100644
--- a/mojo/edk/system/message_pipe_dispatcher.cc
+++ b/mojo/edk/system/message_pipe_dispatcher.cc
@@ -639,7 +639,7 @@ HandleSignalsState MessagePipeDispatcher::GetHandleSignalsStateNoLock() const {
 }
 
 void MessagePipeDispatcher::OnPortStatusChanged() {
-  RequestContext request_context;
+  DCHECK(RequestContext::current());
 
   base::AutoLock lock(signal_lock_);
 
diff --git a/mojo/edk/system/node_channel.cc b/mojo/edk/system/node_channel.cc
index 7e76fe4..1be5e47 100644
--- a/mojo/edk/system/node_channel.cc
+++ b/mojo/edk/system/node_channel.cc
@@ -12,6 +12,7 @@
 #include "base/location.h"
 #include "base/logging.h"
 #include "mojo/edk/system/channel.h"
+#include "mojo/edk/system/request_context.h"
 
 #if defined(OS_MACOSX) && !defined(OS_IOS)
 #include "mojo/edk/system/mach_port_relay.h"
@@ -384,6 +385,8 @@ void NodeChannel::OnChannelMessage(const void* payload,
                                    ScopedPlatformHandleVectorPtr handles) {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
+  RequestContext request_context(RequestContext::Source::SYSTEM);
+
 #if defined(OS_WIN)
   // If we receive handles from a known process, rewrite them to our own
   // process. This can occur when a privileged node receives handles directly
@@ -585,6 +588,8 @@ void NodeChannel::OnChannelMessage(const void* payload,
 void NodeChannel::OnChannelError() {
   DCHECK(io_task_runner_->RunsTasksOnCurrentThread());
 
+  RequestContext request_context(RequestContext::Source::SYSTEM);
+
   ShutDown();
   // |OnChannelError()| may cause |this| to be destroyed, but still need access
   // to the name name after that destruction. So may a copy of
diff --git a/mojo/edk/system/request_context.cc b/mojo/edk/system/request_context.cc
index c30d8bb..f4f70b5 100644
--- a/mojo/edk/system/request_context.cc
+++ b/mojo/edk/system/request_context.cc
@@ -18,7 +18,10 @@ base::LazyInstance<base::ThreadLocalPointer<RequestContext>>::Leaky
 
 }  // namespace
 
-RequestContext::RequestContext() : tls_context_(g_current_context.Pointer()){
+RequestContext::RequestContext() : RequestContext(Source::LOCAL_API_CALL) {}
+
+RequestContext::RequestContext(Source source)
+    : source_(source), tls_context_(g_current_context.Pointer()) {
   // We allow nested RequestContexts to exist as long as they aren't actually
   // used for anything.
   if (!tls_context_->Get())
@@ -26,23 +29,34 @@ RequestContext::RequestContext() : tls_context_(g_current_context.Pointer()){
 }
 
 RequestContext::~RequestContext() {
-  // NOTE: Callbacks invoked by this destructor are allowed to initiate new
-  // EDK requests on this thread, so we need to reset the thread-local context
-  // pointer before calling them.
-  if (IsCurrent())
+  if (IsCurrent()) {
+    // NOTE: Callbacks invoked by this destructor are allowed to initiate new
+    // EDK requests on this thread, so we need to reset the thread-local context
+    // pointer before calling them. We persist the original notification source
+    // since we're starting over at the bottom of the stack.
     tls_context_->Set(nullptr);
 
-  for (const WatchNotifyFinalizer& watch :
-      watch_notify_finalizers_.container()) {
-    // Establish a new request context for the extent of each callback to ensure
-    // that they don't themselves invoke callbacks while holding a watcher lock.
-    RequestContext request_context;
-    watch.watcher->MaybeInvokeCallback(watch.result, watch.state);
+    MojoWatchNotificationFlags flags = MOJO_WATCH_NOTIFICATION_FLAG_NONE;
+    if (source_ == Source::SYSTEM)
+      flags |= MOJO_WATCH_NOTIFICATION_FLAG_FROM_SYSTEM;
+
+    for (const WatchNotifyFinalizer& watch :
+        watch_notify_finalizers_.container()) {
+      // Establish a new request context for the extent of each callback to
+      // ensure that they don't themselves invoke callbacks while holding a
+      // watcher lock.
+      RequestContext request_context(source_);
+      watch.watcher->MaybeInvokeCallback(watch.result, watch.state, flags);
+    }
+
+    for (const scoped_refptr<Watcher>& watcher :
+            watch_cancel_finalizers_.container())
+      watcher->Cancel();
+  } else {
+    // It should be impossible for nested contexts to have finalizers.
+    DCHECK(watch_notify_finalizers_.container().empty());
+    DCHECK(watch_cancel_finalizers_.container().empty());
   }
-
-  for (const scoped_refptr<Watcher>& watcher :
-          watch_cancel_finalizers_.container())
-    watcher->Cancel();
 }
 
 // static
diff --git a/mojo/edk/system/request_context.h b/mojo/edk/system/request_context.h
index 56b8498..394e455 100644
--- a/mojo/edk/system/request_context.h
+++ b/mojo/edk/system/request_context.h
@@ -8,6 +8,7 @@
 #include "base/containers/stack_container.h"
 #include "base/macros.h"
 #include "mojo/edk/system/handle_signals_state.h"
+#include "mojo/edk/system/system_impl_export.h"
 #include "mojo/edk/system/watcher.h"
 
 namespace base {
@@ -27,14 +28,25 @@ namespace edk {
 // for any reason. Therefore it is important to always use
 // |RequestContext::current()| rather than referring to any local instance
 // directly.
-class RequestContext {
+class MOJO_SYSTEM_IMPL_EXPORT RequestContext {
  public:
+  // Identifies the source of the current stack frame's RequestContext.
+  enum class Source {
+    LOCAL_API_CALL,
+    SYSTEM,
+  };
+
+  // Constructs a RequestContext with a LOCAL_API_CALL Source.
   RequestContext();
+
+  explicit RequestContext(Source source);
   ~RequestContext();
 
   // Returns the current thread-local RequestContext.
   static RequestContext* current();
 
+  Source source() const { return source_; }
+
   // Adds a finalizer to this RequestContext corresponding to a watch callback
   // which should be triggered in response to some handle state change. If
   // the Watcher hasn't been cancelled by the time this RequestContext is
@@ -74,6 +86,8 @@ class RequestContext {
   using WatchCancelFinalizerList =
       base::StackVector<scoped_refptr<Watcher>, kStaticWatchFinalizersCapacity>;
 
+  const Source source_;
+
   WatchNotifyFinalizerList watch_notify_finalizers_;
   WatchCancelFinalizerList watch_cancel_finalizers_;
 
diff --git a/mojo/edk/system/wait_set_dispatcher_unittest.cc b/mojo/edk/system/wait_set_dispatcher_unittest.cc
index a90a356..2a42be1 100644
--- a/mojo/edk/system/wait_set_dispatcher_unittest.cc
+++ b/mojo/edk/system/wait_set_dispatcher_unittest.cc
@@ -14,6 +14,7 @@
 #include "mojo/edk/embedder/embedder_internal.h"
 #include "mojo/edk/system/core.h"
 #include "mojo/edk/system/message_pipe_dispatcher.h"
+#include "mojo/edk/system/request_context.h"
 #include "mojo/edk/system/test_utils.h"
 #include "mojo/edk/system/waiter.h"
 #include "testing/gtest/include/gtest/gtest.h"
@@ -78,6 +79,10 @@ class WaitSetDispatcherTest : public ::testing::Test {
   scoped_refptr<MessagePipeDispatcher> dispatcher1_;
 
  private:
+  // We keep an active RequestContext for the duration of each test. It's unused
+  // since these tests don't rely on the MojoWatch API.
+  const RequestContext request_context_;
+
   static uint64_t pipe_id_generator_;
   DispatcherVector dispatchers_to_close_;
 
diff --git a/mojo/edk/system/watch_unittest.cc b/mojo/edk/system/watch_unittest.cc
index f60c8a8..f6a748b 100644
--- a/mojo/edk/system/watch_unittest.cc
+++ b/mojo/edk/system/watch_unittest.cc
@@ -17,7 +17,8 @@ namespace {
 
 void IgnoreResult(uintptr_t context,
                   MojoResult result,
-                  MojoHandleSignalsState signals) {
+                  MojoHandleSignalsState signals,
+                  MojoWatchNotificationFlags flags) {
 }
 
 // A test helper class for watching a handle. The WatchHelper instance is used
@@ -56,11 +57,13 @@ class WatchHelper {
  private:
   static void OnNotify(uintptr_t context,
                        MojoResult result,
-                       MojoHandleSignalsState state) {
+                       MojoHandleSignalsState state,
+                       MojoWatchNotificationFlags flags) {
     WatchHelper* watcher = reinterpret_cast<WatchHelper*>(context);
     CHECK(watcher->watching_);
     if (result == MOJO_RESULT_CANCELLED)
       watcher->watching_ = false;
+    CHECK_EQ(flags, MOJO_WATCH_NOTIFICATION_FLAG_NONE);
     watcher->callback_(result, state);
   }
 
diff --git a/mojo/edk/system/watcher.cc b/mojo/edk/system/watcher.cc
index 4bc9dbb..25c2276 100644
--- a/mojo/edk/system/watcher.cc
+++ b/mojo/edk/system/watcher.cc
@@ -15,12 +15,13 @@ Watcher::Watcher(MojoHandleSignals signals, const WatchCallback& callback)
 }
 
 void Watcher::MaybeInvokeCallback(MojoResult result,
-                                  const HandleSignalsState& state) {
+                                  const HandleSignalsState& state,
+                                  MojoWatchNotificationFlags flags) {
   base::AutoLock lock(lock_);
   if (is_cancelled_)
     return;
 
-  callback_.Run(result, state);
+  callback_.Run(result, state, flags);
 }
 
 void Watcher::NotifyForStateChange(const HandleSignalsState& signals_state) {
@@ -29,9 +30,9 @@ void Watcher::NotifyForStateChange(const HandleSignalsState& signals_state) {
     request_context->AddWatchNotifyFinalizer(
         make_scoped_refptr(this), MOJO_RESULT_OK, signals_state);
   } else if (!signals_state.can_satisfy(signals_)) {
-    request_context->AddWatchNotifyFinalizer(make_scoped_refptr(this),
-                                             MOJO_RESULT_FAILED_PRECONDITION,
-                                             signals_state);
+    request_context->AddWatchNotifyFinalizer(
+        make_scoped_refptr(this), MOJO_RESULT_FAILED_PRECONDITION,
+        signals_state);
   }
 }
 
diff --git a/mojo/edk/system/watcher.h b/mojo/edk/system/watcher.h
index 119355e..b6dc2e4 100644
--- a/mojo/edk/system/watcher.h
+++ b/mojo/edk/system/watcher.h
@@ -29,8 +29,9 @@ struct HandleSignalsState;
 // its cancellation, which is why it's ref-counted.
 class Watcher : public base::RefCountedThreadSafe<Watcher> {
  public:
-  using WatchCallback =
-      base::Callback<void(MojoResult, const HandleSignalsState&)>;
+  using WatchCallback = base::Callback<void(MojoResult,
+                                            const HandleSignalsState&,
+                                            MojoWatchNotificationFlags)>;
 
   // Constructs a new Watcher which watches for |signals| to be satisfied on a
   // handle and which invokes |callback| either when one such signal is
@@ -39,7 +40,9 @@ class Watcher : public base::RefCountedThreadSafe<Watcher> {
 
   // Runs the Watcher's callback with the given arguments if it hasn't been
   // cancelled yet.
-  void MaybeInvokeCallback(MojoResult result, const HandleSignalsState& state);
+  void MaybeInvokeCallback(MojoResult result,
+                           const HandleSignalsState& state,
+                           MojoWatchNotificationFlags flags);
 
   // Notifies the Watcher of a state change. This may result in the Watcher
   // adding a finalizer to the current RequestContext to invoke its callback,
diff --git a/mojo/public/c/system/functions.h b/mojo/public/c/system/functions.h
index 6c7e2df..bd4daed 100644
--- a/mojo/public/c/system/functions.h
+++ b/mojo/public/c/system/functions.h
@@ -24,7 +24,8 @@ extern "C" {
 // documentation for |MojoWatch()| for more details.
 typedef void (*MojoWatchCallback)(uintptr_t context,
                                   MojoResult result,
-                                  struct MojoHandleSignalsState signals_state);
+                                  struct MojoHandleSignalsState signals_state,
+                                  MojoWatchNotificationFlags flags);
 
 // Note: Pointer parameters that are labelled "optional" may be null (at least
 // under some circumstances). Non-const pointer parameters are also labeled
diff --git a/mojo/public/c/system/types.h b/mojo/public/c/system/types.h
index 4574d74..6d21b12 100644
--- a/mojo/public/c/system/types.h
+++ b/mojo/public/c/system/types.h
@@ -182,4 +182,24 @@ struct MOJO_ALIGNAS(4) MojoHandleSignalsState {
 MOJO_STATIC_ASSERT(sizeof(MojoHandleSignalsState) == 8,
                    "MojoHandleSignalsState has wrong size");
 
+// |MojoWatchNotificationFlags|: Passed to a callback invoked as a result of
+// signals being raised on a handle watched by |MojoWatch()|. May take the
+// following values:
+//   |MOJO_WATCH_NOTIFICATION_FLAG_FROM_SYSTEM| - The callback is being invoked
+//       as a result of a system-level event rather than a direct API call from
+//       user code. This may be used as an indication that user code is safe to
+//       call without fear of reentry.
+
+typedef uint32_t MojoWatchNotificationFlags;
+
+#ifdef __cplusplus
+const MojoWatchNotificationFlags MOJO_WATCH_NOTIFICATION_FLAG_NONE = 0;
+const MojoWatchNotificationFlags MOJO_WATCH_NOTIFICATION_FLAG_FROM_SYSTEM =
+    1 << 0;
+#else
+#define MOJO_WATCH_NOTIFICATION_FLAG_NONE ((MojoWatchNotificationFlags)0)
+#define MOJO_WATCH_NOTIFICATION_FLAG_FROM_SYSTEM \
+    ((MojoWatchNotificationFlags)1 << 0);
+#endif
+
 #endif  // MOJO_PUBLIC_C_SYSTEM_TYPES_H_
diff --git a/mojo/public/cpp/system/watcher.cc b/mojo/public/cpp/system/watcher.cc
index 5723533..a3a00a9 100644
--- a/mojo/public/cpp/system/watcher.cc
+++ b/mojo/public/cpp/system/watcher.cc
@@ -121,16 +121,22 @@ void Watcher::OnHandleReady(MojoResult result) {
 // static
 void Watcher::CallOnHandleReady(uintptr_t context,
                                 MojoResult result,
-                                MojoHandleSignalsState signals_state) {
+                                MojoHandleSignalsState signals_state,
+                                MojoWatchNotificationFlags flags) {
   // NOTE: It is safe to assume the Watcher still exists because this callback
   // will never be run after the Watcher's destructor.
   //
   // TODO: Maybe we should also expose |signals_state| throught he Watcher API.
   // Current HandleWatcher users have no need for it, so it's omitted here.
   Watcher* watcher = reinterpret_cast<Watcher*>(context);
-  watcher->task_runner_->PostTask(
-      FROM_HERE,
-      base::Bind(&Watcher::OnHandleReady, watcher->weak_self_, result));
+  if ((flags & MOJO_WATCH_NOTIFICATION_FLAG_FROM_SYSTEM) &&
+      watcher->task_runner_->RunsTasksOnCurrentThread()) {
+    watcher->OnHandleReady(result);
+  } else {
+    watcher->task_runner_->PostTask(
+        FROM_HERE,
+        base::Bind(&Watcher::OnHandleReady, watcher->weak_self_, result));
+  }
 }
 
 }  // namespace mojo
diff --git a/mojo/public/cpp/system/watcher.h b/mojo/public/cpp/system/watcher.h
index cd068aa..bc82391 100644
--- a/mojo/public/cpp/system/watcher.h
+++ b/mojo/public/cpp/system/watcher.h
@@ -84,7 +84,8 @@ class Watcher {
 
   static void CallOnHandleReady(uintptr_t context,
                                 MojoResult result,
-                                MojoHandleSignalsState signals_state);
+                                MojoHandleSignalsState signals_state,
+                                MojoWatchNotificationFlags flags);
 
   base::ThreadChecker thread_checker_;