Redirect HTML resource bytes directly to parser thread (Chrome side)

Blink side: https://codereview.chromium.org/100563004/

Note: This can't land until the Blink-side has landed.

R=jamesr@chromium.org,jam@chromium.org
BUG=277886

Review URL: https://codereview.chromium.org/109283006

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@275655 0039d316-1c4b-4281-b951-d872f2087c98

content/child/resource_dispatcher.cc

@@ -19,6 +19,7 @@
 #include "content/child/request_info.h"
 #include "content/child/site_isolation_policy.h"
 #include "content/child/sync_load_response.h"
+#include "content/child/threaded_data_provider.h"
 #include "content/common/inter_process_time_ticks_converter.h"
 #include "content/common/resource_messages.h"
 #include "content/public/child/request_peer.h"
@@ -80,6 +81,8 @@ class IPCResourceLoaderBridge : public ResourceLoaderBridge {
   virtual void SetDefersLoading(bool value) OVERRIDE;
   virtual void DidChangePriority(net::RequestPriority new_priority,
                                  int intra_priority_value) OVERRIDE;
+  virtual bool AttachThreadedDataReceiver(
+      blink::WebThreadedDataReceiver* threaded_data_receiver) OVERRIDE;
   virtual void SyncLoad(SyncLoadResponse* response) OVERRIDE;
 
  private:
@@ -220,6 +223,17 @@ void IPCResourceLoaderBridge::DidChangePriority(
       request_id_, new_priority, intra_priority_value);
 }
 
+bool IPCResourceLoaderBridge::AttachThreadedDataReceiver(
+    blink::WebThreadedDataReceiver* threaded_data_receiver) {
+  if (request_id_ < 0) {
+    NOTREACHED() << "Trying to attach threaded receiver on unstarted request";
+    return false;
+  }
+
+  return dispatcher_->AttachThreadedDataReceiver(request_id_,
+                                                 threaded_data_receiver);
+}
+
 void IPCResourceLoaderBridge::SyncLoad(SyncLoadResponse* response) {
   if (request_id_ != -1) {
     NOTREACHED() << "Starting a request twice";
@@ -405,6 +419,7 @@ void ResourceDispatcher::OnReceivedData(int request_id,
   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) {
     CHECK(base::SharedMemory::IsHandleValid(request_info->buffer->handle()));
     CHECK_GE(request_info->buffer_size, data_offset + data_length);
@@ -416,9 +431,10 @@ void ResourceDispatcher::OnReceivedData(int request_id,
 
     base::TimeTicks time_start = base::TimeTicks::Now();
 
-    const char* data_ptr = static_cast<char*>(request_info->buffer->memory());
-    CHECK(data_ptr);
-    CHECK(data_ptr + data_offset);
+    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;
 
     // Check whether this response data is compliant with our cross-site
     // document blocking policy. We only do this for the first packet.
@@ -426,22 +442,31 @@ void ResourceDispatcher::OnReceivedData(int request_id,
     if (request_info->site_isolation_metadata.get()) {
       request_info->blocked_response =
           SiteIsolationPolicy::ShouldBlockResponse(
-              request_info->site_isolation_metadata, data_ptr + data_offset,
-              data_length, &alternative_data);
+              request_info->site_isolation_metadata, data_ptr, data_length,
+              &alternative_data);
       request_info->site_isolation_metadata.reset();
-    }
 
-    // When the response is not blocked.
-    if (!request_info->blocked_response) {
-      request_info->peer->OnReceivedData(
-          data_ptr + data_offset, data_length, encoded_data_length);
-    } else if (alternative_data.size() > 0) {
-      // When the response is blocked, and when we have any alternative data to
+      // When the response is blocked we may have any alternative data to
       // send to the renderer. When |alternative_data| is zero-sized, we do not
       // call peer's callback.
-      request_info->peer->OnReceivedData(alternative_data.data(),
-                                         alternative_data.size(),
-                                         alternative_data.size());
+      if (request_info->blocked_response && !alternative_data.empty()) {
+        data_ptr = alternative_data.data();
+        data_length = alternative_data.size();
+        encoded_data_length = alternative_data.size();
+      }
+    }
+
+    if (!request_info->blocked_response || !alternative_data.empty()) {
+      if (request_info->threaded_data_provider) {
+        request_info->threaded_data_provider->OnReceivedDataOnForegroundThread(
+            data_ptr, data_length, encoded_data_length);
+        // A threaded data provider will take care of its own ACKing, as the
+        // data may be processed later on another thread.
+        send_ack = false;
+      } else {
+        request_info->peer->OnReceivedData(
+            data_ptr, data_length, encoded_data_length);
+      }
     }
 
     UMA_HISTOGRAM_TIMES("ResourceDispatcher.OnReceivedDataTime",
@@ -449,7 +474,8 @@ void ResourceDispatcher::OnReceivedData(int request_id,
   }
 
   // Acknowledge the reception of this data.
-  message_sender_->Send(new ResourceHostMsg_DataReceived_ACK(request_id));
+  if (send_ack)
+    message_sender_->Send(new ResourceHostMsg_DataReceived_ACK(request_id));
 }
 
 void ResourceDispatcher::OnDownloadedData(int request_id,
@@ -623,8 +649,21 @@ void ResourceDispatcher::DidChangePriority(int request_id,
       request_id, new_priority, intra_priority_value));
 }
 
+bool ResourceDispatcher::AttachThreadedDataReceiver(
+    int request_id, blink::WebThreadedDataReceiver* threaded_data_receiver) {
+  PendingRequestInfo* request_info = GetPendingRequestInfo(request_id);
+  DCHECK(request_info);
+  DCHECK(!request_info->threaded_data_provider);
+  request_info->threaded_data_provider = new ThreadedDataProvider(
+      request_id, threaded_data_receiver, request_info->buffer,
+      request_info->buffer_size);
+
+  return true;
+}
+
 ResourceDispatcher::PendingRequestInfo::PendingRequestInfo()
     : peer(NULL),
+      threaded_data_provider(NULL),
       resource_type(ResourceType::SUB_RESOURCE),
       is_deferred(false),
       download_to_file(false),
@@ -640,6 +679,7 @@ ResourceDispatcher::PendingRequestInfo::PendingRequestInfo(
     const GURL& request_url,
     bool download_to_file)
     : peer(peer),
+      threaded_data_provider(NULL),
       resource_type(resource_type),
       origin_pid(origin_pid),
       is_deferred(false),
@@ -650,7 +690,10 @@ ResourceDispatcher::PendingRequestInfo::PendingRequestInfo(
       request_start(base::TimeTicks::Now()),
       blocked_response(false) {}
 
-ResourceDispatcher::PendingRequestInfo::~PendingRequestInfo() {}
+ResourceDispatcher::PendingRequestInfo::~PendingRequestInfo() {
+  if (threaded_data_provider)
+    threaded_data_provider->Stop();
+}
 
 void ResourceDispatcher::DispatchMessage(const IPC::Message& message) {
   IPC_BEGIN_MESSAGE_MAP(ResourceDispatcher, message)

content/child/resource_dispatcher.h

@@ -23,6 +23,10 @@
 
 struct ResourceMsg_RequestCompleteData;
 
+namespace blink {
+class WebThreadedDataReceiver;
+}
+
 namespace webkit_glue {
 class ResourceLoaderBridge;
 }
@@ -30,6 +34,7 @@ class ResourceLoaderBridge;
 namespace content {
 class RequestPeer;
 class ResourceDispatcherDelegate;
+class ThreadedDataProvider;
 struct ResourceResponseInfo;
 struct RequestInfo;
 struct ResourceResponseHead;
@@ -76,6 +81,11 @@ class CONTENT_EXPORT ResourceDispatcher : public IPC::Listener {
                          net::RequestPriority new_priority,
                          int intra_priority_value);
 
+  // The provided data receiver will receive incoming resource data rather
+  // than the resource bridge.
+  bool AttachThreadedDataReceiver(
+      int request_id, blink::WebThreadedDataReceiver* threaded_data_receiver);
+
   IPC::Sender* message_sender() const { return message_sender_; }
 
   // This does not take ownership of the delegate. It is expected that the
@@ -106,6 +116,7 @@ class CONTENT_EXPORT ResourceDispatcher : public IPC::Listener {
     ~PendingRequestInfo();
 
     RequestPeer* peer;
+    ThreadedDataProvider* threaded_data_provider;
     ResourceType::Type resource_type;
     // The PID of the original process which issued this request. This gets
     // non-zero only for a request proxied by another renderer, particularly

content/child/threaded_data_provider.h

+// Copyright 2014 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.
+
+#ifndef CONTENT_CHILD_THREADEDDATAPROVIDER_IMPL_H_
+#define CONTENT_CHILD_THREADEDDATAPROVIDER_IMPL_H_
+
+#include "base/compiler_specific.h"
+#include "base/memory/linked_ptr.h"
+#include "base/memory/ref_counted.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/memory/shared_memory.h"
+#include "base/memory/weak_ptr.h"
+#include "ipc/ipc_channel.h"
+#include "ipc/message_filter.h"
+
+namespace blink {
+class WebThreadedDataReceiver;
+}
+
+namespace IPC {
+class SyncChannel;
+}
+
+namespace webkit_glue {
+class WebThreadImpl;
+}
+
+namespace content {
+class ResourceDispatcher;
+class WebThreadImpl;
+
+class ThreadedDataProvider {
+ public:
+  ThreadedDataProvider(
+      int request_id,
+      blink::WebThreadedDataReceiver* threaded_data_receiver,
+      linked_ptr<base::SharedMemory> shm_buffer,
+      int shm_size);
+  virtual ~ThreadedDataProvider();
+
+  void Stop();
+  void OnReceivedDataOnBackgroundThread(int data_offset,
+                                        int data_length,
+                                        int encoded_data_length);
+
+  void OnReceivedDataOnForegroundThread(const char* data,
+                                        int data_length,
+                                        int encoded_data_length);
+
+  void OnResourceMessageFilterAddedMainThread();
+
+ private:
+  void StopOnBackgroundThread();
+  void OnResourceMessageFilterAddedBackgroundThread();
+  void ForwardAndACKData(const char* data, int data_length);
+
+  scoped_refptr<IPC::MessageFilter> filter_;
+  int request_id_;
+  linked_ptr<base::SharedMemory> shm_buffer_;
+  int shm_size_;
+  scoped_ptr<base::WeakPtrFactory<ThreadedDataProvider> >
+      background_thread_weak_factory_;
+  base::WeakPtrFactory<ThreadedDataProvider>
+      main_thread_weak_factory_;
+  WebThreadImpl& background_thread_;
+  IPC::SyncChannel* ipc_channel_;
+  blink::WebThreadedDataReceiver* threaded_data_receiver_;
+  bool resource_filter_active_;
+  base::MessageLoop* main_thread_message_loop_;
+
+  struct QueuedSharedMemoryData {
+    const char* data;
+    int length;
+  };
+  std::vector<QueuedSharedMemoryData> queued_data_;
+
+  DISALLOW_COPY_AND_ASSIGN(ThreadedDataProvider);
+};
+
+}  // namespace content
+
+#endif  // CONTENT_CHILD_THREADEDDATAPROVIDER_IMPL_H_

content/child/web_url_loader_impl.cc

@@ -218,6 +218,8 @@ class WebURLLoaderImpl::Context : public base::RefCounted<Context>,
   void SetDefersLoading(bool value);
   void DidChangePriority(WebURLRequest::Priority new_priority,
                          int intra_priority_value);
+  bool AttachThreadedDataReceiver(
+      blink::WebThreadedDataReceiver* threaded_data_receiver);
   void Start(const WebURLRequest& request,
              SyncLoadResponse* sync_load_response);
 
@@ -295,6 +297,14 @@ void WebURLLoaderImpl::Context::DidChangePriority(
         ConvertWebKitPriorityToNetPriority(new_priority), intra_priority_value);
 }
 
+bool WebURLLoaderImpl::Context::AttachThreadedDataReceiver(
+    blink::WebThreadedDataReceiver* threaded_data_receiver) {
+  if (bridge_)
+    return bridge_->AttachThreadedDataReceiver(threaded_data_receiver);
+
+  return false;
+}
+
 void WebURLLoaderImpl::Context::Start(const WebURLRequest& request,
                                       SyncLoadResponse* sync_load_response) {
   DCHECK(!bridge_.get());
@@ -868,4 +878,9 @@ void WebURLLoaderImpl::didChangePriority(WebURLRequest::Priority new_priority,
   context_->DidChangePriority(new_priority, intra_priority_value);
 }
 
+bool WebURLLoaderImpl::attachThreadedDataReceiver(
+    blink::WebThreadedDataReceiver* threaded_data_receiver) {
+  return context_->AttachThreadedDataReceiver(threaded_data_receiver);
+}
+
 }  // namespace content

content/child/web_url_loader_impl.h

@@ -39,6 +39,8 @@ class WebURLLoaderImpl : public blink::WebURLLoader {
   virtual void setDefersLoading(bool value);
   virtual void didChangePriority(blink::WebURLRequest::Priority new_priority,
                                  int intra_priority_value);
+  virtual bool attachThreadedDataReceiver(
+      blink::WebThreadedDataReceiver* threaded_data_receiver);
 
  private:
   class Context;

content/child/webthread_impl.h

@@ -14,7 +14,7 @@
 
 namespace content {
 
-class WebThreadBase : public blink::WebThread {
+class CONTENT_EXPORT WebThreadBase : public blink::WebThread {
  public:
   virtual ~WebThreadBase();
 
@@ -33,10 +33,10 @@ class WebThreadBase : public blink::WebThread {
   TaskObserverMap task_observer_map_;
 };
 
-class WebThreadImpl : public WebThreadBase {
+class CONTENT_EXPORT WebThreadImpl : public WebThreadBase {
  public:
-  CONTENT_EXPORT explicit WebThreadImpl(const char* name);
-  CONTENT_EXPORT virtual ~WebThreadImpl();
+  explicit WebThreadImpl(const char* name);
+  virtual ~WebThreadImpl();
 
   virtual void postTask(Task* task);
   virtual void postDelayedTask(Task* task, long long delay_ms);
@@ -46,8 +46,9 @@ class WebThreadImpl : public WebThreadBase {
 
   base::MessageLoop* message_loop() const { return thread_->message_loop(); }
 
- private:
   virtual bool isCurrentThread() const OVERRIDE;
+
+ private:
   scoped_ptr<base::Thread> thread_;
 };
 
@@ -57,11 +58,11 @@ class WebThreadImplForMessageLoop : public WebThreadBase {
       base::MessageLoopProxy* message_loop);
   CONTENT_EXPORT virtual ~WebThreadImplForMessageLoop();
 
-  virtual void postTask(Task* task);
-  virtual void postDelayedTask(Task* task, long long delay_ms);
+  virtual void postTask(Task* task) OVERRIDE;
+  virtual void postDelayedTask(Task* task, long long delay_ms) OVERRIDE;
 
-  virtual void enterRunLoop();
-  virtual void exitRunLoop();
+  virtual void enterRunLoop() OVERRIDE;
+  virtual void exitRunLoop() OVERRIDE;
 
  private:
   virtual bool isCurrentThread() const OVERRIDE;

content/content_child.gypi

@@ -213,6 +213,8 @@
       'child/webfileutilities_impl.h',
       'child/webmessageportchannel_impl.cc',
       'child/webmessageportchannel_impl.h',
+      'child/threaded_data_provider.cc',
+      'child/threaded_data_provider.h',
       'child/websocket_bridge.cc',
       'child/websocket_bridge.h',
       'child/websocket_dispatcher.cc',

webkit/child/resource_loader_bridge.h

@@ -21,6 +21,10 @@
 #include "net/base/request_priority.h"
 #include "webkit/child/webkit_child_export.h"
 
+namespace blink {
+class WebThreadedDataReceiver;
+}
+
 // TODO(pilgrim) remove this once resource loader is moved to content
 // http://crbug.com/338338
 namespace content {
@@ -61,6 +65,11 @@ class ResourceLoaderBridge {
   virtual void DidChangePriority(net::RequestPriority new_priority,
                                  int intra_priority_value) = 0;
 
+  // Call this method to attach a data receiver which will receive resource data
+  // on its own thread.
+  virtual bool AttachThreadedDataReceiver(
+      blink::WebThreadedDataReceiver* threaded_data_receiver) = 0;
+
   // Call this method to load the resource synchronously (i.e., in one shot).
   // This is an alternative to the Start method.  Be warned that this method
   // will block the calling thread until the resource is fully downloaded or an