Mojo C++ bindings: introduce MultiplexRouter and related classes.

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

 // Copyright 2013 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/connector.h"
 
-#include "mojo/public/cpp/environment/logging.h"
+#include "base/logging.h"
+#include "base/macros.h"
+#include "base/synchronization/lock.h"
 
 namespace mojo {
 namespace internal {
 
+namespace {
+
+// Similar to base::AutoLock, except that it does nothing if |lock| passed into
+// the constructor is null.
+class MayAutoLock {
+ public:
+  explicit MayAutoLock(base::Lock* lock) : lock_(lock) {
+    if (lock_)
+      lock_->Acquire();
+  }
+
+  ~MayAutoLock() {
+    if (lock_) {
+      lock_->AssertAcquired();
+      lock_->Release();
+    }
+  }
+
+ private:
+  base::Lock* lock_;
+  DISALLOW_COPY_AND_ASSIGN(MayAutoLock);
+};
+
+}  // namespace
+
 // ----------------------------------------------------------------------------
 
 Connector::Connector(ScopedMessagePipeHandle message_pipe,
+                     SendingThreadSafetyType sending_thread_safety_type,
                      const MojoAsyncWaiter* waiter)
     : waiter_(waiter),
       message_pipe_(message_pipe.Pass()),
       incoming_receiver_(nullptr),
       async_wait_id_(0),
       error_(false),
       drop_writes_(false),
       enforce_errors_from_incoming_receiver_(true),
       paused_(false),
-      destroyed_flag_(nullptr) {
+      destroyed_flag_(nullptr),
+      lock_(sending_thread_safety_type == MULTI_THREADED_SEND ? new base::Lock
+                                                              : nullptr) {
   // 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();
 }
 
 Connector::~Connector() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
   if (destroyed_flag_)
     *destroyed_flag_ = true;
 
   CancelWait();
 }
 
 void Connector::CloseMessagePipe() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
   CancelWait();
+  MayAutoLock locker(lock_.get());
   Close(message_pipe_.Pass());
 }
 
 ScopedMessagePipeHandle Connector::PassMessagePipe() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
   CancelWait();
+  MayAutoLock locker(lock_.get());
   return message_pipe_.Pass();
 }
 
 void Connector::RaiseError() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
   HandleError(true, true);
 }
 
 bool Connector::WaitForIncomingMessage(MojoDeadline deadline) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
   if (error_)
     return false;
 
   ResumeIncomingMethodCallProcessing();
 
   MojoResult rv =
       Wait(message_pipe_.get(), MOJO_HANDLE_SIGNAL_READABLE, deadline, nullptr);
   if (rv == MOJO_RESULT_SHOULD_WAIT || rv == MOJO_RESULT_DEADLINE_EXCEEDED)
     return false;
   if (rv != MOJO_RESULT_OK) {
     // Users that call WaitForIncomingMessage() should expect their code to be
     // re-entered, so we call the error handler synchronously.
     HandleError(rv != MOJO_RESULT_FAILED_PRECONDITION, false);
     return false;
   }
   mojo_ignore_result(ReadSingleMessage(&rv));
   return (rv == MOJO_RESULT_OK);
 }
 
 void Connector::PauseIncomingMethodCallProcessing() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
   if (paused_)
     return;
 
   paused_ = true;
   CancelWait();
 }
 
 void Connector::ResumeIncomingMethodCallProcessing() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
   if (!paused_)
     return;
 
   paused_ = false;
   WaitToReadMore();
 }
 
 bool Connector::Accept(Message* message) {
+  DCHECK(lock_ || thread_checker_.CalledOnValidThread());
+
+  // It shouldn't hurt even if |error_| may be changed by a different thread at
+  // the same time. The outcome is that we may write into |message_pipe_| after
+  // encountering an error, which should be fine.
   if (error_)
     return false;
 
-  MOJO_CHECK(message_pipe_.is_valid());
-  if (drop_writes_)
+  MayAutoLock locker(lock_.get());
+
+  if (!message_pipe_.is_valid() || drop_writes_)
     return true;
 
   MojoResult rv =
       WriteMessageRaw(message_pipe_.get(),
                       message->data(),
                       message->data_num_bytes(),
                       message->mutable_handles()->empty()
                           ? nullptr
                           : reinterpret_cast<const MojoHandle*>(
                                 &message->mutable_handles()->front()),
@@ skipping 14 matching lines @@
       drop_writes_ = true;
       break;
     case MOJO_RESULT_BUSY:
       // We'd get a "busy" result if one of the message's handles is:
       //   - |message_pipe_|'s own handle;
       //   - simultaneously being used on another thread; or
       //   - in a "busy" state that prohibits it from being transferred (e.g.,
       //     a data pipe handle in the middle of a two-phase read/write,
       //     regardless of which thread that two-phase read/write is happening
       //     on).
-      // TODO(vtl): I wonder if this should be a |MOJO_DCHECK()|. (But, until
+      // TODO(vtl): I wonder if this should be a |DCHECK()|. (But, until
       // crbug.com/389666, etc. are resolved, this will make tests fail quickly
       // rather than hanging.)
-      MOJO_CHECK(false) << "Race condition or other bug detected";
+      CHECK(false) << "Race condition or other bug detected";
       return false;
     default:
       // This particular write was rejected, presumably because of bad input.
       // The pipe is not necessarily in a bad state.
       return false;
   }
   return true;
 }
 
 // static
 void Connector::CallOnHandleReady(void* closure, MojoResult result) {
   Connector* self = static_cast<Connector*>(closure);
   self->OnHandleReady(result);
 }
 
 void Connector::OnHandleReady(MojoResult result) {
-  MOJO_CHECK(async_wait_id_ != 0);
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  CHECK(async_wait_id_ != 0);
   async_wait_id_ = 0;
   if (result != MOJO_RESULT_OK) {
     HandleError(result != MOJO_RESULT_FAILED_PRECONDITION, false);
     return;
   }
   ReadAllAvailableMessages();
   // At this point, this object might have been deleted. Return.
 }
 
 void Connector::WaitToReadMore() {
-  MOJO_CHECK(!async_wait_id_);
+  CHECK(!async_wait_id_);
   async_wait_id_ = waiter_->AsyncWait(message_pipe_.get().value(),
                                       MOJO_HANDLE_SIGNAL_READABLE,
                                       MOJO_DEADLINE_INDEFINITE,
                                       &Connector::CallOnHandleReady,
                                       this);
 }
 
 bool Connector::ReadSingleMessage(MojoResult* read_result) {
   bool receiver_result = false;
 
@@ skipping 61 matching lines @@
     force_pipe_reset = true;
 
   if (paused_) {
     // If the user has paused receiving messages, we shouldn't call the error
     // handler right away. We need to wait until the user starts receiving
     // messages again.
     force_async_handler = true;
   }
 
   if (force_pipe_reset) {
-    CloseMessagePipe();
+    CancelWait();
+    MayAutoLock locker(lock_.get());
+    Close(message_pipe_.Pass());
     MessagePipe dummy_pipe;
     message_pipe_ = dummy_pipe.handle0.Pass();
   } else {
     CancelWait();
   }
 
   if (force_async_handler) {
     // |dummy_pipe.handle1| has been destructed. Reading the pipe will
     // eventually cause a read error on |message_pipe_| and set error state.
     if (!paused_)
       WaitToReadMore();
   } else {
     error_ = true;
     connection_error_handler_.Run();
   }
 }
 
 }  // namespace internal
 }  // namespace mojo