/*
- * Copyright (c) 2008-2013 Apple Inc. All rights reserved.
+ * Copyright (c) 2008-2016 Apple Inc. All rights reserved.
*
* @APPLE_APACHE_LICENSE_HEADER_START@
*
*/
#include "internal.h"
-#if HAVE_MACH
-#include "protocol.h"
-#include "protocolServer.h"
-#endif
-#include <sys/mount.h>
-
-static void _dispatch_source_merge_kevent(dispatch_source_t ds,
- const struct kevent64_s *ke);
-static bool _dispatch_kevent_register(dispatch_kevent_t *dkp, uint32_t *flgp);
-static void _dispatch_kevent_unregister(dispatch_kevent_t dk, uint32_t flg);
-static bool _dispatch_kevent_resume(dispatch_kevent_t dk, uint32_t new_flags,
- uint32_t del_flags);
-static void _dispatch_kevent_drain(struct kevent64_s *ke);
-static void _dispatch_kevent_merge(struct kevent64_s *ke);
-static void _dispatch_timers_kevent(struct kevent64_s *ke);
-static void _dispatch_timers_unregister(dispatch_source_t ds,
- dispatch_kevent_t dk);
-static void _dispatch_timers_update(dispatch_source_t ds);
-static void _dispatch_timer_aggregates_check(void);
-static void _dispatch_timer_aggregates_register(dispatch_source_t ds);
-static void _dispatch_timer_aggregates_update(dispatch_source_t ds,
- unsigned int tidx);
-static void _dispatch_timer_aggregates_unregister(dispatch_source_t ds,
- unsigned int tidx);
+
+static void _dispatch_source_handler_free(dispatch_source_t ds, long kind);
+static void _dispatch_source_set_interval(dispatch_source_t ds, uint64_t interval);
+
+#define DISPATCH_TIMERS_UNREGISTER 0x1
+#define DISPATCH_TIMERS_RETAIN_2 0x2
+static void _dispatch_timers_update(dispatch_unote_t du, uint32_t flags);
+static void _dispatch_timers_unregister(dispatch_timer_source_refs_t dt);
+
+static void _dispatch_source_timer_configure(dispatch_source_t ds);
static inline unsigned long _dispatch_source_timer_data(
- dispatch_source_refs_t dr, unsigned long prev);
-static long _dispatch_kq_update(const struct kevent64_s *);
-static void _dispatch_memorystatus_init(void);
-#if HAVE_MACH
-static void _dispatch_mach_host_calendar_change_register(void);
-static void _dispatch_mach_recv_msg_buf_init(void);
-static kern_return_t _dispatch_kevent_machport_resume(dispatch_kevent_t dk,
- uint32_t new_flags, uint32_t del_flags);
-static kern_return_t _dispatch_kevent_mach_notify_resume(dispatch_kevent_t dk,
- uint32_t new_flags, uint32_t del_flags);
-static inline void _dispatch_kevent_mach_portset(struct kevent64_s *ke);
-#else
-static inline void _dispatch_mach_host_calendar_change_register(void) {}
-static inline void _dispatch_mach_recv_msg_buf_init(void) {}
-#endif
-static const char * _evfiltstr(short filt);
-#if DISPATCH_DEBUG
-static void _dispatch_kevent_debug(struct kevent64_s* kev, const char* str);
-static void _dispatch_kevent_debugger(void *context);
-#define DISPATCH_ASSERT_ON_MANAGER_QUEUE() \
- dispatch_assert(_dispatch_queue_get_current() == &_dispatch_mgr_q)
-#else
-static inline void
-_dispatch_kevent_debug(struct kevent64_s* kev DISPATCH_UNUSED,
- const char* str DISPATCH_UNUSED) {}
-#define DISPATCH_ASSERT_ON_MANAGER_QUEUE()
-#endif
+ dispatch_source_t ds, dispatch_unote_t du);
#pragma mark -
#pragma mark dispatch_source_t
dispatch_source_t
-dispatch_source_create(dispatch_source_type_t type,
- uintptr_t handle,
- unsigned long mask,
- dispatch_queue_t q)
+dispatch_source_create(dispatch_source_type_t dst, uintptr_t handle,
+ unsigned long mask, dispatch_queue_t dq)
{
- const struct kevent64_s *proto_kev = &type->ke;
+ dispatch_source_refs_t dr;
dispatch_source_t ds;
- dispatch_kevent_t dk;
-
- // input validation
- if (type == NULL || (mask & ~type->mask)) {
- return NULL;
- }
- switch (type->ke.filter) {
- case EVFILT_SIGNAL:
- if (handle >= NSIG) {
- return NULL;
- }
- break;
- case EVFILT_FS:
-#if DISPATCH_USE_VM_PRESSURE
- case EVFILT_VM:
-#endif
-#if DISPATCH_USE_MEMORYSTATUS
- case EVFILT_MEMORYSTATUS:
-#endif
- case DISPATCH_EVFILT_CUSTOM_ADD:
- case DISPATCH_EVFILT_CUSTOM_OR:
- if (handle) {
- return NULL;
- }
- break;
- case DISPATCH_EVFILT_TIMER:
- if (!!handle ^ !!type->ke.ident) {
- return NULL;
- }
- break;
- default:
- break;
+ dr = dux_create(dst, handle, mask)._dr;
+ if (unlikely(!dr)) {
+ return DISPATCH_BAD_INPUT;
}
- ds = _dispatch_alloc(DISPATCH_VTABLE(source),
+ ds = _dispatch_object_alloc(DISPATCH_VTABLE(source),
sizeof(struct dispatch_source_s));
// Initialize as a queue first, then override some settings below.
- _dispatch_queue_init((dispatch_queue_t)ds);
+ _dispatch_queue_init(ds->_as_dq, DQF_LEGACY, 1,
+ DISPATCH_QUEUE_INACTIVE | DISPATCH_QUEUE_ROLE_INNER);
ds->dq_label = "source";
-
ds->do_ref_cnt++; // the reference the manager queue holds
- ds->do_ref_cnt++; // since source is created suspended
- ds->do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_INTERVAL;
- // The initial target queue is the manager queue, in order to get
- // the source installed. <rdar://problem/8928171>
- ds->do_targetq = &_dispatch_mgr_q;
-
- dk = _dispatch_calloc(1ul, sizeof(struct dispatch_kevent_s));
- dk->dk_kevent = *proto_kev;
- dk->dk_kevent.ident = handle;
- dk->dk_kevent.flags |= EV_ADD|EV_ENABLE;
- dk->dk_kevent.fflags |= (uint32_t)mask;
- dk->dk_kevent.udata = (uintptr_t)dk;
- TAILQ_INIT(&dk->dk_sources);
-
- ds->ds_dkev = dk;
- ds->ds_pending_data_mask = dk->dk_kevent.fflags;
- ds->ds_ident_hack = (uintptr_t)dk->dk_kevent.ident;
- if ((EV_DISPATCH|EV_ONESHOT) & proto_kev->flags) {
- ds->ds_is_level = true;
- ds->ds_needs_rearm = true;
- } else if (!(EV_CLEAR & proto_kev->flags)) {
- // we cheat and use EV_CLEAR to mean a "flag thingy"
- ds->ds_is_adder = true;
- }
- // Some sources require special processing
- if (type->init != NULL) {
- type->init(ds, type, handle, mask, q);
- }
- dispatch_assert(!(ds->ds_is_level && ds->ds_is_adder));
-
- if (fastpath(!ds->ds_refs)) {
- ds->ds_refs = _dispatch_calloc(1ul,
- sizeof(struct dispatch_source_refs_s));
- }
- ds->ds_refs->dr_source_wref = _dispatch_ptr2wref(ds);
-
- // First item on the queue sets the user-specified target queue
- dispatch_set_target_queue(ds, q);
+ ds->ds_refs = dr;
+ dr->du_owner_wref = _dispatch_ptr2wref(ds);
+
+ if (slowpath(!dq)) {
+ dq = _dispatch_get_root_queue(DISPATCH_QOS_DEFAULT, true);
+ } else {
+ _dispatch_retain((dispatch_queue_t _Nonnull)dq);
+ }
+ ds->do_targetq = dq;
+ if (dr->du_is_timer && (dr->du_fflags & DISPATCH_TIMER_INTERVAL)) {
+ _dispatch_source_set_interval(ds, handle);
+ }
_dispatch_object_debug(ds, "%s", __func__);
return ds;
}
void
-_dispatch_source_dispose(dispatch_source_t ds)
+_dispatch_source_dispose(dispatch_source_t ds, bool *allow_free)
{
_dispatch_object_debug(ds, "%s", __func__);
- free(ds->ds_refs);
- _dispatch_queue_destroy(ds);
+ _dispatch_source_handler_free(ds, DS_REGISTN_HANDLER);
+ _dispatch_source_handler_free(ds, DS_EVENT_HANDLER);
+ _dispatch_source_handler_free(ds, DS_CANCEL_HANDLER);
+ _dispatch_unote_dispose(ds->ds_refs);
+ ds->ds_refs = NULL;
+ _dispatch_queue_destroy(ds->_as_dq, allow_free);
}
void
_dispatch_source_xref_dispose(dispatch_source_t ds)
{
- _dispatch_wakeup(ds);
-}
-
-void
-dispatch_source_cancel(dispatch_source_t ds)
-{
- _dispatch_object_debug(ds, "%s", __func__);
- // Right after we set the cancel flag, someone else
- // could potentially invoke the source, do the cancelation,
- // unregister the source, and deallocate it. We would
- // need to therefore retain/release before setting the bit
-
- _dispatch_retain(ds);
- (void)dispatch_atomic_or2o(ds, ds_atomic_flags, DSF_CANCELED, relaxed);
- _dispatch_wakeup(ds);
- _dispatch_release(ds);
+ dispatch_queue_flags_t dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
+ if (unlikely(!(dqf & (DQF_LEGACY|DSF_CANCELED)))) {
+ DISPATCH_CLIENT_CRASH(ds, "Release of a source that has not been "
+ "cancelled, but has a mandatory cancel handler");
+ }
+ dx_wakeup(ds, 0, DISPATCH_WAKEUP_MAKE_DIRTY);
}
long
dispatch_source_testcancel(dispatch_source_t ds)
{
- return (bool)(ds->ds_atomic_flags & DSF_CANCELED);
+ return (bool)(ds->dq_atomic_flags & DSF_CANCELED);
}
unsigned long
dispatch_source_get_mask(dispatch_source_t ds)
{
- unsigned long mask = ds->ds_pending_data_mask;
- if (ds->ds_vmpressure_override) {
- mask = NOTE_VM_PRESSURE;
+ dispatch_source_refs_t dr = ds->ds_refs;
+ if (ds->dq_atomic_flags & DSF_CANCELED) {
+ return 0;
+ }
+#if DISPATCH_USE_MEMORYSTATUS
+ if (dr->du_vmpressure_override) {
+ return NOTE_VM_PRESSURE;
}
#if TARGET_IPHONE_SIMULATOR
- else if (ds->ds_memorystatus_override) {
- mask = NOTE_MEMORYSTATUS_PRESSURE_WARN;
+ if (dr->du_memorypressure_override) {
+ return NOTE_MEMORYSTATUS_PRESSURE_WARN;
}
#endif
- return mask;
+#endif // DISPATCH_USE_MEMORYSTATUS
+ return dr->du_fflags;
}
uintptr_t
dispatch_source_get_handle(dispatch_source_t ds)
{
- unsigned int handle = (unsigned int)ds->ds_ident_hack;
+ dispatch_source_refs_t dr = ds->ds_refs;
#if TARGET_IPHONE_SIMULATOR
- if (ds->ds_memorystatus_override) {
- handle = 0;
+ if (dr->du_memorypressure_override) {
+ return 0;
}
#endif
- return handle;
+ return dr->du_ident;
}
unsigned long
dispatch_source_get_data(dispatch_source_t ds)
{
- unsigned long data = ds->ds_data;
- if (ds->ds_vmpressure_override) {
- data = NOTE_VM_PRESSURE;
+#if DISPATCH_USE_MEMORYSTATUS
+ dispatch_source_refs_t dr = ds->ds_refs;
+ if (dr->du_vmpressure_override) {
+ return NOTE_VM_PRESSURE;
}
#if TARGET_IPHONE_SIMULATOR
- else if (ds->ds_memorystatus_override) {
- data = NOTE_MEMORYSTATUS_PRESSURE_WARN;
+ if (dr->du_memorypressure_override) {
+ return NOTE_MEMORYSTATUS_PRESSURE_WARN;
}
#endif
- return data;
+#endif // DISPATCH_USE_MEMORYSTATUS
+ uint64_t value = os_atomic_load2o(ds, ds_data, relaxed);
+ return (unsigned long)(
+ ds->ds_refs->du_data_action == DISPATCH_UNOTE_ACTION_DATA_OR_STATUS_SET
+ ? DISPATCH_SOURCE_GET_DATA(value) : value);
+}
+
+size_t
+dispatch_source_get_extended_data(dispatch_source_t ds,
+ dispatch_source_extended_data_t edata, size_t size)
+{
+ size_t target_size = MIN(size,
+ sizeof(struct dispatch_source_extended_data_s));
+ if (size > 0) {
+ unsigned long data, status = 0;
+ if (ds->ds_refs->du_data_action
+ == DISPATCH_UNOTE_ACTION_DATA_OR_STATUS_SET) {
+ uint64_t combined = os_atomic_load(&ds->ds_data, relaxed);
+ data = DISPATCH_SOURCE_GET_DATA(combined);
+ status = DISPATCH_SOURCE_GET_STATUS(combined);
+ } else {
+ data = dispatch_source_get_data(ds);
+ }
+ if (size >= offsetof(struct dispatch_source_extended_data_s, data)
+ + sizeof(edata->data)) {
+ edata->data = data;
+ }
+ if (size >= offsetof(struct dispatch_source_extended_data_s, status)
+ + sizeof(edata->status)) {
+ edata->status = status;
+ }
+ if (size > sizeof(struct dispatch_source_extended_data_s)) {
+ memset(
+ (char *)edata + sizeof(struct dispatch_source_extended_data_s),
+ 0, size - sizeof(struct dispatch_source_extended_data_s));
+ }
+ }
+ return target_size;
}
+DISPATCH_NOINLINE
void
-dispatch_source_merge_data(dispatch_source_t ds, unsigned long val)
+_dispatch_source_merge_data(dispatch_source_t ds, pthread_priority_t pp,
+ unsigned long val)
{
- struct kevent64_s kev = {
- .fflags = (typeof(kev.fflags))val,
- .data = (typeof(kev.data))val,
- };
+ dispatch_queue_flags_t dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
+ int filter = ds->ds_refs->du_filter;
+
+ if (unlikely(dqf & (DSF_CANCELED | DSF_DELETED))) {
+ return;
+ }
+
+ switch (filter) {
+ case DISPATCH_EVFILT_CUSTOM_ADD:
+ os_atomic_add2o(ds, ds_pending_data, val, relaxed);
+ break;
+ case DISPATCH_EVFILT_CUSTOM_OR:
+ os_atomic_or2o(ds, ds_pending_data, val, relaxed);
+ break;
+ case DISPATCH_EVFILT_CUSTOM_REPLACE:
+ os_atomic_store2o(ds, ds_pending_data, val, relaxed);
+ break;
+ default:
+ DISPATCH_CLIENT_CRASH(filter, "Invalid source type");
+ }
- dispatch_assert(
- ds->ds_dkev->dk_kevent.filter == DISPATCH_EVFILT_CUSTOM_ADD ||
- ds->ds_dkev->dk_kevent.filter == DISPATCH_EVFILT_CUSTOM_OR);
+ dx_wakeup(ds, _dispatch_qos_from_pp(pp), DISPATCH_WAKEUP_MAKE_DIRTY);
+}
- _dispatch_source_merge_kevent(ds, &kev);
+void
+dispatch_source_merge_data(dispatch_source_t ds, unsigned long val)
+{
+ _dispatch_source_merge_data(ds, 0, val);
}
#pragma mark -
DISPATCH_ALWAYS_INLINE
static inline dispatch_continuation_t
-_dispatch_source_handler_alloc(dispatch_source_t ds, void *handler, long kind,
+_dispatch_source_get_handler(dispatch_source_refs_t dr, long kind)
+{
+ return os_atomic_load(&dr->ds_handler[kind], relaxed);
+}
+#define _dispatch_source_get_event_handler(dr) \
+ _dispatch_source_get_handler(dr, DS_EVENT_HANDLER)
+#define _dispatch_source_get_cancel_handler(dr) \
+ _dispatch_source_get_handler(dr, DS_CANCEL_HANDLER)
+#define _dispatch_source_get_registration_handler(dr) \
+ _dispatch_source_get_handler(dr, DS_REGISTN_HANDLER)
+
+DISPATCH_ALWAYS_INLINE
+static inline dispatch_continuation_t
+_dispatch_source_handler_alloc(dispatch_source_t ds, void *func, long kind,
bool block)
{
+ // sources don't propagate priority by default
+ const dispatch_block_flags_t flags =
+ DISPATCH_BLOCK_HAS_PRIORITY | DISPATCH_BLOCK_NO_VOUCHER;
dispatch_continuation_t dc = _dispatch_continuation_alloc();
- if (handler) {
- dc->do_vtable = (void *)((block ? DISPATCH_OBJ_BLOCK_RELEASE_BIT :
- DISPATCH_OBJ_CTXT_FETCH_BIT) | (kind != DS_EVENT_HANDLER ?
- DISPATCH_OBJ_ASYNC_BIT : 0l));
- dc->dc_priority = 0;
- dc->dc_voucher = NULL;
+ if (func) {
+ uintptr_t dc_flags = 0;
+
+ if (kind != DS_EVENT_HANDLER) {
+ dc_flags |= DISPATCH_OBJ_CONSUME_BIT;
+ }
if (block) {
#ifdef __BLOCKS__
- if (slowpath(_dispatch_block_has_private_data(handler))) {
- // sources don't propagate priority by default
- dispatch_block_flags_t flags = DISPATCH_BLOCK_NO_QOS_CLASS;
- flags |= _dispatch_block_get_flags(handler);
- _dispatch_continuation_priority_set(dc,
- _dispatch_block_get_priority(handler), flags);
- }
- if (kind != DS_EVENT_HANDLER) {
- dc->dc_func = _dispatch_call_block_and_release;
- } else {
- dc->dc_func = _dispatch_Block_invoke(handler);
- }
- dc->dc_ctxt = _dispatch_Block_copy(handler);
+ _dispatch_continuation_init(dc, ds, func, 0, flags, dc_flags);
#endif /* __BLOCKS__ */
} else {
- dc->dc_func = handler;
- dc->dc_ctxt = ds->do_ctxt;
+ dc_flags |= DISPATCH_OBJ_CTXT_FETCH_BIT;
+ _dispatch_continuation_init_f(dc, ds, ds->do_ctxt, func,
+ 0, flags, dc_flags);
}
- _dispatch_trace_continuation_push((dispatch_queue_t)ds, dc);
+ _dispatch_trace_continuation_push(ds->_as_dq, dc);
} else {
+ dc->dc_flags = 0;
dc->dc_func = NULL;
}
- dc->dc_data = (void*)kind;
return dc;
}
-static inline void
-_dispatch_source_handler_replace(dispatch_source_refs_t dr, long kind,
- dispatch_continuation_t dc_new)
+DISPATCH_NOINLINE
+static void
+_dispatch_source_handler_dispose(dispatch_continuation_t dc)
{
- dispatch_continuation_t dc = dr->ds_handler[kind];
- if (dc) {
#ifdef __BLOCKS__
- if ((long)dc->do_vtable & DISPATCH_OBJ_BLOCK_RELEASE_BIT) {
- Block_release(dc->dc_ctxt);
- }
+ if (dc->dc_flags & DISPATCH_OBJ_BLOCK_BIT) {
+ Block_release(dc->dc_ctxt);
+ }
#endif /* __BLOCKS__ */
- if (dc->dc_voucher) {
- _voucher_release(dc->dc_voucher);
- dc->dc_voucher = NULL;
- }
- _dispatch_continuation_free(dc);
+ if (dc->dc_voucher) {
+ _voucher_release(dc->dc_voucher);
+ dc->dc_voucher = VOUCHER_INVALID;
}
- dr->ds_handler[kind] = dc_new;
+ _dispatch_continuation_free(dc);
+}
+
+DISPATCH_ALWAYS_INLINE
+static inline dispatch_continuation_t
+_dispatch_source_handler_take(dispatch_source_t ds, long kind)
+{
+ return os_atomic_xchg(&ds->ds_refs->ds_handler[kind], NULL, relaxed);
+}
+
+DISPATCH_ALWAYS_INLINE
+static inline void
+_dispatch_source_handler_free(dispatch_source_t ds, long kind)
+{
+ dispatch_continuation_t dc = _dispatch_source_handler_take(ds, kind);
+ if (dc) _dispatch_source_handler_dispose(dc);
}
+DISPATCH_ALWAYS_INLINE
static inline void
-_dispatch_source_handler_free(dispatch_source_refs_t dr, long kind)
+_dispatch_source_handler_replace(dispatch_source_t ds, long kind,
+ dispatch_continuation_t dc)
{
- _dispatch_source_handler_replace(dr, kind, NULL);
+ if (!dc->dc_func) {
+ _dispatch_continuation_free(dc);
+ dc = NULL;
+ } else if (dc->dc_flags & DISPATCH_OBJ_CTXT_FETCH_BIT) {
+ dc->dc_ctxt = ds->do_ctxt;
+ }
+ dc = os_atomic_xchg(&ds->ds_refs->ds_handler[kind], dc, release);
+ if (dc) _dispatch_source_handler_dispose(dc);
}
+DISPATCH_NOINLINE
static void
-_dispatch_source_set_handler(void *context)
+_dispatch_source_set_handler_slow(void *context)
{
dispatch_source_t ds = (dispatch_source_t)_dispatch_queue_get_current();
dispatch_assert(dx_type(ds) == DISPATCH_SOURCE_KEVENT_TYPE);
+
dispatch_continuation_t dc = context;
long kind = (long)dc->dc_data;
- dc->dc_data = 0;
- if (!dc->dc_func) {
- _dispatch_continuation_free(dc);
- dc = NULL;
- } else if ((long)dc->do_vtable & DISPATCH_OBJ_CTXT_FETCH_BIT) {
- dc->dc_ctxt = ds->do_ctxt;
+ dc->dc_data = NULL;
+ _dispatch_source_handler_replace(ds, kind, dc);
+}
+
+DISPATCH_NOINLINE
+static void
+_dispatch_source_set_handler(dispatch_source_t ds, long kind,
+ dispatch_continuation_t dc)
+{
+ dispatch_assert(dx_type(ds) == DISPATCH_SOURCE_KEVENT_TYPE);
+ if (_dispatch_queue_try_inactive_suspend(ds->_as_dq)) {
+ _dispatch_source_handler_replace(ds, kind, dc);
+ return dx_vtable(ds)->do_resume(ds, false);
}
- _dispatch_source_handler_replace(ds->ds_refs, kind, dc);
- if (kind == DS_EVENT_HANDLER && dc && dc->dc_priority) {
-#if HAVE_PTHREAD_WORKQUEUE_QOS
- ds->dq_priority = dc->dc_priority & ~_PTHREAD_PRIORITY_FLAGS_MASK;
- _dispatch_queue_set_override_priority((dispatch_queue_t)ds);
-#endif
+ if (unlikely(!_dispatch_queue_is_legacy(ds->_as_dq))) {
+ DISPATCH_CLIENT_CRASH(kind, "Cannot change a handler of this source "
+ "after it has been activated");
+ }
+ _dispatch_ktrace1(DISPATCH_PERF_post_activate_mutation, ds);
+ if (kind == DS_REGISTN_HANDLER) {
+ _dispatch_bug_deprecated("Setting registration handler after "
+ "the source has been activated");
}
+ dc->dc_data = (void *)kind;
+ _dispatch_barrier_trysync_or_async_f(ds->_as_dq, dc,
+ _dispatch_source_set_handler_slow, 0);
}
#ifdef __BLOCKS__
{
dispatch_continuation_t dc;
dc = _dispatch_source_handler_alloc(ds, handler, DS_EVENT_HANDLER, true);
- _dispatch_barrier_trysync_f((dispatch_queue_t)ds, dc,
- _dispatch_source_set_handler);
+ _dispatch_source_set_handler(ds, DS_EVENT_HANDLER, dc);
}
#endif /* __BLOCKS__ */
{
dispatch_continuation_t dc;
dc = _dispatch_source_handler_alloc(ds, handler, DS_EVENT_HANDLER, false);
- _dispatch_barrier_trysync_f((dispatch_queue_t)ds, dc,
- _dispatch_source_set_handler);
+ _dispatch_source_set_handler(ds, DS_EVENT_HANDLER, dc);
}
-void
-_dispatch_source_set_event_handler_with_context_f(dispatch_source_t ds,
- void *ctxt, dispatch_function_t handler)
+#ifdef __BLOCKS__
+DISPATCH_NOINLINE
+static void
+_dispatch_source_set_cancel_handler(dispatch_source_t ds,
+ dispatch_block_t handler)
{
dispatch_continuation_t dc;
- dc = _dispatch_source_handler_alloc(ds, handler, DS_EVENT_HANDLER, false);
- dc->do_vtable = (void *)((long)dc->do_vtable &~DISPATCH_OBJ_CTXT_FETCH_BIT);
- dc->dc_other = dc->dc_ctxt;
- dc->dc_ctxt = ctxt;
- _dispatch_barrier_trysync_f((dispatch_queue_t)ds, dc,
- _dispatch_source_set_handler);
+ dc = _dispatch_source_handler_alloc(ds, handler, DS_CANCEL_HANDLER, true);
+ _dispatch_source_set_handler(ds, DS_CANCEL_HANDLER, dc);
}
-#ifdef __BLOCKS__
void
dispatch_source_set_cancel_handler(dispatch_source_t ds,
dispatch_block_t handler)
{
- dispatch_continuation_t dc;
- dc = _dispatch_source_handler_alloc(ds, handler, DS_CANCEL_HANDLER, true);
- _dispatch_barrier_trysync_f((dispatch_queue_t)ds, dc,
- _dispatch_source_set_handler);
+ if (unlikely(!_dispatch_queue_is_legacy(ds->_as_dq))) {
+ DISPATCH_CLIENT_CRASH(0, "Cannot set a non mandatory handler on "
+ "this source");
+ }
+ return _dispatch_source_set_cancel_handler(ds, handler);
}
-#endif /* __BLOCKS__ */
void
-dispatch_source_set_cancel_handler_f(dispatch_source_t ds,
+dispatch_source_set_mandatory_cancel_handler(dispatch_source_t ds,
+ dispatch_block_t handler)
+{
+ _dispatch_queue_atomic_flags_clear(ds->_as_dq, DQF_LEGACY);
+ return _dispatch_source_set_cancel_handler(ds, handler);
+}
+#endif /* __BLOCKS__ */
+
+DISPATCH_NOINLINE
+static void
+_dispatch_source_set_cancel_handler_f(dispatch_source_t ds,
dispatch_function_t handler)
{
dispatch_continuation_t dc;
dc = _dispatch_source_handler_alloc(ds, handler, DS_CANCEL_HANDLER, false);
- _dispatch_barrier_trysync_f((dispatch_queue_t)ds, dc,
- _dispatch_source_set_handler);
+ _dispatch_source_set_handler(ds, DS_CANCEL_HANDLER, dc);
+}
+
+void
+dispatch_source_set_cancel_handler_f(dispatch_source_t ds,
+ dispatch_function_t handler)
+{
+ if (unlikely(!_dispatch_queue_is_legacy(ds->_as_dq))) {
+ DISPATCH_CLIENT_CRASH(0, "Cannot set a non mandatory handler on "
+ "this source");
+ }
+ return _dispatch_source_set_cancel_handler_f(ds, handler);
+}
+
+void
+dispatch_source_set_mandatory_cancel_handler_f(dispatch_source_t ds,
+ dispatch_function_t handler)
+{
+ _dispatch_queue_atomic_flags_clear(ds->_as_dq, DQF_LEGACY);
+ return _dispatch_source_set_cancel_handler_f(ds, handler);
}
#ifdef __BLOCKS__
{
dispatch_continuation_t dc;
dc = _dispatch_source_handler_alloc(ds, handler, DS_REGISTN_HANDLER, true);
- _dispatch_barrier_trysync_f((dispatch_queue_t)ds, dc,
- _dispatch_source_set_handler);
+ _dispatch_source_set_handler(ds, DS_REGISTN_HANDLER, dc);
}
#endif /* __BLOCKS__ */
{
dispatch_continuation_t dc;
dc = _dispatch_source_handler_alloc(ds, handler, DS_REGISTN_HANDLER, false);
- _dispatch_barrier_trysync_f((dispatch_queue_t)ds, dc,
- _dispatch_source_set_handler);
+ _dispatch_source_set_handler(ds, DS_REGISTN_HANDLER, dc);
}
#pragma mark -
#pragma mark dispatch_source_invoke
+bool
+_dispatch_source_will_reenable_kevent_4NW(dispatch_source_t ds)
+{
+ uint64_t dq_state = os_atomic_load2o(ds, dq_state, relaxed);
+ dispatch_queue_flags_t dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
+
+ if (unlikely(!_dq_state_drain_locked_by_self(dq_state))) {
+ DISPATCH_CLIENT_CRASH(0, "_dispatch_source_will_reenable_kevent_4NW "
+ "not called from within the event handler");
+ }
+
+ return _dispatch_unote_needs_rearm(ds->ds_refs) && !(dqf & DSF_ARMED);
+}
+
static void
-_dispatch_source_registration_callout(dispatch_source_t ds)
+_dispatch_source_registration_callout(dispatch_source_t ds, dispatch_queue_t cq,
+ dispatch_invoke_flags_t flags)
{
- dispatch_source_refs_t dr = ds->ds_refs;
- dispatch_continuation_t dc = dr->ds_handler[DS_REGISTN_HANDLER];
- if ((ds->ds_atomic_flags & DSF_CANCELED) || (ds->do_xref_cnt == -1)) {
+ dispatch_continuation_t dc;
+
+ dc = _dispatch_source_handler_take(ds, DS_REGISTN_HANDLER);
+ if (ds->dq_atomic_flags & (DSF_CANCELED | DQF_RELEASED)) {
// no registration callout if source is canceled rdar://problem/8955246
- return _dispatch_source_handler_free(dr, DS_REGISTN_HANDLER);
+ return _dispatch_source_handler_dispose(dc);
}
- pthread_priority_t old_dp = _dispatch_set_defaultpriority(ds->dq_priority);
- if ((long)dc->do_vtable & DISPATCH_OBJ_CTXT_FETCH_BIT) {
+ if (dc->dc_flags & DISPATCH_OBJ_CTXT_FETCH_BIT) {
dc->dc_ctxt = ds->do_ctxt;
}
- _dispatch_continuation_pop(dc);
- dr->ds_handler[DS_REGISTN_HANDLER] = NULL;
- _dispatch_reset_defaultpriority(old_dp);
+ _dispatch_continuation_pop(dc, NULL, flags, cq);
}
static void
-_dispatch_source_cancel_callout(dispatch_source_t ds)
+_dispatch_source_cancel_callout(dispatch_source_t ds, dispatch_queue_t cq,
+ dispatch_invoke_flags_t flags)
{
- dispatch_source_refs_t dr = ds->ds_refs;
- dispatch_continuation_t dc = dr->ds_handler[DS_CANCEL_HANDLER];
- ds->ds_pending_data_mask = 0;
+ dispatch_continuation_t dc;
+
+ dc = _dispatch_source_handler_take(ds, DS_CANCEL_HANDLER);
ds->ds_pending_data = 0;
ds->ds_data = 0;
- _dispatch_source_handler_free(dr, DS_EVENT_HANDLER);
- _dispatch_source_handler_free(dr, DS_REGISTN_HANDLER);
+ _dispatch_source_handler_free(ds, DS_EVENT_HANDLER);
+ _dispatch_source_handler_free(ds, DS_REGISTN_HANDLER);
if (!dc) {
return;
}
- if (!(ds->ds_atomic_flags & DSF_CANCELED)) {
- return _dispatch_source_handler_free(dr, DS_CANCEL_HANDLER);
+ if (!(ds->dq_atomic_flags & DSF_CANCELED)) {
+ return _dispatch_source_handler_dispose(dc);
}
- pthread_priority_t old_dp = _dispatch_set_defaultpriority(ds->dq_priority);
- if ((long)dc->do_vtable & DISPATCH_OBJ_CTXT_FETCH_BIT) {
+ if (dc->dc_flags & DISPATCH_OBJ_CTXT_FETCH_BIT) {
dc->dc_ctxt = ds->do_ctxt;
}
- _dispatch_continuation_pop(dc);
- dr->ds_handler[DS_CANCEL_HANDLER] = NULL;
- _dispatch_reset_defaultpriority(old_dp);
+ _dispatch_continuation_pop(dc, NULL, flags, cq);
}
static void
-_dispatch_source_latch_and_call(dispatch_source_t ds)
+_dispatch_source_latch_and_call(dispatch_source_t ds, dispatch_queue_t cq,
+ dispatch_invoke_flags_t flags)
{
- unsigned long prev;
+ dispatch_source_refs_t dr = ds->ds_refs;
+ dispatch_continuation_t dc = _dispatch_source_get_handler(dr, DS_EVENT_HANDLER);
+ uint64_t prev;
- if ((ds->ds_atomic_flags & DSF_CANCELED) || (ds->do_xref_cnt == -1)) {
- return;
+ if (dr->du_is_timer && !(dr->du_fflags & DISPATCH_TIMER_AFTER)) {
+ prev = _dispatch_source_timer_data(ds, dr);
+ } else {
+ prev = os_atomic_xchg2o(ds, ds_pending_data, 0, relaxed);
}
- dispatch_source_refs_t dr = ds->ds_refs;
- dispatch_continuation_t dc = dr->ds_handler[DS_EVENT_HANDLER];
- prev = dispatch_atomic_xchg2o(ds, ds_pending_data, 0, relaxed);
- if (ds->ds_is_level) {
+ if (dr->du_data_action == DISPATCH_UNOTE_ACTION_DATA_SET) {
ds->ds_data = ~prev;
- } else if (ds->ds_is_timer && ds_timer(dr).target && prev) {
- ds->ds_data = _dispatch_source_timer_data(dr, prev);
} else {
ds->ds_data = prev;
}
- if (!dispatch_assume(prev) || !dc) {
+ if (!dispatch_assume(prev != 0) || !dc) {
return;
}
- pthread_priority_t old_dp = _dispatch_set_defaultpriority(ds->dq_priority);
- _dispatch_trace_continuation_pop(_dispatch_queue_get_current(), dc);
- voucher_t voucher = dc->dc_voucher ? _voucher_retain(dc->dc_voucher) : NULL;
- _dispatch_continuation_voucher_adopt(dc); // consumes voucher reference
- _dispatch_client_callout(dc->dc_ctxt, dc->dc_func);
- _dispatch_introspection_queue_item_complete(dc);
- if (voucher) dc->dc_voucher = voucher;
- _dispatch_reset_defaultpriority(old_dp);
+ _dispatch_continuation_pop(dc, NULL, flags, cq);
+ if (dr->du_is_timer && (dr->du_fflags & DISPATCH_TIMER_AFTER)) {
+ _dispatch_source_handler_free(ds, DS_EVENT_HANDLER);
+ dispatch_release(ds); // dispatch_after sources are one-shot
+ }
}
+DISPATCH_NOINLINE
static void
-_dispatch_source_kevent_unregister(dispatch_source_t ds)
+_dispatch_source_refs_finalize_unregistration(dispatch_source_t ds)
+{
+ dispatch_queue_flags_t dqf;
+ dispatch_source_refs_t dr = ds->ds_refs;
+
+ dqf = _dispatch_queue_atomic_flags_set_and_clear_orig(ds->_as_dq,
+ DSF_DELETED, DSF_ARMED | DSF_DEFERRED_DELETE | DSF_CANCEL_WAITER);
+ if (dqf & DSF_CANCEL_WAITER) {
+ _dispatch_wake_by_address(&ds->dq_atomic_flags);
+ }
+ _dispatch_debug("kevent-source[%p]: disarmed kevent[%p]", ds, dr);
+ _dispatch_release_tailcall(ds); // the retain is done at creation time
+}
+
+void
+_dispatch_source_refs_unregister(dispatch_source_t ds, uint32_t options)
{
_dispatch_object_debug(ds, "%s", __func__);
- dispatch_kevent_t dk = ds->ds_dkev;
- ds->ds_dkev = NULL;
- switch (dk->dk_kevent.filter) {
- case DISPATCH_EVFILT_TIMER:
- _dispatch_timers_unregister(ds, dk);
- break;
- default:
- TAILQ_REMOVE(&dk->dk_sources, ds->ds_refs, dr_list);
- _dispatch_kevent_unregister(dk, (uint32_t)ds->ds_pending_data_mask);
- break;
+ dispatch_queue_flags_t dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
+ dispatch_source_refs_t dr = ds->ds_refs;
+
+ if (dr->du_is_timer) {
+ // Because of the optimization to unregister fired oneshot timers
+ // from the target queue, we can't trust _dispatch_unote_registered()
+ // to tell the truth, it may not have happened yet
+ if (dqf & DSF_ARMED) {
+ _dispatch_timers_unregister(ds->ds_timer_refs);
+ _dispatch_release_2(ds);
+ }
+ dr->du_ident = DISPATCH_TIMER_IDENT_CANCELED;
+ } else {
+ if (_dispatch_unote_needs_rearm(dr) && !(dqf & DSF_ARMED)) {
+ options |= DU_UNREGISTER_IMMEDIATE_DELETE;
+ }
+ if (!_dispatch_unote_unregister(dr, options)) {
+ _dispatch_debug("kevent-source[%p]: deferred delete kevent[%p]",
+ ds, dr);
+ _dispatch_queue_atomic_flags_set(ds->_as_dq, DSF_DEFERRED_DELETE);
+ return; // deferred unregistration
+ }
}
- (void)dispatch_atomic_and2o(ds, ds_atomic_flags, ~DSF_ARMED, relaxed);
- ds->ds_needs_rearm = false; // re-arm is pointless and bad now
- _dispatch_release(ds); // the retain is done at creation time
+ ds->ds_is_installed = true;
+ _dispatch_source_refs_finalize_unregistration(ds);
}
-static void
-_dispatch_source_kevent_resume(dispatch_source_t ds, uint32_t new_flags)
+DISPATCH_ALWAYS_INLINE
+static inline bool
+_dispatch_source_tryarm(dispatch_source_t ds)
{
- switch (ds->ds_dkev->dk_kevent.filter) {
- case DISPATCH_EVFILT_TIMER:
- return _dispatch_timers_update(ds);
- case EVFILT_MACHPORT:
- if (ds->ds_pending_data_mask & DISPATCH_MACH_RECV_MESSAGE) {
- new_flags |= DISPATCH_MACH_RECV_MESSAGE; // emulate EV_DISPATCH
+ dispatch_queue_flags_t oqf, nqf;
+ return os_atomic_rmw_loop2o(ds, dq_atomic_flags, oqf, nqf, relaxed, {
+ if (oqf & (DSF_DEFERRED_DELETE | DSF_DELETED)) {
+ // the test is inside the loop because it's convenient but the
+ // result should not change for the duration of the rmw_loop
+ os_atomic_rmw_loop_give_up(break);
}
- break;
+ nqf = oqf | DSF_ARMED;
+ });
+}
+
+DISPATCH_ALWAYS_INLINE
+static inline bool
+_dispatch_source_refs_resume(dispatch_source_t ds)
+{
+ dispatch_source_refs_t dr = ds->ds_refs;
+ if (dr->du_is_timer) {
+ _dispatch_timers_update(dr, 0);
+ return true;
}
- if (_dispatch_kevent_resume(ds->ds_dkev, new_flags, 0)) {
- _dispatch_source_kevent_unregister(ds);
+ if (unlikely(!_dispatch_source_tryarm(ds))) {
+ return false;
}
+ _dispatch_unote_resume(dr);
+ _dispatch_debug("kevent-source[%p]: rearmed kevent[%p]", ds, dr);
+ return true;
}
-static void
-_dispatch_source_kevent_register(dispatch_source_t ds)
+void
+_dispatch_source_refs_register(dispatch_source_t ds, dispatch_wlh_t wlh,
+ dispatch_priority_t pri)
{
- dispatch_assert_zero(ds->ds_is_installed);
- switch (ds->ds_dkev->dk_kevent.filter) {
- case DISPATCH_EVFILT_TIMER:
- return _dispatch_timers_update(ds);
- }
- uint32_t flags;
- bool do_resume = _dispatch_kevent_register(&ds->ds_dkev, &flags);
- TAILQ_INSERT_TAIL(&ds->ds_dkev->dk_sources, ds->ds_refs, dr_list);
- if (do_resume || ds->ds_needs_rearm) {
- _dispatch_source_kevent_resume(ds, flags);
- }
- (void)dispatch_atomic_or2o(ds, ds_atomic_flags, DSF_ARMED, relaxed);
+ dispatch_source_refs_t dr = ds->ds_refs;
+ dispatch_priority_t kbp;
+
+ dispatch_assert(!ds->ds_is_installed);
+
+ if (dr->du_is_timer) {
+ dispatch_queue_t dq = ds->_as_dq;
+ kbp = _dispatch_queue_compute_priority_and_wlh(dq, NULL);
+ // aggressively coalesce background/maintenance QoS timers
+ // <rdar://problem/12200216&27342536>
+ if (_dispatch_qos_is_background(_dispatch_priority_qos(kbp))) {
+ if (dr->du_fflags & DISPATCH_TIMER_STRICT) {
+ _dispatch_ktrace1(DISPATCH_PERF_strict_bg_timer, ds);
+ } else {
+ dr->du_fflags |= DISPATCH_TIMER_BACKGROUND;
+ dr->du_ident = _dispatch_source_timer_idx(dr);
+ }
+ }
+ _dispatch_timers_update(dr, 0);
+ return;
+ }
+
+ if (unlikely(!_dispatch_source_tryarm(ds) ||
+ !_dispatch_unote_register(dr, wlh, pri))) {
+ // Do the parts of dispatch_source_refs_unregister() that
+ // are required after this partial initialization.
+ _dispatch_source_refs_finalize_unregistration(ds);
+ } else {
+ _dispatch_debug("kevent-source[%p]: armed kevent[%p]", ds, dr);
+ }
_dispatch_object_debug(ds, "%s", __func__);
}
+static void
+_dispatch_source_set_event_handler_context(void *ctxt)
+{
+ dispatch_source_t ds = ctxt;
+ dispatch_continuation_t dc = _dispatch_source_get_event_handler(ds->ds_refs);
+
+ if (dc && (dc->dc_flags & DISPATCH_OBJ_CTXT_FETCH_BIT)) {
+ dc->dc_ctxt = ds->do_ctxt;
+ }
+}
+
+DISPATCH_ALWAYS_INLINE
+static inline void
+_dispatch_source_install(dispatch_source_t ds, dispatch_wlh_t wlh,
+ dispatch_priority_t pri)
+{
+ _dispatch_source_refs_register(ds, wlh, pri);
+ ds->ds_is_installed = true;
+}
+
+void
+_dispatch_source_finalize_activation(dispatch_source_t ds, bool *allow_resume)
+{
+ dispatch_continuation_t dc;
+ dispatch_source_refs_t dr = ds->ds_refs;
+ dispatch_priority_t pri;
+ dispatch_wlh_t wlh;
+
+ if (unlikely(dr->du_is_direct &&
+ (_dispatch_queue_atomic_flags(ds->_as_dq) & DSF_CANCELED))) {
+ return _dispatch_source_refs_unregister(ds, 0);
+ }
+
+ dc = _dispatch_source_get_event_handler(dr);
+ if (dc) {
+ if (_dispatch_object_is_barrier(dc)) {
+ _dispatch_queue_atomic_flags_set(ds->_as_dq, DQF_BARRIER_BIT);
+ }
+ ds->dq_priority = _dispatch_priority_from_pp_strip_flags(dc->dc_priority);
+ if (dc->dc_flags & DISPATCH_OBJ_CTXT_FETCH_BIT) {
+ _dispatch_barrier_async_detached_f(ds->_as_dq, ds,
+ _dispatch_source_set_event_handler_context);
+ }
+ }
+
+ // call "super"
+ _dispatch_queue_finalize_activation(ds->_as_dq, allow_resume);
+
+ if (dr->du_is_direct && !ds->ds_is_installed) {
+ dispatch_queue_t dq = ds->_as_dq;
+ pri = _dispatch_queue_compute_priority_and_wlh(dq, &wlh);
+ if (pri) _dispatch_source_install(ds, wlh, pri);
+ }
+}
+
DISPATCH_ALWAYS_INLINE
-static inline dispatch_queue_t
-_dispatch_source_invoke2(dispatch_object_t dou,
- _dispatch_thread_semaphore_t *sema_ptr DISPATCH_UNUSED)
+static inline dispatch_queue_wakeup_target_t
+_dispatch_source_invoke2(dispatch_object_t dou, dispatch_invoke_context_t dic,
+ dispatch_invoke_flags_t flags, uint64_t *owned)
{
dispatch_source_t ds = dou._ds;
- if (slowpath(_dispatch_queue_drain(ds))) {
- DISPATCH_CLIENT_CRASH("Sync onto source");
+ dispatch_queue_wakeup_target_t retq = DISPATCH_QUEUE_WAKEUP_NONE;
+ dispatch_queue_t dq = _dispatch_queue_get_current();
+ dispatch_source_refs_t dr = ds->ds_refs;
+ dispatch_queue_flags_t dqf;
+
+ if (!(flags & DISPATCH_INVOKE_MANAGER_DRAIN) &&
+ _dispatch_unote_wlh_changed(dr, _dispatch_get_wlh())) {
+ dqf = _dispatch_queue_atomic_flags_set_orig(ds->_as_dq,
+ DSF_WLH_CHANGED);
+ if (!(dqf & DSF_WLH_CHANGED)) {
+ _dispatch_bug_deprecated("Changing target queue "
+ "hierarchy after source was activated");
+ }
+ }
+
+ if (_dispatch_queue_class_probe(ds)) {
+ // Intentionally always drain even when on the manager queue
+ // and not the source's regular target queue: we need to be able
+ // to drain timer setting and the like there.
+ dispatch_with_disabled_narrowing(dic, {
+ retq = _dispatch_queue_serial_drain(ds->_as_dq, dic, flags, owned);
+ });
}
// This function performs all source actions. Each action is responsible
// current queue is not the correct queue for this action, the correct queue
// will be returned and the invoke will be re-driven on that queue.
- // The order of tests here in invoke and in probe should be consistent.
+ // The order of tests here in invoke and in wakeup should be consistent.
- dispatch_queue_t dq = _dispatch_queue_get_current();
- dispatch_source_refs_t dr = ds->ds_refs;
+ dispatch_queue_t dkq = &_dispatch_mgr_q;
+ bool prevent_starvation = false;
- if (!ds->ds_is_installed) {
- // The source needs to be installed on the manager queue.
- if (dq != &_dispatch_mgr_q) {
- return &_dispatch_mgr_q;
- }
- _dispatch_source_kevent_register(ds);
- ds->ds_is_installed = true;
- if (dr->ds_handler[DS_REGISTN_HANDLER]) {
- return ds->do_targetq;
+ if (dr->du_is_direct) {
+ dkq = ds->do_targetq;
+ }
+
+ if (dr->du_is_timer &&
+ os_atomic_load2o(ds, ds_timer_refs->dt_pending_config, relaxed)) {
+ dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
+ if (!(dqf & (DSF_CANCELED | DQF_RELEASED))) {
+ // timer has to be configured on the kevent queue
+ if (dq != dkq) {
+ return dkq;
+ }
+ _dispatch_source_timer_configure(ds);
}
- if (slowpath(ds->do_xref_cnt == -1)) {
- return &_dispatch_mgr_q; // rdar://problem/9558246
+ }
+
+ if (!ds->ds_is_installed) {
+ // The source needs to be installed on the kevent queue.
+ if (dq != dkq) {
+ return dkq;
}
- } else if (slowpath(DISPATCH_OBJECT_SUSPENDED(ds))) {
+ _dispatch_source_install(ds, _dispatch_get_wlh(),
+ _dispatch_get_basepri());
+ }
+
+ if (unlikely(DISPATCH_QUEUE_IS_SUSPENDED(ds))) {
// Source suspended by an item drained from the source queue.
- return NULL;
- } else if (dr->ds_handler[DS_REGISTN_HANDLER]) {
+ return ds->do_targetq;
+ }
+
+ if (_dispatch_source_get_registration_handler(dr)) {
// The source has been registered and the registration handler needs
// to be delivered on the target queue.
if (dq != ds->do_targetq) {
return ds->do_targetq;
}
// clears ds_registration_handler
- _dispatch_source_registration_callout(ds);
- if (slowpath(ds->do_xref_cnt == -1)) {
- return &_dispatch_mgr_q; // rdar://problem/9558246
+ _dispatch_source_registration_callout(ds, dq, flags);
+ }
+
+ dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
+ if ((dqf & DSF_DEFERRED_DELETE) && !(dqf & DSF_ARMED)) {
+unregister_event:
+ // DSF_DELETE: Pending source kevent unregistration has been completed
+ // !DSF_ARMED: event was delivered and can safely be unregistered
+ if (dq != dkq) {
+ return dkq;
+ }
+ _dispatch_source_refs_unregister(ds, DU_UNREGISTER_IMMEDIATE_DELETE);
+ dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
+ }
+
+ if (!(dqf & (DSF_CANCELED | DQF_RELEASED)) &&
+ os_atomic_load2o(ds, ds_pending_data, relaxed)) {
+ // The source has pending data to deliver via the event handler callback
+ // on the target queue. Some sources need to be rearmed on the kevent
+ // queue after event delivery.
+ if (dq == ds->do_targetq) {
+ _dispatch_source_latch_and_call(ds, dq, flags);
+ dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
+
+ // starvation avoidance: if the source triggers itself then force a
+ // re-queue to give other things already queued on the target queue
+ // a chance to run.
+ //
+ // however, if the source is directly targeting an overcommit root
+ // queue, this would requeue the source and ask for a new overcommit
+ // thread right away.
+ prevent_starvation = dq->do_targetq ||
+ !(dq->dq_priority & DISPATCH_PRIORITY_FLAG_OVERCOMMIT);
+ if (prevent_starvation &&
+ os_atomic_load2o(ds, ds_pending_data, relaxed)) {
+ retq = ds->do_targetq;
+ }
+ } else {
+ // there is no point trying to be eager, the next thing to do is
+ // to deliver the event
+ return ds->do_targetq;
}
- } else if ((ds->ds_atomic_flags & DSF_CANCELED) || (ds->do_xref_cnt == -1)){
+ }
+
+ if ((dqf & (DSF_CANCELED | DQF_RELEASED)) && !(dqf & DSF_DEFERRED_DELETE)) {
// The source has been cancelled and needs to be uninstalled from the
- // manager queue. After uninstallation, the cancellation handler needs
+ // kevent queue. After uninstallation, the cancellation handler needs
// to be delivered to the target queue.
- if (ds->ds_dkev) {
- if (dq != &_dispatch_mgr_q) {
- return &_dispatch_mgr_q;
+ if (!(dqf & DSF_DELETED)) {
+ if (dr->du_is_timer && !(dqf & DSF_ARMED)) {
+ // timers can cheat if not armed because there's nothing left
+ // to do on the manager queue and unregistration can happen
+ // on the regular target queue
+ } else if (dq != dkq) {
+ return dkq;
}
- _dispatch_source_kevent_unregister(ds);
- }
- if (dr->ds_handler[DS_EVENT_HANDLER] ||
- dr->ds_handler[DS_CANCEL_HANDLER] ||
- dr->ds_handler[DS_REGISTN_HANDLER]) {
- if (dq != ds->do_targetq) {
- return ds->do_targetq;
+ _dispatch_source_refs_unregister(ds, 0);
+ dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
+ if (unlikely(dqf & DSF_DEFERRED_DELETE)) {
+ if (!(dqf & DSF_ARMED)) {
+ goto unregister_event;
+ }
+ // we need to wait for the EV_DELETE
+ return retq ? retq : DISPATCH_QUEUE_WAKEUP_WAIT_FOR_EVENT;
}
}
- _dispatch_source_cancel_callout(ds);
- } else if (ds->ds_pending_data) {
- // The source has pending data to deliver via the event handler callback
- // on the target queue. Some sources need to be rearmed on the manager
- // queue after event delivery.
- if (dq != ds->do_targetq) {
+ if (dq != ds->do_targetq && (_dispatch_source_get_event_handler(dr) ||
+ _dispatch_source_get_cancel_handler(dr) ||
+ _dispatch_source_get_registration_handler(dr))) {
+ retq = ds->do_targetq;
+ } else {
+ _dispatch_source_cancel_callout(ds, dq, flags);
+ dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
+ }
+ prevent_starvation = false;
+ }
+
+ if (_dispatch_unote_needs_rearm(dr) &&
+ !(dqf & (DSF_ARMED|DSF_DELETED|DSF_CANCELED|DQF_RELEASED))) {
+ // The source needs to be rearmed on the kevent queue.
+ if (dq != dkq) {
+ return dkq;
+ }
+ if (unlikely(dqf & DSF_DEFERRED_DELETE)) {
+ // no need for resume when we can directly unregister the kevent
+ goto unregister_event;
+ }
+ if (unlikely(DISPATCH_QUEUE_IS_SUSPENDED(ds))) {
+ // do not try to rearm the kevent if the source is suspended
+ // from the source handler
+ return ds->do_targetq;
+ }
+ if (prevent_starvation && dr->du_wlh == DISPATCH_WLH_ANON) {
+ // keep the old behavior to force re-enqueue to our target queue
+ // for the rearm.
+ //
+ // if the handler didn't run, or this is a pending delete
+ // or our target queue is a global queue, then starvation is
+ // not a concern and we can rearm right away.
return ds->do_targetq;
}
- _dispatch_source_latch_and_call(ds);
- if (ds->ds_needs_rearm) {
- return &_dispatch_mgr_q;
+ if (unlikely(!_dispatch_source_refs_resume(ds))) {
+ goto unregister_event;
}
- } else if (ds->ds_needs_rearm && !(ds->ds_atomic_flags & DSF_ARMED)) {
- // The source needs to be rearmed on the manager queue.
- if (dq != &_dispatch_mgr_q) {
- return &_dispatch_mgr_q;
+ if (!prevent_starvation && _dispatch_wlh_should_poll_unote(dr)) {
+ // try to redrive the drain from under the lock for sources
+ // targeting an overcommit root queue to avoid parking
+ // when the next event has already fired
+ _dispatch_event_loop_drain(KEVENT_FLAG_IMMEDIATE);
}
- _dispatch_source_kevent_resume(ds, 0);
- (void)dispatch_atomic_or2o(ds, ds_atomic_flags, DSF_ARMED, relaxed);
}
- return NULL;
+ return retq;
}
DISPATCH_NOINLINE
void
-_dispatch_source_invoke(dispatch_source_t ds)
+_dispatch_source_invoke(dispatch_source_t ds, dispatch_invoke_context_t dic,
+ dispatch_invoke_flags_t flags)
{
- _dispatch_queue_class_invoke(ds, _dispatch_source_invoke2);
+ _dispatch_queue_class_invoke(ds, dic, flags,
+ DISPATCH_INVOKE_DISALLOW_SYNC_WAITERS, _dispatch_source_invoke2);
}
-unsigned long
-_dispatch_source_probe(dispatch_source_t ds)
+void
+_dispatch_source_wakeup(dispatch_source_t ds, dispatch_qos_t qos,
+ dispatch_wakeup_flags_t flags)
{
// This function determines whether the source needs to be invoked.
- // The order of tests here in probe and in invoke should be consistent.
+ // The order of tests here in wakeup and in invoke should be consistent.
dispatch_source_refs_t dr = ds->ds_refs;
- if (!ds->ds_is_installed) {
- // The source needs to be installed on the manager queue.
- return true;
- } else if (dr->ds_handler[DS_REGISTN_HANDLER]) {
+ dispatch_queue_wakeup_target_t dkq = DISPATCH_QUEUE_WAKEUP_MGR;
+ dispatch_queue_wakeup_target_t tq = DISPATCH_QUEUE_WAKEUP_NONE;
+ dispatch_queue_flags_t dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
+ bool deferred_delete = (dqf & DSF_DEFERRED_DELETE);
+
+ if (dr->du_is_direct) {
+ dkq = DISPATCH_QUEUE_WAKEUP_TARGET;
+ }
+
+ if (!(dqf & (DSF_CANCELED | DQF_RELEASED)) && dr->du_is_timer &&
+ os_atomic_load2o(ds, ds_timer_refs->dt_pending_config, relaxed)) {
+ // timer has to be configured on the kevent queue
+ tq = dkq;
+ } else if (!ds->ds_is_installed) {
+ // The source needs to be installed on the kevent queue.
+ tq = dkq;
+ } else if (_dispatch_source_get_registration_handler(dr)) {
// The registration handler needs to be delivered to the target queue.
- return true;
- } else if ((ds->ds_atomic_flags & DSF_CANCELED) || (ds->do_xref_cnt == -1)){
- // The source needs to be uninstalled from the manager queue, or the
+ tq = DISPATCH_QUEUE_WAKEUP_TARGET;
+ } else if (deferred_delete && !(dqf & DSF_ARMED)) {
+ // Pending source kevent unregistration has been completed
+ // or EV_ONESHOT event can be acknowledged
+ tq = dkq;
+ } else if (!(dqf & (DSF_CANCELED | DQF_RELEASED)) &&
+ os_atomic_load2o(ds, ds_pending_data, relaxed)) {
+ // The source has pending data to deliver to the target queue.
+ tq = DISPATCH_QUEUE_WAKEUP_TARGET;
+ } else if ((dqf & (DSF_CANCELED | DQF_RELEASED)) && !deferred_delete) {
+ // The source needs to be uninstalled from the kevent queue, or the
// cancellation handler needs to be delivered to the target queue.
// Note: cancellation assumes installation.
- if (ds->ds_dkev || dr->ds_handler[DS_EVENT_HANDLER] ||
- dr->ds_handler[DS_CANCEL_HANDLER] ||
- dr->ds_handler[DS_REGISTN_HANDLER]) {
- return true;
+ if (!(dqf & DSF_DELETED)) {
+ if (dr->du_is_timer && !(dqf & DSF_ARMED)) {
+ // timers can cheat if not armed because there's nothing left
+ // to do on the manager queue and unregistration can happen
+ // on the regular target queue
+ tq = DISPATCH_QUEUE_WAKEUP_TARGET;
+ } else {
+ tq = dkq;
+ }
+ } else if (_dispatch_source_get_event_handler(dr) ||
+ _dispatch_source_get_cancel_handler(dr) ||
+ _dispatch_source_get_registration_handler(dr)) {
+ tq = DISPATCH_QUEUE_WAKEUP_TARGET;
}
- } else if (ds->ds_pending_data) {
- // The source has pending data to deliver to the target queue.
- return true;
- } else if (ds->ds_needs_rearm && !(ds->ds_atomic_flags & DSF_ARMED)) {
- // The source needs to be rearmed on the manager queue.
- return true;
+ } else if (_dispatch_unote_needs_rearm(dr) &&
+ !(dqf & (DSF_ARMED|DSF_DELETED|DSF_CANCELED|DQF_RELEASED))) {
+ // The source needs to be rearmed on the kevent queue.
+ tq = dkq;
}
- return _dispatch_queue_class_probe(ds);
-}
-
-static void
-_dispatch_source_merge_kevent(dispatch_source_t ds, const struct kevent64_s *ke)
-{
- if ((ds->ds_atomic_flags & DSF_CANCELED) || (ds->do_xref_cnt == -1)) {
- return;
+ if (!tq && _dispatch_queue_class_probe(ds)) {
+ tq = DISPATCH_QUEUE_WAKEUP_TARGET;
}
- if (ds->ds_is_level) {
- // ke->data is signed and "negative available data" makes no sense
- // zero bytes happens when EV_EOF is set
- // 10A268 does not fail this assert with EVFILT_READ and a 10 GB file
- dispatch_assert(ke->data >= 0l);
- dispatch_atomic_store2o(ds, ds_pending_data, ~(unsigned long)ke->data,
- relaxed);
- } else if (ds->ds_is_adder) {
- (void)dispatch_atomic_add2o(ds, ds_pending_data,
- (unsigned long)ke->data, relaxed);
- } else if (ke->fflags & ds->ds_pending_data_mask) {
- (void)dispatch_atomic_or2o(ds, ds_pending_data,
- ke->fflags & ds->ds_pending_data_mask, relaxed);
- }
- // EV_DISPATCH and EV_ONESHOT sources are no longer armed after delivery
- if (ds->ds_needs_rearm) {
- (void)dispatch_atomic_and2o(ds, ds_atomic_flags, ~DSF_ARMED, relaxed);
- }
-
- _dispatch_wakeup(ds);
-}
-
-#pragma mark -
-#pragma mark dispatch_kevent_t
-
-#if DISPATCH_USE_GUARDED_FD_CHANGE_FDGUARD
-static void _dispatch_kevent_guard(dispatch_kevent_t dk);
-static void _dispatch_kevent_unguard(dispatch_kevent_t dk);
-#else
-static inline void _dispatch_kevent_guard(dispatch_kevent_t dk) { (void)dk; }
-static inline void _dispatch_kevent_unguard(dispatch_kevent_t dk) { (void)dk; }
-#endif
-
-static struct dispatch_kevent_s _dispatch_kevent_data_or = {
- .dk_kevent = {
- .filter = DISPATCH_EVFILT_CUSTOM_OR,
- .flags = EV_CLEAR,
- },
- .dk_sources = TAILQ_HEAD_INITIALIZER(_dispatch_kevent_data_or.dk_sources),
-};
-static struct dispatch_kevent_s _dispatch_kevent_data_add = {
- .dk_kevent = {
- .filter = DISPATCH_EVFILT_CUSTOM_ADD,
- },
- .dk_sources = TAILQ_HEAD_INITIALIZER(_dispatch_kevent_data_add.dk_sources),
-};
-
-#define DSL_HASH(x) ((x) & (DSL_HASH_SIZE - 1))
-
-DISPATCH_CACHELINE_ALIGN
-static TAILQ_HEAD(, dispatch_kevent_s) _dispatch_sources[DSL_HASH_SIZE];
-
-static void
-_dispatch_kevent_init()
-{
- unsigned int i;
- for (i = 0; i < DSL_HASH_SIZE; i++) {
- TAILQ_INIT(&_dispatch_sources[i]);
- }
-
- TAILQ_INSERT_TAIL(&_dispatch_sources[0],
- &_dispatch_kevent_data_or, dk_list);
- TAILQ_INSERT_TAIL(&_dispatch_sources[0],
- &_dispatch_kevent_data_add, dk_list);
- _dispatch_kevent_data_or.dk_kevent.udata =
- (uintptr_t)&_dispatch_kevent_data_or;
- _dispatch_kevent_data_add.dk_kevent.udata =
- (uintptr_t)&_dispatch_kevent_data_add;
-}
-
-static inline uintptr_t
-_dispatch_kevent_hash(uint64_t ident, short filter)
-{
- uint64_t value;
-#if HAVE_MACH
- value = (filter == EVFILT_MACHPORT ||
- filter == DISPATCH_EVFILT_MACH_NOTIFICATION ?
- MACH_PORT_INDEX(ident) : ident);
-#else
- value = ident;
-#endif
- return DSL_HASH((uintptr_t)value);
-}
-
-static dispatch_kevent_t
-_dispatch_kevent_find(uint64_t ident, short filter)
-{
- uintptr_t hash = _dispatch_kevent_hash(ident, filter);
- dispatch_kevent_t dki;
- TAILQ_FOREACH(dki, &_dispatch_sources[hash], dk_list) {
- if (dki->dk_kevent.ident == ident && dki->dk_kevent.filter == filter) {
- break;
- }
+ if ((tq == DISPATCH_QUEUE_WAKEUP_TARGET) &&
+ ds->do_targetq == &_dispatch_mgr_q) {
+ tq = DISPATCH_QUEUE_WAKEUP_MGR;
}
- return dki;
-}
-static void
-_dispatch_kevent_insert(dispatch_kevent_t dk)
-{
- _dispatch_kevent_guard(dk);
- uintptr_t hash = _dispatch_kevent_hash(dk->dk_kevent.ident,
- dk->dk_kevent.filter);
- TAILQ_INSERT_TAIL(&_dispatch_sources[hash], dk, dk_list);
+ return _dispatch_queue_class_wakeup(ds->_as_dq, qos, flags, tq);
}
-// Find existing kevents, and merge any new flags if necessary
-static bool
-_dispatch_kevent_register(dispatch_kevent_t *dkp, uint32_t *flgp)
+void
+dispatch_source_cancel(dispatch_source_t ds)
{
- dispatch_kevent_t dk, ds_dkev = *dkp;
- uint32_t new_flags;
- bool do_resume = false;
-
- dk = _dispatch_kevent_find(ds_dkev->dk_kevent.ident,
- ds_dkev->dk_kevent.filter);
- if (dk) {
- // If an existing dispatch kevent is found, check to see if new flags
- // need to be added to the existing kevent
- new_flags = ~dk->dk_kevent.fflags & ds_dkev->dk_kevent.fflags;
- dk->dk_kevent.fflags |= ds_dkev->dk_kevent.fflags;
- free(ds_dkev);
- *dkp = dk;
- do_resume = new_flags;
- } else {
- dk = ds_dkev;
- _dispatch_kevent_insert(dk);
- new_flags = dk->dk_kevent.fflags;
- do_resume = true;
- }
- // Re-register the kevent with the kernel if new flags were added
- // by the dispatch kevent
- if (do_resume) {
- dk->dk_kevent.flags |= EV_ADD;
- }
- *flgp = new_flags;
- return do_resume;
-}
+ _dispatch_object_debug(ds, "%s", __func__);
+ // Right after we set the cancel flag, someone else
+ // could potentially invoke the source, do the cancellation,
+ // unregister the source, and deallocate it. We would
+ // need to therefore retain/release before setting the bit
+ _dispatch_retain_2(ds);
-static bool
-_dispatch_kevent_resume(dispatch_kevent_t dk, uint32_t new_flags,
- uint32_t del_flags)
-{
- long r;
- switch (dk->dk_kevent.filter) {
- case DISPATCH_EVFILT_TIMER:
- case DISPATCH_EVFILT_CUSTOM_ADD:
- case DISPATCH_EVFILT_CUSTOM_OR:
- // these types not registered with kevent
- return 0;
-#if HAVE_MACH
- case EVFILT_MACHPORT:
- return _dispatch_kevent_machport_resume(dk, new_flags, del_flags);
- case DISPATCH_EVFILT_MACH_NOTIFICATION:
- return _dispatch_kevent_mach_notify_resume(dk, new_flags, del_flags);
-#endif
- case EVFILT_PROC:
- if (dk->dk_kevent.flags & EV_ONESHOT) {
- return 0;
- }
- // fall through
- default:
- r = _dispatch_kq_update(&dk->dk_kevent);
- if (dk->dk_kevent.flags & EV_DISPATCH) {
- dk->dk_kevent.flags &= ~EV_ADD;
- }
- return r;
+ dispatch_queue_t q = ds->_as_dq;
+ if (_dispatch_queue_atomic_flags_set_orig(q, DSF_CANCELED) & DSF_CANCELED) {
+ _dispatch_release_2_tailcall(ds);
+ } else {
+ dx_wakeup(ds, 0, DISPATCH_WAKEUP_MAKE_DIRTY | DISPATCH_WAKEUP_CONSUME_2);
}
}
-static void
-_dispatch_kevent_dispose(dispatch_kevent_t dk)
+void
+dispatch_source_cancel_and_wait(dispatch_source_t ds)
{
- uintptr_t hash;
+ dispatch_queue_flags_t old_dqf, dqf, new_dqf;
+ dispatch_source_refs_t dr = ds->ds_refs;
- switch (dk->dk_kevent.filter) {
- case DISPATCH_EVFILT_TIMER:
- case DISPATCH_EVFILT_CUSTOM_ADD:
- case DISPATCH_EVFILT_CUSTOM_OR:
- // these sources live on statically allocated lists
- return;
-#if HAVE_MACH
- case EVFILT_MACHPORT:
- _dispatch_kevent_machport_resume(dk, 0, dk->dk_kevent.fflags);
- break;
- case DISPATCH_EVFILT_MACH_NOTIFICATION:
- _dispatch_kevent_mach_notify_resume(dk, 0, dk->dk_kevent.fflags);
- break;
-#endif
- case EVFILT_PROC:
- if (dk->dk_kevent.flags & EV_ONESHOT) {
- break; // implicitly deleted
- }
- // fall through
- default:
- if (~dk->dk_kevent.flags & EV_DELETE) {
- dk->dk_kevent.flags |= EV_DELETE;
- dk->dk_kevent.flags &= ~(EV_ADD|EV_ENABLE);
- _dispatch_kq_update(&dk->dk_kevent);
- }
- break;
+ if (unlikely(_dispatch_source_get_cancel_handler(dr))) {
+ DISPATCH_CLIENT_CRASH(ds, "Source has a cancel handler");
}
- hash = _dispatch_kevent_hash(dk->dk_kevent.ident,
- dk->dk_kevent.filter);
- TAILQ_REMOVE(&_dispatch_sources[hash], dk, dk_list);
- _dispatch_kevent_unguard(dk);
- free(dk);
-}
-
-static void
-_dispatch_kevent_unregister(dispatch_kevent_t dk, uint32_t flg)
-{
- dispatch_source_refs_t dri;
- uint32_t del_flags, fflags = 0;
-
- if (TAILQ_EMPTY(&dk->dk_sources)) {
- _dispatch_kevent_dispose(dk);
- } else {
- TAILQ_FOREACH(dri, &dk->dk_sources, dr_list) {
- dispatch_source_t dsi = _dispatch_source_from_refs(dri);
- uint32_t mask = (uint32_t)dsi->ds_pending_data_mask;
- fflags |= mask;
+ _dispatch_object_debug(ds, "%s", __func__);
+ os_atomic_rmw_loop2o(ds, dq_atomic_flags, old_dqf, new_dqf, relaxed, {
+ new_dqf = old_dqf | DSF_CANCELED;
+ if (old_dqf & DSF_CANCEL_WAITER) {
+ os_atomic_rmw_loop_give_up(break);
}
- del_flags = flg & ~fflags;
- if (del_flags) {
- dk->dk_kevent.flags |= EV_ADD;
- dk->dk_kevent.fflags = fflags;
- _dispatch_kevent_resume(dk, 0, del_flags);
+ if ((old_dqf & DSF_STATE_MASK) == DSF_DELETED) {
+ // just add DSF_CANCELED
+ } else if ((old_dqf & DSF_DEFERRED_DELETE) || !dr->du_is_direct) {
+ new_dqf |= DSF_CANCEL_WAITER;
}
- }
-}
-
-DISPATCH_NOINLINE
-static void
-_dispatch_kevent_proc_exit(struct kevent64_s *ke)
-{
- // EVFILT_PROC may fail with ESRCH when the process exists but is a zombie
- // <rdar://problem/5067725>. As a workaround, we simulate an exit event for
- // any EVFILT_PROC with an invalid pid <rdar://problem/6626350>.
- struct kevent64_s fake;
- fake = *ke;
- fake.flags &= ~EV_ERROR;
- fake.fflags = NOTE_EXIT;
- fake.data = 0;
- _dispatch_kevent_drain(&fake);
-}
+ });
+ dqf = new_dqf;
-DISPATCH_NOINLINE
-static void
-_dispatch_kevent_error(struct kevent64_s *ke)
-{
- _dispatch_kevent_debug(ke, __func__);
- if (ke->data) {
- // log the unexpected error
- _dispatch_bug_kevent_client("kevent", _evfiltstr(ke->filter),
- ke->flags & EV_DELETE ? "delete" :
- ke->flags & EV_ADD ? "add" :
- ke->flags & EV_ENABLE ? "enable" : "monitor",
- (int)ke->data);
+ if (old_dqf & DQF_RELEASED) {
+ DISPATCH_CLIENT_CRASH(ds, "Dispatch source used after last release");
}
-}
-
-static void
-_dispatch_kevent_drain(struct kevent64_s *ke)
-{
-#if DISPATCH_DEBUG
- static dispatch_once_t pred;
- dispatch_once_f(&pred, NULL, _dispatch_kevent_debugger);
-#endif
- if (ke->filter == EVFILT_USER) {
+ if ((old_dqf & DSF_STATE_MASK) == DSF_DELETED) {
return;
}
- if (slowpath(ke->flags & EV_ERROR)) {
- if (ke->filter == EVFILT_PROC) {
- if (ke->flags & EV_DELETE) {
- // Process exited while monitored
- return;
- } else if (ke->data == ESRCH) {
- return _dispatch_kevent_proc_exit(ke);
- }
- }
- return _dispatch_kevent_error(ke);
- }
- _dispatch_kevent_debug(ke, __func__);
- if (ke->filter == EVFILT_TIMER) {
- return _dispatch_timers_kevent(ke);
+ if (dqf & DSF_CANCEL_WAITER) {
+ goto wakeup;
}
-#if HAVE_MACH
- if (ke->filter == EVFILT_MACHPORT) {
- return _dispatch_kevent_mach_portset(ke);
- }
-#endif
- return _dispatch_kevent_merge(ke);
-}
-DISPATCH_NOINLINE
-static void
-_dispatch_kevent_merge(struct kevent64_s *ke)
-{
- dispatch_kevent_t dk;
- dispatch_source_refs_t dri;
-
- dk = (void*)ke->udata;
- dispatch_assert(dk);
-
- if (ke->flags & EV_ONESHOT) {
- dk->dk_kevent.flags |= EV_ONESHOT;
- }
- TAILQ_FOREACH(dri, &dk->dk_sources, dr_list) {
- _dispatch_source_merge_kevent(_dispatch_source_from_refs(dri), ke);
- }
-}
+ // simplified version of _dispatch_queue_drain_try_lock
+ // that also sets the DIRTY bit on failure to lock
+ uint64_t set_owner_and_set_full_width = _dispatch_lock_value_for_self() |
+ DISPATCH_QUEUE_WIDTH_FULL_BIT | DISPATCH_QUEUE_IN_BARRIER;
+ uint64_t old_state, new_state;
-#if DISPATCH_USE_GUARDED_FD_CHANGE_FDGUARD
-static void
-_dispatch_kevent_guard(dispatch_kevent_t dk)
-{
- guardid_t guard;
- const unsigned int guard_flags = GUARD_CLOSE;
- int r, fd_flags = 0;
- switch (dk->dk_kevent.filter) {
- case EVFILT_READ:
- case EVFILT_WRITE:
- case EVFILT_VNODE:
- guard = &dk->dk_kevent;
- r = change_fdguard_np((int)dk->dk_kevent.ident, NULL, 0,
- &guard, guard_flags, &fd_flags);
- if (slowpath(r == -1)) {
- int err = errno;
- if (err != EPERM) {
- (void)dispatch_assume_zero(err);
+ os_atomic_rmw_loop2o(ds, dq_state, old_state, new_state, seq_cst, {
+ new_state = old_state;
+ if (likely(_dq_state_is_runnable(old_state) &&
+ !_dq_state_drain_locked(old_state))) {
+ new_state &= DISPATCH_QUEUE_DRAIN_PRESERVED_BITS_MASK;
+ new_state |= set_owner_and_set_full_width;
+ } else if (old_dqf & DSF_CANCELED) {
+ os_atomic_rmw_loop_give_up(break);
+ } else {
+ // this case needs a release barrier, hence the seq_cst above
+ new_state |= DISPATCH_QUEUE_DIRTY;
+ }
+ });
+
+ if (unlikely(_dq_state_is_suspended(old_state))) {
+ if (unlikely(_dq_state_suspend_cnt(old_state))) {
+ DISPATCH_CLIENT_CRASH(ds, "Source is suspended");
+ }
+ // inactive sources have never been registered and there is no need
+ // to wait here because activation will notice and mark the source
+ // as deleted without ever trying to use the fd or mach port.
+ return dispatch_activate(ds);
+ }
+
+ if (likely(_dq_state_is_runnable(old_state) &&
+ !_dq_state_drain_locked(old_state))) {
+ // same thing _dispatch_source_invoke2() does when handling cancellation
+ dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
+ if (!(dqf & (DSF_DEFERRED_DELETE | DSF_DELETED))) {
+ _dispatch_source_refs_unregister(ds, 0);
+ dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
+ if (likely((dqf & DSF_STATE_MASK) == DSF_DELETED)) {
+ _dispatch_source_cancel_callout(ds, NULL, DISPATCH_INVOKE_NONE);
}
- return;
}
- dk->dk_kevent.ext[0] = guard_flags;
- dk->dk_kevent.ext[1] = fd_flags;
- break;
+ dx_wakeup(ds, 0, DISPATCH_WAKEUP_BARRIER_COMPLETE);
+ } else if (unlikely(_dq_state_drain_locked_by_self(old_state))) {
+ DISPATCH_CLIENT_CRASH(ds, "dispatch_source_cancel_and_wait "
+ "called from a source handler");
+ } else {
+ dispatch_qos_t qos;
+wakeup:
+ qos = _dispatch_qos_from_pp(_dispatch_get_priority());
+ dx_wakeup(ds, qos, DISPATCH_WAKEUP_MAKE_DIRTY);
+ dispatch_activate(ds);
+ }
+
+ dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
+ while (unlikely((dqf & DSF_STATE_MASK) != DSF_DELETED)) {
+ if (unlikely(!(dqf & DSF_CANCEL_WAITER))) {
+ if (!os_atomic_cmpxchgv2o(ds, dq_atomic_flags,
+ dqf, dqf | DSF_CANCEL_WAITER, &dqf, relaxed)) {
+ continue;
+ }
+ dqf |= DSF_CANCEL_WAITER;
+ }
+ _dispatch_wait_on_address(&ds->dq_atomic_flags, dqf, DLOCK_LOCK_NONE);
+ dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
}
}
-static void
-_dispatch_kevent_unguard(dispatch_kevent_t dk)
-{
- guardid_t guard;
- unsigned int guard_flags;
- int r, fd_flags;
- switch (dk->dk_kevent.filter) {
- case EVFILT_READ:
- case EVFILT_WRITE:
- case EVFILT_VNODE:
- guard_flags = (unsigned int)dk->dk_kevent.ext[0];
- if (!guard_flags) {
- return;
- }
- guard = &dk->dk_kevent;
- fd_flags = (int)dk->dk_kevent.ext[1];
- r = change_fdguard_np((int)dk->dk_kevent.ident, &guard,
- guard_flags, NULL, 0, &fd_flags);
- if (slowpath(r == -1)) {
- (void)dispatch_assume_zero(errno);
- return;
- }
- dk->dk_kevent.ext[0] = 0;
- break;
+void
+_dispatch_source_merge_evt(dispatch_unote_t du, uint32_t flags, uintptr_t data,
+ uintptr_t status, pthread_priority_t pp)
+{
+ dispatch_source_refs_t dr = du._dr;
+ dispatch_source_t ds = _dispatch_source_from_refs(dr);
+ dispatch_wakeup_flags_t wflags = 0;
+ dispatch_queue_flags_t dqf;
+
+ if (_dispatch_unote_needs_rearm(dr) || (flags & (EV_DELETE | EV_ONESHOT))) {
+ // once we modify the queue atomic flags below, it will allow concurrent
+ // threads running _dispatch_source_invoke2 to dispose of the source,
+ // so we can't safely borrow the reference we get from the muxnote udata
+ // anymore, and need our own
+ wflags = DISPATCH_WAKEUP_CONSUME_2;
+ _dispatch_retain_2(ds); // rdar://20382435
+ }
+
+ if ((flags & EV_UDATA_SPECIFIC) && (flags & EV_ONESHOT) &&
+ !(flags & EV_DELETE)) {
+ dqf = _dispatch_queue_atomic_flags_set_and_clear(ds->_as_dq,
+ DSF_DEFERRED_DELETE, DSF_ARMED);
+ if (flags & EV_VANISHED) {
+ _dispatch_bug_kevent_client("kevent", dr->du_type->dst_kind,
+ "monitored resource vanished before the source "
+ "cancel handler was invoked", 0);
+ }
+ _dispatch_debug("kevent-source[%p]: %s kevent[%p]", ds,
+ (flags & EV_VANISHED) ? "vanished" :
+ "deferred delete oneshot", dr);
+ } else if (flags & (EV_DELETE | EV_ONESHOT)) {
+ _dispatch_source_refs_unregister(ds, DU_UNREGISTER_ALREADY_DELETED);
+ _dispatch_debug("kevent-source[%p]: deleted kevent[%p]", ds, dr);
+ if (flags & EV_DELETE) goto done;
+ dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
+ } else if (_dispatch_unote_needs_rearm(dr)) {
+ dqf = _dispatch_queue_atomic_flags_clear(ds->_as_dq, DSF_ARMED);
+ _dispatch_debug("kevent-source[%p]: disarmed kevent[%p]", ds, dr);
+ } else {
+ dqf = _dispatch_queue_atomic_flags(ds->_as_dq);
+ }
+
+ if (dqf & (DSF_CANCELED | DQF_RELEASED)) {
+ goto done; // rdar://20204025
}
+
+ dispatch_unote_action_t action = dr->du_data_action;
+ if ((flags & EV_UDATA_SPECIFIC) && (flags & EV_ONESHOT) &&
+ (flags & EV_VANISHED)) {
+ // if the resource behind the ident vanished, the event handler can't
+ // do anything useful anymore, so do not try to call it at all
+ //
+ // Note: if the kernel doesn't support EV_VANISHED we always get it
+ // back unchanged from the flags passed at EV_ADD (registration) time
+ // Since we never ask for both EV_ONESHOT and EV_VANISHED for sources,
+ // if we get both bits it was a real EV_VANISHED delivery
+ os_atomic_store2o(ds, ds_pending_data, 0, relaxed);
+#if HAVE_MACH
+ } else if (dr->du_filter == EVFILT_MACHPORT) {
+ os_atomic_store2o(ds, ds_pending_data, data, relaxed);
+#endif
+ } else if (action == DISPATCH_UNOTE_ACTION_DATA_SET) {
+ os_atomic_store2o(ds, ds_pending_data, data, relaxed);
+ } else if (action == DISPATCH_UNOTE_ACTION_DATA_ADD) {
+ os_atomic_add2o(ds, ds_pending_data, data, relaxed);
+ } else if (data && action == DISPATCH_UNOTE_ACTION_DATA_OR) {
+ os_atomic_or2o(ds, ds_pending_data, data, relaxed);
+ } else if (data && action == DISPATCH_UNOTE_ACTION_DATA_OR_STATUS_SET) {
+ // We combine the data and status into a single 64-bit value.
+ uint64_t odata, ndata;
+ uint64_t value = DISPATCH_SOURCE_COMBINE_DATA_AND_STATUS(data, status);
+ os_atomic_rmw_loop2o(ds, ds_pending_data, odata, ndata, relaxed, {
+ ndata = DISPATCH_SOURCE_GET_DATA(odata) | value;
+ });
+ } else if (data) {
+ DISPATCH_INTERNAL_CRASH(action, "Unexpected source action value");
+ }
+ _dispatch_debug("kevent-source[%p]: merged kevent[%p]", ds, dr);
+
+done:
+ _dispatch_object_debug(ds, "%s", __func__);
+ dx_wakeup(ds, _dispatch_qos_from_pp(pp), wflags | DISPATCH_WAKEUP_MAKE_DIRTY);
}
-#endif // DISPATCH_USE_GUARDED_FD_CHANGE_FDGUARD
#pragma mark -
#pragma mark dispatch_source_timer
#if DISPATCH_USE_DTRACE
-static dispatch_source_refs_t
+static dispatch_timer_source_refs_t
_dispatch_trace_next_timer[DISPATCH_TIMER_QOS_COUNT];
#define _dispatch_trace_next_timer_set(x, q) \
_dispatch_trace_next_timer[(q)] = (x)
#define _dispatch_trace_next_timer_program(d, q) \
_dispatch_trace_timer_program(_dispatch_trace_next_timer[(q)], (d))
-#define _dispatch_trace_next_timer_wake(q) \
- _dispatch_trace_timer_wake(_dispatch_trace_next_timer[(q)])
+DISPATCH_ALWAYS_INLINE
+static inline void
+_dispatch_mgr_trace_timers_wakes(void)
+{
+ uint32_t qos;
+
+ if (_dispatch_timers_will_wake) {
+ if (slowpath(DISPATCH_TIMER_WAKE_ENABLED())) {
+ for (qos = 0; qos < DISPATCH_TIMER_QOS_COUNT; qos++) {
+ if (_dispatch_timers_will_wake & (1 << qos)) {
+ _dispatch_trace_timer_wake(_dispatch_trace_next_timer[qos]);
+ }
+ }
+ }
+ _dispatch_timers_will_wake = 0;
+ }
+}
#else
#define _dispatch_trace_next_timer_set(x, q)
#define _dispatch_trace_next_timer_program(d, q)
-#define _dispatch_trace_next_timer_wake(q)
+#define _dispatch_mgr_trace_timers_wakes()
#endif
#define _dispatch_source_timer_telemetry_enabled() false
DISPATCH_NOINLINE
static void
_dispatch_source_timer_telemetry_slow(dispatch_source_t ds,
- uintptr_t ident, struct dispatch_timer_source_s *values)
+ dispatch_clock_t clock, struct dispatch_timer_source_s *values)
{
if (_dispatch_trace_timer_configure_enabled()) {
- _dispatch_trace_timer_configure(ds, ident, values);
+ _dispatch_trace_timer_configure(ds, clock, values);
}
}
DISPATCH_ALWAYS_INLINE
static inline void
-_dispatch_source_timer_telemetry(dispatch_source_t ds, uintptr_t ident,
+_dispatch_source_timer_telemetry(dispatch_source_t ds, dispatch_clock_t clock,
struct dispatch_timer_source_s *values)
{
if (_dispatch_trace_timer_configure_enabled() ||
_dispatch_source_timer_telemetry_enabled()) {
- _dispatch_source_timer_telemetry_slow(ds, ident, values);
+ _dispatch_source_timer_telemetry_slow(ds, clock, values);
asm(""); // prevent tailcall
}
}
-// approx 1 year (60s * 60m * 24h * 365d)
-#define FOREVER_NSEC 31536000000000000ull
-
-DISPATCH_ALWAYS_INLINE
-static inline uint64_t
-_dispatch_source_timer_now(uint64_t nows[], unsigned int tidx)
-{
- unsigned int tk = DISPATCH_TIMER_KIND(tidx);
- if (nows && fastpath(nows[tk])) {
- return nows[tk];
- }
- uint64_t now;
- switch (tk) {
- case DISPATCH_TIMER_KIND_MACH:
- now = _dispatch_absolute_time();
- break;
- case DISPATCH_TIMER_KIND_WALL:
- now = _dispatch_get_nanoseconds();
- break;
- }
- if (nows) {
- nows[tk] = now;
- }
- return now;
-}
-
-static inline unsigned long
-_dispatch_source_timer_data(dispatch_source_refs_t dr, unsigned long prev)
-{
- // calculate the number of intervals since last fire
- unsigned long data, missed;
- uint64_t now;
- now = _dispatch_source_timer_now(NULL, _dispatch_source_timer_idx(dr));
- missed = (unsigned long)((now - ds_timer(dr).last_fire) /
- ds_timer(dr).interval);
- // correct for missed intervals already delivered last time
- data = prev - ds_timer(dr).missed + missed;
- ds_timer(dr).missed = missed;
- return data;
-}
-
-struct dispatch_set_timer_params {
- dispatch_source_t ds;
- uintptr_t ident;
- struct dispatch_timer_source_s values;
-};
-
+DISPATCH_NOINLINE
static void
-_dispatch_source_set_timer3(void *context)
+_dispatch_source_timer_configure(dispatch_source_t ds)
{
- // Called on the _dispatch_mgr_q
- struct dispatch_set_timer_params *params = context;
- dispatch_source_t ds = params->ds;
- ds->ds_ident_hack = params->ident;
- ds_timer(ds->ds_refs) = params->values;
- // Clear any pending data that might have accumulated on
- // older timer params <rdar://problem/8574886>
- ds->ds_pending_data = 0;
- // Re-arm in case we got disarmed because of pending set_timer suspension
- (void)dispatch_atomic_or2o(ds, ds_atomic_flags, DSF_ARMED, release);
- dispatch_resume(ds);
- // Must happen after resume to avoid getting disarmed due to suspension
- _dispatch_timers_update(ds);
- dispatch_release(ds);
- if (params->values.flags & DISPATCH_TIMER_WALL_CLOCK) {
- _dispatch_mach_host_calendar_change_register();
- }
- free(params);
-}
+ dispatch_timer_source_refs_t dt = ds->ds_timer_refs;
+ dispatch_timer_config_t dtc;
-static void
-_dispatch_source_set_timer2(void *context)
-{
- // Called on the source queue
- struct dispatch_set_timer_params *params = context;
- dispatch_suspend(params->ds);
- _dispatch_barrier_async_detached_f(&_dispatch_mgr_q, params,
- _dispatch_source_set_timer3);
+ dtc = os_atomic_xchg2o(dt, dt_pending_config, NULL, dependency);
+ if (dtc->dtc_clock == DISPATCH_CLOCK_MACH) {
+ dt->du_fflags |= DISPATCH_TIMER_CLOCK_MACH;
+ } else {
+ dt->du_fflags &= ~(uint32_t)DISPATCH_TIMER_CLOCK_MACH;
+ }
+ dt->dt_timer = dtc->dtc_timer;
+ free(dtc);
+ if (ds->ds_is_installed) {
+ // Clear any pending data that might have accumulated on
+ // older timer params <rdar://problem/8574886>
+ os_atomic_store2o(ds, ds_pending_data, 0, relaxed);
+ _dispatch_timers_update(dt, 0);
+ }
}
-DISPATCH_NOINLINE
-static struct dispatch_set_timer_params *
-_dispatch_source_timer_params(dispatch_source_t ds, dispatch_time_t start,
+static dispatch_timer_config_t
+_dispatch_source_timer_config_create(dispatch_time_t start,
uint64_t interval, uint64_t leeway)
{
- struct dispatch_set_timer_params *params;
- params = _dispatch_calloc(1ul, sizeof(struct dispatch_set_timer_params));
- params->ds = ds;
- params->values.flags = ds_timer(ds->ds_refs).flags;
-
- if (interval == 0) {
- // we use zero internally to mean disabled
+ dispatch_timer_config_t dtc;
+ dtc = _dispatch_calloc(1ul, sizeof(struct dispatch_timer_config_s));
+ if (unlikely(interval == 0)) {
+ if (start != DISPATCH_TIME_FOREVER) {
+ _dispatch_bug_deprecated("Setting timer interval to 0 requests "
+ "a 1ns timer, did you mean FOREVER (a one-shot timer)?");
+ }
interval = 1;
} else if ((int64_t)interval < 0) {
// 6866347 - make sure nanoseconds won't overflow
if ((int64_t)start < 0) {
// wall clock
start = (dispatch_time_t)-((int64_t)start);
- params->values.flags |= DISPATCH_TIMER_WALL_CLOCK;
+ dtc->dtc_clock = DISPATCH_CLOCK_WALL;
} else {
// absolute clock
interval = _dispatch_time_nano2mach(interval);
interval = 1;
}
leeway = _dispatch_time_nano2mach(leeway);
- params->values.flags &= ~(unsigned long)DISPATCH_TIMER_WALL_CLOCK;
- }
- params->ident = DISPATCH_TIMER_IDENT(params->values.flags);
- params->values.target = start;
- params->values.deadline = (start < UINT64_MAX - leeway) ?
- start + leeway : UINT64_MAX;
- params->values.interval = interval;
- params->values.leeway = (interval == INT64_MAX || leeway < interval / 2) ?
- leeway : interval / 2;
- return params;
-}
-
-DISPATCH_ALWAYS_INLINE
-static inline void
-_dispatch_source_set_timer(dispatch_source_t ds, dispatch_time_t start,
- uint64_t interval, uint64_t leeway, bool source_sync)
-{
- if (slowpath(!ds->ds_is_timer) ||
- slowpath(ds_timer(ds->ds_refs).flags & DISPATCH_TIMER_INTERVAL)) {
- DISPATCH_CLIENT_CRASH("Attempt to set timer on a non-timer source");
+ dtc->dtc_clock = DISPATCH_CLOCK_MACH;
+ }
+ if (interval < INT64_MAX && leeway > interval / 2) {
+ leeway = interval / 2;
}
- struct dispatch_set_timer_params *params;
- params = _dispatch_source_timer_params(ds, start, interval, leeway);
-
- _dispatch_source_timer_telemetry(ds, params->ident, ¶ms->values);
- // Suspend the source so that it doesn't fire with pending changes
- // The use of suspend/resume requires the external retain/release
- dispatch_retain(ds);
- if (source_sync) {
- return _dispatch_barrier_trysync_f((dispatch_queue_t)ds, params,
- _dispatch_source_set_timer2);
+ dtc->dtc_timer.target = start;
+ dtc->dtc_timer.interval = interval;
+ if (start + leeway < INT64_MAX) {
+ dtc->dtc_timer.deadline = start + leeway;
} else {
- return _dispatch_source_set_timer2(params);
+ dtc->dtc_timer.deadline = INT64_MAX;
}
+ return dtc;
}
+DISPATCH_NOINLINE
void
dispatch_source_set_timer(dispatch_source_t ds, dispatch_time_t start,
uint64_t interval, uint64_t leeway)
{
- _dispatch_source_set_timer(ds, start, interval, leeway, true);
-}
+ dispatch_timer_source_refs_t dt = ds->ds_timer_refs;
+ dispatch_timer_config_t dtc;
-void
-_dispatch_source_set_runloop_timer_4CF(dispatch_source_t ds,
- dispatch_time_t start, uint64_t interval, uint64_t leeway)
-{
- // Don't serialize through the source queue for CF timers <rdar://13833190>
- _dispatch_source_set_timer(ds, start, interval, leeway, false);
+ if (unlikely(!dt->du_is_timer || (dt->du_fflags&DISPATCH_TIMER_INTERVAL))) {
+ DISPATCH_CLIENT_CRASH(ds, "Attempt to set timer on a non-timer source");
+ }
+
+ dtc = _dispatch_source_timer_config_create(start, interval, leeway);
+ _dispatch_source_timer_telemetry(ds, dtc->dtc_clock, &dtc->dtc_timer);
+ dtc = os_atomic_xchg2o(dt, dt_pending_config, dtc, release);
+ if (dtc) free(dtc);
+ dx_wakeup(ds, 0, DISPATCH_WAKEUP_MAKE_DIRTY);
}
-void
+static void
_dispatch_source_set_interval(dispatch_source_t ds, uint64_t interval)
{
- dispatch_source_refs_t dr = ds->ds_refs;
- #define NSEC_PER_FRAME (NSEC_PER_SEC/60)
- const bool animation = ds_timer(dr).flags & DISPATCH_INTERVAL_UI_ANIMATION;
+#define NSEC_PER_FRAME (NSEC_PER_SEC/60)
+// approx 1 year (60s * 60m * 24h * 365d)
+#define FOREVER_NSEC 31536000000000000ull
+
+ dispatch_timer_source_refs_t dr = ds->ds_timer_refs;
+ const bool animation = dr->du_fflags & DISPATCH_INTERVAL_UI_ANIMATION;
if (fastpath(interval <= (animation ? FOREVER_NSEC/NSEC_PER_FRAME :
FOREVER_NSEC/NSEC_PER_MSEC))) {
interval *= animation ? NSEC_PER_FRAME : NSEC_PER_MSEC;
}
interval = _dispatch_time_nano2mach(interval);
uint64_t target = _dispatch_absolute_time() + interval;
- target = (target / interval) * interval;
+ target -= (target % interval);
const uint64_t leeway = animation ?
_dispatch_time_nano2mach(NSEC_PER_FRAME) : interval / 2;
- ds_timer(dr).target = target;
- ds_timer(dr).deadline = target + leeway;
- ds_timer(dr).interval = interval;
- ds_timer(dr).leeway = leeway;
- _dispatch_source_timer_telemetry(ds, ds->ds_ident_hack, &ds_timer(dr));
+ dr->dt_timer.target = target;
+ dr->dt_timer.deadline = target + leeway;
+ dr->dt_timer.interval = interval;
+ _dispatch_source_timer_telemetry(ds, DISPATCH_CLOCK_MACH, &dr->dt_timer);
}
#pragma mark -
-#pragma mark dispatch_timers
+#pragma mark dispatch_after
-#define DISPATCH_TIMER_STRUCT(refs) \
- uint64_t target, deadline; \
- TAILQ_HEAD(, refs) dt_sources
-
-typedef struct dispatch_timer_s {
- DISPATCH_TIMER_STRUCT(dispatch_timer_source_refs_s);
-} *dispatch_timer_t;
-
-#define DISPATCH_TIMER_INITIALIZER(tidx) \
- [tidx] = { \
- .target = UINT64_MAX, \
- .deadline = UINT64_MAX, \
- .dt_sources = TAILQ_HEAD_INITIALIZER( \
- _dispatch_timer[tidx].dt_sources), \
- }
-#define DISPATCH_TIMER_INIT(kind, qos) \
- DISPATCH_TIMER_INITIALIZER(DISPATCH_TIMER_INDEX( \
- DISPATCH_TIMER_KIND_##kind, DISPATCH_TIMER_QOS_##qos))
-
-struct dispatch_timer_s _dispatch_timer[] = {
- DISPATCH_TIMER_INIT(WALL, NORMAL),
- DISPATCH_TIMER_INIT(WALL, CRITICAL),
- DISPATCH_TIMER_INIT(WALL, BACKGROUND),
- DISPATCH_TIMER_INIT(MACH, NORMAL),
- DISPATCH_TIMER_INIT(MACH, CRITICAL),
- DISPATCH_TIMER_INIT(MACH, BACKGROUND),
-};
-#define DISPATCH_TIMER_COUNT \
- ((sizeof(_dispatch_timer) / sizeof(_dispatch_timer[0])))
-
-#define DISPATCH_KEVENT_TIMER_UDATA(tidx) \
- (uintptr_t)&_dispatch_kevent_timer[tidx]
-#ifdef __LP64__
-#define DISPATCH_KEVENT_TIMER_UDATA_INITIALIZER(tidx) \
- .udata = DISPATCH_KEVENT_TIMER_UDATA(tidx)
-#else // __LP64__
-// dynamic initialization in _dispatch_timers_init()
-#define DISPATCH_KEVENT_TIMER_UDATA_INITIALIZER(tidx) \
- .udata = 0
-#endif // __LP64__
-#define DISPATCH_KEVENT_TIMER_INITIALIZER(tidx) \
- [tidx] = { \
- .dk_kevent = { \
- .ident = tidx, \
- .filter = DISPATCH_EVFILT_TIMER, \
- DISPATCH_KEVENT_TIMER_UDATA_INITIALIZER(tidx), \
- }, \
- .dk_sources = TAILQ_HEAD_INITIALIZER( \
- _dispatch_kevent_timer[tidx].dk_sources), \
- }
-#define DISPATCH_KEVENT_TIMER_INIT(kind, qos) \
- DISPATCH_KEVENT_TIMER_INITIALIZER(DISPATCH_TIMER_INDEX( \
- DISPATCH_TIMER_KIND_##kind, DISPATCH_TIMER_QOS_##qos))
-
-struct dispatch_kevent_s _dispatch_kevent_timer[] = {
- DISPATCH_KEVENT_TIMER_INIT(WALL, NORMAL),
- DISPATCH_KEVENT_TIMER_INIT(WALL, CRITICAL),
- DISPATCH_KEVENT_TIMER_INIT(WALL, BACKGROUND),
- DISPATCH_KEVENT_TIMER_INIT(MACH, NORMAL),
- DISPATCH_KEVENT_TIMER_INIT(MACH, CRITICAL),
- DISPATCH_KEVENT_TIMER_INIT(MACH, BACKGROUND),
- DISPATCH_KEVENT_TIMER_INITIALIZER(DISPATCH_TIMER_INDEX_DISARM),
-};
-#define DISPATCH_KEVENT_TIMER_COUNT \
- ((sizeof(_dispatch_kevent_timer) / sizeof(_dispatch_kevent_timer[0])))
-
-#define DISPATCH_KEVENT_TIMEOUT_IDENT_MASK (~0ull << 8)
-#define DISPATCH_KEVENT_TIMEOUT_INITIALIZER(qos, note) \
- [qos] = { \
- .ident = DISPATCH_KEVENT_TIMEOUT_IDENT_MASK|(qos), \
- .filter = EVFILT_TIMER, \
- .flags = EV_ONESHOT, \
- .fflags = NOTE_ABSOLUTE|NOTE_NSECONDS|NOTE_LEEWAY|(note), \
- }
-#define DISPATCH_KEVENT_TIMEOUT_INIT(qos, note) \
- DISPATCH_KEVENT_TIMEOUT_INITIALIZER(DISPATCH_TIMER_QOS_##qos, note)
-
-struct kevent64_s _dispatch_kevent_timeout[] = {
- DISPATCH_KEVENT_TIMEOUT_INIT(NORMAL, 0),
- DISPATCH_KEVENT_TIMEOUT_INIT(CRITICAL, NOTE_CRITICAL),
- DISPATCH_KEVENT_TIMEOUT_INIT(BACKGROUND, NOTE_BACKGROUND),
-};
+DISPATCH_ALWAYS_INLINE
+static inline void
+_dispatch_after(dispatch_time_t when, dispatch_queue_t queue,
+ void *ctxt, void *handler, bool block)
+{
+ dispatch_timer_source_refs_t dt;
+ dispatch_source_t ds;
+ uint64_t leeway, delta;
-#define DISPATCH_KEVENT_COALESCING_WINDOW_INIT(qos, ms) \
- [DISPATCH_TIMER_QOS_##qos] = 2ull * (ms) * NSEC_PER_MSEC
+ if (when == DISPATCH_TIME_FOREVER) {
+#if DISPATCH_DEBUG
+ DISPATCH_CLIENT_CRASH(0, "dispatch_after called with 'when' == infinity");
+#endif
+ return;
+ }
-static const uint64_t _dispatch_kevent_coalescing_window[] = {
- DISPATCH_KEVENT_COALESCING_WINDOW_INIT(NORMAL, 75),
- DISPATCH_KEVENT_COALESCING_WINDOW_INIT(CRITICAL, 1),
- DISPATCH_KEVENT_COALESCING_WINDOW_INIT(BACKGROUND, 100),
-};
+ delta = _dispatch_timeout(when);
+ if (delta == 0) {
+ if (block) {
+ return dispatch_async(queue, handler);
+ }
+ return dispatch_async_f(queue, ctxt, handler);
+ }
+ leeway = delta / 10; // <rdar://problem/13447496>
-#define _dispatch_timers_insert(tidx, dra, dr, dr_list, dta, dt, dt_list) ({ \
- typeof(dr) dri = NULL; typeof(dt) dti; \
- if (tidx != DISPATCH_TIMER_INDEX_DISARM) { \
- TAILQ_FOREACH(dri, &dra[tidx].dk_sources, dr_list) { \
- if (ds_timer(dr).target < ds_timer(dri).target) { \
- break; \
- } \
- } \
- TAILQ_FOREACH(dti, &dta[tidx].dt_sources, dt_list) { \
- if (ds_timer(dt).deadline < ds_timer(dti).deadline) { \
- break; \
- } \
- } \
- if (dti) { \
- TAILQ_INSERT_BEFORE(dti, dt, dt_list); \
- } else { \
- TAILQ_INSERT_TAIL(&dta[tidx].dt_sources, dt, dt_list); \
- } \
- } \
- if (dri) { \
- TAILQ_INSERT_BEFORE(dri, dr, dr_list); \
- } else { \
- TAILQ_INSERT_TAIL(&dra[tidx].dk_sources, dr, dr_list); \
- } \
- })
-
-#define _dispatch_timers_remove(tidx, dk, dra, dr, dr_list, dta, dt, dt_list) \
- ({ \
- if (tidx != DISPATCH_TIMER_INDEX_DISARM) { \
- TAILQ_REMOVE(&dta[tidx].dt_sources, dt, dt_list); \
- } \
- TAILQ_REMOVE(dk ? &(*(dk)).dk_sources : &dra[tidx].dk_sources, dr, \
- dr_list); })
-
-#define _dispatch_timers_check(dra, dta) ({ \
- unsigned int qosm = _dispatch_timers_qos_mask; \
- bool update = false; \
- unsigned int tidx; \
- for (tidx = 0; tidx < DISPATCH_TIMER_COUNT; tidx++) { \
- if (!(qosm & 1 << DISPATCH_TIMER_QOS(tidx))){ \
- continue; \
- } \
- dispatch_timer_source_refs_t dr = (dispatch_timer_source_refs_t) \
- TAILQ_FIRST(&dra[tidx].dk_sources); \
- dispatch_timer_source_refs_t dt = (dispatch_timer_source_refs_t) \
- TAILQ_FIRST(&dta[tidx].dt_sources); \
- uint64_t target = dr ? ds_timer(dr).target : UINT64_MAX; \
- uint64_t deadline = dr ? ds_timer(dt).deadline : UINT64_MAX; \
- if (target != dta[tidx].target) { \
- dta[tidx].target = target; \
- update = true; \
- } \
- if (deadline != dta[tidx].deadline) { \
- dta[tidx].deadline = deadline; \
- update = true; \
- } \
- } \
- update; })
-
-static bool _dispatch_timers_reconfigure, _dispatch_timer_expired;
-static unsigned int _dispatch_timers_qos_mask;
-static bool _dispatch_timers_force_max_leeway;
+ if (leeway < NSEC_PER_MSEC) leeway = NSEC_PER_MSEC;
+ if (leeway > 60 * NSEC_PER_SEC) leeway = 60 * NSEC_PER_SEC;
-static void
-_dispatch_timers_init(void)
-{
-#ifndef __LP64__
- unsigned int tidx;
- for (tidx = 0; tidx < DISPATCH_TIMER_COUNT; tidx++) {
- _dispatch_kevent_timer[tidx].dk_kevent.udata = \
- DISPATCH_KEVENT_TIMER_UDATA(tidx);
+ // this function can and should be optimized to not use a dispatch source
+ ds = dispatch_source_create(&_dispatch_source_type_after, 0, 0, queue);
+ dt = ds->ds_timer_refs;
+
+ dispatch_continuation_t dc = _dispatch_continuation_alloc();
+ if (block) {
+ _dispatch_continuation_init(dc, ds, handler, 0, 0, 0);
+ } else {
+ _dispatch_continuation_init_f(dc, ds, ctxt, handler, 0, 0, 0);
}
-#endif // __LP64__
- if (slowpath(getenv("LIBDISPATCH_TIMERS_FORCE_MAX_LEEWAY"))) {
- _dispatch_timers_force_max_leeway = true;
+ // reference `ds` so that it doesn't show up as a leak
+ dc->dc_data = ds;
+ _dispatch_trace_continuation_push(ds->_as_dq, dc);
+ os_atomic_store2o(dt, ds_handler[DS_EVENT_HANDLER], dc, relaxed);
+
+ if ((int64_t)when < 0) {
+ // wall clock
+ when = (dispatch_time_t)-((int64_t)when);
+ } else {
+ // absolute clock
+ dt->du_fflags |= DISPATCH_TIMER_CLOCK_MACH;
+ leeway = _dispatch_time_nano2mach(leeway);
}
+ dt->dt_timer.target = when;
+ dt->dt_timer.interval = UINT64_MAX;
+ dt->dt_timer.deadline = when + leeway;
+ dispatch_activate(ds);
}
-static inline void
-_dispatch_timers_unregister(dispatch_source_t ds, dispatch_kevent_t dk)
+DISPATCH_NOINLINE
+void
+dispatch_after_f(dispatch_time_t when, dispatch_queue_t queue, void *ctxt,
+ dispatch_function_t func)
{
- dispatch_source_refs_t dr = ds->ds_refs;
- unsigned int tidx = (unsigned int)dk->dk_kevent.ident;
+ _dispatch_after(when, queue, ctxt, func, false);
+}
- if (slowpath(ds_timer_aggregate(ds))) {
- _dispatch_timer_aggregates_unregister(ds, tidx);
- }
- _dispatch_timers_remove(tidx, dk, _dispatch_kevent_timer, dr, dr_list,
- _dispatch_timer, (dispatch_timer_source_refs_t)dr, dt_list);
- if (tidx != DISPATCH_TIMER_INDEX_DISARM) {
- _dispatch_timers_reconfigure = true;
- _dispatch_timers_qos_mask |= 1 << DISPATCH_TIMER_QOS(tidx);
- }
+#ifdef __BLOCKS__
+void
+dispatch_after(dispatch_time_t when, dispatch_queue_t queue,
+ dispatch_block_t work)
+{
+ _dispatch_after(when, queue, NULL, work, true);
}
+#endif
-// Updates the ordered list of timers based on next fire date for changes to ds.
-// Should only be called from the context of _dispatch_mgr_q.
-static void
-_dispatch_timers_update(dispatch_source_t ds)
+#pragma mark -
+#pragma mark dispatch_timers
+
+/*
+ * The dispatch_timer_heap_t structure is a double min-heap of timers,
+ * interleaving the by-target min-heap in the even slots, and the by-deadline
+ * in the odd ones.
+ *
+ * The min element of these is held inline in the dispatch_timer_heap_t
+ * structure, and further entries are held in segments.
+ *
+ * dth_segments is the number of allocated segments.
+ *
+ * Segment 0 has a size of `DISPATCH_HEAP_INIT_SEGMENT_CAPACITY` pointers
+ * Segment k has a size of (DISPATCH_HEAP_INIT_SEGMENT_CAPACITY << (k - 1))
+ *
+ * Segment n (dth_segments - 1) is the last segment and points its final n
+ * entries to previous segments. Its address is held in the `dth_heap` field.
+ *
+ * segment n [ regular timer pointers | n-1 | k | 0 ]
+ * | | |
+ * segment n-1 <---------------------------' | |
+ * segment k <--------------------------------' |
+ * segment 0 <------------------------------------'
+ */
+#define DISPATCH_HEAP_INIT_SEGMENT_CAPACITY 8u
+
+/*
+ * There are two min-heaps stored interleaved in a single array,
+ * even indices are for the by-target min-heap, and odd indices for
+ * the by-deadline one.
+ */
+#define DTH_HEAP_ID_MASK (DTH_ID_COUNT - 1)
+#define DTH_HEAP_ID(idx) ((idx) & DTH_HEAP_ID_MASK)
+#define DTH_IDX_FOR_HEAP_ID(idx, heap_id) \
+ (((idx) & ~DTH_HEAP_ID_MASK) | (heap_id))
+
+DISPATCH_ALWAYS_INLINE
+static inline uint32_t
+_dispatch_timer_heap_capacity(uint32_t segments)
{
- dispatch_kevent_t dk = ds->ds_dkev;
- dispatch_source_refs_t dr = ds->ds_refs;
- unsigned int tidx;
+ if (segments == 0) return 2;
+ uint32_t seg_no = segments - 1;
+ // for C = DISPATCH_HEAP_INIT_SEGMENT_CAPACITY,
+ // 2 + C + SUM(C << (i-1), i = 1..seg_no) - seg_no
+ return 2 + (DISPATCH_HEAP_INIT_SEGMENT_CAPACITY << seg_no) - seg_no;
+}
- DISPATCH_ASSERT_ON_MANAGER_QUEUE();
+DISPATCH_NOINLINE
+static void
+_dispatch_timer_heap_grow(dispatch_timer_heap_t dth)
+{
+ uint32_t seg_capacity = DISPATCH_HEAP_INIT_SEGMENT_CAPACITY;
+ uint32_t seg_no = dth->dth_segments++;
+ void **heap, **heap_prev = dth->dth_heap;
- // Do not reschedule timers unregistered with _dispatch_kevent_unregister()
- if (slowpath(!dk)) {
- return;
+ if (seg_no > 0) {
+ seg_capacity <<= (seg_no - 1);
}
- // Move timers that are disabled, suspended or have missed intervals to the
- // disarmed list, rearm after resume resp. source invoke will reenable them
- if (!ds_timer(dr).target || DISPATCH_OBJECT_SUSPENDED(ds) ||
- ds->ds_pending_data) {
- tidx = DISPATCH_TIMER_INDEX_DISARM;
- (void)dispatch_atomic_and2o(ds, ds_atomic_flags, ~DSF_ARMED, relaxed);
- } else {
- tidx = _dispatch_source_timer_idx(dr);
+ heap = _dispatch_calloc(seg_capacity, sizeof(void *));
+ if (seg_no > 1) {
+ uint32_t prev_seg_no = seg_no - 1;
+ uint32_t prev_seg_capacity = seg_capacity >> 1;
+ memcpy(&heap[seg_capacity - prev_seg_no],
+ &heap_prev[prev_seg_capacity - prev_seg_no],
+ prev_seg_no * sizeof(void *));
}
- if (slowpath(ds_timer_aggregate(ds))) {
- _dispatch_timer_aggregates_register(ds);
+ if (seg_no > 0) {
+ heap[seg_capacity - seg_no] = heap_prev;
}
- if (slowpath(!ds->ds_is_installed)) {
- ds->ds_is_installed = true;
- if (tidx != DISPATCH_TIMER_INDEX_DISARM) {
- (void)dispatch_atomic_or2o(ds, ds_atomic_flags, DSF_ARMED, relaxed);
- }
- _dispatch_object_debug(ds, "%s", __func__);
- ds->ds_dkev = NULL;
- free(dk);
- } else {
- _dispatch_timers_unregister(ds, dk);
+ dth->dth_heap = heap;
+}
+
+DISPATCH_NOINLINE
+static void
+_dispatch_timer_heap_shrink(dispatch_timer_heap_t dth)
+{
+ uint32_t seg_capacity = DISPATCH_HEAP_INIT_SEGMENT_CAPACITY;
+ uint32_t seg_no = --dth->dth_segments;
+ void **heap = dth->dth_heap, **heap_prev = NULL;
+
+ if (seg_no > 0) {
+ seg_capacity <<= (seg_no - 1);
+ heap_prev = heap[seg_capacity - seg_no];
}
- if (tidx != DISPATCH_TIMER_INDEX_DISARM) {
- _dispatch_timers_reconfigure = true;
- _dispatch_timers_qos_mask |= 1 << DISPATCH_TIMER_QOS(tidx);
+ if (seg_no > 1) {
+ uint32_t prev_seg_no = seg_no - 1;
+ uint32_t prev_seg_capacity = seg_capacity >> 1;
+ memcpy(&heap_prev[prev_seg_capacity - prev_seg_no],
+ &heap[seg_capacity - prev_seg_no],
+ prev_seg_no * sizeof(void *));
}
- if (dk != &_dispatch_kevent_timer[tidx]){
- ds->ds_dkev = &_dispatch_kevent_timer[tidx];
+ dth->dth_heap = heap_prev;
+ free(heap);
+}
+
+DISPATCH_ALWAYS_INLINE
+static inline dispatch_timer_source_refs_t *
+_dispatch_timer_heap_get_slot(dispatch_timer_heap_t dth, uint32_t idx)
+{
+ uint32_t seg_no, segments = dth->dth_segments;
+ void **segment;
+
+ if (idx < DTH_ID_COUNT) {
+ return &dth->dth_min[idx];
+ }
+ idx -= DTH_ID_COUNT;
+
+ // Derive the segment number from the index. Naming
+ // DISPATCH_HEAP_INIT_SEGMENT_CAPACITY `C`, the segments index ranges are:
+ // 0: 0 .. (C - 1)
+ // 1: C .. 2 * C - 1
+ // k: 2^(k-1) * C .. 2^k * C - 1
+ // so `k` can be derived from the first bit set in `idx`
+ seg_no = (uint32_t)(__builtin_clz(DISPATCH_HEAP_INIT_SEGMENT_CAPACITY - 1) -
+ __builtin_clz(idx | (DISPATCH_HEAP_INIT_SEGMENT_CAPACITY - 1)));
+ if (seg_no + 1 == segments) {
+ segment = dth->dth_heap;
+ } else {
+ uint32_t seg_capacity = DISPATCH_HEAP_INIT_SEGMENT_CAPACITY;
+ seg_capacity <<= (segments - 2);
+ segment = dth->dth_heap[seg_capacity - seg_no - 1];
}
- _dispatch_timers_insert(tidx, _dispatch_kevent_timer, dr, dr_list,
- _dispatch_timer, (dispatch_timer_source_refs_t)dr, dt_list);
- if (slowpath(ds_timer_aggregate(ds))) {
- _dispatch_timer_aggregates_update(ds, tidx);
+ if (seg_no) {
+ idx -= DISPATCH_HEAP_INIT_SEGMENT_CAPACITY << (seg_no - 1);
}
+ return (dispatch_timer_source_refs_t *)(segment + idx);
}
+DISPATCH_ALWAYS_INLINE
static inline void
-_dispatch_timers_run2(uint64_t nows[], unsigned int tidx)
+_dispatch_timer_heap_set(dispatch_timer_source_refs_t *slot,
+ dispatch_timer_source_refs_t dt, uint32_t idx)
{
- dispatch_source_refs_t dr;
- dispatch_source_t ds;
- uint64_t now, missed;
+ *slot = dt;
+ dt->dt_heap_entry[DTH_HEAP_ID(idx)] = idx;
+}
- now = _dispatch_source_timer_now(nows, tidx);
- while ((dr = TAILQ_FIRST(&_dispatch_kevent_timer[tidx].dk_sources))) {
- ds = _dispatch_source_from_refs(dr);
- // We may find timers on the wrong list due to a pending update from
- // dispatch_source_set_timer. Force an update of the list in that case.
- if (tidx != ds->ds_ident_hack) {
- _dispatch_timers_update(ds);
- continue;
- }
- if (!ds_timer(dr).target) {
- // No configured timers on the list
- break;
- }
- if (ds_timer(dr).target > now) {
- // Done running timers for now.
- break;
- }
- // Remove timers that are suspended or have missed intervals from the
- // list, rearm after resume resp. source invoke will reenable them
- if (DISPATCH_OBJECT_SUSPENDED(ds) || ds->ds_pending_data) {
- _dispatch_timers_update(ds);
- continue;
- }
- // Calculate number of missed intervals.
- missed = (now - ds_timer(dr).target) / ds_timer(dr).interval;
- if (++missed > INT_MAX) {
- missed = INT_MAX;
- }
- if (ds_timer(dr).interval < INT64_MAX) {
- ds_timer(dr).target += missed * ds_timer(dr).interval;
- ds_timer(dr).deadline = ds_timer(dr).target + ds_timer(dr).leeway;
+DISPATCH_ALWAYS_INLINE
+static inline uint32_t
+_dispatch_timer_heap_parent(uint32_t idx)
+{
+ uint32_t heap_id = DTH_HEAP_ID(idx);
+ idx = (idx - DTH_ID_COUNT) / 2; // go to the parent
+ return DTH_IDX_FOR_HEAP_ID(idx, heap_id);
+}
+
+DISPATCH_ALWAYS_INLINE
+static inline uint32_t
+_dispatch_timer_heap_left_child(uint32_t idx)
+{
+ uint32_t heap_id = DTH_HEAP_ID(idx);
+ // 2 * (idx - heap_id) + DTH_ID_COUNT + heap_id
+ return 2 * idx + DTH_ID_COUNT - heap_id;
+}
+
+#if DISPATCH_HAVE_TIMER_COALESCING
+DISPATCH_ALWAYS_INLINE
+static inline uint32_t
+_dispatch_timer_heap_walk_skip(uint32_t idx, uint32_t count)
+{
+ uint32_t heap_id = DTH_HEAP_ID(idx);
+
+ idx -= heap_id;
+ if (unlikely(idx + DTH_ID_COUNT == count)) {
+ // reaching `count` doesn't mean we're done, but there is a weird
+ // corner case if the last item of the heap is a left child:
+ //
+ // /\
+ // / \
+ // / __\
+ // /__/
+ // ^
+ //
+ // The formula below would return the sibling of `idx` which is
+ // out of bounds. Fortunately, the correct answer is the same
+ // as for idx's parent
+ idx = _dispatch_timer_heap_parent(idx);
+ }
+
+ //
+ // When considering the index in a non interleaved, 1-based array
+ // representation of a heap, hence looking at (idx / DTH_ID_COUNT + 1)
+ // for a given idx in our dual-heaps, that index is in one of two forms:
+ //
+ // (a) 1xxxx011111 or (b) 111111111
+ // d i 0 d 0
+ //
+ // The first bit set is the row of the binary tree node (0-based).
+ // The following digits from most to least significant represent the path
+ // to that node, where `0` is a left turn and `1` a right turn.
+ //
+ // For example 0b0101 (5) is a node on row 2 accessed going left then right:
+ //
+ // row 0 1
+ // / .
+ // row 1 2 3
+ // . \ . .
+ // row 2 4 5 6 7
+ // : : : : : : : :
+ //
+ // Skipping a sub-tree in walk order means going to the sibling of the last
+ // node reached after we turned left. If the node was of the form (a),
+ // this node is 1xxxx1, which for the above example is 0b0011 (3).
+ // If the node was of the form (b) then we never took a left, meaning
+ // we reached the last element in traversal order.
+ //
+
+ //
+ // we want to find
+ // - the least significant bit set to 0 in (idx / DTH_ID_COUNT + 1)
+ // - which is offset by log_2(DTH_ID_COUNT) from the position of the least
+ // significant 0 in (idx + DTH_ID_COUNT + DTH_ID_COUNT - 1)
+ // since idx is a multiple of DTH_ID_COUNT and DTH_ID_COUNT a power of 2.
+ // - which in turn is the same as the position of the least significant 1 in
+ // ~(idx + DTH_ID_COUNT + DTH_ID_COUNT - 1)
+ //
+ dispatch_static_assert(powerof2(DTH_ID_COUNT));
+ idx += DTH_ID_COUNT + DTH_ID_COUNT - 1;
+ idx >>= __builtin_ctz(~idx);
+
+ //
+ // `idx` is now either:
+ // - 0 if it was the (b) case above, in which case the walk is done
+ // - 1xxxx0 as the position in a 0 based array representation of a non
+ // interleaved heap, so we just have to compute the interleaved index.
+ //
+ return likely(idx) ? DTH_ID_COUNT * idx + heap_id : UINT32_MAX;
+}
+
+DISPATCH_ALWAYS_INLINE
+static inline uint32_t
+_dispatch_timer_heap_walk_next(uint32_t idx, uint32_t count)
+{
+ //
+ // Goes to the next element in heap walk order, which is the prefix ordered
+ // walk of the tree.
+ //
+ // From a given node, the next item to return is the left child if it
+ // exists, else the first right sibling we find by walking our parent chain,
+ // which is exactly what _dispatch_timer_heap_walk_skip() returns.
+ //
+ uint32_t lchild = _dispatch_timer_heap_left_child(idx);
+ if (lchild < count) {
+ return lchild;
+ }
+ return _dispatch_timer_heap_walk_skip(idx, count);
+}
+
+DISPATCH_NOINLINE
+static uint64_t
+_dispatch_timer_heap_max_target_before(dispatch_timer_heap_t dth, uint64_t limit)
+{
+ dispatch_timer_source_refs_t dri;
+ uint32_t idx = _dispatch_timer_heap_left_child(DTH_TARGET_ID);
+ uint32_t count = dth->dth_count;
+ uint64_t tmp, target = dth->dth_min[DTH_TARGET_ID]->dt_timer.target;
+
+ while (idx < count) {
+ dri = *_dispatch_timer_heap_get_slot(dth, idx);
+ tmp = dri->dt_timer.target;
+ if (tmp > limit) {
+ // skip subtree since none of the targets below can be before limit
+ idx = _dispatch_timer_heap_walk_skip(idx, count);
} else {
- ds_timer(dr).target = UINT64_MAX;
- ds_timer(dr).deadline = UINT64_MAX;
+ target = tmp;
+ idx = _dispatch_timer_heap_walk_next(idx, count);
}
- _dispatch_timers_update(ds);
- ds_timer(dr).last_fire = now;
-
- unsigned long data;
- data = dispatch_atomic_add2o(ds, ds_pending_data,
- (unsigned long)missed, relaxed);
- _dispatch_trace_timer_fire(dr, data, (unsigned long)missed);
- _dispatch_wakeup(ds);
}
+ return target;
}
+#endif // DISPATCH_HAVE_TIMER_COALESCING
DISPATCH_NOINLINE
static void
-_dispatch_timers_run(uint64_t nows[])
-{
- unsigned int tidx;
- for (tidx = 0; tidx < DISPATCH_TIMER_COUNT; tidx++) {
- if (!TAILQ_EMPTY(&_dispatch_kevent_timer[tidx].dk_sources)) {
- _dispatch_timers_run2(nows, tidx);
+_dispatch_timer_heap_resift(dispatch_timer_heap_t dth,
+ dispatch_timer_source_refs_t dt, uint32_t idx)
+{
+ dispatch_static_assert(offsetof(struct dispatch_timer_source_s, target) ==
+ offsetof(struct dispatch_timer_source_s, heap_key[DTH_TARGET_ID]));
+ dispatch_static_assert(offsetof(struct dispatch_timer_source_s, deadline) ==
+ offsetof(struct dispatch_timer_source_s, heap_key[DTH_DEADLINE_ID]));
+#define dth_cmp(hid, dt1, op, dt2) \
+ (((dt1)->dt_timer.heap_key)[hid] op ((dt2)->dt_timer.heap_key)[hid])
+
+ dispatch_timer_source_refs_t *pslot, pdt;
+ dispatch_timer_source_refs_t *cslot, cdt;
+ dispatch_timer_source_refs_t *rslot, rdt;
+ uint32_t cidx, dth_count = dth->dth_count;
+ dispatch_timer_source_refs_t *slot;
+ int heap_id = DTH_HEAP_ID(idx);
+ bool sifted_up = false;
+
+ // try to sift up
+
+ slot = _dispatch_timer_heap_get_slot(dth, idx);
+ while (idx >= DTH_ID_COUNT) {
+ uint32_t pidx = _dispatch_timer_heap_parent(idx);
+ pslot = _dispatch_timer_heap_get_slot(dth, pidx);
+ pdt = *pslot;
+ if (dth_cmp(heap_id, pdt, <=, dt)) {
+ break;
}
+ _dispatch_timer_heap_set(slot, pdt, idx);
+ slot = pslot;
+ idx = pidx;
+ sifted_up = true;
+ }
+ if (sifted_up) {
+ goto done;
}
-}
-static inline unsigned int
-_dispatch_timers_get_delay(uint64_t nows[], struct dispatch_timer_s timer[],
- uint64_t *delay, uint64_t *leeway, int qos)
-{
- unsigned int tidx, ridx = DISPATCH_TIMER_COUNT;
- uint64_t tmp, delta = UINT64_MAX, dldelta = UINT64_MAX;
+ // try to sift down
- for (tidx = 0; tidx < DISPATCH_TIMER_COUNT; tidx++) {
- if (qos >= 0 && qos != DISPATCH_TIMER_QOS(tidx)){
- continue;
- }
- uint64_t target = timer[tidx].target;
- if (target == UINT64_MAX) {
- continue;
- }
- uint64_t deadline = timer[tidx].deadline;
- if (qos >= 0) {
- // Timer pre-coalescing <rdar://problem/13222034>
- uint64_t window = _dispatch_kevent_coalescing_window[qos];
- uint64_t latest = deadline > window ? deadline - window : 0;
- dispatch_source_refs_t dri;
- TAILQ_FOREACH(dri, &_dispatch_kevent_timer[tidx].dk_sources,
- dr_list) {
- tmp = ds_timer(dri).target;
- if (tmp > latest) break;
- target = tmp;
+ while ((cidx = _dispatch_timer_heap_left_child(idx)) < dth_count) {
+ uint32_t ridx = cidx + DTH_ID_COUNT;
+ cslot = _dispatch_timer_heap_get_slot(dth, cidx);
+ cdt = *cslot;
+ if (ridx < dth_count) {
+ rslot = _dispatch_timer_heap_get_slot(dth, ridx);
+ rdt = *rslot;
+ if (dth_cmp(heap_id, cdt, >, rdt)) {
+ cidx = ridx;
+ cdt = rdt;
+ cslot = rslot;
}
}
- uint64_t now = _dispatch_source_timer_now(nows, tidx);
- if (target <= now) {
- delta = 0;
+ if (dth_cmp(heap_id, dt, <=, cdt)) {
break;
}
- tmp = target - now;
- if (DISPATCH_TIMER_KIND(tidx) != DISPATCH_TIMER_KIND_WALL) {
- tmp = _dispatch_time_mach2nano(tmp);
- }
- if (tmp < INT64_MAX && tmp < delta) {
- ridx = tidx;
- delta = tmp;
- }
- dispatch_assert(target <= deadline);
- tmp = deadline - now;
- if (DISPATCH_TIMER_KIND(tidx) != DISPATCH_TIMER_KIND_WALL) {
- tmp = _dispatch_time_mach2nano(tmp);
- }
- if (tmp < INT64_MAX && tmp < dldelta) {
- dldelta = tmp;
- }
+ _dispatch_timer_heap_set(slot, cdt, idx);
+ slot = cslot;
+ idx = cidx;
}
- *delay = delta;
- *leeway = delta && delta < UINT64_MAX ? dldelta - delta : UINT64_MAX;
- return ridx;
+
+done:
+ _dispatch_timer_heap_set(slot, dt, idx);
+#undef dth_cmp
}
-static bool
-_dispatch_timers_program2(uint64_t nows[], struct kevent64_s *ke,
- unsigned int qos)
+DISPATCH_ALWAYS_INLINE
+static void
+_dispatch_timer_heap_insert(dispatch_timer_heap_t dth,
+ dispatch_timer_source_refs_t dt)
{
- unsigned int tidx;
- bool poll;
- uint64_t delay, leeway;
+ uint32_t idx = (dth->dth_count += DTH_ID_COUNT) - DTH_ID_COUNT;
- tidx = _dispatch_timers_get_delay(nows, _dispatch_timer, &delay, &leeway,
- (int)qos);
- poll = (delay == 0);
- if (poll || delay == UINT64_MAX) {
- _dispatch_trace_next_timer_set(NULL, qos);
- if (!ke->data) {
- return poll;
- }
- ke->data = 0;
- ke->flags |= EV_DELETE;
- ke->flags &= ~(EV_ADD|EV_ENABLE);
- } else {
- _dispatch_trace_next_timer_set(
- TAILQ_FIRST(&_dispatch_kevent_timer[tidx].dk_sources), qos);
- _dispatch_trace_next_timer_program(delay, qos);
- delay += _dispatch_source_timer_now(nows, DISPATCH_TIMER_KIND_WALL);
- if (slowpath(_dispatch_timers_force_max_leeway)) {
- ke->data = (int64_t)(delay + leeway);
- ke->ext[1] = 0;
- } else {
- ke->data = (int64_t)delay;
- ke->ext[1] = leeway;
- }
- ke->flags |= EV_ADD|EV_ENABLE;
- ke->flags &= ~EV_DELETE;
+ DISPATCH_TIMER_ASSERT(dt->dt_heap_entry[DTH_TARGET_ID], ==,
+ DTH_INVALID_ID, "target idx");
+ DISPATCH_TIMER_ASSERT(dt->dt_heap_entry[DTH_DEADLINE_ID], ==,
+ DTH_INVALID_ID, "deadline idx");
+
+ if (idx == 0) {
+ dt->dt_heap_entry[DTH_TARGET_ID] = DTH_TARGET_ID;
+ dt->dt_heap_entry[DTH_DEADLINE_ID] = DTH_DEADLINE_ID;
+ dth->dth_min[DTH_TARGET_ID] = dth->dth_min[DTH_DEADLINE_ID] = dt;
+ return;
}
- _dispatch_kq_update(ke);
- return poll;
+
+ if (unlikely(idx + DTH_ID_COUNT >
+ _dispatch_timer_heap_capacity(dth->dth_segments))) {
+ _dispatch_timer_heap_grow(dth);
+ }
+ _dispatch_timer_heap_resift(dth, dt, idx + DTH_TARGET_ID);
+ _dispatch_timer_heap_resift(dth, dt, idx + DTH_DEADLINE_ID);
}
DISPATCH_NOINLINE
-static bool
-_dispatch_timers_program(uint64_t nows[])
+static void
+_dispatch_timer_heap_remove(dispatch_timer_heap_t dth,
+ dispatch_timer_source_refs_t dt)
{
- bool poll = false;
- unsigned int qos, qosm = _dispatch_timers_qos_mask;
- for (qos = 0; qos < DISPATCH_TIMER_QOS_COUNT; qos++) {
- if (!(qosm & 1 << qos)){
- continue;
+ uint32_t idx = (dth->dth_count -= DTH_ID_COUNT);
+
+ DISPATCH_TIMER_ASSERT(dt->dt_heap_entry[DTH_TARGET_ID], !=,
+ DTH_INVALID_ID, "target idx");
+ DISPATCH_TIMER_ASSERT(dt->dt_heap_entry[DTH_DEADLINE_ID], !=,
+ DTH_INVALID_ID, "deadline idx");
+
+ if (idx == 0) {
+ DISPATCH_TIMER_ASSERT(dth->dth_min[DTH_TARGET_ID], ==, dt,
+ "target slot");
+ DISPATCH_TIMER_ASSERT(dth->dth_min[DTH_DEADLINE_ID], ==, dt,
+ "deadline slot");
+ dth->dth_min[DTH_TARGET_ID] = dth->dth_min[DTH_DEADLINE_ID] = NULL;
+ goto clear_heap_entry;
+ }
+
+ for (uint32_t heap_id = 0; heap_id < DTH_ID_COUNT; heap_id++) {
+ dispatch_timer_source_refs_t *slot, last_dt;
+ slot = _dispatch_timer_heap_get_slot(dth, idx + heap_id);
+ last_dt = *slot; *slot = NULL;
+ if (last_dt != dt) {
+ uint32_t removed_idx = dt->dt_heap_entry[heap_id];
+ _dispatch_timer_heap_resift(dth, last_dt, removed_idx);
}
- poll |= _dispatch_timers_program2(nows, &_dispatch_kevent_timeout[qos],
- qos);
}
- return poll;
-}
+ if (unlikely(idx <= _dispatch_timer_heap_capacity(dth->dth_segments - 1))) {
+ _dispatch_timer_heap_shrink(dth);
+ }
-DISPATCH_NOINLINE
-static bool
-_dispatch_timers_configure(void)
-{
- _dispatch_timer_aggregates_check();
- // Find out if there is a new target/deadline on the timer lists
- return _dispatch_timers_check(_dispatch_kevent_timer, _dispatch_timer);
+clear_heap_entry:
+ dt->dt_heap_entry[DTH_TARGET_ID] = DTH_INVALID_ID;
+ dt->dt_heap_entry[DTH_DEADLINE_ID] = DTH_INVALID_ID;
}
-static void
-_dispatch_timers_calendar_change(void)
+DISPATCH_ALWAYS_INLINE
+static inline void
+_dispatch_timer_heap_update(dispatch_timer_heap_t dth,
+ dispatch_timer_source_refs_t dt)
{
- // calendar change may have gone past the wallclock deadline
- _dispatch_timer_expired = true;
- _dispatch_timers_qos_mask = ~0u;
-}
+ DISPATCH_TIMER_ASSERT(dt->dt_heap_entry[DTH_TARGET_ID], !=,
+ DTH_INVALID_ID, "target idx");
+ DISPATCH_TIMER_ASSERT(dt->dt_heap_entry[DTH_DEADLINE_ID], !=,
+ DTH_INVALID_ID, "deadline idx");
-static void
-_dispatch_timers_kevent(struct kevent64_s *ke)
-{
- dispatch_assert(ke->data > 0);
- dispatch_assert((ke->ident & DISPATCH_KEVENT_TIMEOUT_IDENT_MASK) ==
- DISPATCH_KEVENT_TIMEOUT_IDENT_MASK);
- unsigned int qos = ke->ident & ~DISPATCH_KEVENT_TIMEOUT_IDENT_MASK;
- dispatch_assert(qos < DISPATCH_TIMER_QOS_COUNT);
- dispatch_assert(_dispatch_kevent_timeout[qos].data);
- _dispatch_kevent_timeout[qos].data = 0; // kevent deleted via EV_ONESHOT
- _dispatch_timer_expired = true;
- _dispatch_timers_qos_mask |= 1 << qos;
- _dispatch_trace_next_timer_wake(qos);
+
+ _dispatch_timer_heap_resift(dth, dt, dt->dt_heap_entry[DTH_TARGET_ID]);
+ _dispatch_timer_heap_resift(dth, dt, dt->dt_heap_entry[DTH_DEADLINE_ID]);
}
-static inline bool
-_dispatch_mgr_timers(void)
+DISPATCH_ALWAYS_INLINE
+static bool
+_dispatch_timer_heap_has_new_min(dispatch_timer_heap_t dth,
+ uint32_t count, uint32_t mask)
{
- uint64_t nows[DISPATCH_TIMER_KIND_COUNT] = {};
- bool expired = slowpath(_dispatch_timer_expired);
- if (expired) {
- _dispatch_timers_run(nows);
- }
- bool reconfigure = slowpath(_dispatch_timers_reconfigure);
- if (reconfigure || expired) {
- if (reconfigure) {
- reconfigure = _dispatch_timers_configure();
- _dispatch_timers_reconfigure = false;
+ dispatch_timer_source_refs_t dt;
+ bool changed = false;
+ uint64_t tmp;
+ uint32_t tidx;
+
+ for (tidx = 0; tidx < count; tidx++) {
+ if (!(mask & (1u << tidx))) {
+ continue;
}
- if (reconfigure || expired) {
- expired = _dispatch_timer_expired = _dispatch_timers_program(nows);
- expired = expired || _dispatch_mgr_q.dq_items_tail;
+
+ dt = dth[tidx].dth_min[DTH_TARGET_ID];
+ tmp = dt ? dt->dt_timer.target : UINT64_MAX;
+ if (dth[tidx].dth_target != tmp) {
+ dth[tidx].dth_target = tmp;
+ changed = true;
+ }
+ dt = dth[tidx].dth_min[DTH_DEADLINE_ID];
+ tmp = dt ? dt->dt_timer.deadline : UINT64_MAX;
+ if (dth[tidx].dth_deadline != tmp) {
+ dth[tidx].dth_deadline = tmp;
+ changed = true;
}
- _dispatch_timers_qos_mask = 0;
}
- return expired;
+ return changed;
}
-#pragma mark -
-#pragma mark dispatch_timer_aggregate
-
-typedef struct {
- TAILQ_HEAD(, dispatch_timer_source_aggregate_refs_s) dk_sources;
-} dispatch_timer_aggregate_refs_s;
-
-typedef struct dispatch_timer_aggregate_s {
- DISPATCH_STRUCT_HEADER(queue);
- DISPATCH_QUEUE_HEADER;
- TAILQ_ENTRY(dispatch_timer_aggregate_s) dta_list;
- dispatch_timer_aggregate_refs_s
- dta_kevent_timer[DISPATCH_KEVENT_TIMER_COUNT];
- struct {
- DISPATCH_TIMER_STRUCT(dispatch_timer_source_aggregate_refs_s);
- } dta_timer[DISPATCH_TIMER_COUNT];
- struct dispatch_timer_s dta_timer_data[DISPATCH_TIMER_COUNT];
- unsigned int dta_refcount;
-} dispatch_timer_aggregate_s;
-
-typedef TAILQ_HEAD(, dispatch_timer_aggregate_s) dispatch_timer_aggregates_s;
-static dispatch_timer_aggregates_s _dispatch_timer_aggregates =
- TAILQ_HEAD_INITIALIZER(_dispatch_timer_aggregates);
-
-dispatch_timer_aggregate_t
-dispatch_timer_aggregate_create(void)
+static inline void
+_dispatch_timers_unregister(dispatch_timer_source_refs_t dt)
{
- unsigned int tidx;
- dispatch_timer_aggregate_t dta = _dispatch_alloc(DISPATCH_VTABLE(queue),
- sizeof(struct dispatch_timer_aggregate_s));
- _dispatch_queue_init((dispatch_queue_t)dta);
- dta->do_targetq = _dispatch_get_root_queue(
- _DISPATCH_QOS_CLASS_USER_INITIATED, true);
- dta->dq_width = DISPATCH_QUEUE_WIDTH_MAX;
- //FIXME: aggregates need custom vtable
- //dta->dq_label = "timer-aggregate";
- for (tidx = 0; tidx < DISPATCH_KEVENT_TIMER_COUNT; tidx++) {
- TAILQ_INIT(&dta->dta_kevent_timer[tidx].dk_sources);
- }
- for (tidx = 0; tidx < DISPATCH_TIMER_COUNT; tidx++) {
- TAILQ_INIT(&dta->dta_timer[tidx].dt_sources);
- dta->dta_timer[tidx].target = UINT64_MAX;
- dta->dta_timer[tidx].deadline = UINT64_MAX;
- dta->dta_timer_data[tidx].target = UINT64_MAX;
- dta->dta_timer_data[tidx].deadline = UINT64_MAX;
- }
- return (dispatch_timer_aggregate_t)_dispatch_introspection_queue_create(
- (dispatch_queue_t)dta);
-}
-
-typedef struct dispatch_timer_delay_s {
- dispatch_timer_t timer;
- uint64_t delay, leeway;
-} *dispatch_timer_delay_t;
+ uint32_t tidx = dt->du_ident;
+ dispatch_timer_heap_t heap = &_dispatch_timers_heap[tidx];
-static void
-_dispatch_timer_aggregate_get_delay(void *ctxt)
-{
- dispatch_timer_delay_t dtd = ctxt;
- struct { uint64_t nows[DISPATCH_TIMER_KIND_COUNT]; } dtn = {};
- _dispatch_timers_get_delay(dtn.nows, dtd->timer, &dtd->delay, &dtd->leeway,
- -1);
+ _dispatch_timer_heap_remove(heap, dt);
+ _dispatch_timers_reconfigure = true;
+ _dispatch_timers_processing_mask |= 1 << tidx;
+ dispatch_assert(dt->du_wlh == NULL || dt->du_wlh == DISPATCH_WLH_ANON);
+ dt->du_wlh = NULL;
}
-uint64_t
-dispatch_timer_aggregate_get_delay(dispatch_timer_aggregate_t dta,
- uint64_t *leeway_ptr)
+static inline void
+_dispatch_timers_register(dispatch_timer_source_refs_t dt, uint32_t tidx)
{
- struct dispatch_timer_delay_s dtd = {
- .timer = dta->dta_timer_data,
- };
- dispatch_sync_f((dispatch_queue_t)dta, &dtd,
- _dispatch_timer_aggregate_get_delay);
- if (leeway_ptr) {
- *leeway_ptr = dtd.leeway;
+ dispatch_timer_heap_t heap = &_dispatch_timers_heap[tidx];
+ if (_dispatch_unote_registered(dt)) {
+ DISPATCH_TIMER_ASSERT(dt->du_ident, ==, tidx, "tidx");
+ _dispatch_timer_heap_update(heap, dt);
+ } else {
+ dt->du_ident = tidx;
+ _dispatch_timer_heap_insert(heap, dt);
}
- return dtd.delay;
+ _dispatch_timers_reconfigure = true;
+ _dispatch_timers_processing_mask |= 1 << tidx;
+ dispatch_assert(dt->du_wlh == NULL || dt->du_wlh == DISPATCH_WLH_ANON);
+ dt->du_wlh = DISPATCH_WLH_ANON;
}
-static void
-_dispatch_timer_aggregate_update(void *ctxt)
+DISPATCH_ALWAYS_INLINE
+static inline bool
+_dispatch_source_timer_tryarm(dispatch_source_t ds)
{
- dispatch_timer_aggregate_t dta = (void*)_dispatch_queue_get_current();
- dispatch_timer_t dtau = ctxt;
- unsigned int tidx;
- for (tidx = 0; tidx < DISPATCH_TIMER_COUNT; tidx++) {
- dta->dta_timer_data[tidx].target = dtau[tidx].target;
- dta->dta_timer_data[tidx].deadline = dtau[tidx].deadline;
- }
- free(dtau);
+ dispatch_queue_flags_t oqf, nqf;
+ return os_atomic_rmw_loop2o(ds, dq_atomic_flags, oqf, nqf, relaxed, {
+ if (oqf & (DSF_CANCELED | DQF_RELEASED)) {
+ // do not install a cancelled timer
+ os_atomic_rmw_loop_give_up(break);
+ }
+ nqf = oqf | DSF_ARMED;
+ });
}
-DISPATCH_NOINLINE
+// Updates the ordered list of timers based on next fire date for changes to ds.
+// Should only be called from the context of _dispatch_mgr_q.
static void
-_dispatch_timer_aggregates_configure(void)
+_dispatch_timers_update(dispatch_unote_t du, uint32_t flags)
{
- dispatch_timer_aggregate_t dta;
- dispatch_timer_t dtau;
- TAILQ_FOREACH(dta, &_dispatch_timer_aggregates, dta_list) {
- if (!_dispatch_timers_check(dta->dta_kevent_timer, dta->dta_timer)) {
- continue;
+ dispatch_timer_source_refs_t dr = du._dt;
+ dispatch_source_t ds = _dispatch_source_from_refs(dr);
+ const char *verb = "updated";
+ bool will_register, disarm = false;
+
+ DISPATCH_ASSERT_ON_MANAGER_QUEUE();
+
+ if (unlikely(dr->du_ident == DISPATCH_TIMER_IDENT_CANCELED)) {
+ dispatch_assert((flags & DISPATCH_TIMERS_RETAIN_2) == 0);
+ return;
+ }
+
+ // Unregister timers that are unconfigured, disabled, suspended or have
+ // missed intervals. Rearm after dispatch_set_timer(), resume or source
+ // invoke will reenable them
+ will_register = !(flags & DISPATCH_TIMERS_UNREGISTER) &&
+ dr->dt_timer.target < INT64_MAX &&
+ !os_atomic_load2o(ds, ds_pending_data, relaxed) &&
+ !DISPATCH_QUEUE_IS_SUSPENDED(ds) &&
+ !os_atomic_load2o(dr, dt_pending_config, relaxed);
+ if (likely(!_dispatch_unote_registered(dr))) {
+ dispatch_assert((flags & DISPATCH_TIMERS_RETAIN_2) == 0);
+ if (unlikely(!will_register || !_dispatch_source_timer_tryarm(ds))) {
+ return;
}
- dtau = _dispatch_calloc(DISPATCH_TIMER_COUNT, sizeof(*dtau));
- memcpy(dtau, dta->dta_timer, sizeof(dta->dta_timer));
- _dispatch_barrier_async_detached_f((dispatch_queue_t)dta, dtau,
- _dispatch_timer_aggregate_update);
+ verb = "armed";
+ } else if (unlikely(!will_register)) {
+ disarm = true;
+ verb = "disarmed";
}
-}
-static inline void
-_dispatch_timer_aggregates_check(void)
-{
- if (fastpath(TAILQ_EMPTY(&_dispatch_timer_aggregates))) {
- return;
+ // The heap owns a +2 on dispatch sources it references
+ //
+ // _dispatch_timers_run2() also sometimes passes DISPATCH_TIMERS_RETAIN_2
+ // when it wants to take over this +2 at the same time we are unregistering
+ // the timer from the heap.
+ //
+ // Compute our refcount balance according to these rules, if our balance
+ // would become negative we retain the source upfront, if it is positive, we
+ // get rid of the extraneous refcounts after we're done touching the source.
+ int refs = will_register ? -2 : 0;
+ if (_dispatch_unote_registered(dr) && !(flags & DISPATCH_TIMERS_RETAIN_2)) {
+ refs += 2;
+ }
+ if (refs < 0) {
+ dispatch_assert(refs == -2);
+ _dispatch_retain_2(ds);
}
- _dispatch_timer_aggregates_configure();
-}
-static void
-_dispatch_timer_aggregates_register(dispatch_source_t ds)
-{
- dispatch_timer_aggregate_t dta = ds_timer_aggregate(ds);
- if (!dta->dta_refcount++) {
- TAILQ_INSERT_TAIL(&_dispatch_timer_aggregates, dta, dta_list);
+ uint32_t tidx = _dispatch_source_timer_idx(dr);
+ if (unlikely(_dispatch_unote_registered(dr) &&
+ (!will_register || dr->du_ident != tidx))) {
+ _dispatch_timers_unregister(dr);
+ }
+ if (likely(will_register)) {
+ _dispatch_timers_register(dr, tidx);
}
-}
-DISPATCH_NOINLINE
-static void
-_dispatch_timer_aggregates_update(dispatch_source_t ds, unsigned int tidx)
-{
- dispatch_timer_aggregate_t dta = ds_timer_aggregate(ds);
- dispatch_timer_source_aggregate_refs_t dr;
- dr = (dispatch_timer_source_aggregate_refs_t)ds->ds_refs;
- _dispatch_timers_insert(tidx, dta->dta_kevent_timer, dr, dra_list,
- dta->dta_timer, dr, dta_list);
+ if (disarm) {
+ _dispatch_queue_atomic_flags_clear(ds->_as_dq, DSF_ARMED);
+ }
+ _dispatch_debug("kevent-source[%p]: %s timer[%p]", ds, verb, dr);
+ _dispatch_object_debug(ds, "%s", __func__);
+ if (refs > 0) {
+ dispatch_assert(refs == 2);
+ _dispatch_release_2_tailcall(ds);
+ }
}
-DISPATCH_NOINLINE
-static void
-_dispatch_timer_aggregates_unregister(dispatch_source_t ds, unsigned int tidx)
+#define DISPATCH_TIMER_MISSED_MARKER 1ul
+
+DISPATCH_ALWAYS_INLINE
+static inline unsigned long
+_dispatch_source_timer_compute_missed(dispatch_timer_source_refs_t dt,
+ uint64_t now, unsigned long prev)
{
- dispatch_timer_aggregate_t dta = ds_timer_aggregate(ds);
- dispatch_timer_source_aggregate_refs_t dr;
- dr = (dispatch_timer_source_aggregate_refs_t)ds->ds_refs;
- _dispatch_timers_remove(tidx, (dispatch_timer_aggregate_refs_s*)NULL,
- dta->dta_kevent_timer, dr, dra_list, dta->dta_timer, dr, dta_list);
- if (!--dta->dta_refcount) {
- TAILQ_REMOVE(&_dispatch_timer_aggregates, dta, dta_list);
+ uint64_t missed = (now - dt->dt_timer.target) / dt->dt_timer.interval;
+ if (++missed + prev > LONG_MAX) {
+ missed = LONG_MAX - prev;
+ }
+ if (dt->dt_timer.interval < INT64_MAX) {
+ uint64_t push_by = missed * dt->dt_timer.interval;
+ dt->dt_timer.target += push_by;
+ dt->dt_timer.deadline += push_by;
+ } else {
+ dt->dt_timer.target = UINT64_MAX;
+ dt->dt_timer.deadline = UINT64_MAX;
}
+ prev += missed;
+ return prev;
}
-#pragma mark -
-#pragma mark dispatch_select
+DISPATCH_ALWAYS_INLINE
+static inline unsigned long
+_dispatch_source_timer_data(dispatch_source_t ds, dispatch_unote_t du)
+{
+ dispatch_timer_source_refs_t dr = du._dt;
+ unsigned long data, prev, clear_prev = 0;
-static int _dispatch_kq;
+ os_atomic_rmw_loop2o(ds, ds_pending_data, prev, clear_prev, relaxed, {
+ data = prev >> 1;
+ if (unlikely(prev & DISPATCH_TIMER_MISSED_MARKER)) {
+ os_atomic_rmw_loop_give_up(goto handle_missed_intervals);
+ }
+ });
+ return data;
-static unsigned int _dispatch_select_workaround;
-static fd_set _dispatch_rfds;
-static fd_set _dispatch_wfds;
-static uint64_t*_dispatch_rfd_ptrs;
-static uint64_t*_dispatch_wfd_ptrs;
+handle_missed_intervals:
+ // The timer may be in _dispatch_source_invoke2() already for other
+ // reasons such as running the registration handler when ds_pending_data
+ // is changed by _dispatch_timers_run2() without holding the drain lock.
+ //
+ // We hence need dependency ordering to pair with the release barrier
+ // done by _dispatch_timers_run2() when setting the MISSED_MARKER bit.
+ os_atomic_thread_fence(dependency);
+ dr = os_atomic_force_dependency_on(dr, data);
+
+ uint64_t now = _dispatch_time_now(DISPATCH_TIMER_CLOCK(dr->du_ident));
+ if (now >= dr->dt_timer.target) {
+ OS_COMPILER_CAN_ASSUME(dr->dt_timer.interval < INT64_MAX);
+ data = _dispatch_source_timer_compute_missed(dr, now, data);
+ }
+
+ // When we see the MISSED_MARKER the manager has given up on this timer
+ // and expects the handler to call "resume".
+ //
+ // However, it may not have reflected this into the atomic flags yet
+ // so make sure _dispatch_source_invoke2() sees the timer is disarmed
+ //
+ // The subsequent _dispatch_source_refs_resume() will enqueue the source
+ // on the manager and make the changes to `ds_timer` above visible.
+ _dispatch_queue_atomic_flags_clear(ds->_as_dq, DSF_ARMED);
+ os_atomic_store2o(ds, ds_pending_data, 0, relaxed);
+ return data;
+}
-DISPATCH_NOINLINE
-static bool
-_dispatch_select_register(struct kevent64_s *kev)
+static inline void
+_dispatch_timers_run2(dispatch_clock_now_cache_t nows, uint32_t tidx)
{
+ dispatch_timer_source_refs_t dr;
+ dispatch_source_t ds;
+ uint64_t data, pending_data;
+ uint64_t now = _dispatch_time_now_cached(DISPATCH_TIMER_CLOCK(tidx), nows);
+
+ while ((dr = _dispatch_timers_heap[tidx].dth_min[DTH_TARGET_ID])) {
+ DISPATCH_TIMER_ASSERT(dr->du_filter, ==, DISPATCH_EVFILT_TIMER,
+ "invalid filter");
+ DISPATCH_TIMER_ASSERT(dr->du_ident, ==, tidx, "tidx");
+ DISPATCH_TIMER_ASSERT(dr->dt_timer.target, !=, 0, "missing target");
+ ds = _dispatch_source_from_refs(dr);
+ if (dr->dt_timer.target > now) {
+ // Done running timers for now.
+ break;
+ }
+ if (dr->du_fflags & DISPATCH_TIMER_AFTER) {
+ _dispatch_trace_timer_fire(dr, 1, 1);
+ _dispatch_source_merge_evt(dr, EV_ONESHOT, 1, 0, 0);
+ _dispatch_debug("kevent-source[%p]: fired after timer[%p]", ds, dr);
+ _dispatch_object_debug(ds, "%s", __func__);
+ continue;
+ }
- // Must execute on manager queue
- DISPATCH_ASSERT_ON_MANAGER_QUEUE();
-
- // If an EINVAL or ENOENT error occurred while adding/enabling a read or
- // write kevent, assume it was due to a type of filedescriptor not
- // supported by kqueue and fall back to select
- switch (kev->filter) {
- case EVFILT_READ:
- if ((kev->data == EINVAL || kev->data == ENOENT) &&
- dispatch_assume(kev->ident < FD_SETSIZE)) {
- FD_SET((int)kev->ident, &_dispatch_rfds);
- if (slowpath(!_dispatch_rfd_ptrs)) {
- _dispatch_rfd_ptrs = _dispatch_calloc(FD_SETSIZE,
- sizeof(*_dispatch_rfd_ptrs));
- }
- if (!_dispatch_rfd_ptrs[kev->ident]) {
- _dispatch_rfd_ptrs[kev->ident] = kev->udata;
- _dispatch_select_workaround++;
- _dispatch_debug("select workaround used to read fd %d: 0x%lx",
- (int)kev->ident, (long)kev->data);
+ data = os_atomic_load2o(ds, ds_pending_data, relaxed);
+ if (unlikely(data)) {
+ // the release barrier is required to make the changes
+ // to `ds_timer` visible to _dispatch_source_timer_data()
+ if (os_atomic_cmpxchg2o(ds, ds_pending_data, data,
+ data | DISPATCH_TIMER_MISSED_MARKER, release)) {
+ _dispatch_timers_update(dr, DISPATCH_TIMERS_UNREGISTER);
+ continue;
}
}
- return true;
- case EVFILT_WRITE:
- if ((kev->data == EINVAL || kev->data == ENOENT) &&
- dispatch_assume(kev->ident < FD_SETSIZE)) {
- FD_SET((int)kev->ident, &_dispatch_wfds);
- if (slowpath(!_dispatch_wfd_ptrs)) {
- _dispatch_wfd_ptrs = _dispatch_calloc(FD_SETSIZE,
- sizeof(*_dispatch_wfd_ptrs));
- }
- if (!_dispatch_wfd_ptrs[kev->ident]) {
- _dispatch_wfd_ptrs[kev->ident] = kev->udata;
- _dispatch_select_workaround++;
- _dispatch_debug("select workaround used to write fd %d: 0x%lx",
- (int)kev->ident, (long)kev->data);
- }
+
+ data = _dispatch_source_timer_compute_missed(dr, now, 0);
+ _dispatch_timers_update(dr, DISPATCH_TIMERS_RETAIN_2);
+ pending_data = data << 1;
+ if (!_dispatch_unote_registered(dr) && dr->dt_timer.target < INT64_MAX){
+ // if we unregistered because of suspension we have to fake we
+ // missed events.
+ pending_data |= DISPATCH_TIMER_MISSED_MARKER;
+ os_atomic_store2o(ds, ds_pending_data, pending_data, release);
+ } else {
+ os_atomic_store2o(ds, ds_pending_data, pending_data, relaxed);
}
- return true;
+ _dispatch_trace_timer_fire(dr, data, data);
+ _dispatch_debug("kevent-source[%p]: fired timer[%p]", ds, dr);
+ _dispatch_object_debug(ds, "%s", __func__);
+ dx_wakeup(ds, 0, DISPATCH_WAKEUP_MAKE_DIRTY | DISPATCH_WAKEUP_CONSUME_2);
}
- return false;
}
DISPATCH_NOINLINE
-static bool
-_dispatch_select_unregister(const struct kevent64_s *kev)
+static void
+_dispatch_timers_run(dispatch_clock_now_cache_t nows)
{
- // Must execute on manager queue
- DISPATCH_ASSERT_ON_MANAGER_QUEUE();
-
- switch (kev->filter) {
- case EVFILT_READ:
- if (_dispatch_rfd_ptrs && kev->ident < FD_SETSIZE &&
- _dispatch_rfd_ptrs[kev->ident]) {
- FD_CLR((int)kev->ident, &_dispatch_rfds);
- _dispatch_rfd_ptrs[kev->ident] = 0;
- _dispatch_select_workaround--;
- return true;
- }
- break;
- case EVFILT_WRITE:
- if (_dispatch_wfd_ptrs && kev->ident < FD_SETSIZE &&
- _dispatch_wfd_ptrs[kev->ident]) {
- FD_CLR((int)kev->ident, &_dispatch_wfds);
- _dispatch_wfd_ptrs[kev->ident] = 0;
- _dispatch_select_workaround--;
- return true;
+ uint32_t tidx;
+ for (tidx = 0; tidx < DISPATCH_TIMER_COUNT; tidx++) {
+ if (_dispatch_timers_heap[tidx].dth_count) {
+ _dispatch_timers_run2(nows, tidx);
}
- break;
}
- return false;
}
-DISPATCH_NOINLINE
-static bool
-_dispatch_mgr_select(bool poll)
-{
- static const struct timeval timeout_immediately = { 0, 0 };
- fd_set tmp_rfds, tmp_wfds;
- struct kevent64_s kev;
- int err, i, r;
- bool kevent_avail = false;
-
- FD_COPY(&_dispatch_rfds, &tmp_rfds);
- FD_COPY(&_dispatch_wfds, &tmp_wfds);
-
- r = select(FD_SETSIZE, &tmp_rfds, &tmp_wfds, NULL,
- poll ? (struct timeval*)&timeout_immediately : NULL);
- if (slowpath(r == -1)) {
- err = errno;
- if (err != EBADF) {
- if (err != EINTR) {
- (void)dispatch_assume_zero(err);
- }
- return false;
- }
- for (i = 0; i < FD_SETSIZE; i++) {
- if (i == _dispatch_kq) {
- continue;
- }
- if (!FD_ISSET(i, &_dispatch_rfds) && !FD_ISSET(i, &_dispatch_wfds)){
- continue;
- }
- r = dup(i);
- if (dispatch_assume(r != -1)) {
- close(r);
- } else {
- if (_dispatch_rfd_ptrs && _dispatch_rfd_ptrs[i]) {
- FD_CLR(i, &_dispatch_rfds);
- _dispatch_rfd_ptrs[i] = 0;
- _dispatch_select_workaround--;
- }
- if (_dispatch_wfd_ptrs && _dispatch_wfd_ptrs[i]) {
- FD_CLR(i, &_dispatch_wfds);
- _dispatch_wfd_ptrs[i] = 0;
- _dispatch_select_workaround--;
- }
- }
+#if DISPATCH_HAVE_TIMER_COALESCING
+#define DISPATCH_KEVENT_COALESCING_WINDOW_INIT(qos, ms) \
+ [DISPATCH_TIMER_QOS_##qos] = 2ull * (ms) * NSEC_PER_MSEC
+
+static const uint64_t _dispatch_kevent_coalescing_window[] = {
+ DISPATCH_KEVENT_COALESCING_WINDOW_INIT(NORMAL, 75),
+#if DISPATCH_HAVE_TIMER_QOS
+ DISPATCH_KEVENT_COALESCING_WINDOW_INIT(CRITICAL, 1),
+ DISPATCH_KEVENT_COALESCING_WINDOW_INIT(BACKGROUND, 100),
+#endif
+};
+#endif // DISPATCH_HAVE_TIMER_COALESCING
+
+static inline dispatch_timer_delay_s
+_dispatch_timers_get_delay(dispatch_timer_heap_t dth, dispatch_clock_t clock,
+ uint32_t qos, dispatch_clock_now_cache_t nows)
+{
+ uint64_t target = dth->dth_target, deadline = dth->dth_deadline;
+ uint64_t delta = INT64_MAX, dldelta = INT64_MAX;
+ dispatch_timer_delay_s rc;
+
+ dispatch_assert(target <= deadline);
+ if (delta == 0 || target >= INT64_MAX) {
+ goto done;
+ }
+
+ if (qos < DISPATCH_TIMER_QOS_COUNT && dth->dth_count > 2) {
+#if DISPATCH_HAVE_TIMER_COALESCING
+ // Timer pre-coalescing <rdar://problem/13222034>
+ // When we have several timers with this target/deadline bracket:
+ //
+ // Target window Deadline
+ // V <-------V
+ // t1: [...........|.................]
+ // t2: [......|.......]
+ // t3: [..|..........]
+ // t4: | [.............]
+ // ^
+ // Optimal Target
+ //
+ // Coalescing works better if the Target is delayed to "Optimal", by
+ // picking the latest target that isn't too close to the deadline.
+ uint64_t window = _dispatch_kevent_coalescing_window[qos];
+ if (target + window < deadline) {
+ uint64_t latest = deadline - window;
+ target = _dispatch_timer_heap_max_target_before(dth, latest);
}
- return false;
+#endif
}
- if (r > 0) {
- for (i = 0; i < FD_SETSIZE; i++) {
- if (FD_ISSET(i, &tmp_rfds)) {
- if (i == _dispatch_kq) {
- kevent_avail = true;
- continue;
- }
- FD_CLR(i, &_dispatch_rfds); // emulate EV_DISPATCH
- EV_SET64(&kev, i, EVFILT_READ,
- EV_ADD|EV_ENABLE|EV_DISPATCH, 0, 1,
- _dispatch_rfd_ptrs[i], 0, 0);
- _dispatch_kevent_drain(&kev);
- }
- if (FD_ISSET(i, &tmp_wfds)) {
- FD_CLR(i, &_dispatch_wfds); // emulate EV_DISPATCH
- EV_SET64(&kev, i, EVFILT_WRITE,
- EV_ADD|EV_ENABLE|EV_DISPATCH, 0, 1,
- _dispatch_wfd_ptrs[i], 0, 0);
- _dispatch_kevent_drain(&kev);
- }
- }
+
+ uint64_t now = _dispatch_time_now_cached(clock, nows);
+ if (target <= now) {
+ delta = 0;
+ dldelta = 0;
+ goto done;
}
- return kevent_avail;
-}
-#pragma mark -
-#pragma mark dispatch_kqueue
+ uint64_t tmp = target - now;
+ if (clock != DISPATCH_CLOCK_WALL) {
+ tmp = _dispatch_time_mach2nano(tmp);
+ }
+ if (tmp < delta) {
+ delta = tmp;
+ }
-static void
-_dispatch_kq_init(void *context DISPATCH_UNUSED)
+ tmp = deadline - now;
+ if (clock != DISPATCH_CLOCK_WALL) {
+ tmp = _dispatch_time_mach2nano(tmp);
+ }
+ if (tmp < dldelta) {
+ dldelta = tmp;
+ }
+
+done:
+ rc.delay = delta;
+ rc.leeway = delta < INT64_MAX ? dldelta - delta : INT64_MAX;
+ return rc;
+}
+
+static bool
+_dispatch_timers_program2(dispatch_clock_now_cache_t nows, uint32_t tidx)
{
- static const struct kevent64_s kev = {
- .ident = 1,
- .filter = EVFILT_USER,
- .flags = EV_ADD|EV_CLEAR,
- };
+ uint32_t qos = DISPATCH_TIMER_QOS(tidx);
+ dispatch_clock_t clock = DISPATCH_TIMER_CLOCK(tidx);
+ dispatch_timer_heap_t heap = &_dispatch_timers_heap[tidx];
+ dispatch_timer_delay_s range;
- _dispatch_safe_fork = false;
-#if DISPATCH_USE_GUARDED_FD
- guardid_t guard = (uintptr_t)&kev;
- _dispatch_kq = guarded_kqueue_np(&guard, GUARD_CLOSE | GUARD_DUP);
-#else
- _dispatch_kq = kqueue();
-#endif
- if (_dispatch_kq == -1) {
- int err = errno;
- switch (err) {
- case EMFILE:
- DISPATCH_CLIENT_CRASH("kqueue() failure: "
- "process is out of file descriptors");
- break;
- case ENFILE:
- DISPATCH_CLIENT_CRASH("kqueue() failure: "
- "system is out of file descriptors");
- break;
- case ENOMEM:
- DISPATCH_CLIENT_CRASH("kqueue() failure: "
- "kernel is out of memory");
- break;
- default:
- (void)dispatch_assume_zero(err);
- DISPATCH_CRASH("kqueue() failure");
- break;
+ range = _dispatch_timers_get_delay(heap, clock, qos, nows);
+ if (range.delay == 0 || range.delay >= INT64_MAX) {
+ _dispatch_trace_next_timer_set(NULL, qos);
+ if (heap->dth_flags & DTH_ARMED) {
+ _dispatch_event_loop_timer_delete(tidx);
}
- } else if (dispatch_assume(_dispatch_kq < FD_SETSIZE)) {
- // in case we fall back to select()
- FD_SET(_dispatch_kq, &_dispatch_rfds);
+ return range.delay == 0;
}
- (void)dispatch_assume_zero(kevent64(_dispatch_kq, &kev, 1, NULL, 0, 0,
- NULL));
- _dispatch_queue_push(_dispatch_mgr_q.do_targetq, &_dispatch_mgr_q, 0);
+ _dispatch_trace_next_timer_set(heap->dth_min[DTH_TARGET_ID], qos);
+ _dispatch_trace_next_timer_program(range.delay, qos);
+ _dispatch_event_loop_timer_arm(tidx, range, nows);
+ return false;
}
-static int
-_dispatch_get_kq(void)
+DISPATCH_NOINLINE
+static bool
+_dispatch_timers_program(dispatch_clock_now_cache_t nows)
{
- static dispatch_once_t pred;
-
- dispatch_once_f(&pred, NULL, _dispatch_kq_init);
+ bool poll = false;
+ uint32_t tidx, timerm = _dispatch_timers_processing_mask;
- return _dispatch_kq;
+ for (tidx = 0; tidx < DISPATCH_TIMER_COUNT; tidx++) {
+ if (timerm & (1 << tidx)) {
+ poll |= _dispatch_timers_program2(nows, tidx);
+ }
+ }
+ return poll;
}
DISPATCH_NOINLINE
-static long
-_dispatch_kq_update(const struct kevent64_s *kev)
+static bool
+_dispatch_timers_configure(void)
{
- int r;
- struct kevent64_s kev_copy;
+ // Find out if there is a new target/deadline on the timer lists
+ return _dispatch_timer_heap_has_new_min(_dispatch_timers_heap,
+ countof(_dispatch_timers_heap), _dispatch_timers_processing_mask);
+}
- if (slowpath(_dispatch_select_workaround) && (kev->flags & EV_DELETE)) {
- if (_dispatch_select_unregister(kev)) {
- return 0;
- }
+static inline bool
+_dispatch_mgr_timers(void)
+{
+ dispatch_clock_now_cache_s nows = { };
+ bool expired = _dispatch_timers_expired;
+ if (unlikely(expired)) {
+ _dispatch_timers_run(&nows);
}
- kev_copy = *kev;
- // This ensures we don't get a pending kevent back while registering
- // a new kevent
- kev_copy.flags |= EV_RECEIPT;
-retry:
- r = dispatch_assume(kevent64(_dispatch_get_kq(), &kev_copy, 1,
- &kev_copy, 1, 0, NULL));
- if (slowpath(r == -1)) {
- int err = errno;
- switch (err) {
- case EINTR:
- goto retry;
- case EBADF:
- DISPATCH_CLIENT_CRASH("Do not close random Unix descriptors");
- break;
- default:
- (void)dispatch_assume_zero(err);
- break;
+ _dispatch_mgr_trace_timers_wakes();
+ bool reconfigure = _dispatch_timers_reconfigure;
+ if (unlikely(reconfigure || expired)) {
+ if (reconfigure) {
+ reconfigure = _dispatch_timers_configure();
+ _dispatch_timers_reconfigure = false;
}
- return err;
- }
- switch (kev_copy.data) {
- case 0:
- return 0;
- case EBADF:
- case EPERM:
- case EINVAL:
- case ENOENT:
- if ((kev->flags & (EV_ADD|EV_ENABLE)) && !(kev->flags & EV_DELETE)) {
- if (_dispatch_select_register(&kev_copy)) {
- return 0;
- }
+ if (reconfigure || expired) {
+ expired = _dispatch_timers_expired = _dispatch_timers_program(&nows);
}
- // fall through
- default:
- kev_copy.flags |= kev->flags;
- _dispatch_kevent_drain(&kev_copy);
- break;
+ _dispatch_timers_processing_mask = 0;
}
- return (long)kev_copy.data;
+ return expired;
}
#pragma mark -
#pragma mark dispatch_mgr
-static struct kevent64_s *_dispatch_kevent_enable;
-
-static void inline
-_dispatch_mgr_kevent_reenable(struct kevent64_s *ke)
-{
- dispatch_assert(!_dispatch_kevent_enable || _dispatch_kevent_enable == ke);
- _dispatch_kevent_enable = ke;
-}
-
-unsigned long
-_dispatch_mgr_wakeup(dispatch_queue_t dq DISPATCH_UNUSED)
+void
+_dispatch_mgr_queue_push(dispatch_queue_t dq, dispatch_object_t dou,
+ DISPATCH_UNUSED dispatch_qos_t qos)
{
- if (_dispatch_queue_get_current() == &_dispatch_mgr_q) {
- return false;
+ uint64_t dq_state;
+ _dispatch_trace_continuation_push(dq, dou._do);
+ if (unlikely(_dispatch_queue_push_update_tail(dq, dou._do))) {
+ _dispatch_queue_push_update_head(dq, dou._do);
+ dq_state = os_atomic_or2o(dq, dq_state, DISPATCH_QUEUE_DIRTY, release);
+ if (!_dq_state_drain_locked_by_self(dq_state)) {
+ _dispatch_event_loop_poke(DISPATCH_WLH_MANAGER, 0, 0);
+ }
}
-
- static const struct kevent64_s kev = {
- .ident = 1,
- .filter = EVFILT_USER,
- .fflags = NOTE_TRIGGER,
- };
-
-#if DISPATCH_DEBUG && DISPATCH_MGR_QUEUE_DEBUG
- _dispatch_debug("waking up the dispatch manager queue: %p", dq);
-#endif
-
- _dispatch_kq_update(&kev);
-
- return false;
}
-DISPATCH_NOINLINE
-static void
-_dispatch_mgr_init(void)
+DISPATCH_NORETURN
+void
+_dispatch_mgr_queue_wakeup(DISPATCH_UNUSED dispatch_queue_t dq,
+ DISPATCH_UNUSED dispatch_qos_t qos,
+ DISPATCH_UNUSED dispatch_wakeup_flags_t flags)
{
- (void)dispatch_atomic_inc2o(&_dispatch_mgr_q, dq_running, relaxed);
- _dispatch_thread_setspecific(dispatch_queue_key, &_dispatch_mgr_q);
- _dispatch_queue_set_bound_thread(&_dispatch_mgr_q);
- _dispatch_mgr_priority_init();
- _dispatch_kevent_init();
- _dispatch_timers_init();
- _dispatch_mach_recv_msg_buf_init();
- _dispatch_memorystatus_init();
+ DISPATCH_INTERNAL_CRASH(0, "Don't try to wake up or override the manager");
}
+#if DISPATCH_USE_MGR_THREAD
DISPATCH_NOINLINE DISPATCH_NORETURN
static void
_dispatch_mgr_invoke(void)
{
- static const struct timespec timeout_immediately = { 0, 0 };
- struct kevent64_s kev;
+#if DISPATCH_EVENT_BACKEND_KEVENT
+ dispatch_kevent_s evbuf[DISPATCH_DEFERRED_ITEMS_EVENT_COUNT];
+#endif
+ dispatch_deferred_items_s ddi = {
+#if DISPATCH_EVENT_BACKEND_KEVENT
+ .ddi_maxevents = DISPATCH_DEFERRED_ITEMS_EVENT_COUNT,
+ .ddi_eventlist = evbuf,
+#endif
+ };
bool poll;
- int r;
+ _dispatch_deferred_items_set(&ddi);
for (;;) {
_dispatch_mgr_queue_drain();
poll = _dispatch_mgr_timers();
- if (slowpath(_dispatch_select_workaround)) {
- poll = _dispatch_mgr_select(poll);
- if (!poll) continue;
- }
poll = poll || _dispatch_queue_class_probe(&_dispatch_mgr_q);
- r = kevent64(_dispatch_kq, _dispatch_kevent_enable,
- _dispatch_kevent_enable ? 1 : 0, &kev, 1, 0,
- poll ? &timeout_immediately : NULL);
- _dispatch_kevent_enable = NULL;
- if (slowpath(r == -1)) {
- int err = errno;
- switch (err) {
- case EINTR:
- break;
- case EBADF:
- DISPATCH_CLIENT_CRASH("Do not close random Unix descriptors");
- break;
- default:
- (void)dispatch_assume_zero(err);
- break;
- }
- } else if (r) {
- _dispatch_kevent_drain(&kev);
- }
+ _dispatch_event_loop_drain(poll ? KEVENT_FLAG_IMMEDIATE : 0);
}
}
+#endif // DISPATCH_USE_MGR_THREAD
DISPATCH_NORETURN
void
-_dispatch_mgr_thread(dispatch_queue_t dq DISPATCH_UNUSED)
+_dispatch_mgr_thread(dispatch_queue_t dq DISPATCH_UNUSED,
+ dispatch_invoke_context_t dic DISPATCH_UNUSED,
+ dispatch_invoke_flags_t flags DISPATCH_UNUSED)
{
- _dispatch_mgr_init();
+#if DISPATCH_USE_KEVENT_WORKQUEUE
+ if (_dispatch_kevent_workqueue_enabled) {
+ DISPATCH_INTERNAL_CRASH(0, "Manager queue invoked with "
+ "kevent workqueue enabled");
+ }
+#endif
+#if DISPATCH_USE_MGR_THREAD
+ _dispatch_queue_set_current(&_dispatch_mgr_q);
+ _dispatch_mgr_priority_init();
+ _dispatch_queue_mgr_lock(&_dispatch_mgr_q);
// never returns, so burn bridges behind us & clear stack 2k ahead
_dispatch_clear_stack(2048);
_dispatch_mgr_invoke();
-}
-
-#pragma mark -
-#pragma mark dispatch_memorystatus
-
-#if DISPATCH_USE_MEMORYSTATUS_SOURCE
-#define DISPATCH_MEMORYSTATUS_SOURCE_TYPE DISPATCH_SOURCE_TYPE_MEMORYSTATUS
-#define DISPATCH_MEMORYSTATUS_SOURCE_MASK ( \
- DISPATCH_MEMORYSTATUS_PRESSURE_NORMAL | \
- DISPATCH_MEMORYSTATUS_PRESSURE_WARN)
-#elif DISPATCH_USE_VM_PRESSURE_SOURCE
-#define DISPATCH_MEMORYSTATUS_SOURCE_TYPE DISPATCH_SOURCE_TYPE_VM
-#define DISPATCH_MEMORYSTATUS_SOURCE_MASK DISPATCH_VM_PRESSURE
-#endif
-
-#if DISPATCH_USE_MEMORYSTATUS_SOURCE || DISPATCH_USE_VM_PRESSURE_SOURCE
-static dispatch_source_t _dispatch_memorystatus_source;
-
-static void
-_dispatch_memorystatus_handler(void *context DISPATCH_UNUSED)
-{
-#if DISPATCH_USE_MEMORYSTATUS_SOURCE
- unsigned long memorystatus;
- memorystatus = dispatch_source_get_data(_dispatch_memorystatus_source);
- if (memorystatus & DISPATCH_MEMORYSTATUS_PRESSURE_NORMAL) {
- _dispatch_continuation_cache_limit = DISPATCH_CONTINUATION_CACHE_LIMIT;
- _voucher_activity_heap_pressure_normal();
- return;
- }
- _dispatch_continuation_cache_limit =
- DISPATCH_CONTINUATION_CACHE_LIMIT_MEMORYSTATUS_PRESSURE_WARN;
- _voucher_activity_heap_pressure_warn();
#endif
- malloc_zone_pressure_relief(0,0);
-}
-
-static void
-_dispatch_memorystatus_init(void)
-{
- _dispatch_memorystatus_source = dispatch_source_create(
- DISPATCH_MEMORYSTATUS_SOURCE_TYPE, 0,
- DISPATCH_MEMORYSTATUS_SOURCE_MASK,
- _dispatch_get_root_queue(_DISPATCH_QOS_CLASS_DEFAULT, true));
- dispatch_source_set_event_handler_f(_dispatch_memorystatus_source,
- _dispatch_memorystatus_handler);
- dispatch_resume(_dispatch_memorystatus_source);
}
-#else
-static inline void _dispatch_memorystatus_init(void) {}
-#endif // DISPATCH_USE_MEMORYSTATUS_SOURCE || DISPATCH_USE_VM_PRESSURE_SOURCE
-
-#pragma mark -
-#pragma mark dispatch_mach
-#if HAVE_MACH
-
-#if DISPATCH_DEBUG && DISPATCH_MACHPORT_DEBUG
-#define _dispatch_debug_machport(name) \
- dispatch_debug_machport((name), __func__)
-#else
-#define _dispatch_debug_machport(name) ((void)(name))
-#endif
+#if DISPATCH_USE_KEVENT_WORKQUEUE
-// Flags for all notifications that are registered/unregistered when a
-// send-possible notification is requested/delivered
-#define _DISPATCH_MACH_SP_FLAGS (DISPATCH_MACH_SEND_POSSIBLE| \
- DISPATCH_MACH_SEND_DEAD|DISPATCH_MACH_SEND_DELETED)
-#define _DISPATCH_MACH_RECV_FLAGS (DISPATCH_MACH_RECV_MESSAGE| \
- DISPATCH_MACH_RECV_MESSAGE_DIRECT| \
- DISPATCH_MACH_RECV_MESSAGE_DIRECT_ONCE)
-#define _DISPATCH_MACH_RECV_DIRECT_FLAGS ( \
- DISPATCH_MACH_RECV_MESSAGE_DIRECT| \
- DISPATCH_MACH_RECV_MESSAGE_DIRECT_ONCE)
-
-#define _DISPATCH_IS_POWER_OF_TWO(v) (!(v & (v - 1)) && v)
-#define _DISPATCH_HASH(x, y) (_DISPATCH_IS_POWER_OF_TWO(y) ? \
- (MACH_PORT_INDEX(x) & ((y) - 1)) : (MACH_PORT_INDEX(x) % (y)))
-
-#define _DISPATCH_MACHPORT_HASH_SIZE 32
-#define _DISPATCH_MACHPORT_HASH(x) \
- _DISPATCH_HASH((x), _DISPATCH_MACHPORT_HASH_SIZE)
-
-#ifndef MACH_RCV_LARGE_IDENTITY
-#define MACH_RCV_LARGE_IDENTITY 0x00000008
-#endif
-#ifndef MACH_RCV_VOUCHER
-#define MACH_RCV_VOUCHER 0x00000800
-#endif
-#define DISPATCH_MACH_RCV_TRAILER MACH_RCV_TRAILER_CTX
-#define DISPATCH_MACH_RCV_OPTIONS ( \
- MACH_RCV_MSG | MACH_RCV_LARGE | MACH_RCV_LARGE_IDENTITY | \
- MACH_RCV_TRAILER_ELEMENTS(DISPATCH_MACH_RCV_TRAILER) | \
- MACH_RCV_TRAILER_TYPE(MACH_MSG_TRAILER_FORMAT_0)) | \
- MACH_RCV_VOUCHER
-
-#define DISPATCH_MACH_KEVENT_ARMED(dk) ((dk)->dk_kevent.ext[0])
-
-static void _dispatch_kevent_machport_drain(struct kevent64_s *ke);
-static void _dispatch_kevent_mach_msg_drain(struct kevent64_s *ke);
-static void _dispatch_kevent_mach_msg_recv(mach_msg_header_t *hdr);
-static void _dispatch_kevent_mach_msg_destroy(mach_msg_header_t *hdr);
-static void _dispatch_source_merge_mach_msg(dispatch_source_t ds,
- dispatch_source_refs_t dr, dispatch_kevent_t dk,
- mach_msg_header_t *hdr, mach_msg_size_t siz);
-static kern_return_t _dispatch_mach_notify_update(dispatch_kevent_t dk,
- uint32_t new_flags, uint32_t del_flags, uint32_t mask,
- mach_msg_id_t notify_msgid, mach_port_mscount_t notify_sync);
-static void _dispatch_mach_notify_source_invoke(mach_msg_header_t *hdr);
-static void _dispatch_mach_reply_kevent_unregister(dispatch_mach_t dm,
- dispatch_mach_reply_refs_t dmr, bool disconnected);
-static void _dispatch_mach_kevent_unregister(dispatch_mach_t dm);
-static inline void _dispatch_mach_msg_set_options(dispatch_object_t dou,
- mach_msg_option_t options);
-static void _dispatch_mach_msg_recv(dispatch_mach_t dm,
- dispatch_mach_reply_refs_t dmr, mach_msg_header_t *hdr,
- mach_msg_size_t siz);
-static void _dispatch_mach_merge_kevent(dispatch_mach_t dm,
- const struct kevent64_s *ke);
-static inline mach_msg_option_t _dispatch_mach_checkin_options(void);
-
-static const size_t _dispatch_mach_recv_msg_size =
- DISPATCH_MACH_RECEIVE_MAX_INLINE_MESSAGE_SIZE;
-static const size_t dispatch_mach_trailer_size =
- sizeof(dispatch_mach_trailer_t);
-static mach_msg_size_t _dispatch_mach_recv_msg_buf_size;
-static mach_port_t _dispatch_mach_portset, _dispatch_mach_recv_portset;
-static mach_port_t _dispatch_mach_notify_port;
-static struct kevent64_s _dispatch_mach_recv_kevent = {
- .filter = EVFILT_MACHPORT,
- .flags = EV_ADD|EV_ENABLE|EV_DISPATCH,
- .fflags = DISPATCH_MACH_RCV_OPTIONS,
-};
-static dispatch_source_t _dispatch_mach_notify_source;
-static const
-struct dispatch_source_type_s _dispatch_source_type_mach_recv_direct = {
- .ke = {
- .filter = EVFILT_MACHPORT,
- .flags = EV_CLEAR,
- .fflags = DISPATCH_MACH_RECV_MESSAGE_DIRECT,
- },
-};
+#define DISPATCH_KEVENT_WORKER_IS_NOT_MANAGER ((dispatch_priority_t)~0u)
-static void
-_dispatch_mach_recv_msg_buf_init(void)
-{
- mach_vm_size_t vm_size = mach_vm_round_page(
- _dispatch_mach_recv_msg_size + dispatch_mach_trailer_size);
- _dispatch_mach_recv_msg_buf_size = (mach_msg_size_t)vm_size;
- mach_vm_address_t vm_addr = vm_page_size;
- kern_return_t kr;
-
- while (slowpath(kr = mach_vm_allocate(mach_task_self(), &vm_addr, vm_size,
- VM_FLAGS_ANYWHERE))) {
- if (kr != KERN_NO_SPACE) {
- (void)dispatch_assume_zero(kr);
- DISPATCH_CLIENT_CRASH("Could not allocate mach msg receive buffer");
- }
- _dispatch_temporary_resource_shortage();
- vm_addr = vm_page_size;
- }
- _dispatch_mach_recv_kevent.ext[0] = (uintptr_t)vm_addr;
- _dispatch_mach_recv_kevent.ext[1] = vm_size;
-}
+_Static_assert(WORKQ_KEVENT_EVENT_BUFFER_LEN >=
+ DISPATCH_DEFERRED_ITEMS_EVENT_COUNT,
+ "our list should not be longer than the kernel's");
-static inline void*
-_dispatch_get_mach_recv_msg_buf(void)
+DISPATCH_ALWAYS_INLINE
+static inline dispatch_priority_t
+_dispatch_wlh_worker_thread_init(dispatch_wlh_t wlh,
+ dispatch_deferred_items_t ddi)
+{
+ dispatch_assert(wlh);
+ dispatch_priority_t old_dbp;
+
+ pthread_priority_t pp = _dispatch_get_priority();
+ if (!(pp & _PTHREAD_PRIORITY_EVENT_MANAGER_FLAG)) {
+ // If this thread does not have the event manager flag set, don't setup
+ // as the dispatch manager and let the caller know to only process
+ // the delivered events.
+ //
+ // Also add the NEEDS_UNBIND flag so that
+ // _dispatch_priority_compute_update knows it has to unbind
+ pp &= _PTHREAD_PRIORITY_OVERCOMMIT_FLAG | ~_PTHREAD_PRIORITY_FLAGS_MASK;
+ if (wlh == DISPATCH_WLH_ANON) {
+ pp |= _PTHREAD_PRIORITY_NEEDS_UNBIND_FLAG;
+ } else {
+ // pthread sets the flag when it is an event delivery thread
+ // so we need to explicitly clear it
+ pp &= ~(pthread_priority_t)_PTHREAD_PRIORITY_NEEDS_UNBIND_FLAG;
+ }
+ _dispatch_thread_setspecific(dispatch_priority_key,
+ (void *)(uintptr_t)pp);
+ if (wlh != DISPATCH_WLH_ANON) {
+ _dispatch_debug("wlh[%p]: handling events", wlh);
+ } else {
+ ddi->ddi_can_stash = true;
+ }
+ return DISPATCH_KEVENT_WORKER_IS_NOT_MANAGER;
+ }
+
+ if ((pp & _PTHREAD_PRIORITY_SCHED_PRI_FLAG) ||
+ !(pp & ~_PTHREAD_PRIORITY_FLAGS_MASK)) {
+ // When the phtread kext is delivering kevents to us, and pthread
+ // root queues are in use, then the pthread priority TSD is set
+ // to a sched pri with the _PTHREAD_PRIORITY_SCHED_PRI_FLAG bit set.
+ //
+ // Given that this isn't a valid QoS we need to fixup the TSD,
+ // and the best option is to clear the qos/priority bits which tells
+ // us to not do any QoS related calls on this thread.
+ //
+ // However, in that case the manager thread is opted out of QoS,
+ // as far as pthread is concerned, and can't be turned into
+ // something else, so we can't stash.
+ pp &= (pthread_priority_t)_PTHREAD_PRIORITY_FLAGS_MASK;
+ }
+ // Managers always park without mutating to a regular worker thread, and
+ // hence never need to unbind from userland, and when draining a manager,
+ // the NEEDS_UNBIND flag would cause the mutation to happen.
+ // So we need to strip this flag
+ pp &= ~(pthread_priority_t)_PTHREAD_PRIORITY_NEEDS_UNBIND_FLAG;
+ _dispatch_thread_setspecific(dispatch_priority_key, (void *)(uintptr_t)pp);
+
+ // ensure kevents registered from this thread are registered at manager QoS
+ old_dbp = _dispatch_set_basepri(DISPATCH_PRIORITY_FLAG_MANAGER);
+ _dispatch_queue_set_current(&_dispatch_mgr_q);
+ _dispatch_queue_mgr_lock(&_dispatch_mgr_q);
+ return old_dbp;
+}
+
+DISPATCH_ALWAYS_INLINE DISPATCH_WARN_RESULT
+static inline bool
+_dispatch_wlh_worker_thread_reset(dispatch_priority_t old_dbp)
{
- return (void*)_dispatch_mach_recv_kevent.ext[0];
+ bool needs_poll = _dispatch_queue_mgr_unlock(&_dispatch_mgr_q);
+ _dispatch_reset_basepri(old_dbp);
+ _dispatch_reset_basepri_override();
+ _dispatch_queue_set_current(NULL);
+ return needs_poll;
}
+DISPATCH_ALWAYS_INLINE
static void
-_dispatch_mach_recv_portset_init(void *context DISPATCH_UNUSED)
-{
- kern_return_t kr;
-
- kr = mach_port_allocate(mach_task_self(), MACH_PORT_RIGHT_PORT_SET,
- &_dispatch_mach_recv_portset);
- DISPATCH_VERIFY_MIG(kr);
- if (dispatch_assume_zero(kr)) {
- DISPATCH_CLIENT_CRASH(
- "mach_port_allocate() failed: cannot create port set");
- }
- dispatch_assert(_dispatch_get_mach_recv_msg_buf());
- dispatch_assert(dispatch_mach_trailer_size ==
- REQUESTED_TRAILER_SIZE_NATIVE(MACH_RCV_TRAILER_ELEMENTS(
- DISPATCH_MACH_RCV_TRAILER)));
- _dispatch_mach_recv_kevent.ident = _dispatch_mach_recv_portset;
- _dispatch_kq_update(&_dispatch_mach_recv_kevent);
-
- kr = mach_port_allocate(mach_task_self(), MACH_PORT_RIGHT_RECEIVE,
- &_dispatch_mach_notify_port);
- DISPATCH_VERIFY_MIG(kr);
- if (dispatch_assume_zero(kr)) {
- DISPATCH_CLIENT_CRASH(
- "mach_port_allocate() failed: cannot create receive right");
- }
- _dispatch_mach_notify_source = dispatch_source_create(
- &_dispatch_source_type_mach_recv_direct,
- _dispatch_mach_notify_port, 0, &_dispatch_mgr_q);
- static const struct dispatch_continuation_s dc = {
- .dc_func = (void*)_dispatch_mach_notify_source_invoke,
- };
- _dispatch_mach_notify_source->ds_refs->ds_handler[DS_EVENT_HANDLER] =
- (dispatch_continuation_t)&dc;
- dispatch_assert(_dispatch_mach_notify_source);
- dispatch_resume(_dispatch_mach_notify_source);
-}
-
-static mach_port_t
-_dispatch_get_mach_recv_portset(void)
+_dispatch_wlh_worker_thread(dispatch_wlh_t wlh, dispatch_kevent_t events,
+ int *nevents)
{
- static dispatch_once_t pred;
- dispatch_once_f(&pred, NULL, _dispatch_mach_recv_portset_init);
- return _dispatch_mach_recv_portset;
-}
+ _dispatch_introspection_thread_add();
+ DISPATCH_PERF_MON_VAR_INIT
-static void
-_dispatch_mach_portset_init(void *context DISPATCH_UNUSED)
-{
- struct kevent64_s kev = {
- .filter = EVFILT_MACHPORT,
- .flags = EV_ADD,
+ dispatch_deferred_items_s ddi = {
+ .ddi_eventlist = events,
};
- kern_return_t kr;
-
- kr = mach_port_allocate(mach_task_self(), MACH_PORT_RIGHT_PORT_SET,
- &_dispatch_mach_portset);
- DISPATCH_VERIFY_MIG(kr);
- if (dispatch_assume_zero(kr)) {
- DISPATCH_CLIENT_CRASH(
- "mach_port_allocate() failed: cannot create port set");
- }
- kev.ident = _dispatch_mach_portset;
- _dispatch_kq_update(&kev);
-}
-
-static mach_port_t
-_dispatch_get_mach_portset(void)
-{
- static dispatch_once_t pred;
- dispatch_once_f(&pred, NULL, _dispatch_mach_portset_init);
- return _dispatch_mach_portset;
-}
-
-static kern_return_t
-_dispatch_mach_portset_update(dispatch_kevent_t dk, mach_port_t mps)
-{
- mach_port_t mp = (mach_port_t)dk->dk_kevent.ident;
- kern_return_t kr;
-
- _dispatch_debug_machport(mp);
- kr = mach_port_move_member(mach_task_self(), mp, mps);
- if (slowpath(kr)) {
- DISPATCH_VERIFY_MIG(kr);
- switch (kr) {
- case KERN_INVALID_RIGHT:
- if (mps) {
- _dispatch_bug_mach_client("_dispatch_kevent_machport_enable: "
- "mach_port_move_member() failed ", kr);
- break;
- }
- //fall through
- case KERN_INVALID_NAME:
-#if DISPATCH_DEBUG
- _dispatch_log("Corruption: Mach receive right 0x%x destroyed "
- "prematurely", mp);
-#endif
- break;
- default:
- (void)dispatch_assume_zero(kr);
- break;
- }
- }
- return mps ? kr : 0;
-}
+ dispatch_priority_t old_dbp;
-static void
-_dispatch_kevent_mach_recv_reenable(struct kevent64_s *ke DISPATCH_UNUSED)
-{
-#if (TARGET_IPHONE_SIMULATOR && \
- IPHONE_SIMULATOR_HOST_MIN_VERSION_REQUIRED < 1090) || \
- (!TARGET_OS_IPHONE && __MAC_OS_X_VERSION_MIN_REQUIRED < 1090)
- // delete and re-add kevent to workaround <rdar://problem/13924256>
- if (ke->ext[1] != _dispatch_mach_recv_kevent.ext[1]) {
- struct kevent64_s kev = _dispatch_mach_recv_kevent;
- kev.flags = EV_DELETE;
- _dispatch_kq_update(&kev);
+ old_dbp = _dispatch_wlh_worker_thread_init(wlh, &ddi);
+ if (old_dbp == DISPATCH_KEVENT_WORKER_IS_NOT_MANAGER) {
+ _dispatch_perfmon_start_impl(true);
+ } else {
+ dispatch_assert(wlh == DISPATCH_WLH_ANON);
+ wlh = DISPATCH_WLH_ANON;
}
-#endif
- _dispatch_mgr_kevent_reenable(&_dispatch_mach_recv_kevent);
-}
+ _dispatch_deferred_items_set(&ddi);
+ _dispatch_event_loop_merge(events, *nevents);
-static kern_return_t
-_dispatch_kevent_machport_resume(dispatch_kevent_t dk, uint32_t new_flags,
- uint32_t del_flags)
-{
- kern_return_t kr = 0;
- dispatch_assert_zero(new_flags & del_flags);
- if ((new_flags & _DISPATCH_MACH_RECV_FLAGS) ||
- (del_flags & _DISPATCH_MACH_RECV_FLAGS)) {
- mach_port_t mps;
- if (new_flags & _DISPATCH_MACH_RECV_DIRECT_FLAGS) {
- mps = _dispatch_get_mach_recv_portset();
- } else if ((new_flags & DISPATCH_MACH_RECV_MESSAGE) ||
- ((del_flags & _DISPATCH_MACH_RECV_DIRECT_FLAGS) &&
- (dk->dk_kevent.fflags & DISPATCH_MACH_RECV_MESSAGE))) {
- mps = _dispatch_get_mach_portset();
+ if (old_dbp != DISPATCH_KEVENT_WORKER_IS_NOT_MANAGER) {
+ _dispatch_mgr_queue_drain();
+ bool poll = _dispatch_mgr_timers();
+ if (_dispatch_wlh_worker_thread_reset(old_dbp)) {
+ poll = true;
+ }
+ if (poll) _dispatch_event_loop_poke(DISPATCH_WLH_MANAGER, 0, 0);
+ } else if (ddi.ddi_stashed_dou._do) {
+ _dispatch_debug("wlh[%p]: draining deferred item %p", wlh,
+ ddi.ddi_stashed_dou._do);
+ if (wlh == DISPATCH_WLH_ANON) {
+ dispatch_assert(ddi.ddi_nevents == 0);
+ _dispatch_deferred_items_set(NULL);
+ _dispatch_root_queue_drain_deferred_item(&ddi
+ DISPATCH_PERF_MON_ARGS);
} else {
- mps = MACH_PORT_NULL;
+ _dispatch_root_queue_drain_deferred_wlh(&ddi
+ DISPATCH_PERF_MON_ARGS);
}
- kr = _dispatch_mach_portset_update(dk, mps);
}
- return kr;
-}
-
-static kern_return_t
-_dispatch_kevent_mach_notify_resume(dispatch_kevent_t dk, uint32_t new_flags,
- uint32_t del_flags)
-{
- kern_return_t kr = 0;
- dispatch_assert_zero(new_flags & del_flags);
- if ((new_flags & _DISPATCH_MACH_SP_FLAGS) ||
- (del_flags & _DISPATCH_MACH_SP_FLAGS)) {
- // Requesting a (delayed) non-sync send-possible notification
- // registers for both immediate dead-name notification and delayed-arm
- // send-possible notification for the port.
- // The send-possible notification is armed when a mach_msg() with the
- // the MACH_SEND_NOTIFY to the port times out.
- // If send-possible is unavailable, fall back to immediate dead-name
- // registration rdar://problem/2527840&9008724
- kr = _dispatch_mach_notify_update(dk, new_flags, del_flags,
- _DISPATCH_MACH_SP_FLAGS, MACH_NOTIFY_SEND_POSSIBLE,
- MACH_NOTIFY_SEND_POSSIBLE == MACH_NOTIFY_DEAD_NAME ? 1 : 0);
- }
- return kr;
-}
-static inline void
-_dispatch_kevent_mach_portset(struct kevent64_s *ke)
-{
- if (ke->ident == _dispatch_mach_recv_portset) {
- return _dispatch_kevent_mach_msg_drain(ke);
- } else if (ke->ident == _dispatch_mach_portset) {
- return _dispatch_kevent_machport_drain(ke);
- } else {
- return _dispatch_kevent_error(ke);
+ _dispatch_deferred_items_set(NULL);
+ if (old_dbp == DISPATCH_KEVENT_WORKER_IS_NOT_MANAGER &&
+ !ddi.ddi_stashed_dou._do) {
+ _dispatch_perfmon_end(perfmon_thread_event_no_steal);
}
+ _dispatch_debug("returning %d deferred kevents", ddi.ddi_nevents);
+ *nevents = ddi.ddi_nevents;
}
DISPATCH_NOINLINE
-static void
-_dispatch_kevent_machport_drain(struct kevent64_s *ke)
+void
+_dispatch_kevent_worker_thread(dispatch_kevent_t *events, int *nevents)
{
- mach_port_t name = (mach_port_name_t)ke->data;
- dispatch_kevent_t dk;
- struct kevent64_s kev;
-
- _dispatch_debug_machport(name);
- dk = _dispatch_kevent_find(name, EVFILT_MACHPORT);
- if (!dispatch_assume(dk)) {
+ if (!events && !nevents) {
+ // events for worker thread request have already been delivered earlier
return;
}
- _dispatch_mach_portset_update(dk, MACH_PORT_NULL); // emulate EV_DISPATCH
-
- EV_SET64(&kev, name, EVFILT_MACHPORT, EV_ADD|EV_ENABLE|EV_DISPATCH,
- DISPATCH_MACH_RECV_MESSAGE, 0, (uintptr_t)dk, 0, 0);
- _dispatch_kevent_debug(&kev, __func__);
- _dispatch_kevent_merge(&kev);
-}
-
-DISPATCH_NOINLINE
-static void
-_dispatch_kevent_mach_msg_drain(struct kevent64_s *ke)
-{
- mach_msg_header_t *hdr = (mach_msg_header_t*)ke->ext[0];
- mach_msg_size_t siz, msgsiz;
- mach_msg_return_t kr = (mach_msg_return_t)ke->fflags;
-
- _dispatch_kevent_mach_recv_reenable(ke);
- if (!dispatch_assume(hdr)) {
- DISPATCH_CRASH("EVFILT_MACHPORT with no message");
- }
- if (fastpath(!kr)) {
- return _dispatch_kevent_mach_msg_recv(hdr);
- } else if (kr != MACH_RCV_TOO_LARGE) {
- goto out;
- }
- if (!dispatch_assume(ke->ext[1] <= UINT_MAX -
- dispatch_mach_trailer_size)) {
- DISPATCH_CRASH("EVFILT_MACHPORT with overlarge message");
- }
- siz = (mach_msg_size_t)ke->ext[1] + dispatch_mach_trailer_size;
- hdr = malloc(siz);
- if (ke->data) {
- if (!dispatch_assume(hdr)) {
- // Kernel will discard message too large to fit
- hdr = _dispatch_get_mach_recv_msg_buf();
- siz = _dispatch_mach_recv_msg_buf_size;
- }
- mach_port_t name = (mach_port_name_t)ke->data;
- const mach_msg_option_t options = ((DISPATCH_MACH_RCV_OPTIONS |
- MACH_RCV_TIMEOUT) & ~MACH_RCV_LARGE);
- kr = mach_msg(hdr, options, 0, siz, name, MACH_MSG_TIMEOUT_NONE,
- MACH_PORT_NULL);
- if (fastpath(!kr)) {
- return _dispatch_kevent_mach_msg_recv(hdr);
- } else if (kr == MACH_RCV_TOO_LARGE) {
- _dispatch_log("BUG in libdispatch client: "
- "_dispatch_kevent_mach_msg_drain: dropped message too "
- "large to fit in memory: id = 0x%x, size = %lld",
- hdr->msgh_id, ke->ext[1]);
- kr = MACH_MSG_SUCCESS;
- }
- } else {
- // We don't know which port in the portset contains the large message,
- // so need to receive all messages pending on the portset to ensure the
- // large message is drained. <rdar://problem/13950432>
- bool received = false;
- for (;;) {
- if (!dispatch_assume(hdr)) {
- DISPATCH_CLIENT_CRASH("Message too large to fit in memory");
- }
- const mach_msg_option_t options = (DISPATCH_MACH_RCV_OPTIONS |
- MACH_RCV_TIMEOUT);
- kr = mach_msg(hdr, options, 0, siz, _dispatch_mach_recv_portset,
- MACH_MSG_TIMEOUT_NONE, MACH_PORT_NULL);
- if ((!kr || kr == MACH_RCV_TOO_LARGE) && !dispatch_assume(
- hdr->msgh_size <= UINT_MAX - dispatch_mach_trailer_size)) {
- DISPATCH_CRASH("Overlarge message");
- }
- if (fastpath(!kr)) {
- msgsiz = hdr->msgh_size + dispatch_mach_trailer_size;
- if (msgsiz < siz) {
- void *shrink = realloc(hdr, msgsiz);
- if (shrink) hdr = shrink;
- }
- _dispatch_kevent_mach_msg_recv(hdr);
- hdr = NULL;
- received = true;
- } else if (kr == MACH_RCV_TOO_LARGE) {
- siz = hdr->msgh_size + dispatch_mach_trailer_size;
- } else {
- if (kr == MACH_RCV_TIMED_OUT && received) {
- kr = MACH_MSG_SUCCESS;
- }
- break;
- }
- hdr = reallocf(hdr, siz);
- }
- }
- if (hdr != _dispatch_get_mach_recv_msg_buf()) {
- free(hdr);
- }
-out:
- if (slowpath(kr)) {
- _dispatch_bug_mach_client("_dispatch_kevent_mach_msg_drain: "
- "message reception failed", kr);
- }
+ if (!dispatch_assume(*nevents && *events)) return;
+ _dispatch_adopt_wlh_anon();
+ _dispatch_wlh_worker_thread(DISPATCH_WLH_ANON, *events, nevents);
+ _dispatch_reset_wlh();
}
-static void
-_dispatch_kevent_mach_msg_recv(mach_msg_header_t *hdr)
-{
- dispatch_source_refs_t dri;
- dispatch_kevent_t dk;
- mach_port_t name = hdr->msgh_local_port;
- mach_msg_size_t siz = hdr->msgh_size + dispatch_mach_trailer_size;
-
- if (!dispatch_assume(hdr->msgh_size <= UINT_MAX -
- dispatch_mach_trailer_size)) {
- _dispatch_bug_client("_dispatch_kevent_mach_msg_recv: "
- "received overlarge message");
- return _dispatch_kevent_mach_msg_destroy(hdr);
- }
- if (!dispatch_assume(name)) {
- _dispatch_bug_client("_dispatch_kevent_mach_msg_recv: "
- "received message with MACH_PORT_NULL port");
- return _dispatch_kevent_mach_msg_destroy(hdr);
- }
- _dispatch_debug_machport(name);
- dk = _dispatch_kevent_find(name, EVFILT_MACHPORT);
- if (!dispatch_assume(dk)) {
- _dispatch_bug_client("_dispatch_kevent_mach_msg_recv: "
- "received message with unknown kevent");
- return _dispatch_kevent_mach_msg_destroy(hdr);
- }
- _dispatch_kevent_debug(&dk->dk_kevent, __func__);
- TAILQ_FOREACH(dri, &dk->dk_sources, dr_list) {
- dispatch_source_t dsi = _dispatch_source_from_refs(dri);
- if (dsi->ds_pending_data_mask & _DISPATCH_MACH_RECV_DIRECT_FLAGS) {
- return _dispatch_source_merge_mach_msg(dsi, dri, dk, hdr, siz);
- }
- }
- _dispatch_bug_client("_dispatch_kevent_mach_msg_recv: "
- "received message with no listeners");
- return _dispatch_kevent_mach_msg_destroy(hdr);
-}
-static void
-_dispatch_kevent_mach_msg_destroy(mach_msg_header_t *hdr)
-{
- if (hdr) {
- mach_msg_destroy(hdr);
- if (hdr != _dispatch_get_mach_recv_msg_buf()) {
- free(hdr);
- }
- }
-}
+#endif // DISPATCH_USE_KEVENT_WORKQUEUE
+#pragma mark -
+#pragma mark dispatch_source_debug
-static void
-_dispatch_source_merge_mach_msg(dispatch_source_t ds, dispatch_source_refs_t dr,
- dispatch_kevent_t dk, mach_msg_header_t *hdr, mach_msg_size_t siz)
+static size_t
+_dispatch_source_debug_attr(dispatch_source_t ds, char* buf, size_t bufsiz)
{
- if (ds == _dispatch_mach_notify_source) {
- _dispatch_mach_notify_source_invoke(hdr);
- return _dispatch_kevent_mach_msg_destroy(hdr);
- }
- dispatch_mach_reply_refs_t dmr = NULL;
- if (dk->dk_kevent.fflags & DISPATCH_MACH_RECV_MESSAGE_DIRECT_ONCE) {
- dmr = (dispatch_mach_reply_refs_t)dr;
- }
- return _dispatch_mach_msg_recv((dispatch_mach_t)ds, dmr, hdr, siz);
+ dispatch_queue_t target = ds->do_targetq;
+ dispatch_source_refs_t dr = ds->ds_refs;
+ return dsnprintf(buf, bufsiz, "target = %s[%p], ident = 0x%x, "
+ "mask = 0x%x, pending_data = 0x%llx, registered = %d, "
+ "armed = %d, deleted = %d%s, canceled = %d, ",
+ target && target->dq_label ? target->dq_label : "", target,
+ dr->du_ident, dr->du_fflags, (unsigned long long)ds->ds_pending_data,
+ ds->ds_is_installed, (bool)(ds->dq_atomic_flags & DSF_ARMED),
+ (bool)(ds->dq_atomic_flags & DSF_DELETED),
+ (ds->dq_atomic_flags & DSF_DEFERRED_DELETE) ? " (pending)" : "",
+ (bool)(ds->dq_atomic_flags & DSF_CANCELED));
}
-DISPATCH_ALWAYS_INLINE
-static inline void
-_dispatch_mach_notify_merge(mach_port_t name, uint32_t flag, bool final)
+static size_t
+_dispatch_timer_debug_attr(dispatch_source_t ds, char* buf, size_t bufsiz)
{
- dispatch_source_refs_t dri, dr_next;
- dispatch_kevent_t dk;
- struct kevent64_s kev;
- bool unreg;
-
- dk = _dispatch_kevent_find(name, DISPATCH_EVFILT_MACH_NOTIFICATION);
- if (!dk) {
- return;
- }
-
- // Update notification registration state.
- dk->dk_kevent.data &= ~_DISPATCH_MACH_SP_FLAGS;
- EV_SET64(&kev, name, DISPATCH_EVFILT_MACH_NOTIFICATION, EV_ADD|EV_ENABLE,
- flag, 0, (uintptr_t)dk, 0, 0);
- if (final) {
- // This can never happen again
- unreg = true;
- } else {
- // Re-register for notification before delivery
- unreg = _dispatch_kevent_resume(dk, flag, 0);
- }
- DISPATCH_MACH_KEVENT_ARMED(dk) = 0;
- TAILQ_FOREACH_SAFE(dri, &dk->dk_sources, dr_list, dr_next) {
- dispatch_source_t dsi = _dispatch_source_from_refs(dri);
- if (dx_type(dsi) == DISPATCH_MACH_CHANNEL_TYPE) {
- dispatch_mach_t dm = (dispatch_mach_t)dsi;
- _dispatch_mach_merge_kevent(dm, &kev);
- if (unreg && dm->dm_dkev) {
- _dispatch_mach_kevent_unregister(dm);
- }
- } else {
- _dispatch_source_merge_kevent(dsi, &kev);
- if (unreg) {
- _dispatch_source_kevent_unregister(dsi);
- }
- }
- if (!dr_next || DISPATCH_MACH_KEVENT_ARMED(dk)) {
- // current merge is last in list (dk might have been freed)
- // or it re-armed the notification
- return;
- }
- }
+ dispatch_timer_source_refs_t dr = ds->ds_timer_refs;
+ return dsnprintf(buf, bufsiz, "timer = { target = 0x%llx, deadline = 0x%llx"
+ ", interval = 0x%llx, flags = 0x%x }, ",
+ (unsigned long long)dr->dt_timer.target,
+ (unsigned long long)dr->dt_timer.deadline,
+ (unsigned long long)dr->dt_timer.interval, dr->du_fflags);
}
-static kern_return_t
-_dispatch_mach_notify_update(dispatch_kevent_t dk, uint32_t new_flags,
- uint32_t del_flags, uint32_t mask, mach_msg_id_t notify_msgid,
- mach_port_mscount_t notify_sync)
+size_t
+_dispatch_source_debug(dispatch_source_t ds, char *buf, size_t bufsiz)
{
- mach_port_t previous, port = (mach_port_t)dk->dk_kevent.ident;
- typeof(dk->dk_kevent.data) prev = dk->dk_kevent.data;
- kern_return_t kr, krr = 0;
-
- // Update notification registration state.
- dk->dk_kevent.data |= (new_flags | dk->dk_kevent.fflags) & mask;
- dk->dk_kevent.data &= ~(del_flags & mask);
-
- _dispatch_debug_machport(port);
- if ((dk->dk_kevent.data & mask) && !(prev & mask)) {
- // initialize _dispatch_mach_notify_port:
- (void)_dispatch_get_mach_recv_portset();
- _dispatch_debug("machport[0x%08x]: registering for send-possible "
- "notification", port);
- previous = MACH_PORT_NULL;
- krr = mach_port_request_notification(mach_task_self(), port,
- notify_msgid, notify_sync, _dispatch_mach_notify_port,
- MACH_MSG_TYPE_MAKE_SEND_ONCE, &previous);
- DISPATCH_VERIFY_MIG(krr);
-
- switch(krr) {
- case KERN_INVALID_NAME:
- case KERN_INVALID_RIGHT:
- // Supress errors & clear registration state
- dk->dk_kevent.data &= ~mask;
- break;
- default:
- // Else, we dont expect any errors from mach. Log any errors
- if (dispatch_assume_zero(krr)) {
- // log the error & clear registration state
- dk->dk_kevent.data &= ~mask;
- } else if (dispatch_assume_zero(previous)) {
- // Another subsystem has beat libdispatch to requesting the
- // specified Mach notification on this port. We should
- // technically cache the previous port and message it when the
- // kernel messages our port. Or we can just say screw those
- // subsystems and deallocate the previous port.
- // They should adopt libdispatch :-P
- kr = mach_port_deallocate(mach_task_self(), previous);
- DISPATCH_VERIFY_MIG(kr);
- (void)dispatch_assume_zero(kr);
- previous = MACH_PORT_NULL;
- }
- }
- } else if (!(dk->dk_kevent.data & mask) && (prev & mask)) {
- _dispatch_debug("machport[0x%08x]: unregistering for send-possible "
- "notification", port);
- previous = MACH_PORT_NULL;
- kr = mach_port_request_notification(mach_task_self(), port,
- notify_msgid, notify_sync, MACH_PORT_NULL,
- MACH_MSG_TYPE_MOVE_SEND_ONCE, &previous);
- DISPATCH_VERIFY_MIG(kr);
-
- switch (kr) {
- case KERN_INVALID_NAME:
- case KERN_INVALID_RIGHT:
- case KERN_INVALID_ARGUMENT:
- break;
- default:
- if (dispatch_assume_zero(kr)) {
- // log the error
- }
- }
- } else {
- return 0;
- }
- if (slowpath(previous)) {
- // the kernel has not consumed the send-once right yet
- (void)dispatch_assume_zero(
- _dispatch_send_consume_send_once_right(previous));
+ dispatch_source_refs_t dr = ds->ds_refs;
+ size_t offset = 0;
+ offset += dsnprintf(&buf[offset], bufsiz - offset, "%s[%p] = { ",
+ dx_kind(ds), ds);
+ offset += _dispatch_object_debug_attr(ds, &buf[offset], bufsiz - offset);
+ offset += _dispatch_source_debug_attr(ds, &buf[offset], bufsiz - offset);
+ if (dr->du_is_timer) {
+ offset += _dispatch_timer_debug_attr(ds, &buf[offset], bufsiz - offset);
}
- return krr;
-}
-
-static void
-_dispatch_mach_host_notify_update(void *context DISPATCH_UNUSED)
-{
- (void)_dispatch_get_mach_recv_portset();
- _dispatch_debug("registering for calendar-change notification");
- kern_return_t kr = host_request_notification(_dispatch_get_mach_host_port(),
- HOST_NOTIFY_CALENDAR_CHANGE, _dispatch_mach_notify_port);
- DISPATCH_VERIFY_MIG(kr);
- (void)dispatch_assume_zero(kr);
+ offset += dsnprintf(&buf[offset], bufsiz - offset, "kevent = %p%s, "
+ "filter = %s }", dr, dr->du_is_direct ? " (direct)" : "",
+ dr->du_type->dst_kind);
+ return offset;
}
-
-static void
-_dispatch_mach_host_calendar_change_register(void)
-{
- static dispatch_once_t pred;
- dispatch_once_f(&pred, NULL, _dispatch_mach_host_notify_update);
-}
-
-static void
-_dispatch_mach_notify_source_invoke(mach_msg_header_t *hdr)
-{
- mig_reply_error_t reply;
- dispatch_assert(sizeof(mig_reply_error_t) == sizeof(union
- __ReplyUnion___dispatch_libdispatch_internal_protocol_subsystem));
- dispatch_assert(sizeof(mig_reply_error_t) < _dispatch_mach_recv_msg_size);
- boolean_t success = libdispatch_internal_protocol_server(hdr, &reply.Head);
- if (!success && reply.RetCode == MIG_BAD_ID && hdr->msgh_id == 950) {
- // host_notify_reply.defs: host_calendar_changed
- _dispatch_debug("calendar-change notification");
- _dispatch_timers_calendar_change();
- _dispatch_mach_host_notify_update(NULL);
- success = TRUE;
- reply.RetCode = KERN_SUCCESS;
- }
- if (dispatch_assume(success) && reply.RetCode != MIG_NO_REPLY) {
- (void)dispatch_assume_zero(reply.RetCode);
- }
-}
-
-kern_return_t
-_dispatch_mach_notify_port_deleted(mach_port_t notify DISPATCH_UNUSED,
- mach_port_name_t name)
-{
-#if DISPATCH_DEBUG
- _dispatch_log("Corruption: Mach send/send-once/dead-name right 0x%x "
- "deleted prematurely", name);
-#endif
-
- _dispatch_debug_machport(name);
- _dispatch_mach_notify_merge(name, DISPATCH_MACH_SEND_DELETED, true);
-
- return KERN_SUCCESS;
-}
-
-kern_return_t
-_dispatch_mach_notify_dead_name(mach_port_t notify DISPATCH_UNUSED,
- mach_port_name_t name)
-{
- kern_return_t kr;
-
- _dispatch_debug("machport[0x%08x]: dead-name notification", name);
- _dispatch_debug_machport(name);
- _dispatch_mach_notify_merge(name, DISPATCH_MACH_SEND_DEAD, true);
-
- // the act of receiving a dead name notification allocates a dead-name
- // right that must be deallocated
- kr = mach_port_deallocate(mach_task_self(), name);
- DISPATCH_VERIFY_MIG(kr);
- //(void)dispatch_assume_zero(kr);
-
- return KERN_SUCCESS;
-}
-
-kern_return_t
-_dispatch_mach_notify_send_possible(mach_port_t notify DISPATCH_UNUSED,
- mach_port_name_t name)
-{
- _dispatch_debug("machport[0x%08x]: send-possible notification", name);
- _dispatch_debug_machport(name);
- _dispatch_mach_notify_merge(name, DISPATCH_MACH_SEND_POSSIBLE, false);
-
- return KERN_SUCCESS;
-}
-
-#pragma mark -
-#pragma mark dispatch_mach_t
-
-#define DISPATCH_MACH_NEVER_CONNECTED (UINT32_MAX/2)
-#define DISPATCH_MACH_REGISTER_FOR_REPLY 0x2
-#define DISPATCH_MACH_OPTIONS_MASK 0xffff
-
-static mach_port_t _dispatch_mach_msg_get_remote_port(dispatch_object_t dou);
-static void _dispatch_mach_msg_disconnected(dispatch_mach_t dm,
- mach_port_t local_port, mach_port_t remote_port);
-static dispatch_mach_msg_t _dispatch_mach_msg_create_reply_disconnected(
- dispatch_object_t dou, dispatch_mach_reply_refs_t dmr);
-static bool _dispatch_mach_reconnect_invoke(dispatch_mach_t dm,
- dispatch_object_t dou);
-static inline mach_msg_header_t* _dispatch_mach_msg_get_msg(
- dispatch_mach_msg_t dmsg);
-static void _dispatch_mach_push(dispatch_object_t dm, dispatch_object_t dou,
- pthread_priority_t pp);
-
-static dispatch_mach_t
-_dispatch_mach_create(const char *label, dispatch_queue_t q, void *context,
- dispatch_mach_handler_function_t handler, bool handler_is_block)
-{
- dispatch_mach_t dm;
- dispatch_mach_refs_t dr;
-
- dm = _dispatch_alloc(DISPATCH_VTABLE(mach),
- sizeof(struct dispatch_mach_s));
- _dispatch_queue_init((dispatch_queue_t)dm);
- dm->dq_label = label;
-
- dm->do_ref_cnt++; // the reference _dispatch_mach_cancel_invoke holds
- dm->do_ref_cnt++; // since channel is created suspended
- dm->do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_INTERVAL;
- dm->do_targetq = &_dispatch_mgr_q;
-
- dr = _dispatch_calloc(1ul, sizeof(struct dispatch_mach_refs_s));
- dr->dr_source_wref = _dispatch_ptr2wref(dm);
- dr->dm_handler_func = handler;
- dr->dm_handler_ctxt = context;
- dm->ds_refs = dr;
- dm->dm_handler_is_block = handler_is_block;
-
- dm->dm_refs = _dispatch_calloc(1ul,
- sizeof(struct dispatch_mach_send_refs_s));
- dm->dm_refs->dr_source_wref = _dispatch_ptr2wref(dm);
- dm->dm_refs->dm_disconnect_cnt = DISPATCH_MACH_NEVER_CONNECTED;
- TAILQ_INIT(&dm->dm_refs->dm_replies);
-
- // First item on the channel sets the user-specified target queue
- dispatch_set_target_queue(dm, q);
- _dispatch_object_debug(dm, "%s", __func__);
- return dm;
-}
-
-dispatch_mach_t
-dispatch_mach_create(const char *label, dispatch_queue_t q,
- dispatch_mach_handler_t handler)
-{
- dispatch_block_t bb = _dispatch_Block_copy((void*)handler);
- return _dispatch_mach_create(label, q, bb,
- (dispatch_mach_handler_function_t)_dispatch_Block_invoke(bb), true);
-}
-
-dispatch_mach_t
-dispatch_mach_create_f(const char *label, dispatch_queue_t q, void *context,
- dispatch_mach_handler_function_t handler)
-{
- return _dispatch_mach_create(label, q, context, handler, false);
-}
-
-void
-_dispatch_mach_dispose(dispatch_mach_t dm)
-{
- _dispatch_object_debug(dm, "%s", __func__);
- dispatch_mach_refs_t dr = dm->ds_refs;
- if (dm->dm_handler_is_block && dr->dm_handler_ctxt) {
- Block_release(dr->dm_handler_ctxt);
- }
- free(dr);
- free(dm->dm_refs);
- _dispatch_queue_destroy(dm);
-}
-
-void
-dispatch_mach_connect(dispatch_mach_t dm, mach_port_t receive,
- mach_port_t send, dispatch_mach_msg_t checkin)
-{
- dispatch_mach_send_refs_t dr = dm->dm_refs;
- dispatch_kevent_t dk;
-
- if (MACH_PORT_VALID(receive)) {
- dk = _dispatch_calloc(1ul, sizeof(struct dispatch_kevent_s));
- dk->dk_kevent = _dispatch_source_type_mach_recv_direct.ke;
- dk->dk_kevent.ident = receive;
- dk->dk_kevent.flags |= EV_ADD|EV_ENABLE;
- dk->dk_kevent.udata = (uintptr_t)dk;
- TAILQ_INIT(&dk->dk_sources);
- dm->ds_dkev = dk;
- dm->ds_pending_data_mask = dk->dk_kevent.fflags;
- _dispatch_retain(dm); // the reference the manager queue holds
- }
- dr->dm_send = send;
- if (MACH_PORT_VALID(send)) {
- if (checkin) {
- dispatch_retain(checkin);
- mach_msg_option_t options = _dispatch_mach_checkin_options();
- _dispatch_mach_msg_set_options(checkin, options);
- dr->dm_checkin_port = _dispatch_mach_msg_get_remote_port(checkin);
- }
- dr->dm_checkin = checkin;
- }
- // monitor message reply ports
- dm->ds_pending_data_mask |= DISPATCH_MACH_RECV_MESSAGE_DIRECT_ONCE;
- if (slowpath(!dispatch_atomic_cmpxchg2o(dr, dm_disconnect_cnt,
- DISPATCH_MACH_NEVER_CONNECTED, 0, release))) {
- DISPATCH_CLIENT_CRASH("Channel already connected");
- }
- _dispatch_object_debug(dm, "%s", __func__);
- return dispatch_resume(dm);
-}
-
-DISPATCH_NOINLINE
-static void
-_dispatch_mach_reply_kevent_unregister(dispatch_mach_t dm,
- dispatch_mach_reply_refs_t dmr, bool disconnected)
-{
- dispatch_mach_msg_t dmsgr = NULL;
- if (disconnected) {
- dmsgr = _dispatch_mach_msg_create_reply_disconnected(NULL, dmr);
- }
- dispatch_kevent_t dk = dmr->dmr_dkev;
- TAILQ_REMOVE(&dk->dk_sources, (dispatch_source_refs_t)dmr, dr_list);
- _dispatch_kevent_unregister(dk, DISPATCH_MACH_RECV_MESSAGE_DIRECT_ONCE);
- TAILQ_REMOVE(&dm->dm_refs->dm_replies, dmr, dmr_list);
- if (dmr->dmr_voucher) _voucher_release(dmr->dmr_voucher);
- free(dmr);
- if (dmsgr) _dispatch_mach_push(dm, dmsgr, dmsgr->dmsg_priority);
-}
-
-DISPATCH_NOINLINE
-static void
-_dispatch_mach_reply_kevent_register(dispatch_mach_t dm, mach_port_t reply,
- dispatch_mach_msg_t dmsg)
-{
- dispatch_kevent_t dk;
- dispatch_mach_reply_refs_t dmr;
-
- dk = _dispatch_calloc(1ul, sizeof(struct dispatch_kevent_s));
- dk->dk_kevent = _dispatch_source_type_mach_recv_direct.ke;
- dk->dk_kevent.ident = reply;
- dk->dk_kevent.flags |= EV_ADD|EV_ENABLE;
- dk->dk_kevent.fflags = DISPATCH_MACH_RECV_MESSAGE_DIRECT_ONCE;
- dk->dk_kevent.udata = (uintptr_t)dk;
- TAILQ_INIT(&dk->dk_sources);
-
- dmr = _dispatch_calloc(1ul, sizeof(struct dispatch_mach_reply_refs_s));
- dmr->dr_source_wref = _dispatch_ptr2wref(dm);
- dmr->dmr_dkev = dk;
- if (dmsg->dmsg_voucher) {
- dmr->dmr_voucher =_voucher_retain(dmsg->dmsg_voucher);
- }
- dmr->dmr_priority = dmsg->dmsg_priority;
- // make reply context visible to leaks rdar://11777199
- dmr->dmr_ctxt = dmsg->do_ctxt;
-
- _dispatch_debug("machport[0x%08x]: registering for reply, ctxt %p", reply,
- dmsg->do_ctxt);
- uint32_t flags;
- bool do_resume = _dispatch_kevent_register(&dmr->dmr_dkev, &flags);
- TAILQ_INSERT_TAIL(&dmr->dmr_dkev->dk_sources, (dispatch_source_refs_t)dmr,
- dr_list);
- TAILQ_INSERT_TAIL(&dm->dm_refs->dm_replies, dmr, dmr_list);
- if (do_resume && _dispatch_kevent_resume(dmr->dmr_dkev, flags, 0)) {
- _dispatch_mach_reply_kevent_unregister(dm, dmr, true);
- }
-}
-
-DISPATCH_NOINLINE
-static void
-_dispatch_mach_kevent_unregister(dispatch_mach_t dm)
-{
- dispatch_kevent_t dk = dm->dm_dkev;
- dm->dm_dkev = NULL;
- TAILQ_REMOVE(&dk->dk_sources, (dispatch_source_refs_t)dm->dm_refs,
- dr_list);
- dm->ds_pending_data_mask &= ~(unsigned long)
- (DISPATCH_MACH_SEND_POSSIBLE|DISPATCH_MACH_SEND_DEAD);
- _dispatch_kevent_unregister(dk,
- DISPATCH_MACH_SEND_POSSIBLE|DISPATCH_MACH_SEND_DEAD);
-}
-
-DISPATCH_NOINLINE
-static void
-_dispatch_mach_kevent_register(dispatch_mach_t dm, mach_port_t send)
-{
- dispatch_kevent_t dk;
-
- dk = _dispatch_calloc(1ul, sizeof(struct dispatch_kevent_s));
- dk->dk_kevent = _dispatch_source_type_mach_send.ke;
- dk->dk_kevent.ident = send;
- dk->dk_kevent.flags |= EV_ADD|EV_ENABLE;
- dk->dk_kevent.fflags = DISPATCH_MACH_SEND_POSSIBLE|DISPATCH_MACH_SEND_DEAD;
- dk->dk_kevent.udata = (uintptr_t)dk;
- TAILQ_INIT(&dk->dk_sources);
-
- dm->ds_pending_data_mask |= dk->dk_kevent.fflags;
-
- uint32_t flags;
- bool do_resume = _dispatch_kevent_register(&dk, &flags);
- TAILQ_INSERT_TAIL(&dk->dk_sources,
- (dispatch_source_refs_t)dm->dm_refs, dr_list);
- dm->dm_dkev = dk;
- if (do_resume && _dispatch_kevent_resume(dm->dm_dkev, flags, 0)) {
- _dispatch_mach_kevent_unregister(dm);
- }
-}
-
-static inline void
-_dispatch_mach_push(dispatch_object_t dm, dispatch_object_t dou,
- pthread_priority_t pp)
-{
- return _dispatch_queue_push(dm._dq, dou, pp);
-}
-
-static inline void
-_dispatch_mach_msg_set_options(dispatch_object_t dou, mach_msg_option_t options)
-{
- dou._do->do_suspend_cnt = (unsigned int)options;
-}
-
-static inline mach_msg_option_t
-_dispatch_mach_msg_get_options(dispatch_object_t dou)
-{
- mach_msg_option_t options = (mach_msg_option_t)dou._do->do_suspend_cnt;
- return options;
-}
-
-static inline void
-_dispatch_mach_msg_set_reason(dispatch_object_t dou, mach_error_t err,
- unsigned long reason)
-{
- dispatch_assert_zero(reason & ~(unsigned long)code_emask);
- dou._do->do_suspend_cnt = (unsigned int)((err || !reason) ? err :
- err_local|err_sub(0x3e0)|(mach_error_t)reason);
-}
-
-static inline unsigned long
-_dispatch_mach_msg_get_reason(dispatch_object_t dou, mach_error_t *err_ptr)
-{
- mach_error_t err = (mach_error_t)dou._do->do_suspend_cnt;
- dou._do->do_suspend_cnt = 0;
- if ((err & system_emask) == err_local && err_get_sub(err) == 0x3e0) {
- *err_ptr = 0;
- return err_get_code(err);
- }
- *err_ptr = err;
- return err ? DISPATCH_MACH_MESSAGE_SEND_FAILED : DISPATCH_MACH_MESSAGE_SENT;
-}
-
-static void
-_dispatch_mach_msg_recv(dispatch_mach_t dm, dispatch_mach_reply_refs_t dmr,
- mach_msg_header_t *hdr, mach_msg_size_t siz)
-{
- _dispatch_debug_machport(hdr->msgh_remote_port);
- _dispatch_debug("machport[0x%08x]: received msg id 0x%x, reply on 0x%08x",
- hdr->msgh_local_port, hdr->msgh_id, hdr->msgh_remote_port);
- if (slowpath(dm->ds_atomic_flags & DSF_CANCELED)) {
- return _dispatch_kevent_mach_msg_destroy(hdr);
- }
- dispatch_mach_msg_t dmsg;
- voucher_t voucher;
- pthread_priority_t priority;
- void *ctxt = NULL;
- if (dmr) {
- _voucher_mach_msg_clear(hdr, false); // deallocate reply message voucher
- voucher = dmr->dmr_voucher;
- dmr->dmr_voucher = NULL; // transfer reference
- priority = dmr->dmr_priority;
- ctxt = dmr->dmr_ctxt;
- _dispatch_mach_reply_kevent_unregister(dm, dmr, false);
- } else {
- voucher = voucher_create_with_mach_msg(hdr);
- priority = _voucher_get_priority(voucher);
- }
- dispatch_mach_msg_destructor_t destructor;
- destructor = (hdr == _dispatch_get_mach_recv_msg_buf()) ?
- DISPATCH_MACH_MSG_DESTRUCTOR_DEFAULT :
- DISPATCH_MACH_MSG_DESTRUCTOR_FREE;
- dmsg = dispatch_mach_msg_create(hdr, siz, destructor, NULL);
- dmsg->dmsg_voucher = voucher;
- dmsg->dmsg_priority = priority;
- dmsg->do_ctxt = ctxt;
- _dispatch_mach_msg_set_reason(dmsg, 0, DISPATCH_MACH_MESSAGE_RECEIVED);
- _dispatch_voucher_debug("mach-msg[%p] create", voucher, dmsg);
- _dispatch_voucher_ktrace_dmsg_push(dmsg);
- return _dispatch_mach_push(dm, dmsg, dmsg->dmsg_priority);
-}
-
-static inline mach_port_t
-_dispatch_mach_msg_get_remote_port(dispatch_object_t dou)
-{
- mach_msg_header_t *hdr = _dispatch_mach_msg_get_msg(dou._dmsg);
- mach_port_t remote = hdr->msgh_remote_port;
- return remote;
-}
-
-static inline void
-_dispatch_mach_msg_disconnected(dispatch_mach_t dm, mach_port_t local_port,
- mach_port_t remote_port)
-{
- mach_msg_header_t *hdr;
- dispatch_mach_msg_t dmsg;
- dmsg = dispatch_mach_msg_create(NULL, sizeof(mach_msg_header_t),
- DISPATCH_MACH_MSG_DESTRUCTOR_DEFAULT, &hdr);
- if (local_port) hdr->msgh_local_port = local_port;
- if (remote_port) hdr->msgh_remote_port = remote_port;
- _dispatch_mach_msg_set_reason(dmsg, 0, DISPATCH_MACH_DISCONNECTED);
- return _dispatch_mach_push(dm, dmsg, dmsg->dmsg_priority);
-}
-
-static inline dispatch_mach_msg_t
-_dispatch_mach_msg_create_reply_disconnected(dispatch_object_t dou,
- dispatch_mach_reply_refs_t dmr)
-{
- dispatch_mach_msg_t dmsg = dou._dmsg, dmsgr;
- if (dmsg && !dmsg->dmsg_reply) return NULL;
- mach_msg_header_t *hdr;
- dmsgr = dispatch_mach_msg_create(NULL, sizeof(mach_msg_header_t),
- DISPATCH_MACH_MSG_DESTRUCTOR_DEFAULT, &hdr);
- if (dmsg) {
- hdr->msgh_local_port = dmsg->dmsg_reply;
- if (dmsg->dmsg_voucher) {
- dmsgr->dmsg_voucher = _voucher_retain(dmsg->dmsg_voucher);
- }
- dmsgr->dmsg_priority = dmsg->dmsg_priority;
- dmsgr->do_ctxt = dmsg->do_ctxt;
- } else {
- hdr->msgh_local_port = (mach_port_t)dmr->dmr_dkev->dk_kevent.ident;
- dmsgr->dmsg_voucher = dmr->dmr_voucher;
- dmr->dmr_voucher = NULL; // transfer reference
- dmsgr->dmsg_priority = dmr->dmr_priority;
- dmsgr->do_ctxt = dmr->dmr_ctxt;
- }
- _dispatch_mach_msg_set_reason(dmsgr, 0, DISPATCH_MACH_DISCONNECTED);
- return dmsgr;
-}
-
-DISPATCH_NOINLINE
-static void
-_dispatch_mach_msg_not_sent(dispatch_mach_t dm, dispatch_object_t dou)
-{
- dispatch_mach_msg_t dmsg = dou._dmsg, dmsgr;
- dmsgr = _dispatch_mach_msg_create_reply_disconnected(dmsg, NULL);
- _dispatch_mach_msg_set_reason(dmsg, 0, DISPATCH_MACH_MESSAGE_NOT_SENT);
- _dispatch_mach_push(dm, dmsg, dmsg->dmsg_priority);
- if (dmsgr) _dispatch_mach_push(dm, dmsgr, dmsgr->dmsg_priority);
-}
-
-DISPATCH_NOINLINE
-static dispatch_object_t
-_dispatch_mach_msg_send(dispatch_mach_t dm, dispatch_object_t dou)
-{
- dispatch_mach_send_refs_t dr = dm->dm_refs;
- dispatch_mach_msg_t dmsg = dou._dmsg, dmsgr = NULL;
- voucher_t voucher = dmsg->dmsg_voucher;
- mach_voucher_t ipc_kvoucher = MACH_VOUCHER_NULL;
- bool clear_voucher = false, kvoucher_move_send = false;
- dr->dm_needs_mgr = 0;
- if (slowpath(dr->dm_checkin) && dmsg != dr->dm_checkin) {
- // send initial checkin message
- if (dm->dm_dkev && slowpath(_dispatch_queue_get_current() !=
- &_dispatch_mgr_q)) {
- // send kevent must be uninstalled on the manager queue
- dr->dm_needs_mgr = 1;
- goto out;
- }
- dr->dm_checkin = _dispatch_mach_msg_send(dm, dr->dm_checkin)._dmsg;
- if (slowpath(dr->dm_checkin)) {
- goto out;
- }
- }
- mach_msg_header_t *msg = _dispatch_mach_msg_get_msg(dmsg);
- mach_msg_return_t kr = 0;
- mach_port_t reply = dmsg->dmsg_reply;
- mach_msg_option_t opts = 0, msg_opts = _dispatch_mach_msg_get_options(dmsg);
- if (!slowpath(msg_opts & DISPATCH_MACH_REGISTER_FOR_REPLY)) {
- opts = MACH_SEND_MSG | (msg_opts & ~DISPATCH_MACH_OPTIONS_MASK);
- if (MACH_MSGH_BITS_REMOTE(msg->msgh_bits) !=
- MACH_MSG_TYPE_MOVE_SEND_ONCE) {
- if (dmsg != dr->dm_checkin) {
- msg->msgh_remote_port = dr->dm_send;
- }
- if (_dispatch_queue_get_current() == &_dispatch_mgr_q) {
- if (slowpath(!dm->dm_dkev)) {
- _dispatch_mach_kevent_register(dm, msg->msgh_remote_port);
- }
- if (fastpath(dm->dm_dkev)) {
- if (DISPATCH_MACH_KEVENT_ARMED(dm->dm_dkev)) {
- goto out;
- }
- opts |= MACH_SEND_NOTIFY;
- }
- }
- opts |= MACH_SEND_TIMEOUT;
- if (dmsg->dmsg_priority != _voucher_get_priority(voucher)) {
- ipc_kvoucher = _voucher_create_mach_voucher_with_priority(
- voucher, dmsg->dmsg_priority);
- }
- _dispatch_voucher_debug("mach-msg[%p] msg_set", voucher, dmsg);
- if (ipc_kvoucher) {
- kvoucher_move_send = true;
- clear_voucher = _voucher_mach_msg_set_mach_voucher(msg,
- ipc_kvoucher, kvoucher_move_send);
- } else {
- clear_voucher = _voucher_mach_msg_set(msg, voucher);
- }
- }
- _voucher_activity_trace_msg(voucher, msg, send);
- _dispatch_debug_machport(msg->msgh_remote_port);
- if (reply) _dispatch_debug_machport(reply);
- kr = mach_msg(msg, opts, msg->msgh_size, 0, MACH_PORT_NULL, 0,
- MACH_PORT_NULL);
- _dispatch_debug("machport[0x%08x]: sent msg id 0x%x, ctxt %p, "
- "opts 0x%x, msg_opts 0x%x, kvoucher 0x%08x, reply on 0x%08x: "
- "%s - 0x%x", msg->msgh_remote_port, msg->msgh_id, dmsg->do_ctxt,
- opts, msg_opts, msg->msgh_voucher_port, reply,
- mach_error_string(kr), kr);
- if (clear_voucher) {
- if (kr == MACH_SEND_INVALID_VOUCHER && msg->msgh_voucher_port) {
- DISPATCH_CRASH("Voucher port corruption");
- }
- mach_voucher_t kv;
- kv = _voucher_mach_msg_clear(msg, kvoucher_move_send);
- if (kvoucher_move_send) ipc_kvoucher = kv;
- }
- }
- if (kr == MACH_SEND_TIMED_OUT && (opts & MACH_SEND_TIMEOUT)) {
- if (opts & MACH_SEND_NOTIFY) {
- _dispatch_debug("machport[0x%08x]: send-possible notification "
- "armed", (mach_port_t)dm->dm_dkev->dk_kevent.ident);
- DISPATCH_MACH_KEVENT_ARMED(dm->dm_dkev) = 1;
- } else {
- // send kevent must be installed on the manager queue
- dr->dm_needs_mgr = 1;
- }
- if (ipc_kvoucher) {
- _dispatch_kvoucher_debug("reuse on re-send", ipc_kvoucher);
- voucher_t ipc_voucher;
- ipc_voucher = _voucher_create_with_priority_and_mach_voucher(
- voucher, dmsg->dmsg_priority, ipc_kvoucher);
- _dispatch_voucher_debug("mach-msg[%p] replace voucher[%p]",
- ipc_voucher, dmsg, voucher);
- if (dmsg->dmsg_voucher) _voucher_release(dmsg->dmsg_voucher);
- dmsg->dmsg_voucher = ipc_voucher;
- }
- goto out;
- } else if (ipc_kvoucher && (kr || !kvoucher_move_send)) {
- _voucher_dealloc_mach_voucher(ipc_kvoucher);
- }
- if (fastpath(!kr) && reply &&
- !(dm->ds_dkev && dm->ds_dkev->dk_kevent.ident == reply)) {
- if (_dispatch_queue_get_current() != &_dispatch_mgr_q) {
- // reply receive kevent must be installed on the manager queue
- dr->dm_needs_mgr = 1;
- _dispatch_mach_msg_set_options(dmsg, msg_opts |
- DISPATCH_MACH_REGISTER_FOR_REPLY);
- goto out;
- }
- _dispatch_mach_reply_kevent_register(dm, reply, dmsg);
- }
- if (slowpath(dmsg == dr->dm_checkin) && dm->dm_dkev) {
- _dispatch_mach_kevent_unregister(dm);
- }
- if (slowpath(kr)) {
- // Send failed, so reply was never connected <rdar://problem/14309159>
- dmsgr = _dispatch_mach_msg_create_reply_disconnected(dmsg, NULL);
- }
- _dispatch_mach_msg_set_reason(dmsg, kr, 0);
- _dispatch_mach_push(dm, dmsg, dmsg->dmsg_priority);
- if (dmsgr) _dispatch_mach_push(dm, dmsgr, dmsgr->dmsg_priority);
- dmsg = NULL;
-out:
- return (dispatch_object_t)dmsg;
-}
-
-DISPATCH_ALWAYS_INLINE
-static inline void
-_dispatch_mach_send_push_wakeup(dispatch_mach_t dm, dispatch_object_t dou,
- bool wakeup)
-{
- dispatch_mach_send_refs_t dr = dm->dm_refs;
- struct dispatch_object_s *prev, *dc = dou._do;
- dc->do_next = NULL;
-
- prev = dispatch_atomic_xchg2o(dr, dm_tail, dc, release);
- if (fastpath(prev)) {
- prev->do_next = dc;
- } else {
- dr->dm_head = dc;
- }
- if (wakeup || !prev) {
- _dispatch_wakeup(dm);
- }
-}
-
-DISPATCH_ALWAYS_INLINE
-static inline void
-_dispatch_mach_send_push(dispatch_mach_t dm, dispatch_object_t dou)
-{
- return _dispatch_mach_send_push_wakeup(dm, dou, false);
-}
-
-DISPATCH_NOINLINE
-static void
-_dispatch_mach_send_drain(dispatch_mach_t dm)
-{
- dispatch_mach_send_refs_t dr = dm->dm_refs;
- struct dispatch_object_s *dc = NULL, *next_dc = NULL;
- while (dr->dm_tail) {
- _dispatch_wait_until(dc = fastpath(dr->dm_head));
- do {
- next_dc = fastpath(dc->do_next);
- dr->dm_head = next_dc;
- if (!next_dc && !dispatch_atomic_cmpxchg2o(dr, dm_tail, dc, NULL,
- relaxed)) {
- _dispatch_wait_until(next_dc = fastpath(dc->do_next));
- dr->dm_head = next_dc;
- }
- if (!DISPATCH_OBJ_IS_VTABLE(dc)) {
- if ((long)dc->do_vtable & DISPATCH_OBJ_BARRIER_BIT) {
- // send barrier
- // leave send queue locked until barrier has completed
- return _dispatch_mach_push(dm, dc,
- ((dispatch_continuation_t)dc)->dc_priority);
- }
-#if DISPATCH_MACH_SEND_SYNC
- if (slowpath((long)dc->do_vtable & DISPATCH_OBJ_SYNC_SLOW_BIT)){
- _dispatch_thread_semaphore_signal(
- (_dispatch_thread_semaphore_t)dc->do_ctxt);
- continue;
- }
-#endif // DISPATCH_MACH_SEND_SYNC
- if (slowpath(!_dispatch_mach_reconnect_invoke(dm, dc))) {
- goto out;
- }
- continue;
- }
- _dispatch_voucher_ktrace_dmsg_pop((dispatch_mach_msg_t)dc);
- if (slowpath(dr->dm_disconnect_cnt) ||
- slowpath(dm->ds_atomic_flags & DSF_CANCELED)) {
- _dispatch_mach_msg_not_sent(dm, dc);
- continue;
- }
- if (slowpath(dc = _dispatch_mach_msg_send(dm, dc)._do)) {
- goto out;
- }
- } while ((dc = next_dc));
- }
-out:
- // if this is not a complete drain, we must undo some things
- if (slowpath(dc)) {
- if (!next_dc &&
- !dispatch_atomic_cmpxchg2o(dr, dm_tail, NULL, dc, relaxed)) {
- // wait for enqueue slow path to finish
- _dispatch_wait_until(next_dc = fastpath(dr->dm_head));
- dc->do_next = next_dc;
- }
- dr->dm_head = dc;
- }
- (void)dispatch_atomic_dec2o(dr, dm_sending, release);
- _dispatch_wakeup(dm);
-}
-
-static inline void
-_dispatch_mach_send(dispatch_mach_t dm)
-{
- dispatch_mach_send_refs_t dr = dm->dm_refs;
- if (!fastpath(dr->dm_tail) || !fastpath(dispatch_atomic_cmpxchg2o(dr,
- dm_sending, 0, 1, acquire))) {
- return;
- }
- _dispatch_object_debug(dm, "%s", __func__);
- _dispatch_mach_send_drain(dm);
-}
-
-DISPATCH_NOINLINE
-static void
-_dispatch_mach_merge_kevent(dispatch_mach_t dm, const struct kevent64_s *ke)
-{
- if (!(ke->fflags & dm->ds_pending_data_mask)) {
- return;
- }
- _dispatch_mach_send(dm);
-}
-
-static inline mach_msg_option_t
-_dispatch_mach_checkin_options(void)
-{
- mach_msg_option_t options = 0;
-#if DISPATCH_USE_CHECKIN_NOIMPORTANCE
- options = MACH_SEND_NOIMPORTANCE; // <rdar://problem/16996737>
-#endif
- return options;
-}
-
-
-static inline mach_msg_option_t
-_dispatch_mach_send_options(void)
-{
- mach_msg_option_t options = 0;
- return options;
-}
-
-DISPATCH_NOINLINE
-void
-dispatch_mach_send(dispatch_mach_t dm, dispatch_mach_msg_t dmsg,
- mach_msg_option_t options)
-{
- dispatch_mach_send_refs_t dr = dm->dm_refs;
- if (slowpath(dmsg->do_next != DISPATCH_OBJECT_LISTLESS)) {
- DISPATCH_CLIENT_CRASH("Message already enqueued");
- }
- dispatch_retain(dmsg);
- dispatch_assert_zero(options & DISPATCH_MACH_OPTIONS_MASK);
- options |= _dispatch_mach_send_options();
- _dispatch_mach_msg_set_options(dmsg, options & ~DISPATCH_MACH_OPTIONS_MASK);
- mach_msg_header_t *msg = _dispatch_mach_msg_get_msg(dmsg);
- dmsg->dmsg_reply = (MACH_MSGH_BITS_LOCAL(msg->msgh_bits) ==
- MACH_MSG_TYPE_MAKE_SEND_ONCE &&
- MACH_PORT_VALID(msg->msgh_local_port) ? msg->msgh_local_port :
- MACH_PORT_NULL);
- bool is_reply = (MACH_MSGH_BITS_REMOTE(msg->msgh_bits) ==
- MACH_MSG_TYPE_MOVE_SEND_ONCE);
- dmsg->dmsg_priority = _dispatch_priority_propagate();
- dmsg->dmsg_voucher = _voucher_copy();
- _dispatch_voucher_debug("mach-msg[%p] set", dmsg->dmsg_voucher, dmsg);
- if ((!is_reply && slowpath(dr->dm_tail)) ||
- slowpath(dr->dm_disconnect_cnt) ||
- slowpath(dm->ds_atomic_flags & DSF_CANCELED) ||
- slowpath(!dispatch_atomic_cmpxchg2o(dr, dm_sending, 0, 1,
- acquire))) {
- _dispatch_voucher_ktrace_dmsg_push(dmsg);
- return _dispatch_mach_send_push(dm, dmsg);
- }
- if (slowpath(dmsg = _dispatch_mach_msg_send(dm, dmsg)._dmsg)) {
- (void)dispatch_atomic_dec2o(dr, dm_sending, release);
- _dispatch_voucher_ktrace_dmsg_push(dmsg);
- return _dispatch_mach_send_push_wakeup(dm, dmsg, true);
- }
- if (!is_reply && slowpath(dr->dm_tail)) {
- return _dispatch_mach_send_drain(dm);
- }
- (void)dispatch_atomic_dec2o(dr, dm_sending, release);
- _dispatch_wakeup(dm);
-}
-
-static void
-_dispatch_mach_disconnect(dispatch_mach_t dm)
-{
- dispatch_mach_send_refs_t dr = dm->dm_refs;
- if (dm->dm_dkev) {
- _dispatch_mach_kevent_unregister(dm);
- }
- if (MACH_PORT_VALID(dr->dm_send)) {
- _dispatch_mach_msg_disconnected(dm, MACH_PORT_NULL, dr->dm_send);
- }
- dr->dm_send = MACH_PORT_NULL;
- if (dr->dm_checkin) {
- _dispatch_mach_msg_not_sent(dm, dr->dm_checkin);
- dr->dm_checkin = NULL;
- }
- if (!TAILQ_EMPTY(&dm->dm_refs->dm_replies)) {
- dispatch_mach_reply_refs_t dmr, tmp;
- TAILQ_FOREACH_SAFE(dmr, &dm->dm_refs->dm_replies, dmr_list, tmp){
- _dispatch_mach_reply_kevent_unregister(dm, dmr, true);
- }
- }
-}
-
-DISPATCH_NOINLINE
-static bool
-_dispatch_mach_cancel(dispatch_mach_t dm)
-{
- dispatch_mach_send_refs_t dr = dm->dm_refs;
- if (!fastpath(dispatch_atomic_cmpxchg2o(dr, dm_sending, 0, 1, acquire))) {
- return false;
- }
- _dispatch_object_debug(dm, "%s", __func__);
- _dispatch_mach_disconnect(dm);
- if (dm->ds_dkev) {
- mach_port_t local_port = (mach_port_t)dm->ds_dkev->dk_kevent.ident;
- _dispatch_source_kevent_unregister((dispatch_source_t)dm);
- _dispatch_mach_msg_disconnected(dm, local_port, MACH_PORT_NULL);
- }
- (void)dispatch_atomic_dec2o(dr, dm_sending, release);
- return true;
-}
-
-DISPATCH_NOINLINE
-static bool
-_dispatch_mach_reconnect_invoke(dispatch_mach_t dm, dispatch_object_t dou)
-{
- if (dm->dm_dkev || !TAILQ_EMPTY(&dm->dm_refs->dm_replies)) {
- if (slowpath(_dispatch_queue_get_current() != &_dispatch_mgr_q)) {
- // send/reply kevents must be uninstalled on the manager queue
- return false;
- }
- }
- _dispatch_mach_disconnect(dm);
- dispatch_mach_send_refs_t dr = dm->dm_refs;
- dr->dm_checkin = dou._dc->dc_data;
- dr->dm_send = (mach_port_t)dou._dc->dc_other;
- _dispatch_continuation_free(dou._dc);
- (void)dispatch_atomic_dec2o(dr, dm_disconnect_cnt, relaxed);
- _dispatch_object_debug(dm, "%s", __func__);
- return true;
-}
-
-DISPATCH_NOINLINE
-void
-dispatch_mach_reconnect(dispatch_mach_t dm, mach_port_t send,
- dispatch_mach_msg_t checkin)
-{
- dispatch_mach_send_refs_t dr = dm->dm_refs;
- (void)dispatch_atomic_inc2o(dr, dm_disconnect_cnt, relaxed);
- if (MACH_PORT_VALID(send) && checkin) {
- dispatch_retain(checkin);
- mach_msg_option_t options = _dispatch_mach_checkin_options();
- _dispatch_mach_msg_set_options(checkin, options);
- dr->dm_checkin_port = _dispatch_mach_msg_get_remote_port(checkin);
- } else {
- checkin = NULL;
- dr->dm_checkin_port = MACH_PORT_NULL;
- }
- dispatch_continuation_t dc = _dispatch_continuation_alloc();
- dc->do_vtable = (void *)(DISPATCH_OBJ_ASYNC_BIT);
- dc->dc_func = (void*)_dispatch_mach_reconnect_invoke;
- dc->dc_ctxt = dc;
- dc->dc_data = checkin;
- dc->dc_other = (void*)(uintptr_t)send;
- return _dispatch_mach_send_push(dm, dc);
-}
-
-#if DISPATCH_MACH_SEND_SYNC
-DISPATCH_NOINLINE
-static void
-_dispatch_mach_send_sync_slow(dispatch_mach_t dm)
-{
- _dispatch_thread_semaphore_t sema = _dispatch_get_thread_semaphore();
- struct dispatch_object_s dc = {
- .do_vtable = (void *)(DISPATCH_OBJ_SYNC_SLOW_BIT),
- .do_ctxt = (void*)sema,
- };
- _dispatch_mach_send_push(dm, &dc);
- _dispatch_thread_semaphore_wait(sema);
- _dispatch_put_thread_semaphore(sema);
-}
-#endif // DISPATCH_MACH_SEND_SYNC
-
-DISPATCH_NOINLINE
-mach_port_t
-dispatch_mach_get_checkin_port(dispatch_mach_t dm)
-{
- dispatch_mach_send_refs_t dr = dm->dm_refs;
- if (slowpath(dm->ds_atomic_flags & DSF_CANCELED)) {
- return MACH_PORT_DEAD;
- }
- return dr->dm_checkin_port;
-}
-
-DISPATCH_NOINLINE
-static void
-_dispatch_mach_connect_invoke(dispatch_mach_t dm)
-{
- dispatch_mach_refs_t dr = dm->ds_refs;
- _dispatch_client_callout4(dr->dm_handler_ctxt,
- DISPATCH_MACH_CONNECTED, NULL, 0, dr->dm_handler_func);
- dm->dm_connect_handler_called = 1;
-}
-
-DISPATCH_NOINLINE
-void
-_dispatch_mach_msg_invoke(dispatch_mach_msg_t dmsg)
-{
- dispatch_mach_t dm = (dispatch_mach_t)_dispatch_queue_get_current();
- dispatch_mach_refs_t dr = dm->ds_refs;
- mach_error_t err;
- unsigned long reason = _dispatch_mach_msg_get_reason(dmsg, &err);
-
- dmsg->do_next = DISPATCH_OBJECT_LISTLESS;
- _dispatch_thread_setspecific(dispatch_queue_key, dm->do_targetq);
- _dispatch_voucher_ktrace_dmsg_pop(dmsg);
- _dispatch_voucher_debug("mach-msg[%p] adopt", dmsg->dmsg_voucher, dmsg);
- _dispatch_adopt_priority_and_replace_voucher(dmsg->dmsg_priority,
- dmsg->dmsg_voucher, DISPATCH_PRIORITY_ENFORCE);
- dmsg->dmsg_voucher = NULL;
- if (slowpath(!dm->dm_connect_handler_called)) {
- _dispatch_mach_connect_invoke(dm);
- }
- _dispatch_client_callout4(dr->dm_handler_ctxt, reason, dmsg, err,
- dr->dm_handler_func);
- _dispatch_thread_setspecific(dispatch_queue_key, (dispatch_queue_t)dm);
- _dispatch_introspection_queue_item_complete(dmsg);
- dispatch_release(dmsg);
-}
-
-DISPATCH_NOINLINE
-void
-_dispatch_mach_barrier_invoke(void *ctxt)
-{
- dispatch_mach_t dm = (dispatch_mach_t)_dispatch_queue_get_current();
- dispatch_mach_refs_t dr = dm->ds_refs;
- struct dispatch_continuation_s *dc = ctxt;
- void *context = dc->dc_data;
- dispatch_function_t barrier = dc->dc_other;
- bool send_barrier = ((long)dc->do_vtable & DISPATCH_OBJ_BARRIER_BIT);
-
- _dispatch_thread_setspecific(dispatch_queue_key, dm->do_targetq);
- if (slowpath(!dm->dm_connect_handler_called)) {
- _dispatch_mach_connect_invoke(dm);
- }
- _dispatch_client_callout(context, barrier);
- _dispatch_client_callout4(dr->dm_handler_ctxt,
- DISPATCH_MACH_BARRIER_COMPLETED, NULL, 0, dr->dm_handler_func);
- _dispatch_thread_setspecific(dispatch_queue_key, (dispatch_queue_t)dm);
- if (send_barrier) {
- (void)dispatch_atomic_dec2o(dm->dm_refs, dm_sending, release);
- }
-}
-
-DISPATCH_NOINLINE
-void
-dispatch_mach_send_barrier_f(dispatch_mach_t dm, void *context,
- dispatch_function_t barrier)
-{
- dispatch_continuation_t dc = _dispatch_continuation_alloc();
- dc->do_vtable = (void *)(DISPATCH_OBJ_ASYNC_BIT | DISPATCH_OBJ_BARRIER_BIT);
- dc->dc_func = _dispatch_mach_barrier_invoke;
- dc->dc_ctxt = dc;
- dc->dc_data = context;
- dc->dc_other = barrier;
- _dispatch_continuation_voucher_set(dc, 0);
- _dispatch_continuation_priority_set(dc, 0, 0);
-
- dispatch_mach_send_refs_t dr = dm->dm_refs;
- if (slowpath(dr->dm_tail) || slowpath(!dispatch_atomic_cmpxchg2o(dr,
- dm_sending, 0, 1, acquire))) {
- return _dispatch_mach_send_push(dm, dc);
- }
- // leave send queue locked until barrier has completed
- return _dispatch_mach_push(dm, dc, dc->dc_priority);
-}
-
-DISPATCH_NOINLINE
-void
-dispatch_mach_receive_barrier_f(dispatch_mach_t dm, void *context,
- dispatch_function_t barrier)
-{
- dispatch_continuation_t dc = _dispatch_continuation_alloc();
- dc->do_vtable = (void *)(DISPATCH_OBJ_ASYNC_BIT);
- dc->dc_func = _dispatch_mach_barrier_invoke;
- dc->dc_ctxt = dc;
- dc->dc_data = context;
- dc->dc_other = barrier;
- _dispatch_continuation_voucher_set(dc, 0);
- _dispatch_continuation_priority_set(dc, 0, 0);
-
- return _dispatch_mach_push(dm, dc, dc->dc_priority);
-}
-
-DISPATCH_NOINLINE
-void
-dispatch_mach_send_barrier(dispatch_mach_t dm, dispatch_block_t barrier)
-{
- dispatch_mach_send_barrier_f(dm, _dispatch_Block_copy(barrier),
- _dispatch_call_block_and_release);
-}
-
-DISPATCH_NOINLINE
-void
-dispatch_mach_receive_barrier(dispatch_mach_t dm, dispatch_block_t barrier)
-{
- dispatch_mach_receive_barrier_f(dm, _dispatch_Block_copy(barrier),
- _dispatch_call_block_and_release);
-}
-
-DISPATCH_NOINLINE
-static void
-_dispatch_mach_cancel_invoke(dispatch_mach_t dm)
-{
- dispatch_mach_refs_t dr = dm->ds_refs;
- if (slowpath(!dm->dm_connect_handler_called)) {
- _dispatch_mach_connect_invoke(dm);
- }
- _dispatch_client_callout4(dr->dm_handler_ctxt,
- DISPATCH_MACH_CANCELED, NULL, 0, dr->dm_handler_func);
- dm->dm_cancel_handler_called = 1;
- _dispatch_release(dm); // the retain is done at creation time
-}
-
-DISPATCH_NOINLINE
-void
-dispatch_mach_cancel(dispatch_mach_t dm)
-{
- dispatch_source_cancel((dispatch_source_t)dm);
-}
-
-DISPATCH_ALWAYS_INLINE
-static inline dispatch_queue_t
-_dispatch_mach_invoke2(dispatch_object_t dou,
- _dispatch_thread_semaphore_t *sema_ptr DISPATCH_UNUSED)
-{
- dispatch_mach_t dm = dou._dm;
-
- // This function performs all mach channel actions. Each action is
- // responsible for verifying that it takes place on the appropriate queue.
- // If the current queue is not the correct queue for this action, the
- // correct queue will be returned and the invoke will be re-driven on that
- // queue.
-
- // The order of tests here in invoke and in probe should be consistent.
-
- dispatch_queue_t dq = _dispatch_queue_get_current();
- dispatch_mach_send_refs_t dr = dm->dm_refs;
-
- if (slowpath(!dm->ds_is_installed)) {
- // The channel needs to be installed on the manager queue.
- if (dq != &_dispatch_mgr_q) {
- return &_dispatch_mgr_q;
- }
- if (dm->ds_dkev) {
- _dispatch_source_kevent_register((dispatch_source_t)dm);
- }
- dm->ds_is_installed = true;
- _dispatch_mach_send(dm);
- // Apply initial target queue change
- _dispatch_queue_drain(dou);
- if (dm->dq_items_tail) {
- return dm->do_targetq;
- }
- } else if (dm->dq_items_tail) {
- // The channel has pending messages to deliver to the target queue.
- if (dq != dm->do_targetq) {
- return dm->do_targetq;
- }
- dispatch_queue_t tq = dm->do_targetq;
- if (slowpath(_dispatch_queue_drain(dou))) {
- DISPATCH_CLIENT_CRASH("Sync onto mach channel");
- }
- if (slowpath(tq != dm->do_targetq)) {
- // An item on the channel changed the target queue
- return dm->do_targetq;
- }
- } else if (dr->dm_sending) {
- // Sending and uninstallation below require the send lock, the channel
- // will be woken up when the lock is dropped <rdar://15132939&15203957>
- return NULL;
- } else if (dr->dm_tail) {
- if (slowpath(dr->dm_needs_mgr) || (slowpath(dr->dm_disconnect_cnt) &&
- (dm->dm_dkev || !TAILQ_EMPTY(&dm->dm_refs->dm_replies)))) {
- // Send/reply kevents need to be installed or uninstalled
- if (dq != &_dispatch_mgr_q) {
- return &_dispatch_mgr_q;
- }
- }
- if (!(dm->dm_dkev && DISPATCH_MACH_KEVENT_ARMED(dm->dm_dkev)) ||
- (dm->ds_atomic_flags & DSF_CANCELED) || dr->dm_disconnect_cnt) {
- // The channel has pending messages to send.
- _dispatch_mach_send(dm);
- }
- } else if (dm->ds_atomic_flags & DSF_CANCELED){
- // The channel has been cancelled and needs to be uninstalled from the
- // manager queue. After uninstallation, the cancellation handler needs
- // to be delivered to the target queue.
- if (dm->ds_dkev || dm->dm_dkev || dr->dm_send ||
- !TAILQ_EMPTY(&dm->dm_refs->dm_replies)) {
- if (dq != &_dispatch_mgr_q) {
- return &_dispatch_mgr_q;
- }
- if (!_dispatch_mach_cancel(dm)) {
- return NULL;
- }
- }
- if (!dm->dm_cancel_handler_called) {
- if (dq != dm->do_targetq) {
- return dm->do_targetq;
- }
- _dispatch_mach_cancel_invoke(dm);
- }
- }
- return NULL;
-}
-
-DISPATCH_NOINLINE
-void
-_dispatch_mach_invoke(dispatch_mach_t dm)
-{
- _dispatch_queue_class_invoke(dm, _dispatch_mach_invoke2);
-}
-
-unsigned long
-_dispatch_mach_probe(dispatch_mach_t dm)
-{
- // This function determines whether the mach channel needs to be invoked.
- // The order of tests here in probe and in invoke should be consistent.
-
- dispatch_mach_send_refs_t dr = dm->dm_refs;
-
- if (slowpath(!dm->ds_is_installed)) {
- // The channel needs to be installed on the manager queue.
- return true;
- } else if (_dispatch_queue_class_probe(dm)) {
- // The source has pending messages to deliver to the target queue.
- return true;
- } else if (dr->dm_sending) {
- // Sending and uninstallation below require the send lock, the channel
- // will be woken up when the lock is dropped <rdar://15132939&15203957>
- return false;
- } else if (dr->dm_tail &&
- (!(dm->dm_dkev && DISPATCH_MACH_KEVENT_ARMED(dm->dm_dkev)) ||
- (dm->ds_atomic_flags & DSF_CANCELED) || dr->dm_disconnect_cnt)) {
- // The channel has pending messages to send.
- return true;
- } else if (dm->ds_atomic_flags & DSF_CANCELED) {
- if (dm->ds_dkev || dm->dm_dkev || dr->dm_send ||
- !TAILQ_EMPTY(&dm->dm_refs->dm_replies) ||
- !dm->dm_cancel_handler_called) {
- // The channel needs to be uninstalled from the manager queue, or
- // the cancellation handler needs to be delivered to the target
- // queue.
- return true;
- }
- }
- // Nothing to do.
- return false;
-}
-
-#pragma mark -
-#pragma mark dispatch_mach_msg_t
-
-dispatch_mach_msg_t
-dispatch_mach_msg_create(mach_msg_header_t *msg, size_t size,
- dispatch_mach_msg_destructor_t destructor, mach_msg_header_t **msg_ptr)
-{
- if (slowpath(size < sizeof(mach_msg_header_t)) ||
- slowpath(destructor && !msg)) {
- DISPATCH_CLIENT_CRASH("Empty message");
- }
- dispatch_mach_msg_t dmsg = _dispatch_alloc(DISPATCH_VTABLE(mach_msg),
- sizeof(struct dispatch_mach_msg_s) +
- (destructor ? 0 : size - sizeof(dmsg->dmsg_msg)));
- if (destructor) {
- dmsg->dmsg_msg = msg;
- } else if (msg) {
- memcpy(dmsg->dmsg_buf, msg, size);
- }
- dmsg->do_next = DISPATCH_OBJECT_LISTLESS;
- dmsg->do_targetq = _dispatch_get_root_queue(_DISPATCH_QOS_CLASS_DEFAULT,
- false);
- dmsg->dmsg_destructor = destructor;
- dmsg->dmsg_size = size;
- if (msg_ptr) {
- *msg_ptr = _dispatch_mach_msg_get_msg(dmsg);
- }
- return dmsg;
-}
-
-void
-_dispatch_mach_msg_dispose(dispatch_mach_msg_t dmsg)
-{
- if (dmsg->dmsg_voucher) {
- _voucher_release(dmsg->dmsg_voucher);
- dmsg->dmsg_voucher = NULL;
- }
- switch (dmsg->dmsg_destructor) {
- case DISPATCH_MACH_MSG_DESTRUCTOR_DEFAULT:
- break;
- case DISPATCH_MACH_MSG_DESTRUCTOR_FREE:
- free(dmsg->dmsg_msg);
- break;
- case DISPATCH_MACH_MSG_DESTRUCTOR_VM_DEALLOCATE: {
- mach_vm_size_t vm_size = dmsg->dmsg_size;
- mach_vm_address_t vm_addr = (uintptr_t)dmsg->dmsg_msg;
- (void)dispatch_assume_zero(mach_vm_deallocate(mach_task_self(),
- vm_addr, vm_size));
- break;
- }}
-}
-
-static inline mach_msg_header_t*
-_dispatch_mach_msg_get_msg(dispatch_mach_msg_t dmsg)
-{
- return dmsg->dmsg_destructor ? dmsg->dmsg_msg :
- (mach_msg_header_t*)dmsg->dmsg_buf;
-}
-
-mach_msg_header_t*
-dispatch_mach_msg_get_msg(dispatch_mach_msg_t dmsg, size_t *size_ptr)
-{
- if (size_ptr) {
- *size_ptr = dmsg->dmsg_size;
- }
- return _dispatch_mach_msg_get_msg(dmsg);
-}
-
-size_t
-_dispatch_mach_msg_debug(dispatch_mach_msg_t dmsg, char* buf, size_t bufsiz)
-{
- size_t offset = 0;
- offset += dsnprintf(&buf[offset], bufsiz - offset, "%s[%p] = { ",
- dx_kind(dmsg), dmsg);
- offset += dsnprintf(&buf[offset], bufsiz - offset, "xrefcnt = 0x%x, "
- "refcnt = 0x%x, ", dmsg->do_xref_cnt + 1, dmsg->do_ref_cnt + 1);
- offset += dsnprintf(&buf[offset], bufsiz - offset, "opts/err = 0x%x, "
- "msgh[%p] = { ", dmsg->do_suspend_cnt, dmsg->dmsg_buf);
- mach_msg_header_t *hdr = _dispatch_mach_msg_get_msg(dmsg);
- if (hdr->msgh_id) {
- offset += dsnprintf(&buf[offset], bufsiz - offset, "id 0x%x, ",
- hdr->msgh_id);
- }
- if (hdr->msgh_size) {
- offset += dsnprintf(&buf[offset], bufsiz - offset, "size %u, ",
- hdr->msgh_size);
- }
- if (hdr->msgh_bits) {
- offset += dsnprintf(&buf[offset], bufsiz - offset, "bits <l %u, r %u",
- MACH_MSGH_BITS_LOCAL(hdr->msgh_bits),
- MACH_MSGH_BITS_REMOTE(hdr->msgh_bits));
- if (MACH_MSGH_BITS_OTHER(hdr->msgh_bits)) {
- offset += dsnprintf(&buf[offset], bufsiz - offset, ", o 0x%x",
- MACH_MSGH_BITS_OTHER(hdr->msgh_bits));
- }
- offset += dsnprintf(&buf[offset], bufsiz - offset, ">, ");
- }
- if (hdr->msgh_local_port && hdr->msgh_remote_port) {
- offset += dsnprintf(&buf[offset], bufsiz - offset, "local 0x%x, "
- "remote 0x%x", hdr->msgh_local_port, hdr->msgh_remote_port);
- } else if (hdr->msgh_local_port) {
- offset += dsnprintf(&buf[offset], bufsiz - offset, "local 0x%x",
- hdr->msgh_local_port);
- } else if (hdr->msgh_remote_port) {
- offset += dsnprintf(&buf[offset], bufsiz - offset, "remote 0x%x",
- hdr->msgh_remote_port);
- } else {
- offset += dsnprintf(&buf[offset], bufsiz - offset, "no ports");
- }
- offset += dsnprintf(&buf[offset], bufsiz - offset, " } }");
- return offset;
-}
-
-#pragma mark -
-#pragma mark dispatch_mig_server
-
-mach_msg_return_t
-dispatch_mig_server(dispatch_source_t ds, size_t maxmsgsz,
- dispatch_mig_callback_t callback)
-{
- mach_msg_options_t options = MACH_RCV_MSG | MACH_RCV_TIMEOUT
- | MACH_RCV_TRAILER_ELEMENTS(MACH_RCV_TRAILER_CTX)
- | MACH_RCV_TRAILER_TYPE(MACH_MSG_TRAILER_FORMAT_0) | MACH_RCV_VOUCHER;
- mach_msg_options_t tmp_options;
- mig_reply_error_t *bufTemp, *bufRequest, *bufReply;
- mach_msg_return_t kr = 0;
- uint64_t assertion_token = 0;
- unsigned int cnt = 1000; // do not stall out serial queues
- boolean_t demux_success;
- bool received = false;
- size_t rcv_size = maxmsgsz + MAX_TRAILER_SIZE;
-
- // XXX FIXME -- allocate these elsewhere
- bufRequest = alloca(rcv_size);
- bufReply = alloca(rcv_size);
- bufReply->Head.msgh_size = 0;
- bufRequest->RetCode = 0;
-
-#if DISPATCH_DEBUG
- options |= MACH_RCV_LARGE; // rdar://problem/8422992
-#endif
- tmp_options = options;
- // XXX FIXME -- change this to not starve out the target queue
- for (;;) {
- if (DISPATCH_OBJECT_SUSPENDED(ds) || (--cnt == 0)) {
- options &= ~MACH_RCV_MSG;
- tmp_options &= ~MACH_RCV_MSG;
-
- if (!(tmp_options & MACH_SEND_MSG)) {
- goto out;
- }
- }
- kr = mach_msg(&bufReply->Head, tmp_options, bufReply->Head.msgh_size,
- (mach_msg_size_t)rcv_size, (mach_port_t)ds->ds_ident_hack, 0,0);
-
- tmp_options = options;
-
- if (slowpath(kr)) {
- switch (kr) {
- case MACH_SEND_INVALID_DEST:
- case MACH_SEND_TIMED_OUT:
- if (bufReply->Head.msgh_bits & MACH_MSGH_BITS_COMPLEX) {
- mach_msg_destroy(&bufReply->Head);
- }
- break;
- case MACH_RCV_TIMED_OUT:
- // Don't return an error if a message was sent this time or
- // a message was successfully received previously
- // rdar://problems/7363620&7791738
- if(bufReply->Head.msgh_remote_port || received) {
- kr = MACH_MSG_SUCCESS;
- }
- break;
- case MACH_RCV_INVALID_NAME:
- break;
-#if DISPATCH_DEBUG
- case MACH_RCV_TOO_LARGE:
- // receive messages that are too large and log their id and size
- // rdar://problem/8422992
- tmp_options &= ~MACH_RCV_LARGE;
- size_t large_size = bufReply->Head.msgh_size + MAX_TRAILER_SIZE;
- void *large_buf = malloc(large_size);
- if (large_buf) {
- rcv_size = large_size;
- bufReply = large_buf;
- }
- if (!mach_msg(&bufReply->Head, tmp_options, 0,
- (mach_msg_size_t)rcv_size,
- (mach_port_t)ds->ds_ident_hack, 0, 0)) {
- _dispatch_log("BUG in libdispatch client: "
- "dispatch_mig_server received message larger than "
- "requested size %zd: id = 0x%x, size = %d",
- maxmsgsz, bufReply->Head.msgh_id,
- bufReply->Head.msgh_size);
- }
- if (large_buf) {
- free(large_buf);
- }
- // fall through
-#endif
- default:
- _dispatch_bug_mach_client(
- "dispatch_mig_server: mach_msg() failed", kr);
- break;
- }
- goto out;
- }
-
- if (!(tmp_options & MACH_RCV_MSG)) {
- goto out;
- }
-
- if (assertion_token) {
-#if DISPATCH_USE_IMPORTANCE_ASSERTION
- int r = proc_importance_assertion_complete(assertion_token);
- (void)dispatch_assume_zero(r);
-#endif
- assertion_token = 0;
- }
- received = true;
-
- bufTemp = bufRequest;
- bufRequest = bufReply;
- bufReply = bufTemp;
-
-#if DISPATCH_USE_IMPORTANCE_ASSERTION
- int r = proc_importance_assertion_begin_with_msg(&bufRequest->Head,
- NULL, &assertion_token);
- if (r && slowpath(r != EIO)) {
- (void)dispatch_assume_zero(r);
- }
-#endif
- _voucher_replace(voucher_create_with_mach_msg(&bufRequest->Head));
- demux_success = callback(&bufRequest->Head, &bufReply->Head);
-
- if (!demux_success) {
- // destroy the request - but not the reply port
- bufRequest->Head.msgh_remote_port = 0;
- mach_msg_destroy(&bufRequest->Head);
- } else if (!(bufReply->Head.msgh_bits & MACH_MSGH_BITS_COMPLEX)) {
- // if MACH_MSGH_BITS_COMPLEX is _not_ set, then bufReply->RetCode
- // is present
- if (slowpath(bufReply->RetCode)) {
- if (bufReply->RetCode == MIG_NO_REPLY) {
- continue;
- }
-
- // destroy the request - but not the reply port
- bufRequest->Head.msgh_remote_port = 0;
- mach_msg_destroy(&bufRequest->Head);
- }
- }
-
- if (bufReply->Head.msgh_remote_port) {
- tmp_options |= MACH_SEND_MSG;
- if (MACH_MSGH_BITS_REMOTE(bufReply->Head.msgh_bits) !=
- MACH_MSG_TYPE_MOVE_SEND_ONCE) {
- tmp_options |= MACH_SEND_TIMEOUT;
- }
- }
- }
-
-out:
- if (assertion_token) {
-#if DISPATCH_USE_IMPORTANCE_ASSERTION
- int r = proc_importance_assertion_complete(assertion_token);
- (void)dispatch_assume_zero(r);
-#endif
- }
-
- return kr;
-}
-
-#endif /* HAVE_MACH */
-
-#pragma mark -
-#pragma mark dispatch_source_debug
-
-DISPATCH_NOINLINE
-static const char *
-_evfiltstr(short filt)
-{
- switch (filt) {
-#define _evfilt2(f) case (f): return #f
- _evfilt2(EVFILT_READ);
- _evfilt2(EVFILT_WRITE);
- _evfilt2(EVFILT_AIO);
- _evfilt2(EVFILT_VNODE);
- _evfilt2(EVFILT_PROC);
- _evfilt2(EVFILT_SIGNAL);
- _evfilt2(EVFILT_TIMER);
-#ifdef EVFILT_VM
- _evfilt2(EVFILT_VM);
-#endif
-#ifdef EVFILT_MEMORYSTATUS
- _evfilt2(EVFILT_MEMORYSTATUS);
-#endif
-#if HAVE_MACH
- _evfilt2(EVFILT_MACHPORT);
- _evfilt2(DISPATCH_EVFILT_MACH_NOTIFICATION);
-#endif
- _evfilt2(EVFILT_FS);
- _evfilt2(EVFILT_USER);
-
- _evfilt2(DISPATCH_EVFILT_TIMER);
- _evfilt2(DISPATCH_EVFILT_CUSTOM_ADD);
- _evfilt2(DISPATCH_EVFILT_CUSTOM_OR);
- default:
- return "EVFILT_missing";
- }
-}
-
-static size_t
-_dispatch_source_debug_attr(dispatch_source_t ds, char* buf, size_t bufsiz)
-{
- dispatch_queue_t target = ds->do_targetq;
- return dsnprintf(buf, bufsiz, "target = %s[%p], ident = 0x%lx, "
- "pending_data = 0x%lx, pending_data_mask = 0x%lx, ",
- target && target->dq_label ? target->dq_label : "", target,
- ds->ds_ident_hack, ds->ds_pending_data, ds->ds_pending_data_mask);
-}
-
-static size_t
-_dispatch_timer_debug_attr(dispatch_source_t ds, char* buf, size_t bufsiz)
-{
- dispatch_source_refs_t dr = ds->ds_refs;
- return dsnprintf(buf, bufsiz, "timer = { target = 0x%llx, deadline = 0x%llx,"
- " last_fire = 0x%llx, interval = 0x%llx, flags = 0x%lx }, ",
- ds_timer(dr).target, ds_timer(dr).deadline, ds_timer(dr).last_fire,
- ds_timer(dr).interval, ds_timer(dr).flags);
-}
-
-size_t
-_dispatch_source_debug(dispatch_source_t ds, char* buf, size_t bufsiz)
-{
- size_t offset = 0;
- offset += dsnprintf(&buf[offset], bufsiz - offset, "%s[%p] = { ",
- dx_kind(ds), ds);
- offset += _dispatch_object_debug_attr(ds, &buf[offset], bufsiz - offset);
- offset += _dispatch_source_debug_attr(ds, &buf[offset], bufsiz - offset);
- if (ds->ds_is_timer) {
- offset += _dispatch_timer_debug_attr(ds, &buf[offset], bufsiz - offset);
- }
- offset += dsnprintf(&buf[offset], bufsiz - offset, "filter = %s }",
- ds->ds_dkev ? _evfiltstr(ds->ds_dkev->dk_kevent.filter) : "????");
- return offset;
-}
-
-static size_t
-_dispatch_mach_debug_attr(dispatch_mach_t dm, char* buf, size_t bufsiz)
-{
- dispatch_queue_t target = dm->do_targetq;
- return dsnprintf(buf, bufsiz, "target = %s[%p], receive = 0x%x, "
- "send = 0x%x, send-possible = 0x%x%s, checkin = 0x%x%s, "
- "sending = %d, disconnected = %d, canceled = %d ",
- target && target->dq_label ? target->dq_label : "", target,
- dm->ds_dkev ?(mach_port_t)dm->ds_dkev->dk_kevent.ident:0,
- dm->dm_refs->dm_send,
- dm->dm_dkev ?(mach_port_t)dm->dm_dkev->dk_kevent.ident:0,
- dm->dm_dkev && DISPATCH_MACH_KEVENT_ARMED(dm->dm_dkev) ?
- " (armed)" : "", dm->dm_refs->dm_checkin_port,
- dm->dm_refs->dm_checkin ? " (pending)" : "",
- dm->dm_refs->dm_sending, dm->dm_refs->dm_disconnect_cnt,
- (bool)(dm->ds_atomic_flags & DSF_CANCELED));
-}
-size_t
-_dispatch_mach_debug(dispatch_mach_t dm, char* buf, size_t bufsiz)
-{
- size_t offset = 0;
- offset += dsnprintf(&buf[offset], bufsiz - offset, "%s[%p] = { ",
- dm->dq_label && !dm->dm_cancel_handler_called ? dm->dq_label :
- dx_kind(dm), dm);
- offset += _dispatch_object_debug_attr(dm, &buf[offset], bufsiz - offset);
- offset += _dispatch_mach_debug_attr(dm, &buf[offset], bufsiz - offset);
- offset += dsnprintf(&buf[offset], bufsiz - offset, "}");
- return offset;
-}
-
-#if DISPATCH_DEBUG
-static void
-_dispatch_kevent_debug(struct kevent64_s* kev, const char* str)
-{
- _dispatch_log("kevent[%p] = { ident = 0x%llx, filter = %s, flags = 0x%x, "
- "fflags = 0x%x, data = 0x%llx, udata = 0x%llx, ext[0] = 0x%llx, "
- "ext[1] = 0x%llx }: %s", kev, kev->ident, _evfiltstr(kev->filter),
- kev->flags, kev->fflags, kev->data, kev->udata, kev->ext[0],
- kev->ext[1], str);
-}
-
-static void
-_dispatch_kevent_debugger2(void *context)
-{
- struct sockaddr sa;
- socklen_t sa_len = sizeof(sa);
- int c, fd = (int)(long)context;
- unsigned int i;
- dispatch_kevent_t dk;
- dispatch_source_t ds;
- dispatch_source_refs_t dr;
- FILE *debug_stream;
-
- c = accept(fd, &sa, &sa_len);
- if (c == -1) {
- if (errno != EAGAIN) {
- (void)dispatch_assume_zero(errno);
- }
- return;
- }
-#if 0
- int r = fcntl(c, F_SETFL, 0); // disable non-blocking IO
- if (r == -1) {
- (void)dispatch_assume_zero(errno);
- }
-#endif
- debug_stream = fdopen(c, "a");
- if (!dispatch_assume(debug_stream)) {
- close(c);
- return;
- }
-
- fprintf(debug_stream, "HTTP/1.0 200 OK\r\n");
- fprintf(debug_stream, "Content-type: text/html\r\n");
- fprintf(debug_stream, "Pragma: nocache\r\n");
- fprintf(debug_stream, "\r\n");
- fprintf(debug_stream, "<html>\n");
- fprintf(debug_stream, "<head><title>PID %u</title></head>\n", getpid());
- fprintf(debug_stream, "<body>\n<ul>\n");
-
- //fprintf(debug_stream, "<tr><td>DK</td><td>DK</td><td>DK</td><td>DK</td>"
- // "<td>DK</td><td>DK</td><td>DK</td></tr>\n");
-
- for (i = 0; i < DSL_HASH_SIZE; i++) {
- if (TAILQ_EMPTY(&_dispatch_sources[i])) {
- continue;
- }
- TAILQ_FOREACH(dk, &_dispatch_sources[i], dk_list) {
- fprintf(debug_stream, "\t<br><li>DK %p ident %lu filter %s flags "
- "0x%hx fflags 0x%x data 0x%lx udata %p\n",
- dk, (unsigned long)dk->dk_kevent.ident,
- _evfiltstr(dk->dk_kevent.filter), dk->dk_kevent.flags,
- dk->dk_kevent.fflags, (unsigned long)dk->dk_kevent.data,
- (void*)dk->dk_kevent.udata);
- fprintf(debug_stream, "\t\t<ul>\n");
- TAILQ_FOREACH(dr, &dk->dk_sources, dr_list) {
- ds = _dispatch_source_from_refs(dr);
- fprintf(debug_stream, "\t\t\t<li>DS %p refcnt 0x%x suspend "
- "0x%x data 0x%lx mask 0x%lx flags 0x%x</li>\n",
- ds, ds->do_ref_cnt + 1, ds->do_suspend_cnt,
- ds->ds_pending_data, ds->ds_pending_data_mask,
- ds->ds_atomic_flags);
- if (ds->do_suspend_cnt == DISPATCH_OBJECT_SUSPEND_LOCK) {
- dispatch_queue_t dq = ds->do_targetq;
- fprintf(debug_stream, "\t\t<br>DQ: %p refcnt 0x%x suspend "
- "0x%x label: %s\n", dq, dq->do_ref_cnt + 1,
- dq->do_suspend_cnt, dq->dq_label ? dq->dq_label:"");
- }
- }
- fprintf(debug_stream, "\t\t</ul>\n");
- fprintf(debug_stream, "\t</li>\n");
- }
- }
- fprintf(debug_stream, "</ul>\n</body>\n</html>\n");
- fflush(debug_stream);
- fclose(debug_stream);
-}
-
-static void
-_dispatch_kevent_debugger2_cancel(void *context)
-{
- int ret, fd = (int)(long)context;
-
- ret = close(fd);
- if (ret != -1) {
- (void)dispatch_assume_zero(errno);
- }
-}
-
-static void
-_dispatch_kevent_debugger(void *context DISPATCH_UNUSED)
-{
- union {
- struct sockaddr_in sa_in;
- struct sockaddr sa;
- } sa_u = {
- .sa_in = {
- .sin_family = AF_INET,
- .sin_addr = { htonl(INADDR_LOOPBACK), },
- },
- };
- dispatch_source_t ds;
- const char *valstr;
- int val, r, fd, sock_opt = 1;
- socklen_t slen = sizeof(sa_u);
-
- if (issetugid()) {
- return;
- }
- valstr = getenv("LIBDISPATCH_DEBUGGER");
- if (!valstr) {
- return;
- }
- val = atoi(valstr);
- if (val == 2) {
- sa_u.sa_in.sin_addr.s_addr = 0;
- }
- fd = socket(PF_INET, SOCK_STREAM, 0);
- if (fd == -1) {
- (void)dispatch_assume_zero(errno);
- return;
- }
- r = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (void *)&sock_opt,
- (socklen_t) sizeof sock_opt);
- if (r == -1) {
- (void)dispatch_assume_zero(errno);
- goto out_bad;
- }
-#if 0
- r = fcntl(fd, F_SETFL, O_NONBLOCK);
- if (r == -1) {
- (void)dispatch_assume_zero(errno);
- goto out_bad;
- }
-#endif
- r = bind(fd, &sa_u.sa, sizeof(sa_u));
- if (r == -1) {
- (void)dispatch_assume_zero(errno);
- goto out_bad;
- }
- r = listen(fd, SOMAXCONN);
- if (r == -1) {
- (void)dispatch_assume_zero(errno);
- goto out_bad;
- }
- r = getsockname(fd, &sa_u.sa, &slen);
- if (r == -1) {
- (void)dispatch_assume_zero(errno);
- goto out_bad;
- }
-
- ds = dispatch_source_create(DISPATCH_SOURCE_TYPE_READ, (uintptr_t)fd, 0,
- &_dispatch_mgr_q);
- if (dispatch_assume(ds)) {
- _dispatch_log("LIBDISPATCH: debug port: %hu",
- (in_port_t)ntohs(sa_u.sa_in.sin_port));
-
- /* ownership of fd transfers to ds */
- dispatch_set_context(ds, (void *)(long)fd);
- dispatch_source_set_event_handler_f(ds, _dispatch_kevent_debugger2);
- dispatch_source_set_cancel_handler_f(ds,
- _dispatch_kevent_debugger2_cancel);
- dispatch_resume(ds);
-
- return;
- }
-out_bad:
- close(fd);
-}
-
-#if HAVE_MACH
-
-#ifndef MACH_PORT_TYPE_SPREQUEST
-#define MACH_PORT_TYPE_SPREQUEST 0x40000000
-#endif
-
-DISPATCH_NOINLINE
-void
-dispatch_debug_machport(mach_port_t name, const char* str)
-{
- mach_port_type_t type;
- mach_msg_bits_t ns = 0, nr = 0, nso = 0, nd = 0;
- unsigned int dnreqs = 0, dnrsiz;
- kern_return_t kr = mach_port_type(mach_task_self(), name, &type);
- if (kr) {
- _dispatch_log("machport[0x%08x] = { error(0x%x) \"%s\" }: %s", name,
- kr, mach_error_string(kr), str);
- return;
- }
- if (type & MACH_PORT_TYPE_SEND) {
- (void)dispatch_assume_zero(mach_port_get_refs(mach_task_self(), name,
- MACH_PORT_RIGHT_SEND, &ns));
- }
- if (type & MACH_PORT_TYPE_SEND_ONCE) {
- (void)dispatch_assume_zero(mach_port_get_refs(mach_task_self(), name,
- MACH_PORT_RIGHT_SEND_ONCE, &nso));
- }
- if (type & MACH_PORT_TYPE_DEAD_NAME) {
- (void)dispatch_assume_zero(mach_port_get_refs(mach_task_self(), name,
- MACH_PORT_RIGHT_DEAD_NAME, &nd));
- }
- if (type & (MACH_PORT_TYPE_RECEIVE|MACH_PORT_TYPE_SEND)) {
- kr = mach_port_dnrequest_info(mach_task_self(), name, &dnrsiz, &dnreqs);
- if (kr != KERN_INVALID_RIGHT) (void)dispatch_assume_zero(kr);
- }
- if (type & MACH_PORT_TYPE_RECEIVE) {
- mach_port_status_t status = { .mps_pset = 0, };
- mach_msg_type_number_t cnt = MACH_PORT_RECEIVE_STATUS_COUNT;
- (void)dispatch_assume_zero(mach_port_get_refs(mach_task_self(), name,
- MACH_PORT_RIGHT_RECEIVE, &nr));
- (void)dispatch_assume_zero(mach_port_get_attributes(mach_task_self(),
- name, MACH_PORT_RECEIVE_STATUS, (void*)&status, &cnt));
- _dispatch_log("machport[0x%08x] = { R(%03u) S(%03u) SO(%03u) D(%03u) "
- "dnreqs(%03u) spreq(%s) nsreq(%s) pdreq(%s) srights(%s) "
- "sorights(%03u) qlim(%03u) msgcount(%03u) mkscount(%03u) "
- "seqno(%03u) }: %s", name, nr, ns, nso, nd, dnreqs,
- type & MACH_PORT_TYPE_SPREQUEST ? "Y":"N",
- status.mps_nsrequest ? "Y":"N", status.mps_pdrequest ? "Y":"N",
- status.mps_srights ? "Y":"N", status.mps_sorights,
- status.mps_qlimit, status.mps_msgcount, status.mps_mscount,
- status.mps_seqno, str);
- } else if (type & (MACH_PORT_TYPE_SEND|MACH_PORT_TYPE_SEND_ONCE|
- MACH_PORT_TYPE_DEAD_NAME)) {
- _dispatch_log("machport[0x%08x] = { R(%03u) S(%03u) SO(%03u) D(%03u) "
- "dnreqs(%03u) spreq(%s) }: %s", name, nr, ns, nso, nd, dnreqs,
- type & MACH_PORT_TYPE_SPREQUEST ? "Y":"N", str);
- } else {
- _dispatch_log("machport[0x%08x] = { type(0x%08x) }: %s", name, type,
- str);
- }
-}
-
-#endif // HAVE_MACH
-
-#endif // DISPATCH_DEBUG
projects / apple / libdispatch.git / blobdiff