-
Notifications
You must be signed in to change notification settings - Fork 10.2k
/
DispatchGlobalExecutor.inc
320 lines (268 loc) · 12.1 KB
/
DispatchGlobalExecutor.inc
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
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
///===--- DispatchGlobalExecutor.inc ------------------------*- C++ -*--===///
///
/// This source file is part of the Swift.org open source project
///
/// Copyright (c) 2014 - 2020 Apple Inc. and the Swift project authors
/// Licensed under Apache License v2.0 with Runtime Library Exception
///
/// See https:///swift.org/LICENSE.txt for license information
/// See https:///swift.org/CONTRIBUTORS.txt for the list of Swift project authors
///
///===------------------------------------------------------------------===///
///
/// The implementation of the global executor when using Dispatch.
///
/// This file is included into GlobalExecutor.cpp only when Dispatch
/// integration is enabled. It is expected to define the following
/// functions:
/// swift_task_enqueueGlobalImpl
/// swift_task_enqueueGlobalWithDelayImpl
/// swift_task_enqueueMainExecutorImpl
/// as well as any Dispatch-specific functions for the runtime.
///
///===------------------------------------------------------------------===///
#if SWIFT_CONCURRENCY_ENABLE_DISPATCH
#include <dispatch/dispatch.h>
#if !defined(_WIN32)
#include <dlfcn.h>
#endif
#endif
// Ensure that Job's layout is compatible with what Dispatch expects.
// Note: MinimalDispatchObjectHeader just has the fields we care about, it is
// not complete and should not be used for anything other than these asserts.
struct MinimalDispatchObjectHeader {
const void *VTable;
int Opaque0;
int Opaque1;
void *Linkage;
};
static_assert(
offsetof(Job, metadata) == offsetof(MinimalDispatchObjectHeader, VTable),
"Job Metadata field must match location of Dispatch VTable field.");
static_assert(offsetof(Job, SchedulerPrivate[Job::DispatchLinkageIndex]) ==
offsetof(MinimalDispatchObjectHeader, Linkage),
"Dispatch Linkage field must match Job "
"SchedulerPrivate[DispatchLinkageIndex].");
/// The function passed to dispatch_async_f to execute a job.
static void __swift_run_job(void *_job) {
Job *job = (Job*) _job;
auto metadata =
reinterpret_cast<const DispatchClassMetadata *>(job->metadata);
metadata->VTableInvoke(job, nullptr, 0);
}
/// The type of a function pointer for enqueueing a Job object onto a dispatch
/// queue.
typedef void (*dispatchEnqueueFuncType)(dispatch_queue_t queue, void *obj,
dispatch_qos_class_t qos);
/// Initialize dispatchEnqueueFunc and then call through to the proper
/// implementation.
static void initializeDispatchEnqueueFunc(dispatch_queue_t queue, void *obj,
dispatch_qos_class_t qos);
/// A function pointer to the function used to enqueue a Job onto a dispatch
/// queue. Initially set to initializeDispatchEnqueueFunc, so that the first
/// call will initialize it. initializeDispatchEnqueueFunc sets it to point
/// either to dispatch_async_swift_job when it's available, otherwise to
/// dispatchEnqueueDispatchAsync.
static std::atomic<dispatchEnqueueFuncType> dispatchEnqueueFunc{
initializeDispatchEnqueueFunc};
/// A small adapter that dispatches a Job onto a queue using dispatch_async_f.
static void dispatchEnqueueDispatchAsync(dispatch_queue_t queue, void *obj,
dispatch_qos_class_t qos) {
dispatch_async_f(queue, obj, __swift_run_job);
}
static void initializeDispatchEnqueueFunc(dispatch_queue_t queue, void *obj,
dispatch_qos_class_t qos) {
dispatchEnqueueFuncType func = nullptr;
// Always fall back to plain dispatch_async_f on Windows for now, and
// also for back-deployed concurrency.
#if !defined(_WIN32) && !defined(SWIFT_CONCURRENCY_BACK_DEPLOYMENT)
if (runtime::environment::concurrencyEnableJobDispatchIntegration())
func = reinterpret_cast<dispatchEnqueueFuncType>(
dlsym(RTLD_NEXT, "dispatch_async_swift_job"));
#endif
if (!func)
func = dispatchEnqueueDispatchAsync;
dispatchEnqueueFunc.store(func, std::memory_order_relaxed);
func(queue, obj, qos);
}
/// Enqueue a Job onto a dispatch queue using dispatchEnqueueFunc.
static void dispatchEnqueue(dispatch_queue_t queue, Job *job,
dispatch_qos_class_t qos, void *executorQueue) {
job->SchedulerPrivate[Job::DispatchQueueIndex] = executorQueue;
dispatchEnqueueFunc.load(std::memory_order_relaxed)(queue, job, qos);
}
static constexpr size_t globalQueueCacheCount =
static_cast<size_t>(JobPriority::UserInteractive) + 1;
static std::atomic<dispatch_queue_t> globalQueueCache[globalQueueCacheCount];
#if defined(SWIFT_CONCURRENCY_BACK_DEPLOYMENT) || !defined(__APPLE__)
extern "C" void dispatch_queue_set_width(dispatch_queue_t dq, long width);
#endif
static dispatch_queue_t getGlobalQueue(JobPriority priority) {
size_t numericPriority = static_cast<size_t>(priority);
if (numericPriority >= globalQueueCacheCount)
swift_Concurrency_fatalError(0, "invalid job priority %#zx");
#ifdef SWIFT_CONCURRENCY_BACK_DEPLOYMENT
std::memory_order loadOrder = std::memory_order_acquire;
#else
std::memory_order loadOrder = std::memory_order_relaxed;
#endif
auto *ptr = &globalQueueCache[numericPriority];
auto queue = ptr->load(loadOrder);
if (SWIFT_LIKELY(queue))
return queue;
#if defined(SWIFT_CONCURRENCY_BACK_DEPLOYMENT) || !defined(__APPLE__)
const int DISPATCH_QUEUE_WIDTH_MAX_LOGICAL_CPUS = -3;
// Create a new cooperative concurrent queue and swap it in.
dispatch_queue_attr_t newQueueAttr = dispatch_queue_attr_make_with_qos_class(
DISPATCH_QUEUE_CONCURRENT, (dispatch_qos_class_t)priority, 0);
dispatch_queue_t newQueue = dispatch_queue_create(
"Swift global concurrent queue", newQueueAttr);
dispatch_queue_set_width(newQueue, DISPATCH_QUEUE_WIDTH_MAX_LOGICAL_CPUS);
if (!ptr->compare_exchange_strong(queue, newQueue,
/*success*/ std::memory_order_release,
/*failure*/ std::memory_order_acquire)) {
dispatch_release(newQueue);
return queue;
}
return newQueue;
#else
// If we don't have a queue cached for this priority, cache it now. This may
// race with other threads doing this at the same time for this priority, but
// that's OK, they'll all end up writing the same value.
queue = dispatch_get_global_queue((dispatch_qos_class_t)priority,
/*flags*/ 0);
// Unconditionally store it back in the cache. If we raced with another
// thread, we'll just overwrite the entry with the same value.
ptr->store(queue, std::memory_order_relaxed);
#endif
return queue;
}
SWIFT_CC(swift)
static void swift_task_enqueueGlobalImpl(Job *job) {
assert(job && "no job provided");
// We really want four things from the global execution service:
// - Enqueuing work should have minimal runtime and memory overhead.
// - Adding work should never result in an "explosion" where many
// more threads are created than the available cores.
// - Jobs should run on threads with an appropriate priority.
// - Thread priorities should temporarily elevatable to avoid
// priority inversions.
//
// Of these, the first two are the most important. Many programs
// do not rely on high-usage priority scheduling, and many priority
// inversions can be avoided at a higher level (albeit with some
// performance cost, e.g. by creating higher-priority tasks to run
// critical sections that contend with high-priority work). In
// contrast, if the async feature adds too much overhead, or if
// heavy use of it leads to thread explosions and memory exhaustion,
// programmers will have no choice but to stop using it. So if
// goals are in conflict, it's best to focus on core properties over
// priority-inversion avoidance.
// We currently use Dispatch for our thread pool on all platforms.
// Dispatch currently backs its serial queues with a global
// concurrent queue that is prone to thread explosions when a flood
// of jobs are added to it. That problem does not apply equally
// to the global concurrent queues returned by dispatch_get_global_queue,
// which are not strictly CPU-limited but are at least much more
// cautious about adding new threads. We cannot safely elevate
// the priorities of work added to this queue using Dispatch's public
// API, but as discussed above, that is less important than avoiding
// performance problems.
JobPriority priority = job->getPriority();
auto queue = getGlobalQueue(priority);
dispatchEnqueue(queue, job, (dispatch_qos_class_t)priority,
DISPATCH_QUEUE_GLOBAL_EXECUTOR);
}
SWIFT_CC(swift)
static void swift_task_enqueueGlobalWithDelayImpl(JobDelay delay,
Job *job) {
assert(job && "no job provided");
dispatch_function_t dispatchFunction = &__swift_run_job;
void *dispatchContext = job;
JobPriority priority = job->getPriority();
auto queue = getGlobalQueue(priority);
job->SchedulerPrivate[Job::DispatchQueueIndex] =
DISPATCH_QUEUE_GLOBAL_EXECUTOR;
dispatch_time_t when = dispatch_time(DISPATCH_TIME_NOW, delay);
dispatch_after_f(when, queue, dispatchContext, dispatchFunction);
}
// TODO: The following is cribbed from libdispatch, we should replace it with an official API
typedef enum {
DISPATCH_CLOCK_UPTIME,
DISPATCH_CLOCK_MONOTONIC,
DISPATCH_CLOCK_WALL,
#define DISPATCH_CLOCK_COUNT (DISPATCH_CLOCK_WALL + 1)
} dispatch_clock_t;
#define DISPATCH_UP_OR_MONOTONIC_TIME_MASK (1ULL << 63)
#define DISPATCH_WALLTIME_MASK (1ULL << 62)
#define DISPATCH_TIME_MAX_VALUE (DISPATCH_WALLTIME_MASK - 1)
static inline uint64_t
_dispatch_time_nano2mach(uint64_t value) {
#if HAS_MACH_TIME
struct mach_timebase_info info = calculateTimebase();
return (value * info.denom) / info.numer;
#else
return value;
#endif
}
static inline dispatch_time_t
_dispatch_clock_and_value_to_time(dispatch_clock_t clock, uint64_t value)
{
if (value >= DISPATCH_TIME_MAX_VALUE) {
return DISPATCH_TIME_FOREVER;
}
switch (clock) {
case DISPATCH_CLOCK_WALL:
return -_dispatch_time_nano2mach(value);
case DISPATCH_CLOCK_UPTIME:
return _dispatch_time_nano2mach(value);
case DISPATCH_CLOCK_MONOTONIC:
return _dispatch_time_nano2mach(value) | DISPATCH_UP_OR_MONOTONIC_TIME_MASK;
}
__builtin_unreachable();
}
// END: REPLACEMENT
SWIFT_CC(swift)
static void swift_task_enqueueGlobalWithDeadlineImpl(long long sec,
long long nsec,
long long tsec,
long long tnsec,
int clock, Job *job) {
assert(job && "no job provided");
dispatch_function_t dispatchFunction = &__swift_run_job;
void *dispatchContext = job;
JobPriority priority = job->getPriority();
auto queue = getGlobalQueue(priority);
job->SchedulerPrivate[Job::DispatchQueueIndex] =
DISPATCH_QUEUE_GLOBAL_EXECUTOR;
dispatch_time_t when;
switch (clock) {
case swift_clock_id_continuous: {
uint64_t q = sec * NSEC_PER_SEC + nsec;
when = _dispatch_clock_and_value_to_time(DISPATCH_CLOCK_MONOTONIC, q);
break;
}
case swift_clock_id_suspending: {
uint64_t q = sec * NSEC_PER_SEC + nsec;
when = _dispatch_clock_and_value_to_time(DISPATCH_CLOCK_UPTIME, q);
break;
}
}
// TODO: this should pass the leeway/tolerance along when it is not -1 nanoseconds
// either a dispatch_source can be created or a better dispatch_after_f can be made for this
dispatch_after_f(when, queue, dispatchContext, dispatchFunction);
}
SWIFT_CC(swift)
static void swift_task_enqueueMainExecutorImpl(Job *job) {
assert(job && "no job provided");
JobPriority priority = job->getPriority();
// This is an inline function that compiles down to a pointer to a global.
auto mainQueue = dispatch_get_main_queue();
dispatchEnqueue(mainQueue, job, (dispatch_qos_class_t)priority, mainQueue);
}
void swift::swift_task_enqueueOnDispatchQueue(Job *job,
HeapObject *_queue) {
JobPriority priority = job->getPriority();
auto queue = reinterpret_cast<dispatch_queue_t>(_queue);
dispatchEnqueue(queue, job, (dispatch_qos_class_t)priority, queue);
}