1 /*
2 * Copyright (c) 2008-2013 Apple Inc. All rights reserved.
3 *
4 * @APPLE_APACHE_LICENSE_HEADER_START@
5 *
6 * Licensed under the Apache License, Version 2.0 (the "License");
7 * you may not use this file except in compliance with the License.
8 * You may obtain a copy of the License at
9 *
10 * http://www.apache.org/licenses/LICENSE-2.0
11 *
12 * Unless required by applicable law or agreed to in writing, software
13 * distributed under the License is distributed on an "AS IS" BASIS,
14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 * See the License for the specific language governing permissions and
16 * limitations under the License.
17 *
18 * @APPLE_APACHE_LICENSE_HEADER_END@
19 */
20
21
22 #include "internal.h"
23 #if HAVE_MACH
24 #include "protocol.h"
25 #endif
26
27 #if (!HAVE_PTHREAD_WORKQUEUES || DISPATCH_DEBUG) && \
28 !defined(DISPATCH_ENABLE_THREAD_POOL)
29 #define DISPATCH_ENABLE_THREAD_POOL 1
30 #endif
31 #if DISPATCH_ENABLE_PTHREAD_ROOT_QUEUES || DISPATCH_ENABLE_THREAD_POOL
32 #define DISPATCH_USE_PTHREAD_POOL 1
33 #endif
34 #if HAVE_PTHREAD_WORKQUEUES && (!HAVE_PTHREAD_WORKQUEUE_QOS || DISPATCH_DEBUG) \
35 && !defined(DISPATCH_USE_NOQOS_WORKQUEUE_FALLBACK)
36 #define DISPATCH_USE_NOQOS_WORKQUEUE_FALLBACK 1
37 #endif
38 #if HAVE_PTHREAD_WORKQUEUES && DISPATCH_USE_NOQOS_WORKQUEUE_FALLBACK && \
39 !HAVE_PTHREAD_WORKQUEUE_SETDISPATCH_NP && \
40 !defined(DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK)
41 #define DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK 1
42 #endif
43 #if HAVE_PTHREAD_WORKQUEUE_QOS && !DISPATCH_USE_NOQOS_WORKQUEUE_FALLBACK
44 #undef HAVE_PTHREAD_WORKQUEUE_SETDISPATCH_NP
45 #define HAVE_PTHREAD_WORKQUEUE_SETDISPATCH_NP 0
46 #endif
47 #if HAVE_PTHREAD_WORKQUEUES && DISPATCH_USE_PTHREAD_POOL && \
48 !DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK
49 #define pthread_workqueue_t void*
50 #endif
51
52 static void _dispatch_cache_cleanup(void *value);
53 static void _dispatch_async_f_redirect(dispatch_queue_t dq,
54 dispatch_continuation_t dc, pthread_priority_t pp);
55 static void _dispatch_queue_cleanup(void *ctxt);
56 static inline void _dispatch_queue_wakeup_global2(dispatch_queue_t dq,
57 unsigned int n);
58 static inline void _dispatch_queue_wakeup_global(dispatch_queue_t dq);
59 static inline _dispatch_thread_semaphore_t
60 _dispatch_queue_drain_one_barrier_sync(dispatch_queue_t dq);
61 static inline bool _dispatch_queue_prepare_override(dispatch_queue_t dq,
62 dispatch_queue_t tq, pthread_priority_t p);
63 static inline void _dispatch_queue_push_override(dispatch_queue_t dq,
64 dispatch_queue_t tq, pthread_priority_t p);
65 #if HAVE_PTHREAD_WORKQUEUES
66 static void _dispatch_worker_thread4(void *context);
67 #if HAVE_PTHREAD_WORKQUEUE_QOS
68 static void _dispatch_worker_thread3(pthread_priority_t priority);
69 #endif
70 #if HAVE_PTHREAD_WORKQUEUE_SETDISPATCH_NP
71 static void _dispatch_worker_thread2(int priority, int options, void *context);
72 #endif
73 #endif
74 #if DISPATCH_USE_PTHREAD_POOL
75 static void *_dispatch_worker_thread(void *context);
76 static int _dispatch_pthread_sigmask(int how, sigset_t *set, sigset_t *oset);
77 #endif
78
79 #if DISPATCH_COCOA_COMPAT
80 static dispatch_once_t _dispatch_main_q_port_pred;
81 static dispatch_queue_t _dispatch_main_queue_wakeup(void);
82 unsigned long _dispatch_runloop_queue_wakeup(dispatch_queue_t dq);
83 static void _dispatch_runloop_queue_port_init(void *ctxt);
84 static void _dispatch_runloop_queue_port_dispose(dispatch_queue_t dq);
85 #endif
86
87 static void _dispatch_root_queues_init(void *context);
88 static dispatch_once_t _dispatch_root_queues_pred;
89
90 #pragma mark -
91 #pragma mark dispatch_root_queue
92
93 struct dispatch_pthread_root_queue_context_s {
94 pthread_attr_t dpq_thread_attr;
95 dispatch_block_t dpq_thread_configure;
96 struct dispatch_semaphore_s dpq_thread_mediator;
97 };
98 typedef struct dispatch_pthread_root_queue_context_s *
99 dispatch_pthread_root_queue_context_t;
100
101 #if DISPATCH_ENABLE_THREAD_POOL
102 static struct dispatch_pthread_root_queue_context_s
103 _dispatch_pthread_root_queue_contexts[] = {
104 [DISPATCH_ROOT_QUEUE_IDX_MAINTENANCE_QOS] = {
105 .dpq_thread_mediator = {
106 .do_vtable = DISPATCH_VTABLE(semaphore),
107 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
108 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
109 }},
110 [DISPATCH_ROOT_QUEUE_IDX_MAINTENANCE_QOS_OVERCOMMIT] = {
111 .dpq_thread_mediator = {
112 .do_vtable = DISPATCH_VTABLE(semaphore),
113 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
114 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
115 }},
116 [DISPATCH_ROOT_QUEUE_IDX_BACKGROUND_QOS] = {
117 .dpq_thread_mediator = {
118 .do_vtable = DISPATCH_VTABLE(semaphore),
119 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
120 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
121 }},
122 [DISPATCH_ROOT_QUEUE_IDX_BACKGROUND_QOS_OVERCOMMIT] = {
123 .dpq_thread_mediator = {
124 .do_vtable = DISPATCH_VTABLE(semaphore),
125 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
126 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
127 }},
128 [DISPATCH_ROOT_QUEUE_IDX_UTILITY_QOS] = {
129 .dpq_thread_mediator = {
130 .do_vtable = DISPATCH_VTABLE(semaphore),
131 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
132 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
133 }},
134 [DISPATCH_ROOT_QUEUE_IDX_UTILITY_QOS_OVERCOMMIT] = {
135 .dpq_thread_mediator = {
136 .do_vtable = DISPATCH_VTABLE(semaphore),
137 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
138 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
139 }},
140 [DISPATCH_ROOT_QUEUE_IDX_DEFAULT_QOS] = {
141 .dpq_thread_mediator = {
142 .do_vtable = DISPATCH_VTABLE(semaphore),
143 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
144 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
145 }},
146 [DISPATCH_ROOT_QUEUE_IDX_DEFAULT_QOS_OVERCOMMIT] = {
147 .dpq_thread_mediator = {
148 .do_vtable = DISPATCH_VTABLE(semaphore),
149 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
150 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
151 }},
152 [DISPATCH_ROOT_QUEUE_IDX_USER_INITIATED_QOS] = {
153 .dpq_thread_mediator = {
154 .do_vtable = DISPATCH_VTABLE(semaphore),
155 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
156 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
157 }},
158 [DISPATCH_ROOT_QUEUE_IDX_USER_INITIATED_QOS_OVERCOMMIT] = {
159 .dpq_thread_mediator = {
160 .do_vtable = DISPATCH_VTABLE(semaphore),
161 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
162 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
163 }},
164 [DISPATCH_ROOT_QUEUE_IDX_USER_INTERACTIVE_QOS] = {
165 .dpq_thread_mediator = {
166 .do_vtable = DISPATCH_VTABLE(semaphore),
167 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
168 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
169 }},
170 [DISPATCH_ROOT_QUEUE_IDX_USER_INTERACTIVE_QOS_OVERCOMMIT] = {
171 .dpq_thread_mediator = {
172 .do_vtable = DISPATCH_VTABLE(semaphore),
173 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
174 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
175 }},
176 };
177 #endif
178
179 #define MAX_PTHREAD_COUNT 255
180
181 struct dispatch_root_queue_context_s {
182 union {
183 struct {
184 unsigned int volatile dgq_pending;
185 #if HAVE_PTHREAD_WORKQUEUES
186 qos_class_t dgq_qos;
187 int dgq_wq_priority, dgq_wq_options;
188 #if DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK || DISPATCH_USE_PTHREAD_POOL
189 pthread_workqueue_t dgq_kworkqueue;
190 #endif
191 #endif // HAVE_PTHREAD_WORKQUEUES
192 #if DISPATCH_USE_PTHREAD_POOL
193 void *dgq_ctxt;
194 uint32_t volatile dgq_thread_pool_size;
195 #endif
196 };
197 char _dgq_pad[DISPATCH_CACHELINE_SIZE];
198 };
199 };
200 typedef struct dispatch_root_queue_context_s *dispatch_root_queue_context_t;
201
202 DISPATCH_CACHELINE_ALIGN
203 static struct dispatch_root_queue_context_s _dispatch_root_queue_contexts[] = {
204 [DISPATCH_ROOT_QUEUE_IDX_MAINTENANCE_QOS] = {{{
205 #if HAVE_PTHREAD_WORKQUEUES
206 .dgq_qos = _DISPATCH_QOS_CLASS_MAINTENANCE,
207 .dgq_wq_priority = WORKQ_BG_PRIOQUEUE,
208 .dgq_wq_options = 0,
209 #endif
210 #if DISPATCH_ENABLE_THREAD_POOL
211 .dgq_ctxt = &_dispatch_pthread_root_queue_contexts[
212 DISPATCH_ROOT_QUEUE_IDX_MAINTENANCE_QOS],
213 #endif
214 }}},
215 [DISPATCH_ROOT_QUEUE_IDX_MAINTENANCE_QOS_OVERCOMMIT] = {{{
216 #if HAVE_PTHREAD_WORKQUEUES
217 .dgq_qos = _DISPATCH_QOS_CLASS_MAINTENANCE,
218 .dgq_wq_priority = WORKQ_BG_PRIOQUEUE,
219 .dgq_wq_options = WORKQ_ADDTHREADS_OPTION_OVERCOMMIT,
220 #endif
221 #if DISPATCH_ENABLE_THREAD_POOL
222 .dgq_ctxt = &_dispatch_pthread_root_queue_contexts[
223 DISPATCH_ROOT_QUEUE_IDX_MAINTENANCE_QOS_OVERCOMMIT],
224 #endif
225 }}},
226 [DISPATCH_ROOT_QUEUE_IDX_BACKGROUND_QOS] = {{{
227 #if HAVE_PTHREAD_WORKQUEUES
228 .dgq_qos = _DISPATCH_QOS_CLASS_BACKGROUND,
229 .dgq_wq_priority = WORKQ_BG_PRIOQUEUE,
230 .dgq_wq_options = 0,
231 #endif
232 #if DISPATCH_ENABLE_THREAD_POOL
233 .dgq_ctxt = &_dispatch_pthread_root_queue_contexts[
234 DISPATCH_ROOT_QUEUE_IDX_BACKGROUND_QOS],
235 #endif
236 }}},
237 [DISPATCH_ROOT_QUEUE_IDX_BACKGROUND_QOS_OVERCOMMIT] = {{{
238 #if HAVE_PTHREAD_WORKQUEUES
239 .dgq_qos = _DISPATCH_QOS_CLASS_BACKGROUND,
240 .dgq_wq_priority = WORKQ_BG_PRIOQUEUE,
241 .dgq_wq_options = WORKQ_ADDTHREADS_OPTION_OVERCOMMIT,
242 #endif
243 #if DISPATCH_ENABLE_THREAD_POOL
244 .dgq_ctxt = &_dispatch_pthread_root_queue_contexts[
245 DISPATCH_ROOT_QUEUE_IDX_BACKGROUND_QOS_OVERCOMMIT],
246 #endif
247 }}},
248 [DISPATCH_ROOT_QUEUE_IDX_UTILITY_QOS] = {{{
249 #if HAVE_PTHREAD_WORKQUEUES
250 .dgq_qos = _DISPATCH_QOS_CLASS_UTILITY,
251 .dgq_wq_priority = WORKQ_LOW_PRIOQUEUE,
252 .dgq_wq_options = 0,
253 #endif
254 #if DISPATCH_ENABLE_THREAD_POOL
255 .dgq_ctxt = &_dispatch_pthread_root_queue_contexts[
256 DISPATCH_ROOT_QUEUE_IDX_UTILITY_QOS],
257 #endif
258 }}},
259 [DISPATCH_ROOT_QUEUE_IDX_UTILITY_QOS_OVERCOMMIT] = {{{
260 #if HAVE_PTHREAD_WORKQUEUES
261 .dgq_qos = _DISPATCH_QOS_CLASS_UTILITY,
262 .dgq_wq_priority = WORKQ_LOW_PRIOQUEUE,
263 .dgq_wq_options = WORKQ_ADDTHREADS_OPTION_OVERCOMMIT,
264 #endif
265 #if DISPATCH_ENABLE_THREAD_POOL
266 .dgq_ctxt = &_dispatch_pthread_root_queue_contexts[
267 DISPATCH_ROOT_QUEUE_IDX_UTILITY_QOS_OVERCOMMIT],
268 #endif
269 }}},
270 [DISPATCH_ROOT_QUEUE_IDX_DEFAULT_QOS] = {{{
271 #if HAVE_PTHREAD_WORKQUEUES
272 .dgq_qos = _DISPATCH_QOS_CLASS_DEFAULT,
273 .dgq_wq_priority = WORKQ_DEFAULT_PRIOQUEUE,
274 .dgq_wq_options = 0,
275 #endif
276 #if DISPATCH_ENABLE_THREAD_POOL
277 .dgq_ctxt = &_dispatch_pthread_root_queue_contexts[
278 DISPATCH_ROOT_QUEUE_IDX_DEFAULT_QOS],
279 #endif
280 }}},
281 [DISPATCH_ROOT_QUEUE_IDX_DEFAULT_QOS_OVERCOMMIT] = {{{
282 #if HAVE_PTHREAD_WORKQUEUES
283 .dgq_qos = _DISPATCH_QOS_CLASS_DEFAULT,
284 .dgq_wq_priority = WORKQ_DEFAULT_PRIOQUEUE,
285 .dgq_wq_options = WORKQ_ADDTHREADS_OPTION_OVERCOMMIT,
286 #endif
287 #if DISPATCH_ENABLE_THREAD_POOL
288 .dgq_ctxt = &_dispatch_pthread_root_queue_contexts[
289 DISPATCH_ROOT_QUEUE_IDX_DEFAULT_QOS_OVERCOMMIT],
290 #endif
291 }}},
292 [DISPATCH_ROOT_QUEUE_IDX_USER_INITIATED_QOS] = {{{
293 #if HAVE_PTHREAD_WORKQUEUES
294 .dgq_qos = _DISPATCH_QOS_CLASS_USER_INITIATED,
295 .dgq_wq_priority = WORKQ_HIGH_PRIOQUEUE,
296 .dgq_wq_options = 0,
297 #endif
298 #if DISPATCH_ENABLE_THREAD_POOL
299 .dgq_ctxt = &_dispatch_pthread_root_queue_contexts[
300 DISPATCH_ROOT_QUEUE_IDX_USER_INITIATED_QOS],
301 #endif
302 }}},
303 [DISPATCH_ROOT_QUEUE_IDX_USER_INITIATED_QOS_OVERCOMMIT] = {{{
304 #if HAVE_PTHREAD_WORKQUEUES
305 .dgq_qos = _DISPATCH_QOS_CLASS_USER_INITIATED,
306 .dgq_wq_priority = WORKQ_HIGH_PRIOQUEUE,
307 .dgq_wq_options = WORKQ_ADDTHREADS_OPTION_OVERCOMMIT,
308 #endif
309 #if DISPATCH_ENABLE_THREAD_POOL
310 .dgq_ctxt = &_dispatch_pthread_root_queue_contexts[
311 DISPATCH_ROOT_QUEUE_IDX_USER_INITIATED_QOS_OVERCOMMIT],
312 #endif
313 }}},
314 [DISPATCH_ROOT_QUEUE_IDX_USER_INTERACTIVE_QOS] = {{{
315 #if HAVE_PTHREAD_WORKQUEUES
316 .dgq_qos = _DISPATCH_QOS_CLASS_USER_INTERACTIVE,
317 .dgq_wq_priority = WORKQ_HIGH_PRIOQUEUE,
318 .dgq_wq_options = 0,
319 #endif
320 #if DISPATCH_ENABLE_THREAD_POOL
321 .dgq_ctxt = &_dispatch_pthread_root_queue_contexts[
322 DISPATCH_ROOT_QUEUE_IDX_USER_INTERACTIVE_QOS],
323 #endif
324 }}},
325 [DISPATCH_ROOT_QUEUE_IDX_USER_INTERACTIVE_QOS_OVERCOMMIT] = {{{
326 #if HAVE_PTHREAD_WORKQUEUES
327 .dgq_qos = _DISPATCH_QOS_CLASS_USER_INTERACTIVE,
328 .dgq_wq_priority = WORKQ_HIGH_PRIOQUEUE,
329 .dgq_wq_options = WORKQ_ADDTHREADS_OPTION_OVERCOMMIT,
330 #endif
331 #if DISPATCH_ENABLE_THREAD_POOL
332 .dgq_ctxt = &_dispatch_pthread_root_queue_contexts[
333 DISPATCH_ROOT_QUEUE_IDX_USER_INTERACTIVE_QOS_OVERCOMMIT],
334 #endif
335 }}},
336 };
337
338 // 6618342 Contact the team that owns the Instrument DTrace probe before
339 // renaming this symbol
340 // dq_running is set to 2 so that barrier operations go through the slow path
341 DISPATCH_CACHELINE_ALIGN
342 struct dispatch_queue_s _dispatch_root_queues[] = {
343 [DISPATCH_ROOT_QUEUE_IDX_MAINTENANCE_QOS] = {
344 .do_vtable = DISPATCH_VTABLE(queue_root),
345 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
346 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
347 .do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK,
348 .do_ctxt = &_dispatch_root_queue_contexts[
349 DISPATCH_ROOT_QUEUE_IDX_MAINTENANCE_QOS],
350 .dq_label = "com.apple.root.maintenance-qos",
351 .dq_running = 2,
352 .dq_width = DISPATCH_QUEUE_WIDTH_MAX,
353 .dq_serialnum = 4,
354 },
355 [DISPATCH_ROOT_QUEUE_IDX_MAINTENANCE_QOS_OVERCOMMIT] = {
356 .do_vtable = DISPATCH_VTABLE(queue_root),
357 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
358 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
359 .do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK,
360 .do_ctxt = &_dispatch_root_queue_contexts[
361 DISPATCH_ROOT_QUEUE_IDX_MAINTENANCE_QOS_OVERCOMMIT],
362 .dq_label = "com.apple.root.maintenance-qos.overcommit",
363 .dq_running = 2,
364 .dq_width = DISPATCH_QUEUE_WIDTH_MAX,
365 .dq_serialnum = 5,
366 },
367 [DISPATCH_ROOT_QUEUE_IDX_BACKGROUND_QOS] = {
368 .do_vtable = DISPATCH_VTABLE(queue_root),
369 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
370 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
371 .do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK,
372 .do_ctxt = &_dispatch_root_queue_contexts[
373 DISPATCH_ROOT_QUEUE_IDX_BACKGROUND_QOS],
374 .dq_label = "com.apple.root.background-qos",
375 .dq_running = 2,
376 .dq_width = DISPATCH_QUEUE_WIDTH_MAX,
377 .dq_serialnum = 6,
378 },
379 [DISPATCH_ROOT_QUEUE_IDX_BACKGROUND_QOS_OVERCOMMIT] = {
380 .do_vtable = DISPATCH_VTABLE(queue_root),
381 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
382 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
383 .do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK,
384 .do_ctxt = &_dispatch_root_queue_contexts[
385 DISPATCH_ROOT_QUEUE_IDX_BACKGROUND_QOS_OVERCOMMIT],
386 .dq_label = "com.apple.root.background-qos.overcommit",
387 .dq_running = 2,
388 .dq_width = DISPATCH_QUEUE_WIDTH_MAX,
389 .dq_serialnum = 7,
390 },
391 [DISPATCH_ROOT_QUEUE_IDX_UTILITY_QOS] = {
392 .do_vtable = DISPATCH_VTABLE(queue_root),
393 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
394 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
395 .do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK,
396 .do_ctxt = &_dispatch_root_queue_contexts[
397 DISPATCH_ROOT_QUEUE_IDX_UTILITY_QOS],
398 .dq_label = "com.apple.root.utility-qos",
399 .dq_running = 2,
400 .dq_width = DISPATCH_QUEUE_WIDTH_MAX,
401 .dq_serialnum = 8,
402 },
403 [DISPATCH_ROOT_QUEUE_IDX_UTILITY_QOS_OVERCOMMIT] = {
404 .do_vtable = DISPATCH_VTABLE(queue_root),
405 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
406 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
407 .do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK,
408 .do_ctxt = &_dispatch_root_queue_contexts[
409 DISPATCH_ROOT_QUEUE_IDX_UTILITY_QOS_OVERCOMMIT],
410 .dq_label = "com.apple.root.utility-qos.overcommit",
411 .dq_running = 2,
412 .dq_width = DISPATCH_QUEUE_WIDTH_MAX,
413 .dq_serialnum = 9,
414 },
415 [DISPATCH_ROOT_QUEUE_IDX_DEFAULT_QOS] = {
416 .do_vtable = DISPATCH_VTABLE(queue_root),
417 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
418 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
419 .do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK,
420 .do_ctxt = &_dispatch_root_queue_contexts[
421 DISPATCH_ROOT_QUEUE_IDX_DEFAULT_QOS],
422 .dq_label = "com.apple.root.default-qos",
423 .dq_running = 2,
424 .dq_width = DISPATCH_QUEUE_WIDTH_MAX,
425 .dq_serialnum = 10,
426 },
427 [DISPATCH_ROOT_QUEUE_IDX_DEFAULT_QOS_OVERCOMMIT] = {
428 .do_vtable = DISPATCH_VTABLE(queue_root),
429 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
430 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
431 .do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK,
432 .do_ctxt = &_dispatch_root_queue_contexts[
433 DISPATCH_ROOT_QUEUE_IDX_DEFAULT_QOS_OVERCOMMIT],
434 .dq_label = "com.apple.root.default-qos.overcommit",
435 .dq_running = 2,
436 .dq_width = DISPATCH_QUEUE_WIDTH_MAX,
437 .dq_serialnum = 11,
438 },
439 [DISPATCH_ROOT_QUEUE_IDX_USER_INITIATED_QOS] = {
440 .do_vtable = DISPATCH_VTABLE(queue_root),
441 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
442 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
443 .do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK,
444 .do_ctxt = &_dispatch_root_queue_contexts[
445 DISPATCH_ROOT_QUEUE_IDX_USER_INITIATED_QOS],
446 .dq_label = "com.apple.root.user-initiated-qos",
447 .dq_running = 2,
448 .dq_width = DISPATCH_QUEUE_WIDTH_MAX,
449 .dq_serialnum = 12,
450 },
451 [DISPATCH_ROOT_QUEUE_IDX_USER_INITIATED_QOS_OVERCOMMIT] = {
452 .do_vtable = DISPATCH_VTABLE(queue_root),
453 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
454 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
455 .do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK,
456 .do_ctxt = &_dispatch_root_queue_contexts[
457 DISPATCH_ROOT_QUEUE_IDX_USER_INITIATED_QOS_OVERCOMMIT],
458 .dq_label = "com.apple.root.user-initiated-qos.overcommit",
459 .dq_running = 2,
460 .dq_width = DISPATCH_QUEUE_WIDTH_MAX,
461 .dq_serialnum = 13,
462 },
463 [DISPATCH_ROOT_QUEUE_IDX_USER_INTERACTIVE_QOS] = {
464 .do_vtable = DISPATCH_VTABLE(queue_root),
465 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
466 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
467 .do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK,
468 .do_ctxt = &_dispatch_root_queue_contexts[
469 DISPATCH_ROOT_QUEUE_IDX_USER_INTERACTIVE_QOS],
470 .dq_label = "com.apple.root.user-interactive-qos",
471 .dq_running = 2,
472 .dq_width = DISPATCH_QUEUE_WIDTH_MAX,
473 .dq_serialnum = 14,
474 },
475 [DISPATCH_ROOT_QUEUE_IDX_USER_INTERACTIVE_QOS_OVERCOMMIT] = {
476 .do_vtable = DISPATCH_VTABLE(queue_root),
477 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
478 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
479 .do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK,
480 .do_ctxt = &_dispatch_root_queue_contexts[
481 DISPATCH_ROOT_QUEUE_IDX_USER_INTERACTIVE_QOS_OVERCOMMIT],
482 .dq_label = "com.apple.root.user-interactive-qos.overcommit",
483 .dq_running = 2,
484 .dq_width = DISPATCH_QUEUE_WIDTH_MAX,
485 .dq_serialnum = 15,
486 },
487 };
488
489 #if HAVE_PTHREAD_WORKQUEUE_SETDISPATCH_NP
490 static const dispatch_queue_t _dispatch_wq2root_queues[][2] = {
491 [WORKQ_BG_PRIOQUEUE][0] = &_dispatch_root_queues[
492 DISPATCH_ROOT_QUEUE_IDX_BACKGROUND_QOS],
493 [WORKQ_BG_PRIOQUEUE][WORKQ_ADDTHREADS_OPTION_OVERCOMMIT] =
494 &_dispatch_root_queues[
495 DISPATCH_ROOT_QUEUE_IDX_BACKGROUND_QOS_OVERCOMMIT],
496 [WORKQ_LOW_PRIOQUEUE][0] = &_dispatch_root_queues[
497 DISPATCH_ROOT_QUEUE_IDX_UTILITY_QOS],
498 [WORKQ_LOW_PRIOQUEUE][WORKQ_ADDTHREADS_OPTION_OVERCOMMIT] =
499 &_dispatch_root_queues[
500 DISPATCH_ROOT_QUEUE_IDX_UTILITY_QOS_OVERCOMMIT],
501 [WORKQ_DEFAULT_PRIOQUEUE][0] = &_dispatch_root_queues[
502 DISPATCH_ROOT_QUEUE_IDX_DEFAULT_QOS],
503 [WORKQ_DEFAULT_PRIOQUEUE][WORKQ_ADDTHREADS_OPTION_OVERCOMMIT] =
504 &_dispatch_root_queues[
505 DISPATCH_ROOT_QUEUE_IDX_DEFAULT_QOS_OVERCOMMIT],
506 [WORKQ_HIGH_PRIOQUEUE][0] = &_dispatch_root_queues[
507 DISPATCH_ROOT_QUEUE_IDX_USER_INITIATED_QOS],
508 [WORKQ_HIGH_PRIOQUEUE][WORKQ_ADDTHREADS_OPTION_OVERCOMMIT] =
509 &_dispatch_root_queues[
510 DISPATCH_ROOT_QUEUE_IDX_USER_INITIATED_QOS_OVERCOMMIT],
511 };
512 #endif // HAVE_PTHREAD_WORKQUEUE_SETDISPATCH_NP
513
514 #define DISPATCH_PRIORITY_COUNT 5
515
516 enum {
517 // No DISPATCH_PRIORITY_IDX_MAINTENANCE define because there is no legacy
518 // maintenance priority
519 DISPATCH_PRIORITY_IDX_BACKGROUND = 0,
520 DISPATCH_PRIORITY_IDX_NON_INTERACTIVE,
521 DISPATCH_PRIORITY_IDX_LOW,
522 DISPATCH_PRIORITY_IDX_DEFAULT,
523 DISPATCH_PRIORITY_IDX_HIGH,
524 };
525
526 static qos_class_t _dispatch_priority2qos[] = {
527 [DISPATCH_PRIORITY_IDX_BACKGROUND] = _DISPATCH_QOS_CLASS_BACKGROUND,
528 [DISPATCH_PRIORITY_IDX_NON_INTERACTIVE] = _DISPATCH_QOS_CLASS_UTILITY,
529 [DISPATCH_PRIORITY_IDX_LOW] = _DISPATCH_QOS_CLASS_UTILITY,
530 [DISPATCH_PRIORITY_IDX_DEFAULT] = _DISPATCH_QOS_CLASS_DEFAULT,
531 [DISPATCH_PRIORITY_IDX_HIGH] = _DISPATCH_QOS_CLASS_USER_INITIATED,
532 };
533
534 #if HAVE_PTHREAD_WORKQUEUE_QOS
535 static const int _dispatch_priority2wq[] = {
536 [DISPATCH_PRIORITY_IDX_BACKGROUND] = WORKQ_BG_PRIOQUEUE,
537 [DISPATCH_PRIORITY_IDX_NON_INTERACTIVE] = WORKQ_NON_INTERACTIVE_PRIOQUEUE,
538 [DISPATCH_PRIORITY_IDX_LOW] = WORKQ_LOW_PRIOQUEUE,
539 [DISPATCH_PRIORITY_IDX_DEFAULT] = WORKQ_DEFAULT_PRIOQUEUE,
540 [DISPATCH_PRIORITY_IDX_HIGH] = WORKQ_HIGH_PRIOQUEUE,
541 };
542 #endif
543
544 #if DISPATCH_ENABLE_PTHREAD_ROOT_QUEUES
545 static struct dispatch_queue_s _dispatch_mgr_root_queue;
546 #else
547 #define _dispatch_mgr_root_queue \
548 _dispatch_root_queues[DISPATCH_ROOT_QUEUE_IDX_HIGH_OVERCOMMIT_PRIORITY]
549 #endif
550
551 // 6618342 Contact the team that owns the Instrument DTrace probe before
552 // renaming this symbol
553 DISPATCH_CACHELINE_ALIGN
554 struct dispatch_queue_s _dispatch_mgr_q = {
555 .do_vtable = DISPATCH_VTABLE(queue_mgr),
556 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
557 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
558 .do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK,
559 .do_targetq = &_dispatch_mgr_root_queue,
560 .dq_label = "com.apple.libdispatch-manager",
561 .dq_width = 1,
562 .dq_is_thread_bound = 1,
563 .dq_serialnum = 2,
564 };
565
566 dispatch_queue_t
dispatch_get_global_queue(long priority,unsigned long flags)567 dispatch_get_global_queue(long priority, unsigned long flags)
568 {
569 if (flags & ~(unsigned long)DISPATCH_QUEUE_OVERCOMMIT) {
570 return NULL;
571 }
572 dispatch_once_f(&_dispatch_root_queues_pred, NULL,
573 _dispatch_root_queues_init);
574 qos_class_t qos;
575 switch (priority) {
576 #if !RDAR_17878963 || DISPATCH_USE_NOQOS_WORKQUEUE_FALLBACK
577 case _DISPATCH_QOS_CLASS_MAINTENANCE:
578 if (!_dispatch_root_queues[DISPATCH_ROOT_QUEUE_IDX_MAINTENANCE_QOS]
579 .dq_priority) {
580 // map maintenance to background on old kernel
581 qos = _dispatch_priority2qos[DISPATCH_PRIORITY_IDX_BACKGROUND];
582 } else {
583 qos = (qos_class_t)priority;
584 }
585 break;
586 #endif // RDAR_17878963 || DISPATCH_USE_NOQOS_WORKQUEUE_FALLBACK
587 case DISPATCH_QUEUE_PRIORITY_BACKGROUND:
588 qos = _dispatch_priority2qos[DISPATCH_PRIORITY_IDX_BACKGROUND];
589 break;
590 case DISPATCH_QUEUE_PRIORITY_NON_INTERACTIVE:
591 qos = _dispatch_priority2qos[DISPATCH_PRIORITY_IDX_NON_INTERACTIVE];
592 break;
593 case DISPATCH_QUEUE_PRIORITY_LOW:
594 qos = _dispatch_priority2qos[DISPATCH_PRIORITY_IDX_LOW];
595 break;
596 case DISPATCH_QUEUE_PRIORITY_DEFAULT:
597 qos = _dispatch_priority2qos[DISPATCH_PRIORITY_IDX_DEFAULT];
598 break;
599 case DISPATCH_QUEUE_PRIORITY_HIGH:
600 qos = _dispatch_priority2qos[DISPATCH_PRIORITY_IDX_HIGH];
601 break;
602 case _DISPATCH_QOS_CLASS_USER_INTERACTIVE:
603 #if DISPATCH_USE_NOQOS_WORKQUEUE_FALLBACK
604 if (!_dispatch_root_queues[DISPATCH_ROOT_QUEUE_IDX_USER_INTERACTIVE_QOS]
605 .dq_priority) {
606 qos = _dispatch_priority2qos[DISPATCH_PRIORITY_IDX_HIGH];
607 break;
608 }
609 #endif
610 // fallthrough
611 default:
612 qos = (qos_class_t)priority;
613 break;
614 }
615 return _dispatch_get_root_queue(qos, flags & DISPATCH_QUEUE_OVERCOMMIT);
616 }
617
618 DISPATCH_ALWAYS_INLINE
619 static inline dispatch_queue_t
_dispatch_get_current_queue(void)620 _dispatch_get_current_queue(void)
621 {
622 return _dispatch_queue_get_current() ?:
623 _dispatch_get_root_queue(_DISPATCH_QOS_CLASS_DEFAULT, true);
624 }
625
626 dispatch_queue_t
dispatch_get_current_queue(void)627 dispatch_get_current_queue(void)
628 {
629 return _dispatch_get_current_queue();
630 }
631
632 DISPATCH_ALWAYS_INLINE
633 static inline bool
_dispatch_queue_targets_queue(dispatch_queue_t dq1,dispatch_queue_t dq2)634 _dispatch_queue_targets_queue(dispatch_queue_t dq1, dispatch_queue_t dq2)
635 {
636 while (dq1) {
637 if (dq1 == dq2) {
638 return true;
639 }
640 dq1 = dq1->do_targetq;
641 }
642 return false;
643 }
644
645 #define DISPATCH_ASSERT_QUEUE_MESSAGE "BUG in client of libdispatch: " \
646 "Assertion failed: Block was run on an unexpected queue"
647
648 DISPATCH_NOINLINE
649 static void
_dispatch_assert_queue_fail(dispatch_queue_t dq,bool expected)650 _dispatch_assert_queue_fail(dispatch_queue_t dq, bool expected)
651 {
652 char *msg;
653 asprintf(&msg, "%s\n%s queue: 0x%p[%s]", DISPATCH_ASSERT_QUEUE_MESSAGE,
654 expected ? "Expected" : "Unexpected", dq, dq->dq_label ?
655 dq->dq_label : "");
656 _dispatch_log("%s", msg);
657 _dispatch_set_crash_log_message(msg);
658 _dispatch_hardware_crash();
659 free(msg);
660 }
661
662 void
dispatch_assert_queue(dispatch_queue_t dq)663 dispatch_assert_queue(dispatch_queue_t dq)
664 {
665 if (slowpath(!dq) || slowpath(!(dx_metatype(dq) == _DISPATCH_QUEUE_TYPE))) {
666 DISPATCH_CLIENT_CRASH("invalid queue passed to "
667 "dispatch_assert_queue()");
668 }
669 dispatch_queue_t cq = _dispatch_queue_get_current();
670 if (fastpath(cq) && fastpath(_dispatch_queue_targets_queue(cq, dq))) {
671 return;
672 }
673 _dispatch_assert_queue_fail(dq, true);
674 }
675
676 void
dispatch_assert_queue_not(dispatch_queue_t dq)677 dispatch_assert_queue_not(dispatch_queue_t dq)
678 {
679 if (slowpath(!dq) || slowpath(!(dx_metatype(dq) == _DISPATCH_QUEUE_TYPE))) {
680 DISPATCH_CLIENT_CRASH("invalid queue passed to "
681 "dispatch_assert_queue_not()");
682 }
683 dispatch_queue_t cq = _dispatch_queue_get_current();
684 if (slowpath(cq) && slowpath(_dispatch_queue_targets_queue(cq, dq))) {
685 _dispatch_assert_queue_fail(dq, false);
686 }
687 }
688
689 #if DISPATCH_DEBUG && DISPATCH_ROOT_QUEUE_DEBUG
690 #define _dispatch_root_queue_debug(...) _dispatch_debug(__VA_ARGS__)
691 #define _dispatch_debug_root_queue(...) dispatch_debug_queue(__VA_ARGS__)
692 #else
693 #define _dispatch_root_queue_debug(...)
694 #define _dispatch_debug_root_queue(...)
695 #endif
696
697 #pragma mark -
698 #pragma mark dispatch_init
699
700 #if HAVE_PTHREAD_WORKQUEUE_QOS
701 int _dispatch_set_qos_class_enabled;
702 pthread_priority_t _dispatch_background_priority;
703 pthread_priority_t _dispatch_user_initiated_priority;
704
705 static void
_dispatch_root_queues_init_qos(int supported)706 _dispatch_root_queues_init_qos(int supported)
707 {
708 pthread_priority_t p;
709 qos_class_t qos;
710 unsigned int i;
711 for (i = 0; i < DISPATCH_PRIORITY_COUNT; i++) {
712 p = _pthread_qos_class_encode_workqueue(_dispatch_priority2wq[i], 0);
713 qos = _pthread_qos_class_decode(p, NULL, NULL);
714 dispatch_assert(qos != _DISPATCH_QOS_CLASS_UNSPECIFIED);
715 _dispatch_priority2qos[i] = qos;
716 }
717 for (i = 0; i < DISPATCH_ROOT_QUEUE_COUNT; i++) {
718 qos = _dispatch_root_queue_contexts[i].dgq_qos;
719 if (qos == _DISPATCH_QOS_CLASS_MAINTENANCE &&
720 !(supported & WORKQ_FEATURE_MAINTENANCE)) {
721 continue;
722 }
723 unsigned long flags = i & 1 ? _PTHREAD_PRIORITY_OVERCOMMIT_FLAG : 0;
724 flags |= _PTHREAD_PRIORITY_ROOTQUEUE_FLAG;
725 if (i == DISPATCH_ROOT_QUEUE_IDX_DEFAULT_QOS ||
726 i == DISPATCH_ROOT_QUEUE_IDX_DEFAULT_QOS_OVERCOMMIT) {
727 flags |= _PTHREAD_PRIORITY_DEFAULTQUEUE_FLAG;
728 }
729 p = _pthread_qos_class_encode(qos, 0, flags);
730 _dispatch_root_queues[i].dq_priority = p;
731 }
732 p = _pthread_qos_class_encode(qos_class_main(), 0, 0);
733 _dispatch_main_q.dq_priority = p;
734 _dispatch_queue_set_override_priority(&_dispatch_main_q);
735 _dispatch_background_priority = _dispatch_root_queues[
736 DISPATCH_ROOT_QUEUE_IDX_BACKGROUND_QOS].dq_priority &
737 ~_PTHREAD_PRIORITY_FLAGS_MASK;
738 _dispatch_user_initiated_priority = _dispatch_root_queues[
739 DISPATCH_ROOT_QUEUE_IDX_USER_INITIATED_QOS].dq_priority &
740 ~_PTHREAD_PRIORITY_FLAGS_MASK;
741 if (!slowpath(getenv("LIBDISPATCH_DISABLE_SET_QOS"))) {
742 _dispatch_set_qos_class_enabled = 1;
743 }
744 }
745 #endif
746
747 static inline bool
_dispatch_root_queues_init_workq(void)748 _dispatch_root_queues_init_workq(void)
749 {
750 bool result = false;
751 #if HAVE_PTHREAD_WORKQUEUES
752 bool disable_wq = false;
753 #if DISPATCH_ENABLE_THREAD_POOL && DISPATCH_DEBUG
754 disable_wq = slowpath(getenv("LIBDISPATCH_DISABLE_KWQ"));
755 #endif
756 int r;
757 #if HAVE_PTHREAD_WORKQUEUE_QOS
758 bool disable_qos = false;
759 #if DISPATCH_DEBUG
760 disable_qos = slowpath(getenv("LIBDISPATCH_DISABLE_QOS"));
761 #endif
762 if (!disable_qos && !disable_wq) {
763 r = _pthread_workqueue_supported();
764 int supported = r;
765 if (r & WORKQ_FEATURE_FINEPRIO) {
766 r = _pthread_workqueue_init(_dispatch_worker_thread3,
767 offsetof(struct dispatch_queue_s, dq_serialnum), 0);
768 result = !r;
769 if (result) _dispatch_root_queues_init_qos(supported);
770 }
771 }
772 #endif // HAVE_PTHREAD_WORKQUEUE_QOS
773 #if HAVE_PTHREAD_WORKQUEUE_SETDISPATCH_NP
774 if (!result && !disable_wq) {
775 #if PTHREAD_WORKQUEUE_SPI_VERSION >= 20121218
776 pthread_workqueue_setdispatchoffset_np(
777 offsetof(struct dispatch_queue_s, dq_serialnum));
778 #endif
779 r = pthread_workqueue_setdispatch_np(_dispatch_worker_thread2);
780 #if !DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK
781 (void)dispatch_assume_zero(r);
782 #endif
783 result = !r;
784 }
785 #endif // HAVE_PTHREAD_WORKQUEUE_SETDISPATCH_NP
786 #if DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK || DISPATCH_USE_PTHREAD_POOL
787 if (!result) {
788 #if DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK
789 pthread_workqueue_attr_t pwq_attr;
790 if (!disable_wq) {
791 r = pthread_workqueue_attr_init_np(&pwq_attr);
792 (void)dispatch_assume_zero(r);
793 }
794 #endif
795 int i;
796 for (i = 0; i < DISPATCH_ROOT_QUEUE_COUNT; i++) {
797 pthread_workqueue_t pwq = NULL;
798 dispatch_root_queue_context_t qc;
799 qc = &_dispatch_root_queue_contexts[i];
800 #if DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK
801 if (!disable_wq) {
802 r = pthread_workqueue_attr_setqueuepriority_np(&pwq_attr,
803 qc->dgq_wq_priority);
804 (void)dispatch_assume_zero(r);
805 r = pthread_workqueue_attr_setovercommit_np(&pwq_attr,
806 qc->dgq_wq_options &
807 WORKQ_ADDTHREADS_OPTION_OVERCOMMIT);
808 (void)dispatch_assume_zero(r);
809 r = pthread_workqueue_create_np(&pwq, &pwq_attr);
810 (void)dispatch_assume_zero(r);
811 result = result || dispatch_assume(pwq);
812 }
813 #endif // DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK
814 qc->dgq_kworkqueue = pwq ? pwq : (void*)(~0ul);
815 }
816 #if DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK
817 if (!disable_wq) {
818 r = pthread_workqueue_attr_destroy_np(&pwq_attr);
819 (void)dispatch_assume_zero(r);
820 }
821 #endif
822 }
823 #endif // DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK || DISPATCH_ENABLE_THREAD_POOL
824 #endif // HAVE_PTHREAD_WORKQUEUES
825 return result;
826 }
827
828 #if DISPATCH_USE_PTHREAD_POOL
829 static inline void
_dispatch_root_queue_init_pthread_pool(dispatch_root_queue_context_t qc,uint8_t pool_size,bool overcommit)830 _dispatch_root_queue_init_pthread_pool(dispatch_root_queue_context_t qc,
831 uint8_t pool_size, bool overcommit)
832 {
833 dispatch_pthread_root_queue_context_t pqc = qc->dgq_ctxt;
834 uint32_t thread_pool_size = overcommit ? MAX_PTHREAD_COUNT :
835 dispatch_hw_config(active_cpus);
836 if (slowpath(pool_size) && pool_size < thread_pool_size) {
837 thread_pool_size = pool_size;
838 }
839 qc->dgq_thread_pool_size = thread_pool_size;
840 #ifndef __FreeBSD__
841 if (qc->dgq_qos) {
842 (void)dispatch_assume_zero(pthread_attr_init(&pqc->dpq_thread_attr));
843 (void)dispatch_assume_zero(pthread_attr_setdetachstate(
844 &pqc->dpq_thread_attr, PTHREAD_CREATE_DETACHED));
845 #if HAVE_PTHREAD_WORKQUEUE_QOS
846 (void)dispatch_assume_zero(pthread_attr_set_qos_class_np(
847 &pqc->dpq_thread_attr, qc->dgq_qos, 0));
848 #endif
849 }
850 #endif
851 #if USE_MACH_SEM
852 // override the default FIFO behavior for the pool semaphores
853 kern_return_t kr = semaphore_create(mach_task_self(),
854 &pqc->dpq_thread_mediator.dsema_port, SYNC_POLICY_LIFO, 0);
855 DISPATCH_VERIFY_MIG(kr);
856 (void)dispatch_assume_zero(kr);
857 (void)dispatch_assume(pqc->dpq_thread_mediator.dsema_port);
858 #elif USE_POSIX_SEM
859 /* XXXRW: POSIX semaphores don't support LIFO? */
860 int ret = sem_init(&pqc->dpq_thread_mediator.dsema_sem, 0, 0);
861 (void)dispatch_assume_zero(ret);
862 #endif
863 }
864 #endif // DISPATCH_USE_PTHREAD_POOL
865
866 static dispatch_once_t _dispatch_root_queues_pred;
867
868 static void
_dispatch_root_queues_init(void * context DISPATCH_UNUSED)869 _dispatch_root_queues_init(void *context DISPATCH_UNUSED)
870 {
871 _dispatch_safe_fork = false;
872 if (!_dispatch_root_queues_init_workq()) {
873 #if DISPATCH_ENABLE_THREAD_POOL
874 int i;
875 for (i = 0; i < DISPATCH_ROOT_QUEUE_COUNT; i++) {
876 bool overcommit = true;
877 #if TARGET_OS_EMBEDDED
878 // some software hangs if the non-overcommitting queues do not
879 // overcommit when threads block. Someday, this behavior should
880 // apply to all platforms
881 if (!(i & 1)) {
882 overcommit = false;
883 }
884 #endif
885 _dispatch_root_queue_init_pthread_pool(
886 &_dispatch_root_queue_contexts[i], 0, overcommit);
887 }
888 #else
889 DISPATCH_CRASH("Root queue initialization failed");
890 #endif // DISPATCH_ENABLE_THREAD_POOL
891 }
892 }
893
894 #define countof(x) (sizeof(x) / sizeof(x[0]))
895
896 DISPATCH_EXPORT DISPATCH_NOTHROW
897 void
libdispatch_init(void)898 libdispatch_init(void)
899 {
900 dispatch_assert(DISPATCH_QUEUE_QOS_COUNT == 6);
901 dispatch_assert(DISPATCH_ROOT_QUEUE_COUNT == 12);
902
903 dispatch_assert(DISPATCH_QUEUE_PRIORITY_LOW ==
904 -DISPATCH_QUEUE_PRIORITY_HIGH);
905 dispatch_assert(countof(_dispatch_root_queues) ==
906 DISPATCH_ROOT_QUEUE_COUNT);
907 dispatch_assert(countof(_dispatch_root_queue_contexts) ==
908 DISPATCH_ROOT_QUEUE_COUNT);
909 dispatch_assert(countof(_dispatch_priority2qos) ==
910 DISPATCH_PRIORITY_COUNT);
911 #if HAVE_PTHREAD_WORKQUEUE_QOS
912 dispatch_assert(countof(_dispatch_priority2wq) ==
913 DISPATCH_PRIORITY_COUNT);
914 #endif
915 #if HAVE_PTHREAD_WORKQUEUE_SETDISPATCH_NP
916 dispatch_assert(sizeof(_dispatch_wq2root_queues) /
917 sizeof(_dispatch_wq2root_queues[0][0]) ==
918 WORKQ_NUM_PRIOQUEUE * 2);
919 #endif
920 #if DISPATCH_ENABLE_THREAD_POOL
921 dispatch_assert(countof(_dispatch_pthread_root_queue_contexts) ==
922 DISPATCH_ROOT_QUEUE_COUNT);
923 #endif
924
925 dispatch_assert(offsetof(struct dispatch_continuation_s, do_next) ==
926 offsetof(struct dispatch_object_s, do_next));
927 dispatch_assert(sizeof(struct dispatch_apply_s) <=
928 DISPATCH_CONTINUATION_SIZE);
929 dispatch_assert(sizeof(struct dispatch_queue_s) % DISPATCH_CACHELINE_SIZE
930 == 0);
931 dispatch_assert(sizeof(struct dispatch_root_queue_context_s) %
932 DISPATCH_CACHELINE_SIZE == 0);
933
934 _dispatch_thread_key_create(&dispatch_queue_key, _dispatch_queue_cleanup);
935 _dispatch_thread_key_create(&dispatch_voucher_key, _voucher_thread_cleanup);
936 _dispatch_thread_key_create(&dispatch_cache_key, _dispatch_cache_cleanup);
937 _dispatch_thread_key_create(&dispatch_io_key, NULL);
938 _dispatch_thread_key_create(&dispatch_apply_key, NULL);
939 _dispatch_thread_key_create(&dispatch_defaultpriority_key, NULL);
940 #if DISPATCH_PERF_MON && !DISPATCH_INTROSPECTION
941 _dispatch_thread_key_create(&dispatch_bcounter_key, NULL);
942 #endif
943 #if !DISPATCH_USE_OS_SEMAPHORE_CACHE
944 _dispatch_thread_key_create(&dispatch_sema4_key,
945 (void (*)(void *))_dispatch_thread_semaphore_dispose);
946 #endif
947
948 #if DISPATCH_USE_RESOLVERS // rdar://problem/8541707
949 _dispatch_main_q.do_targetq = &_dispatch_root_queues[
950 DISPATCH_ROOT_QUEUE_IDX_DEFAULT_QOS_OVERCOMMIT];
951 #endif
952
953 _dispatch_thread_setspecific(dispatch_queue_key, &_dispatch_main_q);
954 _dispatch_queue_set_bound_thread(&_dispatch_main_q);
955
956 #if DISPATCH_USE_PTHREAD_ATFORK
957 (void)dispatch_assume_zero(pthread_atfork(dispatch_atfork_prepare,
958 dispatch_atfork_parent, dispatch_atfork_child));
959 #endif
960
961 _dispatch_hw_config_init();
962 _dispatch_vtable_init();
963 _os_object_init();
964 _voucher_init();
965 _dispatch_introspection_init();
966 }
967
968 #if HAVE_MACH
969 static dispatch_once_t _dispatch_mach_host_port_pred;
970 static mach_port_t _dispatch_mach_host_port;
971
972 static void
_dispatch_mach_host_port_init(void * ctxt DISPATCH_UNUSED)973 _dispatch_mach_host_port_init(void *ctxt DISPATCH_UNUSED)
974 {
975 kern_return_t kr;
976 mach_port_t mp, mhp = mach_host_self();
977 kr = host_get_host_port(mhp, &mp);
978 DISPATCH_VERIFY_MIG(kr);
979 if (!kr) {
980 // mach_host_self returned the HOST_PRIV port
981 kr = mach_port_deallocate(mach_task_self(), mhp);
982 DISPATCH_VERIFY_MIG(kr);
983 (void)dispatch_assume_zero(kr);
984 mhp = mp;
985 } else if (kr != KERN_INVALID_ARGUMENT) {
986 (void)dispatch_assume_zero(kr);
987 }
988 if (!dispatch_assume(mhp)) {
989 DISPATCH_CRASH("Could not get unprivileged host port");
990 }
991 _dispatch_mach_host_port = mhp;
992 }
993
994 mach_port_t
_dispatch_get_mach_host_port(void)995 _dispatch_get_mach_host_port(void)
996 {
997 dispatch_once_f(&_dispatch_mach_host_port_pred, NULL,
998 _dispatch_mach_host_port_init);
999 return _dispatch_mach_host_port;
1000 }
1001 #endif
1002
1003 DISPATCH_EXPORT DISPATCH_NOTHROW
1004 void
dispatch_atfork_child(void)1005 dispatch_atfork_child(void)
1006 {
1007 void *crash = (void *)0x100;
1008 size_t i;
1009
1010 #if HAVE_MACH
1011 _dispatch_mach_host_port_pred = 0;
1012 _dispatch_mach_host_port = MACH_VOUCHER_NULL;
1013 #endif
1014 _voucher_atfork_child();
1015 if (_dispatch_safe_fork) {
1016 return;
1017 }
1018 _dispatch_child_of_unsafe_fork = true;
1019
1020 _dispatch_main_q.dq_items_head = crash;
1021 _dispatch_main_q.dq_items_tail = crash;
1022
1023 _dispatch_mgr_q.dq_items_head = crash;
1024 _dispatch_mgr_q.dq_items_tail = crash;
1025
1026 for (i = 0; i < DISPATCH_ROOT_QUEUE_COUNT; i++) {
1027 _dispatch_root_queues[i].dq_items_head = crash;
1028 _dispatch_root_queues[i].dq_items_tail = crash;
1029 }
1030 }
1031
1032 #pragma mark -
1033 #pragma mark dispatch_queue_attr_t
1034
1035 DISPATCH_ALWAYS_INLINE
1036 static inline bool
_dispatch_qos_class_valid(dispatch_qos_class_t qos_class,int relative_priority)1037 _dispatch_qos_class_valid(dispatch_qos_class_t qos_class, int relative_priority)
1038 {
1039 qos_class_t qos = (qos_class_t)qos_class;
1040 switch (qos) {
1041 case _DISPATCH_QOS_CLASS_MAINTENANCE:
1042 case _DISPATCH_QOS_CLASS_BACKGROUND:
1043 case _DISPATCH_QOS_CLASS_UTILITY:
1044 case _DISPATCH_QOS_CLASS_DEFAULT:
1045 case _DISPATCH_QOS_CLASS_USER_INITIATED:
1046 case _DISPATCH_QOS_CLASS_USER_INTERACTIVE:
1047 case _DISPATCH_QOS_CLASS_UNSPECIFIED:
1048 break;
1049 default:
1050 return false;
1051 }
1052 if (relative_priority > 0 || relative_priority < QOS_MIN_RELATIVE_PRIORITY){
1053 return false;
1054 }
1055 return true;
1056 }
1057
1058 #define DISPATCH_QUEUE_ATTR_QOS2IDX_INITIALIZER(qos) \
1059 [_DISPATCH_QOS_CLASS_##qos] = DQA_INDEX_QOS_CLASS_##qos
1060
1061 static const
1062 _dispatch_queue_attr_index_qos_class_t _dispatch_queue_attr_qos2idx[] = {
1063 DISPATCH_QUEUE_ATTR_QOS2IDX_INITIALIZER(UNSPECIFIED),
1064 DISPATCH_QUEUE_ATTR_QOS2IDX_INITIALIZER(MAINTENANCE),
1065 DISPATCH_QUEUE_ATTR_QOS2IDX_INITIALIZER(BACKGROUND),
1066 DISPATCH_QUEUE_ATTR_QOS2IDX_INITIALIZER(UTILITY),
1067 DISPATCH_QUEUE_ATTR_QOS2IDX_INITIALIZER(DEFAULT),
1068 DISPATCH_QUEUE_ATTR_QOS2IDX_INITIALIZER(USER_INITIATED),
1069 DISPATCH_QUEUE_ATTR_QOS2IDX_INITIALIZER(USER_INTERACTIVE),
1070 };
1071
1072 #define DISPATCH_QUEUE_ATTR_OVERCOMMIT2IDX(overcommit) \
1073 (overcommit ? DQA_INDEX_OVERCOMMIT : DQA_INDEX_NON_OVERCOMMIT)
1074
1075 #define DISPATCH_QUEUE_ATTR_CONCURRENT2IDX(concurrent) \
1076 (concurrent ? DQA_INDEX_CONCURRENT : DQA_INDEX_SERIAL)
1077
1078 #define DISPATCH_QUEUE_ATTR_PRIO2IDX(prio) (-(prio))
1079
1080 #define DISPATCH_QUEUE_ATTR_QOS2IDX(qos) (_dispatch_queue_attr_qos2idx[(qos)])
1081
1082 static inline dispatch_queue_attr_t
_dispatch_get_queue_attr(qos_class_t qos,int prio,bool overcommit,bool concurrent)1083 _dispatch_get_queue_attr(qos_class_t qos, int prio, bool overcommit,
1084 bool concurrent)
1085 {
1086 return (dispatch_queue_attr_t)&_dispatch_queue_attrs
1087 [DISPATCH_QUEUE_ATTR_QOS2IDX(qos)]
1088 [DISPATCH_QUEUE_ATTR_PRIO2IDX(prio)]
1089 [DISPATCH_QUEUE_ATTR_OVERCOMMIT2IDX(overcommit)]
1090 [DISPATCH_QUEUE_ATTR_CONCURRENT2IDX(concurrent)];
1091 }
1092
1093 dispatch_queue_attr_t
dispatch_queue_attr_make_with_qos_class(dispatch_queue_attr_t dqa,dispatch_qos_class_t qos_class,int relative_priority)1094 dispatch_queue_attr_make_with_qos_class(dispatch_queue_attr_t dqa,
1095 dispatch_qos_class_t qos_class, int relative_priority)
1096 {
1097 if (!_dispatch_qos_class_valid(qos_class, relative_priority)) return NULL;
1098 if (!slowpath(dqa)) {
1099 dqa = _dispatch_get_queue_attr(0, 0, false, false);
1100 } else if (dqa->do_vtable != DISPATCH_VTABLE(queue_attr)) {
1101 DISPATCH_CLIENT_CRASH("Invalid queue attribute");
1102 }
1103 return _dispatch_get_queue_attr(qos_class, relative_priority,
1104 dqa->dqa_overcommit, dqa->dqa_concurrent);
1105 }
1106
1107 dispatch_queue_attr_t
dispatch_queue_attr_make_with_overcommit(dispatch_queue_attr_t dqa,bool overcommit)1108 dispatch_queue_attr_make_with_overcommit(dispatch_queue_attr_t dqa,
1109 bool overcommit)
1110 {
1111 if (!slowpath(dqa)) {
1112 dqa = _dispatch_get_queue_attr(0, 0, false, false);
1113 } else if (dqa->do_vtable != DISPATCH_VTABLE(queue_attr)) {
1114 DISPATCH_CLIENT_CRASH("Invalid queue attribute");
1115 }
1116 return _dispatch_get_queue_attr(dqa->dqa_qos_class,
1117 dqa->dqa_relative_priority, overcommit, dqa->dqa_concurrent);
1118 }
1119
1120 #pragma mark -
1121 #pragma mark dispatch_queue_t
1122
1123 // skip zero
1124 // 1 - main_q
1125 // 2 - mgr_q
1126 // 3 - mgr_root_q
1127 // 4,5,6,7,8,9,10,11,12,13,14,15 - global queues
1128 // we use 'xadd' on Intel, so the initial value == next assigned
1129 unsigned long volatile _dispatch_queue_serial_numbers = 16;
1130
1131 dispatch_queue_t
dispatch_queue_create_with_target(const char * label,dispatch_queue_attr_t dqa,dispatch_queue_t tq)1132 dispatch_queue_create_with_target(const char *label, dispatch_queue_attr_t dqa,
1133 dispatch_queue_t tq)
1134 {
1135 #if DISPATCH_USE_NOQOS_WORKQUEUE_FALLBACK
1136 // Be sure the root queue priorities are set
1137 dispatch_once_f(&_dispatch_root_queues_pred, NULL,
1138 _dispatch_root_queues_init);
1139 #endif
1140 bool disallow_tq = (slowpath(dqa) && dqa != DISPATCH_QUEUE_CONCURRENT);
1141 if (!slowpath(dqa)) {
1142 dqa = _dispatch_get_queue_attr(0, 0, false, false);
1143 } else if (dqa->do_vtable != DISPATCH_VTABLE(queue_attr)) {
1144 DISPATCH_CLIENT_CRASH("Invalid queue attribute");
1145 }
1146 dispatch_queue_t dq = _dispatch_alloc(DISPATCH_VTABLE(queue),
1147 sizeof(struct dispatch_queue_s) - DISPATCH_QUEUE_CACHELINE_PAD);
1148 _dispatch_queue_init(dq);
1149 if (label) {
1150 dq->dq_label = strdup(label);
1151 }
1152 qos_class_t qos = dqa->dqa_qos_class;
1153 bool overcommit = dqa->dqa_overcommit;
1154 #if HAVE_PTHREAD_WORKQUEUE_QOS
1155 dq->dq_priority = _pthread_qos_class_encode(qos, dqa->dqa_relative_priority,
1156 overcommit);
1157 #endif
1158 if (dqa->dqa_concurrent) {
1159 dq->dq_width = DISPATCH_QUEUE_WIDTH_MAX;
1160 } else {
1161 // Default serial queue target queue is overcommit!
1162 overcommit = true;
1163 }
1164 if (!tq) {
1165 if (qos == _DISPATCH_QOS_CLASS_UNSPECIFIED) {
1166 qos = _DISPATCH_QOS_CLASS_DEFAULT;
1167 }
1168 #if DISPATCH_USE_NOQOS_WORKQUEUE_FALLBACK
1169 if (qos == _DISPATCH_QOS_CLASS_USER_INTERACTIVE &&
1170 !_dispatch_root_queues[
1171 DISPATCH_ROOT_QUEUE_IDX_USER_INTERACTIVE_QOS].dq_priority) {
1172 qos = _DISPATCH_QOS_CLASS_USER_INITIATED;
1173 }
1174 #endif
1175 bool maintenance_fallback = false;
1176 #if DISPATCH_USE_NOQOS_WORKQUEUE_FALLBACK
1177 maintenance_fallback = true;
1178 #endif // DISPATCH_USE_NOQOS_WORKQUEUE_FALLBACK
1179 if (maintenance_fallback) {
1180 if (qos == _DISPATCH_QOS_CLASS_MAINTENANCE &&
1181 !_dispatch_root_queues[
1182 DISPATCH_ROOT_QUEUE_IDX_MAINTENANCE_QOS].dq_priority) {
1183 qos = _DISPATCH_QOS_CLASS_BACKGROUND;
1184 }
1185 }
1186
1187 tq = _dispatch_get_root_queue(qos, overcommit);
1188 if (slowpath(!tq)) {
1189 DISPATCH_CLIENT_CRASH("Invalid queue attribute");
1190 }
1191 } else {
1192 _dispatch_retain(tq);
1193 if (disallow_tq) {
1194 // TODO: override target queue's qos/overcommit ?
1195 DISPATCH_CLIENT_CRASH("Invalid combination of target queue & "
1196 "queue attribute");
1197 }
1198 _dispatch_queue_priority_inherit_from_target(dq, tq);
1199 }
1200 _dispatch_queue_set_override_priority(dq);
1201 dq->do_targetq = tq;
1202 _dispatch_object_debug(dq, "%s", __func__);
1203 return _dispatch_introspection_queue_create(dq);
1204 }
1205
1206 dispatch_queue_t
dispatch_queue_create(const char * label,dispatch_queue_attr_t attr)1207 dispatch_queue_create(const char *label, dispatch_queue_attr_t attr)
1208 {
1209 return dispatch_queue_create_with_target(label, attr,
1210 DISPATCH_TARGET_QUEUE_DEFAULT);
1211 }
1212
1213 void
_dispatch_queue_destroy(dispatch_object_t dou)1214 _dispatch_queue_destroy(dispatch_object_t dou)
1215 {
1216 dispatch_queue_t dq = dou._dq;
1217 if (slowpath(dq == _dispatch_queue_get_current())) {
1218 DISPATCH_CRASH("Release of a queue by itself");
1219 }
1220 if (slowpath(dq->dq_items_tail)) {
1221 DISPATCH_CRASH("Release of a queue while items are enqueued");
1222 }
1223
1224 // trash the tail queue so that use after free will crash
1225 dq->dq_items_tail = (void *)0x200;
1226
1227 dispatch_queue_t dqsq = dispatch_atomic_xchg2o(dq, dq_specific_q,
1228 (void *)0x200, relaxed);
1229 if (dqsq) {
1230 _dispatch_release(dqsq);
1231 }
1232 }
1233
1234 // 6618342 Contact the team that owns the Instrument DTrace probe before
1235 // renaming this symbol
1236 void
_dispatch_queue_dispose(dispatch_queue_t dq)1237 _dispatch_queue_dispose(dispatch_queue_t dq)
1238 {
1239 _dispatch_object_debug(dq, "%s", __func__);
1240 _dispatch_introspection_queue_dispose(dq);
1241 if (dq->dq_label) {
1242 free((void*)dq->dq_label);
1243 }
1244 _dispatch_queue_destroy(dq);
1245 }
1246
1247 const char *
dispatch_queue_get_label(dispatch_queue_t dq)1248 dispatch_queue_get_label(dispatch_queue_t dq)
1249 {
1250 if (slowpath(dq == DISPATCH_CURRENT_QUEUE_LABEL)) {
1251 dq = _dispatch_get_current_queue();
1252 }
1253 return dq->dq_label ? dq->dq_label : "";
1254 }
1255
1256 qos_class_t
dispatch_queue_get_qos_class(dispatch_queue_t dq,int * relative_priority_ptr)1257 dispatch_queue_get_qos_class(dispatch_queue_t dq, int *relative_priority_ptr)
1258 {
1259 qos_class_t qos = _DISPATCH_QOS_CLASS_UNSPECIFIED;
1260 int relative_priority = 0;
1261 #if HAVE_PTHREAD_WORKQUEUE_QOS
1262 pthread_priority_t dqp = dq->dq_priority;
1263 if (dqp & _PTHREAD_PRIORITY_INHERIT_FLAG) dqp = 0;
1264 qos = _pthread_qos_class_decode(dqp, &relative_priority, NULL);
1265 #else
1266 (void)dq;
1267 #endif
1268 if (relative_priority_ptr) *relative_priority_ptr = relative_priority;
1269 return qos;
1270 }
1271
1272 static void
_dispatch_queue_set_width2(void * ctxt)1273 _dispatch_queue_set_width2(void *ctxt)
1274 {
1275 int w = (int)(intptr_t)ctxt; // intentional truncation
1276 uint32_t tmp;
1277 dispatch_queue_t dq = _dispatch_queue_get_current();
1278
1279 if (w == 1 || w == 0) {
1280 dq->dq_width = 1;
1281 _dispatch_object_debug(dq, "%s", __func__);
1282 return;
1283 }
1284 if (w > 0) {
1285 tmp = (unsigned int)w;
1286 } else switch (w) {
1287 case DISPATCH_QUEUE_WIDTH_MAX_PHYSICAL_CPUS:
1288 tmp = dispatch_hw_config(physical_cpus);
1289 break;
1290 case DISPATCH_QUEUE_WIDTH_ACTIVE_CPUS:
1291 tmp = dispatch_hw_config(active_cpus);
1292 break;
1293 default:
1294 // fall through
1295 case DISPATCH_QUEUE_WIDTH_MAX_LOGICAL_CPUS:
1296 tmp = dispatch_hw_config(logical_cpus);
1297 break;
1298 }
1299 if (tmp > DISPATCH_QUEUE_WIDTH_MAX / 2) {
1300 tmp = DISPATCH_QUEUE_WIDTH_MAX / 2;
1301 }
1302 // multiply by two since the running count is inc/dec by two
1303 // (the low bit == barrier)
1304 dq->dq_width = (typeof(dq->dq_width))(tmp * 2);
1305 _dispatch_object_debug(dq, "%s", __func__);
1306 }
1307
1308 void
dispatch_queue_set_width(dispatch_queue_t dq,long width)1309 dispatch_queue_set_width(dispatch_queue_t dq, long width)
1310 {
1311 if (slowpath(dq->do_ref_cnt == DISPATCH_OBJECT_GLOBAL_REFCNT) ||
1312 slowpath(dx_type(dq) == DISPATCH_QUEUE_ROOT_TYPE)) {
1313 return;
1314 }
1315 _dispatch_barrier_trysync_f(dq, (void*)(intptr_t)width,
1316 _dispatch_queue_set_width2);
1317 }
1318
1319 // 6618342 Contact the team that owns the Instrument DTrace probe before
1320 // renaming this symbol
1321 static void
_dispatch_set_target_queue2(void * ctxt)1322 _dispatch_set_target_queue2(void *ctxt)
1323 {
1324 dispatch_queue_t prev_dq, dq = _dispatch_queue_get_current(), tq = ctxt;
1325 mach_port_t th;
1326
1327 while (!dispatch_atomic_cmpxchgv2o(dq, dq_tqthread, MACH_PORT_NULL,
1328 _dispatch_thread_port(), &th, acquire)) {
1329 _dispatch_thread_switch(th, DISPATCH_YIELD_THREAD_SWITCH_OPTION,
1330 DISPATCH_CONTENTION_USLEEP_START);
1331 }
1332 _dispatch_queue_priority_inherit_from_target(dq, tq);
1333 prev_dq = dq->do_targetq;
1334 dq->do_targetq = tq;
1335 _dispatch_release(prev_dq);
1336 _dispatch_object_debug(dq, "%s", __func__);
1337 dispatch_atomic_store2o(dq, dq_tqthread, MACH_PORT_NULL, release);
1338 }
1339
1340 void
dispatch_set_target_queue(dispatch_object_t dou,dispatch_queue_t dq)1341 dispatch_set_target_queue(dispatch_object_t dou, dispatch_queue_t dq)
1342 {
1343 DISPATCH_OBJECT_TFB(_dispatch_objc_set_target_queue, dou, dq);
1344 if (slowpath(dou._do->do_ref_cnt == DISPATCH_OBJECT_GLOBAL_REFCNT) ||
1345 slowpath(dx_type(dou._do) == DISPATCH_QUEUE_ROOT_TYPE)) {
1346 return;
1347 }
1348 unsigned long type = dx_metatype(dou._do);
1349 if (slowpath(!dq)) {
1350 bool is_concurrent_q = (type == _DISPATCH_QUEUE_TYPE &&
1351 slowpath(dou._dq->dq_width > 1));
1352 dq = _dispatch_get_root_queue(_DISPATCH_QOS_CLASS_DEFAULT,
1353 !is_concurrent_q);
1354 }
1355 // TODO: put into the vtable
1356 switch(type) {
1357 case _DISPATCH_QUEUE_TYPE:
1358 case _DISPATCH_SOURCE_TYPE:
1359 _dispatch_retain(dq);
1360 return _dispatch_barrier_trysync_f(dou._dq, dq,
1361 _dispatch_set_target_queue2);
1362 case _DISPATCH_IO_TYPE:
1363 return _dispatch_io_set_target_queue(dou._dchannel, dq);
1364 default: {
1365 dispatch_queue_t prev_dq;
1366 _dispatch_retain(dq);
1367 prev_dq = dispatch_atomic_xchg2o(dou._do, do_targetq, dq, release);
1368 if (prev_dq) _dispatch_release(prev_dq);
1369 _dispatch_object_debug(dou._do, "%s", __func__);
1370 return;
1371 }
1372 }
1373 }
1374
1375 #pragma mark -
1376 #pragma mark dispatch_pthread_root_queue
1377
1378 #if DISPATCH_ENABLE_PTHREAD_ROOT_QUEUES
1379 static struct dispatch_pthread_root_queue_context_s
1380 _dispatch_mgr_root_queue_pthread_context;
1381 static struct dispatch_root_queue_context_s
1382 _dispatch_mgr_root_queue_context = {{{
1383 #if HAVE_PTHREAD_WORKQUEUES
1384 .dgq_kworkqueue = (void*)(~0ul),
1385 #endif
1386 .dgq_ctxt = &_dispatch_mgr_root_queue_pthread_context,
1387 .dgq_thread_pool_size = 1,
1388 }}};
1389 static struct dispatch_queue_s _dispatch_mgr_root_queue = {
1390 .do_vtable = DISPATCH_VTABLE(queue_root),
1391 .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
1392 .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
1393 .do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK,
1394 .do_ctxt = &_dispatch_mgr_root_queue_context,
1395 .dq_label = "com.apple.root.libdispatch-manager",
1396 .dq_running = 2,
1397 .dq_width = DISPATCH_QUEUE_WIDTH_MAX,
1398 .dq_serialnum = 3,
1399 };
1400 static struct {
1401 volatile int prio;
1402 int default_prio;
1403 int policy;
1404 pthread_t tid;
1405 } _dispatch_mgr_sched;
1406 static dispatch_once_t _dispatch_mgr_sched_pred;
1407
1408 static void
_dispatch_mgr_sched_init(void * ctxt DISPATCH_UNUSED)1409 _dispatch_mgr_sched_init(void *ctxt DISPATCH_UNUSED)
1410 {
1411 struct sched_param param;
1412 pthread_attr_t *attr;
1413 attr = &_dispatch_mgr_root_queue_pthread_context.dpq_thread_attr;
1414 (void)dispatch_assume_zero(pthread_attr_init(attr));
1415 (void)dispatch_assume_zero(pthread_attr_getschedpolicy(attr,
1416 &_dispatch_mgr_sched.policy));
1417 (void)dispatch_assume_zero(pthread_attr_getschedparam(attr, ¶m));
1418 // legacy priority calls allowed when requesting above default priority
1419 _dispatch_mgr_sched.default_prio = param.sched_priority;
1420 _dispatch_mgr_sched.prio = _dispatch_mgr_sched.default_prio;
1421 }
1422
1423 DISPATCH_NOINLINE
1424 static pthread_t *
_dispatch_mgr_root_queue_init(void)1425 _dispatch_mgr_root_queue_init(void)
1426 {
1427 dispatch_once_f(&_dispatch_mgr_sched_pred, NULL, _dispatch_mgr_sched_init);
1428 struct sched_param param;
1429 pthread_attr_t *attr;
1430 attr = &_dispatch_mgr_root_queue_pthread_context.dpq_thread_attr;
1431 (void)dispatch_assume_zero(pthread_attr_setdetachstate(attr,
1432 PTHREAD_CREATE_DETACHED));
1433 #if !DISPATCH_DEBUG
1434 (void)dispatch_assume_zero(pthread_attr_setstacksize(attr, 64 * 1024));
1435 #endif
1436 #if HAVE_PTHREAD_WORKQUEUE_QOS
1437 if (_dispatch_set_qos_class_enabled) {
1438 qos_class_t qos = qos_class_main();
1439 (void)dispatch_assume_zero(pthread_attr_set_qos_class_np(attr, qos, 0));
1440 _dispatch_mgr_q.dq_priority = _pthread_qos_class_encode(qos, 0, 0);
1441 _dispatch_queue_set_override_priority(&_dispatch_mgr_q);
1442 }
1443 #endif
1444 param.sched_priority = _dispatch_mgr_sched.prio;
1445 if (param.sched_priority > _dispatch_mgr_sched.default_prio) {
1446 (void)dispatch_assume_zero(pthread_attr_setschedparam(attr, ¶m));
1447 }
1448 return &_dispatch_mgr_sched.tid;
1449 }
1450
1451 static inline void
_dispatch_mgr_priority_apply(void)1452 _dispatch_mgr_priority_apply(void)
1453 {
1454 struct sched_param param;
1455 do {
1456 param.sched_priority = _dispatch_mgr_sched.prio;
1457 if (param.sched_priority > _dispatch_mgr_sched.default_prio) {
1458 (void)dispatch_assume_zero(pthread_setschedparam(
1459 _dispatch_mgr_sched.tid, _dispatch_mgr_sched.policy,
1460 ¶m));
1461 }
1462 } while (_dispatch_mgr_sched.prio > param.sched_priority);
1463 }
1464
1465 DISPATCH_NOINLINE
1466 void
_dispatch_mgr_priority_init(void)1467 _dispatch_mgr_priority_init(void)
1468 {
1469 struct sched_param param;
1470 pthread_attr_t *attr;
1471 attr = &_dispatch_mgr_root_queue_pthread_context.dpq_thread_attr;
1472 (void)dispatch_assume_zero(pthread_attr_getschedparam(attr, ¶m));
1473 if (slowpath(_dispatch_mgr_sched.prio > param.sched_priority)) {
1474 return _dispatch_mgr_priority_apply();
1475 }
1476 }
1477
1478 DISPATCH_NOINLINE
1479 static void
_dispatch_mgr_priority_raise(const pthread_attr_t * attr)1480 _dispatch_mgr_priority_raise(const pthread_attr_t *attr)
1481 {
1482 dispatch_once_f(&_dispatch_mgr_sched_pred, NULL, _dispatch_mgr_sched_init);
1483 struct sched_param param;
1484 (void)dispatch_assume_zero(pthread_attr_getschedparam(attr, ¶m));
1485 int p = _dispatch_mgr_sched.prio;
1486 do if (p >= param.sched_priority) {
1487 return;
1488 } while (slowpath(!dispatch_atomic_cmpxchgvw2o(&_dispatch_mgr_sched, prio,
1489 p, param.sched_priority, &p, relaxed)));
1490 if (_dispatch_mgr_sched.tid) {
1491 return _dispatch_mgr_priority_apply();
1492 }
1493 }
1494
1495 dispatch_queue_t
dispatch_pthread_root_queue_create(const char * label,unsigned long flags,const pthread_attr_t * attr,dispatch_block_t configure)1496 dispatch_pthread_root_queue_create(const char *label, unsigned long flags,
1497 const pthread_attr_t *attr, dispatch_block_t configure)
1498 {
1499 dispatch_queue_t dq;
1500 dispatch_root_queue_context_t qc;
1501 dispatch_pthread_root_queue_context_t pqc;
1502 size_t dqs;
1503 uint8_t pool_size = flags & _DISPATCH_PTHREAD_ROOT_QUEUE_FLAG_POOL_SIZE ?
1504 (uint8_t)(flags & ~_DISPATCH_PTHREAD_ROOT_QUEUE_FLAG_POOL_SIZE) : 0;
1505
1506 dqs = sizeof(struct dispatch_queue_s) - DISPATCH_QUEUE_CACHELINE_PAD;
1507 dq = _dispatch_alloc(DISPATCH_VTABLE(queue_root), dqs +
1508 sizeof(struct dispatch_root_queue_context_s) +
1509 sizeof(struct dispatch_pthread_root_queue_context_s));
1510 qc = (void*)((uint8_t *)dq + dqs);
1511 pqc = (void*)((uint8_t *)qc + sizeof(struct dispatch_root_queue_context_s));
1512
1513 _dispatch_queue_init(dq);
1514 if (label) {
1515 dq->dq_label = strdup(label);
1516 }
1517
1518 dq->do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK;
1519 dq->do_ctxt = qc;
1520 dq->do_targetq = NULL;
1521 dq->dq_running = 2;
1522 dq->dq_width = DISPATCH_QUEUE_WIDTH_MAX;
1523
1524 pqc->dpq_thread_mediator.do_vtable = DISPATCH_VTABLE(semaphore);
1525 qc->dgq_ctxt = pqc;
1526 #if HAVE_PTHREAD_WORKQUEUES
1527 qc->dgq_kworkqueue = (void*)(~0ul);
1528 #endif
1529 _dispatch_root_queue_init_pthread_pool(qc, pool_size, true);
1530
1531 if (attr) {
1532 memcpy(&pqc->dpq_thread_attr, attr, sizeof(pthread_attr_t));
1533 #if HAVE_PTHREAD_WORKQUEUE_QOS
1534 qos_class_t qos = 0;
1535 if (!pthread_attr_get_qos_class_np(&pqc->dpq_thread_attr, &qos, NULL)
1536 && qos > _DISPATCH_QOS_CLASS_DEFAULT) {
1537 DISPATCH_CLIENT_CRASH("pthread root queues do not support "
1538 "explicit QoS attributes");
1539 }
1540 #endif
1541 _dispatch_mgr_priority_raise(&pqc->dpq_thread_attr);
1542 } else {
1543 (void)dispatch_assume_zero(pthread_attr_init(&pqc->dpq_thread_attr));
1544 }
1545 (void)dispatch_assume_zero(pthread_attr_setdetachstate(
1546 &pqc->dpq_thread_attr, PTHREAD_CREATE_DETACHED));
1547 if (configure) {
1548 pqc->dpq_thread_configure = _dispatch_Block_copy(configure);
1549 }
1550 _dispatch_object_debug(dq, "%s", __func__);
1551 return _dispatch_introspection_queue_create(dq);
1552 }
1553 #endif
1554
1555 void
_dispatch_pthread_root_queue_dispose(dispatch_queue_t dq)1556 _dispatch_pthread_root_queue_dispose(dispatch_queue_t dq)
1557 {
1558 if (slowpath(dq->do_ref_cnt == DISPATCH_OBJECT_GLOBAL_REFCNT)) {
1559 DISPATCH_CRASH("Global root queue disposed");
1560 }
1561 _dispatch_object_debug(dq, "%s", __func__);
1562 _dispatch_introspection_queue_dispose(dq);
1563 #if DISPATCH_USE_PTHREAD_POOL
1564 dispatch_root_queue_context_t qc = dq->do_ctxt;
1565 dispatch_pthread_root_queue_context_t pqc = qc->dgq_ctxt;
1566
1567 pthread_attr_destroy(&pqc->dpq_thread_attr);
1568 _dispatch_semaphore_dispose(&pqc->dpq_thread_mediator);
1569 if (pqc->dpq_thread_configure) {
1570 Block_release(pqc->dpq_thread_configure);
1571 }
1572 dq->do_targetq = _dispatch_get_root_queue(_DISPATCH_QOS_CLASS_DEFAULT,
1573 false);
1574 #endif
1575 if (dq->dq_label) {
1576 free((void*)dq->dq_label);
1577 }
1578 _dispatch_queue_destroy(dq);
1579 }
1580
1581 #pragma mark -
1582 #pragma mark dispatch_queue_specific
1583
1584 struct dispatch_queue_specific_queue_s {
1585 DISPATCH_STRUCT_HEADER(queue_specific_queue);
1586 DISPATCH_QUEUE_HEADER;
1587 TAILQ_HEAD(dispatch_queue_specific_head_s,
1588 dispatch_queue_specific_s) dqsq_contexts;
1589 };
1590
1591 struct dispatch_queue_specific_s {
1592 const void *dqs_key;
1593 void *dqs_ctxt;
1594 dispatch_function_t dqs_destructor;
1595 TAILQ_ENTRY(dispatch_queue_specific_s) dqs_list;
1596 };
1597 DISPATCH_DECL(dispatch_queue_specific);
1598
1599 void
_dispatch_queue_specific_queue_dispose(dispatch_queue_specific_queue_t dqsq)1600 _dispatch_queue_specific_queue_dispose(dispatch_queue_specific_queue_t dqsq)
1601 {
1602 dispatch_queue_specific_t dqs, tmp;
1603
1604 TAILQ_FOREACH_SAFE(dqs, &dqsq->dqsq_contexts, dqs_list, tmp) {
1605 if (dqs->dqs_destructor) {
1606 dispatch_async_f(_dispatch_get_root_queue(
1607 _DISPATCH_QOS_CLASS_DEFAULT, false), dqs->dqs_ctxt,
1608 dqs->dqs_destructor);
1609 }
1610 free(dqs);
1611 }
1612 _dispatch_queue_destroy((dispatch_queue_t)dqsq);
1613 }
1614
1615 static void
_dispatch_queue_init_specific(dispatch_queue_t dq)1616 _dispatch_queue_init_specific(dispatch_queue_t dq)
1617 {
1618 dispatch_queue_specific_queue_t dqsq;
1619
1620 dqsq = _dispatch_alloc(DISPATCH_VTABLE(queue_specific_queue),
1621 sizeof(struct dispatch_queue_specific_queue_s));
1622 _dispatch_queue_init((dispatch_queue_t)dqsq);
1623 dqsq->do_xref_cnt = -1;
1624 dqsq->do_targetq = _dispatch_get_root_queue(
1625 _DISPATCH_QOS_CLASS_USER_INITIATED, true);
1626 dqsq->dq_width = DISPATCH_QUEUE_WIDTH_MAX;
1627 dqsq->dq_label = "queue-specific";
1628 TAILQ_INIT(&dqsq->dqsq_contexts);
1629 if (slowpath(!dispatch_atomic_cmpxchg2o(dq, dq_specific_q, NULL,
1630 (dispatch_queue_t)dqsq, release))) {
1631 _dispatch_release((dispatch_queue_t)dqsq);
1632 }
1633 }
1634
1635 static void
_dispatch_queue_set_specific(void * ctxt)1636 _dispatch_queue_set_specific(void *ctxt)
1637 {
1638 dispatch_queue_specific_t dqs, dqsn = ctxt;
1639 dispatch_queue_specific_queue_t dqsq =
1640 (dispatch_queue_specific_queue_t)_dispatch_queue_get_current();
1641
1642 TAILQ_FOREACH(dqs, &dqsq->dqsq_contexts, dqs_list) {
1643 if (dqs->dqs_key == dqsn->dqs_key) {
1644 // Destroy previous context for existing key
1645 if (dqs->dqs_destructor) {
1646 dispatch_async_f(_dispatch_get_root_queue(
1647 _DISPATCH_QOS_CLASS_DEFAULT, false), dqs->dqs_ctxt,
1648 dqs->dqs_destructor);
1649 }
1650 if (dqsn->dqs_ctxt) {
1651 // Copy new context for existing key
1652 dqs->dqs_ctxt = dqsn->dqs_ctxt;
1653 dqs->dqs_destructor = dqsn->dqs_destructor;
1654 } else {
1655 // Remove context storage for existing key
1656 TAILQ_REMOVE(&dqsq->dqsq_contexts, dqs, dqs_list);
1657 free(dqs);
1658 }
1659 return free(dqsn);
1660 }
1661 }
1662 // Insert context storage for new key
1663 TAILQ_INSERT_TAIL(&dqsq->dqsq_contexts, dqsn, dqs_list);
1664 }
1665
1666 DISPATCH_NOINLINE
1667 void
dispatch_queue_set_specific(dispatch_queue_t dq,const void * key,void * ctxt,dispatch_function_t destructor)1668 dispatch_queue_set_specific(dispatch_queue_t dq, const void *key,
1669 void *ctxt, dispatch_function_t destructor)
1670 {
1671 if (slowpath(!key)) {
1672 return;
1673 }
1674 dispatch_queue_specific_t dqs;
1675
1676 dqs = _dispatch_calloc(1, sizeof(struct dispatch_queue_specific_s));
1677 dqs->dqs_key = key;
1678 dqs->dqs_ctxt = ctxt;
1679 dqs->dqs_destructor = destructor;
1680 if (slowpath(!dq->dq_specific_q)) {
1681 _dispatch_queue_init_specific(dq);
1682 }
1683 _dispatch_barrier_trysync_f(dq->dq_specific_q, dqs,
1684 _dispatch_queue_set_specific);
1685 }
1686
1687 static void
_dispatch_queue_get_specific(void * ctxt)1688 _dispatch_queue_get_specific(void *ctxt)
1689 {
1690 void **ctxtp = ctxt;
1691 void *key = *ctxtp;
1692 dispatch_queue_specific_queue_t dqsq =
1693 (dispatch_queue_specific_queue_t)_dispatch_queue_get_current();
1694 dispatch_queue_specific_t dqs;
1695
1696 TAILQ_FOREACH(dqs, &dqsq->dqsq_contexts, dqs_list) {
1697 if (dqs->dqs_key == key) {
1698 *ctxtp = dqs->dqs_ctxt;
1699 return;
1700 }
1701 }
1702 *ctxtp = NULL;
1703 }
1704
1705 DISPATCH_NOINLINE
1706 void *
dispatch_queue_get_specific(dispatch_queue_t dq,const void * key)1707 dispatch_queue_get_specific(dispatch_queue_t dq, const void *key)
1708 {
1709 if (slowpath(!key)) {
1710 return NULL;
1711 }
1712 void *ctxt = NULL;
1713
1714 if (fastpath(dq->dq_specific_q)) {
1715 ctxt = (void *)key;
1716 dispatch_sync_f(dq->dq_specific_q, &ctxt, _dispatch_queue_get_specific);
1717 }
1718 return ctxt;
1719 }
1720
1721 DISPATCH_NOINLINE
1722 void *
dispatch_get_specific(const void * key)1723 dispatch_get_specific(const void *key)
1724 {
1725 if (slowpath(!key)) {
1726 return NULL;
1727 }
1728 void *ctxt = NULL;
1729 dispatch_queue_t dq = _dispatch_queue_get_current();
1730
1731 while (slowpath(dq)) {
1732 if (slowpath(dq->dq_specific_q)) {
1733 ctxt = (void *)key;
1734 dispatch_sync_f(dq->dq_specific_q, &ctxt,
1735 _dispatch_queue_get_specific);
1736 if (ctxt) break;
1737 }
1738 dq = dq->do_targetq;
1739 }
1740 return ctxt;
1741 }
1742
1743 #pragma mark -
1744 #pragma mark dispatch_queue_debug
1745
1746 size_t
_dispatch_queue_debug_attr(dispatch_queue_t dq,char * buf,size_t bufsiz)1747 _dispatch_queue_debug_attr(dispatch_queue_t dq, char* buf, size_t bufsiz)
1748 {
1749 size_t offset = 0;
1750 dispatch_queue_t target = dq->do_targetq;
1751 offset += dsnprintf(buf, bufsiz, "target = %s[%p], width = 0x%x, "
1752 "running = 0x%x, barrier = %d ", target && target->dq_label ?
1753 target->dq_label : "", target, dq->dq_width / 2,
1754 dq->dq_running / 2, dq->dq_running & 1);
1755 if (dq->dq_is_thread_bound) {
1756 offset += dsnprintf(buf, bufsiz, ", thread = 0x%x ",
1757 _dispatch_queue_get_bound_thread(dq));
1758 }
1759 return offset;
1760 }
1761
1762 size_t
dispatch_queue_debug(dispatch_queue_t dq,char * buf,size_t bufsiz)1763 dispatch_queue_debug(dispatch_queue_t dq, char* buf, size_t bufsiz)
1764 {
1765 size_t offset = 0;
1766 offset += dsnprintf(&buf[offset], bufsiz - offset, "%s[%p] = { ",
1767 dq->dq_label ? dq->dq_label : dx_kind(dq), dq);
1768 offset += _dispatch_object_debug_attr(dq, &buf[offset], bufsiz - offset);
1769 offset += _dispatch_queue_debug_attr(dq, &buf[offset], bufsiz - offset);
1770 offset += dsnprintf(&buf[offset], bufsiz - offset, "}");
1771 return offset;
1772 }
1773
1774 #if DISPATCH_DEBUG
1775 void
dispatch_debug_queue(dispatch_queue_t dq,const char * str)1776 dispatch_debug_queue(dispatch_queue_t dq, const char* str) {
1777 if (fastpath(dq)) {
1778 _dispatch_object_debug(dq, "%s", str);
1779 } else {
1780 _dispatch_log("queue[NULL]: %s", str);
1781 }
1782 }
1783 #endif
1784
1785 #if DISPATCH_PERF_MON && !DISPATCH_INTROSPECTION
1786 static OSSpinLock _dispatch_stats_lock;
1787 static struct {
1788 uint64_t time_total;
1789 uint64_t count_total;
1790 uint64_t thread_total;
1791 } _dispatch_stats[65]; // ffs*/fls*() returns zero when no bits are set
1792
1793 static void
_dispatch_queue_merge_stats(uint64_t start)1794 _dispatch_queue_merge_stats(uint64_t start)
1795 {
1796 uint64_t delta = _dispatch_absolute_time() - start;
1797 unsigned long count;
1798
1799 count = (unsigned long)_dispatch_thread_getspecific(dispatch_bcounter_key);
1800 _dispatch_thread_setspecific(dispatch_bcounter_key, NULL);
1801
1802 int bucket = flsl((long)count);
1803
1804 // 64-bit counters on 32-bit require a lock or a queue
1805 OSSpinLockLock(&_dispatch_stats_lock);
1806
1807 _dispatch_stats[bucket].time_total += delta;
1808 _dispatch_stats[bucket].count_total += count;
1809 _dispatch_stats[bucket].thread_total++;
1810
1811 OSSpinLockUnlock(&_dispatch_stats_lock);
1812 }
1813 #endif
1814
1815 #pragma mark -
1816 #pragma mark dispatch_continuation_t
1817
1818 static void
_dispatch_force_cache_cleanup(void)1819 _dispatch_force_cache_cleanup(void)
1820 {
1821 dispatch_continuation_t dc;
1822 dc = _dispatch_thread_getspecific(dispatch_cache_key);
1823 if (dc) {
1824 _dispatch_thread_setspecific(dispatch_cache_key, NULL);
1825 _dispatch_cache_cleanup(dc);
1826 }
1827 }
1828
1829 DISPATCH_NOINLINE
1830 static void
_dispatch_cache_cleanup(void * value)1831 _dispatch_cache_cleanup(void *value)
1832 {
1833 dispatch_continuation_t dc, next_dc = value;
1834
1835 while ((dc = next_dc)) {
1836 next_dc = dc->do_next;
1837 _dispatch_continuation_free_to_heap(dc);
1838 }
1839 }
1840
1841 #if DISPATCH_USE_MEMORYSTATUS_SOURCE
1842 int _dispatch_continuation_cache_limit = DISPATCH_CONTINUATION_CACHE_LIMIT;
1843
1844 DISPATCH_NOINLINE
1845 void
_dispatch_continuation_free_to_cache_limit(dispatch_continuation_t dc)1846 _dispatch_continuation_free_to_cache_limit(dispatch_continuation_t dc)
1847 {
1848 _dispatch_continuation_free_to_heap(dc);
1849 dispatch_continuation_t next_dc;
1850 dc = _dispatch_thread_getspecific(dispatch_cache_key);
1851 int cnt;
1852 if (!dc || (cnt = dc->dc_cache_cnt -
1853 _dispatch_continuation_cache_limit) <= 0){
1854 return;
1855 }
1856 do {
1857 next_dc = dc->do_next;
1858 _dispatch_continuation_free_to_heap(dc);
1859 } while (--cnt && (dc = next_dc));
1860 _dispatch_thread_setspecific(dispatch_cache_key, next_dc);
1861 }
1862 #endif
1863
1864 DISPATCH_ALWAYS_INLINE_NDEBUG
1865 static inline void
_dispatch_continuation_redirect(dispatch_queue_t dq,dispatch_object_t dou)1866 _dispatch_continuation_redirect(dispatch_queue_t dq, dispatch_object_t dou)
1867 {
1868 dispatch_continuation_t dc = dou._dc;
1869
1870 (void)dispatch_atomic_add2o(dq, dq_running, 2, acquire);
1871 if (!DISPATCH_OBJ_IS_VTABLE(dc) &&
1872 (long)dc->do_vtable & DISPATCH_OBJ_SYNC_SLOW_BIT) {
1873 _dispatch_trace_continuation_pop(dq, dou);
1874 _dispatch_thread_semaphore_signal(
1875 (_dispatch_thread_semaphore_t)dc->dc_other);
1876 _dispatch_introspection_queue_item_complete(dou);
1877 } else {
1878 _dispatch_async_f_redirect(dq, dc,
1879 _dispatch_queue_get_override_priority(dq));
1880 }
1881 _dispatch_perfmon_workitem_inc();
1882 }
1883
1884 #pragma mark -
1885 #pragma mark dispatch_block_create
1886
1887 #if __BLOCKS__
1888
1889 DISPATCH_ALWAYS_INLINE
1890 static inline bool
_dispatch_block_flags_valid(dispatch_block_flags_t flags)1891 _dispatch_block_flags_valid(dispatch_block_flags_t flags)
1892 {
1893 return ((flags & ~DISPATCH_BLOCK_API_MASK) == 0);
1894 }
1895
1896 DISPATCH_ALWAYS_INLINE
1897 static inline dispatch_block_flags_t
_dispatch_block_normalize_flags(dispatch_block_flags_t flags)1898 _dispatch_block_normalize_flags(dispatch_block_flags_t flags)
1899 {
1900 if (flags & (DISPATCH_BLOCK_NO_VOUCHER|DISPATCH_BLOCK_DETACHED)) {
1901 flags |= DISPATCH_BLOCK_HAS_VOUCHER;
1902 }
1903 if (flags & (DISPATCH_BLOCK_NO_QOS_CLASS|DISPATCH_BLOCK_DETACHED)) {
1904 flags |= DISPATCH_BLOCK_HAS_PRIORITY;
1905 }
1906 return flags;
1907 }
1908
1909 static inline dispatch_block_t
_dispatch_block_create_with_voucher_and_priority(dispatch_block_flags_t flags,voucher_t voucher,pthread_priority_t pri,dispatch_block_t block)1910 _dispatch_block_create_with_voucher_and_priority(dispatch_block_flags_t flags,
1911 voucher_t voucher, pthread_priority_t pri, dispatch_block_t block)
1912 {
1913 flags = _dispatch_block_normalize_flags(flags);
1914 voucher_t cv = NULL;
1915 bool assign = (flags & DISPATCH_BLOCK_ASSIGN_CURRENT);
1916 if (assign && !(flags & DISPATCH_BLOCK_HAS_VOUCHER)) {
1917 voucher = cv = voucher_copy();
1918 flags |= DISPATCH_BLOCK_HAS_VOUCHER;
1919 }
1920 if (assign && !(flags & DISPATCH_BLOCK_HAS_PRIORITY)) {
1921 pri = _dispatch_priority_propagate();
1922 flags |= DISPATCH_BLOCK_HAS_PRIORITY;
1923 }
1924 dispatch_block_t db = _dispatch_block_create(flags, voucher, pri, block);
1925 if (cv) _voucher_release(cv);
1926 #if DISPATCH_DEBUG
1927 dispatch_assert(_dispatch_block_get_data(db));
1928 #endif
1929 return db;
1930 }
1931
1932 dispatch_block_t
dispatch_block_create(dispatch_block_flags_t flags,dispatch_block_t block)1933 dispatch_block_create(dispatch_block_flags_t flags, dispatch_block_t block)
1934 {
1935 if (!_dispatch_block_flags_valid(flags)) return NULL;
1936 return _dispatch_block_create_with_voucher_and_priority(flags, NULL, 0,
1937 block);
1938 }
1939
1940 dispatch_block_t
dispatch_block_create_with_qos_class(dispatch_block_flags_t flags,dispatch_qos_class_t qos_class,int relative_priority,dispatch_block_t block)1941 dispatch_block_create_with_qos_class(dispatch_block_flags_t flags,
1942 dispatch_qos_class_t qos_class, int relative_priority,
1943 dispatch_block_t block)
1944 {
1945 if (!_dispatch_block_flags_valid(flags)) return NULL;
1946 if (!_dispatch_qos_class_valid(qos_class, relative_priority)) return NULL;
1947 flags |= DISPATCH_BLOCK_HAS_PRIORITY;
1948 pthread_priority_t pri = 0;
1949 #if HAVE_PTHREAD_WORKQUEUE_QOS
1950 pri = _pthread_qos_class_encode(qos_class, relative_priority, 0);
1951 #endif
1952 return _dispatch_block_create_with_voucher_and_priority(flags, NULL,
1953 pri, block);
1954 }
1955
1956 dispatch_block_t
dispatch_block_create_with_voucher(dispatch_block_flags_t flags,voucher_t voucher,dispatch_block_t block)1957 dispatch_block_create_with_voucher(dispatch_block_flags_t flags,
1958 voucher_t voucher, dispatch_block_t block)
1959 {
1960 if (!_dispatch_block_flags_valid(flags)) return NULL;
1961 flags |= DISPATCH_BLOCK_HAS_VOUCHER;
1962 return _dispatch_block_create_with_voucher_and_priority(flags, voucher, 0,
1963 block);
1964 }
1965
1966 dispatch_block_t
dispatch_block_create_with_voucher_and_qos_class(dispatch_block_flags_t flags,voucher_t voucher,dispatch_qos_class_t qos_class,int relative_priority,dispatch_block_t block)1967 dispatch_block_create_with_voucher_and_qos_class(dispatch_block_flags_t flags,
1968 voucher_t voucher, dispatch_qos_class_t qos_class,
1969 int relative_priority, dispatch_block_t block)
1970 {
1971 if (!_dispatch_block_flags_valid(flags)) return NULL;
1972 if (!_dispatch_qos_class_valid(qos_class, relative_priority)) return NULL;
1973 flags |= (DISPATCH_BLOCK_HAS_VOUCHER|DISPATCH_BLOCK_HAS_PRIORITY);
1974 pthread_priority_t pri = 0;
1975 #if HAVE_PTHREAD_WORKQUEUE_QOS
1976 pri = _pthread_qos_class_encode(qos_class, relative_priority, 0);
1977 #endif
1978 return _dispatch_block_create_with_voucher_and_priority(flags, voucher,
1979 pri, block);
1980 }
1981
1982 void
dispatch_block_perform(dispatch_block_flags_t flags,dispatch_block_t block)1983 dispatch_block_perform(dispatch_block_flags_t flags, dispatch_block_t block)
1984 {
1985 if (!_dispatch_block_flags_valid(flags)) {
1986 DISPATCH_CLIENT_CRASH("Invalid flags passed to "
1987 "dispatch_block_perform()");
1988 }
1989 flags = _dispatch_block_normalize_flags(flags);
1990 struct dispatch_block_private_data_s dbpds =
1991 DISPATCH_BLOCK_PRIVATE_DATA_INITIALIZER(flags, NULL, 0, block);
1992 dbpds.dbpd_atomic_flags |= DBF_PERFORM; // no group_leave at end of invoke
1993 return _dispatch_block_invoke(&dbpds);
1994 }
1995
1996 #define _dbpd_group(dbpd) ((dispatch_group_t)&(dbpd)->dbpd_group)
1997
1998 void
_dispatch_block_invoke(const struct dispatch_block_private_data_s * dbcpd)1999 _dispatch_block_invoke(const struct dispatch_block_private_data_s *dbcpd)
2000 {
2001 dispatch_block_private_data_t dbpd = (dispatch_block_private_data_t)dbcpd;
2002 dispatch_block_flags_t flags = dbpd->dbpd_flags;
2003 unsigned int atomic_flags = dbpd->dbpd_atomic_flags;
2004 if (slowpath(atomic_flags & DBF_WAITED)) {
2005 DISPATCH_CLIENT_CRASH("A block object may not be both run more "
2006 "than once and waited for");
2007 }
2008 if (atomic_flags & DBF_CANCELED) goto out;
2009
2010 pthread_priority_t op = DISPATCH_NO_PRIORITY, p = DISPATCH_NO_PRIORITY;
2011 unsigned long override = 0;
2012 if (flags & DISPATCH_BLOCK_HAS_PRIORITY) {
2013 op = _dispatch_get_priority();
2014 p = dbpd->dbpd_priority;
2015 override = (flags & DISPATCH_BLOCK_ENFORCE_QOS_CLASS) ||
2016 !(flags & DISPATCH_BLOCK_INHERIT_QOS_CLASS) ?
2017 DISPATCH_PRIORITY_ENFORCE : 0;
2018 }
2019 voucher_t ov, v = DISPATCH_NO_VOUCHER;
2020 if (flags & DISPATCH_BLOCK_HAS_VOUCHER) {
2021 v = dbpd->dbpd_voucher;
2022 if (v) _voucher_retain(v);
2023 }
2024 ov = _dispatch_adopt_priority_and_voucher(p, v, override);
2025 dbpd->dbpd_thread = _dispatch_thread_port();
2026 dbpd->dbpd_block();
2027 _dispatch_set_priority_and_replace_voucher(op, ov);
2028 out:
2029 if ((atomic_flags & DBF_PERFORM) == 0) {
2030 if (dispatch_atomic_inc2o(dbpd, dbpd_performed, acquire) == 1) {
2031 dispatch_group_leave(_dbpd_group(dbpd));
2032 }
2033 }
2034 }
2035
2036 static void
_dispatch_block_sync_invoke(void * block)2037 _dispatch_block_sync_invoke(void *block)
2038 {
2039 dispatch_block_t b = block;
2040 dispatch_block_private_data_t dbpd = _dispatch_block_get_data(b);
2041 dispatch_block_flags_t flags = dbpd->dbpd_flags;
2042 unsigned int atomic_flags = dbpd->dbpd_atomic_flags;
2043 if (slowpath(atomic_flags & DBF_WAITED)) {
2044 DISPATCH_CLIENT_CRASH("A block object may not be both run more "
2045 "than once and waited for");
2046 }
2047 if (atomic_flags & DBF_CANCELED) goto out;
2048
2049 pthread_priority_t op = DISPATCH_NO_PRIORITY, p = DISPATCH_NO_PRIORITY;
2050 unsigned long override = 0;
2051 if (flags & DISPATCH_BLOCK_HAS_PRIORITY) {
2052 op = _dispatch_get_priority();
2053 p = dbpd->dbpd_priority;
2054 override = (flags & DISPATCH_BLOCK_ENFORCE_QOS_CLASS) ||
2055 !(flags & DISPATCH_BLOCK_INHERIT_QOS_CLASS) ?
2056 DISPATCH_PRIORITY_ENFORCE : 0;
2057 }
2058 voucher_t ov, v = DISPATCH_NO_VOUCHER;
2059 if (flags & DISPATCH_BLOCK_HAS_VOUCHER) {
2060 v = dbpd->dbpd_voucher;
2061 if (v) _voucher_retain(v);
2062 }
2063 ov = _dispatch_adopt_priority_and_voucher(p, v, override);
2064 dbpd->dbpd_block();
2065 _dispatch_set_priority_and_replace_voucher(op, ov);
2066 out:
2067 if ((atomic_flags & DBF_PERFORM) == 0) {
2068 if (dispatch_atomic_inc2o(dbpd, dbpd_performed, acquire) == 1) {
2069 dispatch_group_leave(_dbpd_group(dbpd));
2070 }
2071 }
2072
2073 dispatch_queue_t dq = _dispatch_queue_get_current();
2074 if (dispatch_atomic_cmpxchg2o(dbpd, dbpd_queue, dq, NULL, acquire)) {
2075 // balances dispatch_{,barrier_,}sync
2076 _dispatch_release(dq);
2077 }
2078 }
2079
2080 static void
_dispatch_block_async_invoke_and_release(void * block)2081 _dispatch_block_async_invoke_and_release(void *block)
2082 {
2083 dispatch_block_t b = block;
2084 dispatch_block_private_data_t dbpd = _dispatch_block_get_data(b);
2085 dispatch_block_flags_t flags = dbpd->dbpd_flags;
2086 unsigned int atomic_flags = dbpd->dbpd_atomic_flags;
2087 if (slowpath(atomic_flags & DBF_WAITED)) {
2088 DISPATCH_CLIENT_CRASH("A block object may not be both run more "
2089 "than once and waited for");
2090 }
2091 if (atomic_flags & DBF_CANCELED) goto out;
2092
2093 pthread_priority_t p = DISPATCH_NO_PRIORITY;
2094 unsigned long override = 0;
2095 if (flags & DISPATCH_BLOCK_HAS_PRIORITY) {
2096 override = (flags & DISPATCH_BLOCK_ENFORCE_QOS_CLASS) ?
2097 DISPATCH_PRIORITY_ENFORCE : 0;
2098 p = dbpd->dbpd_priority;
2099 }
2100 voucher_t v = DISPATCH_NO_VOUCHER;
2101 if (flags & DISPATCH_BLOCK_HAS_VOUCHER) {
2102 v = dbpd->dbpd_voucher;
2103 if (v) _voucher_retain(v);
2104 }
2105 _dispatch_adopt_priority_and_replace_voucher(p, v, override);
2106 dbpd->dbpd_block();
2107 out:
2108 if ((atomic_flags & DBF_PERFORM) == 0) {
2109 if (dispatch_atomic_inc2o(dbpd, dbpd_performed, acquire) == 1) {
2110 dispatch_group_leave(_dbpd_group(dbpd));
2111 }
2112 }
2113 dispatch_queue_t dq = _dispatch_queue_get_current();
2114 if (dispatch_atomic_cmpxchg2o(dbpd, dbpd_queue, dq, NULL, acquire)) {
2115 // balances dispatch_{,barrier_,group_}async
2116 _dispatch_release(dq);
2117 }
2118 Block_release(b);
2119 }
2120
2121 void
dispatch_block_cancel(dispatch_block_t db)2122 dispatch_block_cancel(dispatch_block_t db)
2123 {
2124 dispatch_block_private_data_t dbpd = _dispatch_block_get_data(db);
2125 if (!dbpd) {
2126 DISPATCH_CLIENT_CRASH("Invalid block object passed to "
2127 "dispatch_block_cancel()");
2128 }
2129 (void)dispatch_atomic_or2o(dbpd, dbpd_atomic_flags, DBF_CANCELED, relaxed);
2130 }
2131
2132 long
dispatch_block_testcancel(dispatch_block_t db)2133 dispatch_block_testcancel(dispatch_block_t db)
2134 {
2135 dispatch_block_private_data_t dbpd = _dispatch_block_get_data(db);
2136 if (!dbpd) {
2137 DISPATCH_CLIENT_CRASH("Invalid block object passed to "
2138 "dispatch_block_testcancel()");
2139 }
2140 return (bool)(dbpd->dbpd_atomic_flags & DBF_CANCELED);
2141 }
2142
2143 long
dispatch_block_wait(dispatch_block_t db,dispatch_time_t timeout)2144 dispatch_block_wait(dispatch_block_t db, dispatch_time_t timeout)
2145 {
2146 dispatch_block_private_data_t dbpd = _dispatch_block_get_data(db);
2147 if (!dbpd) {
2148 DISPATCH_CLIENT_CRASH("Invalid block object passed to "
2149 "dispatch_block_wait()");
2150 }
2151
2152 unsigned int flags = dispatch_atomic_or_orig2o(dbpd, dbpd_atomic_flags,
2153 DBF_WAITING, relaxed);
2154 if (slowpath(flags & (DBF_WAITED | DBF_WAITING))) {
2155 DISPATCH_CLIENT_CRASH("A block object may not be waited for "
2156 "more than once");
2157 }
2158
2159 // <rdar://problem/17703192> If we know the queue where this block is
2160 // enqueued, or the thread that's executing it, then we should boost
2161 // it here.
2162
2163 pthread_priority_t pp = _dispatch_get_priority();
2164
2165 dispatch_queue_t boost_dq;
2166 boost_dq = dispatch_atomic_xchg2o(dbpd, dbpd_queue, NULL, acquire);
2167 if (boost_dq) {
2168 // release balances dispatch_{,barrier_,group_}async.
2169 // Can't put the queue back in the timeout case: the block might
2170 // finish after we fell out of group_wait and see our NULL, so
2171 // neither of us would ever release. Side effect: After a _wait
2172 // that times out, subsequent waits will not boost the qos of the
2173 // still-running block.
2174 _dispatch_queue_wakeup_with_qos_and_release(boost_dq, pp);
2175 }
2176
2177 mach_port_t boost_th = dbpd->dbpd_thread;
2178 if (boost_th) {
2179 _dispatch_thread_override_start(boost_th, pp);
2180 }
2181
2182 int performed = dispatch_atomic_load2o(dbpd, dbpd_performed, relaxed);
2183 if (slowpath(performed > 1 || (boost_th && boost_dq))) {
2184 DISPATCH_CLIENT_CRASH("A block object may not be both run more "
2185 "than once and waited for");
2186 }
2187
2188 long ret = dispatch_group_wait(_dbpd_group(dbpd), timeout);
2189
2190 if (boost_th) {
2191 _dispatch_thread_override_end(boost_th);
2192 }
2193
2194 if (ret) {
2195 // timed out: reverse our changes
2196 (void)dispatch_atomic_and2o(dbpd, dbpd_atomic_flags,
2197 ~DBF_WAITING, relaxed);
2198 } else {
2199 (void)dispatch_atomic_or2o(dbpd, dbpd_atomic_flags,
2200 DBF_WAITED, relaxed);
2201 // don't need to re-test here: the second call would see
2202 // the first call's WAITING
2203 }
2204
2205 return ret;
2206 }
2207
2208 void
dispatch_block_notify(dispatch_block_t db,dispatch_queue_t queue,dispatch_block_t notification_block)2209 dispatch_block_notify(dispatch_block_t db, dispatch_queue_t queue,
2210 dispatch_block_t notification_block)
2211 {
2212 dispatch_block_private_data_t dbpd = _dispatch_block_get_data(db);
2213 if (!dbpd) {
2214 DISPATCH_CLIENT_CRASH("Invalid block object passed to "
2215 "dispatch_block_notify()");
2216 }
2217 int performed = dispatch_atomic_load2o(dbpd, dbpd_performed, relaxed);
2218 if (slowpath(performed > 1)) {
2219 DISPATCH_CLIENT_CRASH("A block object may not be both run more "
2220 "than once and observed");
2221 }
2222
2223 return dispatch_group_notify(_dbpd_group(dbpd), queue, notification_block);
2224 }
2225
2226 #endif // __BLOCKS__
2227
2228 #pragma mark -
2229 #pragma mark dispatch_barrier_async
2230
2231 DISPATCH_NOINLINE
2232 static void
_dispatch_barrier_async_f_slow(dispatch_queue_t dq,void * ctxt,dispatch_function_t func,pthread_priority_t pp,dispatch_block_flags_t flags)2233 _dispatch_barrier_async_f_slow(dispatch_queue_t dq, void *ctxt,
2234 dispatch_function_t func, pthread_priority_t pp,
2235 dispatch_block_flags_t flags)
2236 {
2237 dispatch_continuation_t dc = _dispatch_continuation_alloc_from_heap();
2238
2239 dc->do_vtable = (void *)(DISPATCH_OBJ_ASYNC_BIT | DISPATCH_OBJ_BARRIER_BIT);
2240 dc->dc_func = func;
2241 dc->dc_ctxt = ctxt;
2242 _dispatch_continuation_voucher_set(dc, flags);
2243 _dispatch_continuation_priority_set(dc, pp, flags);
2244
2245 pp = _dispatch_continuation_get_override_priority(dq, dc);
2246
2247 _dispatch_queue_push(dq, dc, pp);
2248 }
2249
2250 DISPATCH_ALWAYS_INLINE
2251 static inline void
_dispatch_barrier_async_f2(dispatch_queue_t dq,void * ctxt,dispatch_function_t func,pthread_priority_t pp,dispatch_block_flags_t flags)2252 _dispatch_barrier_async_f2(dispatch_queue_t dq, void *ctxt,
2253 dispatch_function_t func, pthread_priority_t pp,
2254 dispatch_block_flags_t flags)
2255 {
2256 dispatch_continuation_t dc;
2257
2258 dc = fastpath(_dispatch_continuation_alloc_cacheonly());
2259 if (!dc) {
2260 return _dispatch_barrier_async_f_slow(dq, ctxt, func, pp, flags);
2261 }
2262
2263 dc->do_vtable = (void *)(DISPATCH_OBJ_ASYNC_BIT | DISPATCH_OBJ_BARRIER_BIT);
2264 dc->dc_func = func;
2265 dc->dc_ctxt = ctxt;
2266 _dispatch_continuation_voucher_set(dc, flags);
2267 _dispatch_continuation_priority_set(dc, pp, flags);
2268
2269 pp = _dispatch_continuation_get_override_priority(dq, dc);
2270
2271 _dispatch_queue_push(dq, dc, pp);
2272 }
2273
2274 DISPATCH_NOINLINE
2275 static void
_dispatch_barrier_async_f(dispatch_queue_t dq,void * ctxt,dispatch_function_t func,pthread_priority_t pp,dispatch_block_flags_t flags)2276 _dispatch_barrier_async_f(dispatch_queue_t dq, void *ctxt,
2277 dispatch_function_t func, pthread_priority_t pp,
2278 dispatch_block_flags_t flags)
2279 {
2280 return _dispatch_barrier_async_f2(dq, ctxt, func, pp, flags);
2281 }
2282
2283 DISPATCH_NOINLINE
2284 void
dispatch_barrier_async_f(dispatch_queue_t dq,void * ctxt,dispatch_function_t func)2285 dispatch_barrier_async_f(dispatch_queue_t dq, void *ctxt,
2286 dispatch_function_t func)
2287 {
2288 return _dispatch_barrier_async_f2(dq, ctxt, func, 0, 0);
2289 }
2290
2291 DISPATCH_NOINLINE
2292 void
_dispatch_barrier_async_detached_f(dispatch_queue_t dq,void * ctxt,dispatch_function_t func)2293 _dispatch_barrier_async_detached_f(dispatch_queue_t dq, void *ctxt,
2294 dispatch_function_t func)
2295 {
2296 return _dispatch_barrier_async_f2(dq, ctxt, func, 0,
2297 DISPATCH_BLOCK_NO_QOS_CLASS|DISPATCH_BLOCK_NO_VOUCHER);
2298 }
2299
2300 #ifdef __BLOCKS__
2301 void
2302 dispatch_barrier_async(dispatch_queue_t dq, void (^work)(void))
2303 {
2304 dispatch_function_t func = _dispatch_call_block_and_release;
2305 pthread_priority_t pp = 0;
2306 dispatch_block_flags_t flags = 0;
2307 if (slowpath(_dispatch_block_has_private_data(work))) {
2308 func = _dispatch_block_async_invoke_and_release;
2309 pp = _dispatch_block_get_priority(work);
2310 flags = _dispatch_block_get_flags(work);
2311 // balanced in d_block_async_invoke_and_release or d_block_wait
2312 if (dispatch_atomic_cmpxchg2o(_dispatch_block_get_data(work),
2313 dbpd_queue, NULL, dq, release)) {
2314 _dispatch_retain(dq);
2315 }
2316 }
2317 _dispatch_barrier_async_f(dq, _dispatch_Block_copy(work), func, pp, flags);
2318 }
2319 #endif
2320
2321 #pragma mark -
2322 #pragma mark dispatch_async
2323
2324 void
_dispatch_async_redirect_invoke(void * ctxt)2325 _dispatch_async_redirect_invoke(void *ctxt)
2326 {
2327 struct dispatch_continuation_s *dc = ctxt;
2328 struct dispatch_continuation_s *other_dc = dc->dc_other;
2329 dispatch_queue_t old_dq, dq = dc->dc_data, rq;
2330
2331 old_dq = _dispatch_thread_getspecific(dispatch_queue_key);
2332 _dispatch_thread_setspecific(dispatch_queue_key, dq);
2333 pthread_priority_t old_dp = _dispatch_set_defaultpriority(dq->dq_priority);
2334 _dispatch_continuation_pop(other_dc);
2335 _dispatch_reset_defaultpriority(old_dp);
2336 _dispatch_thread_setspecific(dispatch_queue_key, old_dq);
2337
2338 rq = dq->do_targetq;
2339 while (slowpath(rq->do_targetq) && rq != old_dq) {
2340 if (dispatch_atomic_sub2o(rq, dq_running, 2, relaxed) == 0) {
2341 _dispatch_queue_wakeup(rq);
2342 }
2343 rq = rq->do_targetq;
2344 }
2345
2346 if (dispatch_atomic_sub2o(dq, dq_running, 2, relaxed) == 0) {
2347 _dispatch_queue_wakeup(dq);
2348 }
2349 _dispatch_release(dq);
2350 }
2351
2352 static inline void
_dispatch_async_f_redirect2(dispatch_queue_t dq,dispatch_continuation_t dc,pthread_priority_t pp)2353 _dispatch_async_f_redirect2(dispatch_queue_t dq, dispatch_continuation_t dc,
2354 pthread_priority_t pp)
2355 {
2356 uint32_t running = 2;
2357
2358 // Find the queue to redirect to
2359 do {
2360 if (slowpath(dq->dq_items_tail) ||
2361 slowpath(DISPATCH_OBJECT_SUSPENDED(dq)) ||
2362 slowpath(dq->dq_width == 1)) {
2363 break;
2364 }
2365 running = dispatch_atomic_add2o(dq, dq_running, 2, relaxed);
2366 if (slowpath(running & 1) || slowpath(running > dq->dq_width)) {
2367 running = dispatch_atomic_sub2o(dq, dq_running, 2, relaxed);
2368 break;
2369 }
2370 dq = dq->do_targetq;
2371 } while (slowpath(dq->do_targetq));
2372
2373 _dispatch_queue_push_wakeup(dq, dc, pp, running == 0);
2374 }
2375
2376 DISPATCH_NOINLINE
2377 static void
_dispatch_async_f_redirect(dispatch_queue_t dq,dispatch_continuation_t other_dc,pthread_priority_t pp)2378 _dispatch_async_f_redirect(dispatch_queue_t dq,
2379 dispatch_continuation_t other_dc, pthread_priority_t pp)
2380 {
2381 dispatch_continuation_t dc = _dispatch_continuation_alloc();
2382
2383 dc->do_vtable = (void *)DISPATCH_OBJ_ASYNC_BIT;
2384 dc->dc_func = _dispatch_async_redirect_invoke;
2385 dc->dc_ctxt = dc;
2386 dc->dc_data = dq;
2387 dc->dc_other = other_dc;
2388 dc->dc_priority = 0;
2389 dc->dc_voucher = NULL;
2390
2391 _dispatch_retain(dq);
2392 dq = dq->do_targetq;
2393 if (slowpath(dq->do_targetq)) {
2394 return _dispatch_async_f_redirect2(dq, dc, pp);
2395 }
2396
2397 _dispatch_queue_push(dq, dc, pp);
2398 }
2399
2400 DISPATCH_NOINLINE
2401 static void
_dispatch_async_f2(dispatch_queue_t dq,dispatch_continuation_t dc,pthread_priority_t pp)2402 _dispatch_async_f2(dispatch_queue_t dq, dispatch_continuation_t dc,
2403 pthread_priority_t pp)
2404 {
2405 uint32_t running = 2;
2406
2407 do {
2408 if (slowpath(dq->dq_items_tail)
2409 || slowpath(DISPATCH_OBJECT_SUSPENDED(dq))) {
2410 break;
2411 }
2412 running = dispatch_atomic_add2o(dq, dq_running, 2, relaxed);
2413 if (slowpath(running > dq->dq_width)) {
2414 running = dispatch_atomic_sub2o(dq, dq_running, 2, relaxed);
2415 break;
2416 }
2417 if (!slowpath(running & 1)) {
2418 return _dispatch_async_f_redirect(dq, dc, pp);
2419 }
2420 running = dispatch_atomic_sub2o(dq, dq_running, 2, relaxed);
2421 // We might get lucky and find that the barrier has ended by now
2422 } while (!(running & 1));
2423
2424 _dispatch_queue_push_wakeup(dq, dc, pp, running == 0);
2425 }
2426
2427 DISPATCH_NOINLINE
2428 static void
_dispatch_async_f_slow(dispatch_queue_t dq,void * ctxt,dispatch_function_t func,pthread_priority_t pp,dispatch_block_flags_t flags)2429 _dispatch_async_f_slow(dispatch_queue_t dq, void *ctxt,
2430 dispatch_function_t func, pthread_priority_t pp,
2431 dispatch_block_flags_t flags)
2432 {
2433 dispatch_continuation_t dc = _dispatch_continuation_alloc_from_heap();
2434
2435 dc->do_vtable = (void *)DISPATCH_OBJ_ASYNC_BIT;
2436 dc->dc_func = func;
2437 dc->dc_ctxt = ctxt;
2438 _dispatch_continuation_voucher_set(dc, flags);
2439 _dispatch_continuation_priority_set(dc, pp, flags);
2440
2441 pp = _dispatch_continuation_get_override_priority(dq, dc);
2442
2443 // No fastpath/slowpath hint because we simply don't know
2444 if (dq->do_targetq) {
2445 return _dispatch_async_f2(dq, dc, pp);
2446 }
2447
2448 _dispatch_queue_push(dq, dc, pp);
2449 }
2450
2451 DISPATCH_ALWAYS_INLINE
2452 static inline void
_dispatch_async_f(dispatch_queue_t dq,void * ctxt,dispatch_function_t func,pthread_priority_t pp,dispatch_block_flags_t flags)2453 _dispatch_async_f(dispatch_queue_t dq, void *ctxt, dispatch_function_t func,
2454 pthread_priority_t pp, dispatch_block_flags_t flags)
2455 {
2456 dispatch_continuation_t dc;
2457
2458 // No fastpath/slowpath hint because we simply don't know
2459 if (dq->dq_width == 1 || flags & DISPATCH_BLOCK_BARRIER) {
2460 return _dispatch_barrier_async_f(dq, ctxt, func, pp, flags);
2461 }
2462
2463 dc = fastpath(_dispatch_continuation_alloc_cacheonly());
2464 if (!dc) {
2465 return _dispatch_async_f_slow(dq, ctxt, func, pp, flags);
2466 }
2467
2468 dc->do_vtable = (void *)DISPATCH_OBJ_ASYNC_BIT;
2469 dc->dc_func = func;
2470 dc->dc_ctxt = ctxt;
2471 _dispatch_continuation_voucher_set(dc, flags);
2472 _dispatch_continuation_priority_set(dc, pp, flags);
2473
2474 pp = _dispatch_continuation_get_override_priority(dq, dc);
2475
2476 // No fastpath/slowpath hint because we simply don't know
2477 if (dq->do_targetq) {
2478 return _dispatch_async_f2(dq, dc, pp);
2479 }
2480
2481 _dispatch_queue_push(dq, dc, pp);
2482 }
2483
2484 DISPATCH_NOINLINE
2485 void
dispatch_async_f(dispatch_queue_t dq,void * ctxt,dispatch_function_t func)2486 dispatch_async_f(dispatch_queue_t dq, void *ctxt, dispatch_function_t func)
2487 {
2488 return _dispatch_async_f(dq, ctxt, func, 0, 0);
2489 }
2490
2491 #ifdef __BLOCKS__
2492 void
2493 dispatch_async(dispatch_queue_t dq, void (^work)(void))
2494 {
2495 dispatch_function_t func = _dispatch_call_block_and_release;
2496 dispatch_block_flags_t flags = 0;
2497 pthread_priority_t pp = 0;
2498 if (slowpath(_dispatch_block_has_private_data(work))) {
2499 func = _dispatch_block_async_invoke_and_release;
2500 pp = _dispatch_block_get_priority(work);
2501 flags = _dispatch_block_get_flags(work);
2502 // balanced in d_block_async_invoke_and_release or d_block_wait
2503 if (dispatch_atomic_cmpxchg2o(_dispatch_block_get_data(work),
2504 dbpd_queue, NULL, dq, release)) {
2505 _dispatch_retain(dq);
2506 }
2507 }
2508 _dispatch_async_f(dq, _dispatch_Block_copy(work), func, pp, flags);
2509 }
2510 #endif
2511
2512 #pragma mark -
2513 #pragma mark dispatch_group_async
2514
2515 DISPATCH_ALWAYS_INLINE
2516 static inline void
_dispatch_group_async_f(dispatch_group_t dg,dispatch_queue_t dq,void * ctxt,dispatch_function_t func,pthread_priority_t pp,dispatch_block_flags_t flags)2517 _dispatch_group_async_f(dispatch_group_t dg, dispatch_queue_t dq, void *ctxt,
2518 dispatch_function_t func, pthread_priority_t pp,
2519 dispatch_block_flags_t flags)
2520 {
2521 dispatch_continuation_t dc;
2522
2523 _dispatch_retain(dg);
2524 dispatch_group_enter(dg);
2525
2526 dc = _dispatch_continuation_alloc();
2527
2528 unsigned long barrier = (flags & DISPATCH_BLOCK_BARRIER) ?
2529 DISPATCH_OBJ_BARRIER_BIT : 0;
2530 dc->do_vtable = (void *)(DISPATCH_OBJ_ASYNC_BIT | DISPATCH_OBJ_GROUP_BIT |
2531 barrier);
2532 dc->dc_func = func;
2533 dc->dc_ctxt = ctxt;
2534 dc->dc_data = dg;
2535 _dispatch_continuation_voucher_set(dc, flags);
2536 _dispatch_continuation_priority_set(dc, pp, flags);
2537
2538 pp = _dispatch_continuation_get_override_priority(dq, dc);
2539
2540 // No fastpath/slowpath hint because we simply don't know
2541 if (dq->dq_width != 1 && !barrier && dq->do_targetq) {
2542 return _dispatch_async_f2(dq, dc, pp);
2543 }
2544
2545 _dispatch_queue_push(dq, dc, pp);
2546 }
2547
2548 DISPATCH_NOINLINE
2549 void
dispatch_group_async_f(dispatch_group_t dg,dispatch_queue_t dq,void * ctxt,dispatch_function_t func)2550 dispatch_group_async_f(dispatch_group_t dg, dispatch_queue_t dq, void *ctxt,
2551 dispatch_function_t func)
2552 {
2553 return _dispatch_group_async_f(dg, dq, ctxt, func, 0, 0);
2554 }
2555
2556 #ifdef __BLOCKS__
2557 void
dispatch_group_async(dispatch_group_t dg,dispatch_queue_t dq,dispatch_block_t db)2558 dispatch_group_async(dispatch_group_t dg, dispatch_queue_t dq,
2559 dispatch_block_t db)
2560 {
2561 dispatch_function_t func = _dispatch_call_block_and_release;
2562 dispatch_block_flags_t flags = 0;
2563 pthread_priority_t pp = 0;
2564 if (slowpath(_dispatch_block_has_private_data(db))) {
2565 func = _dispatch_block_async_invoke_and_release;
2566 pp = _dispatch_block_get_priority(db);
2567 flags = _dispatch_block_get_flags(db);
2568 // balanced in d_block_async_invoke_and_release or d_block_wait
2569 if (dispatch_atomic_cmpxchg2o(_dispatch_block_get_data(db),
2570 dbpd_queue, NULL, dq, release)) {
2571 _dispatch_retain(dq);
2572 }
2573 }
2574 _dispatch_group_async_f(dg, dq, _dispatch_Block_copy(db), func, pp, flags);
2575 }
2576 #endif
2577
2578 #pragma mark -
2579 #pragma mark dispatch_function_invoke
2580
2581 static void _dispatch_sync_f(dispatch_queue_t dq, void *ctxt,
2582 dispatch_function_t func, pthread_priority_t pp);
2583
2584 DISPATCH_ALWAYS_INLINE
2585 static inline void
_dispatch_function_invoke(dispatch_queue_t dq,void * ctxt,dispatch_function_t func)2586 _dispatch_function_invoke(dispatch_queue_t dq, void *ctxt,
2587 dispatch_function_t func)
2588 {
2589 dispatch_queue_t old_dq = _dispatch_thread_getspecific(dispatch_queue_key);
2590 _dispatch_thread_setspecific(dispatch_queue_key, dq);
2591 _dispatch_client_callout(ctxt, func);
2592 _dispatch_perfmon_workitem_inc();
2593 _dispatch_thread_setspecific(dispatch_queue_key, old_dq);
2594 }
2595
2596 void
_dispatch_sync_recurse_invoke(void * ctxt)2597 _dispatch_sync_recurse_invoke(void *ctxt)
2598 {
2599 dispatch_continuation_t dc = ctxt;
2600 _dispatch_function_invoke(dc->dc_data, dc->dc_ctxt, dc->dc_func);
2601 }
2602
2603 DISPATCH_ALWAYS_INLINE
2604 static inline void
_dispatch_function_recurse(dispatch_queue_t dq,void * ctxt,dispatch_function_t func,pthread_priority_t pp)2605 _dispatch_function_recurse(dispatch_queue_t dq, void *ctxt,
2606 dispatch_function_t func, pthread_priority_t pp)
2607 {
2608 struct dispatch_continuation_s dc = {
2609 .dc_data = dq,
2610 .dc_func = func,
2611 .dc_ctxt = ctxt,
2612 };
2613 _dispatch_sync_f(dq->do_targetq, &dc, _dispatch_sync_recurse_invoke, pp);
2614 }
2615
2616 #pragma mark -
2617 #pragma mark dispatch_barrier_sync
2618
2619 static void _dispatch_sync_f_invoke(dispatch_queue_t dq, void *ctxt,
2620 dispatch_function_t func);
2621
2622 DISPATCH_ALWAYS_INLINE_NDEBUG
2623 static inline _dispatch_thread_semaphore_t
_dispatch_barrier_sync_f_pop(dispatch_queue_t dq,dispatch_object_t dou,bool lock)2624 _dispatch_barrier_sync_f_pop(dispatch_queue_t dq, dispatch_object_t dou,
2625 bool lock)
2626 {
2627 _dispatch_thread_semaphore_t sema;
2628 dispatch_continuation_t dc = dou._dc;
2629 mach_port_t th;
2630
2631 if (DISPATCH_OBJ_IS_VTABLE(dc) || ((long)dc->do_vtable &
2632 (DISPATCH_OBJ_BARRIER_BIT | DISPATCH_OBJ_SYNC_SLOW_BIT)) !=
2633 (DISPATCH_OBJ_BARRIER_BIT | DISPATCH_OBJ_SYNC_SLOW_BIT)) {
2634 return 0;
2635 }
2636 _dispatch_trace_continuation_pop(dq, dc);
2637 _dispatch_perfmon_workitem_inc();
2638
2639 th = (mach_port_t)dc->dc_data;
2640 dc = dc->dc_ctxt;
2641 dq = dc->dc_data;
2642 sema = (_dispatch_thread_semaphore_t)dc->dc_other;
2643 if (lock) {
2644 (void)dispatch_atomic_add2o(dq, do_suspend_cnt,
2645 DISPATCH_OBJECT_SUSPEND_INTERVAL, relaxed);
2646 // rdar://problem/9032024 running lock must be held until sync_f_slow
2647 // returns
2648 (void)dispatch_atomic_add2o(dq, dq_running, 2, relaxed);
2649 }
2650 _dispatch_introspection_queue_item_complete(dou);
2651 _dispatch_wqthread_override_start(th,
2652 _dispatch_queue_get_override_priority(dq));
2653 return sema ? sema : MACH_PORT_DEAD;
2654 }
2655
2656 static void
_dispatch_barrier_sync_f_slow_invoke(void * ctxt)2657 _dispatch_barrier_sync_f_slow_invoke(void *ctxt)
2658 {
2659 dispatch_continuation_t dc = ctxt;
2660 dispatch_queue_t dq = dc->dc_data;
2661 _dispatch_thread_semaphore_t sema;
2662 sema = (_dispatch_thread_semaphore_t)dc->dc_other;
2663
2664 dispatch_assert(dq == _dispatch_queue_get_current());
2665 #if DISPATCH_COCOA_COMPAT
2666 if (slowpath(dq->dq_is_thread_bound)) {
2667 // The queue is bound to a non-dispatch thread (e.g. main thread)
2668 _dispatch_continuation_voucher_adopt(dc);
2669 _dispatch_client_callout(dc->dc_ctxt, dc->dc_func);
2670 dispatch_atomic_store2o(dc, dc_func, NULL, release);
2671 _dispatch_thread_semaphore_signal(sema); // release
2672 return;
2673 }
2674 #endif
2675 (void)dispatch_atomic_add2o(dq, do_suspend_cnt,
2676 DISPATCH_OBJECT_SUSPEND_INTERVAL, relaxed);
2677 // rdar://9032024 running lock must be held until sync_f_slow returns
2678 (void)dispatch_atomic_add2o(dq, dq_running, 2, relaxed);
2679 _dispatch_thread_semaphore_signal(sema); // release
2680 }
2681
2682 DISPATCH_NOINLINE
2683 static void
_dispatch_barrier_sync_f_slow(dispatch_queue_t dq,void * ctxt,dispatch_function_t func,pthread_priority_t pp)2684 _dispatch_barrier_sync_f_slow(dispatch_queue_t dq, void *ctxt,
2685 dispatch_function_t func, pthread_priority_t pp)
2686 {
2687 if (slowpath(!dq->do_targetq)) {
2688 // the global concurrent queues do not need strict ordering
2689 (void)dispatch_atomic_add2o(dq, dq_running, 2, relaxed);
2690 return _dispatch_sync_f_invoke(dq, ctxt, func);
2691 }
2692 if (!pp) pp = (_dispatch_get_priority() | _PTHREAD_PRIORITY_ENFORCE_FLAG);
2693 _dispatch_thread_semaphore_t sema = _dispatch_get_thread_semaphore();
2694 struct dispatch_continuation_s dc = {
2695 .dc_data = dq,
2696 #if DISPATCH_COCOA_COMPAT
2697 .dc_func = func,
2698 .dc_ctxt = ctxt,
2699 #endif
2700 .dc_other = (void*)sema,
2701 };
2702 #if DISPATCH_COCOA_COMPAT
2703 // It's preferred to execute synchronous blocks on the current thread
2704 // due to thread-local side effects, garbage collection, etc. However,
2705 // blocks submitted to the main thread MUST be run on the main thread
2706 if (slowpath(dq->dq_is_thread_bound)) {
2707 _dispatch_continuation_voucher_set(&dc, 0);
2708 }
2709 #endif
2710 struct dispatch_continuation_s dbss = {
2711 .do_vtable = (void *)(DISPATCH_OBJ_BARRIER_BIT |
2712 DISPATCH_OBJ_SYNC_SLOW_BIT),
2713 .dc_func = _dispatch_barrier_sync_f_slow_invoke,
2714 .dc_ctxt = &dc,
2715 .dc_data = (void*)(uintptr_t)_dispatch_thread_port(),
2716 .dc_priority = pp,
2717 };
2718 _dispatch_queue_push(dq, &dbss,
2719 _dispatch_continuation_get_override_priority(dq, &dbss));
2720
2721 _dispatch_thread_semaphore_wait(sema); // acquire
2722 _dispatch_put_thread_semaphore(sema);
2723
2724 #if DISPATCH_COCOA_COMPAT
2725 // Queue bound to a non-dispatch thread
2726 if (dc.dc_func == NULL) {
2727 return;
2728 }
2729 #endif
2730
2731 _dispatch_queue_set_thread(dq);
2732 if (slowpath(dq->do_targetq->do_targetq)) {
2733 _dispatch_function_recurse(dq, ctxt, func, pp);
2734 } else {
2735 _dispatch_function_invoke(dq, ctxt, func);
2736 }
2737 _dispatch_queue_clear_thread(dq);
2738
2739 if (fastpath(dq->do_suspend_cnt < 2 * DISPATCH_OBJECT_SUSPEND_INTERVAL) &&
2740 dq->dq_running == 2) {
2741 // rdar://problem/8290662 "lock transfer"
2742 sema = _dispatch_queue_drain_one_barrier_sync(dq);
2743 if (sema) {
2744 _dispatch_thread_semaphore_signal(sema); // release
2745 return;
2746 }
2747 }
2748 (void)dispatch_atomic_sub2o(dq, do_suspend_cnt,
2749 DISPATCH_OBJECT_SUSPEND_INTERVAL, release);
2750 if (slowpath(dispatch_atomic_sub2o(dq, dq_running, 2, relaxed) == 0)) {
2751 _dispatch_queue_wakeup(dq);
2752 }
2753 }
2754
2755 DISPATCH_NOINLINE
2756 static void
_dispatch_barrier_sync_f2(dispatch_queue_t dq)2757 _dispatch_barrier_sync_f2(dispatch_queue_t dq)
2758 {
2759 if (!slowpath(DISPATCH_OBJECT_SUSPENDED(dq))) {
2760 // rdar://problem/8290662 "lock transfer"
2761 _dispatch_thread_semaphore_t sema;
2762 sema = _dispatch_queue_drain_one_barrier_sync(dq);
2763 if (sema) {
2764 (void)dispatch_atomic_add2o(dq, do_suspend_cnt,
2765 DISPATCH_OBJECT_SUSPEND_INTERVAL, relaxed);
2766 // rdar://9032024 running lock must be held until sync_f_slow
2767 // returns: increment by 2 and decrement by 1
2768 (void)dispatch_atomic_inc2o(dq, dq_running, relaxed);
2769 _dispatch_thread_semaphore_signal(sema);
2770 return;
2771 }
2772 }
2773 if (slowpath(dispatch_atomic_dec2o(dq, dq_running, release) == 0)) {
2774 _dispatch_queue_wakeup(dq);
2775 }
2776 }
2777
2778 DISPATCH_NOINLINE
2779 static void
_dispatch_barrier_sync_f_invoke(dispatch_queue_t dq,void * ctxt,dispatch_function_t func)2780 _dispatch_barrier_sync_f_invoke(dispatch_queue_t dq, void *ctxt,
2781 dispatch_function_t func)
2782 {
2783 _dispatch_queue_set_thread(dq);
2784 _dispatch_function_invoke(dq, ctxt, func);
2785 _dispatch_queue_clear_thread(dq);
2786 if (slowpath(dq->dq_items_tail)) {
2787 return _dispatch_barrier_sync_f2(dq);
2788 }
2789 if (slowpath(dispatch_atomic_dec2o(dq, dq_running, release) == 0)) {
2790 _dispatch_queue_wakeup(dq);
2791 }
2792 }
2793
2794 DISPATCH_NOINLINE
2795 static void
_dispatch_barrier_sync_f_recurse(dispatch_queue_t dq,void * ctxt,dispatch_function_t func,pthread_priority_t pp)2796 _dispatch_barrier_sync_f_recurse(dispatch_queue_t dq, void *ctxt,
2797 dispatch_function_t func, pthread_priority_t pp)
2798 {
2799 _dispatch_queue_set_thread(dq);
2800 _dispatch_function_recurse(dq, ctxt, func, pp);
2801 _dispatch_queue_clear_thread(dq);
2802 if (slowpath(dq->dq_items_tail)) {
2803 return _dispatch_barrier_sync_f2(dq);
2804 }
2805 if (slowpath(dispatch_atomic_dec2o(dq, dq_running, release) == 0)) {
2806 _dispatch_queue_wakeup(dq);
2807 }
2808 }
2809
2810 DISPATCH_NOINLINE
2811 static void
_dispatch_barrier_sync_f(dispatch_queue_t dq,void * ctxt,dispatch_function_t func,pthread_priority_t pp)2812 _dispatch_barrier_sync_f(dispatch_queue_t dq, void *ctxt,
2813 dispatch_function_t func, pthread_priority_t pp)
2814 {
2815 // 1) ensure that this thread hasn't enqueued anything ahead of this call
2816 // 2) the queue is not suspended
2817 if (slowpath(dq->dq_items_tail) || slowpath(DISPATCH_OBJECT_SUSPENDED(dq))){
2818 return _dispatch_barrier_sync_f_slow(dq, ctxt, func, pp);
2819 }
2820 if (slowpath(!dispatch_atomic_cmpxchg2o(dq, dq_running, 0, 1, acquire))) {
2821 // global concurrent queues and queues bound to non-dispatch threads
2822 // always fall into the slow case
2823 return _dispatch_barrier_sync_f_slow(dq, ctxt, func, pp);
2824 }
2825 if (slowpath(dq->do_targetq->do_targetq)) {
2826 return _dispatch_barrier_sync_f_recurse(dq, ctxt, func, pp);
2827 }
2828 _dispatch_barrier_sync_f_invoke(dq, ctxt, func);
2829 }
2830
2831 DISPATCH_NOINLINE
2832 void
dispatch_barrier_sync_f(dispatch_queue_t dq,void * ctxt,dispatch_function_t func)2833 dispatch_barrier_sync_f(dispatch_queue_t dq, void *ctxt,
2834 dispatch_function_t func)
2835 {
2836 // 1) ensure that this thread hasn't enqueued anything ahead of this call
2837 // 2) the queue is not suspended
2838 if (slowpath(dq->dq_items_tail) || slowpath(DISPATCH_OBJECT_SUSPENDED(dq))){
2839 return _dispatch_barrier_sync_f_slow(dq, ctxt, func, 0);
2840 }
2841 if (slowpath(!dispatch_atomic_cmpxchg2o(dq, dq_running, 0, 1, acquire))) {
2842 // global concurrent queues and queues bound to non-dispatch threads
2843 // always fall into the slow case
2844 return _dispatch_barrier_sync_f_slow(dq, ctxt, func, 0);
2845 }
2846 if (slowpath(dq->do_targetq->do_targetq)) {
2847 return _dispatch_barrier_sync_f_recurse(dq, ctxt, func, 0);
2848 }
2849 _dispatch_barrier_sync_f_invoke(dq, ctxt, func);
2850 }
2851
2852 #ifdef __BLOCKS__
2853 DISPATCH_NOINLINE
2854 static void
2855 _dispatch_barrier_sync_slow(dispatch_queue_t dq, void (^work)(void))
2856 {
2857 bool has_pd = _dispatch_block_has_private_data(work);
2858 dispatch_function_t func = _dispatch_Block_invoke(work);
2859 pthread_priority_t pp = 0;
2860 if (has_pd) {
2861 func = _dispatch_block_sync_invoke;
2862 pp = _dispatch_block_get_priority(work);
2863 dispatch_block_flags_t flags = _dispatch_block_get_flags(work);
2864 if (flags & DISPATCH_BLOCK_HAS_PRIORITY) {
2865 pthread_priority_t tp = _dispatch_get_priority();
2866 if (pp < tp) {
2867 pp = tp | _PTHREAD_PRIORITY_ENFORCE_FLAG;
2868 } else if (!(flags & DISPATCH_BLOCK_INHERIT_QOS_CLASS)) {
2869 pp |= _PTHREAD_PRIORITY_ENFORCE_FLAG;
2870 }
2871 }
2872 // balanced in d_block_sync_invoke or d_block_wait
2873 if (dispatch_atomic_cmpxchg2o(_dispatch_block_get_data(work),
2874 dbpd_queue, NULL, dq, release)) {
2875 _dispatch_retain(dq);
2876 }
2877 #if DISPATCH_COCOA_COMPAT
2878 } else if (dq->dq_is_thread_bound && dispatch_begin_thread_4GC) {
2879 // Blocks submitted to the main queue MUST be run on the main thread,
2880 // under GC we must Block_copy in order to notify the thread-local
2881 // garbage collector that the objects are transferring to another thread
2882 // rdar://problem/7176237&7181849&7458685
2883 work = _dispatch_Block_copy(work);
2884 func = _dispatch_call_block_and_release;
2885 }
2886 #endif
2887 _dispatch_barrier_sync_f(dq, work, func, pp);
2888 }
2889
2890 void
2891 dispatch_barrier_sync(dispatch_queue_t dq, void (^work)(void))
2892 {
2893 if (slowpath(dq->dq_is_thread_bound) ||
2894 slowpath(_dispatch_block_has_private_data(work))) {
2895 return _dispatch_barrier_sync_slow(dq, work);
2896 }
2897 dispatch_barrier_sync_f(dq, work, _dispatch_Block_invoke(work));
2898 }
2899 #endif
2900
2901 DISPATCH_NOINLINE
2902 static void
_dispatch_barrier_trysync_f_invoke(dispatch_queue_t dq,void * ctxt,dispatch_function_t func)2903 _dispatch_barrier_trysync_f_invoke(dispatch_queue_t dq, void *ctxt,
2904 dispatch_function_t func)
2905 {
2906 _dispatch_queue_set_thread(dq);
2907 _dispatch_function_invoke(dq, ctxt, func);
2908 _dispatch_queue_clear_thread(dq);
2909 if (slowpath(dispatch_atomic_dec2o(dq, dq_running, release) == 0)) {
2910 _dispatch_queue_wakeup(dq);
2911 }
2912 }
2913
2914 DISPATCH_NOINLINE
2915 void
_dispatch_barrier_trysync_f(dispatch_queue_t dq,void * ctxt,dispatch_function_t func)2916 _dispatch_barrier_trysync_f(dispatch_queue_t dq, void *ctxt,
2917 dispatch_function_t func)
2918 {
2919 // Use for mutation of queue-/source-internal state only, ignores target
2920 // queue hierarchy!
2921 if (slowpath(dq->dq_items_tail) || slowpath(DISPATCH_OBJECT_SUSPENDED(dq))
2922 || slowpath(!dispatch_atomic_cmpxchg2o(dq, dq_running, 0, 1,
2923 acquire))) {
2924 return _dispatch_barrier_async_detached_f(dq, ctxt, func);
2925 }
2926 _dispatch_barrier_trysync_f_invoke(dq, ctxt, func);
2927 }
2928
2929 #pragma mark -
2930 #pragma mark dispatch_sync
2931
2932 DISPATCH_NOINLINE
2933 static void
_dispatch_sync_f_slow(dispatch_queue_t dq,void * ctxt,dispatch_function_t func,pthread_priority_t pp,bool wakeup)2934 _dispatch_sync_f_slow(dispatch_queue_t dq, void *ctxt, dispatch_function_t func,
2935 pthread_priority_t pp, bool wakeup)
2936 {
2937 if (!pp) pp = (_dispatch_get_priority() | _PTHREAD_PRIORITY_ENFORCE_FLAG);
2938 _dispatch_thread_semaphore_t sema = _dispatch_get_thread_semaphore();
2939 struct dispatch_continuation_s dc = {
2940 .do_vtable = (void*)DISPATCH_OBJ_SYNC_SLOW_BIT,
2941 #if DISPATCH_INTROSPECTION
2942 .dc_func = func,
2943 .dc_ctxt = ctxt,
2944 .dc_data = (void*)(uintptr_t)_dispatch_thread_port(),
2945 #endif
2946 .dc_other = (void*)sema,
2947 .dc_priority = pp,
2948 };
2949 _dispatch_queue_push_wakeup(dq, &dc,
2950 _dispatch_continuation_get_override_priority(dq, &dc), wakeup);
2951
2952 _dispatch_thread_semaphore_wait(sema);
2953 _dispatch_put_thread_semaphore(sema);
2954
2955 if (slowpath(dq->do_targetq->do_targetq)) {
2956 _dispatch_function_recurse(dq, ctxt, func, pp);
2957 } else {
2958 _dispatch_function_invoke(dq, ctxt, func);
2959 }
2960
2961 if (slowpath(dispatch_atomic_sub2o(dq, dq_running, 2, relaxed) == 0)) {
2962 _dispatch_queue_wakeup(dq);
2963 }
2964 }
2965
2966 DISPATCH_NOINLINE
2967 static void
_dispatch_sync_f_invoke(dispatch_queue_t dq,void * ctxt,dispatch_function_t func)2968 _dispatch_sync_f_invoke(dispatch_queue_t dq, void *ctxt,
2969 dispatch_function_t func)
2970 {
2971 _dispatch_function_invoke(dq, ctxt, func);
2972 if (slowpath(dispatch_atomic_sub2o(dq, dq_running, 2, relaxed) == 0)) {
2973 _dispatch_queue_wakeup(dq);
2974 }
2975 }
2976
2977 DISPATCH_NOINLINE
2978 static void
_dispatch_sync_f_recurse(dispatch_queue_t dq,void * ctxt,dispatch_function_t func,pthread_priority_t pp)2979 _dispatch_sync_f_recurse(dispatch_queue_t dq, void *ctxt,
2980 dispatch_function_t func, pthread_priority_t pp)
2981 {
2982 _dispatch_function_recurse(dq, ctxt, func, pp);
2983 if (slowpath(dispatch_atomic_sub2o(dq, dq_running, 2, relaxed) == 0)) {
2984 _dispatch_queue_wakeup(dq);
2985 }
2986 }
2987
2988 static inline void
_dispatch_sync_f2(dispatch_queue_t dq,void * ctxt,dispatch_function_t func,pthread_priority_t pp)2989 _dispatch_sync_f2(dispatch_queue_t dq, void *ctxt, dispatch_function_t func,
2990 pthread_priority_t pp)
2991 {
2992 // 1) ensure that this thread hasn't enqueued anything ahead of this call
2993 // 2) the queue is not suspended
2994 if (slowpath(dq->dq_items_tail) || slowpath(DISPATCH_OBJECT_SUSPENDED(dq))){
2995 return _dispatch_sync_f_slow(dq, ctxt, func, pp, false);
2996 }
2997 uint32_t running = dispatch_atomic_add2o(dq, dq_running, 2, relaxed);
2998 // re-check suspension after barrier check <rdar://problem/15242126>
2999 if (slowpath(running & 1) || _dispatch_object_suspended(dq)) {
3000 running = dispatch_atomic_sub2o(dq, dq_running, 2, relaxed);
3001 return _dispatch_sync_f_slow(dq, ctxt, func, pp, running == 0);
3002 }
3003 if (slowpath(dq->do_targetq->do_targetq)) {
3004 return _dispatch_sync_f_recurse(dq, ctxt, func, pp);
3005 }
3006 _dispatch_sync_f_invoke(dq, ctxt, func);
3007 }
3008
3009 DISPATCH_NOINLINE
3010 static void
_dispatch_sync_f(dispatch_queue_t dq,void * ctxt,dispatch_function_t func,pthread_priority_t pp)3011 _dispatch_sync_f(dispatch_queue_t dq, void *ctxt, dispatch_function_t func,
3012 pthread_priority_t pp)
3013 {
3014 if (fastpath(dq->dq_width == 1)) {
3015 return _dispatch_barrier_sync_f(dq, ctxt, func, pp);
3016 }
3017 if (slowpath(!dq->do_targetq)) {
3018 // the global concurrent queues do not need strict ordering
3019 (void)dispatch_atomic_add2o(dq, dq_running, 2, relaxed);
3020 return _dispatch_sync_f_invoke(dq, ctxt, func);
3021 }
3022 _dispatch_sync_f2(dq, ctxt, func, pp);
3023 }
3024
3025 DISPATCH_NOINLINE
3026 void
dispatch_sync_f(dispatch_queue_t dq,void * ctxt,dispatch_function_t func)3027 dispatch_sync_f(dispatch_queue_t dq, void *ctxt, dispatch_function_t func)
3028 {
3029 if (fastpath(dq->dq_width == 1)) {
3030 return dispatch_barrier_sync_f(dq, ctxt, func);
3031 }
3032 if (slowpath(!dq->do_targetq)) {
3033 // the global concurrent queues do not need strict ordering
3034 (void)dispatch_atomic_add2o(dq, dq_running, 2, relaxed);
3035 return _dispatch_sync_f_invoke(dq, ctxt, func);
3036 }
3037 _dispatch_sync_f2(dq, ctxt, func, 0);
3038 }
3039
3040 #ifdef __BLOCKS__
3041 DISPATCH_NOINLINE
3042 static void
3043 _dispatch_sync_slow(dispatch_queue_t dq, void (^work)(void))
3044 {
3045 bool has_pd = _dispatch_block_has_private_data(work);
3046 if (has_pd && (_dispatch_block_get_flags(work) & DISPATCH_BLOCK_BARRIER)) {
3047 return _dispatch_barrier_sync_slow(dq, work);
3048 }
3049 dispatch_function_t func = _dispatch_Block_invoke(work);
3050 pthread_priority_t pp = 0;
3051 if (has_pd) {
3052 func = _dispatch_block_sync_invoke;
3053 pp = _dispatch_block_get_priority(work);
3054 dispatch_block_flags_t flags = _dispatch_block_get_flags(work);
3055 if (flags & DISPATCH_BLOCK_HAS_PRIORITY) {
3056 pthread_priority_t tp = _dispatch_get_priority();
3057 if (pp < tp) {
3058 pp = tp | _PTHREAD_PRIORITY_ENFORCE_FLAG;
3059 } else if (!(flags & DISPATCH_BLOCK_INHERIT_QOS_CLASS)) {
3060 pp |= _PTHREAD_PRIORITY_ENFORCE_FLAG;
3061 }
3062 }
3063 // balanced in d_block_sync_invoke or d_block_wait
3064 if (dispatch_atomic_cmpxchg2o(_dispatch_block_get_data(work),
3065 dbpd_queue, NULL, dq, release)) {
3066 _dispatch_retain(dq);
3067 }
3068 #if DISPATCH_COCOA_COMPAT
3069 } else if (dq->dq_is_thread_bound && dispatch_begin_thread_4GC) {
3070 // Blocks submitted to the main queue MUST be run on the main thread,
3071 // under GC we must Block_copy in order to notify the thread-local
3072 // garbage collector that the objects are transferring to another thread
3073 // rdar://problem/7176237&7181849&7458685
3074 work = _dispatch_Block_copy(work);
3075 func = _dispatch_call_block_and_release;
3076 #endif
3077 }
3078 if (slowpath(!dq->do_targetq)) {
3079 // the global concurrent queues do not need strict ordering
3080 (void)dispatch_atomic_add2o(dq, dq_running, 2, relaxed);
3081 return _dispatch_sync_f_invoke(dq, work, func);
3082 }
3083 _dispatch_sync_f2(dq, work, func, pp);
3084 }
3085
3086 void
3087 dispatch_sync(dispatch_queue_t dq, void (^work)(void))
3088 {
3089 if (fastpath(dq->dq_width == 1)) {
3090 return dispatch_barrier_sync(dq, work);
3091 }
3092 if (slowpath(dq->dq_is_thread_bound) ||
3093 slowpath(_dispatch_block_has_private_data(work)) ) {
3094 return _dispatch_sync_slow(dq, work);
3095 }
3096 dispatch_sync_f(dq, work, _dispatch_Block_invoke(work));
3097 }
3098 #endif
3099
3100 #pragma mark -
3101 #pragma mark dispatch_after
3102
3103 void
_dispatch_after_timer_callback(void * ctxt)3104 _dispatch_after_timer_callback(void *ctxt)
3105 {
3106 dispatch_continuation_t dc = ctxt, dc1;
3107 dispatch_source_t ds = dc->dc_data;
3108 dc1 = _dispatch_continuation_free_cacheonly(dc);
3109 _dispatch_client_callout(dc->dc_ctxt, dc->dc_func);
3110 dispatch_source_cancel(ds);
3111 dispatch_release(ds);
3112 if (slowpath(dc1)) {
3113 _dispatch_continuation_free_to_cache_limit(dc1);
3114 }
3115 }
3116
3117 DISPATCH_NOINLINE
3118 void
dispatch_after_f(dispatch_time_t when,dispatch_queue_t queue,void * ctxt,dispatch_function_t func)3119 dispatch_after_f(dispatch_time_t when, dispatch_queue_t queue, void *ctxt,
3120 dispatch_function_t func)
3121 {
3122 uint64_t delta, leeway;
3123 dispatch_source_t ds;
3124
3125 if (when == DISPATCH_TIME_FOREVER) {
3126 #if DISPATCH_DEBUG
3127 DISPATCH_CLIENT_CRASH(
3128 "dispatch_after_f() called with 'when' == infinity");
3129 #endif
3130 return;
3131 }
3132
3133 delta = _dispatch_timeout(when);
3134 if (delta == 0) {
3135 return dispatch_async_f(queue, ctxt, func);
3136 }
3137 leeway = delta / 10; // <rdar://problem/13447496>
3138 if (leeway < NSEC_PER_MSEC) leeway = NSEC_PER_MSEC;
3139 if (leeway > 60 * NSEC_PER_SEC) leeway = 60 * NSEC_PER_SEC;
3140
3141 // this function can and should be optimized to not use a dispatch source
3142 ds = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, 0, queue);
3143 dispatch_assert(ds);
3144
3145 // TODO: don't use a separate continuation & voucher
3146 dispatch_continuation_t dc = _dispatch_continuation_alloc();
3147 dc->do_vtable = (void *)(DISPATCH_OBJ_ASYNC_BIT);
3148 dc->dc_func = func;
3149 dc->dc_ctxt = ctxt;
3150 dc->dc_data = ds;
3151
3152 dispatch_set_context(ds, dc);
3153 dispatch_source_set_event_handler_f(ds, _dispatch_after_timer_callback);
3154 dispatch_source_set_timer(ds, when, DISPATCH_TIME_FOREVER, leeway);
3155 dispatch_resume(ds);
3156 }
3157
3158 #ifdef __BLOCKS__
3159 void
dispatch_after(dispatch_time_t when,dispatch_queue_t queue,dispatch_block_t work)3160 dispatch_after(dispatch_time_t when, dispatch_queue_t queue,
3161 dispatch_block_t work)
3162 {
3163 // test before the copy of the block
3164 if (when == DISPATCH_TIME_FOREVER) {
3165 #if DISPATCH_DEBUG
3166 DISPATCH_CLIENT_CRASH(
3167 "dispatch_after() called with 'when' == infinity");
3168 #endif
3169 return;
3170 }
3171 dispatch_after_f(when, queue, _dispatch_Block_copy(work),
3172 _dispatch_call_block_and_release);
3173 }
3174 #endif
3175
3176 #pragma mark -
3177 #pragma mark dispatch_queue_push
3178
3179 DISPATCH_ALWAYS_INLINE
3180 static inline void
_dispatch_queue_push_list_slow2(dispatch_queue_t dq,pthread_priority_t pp,struct dispatch_object_s * obj,bool retained)3181 _dispatch_queue_push_list_slow2(dispatch_queue_t dq, pthread_priority_t pp,
3182 struct dispatch_object_s *obj, bool retained)
3183 {
3184 // The queue must be retained before dq_items_head is written in order
3185 // to ensure that the reference is still valid when _dispatch_wakeup is
3186 // called. Otherwise, if preempted between the assignment to
3187 // dq_items_head and _dispatch_wakeup, the blocks submitted to the
3188 // queue may release the last reference to the queue when invoked by
3189 // _dispatch_queue_drain. <rdar://problem/6932776>
3190 if (!retained) _dispatch_retain(dq);
3191 dq->dq_items_head = obj;
3192 return _dispatch_queue_wakeup_with_qos_and_release(dq, pp);
3193 }
3194
3195 DISPATCH_NOINLINE
3196 void
_dispatch_queue_push_list_slow(dispatch_queue_t dq,pthread_priority_t pp,struct dispatch_object_s * obj,unsigned int n,bool retained)3197 _dispatch_queue_push_list_slow(dispatch_queue_t dq, pthread_priority_t pp,
3198 struct dispatch_object_s *obj, unsigned int n, bool retained)
3199 {
3200 if (dx_type(dq) == DISPATCH_QUEUE_ROOT_TYPE && !dq->dq_is_thread_bound) {
3201 dispatch_assert(!retained);
3202 dispatch_atomic_store2o(dq, dq_items_head, obj, relaxed);
3203 return _dispatch_queue_wakeup_global2(dq, n);
3204 }
3205 _dispatch_queue_push_list_slow2(dq, pp, obj, retained);
3206 }
3207
3208 DISPATCH_NOINLINE
3209 void
_dispatch_queue_push_slow(dispatch_queue_t dq,pthread_priority_t pp,struct dispatch_object_s * obj,bool retained)3210 _dispatch_queue_push_slow(dispatch_queue_t dq, pthread_priority_t pp,
3211 struct dispatch_object_s *obj, bool retained)
3212 {
3213 if (dx_type(dq) == DISPATCH_QUEUE_ROOT_TYPE && !dq->dq_is_thread_bound) {
3214 dispatch_assert(!retained);
3215 dispatch_atomic_store2o(dq, dq_items_head, obj, relaxed);
3216 return _dispatch_queue_wakeup_global(dq);
3217 }
3218 _dispatch_queue_push_list_slow2(dq, pp, obj, retained);
3219 }
3220
3221 #pragma mark -
3222 #pragma mark dispatch_queue_probe
3223
3224 unsigned long
_dispatch_queue_probe(dispatch_queue_t dq)3225 _dispatch_queue_probe(dispatch_queue_t dq)
3226 {
3227 return _dispatch_queue_class_probe(dq);
3228 }
3229
3230 #if DISPATCH_COCOA_COMPAT
3231 unsigned long
_dispatch_runloop_queue_probe(dispatch_queue_t dq)3232 _dispatch_runloop_queue_probe(dispatch_queue_t dq)
3233 {
3234 if (_dispatch_queue_class_probe(dq)) {
3235 if (dq->do_xref_cnt == -1) return true; // <rdar://problem/14026816>
3236 return _dispatch_runloop_queue_wakeup(dq);
3237 }
3238 return false;
3239 }
3240 #endif
3241
3242 unsigned long
_dispatch_mgr_queue_probe(dispatch_queue_t dq)3243 _dispatch_mgr_queue_probe(dispatch_queue_t dq)
3244 {
3245 if (_dispatch_queue_class_probe(dq)) {
3246 return _dispatch_mgr_wakeup(dq);
3247 }
3248 return false;
3249 }
3250
3251 unsigned long
_dispatch_root_queue_probe(dispatch_queue_t dq)3252 _dispatch_root_queue_probe(dispatch_queue_t dq)
3253 {
3254 _dispatch_queue_wakeup_global(dq);
3255 return false;
3256 }
3257
3258 #pragma mark -
3259 #pragma mark dispatch_wakeup
3260
3261 // 6618342 Contact the team that owns the Instrument DTrace probe before
3262 // renaming this symbol
3263 dispatch_queue_t
_dispatch_wakeup(dispatch_object_t dou)3264 _dispatch_wakeup(dispatch_object_t dou)
3265 {
3266 unsigned long type = dx_metatype(dou._do);
3267 if (type == _DISPATCH_QUEUE_TYPE || type == _DISPATCH_SOURCE_TYPE) {
3268 return _dispatch_queue_wakeup(dou._dq);
3269 }
3270 if (_dispatch_object_suspended(dou)) {
3271 return NULL;
3272 }
3273 if (!dx_probe(dou._do)) {
3274 return NULL;
3275 }
3276 if (!dispatch_atomic_cmpxchg2o(dou._do, do_suspend_cnt, 0,
3277 DISPATCH_OBJECT_SUSPEND_LOCK, acquire)) {
3278 return NULL;
3279 }
3280 _dispatch_retain(dou._do);
3281 dispatch_queue_t tq = dou._do->do_targetq;
3282 _dispatch_queue_push(tq, dou._do, 0);
3283 return tq; // libdispatch does not need this, but the Instrument DTrace
3284 // probe does
3285 }
3286
3287 #if DISPATCH_COCOA_COMPAT
3288 static inline void
_dispatch_runloop_queue_wakeup_thread(dispatch_queue_t dq)3289 _dispatch_runloop_queue_wakeup_thread(dispatch_queue_t dq)
3290 {
3291 mach_port_t mp = (mach_port_t)dq->do_ctxt;
3292 if (!mp) {
3293 return;
3294 }
3295 kern_return_t kr = _dispatch_send_wakeup_runloop_thread(mp, 0);
3296 switch (kr) {
3297 case MACH_SEND_TIMEOUT:
3298 case MACH_SEND_TIMED_OUT:
3299 case MACH_SEND_INVALID_DEST:
3300 break;
3301 default:
3302 (void)dispatch_assume_zero(kr);
3303 break;
3304 }
3305 }
3306
3307 DISPATCH_NOINLINE DISPATCH_WEAK
3308 unsigned long
_dispatch_runloop_queue_wakeup(dispatch_queue_t dq)3309 _dispatch_runloop_queue_wakeup(dispatch_queue_t dq)
3310 {
3311 _dispatch_runloop_queue_wakeup_thread(dq);
3312 return false;
3313 }
3314
3315 DISPATCH_NOINLINE
3316 static dispatch_queue_t
_dispatch_main_queue_wakeup(void)3317 _dispatch_main_queue_wakeup(void)
3318 {
3319 dispatch_queue_t dq = &_dispatch_main_q;
3320 if (!dq->dq_is_thread_bound) {
3321 return NULL;
3322 }
3323 dispatch_once_f(&_dispatch_main_q_port_pred, dq,
3324 _dispatch_runloop_queue_port_init);
3325 _dispatch_runloop_queue_wakeup_thread(dq);
3326 return NULL;
3327 }
3328 #endif
3329
3330 DISPATCH_NOINLINE
3331 static void
_dispatch_queue_wakeup_global_slow(dispatch_queue_t dq,unsigned int n)3332 _dispatch_queue_wakeup_global_slow(dispatch_queue_t dq, unsigned int n)
3333 {
3334 dispatch_root_queue_context_t qc = dq->do_ctxt;
3335 uint32_t i = n;
3336 int r;
3337
3338 _dispatch_debug_root_queue(dq, __func__);
3339 dispatch_once_f(&_dispatch_root_queues_pred, NULL,
3340 _dispatch_root_queues_init);
3341
3342 #if HAVE_PTHREAD_WORKQUEUES
3343 #if DISPATCH_USE_PTHREAD_POOL
3344 if (qc->dgq_kworkqueue != (void*)(~0ul))
3345 #endif
3346 {
3347 _dispatch_root_queue_debug("requesting new worker thread for global "
3348 "queue: %p", dq);
3349 #if DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK
3350 if (qc->dgq_kworkqueue) {
3351 pthread_workitem_handle_t wh;
3352 unsigned int gen_cnt;
3353 do {
3354 r = pthread_workqueue_additem_np(qc->dgq_kworkqueue,
3355 _dispatch_worker_thread4, dq, &wh, &gen_cnt);
3356 (void)dispatch_assume_zero(r);
3357 } while (--i);
3358 return;
3359 }
3360 #endif // DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK
3361 #if HAVE_PTHREAD_WORKQUEUE_SETDISPATCH_NP
3362 if (!dq->dq_priority) {
3363 r = pthread_workqueue_addthreads_np(qc->dgq_wq_priority,
3364 qc->dgq_wq_options, (int)i);
3365 (void)dispatch_assume_zero(r);
3366 return;
3367 }
3368 #endif
3369 #if HAVE_PTHREAD_WORKQUEUE_QOS
3370 r = _pthread_workqueue_addthreads((int)i, dq->dq_priority);
3371 (void)dispatch_assume_zero(r);
3372 #endif
3373 return;
3374 }
3375 #endif // HAVE_PTHREAD_WORKQUEUES
3376 #if DISPATCH_USE_PTHREAD_POOL
3377 dispatch_pthread_root_queue_context_t pqc = qc->dgq_ctxt;
3378 if (fastpath(pqc->dpq_thread_mediator.do_vtable)) {
3379 while (dispatch_semaphore_signal(&pqc->dpq_thread_mediator)) {
3380 if (!--i) {
3381 return;
3382 }
3383 }
3384 }
3385 uint32_t j, t_count;
3386 // seq_cst with atomic store to tail <rdar://problem/16932833>
3387 t_count = dispatch_atomic_load2o(qc, dgq_thread_pool_size, seq_cst);
3388 do {
3389 if (!t_count) {
3390 _dispatch_root_queue_debug("pthread pool is full for root queue: "
3391 "%p", dq);
3392 return;
3393 }
3394 j = i > t_count ? t_count : i;
3395 } while (!dispatch_atomic_cmpxchgvw2o(qc, dgq_thread_pool_size, t_count,
3396 t_count - j, &t_count, acquire));
3397
3398 pthread_attr_t *attr = &pqc->dpq_thread_attr;
3399 pthread_t tid, *pthr = &tid;
3400 #if DISPATCH_ENABLE_PTHREAD_ROOT_QUEUES
3401 if (slowpath(dq == &_dispatch_mgr_root_queue)) {
3402 pthr = _dispatch_mgr_root_queue_init();
3403 }
3404 #endif
3405 do {
3406 _dispatch_retain(dq);
3407 while ((r = pthread_create(pthr, attr, _dispatch_worker_thread, dq))) {
3408 if (r != EAGAIN) {
3409 (void)dispatch_assume_zero(r);
3410 }
3411 _dispatch_temporary_resource_shortage();
3412 }
3413 } while (--j);
3414 #endif // DISPATCH_USE_PTHREAD_POOL
3415 }
3416
3417 static inline void
_dispatch_queue_wakeup_global2(dispatch_queue_t dq,unsigned int n)3418 _dispatch_queue_wakeup_global2(dispatch_queue_t dq, unsigned int n)
3419 {
3420 if (!_dispatch_queue_class_probe(dq)) {
3421 return;
3422 }
3423 #if HAVE_PTHREAD_WORKQUEUES
3424 dispatch_root_queue_context_t qc = dq->do_ctxt;
3425 if (
3426 #if DISPATCH_USE_PTHREAD_POOL
3427 (qc->dgq_kworkqueue != (void*)(~0ul)) &&
3428 #endif
3429 !dispatch_atomic_cmpxchg2o(qc, dgq_pending, 0, n, relaxed)) {
3430 _dispatch_root_queue_debug("worker thread request still pending for "
3431 "global queue: %p", dq);
3432 return;
3433 }
3434 #endif // HAVE_PTHREAD_WORKQUEUES
3435 return _dispatch_queue_wakeup_global_slow(dq, n);
3436 }
3437
3438 static inline void
_dispatch_queue_wakeup_global(dispatch_queue_t dq)3439 _dispatch_queue_wakeup_global(dispatch_queue_t dq)
3440 {
3441 return _dispatch_queue_wakeup_global2(dq, 1);
3442 }
3443
3444 #pragma mark -
3445 #pragma mark dispatch_queue_invoke
3446
3447 DISPATCH_ALWAYS_INLINE
3448 static inline dispatch_queue_t
dispatch_queue_invoke2(dispatch_object_t dou,_dispatch_thread_semaphore_t * sema_ptr)3449 dispatch_queue_invoke2(dispatch_object_t dou,
3450 _dispatch_thread_semaphore_t *sema_ptr)
3451 {
3452 dispatch_queue_t dq = dou._dq;
3453 dispatch_queue_t otq = dq->do_targetq;
3454 dispatch_queue_t cq = _dispatch_queue_get_current();
3455
3456 if (slowpath(cq != otq)) {
3457 return otq;
3458 }
3459
3460 *sema_ptr = _dispatch_queue_drain(dq);
3461
3462 if (slowpath(otq != dq->do_targetq)) {
3463 // An item on the queue changed the target queue
3464 return dq->do_targetq;
3465 }
3466 return NULL;
3467 }
3468
3469 // 6618342 Contact the team that owns the Instrument DTrace probe before
3470 // renaming this symbol
3471 DISPATCH_NOINLINE
3472 void
_dispatch_queue_invoke(dispatch_queue_t dq)3473 _dispatch_queue_invoke(dispatch_queue_t dq)
3474 {
3475 _dispatch_queue_class_invoke(dq, dispatch_queue_invoke2);
3476 }
3477
3478 #pragma mark -
3479 #pragma mark dispatch_queue_drain
3480
3481 DISPATCH_ALWAYS_INLINE
3482 static inline struct dispatch_object_s*
_dispatch_queue_head(dispatch_queue_t dq)3483 _dispatch_queue_head(dispatch_queue_t dq)
3484 {
3485 struct dispatch_object_s *dc;
3486 _dispatch_wait_until(dc = fastpath(dq->dq_items_head));
3487 return dc;
3488 }
3489
3490 DISPATCH_ALWAYS_INLINE
3491 static inline struct dispatch_object_s*
_dispatch_queue_next(dispatch_queue_t dq,struct dispatch_object_s * dc)3492 _dispatch_queue_next(dispatch_queue_t dq, struct dispatch_object_s *dc)
3493 {
3494 struct dispatch_object_s *next_dc;
3495 next_dc = fastpath(dc->do_next);
3496 dq->dq_items_head = next_dc;
3497 if (!next_dc && !dispatch_atomic_cmpxchg2o(dq, dq_items_tail, dc, NULL,
3498 relaxed)) {
3499 _dispatch_wait_until(next_dc = fastpath(dc->do_next));
3500 dq->dq_items_head = next_dc;
3501 }
3502 return next_dc;
3503 }
3504
3505 _dispatch_thread_semaphore_t
_dispatch_queue_drain(dispatch_object_t dou)3506 _dispatch_queue_drain(dispatch_object_t dou)
3507 {
3508 dispatch_queue_t dq = dou._dq, orig_tq, old_dq;
3509 old_dq = _dispatch_thread_getspecific(dispatch_queue_key);
3510 struct dispatch_object_s *dc, *next_dc;
3511 _dispatch_thread_semaphore_t sema = 0;
3512
3513 // Continue draining sources after target queue change rdar://8928171
3514 bool check_tq = (dx_type(dq) != DISPATCH_SOURCE_KEVENT_TYPE);
3515
3516 orig_tq = dq->do_targetq;
3517
3518 _dispatch_thread_setspecific(dispatch_queue_key, dq);
3519 pthread_priority_t old_dp = _dispatch_set_defaultpriority(dq->dq_priority);
3520
3521 pthread_priority_t op = _dispatch_queue_get_override_priority(dq);
3522 pthread_priority_t dp = _dispatch_get_defaultpriority();
3523 dp &= _PTHREAD_PRIORITY_QOS_CLASS_MASK;
3524 if (op > dp) {
3525 _dispatch_wqthread_override_start(dq->dq_thread, op);
3526 }
3527
3528 //dispatch_debug_queue(dq, __func__);
3529
3530 while (dq->dq_items_tail) {
3531 dc = _dispatch_queue_head(dq);
3532 do {
3533 if (DISPATCH_OBJECT_SUSPENDED(dq)) {
3534 goto out;
3535 }
3536 if (dq->dq_running > dq->dq_width) {
3537 goto out;
3538 }
3539 if (slowpath(orig_tq != dq->do_targetq) && check_tq) {
3540 goto out;
3541 }
3542 bool redirect = false;
3543 if (!fastpath(dq->dq_width == 1)) {
3544 if (!DISPATCH_OBJ_IS_VTABLE(dc) &&
3545 (long)dc->do_vtable & DISPATCH_OBJ_BARRIER_BIT) {
3546 if (dq->dq_running > 1) {
3547 goto out;
3548 }
3549 } else {
3550 redirect = true;
3551 }
3552 }
3553 next_dc = _dispatch_queue_next(dq, dc);
3554 if (redirect) {
3555 _dispatch_continuation_redirect(dq, dc);
3556 continue;
3557 }
3558 if ((sema = _dispatch_barrier_sync_f_pop(dq, dc, true))) {
3559 goto out;
3560 }
3561 _dispatch_continuation_pop(dc);
3562 _dispatch_perfmon_workitem_inc();
3563 } while ((dc = next_dc));
3564 }
3565
3566 out:
3567 _dispatch_reset_defaultpriority(old_dp);
3568 _dispatch_thread_setspecific(dispatch_queue_key, old_dq);
3569 return sema;
3570 }
3571
3572 #if DISPATCH_COCOA_COMPAT
3573 static void
_dispatch_main_queue_drain(void)3574 _dispatch_main_queue_drain(void)
3575 {
3576 dispatch_queue_t dq = &_dispatch_main_q;
3577 if (!dq->dq_items_tail) {
3578 return;
3579 }
3580 struct dispatch_continuation_s marker = {
3581 .do_vtable = NULL,
3582 };
3583 struct dispatch_object_s *dmarker = (void*)▮
3584 _dispatch_queue_push_notrace(dq, dmarker, 0);
3585
3586 _dispatch_perfmon_start();
3587 dispatch_queue_t old_dq = _dispatch_thread_getspecific(dispatch_queue_key);
3588 _dispatch_thread_setspecific(dispatch_queue_key, dq);
3589 pthread_priority_t old_pri = _dispatch_get_priority();
3590 pthread_priority_t old_dp = _dispatch_set_defaultpriority(old_pri);
3591 voucher_t voucher = _voucher_copy();
3592
3593 struct dispatch_object_s *dc, *next_dc;
3594 dc = _dispatch_queue_head(dq);
3595 do {
3596 next_dc = _dispatch_queue_next(dq, dc);
3597 if (dc == dmarker) {
3598 goto out;
3599 }
3600 _dispatch_continuation_pop(dc);
3601 _dispatch_perfmon_workitem_inc();
3602 } while ((dc = next_dc));
3603 DISPATCH_CRASH("Main queue corruption");
3604
3605 out:
3606 if (next_dc) {
3607 _dispatch_main_queue_wakeup();
3608 }
3609 _dispatch_voucher_debug("main queue restore", voucher);
3610 _dispatch_set_priority_and_replace_voucher(old_pri, voucher);
3611 _dispatch_queue_reset_override_priority(dq);
3612 _dispatch_reset_defaultpriority(old_dp);
3613 _dispatch_thread_setspecific(dispatch_queue_key, old_dq);
3614 _dispatch_perfmon_end();
3615 _dispatch_force_cache_cleanup();
3616 }
3617
3618 static bool
_dispatch_runloop_queue_drain_one(dispatch_queue_t dq)3619 _dispatch_runloop_queue_drain_one(dispatch_queue_t dq)
3620 {
3621 if (!dq->dq_items_tail) {
3622 return false;
3623 }
3624 _dispatch_perfmon_start();
3625 dispatch_queue_t old_dq = _dispatch_thread_getspecific(dispatch_queue_key);
3626 _dispatch_thread_setspecific(dispatch_queue_key, dq);
3627 pthread_priority_t old_pri = _dispatch_get_priority();
3628 pthread_priority_t old_dp = _dispatch_set_defaultpriority(old_pri);
3629 voucher_t voucher = _voucher_copy();
3630
3631 struct dispatch_object_s *dc, *next_dc;
3632 dc = _dispatch_queue_head(dq);
3633 next_dc = _dispatch_queue_next(dq, dc);
3634 _dispatch_continuation_pop(dc);
3635 _dispatch_perfmon_workitem_inc();
3636
3637 _dispatch_voucher_debug("runloop queue restore", voucher);
3638 _dispatch_set_priority_and_replace_voucher(old_pri, voucher);
3639 _dispatch_reset_defaultpriority(old_dp);
3640 _dispatch_thread_setspecific(dispatch_queue_key, old_dq);
3641 _dispatch_perfmon_end();
3642 _dispatch_force_cache_cleanup();
3643 return next_dc;
3644 }
3645 #endif
3646
3647 DISPATCH_ALWAYS_INLINE_NDEBUG
3648 static inline _dispatch_thread_semaphore_t
_dispatch_queue_drain_one_barrier_sync(dispatch_queue_t dq)3649 _dispatch_queue_drain_one_barrier_sync(dispatch_queue_t dq)
3650 {
3651 // rdar://problem/8290662 "lock transfer"
3652 struct dispatch_object_s *dc;
3653 _dispatch_thread_semaphore_t sema;
3654
3655 // queue is locked, or suspended and not being drained
3656 dc = dq->dq_items_head;
3657 if (slowpath(!dc) || !(sema = _dispatch_barrier_sync_f_pop(dq, dc, false))){
3658 return 0;
3659 }
3660 // dequeue dc, it is a barrier sync
3661 (void)_dispatch_queue_next(dq, dc);
3662 return sema;
3663 }
3664
3665 void
_dispatch_mgr_queue_drain(void)3666 _dispatch_mgr_queue_drain(void)
3667 {
3668 dispatch_queue_t dq = &_dispatch_mgr_q;
3669 if (!dq->dq_items_tail) {
3670 return _dispatch_force_cache_cleanup();
3671 }
3672 _dispatch_perfmon_start();
3673 if (slowpath(_dispatch_queue_drain(dq))) {
3674 DISPATCH_CRASH("Sync onto manager queue");
3675 }
3676 _dispatch_voucher_debug("mgr queue clear", NULL);
3677 _voucher_clear();
3678 _dispatch_queue_reset_override_priority(dq);
3679 _dispatch_reset_defaultpriority_override();
3680 _dispatch_perfmon_end();
3681 _dispatch_force_cache_cleanup();
3682 }
3683
3684 #pragma mark -
3685 #pragma mark _dispatch_queue_wakeup_with_qos
3686
3687 DISPATCH_NOINLINE
3688 static dispatch_queue_t
_dispatch_queue_wakeup_with_qos_slow(dispatch_queue_t dq,pthread_priority_t pp,bool retained)3689 _dispatch_queue_wakeup_with_qos_slow(dispatch_queue_t dq, pthread_priority_t pp,
3690 bool retained)
3691 {
3692 if (!dx_probe(dq) && (dq->dq_is_thread_bound || !dq->dq_thread)) {
3693 if (retained) _dispatch_release(dq);
3694 return NULL;
3695 }
3696 pp &= _PTHREAD_PRIORITY_QOS_CLASS_MASK;
3697 bool override = _dispatch_queue_override_priority(dq, pp);
3698 if (override && dq->dq_running > 1) {
3699 override = false;
3700 }
3701
3702 if (!dispatch_atomic_cmpxchg2o(dq, do_suspend_cnt, 0,
3703 DISPATCH_OBJECT_SUSPEND_LOCK, acquire)) {
3704 #if DISPATCH_COCOA_COMPAT
3705 if (dq == &_dispatch_main_q && dq->dq_is_thread_bound) {
3706 return _dispatch_main_queue_wakeup();
3707 }
3708 #endif
3709 if (override) {
3710 mach_port_t th;
3711 // <rdar://problem/17735825> to traverse the tq chain safely we must
3712 // lock it to ensure it cannot change, unless the queue is running
3713 // and we can just override the thread itself
3714 if (dq->dq_thread) {
3715 _dispatch_wqthread_override_start(dq->dq_thread, pp);
3716 } else if (!dispatch_atomic_cmpxchgv2o(dq, dq_tqthread,
3717 MACH_PORT_NULL, _dispatch_thread_port(), &th, acquire)) {
3718 // already locked, override the owner, trysync will do a queue
3719 // wakeup when it returns.
3720 _dispatch_wqthread_override_start(th, pp);
3721 } else {
3722 dispatch_queue_t tq = dq->do_targetq;
3723 if (_dispatch_queue_prepare_override(dq, tq, pp)) {
3724 _dispatch_queue_push_override(dq, tq, pp);
3725 } else {
3726 _dispatch_queue_wakeup_with_qos(tq, pp);
3727 }
3728 dispatch_atomic_store2o(dq, dq_tqthread, MACH_PORT_NULL,
3729 release);
3730 }
3731 }
3732 if (retained) _dispatch_release(dq);
3733 return NULL;
3734 }
3735 dispatch_queue_t tq = dq->do_targetq;
3736 if (!retained) _dispatch_retain(dq);
3737 if (override) {
3738 override = _dispatch_queue_prepare_override(dq, tq, pp);
3739 }
3740 _dispatch_queue_push(tq, dq, pp);
3741 if (override) {
3742 _dispatch_queue_push_override(dq, tq, pp);
3743 }
3744 return tq; // libdispatch does not need this, but the Instrument DTrace
3745 // probe does
3746 }
3747
3748 DISPATCH_ALWAYS_INLINE
3749 static inline dispatch_queue_t
_dispatch_queue_wakeup_with_qos2(dispatch_queue_t dq,pthread_priority_t pp,bool retained)3750 _dispatch_queue_wakeup_with_qos2(dispatch_queue_t dq, pthread_priority_t pp,
3751 bool retained)
3752 {
3753 if (_dispatch_object_suspended(dq)) {
3754 _dispatch_queue_override_priority(dq, pp);
3755 if (retained) _dispatch_release(dq);
3756 return NULL;
3757 }
3758 return _dispatch_queue_wakeup_with_qos_slow(dq, pp, retained);
3759 }
3760
3761 DISPATCH_NOINLINE
3762 void
_dispatch_queue_wakeup_with_qos_and_release(dispatch_queue_t dq,pthread_priority_t pp)3763 _dispatch_queue_wakeup_with_qos_and_release(dispatch_queue_t dq,
3764 pthread_priority_t pp)
3765 {
3766 (void)_dispatch_queue_wakeup_with_qos2(dq, pp, true);
3767 }
3768
3769 DISPATCH_NOINLINE
3770 void
_dispatch_queue_wakeup_with_qos(dispatch_queue_t dq,pthread_priority_t pp)3771 _dispatch_queue_wakeup_with_qos(dispatch_queue_t dq, pthread_priority_t pp)
3772 {
3773 (void)_dispatch_queue_wakeup_with_qos2(dq, pp, false);
3774 }
3775
3776 DISPATCH_NOINLINE
3777 dispatch_queue_t
_dispatch_queue_wakeup(dispatch_queue_t dq)3778 _dispatch_queue_wakeup(dispatch_queue_t dq)
3779 {
3780 return _dispatch_queue_wakeup_with_qos2(dq,
3781 _dispatch_queue_get_override_priority(dq), false);
3782 }
3783
3784 #if HAVE_PTHREAD_WORKQUEUE_QOS
3785 static void
_dispatch_queue_override_invoke(void * ctxt)3786 _dispatch_queue_override_invoke(void *ctxt)
3787 {
3788 dispatch_continuation_t dc = (dispatch_continuation_t)ctxt;
3789 dispatch_queue_t dq = dc->dc_data;
3790 pthread_priority_t p = 0;
3791
3792 if (!slowpath(DISPATCH_OBJECT_SUSPENDED(dq)) &&
3793 fastpath(dispatch_atomic_cmpxchg2o(dq, dq_running, 0, 1, acquire))) {
3794 _dispatch_queue_set_thread(dq);
3795
3796 _dispatch_object_debug(dq, "stolen onto thread 0x%x, 0x%lx",
3797 dq->dq_thread, _dispatch_get_defaultpriority());
3798
3799 pthread_priority_t old_dp = _dispatch_get_defaultpriority();
3800 _dispatch_reset_defaultpriority(dc->dc_priority);
3801
3802 dispatch_queue_t tq = NULL;
3803 _dispatch_thread_semaphore_t sema = 0;
3804 tq = dispatch_queue_invoke2(dq, &sema);
3805
3806 _dispatch_queue_clear_thread(dq);
3807 _dispatch_reset_defaultpriority(old_dp);
3808
3809 uint32_t running = dispatch_atomic_dec2o(dq, dq_running, release);
3810 if (sema) {
3811 _dispatch_thread_semaphore_signal(sema);
3812 } else if (!tq && running == 0) {
3813 p = _dispatch_queue_reset_override_priority(dq);
3814 if (p > (dq->dq_priority & _PTHREAD_PRIORITY_QOS_CLASS_MASK)) {
3815 _dispatch_wqthread_override_reset();
3816 }
3817 }
3818 _dispatch_introspection_queue_item_complete(dq);
3819 if (running == 0) {
3820 return _dispatch_queue_wakeup_with_qos_and_release(dq, p);
3821 }
3822 } else {
3823 mach_port_t th = dq->dq_thread;
3824 if (th) {
3825 p = _dispatch_queue_get_override_priority(dq);
3826 _dispatch_object_debug(dq, "overriding thr 0x%x to priority 0x%lx",
3827 th, p);
3828 _dispatch_wqthread_override_start(th, p);
3829 }
3830 }
3831 _dispatch_release(dq); // added when we pushed the override block
3832 }
3833 #endif
3834
3835 static inline bool
_dispatch_queue_prepare_override(dispatch_queue_t dq,dispatch_queue_t tq,pthread_priority_t p)3836 _dispatch_queue_prepare_override(dispatch_queue_t dq, dispatch_queue_t tq,
3837 pthread_priority_t p)
3838 {
3839 #if HAVE_PTHREAD_WORKQUEUE_QOS
3840 if (dx_type(tq) != DISPATCH_QUEUE_ROOT_TYPE || !tq->dq_priority) {
3841 return false;
3842 }
3843 if (p <= (dq->dq_priority & _PTHREAD_PRIORITY_QOS_CLASS_MASK)) {
3844 return false;
3845 }
3846 if (p <= (tq->dq_priority & _PTHREAD_PRIORITY_QOS_CLASS_MASK)) {
3847 return false;
3848 }
3849 _dispatch_retain(dq);
3850 return true;
3851 #else
3852 (void)dq; (void)tq; (void)p;
3853 return false;
3854 #endif
3855 }
3856
3857 static inline void
_dispatch_queue_push_override(dispatch_queue_t dq,dispatch_queue_t tq,pthread_priority_t p)3858 _dispatch_queue_push_override(dispatch_queue_t dq, dispatch_queue_t tq,
3859 pthread_priority_t p)
3860 {
3861 #if HAVE_PTHREAD_WORKQUEUE_QOS
3862 unsigned int qosbit, idx, overcommit;
3863 overcommit = (tq->dq_priority & _PTHREAD_PRIORITY_OVERCOMMIT_FLAG) ? 1 : 0;
3864 qosbit = (p & _PTHREAD_PRIORITY_QOS_CLASS_MASK) >>
3865 _PTHREAD_PRIORITY_QOS_CLASS_SHIFT;
3866 idx = (unsigned int)__builtin_ffs((int)qosbit);
3867 if (!idx || idx > DISPATCH_QUEUE_QOS_COUNT) {
3868 DISPATCH_CRASH("Corrupted override priority");
3869 }
3870 dispatch_queue_t rq = &_dispatch_root_queues[((idx-1) << 1) | overcommit];
3871
3872 dispatch_continuation_t dc = _dispatch_continuation_alloc();
3873 dc->do_vtable = (void *)(DISPATCH_OBJ_ASYNC_BIT | DISPATCH_OBJ_BARRIER_BIT);
3874 dc->dc_func = _dispatch_queue_override_invoke;
3875 dc->dc_ctxt = dc;
3876 dc->dc_priority = tq->dq_priority;
3877 dc->dc_voucher = NULL;
3878 dc->dc_data = dq;
3879 // dq retained by _dispatch_queue_prepare_override
3880
3881 _dispatch_queue_push(rq, dc, 0);
3882 #else
3883 (void)dq; (void)tq; (void)p;
3884 #endif
3885 }
3886
3887 #pragma mark -
3888 #pragma mark dispatch_root_queue_drain
3889
3890 DISPATCH_NOINLINE
3891 static bool
_dispatch_queue_concurrent_drain_one_slow(dispatch_queue_t dq)3892 _dispatch_queue_concurrent_drain_one_slow(dispatch_queue_t dq)
3893 {
3894 dispatch_root_queue_context_t qc = dq->do_ctxt;
3895 struct dispatch_object_s *const mediator = (void *)~0ul;
3896 bool pending = false, available = true;
3897 unsigned int sleep_time = DISPATCH_CONTENTION_USLEEP_START;
3898
3899 do {
3900 // Spin for a short while in case the contention is temporary -- e.g.
3901 // when starting up after dispatch_apply, or when executing a few
3902 // short continuations in a row.
3903 if (_dispatch_contention_wait_until(dq->dq_items_head != mediator)) {
3904 goto out;
3905 }
3906 // Since we have serious contention, we need to back off.
3907 if (!pending) {
3908 // Mark this queue as pending to avoid requests for further threads
3909 (void)dispatch_atomic_inc2o(qc, dgq_pending, relaxed);
3910 pending = true;
3911 }
3912 _dispatch_contention_usleep(sleep_time);
3913 if (fastpath(dq->dq_items_head != mediator)) goto out;
3914 sleep_time *= 2;
3915 } while (sleep_time < DISPATCH_CONTENTION_USLEEP_MAX);
3916
3917 // The ratio of work to libdispatch overhead must be bad. This
3918 // scenario implies that there are too many threads in the pool.
3919 // Create a new pending thread and then exit this thread.
3920 // The kernel will grant a new thread when the load subsides.
3921 _dispatch_debug("contention on global queue: %p", dq);
3922 available = false;
3923 out:
3924 if (pending) {
3925 (void)dispatch_atomic_dec2o(qc, dgq_pending, relaxed);
3926 }
3927 if (!available) {
3928 _dispatch_queue_wakeup_global(dq);
3929 }
3930 return available;
3931 }
3932
3933 DISPATCH_ALWAYS_INLINE
3934 static inline bool
_dispatch_queue_concurrent_drain_one2(dispatch_queue_t dq)3935 _dispatch_queue_concurrent_drain_one2(dispatch_queue_t dq)
3936 {
3937 // Wait for queue head and tail to be both non-empty or both empty
3938 bool available; // <rdar://problem/15917893>
3939 _dispatch_wait_until((dq->dq_items_head != NULL) ==
3940 (available = (dq->dq_items_tail != NULL)));
3941 return available;
3942 }
3943
3944 DISPATCH_ALWAYS_INLINE_NDEBUG
3945 static inline struct dispatch_object_s *
_dispatch_queue_concurrent_drain_one(dispatch_queue_t dq)3946 _dispatch_queue_concurrent_drain_one(dispatch_queue_t dq)
3947 {
3948 struct dispatch_object_s *head, *next, *const mediator = (void *)~0ul;
3949
3950 start:
3951 // The mediator value acts both as a "lock" and a signal
3952 head = dispatch_atomic_xchg2o(dq, dq_items_head, mediator, relaxed);
3953
3954 if (slowpath(head == NULL)) {
3955 // The first xchg on the tail will tell the enqueueing thread that it
3956 // is safe to blindly write out to the head pointer. A cmpxchg honors
3957 // the algorithm.
3958 if (slowpath(!dispatch_atomic_cmpxchg2o(dq, dq_items_head, mediator,
3959 NULL, relaxed))) {
3960 goto start;
3961 }
3962 if (slowpath(dq->dq_items_tail) && // <rdar://problem/14416349>
3963 _dispatch_queue_concurrent_drain_one2(dq)) {
3964 goto start;
3965 }
3966 _dispatch_root_queue_debug("no work on global queue: %p", dq);
3967 return NULL;
3968 }
3969
3970 if (slowpath(head == mediator)) {
3971 // This thread lost the race for ownership of the queue.
3972 if (fastpath(_dispatch_queue_concurrent_drain_one_slow(dq))) {
3973 goto start;
3974 }
3975 return NULL;
3976 }
3977
3978 // Restore the head pointer to a sane value before returning.
3979 // If 'next' is NULL, then this item _might_ be the last item.
3980 next = fastpath(head->do_next);
3981
3982 if (slowpath(!next)) {
3983 dispatch_atomic_store2o(dq, dq_items_head, NULL, relaxed);
3984
3985 if (dispatch_atomic_cmpxchg2o(dq, dq_items_tail, head, NULL, relaxed)) {
3986 // both head and tail are NULL now
3987 goto out;
3988 }
3989 // There must be a next item now.
3990 _dispatch_wait_until(next = head->do_next);
3991 }
3992
3993 dispatch_atomic_store2o(dq, dq_items_head, next, relaxed);
3994 _dispatch_queue_wakeup_global(dq);
3995 out:
3996 return head;
3997 }
3998
3999 static void
_dispatch_root_queue_drain(dispatch_queue_t dq)4000 _dispatch_root_queue_drain(dispatch_queue_t dq)
4001 {
4002 #if DISPATCH_DEBUG
4003 if (_dispatch_thread_getspecific(dispatch_queue_key)) {
4004 DISPATCH_CRASH("Premature thread recycling");
4005 }
4006 #endif
4007 _dispatch_thread_setspecific(dispatch_queue_key, dq);
4008 pthread_priority_t old_pri = _dispatch_get_priority();
4009 pthread_priority_t pri = dq->dq_priority ? dq->dq_priority : old_pri;
4010 pthread_priority_t old_dp = _dispatch_set_defaultpriority(pri);
4011
4012 #if DISPATCH_COCOA_COMPAT
4013 // ensure that high-level memory management techniques do not leak/crash
4014 if (dispatch_begin_thread_4GC) {
4015 dispatch_begin_thread_4GC();
4016 }
4017 void *pool = _dispatch_autorelease_pool_push();
4018 #endif // DISPATCH_COCOA_COMPAT
4019
4020 _dispatch_perfmon_start();
4021 struct dispatch_object_s *item;
4022 bool reset = false;
4023 while ((item = fastpath(_dispatch_queue_concurrent_drain_one(dq)))) {
4024 if (reset) _dispatch_wqthread_override_reset();
4025 _dispatch_continuation_pop(item);
4026 reset = _dispatch_reset_defaultpriority_override();
4027 }
4028 _dispatch_voucher_debug("root queue clear", NULL);
4029 _dispatch_set_priority_and_replace_voucher(old_pri, NULL);
4030 _dispatch_reset_defaultpriority(old_dp);
4031 _dispatch_perfmon_end();
4032
4033 #if DISPATCH_COCOA_COMPAT
4034 _dispatch_autorelease_pool_pop(pool);
4035 if (dispatch_end_thread_4GC) {
4036 dispatch_end_thread_4GC();
4037 }
4038 #endif // DISPATCH_COCOA_COMPAT
4039
4040 _dispatch_thread_setspecific(dispatch_queue_key, NULL);
4041 }
4042
4043 #pragma mark -
4044 #pragma mark dispatch_worker_thread
4045
4046 #if HAVE_PTHREAD_WORKQUEUES
4047 static void
_dispatch_worker_thread4(void * context)4048 _dispatch_worker_thread4(void *context)
4049 {
4050 dispatch_queue_t dq = context;
4051 dispatch_root_queue_context_t qc = dq->do_ctxt;
4052
4053 _dispatch_introspection_thread_add();
4054 int pending = (int)dispatch_atomic_dec2o(qc, dgq_pending, relaxed);
4055 dispatch_assert(pending >= 0);
4056 _dispatch_root_queue_drain(dq);
4057 __asm__(""); // prevent tailcall (for Instrument DTrace probe)
4058 }
4059
4060 #if HAVE_PTHREAD_WORKQUEUE_QOS
4061 static void
_dispatch_worker_thread3(pthread_priority_t priority)4062 _dispatch_worker_thread3(pthread_priority_t priority)
4063 {
4064 // Reset priority TSD to workaround <rdar://problem/17825261>
4065 _dispatch_thread_setspecific(dispatch_priority_key,
4066 (void*)(uintptr_t)(priority & ~_PTHREAD_PRIORITY_FLAGS_MASK));
4067 unsigned int overcommit, qosbit, idx;
4068 overcommit = (priority & _PTHREAD_PRIORITY_OVERCOMMIT_FLAG) ? 1 : 0;
4069 qosbit = (priority & _PTHREAD_PRIORITY_QOS_CLASS_MASK) >>
4070 _PTHREAD_PRIORITY_QOS_CLASS_SHIFT;
4071 if (!_dispatch_root_queues[DISPATCH_ROOT_QUEUE_IDX_MAINTENANCE_QOS].
4072 dq_priority) {
4073 // If kernel doesn't support maintenance, bottom bit is background.
4074 // Shift to our idea of where background bit is.
4075 qosbit <<= 1;
4076 }
4077 idx = (unsigned int)__builtin_ffs((int)qosbit);
4078 dispatch_assert(idx > 0 && idx < DISPATCH_QUEUE_QOS_COUNT+1);
4079 dispatch_queue_t dq = &_dispatch_root_queues[((idx-1) << 1) | overcommit];
4080 return _dispatch_worker_thread4(dq);
4081 }
4082 #endif // HAVE_PTHREAD_WORKQUEUE_QOS
4083
4084 #if HAVE_PTHREAD_WORKQUEUE_SETDISPATCH_NP
4085 // 6618342 Contact the team that owns the Instrument DTrace probe before
4086 // renaming this symbol
4087 static void
_dispatch_worker_thread2(int priority,int options,void * context DISPATCH_UNUSED)4088 _dispatch_worker_thread2(int priority, int options,
4089 void *context DISPATCH_UNUSED)
4090 {
4091 dispatch_assert(priority >= 0 && priority < WORKQ_NUM_PRIOQUEUE);
4092 dispatch_assert(!(options & ~WORKQ_ADDTHREADS_OPTION_OVERCOMMIT));
4093 dispatch_queue_t dq = _dispatch_wq2root_queues[priority][options];
4094
4095 return _dispatch_worker_thread4(dq);
4096 }
4097 #endif // HAVE_PTHREAD_WORKQUEUE_SETDISPATCH_NP
4098 #endif // HAVE_PTHREAD_WORKQUEUES
4099
4100 #if DISPATCH_USE_PTHREAD_POOL
4101 // 6618342 Contact the team that owns the Instrument DTrace probe before
4102 // renaming this symbol
4103 static void *
_dispatch_worker_thread(void * context)4104 _dispatch_worker_thread(void *context)
4105 {
4106 dispatch_queue_t dq = context;
4107 dispatch_root_queue_context_t qc = dq->do_ctxt;
4108 dispatch_pthread_root_queue_context_t pqc = qc->dgq_ctxt;
4109
4110 if (pqc->dpq_thread_configure) {
4111 pqc->dpq_thread_configure();
4112 }
4113
4114 sigset_t mask;
4115 int r;
4116 // workaround tweaks the kernel workqueue does for us
4117 r = sigfillset(&mask);
4118 (void)dispatch_assume_zero(r);
4119 r = _dispatch_pthread_sigmask(SIG_BLOCK, &mask, NULL);
4120 (void)dispatch_assume_zero(r);
4121 _dispatch_introspection_thread_add();
4122
4123 const int64_t timeout = 5ull * NSEC_PER_SEC;
4124 do {
4125 _dispatch_root_queue_drain(dq);
4126 } while (dispatch_semaphore_wait(&pqc->dpq_thread_mediator,
4127 dispatch_time(0, timeout)) == 0);
4128
4129 (void)dispatch_atomic_inc2o(qc, dgq_thread_pool_size, release);
4130 _dispatch_queue_wakeup_global(dq);
4131 _dispatch_release(dq);
4132
4133 return NULL;
4134 }
4135
4136 int
_dispatch_pthread_sigmask(int how,sigset_t * set,sigset_t * oset)4137 _dispatch_pthread_sigmask(int how, sigset_t *set, sigset_t *oset)
4138 {
4139 int r;
4140
4141 /* Workaround: 6269619 Not all signals can be delivered on any thread */
4142
4143 r = sigdelset(set, SIGILL);
4144 (void)dispatch_assume_zero(r);
4145 r = sigdelset(set, SIGTRAP);
4146 (void)dispatch_assume_zero(r);
4147 #if HAVE_DECL_SIGEMT
4148 r = sigdelset(set, SIGEMT);
4149 (void)dispatch_assume_zero(r);
4150 #endif
4151 r = sigdelset(set, SIGFPE);
4152 (void)dispatch_assume_zero(r);
4153 r = sigdelset(set, SIGBUS);
4154 (void)dispatch_assume_zero(r);
4155 r = sigdelset(set, SIGSEGV);
4156 (void)dispatch_assume_zero(r);
4157 r = sigdelset(set, SIGSYS);
4158 (void)dispatch_assume_zero(r);
4159 r = sigdelset(set, SIGPIPE);
4160 (void)dispatch_assume_zero(r);
4161
4162 return pthread_sigmask(how, set, oset);
4163 }
4164 #endif // DISPATCH_USE_PTHREAD_POOL
4165
4166 #pragma mark -
4167 #pragma mark dispatch_runloop_queue
4168
4169 static bool _dispatch_program_is_probably_callback_driven;
4170
4171 #if DISPATCH_COCOA_COMPAT
4172
4173 dispatch_queue_t
_dispatch_runloop_root_queue_create_4CF(const char * label,unsigned long flags)4174 _dispatch_runloop_root_queue_create_4CF(const char *label, unsigned long flags)
4175 {
4176 dispatch_queue_t dq;
4177 size_t dqs;
4178
4179 if (slowpath(flags)) {
4180 return NULL;
4181 }
4182 dqs = sizeof(struct dispatch_queue_s) - DISPATCH_QUEUE_CACHELINE_PAD;
4183 dq = _dispatch_alloc(DISPATCH_VTABLE(queue_runloop), dqs);
4184 _dispatch_queue_init(dq);
4185 dq->do_targetq = _dispatch_get_root_queue(_DISPATCH_QOS_CLASS_DEFAULT,true);
4186 dq->dq_label = label ? label : "runloop-queue"; // no-copy contract
4187 dq->do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK;
4188 dq->dq_running = 1;
4189 dq->dq_is_thread_bound = 1;
4190 _dispatch_runloop_queue_port_init(dq);
4191 _dispatch_queue_set_bound_thread(dq);
4192 _dispatch_object_debug(dq, "%s", __func__);
4193 return _dispatch_introspection_queue_create(dq);
4194 }
4195
4196 void
_dispatch_runloop_queue_xref_dispose(dispatch_queue_t dq)4197 _dispatch_runloop_queue_xref_dispose(dispatch_queue_t dq)
4198 {
4199 _dispatch_object_debug(dq, "%s", __func__);
4200 (void)dispatch_atomic_dec2o(dq, dq_running, relaxed);
4201 unsigned int suspend_cnt = dispatch_atomic_sub2o(dq, do_suspend_cnt,
4202 DISPATCH_OBJECT_SUSPEND_LOCK, release);
4203 _dispatch_queue_clear_bound_thread(dq);
4204 if (suspend_cnt == 0) {
4205 _dispatch_queue_wakeup(dq);
4206 }
4207 }
4208
4209 void
_dispatch_runloop_queue_dispose(dispatch_queue_t dq)4210 _dispatch_runloop_queue_dispose(dispatch_queue_t dq)
4211 {
4212 _dispatch_object_debug(dq, "%s", __func__);
4213 _dispatch_introspection_queue_dispose(dq);
4214 _dispatch_runloop_queue_port_dispose(dq);
4215 _dispatch_queue_destroy(dq);
4216 }
4217
4218 bool
_dispatch_runloop_root_queue_perform_4CF(dispatch_queue_t dq)4219 _dispatch_runloop_root_queue_perform_4CF(dispatch_queue_t dq)
4220 {
4221 if (slowpath(dq->do_vtable != DISPATCH_VTABLE(queue_runloop))) {
4222 DISPATCH_CLIENT_CRASH("Not a runloop queue");
4223 }
4224 dispatch_retain(dq);
4225 bool r = _dispatch_runloop_queue_drain_one(dq);
4226 dispatch_release(dq);
4227 return r;
4228 }
4229
4230 void
_dispatch_runloop_root_queue_wakeup_4CF(dispatch_queue_t dq)4231 _dispatch_runloop_root_queue_wakeup_4CF(dispatch_queue_t dq)
4232 {
4233 if (slowpath(dq->do_vtable != DISPATCH_VTABLE(queue_runloop))) {
4234 DISPATCH_CLIENT_CRASH("Not a runloop queue");
4235 }
4236 _dispatch_runloop_queue_probe(dq);
4237 }
4238
4239 mach_port_t
_dispatch_runloop_root_queue_get_port_4CF(dispatch_queue_t dq)4240 _dispatch_runloop_root_queue_get_port_4CF(dispatch_queue_t dq)
4241 {
4242 if (slowpath(dq->do_vtable != DISPATCH_VTABLE(queue_runloop))) {
4243 DISPATCH_CLIENT_CRASH("Not a runloop queue");
4244 }
4245 return (mach_port_t)dq->do_ctxt;
4246 }
4247
4248 static void
_dispatch_runloop_queue_port_init(void * ctxt)4249 _dispatch_runloop_queue_port_init(void *ctxt)
4250 {
4251 dispatch_queue_t dq = (dispatch_queue_t)ctxt;
4252 mach_port_t mp;
4253 kern_return_t kr;
4254
4255 _dispatch_safe_fork = false;
4256 kr = mach_port_allocate(mach_task_self(), MACH_PORT_RIGHT_RECEIVE, &mp);
4257 DISPATCH_VERIFY_MIG(kr);
4258 (void)dispatch_assume_zero(kr);
4259 kr = mach_port_insert_right(mach_task_self(), mp, mp,
4260 MACH_MSG_TYPE_MAKE_SEND);
4261 DISPATCH_VERIFY_MIG(kr);
4262 (void)dispatch_assume_zero(kr);
4263 if (dq != &_dispatch_main_q) {
4264 struct mach_port_limits limits = {
4265 .mpl_qlimit = 1,
4266 };
4267 kr = mach_port_set_attributes(mach_task_self(), mp,
4268 MACH_PORT_LIMITS_INFO, (mach_port_info_t)&limits,
4269 sizeof(limits));
4270 DISPATCH_VERIFY_MIG(kr);
4271 (void)dispatch_assume_zero(kr);
4272 }
4273 dq->do_ctxt = (void*)(uintptr_t)mp;
4274
4275 _dispatch_program_is_probably_callback_driven = true;
4276 }
4277
4278 static void
_dispatch_runloop_queue_port_dispose(dispatch_queue_t dq)4279 _dispatch_runloop_queue_port_dispose(dispatch_queue_t dq)
4280 {
4281 mach_port_t mp = (mach_port_t)dq->do_ctxt;
4282 if (!mp) {
4283 return;
4284 }
4285 dq->do_ctxt = NULL;
4286 kern_return_t kr = mach_port_deallocate(mach_task_self(), mp);
4287 DISPATCH_VERIFY_MIG(kr);
4288 (void)dispatch_assume_zero(kr);
4289 #ifndef __FreeBSD__
4290 /* XXX: https://bugs.freenas.org/issues/10145 */
4291 kr = mach_port_mod_refs(mach_task_self(), mp, MACH_PORT_RIGHT_RECEIVE, -1);
4292 DISPATCH_VERIFY_MIG(kr);
4293 (void)dispatch_assume_zero(kr);
4294 #endif
4295 }
4296
4297 #pragma mark -
4298 #pragma mark dispatch_main_queue
4299
4300 mach_port_t
_dispatch_get_main_queue_port_4CF(void)4301 _dispatch_get_main_queue_port_4CF(void)
4302 {
4303 dispatch_queue_t dq = &_dispatch_main_q;
4304 dispatch_once_f(&_dispatch_main_q_port_pred, dq,
4305 _dispatch_runloop_queue_port_init);
4306 return (mach_port_t)dq->do_ctxt;
4307 }
4308
4309 static bool main_q_is_draining;
4310
4311 // 6618342 Contact the team that owns the Instrument DTrace probe before
4312 // renaming this symbol
4313 DISPATCH_NOINLINE
4314 static void
_dispatch_queue_set_mainq_drain_state(bool arg)4315 _dispatch_queue_set_mainq_drain_state(bool arg)
4316 {
4317 main_q_is_draining = arg;
4318 }
4319
4320 void
_dispatch_main_queue_callback_4CF(mach_msg_header_t * msg DISPATCH_UNUSED)4321 _dispatch_main_queue_callback_4CF(mach_msg_header_t *msg DISPATCH_UNUSED)
4322 {
4323 if (main_q_is_draining) {
4324 return;
4325 }
4326 _dispatch_queue_set_mainq_drain_state(true);
4327 _dispatch_main_queue_drain();
4328 _dispatch_queue_set_mainq_drain_state(false);
4329 }
4330
4331 #endif
4332
4333 void
dispatch_main(void)4334 dispatch_main(void)
4335 {
4336 #if HAVE_PTHREAD_MAIN_NP
4337 if (pthread_main_np()) {
4338 #endif
4339 _dispatch_object_debug(&_dispatch_main_q, "%s", __func__);
4340 _dispatch_program_is_probably_callback_driven = true;
4341 pthread_exit(NULL);
4342 DISPATCH_CRASH("pthread_exit() returned");
4343 #if HAVE_PTHREAD_MAIN_NP
4344 }
4345 DISPATCH_CLIENT_CRASH("dispatch_main() must be called on the main thread");
4346 #endif
4347 }
4348
4349 DISPATCH_NOINLINE DISPATCH_NORETURN
4350 static void
_dispatch_sigsuspend(void)4351 _dispatch_sigsuspend(void)
4352 {
4353 static const sigset_t mask;
4354
4355 for (;;) {
4356 sigsuspend(&mask);
4357 }
4358 }
4359
4360 DISPATCH_NORETURN
4361 static void
_dispatch_sig_thread(void * ctxt DISPATCH_UNUSED)4362 _dispatch_sig_thread(void *ctxt DISPATCH_UNUSED)
4363 {
4364 // never returns, so burn bridges behind us
4365 _dispatch_clear_stack(0);
4366 _dispatch_sigsuspend();
4367 }
4368
4369 DISPATCH_NOINLINE
4370 static void
_dispatch_queue_cleanup2(void)4371 _dispatch_queue_cleanup2(void)
4372 {
4373 dispatch_queue_t dq = &_dispatch_main_q;
4374 (void)dispatch_atomic_dec2o(dq, dq_running, relaxed);
4375 unsigned int suspend_cnt = dispatch_atomic_sub2o(dq, do_suspend_cnt,
4376 DISPATCH_OBJECT_SUSPEND_LOCK, release);
4377 dq->dq_is_thread_bound = 0;
4378 if (suspend_cnt == 0) {
4379 _dispatch_queue_wakeup(dq);
4380 }
4381
4382 // overload the "probably" variable to mean that dispatch_main() or
4383 // similar non-POSIX API was called
4384 // this has to run before the DISPATCH_COCOA_COMPAT below
4385 if (_dispatch_program_is_probably_callback_driven) {
4386 _dispatch_barrier_async_detached_f(_dispatch_get_root_queue(
4387 _DISPATCH_QOS_CLASS_DEFAULT, true), NULL, _dispatch_sig_thread);
4388 sleep(1); // workaround 6778970
4389 }
4390
4391 #if DISPATCH_COCOA_COMPAT
4392 dispatch_once_f(&_dispatch_main_q_port_pred, dq,
4393 _dispatch_runloop_queue_port_init);
4394 _dispatch_runloop_queue_port_dispose(dq);
4395 #endif
4396 }
4397
4398 static void
_dispatch_queue_cleanup(void * ctxt)4399 _dispatch_queue_cleanup(void *ctxt)
4400 {
4401 if (ctxt == &_dispatch_main_q) {
4402 return _dispatch_queue_cleanup2();
4403 }
4404 // POSIX defines that destructors are only called if 'ctxt' is non-null
4405 DISPATCH_CRASH("Premature thread exit while a dispatch queue is running");
4406 }
4407