Skip to content

Slice A: shared-code seams for a cooperative single-threaded event backend #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
scottmarchant wants to merge 1 commit into from
Closed

Slice A: shared-code seams for a cooperative single-threaded event backend #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions src/event/event_internal.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -699,6 +699,26 @@ void _dispatch_event_loop_timer_delete(dispatch_timer_heap_t dth, uint32_t tidx)

void _dispatch_event_loop_drain_timers(dispatch_timer_heap_t dth, uint32_t count);

#if DISPATCH_EVENT_BACKEND_WASI
// Bracket a caller-held critical section entered outside any drain (an
// inline-executed sync body, a once initializer, an object dispose, the
// specifics-hash mutation): pokes inside only record pending work, and the
// outermost undefer flushes it.
void _dispatch_wasi_defer_pokes(void);
void _dispatch_wasi_undefer_pokes(void);
#endif // DISPATCH_EVENT_BACKEND_WASI

// No-op on threaded platforms: pokes there wake other workers and never run
// client code on the submitting stack, so critical sections need no bracket.
// A cooperative single-threaded event backend defines the real versions.
#if DISPATCH_EVENT_BACKEND_WASI
#define _dispatch_cooperative_pokes_defer() _dispatch_wasi_defer_pokes()
#define _dispatch_cooperative_pokes_undefer() _dispatch_wasi_undefer_pokes()
#else
#define _dispatch_cooperative_pokes_defer() ((void)0)
#define _dispatch_cooperative_pokes_undefer() ((void)0)
#endif

DISPATCH_ALWAYS_INLINE
static inline void
_dispatch_timers_heap_dirty(dispatch_timer_heap_t dth, uint32_t tidx)
Expand Down
6 changes: 6 additions & 0 deletions src/object.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -230,13 +230,19 @@ _dispatch_dispose(dispatch_object_t dou)
tq = _dispatch_get_root_queue(DISPATCH_QOS_DEFAULT, false)->_as_dq;
}

// Dispose paths submit detached work (e.g. queue-specific destructor
// batches) while the object is partially torn down: on cooperative
// single-threaded targets those pokes must defer until the object is
// fully disposed and freed rather than run client code mid-dispose.
_dispatch_cooperative_pokes_defer();
dx_dispose(dou._do, &allow_free);

// Past this point, the only thing left of the object is its memory
if (likely(allow_free)) {
_dispatch_object_finalize(dou);
_dispatch_object_dealloc(dou);
}
_dispatch_cooperative_pokes_undefer();
if (func && ctxt) {
dispatch_async_f(tq, ctxt, func);
}
Expand Down
6 changes: 6 additions & 0 deletions src/once.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,8 +43,14 @@ static void
_dispatch_once_callout(dispatch_once_gate_t l, void *ctxt,
dispatch_function_t func)
{
// The initializer runs with the once gate held: on cooperative
// single-threaded targets, pokes from inside it must defer to after the
// gate is broadcast (a drained item re-entering this dispatch_once would
// otherwise crash where threaded platforms simply wait).
_dispatch_cooperative_pokes_defer();
_dispatch_client_callout(ctxt, func);
_dispatch_once_gate_broadcast(l);
_dispatch_cooperative_pokes_undefer();
}

DISPATCH_NOINLINE
Expand Down
63 changes: 58 additions & 5 deletions src/queue.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1718,7 +1718,11 @@ _dispatch_barrier_trysync_or_async_f(dispatch_lane_t dq, void *ctxt,
if (flags & DISPATCH_BARRIER_TRYSYNC_SUSPEND) {
_dispatch_retain_2(dq); // see _dispatch_lane_suspend
}
// the invoke runs with the barrier lock held; see
// _dispatch_barrier_sync_f_inline
_dispatch_cooperative_pokes_defer();
_dispatch_barrier_trysync_or_async_f_complete(dq, ctxt, func, flags);
_dispatch_cooperative_pokes_undefer();
}

#pragma mark -
Expand Down Expand Up @@ -1794,7 +1798,7 @@ _dispatch_sync_recurse(dispatch_lane_t dq, void *ctxt,

DISPATCH_ALWAYS_INLINE
static inline void
_dispatch_barrier_sync_f_inline(dispatch_queue_t dq, void *ctxt,
_dispatch_barrier_sync_f_inline_impl(dispatch_queue_t dq, void *ctxt,
dispatch_function_t func, uintptr_t dc_flags)
{
dispatch_tid tid = _dispatch_tid_self();
Expand Down Expand Up @@ -1828,6 +1832,19 @@ _dispatch_barrier_sync_f_inline(dispatch_queue_t dq, void *ctxt,
dq, ctxt, func, dc_flags | DC_FLAG_BARRIER)));
}

DISPATCH_ALWAYS_INLINE
static inline void
_dispatch_barrier_sync_f_inline(dispatch_queue_t dq, void *ctxt,
dispatch_function_t func, uintptr_t dc_flags)
{
// The body (and thus the client callout, on the inline paths) runs with
// the queue's barrier lock held: on cooperative single-threaded targets,
// pokes from inside it must defer to after the sync completes.
_dispatch_cooperative_pokes_defer();
_dispatch_barrier_sync_f_inline_impl(dq, ctxt, func, dc_flags);
_dispatch_cooperative_pokes_undefer();
}

DISPATCH_NOINLINE
static void
_dispatch_barrier_sync_f(dispatch_queue_t dq, void *ctxt,
Expand All @@ -1846,7 +1863,7 @@ dispatch_barrier_sync_f(dispatch_queue_t dq, void *ctxt,

DISPATCH_ALWAYS_INLINE
static inline void
_dispatch_sync_f_inline(dispatch_queue_t dq, void *ctxt,
_dispatch_sync_f_inline_impl(dispatch_queue_t dq, void *ctxt,
dispatch_function_t func, uintptr_t dc_flags)
{
if (likely(dq->dq_width == 1)) {
Expand All @@ -1872,6 +1889,17 @@ _dispatch_sync_f_inline(dispatch_queue_t dq, void *ctxt,
_dispatch_trace_item_sync_push_pop(dq, ctxt, func, dc_flags)));
}

DISPATCH_ALWAYS_INLINE
static inline void
_dispatch_sync_f_inline(dispatch_queue_t dq, void *ctxt,
dispatch_function_t func, uintptr_t dc_flags)
{
// see _dispatch_barrier_sync_f_inline
_dispatch_cooperative_pokes_defer();
_dispatch_sync_f_inline_impl(dq, ctxt, func, dc_flags);
_dispatch_cooperative_pokes_undefer();
}

DISPATCH_NOINLINE
static void
_dispatch_sync_f(dispatch_queue_t dq, void *ctxt, dispatch_function_t func,
Expand Down Expand Up @@ -2099,7 +2127,11 @@ _dispatch_async_and_wait_f(dispatch_queue_t dq,
.dsc_waiter = tid,
};

return _dispatch_async_and_wait_recurse(dq, &dsc, tid, dc_flags);
// see _dispatch_barrier_sync_f_inline: the invoke can run inline with
// the acquired width/barrier held
_dispatch_cooperative_pokes_defer();
_dispatch_async_and_wait_recurse(dq, &dsc, tid, dc_flags);
_dispatch_cooperative_pokes_undefer();
}

DISPATCH_NOINLINE
Expand Down Expand Up @@ -2175,7 +2207,10 @@ _dispatch_async_and_wait_block_with_privdata(dispatch_queue_t dq,
.dsc_waiter = tid,
};

return _dispatch_async_and_wait_recurse(dq, &dsc, tid, dc_flags);
// see _dispatch_async_and_wait_f
_dispatch_cooperative_pokes_defer();
_dispatch_async_and_wait_recurse(dq, &dsc, tid, dc_flags);
_dispatch_cooperative_pokes_undefer();
}

void
Expand Down Expand Up @@ -2318,6 +2353,10 @@ dispatch_queue_set_specific(dispatch_queue_t dq, const void *key,
return;
}

// On cooperative single-threaded targets the destructor push below could
// otherwise run the client destructor on this stack while dqsh_lock is
// held; defer pokes across the critical section so it runs after unlock.
_dispatch_cooperative_pokes_defer();
_dispatch_unfair_lock_lock(&dqsh->dqsh_lock);
dqs = _dispatch_queue_specific_find(dqsh, key);
if (dqs) {
Expand All @@ -2341,6 +2380,7 @@ dispatch_queue_set_specific(dispatch_queue_t dq, const void *key,
}

_dispatch_unfair_lock_unlock(&dqsh->dqsh_lock);
_dispatch_cooperative_pokes_undefer();
}

DISPATCH_ALWAYS_INLINE
Expand Down Expand Up @@ -6877,6 +6917,13 @@ _dispatch_main_queue_update_priority_from_thread(void)
}
}

#endif // DISPATCH_COCOA_COMPAT
#if DISPATCH_COCOA_COMPAT || defined(__wasi__)
// Shared between the CFRunLoop callback path (DISPATCH_COCOA_COMPAT) and a
// cooperative single-threaded drain, which owns the thread-bound main
// queue's drain lock for the lifetime of the program. The runloop-handle
// initialization and the thread-QoS override propagation are runloop/Darwin
// machinery and compile only for COCOA_COMPAT.
static void
_dispatch_main_queue_drain(dispatch_queue_main_t dq)
{
Expand All @@ -6898,8 +6945,10 @@ _dispatch_main_queue_drain(dispatch_queue_main_t dq)
" from the wrong thread");
}

#if DISPATCH_COCOA_COMPAT
dispatch_once_f(&_dispatch_main_q_handle_pred, dq,
_dispatch_runloop_queue_handle_init);
#endif

// <rdar://problem/23256682> hide the frame chaining when CFRunLoop
// drains the main runloop, as this should not be observable that way
Expand All @@ -6908,12 +6957,14 @@ _dispatch_main_queue_drain(dispatch_queue_main_t dq)

pthread_priority_t pp = _dispatch_get_priority();
dispatch_priority_t pri = _dispatch_priority_from_pp(pp);
dispatch_qos_t qos = _dispatch_priority_qos(pri);
voucher_t voucher = _voucher_copy();

#if DISPATCH_COCOA_COMPAT
dispatch_qos_t qos = _dispatch_priority_qos(pri);
if (unlikely(qos != _dispatch_priority_qos(dq->dq_priority))) {
_dispatch_main_queue_update_priority_from_thread();
}
#endif
dispatch_priority_t old_dbp = _dispatch_set_basepri(pri);
_dispatch_set_basepri_override_qos(DISPATCH_QOS_SATURATED);

Expand All @@ -6936,6 +6987,8 @@ _dispatch_main_queue_drain(dispatch_queue_main_t dq)
_dispatch_force_cache_cleanup();
_dispatch_perfmon_end_notrace();
}
#endif // DISPATCH_COCOA_COMPAT || defined(__wasi__)
#if DISPATCH_COCOA_COMPAT

static bool
_dispatch_runloop_queue_drain_one(dispatch_lane_t dq)
Expand Down