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.
5 #include "content/child/threaded_data_provider.h"
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"
21 class DataProviderMessageFilter
: public IPC::MessageFilter
{
23 DataProviderMessageFilter(
24 const scoped_refptr
<base::MessageLoopProxy
>& io_message_loop
,
25 scoped_refptr
<base::SingleThreadTaskRunner
> main_thread_task_runner
,
26 const WebThreadImpl
& background_thread
,
27 const base::WeakPtr
<ThreadedDataProvider
>&
28 background_thread_resource_provider
,
29 const base::WeakPtr
<ThreadedDataProvider
>& main_thread_resource_provider
,
32 // IPC::ChannelProxy::MessageFilter
33 void OnFilterAdded(IPC::Sender
* sender
) final
;
34 bool OnMessageReceived(const IPC::Message
& message
) final
;
37 ~DataProviderMessageFilter() override
{}
39 void OnReceivedData(int request_id
, int data_offset
, int data_length
,
40 int encoded_data_length
);
42 const scoped_refptr
<base::MessageLoopProxy
> io_message_loop_
;
43 scoped_refptr
<base::SingleThreadTaskRunner
> main_thread_task_runner_
;
44 const WebThreadImpl
& 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_
;
54 DataProviderMessageFilter::DataProviderMessageFilter(
55 const scoped_refptr
<base::MessageLoopProxy
>& io_message_loop
,
56 scoped_refptr
<base::SingleThreadTaskRunner
> main_thread_task_runner
,
57 const WebThreadImpl
& background_thread
,
58 const base::WeakPtr
<ThreadedDataProvider
>&
59 background_thread_resource_provider
,
60 const base::WeakPtr
<ThreadedDataProvider
>& main_thread_resource_provider
,
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());
71 void DataProviderMessageFilter::OnFilterAdded(IPC::Sender
* sender
) {
72 DCHECK(io_message_loop_
->BelongsToCurrentThread());
74 main_thread_task_runner_
->PostTask(
76 base::Bind(&ThreadedDataProvider::OnResourceMessageFilterAddedMainThread
,
77 main_thread_resource_provider_
));
80 bool DataProviderMessageFilter::OnMessageReceived(
81 const IPC::Message
& message
) {
82 DCHECK(io_message_loop_
->BelongsToCurrentThread());
84 if (message
.type() != ResourceMsg_DataReceived::ID
)
89 PickleIterator
iter(message
);
90 if (!iter
.ReadInt(&request_id
)) {
91 NOTREACHED() << "malformed resource message";
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
));
106 void DataProviderMessageFilter::OnReceivedData(int request_id
,
109 int encoded_data_length
) {
110 DCHECK(io_message_loop_
->BelongsToCurrentThread());
111 background_thread_
.message_loop()->PostTask(FROM_HERE
, base::Bind(
112 &ThreadedDataProvider::OnReceivedDataOnBackgroundThread
,
113 background_thread_resource_provider_
,
114 data_offset
, data_length
, encoded_data_length
));
117 } // anonymous namespace
119 ThreadedDataProvider::ThreadedDataProvider(
121 blink::WebThreadedDataReceiver
* threaded_data_receiver
,
122 linked_ptr
<base::SharedMemory
> shm_buffer
,
124 scoped_refptr
<base::SingleThreadTaskRunner
> main_thread_task_runner
)
125 : request_id_(request_id
),
126 shm_buffer_(shm_buffer
),
128 background_thread_(static_cast<WebThreadImpl
&>(
129 *threaded_data_receiver
->backgroundThread())),
130 ipc_channel_(ChildThread::current()->channel()),
131 threaded_data_receiver_(threaded_data_receiver
),
132 resource_filter_active_(false),
133 main_thread_task_runner_(main_thread_task_runner
),
134 main_thread_weak_factory_(this) {
135 DCHECK(ChildThread::current());
136 DCHECK(ipc_channel_
);
137 DCHECK(threaded_data_receiver_
);
138 DCHECK(main_thread_task_runner_
.get());
140 background_thread_weak_factory_
.reset(
141 new base::WeakPtrFactory
<ThreadedDataProvider
>(this));
143 filter_
= new DataProviderMessageFilter(
144 ChildProcess::current()->io_message_loop_proxy(),
145 main_thread_task_runner_
, background_thread_
,
146 background_thread_weak_factory_
->GetWeakPtr(),
147 main_thread_weak_factory_
.GetWeakPtr(), request_id
);
149 ChildThread::current()->channel()->AddFilter(filter_
.get());
152 ThreadedDataProvider::~ThreadedDataProvider() {
153 DCHECK(ChildThread::current());
155 ChildThread::current()->channel()->RemoveFilter(filter_
.get());
157 delete threaded_data_receiver_
;
160 void DestructOnMainThread(ThreadedDataProvider
* data_provider
) {
161 DCHECK(ChildThread::current());
163 // The ThreadedDataProvider must be destructed on the main thread to
164 // be threadsafe when removing the message filter and releasing the shared
166 delete data_provider
;
169 void ThreadedDataProvider::Stop() {
170 DCHECK(ChildThread::current());
172 // Make sure we don't get called by on the main thread anymore via weak
173 // pointers we've passed to the filter.
174 main_thread_weak_factory_
.InvalidateWeakPtrs();
176 blink::WebThread
* current_background_thread
=
177 threaded_data_receiver_
->backgroundThread();
179 // We can't destroy this instance directly; we need to bounce a message over
180 // to the background thread and back to make sure nothing else will access it
181 // there, before we can destruct it. We also need to make sure the background
182 // thread is still alive, since Blink could have shut down at this point
183 // and freed the thread.
184 if (current_background_thread
) {
185 // We should never end up with a different parser thread than from when the
186 // ThreadedDataProvider gets created.
187 DCHECK(current_background_thread
==
188 static_cast<WebThreadImpl
*>(&background_thread_
));
189 background_thread_
.message_loop()->PostTask(FROM_HERE
,
190 base::Bind(&ThreadedDataProvider::StopOnBackgroundThread
,
191 base::Unretained(this)));
195 void ThreadedDataProvider::StopOnBackgroundThread() {
196 DCHECK(background_thread_
.isCurrentThread());
197 DCHECK(background_thread_weak_factory_
);
199 // When this happens, the provider should no longer be called on the
200 // background thread as it's about to be destroyed on the main thread.
201 // Destructing the weak pointer factory means invalidating the weak pointers
202 // which means no callbacks from the filter will happen and nothing else will
203 // use this instance on the background thread.
204 background_thread_weak_factory_
.reset(NULL
);
205 main_thread_task_runner_
->PostTask(FROM_HERE
,
206 base::Bind(&DestructOnMainThread
, this));
209 void ThreadedDataProvider::OnResourceMessageFilterAddedMainThread() {
210 DCHECK(ChildThread::current());
211 DCHECK(background_thread_weak_factory_
);
213 // We bounce this message from the I/O thread via the main thread and then
214 // to our background thread, following the same path as incoming data before
215 // our filter gets added, to make sure there's nothing still incoming.
216 background_thread_
.message_loop()->PostTask(FROM_HERE
,
218 &ThreadedDataProvider::OnResourceMessageFilterAddedBackgroundThread
,
219 background_thread_weak_factory_
->GetWeakPtr()));
222 void ThreadedDataProvider::OnResourceMessageFilterAddedBackgroundThread() {
223 DCHECK(background_thread_
.isCurrentThread());
224 resource_filter_active_
= true;
226 // At this point we know no more data is going to arrive from the main thread,
227 // so we can process any data we've received directly from the I/O thread
229 if (!queued_data_
.empty()) {
230 std::vector
<QueuedSharedMemoryData
>::iterator iter
= queued_data_
.begin();
231 for (; iter
!= queued_data_
.end(); ++iter
) {
232 ForwardAndACKData(iter
->data
, iter
->length
);
235 queued_data_
.clear();
239 void ThreadedDataProvider::OnReceivedDataOnBackgroundThread(
240 int data_offset
, int data_length
, int encoded_data_length
) {
241 DCHECK(background_thread_
.isCurrentThread());
242 DCHECK(shm_buffer_
!= NULL
);
244 CHECK_GE(shm_size_
, data_offset
+ data_length
);
245 const char* data_ptr
= static_cast<char*>(shm_buffer_
->memory());
247 CHECK(data_ptr
+ data_offset
);
249 if (resource_filter_active_
) {
250 ForwardAndACKData(data_ptr
+ data_offset
, data_length
);
252 // There's a brief interval between the point where we know the filter
253 // has been installed on the I/O thread, and when we know for sure there's
254 // no more data coming in from the main thread (from before the filter
255 // got added). If we get any data during that interval, we need to queue
256 // it until we're certain we've processed all the main thread data to make
257 // sure we forward (and ACK) everything in the right order.
258 QueuedSharedMemoryData queued_data
;
259 queued_data
.data
= data_ptr
+ data_offset
;
260 queued_data
.length
= data_length
;
261 queued_data_
.push_back(queued_data
);
265 void ThreadedDataProvider::OnReceivedDataOnForegroundThread(
266 const char* data
, int data_length
, int encoded_data_length
) {
267 DCHECK(ChildThread::current());
269 background_thread_
.message_loop()->PostTask(FROM_HERE
,
270 base::Bind(&ThreadedDataProvider::ForwardAndACKData
,
271 base::Unretained(this),
275 void ThreadedDataProvider::ForwardAndACKData(const char* data
,
277 DCHECK(background_thread_
.isCurrentThread());
279 // TODO(oysteine): SiteIsolationPolicy needs to be be checked
280 // here before we pass the data to the data provider
281 // (or earlier on the I/O thread), otherwise once SiteIsolationPolicy does
282 // actual blocking as opposed to just UMA logging this will bypass it.
283 threaded_data_receiver_
->acceptData(data
, data_length
);
284 ipc_channel_
->Send(new ResourceHostMsg_DataReceived_ACK(request_id_
));
287 } // namespace content