WIP: A couple of more ChannelMojo speedup ideas

third_party/mojo/src/mojo/edk/system/message_pipe.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/message_pipe.h"
 
 #include "base/logging.h"
 #include "mojo/edk/system/channel.h"
 #include "mojo/edk/system/channel_endpoint.h"
 #include "mojo/edk/system/channel_endpoint_id.h"
@@ skipping 188 matching lines @@
                                  Awakable* awakable,
                                  HandleSignalsState* signals_state) {
   DCHECK(port == 0 || port == 1);
 
   base::AutoLock locker(lock_);
   DCHECK(endpoints_[port]);
 
   endpoints_[port]->RemoveAwakable(awakable, signals_state);
 }
 
+MojoResult MessagePipe::SetAsyncMessageCallback(
+    unsigned port,
+    const AsyncMessageCallback& callback) {
+  base::AutoLock locker(lock_);
+  DCHECK(endpoints_[port]);
+  return endpoints_[port]->SetAsyncMessageCallback(callback);
+}
+
 void MessagePipe::StartSerialize(unsigned /*port*/,
                                  Channel* channel,
                                  size_t* max_size,
                                  size_t* max_platform_handles) {
   *max_size = channel->GetSerializedEndpointSize();
   *max_platform_handles = 0;
 }
 
 bool MessagePipe::EndSerialize(
     unsigned port,
@@ skipping 47 matching lines @@
     endpoints_[peer_port].reset();
   }
 
   endpoints_[port]->Close();
   endpoints_[port].reset(replacement_endpoint);
 
   *actual_size = channel->GetSerializedEndpointSize();
   return true;
 }
 
-bool MessagePipe::OnReadMessage(unsigned port, MessageInTransit* message) {
+bool MessagePipe::OnReadMessage(
+    unsigned port,
+    MessageInTransit::ReadContext& reading_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 = EnqueueMessageNoLock(GetPeerPort(port),
-                                           make_scoped_ptr(message), nullptr);
+  MojoResult result =
+      DispatchOnReadMessageNoLock(GetPeerPort(port), reading_message);
   DLOG_IF(WARNING, result != MOJO_RESULT_OK)
-      << "EnqueueMessageNoLock() failed (result  = " << result << ")";
+      << "DispatchOnReadMessageNoLock() failed (result  = " << result << ")";
   return true;
 }
 
 void MessagePipe::OnDetachFromChannel(unsigned port) {
   Close(port);
 }
 
 MessagePipe::MessagePipe() {
 }
 
 MessagePipe::~MessagePipe() {
   // Owned by the dispatchers. The owning dispatchers should only release us via
   // their |Close()| method, which should inform us of being closed via our
   // |Close()|. Thus these should already be null.
   DCHECK(!endpoints_[0]);
   DCHECK(!endpoints_[1]);
 }
 
+MojoResult MessagePipe::DispatchOnReadMessageNoLock(
+    unsigned port,
+    MessageInTransit::ReadContext& reading_message) {
+  DCHECK(port == 0 || port == 1);
+
+  DCHECK_EQ(reading_message.view().type(), MessageInTransit::kTypeEndpoint);
+  DCHECK(endpoints_[GetPeerPort(port)]);
+
+  // The destination port need not be open, unlike the source port.
+  if (!endpoints_[port])
+    return MOJO_RESULT_FAILED_PRECONDITION;
+
+  // The endpoint's |EnqueueMessage()| may not report failure.
+  endpoints_[port]->OnReadMessage(reading_message);
+  return MOJO_RESULT_OK;
+}
+
 MojoResult MessagePipe::EnqueueMessageNoLock(
     unsigned port,
     scoped_ptr<MessageInTransit> message,
     std::vector<DispatcherTransport>* transports) {
   DCHECK(port == 0 || port == 1);
   DCHECK(message);
 
   DCHECK_EQ(message->type(), MessageInTransit::kTypeEndpoint);
   DCHECK(endpoints_[GetPeerPort(port)]);
 
@@ skipping 51 matching lines @@
       LOG(WARNING) << "Enqueueing null dispatcher";
       dispatchers->push_back(nullptr);
     }
   }
   message->SetDispatchers(dispatchers.Pass());
   return MOJO_RESULT_OK;
 }
 
 }  // namespace system
 }  // namespace mojo