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

content/child/webparserresourcebridge_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/webparserresourcebridge_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 {
+
+class ParserResourceMessageFilter : public IPC::ChannelProxy::MessageFilter {
+ public:
+  ParserResourceMessageFilter(
+      const scoped_refptr<base::MessageLoopProxy>& io_message_loop,
+      base::MessageLoop* main_thread_message_loop,
+      base::WeakPtr<ResourceDispatcher> resource_dispatcher,
+      webkit_glue::WebThreadImpl& parser_thread,
+      base::WeakPtr<WebParserResourceBridgeImpl> parser_bridge,
+      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_;
+  // This weakptr can only be dereferenced on the main thread.
+  base::WeakPtr<ResourceDispatcher> resource_dispatcher_;
+  webkit_glue::WebThreadImpl& parser_thread_;
+  // This weakptr can only be dereferenced on the parser thread.
+  base::WeakPtr<WebParserResourceBridgeImpl> parser_bridge_;
+  int request_id_;
+};
+
+ParserResourceMessageFilter::ParserResourceMessageFilter(
+    const scoped_refptr<base::MessageLoopProxy>& io_message_loop,
+    base::MessageLoop* main_thread_message_loop,
+    base::WeakPtr<ResourceDispatcher> resource_dispatcher,
+    webkit_glue::WebThreadImpl& parser_thread,
+    base::WeakPtr<WebParserResourceBridgeImpl> parser_bridge,
+    int request_id)
+    : io_message_loop_(io_message_loop)
+    , main_thread_message_loop_(main_thread_message_loop)

[jam, 2013/12/17 00:44:40]

nit: google style is comma on previous lines

[oystein (OOO til 10th of July), 2013/12/17 01:07:27]

Done.

+    , resource_dispatcher_(resource_dispatcher)
+    , parser_thread_(parser_thread)
+    , parser_bridge_(parser_bridge)
+    , 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(
+          &ResourceDispatcher::OnParserResourceMessageFilterAdded,
+          resource_dispatcher_,
+          request_id_));
+}
+
+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(
+      &WebParserResourceBridgeImpl::OnReceivedData,
+      parser_bridge_,
+      data_offset, data_length, encoded_data_length));
+}
+
+}  // anonymous namespace
+
+webkit_glue::WebThreadImpl& WebParserResourceBridgeImpl::parser_thread() {
+  static webkit_glue::WebThreadImpl* thread = NULL;
+  // Do we care about this leaking on shutdown?

[jam, 2013/12/17 00:44:40]

no

[oystein (OOO til 10th of July), 2013/12/17 01:07:27]

Done.

+  if (thread == NULL)
+    thread = new webkit_glue::WebThreadImpl("HTMLParserThread");
+
+  return *thread;
+}
+
+WebParserResourceBridgeImpl::WebParserResourceBridgeImpl(
+    int request_id, base::WeakPtr<ResourceDispatcher> resource_dispatcher,
+    base::SharedMemoryHandle shm_handle, int shm_size)
+    : peer_(NULL)
+    , shm_handle_(shm_handle)
+    , shm_size_(shm_size)

[jam, 2013/12/17 00:44:40]

ditto

[oystein (OOO til 10th of July), 2013/12/17 01:07:27]

Done.

+    , weak_factory_(this) {
+  DCHECK(ChildThread::current() != NULL);
+
+  filter_ = new ParserResourceMessageFilter(
+      ChildProcess::current()->io_message_loop_proxy(),
+      ChildThread::current()->message_loop(),
+      resource_dispatcher,
+      parser_thread(),
+      weak_factory_.GetWeakPtr(),
+      request_id);
+
+  ChildThread::current()->channel()->AddFilter(filter_.get());
+}
+
+WebParserResourceBridgeImpl::~WebParserResourceBridgeImpl() {
+  DCHECK(ChildThread::current() != NULL);
+
+  // 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);
+  // If this fails, we're leaking some shared memory.
+  DCHECK(shm_handle_ == base::SharedMemory::NULLHandle());
+}
+
+blink::WebThread* WebParserResourceBridgeImpl::getParserThread() {
+  return &parser_thread();
+}
+
+void WebParserResourceBridgeImpl::setPeer(
+    blink::WebParserResourceBridge::Peer* peer) {
+  DCHECK(parser_thread().isCurrentThread());
+  peer_ = peer;
+
+  if (peer_ == NULL) {

[jam, 2013/12/17 00:44:40]

nit: if (!peer_). also above and below

[oystein (OOO til 10th of July), 2013/12/17 01:07:27]

Done.

+    // When this happens, the bridge 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.
+    weak_factory_.InvalidateWeakPtrs();
+  } else {
+    CreateSharedMemoryBuffer();
+  }
+}
+
+void WebParserResourceBridgeImpl::OnReceivedData(int data_offset,
+                                                 int data_length,
+                                                 int encoded_data_length) {
+  DCHECK(parser_thread().isCurrentThread());
+  DCHECK(shm_buffer_ != NULL);
+  DCHECK(peer_ != NULL);
+
+  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
+  // here before we pass it to the parser.
+  peer_->OnReceivedData(data_ptr + data_offset, data_length);
+}
+
+void WebParserResourceBridgeImpl::CreateSharedMemoryBuffer() {
+  DCHECK(parser_thread().isCurrentThread());
+  // This assumes OnSetDataBuffer has been received by the main thread when
+  // the bridge is created; once the creation is moved to OnReceivedResponse
+  // rather than the first OnReceiveData, that will no longer always be true
+  // and the filter will have to listen to OnSetDataBuffer itself and forward
+  // the handle to the bridge.
+  DCHECK(base::SharedMemory::IsHandleValid(shm_handle_) || shm_size_ == 0);
+
+  shm_buffer_.reset(new base::SharedMemory(shm_handle_, true));  // read only
+  shm_handle_ = base::SharedMemory::NULLHandle();
+
+  bool ok = shm_buffer_->Map(shm_size_);
+  DCHECK(ok);
+}
+
+}  // namespace content