Fix/reland 505555fa8031fc06d67e9ed43e3021d8d84cb44f (crrev.com/773983002). This adds ChannelEndpoint::ResetChannelNoLock(). I needed to set is_detached_from_channel_ in a bunch of other places. R=yzshen@chromium.org Review URL: https://codereview.chromium.org/773113006
diff --git a/mojo/edk/system/channel_endpoint.cc b/mojo/edk/system/channel_endpoint.cc index a8bbbb2..5da4056 100644 --- a/mojo/edk/system/channel_endpoint.cc +++ b/mojo/edk/system/channel_endpoint.cc
@@ -5,6 +5,7 @@ #include "mojo/edk/system/channel_endpoint.h" #include "base/logging.h" +#include "base/threading/platform_thread.h" #include "mojo/edk/system/channel.h" #include "mojo/edk/system/channel_endpoint_client.h" @@ -14,7 +15,10 @@ ChannelEndpoint::ChannelEndpoint(ChannelEndpointClient* client, unsigned client_port, MessageInTransitQueue* message_queue) - : client_(client), client_port_(client_port), channel_(nullptr) { + : client_(client), + client_port_(client_port), + channel_(nullptr), + is_detached_from_channel_(false) { DCHECK(client_.get() || message_queue); if (message_queue) @@ -39,21 +43,27 @@ return WriteMessageNoLock(message.Pass()); } -void ChannelEndpoint::DetachFromClient() { - { - base::AutoLock locker(lock_); - DCHECK(client_.get()); - client_ = nullptr; +bool ChannelEndpoint::ReplaceClient(ChannelEndpointClient* client, + unsigned client_port) { + DCHECK(client); - if (!channel_) - return; - DCHECK(local_id_.is_valid()); - DCHECK(remote_id_.is_valid()); - channel_->DetachEndpoint(this, local_id_, remote_id_); - channel_ = nullptr; - local_id_ = ChannelEndpointId(); - remote_id_ = ChannelEndpointId(); - } + base::AutoLock locker(lock_); + DCHECK(client_.get()); + DCHECK(client != client_.get() || client_port != client_port_); + client_ = client; + client_port_ = client_port; + return !is_detached_from_channel_; +} + +void ChannelEndpoint::DetachFromClient() { + base::AutoLock locker(lock_); + DCHECK(client_.get()); + client_ = nullptr; + + if (!channel_) + return; + channel_->DetachEndpoint(this, local_id_, remote_id_); + ResetChannelNoLock(); } void ChannelEndpoint::AttachAndRun(Channel* channel, @@ -78,30 +88,50 @@ if (!client_.get()) { channel_->DetachEndpoint(this, local_id_, remote_id_); - channel_ = nullptr; - local_id_ = ChannelEndpointId(); - remote_id_ = ChannelEndpointId(); + ResetChannelNoLock(); } } void ChannelEndpoint::OnReadMessage(scoped_ptr<MessageInTransit> message) { scoped_refptr<ChannelEndpointClient> client; - unsigned client_port; - { - base::AutoLock locker(lock_); - DCHECK(channel_); - if (!client_.get()) { - // This isn't a failure per se. (It just means that, e.g., the other end - // of the message point closed first.) - return; + unsigned client_port = 0; + + // This loop is to make |ReplaceClient()| work. We can't call the client's + // |OnReadMessage()| under our lock, so by the time we do that, |client| may + // no longer be our client. + // + // In that case, |client| must return false. We'll then yield, and retry with + // the new client. (Theoretically, the client could be replaced again.) + // + // This solution isn't terribly elegant, but it's the least costly way of + // handling/avoiding this (very unlikely) race. (Other solutions -- e.g., + // adding a client message queue, which the client only fetches messages from + // -- impose significant cost in the common case.) + for (;;) { + { + base::AutoLock locker(lock_); + if (!channel_ || !client_.get()) { + // This isn't a failure per se. (It just means that, e.g., the other end + // of the message point closed first.) + return; + } + + // If we get here in a second (third, etc.) iteration of the loop, it's + // because |ReplaceClient()| was called. + DCHECK(client_ != client || client_port_ != client_port); + + // Take a ref, and call |OnReadMessage()| outside the lock. + client = client_; + client_port = client_port_; } - // Take a ref, and call |OnReadMessage()| outside the lock. - client = client_; - client_port = client_port_; - } + if (client->OnReadMessage(client_port, message.get())) { + ignore_result(message.release()); + break; + } - client->OnReadMessage(client_port, message.Pass()); + base::PlatformThread::YieldCurrentThread(); + } } void ChannelEndpoint::DetachFromChannel() { @@ -119,15 +149,17 @@ // |channel_| may already be null if we already detached from the channel in // |DetachFromClient()| by calling |Channel::DetachEndpoint()| (and there // are racing detaches). - if (channel_) { - DCHECK(local_id_.is_valid()); - DCHECK(remote_id_.is_valid()); - channel_ = nullptr; - local_id_ = ChannelEndpointId(); - remote_id_ = ChannelEndpointId(); - } + if (channel_) + ResetChannelNoLock(); + else + DCHECK(is_detached_from_channel_); } + // If |ReplaceClient()| is called (from another thread) after the above locked + // section but before we call |OnDetachFromChannel()|, |ReplaceClient()| + // return false to notify the caller that the channel was already detached. + // (The old client has to accept the arguably-spurious call to + // |OnDetachFromChannel()|.) if (client.get()) client->OnDetachFromChannel(client_port); } @@ -154,5 +186,17 @@ return channel_->WriteMessage(message.Pass()); } +void ChannelEndpoint::ResetChannelNoLock() { + DCHECK(channel_); + DCHECK(local_id_.is_valid()); + DCHECK(remote_id_.is_valid()); + DCHECK(!is_detached_from_channel_); + + channel_ = nullptr; + local_id_ = ChannelEndpointId(); + remote_id_ = ChannelEndpointId(); + is_detached_from_channel_ = true; +} + } // namespace system } // namespace mojo
diff --git a/mojo/edk/system/channel_endpoint.h b/mojo/edk/system/channel_endpoint.h index 71b6f9b..90ee86b 100644 --- a/mojo/edk/system/channel_endpoint.h +++ b/mojo/edk/system/channel_endpoint.h
@@ -130,6 +130,15 @@ // called.) bool EnqueueMessage(scoped_ptr<MessageInTransit> message); + // Called to *replace* current client with a new client (which must differ + // from the existing client). This must not be called after + // |DetachFromClient()| has been called. + // + // This returns true in the typical case, and false if this endpoint has been + // detached from the channel, in which case the caller should probably call + // its (new) client's |OnDetachFromChannel()|. + bool ReplaceClient(ChannelEndpointClient* client, unsigned client_port); + // Called before the |ChannelEndpointClient| gives up its reference to this // object. void DetachFromClient(); @@ -156,6 +165,10 @@ // Must be called with |lock_| held. bool WriteMessageNoLock(scoped_ptr<MessageInTransit> message); + // Resets |channel_| to null (and sets |is_detached_from_channel_|). This may + // only be called if |channel_| is non-null. Must be called with |lock_| held. + void ResetChannelNoLock(); + // Protects the members below. base::Lock lock_; @@ -168,6 +181,9 @@ // WARNING: |ChannelEndpointClient| methods must not be called under |lock_|. // Thus to make such a call, a reference must first be taken under |lock_| and // the lock released. + // WARNING: Beware of interactions with |ReplaceClient()|. By the time the + // call is made, the client may have changed. This must be detected and dealt + // with. scoped_refptr<ChannelEndpointClient> client_; unsigned client_port_; @@ -177,6 +193,9 @@ Channel* channel_; ChannelEndpointId local_id_; ChannelEndpointId remote_id_; + // This distinguishes the two cases of |channel| being null: not yet attached + // versus detached. + bool is_detached_from_channel_; // This queue is used before we're running on a channel and ready to send // messages to the channel.
diff --git a/mojo/edk/system/channel_endpoint_client.h b/mojo/edk/system/channel_endpoint_client.h index 7a7d5b8..c758e9c 100644 --- a/mojo/edk/system/channel_endpoint_client.h +++ b/mojo/edk/system/channel_endpoint_client.h
@@ -38,8 +38,9 @@ // called by |Channel| when it receives a message for the |ChannelEndpoint|. // (|port| is the value passed to |ChannelEndpoint|'s constructor as // |client_port|.) - virtual void OnReadMessage(unsigned port, - scoped_ptr<MessageInTransit> message) = 0; + // + // This should return true if it accepted (and took ownership of) |message|. + virtual bool OnReadMessage(unsigned port, MessageInTransit* message) = 0; // Called by |ChannelEndpoint| when the |Channel| is relinquishing its pointer // to the |ChannelEndpoint| (and vice versa). After this is called,
diff --git a/mojo/edk/system/message_pipe.cc b/mojo/edk/system/message_pipe.cc index ee5036f..2bf313f 100644 --- a/mojo/edk/system/message_pipe.cc +++ b/mojo/edk/system/message_pipe.cc
@@ -139,7 +139,9 @@ std::vector<DispatcherTransport>* transports, MojoWriteMessageFlags flags) { DCHECK(port == 0 || port == 1); - return EnqueueMessage( + + base::AutoLock locker(lock_); + return EnqueueMessageNoLock( GetPeerPort(port), make_scoped_ptr(new MessageInTransit( MessageInTransit::kTypeEndpoint, @@ -287,16 +289,27 @@ return true; } -void MessagePipe::OnReadMessage(unsigned port, - scoped_ptr<MessageInTransit> message) { +bool MessagePipe::OnReadMessage(unsigned port, MessageInTransit* message) { + base::AutoLock locker(lock_); + + if (!endpoints_[port]) { + // This will happen only on the rare occasion that the call to + // |OnReadMessage()| is racing with us calling + // |ChannelEndpoint::ReplaceClient()|, in which case we reject the message, + // and the |ChannelEndpoint| can retry (calling the new client's + // |OnReadMessage()|). + return false; + } + // This is called when the |ChannelEndpoint| for the // |ProxyMessagePipeEndpoint| |port| receives a message (from the |Channel|). // We need to pass this message on to its peer port (typically a // |LocalMessagePipeEndpoint|). - MojoResult result = - EnqueueMessage(GetPeerPort(port), message.Pass(), nullptr); + MojoResult result = EnqueueMessageNoLock(GetPeerPort(port), + make_scoped_ptr(message), nullptr); DLOG_IF(WARNING, result != MOJO_RESULT_OK) - << "EnqueueMessage() failed (result = " << result << ")"; + << "EnqueueMessageNoLock() failed (result = " << result << ")"; + return true; } void MessagePipe::OnDetachFromChannel(unsigned port) { @@ -314,7 +327,7 @@ DCHECK(!endpoints_[1]); } -MojoResult MessagePipe::EnqueueMessage( +MojoResult MessagePipe::EnqueueMessageNoLock( unsigned port, scoped_ptr<MessageInTransit> message, std::vector<DispatcherTransport>* transports) { @@ -322,8 +335,6 @@ DCHECK(message); DCHECK_EQ(message->type(), MessageInTransit::kTypeEndpoint); - - base::AutoLock locker(lock_); DCHECK(endpoints_[GetPeerPort(port)]); // The destination port need not be open, unlike the source port.
diff --git a/mojo/edk/system/message_pipe.h b/mojo/edk/system/message_pipe.h index e8f97d0..791eece 100644 --- a/mojo/edk/system/message_pipe.h +++ b/mojo/edk/system/message_pipe.h
@@ -107,8 +107,7 @@ embedder::PlatformHandleVector* platform_handles); // |ChannelEndpointClient| methods: - void OnReadMessage(unsigned port, - scoped_ptr<MessageInTransit> message) override; + bool OnReadMessage(unsigned port, MessageInTransit* message) override; void OnDetachFromChannel(unsigned port) override; private: @@ -117,12 +116,12 @@ // This is used internally by |WriteMessage()| and by |OnReadMessage()|. // |transports| may be non-null only if it's nonempty and |message| has no - // dispatchers attached. - MojoResult EnqueueMessage(unsigned port, - scoped_ptr<MessageInTransit> message, - std::vector<DispatcherTransport>* transports); + // dispatchers attached. Must be called with |lock_| held. + MojoResult EnqueueMessageNoLock(unsigned port, + scoped_ptr<MessageInTransit> message, + std::vector<DispatcherTransport>* transports); - // Helper for |EnqueueMessage()|. Must be called with |lock_| held. + // Helper for |EnqueueMessageNoLock()|. Must be called with |lock_| held. MojoResult AttachTransportsNoLock( unsigned port, MessageInTransit* message,