Mojo: Fix deadlock in quota control messaging

If an intraprocess pipe calls SetQuota on one endpoint, it's possible
for this to result in a deadlock due to Mojo internals re-entering the
endpoint's MessagePipeDispatcher while already holding its internal
lock.

This fixes the issue about deferring the potential re-entry until the
lock is released.

Fixed: 1137988
Change-Id: I814415a3ce155a991a2e6d2cfd07fa309004eeda
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2485147
Commit-Queue: Ken Rockot <rockot@google.com>
Reviewed-by: Sigurður Ásgeirsson <siggi@chromium.org>
Cr-Commit-Position: refs/heads/master@{#819221}

mojo/core/message_pipe_dispatcher.cc

@@ -199,42 +199,51 @@ MojoResult MessagePipeDispatcher::ReadMessage(
 }
 
 MojoResult MessagePipeDispatcher::SetQuota(MojoQuotaType type, uint64_t limit) {
-  base::AutoLock lock(signal_lock_);
-
-  switch (type) {
-    case MOJO_QUOTA_TYPE_RECEIVE_QUEUE_LENGTH:
-      if (limit == MOJO_QUOTA_LIMIT_NONE)
-        receive_queue_length_limit_.reset();
-      else
-        receive_queue_length_limit_ = limit;
-      break;
-
-    case MOJO_QUOTA_TYPE_RECEIVE_QUEUE_MEMORY_SIZE:
-      if (limit == MOJO_QUOTA_LIMIT_NONE)
-        receive_queue_memory_size_limit_.reset();
-      else
-        receive_queue_memory_size_limit_ = limit;
-      break;
-
-    case MOJO_QUOTA_TYPE_UNREAD_MESSAGE_COUNT:
-      if (limit == MOJO_QUOTA_LIMIT_NONE) {
-        unread_message_count_limit_.reset();
-        node_controller_->node()->SetAcknowledgeRequestInterval(port_, 0);
-      } else {
-        unread_message_count_limit_ = limit;
-        // Setting the acknowledge request interval for the port to half the
-        // unread quota limit, means the ack roundtrip has half the window to
-        // catch up with sent messages. In other words, if the producer is
-        // producing messages at a steady rate of limit/2 packets per message
-        // round trip or lower, the quota limit won't be exceeded. This is
-        // assuming the consumer is consuming messages at the same rate.
-        node_controller_->node()->SetAcknowledgeRequestInterval(
-            port_, (limit + 1) / 2);
-      }
-      break;
+  base::Optional<uint64_t> new_ack_request_interval;
+  {
+    base::AutoLock lock(signal_lock_);
+    switch (type) {
+      case MOJO_QUOTA_TYPE_RECEIVE_QUEUE_LENGTH:
+        if (limit == MOJO_QUOTA_LIMIT_NONE)
+          receive_queue_length_limit_.reset();
+        else
+          receive_queue_length_limit_ = limit;
+        break;
+
+      case MOJO_QUOTA_TYPE_RECEIVE_QUEUE_MEMORY_SIZE:
+        if (limit == MOJO_QUOTA_LIMIT_NONE)
+          receive_queue_memory_size_limit_.reset();
+        else
+          receive_queue_memory_size_limit_ = limit;
+        break;
+
+      case MOJO_QUOTA_TYPE_UNREAD_MESSAGE_COUNT:
+        if (limit == MOJO_QUOTA_LIMIT_NONE) {
+          unread_message_count_limit_.reset();
+          new_ack_request_interval = 0;
+        } else {
+          unread_message_count_limit_ = limit;
+          // Setting the acknowledge request interval for the port to half the
+          // unread quota limit, means the ack roundtrip has half the window to
+          // catch up with sent messages. In other words, if the producer is
+          // producing messages at a steady rate of limit/2 packets per message
+          // round trip or lower, the quota limit won't be exceeded. This is
+          // assuming the consumer is consuming messages at the same rate.
+          new_ack_request_interval = (limit + 1) / 2;
+        }
+        break;
+
+      default:
+        return MOJO_RESULT_INVALID_ARGUMENT;
+    }
+  }
 
-    default:
-      return MOJO_RESULT_INVALID_ARGUMENT;
+  if (new_ack_request_interval.has_value()) {
+    // NOTE: It is not safe to call into SetAcknowledgeRequestInterval while
+    // holding a `signal_lock_`, as it may re-enter this object when the peer is
+    // in the same process.
+    node_controller_->node()->SetAcknowledgeRequestInterval(
+        port_, *new_ack_request_interval);
   }
 
   return MOJO_RESULT_OK;