Use mojo for Chrome Loading, Part 1

content/browser/loader/mojo_async_resource_handler.cc

+// Copyright 2016 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 "content/browser/loader/mojo_async_resource_handler.h"
+
+#include <utility>
+
+#include "base/command_line.h"
+#include "base/containers/hash_tables.h"
+#include "base/logging.h"
+#include "base/macros.h"
+#include "base/strings/string_number_conversions.h"
+#include "base/time/time.h"
+#include "content/browser/loader/netlog_observer.h"
+#include "content/browser/loader/resource_dispatcher_host_impl.h"
+#include "content/browser/loader/resource_request_info_impl.h"
+#include "content/common/resource_request_completion_status.h"
+#include "content/public/browser/resource_dispatcher_host_delegate.h"
+#include "content/public/common/resource_response.h"
+#include "mojo/public/c/system/data_pipe.h"
+#include "mojo/public/cpp/system/data_pipe.h"
+#include "net/base/io_buffer.h"
+#include "net/base/load_flags.h"
+#include "net/base/mime_sniffer.h"
+#include "net/log/net_log.h"
+#include "net/url_request/redirect_info.h"
+
+namespace content {
+namespace {
+
+int g_max_allocation_size = 1024 * 32;
+// MimeTypeResourceHandler *implicitly* requires that the buffer size
+// returned from OnWillRead should be larger than certain size.
+// TODO(yhirano): Fix MimeTypeResourceHandler.
+size_t kMinAllocationSize = 2 * net::kMaxBytesToSniff;
+
+void GetNumericArg(const std::string& name, int* result) {
+  const std::string& value =
+      base::CommandLine::ForCurrentProcess()->GetSwitchValueASCII(name);
+  if (!value.empty())
+    base::StringToInt(value, result);
+}
+
+void InitializeResourceBufferConstants() {
+  static bool did_init = false;
+  if (did_init)
+    return;
+  did_init = true;
+
+  GetNumericArg("resource-buffer-max-allocation-size", &g_max_allocation_size);
+}
+
+}  // namespace
+
+// This class is for sharing the ownership of a ScopedDataPipeProducerHandle.
+class MojoAsyncResourceHandler::SharedWriter final
+    : public base::RefCountedThreadSafe<SharedWriter> {
+ public:
+  explicit SharedWriter(mojo::ScopedDataPipeProducerHandle writer)
+      : writer_(std::move(writer)) {}
+  mojo::DataPipeProducerHandle writer() { return writer_.get(); }
+
+ private:
+  friend class base::RefCountedThreadSafe<SharedWriter>;
+  ~SharedWriter() {}
+
+  const mojo::ScopedDataPipeProducerHandle writer_;
+
+  DISALLOW_COPY_AND_ASSIGN(SharedWriter);
+};
+
+// This class is a IOBuffer subclass for data gotten from a
+// ScopedDataPipeProducerHandle.
+class MojoAsyncResourceHandler::WriterIOBuffer final

[mmenke, 2016/07/14 21:23:04]

Hrm...Wonder if it would be cleaner to merge this and SharedWriter.  We'd
probably have to make this class implement AllocateBuffer itself, and it would
still have to expose writer_ as well (Or make AllocateBuffer add a pointer to a
method to call back on async completion, and add an EndWriting method).

Trying to figure out how to simplify
MojoAsyncResourceHandler::OnReadCompleted/MojoAsyncResourceHandler::ResumeIfDeferred,
and more generally, clean up the code a bit.

[yhirano, 2016/07/15 11:47:55]

Are you proposing to return the same IOBuffer pointer from OnWillRead by placing
AllocateBuffer in this class? That sounds confusing to me.

As sometimes OnWillRead returns a net::WrappedIOBuffer, I don't think merging
this class with SharedWriter is a good idea. 

[mmenke, 2016/07/15 14:54:07]

Yea, that's what I was thinking.  Currently CopyReadData and ResumeIfDeferred
are quite hard to follow, and just trying to figure out if we can fix that.  May
make things worse, though.  If you prefer it as-is, that's fine.

+    : public net::IOBufferWithSize {
+ public:
+  // |data| and |size| should be gotten from |writer| via BeginWriteDataRaw.
+  // They will be accesible via IOBuffer methods. As |writer| is stored in this
+  // instance, |data| will be kept valid as long as the following conditions
+  // hold:
+  //  1. |data| is not invalidated via EndWriteDataRaw.
+  //  2. |this| instance is alive.
+  WriterIOBuffer(scoped_refptr<MojoAsyncResourceHandler::SharedWriter> writer,
+                 void* data,
+                 size_t size)
+      : net::IOBufferWithSize(static_cast<char*>(data), size),
+        writer_(std::move(writer)) {}
+
+ private:
+  ~WriterIOBuffer() override {
+    // Avoid deleting |data_| in the IOBuffer destructor.
+    data_ = nullptr;
+  }
+
+  // This member is for keeping the writer alive.
+  scoped_refptr<MojoAsyncResourceHandler::SharedWriter> writer_;
+
+  DISALLOW_COPY_AND_ASSIGN(WriterIOBuffer);
+};
+
+MojoAsyncResourceHandler::MojoAsyncResourceHandler(
+    net::URLRequest* request,
+    ResourceDispatcherHostImpl* rdh,
+    mojo::InterfaceRequest<mojom::URLLoader> mojo_request,
+    mojom::URLLoaderClientPtr url_loader_client)
+    : ResourceHandler(request),
+      rdh_(rdh),
+      binding_(this, std::move(mojo_request)),
+      url_loader_client_(std::move(url_loader_client)) {
+  DCHECK(url_loader_client_);
+  InitializeResourceBufferConstants();
+}
+
+MojoAsyncResourceHandler::~MojoAsyncResourceHandler() {
+  if (has_checked_for_sufficient_resources_)
+    rdh_->FinishedWithResourcesForRequest(request());
+}
+
+bool MojoAsyncResourceHandler::OnRequestRedirected(
+    const net::RedirectInfo& redirect_info,
+    ResourceResponse* response,
+    bool* defer) {
+  // Not implemented.
+  return false;
+}
+
+bool MojoAsyncResourceHandler::OnResponseStarted(ResourceResponse* response,
+                                                 bool* defer) {
+  const ResourceRequestInfoImpl* info = GetRequestInfo();
+
+  if (rdh_->delegate()) {
+    rdh_->delegate()->OnResponseStarted(request(), info->GetContext(),
+                                        response);
+  }
+
+  NetLogObserver::PopulateResponseInfo(request(), response);
+
+  response->head.request_start = request()->creation_time();
+  response->head.response_start = base::TimeTicks::Now();
+  sent_received_response_message_ = true;
+  url_loader_client_->OnReceiveResponse(response->head);
+  return true;
+}
+
+bool MojoAsyncResourceHandler::OnWillStart(const GURL& url, bool* defer) {
+  return true;
+}
+
+bool MojoAsyncResourceHandler::OnBeforeNetworkStart(const GURL& url,
+                                                    bool* defer) {

[mmenke, 2016/07/14 21:23:04]

note:  This method had been removed from the interface, I believe.

[yhirano, 2016/07/19 09:42:47]

OnBeforeNetworkStart was removed from URLRequest::Delegate but not from
ResourceHandler, I think.

+  return true;
+}
+
+bool MojoAsyncResourceHandler::OnWillRead(scoped_refptr<net::IOBuffer>* buf,
+                                          int* buf_size,
+                                          int min_size) {
+  DCHECK_EQ(-1, min_size);
+
+  if (!CheckForSufficientResource())
+    return false;
+
+  if (!shared_writer_) {
+    MojoCreateDataPipeOptions options;
+    options.struct_size = sizeof(MojoCreateDataPipeOptions);
+    options.flags = MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE;
+    options.element_num_bytes = 1;
+    options.capacity_num_bytes = g_max_allocation_size;

[mmenke, 2016/07/14 21:23:04]

This is the total capacity of the channel, not the number of bytes returned by
each BeginWrite call, right?  Should this be the total shared memory size of the
AsyncResourceHandler (512k) instead?  Happy to see this tuned later, but think
we should try and match AsyncResourceHandler's behavior for now, to try to avoid
a perf regression due to changed behavior.  And I rather suspect using a 32k
total buffer capacity would result in a perf regression.

In that same vein, we'd change buf_size to std::min(buffer_->size(),
g_max_allocation_size);

We would still potentially end up reading sizes less than kMinAllocationSize in
the old code, but the behavior would more closely match that of
AsyncResourceHandler.

[yhirano, 2016/07/19 09:42:47]

I renamed it to g_allocation_size and changed the default value.
I don't understand the benefit: Can you explain it? if |available| is greater
than |g_max_allocation_size|, exposing smaller size will not save any memory.

[mmenke, 2016/07/19 15:50:28]

It's not about memory, it's about performance.  Let's suppose it takes 10
milliseconds to process 32k of data and each IPC call introduces 100
milliseconds of latency (These numbers are way too big, just using them to
demonstrate my point).

Suppose we have a 1MB response.

If we send it in 2 512k chunks, then it takes 100+160+100 milliseconds to send
the first chunk, for it to be processed, and for us to learn that the renderer
needs more data.  The same for the second chunk, so that's 360*2 = 720
milliseconds to handle the body and which point we try to read again, learn the
file was complete.  Then we tell the renderer the response is complete, which
has another 100 millisecond delay, so that's about 820 ms of latency from when
we read the first 512k to when the request completes.  All of this is blocking
time, during which we can't do anything else, so add to that any internal work
the network stack does when we read the data.  It can't be done in parallel,
since we don't have a pending read while we're waiting to pass the data around.

Now consider if we read the data in 32 32k chunks.  When we read the first
chunk, we post an IPC, which takes 100 milliseconds to reach the renderer, 10
milliseconds of processing, and 100 milliseconds for the browser process to
learn it was consumed...But we can be doing the next 32k read when that's
happening, so before we hear back, and maybe before we renderer even received
the last 32k, we're sending 32k more of data.  Regardless of how fast the
network is, we learn the renderer processed the first 32k before the time the
renderer starts processing the next 11th 32k chunk.  So basically, we're never
stuck waiting for an IPC to tell us the renderer has finished handling an
earlier read.

So the end result is 100 + 320 + 100 = 520 millseconds (We don't even need the
extra half-trip to learn the response is complete) instead of 720 millisecond. 
Make sense?

[mmenke, 2016/07/19 20:50:26]

It's not necessarily the case that a bunch of 32k writes are faster (There are
more context switches, for instance), but if we see a performance regression
with the Mojo code, I want to minimize potential causes to investigate.

[yhirano, 2016/07/20 13:38:36]

Thanks, I see. I added kMaxChunkSize.

+    mojo::DataPipe data_pipe(options);
+
+    url_loader_client_->OnStartLoadingResponseBody(
+        std::move(data_pipe.consumer_handle));
+    if (!data_pipe.producer_handle.is_valid())
+      return false;
+
+    shared_writer_ = new SharedWriter(std::move(data_pipe.producer_handle));
+
+    void* data = nullptr;
+    uint32_t available = 0;
+    MojoResult result = BeginWrite(&data, &available);
+
+    if (result != MOJO_RESULT_OK && result != MOJO_RESULT_SHOULD_WAIT)
+      return false;
+    if (result == MOJO_RESULT_OK && available >= kMinAllocationSize) {
+      *buf = buffer_ = new WriterIOBuffer(shared_writer_, data, available);
+      *buf_size = buffer_->size();
+      return true;
+    }
+
+    if (result == MOJO_RESULT_OK) {
+      // The allocated buffer is too small.
+      if (EndWrite(0) != MOJO_RESULT_OK)
+        return false;
+    }
+    DCHECK(!is_using_io_buffer_not_from_writer_);
+    is_using_io_buffer_not_from_writer_ = true;
+    buffer_ = new net::IOBufferWithSize(kMinAllocationSize);
+  }
+
+  if (is_using_io_buffer_not_from_writer_) {
+    DCHECK_EQ(0u, buffer_offset_);
+    *buf = new net::WrappedIOBuffer(buffer_->data());

[mmenke, 2016/07/15 14:54:07]

Hrm...Why can't we just return buffer_, like we do in the fallthrough case?

[yhirano, 2016/07/19 09:42:47]

You're right. fixed.

+    *buf_size = buffer_->size();
+    return true;
+  }
+
+  *buf = buffer_;
+  *buf_size = buffer_->size();
+  return true;
+}
+
+bool MojoAsyncResourceHandler::OnReadCompleted(int bytes_read, bool* defer) {

[mmenke, 2016/07/14 21:23:05]

I think most branches here could use some documentation.

[yhirano, 2016/07/15 11:47:55]

Personally PS46 version is much easier to explain for me.

[mmenke, 2016/07/15 14:54:07]

I find them both extremely obscure, largely because of the
is_using_io_buffer_not_from_writer_ stuff.

Longer term, I think we should probably make
MojoAsyncResourceHandler::OnWillRead be able to complete asynchronously, and get
rid of is_using_io_buffer_not_from_writer_, but certainly not in this CL. 
Unclear how difficult making OnWillRead async will be, but it will touch a fair
bit of code.

[yhirano, 2016/07/19 09:42:47]

I agree that both are hard to understand and changing OnWillRead may mitigate
the pain (and I think MimeTypeResourceHandler's implicit expectation should be
fixed too).

I reverted the code back to PS46 version and added some comments.

+  DCHECK_GE(bytes_read, 0);
+  DCHECK(buffer_);
+
+  if (!bytes_read)
+    return true;
+
+  if (is_using_io_buffer_not_from_writer_) {
+    // Couldn't allocate a buffer on the data pipe in OnWillRead.
+    DCHECK(!did_defer_);
+    DCHECK_EQ(0u, buffer_bytes_read_);
+    buffer_bytes_read_ = bytes_read;
+    if (!CopyReadData(defer))
+      return false;
+  } else {
+    if (EndWrite(bytes_read) != MOJO_RESULT_OK)
+      return false;
+  }
+  if (!is_using_io_buffer_not_from_writer_) {
+    DCHECK(!*defer);
+    if (!AllocateBuffer(defer))
+      return false;
+  }
+  if (*defer) {
+    did_defer_ = true;
+    OnDefer();
+  }
+  return true;
+}
+
+void MojoAsyncResourceHandler::OnDataDownloaded(int bytes_downloaded) {
+  // Not implemented.
+}
+
+void MojoAsyncResourceHandler::FollowRedirect() {
+  NOTIMPLEMENTED();
+}
+
+void MojoAsyncResourceHandler::Cancel() {
+  NOTIMPLEMENTED();
+}
+
+void MojoAsyncResourceHandler::SetAllocationSizeForTesting(size_t size) {
+  g_max_allocation_size = size;

[mmenke, 2016/07/14 21:23:05]

Can we just set this via the command line flag?  Each test runs in its own
process, right?  If not...the current unittests don't correctly restore it to
its original size on completion, and they should.

[yhirano, 2016/07/19 09:42:47]

I think some test cases share the same process. I added
SetAllocationSizeForTesting(kDefaultAllocationSize) in TearDown functions.

+}
+
+MojoResult MojoAsyncResourceHandler::BeginWrite(void** data,
+                                                uint32_t* available) {
+  return mojo::BeginWriteDataRaw(shared_writer_->writer(), data, available,
+                                 MOJO_WRITE_DATA_FLAG_NONE);
+}
+
+MojoResult MojoAsyncResourceHandler::EndWrite(uint32_t written) {
+  return mojo::EndWriteDataRaw(shared_writer_->writer(), written);
+}
+
+void MojoAsyncResourceHandler::OnResponseCompleted(
+    const net::URLRequestStatus& status,
+    const std::string& security_info,
+    bool* defer) {
+  shared_writer_ = nullptr;
+  buffer_ = nullptr;
+  handle_watcher_.Stop();
+
+  const ResourceRequestInfoImpl* info = GetRequestInfo();
+
+  // TODO(gavinp): Remove this CHECK when we figure out the cause of
+  // http://crbug.com/124680 . This check mirrors closely check in
+  // WebURLLoaderImpl::OnCompletedRequest that routes this message to a WebCore
+  // ResourceHandleInternal which asserts on its state and crashes. By crashing
+  // when the message is sent, we should get better crash reports.
+  CHECK(status.status() != net::URLRequestStatus::SUCCESS ||
+        sent_received_response_message_);
+
+  int error_code = status.error();
+  bool was_ignored_by_handler = info->WasIgnoredByHandler();
+
+  DCHECK(status.status() != net::URLRequestStatus::IO_PENDING);
+  // If this check fails, then we're in an inconsistent state because all
+  // requests ignored by the handler should be canceled (which should result in
+  // the ERR_ABORTED error code).
+  DCHECK(!was_ignored_by_handler || error_code == net::ERR_ABORTED);
+
+  ResourceRequestCompletionStatus request_complete_data;
+  request_complete_data.error_code = error_code;
+  request_complete_data.was_ignored_by_handler = was_ignored_by_handler;
+  request_complete_data.exists_in_cache = request()->response_info().was_cached;
+  request_complete_data.security_info = security_info;
+  request_complete_data.completion_time = base::TimeTicks::Now();
+  request_complete_data.encoded_data_length =
+      request()->GetTotalReceivedBytes();
+
+  url_loader_client_->OnComplete(request_complete_data);
+}
+
+bool MojoAsyncResourceHandler::CopyReadData(bool* defer) {
+  while (true) {
+    if (buffer_bytes_read_ == 0) {
+      // All bytes are copied.
+      buffer_offset_ = 0;
+      is_using_io_buffer_not_from_writer_ = false;
+      buffer_ = nullptr;
+      return true;
+    }
+    void* data = nullptr;
+    uint32_t available = 0;
+    MojoResult result = BeginWrite(&data, &available);
+    if (result == MOJO_RESULT_SHOULD_WAIT) {
+      *defer = true;
+      return true;
+    }
+    if (result != MOJO_RESULT_OK)
+      return false;
+
+    size_t copied_size =
+        std::min(static_cast<uint32_t>(buffer_bytes_read_), available);
+    memcpy(data, buffer_->data() + buffer_offset_, copied_size);
+    buffer_offset_ += copied_size;
+    buffer_bytes_read_ -= copied_size;
+    result = EndWrite(copied_size);
+    if (result != MOJO_RESULT_OK)
+      return false;
+  }
+}
+
+bool MojoAsyncResourceHandler::AllocateBuffer(bool* defer) {
+  void* data = nullptr;
+  uint32_t available = 0;
+  MojoResult result = BeginWrite(&data, &available);
+  if (result == MOJO_RESULT_SHOULD_WAIT) {
+    *defer = true;
+    return true;
+  }
+  if (result != MOJO_RESULT_OK)
+    return false;
+  buffer_ = new WriterIOBuffer(shared_writer_, data, available);
+  return true;
+}
+
+void MojoAsyncResourceHandler::ResumeIfDeferred() {

[mmenke, 2016/07/14 21:23:05]

I think most branches here could use some documentation.

[yhirano, 2016/07/19 09:42:47]

Done.

+  DCHECK(did_defer_);
+  // did_defer_ is set only in OnReadCompleted (and here).
+  did_defer_ = false;
+  if (is_using_io_buffer_not_from_writer_) {
+    DCHECK_GT(buffer_bytes_read_, 0u);
+    if (!CopyReadData(&did_defer_)) {
+      controller()->CancelWithError(net::ERR_FAILED);
+      return;
+    }
+  }
+  if (!is_using_io_buffer_not_from_writer_) {
+    DCHECK(!did_defer_);
+    if (!AllocateBuffer(&did_defer_)) {
+      controller()->CancelWithError(net::ERR_FAILED);
+      return;
+    }
+  }
+  if (did_defer_) {
+    // Continue waiting.
+    handle_watcher_.Start(shared_writer_->writer(), MOJO_HANDLE_SIGNAL_WRITABLE,
+                          MOJO_DEADLINE_INDEFINITE,
+                          base::Bind(&MojoAsyncResourceHandler::OnWritable,
+                                     base::Unretained(this)));
+    return;
+  }
+  request()->LogUnblocked();
+  controller()->Resume();
+}
+
+void MojoAsyncResourceHandler::OnDefer() {
+  request()->LogBlockedBy("MojoAsyncResourceHandler");
+  handle_watcher_.Start(shared_writer_->writer(), MOJO_HANDLE_SIGNAL_WRITABLE,
+                        MOJO_DEADLINE_INDEFINITE,
+                        base::Bind(&MojoAsyncResourceHandler::OnWritable,
+                                   base::Unretained(this)));
+}
+
+bool MojoAsyncResourceHandler::CheckForSufficientResource() {
+  if (has_checked_for_sufficient_resources_)
+    return true;
+  has_checked_for_sufficient_resources_ = true;
+
+  if (rdh_->HasSufficientResourcesForRequest(request()))
+    return true;
+
+  controller()->CancelWithError(net::ERR_INSUFFICIENT_RESOURCES);
+  return false;
+}
+
+void MojoAsyncResourceHandler::OnWritable(MojoResult unused) {
+  ResumeIfDeferred();
+}
+
+}  // namespace content