WIP: A couple of more ChannelMojo speedup ideas

ipc/mojo/async_handle_waiter.cc

 // Copyright 2015 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 "ipc/mojo/async_handle_waiter.h"
 
 #include "base/atomic_ref_count.h"
 #include "base/bind.h"
 #include "base/bind_helpers.h"
 #include "base/location.h"
@@ skipping 15 matching lines @@
 class AsyncHandleWaiter::Context
     : public base::RefCountedThreadSafe<AsyncHandleWaiter::Context,
                                         AsyncHandleWaiterContextTraits>,
       public base::MessageLoopForIO::IOObserver {
  public:
   Context(base::WeakPtr<AsyncHandleWaiter> waiter)
       : io_runner_(base::MessageLoopForIO::current()->task_runner()),
         waiter_(waiter),
         last_result_(MOJO_RESULT_INTERNAL),
         processing_(false),
-        should_invoke_callback_(false) {
+        should_invoke_callback_(false),
+        should_invoke_message_callback_(false) {
     base::MessageLoopForIO::current()->AddIOObserver(this);
   }
 
+  bool MessageWasReceived(const void* bytes, uint32_t num_bytes) {
+    DCHECK(IsCalledFromIOHandler());
+
+    // For simplicity, we don't steal more than one message at a time.
+    if (!message_body_.empty())
+      return false;
+
+    should_invoke_message_callback_ = true;
+    message_body_.resize(num_bytes);
+    memcpy(&message_body_[0], bytes, num_bytes);
+    return true;
+  }
+
   void HandleIsReady(MojoResult result) {
     last_result_ = result;
 
     // If the signaling happens in the IO handler, use |IOObserver| callback
     // to invoke the callback.
     if (IsCalledFromIOHandler()) {
       should_invoke_callback_ = true;
       return;
     }
 
@@ skipping 21 matching lines @@
   }
 
   // Called from |io_runner_| thus safe to touch |waiter_|.
   void InvokeWaiterCallback() {
     MojoResult result = last_result_;
     last_result_ = MOJO_RESULT_INTERNAL;
     if (waiter_)
       waiter_->InvokeCallback(result);
   }
 
+  void InvokeWaiterMessageCallback() {
+    if (waiter_) {
+      DCHECK(!message_body_.empty());
+      waiter_->InvokeMessageCallback(
+          &message_body_[0], static_cast<uint32_t>(message_body_.size()));
+    }
+
+    message_body_.clear();
+  }
+
   // IOObserver implementation:
 
   void WillProcessIOEvent() override {
     DCHECK(!should_invoke_callback_);
     DCHECK(!processing_);
     processing_ = true;
   }
 
   void DidProcessIOEvent() override {
     DCHECK(processing_);
 
     // The zero |waiter_| indicates that |this| have lost the owner and can be
     // under destruction. So we cannot wrap it with a |scoped_refptr| anymore.
     if (!waiter_) {
-      should_invoke_callback_ = false;
+      should_invoke_callback_ = should_invoke_message_callback_ = false;
       processing_ = false;
       return;
     }
 
     // We have to protect |this| because |AsyncHandleWaiter| can be
     // deleted during the callback.
     scoped_refptr<Context> protect(this);
-    while (should_invoke_callback_) {
-      should_invoke_callback_ = false;
-      InvokeWaiterCallback();
+    while (should_invoke_callback_ || should_invoke_message_callback_) {
+      if (should_invoke_callback_) {
+        should_invoke_callback_ = false;
+        InvokeWaiterCallback();
+      }
+
+      if (should_invoke_message_callback_) {
+        should_invoke_message_callback_ = false;
+        InvokeWaiterMessageCallback();
+      }
     }
 
     processing_ = false;
   }
 
   // Only |io_runner_| is accessed from arbitrary threads.  Others are touched
   // only from the IO thread.
   const scoped_refptr<base::TaskRunner> io_runner_;
 
   const base::WeakPtr<AsyncHandleWaiter> waiter_;
   MojoResult last_result_;
+  std::vector<unsigned char> message_body_;
   bool processing_;
-  bool should_invoke_callback_;
+  bool should_invoke_callback_;          // XXX: Rename
+  bool should_invoke_message_callback_;  // XXX: Rename
 
   DISALLOW_COPY_AND_ASSIGN(Context);
 };
 
-AsyncHandleWaiter::AsyncHandleWaiter(base::Callback<void(MojoResult)> callback)
-    : callback_(callback),
-      weak_factory_(this) {
+AsyncHandleWaiter::AsyncHandleWaiter(AsyncHandleWaiter::Delegate* delegate)
+    : delegate_(delegate), weak_factory_(this) {
   context_ = new Context(weak_factory_.GetWeakPtr());
+  context_callback_ = base::Bind(&Context::HandleIsReady, context_);
+  context_message_callback_ =
+      base::Bind(&Context::MessageWasReceived, context_);
 }
 
 AsyncHandleWaiter::~AsyncHandleWaiter() {
 }
 
+void AsyncHandleWaiter::ClearMessageCallback(MojoHandle handle) {
+  MojoResult rv = mojo::embedder::SetAsyncMessageCallback(
+      handle, base::Callback<bool(const void*, uint32_t)>());
+  DCHECK_EQ(rv, MOJO_RESULT_OK);
+}
+
+void AsyncHandleWaiter::SetMessageCallback(MojoHandle handle) {
+  MojoResult rv = mojo::embedder::SetAsyncMessageCallback(
+      handle, context_message_callback_);
+  DCHECK_EQ(rv, MOJO_RESULT_OK);
+}
+
 MojoResult AsyncHandleWaiter::Wait(MojoHandle handle,
                                    MojoHandleSignals signals) {
-  return mojo::embedder::AsyncWait(
-      handle, signals, base::Bind(&Context::HandleIsReady, context_));
+  return mojo::embedder::AsyncWait(handle, signals, context_callback_);
 }
 
 void AsyncHandleWaiter::InvokeCallback(MojoResult result) {
-  callback_.Run(result);
+  delegate_->PipeIsReady(result);
+}
+
+void AsyncHandleWaiter::InvokeMessageCallback(const void* bytes,
+                                              uint32_t num_bytes) {
+  delegate_->MessageWasArrived(bytes, num_bytes);
 }
 
 // static
 void AsyncHandleWaiterContextTraits::Destruct(
     const AsyncHandleWaiter::Context* context) {
   context->io_runner_->PostTask(
       FROM_HERE,
       base::Bind(&base::DeletePointer<const AsyncHandleWaiter::Context>,
                  base::Unretained(context)));
 }
 
 }  // namespace internal
 }  // namespace IPC