scheduler: Implement task observers

content/child/threaded_data_provider.cc

 // 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_impl.h"
 #include "content/child/resource_dispatcher.h"
 #include "content/child/thread_safe_sender.h"
@@ skipping 90 matching lines @@
   }
 
   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));
+  background_thread_.TaskRunner()->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,
     scoped_refptr<base::SingleThreadTaskRunner> main_thread_task_runner)
@@ skipping 54 matching lines @@
   // 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. We also need to make sure the background
   // thread is still alive, since Blink could have shut down at this point
   // and freed the thread.
   if (current_background_thread) {
     // We should never end up with a different parser thread than from when the
     // ThreadedDataProvider gets created.
     DCHECK(current_background_thread ==
         static_cast<WebThreadImpl*>(&background_thread_));
-    background_thread_.message_loop()->PostTask(FROM_HERE,
-        base::Bind(&ThreadedDataProvider::StopOnBackgroundThread,
-                   base::Unretained(this)));
+    background_thread_.TaskRunner()->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_task_runner_->PostTask(FROM_HERE,
       base::Bind(&ThreadedDataProvider::DestructOnMainThread,
                  base::Unretained(this)));
 }
 
 void ThreadedDataProvider::OnRequestCompleteForegroundThread(
       base::WeakPtr<ResourceDispatcher> resource_dispatcher,
       const ResourceMsg_RequestCompleteData& request_complete_data,
       const base::TimeTicks& renderer_completion_time) {
   DCHECK(ChildThreadImpl::current());
 
-  background_thread_.message_loop()->PostTask(FROM_HERE,
+  background_thread_.TaskRunner()->PostTask(
+      FROM_HERE,
       base::Bind(&ThreadedDataProvider::OnRequestCompleteBackgroundThread,
                  base::Unretained(this), resource_dispatcher,
                  request_complete_data, renderer_completion_time));
 }
 
 void ThreadedDataProvider::OnRequestCompleteBackgroundThread(
       base::WeakPtr<ResourceDispatcher> resource_dispatcher,
       const ResourceMsg_RequestCompleteData& request_complete_data,
       const base::TimeTicks& renderer_completion_time) {
   DCHECK(background_thread_.isCurrentThread());
 
   main_thread_task_runner_->PostTask(FROM_HERE,
       base::Bind(
           &ResourceDispatcher::CompletedRequestAfterBackgroundThreadFlush,
           resource_dispatcher,
           request_id_,
           request_complete_data,
           renderer_completion_time));
 }
 
 void ThreadedDataProvider::OnResourceMessageFilterAddedMainThread() {
   DCHECK(ChildThreadImpl::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,
+  background_thread_.TaskRunner()->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,
@@ skipping 33 matching lines @@
     queued_data.length = data_length;
     queued_data.encoded_length = encoded_data_length;
     queued_data_.push_back(queued_data);
   }
 }
 
 void ThreadedDataProvider::OnReceivedDataOnForegroundThread(
     const char* data, int data_length, int encoded_data_length) {
   DCHECK(ChildThreadImpl::current());
 
-  background_thread_.message_loop()->PostTask(FROM_HERE,
-      base::Bind(&ThreadedDataProvider::ForwardAndACKData,
-                 base::Unretained(this),
-                 data, data_length, encoded_data_length));
+  background_thread_.TaskRunner()->PostTask(
+      FROM_HERE, base::Bind(&ThreadedDataProvider::ForwardAndACKData,
+                            base::Unretained(this), data, data_length,
+                            encoded_data_length));
 }
 
 void ThreadedDataProvider::ForwardAndACKData(const char* data,
                                              int data_length,
                                              int encoded_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
@@ skipping 23 matching lines @@
     DCHECK(threaded_data_receiver_->needsMainthreadDataCopy());
     DCHECK_EQ((size_t)data_length, data_copy->size());
   }
 
   threaded_data_receiver_->acceptMainthreadDataNotification(
       (data_copy && !data_copy->empty()) ? &data_copy->front() : NULL,
       data_length, encoded_data_length);
 }
 
 }  // namespace content