PROTOTYPE: PlzNavigate: use a Mojo data pipe to stream navigation data to the renderer.

content/browser/loader/mojo_stream_writer.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_stream_writer.h"
+
+#include "base/bind.h"
+#include "base/guid.h"
+#include "base/numerics/safe_conversions.h"
+#include "content/browser/streams/stream.h"
+#include "content/browser/streams/stream_registry.h"
+#include "content/public/browser/resource_controller.h"
+#include "net/base/io_buffer.h"
+#include "url/gurl.h"
+#include "url/url_constants.h"
+
+namespace content {
+
+MojoStreamWriter::MojoStreamWriter()
+    : controller_(nullptr), immediate_mode_(false), buffer_(nullptr),
+      buffer_size_(-1) {
+}
+
+MojoStreamWriter::~MojoStreamWriter() {
+  if (data_producer_handle_.is_valid())
+    Finalize();
+}
+
+void MojoStreamWriter::InitializeStream(
+      mojo::ScopedDataPipeConsumerHandle* data_consumer_handle) {
+  LOG(ERROR) << "MojoStreamWriter@" << this << " InitializeStream";
+  DCHECK(!data_producer_handle_.is_valid());
+
+  // TODO(carlosk): why not have a default constructor for this struct???
+  // The user having to default initialize it is bad.

[Ken Rockot(use gerrit already), 2016/02/11 16:40:39]

The API at this level (mojo/public/c/system) is specifically intended to be
C-compatible, hence no constructor.

The public/cpp mojo::DataPipe however could have a nicer constructor.

[carlosk, 2016/02/16 15:34:07]

Or the one in mojo/public/cpp/system/data_pipe.h could have a nice C++11 helper
function to build the default options struct.

+  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 = kReadBufSize;
+
+  mojo::DataPipe data_pipe(options);
+  data_producer_handle_ = std::move(data_pipe.producer_handle);
+  *data_consumer_handle = std::move(data_pipe.consumer_handle);
+
+  // TODO(carlosk): hack to comply with the "deferring" interface.
+  buffer_size_ = kReadBufSize;
+  // TODO(carlosk): Should make sure MOJO_WRITE_DATA_FLAG_ALL_OR_NONE is in fact
+  // needed here and in the Begin* calls below.

[Ken Rockot(use gerrit already), 2016/02/11 16:40:39]

The flag has no effect on Begin* (i.e. two-phase) calls. Two-phase reads and
writes give you whatever they can, i.e. the minimum between what you asked for
and what capacity is left before reads/writes would have to wrap around the ring
boundary.

[carlosk, 2016/02/16 15:34:07]

This is not what the method documentation for MojoBeginWriteData says:
https://code.google.com/p/chromium/codesearch#chromium/src/mojo/public/c/syst...
. Is this accurate?

Is that comment not up to date? As I'm basing this code off of StreamWriter it
enforces that the returned buffer size must be greater or equal to the requested
size. I'm just trying to mimic that behavior.

+  CHECK_EQ(mojo::BeginWriteDataRaw(
+    data_producer_handle_.get(), &buffer_, &buffer_size_,
+    MOJO_WRITE_DATA_FLAG_ALL_OR_NONE), MOJO_RESULT_OK);

[Ken Rockot(use gerrit already), 2016/02/11 16:40:39]

I haven't dug into the details of how MojoStreamWriter is used, but it looks
like you're only calling EndWriteData when the consumer signals that it's read
data. This should mean that no data can ever be read or written. BeginWriteData
essentially just locks a portion of the pipe buffer. EndWriteData makes whatever
you wrote available to the consumer. Did I just miss something?

[carlosk, 2016/02/16 15:34:07]

These overridden observer method names are really confusing. OnWillRead is
called when data from the network will be read, so the buffer to write it to
must be ready. OnReadCompleted after the data was read from the network and it's
supposedly now written to the buffer.

So I think I am indeed using these correctly: BeginWriteDataRaw to allocate the
buffer and EndWriteDataRaw to signal that data has been written to the buffer
and can now be sent to through the pipe.

+}
+
+void MojoStreamWriter::OnWillRead(scoped_refptr<net::IOBuffer>* buf,
+                              int* buf_size,
+                              int min_size) {
+  LOG(ERROR) << "MojoStreamWriter@" << this << " OnWillRead";
+
+  DCHECK(buf);
+  DCHECK(buf_size);
+  DCHECK_LE(min_size, base::checked_cast<int>(kReadBufSize));
+  *buf = new net::WrappedIOBuffer(static_cast<char*>(buffer_));
+  *buf_size = base::checked_cast<int>(buffer_size_);
+  buffer_ = nullptr;
+  buffer_size_ = -1;
+
+  // void* buffer = nullptr;
+  // uint32_t size = kReadBufSize;
+  // // TODO(carlosk): check if we need and how to handle error conditions. For
+  // // instance, we might need to move the bool* defer argument from
+  // // OnReadCompleted to this method instead for the case when
+  // // MOJO_RESULT_SHOULD_WAIT is returned indicating Mojo can't currently
+  // // allocate a buffer (somewhat similar to stream_->can_add_data()).
+
+  // CHECK_EQ(mojo::BeginWriteDataRaw(
+  //   data_producer_handle_.get(), &buffer, &size,
+  //   MOJO_WRITE_DATA_FLAG_ALL_OR_NONE), MOJO_RESULT_OK);
+
+  // // TODO(carlosk): Maybe the buffer pointer should still be stored locally to
+  // // be enable the use of the CHECK in OnReadCompleted that would make sure
+  // // this method was called before that one.
+  // *buf = new net::WrappedIOBuffer(static_cast<char*>(buffer));
+  // *buf_size = base::checked_cast<int>(size);
+}
+
+void MojoStreamWriter::OnReadCompleted(int bytes_read, bool* defer) {
+  LOG(ERROR) << "MojoStreamWriter@" << this << " OnReadCompleted";
+  if (!bytes_read)
+    return;
+
+  // We have more data to read.
+  // DCHECK(read_buffer_.get());
+
+  // TODO(carlosk): we might need immediate_mode support for Mojo Data Pipes.
+  CHECK(mojo::EndWriteDataRaw(
+      data_producer_handle_.get(), bytes_read) == MOJO_RESULT_OK);
+
+  // TODO(carlosk): hack to comply with the "deferring" interface.
+  buffer_size_ = kReadBufSize;
+  MojoResult result = mojo::BeginWriteDataRaw(
+    data_producer_handle_.get(), &buffer_, &buffer_size_,
+    MOJO_WRITE_DATA_FLAG_ALL_OR_NONE);
+
+  if (result == MOJO_RESULT_SHOULD_WAIT) {
+    handle_watcher_.Start(
+        data_producer_handle_.get(), MOJO_HANDLE_SIGNAL_WRITABLE,
+        MOJO_DEADLINE_INDEFINITE,
+        base::Bind(&MojoStreamWriter::OnSpaceAvailable, base::Unretained(this)));
+    *defer = true;
+    LOG(ERROR) << "MojoStreamWriter@" << this << " OnReadCompleted - deferring!";
+  } else {
+    CHECK_EQ(result, MOJO_RESULT_OK);
+  }
+}
+
+void MojoStreamWriter::Finalize() {
+  DCHECK(data_producer_handle_.is_valid());
+  if (buffer_ != nullptr) {
+    CHECK_EQ(mojo::EndWriteDataRaw(data_producer_handle_.get(), 0), MOJO_RESULT_OK);
+    buffer_size_ = -1;
+  }
+  data_producer_handle_.reset();
+}
+
+void MojoStreamWriter::OnSpaceAvailable(MojoResult result) {
+  LOG(ERROR) << "MojoStreamWriter@" << this << " OnSpaceAvailable - resuming!";
+  CHECK(!data_producer_handle_.is_valid() || result == MOJO_RESULT_OK);
+  controller_->Resume();
+}
+
+// void MojoStreamWriter::OnClose() {
+//   controller_->Cancel();
+// }
+
+}  // namespace content