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

content/child/resource_dispatcher.cc

 // Copyright (c) 2012 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.
 
 // See http://dev.chromium.org/developers/design-documents/multi-process-resource-loading
 
 #include "content/child/resource_dispatcher.h"
 
 #include <utility>
 
 #include "base/bind.h"
 #include "base/compiler_specific.h"
 #include "base/debug/alias.h"
 #include "base/debug/dump_without_crashing.h"
 #include "base/debug/stack_trace.h"
 #include "base/feature_list.h"
 #include "base/files/file_path.h"
 #include "base/memory/shared_memory.h"
 #include "base/message_loop/message_loop.h"
 #include "base/metrics/histogram.h"
 #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"
 #include "content/child/shared_memory_received_data_factory.h"
 #include "content/child/site_isolation_stats_gatherer.h"
 #include "content/child/sync_load_response.h"
 #include "content/common/inter_process_time_ticks_converter.h"
 #include "content/common/navigation_params.h"
 #include "content/common/resource_messages.h"
 #include "content/public/child/fixed_received_data.h"
@@ skipping 150 matching lines @@
 }
 
 void ResourceDispatcher::OnSetDataBuffer(int request_id,
                                          base::SharedMemoryHandle shm_handle,
                                          int shm_size,
                                          base::ProcessId renderer_pid) {
   TRACE_EVENT0("loader", "ResourceDispatcher::OnSetDataBuffer");
   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));
 
   request_info->buffer.reset(
       new base::SharedMemory(shm_handle, true));  // read only
   request_info->received_data_factory =
       make_scoped_refptr(new SharedMemoryReceivedDataFactory(
           message_sender_, request_id, request_info->buffer));
 
   bool ok = request_info->buffer->Map(shm_size);
   if (!ok) {
     // Added to help debug crbug/160401.
     base::ProcessId renderer_pid_copy = renderer_pid;
     base::debug::Alias(&renderer_pid_copy);
 
     base::SharedMemoryHandle shm_handle_copy = shm_handle;
     base::debug::Alias(&shm_handle_copy);
 
     CrashOnMapFailure();
     return;
   }
 
   // TODO(erikchen): Temporary debugging. http://crbug.com/527588.
   CHECK_GE(shm_size, 0);
   CHECK_LE(shm_size, 512 * 1024);
   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,
     int encoded_data_length) {
   TRACE_EVENT0("loader", "ResourceDispatcher::OnReceivedInlinedDataChunk");
   DCHECK(!data.empty());
   DCHECK(base::FeatureList::IsEnabled(features::kOptimizeIPCForSmallResource));
 
   PendingRequestInfo* request_info = GetPendingRequestInfo(request_id);
   if (!request_info || data.empty())
@@ skipping 16 matching lines @@
 
 void ResourceDispatcher::OnReceivedData(int request_id,
                                         int data_offset,
                                         int data_length,
                                         int encoded_data_length) {
   TRACE_EVENT0("loader", "ResourceDispatcher::OnReceivedData");
   DCHECK_GT(data_length, 0);
   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);
 
     base::TimeTicks time_start = base::TimeTicks::Now();
 
     const char* data_start = static_cast<char*>(request_info->buffer->memory());
     CHECK(data_start);
     CHECK(data_start + data_offset);
     const char* data_ptr = data_start + data_offset;
 
@@ skipping 110 matching lines @@
   // (crbug.com/547047)
   peer->OnCompletedRequest(request_complete_data.error_code,
                            request_complete_data.was_ignored_by_handler,
                            request_complete_data.exists_in_cache,
                            request_complete_data.security_info,
                            renderer_completion_time,
                            request_complete_data.encoded_data_length);
 }
 
 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();
 
   bool release_downloaded_file = request_info->download_to_file;
 
   ReleaseResourcesInMessageQueue(&request_info->deferred_message_queue);
 
   // Always delete the pending_request asyncly so that cancelling the request
   // doesn't delete the request context info while its response is still being
   // handled.
@@ skipping 82 matching lines @@
     const GURL& frame_origin,
     const GURL& request_url,
     bool download_to_file)
     : peer(std::move(peer)),
       resource_type(resource_type),
       origin_pid(origin_pid),
       url(request_url),
       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() {
 }
 
 void ResourceDispatcher::DispatchMessage(const IPC::Message& message) {
   IPC_BEGIN_MESSAGE_MAP(ResourceDispatcher, message)
     IPC_MESSAGE_HANDLER(ResourceMsg_UploadProgress, OnUploadProgress)
     IPC_MESSAGE_HANDLER(ResourceMsg_ReceivedResponse, OnReceivedResponse)
     IPC_MESSAGE_HANDLER(ResourceMsg_ReceivedCachedMetadata,
                         OnReceivedCachedMetadata)
@@ skipping 62 matching lines @@
   response->request_time = result.request_time;
   response->response_time = result.response_time;
   response->encoded_data_length = result.encoded_data_length;
   response->load_timing = result.load_timing;
   response->devtools_info = result.devtools_info;
   response->data.swap(result.data);
   response->download_file_path = result.download_file_path;
   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));
 
   if (resource_scheduling_filter_.get() &&
       request_info.loading_web_task_runner) {
     resource_scheduling_filter_->SetRequestIdTaskRunner(
         request_id,
         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;
 }
 
 void ResourceDispatcher::ToResourceResponseInfo(
     const PendingRequestInfo& request_info,
     const ResourceResponseHead& browser_info,
     ResourceResponseInfo* renderer_info) const {
   *renderer_info = browser_info;
   if (request_info.request_start.is_null() ||
@@ skipping 198 matching lines @@
     *frame_origin = extra_data->frame_origin();
   return request;
 }
 
 void ResourceDispatcher::SetResourceSchedulingFilter(
     scoped_refptr<ResourceSchedulingFilter> resource_scheduling_filter) {
   resource_scheduling_filter_ = resource_scheduling_filter;
 }
 
 }  // namespace content