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 "content/child/child_process.h"
   8 #include "content/child/child_thread.h"
   9 #include "content/child/resource_dispatcher.h"
  10 #include "content/child/thread_safe_sender.h"
  11 #include "content/child/webthread_impl.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       base::MessageLoop* main_thread_message_loop,
  26       const WebThreadImpl& background_thread,
  27       const base::WeakPtr<ThreadedDataProvider>&
  28           background_thread_resource_provider,
  29       const base::WeakPtr<ThreadedDataProvider>&
  30           main_thread_resource_provider,
  31       int request_id);
  32
  33   // IPC::ChannelProxy::MessageFilter
  34   virtual void OnFilterAdded(IPC::Sender* sender) OVERRIDE FINAL;
  35   virtual bool OnMessageReceived(const IPC::Message& message) OVERRIDE FINAL;
  36
  37  private:
  38   virtual ~DataProviderMessageFilter() { }
  39
  40   void OnReceivedData(int request_id, int data_offset, int data_length,
  41                       int encoded_data_length);
  42
  43   const scoped_refptr<base::MessageLoopProxy> io_message_loop_;
  44   base::MessageLoop* main_thread_message_loop_;
  45   const WebThreadImpl& background_thread_;
  46   // This weakptr can only be dereferenced on the background thread.
  47   base::WeakPtr<ThreadedDataProvider>
  48       background_thread_resource_provider_;
  49   // This weakptr can only be dereferenced on the main thread.
  50   base::WeakPtr<ThreadedDataProvider>
  51       main_thread_resource_provider_;
  52   int request_id_;
  53 };
  54
  55 DataProviderMessageFilter::DataProviderMessageFilter(
  56     const scoped_refptr<base::MessageLoopProxy>& io_message_loop,
  57     base::MessageLoop* main_thread_message_loop,
  58     const WebThreadImpl& background_thread,
  59     const base::WeakPtr<ThreadedDataProvider>&
  60         background_thread_resource_provider,
  61     const base::WeakPtr<ThreadedDataProvider>&
  62         main_thread_resource_provider,
  63     int request_id)
  64     : io_message_loop_(io_message_loop),
  65       main_thread_message_loop_(main_thread_message_loop),
  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_message_loop != NULL);
  71 }
  72
  73 void DataProviderMessageFilter::OnFilterAdded(IPC::Sender* sender) {
  74   DCHECK(io_message_loop_->BelongsToCurrentThread());
  75
  76   main_thread_message_loop_->PostTask(FROM_HERE,
  77       base::Bind(
  78           &ThreadedDataProvider::OnResourceMessageFilterAddedMainThread,
  79           main_thread_resource_provider_));
  80 }
  81
  82 bool DataProviderMessageFilter::OnMessageReceived(
  83     const IPC::Message& message) {
  84   DCHECK(io_message_loop_->BelongsToCurrentThread());
  85
  86   if (message.type() != ResourceMsg_DataReceived::ID)
  87     return false;
  88
  89   int request_id;
  90
  91   PickleIterator iter(message);
  92   if (!message.ReadInt(&iter, &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(arg.a, arg.b, arg.c, arg.d);
 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_message_loop_->BelongsToCurrentThread());
 113   background_thread_.message_loop()->PostTask(FROM_HERE, base::Bind(
 114       &ThreadedDataProvider::OnReceivedDataOnBackgroundThread,
 115       background_thread_resource_provider_,
 116       data_offset, data_length, encoded_data_length));
 117 }
 118
 119 }  // anonymous namespace
 120
 121 ThreadedDataProvider::ThreadedDataProvider(
 122     int request_id, blink::WebThreadedDataReceiver* threaded_data_receiver,
 123     linked_ptr<base::SharedMemory> shm_buffer, int shm_size)
 124     : request_id_(request_id),
 125       shm_buffer_(shm_buffer),
 126       shm_size_(shm_size),
 127       main_thread_weak_factory_(this),
 128       background_thread_(
 129           static_cast<WebThreadImpl&>(
 130               *threaded_data_receiver->backgroundThread())),
 131       ipc_channel_(ChildThread::current()->channel()),
 132       threaded_data_receiver_(threaded_data_receiver),
 133       resource_filter_active_(false),
 134       main_thread_message_loop_(ChildThread::current()->message_loop()) {
 135   DCHECK(ChildThread::current());
 136   DCHECK(ipc_channel_);
 137   DCHECK(threaded_data_receiver_);
 138   DCHECK(main_thread_message_loop_);
 139
 140   background_thread_weak_factory_.reset(
 141       new base::WeakPtrFactory<ThreadedDataProvider>(this));
 142
 143   filter_ = new DataProviderMessageFilter(
 144       ChildProcess::current()->io_message_loop_proxy(),
 145       main_thread_message_loop_,
 146       background_thread_,
 147       background_thread_weak_factory_->GetWeakPtr(),
 148       main_thread_weak_factory_.GetWeakPtr(),
 149       request_id);
 150
 151   ChildThread::current()->channel()->AddFilter(filter_);
 152 }
 153
 154 ThreadedDataProvider::~ThreadedDataProvider() {
 155   DCHECK(ChildThread::current());
 156
 157   ChildThread::current()->channel()->RemoveFilter(filter_);
 158
 159   delete threaded_data_receiver_;
 160 }
 161
 162 void DestructOnMainThread(ThreadedDataProvider* data_provider) {
 163   DCHECK(ChildThread::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 data_provider;
 169 }
 170
 171 void ThreadedDataProvider::Stop() {
 172   DCHECK(ChildThread::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<WebThreadImpl*>(&background_thread_));
 191     background_thread_.message_loop()->PostTask(FROM_HERE,
 192         base::Bind(&ThreadedDataProvider::StopOnBackgroundThread,
 193                    base::Unretained(this)));
 194   }
 195 }
 196
 197 void ThreadedDataProvider::StopOnBackgroundThread() {
 198   DCHECK(background_thread_.isCurrentThread());
 199   DCHECK(background_thread_weak_factory_);
 200
 201   // When this happens, the provider should no longer be called on the
 202   // background thread as it's about to be destroyed on the main thread.
 203   // Destructing the weak pointer factory means invalidating the weak pointers
 204   // which means no callbacks from the filter will happen and nothing else will
 205   // use this instance on the background thread.
 206   background_thread_weak_factory_.reset(NULL);
 207   main_thread_message_loop_->PostTask(FROM_HERE,
 208       base::Bind(&DestructOnMainThread, this));
 209 }
 210
 211 void ThreadedDataProvider::OnResourceMessageFilterAddedMainThread() {
 212   DCHECK(ChildThread::current());
 213   DCHECK(background_thread_weak_factory_);
 214
 215   // We bounce this message from the I/O thread via the main thread and then
 216   // to our background thread, following the same path as incoming data before
 217   // our filter gets added, to make sure there's nothing still incoming.
 218   background_thread_.message_loop()->PostTask(FROM_HERE,
 219       base::Bind(
 220           &ThreadedDataProvider::OnResourceMessageFilterAddedBackgroundThread,
 221           background_thread_weak_factory_->GetWeakPtr()));
 222 }
 223
 224 void ThreadedDataProvider::OnResourceMessageFilterAddedBackgroundThread() {
 225   DCHECK(background_thread_.isCurrentThread());
 226   resource_filter_active_ = true;
 227
 228   // At this point we know no more data is going to arrive from the main thread,
 229   // so we can process any data we've received directly from the I/O thread
 230   // in the meantime.
 231   if (!queued_data_.empty()) {
 232     std::vector<QueuedSharedMemoryData>::iterator iter = queued_data_.begin();
 233     for (; iter != queued_data_.end(); ++iter) {
 234       ForwardAndACKData(iter->data, iter->length);
 235     }
 236
 237     queued_data_.clear();
 238   }
 239 }
 240
 241 void ThreadedDataProvider::OnReceivedDataOnBackgroundThread(
 242     int data_offset, int data_length, int encoded_data_length) {
 243   DCHECK(background_thread_.isCurrentThread());
 244   DCHECK(shm_buffer_ != NULL);
 245
 246   CHECK_GE(shm_size_, data_offset + data_length);
 247   const char* data_ptr = static_cast<char*>(shm_buffer_->memory());
 248   CHECK(data_ptr);
 249   CHECK(data_ptr + data_offset);
 250
 251   if (resource_filter_active_) {
 252     ForwardAndACKData(data_ptr + data_offset, data_length);
 253   } else {
 254     // There's a brief interval between the point where we know the filter
 255     // has been installed on the I/O thread, and when we know for sure there's
 256     // no more data coming in from the main thread (from before the filter
 257     // got added). If we get any data during that interval, we need to queue
 258     // it until we're certain we've processed all the main thread data to make
 259     // sure we forward (and ACK) everything in the right order.
 260     QueuedSharedMemoryData queued_data;
 261     queued_data.data = data_ptr + data_offset;
 262     queued_data.length = data_length;
 263     queued_data_.push_back(queued_data);
 264   }
 265 }
 266
 267 void ThreadedDataProvider::OnReceivedDataOnForegroundThread(
 268     const char* data, int data_length, int encoded_data_length) {
 269   DCHECK(ChildThread::current());
 270
 271   background_thread_.message_loop()->PostTask(FROM_HERE,
 272       base::Bind(&ThreadedDataProvider::ForwardAndACKData,
 273                  base::Unretained(this),
 274                  data, data_length));
 275 }
 276
 277 void ThreadedDataProvider::ForwardAndACKData(const char* data,
 278                                              int data_length) {
 279   DCHECK(background_thread_.isCurrentThread());
 280
 281   // TODO(oysteine): SiteIsolationPolicy needs to be be checked
 282   // here before we pass the data to the data provider
 283   // (or earlier on the I/O thread), otherwise once SiteIsolationPolicy does
 284   // actual blocking as opposed to just UMA logging this will bypass it.
 285   threaded_data_receiver_->acceptData(data, data_length);
 286   ipc_channel_->Send(new ResourceHostMsg_DataReceived_ACK(request_id_));
 287 }
 288
 289 }  // namespace content