content/child/threaded_data_provider.cc

   1 // Copyright 2014 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "content/child/threaded_data_provider.h"
   6
   7 #include "base/location.h"
   8 #include "base/single_thread_task_runner.h"
   9 #include "components/scheduler/child/webthread_impl_for_worker_scheduler.h"
  10 #include "content/child/child_process.h"
  11 #include "content/child/child_thread_impl.h"
  12 #include "content/child/resource_dispatcher.h"
  13 #include "content/child/thread_safe_sender.h"
  14 #include "content/common/resource_messages.h"
  15 #include "ipc/ipc_sync_channel.h"
  16 #include "third_party/WebKit/public/platform/WebThread.h"
  17 #include "third_party/WebKit/public/platform/WebThreadedDataReceiver.h"
  18
  19 namespace content {
  20
  21 namespace {
  22
  23 class DataProviderMessageFilter : public IPC::MessageFilter {
  24  public:
  25   DataProviderMessageFilter(
  26       scoped_refptr<base::SingleThreadTaskRunner> io_task_runner,
  27       scoped_refptr<base::SingleThreadTaskRunner> main_thread_task_runner,
  28       const scheduler::WebThreadImplForWorkerScheduler& background_thread,
  29       const base::WeakPtr<ThreadedDataProvider>&
  30           background_thread_resource_provider,
  31       const base::WeakPtr<ThreadedDataProvider>& main_thread_resource_provider,
  32       int request_id);
  33
  34   // IPC::ChannelProxy::MessageFilter
  35   void OnFilterAdded(IPC::Sender* sender) final;
  36   bool OnMessageReceived(const IPC::Message& message) final;
  37
  38  private:
  39   ~DataProviderMessageFilter() override {}
  40
  41   void OnReceivedData(int request_id, int data_offset, int data_length,
  42                       int encoded_data_length);
  43
  44   const scoped_refptr<base::SingleThreadTaskRunner> io_task_runner_;
  45   scoped_refptr<base::SingleThreadTaskRunner> main_thread_task_runner_;
  46   const scheduler::WebThreadImplForWorkerScheduler& background_thread_;
  47   // This weakptr can only be dereferenced on the background thread.
  48   base::WeakPtr<ThreadedDataProvider>
  49       background_thread_resource_provider_;
  50   // This weakptr can only be dereferenced on the main thread.
  51   base::WeakPtr<ThreadedDataProvider>
  52       main_thread_resource_provider_;
  53   int request_id_;
  54 };
  55
  56 DataProviderMessageFilter::DataProviderMessageFilter(
  57     scoped_refptr<base::SingleThreadTaskRunner> io_task_runner,
  58     scoped_refptr<base::SingleThreadTaskRunner> main_thread_task_runner,
  59     const scheduler::WebThreadImplForWorkerScheduler& background_thread,
  60     const base::WeakPtr<ThreadedDataProvider>&
  61         background_thread_resource_provider,
  62     const base::WeakPtr<ThreadedDataProvider>& main_thread_resource_provider,
  63     int request_id)
  64     : io_task_runner_(io_task_runner),
  65       main_thread_task_runner_(main_thread_task_runner),
  66       background_thread_(background_thread),
  67       background_thread_resource_provider_(background_thread_resource_provider),
  68       main_thread_resource_provider_(main_thread_resource_provider),
  69       request_id_(request_id) {
  70   DCHECK(main_thread_task_runner_.get());
  71 }
  72
  73 void DataProviderMessageFilter::OnFilterAdded(IPC::Sender* sender) {
  74   DCHECK(io_task_runner_->BelongsToCurrentThread());
  75
  76   main_thread_task_runner_->PostTask(
  77       FROM_HERE,
  78       base::Bind(&ThreadedDataProvider::OnResourceMessageFilterAddedMainThread,
  79                  main_thread_resource_provider_));
  80 }
  81
  82 bool DataProviderMessageFilter::OnMessageReceived(
  83     const IPC::Message& message) {
  84   DCHECK(io_task_runner_->BelongsToCurrentThread());
  85
  86   if (message.type() != ResourceMsg_DataReceived::ID)
  87     return false;
  88
  89   int request_id;
  90
  91   PickleIterator iter(message);
  92   if (!iter.ReadInt(&request_id)) {
  93     NOTREACHED() << "malformed resource message";
  94     return true;
  95   }
  96
  97   if (request_id == request_id_) {
  98     ResourceMsg_DataReceived::Schema::Param arg;
  99     if (ResourceMsg_DataReceived::Read(&message, &arg)) {
 100       OnReceivedData(get<0>(arg), get<1>(arg), get<2>(arg), get<3>(arg));
 101       return true;
 102     }
 103   }
 104
 105   return false;
 106 }
 107
 108 void DataProviderMessageFilter::OnReceivedData(int request_id,
 109                                                int data_offset,
 110                                                int data_length,
 111                                                int encoded_data_length) {
 112   DCHECK(io_task_runner_->BelongsToCurrentThread());
 113   background_thread_.TaskRunner()->PostTask(
 114       FROM_HERE,
 115       base::Bind(&ThreadedDataProvider::OnReceivedDataOnBackgroundThread,
 116                  background_thread_resource_provider_, data_offset, data_length,
 117                  encoded_data_length));
 118 }
 119
 120 }  // anonymous namespace
 121
 122 ThreadedDataProvider::ThreadedDataProvider(
 123     int request_id,
 124     blink::WebThreadedDataReceiver* threaded_data_receiver,
 125     linked_ptr<base::SharedMemory> shm_buffer,
 126     int shm_size,
 127     scoped_refptr<base::SingleThreadTaskRunner> main_thread_task_runner)
 128     : request_id_(request_id),
 129       shm_buffer_(shm_buffer),
 130       shm_size_(shm_size),
 131       background_thread_(
 132           static_cast<scheduler::WebThreadImplForWorkerScheduler&>(
 133               *threaded_data_receiver->backgroundThread())),
 134       ipc_channel_(ChildThreadImpl::current()->channel()),
 135       threaded_data_receiver_(threaded_data_receiver),
 136       resource_filter_active_(false),
 137       main_thread_task_runner_(main_thread_task_runner),
 138       main_thread_weak_factory_(this) {
 139   DCHECK(ChildThreadImpl::current());
 140   DCHECK(ipc_channel_);
 141   DCHECK(threaded_data_receiver_);
 142   DCHECK(main_thread_task_runner_.get());
 143
 144   background_thread_weak_factory_.reset(
 145       new base::WeakPtrFactory<ThreadedDataProvider>(this));
 146
 147   filter_ = new DataProviderMessageFilter(
 148       ChildProcess::current()->io_task_runner(), main_thread_task_runner_,
 149       background_thread_, background_thread_weak_factory_->GetWeakPtr(),
 150       main_thread_weak_factory_.GetWeakPtr(), request_id);
 151
 152   ChildThreadImpl::current()->channel()->AddFilter(filter_.get());
 153 }
 154
 155 ThreadedDataProvider::~ThreadedDataProvider() {
 156   DCHECK(ChildThreadImpl::current());
 157
 158   ChildThreadImpl::current()->channel()->RemoveFilter(filter_.get());
 159
 160   delete threaded_data_receiver_;
 161 }
 162
 163 void ThreadedDataProvider::DestructOnMainThread() {
 164   DCHECK(ChildThreadImpl::current());
 165
 166   // The ThreadedDataProvider must be destructed on the main thread to
 167   // be threadsafe when removing the message filter and releasing the shared
 168   // memory buffer.
 169   delete this;
 170 }
 171
 172 void ThreadedDataProvider::Stop() {
 173   DCHECK(ChildThreadImpl::current());
 174
 175   // Make sure we don't get called by on the main thread anymore via weak
 176   // pointers we've passed to the filter.
 177   main_thread_weak_factory_.InvalidateWeakPtrs();
 178
 179   blink::WebThread* current_background_thread =
 180       threaded_data_receiver_->backgroundThread();
 181
 182   // We can't destroy this instance directly; we need to bounce a message over
 183   // to the background thread and back to make sure nothing else will access it
 184   // there, before we can destruct it. We also need to make sure the background
 185   // thread is still alive, since Blink could have shut down at this point
 186   // and freed the thread.
 187   if (current_background_thread) {
 188     // We should never end up with a different parser thread than from when the
 189     // ThreadedDataProvider gets created.
 190     DCHECK(current_background_thread ==
 191            static_cast<scheduler::WebThreadImplForWorkerScheduler*>(
 192                &background_thread_));
 193     background_thread_.TaskRunner()->PostTask(
 194         FROM_HERE, base::Bind(&ThreadedDataProvider::StopOnBackgroundThread,
 195                               base::Unretained(this)));
 196   }
 197 }
 198
 199 void ThreadedDataProvider::StopOnBackgroundThread() {
 200   DCHECK(background_thread_.isCurrentThread());
 201   DCHECK(background_thread_weak_factory_);
 202
 203   // When this happens, the provider should no longer be called on the
 204   // background thread as it's about to be destroyed on the main thread.
 205   // Destructing the weak pointer factory means invalidating the weak pointers
 206   // which means no callbacks from the filter will happen and nothing else will
 207   // use this instance on the background thread.
 208   background_thread_weak_factory_.reset(NULL);
 209   main_thread_task_runner_->PostTask(FROM_HERE,
 210       base::Bind(&ThreadedDataProvider::DestructOnMainThread,
 211                  base::Unretained(this)));
 212 }
 213
 214 void ThreadedDataProvider::OnRequestCompleteForegroundThread(
 215       base::WeakPtr<ResourceDispatcher> resource_dispatcher,
 216       const ResourceMsg_RequestCompleteData& request_complete_data,
 217       const base::TimeTicks& renderer_completion_time) {
 218   DCHECK(ChildThreadImpl::current());
 219
 220   background_thread_.TaskRunner()->PostTask(
 221       FROM_HERE,
 222       base::Bind(&ThreadedDataProvider::OnRequestCompleteBackgroundThread,
 223                  base::Unretained(this), resource_dispatcher,
 224                  request_complete_data, renderer_completion_time));
 225 }
 226
 227 void ThreadedDataProvider::OnRequestCompleteBackgroundThread(
 228       base::WeakPtr<ResourceDispatcher> resource_dispatcher,
 229       const ResourceMsg_RequestCompleteData& request_complete_data,
 230       const base::TimeTicks& renderer_completion_time) {
 231   DCHECK(background_thread_.isCurrentThread());
 232
 233   main_thread_task_runner_->PostTask(FROM_HERE,
 234       base::Bind(
 235           &ResourceDispatcher::CompletedRequestAfterBackgroundThreadFlush,
 236           resource_dispatcher,
 237           request_id_,
 238           request_complete_data,
 239           renderer_completion_time));
 240 }
 241
 242 void ThreadedDataProvider::OnResourceMessageFilterAddedMainThread() {
 243   DCHECK(ChildThreadImpl::current());
 244   DCHECK(background_thread_weak_factory_);
 245
 246   // We bounce this message from the I/O thread via the main thread and then
 247   // to our background thread, following the same path as incoming data before
 248   // our filter gets added, to make sure there's nothing still incoming.
 249   background_thread_.TaskRunner()->PostTask(
 250       FROM_HERE,
 251       base::Bind(
 252           &ThreadedDataProvider::OnResourceMessageFilterAddedBackgroundThread,
 253           background_thread_weak_factory_->GetWeakPtr()));
 254 }
 255
 256 void ThreadedDataProvider::OnResourceMessageFilterAddedBackgroundThread() {
 257   DCHECK(background_thread_.isCurrentThread());
 258   resource_filter_active_ = true;
 259
 260   // At this point we know no more data is going to arrive from the main thread,
 261   // so we can process any data we've received directly from the I/O thread
 262   // in the meantime.
 263   if (!queued_data_.empty()) {
 264     std::vector<QueuedSharedMemoryData>::iterator iter = queued_data_.begin();
 265     for (; iter != queued_data_.end(); ++iter) {
 266       ForwardAndACKData(iter->data, iter->length, iter->encoded_length);
 267     }
 268
 269     queued_data_.clear();
 270   }
 271 }
 272
 273 void ThreadedDataProvider::OnReceivedDataOnBackgroundThread(
 274     int data_offset, int data_length, int encoded_data_length) {
 275   DCHECK(background_thread_.isCurrentThread());
 276   DCHECK(shm_buffer_ != NULL);
 277
 278   CHECK_GE(shm_size_, data_offset + data_length);
 279   const char* data_ptr = static_cast<char*>(shm_buffer_->memory());
 280   CHECK(data_ptr);
 281   CHECK(data_ptr + data_offset);
 282
 283   if (resource_filter_active_) {
 284     ForwardAndACKData(data_ptr + data_offset, data_length, encoded_data_length);
 285   } else {
 286     // There's a brief interval between the point where we know the filter
 287     // has been installed on the I/O thread, and when we know for sure there's
 288     // no more data coming in from the main thread (from before the filter
 289     // got added). If we get any data during that interval, we need to queue
 290     // it until we're certain we've processed all the main thread data to make
 291     // sure we forward (and ACK) everything in the right order.
 292     QueuedSharedMemoryData queued_data;
 293     queued_data.data = data_ptr + data_offset;
 294     queued_data.length = data_length;
 295     queued_data.encoded_length = encoded_data_length;
 296     queued_data_.push_back(queued_data);
 297   }
 298 }
 299
 300 void ThreadedDataProvider::OnReceivedDataOnForegroundThread(
 301     const char* data, int data_length, int encoded_data_length) {
 302   DCHECK(ChildThreadImpl::current());
 303
 304   background_thread_.TaskRunner()->PostTask(
 305       FROM_HERE, base::Bind(&ThreadedDataProvider::ForwardAndACKData,
 306                             base::Unretained(this), data, data_length,
 307                             encoded_data_length));
 308 }
 309
 310 void ThreadedDataProvider::ForwardAndACKData(const char* data,
 311                                              int data_length,
 312                                              int encoded_data_length) {
 313   DCHECK(background_thread_.isCurrentThread());
 314
 315   // TODO(oysteine): SiteIsolationPolicy needs to be be checked
 316   // here before we pass the data to the data provider
 317   // (or earlier on the I/O thread), otherwise once SiteIsolationPolicy does
 318   // actual blocking as opposed to just UMA logging this will bypass it.
 319   threaded_data_receiver_->acceptData(data, data_length);
 320
 321   scoped_ptr<std::vector<char>> data_copy;
 322   if (threaded_data_receiver_->needsMainthreadDataCopy()) {
 323     data_copy.reset(new std::vector<char>(data, data + data_length));
 324   }
 325
 326   main_thread_task_runner_->PostTask(FROM_HERE,
 327       base::Bind(&ThreadedDataProvider::DataNotifyForegroundThread,
 328           base::Unretained(this),
 329           base::Passed(&data_copy),
 330           data_length,
 331           encoded_data_length));
 332
 333   ipc_channel_->Send(new ResourceHostMsg_DataReceived_ACK(request_id_));
 334 }
 335
 336 void ThreadedDataProvider::DataNotifyForegroundThread(
 337     scoped_ptr<std::vector<char> > data_copy,
 338     int data_length,
 339     int encoded_data_length) {
 340   if (data_copy) {
 341     DCHECK(threaded_data_receiver_->needsMainthreadDataCopy());
 342     DCHECK_EQ((size_t)data_length, data_copy->size());
 343   }
 344
 345   threaded_data_receiver_->acceptMainthreadDataNotification(
 346       (data_copy && !data_copy->empty()) ? &data_copy->front() : NULL,
 347       data_length, encoded_data_length);
 348 }
 349
 350 }  // namespace content