Index: content/child/threaded_data_provider.cc |
diff --git a/content/child/threaded_data_provider.cc b/content/child/threaded_data_provider.cc |
new file mode 100644 |
index 0000000000000000000000000000000000000000..84fa65402594b1a9b6d63ab7650a35809f3d41df |
--- /dev/null |
+++ b/content/child/threaded_data_provider.cc |
@@ -0,0 +1,278 @@ |
+// 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" |
+#include "content/child/webthread_impl.h" |
+#include "content/common/resource_messages.h" |
+#include "ipc/ipc_sync_channel.h" |
+#include "third_party/WebKit/public/platform/WebThread.h" |
+#include "third_party/WebKit/public/platform/WebThreadedDataReceiver.h" |
+ |
+namespace content { |
+ |
+namespace { |
+ |
+class DataProviderMessageFilter : public IPC::MessageFilter { |
+ public: |
+ DataProviderMessageFilter( |
+ const scoped_refptr<base::MessageLoopProxy>& io_message_loop, |
+ base::MessageLoop* main_thread_message_loop, |
+ const WebThreadImpl& background_thread, |
+ const base::WeakPtr<ThreadedDataProvider>& |
+ background_thread_resource_provider, |
+ const base::WeakPtr<ThreadedDataProvider>& |
+ main_thread_resource_provider, |
+ int request_id); |
+ |
+ // IPC::ChannelProxy::MessageFilter |
+ virtual void OnFilterAdded(IPC::Channel* channel) OVERRIDE FINAL; |
+ virtual bool OnMessageReceived(const IPC::Message& message) OVERRIDE FINAL; |
+ |
+ private: |
+ virtual ~DataProviderMessageFilter() { } |
+ |
+ void OnReceivedData(int request_id, int data_offset, int data_length, |
+ int encoded_data_length); |
+ |
+ const scoped_refptr<base::MessageLoopProxy> io_message_loop_; |
+ base::MessageLoop* main_thread_message_loop_; |
+ const WebThreadImpl& background_thread_; |
+ // This weakptr can only be dereferenced on the background thread. |
+ base::WeakPtr<ThreadedDataProvider> |
+ background_thread_resource_provider_; |
+ // This weakptr can only be dereferenced on the main thread. |
+ base::WeakPtr<ThreadedDataProvider> |
+ main_thread_resource_provider_; |
+ int request_id_; |
+}; |
+ |
+DataProviderMessageFilter::DataProviderMessageFilter( |
+ const scoped_refptr<base::MessageLoopProxy>& io_message_loop, |
+ base::MessageLoop* main_thread_message_loop, |
+ const WebThreadImpl& background_thread, |
+ const base::WeakPtr<ThreadedDataProvider>& |
+ background_thread_resource_provider, |
+ const base::WeakPtr<ThreadedDataProvider>& |
+ main_thread_resource_provider, |
+ int request_id) |
+ : io_message_loop_(io_message_loop), |
+ main_thread_message_loop_(main_thread_message_loop), |
+ background_thread_(background_thread), |
+ background_thread_resource_provider_(background_thread_resource_provider), |
+ main_thread_resource_provider_(main_thread_resource_provider), |
+ request_id_(request_id) { |
+ DCHECK(main_thread_message_loop != NULL); |
+} |
+ |
+void DataProviderMessageFilter::OnFilterAdded(IPC::Channel* channel) { |
+ DCHECK(io_message_loop_->BelongsToCurrentThread()); |
+ |
+ main_thread_message_loop_->PostTask(FROM_HERE, |
+ base::Bind( |
+ &ThreadedDataProvider::OnResourceMessageFilterAddedMainThread, |
+ main_thread_resource_provider_)); |
+} |
+ |
+bool DataProviderMessageFilter::OnMessageReceived( |
+ const IPC::Message& message) { |
+ DCHECK(io_message_loop_->BelongsToCurrentThread()); |
+ |
+ if (message.type() != ResourceMsg_DataReceived::ID) |
+ return false; |
+ |
+ int request_id; |
+ |
+ PickleIterator iter(message); |
+ if (!message.ReadInt(&iter, &request_id)) { |
+ NOTREACHED() << "malformed resource message"; |
+ return true; |
+ } |
+ |
+ if (request_id == request_id_) { |
+ ResourceMsg_DataReceived::Schema::Param arg; |
+ if (ResourceMsg_DataReceived::Read(&message, &arg)) { |
+ OnReceivedData(arg.a, arg.b, arg.c, arg.d); |
+ return true; |
+ } |
+ } |
+ |
+ return false; |
+} |
+ |
+void DataProviderMessageFilter::OnReceivedData(int request_id, |
+ int data_offset, |
+ int data_length, |
+ int encoded_data_length) { |
+ DCHECK(io_message_loop_->BelongsToCurrentThread()); |
+ background_thread_.message_loop()->PostTask(FROM_HERE, base::Bind( |
+ &ThreadedDataProvider::OnReceivedDataOnBackgroundThread, |
+ background_thread_resource_provider_, |
+ data_offset, data_length, encoded_data_length)); |
+} |
+ |
+} // anonymous namespace |
+ |
+ThreadedDataProvider::ThreadedDataProvider( |
+ int request_id, blink::WebThreadedDataReceiver* threaded_data_receiver, |
+ linked_ptr<base::SharedMemory> shm_buffer, int shm_size) |
+ : request_id_(request_id), |
+ shm_buffer_(shm_buffer), |
+ shm_size_(shm_size), |
+ main_thread_weak_factory_(this), |
+ background_thread_( |
+ static_cast<WebThreadImpl&>( |
+ *threaded_data_receiver->backgroundThread())), |
+ ipc_channel_(ChildThread::current()->channel()), |
+ threaded_data_receiver_(threaded_data_receiver), |
+ resource_filter_active_(false), |
+ main_thread_message_loop_(ChildThread::current()->message_loop()) { |
+ DCHECK(ChildThread::current()); |
+ DCHECK(ipc_channel_); |
+ DCHECK(threaded_data_receiver_); |
+ DCHECK(main_thread_message_loop_); |
+ |
+ background_thread_weak_factory_.reset( |
+ new base::WeakPtrFactory<ThreadedDataProvider>(this)); |
+ |
+ filter_ = new DataProviderMessageFilter( |
+ ChildProcess::current()->io_message_loop_proxy(), |
+ main_thread_message_loop_, |
+ background_thread_, |
+ background_thread_weak_factory_->GetWeakPtr(), |
+ main_thread_weak_factory_.GetWeakPtr(), |
+ request_id); |
+ |
+ ChildThread::current()->channel()->AddFilter(filter_); |
+} |
+ |
+ThreadedDataProvider::~ThreadedDataProvider() { |
+ DCHECK(ChildThread::current()); |
+ |
+ ChildThread::current()->channel()->RemoveFilter(filter_); |
+ |
+ delete threaded_data_receiver_; |
+} |
+ |
+void DestructOnMainThread(ThreadedDataProvider* data_provider) { |
+ DCHECK(ChildThread::current()); |
+ |
+ // The ThreadedDataProvider must be destructed on the main thread to |
+ // be threadsafe when removing the message filter and releasing the shared |
+ // memory buffer. |
+ delete data_provider; |
+} |
+ |
+void ThreadedDataProvider::Stop() { |
+ DCHECK(ChildThread::current()); |
+ |
+ // Make sure we don't get called by on the main thread anymore via weak |
+ // pointers we've passed to the filter. |
+ main_thread_weak_factory_.InvalidateWeakPtrs(); |
+ |
+ // We can't destroy this instance directly; we need to bounce a message over |
+ // to the background thread and back to make sure nothing else will access it |
+ // there, before we can destruct it. |
+ background_thread_.message_loop()->PostTask(FROM_HERE, |
+ base::Bind(&ThreadedDataProvider::StopOnBackgroundThread, |
+ base::Unretained(this))); |
+} |
+ |
+void ThreadedDataProvider::StopOnBackgroundThread() { |
+ DCHECK(background_thread_.isCurrentThread()); |
+ DCHECK(background_thread_weak_factory_); |
+ |
+ // 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_message_loop_->PostTask(FROM_HERE, |
+ base::Bind(&DestructOnMainThread, this)); |
+} |
+ |
+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); |
+ } |
+ |
+ 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); |
+ } 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_.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)); |
+} |
+ |
+void ThreadedDataProvider::ForwardAndACKData(const char* data, |
+ int 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); |
+ ipc_channel_->Send(new ResourceHostMsg_DataReceived_ACK(request_id_)); |
+} |
+ |
+} // namespace content |