Mojo: Add a feature flag to allow sampling message pipe unread message counts.

To test the feature and parameters, invoke chrome with parameters like:
--enable-features="MojoRecordUnreadMessageCount<Study" \
  --force-fieldtrials=Study/Group1 \
  --force-fieldtrial-params=Study.Group1:SampleRate/1/QuotaValue/10

Bug: 1003391
Change-Id: I395047c17fc54afb4caefca5d0788f541a1ddcfc
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1779023
Commit-Queue: Sigurður Ásgeirsson <siggi@chromium.org>
Reviewed-by: Robert Kaplow <rkaplow@chromium.org>
Reviewed-by: Ken Rockot <rockot@google.com>
Cr-Commit-Position: refs/heads/master@{#697346}

mojo/core/node_controller.cc

@@ -14,7 +14,6 @@
 #include "base/logging.h"
 #include "base/macros.h"
 #include "base/message_loop/message_loop_current.h"
-#include "base/metrics/histogram_macros.h"
 #include "base/process/process_handle.h"
 #include "base/rand_util.h"
 #include "base/time/time.h"

mojo/core/user_message_impl.cc

@@ -9,7 +9,6 @@
 
 #include "base/atomicops.h"
 #include "base/memory/ptr_util.h"
-#include "base/metrics/histogram_macros_local.h"
 #include "base/no_destructor.h"
 #include "base/numerics/safe_conversions.h"
 #include "base/numerics/safe_math.h"

mojo/public/cpp/bindings/connector.h

@@ -18,6 +18,7 @@
 #include "base/optional.h"
 #include "base/sequence_checker.h"
 #include "base/sequenced_task_runner.h"
+#include "mojo/public/c/system/quota.h"
 #include "mojo/public/cpp/bindings/connection_group.h"
 #include "mojo/public/cpp/bindings/message.h"
 #include "mojo/public/cpp/bindings/sequence_local_sync_event_watcher.h"
@@ -307,6 +308,11 @@ class COMPONENT_EXPORT(MOJO_CPP_BINDINGS) Connector : public MessageReceiver {
 
   SEQUENCE_CHECKER(sequence_checker_);
 
+  // If this instance was selected for unread message measurement, contains
+  // the max send quota usage seen so far. If this instance was not selected
+  // contains MOJO_QUOTA_LIMIT_NONE as a sentinel value.
+  uint64_t max_unread_message_quota_used_ = MOJO_QUOTA_LIMIT_NONE;
+
   base::Lock connected_lock_;
   bool connected_ = true;
 

mojo/public/cpp/bindings/features.cc

@@ -22,5 +22,12 @@ namespace features {
 const base::Feature kTaskPerMessage{"MojoTaskPerMessage",
                                     base::FEATURE_ENABLED_BY_DEFAULT};
 
+// Enables measurement of MessageChannel unread message counts. When enabled, a
+// small random selection of Connectors enable the unread message count quota
+// on their associated message pipe, and record the highest unread message count
+// seen during the Connector's lifetime.
+const base::Feature kMojoRecordUnreadMessageCount{
+    "MojoRecordUnreadMessageCount", base::FEATURE_DISABLED_BY_DEFAULT};
+
 }  // namespace features
 }  // namespace mojo

mojo/public/cpp/bindings/features.h

@@ -14,6 +14,9 @@ namespace features {
 COMPONENT_EXPORT(MOJO_CPP_BINDINGS_BASE)
 extern const base::Feature kTaskPerMessage;
 
+COMPONENT_EXPORT(MOJO_CPP_BINDINGS_BASE)
+extern const base::Feature kMojoRecordUnreadMessageCount;
+
 }  // namespace features
 }  // namespace mojo
 

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

@@ -12,7 +12,10 @@
 #include "base/macros.h"
 #include "base/memory/ptr_util.h"
 #include "base/message_loop/message_loop_current.h"
+#include "base/metrics/field_trial_params.h"
+#include "base/metrics/histogram_macros.h"
 #include "base/no_destructor.h"
+#include "base/rand_util.h"
 #include "base/run_loop.h"
 #include "base/synchronization/lock.h"
 #include "base/threading/sequence_local_storage_slot.h"
@@ -50,6 +53,30 @@ bool EnableTaskPerMessage() {
   return enable;
 }
 
+const base::FeatureParam<int> kMojoRecordUnreadMessageCountSampleRate = {
+    &features::kMojoRecordUnreadMessageCount, "SampleRate",
+    100  // Sample 1% of Connectors by default.
+};
+
+const base::FeatureParam<int> kMojoRecordUnreadMessageCountQuotaValue = {
+    &features::kMojoRecordUnreadMessageCount, "QuotaValue",
+    100  // Use a 100 message quote by default.
+};
+
+int UnreadMessageCountQuota() {
+  static const bool enabled =
+      base::FeatureList::IsEnabled(features::kMojoRecordUnreadMessageCount);
+  if (!enabled)
+    return 0;
+
+  static const int sample_rate = kMojoRecordUnreadMessageCountSampleRate.Get();
+  if (base::RandInt(0, sample_rate - 1) != 0)
+    return 0;
+
+  static const int quota = kMojoRecordUnreadMessageCountQuotaValue.Get();
+  return quota;
+}
+
 }  // namespace
 
 // Used to efficiently maintain a doubly-linked list of all Connectors
@@ -158,9 +185,27 @@ Connector::Connector(ScopedMessagePipeHandle message_pipe,
   // Even though we don't have an incoming receiver, we still want to monitor
   // the message pipe to know if is closed or encounters an error.
   WaitToReadMore();
+
+  int unread_message_count_quota = UnreadMessageCountQuota();
+  if (unread_message_count_quota != 0) {
+    // This connector has been sampled to record the max unread message count.
+    // Note that setting the quota to N results in over-counting usage by up to
+    // N/2, in addition to overcounting due to message transit delays. As result
+    // it's best to treat the resulting metric as N-granular.
+    MojoResult rv = MojoSetQuota(message_pipe_.get().value(),
+                                 MOJO_QUOTA_TYPE_UNREAD_MESSAGE_COUNT,
+                                 unread_message_count_quota, nullptr);
+    if (rv == MOJO_RESULT_OK)
+      max_unread_message_quota_used_ = 0U;
+  }
 }
 
 Connector::~Connector() {
+  if (max_unread_message_quota_used_ != MOJO_QUOTA_LIMIT_NONE) {
+    UMA_HISTOGRAM_COUNTS_1M("Mojo.Connector.MaxUnreadMessageQuotaUsed",
+                            max_unread_message_quota_used_);
+  }
+
   {
     // Allow for quick destruction on any sequence if the pipe is already
     // closed.
@@ -317,6 +362,15 @@ bool Connector::Accept(Message* message) {
     DCHECK(dump_result);
   }
 #endif
+  if (max_unread_message_quota_used_ != MOJO_QUOTA_LIMIT_NONE) {
+    uint64_t limit = 0;
+    uint64_t usage = 0;
+    MojoResult rv = MojoQueryQuota(message_pipe_.get().value(),
+                                   MOJO_QUOTA_TYPE_UNREAD_MESSAGE_COUNT,
+                                   nullptr, &limit, &usage);
+    if (rv == MOJO_RESULT_OK && usage > max_unread_message_quota_used_)
+      max_unread_message_quota_used_ = usage;
+  }
 
   MojoResult rv =
       WriteMessageNew(message_pipe_.get(), message->TakeMojoMessage(),

tools/metrics/histograms/histograms.xml

@@ -69344,6 +69344,19 @@ uploading your change for review.
   </summary>
 </histogram>
 
+<histogram name="Mojo.Connector.MaxUnreadMessageQuotaUsed" units="messages"
+    expires_after="2020-10-01">
+  <owner>siggi@chromium.org</owner>
+  <owner>rockot@chromium.org</owner>
+  <summary>
+    The maximal unread message quota used for the lifetime of a Connector. This
+    is sampled for a configurable percentage of Connectors only when the feature
+    MojoRecordUnreadMessageCount is enabled. By default 1% of Connectors are
+    sampled, as there's some overhead involved in enabling the unread message
+    quota on a MessagePipe. See //mojo/public/cpp/bindings/lib/connector.cc.
+  </summary>
+</histogram>
+
 <histogram name="Mojo.MachPortRelay.BrokerError"
     enum="MojoMachPortRelayBrokerError" expires_after="M77">
   <obsolete>