Add a base::Feature for Mojo dispatch change

Adds a new base::Feature to control whether or not Mojo uses the more
granular task-per-message dispatch policy as opposed to the old batch
dispatch behavior.

This is disabled by default until M73 branch.

TBR=jam@chromium.org

Bug: 866708
Change-Id: I138a66c0d65106f44f153e02c1cf92d71c27d14d
Reviewed-on: https://chromium-review.googlesource.com/c/1433139
Reviewed-by: Ken Rockot <rockot@google.com>
Reviewed-by: John Abd-El-Malek <jam@chromium.org>
Commit-Queue: Ken Rockot <rockot@google.com>
Cr-Commit-Position: refs/heads/master@{#625514}

mojo/public/cpp/bindings/BUILD.gn

@@ -39,6 +39,8 @@ component("bindings_base") {
     "disconnect_reason.h",
     "enum_traits.h",
     "equals_traits.h",
+    "features.cc",
+    "features.h",
     "interface_data_view.h",
     "interface_id.h",
     "lib/array_internal.cc",

mojo/public/cpp/bindings/connector.h

@@ -270,10 +270,7 @@ class COMPONENT_EXPORT(MOJO_CPP_BINDINGS) Connector : public MessageReceiver {
   bool paused_ = false;
 
   // See |set_force_immediate_dispatch()|.
-  //
-  // TODO(https://866708): Temporarily reverted to batch dispatch mode until M73
-  // branch. Flip this back to |false| after branch.
-  bool force_immediate_dispatch_ = true;
+  bool force_immediate_dispatch_;
 
   // Messages which have been read off the pipe but not yet dispatched. This
   // exists so that we can schedule individual dispatch tasks for each read

mojo/public/cpp/bindings/features.cc

@@ -0,0 +1,28 @@
+// Copyright 2019 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "mojo/public/cpp/bindings/features.h"
+
+namespace mojo {
+namespace features {
+
+// Enables a task to be scheduled for each individual message dispatched to a
+// Mojo binding endpoint (or reply to an InterfacePtr).
+//
+// When disabled, dispatch happens eagerly in batch, so when a binding is
+// scheduled to dispatch messages, it fully flushes and dispatches all queued
+// messages within the extent of a single scheduler task.
+//
+// Enabling this feature allows for more fine-grained performance control
+// through the scheduler, but may initially cause some important edge cases to
+// regress in performance due to high-priority messages seeing increased
+// latency. Ideally we'd address these cases by giving the affected bindings
+// higher-priority TaskRunners.
+//
+// TODO(https://crbug.com/866708): Enable this by default after M73 branch.
+const base::Feature kTaskPerMessage{"MojoTaskPerMessage",
+                                    base::FEATURE_DISABLED_BY_DEFAULT};
+
+}  // namespace features
+}  // namespace mojo

mojo/public/cpp/bindings/features.h

@@ -0,0 +1,20 @@
+// Copyright 2019 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef MOJO_PUBLIC_CPP_BINDINGS_FEATURES_H_
+#define MOJO_PUBLIC_CPP_BINDINGS_FEATURES_H_
+
+#include "base/component_export.h"
+#include "base/feature_list.h"
+
+namespace mojo {
+namespace features {
+
+COMPONENT_EXPORT(MOJO_CPP_BINDINGS_BASE)
+extern const base::Feature kTaskPerMessage;
+
+}  // namespace features
+}  // namespace mojo
+
+#endif  // MOJO_PUBLIC_CPP_BINDINGS_FEATURES_H_

mojo/public/cpp/bindings/lib/connector.cc

@@ -17,6 +17,7 @@
 #include "base/synchronization/lock.h"
 #include "base/threading/sequence_local_storage_slot.h"
 #include "base/trace_event/trace_event.h"
+#include "mojo/public/cpp/bindings/features.h"
 #include "mojo/public/cpp/bindings/lib/may_auto_lock.h"
 #include "mojo/public/cpp/bindings/mojo_buildflags.h"
 #include "mojo/public/cpp/bindings/sync_handle_watcher.h"
@@ -146,6 +147,8 @@ Connector::Connector(ScopedMessagePipeHandle message_pipe,
     : message_pipe_(std::move(message_pipe)),
       task_runner_(std::move(runner)),
       error_(false),
+      force_immediate_dispatch_(
+          !base::FeatureList::IsEnabled(features::kTaskPerMessage)),
       outgoing_serialization_mode_(g_default_outgoing_serialization_mode),
       incoming_serialization_mode_(g_default_incoming_serialization_mode),
       nesting_observer_(RunLoopNestingObserver::GetForThread()),