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

content/child/webthreadedresourceprovider_impl.cc

+// Copyright 2013 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/child/webthreadedresourceprovider_impl.h"
+
+#include "content/child/child_process.h"
+#include "content/child/child_thread.h"
+#include "content/child/resource_dispatcher.h"
+#include "content/child/thread_safe_sender.h"
+#include "content/common/resource_messages.h"
+#include "ipc/ipc_sync_channel.h"
+#include "third_party/WebKit/public/platform/WebThread.h"
+#include "webkit/child/webthread_impl.h"
+
+namespace content {
+
+namespace {
+
+static webkit_glue::WebThreadImpl* thread_ = NULL;
+
+class ParserResourceMessageFilter : public IPC::ChannelProxy::MessageFilter {

[darin (slow to review), 2014/02/11 07:14:23]

I think I would avoid mentioning the parser in this file. It doesn't seem like
there is anything stopping Blink from making use of this infrastructure for
other things besides the parser.

[oystein (OOO til 10th of July), 2014/02/11 21:46:44]

Done! Yeah this has become less and less parser specific in successive
iterations.

+ public:
+  ParserResourceMessageFilter(
+      const scoped_refptr<base::MessageLoopProxy>& io_message_loop,
+      base::MessageLoop* main_thread_message_loop,
+      webkit_glue::WebThreadImpl& parser_thread,
+      base::WeakPtr<WebThreadedResourceProviderImpl>
+          parser_thread_resource_provider,
+      base::WeakPtr<WebThreadedResourceProviderImpl>
+          main_thread_resource_provider,
+      int request_id);
+
+  // IPC::ChannelProxy::MessageFilter
+  virtual void OnFilterAdded(IPC::Channel* channel) OVERRIDE;
+  virtual bool OnMessageReceived(const IPC::Message& message) OVERRIDE;
+
+ private:
+  virtual ~ParserResourceMessageFilter();
+
+  void OnReceivedData(int request_id, int data_offset, int data_length,
+                      int encoded_data_length);
+
+  const scoped_refptr<base::MessageLoopProxy> io_message_loop_;
+  base::MessageLoop* main_thread_message_loop_;
+  webkit_glue::WebThreadImpl& parser_thread_;
+  // This weakptr can only be dereferenced on the parser thread.
+  base::WeakPtr<WebThreadedResourceProviderImpl>
+      parser_thread_resource_provider_;
+  // This weakptr can only be dereferenced on the main thread.
+  base::WeakPtr<WebThreadedResourceProviderImpl>
+      main_thread_resource_provider_;
+  int request_id_;
+};
+
+ParserResourceMessageFilter::ParserResourceMessageFilter(
+    const scoped_refptr<base::MessageLoopProxy>& io_message_loop,
+    base::MessageLoop* main_thread_message_loop,
+    webkit_glue::WebThreadImpl& parser_thread,
+    base::WeakPtr<WebThreadedResourceProviderImpl>
+        parser_thread_resource_provider,
+    base::WeakPtr<WebThreadedResourceProviderImpl>
+        main_thread_resource_provider,
+    int request_id)
+    : io_message_loop_(io_message_loop),
+      main_thread_message_loop_(main_thread_message_loop),
+      parser_thread_(parser_thread),
+      parser_thread_resource_provider_(parser_thread_resource_provider),
+      main_thread_resource_provider_(main_thread_resource_provider),
+      request_id_(request_id) {
+  DCHECK(main_thread_message_loop != NULL);
+}
+
+void ParserResourceMessageFilter::OnFilterAdded(IPC::Channel* channel) {
+  DCHECK(io_message_loop_->BelongsToCurrentThread());
+
+  main_thread_message_loop_->PostTask(FROM_HERE,
+      base::Bind(
+          &WebThreadedResourceProviderImpl::OnParserResourceMessageFilterAdded,
+          main_thread_resource_provider_));
+}
+
+bool ParserResourceMessageFilter::OnMessageReceived(
+    const IPC::Message& message) {
+  DCHECK(io_message_loop_->BelongsToCurrentThread());
+
+  if (message.type() != ResourceMsg_DataReceived::ID)
+    return false;
+
+  int request_id;
+
+  PickleIterator iter(message);
+  if (!message.ReadInt(&iter, &request_id)) {
+    NOTREACHED() << "malformed resource message";
+    return true;
+  }
+
+  if (request_id == request_id_) {
+    ResourceMsg_DataReceived::Schema::Param arg;
+    if (ResourceMsg_DataReceived::Read(&message, &arg))
+      OnReceivedData(arg.a, arg.b, arg.c, arg.d);
+  }
+
+  // Even if the parser thread wants the message (i.e. the request ID matches),
+  // the main thread currently still expects to get it as well. Each chunk will
+  // still only be parsed once.
+  return false;
+}
+
+ParserResourceMessageFilter::~ParserResourceMessageFilter() {
+  DCHECK(io_message_loop_->BelongsToCurrentThread());
+}
+
+void ParserResourceMessageFilter::OnReceivedData(int request_id,
+                                                 int data_offset,
+                                                 int data_length,
+                                                 int encoded_data_length) {
+  DCHECK(io_message_loop_->BelongsToCurrentThread());
+  parser_thread_.message_loop()->PostTask(FROM_HERE, base::Bind(
+      &WebThreadedResourceProviderImpl::OnReceivedData,
+      parser_thread_resource_provider_,
+      data_offset, data_length, encoded_data_length));
+}
+
+}  // anonymous namespace
+
+webkit_glue::WebThreadImpl& parser_thread() {
+  if (thread_ == NULL)
+    thread_ = new webkit_glue::WebThreadImpl("HTMLParserThread");

[darin (slow to review), 2014/02/11 07:14:23]

This code doesn't seem specific to the parser.

[oystein (OOO til 10th of July), 2014/02/11 21:46:44]

Done.

+
+  return *thread_;
+}
+
+void WebThreadedResourceProviderImpl::Cleanup() {
+  delete thread_;
+  thread_ = NULL;
+}
+
+WebThreadedResourceProviderImpl::WebThreadedResourceProviderImpl(
+    int request_id, base::WeakPtr<ResourceDispatcher> resource_dispatcher,
+    linked_ptr<base::SharedMemory> shm_buffer, int shm_size)
+    : backgroundClient_(NULL),
+      foregroundClient_(NULL),
+      request_id_(request_id),
+      resource_dispatcher_(resource_dispatcher),
+      shm_buffer_(shm_buffer),
+      shm_size_(shm_size),
+      parser_thread_weak_factory_(this),
+      main_thread_weak_factory_(this),
+      main_thread_message_loop(ChildThread::current()->message_loop()) {
+  DCHECK(ChildThread::current());
+  DCHECK(main_thread_message_loop);
+
+  filter_ = new ParserResourceMessageFilter(
+      ChildProcess::current()->io_message_loop_proxy(),
+      main_thread_message_loop,
+      parser_thread(),
+      parser_thread_weak_factory_.GetWeakPtr(),
+      main_thread_weak_factory_.GetWeakPtr(),
+      request_id);
+
+  ChildThread::current()->channel()->AddFilter(filter_.get());
+}
+
+WebThreadedResourceProviderImpl::~WebThreadedResourceProviderImpl() {
+  DCHECK(ChildThread::current());
+
+  // Release it from our locally held member variable before
+  // posting a message to remove it from the I/O thread, to
+  // make sure there's no race conditions and it's always destructed
+  // on the I/O thread.
+  // Ideally we'd also assert on refcount = 2 here...
+  IPC::ChannelProxy::MessageFilter* filter = filter_.get();
+  filter_ = scoped_refptr<IPC::ChannelProxy::MessageFilter>();
+  ChildThread::current()->channel()->RemoveFilter(filter);
+
+  delete foregroundClient_;

[darin (slow to review), 2014/02/11 07:14:23]

nit: google variable naming style is foreground_client_

[oystein (OOO til 10th of July), 2014/02/11 21:46:44]

Done.

+}
+
+blink::WebThread* WebThreadedResourceProviderImpl::resourceThread() {
+  return &parser_thread();

[darin (slow to review), 2014/02/11 07:14:23]

It is not clear why resourceThread() is a method since it always returns the
same value.

[oystein (OOO til 10th of July), 2014/02/11 21:46:44]

resourceProviderThread() (renamed now) is implementing a Blink interface method
and needs to return a blink::WebThread* for use inside Blink, whereas
parser_thread (now renamed resource_provider_thread) returns a
webkit_glue::WebThreadImpl& so we can post tasks to it on the Chrome-side.

+}
+
+void WebThreadedResourceProviderImpl::setBackgroundClient(
+    blink::WebThreadedResourceBackgroundClient* backgroundClient) {
+  DCHECK(parser_thread().isCurrentThread());
+  backgroundClient_ = backgroundClient;
+
+  if (!backgroundClient_) {
+    // When this happens, the provider should no longer be called on the
+    // parser thread as it's about to be destroyed on the main thread.
+    // Invalidating the weak pointers means no callbacks from the filter
+    // will happen.
+    parser_thread_weak_factory_.InvalidateWeakPtrs();
+  }
+}
+
+void WebThreadedResourceProviderImpl::setForegroundClient(
+    blink::WebThreadedResourceForegroundClient* foregroundClient) {
+  DCHECK(ChildThread::current());
+  DCHECK(!foregroundClient_ && foregroundClient);
+  foregroundClient_ = foregroundClient;
+}
+
+void WebThreadedResourceProviderImpl::OnParserResourceMessageFilterAdded() {
+  DCHECK(ChildThread::current());
+  DCHECK(foregroundClient_);
+  DCHECK(resource_dispatcher_);
+  foregroundClient_->didSwitchedToBackgroundClient();
+  resource_dispatcher_->StopSendingDataACKs(request_id_);
+}
+
+void WebThreadedResourceProviderImpl::OnReceivedData(int data_offset,
+                                                     int data_length,
+                                                     int encoded_data_length) {
+  DCHECK(parser_thread().isCurrentThread());
+  DCHECK(shm_buffer_ != NULL);
+  DCHECK(backgroundClient_);
+
+  CHECK_GE(shm_size_, data_offset + data_length);
+  const char* data_ptr = static_cast<char*>(shm_buffer_->memory());
+  CHECK(data_ptr);
+  CHECK(data_ptr + data_offset);
+
+  // TODO: XSS validation and other stuff needs to happen to happen

[darin (slow to review), 2014/02/11 07:14:23]

nit: TODO(username)

[oystein (OOO til 10th of July), 2014/02/11 21:46:44]

Done.

+  // here before we pass it to the parser (or earlier on the I/O thread)
+  backgroundClient_->didReceivedData(data_ptr + data_offset, data_length);
+
+  // Sending a message via the main thread here that it can ACK the received
+  // data guarantees that the shared memory buffer won't be released by
+  // the browser process until we're done with it.

[darin (slow to review), 2014/02/11 07:14:23]

true, but why does this message need to be sent from the main thread? you have
the request_id_ so you should be able to send the ACK directly from the IO
thread.

[oystein (OOO til 10th of July), 2014/02/11 21:46:44]

Clarified this in the comment a bit; it's because the incoming data is received
on both the resource thread and the main thread. If we directly ACKed from the
background thread we'd risk invalidating the shared memory buffer while the main
thread is still dealing with the data.

+  main_thread_message_loop->PostTask(FROM_HERE, base::Bind(
+      &ResourceDispatcher::SendDataACK,
+      resource_dispatcher_,
+      request_id_));
+}
+
+}  // namespace content