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

content/child/threadeddataprovider.cc

Index: content/child/threadeddataprovider.cc
diff --git a/content/child/threadeddataprovider.cc b/content/child/threadeddataprovider.cc
new file mode 100644
index 0000000000000000000000000000000000000000..86d30b8f69e80c9937151f444c4a251b03b79025
--- /dev/null
+++ b/content/child/threadeddataprovider.cc
@@ -0,0 +1,283 @@
+// 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/threadeddataprovider.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/child/webthread_impl.h"
+#include "content/common/resource_messages.h"
+#include "ipc/ipc_sync_channel.h"
+#include "third_party/WebKit/public/platform/WebThread.h"
+#include "third_party/WebKit/public/platform/WebThreadedDataReceiver.h"
+
+namespace content {
+
+namespace {
+
+class DataProviderMessageFilter : public IPC::ChannelProxy::MessageFilter {
+ public:
+  DataProviderMessageFilter(
+      const scoped_refptr<base::MessageLoopProxy>& io_message_loop,
+      base::MessageLoop* main_thread_message_loop,
+      WebThreadImpl& background_thread,

[darin (slow to review), 2014/03/19 03:51:28]

nit: google c++ style forbids non-const reference parameters. use a pointer if
this needs to be non-const.

[oystein (OOO til 10th of July), 2014/03/21 19:26:15]

Done.

+      base::WeakPtr<ThreadedDataProvider>

[darin (slow to review), 2014/03/19 03:51:28]

nit: pass by |const base::WeakPtr<...>&| ?

[oystein (OOO til 10th of July), 2014/03/21 19:26:15]

Done.

+          background_thread_resource_provider,
+      base::WeakPtr<ThreadedDataProvider>
+          main_thread_resource_provider,
+      int request_id);
+
+  // IPC::ChannelProxy::MessageFilter
+  virtual void OnFilterAdded(IPC::Channel* channel) OVERRIDE FINAL;
+  virtual bool OnMessageReceived(const IPC::Message& message) OVERRIDE FINAL;
+
+ private:
+  virtual ~DataProviderMessageFilter();
+
+  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_;
+  WebThreadImpl& background_thread_;
+  // This weakptr can only be dereferenced on the background thread.
+  base::WeakPtr<ThreadedDataProvider>
+      background_thread_resource_provider_;
+  // This weakptr can only be dereferenced on the main thread.
+  base::WeakPtr<ThreadedDataProvider>
+      main_thread_resource_provider_;
+  int request_id_;
+};
+
+DataProviderMessageFilter::DataProviderMessageFilter(
+    const scoped_refptr<base::MessageLoopProxy>& io_message_loop,
+    base::MessageLoop* main_thread_message_loop,
+    WebThreadImpl& background_thread,
+    base::WeakPtr<ThreadedDataProvider>
+        background_thread_resource_provider,
+    base::WeakPtr<ThreadedDataProvider>
+        main_thread_resource_provider,
+    int request_id)
+    : io_message_loop_(io_message_loop),
+      main_thread_message_loop_(main_thread_message_loop),
+      background_thread_(background_thread),
+      background_thread_resource_provider_(background_thread_resource_provider),
+      main_thread_resource_provider_(main_thread_resource_provider),
+      request_id_(request_id) {
+  DCHECK(main_thread_message_loop != NULL);
+}
+
+void DataProviderMessageFilter::OnFilterAdded(IPC::Channel* channel) {
+  DCHECK(io_message_loop_->BelongsToCurrentThread());
+
+  main_thread_message_loop_->PostTask(FROM_HERE,
+      base::Bind(
+          &ThreadedDataProvider::OnResourceMessageFilterAddedMainThread,
+          main_thread_resource_provider_));
+}
+
+bool DataProviderMessageFilter::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);
+      return true;
+    }
+  }
+
+  return false;
+}
+
+DataProviderMessageFilter::~DataProviderMessageFilter() {
+  DCHECK(io_message_loop_->BelongsToCurrentThread());
+}
+
+void DataProviderMessageFilter::OnReceivedData(int request_id,
+                                                 int data_offset,

[darin (slow to review), 2014/03/19 03:51:28]

nit: indentation

[oystein (OOO til 10th of July), 2014/03/21 19:26:15]

Done.

+                                                 int data_length,
+                                                 int encoded_data_length) {
+  DCHECK(io_message_loop_->BelongsToCurrentThread());
+  background_thread_.message_loop()->PostTask(FROM_HERE, base::Bind(
+      &ThreadedDataProvider::OnReceivedDataOnBackgroundThread,
+      background_thread_resource_provider_,
+      data_offset, data_length, encoded_data_length));
+}
+
+}  // anonymous namespace
+
+ThreadedDataProvider::ThreadedDataProvider(
+    int request_id, blink::WebThreadedDataReceiver* threaded_data_receiver,
+    linked_ptr<base::SharedMemory> shm_buffer, int shm_size)
+    : request_id_(request_id),
+      shm_buffer_(shm_buffer),
+      shm_size_(shm_size),
+      background_thread_weak_factory_(this),
+      main_thread_weak_factory_(this),
+      // Is this casting kosher, when we have a Chrome-constructed object sent
+      // back up from Blink?
+      background_thread_(
+          static_cast<WebThreadImpl&>(
+              *threaded_data_receiver->backgroundThread())),
+      ipc_channel_(ChildThread::current()->channel()),
+      threaded_data_receiver_(threaded_data_receiver),
+      resource_filter_active_(false),
+      main_thread_message_loop_(ChildThread::current()->message_loop()) {
+  DCHECK(ChildThread::current());
+  DCHECK(ipc_channel_);
+  DCHECK(threaded_data_receiver_);
+  DCHECK(main_thread_message_loop_);
+
+  filter_ = new DataProviderMessageFilter(
+      ChildProcess::current()->io_message_loop_proxy(),
+      main_thread_message_loop_,
+      background_thread_,
+      background_thread_weak_factory_.GetWeakPtr(),
+      main_thread_weak_factory_.GetWeakPtr(),
+      request_id);
+
+  ChildThread::current()->channel()->AddFilter(filter_.get());
+}
+
+ThreadedDataProvider::~ThreadedDataProvider() {
+  DCHECK(ChildThread::current());
+
+  // Release the filter from our locally held member variable before

[darin (slow to review), 2014/03/19 03:51:28]

A better approach here is to use the traits template parameter to
RefCountedThreadSafe<...>. Perhaps all MessageFilter subclasses should be
destroyed on the IO thread? This is probably a yak shave best left for another
CL, but it seems like the current code in this CL is super fragile.

[oystein (OOO til 10th of July), 2014/03/21 19:26:15]

Looking through this again it seems whatever reason there was for making sure
the MessageFilter always gets destructed on the I/O thread isn't there anymore,
probably only present in an earlier version of the patch. So I removed the
sketchiness here.

+  // 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_ = nullptr;

[darin (slow to review), 2014/03/19 03:51:28]

i don't think we use nullptr in chromium code yet.

[oystein (OOO til 10th of July), 2014/03/21 19:26:15]

Done.

+  ChildThread::current()->channel()->RemoveFilter(filter);
+
+  delete threaded_data_receiver_;
+}
+
+void DestructOnMainThread(ThreadedDataProvider* dataProvider) {

[darin (slow to review), 2014/03/19 03:51:28]

nit: dataProvider -> data_provider

(please double check all of the other variable names)

[oystein (OOO til 10th of July), 2014/03/21 19:26:15]

Sorry about all of these; the Chrome<->Blink style context switching still trips
me up.

+  DCHECK(ChildThread::current());
+
+  // The ThreadedDataProvider must be destructed on the main thread to
+  // be threadsafe when removing the message filter and releasing the shared
+  // memory buffer.
+  delete dataProvider;
+}
+
+void ThreadedDataProvider::Stop() {
+  DCHECK(ChildThread::current());
+
+  // We can't destroy this instance directly; we need to bounce a message over
+  // to the background thread and back to make sure nothing else will access it
+  // there, before we can destruct it.
+  background_thread_.message_loop()->PostTask(FROM_HERE,
+      base::Bind(&ThreadedDataProvider::StopOnBackgroundThread,
+                 base::Unretained(this)));
+}
+
+void ThreadedDataProvider::StopOnBackgroundThread() {
+  DCHECK(background_thread_.isCurrentThread());
+
+  // When this happens, the provider should no longer be called on the
+  // background thread as it's about to be destroyed on the main
+  // thread. Invalidating the weak pointers means no callbacks from the filter
+  // will happen and nothing else will use this instance on the
+  // background thread.
+  background_thread_weak_factory_.InvalidateWeakPtrs();
+  main_thread_message_loop_->PostTask(FROM_HERE,
+      base::Bind(&DestructOnMainThread, this));
+}
+
+void ThreadedDataProvider::OnResourceMessageFilterAddedMainThread() {
+  DCHECK(ChildThread::current());
+
+  // We bounce this message from the I/O thread via the main thread and then
+  // to our background thread, following the same path as incoming data before
+  // our filter gets added, to make sure there's nothing still incoming.
+  background_thread_.message_loop()->PostTask(FROM_HERE,
+      base::Bind(
+          &ThreadedDataProvider::OnResourceMessageFilterAddedBackgroundThread,
+          background_thread_weak_factory_.GetWeakPtr()));
+}
+
+void ThreadedDataProvider::OnResourceMessageFilterAddedBackgroundThread() {
+  resource_filter_active_ = true;

[darin (slow to review), 2014/03/19 03:51:28]

nit: add DCHECK(background_thread_.isCurrentThread()) ?

[oystein (OOO til 10th of July), 2014/03/21 19:26:15]

Done.

+
+  // At this point we know no more data is going to arrive from the main thread,
+  // so we can process any data we've received directly from the I/O thread
+  // in the meantime.
+  if (!queued_data_.empty()) {
+    std::vector<QueuedSharedMemoryData>::iterator iter = queued_data_.begin();
+    for (; iter != queued_data_.end(); ++iter)
+      ForwardAndACKData(iter->data, iter->length);
+
+    std::vector<QueuedSharedMemoryData> emptyList;

[darin (slow to review), 2014/03/19 03:51:28]

nit: emptyList -> empty_list

+    // We'll never need this again, so free the memory.
+    queued_data_.swap(emptyList);

[darin (slow to review), 2014/03/19 03:51:28]

nit: you could also just call .clear() on the vector.

[oystein (OOO til 10th of July), 2014/03/21 19:26:15]

AFAIK .clear() doesn't actually free any memory, but I guess it doesn't really
matter in this case. Done.

+  }
+}
+
+void ThreadedDataProvider::OnReceivedDataOnBackgroundThread(
+    int data_offset, int data_length, int encoded_data_length) {
+  DCHECK(background_thread_.isCurrentThread());
+  DCHECK(shm_buffer_ != 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);
+
+  if (resource_filter_active_) {
+    ForwardAndACKData(data_ptr + data_offset, data_length);
+  } else {
+    // There's a brief interval between the point where we know the filter
+    // has been installed on the I/O thread, and when we know for sure there's
+    // no more data coming in from the main thread (from before the filter
+    // got added). If we get any data during that interval, we need to queue
+    // it until we're certain we've processed all the main thread data to make
+    // sure we forward (and ACK) everything in the right order.
+    QueuedSharedMemoryData queuedData;
+    queuedData.data = data_ptr + data_offset;

[darin (slow to review), 2014/03/19 03:51:28]

nit: queuedData -> queued_data

+    queuedData.length = data_length;
+    queued_data_.push_back(queuedData);
+  }
+}
+
+void ThreadedDataProvider::OnReceivedDataOnForegroundThread(
+    const char* data, int data_length, int encoded_data_length) {
+  DCHECK(ChildThread::current());
+
+  background_thread_.message_loop()->PostTask(FROM_HERE,
+      base::Bind(&ThreadedDataProvider::ForwardAndACKData,
+                 base::Unretained(this),
+                 data, data_length));
+}
+
+void ThreadedDataProvider::ForwardAndACKData(const char* data,
+                                             int data_length) {
+  DCHECK(background_thread_.isCurrentThread());
+
+  // TODO(oysteine): SiteIsolationPolicy needs to be be checked

[darin (slow to review), 2014/03/19 03:51:28]

please make sure that creis@ is aware of this.

[oystein (OOO til 10th of July), 2014/03/21 19:26:15]

Yeah we've discussed it in the past, and I landed a warmup CL to this one a
couple of weeks ago which will make the needed fix pretty trivial. I plan on
getting that landed as soon as this is in.

+  // here before we pass the data to the data provider
+  // (or earlier on the I/O thread), otherwise once SiteIsolationPolicy does
+  // actual blocking as opposed to just UMA logging this will bypass it.
+  threaded_data_receiver_->acceptData(data, data_length);
+  ipc_channel_->Send(new ResourceHostMsg_DataReceived_ACK(request_id_));
+}
+
+}  // namespace content