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