]> git.saurik.com Git - apple/libdispatch.git/blob - src/source.c
2b0a9a2a756a7d5b40c4cf856a1367f584e110ee
[apple/libdispatch.git] / src / source.c
1 /*
2 * Copyright (c) 2008-2011 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 #include "internal.h"
22 #if HAVE_MACH
23 #include "protocol.h"
24 #include "protocolServer.h"
25 #endif
26 #include <sys/mount.h>
27
28 static void _dispatch_source_merge_kevent(dispatch_source_t ds,
29 const struct kevent *ke);
30 static void _dispatch_kevent_register(dispatch_source_t ds);
31 static void _dispatch_kevent_unregister(dispatch_source_t ds);
32 static bool _dispatch_kevent_resume(dispatch_kevent_t dk, uint32_t new_flags,
33 uint32_t del_flags);
34 static inline void _dispatch_source_timer_init(void);
35 static void _dispatch_timer_list_update(dispatch_source_t ds);
36 static inline unsigned long _dispatch_source_timer_data(
37 dispatch_source_refs_t dr, unsigned long prev);
38 #if HAVE_MACH
39 static kern_return_t _dispatch_kevent_machport_resume(dispatch_kevent_t dk,
40 uint32_t new_flags, uint32_t del_flags);
41 static void _dispatch_drain_mach_messages(struct kevent *ke);
42 #endif
43 #if DISPATCH_DEBUG
44 static void _dispatch_kevent_debugger(void *context);
45 #endif
46
47 #pragma mark -
48 #pragma mark dispatch_source_t
49
50 dispatch_source_t
51 dispatch_source_create(dispatch_source_type_t type,
52 uintptr_t handle,
53 unsigned long mask,
54 dispatch_queue_t q)
55 {
56 const struct kevent *proto_kev = &type->ke;
57 dispatch_source_t ds = NULL;
58 dispatch_kevent_t dk = NULL;
59
60 // input validation
61 if (type == NULL || (mask & ~type->mask)) {
62 return NULL;
63 }
64
65 switch (type->ke.filter) {
66 case EVFILT_SIGNAL:
67 if (handle >= NSIG) {
68 return NULL;
69 }
70 break;
71 case EVFILT_FS:
72 #if DISPATCH_USE_VM_PRESSURE
73 case EVFILT_VM:
74 #endif
75 case DISPATCH_EVFILT_CUSTOM_ADD:
76 case DISPATCH_EVFILT_CUSTOM_OR:
77 case DISPATCH_EVFILT_TIMER:
78 if (handle) {
79 return NULL;
80 }
81 break;
82 default:
83 break;
84 }
85
86 dk = calloc(1ul, sizeof(struct dispatch_kevent_s));
87 dk->dk_kevent = *proto_kev;
88 dk->dk_kevent.ident = handle;
89 dk->dk_kevent.flags |= EV_ADD|EV_ENABLE;
90 dk->dk_kevent.fflags |= (uint32_t)mask;
91 dk->dk_kevent.udata = dk;
92 TAILQ_INIT(&dk->dk_sources);
93
94 ds = _dispatch_alloc(DISPATCH_VTABLE(source),
95 sizeof(struct dispatch_source_s));
96 // Initialize as a queue first, then override some settings below.
97 _dispatch_queue_init((dispatch_queue_t)ds);
98 strlcpy(ds->dq_label, "source", sizeof(ds->dq_label));
99
100 // Dispatch Object
101 ds->do_ref_cnt++; // the reference the manger queue holds
102 ds->do_ref_cnt++; // since source is created suspended
103 ds->do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_INTERVAL;
104 // The initial target queue is the manager queue, in order to get
105 // the source installed. <rdar://problem/8928171>
106 ds->do_targetq = &_dispatch_mgr_q;
107
108 // Dispatch Source
109 ds->ds_ident_hack = dk->dk_kevent.ident;
110 ds->ds_dkev = dk;
111 ds->ds_pending_data_mask = dk->dk_kevent.fflags;
112 if ((EV_DISPATCH|EV_ONESHOT) & proto_kev->flags) {
113 ds->ds_is_level = true;
114 ds->ds_needs_rearm = true;
115 } else if (!(EV_CLEAR & proto_kev->flags)) {
116 // we cheat and use EV_CLEAR to mean a "flag thingy"
117 ds->ds_is_adder = true;
118 }
119
120 // Some sources require special processing
121 if (type->init != NULL) {
122 type->init(ds, type, handle, mask, q);
123 }
124 if (fastpath(!ds->ds_refs)) {
125 ds->ds_refs = calloc(1ul, sizeof(struct dispatch_source_refs_s));
126 if (slowpath(!ds->ds_refs)) {
127 goto out_bad;
128 }
129 }
130 ds->ds_refs->dr_source_wref = _dispatch_ptr2wref(ds);
131 dispatch_assert(!(ds->ds_is_level && ds->ds_is_adder));
132
133 // First item on the queue sets the user-specified target queue
134 dispatch_set_target_queue(ds, q);
135 #if DISPATCH_DEBUG
136 dispatch_debug(ds, "%s", __func__);
137 #endif
138 return ds;
139
140 out_bad:
141 free(ds);
142 free(dk);
143 return NULL;
144 }
145
146 void
147 _dispatch_source_dispose(dispatch_source_t ds)
148 {
149 free(ds->ds_refs);
150 _dispatch_queue_dispose((dispatch_queue_t)ds);
151 }
152
153 void
154 _dispatch_source_xref_dispose(dispatch_source_t ds)
155 {
156 _dispatch_wakeup(ds);
157 }
158
159 void
160 dispatch_source_cancel(dispatch_source_t ds)
161 {
162 #if DISPATCH_DEBUG
163 dispatch_debug(ds, "%s", __func__);
164 #endif
165 // Right after we set the cancel flag, someone else
166 // could potentially invoke the source, do the cancelation,
167 // unregister the source, and deallocate it. We would
168 // need to therefore retain/release before setting the bit
169
170 _dispatch_retain(ds);
171 (void)dispatch_atomic_or2o(ds, ds_atomic_flags, DSF_CANCELED);
172 _dispatch_wakeup(ds);
173 _dispatch_release(ds);
174 }
175
176 long
177 dispatch_source_testcancel(dispatch_source_t ds)
178 {
179 return (bool)(ds->ds_atomic_flags & DSF_CANCELED);
180 }
181
182
183 unsigned long
184 dispatch_source_get_mask(dispatch_source_t ds)
185 {
186 return ds->ds_pending_data_mask;
187 }
188
189 uintptr_t
190 dispatch_source_get_handle(dispatch_source_t ds)
191 {
192 return (int)ds->ds_ident_hack;
193 }
194
195 unsigned long
196 dispatch_source_get_data(dispatch_source_t ds)
197 {
198 return ds->ds_data;
199 }
200
201 void
202 dispatch_source_merge_data(dispatch_source_t ds, unsigned long val)
203 {
204 struct kevent kev = {
205 .fflags = (typeof(kev.fflags))val,
206 .data = val,
207 };
208
209 dispatch_assert(
210 ds->ds_dkev->dk_kevent.filter == DISPATCH_EVFILT_CUSTOM_ADD ||
211 ds->ds_dkev->dk_kevent.filter == DISPATCH_EVFILT_CUSTOM_OR);
212
213 _dispatch_source_merge_kevent(ds, &kev);
214 }
215
216 #pragma mark -
217 #pragma mark dispatch_source_handler
218
219 #ifdef __BLOCKS__
220 // 6618342 Contact the team that owns the Instrument DTrace probe before
221 // renaming this symbol
222 static void
223 _dispatch_source_set_event_handler2(void *context)
224 {
225 struct Block_layout *bl = context;
226
227 dispatch_source_t ds = (dispatch_source_t)_dispatch_queue_get_current();
228 dispatch_assert(dx_type(ds) == DISPATCH_SOURCE_KEVENT_TYPE);
229 dispatch_source_refs_t dr = ds->ds_refs;
230
231 if (ds->ds_handler_is_block && dr->ds_handler_ctxt) {
232 Block_release(dr->ds_handler_ctxt);
233 }
234 dr->ds_handler_func = bl ? (void *)bl->invoke : NULL;
235 dr->ds_handler_ctxt = bl;
236 ds->ds_handler_is_block = true;
237 }
238
239 void
240 dispatch_source_set_event_handler(dispatch_source_t ds,
241 dispatch_block_t handler)
242 {
243 handler = _dispatch_Block_copy(handler);
244 dispatch_barrier_async_f((dispatch_queue_t)ds, handler,
245 _dispatch_source_set_event_handler2);
246 }
247 #endif /* __BLOCKS__ */
248
249 static void
250 _dispatch_source_set_event_handler_f(void *context)
251 {
252 dispatch_source_t ds = (dispatch_source_t)_dispatch_queue_get_current();
253 dispatch_assert(dx_type(ds) == DISPATCH_SOURCE_KEVENT_TYPE);
254 dispatch_source_refs_t dr = ds->ds_refs;
255
256 #ifdef __BLOCKS__
257 if (ds->ds_handler_is_block && dr->ds_handler_ctxt) {
258 Block_release(dr->ds_handler_ctxt);
259 }
260 #endif
261 dr->ds_handler_func = context;
262 dr->ds_handler_ctxt = ds->do_ctxt;
263 ds->ds_handler_is_block = false;
264 }
265
266 void
267 dispatch_source_set_event_handler_f(dispatch_source_t ds,
268 dispatch_function_t handler)
269 {
270 dispatch_barrier_async_f((dispatch_queue_t)ds, handler,
271 _dispatch_source_set_event_handler_f);
272 }
273
274 #ifdef __BLOCKS__
275 // 6618342 Contact the team that owns the Instrument DTrace probe before
276 // renaming this symbol
277 static void
278 _dispatch_source_set_cancel_handler2(void *context)
279 {
280 dispatch_source_t ds = (dispatch_source_t)_dispatch_queue_get_current();
281 dispatch_assert(dx_type(ds) == DISPATCH_SOURCE_KEVENT_TYPE);
282 dispatch_source_refs_t dr = ds->ds_refs;
283
284 if (ds->ds_cancel_is_block && dr->ds_cancel_handler) {
285 Block_release(dr->ds_cancel_handler);
286 }
287 dr->ds_cancel_handler = context;
288 ds->ds_cancel_is_block = true;
289 }
290
291 void
292 dispatch_source_set_cancel_handler(dispatch_source_t ds,
293 dispatch_block_t handler)
294 {
295 handler = _dispatch_Block_copy(handler);
296 dispatch_barrier_async_f((dispatch_queue_t)ds, handler,
297 _dispatch_source_set_cancel_handler2);
298 }
299 #endif /* __BLOCKS__ */
300
301 static void
302 _dispatch_source_set_cancel_handler_f(void *context)
303 {
304 dispatch_source_t ds = (dispatch_source_t)_dispatch_queue_get_current();
305 dispatch_assert(dx_type(ds) == DISPATCH_SOURCE_KEVENT_TYPE);
306 dispatch_source_refs_t dr = ds->ds_refs;
307
308 #ifdef __BLOCKS__
309 if (ds->ds_cancel_is_block && dr->ds_cancel_handler) {
310 Block_release(dr->ds_cancel_handler);
311 }
312 #endif
313 dr->ds_cancel_handler = context;
314 ds->ds_cancel_is_block = false;
315 }
316
317 void
318 dispatch_source_set_cancel_handler_f(dispatch_source_t ds,
319 dispatch_function_t handler)
320 {
321 dispatch_barrier_async_f((dispatch_queue_t)ds, handler,
322 _dispatch_source_set_cancel_handler_f);
323 }
324
325 #ifdef __BLOCKS__
326 static void
327 _dispatch_source_set_registration_handler2(void *context)
328 {
329 dispatch_source_t ds = (dispatch_source_t)_dispatch_queue_get_current();
330 dispatch_assert(dx_type(ds) == DISPATCH_SOURCE_KEVENT_TYPE);
331 dispatch_source_refs_t dr = ds->ds_refs;
332
333 if (ds->ds_registration_is_block && dr->ds_registration_handler) {
334 Block_release(dr->ds_registration_handler);
335 }
336 dr->ds_registration_handler = context;
337 ds->ds_registration_is_block = true;
338 }
339
340 void
341 dispatch_source_set_registration_handler(dispatch_source_t ds,
342 dispatch_block_t handler)
343 {
344 handler = _dispatch_Block_copy(handler);
345 dispatch_barrier_async_f((dispatch_queue_t)ds, handler,
346 _dispatch_source_set_registration_handler2);
347 }
348 #endif /* __BLOCKS__ */
349
350 static void
351 _dispatch_source_set_registration_handler_f(void *context)
352 {
353 dispatch_source_t ds = (dispatch_source_t)_dispatch_queue_get_current();
354 dispatch_assert(dx_type(ds) == DISPATCH_SOURCE_KEVENT_TYPE);
355 dispatch_source_refs_t dr = ds->ds_refs;
356
357 #ifdef __BLOCKS__
358 if (ds->ds_registration_is_block && dr->ds_registration_handler) {
359 Block_release(dr->ds_registration_handler);
360 }
361 #endif
362 dr->ds_registration_handler = context;
363 ds->ds_registration_is_block = false;
364 }
365
366 void
367 dispatch_source_set_registration_handler_f(dispatch_source_t ds,
368 dispatch_function_t handler)
369 {
370 dispatch_barrier_async_f((dispatch_queue_t)ds, handler,
371 _dispatch_source_set_registration_handler_f);
372 }
373
374 #pragma mark -
375 #pragma mark dispatch_source_invoke
376
377 static void
378 _dispatch_source_registration_callout(dispatch_source_t ds)
379 {
380 dispatch_source_refs_t dr = ds->ds_refs;
381
382 if ((ds->ds_atomic_flags & DSF_CANCELED) || (ds->do_xref_cnt == -1)) {
383 // no registration callout if source is canceled rdar://problem/8955246
384 #ifdef __BLOCKS__
385 if (ds->ds_registration_is_block) {
386 Block_release(dr->ds_registration_handler);
387 }
388 } else if (ds->ds_registration_is_block) {
389 dispatch_block_t b = dr->ds_registration_handler;
390 _dispatch_client_callout_block(b);
391 Block_release(dr->ds_registration_handler);
392 #endif
393 } else {
394 dispatch_function_t f = dr->ds_registration_handler;
395 _dispatch_client_callout(ds->do_ctxt, f);
396 }
397 ds->ds_registration_is_block = false;
398 dr->ds_registration_handler = NULL;
399 }
400
401 static void
402 _dispatch_source_cancel_callout(dispatch_source_t ds)
403 {
404 dispatch_source_refs_t dr = ds->ds_refs;
405
406 ds->ds_pending_data_mask = 0;
407 ds->ds_pending_data = 0;
408 ds->ds_data = 0;
409
410 #ifdef __BLOCKS__
411 if (ds->ds_handler_is_block) {
412 Block_release(dr->ds_handler_ctxt);
413 ds->ds_handler_is_block = false;
414 dr->ds_handler_func = NULL;
415 dr->ds_handler_ctxt = NULL;
416 }
417 if (ds->ds_registration_is_block) {
418 Block_release(dr->ds_registration_handler);
419 ds->ds_registration_is_block = false;
420 dr->ds_registration_handler = NULL;
421 }
422 #endif
423
424 if (!dr->ds_cancel_handler) {
425 return;
426 }
427 if (ds->ds_cancel_is_block) {
428 #ifdef __BLOCKS__
429 dispatch_block_t b = dr->ds_cancel_handler;
430 if (ds->ds_atomic_flags & DSF_CANCELED) {
431 _dispatch_client_callout_block(b);
432 }
433 Block_release(dr->ds_cancel_handler);
434 ds->ds_cancel_is_block = false;
435 #endif
436 } else {
437 dispatch_function_t f = dr->ds_cancel_handler;
438 if (ds->ds_atomic_flags & DSF_CANCELED) {
439 _dispatch_client_callout(ds->do_ctxt, f);
440 }
441 }
442 dr->ds_cancel_handler = NULL;
443 }
444
445 static void
446 _dispatch_source_latch_and_call(dispatch_source_t ds)
447 {
448 unsigned long prev;
449
450 if ((ds->ds_atomic_flags & DSF_CANCELED) || (ds->do_xref_cnt == -1)) {
451 return;
452 }
453 dispatch_source_refs_t dr = ds->ds_refs;
454 prev = dispatch_atomic_xchg2o(ds, ds_pending_data, 0);
455 if (ds->ds_is_level) {
456 ds->ds_data = ~prev;
457 } else if (ds->ds_is_timer && ds_timer(dr).target && prev) {
458 ds->ds_data = _dispatch_source_timer_data(dr, prev);
459 } else {
460 ds->ds_data = prev;
461 }
462 if (dispatch_assume(prev) && dr->ds_handler_func) {
463 _dispatch_client_callout(dr->ds_handler_ctxt, dr->ds_handler_func);
464 }
465 }
466
467 static void
468 _dispatch_source_kevent_resume(dispatch_source_t ds, uint32_t new_flags)
469 {
470 switch (ds->ds_dkev->dk_kevent.filter) {
471 case DISPATCH_EVFILT_TIMER:
472 // called on manager queue only
473 return _dispatch_timer_list_update(ds);
474 case EVFILT_MACHPORT:
475 if (ds->ds_pending_data_mask & DISPATCH_MACH_RECV_MESSAGE) {
476 new_flags |= DISPATCH_MACH_RECV_MESSAGE; // emulate EV_DISPATCH
477 }
478 break;
479 }
480 if (_dispatch_kevent_resume(ds->ds_dkev, new_flags, 0)) {
481 _dispatch_kevent_unregister(ds);
482 }
483 }
484
485 dispatch_queue_t
486 _dispatch_source_invoke(dispatch_source_t ds)
487 {
488 // This function performs all source actions. Each action is responsible
489 // for verifying that it takes place on the appropriate queue. If the
490 // current queue is not the correct queue for this action, the correct queue
491 // will be returned and the invoke will be re-driven on that queue.
492
493 // The order of tests here in invoke and in probe should be consistent.
494
495 dispatch_queue_t dq = _dispatch_queue_get_current();
496 dispatch_source_refs_t dr = ds->ds_refs;
497
498 if (!ds->ds_is_installed) {
499 // The source needs to be installed on the manager queue.
500 if (dq != &_dispatch_mgr_q) {
501 return &_dispatch_mgr_q;
502 }
503 _dispatch_kevent_register(ds);
504 if (dr->ds_registration_handler) {
505 return ds->do_targetq;
506 }
507 if (slowpath(ds->do_xref_cnt == -1)) {
508 return &_dispatch_mgr_q; // rdar://problem/9558246
509 }
510 } else if (slowpath(DISPATCH_OBJECT_SUSPENDED(ds))) {
511 // Source suspended by an item drained from the source queue.
512 return NULL;
513 } else if (dr->ds_registration_handler) {
514 // The source has been registered and the registration handler needs
515 // to be delivered on the target queue.
516 if (dq != ds->do_targetq) {
517 return ds->do_targetq;
518 }
519 // clears ds_registration_handler
520 _dispatch_source_registration_callout(ds);
521 if (slowpath(ds->do_xref_cnt == -1)) {
522 return &_dispatch_mgr_q; // rdar://problem/9558246
523 }
524 } else if ((ds->ds_atomic_flags & DSF_CANCELED) || (ds->do_xref_cnt == -1)){
525 // The source has been cancelled and needs to be uninstalled from the
526 // manager queue. After uninstallation, the cancellation handler needs
527 // to be delivered to the target queue.
528 if (ds->ds_dkev) {
529 if (dq != &_dispatch_mgr_q) {
530 return &_dispatch_mgr_q;
531 }
532 _dispatch_kevent_unregister(ds);
533 }
534 if (dr->ds_cancel_handler || ds->ds_handler_is_block ||
535 ds->ds_registration_is_block) {
536 if (dq != ds->do_targetq) {
537 return ds->do_targetq;
538 }
539 }
540 _dispatch_source_cancel_callout(ds);
541 } else if (ds->ds_pending_data) {
542 // The source has pending data to deliver via the event handler callback
543 // on the target queue. Some sources need to be rearmed on the manager
544 // queue after event delivery.
545 if (dq != ds->do_targetq) {
546 return ds->do_targetq;
547 }
548 _dispatch_source_latch_and_call(ds);
549 if (ds->ds_needs_rearm) {
550 return &_dispatch_mgr_q;
551 }
552 } else if (ds->ds_needs_rearm && !(ds->ds_atomic_flags & DSF_ARMED)) {
553 // The source needs to be rearmed on the manager queue.
554 if (dq != &_dispatch_mgr_q) {
555 return &_dispatch_mgr_q;
556 }
557 _dispatch_source_kevent_resume(ds, 0);
558 (void)dispatch_atomic_or2o(ds, ds_atomic_flags, DSF_ARMED);
559 }
560
561 return NULL;
562 }
563
564 bool
565 _dispatch_source_probe(dispatch_source_t ds)
566 {
567 // This function determines whether the source needs to be invoked.
568 // The order of tests here in probe and in invoke should be consistent.
569
570 dispatch_source_refs_t dr = ds->ds_refs;
571 if (!ds->ds_is_installed) {
572 // The source needs to be installed on the manager queue.
573 return true;
574 } else if (dr->ds_registration_handler) {
575 // The registration handler needs to be delivered to the target queue.
576 return true;
577 } else if ((ds->ds_atomic_flags & DSF_CANCELED) || (ds->do_xref_cnt == -1)){
578 // The source needs to be uninstalled from the manager queue, or the
579 // cancellation handler needs to be delivered to the target queue.
580 // Note: cancellation assumes installation.
581 if (ds->ds_dkev || dr->ds_cancel_handler
582 #ifdef __BLOCKS__
583 || ds->ds_handler_is_block || ds->ds_registration_is_block
584 #endif
585 ) {
586 return true;
587 }
588 } else if (ds->ds_pending_data) {
589 // The source has pending data to deliver to the target queue.
590 return true;
591 } else if (ds->ds_needs_rearm && !(ds->ds_atomic_flags & DSF_ARMED)) {
592 // The source needs to be rearmed on the manager queue.
593 return true;
594 }
595 // Nothing to do.
596 return false;
597 }
598
599 #pragma mark -
600 #pragma mark dispatch_source_kevent
601
602 static void
603 _dispatch_source_merge_kevent(dispatch_source_t ds, const struct kevent *ke)
604 {
605 struct kevent fake;
606
607 if ((ds->ds_atomic_flags & DSF_CANCELED) || (ds->do_xref_cnt == -1)) {
608 return;
609 }
610
611 // EVFILT_PROC may fail with ESRCH when the process exists but is a zombie
612 // <rdar://problem/5067725>. As a workaround, we simulate an exit event for
613 // any EVFILT_PROC with an invalid pid <rdar://problem/6626350>.
614 if (ke->flags & EV_ERROR) {
615 if (ke->filter == EVFILT_PROC && ke->data == ESRCH) {
616 fake = *ke;
617 fake.flags &= ~EV_ERROR;
618 fake.fflags = NOTE_EXIT;
619 fake.data = 0;
620 ke = &fake;
621 #if DISPATCH_USE_VM_PRESSURE
622 } else if (ke->filter == EVFILT_VM && ke->data == ENOTSUP) {
623 // Memory pressure kevent is not supported on all platforms
624 // <rdar://problem/8636227>
625 return;
626 #endif
627 } else {
628 // log the unexpected error
629 (void)dispatch_assume_zero(ke->data);
630 return;
631 }
632 }
633
634 if (ds->ds_is_level) {
635 // ke->data is signed and "negative available data" makes no sense
636 // zero bytes happens when EV_EOF is set
637 // 10A268 does not fail this assert with EVFILT_READ and a 10 GB file
638 dispatch_assert(ke->data >= 0l);
639 ds->ds_pending_data = ~ke->data;
640 } else if (ds->ds_is_adder) {
641 (void)dispatch_atomic_add2o(ds, ds_pending_data, ke->data);
642 } else if (ke->fflags & ds->ds_pending_data_mask) {
643 (void)dispatch_atomic_or2o(ds, ds_pending_data,
644 ke->fflags & ds->ds_pending_data_mask);
645 }
646
647 // EV_DISPATCH and EV_ONESHOT sources are no longer armed after delivery
648 if (ds->ds_needs_rearm) {
649 (void)dispatch_atomic_and2o(ds, ds_atomic_flags, ~DSF_ARMED);
650 }
651
652 _dispatch_wakeup(ds);
653 }
654
655 void
656 _dispatch_source_drain_kevent(struct kevent *ke)
657 {
658 dispatch_kevent_t dk = ke->udata;
659 dispatch_source_refs_t dri;
660
661 #if DISPATCH_DEBUG
662 static dispatch_once_t pred;
663 dispatch_once_f(&pred, NULL, _dispatch_kevent_debugger);
664 #endif
665
666 dispatch_debug_kevents(ke, 1, __func__);
667
668 #if HAVE_MACH
669 if (ke->filter == EVFILT_MACHPORT) {
670 return _dispatch_drain_mach_messages(ke);
671 }
672 #endif
673 dispatch_assert(dk);
674
675 if (ke->flags & EV_ONESHOT) {
676 dk->dk_kevent.flags |= EV_ONESHOT;
677 }
678
679 TAILQ_FOREACH(dri, &dk->dk_sources, dr_list) {
680 _dispatch_source_merge_kevent(_dispatch_source_from_refs(dri), ke);
681 }
682 }
683
684 #pragma mark -
685 #pragma mark dispatch_kevent_t
686
687 static struct dispatch_kevent_s _dispatch_kevent_data_or = {
688 .dk_kevent = {
689 .filter = DISPATCH_EVFILT_CUSTOM_OR,
690 .flags = EV_CLEAR,
691 .udata = &_dispatch_kevent_data_or,
692 },
693 .dk_sources = TAILQ_HEAD_INITIALIZER(_dispatch_kevent_data_or.dk_sources),
694 };
695 static struct dispatch_kevent_s _dispatch_kevent_data_add = {
696 .dk_kevent = {
697 .filter = DISPATCH_EVFILT_CUSTOM_ADD,
698 .udata = &_dispatch_kevent_data_add,
699 },
700 .dk_sources = TAILQ_HEAD_INITIALIZER(_dispatch_kevent_data_add.dk_sources),
701 };
702
703 #if TARGET_OS_EMBEDDED
704 #define DSL_HASH_SIZE 64u // must be a power of two
705 #else
706 #define DSL_HASH_SIZE 256u // must be a power of two
707 #endif
708 #define DSL_HASH(x) ((x) & (DSL_HASH_SIZE - 1))
709
710 DISPATCH_CACHELINE_ALIGN
711 static TAILQ_HEAD(, dispatch_kevent_s) _dispatch_sources[DSL_HASH_SIZE];
712
713 static dispatch_once_t __dispatch_kevent_init_pred;
714
715 static void
716 _dispatch_kevent_init(void *context DISPATCH_UNUSED)
717 {
718 unsigned int i;
719 for (i = 0; i < DSL_HASH_SIZE; i++) {
720 TAILQ_INIT(&_dispatch_sources[i]);
721 }
722
723 TAILQ_INSERT_TAIL(&_dispatch_sources[0],
724 &_dispatch_kevent_data_or, dk_list);
725 TAILQ_INSERT_TAIL(&_dispatch_sources[0],
726 &_dispatch_kevent_data_add, dk_list);
727
728 _dispatch_source_timer_init();
729 }
730
731 static inline uintptr_t
732 _dispatch_kevent_hash(uintptr_t ident, short filter)
733 {
734 uintptr_t value;
735 #if HAVE_MACH
736 value = (filter == EVFILT_MACHPORT ? MACH_PORT_INDEX(ident) : ident);
737 #else
738 value = ident;
739 #endif
740 return DSL_HASH(value);
741 }
742
743 static dispatch_kevent_t
744 _dispatch_kevent_find(uintptr_t ident, short filter)
745 {
746 uintptr_t hash = _dispatch_kevent_hash(ident, filter);
747 dispatch_kevent_t dki;
748
749 TAILQ_FOREACH(dki, &_dispatch_sources[hash], dk_list) {
750 if (dki->dk_kevent.ident == ident && dki->dk_kevent.filter == filter) {
751 break;
752 }
753 }
754 return dki;
755 }
756
757 static void
758 _dispatch_kevent_insert(dispatch_kevent_t dk)
759 {
760 uintptr_t hash = _dispatch_kevent_hash(dk->dk_kevent.ident,
761 dk->dk_kevent.filter);
762
763 TAILQ_INSERT_TAIL(&_dispatch_sources[hash], dk, dk_list);
764 }
765
766 // Find existing kevents, and merge any new flags if necessary
767 static void
768 _dispatch_kevent_register(dispatch_source_t ds)
769 {
770 dispatch_kevent_t dk;
771 typeof(dk->dk_kevent.fflags) new_flags;
772 bool do_resume = false;
773
774 if (ds->ds_is_installed) {
775 return;
776 }
777 ds->ds_is_installed = true;
778
779 dispatch_once_f(&__dispatch_kevent_init_pred,
780 NULL, _dispatch_kevent_init);
781
782 dk = _dispatch_kevent_find(ds->ds_dkev->dk_kevent.ident,
783 ds->ds_dkev->dk_kevent.filter);
784
785 if (dk) {
786 // If an existing dispatch kevent is found, check to see if new flags
787 // need to be added to the existing kevent
788 new_flags = ~dk->dk_kevent.fflags & ds->ds_dkev->dk_kevent.fflags;
789 dk->dk_kevent.fflags |= ds->ds_dkev->dk_kevent.fflags;
790 free(ds->ds_dkev);
791 ds->ds_dkev = dk;
792 do_resume = new_flags;
793 } else {
794 dk = ds->ds_dkev;
795 _dispatch_kevent_insert(dk);
796 new_flags = dk->dk_kevent.fflags;
797 do_resume = true;
798 }
799
800 TAILQ_INSERT_TAIL(&dk->dk_sources, ds->ds_refs, dr_list);
801
802 // Re-register the kevent with the kernel if new flags were added
803 // by the dispatch kevent
804 if (do_resume) {
805 dk->dk_kevent.flags |= EV_ADD;
806 }
807 if (do_resume || ds->ds_needs_rearm) {
808 _dispatch_source_kevent_resume(ds, new_flags);
809 }
810 (void)dispatch_atomic_or2o(ds, ds_atomic_flags, DSF_ARMED);
811 }
812
813 static bool
814 _dispatch_kevent_resume(dispatch_kevent_t dk, uint32_t new_flags,
815 uint32_t del_flags)
816 {
817 long r;
818 switch (dk->dk_kevent.filter) {
819 case DISPATCH_EVFILT_TIMER:
820 case DISPATCH_EVFILT_CUSTOM_ADD:
821 case DISPATCH_EVFILT_CUSTOM_OR:
822 // these types not registered with kevent
823 return 0;
824 #if HAVE_MACH
825 case EVFILT_MACHPORT:
826 return _dispatch_kevent_machport_resume(dk, new_flags, del_flags);
827 #endif
828 case EVFILT_PROC:
829 if (dk->dk_kevent.flags & EV_ONESHOT) {
830 return 0;
831 }
832 // fall through
833 default:
834 r = _dispatch_update_kq(&dk->dk_kevent);
835 if (dk->dk_kevent.flags & EV_DISPATCH) {
836 dk->dk_kevent.flags &= ~EV_ADD;
837 }
838 return r;
839 }
840 }
841
842 static void
843 _dispatch_kevent_dispose(dispatch_kevent_t dk)
844 {
845 uintptr_t hash;
846
847 switch (dk->dk_kevent.filter) {
848 case DISPATCH_EVFILT_TIMER:
849 case DISPATCH_EVFILT_CUSTOM_ADD:
850 case DISPATCH_EVFILT_CUSTOM_OR:
851 // these sources live on statically allocated lists
852 return;
853 #if HAVE_MACH
854 case EVFILT_MACHPORT:
855 _dispatch_kevent_machport_resume(dk, 0, dk->dk_kevent.fflags);
856 break;
857 #endif
858 case EVFILT_PROC:
859 if (dk->dk_kevent.flags & EV_ONESHOT) {
860 break; // implicitly deleted
861 }
862 // fall through
863 default:
864 if (~dk->dk_kevent.flags & EV_DELETE) {
865 dk->dk_kevent.flags |= EV_DELETE;
866 _dispatch_update_kq(&dk->dk_kevent);
867 }
868 break;
869 }
870
871 hash = _dispatch_kevent_hash(dk->dk_kevent.ident,
872 dk->dk_kevent.filter);
873 TAILQ_REMOVE(&_dispatch_sources[hash], dk, dk_list);
874 free(dk);
875 }
876
877 static void
878 _dispatch_kevent_unregister(dispatch_source_t ds)
879 {
880 dispatch_kevent_t dk = ds->ds_dkev;
881 dispatch_source_refs_t dri;
882 uint32_t del_flags, fflags = 0;
883
884 ds->ds_dkev = NULL;
885
886 TAILQ_REMOVE(&dk->dk_sources, ds->ds_refs, dr_list);
887
888 if (TAILQ_EMPTY(&dk->dk_sources)) {
889 _dispatch_kevent_dispose(dk);
890 } else {
891 TAILQ_FOREACH(dri, &dk->dk_sources, dr_list) {
892 dispatch_source_t dsi = _dispatch_source_from_refs(dri);
893 fflags |= (uint32_t)dsi->ds_pending_data_mask;
894 }
895 del_flags = (uint32_t)ds->ds_pending_data_mask & ~fflags;
896 if (del_flags) {
897 dk->dk_kevent.flags |= EV_ADD;
898 dk->dk_kevent.fflags = fflags;
899 _dispatch_kevent_resume(dk, 0, del_flags);
900 }
901 }
902
903 (void)dispatch_atomic_and2o(ds, ds_atomic_flags, ~DSF_ARMED);
904 ds->ds_needs_rearm = false; // re-arm is pointless and bad now
905 _dispatch_release(ds); // the retain is done at creation time
906 }
907
908 #pragma mark -
909 #pragma mark dispatch_timer
910
911 DISPATCH_CACHELINE_ALIGN
912 static struct dispatch_kevent_s _dispatch_kevent_timer[] = {
913 [DISPATCH_TIMER_INDEX_WALL] = {
914 .dk_kevent = {
915 .ident = DISPATCH_TIMER_INDEX_WALL,
916 .filter = DISPATCH_EVFILT_TIMER,
917 .udata = &_dispatch_kevent_timer[DISPATCH_TIMER_INDEX_WALL],
918 },
919 .dk_sources = TAILQ_HEAD_INITIALIZER(
920 _dispatch_kevent_timer[DISPATCH_TIMER_INDEX_WALL].dk_sources),
921 },
922 [DISPATCH_TIMER_INDEX_MACH] = {
923 .dk_kevent = {
924 .ident = DISPATCH_TIMER_INDEX_MACH,
925 .filter = DISPATCH_EVFILT_TIMER,
926 .udata = &_dispatch_kevent_timer[DISPATCH_TIMER_INDEX_MACH],
927 },
928 .dk_sources = TAILQ_HEAD_INITIALIZER(
929 _dispatch_kevent_timer[DISPATCH_TIMER_INDEX_MACH].dk_sources),
930 },
931 [DISPATCH_TIMER_INDEX_DISARM] = {
932 .dk_kevent = {
933 .ident = DISPATCH_TIMER_INDEX_DISARM,
934 .filter = DISPATCH_EVFILT_TIMER,
935 .udata = &_dispatch_kevent_timer[DISPATCH_TIMER_INDEX_DISARM],
936 },
937 .dk_sources = TAILQ_HEAD_INITIALIZER(
938 _dispatch_kevent_timer[DISPATCH_TIMER_INDEX_DISARM].dk_sources),
939 },
940 };
941 // Don't count disarmed timer list
942 #define DISPATCH_TIMER_COUNT ((sizeof(_dispatch_kevent_timer) \
943 / sizeof(_dispatch_kevent_timer[0])) - 1)
944
945 static inline void
946 _dispatch_source_timer_init(void)
947 {
948 TAILQ_INSERT_TAIL(&_dispatch_sources[DSL_HASH(DISPATCH_TIMER_INDEX_WALL)],
949 &_dispatch_kevent_timer[DISPATCH_TIMER_INDEX_WALL], dk_list);
950 TAILQ_INSERT_TAIL(&_dispatch_sources[DSL_HASH(DISPATCH_TIMER_INDEX_MACH)],
951 &_dispatch_kevent_timer[DISPATCH_TIMER_INDEX_MACH], dk_list);
952 TAILQ_INSERT_TAIL(&_dispatch_sources[DSL_HASH(DISPATCH_TIMER_INDEX_DISARM)],
953 &_dispatch_kevent_timer[DISPATCH_TIMER_INDEX_DISARM], dk_list);
954 }
955
956 DISPATCH_ALWAYS_INLINE
957 static inline unsigned int
958 _dispatch_source_timer_idx(dispatch_source_refs_t dr)
959 {
960 return ds_timer(dr).flags & DISPATCH_TIMER_WALL_CLOCK ?
961 DISPATCH_TIMER_INDEX_WALL : DISPATCH_TIMER_INDEX_MACH;
962 }
963
964 DISPATCH_ALWAYS_INLINE
965 static inline uint64_t
966 _dispatch_source_timer_now2(unsigned int timer)
967 {
968 switch (timer) {
969 case DISPATCH_TIMER_INDEX_MACH:
970 return _dispatch_absolute_time();
971 case DISPATCH_TIMER_INDEX_WALL:
972 return _dispatch_get_nanoseconds();
973 default:
974 DISPATCH_CRASH("Invalid timer");
975 }
976 }
977
978 DISPATCH_ALWAYS_INLINE
979 static inline uint64_t
980 _dispatch_source_timer_now(dispatch_source_refs_t dr)
981 {
982 return _dispatch_source_timer_now2(_dispatch_source_timer_idx(dr));
983 }
984
985 // Updates the ordered list of timers based on next fire date for changes to ds.
986 // Should only be called from the context of _dispatch_mgr_q.
987 static void
988 _dispatch_timer_list_update(dispatch_source_t ds)
989 {
990 dispatch_source_refs_t dr = ds->ds_refs, dri = NULL;
991
992 dispatch_assert(_dispatch_queue_get_current() == &_dispatch_mgr_q);
993
994 // do not reschedule timers unregistered with _dispatch_kevent_unregister()
995 if (!ds->ds_dkev) {
996 return;
997 }
998
999 // Ensure the source is on the global kevent lists before it is removed and
1000 // readded below.
1001 _dispatch_kevent_register(ds);
1002
1003 TAILQ_REMOVE(&ds->ds_dkev->dk_sources, dr, dr_list);
1004
1005 // Move timers that are disabled, suspended or have missed intervals to the
1006 // disarmed list, rearm after resume resp. source invoke will reenable them
1007 if (!ds_timer(dr).target || DISPATCH_OBJECT_SUSPENDED(ds) ||
1008 ds->ds_pending_data) {
1009 (void)dispatch_atomic_and2o(ds, ds_atomic_flags, ~DSF_ARMED);
1010 ds->ds_dkev = &_dispatch_kevent_timer[DISPATCH_TIMER_INDEX_DISARM];
1011 TAILQ_INSERT_TAIL(&ds->ds_dkev->dk_sources, (dispatch_source_refs_t)dr,
1012 dr_list);
1013 return;
1014 }
1015
1016 // change the list if the clock type has changed
1017 ds->ds_dkev = &_dispatch_kevent_timer[_dispatch_source_timer_idx(dr)];
1018
1019 TAILQ_FOREACH(dri, &ds->ds_dkev->dk_sources, dr_list) {
1020 if (ds_timer(dri).target == 0 ||
1021 ds_timer(dr).target < ds_timer(dri).target) {
1022 break;
1023 }
1024 }
1025
1026 if (dri) {
1027 TAILQ_INSERT_BEFORE(dri, dr, dr_list);
1028 } else {
1029 TAILQ_INSERT_TAIL(&ds->ds_dkev->dk_sources, dr, dr_list);
1030 }
1031 }
1032
1033 static inline void
1034 _dispatch_run_timers2(unsigned int timer)
1035 {
1036 dispatch_source_refs_t dr;
1037 dispatch_source_t ds;
1038 uint64_t now, missed;
1039
1040 now = _dispatch_source_timer_now2(timer);
1041 while ((dr = TAILQ_FIRST(&_dispatch_kevent_timer[timer].dk_sources))) {
1042 ds = _dispatch_source_from_refs(dr);
1043 // We may find timers on the wrong list due to a pending update from
1044 // dispatch_source_set_timer. Force an update of the list in that case.
1045 if (timer != ds->ds_ident_hack) {
1046 _dispatch_timer_list_update(ds);
1047 continue;
1048 }
1049 if (!ds_timer(dr).target) {
1050 // no configured timers on the list
1051 break;
1052 }
1053 if (ds_timer(dr).target > now) {
1054 // Done running timers for now.
1055 break;
1056 }
1057 // Remove timers that are suspended or have missed intervals from the
1058 // list, rearm after resume resp. source invoke will reenable them
1059 if (DISPATCH_OBJECT_SUSPENDED(ds) || ds->ds_pending_data) {
1060 _dispatch_timer_list_update(ds);
1061 continue;
1062 }
1063 // Calculate number of missed intervals.
1064 missed = (now - ds_timer(dr).target) / ds_timer(dr).interval;
1065 if (++missed > INT_MAX) {
1066 missed = INT_MAX;
1067 }
1068 ds_timer(dr).target += missed * ds_timer(dr).interval;
1069 _dispatch_timer_list_update(ds);
1070 ds_timer(dr).last_fire = now;
1071 (void)dispatch_atomic_add2o(ds, ds_pending_data, (int)missed);
1072 _dispatch_wakeup(ds);
1073 }
1074 }
1075
1076 void
1077 _dispatch_run_timers(void)
1078 {
1079 dispatch_once_f(&__dispatch_kevent_init_pred,
1080 NULL, _dispatch_kevent_init);
1081
1082 unsigned int i;
1083 for (i = 0; i < DISPATCH_TIMER_COUNT; i++) {
1084 if (!TAILQ_EMPTY(&_dispatch_kevent_timer[i].dk_sources)) {
1085 _dispatch_run_timers2(i);
1086 }
1087 }
1088 }
1089
1090 static inline unsigned long
1091 _dispatch_source_timer_data(dispatch_source_refs_t dr, unsigned long prev)
1092 {
1093 // calculate the number of intervals since last fire
1094 unsigned long data, missed;
1095 uint64_t now = _dispatch_source_timer_now(dr);
1096 missed = (unsigned long)((now - ds_timer(dr).last_fire) /
1097 ds_timer(dr).interval);
1098 // correct for missed intervals already delivered last time
1099 data = prev - ds_timer(dr).missed + missed;
1100 ds_timer(dr).missed = missed;
1101 return data;
1102 }
1103
1104 // approx 1 year (60s * 60m * 24h * 365d)
1105 #define FOREVER_NSEC 31536000000000000ull
1106
1107 struct timespec *
1108 _dispatch_get_next_timer_fire(struct timespec *howsoon)
1109 {
1110 // <rdar://problem/6459649>
1111 // kevent(2) does not allow large timeouts, so we use a long timeout
1112 // instead (approximately 1 year).
1113 dispatch_source_refs_t dr = NULL;
1114 unsigned int timer;
1115 uint64_t now, delta_tmp, delta = UINT64_MAX;
1116
1117 for (timer = 0; timer < DISPATCH_TIMER_COUNT; timer++) {
1118 // Timers are kept in order, first one will fire next
1119 dr = TAILQ_FIRST(&_dispatch_kevent_timer[timer].dk_sources);
1120 if (!dr || !ds_timer(dr).target) {
1121 // Empty list or disabled timer
1122 continue;
1123 }
1124 now = _dispatch_source_timer_now(dr);
1125 if (ds_timer(dr).target <= now) {
1126 howsoon->tv_sec = 0;
1127 howsoon->tv_nsec = 0;
1128 return howsoon;
1129 }
1130 // the subtraction cannot go negative because the previous "if"
1131 // verified that the target is greater than now.
1132 delta_tmp = ds_timer(dr).target - now;
1133 if (!(ds_timer(dr).flags & DISPATCH_TIMER_WALL_CLOCK)) {
1134 delta_tmp = _dispatch_time_mach2nano(delta_tmp);
1135 }
1136 if (delta_tmp < delta) {
1137 delta = delta_tmp;
1138 }
1139 }
1140 if (slowpath(delta > FOREVER_NSEC)) {
1141 return NULL;
1142 } else {
1143 howsoon->tv_sec = (time_t)(delta / NSEC_PER_SEC);
1144 howsoon->tv_nsec = (long)(delta % NSEC_PER_SEC);
1145 }
1146 return howsoon;
1147 }
1148
1149 struct dispatch_set_timer_params {
1150 dispatch_source_t ds;
1151 uintptr_t ident;
1152 struct dispatch_timer_source_s values;
1153 };
1154
1155 static void
1156 _dispatch_source_set_timer3(void *context)
1157 {
1158 // Called on the _dispatch_mgr_q
1159 struct dispatch_set_timer_params *params = context;
1160 dispatch_source_t ds = params->ds;
1161 ds->ds_ident_hack = params->ident;
1162 ds_timer(ds->ds_refs) = params->values;
1163 // Clear any pending data that might have accumulated on
1164 // older timer params <rdar://problem/8574886>
1165 ds->ds_pending_data = 0;
1166 _dispatch_timer_list_update(ds);
1167 dispatch_resume(ds);
1168 dispatch_release(ds);
1169 free(params);
1170 }
1171
1172 static void
1173 _dispatch_source_set_timer2(void *context)
1174 {
1175 // Called on the source queue
1176 struct dispatch_set_timer_params *params = context;
1177 dispatch_suspend(params->ds);
1178 dispatch_barrier_async_f(&_dispatch_mgr_q, params,
1179 _dispatch_source_set_timer3);
1180 }
1181
1182 void
1183 dispatch_source_set_timer(dispatch_source_t ds,
1184 dispatch_time_t start,
1185 uint64_t interval,
1186 uint64_t leeway)
1187 {
1188 if (slowpath(!ds->ds_is_timer)) {
1189 DISPATCH_CLIENT_CRASH("Attempt to set timer on a non-timer source");
1190 }
1191
1192 struct dispatch_set_timer_params *params;
1193
1194 // we use zero internally to mean disabled
1195 if (interval == 0) {
1196 interval = 1;
1197 } else if ((int64_t)interval < 0) {
1198 // 6866347 - make sure nanoseconds won't overflow
1199 interval = INT64_MAX;
1200 }
1201 if ((int64_t)leeway < 0) {
1202 leeway = INT64_MAX;
1203 }
1204
1205 if (start == DISPATCH_TIME_NOW) {
1206 start = _dispatch_absolute_time();
1207 } else if (start == DISPATCH_TIME_FOREVER) {
1208 start = INT64_MAX;
1209 }
1210
1211 while (!(params = calloc(1ul, sizeof(struct dispatch_set_timer_params)))) {
1212 sleep(1);
1213 }
1214
1215 params->ds = ds;
1216 params->values.flags = ds_timer(ds->ds_refs).flags;
1217
1218 if ((int64_t)start < 0) {
1219 // wall clock
1220 params->ident = DISPATCH_TIMER_INDEX_WALL;
1221 params->values.target = -((int64_t)start);
1222 params->values.interval = interval;
1223 params->values.leeway = leeway;
1224 params->values.flags |= DISPATCH_TIMER_WALL_CLOCK;
1225 } else {
1226 // absolute clock
1227 params->ident = DISPATCH_TIMER_INDEX_MACH;
1228 params->values.target = start;
1229 params->values.interval = _dispatch_time_nano2mach(interval);
1230
1231 // rdar://problem/7287561 interval must be at least one in
1232 // in order to avoid later division by zero when calculating
1233 // the missed interval count. (NOTE: the wall clock's
1234 // interval is already "fixed" to be 1 or more)
1235 if (params->values.interval < 1) {
1236 params->values.interval = 1;
1237 }
1238
1239 params->values.leeway = _dispatch_time_nano2mach(leeway);
1240 params->values.flags &= ~DISPATCH_TIMER_WALL_CLOCK;
1241 }
1242 // Suspend the source so that it doesn't fire with pending changes
1243 // The use of suspend/resume requires the external retain/release
1244 dispatch_retain(ds);
1245 dispatch_barrier_async_f((dispatch_queue_t)ds, params,
1246 _dispatch_source_set_timer2);
1247 }
1248
1249 #pragma mark -
1250 #pragma mark dispatch_mach
1251
1252 #if HAVE_MACH
1253
1254 #if DISPATCH_DEBUG && DISPATCH_MACHPORT_DEBUG
1255 #define _dispatch_debug_machport(name) \
1256 dispatch_debug_machport((name), __func__)
1257 #else
1258 #define _dispatch_debug_machport(name)
1259 #endif
1260
1261 // Flags for all notifications that are registered/unregistered when a
1262 // send-possible notification is requested/delivered
1263 #define _DISPATCH_MACH_SP_FLAGS (DISPATCH_MACH_SEND_POSSIBLE| \
1264 DISPATCH_MACH_SEND_DEAD|DISPATCH_MACH_SEND_DELETED)
1265
1266 #define _DISPATCH_IS_POWER_OF_TWO(v) (!(v & (v - 1)) && v)
1267 #define _DISPATCH_HASH(x, y) (_DISPATCH_IS_POWER_OF_TWO(y) ? \
1268 (MACH_PORT_INDEX(x) & ((y) - 1)) : (MACH_PORT_INDEX(x) % (y)))
1269
1270 #define _DISPATCH_MACHPORT_HASH_SIZE 32
1271 #define _DISPATCH_MACHPORT_HASH(x) \
1272 _DISPATCH_HASH((x), _DISPATCH_MACHPORT_HASH_SIZE)
1273
1274 static dispatch_source_t _dispatch_mach_notify_source;
1275 static mach_port_t _dispatch_port_set;
1276 static mach_port_t _dispatch_event_port;
1277
1278 static kern_return_t _dispatch_mach_notify_update(dispatch_kevent_t dk,
1279 uint32_t new_flags, uint32_t del_flags, uint32_t mask,
1280 mach_msg_id_t notify_msgid, mach_port_mscount_t notify_sync);
1281
1282 static void
1283 _dispatch_port_set_init(void *context DISPATCH_UNUSED)
1284 {
1285 struct kevent kev = {
1286 .filter = EVFILT_MACHPORT,
1287 .flags = EV_ADD,
1288 };
1289 kern_return_t kr;
1290
1291 kr = mach_port_allocate(mach_task_self(), MACH_PORT_RIGHT_PORT_SET,
1292 &_dispatch_port_set);
1293 DISPATCH_VERIFY_MIG(kr);
1294 if (kr) {
1295 _dispatch_bug_mach_client(
1296 "_dispatch_port_set_init: mach_port_allocate() failed", kr);
1297 DISPATCH_CLIENT_CRASH(
1298 "mach_port_allocate() failed: cannot create port set");
1299 }
1300 kr = mach_port_allocate(mach_task_self(), MACH_PORT_RIGHT_RECEIVE,
1301 &_dispatch_event_port);
1302 DISPATCH_VERIFY_MIG(kr);
1303 if (kr) {
1304 _dispatch_bug_mach_client(
1305 "_dispatch_port_set_init: mach_port_allocate() failed", kr);
1306 DISPATCH_CLIENT_CRASH(
1307 "mach_port_allocate() failed: cannot create receive right");
1308 }
1309 kr = mach_port_move_member(mach_task_self(), _dispatch_event_port,
1310 _dispatch_port_set);
1311 DISPATCH_VERIFY_MIG(kr);
1312 if (kr) {
1313 _dispatch_bug_mach_client(
1314 "_dispatch_port_set_init: mach_port_move_member() failed", kr);
1315 DISPATCH_CLIENT_CRASH("mach_port_move_member() failed");
1316 }
1317
1318 kev.ident = _dispatch_port_set;
1319
1320 _dispatch_update_kq(&kev);
1321 }
1322
1323 static mach_port_t
1324 _dispatch_get_port_set(void)
1325 {
1326 static dispatch_once_t pred;
1327
1328 dispatch_once_f(&pred, NULL, _dispatch_port_set_init);
1329
1330 return _dispatch_port_set;
1331 }
1332
1333 static kern_return_t
1334 _dispatch_kevent_machport_enable(dispatch_kevent_t dk)
1335 {
1336 mach_port_t mp = (mach_port_t)dk->dk_kevent.ident;
1337 kern_return_t kr;
1338
1339 _dispatch_debug_machport(mp);
1340 kr = mach_port_move_member(mach_task_self(), mp, _dispatch_get_port_set());
1341 if (slowpath(kr)) {
1342 DISPATCH_VERIFY_MIG(kr);
1343 switch (kr) {
1344 case KERN_INVALID_NAME:
1345 #if DISPATCH_DEBUG
1346 _dispatch_log("Corruption: Mach receive right 0x%x destroyed "
1347 "prematurely", mp);
1348 #endif
1349 break;
1350 case KERN_INVALID_RIGHT:
1351 _dispatch_bug_mach_client("_dispatch_kevent_machport_enable: "
1352 "mach_port_move_member() failed ", kr);
1353 break;
1354 default:
1355 (void)dispatch_assume_zero(kr);
1356 break;
1357 }
1358 }
1359 return kr;
1360 }
1361
1362 static void
1363 _dispatch_kevent_machport_disable(dispatch_kevent_t dk)
1364 {
1365 mach_port_t mp = (mach_port_t)dk->dk_kevent.ident;
1366 kern_return_t kr;
1367
1368 _dispatch_debug_machport(mp);
1369 kr = mach_port_move_member(mach_task_self(), mp, 0);
1370 if (slowpath(kr)) {
1371 DISPATCH_VERIFY_MIG(kr);
1372 switch (kr) {
1373 case KERN_INVALID_RIGHT:
1374 case KERN_INVALID_NAME:
1375 #if DISPATCH_DEBUG
1376 _dispatch_log("Corruption: Mach receive right 0x%x destroyed "
1377 "prematurely", mp);
1378 #endif
1379 break;
1380 default:
1381 (void)dispatch_assume_zero(kr);
1382 break;
1383 }
1384 }
1385 }
1386
1387 kern_return_t
1388 _dispatch_kevent_machport_resume(dispatch_kevent_t dk, uint32_t new_flags,
1389 uint32_t del_flags)
1390 {
1391 kern_return_t kr_recv = 0, kr_sp = 0;
1392
1393 dispatch_assert_zero(new_flags & del_flags);
1394 if (new_flags & DISPATCH_MACH_RECV_MESSAGE) {
1395 kr_recv = _dispatch_kevent_machport_enable(dk);
1396 } else if (del_flags & DISPATCH_MACH_RECV_MESSAGE) {
1397 _dispatch_kevent_machport_disable(dk);
1398 }
1399 if ((new_flags & _DISPATCH_MACH_SP_FLAGS) ||
1400 (del_flags & _DISPATCH_MACH_SP_FLAGS)) {
1401 // Requesting a (delayed) non-sync send-possible notification
1402 // registers for both immediate dead-name notification and delayed-arm
1403 // send-possible notification for the port.
1404 // The send-possible notification is armed when a mach_msg() with the
1405 // the MACH_SEND_NOTIFY to the port times out.
1406 // If send-possible is unavailable, fall back to immediate dead-name
1407 // registration rdar://problem/2527840&9008724
1408 kr_sp = _dispatch_mach_notify_update(dk, new_flags, del_flags,
1409 _DISPATCH_MACH_SP_FLAGS, MACH_NOTIFY_SEND_POSSIBLE,
1410 MACH_NOTIFY_SEND_POSSIBLE == MACH_NOTIFY_DEAD_NAME ? 1 : 0);
1411 }
1412
1413 return (kr_recv ? kr_recv : kr_sp);
1414 }
1415
1416 void
1417 _dispatch_drain_mach_messages(struct kevent *ke)
1418 {
1419 mach_port_t name = (mach_port_name_t)ke->data;
1420 dispatch_source_refs_t dri;
1421 dispatch_kevent_t dk;
1422 struct kevent kev;
1423
1424 if (!dispatch_assume(name)) {
1425 return;
1426 }
1427 _dispatch_debug_machport(name);
1428 dk = _dispatch_kevent_find(name, EVFILT_MACHPORT);
1429 if (!dispatch_assume(dk)) {
1430 return;
1431 }
1432 _dispatch_kevent_machport_disable(dk); // emulate EV_DISPATCH
1433
1434 EV_SET(&kev, name, EVFILT_MACHPORT, EV_ADD|EV_ENABLE|EV_DISPATCH,
1435 DISPATCH_MACH_RECV_MESSAGE, 0, dk);
1436
1437 TAILQ_FOREACH(dri, &dk->dk_sources, dr_list) {
1438 _dispatch_source_merge_kevent(_dispatch_source_from_refs(dri), &kev);
1439 }
1440 }
1441
1442 static inline void
1443 _dispatch_mach_notify_merge(mach_port_t name, uint32_t flag, uint32_t unreg,
1444 bool final)
1445 {
1446 dispatch_source_refs_t dri;
1447 dispatch_kevent_t dk;
1448 struct kevent kev;
1449
1450 dk = _dispatch_kevent_find(name, EVFILT_MACHPORT);
1451 if (!dk) {
1452 return;
1453 }
1454
1455 // Update notification registration state.
1456 dk->dk_kevent.data &= ~unreg;
1457 if (!final) {
1458 // Re-register for notification before delivery
1459 _dispatch_kevent_resume(dk, flag, 0);
1460 }
1461
1462 EV_SET(&kev, name, EVFILT_MACHPORT, EV_ADD|EV_ENABLE, flag, 0, dk);
1463
1464 TAILQ_FOREACH(dri, &dk->dk_sources, dr_list) {
1465 _dispatch_source_merge_kevent(_dispatch_source_from_refs(dri), &kev);
1466 if (final) {
1467 // this can never happen again
1468 // this must happen after the merge
1469 // this may be racy in the future, but we don't provide a 'setter'
1470 // API for the mask yet
1471 _dispatch_source_from_refs(dri)->ds_pending_data_mask &= ~unreg;
1472 }
1473 }
1474
1475 if (final) {
1476 // no more sources have these flags
1477 dk->dk_kevent.fflags &= ~unreg;
1478 }
1479 }
1480
1481 static kern_return_t
1482 _dispatch_mach_notify_update(dispatch_kevent_t dk, uint32_t new_flags,
1483 uint32_t del_flags, uint32_t mask, mach_msg_id_t notify_msgid,
1484 mach_port_mscount_t notify_sync)
1485 {
1486 mach_port_t previous, port = (mach_port_t)dk->dk_kevent.ident;
1487 typeof(dk->dk_kevent.data) prev = dk->dk_kevent.data;
1488 kern_return_t kr, krr = 0;
1489
1490 // Update notification registration state.
1491 dk->dk_kevent.data |= (new_flags | dk->dk_kevent.fflags) & mask;
1492 dk->dk_kevent.data &= ~(del_flags & mask);
1493
1494 _dispatch_debug_machport(port);
1495 if ((dk->dk_kevent.data & mask) && !(prev & mask)) {
1496 previous = MACH_PORT_NULL;
1497 krr = mach_port_request_notification(mach_task_self(), port,
1498 notify_msgid, notify_sync, _dispatch_event_port,
1499 MACH_MSG_TYPE_MAKE_SEND_ONCE, &previous);
1500 DISPATCH_VERIFY_MIG(krr);
1501
1502 switch(krr) {
1503 case KERN_INVALID_NAME:
1504 case KERN_INVALID_RIGHT:
1505 // Supress errors & clear registration state
1506 dk->dk_kevent.data &= ~mask;
1507 break;
1508 default:
1509 // Else, we dont expect any errors from mach. Log any errors
1510 if (dispatch_assume_zero(krr)) {
1511 // log the error & clear registration state
1512 dk->dk_kevent.data &= ~mask;
1513 } else if (dispatch_assume_zero(previous)) {
1514 // Another subsystem has beat libdispatch to requesting the
1515 // specified Mach notification on this port. We should
1516 // technically cache the previous port and message it when the
1517 // kernel messages our port. Or we can just say screw those
1518 // subsystems and deallocate the previous port.
1519 // They should adopt libdispatch :-P
1520 kr = mach_port_deallocate(mach_task_self(), previous);
1521 DISPATCH_VERIFY_MIG(kr);
1522 (void)dispatch_assume_zero(kr);
1523 previous = MACH_PORT_NULL;
1524 }
1525 }
1526 } else if (!(dk->dk_kevent.data & mask) && (prev & mask)) {
1527 previous = MACH_PORT_NULL;
1528 kr = mach_port_request_notification(mach_task_self(), port,
1529 notify_msgid, notify_sync, MACH_PORT_NULL,
1530 MACH_MSG_TYPE_MOVE_SEND_ONCE, &previous);
1531 DISPATCH_VERIFY_MIG(kr);
1532
1533 switch (kr) {
1534 case KERN_INVALID_NAME:
1535 case KERN_INVALID_RIGHT:
1536 case KERN_INVALID_ARGUMENT:
1537 break;
1538 default:
1539 if (dispatch_assume_zero(kr)) {
1540 // log the error
1541 }
1542 }
1543 } else {
1544 return 0;
1545 }
1546 if (slowpath(previous)) {
1547 // the kernel has not consumed the send-once right yet
1548 (void)dispatch_assume_zero(
1549 _dispatch_send_consume_send_once_right(previous));
1550 }
1551 return krr;
1552 }
1553
1554 static void
1555 _dispatch_mach_notify_source2(void *context)
1556 {
1557 dispatch_source_t ds = context;
1558 size_t maxsz = MAX(sizeof(union
1559 __RequestUnion___dispatch_send_libdispatch_internal_protocol_subsystem),
1560 sizeof(union
1561 __ReplyUnion___dispatch_libdispatch_internal_protocol_subsystem));
1562
1563 dispatch_mig_server(ds, maxsz, libdispatch_internal_protocol_server);
1564 }
1565
1566 void
1567 _dispatch_mach_notify_source_init(void *context DISPATCH_UNUSED)
1568 {
1569 _dispatch_get_port_set();
1570
1571 _dispatch_mach_notify_source = dispatch_source_create(
1572 DISPATCH_SOURCE_TYPE_MACH_RECV, _dispatch_event_port, 0,
1573 &_dispatch_mgr_q);
1574 dispatch_assert(_dispatch_mach_notify_source);
1575 dispatch_set_context(_dispatch_mach_notify_source,
1576 _dispatch_mach_notify_source);
1577 dispatch_source_set_event_handler_f(_dispatch_mach_notify_source,
1578 _dispatch_mach_notify_source2);
1579 dispatch_resume(_dispatch_mach_notify_source);
1580 }
1581
1582 kern_return_t
1583 _dispatch_mach_notify_port_deleted(mach_port_t notify DISPATCH_UNUSED,
1584 mach_port_name_t name)
1585 {
1586 #if DISPATCH_DEBUG
1587 _dispatch_log("Corruption: Mach send/send-once/dead-name right 0x%x "
1588 "deleted prematurely", name);
1589 #endif
1590
1591 _dispatch_debug_machport(name);
1592 _dispatch_mach_notify_merge(name, DISPATCH_MACH_SEND_DELETED,
1593 _DISPATCH_MACH_SP_FLAGS, true);
1594
1595 return KERN_SUCCESS;
1596 }
1597
1598 kern_return_t
1599 _dispatch_mach_notify_dead_name(mach_port_t notify DISPATCH_UNUSED,
1600 mach_port_name_t name)
1601 {
1602 kern_return_t kr;
1603
1604 #if DISPATCH_DEBUG
1605 _dispatch_log("machport[0x%08x]: dead-name notification: %s",
1606 name, __func__);
1607 #endif
1608 _dispatch_debug_machport(name);
1609 _dispatch_mach_notify_merge(name, DISPATCH_MACH_SEND_DEAD,
1610 _DISPATCH_MACH_SP_FLAGS, true);
1611
1612 // the act of receiving a dead name notification allocates a dead-name
1613 // right that must be deallocated
1614 kr = mach_port_deallocate(mach_task_self(), name);
1615 DISPATCH_VERIFY_MIG(kr);
1616 //(void)dispatch_assume_zero(kr);
1617
1618 return KERN_SUCCESS;
1619 }
1620
1621 kern_return_t
1622 _dispatch_mach_notify_send_possible(mach_port_t notify DISPATCH_UNUSED,
1623 mach_port_name_t name)
1624 {
1625 #if DISPATCH_DEBUG
1626 _dispatch_log("machport[0x%08x]: send-possible notification: %s",
1627 name, __func__);
1628 #endif
1629 _dispatch_debug_machport(name);
1630 _dispatch_mach_notify_merge(name, DISPATCH_MACH_SEND_POSSIBLE,
1631 _DISPATCH_MACH_SP_FLAGS, false);
1632
1633 return KERN_SUCCESS;
1634 }
1635
1636 mach_msg_return_t
1637 dispatch_mig_server(dispatch_source_t ds, size_t maxmsgsz,
1638 dispatch_mig_callback_t callback)
1639 {
1640 mach_msg_options_t options = MACH_RCV_MSG | MACH_RCV_TIMEOUT
1641 | MACH_RCV_TRAILER_ELEMENTS(MACH_RCV_TRAILER_CTX)
1642 | MACH_RCV_TRAILER_TYPE(MACH_MSG_TRAILER_FORMAT_0);
1643 mach_msg_options_t tmp_options;
1644 mig_reply_error_t *bufTemp, *bufRequest, *bufReply;
1645 mach_msg_return_t kr = 0;
1646 unsigned int cnt = 1000; // do not stall out serial queues
1647 int demux_success;
1648 bool received = false;
1649 size_t rcv_size = maxmsgsz + MAX_TRAILER_SIZE;
1650
1651 // XXX FIXME -- allocate these elsewhere
1652 bufRequest = alloca(rcv_size);
1653 bufReply = alloca(rcv_size);
1654 bufReply->Head.msgh_size = 0; // make CLANG happy
1655 bufRequest->RetCode = 0;
1656
1657 #if DISPATCH_DEBUG
1658 options |= MACH_RCV_LARGE; // rdar://problem/8422992
1659 #endif
1660 tmp_options = options;
1661 // XXX FIXME -- change this to not starve out the target queue
1662 for (;;) {
1663 if (DISPATCH_OBJECT_SUSPENDED(ds) || (--cnt == 0)) {
1664 options &= ~MACH_RCV_MSG;
1665 tmp_options &= ~MACH_RCV_MSG;
1666
1667 if (!(tmp_options & MACH_SEND_MSG)) {
1668 break;
1669 }
1670 }
1671 kr = mach_msg(&bufReply->Head, tmp_options, bufReply->Head.msgh_size,
1672 (mach_msg_size_t)rcv_size, (mach_port_t)ds->ds_ident_hack, 0,0);
1673
1674 tmp_options = options;
1675
1676 if (slowpath(kr)) {
1677 switch (kr) {
1678 case MACH_SEND_INVALID_DEST:
1679 case MACH_SEND_TIMED_OUT:
1680 if (bufReply->Head.msgh_bits & MACH_MSGH_BITS_COMPLEX) {
1681 mach_msg_destroy(&bufReply->Head);
1682 }
1683 break;
1684 case MACH_RCV_TIMED_OUT:
1685 // Don't return an error if a message was sent this time or
1686 // a message was successfully received previously
1687 // rdar://problems/7363620&7791738
1688 if(bufReply->Head.msgh_remote_port || received) {
1689 kr = MACH_MSG_SUCCESS;
1690 }
1691 break;
1692 case MACH_RCV_INVALID_NAME:
1693 break;
1694 #if DISPATCH_DEBUG
1695 case MACH_RCV_TOO_LARGE:
1696 // receive messages that are too large and log their id and size
1697 // rdar://problem/8422992
1698 tmp_options &= ~MACH_RCV_LARGE;
1699 size_t large_size = bufReply->Head.msgh_size + MAX_TRAILER_SIZE;
1700 void *large_buf = malloc(large_size);
1701 if (large_buf) {
1702 rcv_size = large_size;
1703 bufReply = large_buf;
1704 }
1705 if (!mach_msg(&bufReply->Head, tmp_options, 0,
1706 (mach_msg_size_t)rcv_size,
1707 (mach_port_t)ds->ds_ident_hack, 0, 0)) {
1708 _dispatch_log("BUG in libdispatch client: "
1709 "dispatch_mig_server received message larger than "
1710 "requested size %zd: id = 0x%x, size = %d",
1711 maxmsgsz, bufReply->Head.msgh_id,
1712 bufReply->Head.msgh_size);
1713 }
1714 if (large_buf) {
1715 free(large_buf);
1716 }
1717 // fall through
1718 #endif
1719 default:
1720 _dispatch_bug_mach_client(
1721 "dispatch_mig_server: mach_msg() failed", kr);
1722 break;
1723 }
1724 break;
1725 }
1726
1727 if (!(tmp_options & MACH_RCV_MSG)) {
1728 break;
1729 }
1730 received = true;
1731
1732 bufTemp = bufRequest;
1733 bufRequest = bufReply;
1734 bufReply = bufTemp;
1735
1736 demux_success = callback(&bufRequest->Head, &bufReply->Head);
1737
1738 if (!demux_success) {
1739 // destroy the request - but not the reply port
1740 bufRequest->Head.msgh_remote_port = 0;
1741 mach_msg_destroy(&bufRequest->Head);
1742 } else if (!(bufReply->Head.msgh_bits & MACH_MSGH_BITS_COMPLEX)) {
1743 // if MACH_MSGH_BITS_COMPLEX is _not_ set, then bufReply->RetCode
1744 // is present
1745 if (slowpath(bufReply->RetCode)) {
1746 if (bufReply->RetCode == MIG_NO_REPLY) {
1747 continue;
1748 }
1749
1750 // destroy the request - but not the reply port
1751 bufRequest->Head.msgh_remote_port = 0;
1752 mach_msg_destroy(&bufRequest->Head);
1753 }
1754 }
1755
1756 if (bufReply->Head.msgh_remote_port) {
1757 tmp_options |= MACH_SEND_MSG;
1758 if (MACH_MSGH_BITS_REMOTE(bufReply->Head.msgh_bits) !=
1759 MACH_MSG_TYPE_MOVE_SEND_ONCE) {
1760 tmp_options |= MACH_SEND_TIMEOUT;
1761 }
1762 }
1763 }
1764
1765 return kr;
1766 }
1767
1768 #endif /* HAVE_MACH */
1769
1770 #pragma mark -
1771 #pragma mark dispatch_source_debug
1772
1773 DISPATCH_NOINLINE
1774 static const char *
1775 _evfiltstr(short filt)
1776 {
1777 switch (filt) {
1778 #define _evfilt2(f) case (f): return #f
1779 _evfilt2(EVFILT_READ);
1780 _evfilt2(EVFILT_WRITE);
1781 _evfilt2(EVFILT_AIO);
1782 _evfilt2(EVFILT_VNODE);
1783 _evfilt2(EVFILT_PROC);
1784 _evfilt2(EVFILT_SIGNAL);
1785 _evfilt2(EVFILT_TIMER);
1786 #ifdef EVFILT_VM
1787 _evfilt2(EVFILT_VM);
1788 #endif
1789 #if HAVE_MACH
1790 _evfilt2(EVFILT_MACHPORT);
1791 #endif
1792 _evfilt2(EVFILT_FS);
1793 _evfilt2(EVFILT_USER);
1794
1795 _evfilt2(DISPATCH_EVFILT_TIMER);
1796 _evfilt2(DISPATCH_EVFILT_CUSTOM_ADD);
1797 _evfilt2(DISPATCH_EVFILT_CUSTOM_OR);
1798 default:
1799 return "EVFILT_missing";
1800 }
1801 }
1802
1803 static size_t
1804 _dispatch_source_debug_attr(dispatch_source_t ds, char* buf, size_t bufsiz)
1805 {
1806 dispatch_queue_t target = ds->do_targetq;
1807 return snprintf(buf, bufsiz, "target = %s[%p], pending_data = 0x%lx, "
1808 "pending_data_mask = 0x%lx, ",
1809 target ? target->dq_label : "", target,
1810 ds->ds_pending_data, ds->ds_pending_data_mask);
1811 }
1812
1813 static size_t
1814 _dispatch_timer_debug_attr(dispatch_source_t ds, char* buf, size_t bufsiz)
1815 {
1816 dispatch_source_refs_t dr = ds->ds_refs;
1817 return snprintf(buf, bufsiz, "timer = { target = 0x%llx, "
1818 "last_fire = 0x%llx, interval = 0x%llx, flags = 0x%llx }, ",
1819 ds_timer(dr).target, ds_timer(dr).last_fire, ds_timer(dr).interval,
1820 ds_timer(dr).flags);
1821 }
1822
1823 size_t
1824 _dispatch_source_debug(dispatch_source_t ds, char* buf, size_t bufsiz)
1825 {
1826 size_t offset = 0;
1827 offset += snprintf(&buf[offset], bufsiz - offset, "%s[%p] = { ",
1828 dx_kind(ds), ds);
1829 offset += _dispatch_object_debug_attr(ds, &buf[offset], bufsiz - offset);
1830 offset += _dispatch_source_debug_attr(ds, &buf[offset], bufsiz - offset);
1831 if (ds->ds_is_timer) {
1832 offset += _dispatch_timer_debug_attr(ds, &buf[offset], bufsiz - offset);
1833 }
1834 offset += snprintf(&buf[offset], bufsiz - offset, "filter = %s }",
1835 ds->ds_dkev ? _evfiltstr(ds->ds_dkev->dk_kevent.filter) : "????");
1836 return offset;
1837 }
1838
1839 #if DISPATCH_DEBUG
1840 void
1841 dispatch_debug_kevents(struct kevent* kev, size_t count, const char* str)
1842 {
1843 size_t i;
1844 for (i = 0; i < count; ++i) {
1845 _dispatch_log("kevent[%lu] = { ident = %p, filter = %s, flags = 0x%x, "
1846 "fflags = 0x%x, data = %p, udata = %p }: %s",
1847 i, (void*)kev[i].ident, _evfiltstr(kev[i].filter), kev[i].flags,
1848 kev[i].fflags, (void*)kev[i].data, (void*)kev[i].udata, str);
1849 }
1850 }
1851
1852 static void
1853 _dispatch_kevent_debugger2(void *context)
1854 {
1855 struct sockaddr sa;
1856 socklen_t sa_len = sizeof(sa);
1857 int c, fd = (int)(long)context;
1858 unsigned int i;
1859 dispatch_kevent_t dk;
1860 dispatch_source_t ds;
1861 dispatch_source_refs_t dr;
1862 FILE *debug_stream;
1863
1864 c = accept(fd, &sa, &sa_len);
1865 if (c == -1) {
1866 if (errno != EAGAIN) {
1867 (void)dispatch_assume_zero(errno);
1868 }
1869 return;
1870 }
1871 #if 0
1872 int r = fcntl(c, F_SETFL, 0); // disable non-blocking IO
1873 if (r == -1) {
1874 (void)dispatch_assume_zero(errno);
1875 }
1876 #endif
1877 debug_stream = fdopen(c, "a");
1878 if (!dispatch_assume(debug_stream)) {
1879 close(c);
1880 return;
1881 }
1882
1883 fprintf(debug_stream, "HTTP/1.0 200 OK\r\n");
1884 fprintf(debug_stream, "Content-type: text/html\r\n");
1885 fprintf(debug_stream, "Pragma: nocache\r\n");
1886 fprintf(debug_stream, "\r\n");
1887 fprintf(debug_stream, "<html>\n");
1888 fprintf(debug_stream, "<head><title>PID %u</title></head>\n", getpid());
1889 fprintf(debug_stream, "<body>\n<ul>\n");
1890
1891 //fprintf(debug_stream, "<tr><td>DK</td><td>DK</td><td>DK</td><td>DK</td>"
1892 // "<td>DK</td><td>DK</td><td>DK</td></tr>\n");
1893
1894 for (i = 0; i < DSL_HASH_SIZE; i++) {
1895 if (TAILQ_EMPTY(&_dispatch_sources[i])) {
1896 continue;
1897 }
1898 TAILQ_FOREACH(dk, &_dispatch_sources[i], dk_list) {
1899 fprintf(debug_stream, "\t<br><li>DK %p ident %lu filter %s flags "
1900 "0x%hx fflags 0x%x data 0x%lx udata %p\n",
1901 dk, (unsigned long)dk->dk_kevent.ident,
1902 _evfiltstr(dk->dk_kevent.filter), dk->dk_kevent.flags,
1903 dk->dk_kevent.fflags, (unsigned long)dk->dk_kevent.data,
1904 dk->dk_kevent.udata);
1905 fprintf(debug_stream, "\t\t<ul>\n");
1906 TAILQ_FOREACH(dr, &dk->dk_sources, dr_list) {
1907 ds = _dispatch_source_from_refs(dr);
1908 fprintf(debug_stream, "\t\t\t<li>DS %p refcnt 0x%x suspend "
1909 "0x%x data 0x%lx mask 0x%lx flags 0x%x</li>\n",
1910 ds, ds->do_ref_cnt + 1, ds->do_suspend_cnt,
1911 ds->ds_pending_data, ds->ds_pending_data_mask,
1912 ds->ds_atomic_flags);
1913 if (ds->do_suspend_cnt == DISPATCH_OBJECT_SUSPEND_LOCK) {
1914 dispatch_queue_t dq = ds->do_targetq;
1915 fprintf(debug_stream, "\t\t<br>DQ: %p refcnt 0x%x suspend "
1916 "0x%x label: %s\n", dq, dq->do_ref_cnt + 1,
1917 dq->do_suspend_cnt, dq->dq_label);
1918 }
1919 }
1920 fprintf(debug_stream, "\t\t</ul>\n");
1921 fprintf(debug_stream, "\t</li>\n");
1922 }
1923 }
1924 fprintf(debug_stream, "</ul>\n</body>\n</html>\n");
1925 fflush(debug_stream);
1926 fclose(debug_stream);
1927 }
1928
1929 static void
1930 _dispatch_kevent_debugger2_cancel(void *context)
1931 {
1932 int ret, fd = (int)(long)context;
1933
1934 ret = close(fd);
1935 if (ret != -1) {
1936 (void)dispatch_assume_zero(errno);
1937 }
1938 }
1939
1940 static void
1941 _dispatch_kevent_debugger(void *context DISPATCH_UNUSED)
1942 {
1943 union {
1944 struct sockaddr_in sa_in;
1945 struct sockaddr sa;
1946 } sa_u = {
1947 .sa_in = {
1948 .sin_family = AF_INET,
1949 .sin_addr = { htonl(INADDR_LOOPBACK), },
1950 },
1951 };
1952 dispatch_source_t ds;
1953 const char *valstr;
1954 int val, r, fd, sock_opt = 1;
1955 socklen_t slen = sizeof(sa_u);
1956
1957 if (issetugid()) {
1958 return;
1959 }
1960 valstr = getenv("LIBDISPATCH_DEBUGGER");
1961 if (!valstr) {
1962 return;
1963 }
1964 val = atoi(valstr);
1965 if (val == 2) {
1966 sa_u.sa_in.sin_addr.s_addr = 0;
1967 }
1968 fd = socket(PF_INET, SOCK_STREAM, 0);
1969 if (fd == -1) {
1970 (void)dispatch_assume_zero(errno);
1971 return;
1972 }
1973 r = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (void *)&sock_opt,
1974 (socklen_t) sizeof sock_opt);
1975 if (r == -1) {
1976 (void)dispatch_assume_zero(errno);
1977 goto out_bad;
1978 }
1979 #if 0
1980 r = fcntl(fd, F_SETFL, O_NONBLOCK);
1981 if (r == -1) {
1982 (void)dispatch_assume_zero(errno);
1983 goto out_bad;
1984 }
1985 #endif
1986 r = bind(fd, &sa_u.sa, sizeof(sa_u));
1987 if (r == -1) {
1988 (void)dispatch_assume_zero(errno);
1989 goto out_bad;
1990 }
1991 r = listen(fd, SOMAXCONN);
1992 if (r == -1) {
1993 (void)dispatch_assume_zero(errno);
1994 goto out_bad;
1995 }
1996 r = getsockname(fd, &sa_u.sa, &slen);
1997 if (r == -1) {
1998 (void)dispatch_assume_zero(errno);
1999 goto out_bad;
2000 }
2001
2002 ds = dispatch_source_create(DISPATCH_SOURCE_TYPE_READ, fd, 0,
2003 &_dispatch_mgr_q);
2004 if (dispatch_assume(ds)) {
2005 _dispatch_log("LIBDISPATCH: debug port: %hu",
2006 (in_port_t)ntohs(sa_u.sa_in.sin_port));
2007
2008 /* ownership of fd transfers to ds */
2009 dispatch_set_context(ds, (void *)(long)fd);
2010 dispatch_source_set_event_handler_f(ds, _dispatch_kevent_debugger2);
2011 dispatch_source_set_cancel_handler_f(ds,
2012 _dispatch_kevent_debugger2_cancel);
2013 dispatch_resume(ds);
2014
2015 return;
2016 }
2017 out_bad:
2018 close(fd);
2019 }
2020
2021 #if HAVE_MACH
2022
2023 #ifndef MACH_PORT_TYPE_SPREQUEST
2024 #define MACH_PORT_TYPE_SPREQUEST 0x40000000
2025 #endif
2026
2027 void
2028 dispatch_debug_machport(mach_port_t name, const char* str)
2029 {
2030 mach_port_type_t type;
2031 mach_msg_bits_t ns = 0, nr = 0, nso = 0, nd = 0;
2032 unsigned int dnreqs = 0, dnrsiz;
2033 kern_return_t kr = mach_port_type(mach_task_self(), name, &type);
2034 if (kr) {
2035 _dispatch_log("machport[0x%08x] = { error(0x%x) \"%s\" }: %s", name,
2036 kr, mach_error_string(kr), str);
2037 return;
2038 }
2039 if (type & MACH_PORT_TYPE_SEND) {
2040 (void)dispatch_assume_zero(mach_port_get_refs(mach_task_self(), name,
2041 MACH_PORT_RIGHT_SEND, &ns));
2042 }
2043 if (type & MACH_PORT_TYPE_SEND_ONCE) {
2044 (void)dispatch_assume_zero(mach_port_get_refs(mach_task_self(), name,
2045 MACH_PORT_RIGHT_SEND_ONCE, &nso));
2046 }
2047 if (type & MACH_PORT_TYPE_DEAD_NAME) {
2048 (void)dispatch_assume_zero(mach_port_get_refs(mach_task_self(), name,
2049 MACH_PORT_RIGHT_DEAD_NAME, &nd));
2050 }
2051 if (type & (MACH_PORT_TYPE_RECEIVE|MACH_PORT_TYPE_SEND|
2052 MACH_PORT_TYPE_SEND_ONCE)) {
2053 (void)dispatch_assume_zero(mach_port_dnrequest_info(mach_task_self(),
2054 name, &dnrsiz, &dnreqs));
2055 }
2056 if (type & MACH_PORT_TYPE_RECEIVE) {
2057 mach_port_status_t status = { .mps_pset = 0, };
2058 mach_msg_type_number_t cnt = MACH_PORT_RECEIVE_STATUS_COUNT;
2059 (void)dispatch_assume_zero(mach_port_get_refs(mach_task_self(), name,
2060 MACH_PORT_RIGHT_RECEIVE, &nr));
2061 (void)dispatch_assume_zero(mach_port_get_attributes(mach_task_self(),
2062 name, MACH_PORT_RECEIVE_STATUS, (void*)&status, &cnt));
2063 _dispatch_log("machport[0x%08x] = { R(%03u) S(%03u) SO(%03u) D(%03u) "
2064 "dnreqs(%03u) spreq(%s) nsreq(%s) pdreq(%s) srights(%s) "
2065 "sorights(%03u) qlim(%03u) msgcount(%03u) mkscount(%03u) "
2066 "seqno(%03u) }: %s", name, nr, ns, nso, nd, dnreqs,
2067 type & MACH_PORT_TYPE_SPREQUEST ? "Y":"N",
2068 status.mps_nsrequest ? "Y":"N", status.mps_pdrequest ? "Y":"N",
2069 status.mps_srights ? "Y":"N", status.mps_sorights,
2070 status.mps_qlimit, status.mps_msgcount, status.mps_mscount,
2071 status.mps_seqno, str);
2072 } else if (type & (MACH_PORT_TYPE_SEND|MACH_PORT_TYPE_SEND_ONCE|
2073 MACH_PORT_TYPE_DEAD_NAME)) {
2074 _dispatch_log("machport[0x%08x] = { R(%03u) S(%03u) SO(%03u) D(%03u) "
2075 "dnreqs(%03u) spreq(%s) }: %s", name, nr, ns, nso, nd, dnreqs,
2076 type & MACH_PORT_TYPE_SPREQUEST ? "Y":"N", str);
2077 } else {
2078 _dispatch_log("machport[0x%08x] = { type(0x%08x) }: %s", name, type,
2079 str);
2080 }
2081 }
2082
2083 #endif // HAVE_MACH
2084
2085 #endif // DISPATCH_DEBUG