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
|
/* -*- 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/. */
#ifndef TaskQueue_h_
#define TaskQueue_h_
#include "mozilla/Monitor.h"
#include "mozilla/MozPromise.h"
#include "mozilla/RefPtr.h"
#include "mozilla/TaskDispatcher.h"
#include "mozilla/Unused.h"
#include <queue>
#include "nsThreadUtils.h"
class nsIEventTarget;
class nsIRunnable;
namespace mozilla {
typedef MozPromise<bool, bool, false> ShutdownPromise;
// Abstracts executing runnables in order on an arbitrary event target. The
// runnables dispatched to the TaskQueue will be executed in the order in which
// they're received, and are guaranteed to not be executed concurrently.
// They may be executed on different threads, and a memory barrier is used
// to make this threadsafe for objects that aren't already threadsafe.
//
// Note, since a TaskQueue can also be converted to an nsIEventTarget using
// WrapAsEventTarget() its possible to construct a hierarchy of TaskQueues.
// Consider these three TaskQueues:
//
// TQ1 dispatches to the main thread
// TQ2 dispatches to TQ1
// TQ3 dispatches to TQ1
//
// This ensures there is only ever a single runnable from the entire chain on
// the main thread. It also ensures that TQ2 and TQ3 only have a single runnable
// in TQ1 at any time.
//
// This arrangement lets you prioritize work by dispatching runnables directly
// to TQ1. You can issue many runnables for important work. Meanwhile the TQ2
// and TQ3 work will always execute at most one runnable and then yield.
class TaskQueue : public AbstractThread
{
class EventTargetWrapper;
public:
explicit TaskQueue(already_AddRefed<nsIEventTarget> aTarget,
bool aSupportsTailDispatch = false);
TaskDispatcher& TailDispatcher() override;
TaskQueue* AsTaskQueue() override { return this; }
void Dispatch(already_AddRefed<nsIRunnable> aRunnable,
DispatchFailureHandling aFailureHandling = AssertDispatchSuccess,
DispatchReason aReason = NormalDispatch) override
{
nsCOMPtr<nsIRunnable> r = aRunnable;
{
MonitorAutoLock mon(mQueueMonitor);
nsresult rv = DispatchLocked(/* passed by ref */r, aFailureHandling, aReason);
MOZ_DIAGNOSTIC_ASSERT(aFailureHandling == DontAssertDispatchSuccess || NS_SUCCEEDED(rv));
Unused << rv;
}
// If the ownership of |r| is not transferred in DispatchLocked() due to
// dispatch failure, it will be deleted here outside the lock. We do so
// since the destructor of the runnable might access TaskQueue and result
// in deadlocks.
}
// Puts the queue in a shutdown state and returns immediately. The queue will
// remain alive at least until all the events are drained, because the Runners
// hold a strong reference to the task queue, and one of them is always held
// by the target event queue when the task queue is non-empty.
//
// The returned promise is resolved when the queue goes empty.
RefPtr<ShutdownPromise> BeginShutdown();
// Blocks until all task finish executing.
void AwaitIdle();
// Blocks until the queue is flagged for shutdown and all tasks have finished
// executing.
void AwaitShutdownAndIdle();
bool IsEmpty();
uint32_t ImpreciseLengthForHeuristics();
// Returns true if the current thread is currently running a Runnable in
// the task queue.
bool IsCurrentThreadIn() override;
// Create a new nsIEventTarget wrapper object that dispatches to this
// TaskQueue.
already_AddRefed<nsIEventTarget> WrapAsEventTarget();
protected:
virtual ~TaskQueue();
// Blocks until all task finish executing. Called internally by methods
// that need to wait until the task queue is idle.
// mQueueMonitor must be held.
void AwaitIdleLocked();
nsresult DispatchLocked(nsCOMPtr<nsIRunnable>& aRunnable,
DispatchFailureHandling aFailureHandling,
DispatchReason aReason = NormalDispatch);
void MaybeResolveShutdown()
{
mQueueMonitor.AssertCurrentThreadOwns();
if (mIsShutdown && !mIsRunning) {
mShutdownPromise.ResolveIfExists(true, __func__);
mTarget = nullptr;
}
}
nsCOMPtr<nsIEventTarget> mTarget;
// Monitor that protects the queue and mIsRunning;
Monitor mQueueMonitor;
// Queue of tasks to run.
std::queue<nsCOMPtr<nsIRunnable>> mTasks;
// The thread currently running the task queue. We store a reference
// to this so that IsCurrentThreadIn() can tell if the current thread
// is the thread currently running in the task queue.
//
// This may be read on any thread, but may only be written on mRunningThread.
// The thread can't die while we're running in it, and we only use it for
// pointer-comparison with the current thread anyway - so we make it atomic
// and don't refcount it.
Atomic<nsIThread*> mRunningThread;
// RAII class that gets instantiated for each dispatched task.
class AutoTaskGuard : public AutoTaskDispatcher
{
public:
explicit AutoTaskGuard(TaskQueue* aQueue)
: AutoTaskDispatcher(/* aIsTailDispatcher = */ true), mQueue(aQueue)
, mLastCurrentThread(nullptr)
{
// NB: We don't hold the lock to aQueue here. Don't do anything that
// might require it.
MOZ_ASSERT(!mQueue->mTailDispatcher);
mQueue->mTailDispatcher = this;
mLastCurrentThread = sCurrentThreadTLS.get();
sCurrentThreadTLS.set(aQueue);
MOZ_ASSERT(mQueue->mRunningThread == nullptr);
mQueue->mRunningThread = NS_GetCurrentThread();
}
~AutoTaskGuard()
{
DrainDirectTasks();
MOZ_ASSERT(mQueue->mRunningThread == NS_GetCurrentThread());
mQueue->mRunningThread = nullptr;
sCurrentThreadTLS.set(mLastCurrentThread);
mQueue->mTailDispatcher = nullptr;
}
private:
TaskQueue* mQueue;
AbstractThread* mLastCurrentThread;
};
TaskDispatcher* mTailDispatcher;
// True if we've dispatched an event to the target to execute events from
// the queue.
bool mIsRunning;
// True if we've started our shutdown process.
bool mIsShutdown;
MozPromiseHolder<ShutdownPromise> mShutdownPromise;
class Runner : public Runnable {
public:
explicit Runner(TaskQueue* aQueue)
: mQueue(aQueue)
{
}
NS_IMETHOD Run() override;
private:
RefPtr<TaskQueue> mQueue;
};
};
} // namespace mozilla
#endif // TaskQueue_h_
|
