[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}

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

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_);
 

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

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_);
 

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

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

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_;
 

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_;
 

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);
   }
 

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);
   }
 }
 

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,

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

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_

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

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_;