Threaded data provider: Support main thread data notifications (Chrome side)

content/child/threaded_data_provider.cc

 // 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.
 
 #include "content/child/threaded_data_provider.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"
@@ skipping 188 matching lines @@
   // 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.
   // Destructing the weak pointer factory means invalidating the weak pointers
   // which means no callbacks from the filter will happen and nothing else will
   // use this instance on the background thread.
   background_thread_weak_factory_.reset(NULL);
   main_thread_task_runner_->PostTask(FROM_HERE,
                                      base::Bind(&DestructOnMainThread, this));
 }
 
+void ThreadedDataProvider::OnRequestCompleteForegroundThread(
+      base::WeakPtr<ResourceDispatcher> resource_dispatcher,
+      const ResourceMsg_RequestCompleteData& request_complete_data,
+      const base::TimeTicks& renderer_completion_time) {
+  DCHECK(ChildThread::current());
+
+  background_thread_.message_loop()->PostTask(FROM_HERE,
+      base::Bind(&ThreadedDataProvider::OnRequestCompleteBackgroundThread,
+                 base::Unretained(this), resource_dispatcher,
+                 request_complete_data, renderer_completion_time));
+}
+
+void ThreadedDataProvider::OnRequestCompleteBackgroundThread(
+      base::WeakPtr<ResourceDispatcher> resource_dispatcher,
+      const ResourceMsg_RequestCompleteData& request_complete_data,
+      const base::TimeTicks& renderer_completion_time) {
+  DCHECK(background_thread_.isCurrentThread());
+
+  main_thread_task_runner_->PostTask(FROM_HERE,
+      base::Bind(
+          &ResourceDispatcher::CompletedRequestAfterBackgroundThreadFlush,
+          resource_dispatcher,
+          request_id_,
+          request_complete_data,
+          renderer_completion_time));
+}
+
 void ThreadedDataProvider::OnResourceMessageFilterAddedMainThread() {
   DCHECK(ChildThread::current());
   DCHECK(background_thread_weak_factory_);
 
   // 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() {
   DCHECK(background_thread_.isCurrentThread());
   resource_filter_active_ = true;
 
   // 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);
+      ForwardAndACKData(iter->data, iter->length, iter->encoded_length);
     }
 
     queued_data_.clear();
   }
 }
 
 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);
+    ForwardAndACKData(data_ptr + data_offset, data_length, encoded_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 queued_data;
     queued_data.data = data_ptr + data_offset;
     queued_data.length = data_length;
+    queued_data.encoded_length = encoded_data_length;
     queued_data_.push_back(queued_data);
   }
 }
 
 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));
+                 data, data_length, encoded_data_length));
 }
 
 void ThreadedDataProvider::ForwardAndACKData(const char* data,
-                                             int data_length) {
+                                             int data_length,
+                                             int encoded_data_length) {
   DCHECK(background_thread_.isCurrentThread());
 
   // TODO(oysteine): SiteIsolationPolicy needs to be be checked
   // 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);
+
+  scoped_ptr<std::vector<char> > data_copy;
+  if (threaded_data_receiver_->needsMainthreadDataCopy()) {

[davidben, 2015/01/23 19:46:43]

Can needsMainthreadDataCopy be called on both threads? (Here and line 334.)

[oystein (OOO til 10th of July), 2015/01/23 21:23:47]

Yep, it's threadsafe.

+    data_copy.reset(new std::vector<char>(data_length));
+    memcpy(data_copy->data(), data, data_length);
+  }
+
+  main_thread_task_runner_->PostTask(FROM_HERE,
+      base::Bind(&ThreadedDataProvider::DataNotifyForegroundThread,
+          base::Unretained(this),
+          base::Passed(&data_copy),
+          data_length,
+          encoded_data_length));
+
   ipc_channel_->Send(new ResourceHostMsg_DataReceived_ACK(request_id_));
 }
 
+void ThreadedDataProvider::DataNotifyForegroundThread(
+    scoped_ptr<std::vector<char> > data_copy,
+    int data_length,
+    int encoded_data_length) {
+  DCHECK(!data_copy || threaded_data_receiver_->needsMainthreadDataCopy());

[davidben, 2015/01/23 19:46:43]

Nit: I'd maybe also DCHECK something like
if (data_copy)
  DCHECK_EQ(data_length, data->size());
(Probably will need to cast something to size_t.)

[oystein (OOO til 10th of July), 2015/01/23 21:23:47]

Done.

+
+  threaded_data_receiver_->acceptMainthreadDataNotification(
+      data_copy ? data_copy->data() : NULL, data_length, encoded_data_length);

[davidben, 2015/01/23 19:46:43]

std::vector::data is a C++11-ism. I don't think all of our STLs have it yet.
(Notably stlport on Android.)

[oystein (OOO til 10th of July), 2015/01/23 21:23:47]

Done.

+}
+
 }  // namespace content