1 /*
2 * Copyright (C) 2014 The Android Open Source Project
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17 #include "task_processor.h"
18
19 #include "base/time_utils.h"
20 #include "scoped_thread_state_change-inl.h"
21
22 namespace art {
23 namespace gc {
24
TaskProcessor()25 TaskProcessor::TaskProcessor()
26 : lock_("Task processor lock", kReferenceProcessorLock),
27 cond_("Task processor condition", lock_),
28 is_running_(false),
29 running_thread_(nullptr) {
30 }
31
~TaskProcessor()32 TaskProcessor::~TaskProcessor() {
33 if (!tasks_.empty()) {
34 LOG(WARNING) << "TaskProcessor: Finalizing " << tasks_.size() << " unprocessed tasks.";
35 for (HeapTask* task : tasks_) {
36 task->Finalize();
37 }
38 }
39 }
40
AddTask(Thread * self,HeapTask * task)41 void TaskProcessor::AddTask(Thread* self, HeapTask* task) {
42 ScopedThreadStateChange tsc(self, kWaitingForTaskProcessor);
43 MutexLock mu(self, lock_);
44 tasks_.insert(task);
45 cond_.Signal(self);
46 }
47
GetTask(Thread * self)48 HeapTask* TaskProcessor::GetTask(Thread* self) {
49 ScopedThreadStateChange tsc(self, kWaitingForTaskProcessor);
50 MutexLock mu(self, lock_);
51 while (true) {
52 if (tasks_.empty()) {
53 if (!is_running_) {
54 return nullptr;
55 }
56 cond_.Wait(self); // Empty queue, wait until we are signalled.
57 } else {
58 // Non empty queue, look at the top element and see if we are ready to run it.
59 const uint64_t current_time = NanoTime();
60 HeapTask* task = *tasks_.begin();
61 // If we are shutting down, return the task right away without waiting. Otherwise return the
62 // task if it is late enough.
63 uint64_t target_time = task->GetTargetRunTime();
64 if (!is_running_ || target_time <= current_time) {
65 tasks_.erase(tasks_.begin());
66 return task;
67 }
68 DCHECK_GT(target_time, current_time);
69 // Wait until we hit the target run time.
70 const uint64_t delta_time = target_time - current_time;
71 const uint64_t ms_delta = NsToMs(delta_time);
72 const uint64_t ns_delta = delta_time - MsToNs(ms_delta);
73 cond_.TimedWait(self, static_cast<int64_t>(ms_delta), static_cast<int32_t>(ns_delta));
74 }
75 }
76 UNREACHABLE();
77 }
78
UpdateTargetRunTime(Thread * self,HeapTask * task,uint64_t new_target_time)79 void TaskProcessor::UpdateTargetRunTime(Thread* self, HeapTask* task, uint64_t new_target_time) {
80 MutexLock mu(self, lock_);
81 // Find the task.
82 auto range = tasks_.equal_range(task);
83 for (auto it = range.first; it != range.second; ++it) {
84 if (*it == task) {
85 // Check if the target time was updated, if so re-insert then wait.
86 if (new_target_time != task->GetTargetRunTime()) {
87 tasks_.erase(it);
88 task->SetTargetRunTime(new_target_time);
89 tasks_.insert(task);
90 // If we became the first task then we may need to signal since we changed the task that we
91 // are sleeping on.
92 if (*tasks_.begin() == task) {
93 cond_.Signal(self);
94 }
95 return;
96 }
97 }
98 }
99 }
100
IsRunning() const101 bool TaskProcessor::IsRunning() const {
102 MutexLock mu(Thread::Current(), lock_);
103 return is_running_;
104 }
105
GetRunningThread() const106 Thread* TaskProcessor::GetRunningThread() const {
107 MutexLock mu(Thread::Current(), lock_);
108 return running_thread_;
109 }
110
Stop(Thread * self)111 void TaskProcessor::Stop(Thread* self) {
112 MutexLock mu(self, lock_);
113 is_running_ = false;
114 running_thread_ = nullptr;
115 cond_.Broadcast(self);
116 }
117
Start(Thread * self)118 void TaskProcessor::Start(Thread* self) {
119 MutexLock mu(self, lock_);
120 is_running_ = true;
121 running_thread_ = self;
122 }
123
RunAllTasks(Thread * self)124 void TaskProcessor::RunAllTasks(Thread* self) {
125 while (true) {
126 // Wait and get a task, may be interrupted.
127 HeapTask* task = GetTask(self);
128 if (task != nullptr) {
129 task->Run(self);
130 task->Finalize();
131 } else if (!IsRunning()) {
132 break;
133 }
134 }
135 }
136
137 } // namespace gc
138 } // namespace art
139