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

content/child/resource_dispatcher.cc

Index: content/child/resource_dispatcher.cc
diff --git a/content/child/resource_dispatcher.cc b/content/child/resource_dispatcher.cc
index cbda6c4731052dbe4a471831bd57d203bd4c5e3f..dd15ab46ac69ee5942ff0026e8f43cfb99e77e1c 100644
--- a/content/child/resource_dispatcher.cc
+++ b/content/child/resource_dispatcher.cc
@@ -21,6 +21,7 @@
 #include "base/rand_util.h"
 #include "base/strings/string_util.h"
 #include "build/build_config.h"
+#include "content/child/mojo_pipe_received_data.h"
 #include "content/child/request_extra_data.h"
 #include "content/child/request_info.h"
 #include "content/child/resource_scheduling_filter.h"
@@ -191,6 +192,7 @@ void ResourceDispatcher::OnSetDataBuffer(int request_id,
   PendingRequestInfo* request_info = GetPendingRequestInfo(request_id);
   if (!request_info)
     return;
+  DCHECK(!request_info->mojo_data_pipe_handle.is_valid());
 
   bool shm_valid = base::SharedMemory::IsHandleValid(shm_handle);
   CHECK((shm_valid && shm_size > 0) || (!shm_valid && !shm_size));
@@ -220,6 +222,99 @@ void ResourceDispatcher::OnSetDataBuffer(int request_id,
   request_info->buffer_size = shm_size;
 }
 
+void ResourceDispatcher::WaitForMojoData(int request_id,
+                                         PendingRequestInfo* request_info) {
+  DCHECK(!request_info->mojo_watcher.IsWatching());
+  DCHECK(request_info->mojo_data_pipe_handle.is_valid());
+  LOG(ERROR) << ">>> ResourceDispatcher::WaitForMojoData: request_id = "
+             << request_id;
+  MojoResult rv = request_info->mojo_watcher.Start(
+      request_info->mojo_data_pipe_handle.get(), MOJO_HANDLE_SIGNAL_READABLE,
+      base::Bind(&ResourceDispatcher::ReceivedMojoData,
+                 weak_factory_.GetWeakPtr(), request_id));
+  CHECK_EQ(rv, MOJO_RESULT_OK);
+}
+
+void ResourceDispatcher::ReceivedMojoData(int request_id, MojoResult) {
+  // Note: it seems like we can ignore the MojoResult parameter here and simply
+  // rely on the one returned by mojo::BeginReadDataRaw.
+  LOG(ERROR) << ">>> ResourceDispatcher::ReceivedMojoData: request_id = "
+             << request_id;
+  TRACE_EVENT0("loader", "ResourceDispatcher::OnReceivedData");
+  PendingRequestInfo* request_info = GetPendingRequestInfo(request_id);
+  if (!request_info)
+    return;
+  CHECK(request_info->mojo_data_pipe_handle.is_valid());
+
+  // TODO(carlosk): see if we need to create an equivalent UMA metric as
+  // ResourceDispatcher.OnReceivedDataTime for the handling of Mojo data;
+
+  // Read all available data bundles received.
+  MojoResult rv = MOJO_RESULT_OK;
+  while (rv == MOJO_RESULT_OK) {
+    const void* buffer = nullptr;
+    uint32_t num_bytes = 0;
+    rv = mojo::BeginReadDataRaw(request_info->mojo_data_pipe_handle.get(),
+                                &buffer, &num_bytes, MOJO_READ_DATA_FLAG_NONE);
+    if (rv == MOJO_RESULT_OK) {
+      // There's data to read from the pipe.
+      AcceptMojoMessage(request_info, buffer, num_bytes);
+    }
+  }
+
+  if (rv == MOJO_RESULT_SHOULD_WAIT) {
+    // No more data to be read from the pipe for now.
+    // Apparently there's no need to re-register the watcher...
+    // WaitForMojoData(request_id, request_info);
+  } else if (rv == MOJO_RESULT_FAILED_PRECONDITION) {
+    // Regular EOF from the pipe; maybe errors too?
+    // TODO(carlosk): given the extra data sent along with
+    // ResourceMsg_RequestComplete we would need an IPC for that. But are
+    // there any ordering guarantees between piped data and IPCs? If not we'd
+    // have to create a barrier between EOF and that IPC or receive this data
+    // through the pipe.
+    request_info->mojo_watcher.Cancel();
+    request_info->mojo_data_pipe_handle.reset();
+    // Note: this is dummy data; it should all come from the browser.
+    ResourceMsg_RequestCompleteData request_complete_data;
+    request_complete_data.error_code = net::OK;
+    request_complete_data.was_ignored_by_handler = false;
+    request_complete_data.exists_in_cache = false;
+    request_complete_data.completion_time = base::TimeTicks::Now();
+    request_complete_data.encoded_data_length = request_info->mojo_total_bytes;
+    LOG(ERROR) << ">>> Will call ResourceDispatcher::OnRequestComplete for "
+                  "Mojo: request_id = "
+               << request_id;
+    OnRequestComplete(request_id, request_complete_data);
+  } else {
+    // Something else that I still have to figure out...
+    CHECK(false) << "Unhandled MojoResult: " << rv;
+  }
+}
+
+void ResourceDispatcher::AcceptMojoMessage(PendingRequestInfo* request_info,
+                                           const void* buffer,
+                                           uint32_t num_bytes) {
+  LOG(ERROR) << ">>> ResourceDispatcher::AcceptMojoMessage: num_bytes = "
+             << num_bytes;
+  DCHECK(num_bytes > 0);
+  // Check whether this response data is compliant with our cross-site
+  // document blocking policy. We only do this for the first chunk of data.
+  if (request_info->site_isolation_metadata.get()) {
+    SiteIsolationStatsGatherer::OnReceivedFirstChunk(
+        request_info->site_isolation_metadata, static_cast<const char*>(buffer),
+        num_bytes);
+    request_info->site_isolation_metadata.reset();
+  }
+
+  request_info->mojo_total_bytes += num_bytes;
+  scoped_ptr<RequestPeer::ReceivedData> mojo_data =
+      MojoPipeReceivedData::Create(request_info->mojo_data_pipe_handle.get(),
+                                   buffer, num_bytes);
+
+  request_info->peer->OnReceivedData(std::move(mojo_data));
+}
+
 void ResourceDispatcher::OnReceivedInlinedDataChunk(
     int request_id,
     const std::vector<char>& data,
@@ -256,6 +351,7 @@ void ResourceDispatcher::OnReceivedData(int request_id,
   PendingRequestInfo* request_info = GetPendingRequestInfo(request_id);
   bool send_ack = true;
   if (request_info && data_length > 0) {
+    DCHECK(!request_info->mojo_data_pipe_handle.is_valid());
     CHECK(base::SharedMemory::IsHandleValid(request_info->buffer->handle()));
     CHECK_GE(request_info->buffer_size, data_offset + data_length);
 
@@ -386,9 +482,14 @@ void ResourceDispatcher::OnRequestComplete(
 }
 
 bool ResourceDispatcher::RemovePendingRequest(int request_id) {
+  if (request_id < -1)
+    LOG(ERROR) << "ResourceDispatcher::RemovePendingRequest request_id: "
+               << request_id;
   PendingRequestMap::iterator it = pending_requests_.find(request_id);
   if (it == pending_requests_.end())
     return false;
+  if (request_id < -1)
+    LOG(ERROR) << "ResourceDispatcher::RemovePendingRequest request deleted";
 
   PendingRequestInfo* request_info = it->second.get();
 
@@ -491,8 +592,8 @@ ResourceDispatcher::PendingRequestInfo::PendingRequestInfo(
       frame_origin(frame_origin),
       response_url(request_url),
       download_to_file(download_to_file),
-      request_start(base::TimeTicks::Now()) {
-}
+      request_start(base::TimeTicks::Now()),
+      mojo_total_bytes(0) {}
 
 ResourceDispatcher::PendingRequestInfo::~PendingRequestInfo() {
 }
@@ -575,15 +676,26 @@ void ResourceDispatcher::StartSync(const RequestInfo& request_info,
   response->socket_address = result.socket_address;
 }
 
-int ResourceDispatcher::StartAsync(const RequestInfo& request_info,
-                                   ResourceRequestBody* request_body,
-                                   scoped_ptr<RequestPeer> peer) {
+int ResourceDispatcher::StartAsync(
+    const RequestInfo& request_info,
+    ResourceRequestBody* request_body,
+    scoped_ptr<RequestPeer> peer,
+    mojo::ScopedDataPipeConsumerHandle mojo_data_pipe_handle,
+    int browser_request_id) {
+  DCHECK_EQ(browser_request_id != -1, mojo_data_pipe_handle.is_valid());
   GURL frame_origin;
   scoped_ptr<ResourceHostMsg_Request> request =
       CreateRequest(request_info, request_body, &frame_origin);
 
-  // Compute a unique request_id for this renderer process.
-  int request_id = MakeRequestID();
+  int request_id;
+  if (browser_request_id == -1) {
+    // Compute a unique request_id for this renderer process.
+    request_id = MakeRequestID();
+  } else {
+    // Uses the request_id already created by the browser.
+    CHECK(!GetPendingRequestInfo(browser_request_id));
+    request_id = browser_request_id;
+  }
   pending_requests_[request_id] = make_scoped_ptr(new PendingRequestInfo(
       std::move(peer), request->resource_type, request->origin_pid,
       frame_origin, request->url, request_info.download_to_file));
@@ -595,8 +707,39 @@ int ResourceDispatcher::StartAsync(const RequestInfo& request_info,
         make_scoped_ptr(request_info.loading_web_task_runner->clone()));
   }
 
-  message_sender_->Send(new ResourceHostMsg_RequestResource(
-      request_info.routing_id, request_id, *request));
+  // PlzNavigate: if there is a Mojo data pipe handle set we know this is a
+  // PlzNavigate loading of a navigated document.
+  if (mojo_data_pipe_handle.is_valid()) {
+    LOG(ERROR) << "*** ResourceDispatcher::StartAsync request_id: "
+               << request_id;
+    LOG(ERROR) << "*** ResourceDispatcher::StartAsync url: "
+               << request_info.url;
+    // SetAndListenToMojoDataPipe(request_id, std::move(mojo_data_pipe_handle));
+    PendingRequestInfo* pending_request_info =
+        GetPendingRequestInfo(request_id);
+    pending_request_info->mojo_data_pipe_handle =
+        std::move(mojo_data_pipe_handle);
+
+    // Short circuiting call to OnReceivedResponse to immediately start the
+    // request.
+    // Note: the ResourceResponseHead() can be empty because it is completely
+    // and totally reset in:
+    // - ResourceDispatcher::OnReceivedResponse
+    // - WebURLLoaderImpl::RequestPeerImpl::OnReceivedResponse
+    // -->> WebURLLoaderImpl::Context::OnReceivedResponse (HERE!)
+    OnReceivedResponse(request_id, ResourceResponseHead());
+
+    // Now that the correct state is set we can start listening for data from
+    // the Mojo pipe, if one is set for this pending request.
+    WaitForMojoData(request_id, pending_request_info);
+
+    // TODO(carlosk): sending ResourceHostMsg_RequestResource to the browser
+    // should not be needed but we still have to figure out if any of its side
+    // effects are necessary for the request to actually succeed.
+  } else {
+    message_sender_->Send(new ResourceHostMsg_RequestResource(
+        request_info.routing_id, request_id, *request));
+  }
 
   return request_id;
 }