[apple/xnu.git] / osfmk / kern / mpsc_queue.c

/*
 * Copyright (c) 2018 Apple Inc. All rights reserved.
 *
 * @APPLE_OSREFERENCE_LICENSE_HEADER_START@
 *
 * This file contains Original Code and/or Modifications of Original Code
 * as defined in and that are subject to the Apple Public Source License
 * Version 2.0 (the 'License'). You may not use this file except in
 * compliance with the License. The rights granted to you under the License
 * may not be used to create, or enable the creation or redistribution of,
 * unlawful or unlicensed copies of an Apple operating system, or to
 * circumvent, violate, or enable the circumvention or violation of, any
 * terms of an Apple operating system software license agreement.
 *
 * Please obtain a copy of the License at
 * http://www.opensource.apple.com/apsl/ and read it before using this file.
 *
 * The Original Code and all software distributed under the License are
 * distributed on an 'AS IS' basis, WITHOUT WARRANTY OF ANY KIND, EITHER
 * EXPRESS OR IMPLIED, AND APPLE HEREBY DISCLAIMS ALL SUCH WARRANTIES,
 * INCLUDING WITHOUT LIMITATION, ANY WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE, QUIET ENJOYMENT OR NON-INFRINGEMENT.
 * Please see the License for the specific language governing rights and
 * limitations under the License.
 *
 * @APPLE_OSREFERENCE_LICENSE_HEADER_END@
 */

#include <machine/machine_cpu.h>
#include <kern/locks.h>
#include <kern/mpsc_queue.h>
#include <kern/thread.h>

#pragma mark Single Consumer calls

__attribute__((noinline))
static mpsc_queue_chain_t
_mpsc_queue_wait_for_enqueuer(struct mpsc_queue_chain *_Atomic *ptr)
{
	return hw_wait_while_equals((void **)ptr, NULL);
}

void
mpsc_queue_restore_batch(mpsc_queue_head_t q, mpsc_queue_chain_t first,
    mpsc_queue_chain_t last)
{
	mpsc_queue_chain_t head = os_atomic_load(&q->mpqh_head.mpqc_next, relaxed);

	os_atomic_store(&last->mpqc_next, head, relaxed);

	if (head == NULL &&
	    !os_atomic_cmpxchg(&q->mpqh_tail, &q->mpqh_head, last, release)) {
		head = os_atomic_load(&q->mpqh_head.mpqc_next, relaxed);
		if (__improbable(head == NULL)) {
			head = _mpsc_queue_wait_for_enqueuer(&q->mpqh_head.mpqc_next);
		}
		os_atomic_store(&last->mpqc_next, head, relaxed);
	}

	os_atomic_store(&q->mpqh_head.mpqc_next, first, relaxed);
}

mpsc_queue_chain_t
mpsc_queue_dequeue_batch(mpsc_queue_head_t q, mpsc_queue_chain_t *tail_out,
    os_atomic_dependency_t dependency)
{
	mpsc_queue_chain_t head, tail;

	q = os_atomic_inject_dependency(q, dependency);

	tail = os_atomic_load(&q->mpqh_tail, relaxed);
	if (__improbable(tail == &q->mpqh_head)) {
		*tail_out = NULL;
		return NULL;
	}

	head = os_atomic_load(&q->mpqh_head.mpqc_next, relaxed);
	if (__improbable(head == NULL)) {
		head = _mpsc_queue_wait_for_enqueuer(&q->mpqh_head.mpqc_next);
	}
	os_atomic_store(&q->mpqh_head.mpqc_next, NULL, relaxed);
	/*
	 * 22708742: set tail to &q->mpqh_head with release, so that NULL write
	 * to head above doesn't clobber the head set by concurrent enqueuer
	 *
	 * The other half of the seq_cst is required to pair with any enqueuer that
	 * contributed to an element in this list (pairs with the release fence in
	 * __mpsc_queue_append_update_tail().
	 *
	 * Making this seq_cst instead of acq_rel makes mpsc_queue_append*()
	 * visibility transitive (when items hop from one queue to the next)
	 * which is expected by clients implicitly.
	 *
	 * Note that this is the same number of fences that a traditional lock
	 * would have, but as a once-per-batch cost.
	 */
	*tail_out = os_atomic_xchg(&q->mpqh_tail, &q->mpqh_head, seq_cst);

	return head;
}

mpsc_queue_chain_t
mpsc_queue_batch_next(mpsc_queue_chain_t cur, mpsc_queue_chain_t tail)
{
	mpsc_queue_chain_t elm = NULL;
	if (cur == tail || cur == NULL) {
		return elm;
	}

	elm = os_atomic_load(&cur->mpqc_next, relaxed);
	if (__improbable(elm == NULL)) {
		elm = _mpsc_queue_wait_for_enqueuer(&cur->mpqc_next);
	}
	return elm;
}

#pragma mark "GCD"-like facilities

static void _mpsc_daemon_queue_drain(mpsc_daemon_queue_t, thread_t);
static void _mpsc_daemon_queue_enqueue(mpsc_daemon_queue_t, mpsc_queue_chain_t);

/* thread based queues */

static void
_mpsc_queue_thread_continue(void *param, wait_result_t wr __unused)
{
	mpsc_daemon_queue_t dq = param;
	mpsc_daemon_queue_kind_t kind = dq->mpd_kind;
	thread_t self = dq->mpd_thread;

	__builtin_assume(self != THREAD_NULL);

	if (kind == MPSC_QUEUE_KIND_THREAD_CRITICAL) {
		self->options |= TH_OPT_SYSTEM_CRITICAL;
	}

	assert(dq->mpd_thread == current_thread());
	_mpsc_daemon_queue_drain(dq, self);

	if (kind == MPSC_QUEUE_KIND_THREAD_CRITICAL) {
		self->options &= ~TH_OPT_SYSTEM_CRITICAL;
	}

	thread_block_parameter(_mpsc_queue_thread_continue, dq);
}

static void
_mpsc_queue_thread_wakeup(mpsc_daemon_queue_t dq)
{
	thread_wakeup_thread((event_t)dq, dq->mpd_thread);
}

static kern_return_t
_mpsc_daemon_queue_init_with_thread(mpsc_daemon_queue_t dq,
    mpsc_daemon_invoke_fn_t invoke, int pri, const char *name,
    mpsc_daemon_queue_kind_t kind)
{
	kern_return_t kr;

	*dq = (struct mpsc_daemon_queue){
		.mpd_kind   = kind,
		.mpd_invoke = invoke,
		.mpd_queue  = MPSC_QUEUE_INITIALIZER(dq->mpd_queue),
		.mpd_chain  = { MPSC_QUEUE_NOTQUEUED_MARKER },
	};

	kr = kernel_thread_create(_mpsc_queue_thread_continue, dq, pri,
	    &dq->mpd_thread);
	if (kr == KERN_SUCCESS) {
		thread_set_thread_name(dq->mpd_thread, name);
		thread_start_in_assert_wait(dq->mpd_thread, (event_t)dq, THREAD_UNINT);
		thread_deallocate(dq->mpd_thread);
	}
	return kr;
}

kern_return_t
mpsc_daemon_queue_init_with_thread(mpsc_daemon_queue_t dq,
    mpsc_daemon_invoke_fn_t invoke, int pri, const char *name)
{
	return _mpsc_daemon_queue_init_with_thread(dq, invoke, pri, name,
	           MPSC_QUEUE_KIND_THREAD);
}

/* thread-call based queues */

static void
_mpsc_queue_thread_call_drain(thread_call_param_t arg0,
    thread_call_param_t arg1 __unused)
{
	_mpsc_daemon_queue_drain((mpsc_daemon_queue_t)arg0, NULL);
}

static void
_mpsc_queue_thread_call_wakeup(mpsc_daemon_queue_t dq)
{
	thread_call_enter(dq->mpd_call);
}

void
mpsc_daemon_queue_init_with_thread_call(mpsc_daemon_queue_t dq,
    mpsc_daemon_invoke_fn_t invoke, thread_call_priority_t pri)
{
	*dq = (struct mpsc_daemon_queue){
		.mpd_kind   = MPSC_QUEUE_KIND_THREAD_CALL,
		.mpd_invoke = invoke,
		.mpd_queue  = MPSC_QUEUE_INITIALIZER(dq->mpd_queue),
		.mpd_chain  = { MPSC_QUEUE_NOTQUEUED_MARKER },
	};
	dq->mpd_call = thread_call_allocate_with_options(
		_mpsc_queue_thread_call_drain, dq, pri, THREAD_CALL_OPTIONS_ONCE);
}

/* nested queues */

void
mpsc_daemon_queue_nested_invoke(mpsc_queue_chain_t elm,
    __unused mpsc_daemon_queue_t tq)
{
	mpsc_daemon_queue_t dq;
	dq = mpsc_queue_element(elm, struct mpsc_daemon_queue, mpd_chain);
	_mpsc_daemon_queue_drain(dq, NULL);
}

static void
_mpsc_daemon_queue_nested_wakeup(mpsc_daemon_queue_t dq)
{
	_mpsc_daemon_queue_enqueue(dq->mpd_target, &dq->mpd_chain);
}

void
mpsc_daemon_queue_init_with_target(mpsc_daemon_queue_t dq,
    mpsc_daemon_invoke_fn_t invoke, mpsc_daemon_queue_t target)
{
	*dq = (struct mpsc_daemon_queue){
		.mpd_kind   = MPSC_QUEUE_KIND_NESTED,
		.mpd_invoke = invoke,
		.mpd_target = target,
		.mpd_queue  = MPSC_QUEUE_INITIALIZER(dq->mpd_queue),
		.mpd_chain  = { MPSC_QUEUE_NOTQUEUED_MARKER },
	};
}

/* enqueue, drain & cancelation */

static void
_mpsc_daemon_queue_drain(mpsc_daemon_queue_t dq, thread_t self)
{
	mpsc_daemon_invoke_fn_t invoke = dq->mpd_invoke;
	mpsc_queue_chain_t head, cur, tail;
	mpsc_daemon_queue_state_t st;

again:
	/*
	 * Most of the time we're woken up because we're dirty,
	 * This atomic xor sets DRAINING and clears WAKEUP in a single atomic
	 * in that case.
	 *
	 * However, if we're woken up for cancelation, the state may be reduced to
	 * the CANCELED bit set only, and then the xor will actually set WAKEUP.
	 * We need to correct this and clear it back to avoid looping below.
	 * This is safe to do as no one is allowed to enqueue more work after
	 * cancelation has happened.
	 *
	 * We use `st` as a dependency token to pair with the release fence in
	 * _mpsc_daemon_queue_enqueue() which gives us the guarantee that the update
	 * to the tail of the MPSC queue that made it non empty is visible to us.
	 */
	st = os_atomic_xor(&dq->mpd_state,
	    MPSC_QUEUE_STATE_DRAINING | MPSC_QUEUE_STATE_WAKEUP, dependency);
	assert(st & MPSC_QUEUE_STATE_DRAINING);
	if (__improbable(st & MPSC_QUEUE_STATE_WAKEUP)) {
		assert(st & MPSC_QUEUE_STATE_CANCELED);
		os_atomic_andnot(&dq->mpd_state, MPSC_QUEUE_STATE_WAKEUP, relaxed);
	}

	os_atomic_dependency_t dep = os_atomic_make_dependency((uintptr_t)st);
	while ((head = mpsc_queue_dequeue_batch(&dq->mpd_queue, &tail, dep))) {
		mpsc_queue_batch_foreach_safe(cur, head, tail) {
			os_atomic_store(&cur->mpqc_next,
			    MPSC_QUEUE_NOTQUEUED_MARKER, relaxed);
			invoke(cur, dq);
		}
	}

	if (self) {
		assert_wait((event_t)dq, THREAD_UNINT);
	}

	/*
	 * Unlike GCD no fence is necessary here: there is no concept similar
	 * to "dispatch_sync()" that would require changes this thread made to be
	 * visible to other threads as part of the mpsc_daemon_queue machinery.
	 *
	 * Making updates that happened on the daemon queue visible to other threads
	 * is the responsibility of the client.
	 */
	st = os_atomic_andnot(&dq->mpd_state, MPSC_QUEUE_STATE_DRAINING, relaxed);

	/*
	 * A wakeup has happened while we were draining,
	 * which means that the queue did an [ empty -> non empty ]
	 * transition during our drain.
	 *
	 * Chances are we already observed and drained everything,
	 * but we need to be absolutely sure, so start a drain again
	 * as the enqueuer observed the DRAINING bit and has skipped calling
	 * _mpsc_daemon_queue_wakeup().
	 */
	if (__improbable(st & MPSC_QUEUE_STATE_WAKEUP)) {
		if (self) {
			clear_wait(self, THREAD_AWAKENED);
		}
		goto again;
	}

	/* dereferencing `dq` past this point is unsafe */

	if (__improbable(st & MPSC_QUEUE_STATE_CANCELED)) {
		thread_wakeup(&dq->mpd_state);
		if (self) {
			clear_wait(self, THREAD_AWAKENED);
			thread_terminate_self();
			__builtin_unreachable();
		}
	}
}

static void
_mpsc_daemon_queue_wakeup(mpsc_daemon_queue_t dq)
{
	switch (dq->mpd_kind) {
	case MPSC_QUEUE_KIND_NESTED:
		_mpsc_daemon_queue_nested_wakeup(dq);
		break;
	case MPSC_QUEUE_KIND_THREAD:
	case MPSC_QUEUE_KIND_THREAD_CRITICAL:
		_mpsc_queue_thread_wakeup(dq);
		break;
	case MPSC_QUEUE_KIND_THREAD_CALL:
		_mpsc_queue_thread_call_wakeup(dq);
		break;
	default:
		panic("mpsc_queue[%p]: invalid kind (%d)", dq, dq->mpd_kind);
	}
}

static void
_mpsc_daemon_queue_enqueue(mpsc_daemon_queue_t dq, mpsc_queue_chain_t elm)
{
	mpsc_daemon_queue_state_t st;

	if (mpsc_queue_append(&dq->mpd_queue, elm)) {
		/*
		 * Pairs with the acquire fence in _mpsc_daemon_queue_drain().
		 */
		st = os_atomic_or_orig(&dq->mpd_state, MPSC_QUEUE_STATE_WAKEUP, release);
		if (__improbable(st & MPSC_QUEUE_STATE_CANCELED)) {
			panic("mpsc_queue[%p]: use after cancelation", dq);
		}

		if ((st & (MPSC_QUEUE_STATE_DRAINING | MPSC_QUEUE_STATE_WAKEUP)) == 0) {
			_mpsc_daemon_queue_wakeup(dq);
		}
	}
}

void
mpsc_daemon_enqueue(mpsc_daemon_queue_t dq, mpsc_queue_chain_t elm,
    mpsc_queue_options_t options)
{
	if (options & MPSC_QUEUE_DISABLE_PREEMPTION) {
		disable_preemption();
	}

	_mpsc_daemon_queue_enqueue(dq, elm);

	if (options & MPSC_QUEUE_DISABLE_PREEMPTION) {
		enable_preemption();
	}
}

void
mpsc_daemon_queue_cancel_and_wait(mpsc_daemon_queue_t dq)
{
	mpsc_daemon_queue_state_t st;

	assert_wait((event_t)&dq->mpd_state, THREAD_UNINT);

	st = os_atomic_or_orig(&dq->mpd_state, MPSC_QUEUE_STATE_CANCELED, relaxed);
	if (__improbable(st & MPSC_QUEUE_STATE_CANCELED)) {
		panic("mpsc_queue[%p]: cancelled twice (%x)", dq, st);
	}

	if (dq->mpd_kind == MPSC_QUEUE_KIND_NESTED && st == 0) {
		clear_wait(current_thread(), THREAD_AWAKENED);
	} else {
		disable_preemption();
		_mpsc_daemon_queue_wakeup(dq);
		enable_preemption();
		thread_block(THREAD_CONTINUE_NULL);
	}

	switch (dq->mpd_kind) {
	case MPSC_QUEUE_KIND_NESTED:
		dq->mpd_target = NULL;
		break;
	case MPSC_QUEUE_KIND_THREAD:
	case MPSC_QUEUE_KIND_THREAD_CRITICAL:
		dq->mpd_thread = NULL;
		break;
	case MPSC_QUEUE_KIND_THREAD_CALL:
		thread_call_cancel_wait(dq->mpd_call);
		thread_call_free(dq->mpd_call);
		dq->mpd_call = NULL;
		break;
	default:
		panic("mpsc_queue[%p]: invalid kind (%d)", dq, dq->mpd_kind);
	}
	dq->mpd_kind = MPSC_QUEUE_KIND_UNKNOWN;
}

#pragma mark deferred deallocation daemon

static struct mpsc_daemon_queue thread_deferred_deallocation_queue;

void
thread_deallocate_daemon_init(void)
{
	kern_return_t kr;

	kr = _mpsc_daemon_queue_init_with_thread(&thread_deferred_deallocation_queue,
	    mpsc_daemon_queue_nested_invoke, MINPRI_KERNEL,
	    "daemon.deferred-deallocation", MPSC_QUEUE_KIND_THREAD_CRITICAL);
	if (kr != KERN_SUCCESS) {
		panic("thread_deallocate_daemon_init: creating daemon failed (%d)", kr);
	}
}

void
thread_deallocate_daemon_register_queue(mpsc_daemon_queue_t dq,
    mpsc_daemon_invoke_fn_t invoke)
{
	mpsc_daemon_queue_init_with_target(dq, invoke,
	    &thread_deferred_deallocation_queue);
}
Commit	Line	Data
cb323159 A	1	/*
	2	* Copyright (c) 2018 Apple Inc. All rights reserved.
	3	*
	4	* @APPLE_OSREFERENCE_LICENSE_HEADER_START@
	5	*
	6	* This file contains Original Code and/or Modifications of Original Code
	7	* as defined in and that are subject to the Apple Public Source License
	8	* Version 2.0 (the 'License'). You may not use this file except in
	9	* compliance with the License. The rights granted to you under the License
	10	* may not be used to create, or enable the creation or redistribution of,
	11	* unlawful or unlicensed copies of an Apple operating system, or to
	12	* circumvent, violate, or enable the circumvention or violation of, any
	13	* terms of an Apple operating system software license agreement.
	14	*
	15	* Please obtain a copy of the License at
	16	* http://www.opensource.apple.com/apsl/ and read it before using this file.
	17	*
	18	* The Original Code and all software distributed under the License are
	19	* distributed on an 'AS IS' basis, WITHOUT WARRANTY OF ANY KIND, EITHER
	20	* EXPRESS OR IMPLIED, AND APPLE HEREBY DISCLAIMS ALL SUCH WARRANTIES,
	21	* INCLUDING WITHOUT LIMITATION, ANY WARRANTIES OF MERCHANTABILITY,
	22	* FITNESS FOR A PARTICULAR PURPOSE, QUIET ENJOYMENT OR NON-INFRINGEMENT.
	23	* Please see the License for the specific language governing rights and
	24	* limitations under the License.
	25	*
	26	* @APPLE_OSREFERENCE_LICENSE_HEADER_END@
	27	*/
	28
	29	#include <machine/machine_cpu.h>
	30	#include <kern/locks.h>
	31	#include <kern/mpsc_queue.h>
	32	#include <kern/thread.h>
	33
	34	#pragma mark Single Consumer calls
	35
	36	__attribute__((noinline))
	37	static mpsc_queue_chain_t
	38	_mpsc_queue_wait_for_enqueuer(struct mpsc_queue_chain _Atomic ptr)
	39	{
	40	return hw_wait_while_equals((void **)ptr, NULL);
	41	}
	42
	43	void
	44	mpsc_queue_restore_batch(mpsc_queue_head_t q, mpsc_queue_chain_t first,
	45	mpsc_queue_chain_t last)
	46	{
	47	mpsc_queue_chain_t head = os_atomic_load(&q->mpqh_head.mpqc_next, relaxed);
	48
	49	os_atomic_store(&last->mpqc_next, head, relaxed);
	50
	51	if (head == NULL &&
	52	!os_atomic_cmpxchg(&q->mpqh_tail, &q->mpqh_head, last, release)) {
	53	head = os_atomic_load(&q->mpqh_head.mpqc_next, relaxed);
	54	if (__improbable(head == NULL)) {
	55	head = _mpsc_queue_wait_for_enqueuer(&q->mpqh_head.mpqc_next);
	56	}
	57	os_atomic_store(&last->mpqc_next, head, relaxed);
	58	}
	59
	60	os_atomic_store(&q->mpqh_head.mpqc_next, first, relaxed);
	61	}
	62
	63	mpsc_queue_chain_t
	64	mpsc_queue_dequeue_batch(mpsc_queue_head_t q, mpsc_queue_chain_t *tail_out,
65	os_atomic_dependency_t dependency)
66	{
67	mpsc_queue_chain_t head, tail;
68
69	q = os_atomic_inject_dependency(q, dependency);
70
71	tail = os_atomic_load(&q->mpqh_tail, relaxed);
72	if (__improbable(tail == &q->mpqh_head)) {
73	*tail_out = NULL;
74	return NULL;
75	}
76
77	head = os_atomic_load(&q->mpqh_head.mpqc_next, relaxed);
78	if (__improbable(head == NULL)) {
79	head = _mpsc_queue_wait_for_enqueuer(&q->mpqh_head.mpqc_next);
80	}
81	os_atomic_store(&q->mpqh_head.mpqc_next, NULL, relaxed);
82	/*
83	* 22708742: set tail to &q->mpqh_head with release, so that NULL write
84	* to head above doesn't clobber the head set by concurrent enqueuer
85	*
86	* The other half of the seq_cst is required to pair with any enqueuer that
87	* contributed to an element in this list (pairs with the release fence in
88	* __mpsc_queue_append_update_tail().
89	*
90	* Making this seq_cst instead of acq_rel makes mpsc_queue_append*()
91	* visibility transitive (when items hop from one queue to the next)
92	* which is expected by clients implicitly.
93	*
94	* Note that this is the same number of fences that a traditional lock
95	* would have, but as a once-per-batch cost.
96	*/
97	*tail_out = os_atomic_xchg(&q->mpqh_tail, &q->mpqh_head, seq_cst);
98
99	return head;
100	}
101
102	mpsc_queue_chain_t
103	mpsc_queue_batch_next(mpsc_queue_chain_t cur, mpsc_queue_chain_t tail)
104	{
105	mpsc_queue_chain_t elm = NULL;
106	if (cur == tail \|\| cur == NULL) {
107	return elm;
108	}
109
110	elm = os_atomic_load(&cur->mpqc_next, relaxed);
111	if (__improbable(elm == NULL)) {
112	elm = _mpsc_queue_wait_for_enqueuer(&cur->mpqc_next);
113	}
114	return elm;
115	}
116
117	#pragma mark "GCD"-like facilities
118
119	static void _mpsc_daemon_queue_drain(mpsc_daemon_queue_t, thread_t);
120	static void _mpsc_daemon_queue_enqueue(mpsc_daemon_queue_t, mpsc_queue_chain_t);
121
122	/* thread based queues */
123
124	static void
125	_mpsc_queue_thread_continue(void *param, wait_result_t wr __unused)
126	{
127	mpsc_daemon_queue_t dq = param;
f427ee49 A	128	mpsc_daemon_queue_kind_t kind = dq->mpd_kind;
	129	thread_t self = dq->mpd_thread;
	130
	131	__builtin_assume(self != THREAD_NULL);
	132
	133	if (kind == MPSC_QUEUE_KIND_THREAD_CRITICAL) {
	134	self->options \|= TH_OPT_SYSTEM_CRITICAL;
	135	}
cb323159 A	136
cb323159 A	137	assert(dq->mpd_thread == current_thread());
f427ee49 A	138	_mpsc_daemon_queue_drain(dq, self);
	139
	140	if (kind == MPSC_QUEUE_KIND_THREAD_CRITICAL) {
	141	self->options &= ~TH_OPT_SYSTEM_CRITICAL;
	142	}
	143
cb323159 A	144	thread_block_parameter(_mpsc_queue_thread_continue, dq);
	145	}
	146
	147	static void
	148	_mpsc_queue_thread_wakeup(mpsc_daemon_queue_t dq)
	149	{
	150	thread_wakeup_thread((event_t)dq, dq->mpd_thread);
	151	}
	152
	153	static kern_return_t
	154	_mpsc_daemon_queue_init_with_thread(mpsc_daemon_queue_t dq,
	155	mpsc_daemon_invoke_fn_t invoke, int pri, const char *name,
	156	mpsc_daemon_queue_kind_t kind)
	157	{
	158	kern_return_t kr;
	159
	160	*dq = (struct mpsc_daemon_queue){
	161	.mpd_kind = kind,
	162	.mpd_invoke = invoke,
	163	.mpd_queue = MPSC_QUEUE_INITIALIZER(dq->mpd_queue),
	164	.mpd_chain = { MPSC_QUEUE_NOTQUEUED_MARKER },
	165	};
	166
	167	kr = kernel_thread_create(_mpsc_queue_thread_continue, dq, pri,
	168	&dq->mpd_thread);
	169	if (kr == KERN_SUCCESS) {
	170	thread_set_thread_name(dq->mpd_thread, name);
	171	thread_start_in_assert_wait(dq->mpd_thread, (event_t)dq, THREAD_UNINT);
	172	thread_deallocate(dq->mpd_thread);
	173	}
	174	return kr;
	175	}
	176
	177	kern_return_t
	178	mpsc_daemon_queue_init_with_thread(mpsc_daemon_queue_t dq,
	179	mpsc_daemon_invoke_fn_t invoke, int pri, const char *name)
	180	{
	181	return _mpsc_daemon_queue_init_with_thread(dq, invoke, pri, name,
	182	MPSC_QUEUE_KIND_THREAD);
	183	}
	184
	185	/* thread-call based queues */
	186
	187	static void
	188	_mpsc_queue_thread_call_drain(thread_call_param_t arg0,
	189	thread_call_param_t arg1 __unused)
	190	{
	191	_mpsc_daemon_queue_drain((mpsc_daemon_queue_t)arg0, NULL);
	192	}
	193
	194	static void
	195	_mpsc_queue_thread_call_wakeup(mpsc_daemon_queue_t dq)
	196	{
	197	thread_call_enter(dq->mpd_call);
	198	}
	199
	200	void
	201	mpsc_daemon_queue_init_with_thread_call(mpsc_daemon_queue_t dq,
	202	mpsc_daemon_invoke_fn_t invoke, thread_call_priority_t pri)
	203	{
	204	*dq = (struct mpsc_daemon_queue){
	205	.mpd_kind = MPSC_QUEUE_KIND_THREAD_CALL,
	206	.mpd_invoke = invoke,
	207	.mpd_queue = MPSC_QUEUE_INITIALIZER(dq->mpd_queue),
208	.mpd_chain = { MPSC_QUEUE_NOTQUEUED_MARKER },
209	};
210	dq->mpd_call = thread_call_allocate_with_options(
211	_mpsc_queue_thread_call_drain, dq, pri, THREAD_CALL_OPTIONS_ONCE);
212	}
213
214	/* nested queues */
215
216	void
217	mpsc_daemon_queue_nested_invoke(mpsc_queue_chain_t elm,
218	__unused mpsc_daemon_queue_t tq)
219	{
220	mpsc_daemon_queue_t dq;
221	dq = mpsc_queue_element(elm, struct mpsc_daemon_queue, mpd_chain);
222	_mpsc_daemon_queue_drain(dq, NULL);
223	}
224
225	static void
226	_mpsc_daemon_queue_nested_wakeup(mpsc_daemon_queue_t dq)
227	{
228	_mpsc_daemon_queue_enqueue(dq->mpd_target, &dq->mpd_chain);
229	}
230
231	void
232	mpsc_daemon_queue_init_with_target(mpsc_daemon_queue_t dq,
233	mpsc_daemon_invoke_fn_t invoke, mpsc_daemon_queue_t target)
234	{
235	*dq = (struct mpsc_daemon_queue){
236	.mpd_kind = MPSC_QUEUE_KIND_NESTED,
237	.mpd_invoke = invoke,
238	.mpd_target = target,
239	.mpd_queue = MPSC_QUEUE_INITIALIZER(dq->mpd_queue),
240	.mpd_chain = { MPSC_QUEUE_NOTQUEUED_MARKER },
241	};
242	}
243
244	/* enqueue, drain & cancelation */
245
246	static void
247	_mpsc_daemon_queue_drain(mpsc_daemon_queue_t dq, thread_t self)
248	{
249	mpsc_daemon_invoke_fn_t invoke = dq->mpd_invoke;
cb323159 A	250	mpsc_queue_chain_t head, cur, tail;
	251	mpsc_daemon_queue_state_t st;
	252
cb323159 A	253	again:
	254	/*
	255	* Most of the time we're woken up because we're dirty,
	256	* This atomic xor sets DRAINING and clears WAKEUP in a single atomic
	257	* in that case.
	258	*
	259	* However, if we're woken up for cancelation, the state may be reduced to
	260	* the CANCELED bit set only, and then the xor will actually set WAKEUP.
	261	* We need to correct this and clear it back to avoid looping below.
	262	* This is safe to do as no one is allowed to enqueue more work after
	263	* cancelation has happened.
	264	*
	265	* We use `st` as a dependency token to pair with the release fence in
	266	* _mpsc_daemon_queue_enqueue() which gives us the guarantee that the update
	267	* to the tail of the MPSC queue that made it non empty is visible to us.
	268	*/
	269	st = os_atomic_xor(&dq->mpd_state,
	270	MPSC_QUEUE_STATE_DRAINING \| MPSC_QUEUE_STATE_WAKEUP, dependency);
	271	assert(st & MPSC_QUEUE_STATE_DRAINING);
	272	if (__improbable(st & MPSC_QUEUE_STATE_WAKEUP)) {
	273	assert(st & MPSC_QUEUE_STATE_CANCELED);
	274	os_atomic_andnot(&dq->mpd_state, MPSC_QUEUE_STATE_WAKEUP, relaxed);
	275	}
	276
	277	os_atomic_dependency_t dep = os_atomic_make_dependency((uintptr_t)st);
	278	while ((head = mpsc_queue_dequeue_batch(&dq->mpd_queue, &tail, dep))) {
	279	mpsc_queue_batch_foreach_safe(cur, head, tail) {
	280	os_atomic_store(&cur->mpqc_next,
	281	MPSC_QUEUE_NOTQUEUED_MARKER, relaxed);
	282	invoke(cur, dq);
	283	}
	284	}
	285
	286	if (self) {
	287	assert_wait((event_t)dq, THREAD_UNINT);
	288	}
	289
	290	/*
	291	* Unlike GCD no fence is necessary here: there is no concept similar
	292	* to "dispatch_sync()" that would require changes this thread made to be
	293	* visible to other threads as part of the mpsc_daemon_queue machinery.
	294	*
	295	* Making updates that happened on the daemon queue visible to other threads
	296	* is the responsibility of the client.
	297	*/
	298	st = os_atomic_andnot(&dq->mpd_state, MPSC_QUEUE_STATE_DRAINING, relaxed);
	299
	300	/*
	301	* A wakeup has happened while we were draining,
	302	* which means that the queue did an [ empty -> non empty ]
	303	* transition during our drain.
	304	*
	305	* Chances are we already observed and drained everything,
	306	* but we need to be absolutely sure, so start a drain again
	307	* as the enqueuer observed the DRAINING bit and has skipped calling
	308	* _mpsc_daemon_queue_wakeup().
	309	*/
	310	if (__improbable(st & MPSC_QUEUE_STATE_WAKEUP)) {
	311	if (self) {
	312	clear_wait(self, THREAD_AWAKENED);
	313	}
	314	goto again;
	315	}
	316
317	/* dereferencing `dq` past this point is unsafe */
318
cb323159 A	319	if (__improbable(st & MPSC_QUEUE_STATE_CANCELED)) {
	320	thread_wakeup(&dq->mpd_state);
	321	if (self) {
	322	clear_wait(self, THREAD_AWAKENED);
	323	thread_terminate_self();
	324	__builtin_unreachable();
	325	}
	326	}
	327	}
	328
	329	static void
	330	_mpsc_daemon_queue_wakeup(mpsc_daemon_queue_t dq)
	331	{
	332	switch (dq->mpd_kind) {
	333	case MPSC_QUEUE_KIND_NESTED:
	334	_mpsc_daemon_queue_nested_wakeup(dq);
	335	break;
	336	case MPSC_QUEUE_KIND_THREAD:
	337	case MPSC_QUEUE_KIND_THREAD_CRITICAL:
	338	_mpsc_queue_thread_wakeup(dq);
	339	break;
	340	case MPSC_QUEUE_KIND_THREAD_CALL:
	341	_mpsc_queue_thread_call_wakeup(dq);
	342	break;
	343	default:
	344	panic("mpsc_queue[%p]: invalid kind (%d)", dq, dq->mpd_kind);
	345	}
	346	}
	347
	348	static void
	349	_mpsc_daemon_queue_enqueue(mpsc_daemon_queue_t dq, mpsc_queue_chain_t elm)
	350	{
	351	mpsc_daemon_queue_state_t st;
	352
	353	if (mpsc_queue_append(&dq->mpd_queue, elm)) {
	354	/*
	355	* Pairs with the acquire fence in _mpsc_daemon_queue_drain().
	356	*/
	357	st = os_atomic_or_orig(&dq->mpd_state, MPSC_QUEUE_STATE_WAKEUP, release);
	358	if (__improbable(st & MPSC_QUEUE_STATE_CANCELED)) {
	359	panic("mpsc_queue[%p]: use after cancelation", dq);
	360	}
	361
	362	if ((st & (MPSC_QUEUE_STATE_DRAINING \| MPSC_QUEUE_STATE_WAKEUP)) == 0) {
	363	_mpsc_daemon_queue_wakeup(dq);
	364	}
	365	}
	366	}
	367
	368	void
	369	mpsc_daemon_enqueue(mpsc_daemon_queue_t dq, mpsc_queue_chain_t elm,
	370	mpsc_queue_options_t options)
	371	{
	372	if (options & MPSC_QUEUE_DISABLE_PREEMPTION) {
	373	disable_preemption();
	374	}
	375
	376	_mpsc_daemon_queue_enqueue(dq, elm);
	377
	378	if (options & MPSC_QUEUE_DISABLE_PREEMPTION) {
	379	enable_preemption();
	380	}
	381	}
	382
383	void
384	mpsc_daemon_queue_cancel_and_wait(mpsc_daemon_queue_t dq)
385	{
386	mpsc_daemon_queue_state_t st;
387
388	assert_wait((event_t)&dq->mpd_state, THREAD_UNINT);
389
390	st = os_atomic_or_orig(&dq->mpd_state, MPSC_QUEUE_STATE_CANCELED, relaxed);
391	if (__improbable(st & MPSC_QUEUE_STATE_CANCELED)) {
392	panic("mpsc_queue[%p]: cancelled twice (%x)", dq, st);
393	}
394
395	if (dq->mpd_kind == MPSC_QUEUE_KIND_NESTED && st == 0) {
396	clear_wait(current_thread(), THREAD_AWAKENED);
397	} else {
398	disable_preemption();
399	_mpsc_daemon_queue_wakeup(dq);
400	enable_preemption();
401	thread_block(THREAD_CONTINUE_NULL);
402	}
403
404	switch (dq->mpd_kind) {
405	case MPSC_QUEUE_KIND_NESTED:
406	dq->mpd_target = NULL;
407	break;
408	case MPSC_QUEUE_KIND_THREAD:
409	case MPSC_QUEUE_KIND_THREAD_CRITICAL:
410	dq->mpd_thread = NULL;
411	break;
412	case MPSC_QUEUE_KIND_THREAD_CALL:
413	thread_call_cancel_wait(dq->mpd_call);
414	thread_call_free(dq->mpd_call);
415	dq->mpd_call = NULL;
416	break;
417	default:
418	panic("mpsc_queue[%p]: invalid kind (%d)", dq, dq->mpd_kind);
419	}
420	dq->mpd_kind = MPSC_QUEUE_KIND_UNKNOWN;
421	}
422
423	#pragma mark deferred deallocation daemon
424
425	static struct mpsc_daemon_queue thread_deferred_deallocation_queue;
426
427	void
428	thread_deallocate_daemon_init(void)
429	{
430	kern_return_t kr;
431
432	kr = _mpsc_daemon_queue_init_with_thread(&thread_deferred_deallocation_queue,
433	mpsc_daemon_queue_nested_invoke, MINPRI_KERNEL,
434	"daemon.deferred-deallocation", MPSC_QUEUE_KIND_THREAD_CRITICAL);
435	if (kr != KERN_SUCCESS) {
436	panic("thread_deallocate_daemon_init: creating daemon failed (%d)", kr);
437	}
438	}
439
440	void
441	thread_deallocate_daemon_register_queue(mpsc_daemon_queue_t dq,
442	mpsc_daemon_invoke_fn_t invoke)
443	{
444	mpsc_daemon_queue_init_with_target(dq, invoke,
445	&thread_deferred_deallocation_queue);
446	}