forked from mozilla/gecko-dev
-
Notifications
You must be signed in to change notification settings - Fork 0
/
JobScheduler.cpp
250 lines (206 loc) · 7.05 KB
/
JobScheduler.cpp
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* vim: set ts=8 sts=2 et sw=2 tw=80: */
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
#include "JobScheduler.h"
#include "Logging.h"
namespace mozilla {
namespace gfx {
JobScheduler* JobScheduler::sSingleton = nullptr;
bool JobScheduler::Init(uint32_t aNumThreads, uint32_t aNumQueues) {
MOZ_ASSERT(!sSingleton);
MOZ_ASSERT(aNumThreads >= aNumQueues);
sSingleton = new JobScheduler();
sSingleton->mNextQueue = 0;
for (uint32_t i = 0; i < aNumQueues; ++i) {
sSingleton->mDrawingQueues.push_back(new MultiThreadedJobQueue());
}
for (uint32_t i = 0; i < aNumThreads; ++i) {
sSingleton->mWorkerThreads.push_back(
WorkerThread::Create(sSingleton->mDrawingQueues[i % aNumQueues]));
}
return true;
}
void JobScheduler::ShutDown() {
MOZ_ASSERT(IsEnabled());
if (!IsEnabled()) {
return;
}
for (auto queue : sSingleton->mDrawingQueues) {
queue->ShutDown();
delete queue;
}
for (WorkerThread* thread : sSingleton->mWorkerThreads) {
// this will block until the thread is joined.
delete thread;
}
sSingleton->mWorkerThreads.clear();
delete sSingleton;
sSingleton = nullptr;
}
JobStatus JobScheduler::ProcessJob(Job* aJob) {
MOZ_ASSERT(aJob);
auto status = aJob->Run();
if (status == JobStatus::Error || status == JobStatus::Complete) {
delete aJob;
}
return status;
}
void JobScheduler::SubmitJob(Job* aJob) {
MOZ_ASSERT(aJob);
RefPtr<SyncObject> start = aJob->GetStartSync();
if (start && start->Register(aJob)) {
// The Job buffer starts with a non-signaled sync object, it
// is now registered in the list of task buffers waiting on the
// sync object, so we should not place it in the queue.
return;
}
GetQueueForJob(aJob)->SubmitJob(aJob);
}
void JobScheduler::Join(SyncObject* aCompletion) {
RefPtr<EventObject> waitForCompletion = new EventObject();
JobScheduler::SubmitJob(new SetEventJob(waitForCompletion, aCompletion));
waitForCompletion->Wait();
}
MultiThreadedJobQueue* JobScheduler::GetQueueForJob(Job* aJob) {
return aJob->IsPinnedToAThread() ? aJob->GetWorkerThread()->GetJobQueue()
: GetDrawingQueue();
}
Job::Job(SyncObject* aStart, SyncObject* aCompletion, WorkerThread* aThread)
: mNextWaitingJob(nullptr),
mStartSync(aStart),
mCompletionSync(aCompletion),
mPinToThread(aThread) {
if (mStartSync) {
mStartSync->AddSubsequent(this);
}
if (mCompletionSync) {
mCompletionSync->AddPrerequisite(this);
}
}
Job::~Job() {
if (mCompletionSync) {
// printf(" -- Job %p dtor completion %p\n", this, mCompletionSync);
mCompletionSync->Signal();
mCompletionSync = nullptr;
}
}
JobStatus SetEventJob::Run() {
mEvent->Set();
return JobStatus::Complete;
}
SetEventJob::SetEventJob(EventObject* aEvent, SyncObject* aStart,
SyncObject* aCompletion, WorkerThread* aWorker)
: Job(aStart, aCompletion, aWorker), mEvent(aEvent) {}
SetEventJob::~SetEventJob() {}
SyncObject::SyncObject(uint32_t aNumPrerequisites)
: mSignals(aNumPrerequisites),
mFirstWaitingJob(nullptr)
#ifdef DEBUG
,
mNumPrerequisites(aNumPrerequisites),
mAddedPrerequisites(0)
#endif
{
}
SyncObject::~SyncObject() { MOZ_ASSERT(mFirstWaitingJob == nullptr); }
bool SyncObject::Register(Job* aJob) {
MOZ_ASSERT(aJob);
// For now, ensure that when we schedule the first subsequent, we have already
// created all of the prerequisites. This is an arbitrary restriction because
// we specify the number of prerequisites in the constructor, but in the
// typical scenario, if the assertion FreezePrerequisite blows up here it
// probably means we got the initial nmber of prerequisites wrong. We can
// decide to remove this restriction if needed.
FreezePrerequisites();
int32_t signals = mSignals;
if (signals > 0) {
AddWaitingJob(aJob);
// Since Register and Signal can be called concurrently, it can happen that
// reading mSignals in Register happens before decrementing mSignals in
// Signal, but SubmitWaitingJobs happens before AddWaitingJob. This ordering
// means the SyncObject ends up in the signaled state with a task sitting in
// the waiting list. To prevent that we check mSignals a second time and
// submit again if signals reached zero in the mean time. We do this instead
// of holding a mutex around mSignals+mJobs to reduce lock contention.
int32_t signals2 = mSignals;
if (signals2 == 0) {
SubmitWaitingJobs();
}
return true;
}
return false;
}
void SyncObject::Signal() {
int32_t signals = --mSignals;
MOZ_ASSERT(signals >= 0);
if (signals == 0) {
SubmitWaitingJobs();
}
}
void SyncObject::AddWaitingJob(Job* aJob) {
// Push (using atomics) the task into the list of waiting tasks.
for (;;) {
Job* first = mFirstWaitingJob;
aJob->mNextWaitingJob = first;
if (mFirstWaitingJob.compareExchange(first, aJob)) {
break;
}
}
}
void SyncObject::SubmitWaitingJobs() {
// Scheduling the tasks can cause code that modifies <this>'s reference
// count to run concurrently, and cause the caller of this function to
// be owned by another thread. We need to make sure the reference count
// does not reach 0 on another thread before the end of this method, so
// hold a strong ref to prevent that!
RefPtr<SyncObject> kungFuDeathGrip(this);
// First atomically swap mFirstWaitingJob and waitingJobs...
Job* waitingJobs = nullptr;
for (;;) {
waitingJobs = mFirstWaitingJob;
if (mFirstWaitingJob.compareExchange(waitingJobs, nullptr)) {
break;
}
}
// ... and submit all of the waiting tasks in waitingJob now that they belong
// to this thread.
while (waitingJobs) {
Job* next = waitingJobs->mNextWaitingJob;
waitingJobs->mNextWaitingJob = nullptr;
JobScheduler::GetQueueForJob(waitingJobs)->SubmitJob(waitingJobs);
waitingJobs = next;
}
}
bool SyncObject::IsSignaled() { return mSignals == 0; }
void SyncObject::FreezePrerequisites() {
MOZ_ASSERT(mAddedPrerequisites == mNumPrerequisites);
}
void SyncObject::AddPrerequisite(Job* aJob) {
MOZ_ASSERT(++mAddedPrerequisites <= mNumPrerequisites);
}
void SyncObject::AddSubsequent(Job* aJob) {}
WorkerThread::WorkerThread(MultiThreadedJobQueue* aJobQueue)
: mQueue(aJobQueue) {
aJobQueue->RegisterThread();
}
void WorkerThread::Run() {
SetName("gfx worker");
for (;;) {
Job* commands = nullptr;
if (!mQueue->WaitForJob(commands)) {
mQueue->UnregisterThread();
return;
}
JobStatus status = JobScheduler::ProcessJob(commands);
if (status == JobStatus::Error) {
// Don't try to handle errors for now, but that's open to discussions.
// I expect errors to be mostly OOM issues.
gfxDevCrash(LogReason::JobStatusError)
<< "Invalid job status " << (int)status;
}
}
}
} // namespace gfx
} // namespace mozilla