mojo/edk/system/dispatcher.cc - Issue 1350023003: Add a Mojo EDK for Chrome that uses one OS pipe per message pipe.

Unified Diff: mojo/edk/system/dispatcher.cc

Issue 1350023003: Add a Mojo EDK for Chrome that uses one OS pipe per message pipe. (Closed) Base URL: https://chromium.googlesource.com/chromium/src.git@master

Patch Set: convert remaining MP tests and simplify RawChannel destruction Created 5 years, 3 months ago

Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.

Jump to:

View side-by-side diff with in-line comments

Download patch

Index: mojo/edk/system/dispatcher.cc

diff --git a/third_party/mojo/src/mojo/edk/system/dispatcher.cc b/mojo/edk/system/dispatcher.cc

similarity index 84%

copy from third_party/mojo/src/mojo/edk/system/dispatcher.cc

copy to mojo/edk/system/dispatcher.cc

index ad362fe5eacd01807a1cc3569355fff19c5ce74e..d80c9d5330499c639d5327d300bb4959467471ae 100644

--- a/third_party/mojo/src/mojo/edk/system/dispatcher.cc

+++ b/mojo/edk/system/dispatcher.cc

@@ -26,18 +26,24 @@ DispatcherTransport DispatcherTryStartTransport(Dispatcher* dispatcher) {

// 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_);

@@ -47,28 +53,25 @@ DispatcherTransport Dispatcher::HandleTableAccess::TryStartTransport(

// 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,

@@ -78,27 +81,29 @@ scoped_refptr<Dispatcher> Dispatcher::TransportDataAccess::Deserialize(

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;

@@ -115,7 +120,7 @@ MojoResult Dispatcher::WriteMessage(

(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;

@@ -130,7 +135,7 @@ MojoResult Dispatcher::ReadMessage(UserPointer<void> bytes,

DCHECK(!num_dispatchers || *num_dispatchers == 0 ||

(dispatchers && dispatchers->empty()));

- MutexLocker locker(&mutex_);

+ base::AutoLock locker(lock_);

if (is_closed_)

return MOJO_RESULT_INVALID_ARGUMENT;

@@ -141,7 +146,7 @@ MojoResult Dispatcher::ReadMessage(UserPointer<void> bytes,

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;

@@ -151,7 +156,7 @@ MojoResult Dispatcher::WriteData(UserPointer<const void> elements,

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;

@@ -159,7 +164,7 @@ MojoResult Dispatcher::BeginWriteData(UserPointer<void*> buffer,

}

MojoResult Dispatcher::EndWriteData(uint32_t num_bytes_written) {

- MutexLocker locker(&mutex_);

+ base::AutoLock locker(lock_);

if (is_closed_)

return MOJO_RESULT_INVALID_ARGUMENT;

@@ -169,7 +174,7 @@ MojoResult Dispatcher::EndWriteData(uint32_t 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;

@@ -179,7 +184,7 @@ MojoResult Dispatcher::ReadData(UserPointer<void> elements,

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;

@@ -187,7 +192,7 @@ MojoResult Dispatcher::BeginReadData(UserPointer<const void*> buffer,

}

MojoResult Dispatcher::EndReadData(uint32_t num_bytes_read) {

- MutexLocker locker(&mutex_);

+ base::AutoLock locker(lock_);

if (is_closed_)

return MOJO_RESULT_INVALID_ARGUMENT;

@@ -197,7 +202,7 @@ MojoResult Dispatcher::EndReadData(uint32_t 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;

@@ -209,7 +214,7 @@ MojoResult Dispatcher::MapBuffer(

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;

@@ -217,7 +222,7 @@ MojoResult Dispatcher::MapBuffer(

}

HandleSignalsState Dispatcher::GetHandleSignalsState() const {

- MutexLocker locker(&mutex_);

+ base::AutoLock locker(lock_);

if (is_closed_)

return HandleSignalsState();

@@ -228,7 +233,7 @@ 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();

@@ -240,7 +245,7 @@ MojoResult Dispatcher::AddAwakable(Awakable* awakable,

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();

@@ -259,14 +264,14 @@ Dispatcher::~Dispatcher() {

}

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.

@@ -277,7 +282,7 @@ MojoResult Dispatcher::WriteMessageImplNoLock(

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;

@@ -289,7 +294,7 @@ MojoResult Dispatcher::ReadMessageImplNoLock(

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;

@@ -298,7 +303,7 @@ MojoResult Dispatcher::ReadMessageImplNoLock(

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;

@@ -308,14 +313,14 @@ 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;

@@ -324,7 +329,7 @@ MojoResult Dispatcher::EndWriteDataImplNoLock(uint32_t /*num_bytes_written*/) {

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;

@@ -334,14 +339,14 @@ 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;

@@ -350,7 +355,7 @@ MojoResult Dispatcher::EndReadDataImplNoLock(uint32_t /*num_bytes_read*/) {

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;

@@ -361,14 +366,14 @@ MojoResult Dispatcher::MapBufferImplNoLock(

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.

@@ -380,7 +385,7 @@ MojoResult Dispatcher::AddAwakableImplNoLock(

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.

@@ -391,7 +396,7 @@ MojoResult Dispatcher::AddAwakableImplNoLock(

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.

@@ -399,8 +404,7 @@ void Dispatcher::RemoveAwakableImplNoLock(Awakable* /*awakable*/,

*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_);

@@ -409,7 +413,6 @@ void Dispatcher::StartSerializeImplNoLock(Channel* /*channel*/,

}

bool Dispatcher::EndSerializeAndCloseImplNoLock(

- Channel* /*channel*/,

void* /*destination*/,

size_t* /*actual_size*/,

embedder::PlatformHandleVector* /*platform_handles*/) {

@@ -421,7 +424,7 @@ bool Dispatcher::EndSerializeAndCloseImplNoLock(

}

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).

@@ -429,7 +432,7 @@ bool Dispatcher::IsBusyNoLock() const {

}

void Dispatcher::CloseNoLock() {

- mutex_.AssertHeld();

+ lock_.AssertAcquired();

DCHECK(!is_closed_);

is_closed_ = true;

@@ -439,7 +442,7 @@ void Dispatcher::CloseNoLock() {

scoped_refptr<Dispatcher>

Dispatcher::CreateEquivalentDispatcherAndCloseNoLock() {

- mutex_.AssertHeld();

+ lock_.AssertAcquired();

DCHECK(!is_closed_);

is_closed_ = true;

@@ -447,25 +450,24 @@ Dispatcher::CreateEquivalentDispatcherAndCloseNoLock() {

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

@@ -477,10 +479,10 @@ bool Dispatcher::EndSerializeAndClose(

// 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);

}

@@ -488,7 +490,10 @@ bool Dispatcher::EndSerializeAndClose(

void DispatcherTransport::End() {

DCHECK(dispatcher_);

- dispatcher_->mutex_.Unlock();

+ dispatcher_->lock_.Release();

+ dispatcher_->TransportEnded();

dispatcher_ = nullptr;

}

« no previous file with comments | « mojo/edk/system/dispatcher.h ('k') | mojo/edk/system/dispatcher_unittest.cc » ('j') | mojo/edk/system/raw_channel.cc » ('J')