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