Add a Mojo EDK for Chrome that uses one OS pipe per message pipe.

mojo/edk/system/dispatcher.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/edk/system/dispatcher.h"
 
 #include "base/logging.h"
 #include "mojo/edk/system/configuration.h"
 #include "mojo/edk/system/data_pipe_consumer_dispatcher.h"
 #include "mojo/edk/system/data_pipe_producer_dispatcher.h"
 #include "mojo/edk/system/message_pipe_dispatcher.h"
 #include "mojo/edk/system/platform_handle_dispatcher.h"
 #include "mojo/edk/system/shared_buffer_dispatcher.h"
 
 namespace mojo {
 namespace system {
 
 namespace test {
 
 // TODO(vtl): Maybe this should be defined in a test-only file instead.
 DispatcherTransport DispatcherTryStartTransport(Dispatcher* dispatcher) {
   return Dispatcher::HandleTableAccess::TryStartTransport(dispatcher);
 }
 
 }  // namespace test
 
 // Dispatcher ------------------------------------------------------------------
 
-// TODO(vtl): The thread-safety analyzer isn't smart enough to deal with the
-// fact that we give up if |TryLock()| fails.
 // static
 DispatcherTransport Dispatcher::HandleTableAccess::TryStartTransport(
-    Dispatcher* dispatcher) MOJO_NO_THREAD_SAFETY_ANALYSIS {
+    Dispatcher* dispatcher) {
   DCHECK(dispatcher);
 
-  if (!dispatcher->mutex_.TryLock())
-    return DispatcherTransport();
+  // if (!dispatcher->lock_.Try())
+  //   return DispatcherTransport();
+  // Our dispatcher implemetnations hop to IO thread on initialization, so it's
+  // valid that while their raw channel os being initialized on IO thread, the
+  // dispatcher is being sent. We handle this by just acquiring the lock.
+
+  // See comment in header for why we need this.
+  dispatcher->TransportStarted();
+
+  dispatcher->lock_.Acquire();
 
   // We shouldn't race with things that close dispatchers, since closing can
-  // only take place either under |handle_table_mutex_| or when the handle is
+  // only take place either under |handle_table_lock_| or when the handle is
   // marked as busy.
   DCHECK(!dispatcher->is_closed_);
 
   return DispatcherTransport(dispatcher);
 }
 
 // static
 void Dispatcher::TransportDataAccess::StartSerialize(
     Dispatcher* dispatcher,
-    Channel* channel,
     size_t* max_size,
     size_t* max_platform_handles) {
   DCHECK(dispatcher);
-  dispatcher->StartSerialize(channel, max_size, max_platform_handles);
+  dispatcher->StartSerialize(max_size, max_platform_handles);
 }
 
 // static
 bool Dispatcher::TransportDataAccess::EndSerializeAndClose(
     Dispatcher* dispatcher,
-    Channel* channel,
     void* destination,
     size_t* actual_size,
     embedder::PlatformHandleVector* platform_handles) {
   DCHECK(dispatcher);
-  return dispatcher->EndSerializeAndClose(channel, destination, actual_size,
+  return dispatcher->EndSerializeAndClose(destination, actual_size,
                                           platform_handles);
 }
 
 // static
 scoped_refptr<Dispatcher> Dispatcher::TransportDataAccess::Deserialize(
-    Channel* channel,
     int32_t type,
     const void* source,
     size_t size,
     embedder::PlatformHandleVector* platform_handles) {
   switch (static_cast<Dispatcher::Type>(type)) {
     case Type::UNKNOWN:
       DVLOG(2) << "Deserializing invalid handle";
       return nullptr;
     case Type::MESSAGE_PIPE:
-      return scoped_refptr<Dispatcher>(
-          MessagePipeDispatcher::Deserialize(channel, source, size));
+      return scoped_refptr<Dispatcher>(MessagePipeDispatcher::Deserialize(
+          source, size, platform_handles));
     case Type::DATA_PIPE_PRODUCER:
       return scoped_refptr<Dispatcher>(
-          DataPipeProducerDispatcher::Deserialize(channel, source, size));
+          DataPipeProducerDispatcher::Deserialize(
+              source, size, platform_handles));
     case Type::DATA_PIPE_CONSUMER:
       return scoped_refptr<Dispatcher>(
-          DataPipeConsumerDispatcher::Deserialize(channel, source, size));
+          DataPipeConsumerDispatcher::Deserialize(
+              source, size, platform_handles));
     case Type::SHARED_BUFFER:
       return scoped_refptr<Dispatcher>(SharedBufferDispatcher::Deserialize(
-          channel, source, size, platform_handles));
+          source, size, platform_handles));
     case Type::PLATFORM_HANDLE:
       return scoped_refptr<Dispatcher>(PlatformHandleDispatcher::Deserialize(
-          channel, source, size, platform_handles));
+          source, size, platform_handles));
   }
   LOG(WARNING) << "Unknown dispatcher type " << type;
   return nullptr;
 }
 
 MojoResult Dispatcher::Close() {
-  MutexLocker locker(&mutex_);
+  base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   CloseNoLock();
   return MOJO_RESULT_OK;
 }
 
 MojoResult Dispatcher::WriteMessage(
     UserPointer<const void> bytes,
     uint32_t num_bytes,
     std::vector<DispatcherTransport>* transports,
     MojoWriteMessageFlags flags) {
   DCHECK(!transports ||
          (transports->size() > 0 &&
           transports->size() < GetConfiguration().max_message_num_handles));
 
-  MutexLocker locker(&mutex_);
+  base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return WriteMessageImplNoLock(bytes, num_bytes, transports, flags);
 }
 
 MojoResult Dispatcher::ReadMessage(UserPointer<void> bytes,
                                    UserPointer<uint32_t> num_bytes,
                                    DispatcherVector* dispatchers,
                                    uint32_t* num_dispatchers,
                                    MojoReadMessageFlags flags) {
   DCHECK(!num_dispatchers || *num_dispatchers == 0 ||
          (dispatchers && dispatchers->empty()));
 
-  MutexLocker locker(&mutex_);
+  base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return ReadMessageImplNoLock(bytes, num_bytes, dispatchers, num_dispatchers,
                                flags);
 }
 
 MojoResult Dispatcher::WriteData(UserPointer<const void> elements,
                                  UserPointer<uint32_t> num_bytes,
                                  MojoWriteDataFlags flags) {
-  MutexLocker locker(&mutex_);
+  base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return WriteDataImplNoLock(elements, num_bytes, flags);
 }
 
 MojoResult Dispatcher::BeginWriteData(UserPointer<void*> buffer,
                                       UserPointer<uint32_t> buffer_num_bytes,
                                       MojoWriteDataFlags flags) {
-  MutexLocker locker(&mutex_);
+  base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return BeginWriteDataImplNoLock(buffer, buffer_num_bytes, flags);
 }
 
 MojoResult Dispatcher::EndWriteData(uint32_t num_bytes_written) {
-  MutexLocker locker(&mutex_);
+  base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return EndWriteDataImplNoLock(num_bytes_written);
 }
 
 MojoResult Dispatcher::ReadData(UserPointer<void> elements,
                                 UserPointer<uint32_t> num_bytes,
                                 MojoReadDataFlags flags) {
-  MutexLocker locker(&mutex_);
+  base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return ReadDataImplNoLock(elements, num_bytes, flags);
 }
 
 MojoResult Dispatcher::BeginReadData(UserPointer<const void*> buffer,
                                      UserPointer<uint32_t> buffer_num_bytes,
                                      MojoReadDataFlags flags) {
-  MutexLocker locker(&mutex_);
+  base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return BeginReadDataImplNoLock(buffer, buffer_num_bytes, flags);
 }
 
 MojoResult Dispatcher::EndReadData(uint32_t num_bytes_read) {
-  MutexLocker locker(&mutex_);
+  base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return EndReadDataImplNoLock(num_bytes_read);
 }
 
 MojoResult Dispatcher::DuplicateBufferHandle(
     UserPointer<const MojoDuplicateBufferHandleOptions> options,
     scoped_refptr<Dispatcher>* new_dispatcher) {
-  MutexLocker locker(&mutex_);
+  base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return DuplicateBufferHandleImplNoLock(options, new_dispatcher);
 }
 
 MojoResult Dispatcher::MapBuffer(
     uint64_t offset,
     uint64_t num_bytes,
     MojoMapBufferFlags flags,
     scoped_ptr<embedder::PlatformSharedBufferMapping>* mapping) {
-  MutexLocker locker(&mutex_);
+  base::AutoLock locker(lock_);
   if (is_closed_)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   return MapBufferImplNoLock(offset, num_bytes, flags, mapping);
 }
 
 HandleSignalsState Dispatcher::GetHandleSignalsState() const {
-  MutexLocker locker(&mutex_);
+  base::AutoLock locker(lock_);
   if (is_closed_)
     return HandleSignalsState();
 
   return GetHandleSignalsStateImplNoLock();
 }
 
 MojoResult Dispatcher::AddAwakable(Awakable* awakable,
                                    MojoHandleSignals signals,
                                    uint32_t context,
                                    HandleSignalsState* signals_state) {
-  MutexLocker locker(&mutex_);
+  base::AutoLock locker(lock_);
   if (is_closed_) {
     if (signals_state)
       *signals_state = HandleSignalsState();
     return MOJO_RESULT_INVALID_ARGUMENT;
   }
 
   return AddAwakableImplNoLock(awakable, signals, context, signals_state);
 }
 
 void Dispatcher::RemoveAwakable(Awakable* awakable,
                                 HandleSignalsState* handle_signals_state) {
-  MutexLocker locker(&mutex_);
+  base::AutoLock locker(lock_);
   if (is_closed_) {
     if (handle_signals_state)
       *handle_signals_state = HandleSignalsState();
     return;
   }
 
   RemoveAwakableImplNoLock(awakable, handle_signals_state);
 }
 
 Dispatcher::Dispatcher() : is_closed_(false) {
 }
 
 Dispatcher::~Dispatcher() {
   // Make sure that |Close()| was called.
   DCHECK(is_closed_);
 }
 
 void Dispatcher::CancelAllAwakablesNoLock() {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(is_closed_);
   // By default, waiting isn't supported. Only dispatchers that can be waited on
   // will do something nontrivial.
 }
 
 void Dispatcher::CloseImplNoLock() {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(is_closed_);
   // This may not need to do anything. Dispatchers should override this to do
   // any actual close-time cleanup necessary.
 }
 
 MojoResult Dispatcher::WriteMessageImplNoLock(
     UserPointer<const void> /*bytes*/,
     uint32_t /*num_bytes*/,
     std::vector<DispatcherTransport>* /*transports*/,
     MojoWriteMessageFlags /*flags*/) {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, not supported. Only needed for message pipe dispatchers.
   return MOJO_RESULT_INVALID_ARGUMENT;
 }
 
 MojoResult Dispatcher::ReadMessageImplNoLock(
     UserPointer<void> /*bytes*/,
     UserPointer<uint32_t> /*num_bytes*/,
     DispatcherVector* /*dispatchers*/,
     uint32_t* /*num_dispatchers*/,
     MojoReadMessageFlags /*flags*/) {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, not supported. Only needed for message pipe dispatchers.
   return MOJO_RESULT_INVALID_ARGUMENT;
 }
 
 MojoResult Dispatcher::WriteDataImplNoLock(UserPointer<const void> /*elements*/,
                                            UserPointer<uint32_t> /*num_bytes*/,
                                            MojoWriteDataFlags /*flags*/) {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, not supported. Only needed for data pipe dispatchers.
   return MOJO_RESULT_INVALID_ARGUMENT;
 }
 
 MojoResult Dispatcher::BeginWriteDataImplNoLock(
     UserPointer<void*> /*buffer*/,
     UserPointer<uint32_t> /*buffer_num_bytes*/,
     MojoWriteDataFlags /*flags*/) {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, not supported. Only needed for data pipe dispatchers.
   return MOJO_RESULT_INVALID_ARGUMENT;
 }
 
 MojoResult Dispatcher::EndWriteDataImplNoLock(uint32_t /*num_bytes_written*/) {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, not supported. Only needed for data pipe dispatchers.
   return MOJO_RESULT_INVALID_ARGUMENT;
 }
 
 MojoResult Dispatcher::ReadDataImplNoLock(UserPointer<void> /*elements*/,
                                           UserPointer<uint32_t> /*num_bytes*/,
                                           MojoReadDataFlags /*flags*/) {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, not supported. Only needed for data pipe dispatchers.
   return MOJO_RESULT_INVALID_ARGUMENT;
 }
 
 MojoResult Dispatcher::BeginReadDataImplNoLock(
     UserPointer<const void*> /*buffer*/,
     UserPointer<uint32_t> /*buffer_num_bytes*/,
     MojoReadDataFlags /*flags*/) {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, not supported. Only needed for data pipe dispatchers.
   return MOJO_RESULT_INVALID_ARGUMENT;
 }
 
 MojoResult Dispatcher::EndReadDataImplNoLock(uint32_t /*num_bytes_read*/) {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, not supported. Only needed for data pipe dispatchers.
   return MOJO_RESULT_INVALID_ARGUMENT;
 }
 
 MojoResult Dispatcher::DuplicateBufferHandleImplNoLock(
     UserPointer<const MojoDuplicateBufferHandleOptions> /*options*/,
     scoped_refptr<Dispatcher>* /*new_dispatcher*/) {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, not supported. Only needed for buffer dispatchers.
   return MOJO_RESULT_INVALID_ARGUMENT;
 }
 
 MojoResult Dispatcher::MapBufferImplNoLock(
     uint64_t /*offset*/,
     uint64_t /*num_bytes*/,
     MojoMapBufferFlags /*flags*/,
     scoped_ptr<embedder::PlatformSharedBufferMapping>* /*mapping*/) {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, not supported. Only needed for buffer dispatchers.
   return MOJO_RESULT_INVALID_ARGUMENT;
 }
 
 HandleSignalsState Dispatcher::GetHandleSignalsStateImplNoLock() const {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, waiting isn't supported. Only dispatchers that can be waited on
   // will do something nontrivial.
   return HandleSignalsState();
 }
 
 MojoResult Dispatcher::AddAwakableImplNoLock(
     Awakable* /*awakable*/,
     MojoHandleSignals /*signals*/,
     uint32_t /*context*/,
     HandleSignalsState* signals_state) {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, waiting isn't supported. Only dispatchers that can be waited on
   // will do something nontrivial.
   if (signals_state)
     *signals_state = HandleSignalsState();
   return MOJO_RESULT_FAILED_PRECONDITION;
 }
 
 void Dispatcher::RemoveAwakableImplNoLock(Awakable* /*awakable*/,
                                           HandleSignalsState* signals_state) {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // By default, waiting isn't supported. Only dispatchers that can be waited on
   // will do something nontrivial.
   if (signals_state)
     *signals_state = HandleSignalsState();
 }
 
-void Dispatcher::StartSerializeImplNoLock(Channel* /*channel*/,
-                                          size_t* max_size,
+void Dispatcher::StartSerializeImplNoLock(size_t* max_size,
                                           size_t* max_platform_handles) {
   DCHECK(HasOneRef());  // Only one ref => no need to take the lock.
   DCHECK(!is_closed_);
   *max_size = 0;
   *max_platform_handles = 0;
 }
 
 bool Dispatcher::EndSerializeAndCloseImplNoLock(
-    Channel* /*channel*/,
     void* /*destination*/,
     size_t* /*actual_size*/,
     embedder::PlatformHandleVector* /*platform_handles*/) {
   DCHECK(HasOneRef());  // Only one ref => no need to take the lock.
   DCHECK(is_closed_);
   // By default, serializing isn't supported, so just close.
   CloseImplNoLock();
   return false;
 }
 
 bool Dispatcher::IsBusyNoLock() const {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(!is_closed_);
   // Most dispatchers support only "atomic" operations, so they are never busy
   // (in this sense).
   return false;
 }
 
 void Dispatcher::CloseNoLock() {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(!is_closed_);
 
   is_closed_ = true;
   CancelAllAwakablesNoLock();
   CloseImplNoLock();
 }
 
 scoped_refptr<Dispatcher>
 Dispatcher::CreateEquivalentDispatcherAndCloseNoLock() {
-  mutex_.AssertHeld();
+  lock_.AssertAcquired();
   DCHECK(!is_closed_);
 
   is_closed_ = true;
   CancelAllAwakablesNoLock();
   return CreateEquivalentDispatcherAndCloseImplNoLock();
 }
 
-void Dispatcher::StartSerialize(Channel* channel,
-                                size_t* max_size,
+void Dispatcher::StartSerialize(size_t* max_size,
                                 size_t* max_platform_handles) {
-  DCHECK(channel);
   DCHECK(max_size);
   DCHECK(max_platform_handles);
-  DCHECK(HasOneRef());  // Only one ref => no need to take the lock.
+  //DCHECK(HasOneRef());  // Only one ref => no need to take the lock.
+  // jam: document exactly when this happens: i.e. when startserialize is called
+  // on main thread.
   DCHECK(!is_closed_);
-  StartSerializeImplNoLock(channel, max_size, max_platform_handles);
+  StartSerializeImplNoLock(max_size, max_platform_handles);
 }
 
 bool Dispatcher::EndSerializeAndClose(
-    Channel* channel,
     void* destination,
     size_t* actual_size,
     embedder::PlatformHandleVector* platform_handles) {
-  DCHECK(channel);
   DCHECK(actual_size);
-  DCHECK(HasOneRef());  // Only one ref => no need to take the lock.
+  //DCHECK(HasOneRef());  // Only one ref => no need to take the lock.
+  // jam: same as startserialize
   DCHECK(!is_closed_);
 
   // Like other |...Close()| methods, we mark ourselves as closed before calling
   // the impl. But there's no need to cancel waiters: we shouldn't have any (and
   // shouldn't be in |Core|'s handle table.
   is_closed_ = true;
 
 #if !defined(NDEBUG)
   // See the comment above |EndSerializeAndCloseImplNoLock()|. In brief: Locking
   // isn't actually needed, but we need to satisfy assertions (which we don't
   // want to remove or weaken).
-  MutexLocker locker(&mutex_);
+  base::AutoLock locker(lock_);
 #endif
 
-  return EndSerializeAndCloseImplNoLock(channel, destination, actual_size,
+  return EndSerializeAndCloseImplNoLock(destination, actual_size,
                                         platform_handles);
 }
 
 // DispatcherTransport ---------------------------------------------------------
 
 void DispatcherTransport::End() {
   DCHECK(dispatcher_);
-  dispatcher_->mutex_.Unlock();
+  dispatcher_->lock_.Release();
+
+  dispatcher_->TransportEnded();
+
   dispatcher_ = nullptr;
 }
 
 }  // namespace system
 }  // namespace mojo