Mojo C++ bindings: fix Pause/Resume for MultiplexRouter.

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

 // Copyright 2015 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/lib/multiplex_router.h"
 
 #include <stdint.h>
 
 #include <utility>
 
@@ skipping 85 matching lines @@
       return;
 
     EnsureEventMessagePipeExists();
     event_signalled_ = true;
     MojoResult result =
         WriteMessageRaw(sync_message_event_sender_.get(), nullptr, 0, nullptr,
                         0, MOJO_WRITE_MESSAGE_FLAG_NONE);
     DCHECK_EQ(MOJO_RESULT_OK, result);
   }
 
+  void ResetSyncMessageSignal() {
+    router_->lock_.AssertAcquired();
+
+    if (!event_signalled_)
+      return;
+
+    DCHECK(sync_message_event_receiver_.is_valid());
+    MojoResult result =
+        ReadMessageRaw(sync_message_event_receiver_.get(), nullptr, nullptr,
+                       nullptr, nullptr, MOJO_READ_MESSAGE_FLAG_MAY_DISCARD);
+    DCHECK_EQ(MOJO_RESULT_OK, result);
+    event_signalled_ = false;
+  }
+
   // ---------------------------------------------------------------------------
   // The following public methods (i.e., InterfaceEndpointController
   // implementation) are called by the client on the same thread as the
   // AttachClient() call. They are called outside of the router's lock.
 
   bool SendMessage(Message* message) override {
     DCHECK(task_runner_->BelongsToCurrentThread());
     message->set_interface_id(id_);
     return router_->connector_.Accept(message);
   }
@@ skipping 76 matching lines @@
     router_->lock_.AssertAcquired();
 
     if (sync_message_event_receiver_.is_valid())
       return;
 
     MojoResult result = CreateMessagePipe(nullptr, &sync_message_event_sender_,
                                           &sync_message_event_receiver_);
     DCHECK_EQ(MOJO_RESULT_OK, result);
   }
 
-  void ResetSyncMessageSignal() {
-    router_->lock_.AssertAcquired();
-
-    if (!event_signalled_)
-      return;
-
-    DCHECK(sync_message_event_receiver_.is_valid());
-    MojoResult result = ReadMessageRaw(sync_message_event_receiver_.get(),
-                                       nullptr, nullptr, nullptr, nullptr,
-                                       MOJO_READ_MESSAGE_FLAG_MAY_DISCARD);
-    DCHECK_EQ(MOJO_RESULT_OK, result);
-    event_signalled_ = false;
-  }
-
   // ---------------------------------------------------------------------------
   // The following members are safe to access from any threads.
 
   MultiplexRouter* const router_;
   const InterfaceId id_;
 
   // ---------------------------------------------------------------------------
   // The following members are accessed under the router's lock.
 
   // Whether the endpoint has been closed.
@@ skipping 63 matching lines @@
       set_interface_id_namespace_bit_(set_interface_id_namesapce_bit),
       header_validator_(this),
       connector_(std::move(message_pipe),
                  Connector::MULTI_THREADED_SEND,
                  std::move(runner)),
       control_message_handler_(this),
       control_message_proxy_(&connector_),
       next_interface_id_value_(1),
       posted_to_process_tasks_(false),
       encountered_error_(false),
+      paused_(false),
       testing_mode_(false) {
   // Always participate in sync handle watching, because even if it doesn't
   // expect sync requests during sync handle watching, it may still need to
   // dispatch messages to associated endpoints on a different thread.
   connector_.AllowWokenUpBySyncWatchOnSameThread();
   connector_.set_incoming_receiver(&header_validator_);
   connector_.set_connection_error_handler(
       base::Bind(&MultiplexRouter::OnPipeConnectionError,
                  base::Unretained(this)));
 }
@@ skipping 140 matching lines @@
 
 void MultiplexRouter::CloseMessagePipe() {
   DCHECK(thread_checker_.CalledOnValidThread());
   connector_.CloseMessagePipe();
   // CloseMessagePipe() above won't trigger connection error handler.
   // Explicitly call OnPipeConnectionError() so that associated endpoints will
   // get notified.
   OnPipeConnectionError();
 }
 
+void MultiplexRouter::PauseIncomingMethodCallProcessing() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  connector_.PauseIncomingMethodCallProcessing();
+
+  base::AutoLock locker(lock_);
+  paused_ = true;
+
+  for (auto iter = endpoints_.begin(); iter != endpoints_.end(); ++iter)
+    iter->second->ResetSyncMessageSignal();
+}
+
+void MultiplexRouter::ResumeIncomingMethodCallProcessing() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  connector_.ResumeIncomingMethodCallProcessing();
+
+  base::AutoLock locker(lock_);
+  paused_ = false;
+
+  for (auto iter = endpoints_.begin(); iter != endpoints_.end(); ++iter) {
+    auto sync_iter = sync_message_tasks_.find(iter->first);
+    if (sync_iter != sync_message_tasks_.end() && !sync_iter->second.empty())
+      iter->second->SignalSyncMessageEvent();
+  }
+
+  ProcessTasks(NO_DIRECT_CLIENT_CALLS, nullptr);
+}
+
 bool MultiplexRouter::HasAssociatedEndpoints() const {
   DCHECK(thread_checker_.CalledOnValidThread());
   base::AutoLock locker(lock_);
 
   if (endpoints_.size() > 1)
     return true;
   if (endpoints_.size() == 0)
     return false;
 
   return !ContainsKey(endpoints_, kMasterInterfaceId);
 }
 
 void MultiplexRouter::EnableTestingMode() {
   DCHECK(thread_checker_.CalledOnValidThread());
   base::AutoLock locker(lock_);
 
   testing_mode_ = true;
   connector_.set_enforce_errors_from_incoming_receiver(false);
 }
 
 bool MultiplexRouter::Accept(Message* message) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   scoped_refptr<MultiplexRouter> protector(this);
   base::AutoLock locker(lock_);
 
+  DCHECK(!paused_);
+
   ClientCallBehavior client_call_behavior =
       connector_.during_sync_handle_watcher_callback()
           ? ALLOW_DIRECT_CLIENT_CALLS_FOR_SYNC_MESSAGES
           : ALLOW_DIRECT_CLIENT_CALLS;
 
   bool processed =
       tasks_.empty() && ProcessIncomingMessage(message, client_call_behavior,
                                                connector_.task_runner());
 
   if (!processed) {
@@ skipping 88 matching lines @@
 }
 
 void MultiplexRouter::ProcessTasks(
     ClientCallBehavior client_call_behavior,
     base::SingleThreadTaskRunner* current_task_runner) {
   lock_.AssertAcquired();
 
   if (posted_to_process_tasks_)
     return;
 
-  while (!tasks_.empty()) {
+  while (!tasks_.empty() && !paused_) {
     std::unique_ptr<Task> task(std::move(tasks_.front()));
     tasks_.pop_front();
 
     InterfaceId id = kInvalidInterfaceId;
     bool sync_message = task->IsMessageTask() && task->message &&
                         task->message->has_flag(Message::kFlagIsSync);
     if (sync_message) {
       id = task->message->interface_id();
       auto& sync_message_queue = sync_message_tasks_[id];
       DCHECK_EQ(task.get(), sync_message_queue.front());
@@ skipping 24 matching lines @@
   }
 }
 
 bool MultiplexRouter::ProcessFirstSyncMessageForEndpoint(InterfaceId id) {
   lock_.AssertAcquired();
 
   auto iter = sync_message_tasks_.find(id);
   if (iter == sync_message_tasks_.end())
     return false;
 
+  if (paused_)
+    return true;
+
   MultiplexRouter::Task* task = iter->second.front();
   iter->second.pop_front();
 
   DCHECK(task->IsMessageTask());
   std::unique_ptr<Message> message(std::move(task->message));
 
-  // Note: after this call, |task| and  |iter| may be invalidated.
+  // Note: after this call, |task| and |iter| may be invalidated.
   bool processed = ProcessIncomingMessage(
       message.get(), ALLOW_DIRECT_CLIENT_CALLS_FOR_SYNC_MESSAGES, nullptr);
   DCHECK(processed);
 
   iter = sync_message_tasks_.find(id);
   if (iter == sync_message_tasks_.end())
     return false;
 
   if (iter->second.empty()) {
     sync_message_tasks_.erase(iter);
     return false;
   }
 
   return true;
 }
 
 bool MultiplexRouter::ProcessNotifyErrorTask(
     Task* task,
     ClientCallBehavior client_call_behavior,
     base::SingleThreadTaskRunner* current_task_runner) {
   DCHECK(!current_task_runner || current_task_runner->BelongsToCurrentThread());
+  DCHECK(!paused_);
+
   lock_.AssertAcquired();
   InterfaceEndpoint* endpoint = task->endpoint_to_notify.get();
   if (!endpoint->client())
     return true;
 
   if (client_call_behavior != ALLOW_DIRECT_CLIENT_CALLS ||
       endpoint->task_runner() != current_task_runner) {
     MaybePostToProcessTasks(endpoint->task_runner());
     return false;
   }
@@ skipping 11 matching lines @@
     client->NotifyError();
   }
   return true;
 }
 
 bool MultiplexRouter::ProcessIncomingMessage(
     Message* message,
     ClientCallBehavior client_call_behavior,
     base::SingleThreadTaskRunner* current_task_runner) {
   DCHECK(!current_task_runner || current_task_runner->BelongsToCurrentThread());
+  DCHECK(!paused_);
   lock_.AssertAcquired();
 
   if (!message) {
     // This is a sync message and has been processed during sync handle
     // watching.
     return true;
   }
 
   if (PipeControlMessageHandler::IsPipeControlMessage(message)) {
     if (!control_message_handler_.Accept(message))
@@ skipping 133 matching lines @@
       *inserted = true;
   } else {
     endpoint = iter->second.get();
   }
 
   return endpoint;
 }
 
 }  // namespace internal
 }  // namespace mojo