Loading...
--- xnu/xnu-12377.121.6/bsd/pthread/pthread_workqueue.c
+++ xnu/xnu-4903.241.1/bsd/pthread/pthread_workqueue.c
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2000-2020 Apple Inc. All rights reserved.
+ * Copyright (c) 2000-2017 Apple Inc. All rights reserved.
*
* @APPLE_OSREFERENCE_LICENSE_HEADER_START@
*
@@ -29,6 +29,9 @@
#include <sys/cdefs.h>
+// <rdar://problem/26158937> panic() should be marked noreturn
+extern void panic(const char *string, ...) __printflike(1,2) __dead2;
+
#include <kern/assert.h>
#include <kern/ast.h>
#include <kern/clock.h>
@@ -36,12 +39,10 @@
#include <kern/kern_types.h>
#include <kern/policy_internal.h>
#include <kern/processor.h>
-#include <kern/sched_prim.h> /* for thread_exception_return */
+#include <kern/sched_prim.h> /* for thread_exception_return */
#include <kern/task.h>
#include <kern/thread.h>
-#include <kern/thread_group.h>
#include <kern/zalloc.h>
-#include <kern/work_interval.h>
#include <mach/kern_return.h>
#include <mach/mach_param.h>
#include <mach/mach_port.h>
@@ -56,9 +57,7 @@
#include <mach/vm_statistics.h>
#include <machine/atomic.h>
#include <machine/machine_routines.h>
-#include <machine/smp.h>
#include <vm/vm_map.h>
-#include <vm/vm_fault_xnu.h>
#include <vm/vm_protos.h>
#include <sys/eventvar.h>
@@ -66,7 +65,7 @@
#include <sys/kernel.h>
#include <sys/lock.h>
#include <sys/param.h>
-#include <sys/proc_info.h> /* for fill_procworkqueue */
+#include <sys/proc_info.h> /* for fill_procworkqueue */
#include <sys/proc_internal.h>
#include <sys/pthread_shims.h>
#include <sys/resourcevar.h>
@@ -83,39 +82,21 @@
#include <os/log.h>
+extern thread_t port_name_to_thread(mach_port_name_t port_name); /* osfmk/kern/ipc_tt.h */
+
static void workq_unpark_continue(void *uth, wait_result_t wr) __dead2;
-
-static void workq_bound_thread_unpark_continue(void *uth, wait_result_t wr) __dead2;
-
-static void workq_bound_thread_initialize_and_unpark_continue(void *uth, wait_result_t wr) __dead2;
-
-static void workq_bound_thread_setup_and_run(struct uthread *uth, int setup_flags) __dead2;
-
-static void workq_schedule_creator(proc_t p, struct workqueue *wq,
- workq_kern_threadreq_flags_t flags);
+static void workq_schedule_creator(proc_t p, struct workqueue *wq, int flags);
static bool workq_threadreq_admissible(struct workqueue *wq, struct uthread *uth,
- workq_threadreq_t req);
+ workq_threadreq_t req);
static uint32_t workq_constrained_allowance(struct workqueue *wq,
- thread_qos_t at_qos, struct uthread *uth,
- bool may_start_timer, bool record_failed_allowance);
-
-static bool _wq_cooperative_queue_refresh_best_req_qos(struct workqueue *wq);
+ thread_qos_t at_qos, struct uthread *uth, bool may_start_timer);
static bool workq_thread_is_busy(uint64_t cur_ts,
- _Atomic uint64_t *lastblocked_tsp);
+ _Atomic uint64_t *lastblocked_tsp);
static int workq_sysctl_handle_usecs SYSCTL_HANDLER_ARGS;
-
-static bool
-workq_schedule_delayed_thread_creation(struct workqueue *wq, int flags);
-
-static inline void
-workq_lock_spin(struct workqueue *wq);
-
-static inline void
-workq_unlock(struct workqueue *wq);
#pragma mark globals
@@ -125,22 +106,20 @@
};
#define WORKQ_SYSCTL_USECS(var, init) \
- static struct workq_usec_var var = { .usecs = init }; \
- SYSCTL_OID(_kern, OID_AUTO, var##_usecs, \
- CTLTYPE_INT | CTLFLAG_RW | CTLFLAG_LOCKED, &var, 0, \
- workq_sysctl_handle_usecs, "I", "")
-
-static LCK_GRP_DECLARE(workq_lck_grp, "workq");
+ static struct workq_usec_var var = { .usecs = init }; \
+ SYSCTL_OID(_kern, OID_AUTO, var##_usecs, \
+ CTLTYPE_INT | CTLFLAG_RW | CTLFLAG_LOCKED, &var, 0, \
+ workq_sysctl_handle_usecs, "I", "")
+
+static lck_grp_t *workq_lck_grp;
+static lck_attr_t *workq_lck_attr;
+static lck_grp_attr_t *workq_lck_grp_attr;
os_refgrp_decl(static, workq_refgrp, "workq", NULL);
-static ZONE_DEFINE(workq_zone_workqueue, "workq.wq",
- sizeof(struct workqueue), ZC_NONE);
-static ZONE_DEFINE(workq_zone_threadreq, "workq.threadreq",
- sizeof(struct workq_threadreq_s), ZC_CACHING);
-
-static struct mpsc_daemon_queue workq_deallocate_queue;
-
-WORKQ_SYSCTL_USECS(wq_stalled_window, WQ_STALLED_WINDOW_USECS);
+static zone_t workq_zone_workqueue;
+static zone_t workq_zone_threadreq;
+
+WORKQ_SYSCTL_USECS(wq_stalled_window, WQ_STALLED_WINDOW_USECS);
WORKQ_SYSCTL_USECS(wq_reduce_pool_window, WQ_REDUCE_POOL_WINDOW_USECS);
WORKQ_SYSCTL_USECS(wq_max_timer_interval, WQ_MAX_TIMER_INTERVAL_USECS);
static uint32_t wq_max_threads = WORKQUEUE_MAXTHREADS;
@@ -149,20 +128,6 @@
static uint16_t wq_death_max_load;
static uint32_t wq_max_parallelism[WORKQ_NUM_QOS_BUCKETS];
-/*
- * This is not a hard limit but the max size we want to aim to hit across the
- * entire cooperative pool. We can oversubscribe the pool due to non-cooperative
- * workers and the max we will oversubscribe the pool by, is a total of
- * wq_max_cooperative_threads * WORKQ_NUM_QOS_BUCKETS.
- */
-static uint32_t wq_max_cooperative_threads;
-
-static inline uint32_t
-wq_cooperative_queue_max_size(struct workqueue *wq)
-{
- return wq->wq_cooperative_queue_has_limited_max_size ? 1 : wq_max_cooperative_threads;
-}
-
#pragma mark sysctls
static int
@@ -171,73 +136,18 @@
#pragma unused(arg2)
struct workq_usec_var *v = arg1;
int error = sysctl_handle_int(oidp, &v->usecs, 0, req);
- if (error || !req->newptr) {
+ if (error || !req->newptr)
return error;
- }
clock_interval_to_absolutetime_interval(v->usecs, NSEC_PER_USEC,
- &v->abstime);
+ &v->abstime);
return 0;
}
SYSCTL_INT(_kern, OID_AUTO, wq_max_threads, CTLFLAG_RW | CTLFLAG_LOCKED,
- &wq_max_threads, 0, "");
+ &wq_max_threads, 0, "");
SYSCTL_INT(_kern, OID_AUTO, wq_max_constrained_threads, CTLFLAG_RW | CTLFLAG_LOCKED,
- &wq_max_constrained_threads, 0, "");
-
-static int
-wq_limit_cooperative_threads_for_proc SYSCTL_HANDLER_ARGS
-{
-#pragma unused(arg1, arg2, oidp)
- int input_pool_size = 0;
- int changed;
- int error = 0;
-
- error = sysctl_io_number(req, 0, sizeof(int), &input_pool_size, &changed);
- if (error || !changed) {
- return error;
- }
-
-#define WQ_COOPERATIVE_POOL_SIZE_DEFAULT 0
-#define WQ_COOPERATIVE_POOL_SIZE_STRICT_PER_QOS -1
-/* Not available currently, but sysctl interface is designed to allow these
- * extra parameters:
- * WQ_COOPERATIVE_POOL_SIZE_STRICT : -2 (across all bucket)
- * WQ_COOPERATIVE_POOL_SIZE_CUSTOM : [1, 512]
- */
-
- if (input_pool_size != WQ_COOPERATIVE_POOL_SIZE_DEFAULT
- && input_pool_size != WQ_COOPERATIVE_POOL_SIZE_STRICT_PER_QOS) {
- error = EINVAL;
- goto out;
- }
-
- proc_t p = req->p;
- struct workqueue *wq = proc_get_wqptr(p);
-
- if (wq != NULL) {
- workq_lock_spin(wq);
- if (wq->wq_reqcount > 0 || wq->wq_nthreads > 0) {
- // Hackily enforce that the workqueue is still new (no requests or
- // threads)
- error = ENOTSUP;
- } else {
- wq->wq_cooperative_queue_has_limited_max_size = (input_pool_size == WQ_COOPERATIVE_POOL_SIZE_STRICT_PER_QOS);
- }
- workq_unlock(wq);
- } else {
- /* This process has no workqueue, calling this syctl makes no sense */
- return ENOTSUP;
- }
-
-out:
- return error;
-}
-
-SYSCTL_PROC(_kern, OID_AUTO, wq_limit_cooperative_threads,
- CTLFLAG_ANYBODY | CTLFLAG_MASKED | CTLFLAG_WR | CTLFLAG_LOCKED | CTLTYPE_INT, 0, 0,
- wq_limit_cooperative_threads_for_proc,
- "I", "Modify the max pool size of the cooperative pool");
+ &wq_max_constrained_threads, 0, "");
#pragma mark p_wqptr
@@ -249,7 +159,7 @@
return os_atomic_load(&p->p_wqptr, relaxed);
}
-struct workqueue *
+static struct workqueue *
proc_get_wqptr(struct proc *p)
{
struct workqueue *wq = proc_get_wqptr_fast(p);
@@ -273,10 +183,10 @@
struct workqueue *wq;
proc_lock(p);
- wq = os_atomic_load(&p->p_wqptr, relaxed);
+ wq = p->p_wqptr;
if (wq == NULL) {
- os_atomic_store(&p->p_wqptr, WQPTR_IS_INITING_VALUE, relaxed);
+ p->p_wqptr = WQPTR_IS_INITING_VALUE;
proc_unlock(p);
return true;
}
@@ -300,7 +210,9 @@
static inline void
workq_thread_wakeup(struct uthread *uth)
{
- thread_wakeup_thread(workq_parked_wait_event(uth), get_machthread(uth));
+ if ((uth->uu_workq_flags & UT_WORKQ_IDLE_CLEANUP) == 0) {
+ thread_wakeup_thread(workq_parked_wait_event(uth), uth->uu_thread);
+ }
}
#pragma mark wq_thactive
@@ -324,15 +236,15 @@
#define WQ_THACTIVE_BUCKET_HALF (1U << (WQ_THACTIVE_BUCKET_WIDTH - 1))
static_assert(sizeof(wq_thactive_t) * CHAR_BIT - WQ_THACTIVE_QOS_SHIFT >= 3,
- "Make sure we have space to encode a QoS");
+ "Make sure we have space to encode a QoS");
static inline wq_thactive_t
_wq_thactive(struct workqueue *wq)
{
- return os_atomic_load_wide(&wq->wq_thactive, relaxed);
-}
-
-static inline uint8_t
+ return os_atomic_load(&wq->wq_thactive, relaxed);
+}
+
+static inline int
_wq_bucket(thread_qos_t qos)
{
// Map both BG and MT to the same bucket by over-shifting down and
@@ -346,7 +258,7 @@
}
#define WQ_THACTIVE_BEST_CONSTRAINED_REQ_QOS(tha) \
- ((thread_qos_t)((tha) >> WQ_THACTIVE_QOS_SHIFT))
+ ((tha) >> WQ_THACTIVE_QOS_SHIFT)
static inline thread_qos_t
_wq_thactive_best_constrained_req_qos(struct workqueue *wq)
@@ -364,7 +276,7 @@
workq_threadreq_t req;
req = priority_queue_max(&wq->wq_constrained_queue,
- struct workq_threadreq_s, tr_entry);
+ struct workq_threadreq_s, tr_entry);
new_qos = req ? req->tr_qos : THREAD_QOS_UNSPECIFIED;
old_qos = _wq_thactive_best_constrained_req_qos(wq);
if (old_qos != new_qos) {
@@ -377,9 +289,9 @@
v = os_atomic_add(&wq->wq_thactive, v, relaxed);
#ifdef __LP64__
WQ_TRACE_WQ(TRACE_wq_thactive_update, wq, (uint64_t)v,
- (uint64_t)(v >> 64), 0);
+ (uint64_t)(v >> 64), 0, 0);
#else
- WQ_TRACE_WQ(TRACE_wq_thactive_update, wq, v, 0, 0);
+ WQ_TRACE_WQ(TRACE_wq_thactive_update, wq, v, 0, 0, 0);
#endif
}
}
@@ -387,9 +299,7 @@
static inline wq_thactive_t
_wq_thactive_offset_for_qos(thread_qos_t qos)
{
- uint8_t bucket = _wq_bucket(qos);
- __builtin_assume(bucket < WORKQ_NUM_BUCKETS);
- return (wq_thactive_t)1 << (bucket * WQ_THACTIVE_BUCKET_WIDTH);
+ return (wq_thactive_t)1 << (_wq_bucket(qos) * WQ_THACTIVE_BUCKET_WIDTH);
}
static inline wq_thactive_t
@@ -408,18 +318,18 @@
static inline void
_wq_thactive_move(struct workqueue *wq,
- thread_qos_t old_qos, thread_qos_t new_qos)
+ thread_qos_t old_qos, thread_qos_t new_qos)
{
wq_thactive_t v = _wq_thactive_offset_for_qos(new_qos) -
- _wq_thactive_offset_for_qos(old_qos);
- os_atomic_add(&wq->wq_thactive, v, relaxed);
+ _wq_thactive_offset_for_qos(old_qos);
+ os_atomic_add_orig(&wq->wq_thactive, v, relaxed);
wq->wq_thscheduled_count[_wq_bucket(old_qos)]--;
wq->wq_thscheduled_count[_wq_bucket(new_qos)]++;
}
static inline uint32_t
_wq_thactive_aggregate_downto_qos(struct workqueue *wq, wq_thactive_t v,
- thread_qos_t qos, uint32_t *busycount, uint32_t *max_busycount)
+ thread_qos_t qos, uint32_t *busycount, uint32_t *max_busycount)
{
uint32_t count = 0, active;
uint64_t curtime;
@@ -434,7 +344,7 @@
*max_busycount = THREAD_QOS_LAST - qos;
}
- uint8_t i = _wq_bucket(qos);
+ int i = _wq_bucket(qos);
v >>= i * WQ_THACTIVE_BUCKET_WIDTH;
for (; i < WORKQ_NUM_QOS_BUCKETS; i++, v >>= WQ_THACTIVE_BUCKET_WIDTH) {
active = v & WQ_THACTIVE_BUCKET_MASK;
@@ -456,24 +366,6 @@
return count;
}
-/* The input qos here should be the requested QoS of the thread, not accounting
- * for any overrides */
-static inline void
-_wq_cooperative_queue_scheduled_count_dec(struct workqueue *wq, thread_qos_t qos)
-{
- __assert_only uint8_t old_scheduled_count = wq->wq_cooperative_queue_scheduled_count[_wq_bucket(qos)]--;
- assert(old_scheduled_count > 0);
-}
-
-/* The input qos here should be the requested QoS of the thread, not accounting
- * for any overrides */
-static inline void
-_wq_cooperative_queue_scheduled_count_inc(struct workqueue *wq, thread_qos_t qos)
-{
- __assert_only uint8_t old_scheduled_count = wq->wq_cooperative_queue_scheduled_count[_wq_bucket(qos)]++;
- assert(old_scheduled_count < UINT8_MAX);
-}
-
#pragma mark wq_flags
static inline uint32_t
@@ -495,43 +387,49 @@
return !wq || _wq_exiting(wq);
}
+struct turnstile *
+workq_turnstile(struct proc *p)
+{
+ struct workqueue *wq = proc_get_wqptr(p);
+ return wq ? wq->wq_turnstile : TURNSTILE_NULL;
+}
#pragma mark workqueue lock
static bool
-workq_lock_is_acquired_kdp(struct workqueue *wq)
-{
- return kdp_lck_ticket_is_acquired(&wq->wq_lock);
+workq_lock_spin_is_acquired_kdp(struct workqueue *wq)
+{
+ return kdp_lck_spin_is_acquired(&wq->wq_lock);
}
static inline void
workq_lock_spin(struct workqueue *wq)
{
- lck_ticket_lock(&wq->wq_lock, &workq_lck_grp);
+ lck_spin_lock(&wq->wq_lock);
}
static inline void
-workq_lock_held(struct workqueue *wq)
-{
- LCK_TICKET_ASSERT_OWNED(&wq->wq_lock);
+workq_lock_held(__assert_only struct workqueue *wq)
+{
+ LCK_SPIN_ASSERT(&wq->wq_lock, LCK_ASSERT_OWNED);
}
static inline bool
workq_lock_try(struct workqueue *wq)
{
- return lck_ticket_lock_try(&wq->wq_lock, &workq_lck_grp);
+ return lck_spin_try_lock(&wq->wq_lock);
}
static inline void
workq_unlock(struct workqueue *wq)
{
- lck_ticket_unlock(&wq->wq_lock);
+ lck_spin_unlock(&wq->wq_lock);
}
#pragma mark idle thread lists
#define WORKQ_POLICY_INIT(qos) \
- (struct uu_workq_policy){ .qos_req = qos, .qos_bucket = qos }
+ (struct uu_workq_policy){ .qos_req = qos, .qos_bucket = qos }
static inline thread_qos_t
workq_pri_bucket(struct uu_workq_policy req)
@@ -551,7 +449,7 @@
workq_threadreq_param_t cur_trp, req_trp = { };
cur_trp.trp_value = uth->uu_save.uus_workq_park_data.workloop_params;
- if (req->tr_flags & WORKQ_TR_FLAG_WL_PARAMS) {
+ if (req->tr_flags & TR_FLAG_WL_PARAMS) {
req_trp = kqueue_threadreq_workloop_param(req);
}
@@ -574,7 +472,7 @@
return true;
}
- if ((req_flags & TRP_POLICY) && req_trp.trp_pol != cur_trp.trp_pol) {
+ if ((req_flags & TRP_POLICY) && cur_trp.trp_pol != cur_trp.trp_pol) {
return true;
}
@@ -588,40 +486,14 @@
return true;
}
- if (req->tr_qos != workq_pri_override(uth->uu_workq_pri)) {
- return true;
- }
-
-#if CONFIG_PREADOPT_TG
- thread_group_qos_t tg = kqr_preadopt_thread_group(req);
- if (KQWL_HAS_VALID_PREADOPTED_TG(tg)) {
- /*
- * Ideally, we'd add check here to see if thread's preadopt TG is same
- * as the thread requests's thread group and short circuit if that is
- * the case. But in the interest of keeping the code clean and not
- * taking the thread lock here, we're going to skip this. We will
- * eventually shortcircuit once we try to set the preadoption thread
- * group on the thread.
- */
- return true;
- }
-#endif
-
- return false;
-}
-
-/* Input thread must be self. Called during self override, resetting overrides
- * or while processing kevents
- *
- * Called with workq lock held. Sometimes also the thread mutex
- */
+ return req->tr_qos != workq_pri_override(uth->uu_workq_pri);
+}
+
static void
workq_thread_update_bucket(proc_t p, struct workqueue *wq, struct uthread *uth,
- struct uu_workq_policy old_pri, struct uu_workq_policy new_pri,
- bool force_run)
-{
- assert(uth == current_uthread());
-
+ struct uu_workq_policy old_pri, struct uu_workq_policy new_pri,
+ bool force_run)
+{
thread_qos_t old_bucket = old_pri.qos_bucket;
thread_qos_t new_bucket = workq_pri_bucket(new_pri);
@@ -632,12 +504,11 @@
new_pri.qos_bucket = new_bucket;
uth->uu_workq_pri = new_pri;
- if (old_pri.qos_override != new_pri.qos_override) {
- thread_set_workq_override(get_machthread(uth), new_pri.qos_override);
- }
-
- if (wq->wq_reqcount &&
- (old_bucket > new_bucket || force_run)) {
+ if (workq_pri_override(old_pri) != new_bucket) {
+ thread_set_workq_override(uth->uu_thread, new_bucket);
+ }
+
+ if (wq->wq_reqcount && (old_bucket > new_bucket || force_run)) {
int flags = WORKQ_THREADREQ_CAN_CREATE_THREADS;
if (old_bucket > new_bucket) {
/*
@@ -665,7 +536,7 @@
assert(uth == current_uthread());
workq_threadreq_param_t trp = { };
- if (req && (req->tr_flags & WORKQ_TR_FLAG_WL_PARAMS)) {
+ if (req && (req->tr_flags & TR_FLAG_WL_PARAMS)) {
trp = kqueue_threadreq_workloop_param(req);
}
@@ -681,47 +552,40 @@
if (trp.trp_flags & TRP_CPUPERCENT) {
thread_set_cpulimit(THREAD_CPULIMIT_BLOCK, trp.trp_cpupercent,
- (uint64_t)trp.trp_refillms * NSEC_PER_SEC);
+ (uint64_t)trp.trp_refillms * NSEC_PER_SEC);
uth->uu_workq_flags |= UT_WORKQ_CPUPERCENT;
}
}
-/*
- * This function is always called with the workq lock, except for the
- * permanently bound workqueue thread, which instead requires the kqlock.
- * See locking model for bound thread's uu_workq_flags.
- */
static void
workq_thread_reset_pri(struct workqueue *wq, struct uthread *uth,
- workq_threadreq_t req, bool unpark)
-{
- thread_t th = get_machthread(uth);
+ workq_threadreq_t req)
+{
+ thread_t th = uth->uu_thread;
thread_qos_t qos = req ? req->tr_qos : WORKQ_THREAD_QOS_CLEANUP;
workq_threadreq_param_t trp = { };
int priority = 31;
int policy = POLICY_TIMESHARE;
- if (req && (req->tr_flags & WORKQ_TR_FLAG_WL_PARAMS)) {
+ if (req && (req->tr_flags & TR_FLAG_WL_PARAMS)) {
trp = kqueue_threadreq_workloop_param(req);
}
uth->uu_workq_pri = WORKQ_POLICY_INIT(qos);
uth->uu_workq_flags &= ~UT_WORKQ_OUTSIDE_QOS;
-
- if (unpark) {
- uth->uu_save.uus_workq_park_data.workloop_params = trp.trp_value;
- // qos sent out to userspace (may differ from uu_workq_pri on param threads)
- uth->uu_save.uus_workq_park_data.qos = qos;
- }
+ uth->uu_save.uus_workq_park_data.workloop_params = trp.trp_value;
+
+ // qos sent out to userspace (may differ from uu_workq_pri on param threads)
+ uth->uu_save.uus_workq_park_data.qos = qos;
if (qos == WORKQ_THREAD_QOS_MANAGER) {
uint32_t mgr_pri = wq->wq_event_manager_priority;
assert(trp.trp_value == 0); // manager qos and thread policy don't mix
- if (_pthread_priority_has_sched_pri(mgr_pri)) {
+ if (mgr_pri & _PTHREAD_PRIORITY_SCHED_PRI_FLAG) {
mgr_pri &= _PTHREAD_PRIORITY_SCHED_PRI_MASK;
thread_set_workq_pri(th, THREAD_QOS_UNSPECIFIED, mgr_pri,
- POLICY_TIMESHARE);
+ POLICY_TIMESHARE);
return;
}
@@ -738,85 +602,6 @@
}
}
-#if CONFIG_PREADOPT_TG
- if (req && (req->tr_flags & WORKQ_TR_FLAG_WORKLOOP)) {
- /*
- * For kqwl permanently configured with a thread group, we can safely borrow
- * +1 ref from kqwl_preadopt_tg. A thread then takes additional +1 ref
- * for itself via thread_set_preadopt_thread_group.
- *
- * In all other cases, we cannot safely read and borrow the reference from the kqwl
- * since it can disappear from under us at any time due to the max-ing logic in
- * kqueue_set_preadopted_thread_group.
- *
- * As such, we do the following dance:
- *
- * 1) cmpxchng and steal the kqwl's preadopt thread group and leave
- * behind with (NULL + QoS). At this point, we have the reference
- * to the thread group from the kqwl.
- * 2) Have the thread set the preadoption thread group on itself.
- * 3) cmpxchng from (NULL + QoS) which we set earlier in (1), back to
- * thread_group + QoS. ie we try to give the reference back to the kqwl.
- * If we fail, that's because a higher QoS thread group was set on the
- * kqwl in kqueue_set_preadopted_thread_group in which case, we need to
- * go back to (1).
- */
-
- _Atomic(struct thread_group *) * tg_loc = kqr_preadopt_thread_group_addr(req);
-
- thread_group_qos_t old_tg, new_tg;
- int ret = 0;
-again:
- ret = os_atomic_rmw_loop(tg_loc, old_tg, new_tg, relaxed, {
- if ((!KQWL_HAS_VALID_PREADOPTED_TG(old_tg)) ||
- KQWL_HAS_PERMANENT_PREADOPTED_TG(old_tg)) {
- os_atomic_rmw_loop_give_up(break);
- }
-
- /*
- * Leave the QoS behind - kqueue_set_preadopted_thread_group will
- * only modify it if there is a higher QoS thread group to attach
- */
- new_tg = (thread_group_qos_t) ((uintptr_t) old_tg & KQWL_PREADOPT_TG_QOS_MASK);
- });
-
- if (ret) {
- /*
- * We successfully took the ref from the kqwl so set it on the
- * thread now
- */
- thread_set_preadopt_thread_group(th, KQWL_GET_PREADOPTED_TG(old_tg));
-
- thread_group_qos_t thread_group_to_expect = new_tg;
- thread_group_qos_t thread_group_to_set = old_tg;
-
- os_atomic_rmw_loop(tg_loc, old_tg, new_tg, relaxed, {
- if (old_tg != thread_group_to_expect) {
- /*
- * There was an intervening write to the kqwl_preadopt_tg,
- * and it has a higher QoS than what we are working with
- * here. Abandon our current adopted thread group and redo
- * the full dance
- */
- thread_group_deallocate_safe(KQWL_GET_PREADOPTED_TG(thread_group_to_set));
- os_atomic_rmw_loop_give_up(goto again);
- }
-
- new_tg = thread_group_to_set;
- });
- } else {
- if (KQWL_HAS_PERMANENT_PREADOPTED_TG(old_tg)) {
- thread_set_preadopt_thread_group(th, KQWL_GET_PREADOPTED_TG(old_tg));
- } else {
- /* Nothing valid on the kqwl, just clear what's on the thread */
- thread_set_preadopt_thread_group(th, NULL);
- }
- }
- } else {
- /* Not even a kqwl, clear what's on the thread */
- thread_set_preadopt_thread_group(th, NULL);
- }
-#endif
thread_set_workq_pri(th, qos, priority, policy);
}
@@ -825,16 +610,15 @@
* every time a servicer is being told about a new max QoS.
*/
void
-workq_thread_set_max_qos(struct proc *p, workq_threadreq_t kqr)
+workq_thread_set_max_qos(struct proc *p, struct kqrequest *kqr)
{
struct uu_workq_policy old_pri, new_pri;
- struct uthread *uth = current_uthread();
+ struct uthread *uth = get_bsdthread_info(kqr->kqr_thread);
struct workqueue *wq = proc_get_wqptr_fast(p);
- thread_qos_t qos = kqr->tr_kq_qos_index;
-
- if (uth->uu_workq_pri.qos_max == qos) {
+ thread_qos_t qos = kqr->kqr_qos_index;
+
+ if (uth->uu_workq_pri.qos_max == qos)
return;
- }
workq_lock_spin(wq);
old_pri = new_pri = uth->uu_workq_pri;
@@ -883,7 +667,7 @@
static inline bool
workq_should_kill_idle_thread(struct workqueue *wq, struct uthread *uth,
- uint64_t now)
+ uint64_t now)
{
uint64_t delay = workq_kill_delay_for_idle_thread(wq);
return now - uth->uu_save.uus_workq_park_data.idle_stamp > delay;
@@ -899,7 +683,7 @@
}
os_atomic_or(&wq->wq_flags, WQ_DEATH_CALL_SCHEDULED, relaxed);
- WQ_TRACE_WQ(TRACE_wq_death_call | DBG_FUNC_NONE, wq, 1, 0, 0);
+ WQ_TRACE_WQ(TRACE_wq_death_call | DBG_FUNC_NONE, wq, 1, 0, 0, 0);
/*
* <rdar://problem/13139182> Due to how long term timers work, the leeway
@@ -908,8 +692,8 @@
* fall into long-term timer list shenanigans.
*/
thread_call_enter_delayed_with_leeway(wq->wq_death_call, NULL, deadline,
- wq_reduce_pool_window.abstime / 10,
- THREAD_CALL_DELAY_LEEWAY | THREAD_CALL_DELAY_USER_BACKGROUND);
+ wq_reduce_pool_window.abstime / 10,
+ THREAD_CALL_DELAY_LEEWAY | THREAD_CALL_DELAY_USER_BACKGROUND);
}
/*
@@ -923,34 +707,29 @@
struct uthread *uth;
assert(wq->wq_thdying_count >= decrement);
- if ((wq->wq_thdying_count -= decrement) > 0) {
+ if ((wq->wq_thdying_count -= decrement) > 0)
return;
- }
-
- if (wq->wq_thidlecount <= 1) {
+
+ if (wq->wq_thidlecount <= 1)
return;
- }
-
- if ((uth = workq_oldest_killable_idle_thread(wq)) == NULL) {
+
+ if ((uth = workq_oldest_killable_idle_thread(wq)) == NULL)
return;
- }
uint64_t now = mach_absolute_time();
uint64_t delay = workq_kill_delay_for_idle_thread(wq);
if (now - uth->uu_save.uus_workq_park_data.idle_stamp > delay) {
WQ_TRACE_WQ(TRACE_wq_thread_terminate | DBG_FUNC_START,
- wq, wq->wq_thidlecount, 0, 0);
+ wq, wq->wq_thidlecount, 0, 0, 0);
wq->wq_thdying_count++;
uth->uu_workq_flags |= UT_WORKQ_DYING;
- if ((uth->uu_workq_flags & UT_WORKQ_IDLE_CLEANUP) == 0) {
- workq_thread_wakeup(uth);
- }
+ workq_thread_wakeup(uth);
return;
}
workq_death_call_schedule(wq,
- uth->uu_save.uus_workq_park_data.idle_stamp + delay);
+ uth->uu_save.uus_workq_park_data.idle_stamp + delay);
}
void
@@ -959,13 +738,11 @@
struct workqueue *wq = proc_get_wqptr_fast(p);
workq_lock_spin(wq);
- if (!workq_thread_is_permanently_bound(uth)) {
- TAILQ_REMOVE(&wq->wq_thrunlist, uth, uu_workq_entry);
- if (uth->uu_workq_flags & UT_WORKQ_DYING) {
- WQ_TRACE_WQ(TRACE_wq_thread_terminate | DBG_FUNC_END,
- wq, wq->wq_thidlecount, 0, 0);
- workq_death_policy_evaluate(wq, 1);
- }
+ TAILQ_REMOVE(&wq->wq_thrunlist, uth, uu_workq_entry);
+ if (uth->uu_workq_flags & UT_WORKQ_DYING) {
+ WQ_TRACE_WQ(TRACE_wq_thread_terminate | DBG_FUNC_END,
+ wq, wq->wq_thidlecount, 0, 0, 0);
+ workq_death_policy_evaluate(wq, 1);
}
if (wq->wq_nthreads-- == wq_max_threads) {
/*
@@ -978,7 +755,7 @@
}
workq_unlock(wq);
- thread_deallocate(get_machthread(uth));
+ thread_deallocate(uth->uu_thread);
}
static void
@@ -987,16 +764,15 @@
struct workqueue *wq = param0;
workq_lock_spin(wq);
- WQ_TRACE_WQ(TRACE_wq_death_call | DBG_FUNC_START, wq, 0, 0, 0);
- os_atomic_andnot(&wq->wq_flags, WQ_DEATH_CALL_SCHEDULED, relaxed);
+ WQ_TRACE_WQ(TRACE_wq_death_call | DBG_FUNC_START, wq, 0, 0, 0, 0);
+ os_atomic_and(&wq->wq_flags, ~WQ_DEATH_CALL_SCHEDULED, relaxed);
workq_death_policy_evaluate(wq, 0);
- WQ_TRACE_WQ(TRACE_wq_death_call | DBG_FUNC_END, wq, 0, 0, 0);
+ WQ_TRACE_WQ(TRACE_wq_death_call | DBG_FUNC_END, wq, 0, 0, 0, 0);
workq_unlock(wq);
}
static struct uthread *
-workq_pop_idle_thread(struct workqueue *wq, uint16_t uu_flags,
- bool *needs_wakeup)
+workq_pop_idle_thread(struct workqueue *wq)
{
struct uthread *uth;
@@ -1009,25 +785,13 @@
TAILQ_INSERT_TAIL(&wq->wq_thrunlist, uth, uu_workq_entry);
assert((uth->uu_workq_flags & UT_WORKQ_RUNNING) == 0);
- uth->uu_workq_flags |= UT_WORKQ_RUNNING | uu_flags;
-
- /* A thread is never woken up as part of the cooperative pool */
- assert((uu_flags & UT_WORKQ_COOPERATIVE) == 0);
-
- if ((uu_flags & UT_WORKQ_OVERCOMMIT) == 0) {
- wq->wq_constrained_threads_scheduled++;
- }
+ uth->uu_workq_flags |= UT_WORKQ_RUNNING | UT_WORKQ_OVERCOMMIT;
wq->wq_threads_scheduled++;
wq->wq_thidlecount--;
if (__improbable(uth->uu_workq_flags & UT_WORKQ_DYING)) {
uth->uu_workq_flags ^= UT_WORKQ_DYING;
workq_death_policy_evaluate(wq, 1);
- *needs_wakeup = false;
- } else if (uth->uu_workq_flags & UT_WORKQ_IDLE_CLEANUP) {
- *needs_wakeup = false;
- } else {
- *needs_wakeup = true;
}
return uth;
}
@@ -1045,7 +809,6 @@
uth->uu_workq_pri = WORKQ_POLICY_INIT(THREAD_QOS_LEGACY);
uth->uu_workq_thport = MACH_PORT_NULL;
uth->uu_workq_stackaddr = 0;
- uth->uu_workq_pthread_kill_allowed = 0;
thread_set_tag(th, THREAD_TAG_PTHREAD | THREAD_TAG_WORKQUEUE);
thread_reset_workq_qos(th, THREAD_QOS_LEGACY);
@@ -1061,13 +824,8 @@
* - dropped and retaken around thread creation
* - return with workq lock held
*/
-static kern_return_t
-workq_add_new_idle_thread(
- proc_t p,
- struct workqueue *wq,
- thread_continue_t continuation,
- bool is_permanently_bound,
- thread_t *new_thread)
+static bool
+workq_add_new_idle_thread(proc_t p, struct workqueue *wq)
{
mach_vm_offset_t th_stackaddr;
kern_return_t kret;
@@ -1077,22 +835,19 @@
workq_unlock(wq);
- vm_map_t vmap = get_task_map(proc_task(p));
+ vm_map_t vmap = get_task_map(p->task);
kret = pthread_functions->workq_create_threadstack(p, vmap, &th_stackaddr);
if (kret != KERN_SUCCESS) {
WQ_TRACE_WQ(TRACE_wq_thread_create_failed | DBG_FUNC_NONE, wq,
- kret, 1, 0);
+ kret, 1, 0, 0);
goto out;
}
- kret = thread_create_workq_waiting(proc_task(p),
- continuation,
- &th,
- is_permanently_bound);
+ kret = thread_create_workq_waiting(p->task, workq_unpark_continue, &th);
if (kret != KERN_SUCCESS) {
WQ_TRACE_WQ(TRACE_wq_thread_create_failed | DBG_FUNC_NONE, wq,
- kret, 0, 0);
+ kret, 0, 0, 0);
pthread_functions->workq_destroy_threadstack(p, vmap, th_stackaddr);
goto out;
}
@@ -1101,20 +856,14 @@
// on success, because it calls workq_thread_init_and_wq_lock() above
struct uthread *uth = get_bsdthread_info(th);
- uth->uu_workq_stackaddr = (user_addr_t)th_stackaddr;
wq->wq_creations++;
- if (!is_permanently_bound) {
- wq->wq_thidlecount++;
- TAILQ_INSERT_TAIL(&wq->wq_thnewlist, uth, uu_workq_entry);
- }
-
- if (new_thread) {
- *new_thread = th;
- }
-
- WQ_TRACE_WQ(TRACE_wq_thread_create | DBG_FUNC_NONE, wq, 0, 0, 0);
- return kret;
+ wq->wq_thidlecount++;
+ uth->uu_workq_stackaddr = th_stackaddr;
+ TAILQ_INSERT_TAIL(&wq->wq_thnewlist, uth, uu_workq_entry);
+
+ WQ_TRACE_WQ(TRACE_wq_thread_create | DBG_FUNC_NONE, wq, 0, 0, 0, 0);
+ return true;
out:
workq_lock_spin(wq);
@@ -1124,54 +873,21 @@
* to do so when it fails.
*/
wq->wq_nthreads--;
- return kret;
-}
-
-static inline bool
-workq_thread_is_overcommit(struct uthread *uth)
-{
- return (uth->uu_workq_flags & UT_WORKQ_OVERCOMMIT) != 0;
-}
-
-static inline bool
-workq_thread_is_nonovercommit(struct uthread *uth)
-{
- return (uth->uu_workq_flags & (UT_WORKQ_OVERCOMMIT |
- UT_WORKQ_COOPERATIVE)) == 0;
-}
-
-static inline bool
-workq_thread_is_cooperative(struct uthread *uth)
-{
- return (uth->uu_workq_flags & UT_WORKQ_COOPERATIVE) != 0;
-}
-
-bool
-workq_thread_is_permanently_bound(struct uthread *uth)
-{
- return (uth->uu_workq_flags & UT_WORKQ_PERMANENT_BIND) != 0;
-}
-
-static inline void
-workq_thread_set_type(struct uthread *uth, uint16_t flags)
-{
- uth->uu_workq_flags &= ~(UT_WORKQ_OVERCOMMIT | UT_WORKQ_COOPERATIVE);
- uth->uu_workq_flags |= flags;
-}
-
+ return false;
+}
#define WORKQ_UNPARK_FOR_DEATH_WAS_IDLE 0x1
__attribute__((noreturn, noinline))
static void
workq_unpark_for_death_and_unlock(proc_t p, struct workqueue *wq,
- struct uthread *uth, uint32_t death_flags, uint32_t setup_flags)
+ struct uthread *uth, uint32_t death_flags)
{
thread_qos_t qos = workq_pri_override(uth->uu_workq_pri);
bool first_use = uth->uu_workq_flags & UT_WORKQ_NEW;
if (qos > WORKQ_THREAD_QOS_CLEANUP) {
- workq_thread_reset_pri(wq, uth, NULL, /*unpark*/ true);
+ workq_thread_reset_pri(wq, uth, NULL);
qos = WORKQ_THREAD_QOS_CLEANUP;
}
@@ -1189,22 +905,15 @@
workq_unlock(wq);
- if (setup_flags & WQ_SETUP_CLEAR_VOUCHER) {
- __assert_only kern_return_t kr;
- kr = thread_set_voucher_name(MACH_PORT_NULL);
- assert(kr == KERN_SUCCESS);
- }
-
uint32_t flags = WQ_FLAG_THREAD_NEWSPI | qos | WQ_FLAG_THREAD_PRIO_QOS;
- thread_t th = get_machthread(uth);
- vm_map_t vmap = get_task_map(proc_task(p));
-
- if (!first_use) {
- flags |= WQ_FLAG_THREAD_REUSE;
- }
+ uint32_t setup_flags = WQ_SETUP_EXIT_THREAD;
+ thread_t th = uth->uu_thread;
+ vm_map_t vmap = get_task_map(p->task);
+
+ if (!first_use) flags |= WQ_FLAG_THREAD_REUSE;
pthread_functions->workq_setup_thread(p, th, vmap, uth->uu_workq_stackaddr,
- uth->uu_workq_thport, 0, WQ_SETUP_EXIT_THREAD, flags);
+ uth->uu_workq_thport, 0, setup_flags, flags);
__builtin_unreachable();
}
@@ -1217,7 +926,7 @@
__attribute__((always_inline))
static inline void
workq_perform_turnstile_operation_locked(struct workqueue *wq,
- void (^operation)(void))
+ void (^operation)(void))
{
workq_lock_held(wq);
wq->wq_turnstile_updater = current_thread();
@@ -1227,76 +936,47 @@
static void
workq_turnstile_update_inheritor(struct workqueue *wq,
- turnstile_inheritor_t inheritor,
- turnstile_update_flags_t flags)
-{
- if (wq->wq_inheritor == inheritor) {
- return;
- }
- wq->wq_inheritor = inheritor;
+ turnstile_inheritor_t inheritor,
+ turnstile_update_flags_t flags)
+{
workq_perform_turnstile_operation_locked(wq, ^{
turnstile_update_inheritor(wq->wq_turnstile, inheritor,
- flags | TURNSTILE_IMMEDIATE_UPDATE);
+ flags | TURNSTILE_IMMEDIATE_UPDATE);
turnstile_update_inheritor_complete(wq->wq_turnstile,
- TURNSTILE_INTERLOCK_HELD);
+ TURNSTILE_INTERLOCK_HELD);
});
}
static void
-workq_push_idle_thread(proc_t p, struct workqueue *wq, struct uthread *uth,
- uint32_t setup_flags)
+workq_push_idle_thread(proc_t p, struct workqueue *wq, struct uthread *uth)
{
uint64_t now = mach_absolute_time();
- bool is_creator = (uth == wq->wq_creator);
-
- if (workq_thread_is_cooperative(uth)) {
- assert(!is_creator);
-
- thread_qos_t thread_qos = uth->uu_workq_pri.qos_req;
- _wq_cooperative_queue_scheduled_count_dec(wq, thread_qos);
-
- /* Before we get here, we always go through
- * workq_select_threadreq_or_park_and_unlock. If we got here, it means
- * that we went through the logic in workq_threadreq_select which
- * did the refresh for the next best cooperative qos while
- * excluding the current thread - we shouldn't need to do it again.
- */
- assert(_wq_cooperative_queue_refresh_best_req_qos(wq) == false);
- } else if (workq_thread_is_nonovercommit(uth)) {
- assert(!is_creator);
-
+
+ uth->uu_workq_flags &= ~UT_WORKQ_RUNNING;
+ if ((uth->uu_workq_flags & UT_WORKQ_OVERCOMMIT) == 0) {
wq->wq_constrained_threads_scheduled--;
}
-
- uth->uu_workq_flags &= ~(UT_WORKQ_RUNNING | UT_WORKQ_OVERCOMMIT | UT_WORKQ_COOPERATIVE);
TAILQ_REMOVE(&wq->wq_thrunlist, uth, uu_workq_entry);
wq->wq_threads_scheduled--;
- if (is_creator) {
+ if (wq->wq_creator == uth) {
+ WQ_TRACE_WQ(TRACE_wq_creator_select, wq, 3, 0,
+ uth->uu_save.uus_workq_park_data.yields, 0);
wq->wq_creator = NULL;
- WQ_TRACE_WQ(TRACE_wq_creator_select, wq, 3, 0,
- uth->uu_save.uus_workq_park_data.yields);
- }
-
- if (wq->wq_inheritor == get_machthread(uth)) {
- assert(wq->wq_creator == NULL);
if (wq->wq_reqcount) {
workq_turnstile_update_inheritor(wq, wq, TURNSTILE_INHERITOR_WORKQ);
} else {
workq_turnstile_update_inheritor(wq, TURNSTILE_INHERITOR_NULL, 0);
}
- }
-
- if (uth->uu_workq_flags & UT_WORKQ_NEW) {
- assert(is_creator || (_wq_flags(wq) & WQ_EXITING));
- TAILQ_INSERT_TAIL(&wq->wq_thnewlist, uth, uu_workq_entry);
- wq->wq_thidlecount++;
- return;
- }
-
- if (!is_creator) {
+ if (uth->uu_workq_flags & UT_WORKQ_NEW) {
+ TAILQ_INSERT_TAIL(&wq->wq_thnewlist, uth, uu_workq_entry);
+ wq->wq_thidlecount++;
+ return;
+ }
+ } else {
_wq_thactive_dec(wq, uth->uu_workq_pri.qos_bucket);
wq->wq_thscheduled_count[_wq_bucket(uth->uu_workq_pri.qos_bucket)]--;
+ assert(!(uth->uu_workq_flags & UT_WORKQ_NEW));
uth->uu_workq_flags |= UT_WORKQ_IDLE_CLEANUP;
}
@@ -1306,8 +986,8 @@
uint16_t cur_idle = wq->wq_thidlecount;
if (cur_idle >= wq_max_constrained_threads ||
- (wq->wq_thdying_count == 0 && oldest &&
- workq_should_kill_idle_thread(wq, oldest, now))) {
+ (wq->wq_thdying_count == 0 && oldest &&
+ workq_should_kill_idle_thread(wq, oldest, now))) {
/*
* Immediately kill threads if we have too may of them.
*
@@ -1323,11 +1003,11 @@
}
WQ_TRACE_WQ(TRACE_wq_thread_terminate | DBG_FUNC_START,
- wq, cur_idle, 0, 0);
+ wq, cur_idle, 0, 0, 0);
wq->wq_thdying_count++;
uth->uu_workq_flags |= UT_WORKQ_DYING;
uth->uu_workq_flags &= ~UT_WORKQ_IDLE_CLEANUP;
- workq_unpark_for_death_and_unlock(p, wq, uth, 0, setup_flags);
+ workq_unpark_for_death_and_unlock(p, wq, uth, 0);
__builtin_unreachable();
}
@@ -1337,7 +1017,7 @@
wq->wq_thidlecount = cur_idle;
if (cur_idle >= wq_death_max_load && tail &&
- tail->uu_save.uus_workq_park_data.has_stack) {
+ tail->uu_save.uus_workq_park_data.has_stack) {
uth->uu_save.uus_workq_park_data.has_stack = false;
TAILQ_INSERT_TAIL(&wq->wq_thidlelist, uth, uu_workq_entry);
} else {
@@ -1353,36 +1033,12 @@
#pragma mark thread requests
-static inline bool
-workq_tr_is_overcommit(workq_tr_flags_t tr_flags)
-{
- return (tr_flags & WORKQ_TR_FLAG_OVERCOMMIT) != 0;
-}
-
-static inline bool
-workq_tr_is_nonovercommit(workq_tr_flags_t tr_flags)
-{
- return (tr_flags & (WORKQ_TR_FLAG_OVERCOMMIT |
- WORKQ_TR_FLAG_COOPERATIVE |
- WORKQ_TR_FLAG_PERMANENT_BIND)) == 0;
-}
-
-static inline bool
-workq_tr_is_cooperative(workq_tr_flags_t tr_flags)
-{
- return (tr_flags & WORKQ_TR_FLAG_COOPERATIVE) != 0;
-}
-
-#define workq_threadreq_is_overcommit(req) workq_tr_is_overcommit((req)->tr_flags)
-#define workq_threadreq_is_nonovercommit(req) workq_tr_is_nonovercommit((req)->tr_flags)
-#define workq_threadreq_is_cooperative(req) workq_tr_is_cooperative((req)->tr_flags)
-
static inline int
workq_priority_for_req(workq_threadreq_t req)
{
thread_qos_t qos = req->tr_qos;
- if (req->tr_flags & WORKQ_TR_FLAG_WL_OUTSIDE_QOS) {
+ if (req->tr_flags & TR_FLAG_WL_OUTSIDE_QOS) {
workq_threadreq_param_t trp = kqueue_threadreq_workloop_param(req);
assert(trp.trp_flags & TRP_PRIORITY);
return trp.trp_pri;
@@ -1390,273 +1046,41 @@
return thread_workq_pri_for_qos(qos);
}
-static inline struct priority_queue_sched_max *
+static inline struct priority_queue *
workq_priority_queue_for_req(struct workqueue *wq, workq_threadreq_t req)
{
- assert(!workq_tr_is_cooperative(req->tr_flags));
-
- if (req->tr_flags & WORKQ_TR_FLAG_WL_OUTSIDE_QOS) {
+ if (req->tr_flags & TR_FLAG_WL_OUTSIDE_QOS) {
return &wq->wq_special_queue;
- } else if (workq_tr_is_overcommit(req->tr_flags)) {
+ } else if (req->tr_flags & TR_FLAG_OVERCOMMIT) {
return &wq->wq_overcommit_queue;
} else {
return &wq->wq_constrained_queue;
}
}
-/* Calculates the number of threads scheduled >= the input QoS */
-static uint64_t
-workq_num_cooperative_threads_scheduled_to_qos_internal(struct workqueue *wq, thread_qos_t qos)
-{
- uint64_t num_cooperative_threads = 0;
-
- for (thread_qos_t cur_qos = WORKQ_THREAD_QOS_MAX; cur_qos >= qos; cur_qos--) {
- uint8_t bucket = _wq_bucket(cur_qos);
- num_cooperative_threads += wq->wq_cooperative_queue_scheduled_count[bucket];
- }
-
- return num_cooperative_threads;
-}
-
-/* Calculates the number of threads scheduled >= the input QoS */
-static uint64_t
-workq_num_cooperative_threads_scheduled_to_qos_locked(struct workqueue *wq, thread_qos_t qos)
-{
- workq_lock_held(wq);
- return workq_num_cooperative_threads_scheduled_to_qos_internal(wq, qos);
-}
-
-static uint64_t
-workq_num_cooperative_threads_scheduled_total(struct workqueue *wq)
-{
- return workq_num_cooperative_threads_scheduled_to_qos_locked(wq, WORKQ_THREAD_QOS_MIN);
-}
-
-static bool
-workq_has_cooperative_thread_requests(struct workqueue *wq)
-{
- for (thread_qos_t qos = WORKQ_THREAD_QOS_MAX; qos >= WORKQ_THREAD_QOS_MIN; qos--) {
- uint8_t bucket = _wq_bucket(qos);
- if (!STAILQ_EMPTY(&wq->wq_cooperative_queue[bucket])) {
- return true;
- }
- }
-
- return false;
-}
-
/*
- * Determines the next QoS bucket we should service next in the cooperative
- * pool. This function will always return a QoS for cooperative pool as long as
- * there are requests to be serviced.
- *
- * Unlike the other thread pools, for the cooperative thread pool the schedule
- * counts for the various buckets in the pool affect the next best request for
- * it.
- *
- * This function is called in the following contexts:
- *
- * a) When determining the best thread QoS for cooperative bucket for the
- * creator/thread reuse
- *
- * b) Once (a) has happened and thread has bound to a thread request, figuring
- * out whether the next best request for this pool has changed so that creator
- * can be scheduled.
- *
- * Returns true if the cooperative queue's best qos changed from previous
- * value.
- */
-static bool
-_wq_cooperative_queue_refresh_best_req_qos(struct workqueue *wq)
-{
- workq_lock_held(wq);
-
- thread_qos_t old_best_req_qos = wq->wq_cooperative_queue_best_req_qos;
-
- /* We determine the next best cooperative thread request based on the
- * following:
- *
- * 1. Take the MAX of the following:
- * a) Highest qos with pending TRs such that number of scheduled
- * threads so far with >= qos is < wq_max_cooperative_threads
- * b) Highest qos bucket with pending TRs but no scheduled threads for that bucket
- *
- * 2. If the result of (1) is UN, then we pick the highest priority amongst
- * pending thread requests in the pool.
- *
- */
- thread_qos_t highest_qos_with_no_scheduled = THREAD_QOS_UNSPECIFIED;
- thread_qos_t highest_qos_req_with_width = THREAD_QOS_UNSPECIFIED;
-
- thread_qos_t highest_qos_req = THREAD_QOS_UNSPECIFIED;
-
- int scheduled_count_till_qos = 0;
-
- for (thread_qos_t qos = WORKQ_THREAD_QOS_MAX; qos >= WORKQ_THREAD_QOS_MIN; qos--) {
- uint8_t bucket = _wq_bucket(qos);
- uint8_t scheduled_count_for_bucket = wq->wq_cooperative_queue_scheduled_count[bucket];
- scheduled_count_till_qos += scheduled_count_for_bucket;
-
- if (!STAILQ_EMPTY(&wq->wq_cooperative_queue[bucket])) {
- if (qos > highest_qos_req) {
- highest_qos_req = qos;
- }
- /*
- * The pool isn't saturated for threads at and above this QoS, and
- * this qos bucket has pending requests
- */
- if (scheduled_count_till_qos < wq_cooperative_queue_max_size(wq)) {
- if (qos > highest_qos_req_with_width) {
- highest_qos_req_with_width = qos;
- }
- }
-
- /*
- * There are no threads scheduled for this bucket but there
- * is work pending, give it at least 1 thread
- */
- if (scheduled_count_for_bucket == 0) {
- if (qos > highest_qos_with_no_scheduled) {
- highest_qos_with_no_scheduled = qos;
- }
- }
- }
- }
-
- wq->wq_cooperative_queue_best_req_qos = MAX(highest_qos_with_no_scheduled, highest_qos_req_with_width);
- if (wq->wq_cooperative_queue_best_req_qos == THREAD_QOS_UNSPECIFIED) {
- wq->wq_cooperative_queue_best_req_qos = highest_qos_req;
- }
-
-#if MACH_ASSERT
- /* Assert that if we are showing up the next best req as UN, then there
- * actually is no thread request in the cooperative pool buckets */
- if (wq->wq_cooperative_queue_best_req_qos == THREAD_QOS_UNSPECIFIED) {
- assert(!workq_has_cooperative_thread_requests(wq));
- }
-#endif
-
- return old_best_req_qos != wq->wq_cooperative_queue_best_req_qos;
-}
-
-/*
- * Returns whether or not the input thread (or creator thread if uth is NULL)
- * should be allowed to work as part of the cooperative pool for the <input qos>
- * bucket.
- *
- * This function is called in a bunch of places:
- * a) Quantum expires for a thread and it is part of the cooperative pool
- * b) When trying to pick a thread request for the creator thread to
- * represent.
- * c) When a thread is trying to pick a thread request to actually bind to
- * and service.
- *
- * Called with workq lock held.
- */
-
-#define WQ_COOPERATIVE_POOL_UNSATURATED 1
-#define WQ_COOPERATIVE_BUCKET_UNSERVICED 2
-#define WQ_COOPERATIVE_POOL_SATURATED_UP_TO_QOS 3
-
-static bool
-workq_cooperative_allowance(struct workqueue *wq, thread_qos_t qos, struct uthread *uth,
- bool may_start_timer)
-{
- workq_lock_held(wq);
-
- bool exclude_thread_as_scheduled = false;
- bool passed_admissions = false;
- uint8_t bucket = _wq_bucket(qos);
-
- if (uth && workq_thread_is_cooperative(uth)) {
- exclude_thread_as_scheduled = true;
- _wq_cooperative_queue_scheduled_count_dec(wq, uth->uu_workq_pri.qos_req);
- }
-
- /*
- * We have not saturated the pool yet, let this thread continue
- */
- uint64_t total_cooperative_threads;
- total_cooperative_threads = workq_num_cooperative_threads_scheduled_total(wq);
- if (total_cooperative_threads < wq_cooperative_queue_max_size(wq)) {
- passed_admissions = true;
- WQ_TRACE(TRACE_wq_cooperative_admission | DBG_FUNC_NONE,
- total_cooperative_threads, qos, passed_admissions,
- WQ_COOPERATIVE_POOL_UNSATURATED);
- goto out;
- }
-
- /*
- * Without this thread, nothing is servicing the bucket which has pending
- * work
- */
- uint64_t bucket_scheduled = wq->wq_cooperative_queue_scheduled_count[bucket];
- if (bucket_scheduled == 0 &&
- !STAILQ_EMPTY(&wq->wq_cooperative_queue[bucket])) {
- passed_admissions = true;
- WQ_TRACE(TRACE_wq_cooperative_admission | DBG_FUNC_NONE,
- total_cooperative_threads, qos, passed_admissions,
- WQ_COOPERATIVE_BUCKET_UNSERVICED);
- goto out;
- }
-
- /*
- * If number of threads at the QoS bucket >= input QoS exceeds the max we want
- * for the pool, deny this thread
- */
- uint64_t aggregate_down_to_qos = workq_num_cooperative_threads_scheduled_to_qos_locked(wq, qos);
- passed_admissions = (aggregate_down_to_qos < wq_cooperative_queue_max_size(wq));
- WQ_TRACE(TRACE_wq_cooperative_admission | DBG_FUNC_NONE, aggregate_down_to_qos,
- qos, passed_admissions, WQ_COOPERATIVE_POOL_SATURATED_UP_TO_QOS);
-
- if (!passed_admissions && may_start_timer) {
- workq_schedule_delayed_thread_creation(wq, 0);
- }
-
-out:
- if (exclude_thread_as_scheduled) {
- _wq_cooperative_queue_scheduled_count_inc(wq, uth->uu_workq_pri.qos_req);
- }
- return passed_admissions;
-}
-
-/*
- * returns true if the best request for the pool changed as a result of
- * enqueuing this thread request.
+ * returns true if the the enqueued request is the highest priority item
+ * in its priority queue.
*/
static bool
workq_threadreq_enqueue(struct workqueue *wq, workq_threadreq_t req)
{
- assert(req->tr_state == WORKQ_TR_STATE_NEW);
-
- req->tr_state = WORKQ_TR_STATE_QUEUED;
+ assert(req->tr_state == TR_STATE_NEW);
+
+ req->tr_state = TR_STATE_QUEUED;
wq->wq_reqcount += req->tr_count;
if (req->tr_qos == WORKQ_THREAD_QOS_MANAGER) {
assert(wq->wq_event_manager_threadreq == NULL);
- assert(req->tr_flags & WORKQ_TR_FLAG_KEVENT);
+ assert(req->tr_flags & TR_FLAG_KEVENT);
assert(req->tr_count == 1);
wq->wq_event_manager_threadreq = req;
return true;
}
-
- if (workq_threadreq_is_cooperative(req)) {
- assert(req->tr_qos != WORKQ_THREAD_QOS_MANAGER);
- assert(req->tr_qos != WORKQ_THREAD_QOS_ABOVEUI);
-
- struct workq_threadreq_tailq *bucket = &wq->wq_cooperative_queue[_wq_bucket(req->tr_qos)];
- STAILQ_INSERT_TAIL(bucket, req, tr_link);
-
- return _wq_cooperative_queue_refresh_best_req_qos(wq);
- }
-
- struct priority_queue_sched_max *q = workq_priority_queue_for_req(wq, req);
-
- priority_queue_entry_set_sched_pri(q, &req->tr_entry,
- workq_priority_for_req(req), false);
-
- if (priority_queue_insert(q, &req->tr_entry)) {
- if (workq_threadreq_is_nonovercommit(req)) {
+ if (priority_queue_insert(workq_priority_queue_for_req(wq, req),
+ &req->tr_entry, workq_priority_for_req(req),
+ PRIORITY_QUEUE_SCHED_PRI_MAX_HEAP_COMPARE)) {
+ if ((req->tr_flags & TR_FLAG_OVERCOMMIT) == 0) {
_wq_thactive_refresh_best_constrained_req_qos(wq);
}
return true;
@@ -1665,120 +1089,158 @@
}
/*
- * returns true if one of the following is true (so as to update creator if
- * needed):
- *
- * (a) the next highest request of the pool we dequeued the request from changed
- * (b) the next highest requests of the pool the current thread used to be a
- * part of, changed
- *
- * For overcommit, special and constrained pools, the next highest QoS for each
- * pool just a MAX of pending requests so tracking (a) is sufficient.
- *
- * But for cooperative thread pool, the next highest QoS for the pool depends on
- * schedule counts in the pool as well. So if the current thread used to be
- * cooperative in it's previous logical run ie (b), then that can also affect
- * cooperative pool's next best QoS requests.
+ * returns true if the the dequeued request was the highest priority item
+ * in its priority queue.
*/
static bool
-workq_threadreq_dequeue(struct workqueue *wq, workq_threadreq_t req,
- bool cooperative_sched_count_changed)
+workq_threadreq_dequeue(struct workqueue *wq, workq_threadreq_t req)
{
wq->wq_reqcount--;
-
- bool next_highest_request_changed = false;
if (--req->tr_count == 0) {
if (req->tr_qos == WORKQ_THREAD_QOS_MANAGER) {
assert(wq->wq_event_manager_threadreq == req);
assert(req->tr_count == 0);
wq->wq_event_manager_threadreq = NULL;
-
- /* If a cooperative thread was the one which picked up the manager
- * thread request, we need to reevaluate the cooperative pool
- * anyways.
- */
- if (cooperative_sched_count_changed) {
- _wq_cooperative_queue_refresh_best_req_qos(wq);
+ return true;
+ }
+ if (priority_queue_remove(workq_priority_queue_for_req(wq, req),
+ &req->tr_entry, PRIORITY_QUEUE_SCHED_PRI_MAX_HEAP_COMPARE)) {
+ if ((req->tr_flags & TR_FLAG_OVERCOMMIT) == 0) {
+ _wq_thactive_refresh_best_constrained_req_qos(wq);
}
return true;
}
-
- if (workq_threadreq_is_cooperative(req)) {
- assert(req->tr_qos != WORKQ_THREAD_QOS_MANAGER);
- assert(req->tr_qos != WORKQ_THREAD_QOS_ABOVEUI);
- /* Account for the fact that BG and MT are coalesced when
- * calculating best request for cooperative pool
- */
- assert(_wq_bucket(req->tr_qos) == _wq_bucket(wq->wq_cooperative_queue_best_req_qos));
-
- struct workq_threadreq_tailq *bucket = &wq->wq_cooperative_queue[_wq_bucket(req->tr_qos)];
- __assert_only workq_threadreq_t head = STAILQ_FIRST(bucket);
-
- assert(head == req);
- STAILQ_REMOVE_HEAD(bucket, tr_link);
-
- /*
- * If the request we're dequeueing is cooperative, then the sched
- * counts definitely changed.
- */
- assert(cooperative_sched_count_changed);
- }
-
- /*
- * We want to do the cooperative pool refresh after dequeueing a
- * cooperative thread request if any (to combine both effects into 1
- * refresh operation)
- */
- if (cooperative_sched_count_changed) {
- next_highest_request_changed = _wq_cooperative_queue_refresh_best_req_qos(wq);
- }
-
- if (!workq_threadreq_is_cooperative(req)) {
- /*
- * All other types of requests are enqueued in priority queues
- */
-
- if (priority_queue_remove(workq_priority_queue_for_req(wq, req),
- &req->tr_entry)) {
- next_highest_request_changed |= true;
- if (workq_threadreq_is_nonovercommit(req)) {
- _wq_thactive_refresh_best_constrained_req_qos(wq);
- }
- }
- }
- }
-
- return next_highest_request_changed;
+ }
+ return false;
}
static void
workq_threadreq_destroy(proc_t p, workq_threadreq_t req)
{
- req->tr_state = WORKQ_TR_STATE_CANCELED;
- if (req->tr_flags & (WORKQ_TR_FLAG_WORKLOOP | WORKQ_TR_FLAG_KEVENT)) {
+ req->tr_state = TR_STATE_IDLE;
+ if (req->tr_flags & (TR_FLAG_WORKLOOP | TR_FLAG_KEVENT)) {
kqueue_threadreq_cancel(p, req);
} else {
zfree(workq_zone_threadreq, req);
}
}
+/*
+ * Mark a thread request as complete. At this point, it is treated as owned by
+ * the submitting subsystem and you should assume it could be freed.
+ *
+ * Called with the workqueue lock held.
+ */
+static void
+workq_threadreq_bind_and_unlock(proc_t p, struct workqueue *wq,
+ workq_threadreq_t req, struct uthread *uth)
+{
+ uint8_t tr_flags = req->tr_flags;
+ bool needs_commit = false;
+ int creator_flags = 0;
+
+ wq->wq_fulfilled++;
+
+ if (req->tr_state == TR_STATE_QUEUED) {
+ workq_threadreq_dequeue(wq, req);
+ creator_flags = WORKQ_THREADREQ_CAN_CREATE_THREADS;
+ }
+
+ if (wq->wq_creator == uth) {
+ WQ_TRACE_WQ(TRACE_wq_creator_select, wq, 4, 0,
+ uth->uu_save.uus_workq_park_data.yields, 0);
+ creator_flags = WORKQ_THREADREQ_CAN_CREATE_THREADS |
+ WORKQ_THREADREQ_CREATOR_TRANSFER;
+ wq->wq_creator = NULL;
+ _wq_thactive_inc(wq, req->tr_qos);
+ wq->wq_thscheduled_count[_wq_bucket(req->tr_qos)]++;
+ } else if (uth->uu_workq_pri.qos_bucket != req->tr_qos) {
+ _wq_thactive_move(wq, uth->uu_workq_pri.qos_bucket, req->tr_qos);
+ }
+ workq_thread_reset_pri(wq, uth, req);
+
+ if (tr_flags & TR_FLAG_OVERCOMMIT) {
+ if ((uth->uu_workq_flags & UT_WORKQ_OVERCOMMIT) == 0) {
+ uth->uu_workq_flags |= UT_WORKQ_OVERCOMMIT;
+ wq->wq_constrained_threads_scheduled--;
+ }
+ } else {
+ if ((uth->uu_workq_flags & UT_WORKQ_OVERCOMMIT) != 0) {
+ uth->uu_workq_flags &= ~UT_WORKQ_OVERCOMMIT;
+ wq->wq_constrained_threads_scheduled++;
+ }
+ }
+
+ if (tr_flags & (TR_FLAG_KEVENT | TR_FLAG_WORKLOOP)) {
+ if (req->tr_state == TR_STATE_NEW) {
+ /*
+ * We're called from workq_kern_threadreq_initiate()
+ * due to an unbind, with the kq req held.
+ */
+ assert(!creator_flags);
+ req->tr_state = TR_STATE_IDLE;
+ kqueue_threadreq_bind(p, req, uth->uu_thread, 0);
+ } else {
+ assert(req->tr_count == 0);
+ workq_perform_turnstile_operation_locked(wq, ^{
+ kqueue_threadreq_bind_prepost(p, req, uth->uu_thread);
+ });
+ needs_commit = true;
+ }
+ req = NULL;
+ } else if (req->tr_count > 0) {
+ req = NULL;
+ }
+
+ if (creator_flags) {
+ /* This can drop the workqueue lock, and take it again */
+ workq_schedule_creator(p, wq, creator_flags);
+ }
+
+ workq_unlock(wq);
+
+ if (req) {
+ zfree(workq_zone_threadreq, req);
+ }
+ if (needs_commit) {
+ kqueue_threadreq_bind_commit(p, uth->uu_thread);
+ }
+
+ /*
+ * Run Thread, Run!
+ */
+ uint32_t upcall_flags = WQ_FLAG_THREAD_NEWSPI;
+ if (uth->uu_workq_pri.qos_bucket == WORKQ_THREAD_QOS_MANAGER) {
+ upcall_flags |= WQ_FLAG_THREAD_EVENT_MANAGER;
+ } else if (tr_flags & TR_FLAG_OVERCOMMIT) {
+ upcall_flags |= WQ_FLAG_THREAD_OVERCOMMIT;
+ }
+ if (tr_flags & TR_FLAG_KEVENT) {
+ upcall_flags |= WQ_FLAG_THREAD_KEVENT;
+ }
+ if (tr_flags & TR_FLAG_WORKLOOP) {
+ upcall_flags |= WQ_FLAG_THREAD_WORKLOOP | WQ_FLAG_THREAD_KEVENT;
+ }
+ uth->uu_save.uus_workq_park_data.upcall_flags = upcall_flags;
+}
+
#pragma mark workqueue thread creation thread calls
static inline bool
workq_thread_call_prepost(struct workqueue *wq, uint32_t sched, uint32_t pend,
- uint32_t fail_mask)
+ uint32_t fail_mask)
{
uint32_t old_flags, new_flags;
os_atomic_rmw_loop(&wq->wq_flags, old_flags, new_flags, acquire, {
if (__improbable(old_flags & (WQ_EXITING | sched | pend | fail_mask))) {
- os_atomic_rmw_loop_give_up(return false);
+ os_atomic_rmw_loop_give_up(return false);
}
if (__improbable(old_flags & WQ_PROC_SUSPENDED)) {
- new_flags = old_flags | pend;
+ new_flags = old_flags | pend;
} else {
- new_flags = old_flags | sched;
+ new_flags = old_flags | sched;
}
});
@@ -1793,8 +1255,8 @@
assert(!preemption_enabled());
if (!workq_thread_call_prepost(wq, WQ_DELAYED_CALL_SCHEDULED,
- WQ_DELAYED_CALL_PENDED, WQ_IMMEDIATE_CALL_PENDED |
- WQ_IMMEDIATE_CALL_SCHEDULED)) {
+ WQ_DELAYED_CALL_PENDED, WQ_IMMEDIATE_CALL_PENDED |
+ WQ_IMMEDIATE_CALL_SCHEDULED)) {
return false;
}
@@ -1805,17 +1267,17 @@
} else if (now - wq->wq_thread_call_last_run <= wq->wq_timer_interval) {
wq->wq_timer_interval *= 2;
if (wq->wq_timer_interval > wq_max_timer_interval.abstime) {
- wq->wq_timer_interval = (uint32_t)wq_max_timer_interval.abstime;
+ wq->wq_timer_interval = wq_max_timer_interval.abstime;
}
} else if (now - wq->wq_thread_call_last_run > 2 * wq->wq_timer_interval) {
wq->wq_timer_interval /= 2;
if (wq->wq_timer_interval < wq_stalled_window.abstime) {
- wq->wq_timer_interval = (uint32_t)wq_stalled_window.abstime;
+ wq->wq_timer_interval = wq_stalled_window.abstime;
}
}
WQ_TRACE_WQ(TRACE_wq_start_add_timer, wq, wq->wq_reqcount,
- _wq_flags(wq), wq->wq_timer_interval);
+ _wq_flags(wq), wq->wq_timer_interval, 0);
thread_call_t call = wq->wq_delayed_call;
uintptr_t arg = WQ_DELAYED_CALL_SCHEDULED;
@@ -1832,9 +1294,9 @@
assert(!preemption_enabled());
if (workq_thread_call_prepost(wq, WQ_IMMEDIATE_CALL_SCHEDULED,
- WQ_IMMEDIATE_CALL_PENDED, 0)) {
+ WQ_IMMEDIATE_CALL_PENDED, 0)) {
WQ_TRACE_WQ(TRACE_wq_start_add_timer, wq, wq->wq_reqcount,
- _wq_flags(wq), 0);
+ _wq_flags(wq), 0, 0);
uintptr_t arg = WQ_IMMEDIATE_CALL_SCHEDULED;
if (thread_call_enter1(wq->wq_immediate_call, (void *)arg)) {
@@ -1848,9 +1310,7 @@
{
struct workqueue *wq = proc_get_wqptr(p);
- if (wq) {
- os_atomic_or(&wq->wq_flags, WQ_PROC_SUSPENDED, relaxed);
- }
+ if (wq) os_atomic_or(&wq->wq_flags, WQ_PROC_SUSPENDED, relaxed);
}
void
@@ -1859,19 +1319,17 @@
struct workqueue *wq = proc_get_wqptr(p);
uint32_t wq_flags;
- if (!wq) {
- return;
- }
-
- wq_flags = os_atomic_andnot_orig(&wq->wq_flags, WQ_PROC_SUSPENDED |
- WQ_DELAYED_CALL_PENDED | WQ_IMMEDIATE_CALL_PENDED, relaxed);
+ if (!wq) return;
+
+ wq_flags = os_atomic_and_orig(&wq->wq_flags, ~(WQ_PROC_SUSPENDED |
+ WQ_DELAYED_CALL_PENDED | WQ_IMMEDIATE_CALL_PENDED), relaxed);
if ((wq_flags & WQ_EXITING) == 0) {
disable_preemption();
if (wq_flags & WQ_IMMEDIATE_CALL_PENDED) {
workq_schedule_immediate_thread_creation(wq);
} else if (wq_flags & WQ_DELAYED_CALL_PENDED) {
workq_schedule_delayed_thread_creation(wq,
- WORKQ_SCHEDULE_DELAYED_THREAD_CREATION_RESTART);
+ WORKQ_SCHEDULE_DELAYED_THREAD_CREATION_RESTART);
}
enable_preemption();
}
@@ -1883,7 +1341,7 @@
static bool
workq_thread_is_busy(uint64_t now, _Atomic uint64_t *lastblocked_tsp)
{
- uint64_t lastblocked_ts = os_atomic_load_wide(lastblocked_tsp, relaxed);
+ uint64_t lastblocked_ts = os_atomic_load(lastblocked_tsp, relaxed);
if (now <= lastblocked_ts) {
/*
* Because the update of the timestamp when a thread blocks
@@ -1910,20 +1368,18 @@
* workq_exit() will set the workqueue to NULL before
* it cancels thread calls.
*/
- if (!wq) {
- return;
- }
+ if (!wq) return;
assert((my_flag == WQ_DELAYED_CALL_SCHEDULED) ||
- (my_flag == WQ_IMMEDIATE_CALL_SCHEDULED));
+ (my_flag == WQ_IMMEDIATE_CALL_SCHEDULED));
WQ_TRACE_WQ(TRACE_wq_add_timer | DBG_FUNC_START, wq, _wq_flags(wq),
- wq->wq_nthreads, wq->wq_thidlecount);
+ wq->wq_nthreads, wq->wq_thidlecount, 0);
workq_lock_spin(wq);
wq->wq_thread_call_last_run = mach_absolute_time();
- os_atomic_andnot(&wq->wq_flags, my_flag, release);
+ os_atomic_and(&wq->wq_flags, ~my_flag, release);
/* This can drop the workqueue lock, and take it again */
workq_schedule_creator(p, wq, WORKQ_THREADREQ_CAN_CREATE_THREADS);
@@ -1931,7 +1387,7 @@
workq_unlock(wq);
WQ_TRACE_WQ(TRACE_wq_add_timer | DBG_FUNC_END, wq, 0,
- wq->wq_nthreads, wq->wq_thidlecount);
+ wq->wq_nthreads, wq->wq_thidlecount, 0);
}
#pragma mark thread state tracking
@@ -1939,9 +1395,9 @@
static void
workq_sched_callback(int type, thread_t thread)
{
- thread_ro_t tro = get_thread_ro(thread);
struct uthread *uth = get_bsdthread_info(thread);
- struct workqueue *wq = proc_get_wqptr(tro->tro_proc);
+ proc_t proc = get_bsdtask_info(get_threadtask(thread));
+ struct workqueue *wq = proc_get_wqptr(proc);
thread_qos_t req_qos, qos = uth->uu_workq_pri.qos_bucket;
wq_thactive_t old_thactive;
bool start_timer = false;
@@ -1965,8 +1421,8 @@
* get scheduled and then block after we start down this path), it's
* not a problem. Either timestamp is adequate, so no need to retry
*/
- os_atomic_store_wide(&wq->wq_lastblocked_ts[_wq_bucket(qos)],
- thread_last_run_time(thread), relaxed);
+ os_atomic_store(&wq->wq_lastblocked_ts[_wq_bucket(qos)],
+ thread_last_run_time(thread), relaxed);
if (req_qos == THREAD_QOS_UNSPECIFIED) {
/*
@@ -1980,7 +1436,7 @@
} else {
uint32_t max_busycount, old_req_count;
old_req_count = _wq_thactive_aggregate_downto_qos(wq, old_thactive,
- req_qos, NULL, &max_busycount);
+ req_qos, NULL, &max_busycount);
/*
* If it is possible that may_start_constrained_thread had refused
* admission due to being over the max concurrency, we may need to
@@ -2000,10 +1456,10 @@
}
if (__improbable(kdebug_enable)) {
__unused uint32_t old = _wq_thactive_aggregate_downto_qos(wq,
- old_thactive, qos, NULL, NULL);
+ old_thactive, qos, NULL, NULL);
WQ_TRACE_WQ(TRACE_wq_thread_block | DBG_FUNC_START, wq,
- old - 1, qos | (req_qos << 8),
- wq->wq_reqcount << 1 | start_timer);
+ old - 1, qos | (req_qos << 8),
+ wq->wq_reqcount << 1 | start_timer, 0);
}
break;
@@ -2019,11 +1475,11 @@
old_thactive = _wq_thactive_inc(wq, qos);
if (__improbable(kdebug_enable)) {
__unused uint32_t old = _wq_thactive_aggregate_downto_qos(wq,
- old_thactive, qos, NULL, NULL);
+ old_thactive, qos, NULL, NULL);
req_qos = WQ_THACTIVE_BEST_CONSTRAINED_REQ_QOS(old_thactive);
WQ_TRACE_WQ(TRACE_wq_thread_block | DBG_FUNC_END, wq,
- old + 1, qos | (req_qos << 8),
- wq->wq_threads_scheduled);
+ old + 1, qos | (req_qos << 8),
+ wq->wq_threads_scheduled, 0);
}
break;
}
@@ -2037,22 +1493,17 @@
os_ref_retain(&wq->wq_refcnt);
}
-static void
-workq_deallocate_queue_invoke(mpsc_queue_chain_t e,
- __assert_only mpsc_daemon_queue_t dq)
-{
- struct workqueue *wq;
+void
+workq_destroy(struct workqueue *wq)
+{
struct turnstile *ts;
- wq = mpsc_queue_element(e, struct workqueue, wq_destroy_link);
- assert(dq == &workq_deallocate_queue);
-
- turnstile_complete((uintptr_t)wq, &wq->wq_turnstile, &ts, TURNSTILE_WORKQS);
+ turnstile_complete((uintptr_t)wq, &wq->wq_turnstile, &ts);
assert(ts);
turnstile_cleanup();
turnstile_deallocate(ts);
- lck_ticket_destroy(&wq->wq_lock, &workq_lck_grp);
+ lck_spin_destroy(&wq->wq_lock, workq_lck_grp);
zfree(workq_zone_workqueue, wq);
}
@@ -2060,8 +1511,7 @@
workq_deallocate(struct workqueue *wq)
{
if (os_ref_release_relaxed(&wq->wq_refcnt) == 0) {
- workq_deallocate_queue_invoke(&wq->wq_destroy_link,
- &workq_deallocate_queue);
+ workq_destroy(wq);
}
}
@@ -2069,8 +1519,7 @@
workq_deallocate_safe(struct workqueue *wq)
{
if (__improbable(os_ref_release_relaxed(&wq->wq_refcnt) == 0)) {
- mpsc_daemon_enqueue(&workq_deallocate_queue, &wq->wq_destroy_link,
- MPSC_QUEUE_DISABLE_PREEMPTION);
+ workq_deallocate_enqueue(wq);
}
}
@@ -2079,7 +1528,7 @@
*/
int
workq_open(struct proc *p, __unused struct workq_open_args *uap,
- __unused int32_t *retval)
+ __unused int32_t *retval)
{
struct workqueue *wq;
int error = 0;
@@ -2089,7 +1538,7 @@
}
if (wq_init_constrained_limit) {
- uint32_t limit, num_cpus = ml_wait_max_cpus();
+ uint32_t limit, num_cpus = ml_get_max_cpus();
/*
* set up the limit for the constrained pool
@@ -2098,9 +1547,8 @@
*/
limit = num_cpus * WORKQUEUE_CONSTRAINED_FACTOR;
- if (limit > wq_max_constrained_threads) {
+ if (limit > wq_max_constrained_threads)
wq_max_constrained_threads = limit;
- }
if (wq_max_threads > WQ_THACTIVE_BUCKET_HALF) {
wq_max_threads = WQ_THACTIVE_BUCKET_HALF;
@@ -2113,10 +1561,8 @@
for (thread_qos_t qos = WORKQ_THREAD_QOS_MIN; qos <= WORKQ_THREAD_QOS_MAX; qos++) {
wq_max_parallelism[_wq_bucket(qos)] =
- qos_max_parallelism(qos, QOS_PARALLELISM_COUNT_LOGICAL);
- }
-
- wq_max_cooperative_threads = num_cpus;
+ qos_max_parallelism(qos, QOS_PARALLELISM_COUNT_LOGICAL);
+ }
wq_init_constrained_limit = 0;
}
@@ -2127,7 +1573,8 @@
goto out;
}
- wq = zalloc_flags(workq_zone_workqueue, Z_WAITOK | Z_ZERO);
+ wq = (struct workqueue *)zalloc(workq_zone_workqueue);
+ bzero(wq, sizeof(struct workqueue));
os_ref_init_count(&wq->wq_refcnt, &workq_refgrp, 1);
@@ -2135,39 +1582,35 @@
thread_qos_t mgr_priority_hint = task_get_default_manager_qos(current_task());
pthread_priority_t pp = _pthread_priority_make_from_thread_qos(mgr_priority_hint, 0, 0);
wq->wq_event_manager_priority = (uint32_t)pp;
- wq->wq_timer_interval = (uint32_t)wq_stalled_window.abstime;
+ wq->wq_timer_interval = wq_stalled_window.abstime;
wq->wq_proc = p;
turnstile_prepare((uintptr_t)wq, &wq->wq_turnstile, turnstile_alloc(),
- TURNSTILE_WORKQS);
+ TURNSTILE_WORKQS);
TAILQ_INIT(&wq->wq_thrunlist);
TAILQ_INIT(&wq->wq_thnewlist);
TAILQ_INIT(&wq->wq_thidlelist);
- priority_queue_init(&wq->wq_overcommit_queue);
- priority_queue_init(&wq->wq_constrained_queue);
- priority_queue_init(&wq->wq_special_queue);
- for (int bucket = 0; bucket < WORKQ_NUM_QOS_BUCKETS; bucket++) {
- STAILQ_INIT(&wq->wq_cooperative_queue[bucket]);
- }
-
- /* We are only using the delayed thread call for the constrained pool
- * which can't have work at >= UI QoS and so we can be fine with a
- * UI QoS thread call.
- */
- wq->wq_delayed_call = thread_call_allocate_with_qos(
- workq_add_new_threads_call, p, THREAD_QOS_USER_INTERACTIVE,
- THREAD_CALL_OPTIONS_ONCE);
+ priority_queue_init(&wq->wq_overcommit_queue,
+ PRIORITY_QUEUE_BUILTIN_MAX_HEAP);
+ priority_queue_init(&wq->wq_constrained_queue,
+ PRIORITY_QUEUE_BUILTIN_MAX_HEAP);
+ priority_queue_init(&wq->wq_special_queue,
+ PRIORITY_QUEUE_BUILTIN_MAX_HEAP);
+
+ wq->wq_delayed_call = thread_call_allocate_with_options(
+ workq_add_new_threads_call, p, THREAD_CALL_PRIORITY_KERNEL,
+ THREAD_CALL_OPTIONS_ONCE);
wq->wq_immediate_call = thread_call_allocate_with_options(
- workq_add_new_threads_call, p, THREAD_CALL_PRIORITY_KERNEL,
- THREAD_CALL_OPTIONS_ONCE);
+ workq_add_new_threads_call, p, THREAD_CALL_PRIORITY_KERNEL,
+ THREAD_CALL_OPTIONS_ONCE);
wq->wq_death_call = thread_call_allocate_with_options(
- workq_kill_old_threads_call, wq,
- THREAD_CALL_PRIORITY_USER, THREAD_CALL_OPTIONS_ONCE);
-
- lck_ticket_init(&wq->wq_lock, &workq_lck_grp);
+ workq_kill_old_threads_call, wq,
+ THREAD_CALL_PRIORITY_USER, THREAD_CALL_OPTIONS_ONCE);
+
+ lck_spin_init(&wq->wq_lock, workq_lck_grp, workq_lck_attr);
WQ_TRACE_WQ(TRACE_wq_create | DBG_FUNC_NONE, wq,
- VM_KERNEL_ADDRHIDE(wq), 0, 0);
+ VM_KERNEL_ADDRHIDE(wq), 0, 0, 0);
proc_set_wqptr(p, wq);
}
out:
@@ -2190,11 +1633,9 @@
uint32_t wq_flags;
workq_threadreq_t mgr_req;
- if (!wq) {
- return;
- }
-
- WQ_TRACE_WQ(TRACE_wq_pthread_exit | DBG_FUNC_START, wq, 0, 0, 0);
+ if (!wq) return;
+
+ WQ_TRACE_WQ(TRACE_wq_pthread_exit|DBG_FUNC_START, wq, 0, 0, 0, 0);
workq_lock_spin(wq);
@@ -2220,8 +1661,7 @@
mgr_req = wq->wq_event_manager_threadreq;
wq->wq_event_manager_threadreq = NULL;
wq->wq_reqcount = 0; /* workq_schedule_creator must not look at queues */
- wq->wq_creator = NULL;
- workq_turnstile_update_inheritor(wq, TURNSTILE_INHERITOR_NULL, 0);
+ workq_turnstile_update_inheritor(wq, NULL, 0);
workq_unlock(wq);
@@ -2233,19 +1673,19 @@
* It is hence safe to do the tear down without holding any lock.
*/
priority_queue_destroy(&wq->wq_overcommit_queue,
- struct workq_threadreq_s, tr_entry, ^(workq_threadreq_t e){
+ struct workq_threadreq_s, tr_entry, ^(void *e){
workq_threadreq_destroy(p, e);
});
priority_queue_destroy(&wq->wq_constrained_queue,
- struct workq_threadreq_s, tr_entry, ^(workq_threadreq_t e){
+ struct workq_threadreq_s, tr_entry, ^(void *e){
workq_threadreq_destroy(p, e);
});
priority_queue_destroy(&wq->wq_special_queue,
- struct workq_threadreq_s, tr_entry, ^(workq_threadreq_t e){
+ struct workq_threadreq_s, tr_entry, ^(void *e){
workq_threadreq_destroy(p, e);
});
- WQ_TRACE(TRACE_wq_pthread_exit | DBG_FUNC_END, 0, 0, 0, 0);
+ WQ_TRACE(TRACE_wq_pthread_exit|DBG_FUNC_END, 0, 0, 0, 0, 0);
}
/*
@@ -2267,7 +1707,7 @@
if (wq != NULL) {
thread_t th = current_thread();
- WQ_TRACE_WQ(TRACE_wq_workqueue_exit | DBG_FUNC_START, wq, 0, 0, 0);
+ WQ_TRACE_WQ(TRACE_wq_workqueue_exit|DBG_FUNC_START, wq, 0, 0, 0, 0);
if (thread_get_tag(th) & THREAD_TAG_WORKQUEUE) {
/*
@@ -2300,41 +1740,32 @@
* idle/new threads should have been interrupted and died on their own
*/
TAILQ_FOREACH_SAFE(uth, &wq->wq_thrunlist, uu_workq_entry, tmp) {
- thread_t mth = get_machthread(uth);
- thread_sched_call(mth, NULL);
- thread_deallocate(mth);
+ thread_sched_call(uth->uu_thread, NULL);
+ thread_deallocate(uth->uu_thread);
}
assert(TAILQ_EMPTY(&wq->wq_thnewlist));
assert(TAILQ_EMPTY(&wq->wq_thidlelist));
WQ_TRACE_WQ(TRACE_wq_destroy | DBG_FUNC_END, wq,
- VM_KERNEL_ADDRHIDE(wq), 0, 0);
+ VM_KERNEL_ADDRHIDE(wq), 0, 0, 0);
workq_deallocate(wq);
- WQ_TRACE(TRACE_wq_workqueue_exit | DBG_FUNC_END, 0, 0, 0, 0);
+ WQ_TRACE(TRACE_wq_workqueue_exit|DBG_FUNC_END, 0, 0, 0, 0, 0);
}
}
#pragma mark bsd thread control
-
-bool
-bsdthread_part_of_cooperative_workqueue(struct uthread *uth)
-{
- return (workq_thread_is_cooperative(uth) || workq_thread_is_nonovercommit(uth)) &&
- (uth->uu_workq_pri.qos_bucket != WORKQ_THREAD_QOS_MANAGER) &&
- (!workq_thread_is_permanently_bound(uth));
-}
static bool
_pthread_priority_to_policy(pthread_priority_t priority,
- thread_qos_policy_data_t *data)
+ thread_qos_policy_data_t *data)
{
data->qos_tier = _pthread_priority_thread_qos(priority);
data->tier_importance = _pthread_priority_relpri(priority);
if (data->qos_tier == THREAD_QOS_UNSPECIFIED || data->tier_importance > 0 ||
- data->tier_importance < THREAD_QOS_MIN_TIER_IMPORTANCE) {
+ data->tier_importance < THREAD_QOS_MIN_TIER_IMPORTANCE) {
return false;
}
return true;
@@ -2342,7 +1773,7 @@
static int
bsdthread_set_self(proc_t p, thread_t th, pthread_priority_t priority,
- mach_port_name_t voucher, enum workq_set_self_flags flags)
+ mach_port_name_t voucher, enum workq_set_self_flags flags)
{
struct uthread *uth = get_bsdthread_info(th);
struct workqueue *wq = proc_get_wqptr(p);
@@ -2351,7 +1782,6 @@
int unbind_rv = 0, qos_rv = 0, voucher_rv = 0, fixedpri_rv = 0;
bool is_wq_thread = (thread_get_tag(th) & THREAD_TAG_WORKQUEUE);
- assert(th == current_thread());
if (flags & WORKQ_SET_SELF_WQ_KEVENT_UNBIND) {
if (!is_wq_thread) {
unbind_rv = EINVAL;
@@ -2363,53 +1793,27 @@
goto qos;
}
- workq_threadreq_t kqr = uth->uu_kqr_bound;
+ struct kqrequest *kqr = uth->uu_kqr_bound;
if (kqr == NULL) {
unbind_rv = EALREADY;
goto qos;
}
- if (kqr->tr_flags & WORKQ_TR_FLAG_WORKLOOP) {
+ if (kqr->kqr_state & KQR_WORKLOOP) {
unbind_rv = EINVAL;
goto qos;
}
- kqueue_threadreq_unbind(p, kqr);
+ kqueue_threadreq_unbind(p, uth->uu_kqr_bound);
}
qos:
- if (flags & (WORKQ_SET_SELF_QOS_FLAG | WORKQ_SET_SELF_QOS_OVERRIDE_FLAG)) {
- assert(flags & WORKQ_SET_SELF_QOS_FLAG);
-
+ if (flags & WORKQ_SET_SELF_QOS_FLAG) {
thread_qos_policy_data_t new_policy;
- thread_qos_t qos_override = THREAD_QOS_UNSPECIFIED;
if (!_pthread_priority_to_policy(priority, &new_policy)) {
qos_rv = EINVAL;
goto voucher;
- }
-
- if (flags & WORKQ_SET_SELF_QOS_OVERRIDE_FLAG) {
- /*
- * If the WORKQ_SET_SELF_QOS_OVERRIDE_FLAG is set, we definitely
- * should have an override QoS in the pthread_priority_t and we should
- * only come into this path for cooperative thread requests
- */
- if (!_pthread_priority_has_override_qos(priority) ||
- !_pthread_priority_is_cooperative(priority)) {
- qos_rv = EINVAL;
- goto voucher;
- }
- qos_override = _pthread_priority_thread_override_qos(priority);
- } else {
- /*
- * If the WORKQ_SET_SELF_QOS_OVERRIDE_FLAG is not set, we definitely
- * should not have an override QoS in the pthread_priority_t
- */
- if (_pthread_priority_has_override_qos(priority)) {
- qos_rv = EINVAL;
- goto voucher;
- }
}
if (!is_wq_thread) {
@@ -2420,10 +1824,9 @@
qos_rv = EPERM;
goto voucher;
}
- } else if (uth->uu_workq_pri.qos_bucket == WORKQ_THREAD_QOS_MANAGER ||
- uth->uu_workq_pri.qos_bucket == WORKQ_THREAD_QOS_ABOVEUI) {
+ } else if (uth->uu_workq_pri.qos_bucket == WORKQ_THREAD_QOS_MANAGER) {
/*
- * Workqueue manager threads or threads above UI can't change QoS
+ * Workqueue manager threads can't change QoS
*/
qos_rv = EINVAL;
goto voucher;
@@ -2431,107 +1834,32 @@
/*
* For workqueue threads, possibly adjust buckets and redrive thread
* requests.
- *
- * Transitions allowed:
- *
- * overcommit --> non-overcommit
- * overcommit --> overcommit
- * non-overcommit --> non-overcommit
- * non-overcommit --> overcommit (to be deprecated later)
- * cooperative --> cooperative
- *
- * All other transitions aren't allowed so reject them.
*/
- if (workq_thread_is_overcommit(uth) && _pthread_priority_is_cooperative(priority)) {
- qos_rv = EINVAL;
- goto voucher;
- } else if (workq_thread_is_cooperative(uth) && !_pthread_priority_is_cooperative(priority)) {
- qos_rv = EINVAL;
- goto voucher;
- } else if (workq_thread_is_nonovercommit(uth) && _pthread_priority_is_cooperative(priority)) {
- qos_rv = EINVAL;
- goto voucher;
- }
-
+ bool old_overcommit = uth->uu_workq_flags & UT_WORKQ_OVERCOMMIT;
+ bool new_overcommit = priority & _PTHREAD_PRIORITY_OVERCOMMIT_FLAG;
struct uu_workq_policy old_pri, new_pri;
bool force_run = false;
- if (qos_override) {
- /*
- * We're in the case of a thread clarifying that it is for eg. not IN
- * req QoS but rather, UT req QoS with IN override. However, this can
- * race with a concurrent override happening to the thread via
- * workq_thread_add_dispatch_override so this needs to be
- * synchronized with the thread mutex.
- */
- thread_mtx_lock(th);
+ workq_lock_spin(wq);
+
+ if (old_overcommit != new_overcommit) {
+ uth->uu_workq_flags ^= UT_WORKQ_OVERCOMMIT;
+ if (old_overcommit) {
+ wq->wq_constrained_threads_scheduled++;
+ } else if (wq->wq_constrained_threads_scheduled-- ==
+ wq_max_constrained_threads) {
+ force_run = true;
+ }
}
- workq_lock_spin(wq);
-
old_pri = new_pri = uth->uu_workq_pri;
- new_pri.qos_req = (thread_qos_t)new_policy.qos_tier;
-
- if (old_pri.qos_override < qos_override) {
- /*
- * Since this can race with a concurrent override via
- * workq_thread_add_dispatch_override, only adjust override value if we
- * are higher - this is a saturating function.
- *
- * We should not be changing the final override values, we should simply
- * be redistributing the current value with a different breakdown of req
- * vs override QoS - assert to that effect. Therefore, buckets should
- * not change.
- */
- new_pri.qos_override = qos_override;
- assert(workq_pri_override(new_pri) == workq_pri_override(old_pri));
- assert(workq_pri_bucket(new_pri) == workq_pri_bucket(old_pri));
- }
-
- /* Adjust schedule counts for various types of transitions */
-
- /* overcommit -> non-overcommit */
- if (workq_thread_is_overcommit(uth) && _pthread_priority_is_nonovercommit(priority)) {
- workq_thread_set_type(uth, 0);
- wq->wq_constrained_threads_scheduled++;
-
- /* non-overcommit -> overcommit */
- } else if (workq_thread_is_nonovercommit(uth) && _pthread_priority_is_overcommit(priority)) {
- workq_thread_set_type(uth, UT_WORKQ_OVERCOMMIT);
- force_run = (wq->wq_constrained_threads_scheduled-- == wq_max_constrained_threads);
-
- /* cooperative -> cooperative */
- } else if (workq_thread_is_cooperative(uth)) {
- _wq_cooperative_queue_scheduled_count_dec(wq, old_pri.qos_req);
- _wq_cooperative_queue_scheduled_count_inc(wq, new_pri.qos_req);
-
- /* We're changing schedule counts within cooperative pool, we
- * need to refresh best cooperative QoS logic again */
- force_run = _wq_cooperative_queue_refresh_best_req_qos(wq);
- }
-
- /*
- * This will set up an override on the thread if any and will also call
- * schedule_creator if needed
- */
+ new_pri.qos_req = new_policy.qos_tier;
workq_thread_update_bucket(p, wq, uth, old_pri, new_pri, force_run);
workq_unlock(wq);
-
- if (qos_override) {
- thread_mtx_unlock(th);
- }
-
- if (workq_thread_is_overcommit(uth)) {
- thread_disarm_workqueue_quantum(th);
- } else {
- /* If the thread changed QoS buckets, the quantum duration
- * may have changed too */
- thread_arm_workqueue_quantum(th);
- }
}
kr = thread_policy_set_internal(th, THREAD_QOS_POLICY,
- (thread_policy_t)&new_policy, THREAD_QOS_POLICY_COUNT);
+ (thread_policy_t)&new_policy, THREAD_QOS_POLICY_COUNT);
if (kr != KERN_SUCCESS) {
qos_rv = EINVAL;
}
@@ -2547,9 +1875,7 @@
}
fixedpri:
- if (qos_rv) {
- goto done;
- }
+ if (qos_rv) goto done;
if (flags & WORKQ_SET_SELF_FIXEDPRIORITY_FLAG) {
thread_extended_policy_data_t extpol = {.timeshare = 0};
@@ -2560,7 +1886,7 @@
}
kr = thread_policy_set_internal(th, THREAD_EXTENDED_POLICY,
- (thread_policy_t)&extpol, THREAD_EXTENDED_POLICY_COUNT);
+ (thread_policy_t)&extpol, THREAD_EXTENDED_POLICY_COUNT);
if (kr != KERN_SUCCESS) {
fixedpri_rv = EINVAL;
goto done;
@@ -2575,7 +1901,7 @@
}
kr = thread_policy_set_internal(th, THREAD_EXTENDED_POLICY,
- (thread_policy_t)&extpol, THREAD_EXTENDED_POLICY_COUNT);
+ (thread_policy_t)&extpol, THREAD_EXTENDED_POLICY_COUNT);
if (kr != KERN_SUCCESS) {
fixedpri_rv = EINVAL;
goto done;
@@ -2604,27 +1930,25 @@
return fixedpri_rv;
}
-
return 0;
}
static int
bsdthread_add_explicit_override(proc_t p, mach_port_name_t kport,
- pthread_priority_t pp, user_addr_t resource)
+ pthread_priority_t pp, user_addr_t resource)
{
thread_qos_t qos = _pthread_priority_thread_qos(pp);
if (qos == THREAD_QOS_UNSPECIFIED) {
return EINVAL;
}
- thread_t th = port_name_to_thread(kport,
- PORT_INTRANS_THREAD_IN_CURRENT_TASK);
+ thread_t th = port_name_to_thread(kport);
if (th == THREAD_NULL) {
return ESRCH;
}
- int rv = proc_thread_qos_add_override(proc_task(p), th, 0, qos, TRUE,
- resource, THREAD_QOS_OVERRIDE_TYPE_PTHREAD_EXPLICIT_OVERRIDE);
+ int rv = proc_thread_qos_add_override(p->task, th, 0, qos, TRUE,
+ resource, THREAD_QOS_OVERRIDE_TYPE_PTHREAD_EXPLICIT_OVERRIDE);
thread_deallocate(th);
return rv;
@@ -2632,16 +1956,15 @@
static int
bsdthread_remove_explicit_override(proc_t p, mach_port_name_t kport,
- user_addr_t resource)
-{
- thread_t th = port_name_to_thread(kport,
- PORT_INTRANS_THREAD_IN_CURRENT_TASK);
+ user_addr_t resource)
+{
+ thread_t th = port_name_to_thread(kport);
if (th == THREAD_NULL) {
return ESRCH;
}
- int rv = proc_thread_qos_remove_override(proc_task(p), th, 0, resource,
- THREAD_QOS_OVERRIDE_TYPE_PTHREAD_EXPLICIT_OVERRIDE);
+ int rv = proc_thread_qos_remove_override(p->task, th, 0, resource,
+ THREAD_QOS_OVERRIDE_TYPE_PTHREAD_EXPLICIT_OVERRIDE);
thread_deallocate(th);
return rv;
@@ -2649,7 +1972,7 @@
static int
workq_thread_add_dispatch_override(proc_t p, mach_port_name_t kport,
- pthread_priority_t pp, user_addr_t ulock_addr)
+ pthread_priority_t pp, user_addr_t ulock_addr)
{
struct uu_workq_policy old_pri, new_pri;
struct workqueue *wq = proc_get_wqptr(p);
@@ -2659,8 +1982,7 @@
return EINVAL;
}
- thread_t thread = port_name_to_thread(kport,
- PORT_INTRANS_THREAD_IN_CURRENT_TASK);
+ thread_t thread = port_name_to_thread(kport);
if (thread == THREAD_NULL) {
return ESRCH;
}
@@ -2672,17 +1994,21 @@
}
WQ_TRACE_WQ(TRACE_wq_override_dispatch | DBG_FUNC_NONE,
- wq, thread_tid(thread), 1, pp);
+ wq, thread_tid(thread), 1, pp, 0);
thread_mtx_lock(thread);
if (ulock_addr) {
- uint32_t val;
+ uint64_t val;
int rc;
- vm_fault_disable();
- rc = copyin_atomic32(ulock_addr, &val);
- vm_fault_enable();
- if (rc == 0 && ulock_owner_value_to_port_name(val) != kport) {
+ /*
+ * Workaround lack of explicit support for 'no-fault copyin'
+ * <rdar://problem/24999882>, as disabling preemption prevents paging in
+ */
+ disable_preemption();
+ rc = copyin_word(ulock_addr, &val, sizeof(kport));
+ enable_preemption();
+ if (rc == 0 && ulock_owner_value_to_port_name((uint32_t)val) != kport) {
goto out;
}
}
@@ -2722,77 +2048,28 @@
return EPERM;
}
- WQ_TRACE_WQ(TRACE_wq_override_reset | DBG_FUNC_NONE, wq, 0, 0, 0);
-
- /*
- * workq_thread_add_dispatch_override takes the thread mutex before doing the
- * copyin to validate the drainer and apply the override. We need to do the
- * same here. See rdar://84472518
- */
- thread_mtx_lock(thread);
+ WQ_TRACE_WQ(TRACE_wq_override_reset | DBG_FUNC_NONE, wq, 0, 0, 0, 0);
workq_lock_spin(wq);
old_pri = new_pri = uth->uu_workq_pri;
new_pri.qos_override = THREAD_QOS_UNSPECIFIED;
workq_thread_update_bucket(p, wq, uth, old_pri, new_pri, false);
workq_unlock(wq);
-
- thread_mtx_unlock(thread);
- return 0;
-}
-
-static int
-workq_thread_allow_kill(__unused proc_t p, thread_t thread, bool enable)
-{
- if (!(thread_get_tag(thread) & THREAD_TAG_WORKQUEUE)) {
- // If the thread isn't a workqueue thread, don't set the
- // kill_allowed bit; however, we still need to return 0
- // instead of an error code since this code is executed
- // on the abort path which needs to not depend on the
- // pthread_t (returning an error depends on pthread_t via
- // cerror_nocancel)
- return 0;
- }
- struct uthread *uth = get_bsdthread_info(thread);
- uth->uu_workq_pthread_kill_allowed = enable;
- return 0;
-}
-
-static int
-workq_allow_sigmask(proc_t p, sigset_t mask)
-{
- if (mask & workq_threadmask) {
- return EINVAL;
- }
-
- proc_lock(p);
- p->p_workq_allow_sigmask |= mask;
- proc_unlock(p);
-
return 0;
}
static int
bsdthread_get_max_parallelism(thread_qos_t qos, unsigned long flags,
- int *retval)
+ int *retval)
{
static_assert(QOS_PARALLELISM_COUNT_LOGICAL ==
- _PTHREAD_QOS_PARALLELISM_COUNT_LOGICAL, "logical");
+ _PTHREAD_QOS_PARALLELISM_COUNT_LOGICAL, "logical");
static_assert(QOS_PARALLELISM_REALTIME ==
- _PTHREAD_QOS_PARALLELISM_REALTIME, "realtime");
- static_assert(QOS_PARALLELISM_CLUSTER_SHARED_RESOURCE ==
- _PTHREAD_QOS_PARALLELISM_CLUSTER_SHARED_RSRC, "cluster shared resource");
-
- if (flags & ~(QOS_PARALLELISM_REALTIME | QOS_PARALLELISM_COUNT_LOGICAL | QOS_PARALLELISM_CLUSTER_SHARED_RESOURCE)) {
+ _PTHREAD_QOS_PARALLELISM_REALTIME, "realtime");
+
+ if (flags & ~(QOS_PARALLELISM_REALTIME | QOS_PARALLELISM_COUNT_LOGICAL)) {
return EINVAL;
}
-
-#if !HAS_ARM_FEAT_SME
- /* No units are present */
- if (flags & QOS_PARALLELISM_CLUSTER_SHARED_RESOURCE) {
- return ENOTSUP;
- }
-#endif /* !HAS_ARM_FEAT_SME */
if (flags & QOS_PARALLELISM_REALTIME) {
if (qos) {
@@ -2806,32 +2083,8 @@
return 0;
}
-static int
-bsdthread_dispatch_apply_attr(__unused struct proc *p, thread_t thread,
- unsigned long flags, uint64_t value1, __unused uint64_t value2)
-{
- uint32_t apply_worker_index;
- kern_return_t kr;
-
- switch (flags) {
- case _PTHREAD_DISPATCH_APPLY_ATTR_CLUSTER_SHARED_RSRC_SET:
- apply_worker_index = (uint32_t)value1;
- kr = thread_shared_rsrc_policy_set(thread, apply_worker_index, CLUSTER_SHARED_RSRC_TYPE_RR, SHARED_RSRC_POLICY_AGENT_DISPATCH);
- /*
- * KERN_INVALID_POLICY indicates that the thread was trying to bind to a
- * cluster which it was not eligible to execute on.
- */
- return (kr == KERN_SUCCESS) ? 0 : ((kr == KERN_INVALID_POLICY) ? ENOTSUP : EINVAL);
- case _PTHREAD_DISPATCH_APPLY_ATTR_CLUSTER_SHARED_RSRC_CLEAR:
- kr = thread_shared_rsrc_policy_clear(thread, CLUSTER_SHARED_RSRC_TYPE_RR, SHARED_RSRC_POLICY_AGENT_DISPATCH);
- return (kr == KERN_SUCCESS) ? 0 : EINVAL;
- default:
- return EINVAL;
- }
-}
-
#define ENSURE_UNUSED(arg) \
- ({ if ((arg) != 0) { return EINVAL; } })
+ ({ if ((arg) != 0) { return EINVAL; } })
int
bsdthread_ctl(struct proc *p, struct bsdthread_ctl_args *uap, int *retval)
@@ -2839,37 +2092,28 @@
switch (uap->cmd) {
case BSDTHREAD_CTL_QOS_OVERRIDE_START:
return bsdthread_add_explicit_override(p, (mach_port_name_t)uap->arg1,
- (pthread_priority_t)uap->arg2, uap->arg3);
+ (pthread_priority_t)uap->arg2, uap->arg3);
case BSDTHREAD_CTL_QOS_OVERRIDE_END:
ENSURE_UNUSED(uap->arg3);
return bsdthread_remove_explicit_override(p, (mach_port_name_t)uap->arg1,
- (user_addr_t)uap->arg2);
+ (user_addr_t)uap->arg2);
case BSDTHREAD_CTL_QOS_OVERRIDE_DISPATCH:
return workq_thread_add_dispatch_override(p, (mach_port_name_t)uap->arg1,
- (pthread_priority_t)uap->arg2, uap->arg3);
+ (pthread_priority_t)uap->arg2, uap->arg3);
case BSDTHREAD_CTL_QOS_OVERRIDE_RESET:
return workq_thread_reset_dispatch_override(p, current_thread());
case BSDTHREAD_CTL_SET_SELF:
return bsdthread_set_self(p, current_thread(),
- (pthread_priority_t)uap->arg1, (mach_port_name_t)uap->arg2,
- (enum workq_set_self_flags)uap->arg3);
+ (pthread_priority_t)uap->arg1, (mach_port_name_t)uap->arg2,
+ (enum workq_set_self_flags)uap->arg3);
case BSDTHREAD_CTL_QOS_MAX_PARALLELISM:
ENSURE_UNUSED(uap->arg3);
return bsdthread_get_max_parallelism((thread_qos_t)uap->arg1,
- (unsigned long)uap->arg2, retval);
- case BSDTHREAD_CTL_WORKQ_ALLOW_KILL:
- ENSURE_UNUSED(uap->arg2);
- ENSURE_UNUSED(uap->arg3);
- return workq_thread_allow_kill(p, current_thread(), (bool)uap->arg1);
- case BSDTHREAD_CTL_DISPATCH_APPLY_ATTR:
- return bsdthread_dispatch_apply_attr(p, current_thread(),
- (unsigned long)uap->arg1, (uint64_t)uap->arg2,
- (uint64_t)uap->arg3);
- case BSDTHREAD_CTL_WORKQ_ALLOW_SIGMASK:
- return workq_allow_sigmask(p, (int)uap->arg1);
+ (unsigned long)uap->arg2, retval);
+
case BSDTHREAD_CTL_SET_QOS:
case BSDTHREAD_CTL_QOS_DISPATCH_ASYNCHRONOUS_OVERRIDE_ADD:
case BSDTHREAD_CTL_QOS_DISPATCH_ASYNCHRONOUS_OVERRIDE_RESET:
@@ -2884,12 +2128,8 @@
#pragma mark workqueue thread manipulation
static void __dead2
-workq_unpark_select_threadreq_or_park_and_unlock(proc_t p, struct workqueue *wq,
- struct uthread *uth, uint32_t setup_flags);
-
-static void __dead2
workq_select_threadreq_or_park_and_unlock(proc_t p, struct workqueue *wq,
- struct uthread *uth, uint32_t setup_flags);
+ struct uthread *uth);
static void workq_setup_and_run(proc_t p, struct uthread *uth, int flags) __dead2;
@@ -2898,8 +2138,8 @@
workq_trace_req_id(workq_threadreq_t req)
{
struct kqworkloop *kqwl;
- if (req->tr_flags & WORKQ_TR_FLAG_WORKLOOP) {
- kqwl = __container_of(req, struct kqworkloop, kqwl_request);
+ if (req->tr_flags & TR_FLAG_WORKLOOP) {
+ kqwl = __container_of(req, struct kqworkloop, kqwl_request.kqr_req);
return kqwl->kqwl_dynamicid;
}
@@ -2911,58 +2151,38 @@
* Entry point for libdispatch to ask for threads
*/
static int
-workq_reqthreads(struct proc *p, uint32_t reqcount, pthread_priority_t pp, bool cooperative)
+workq_reqthreads(struct proc *p, uint32_t reqcount, pthread_priority_t pp)
{
thread_qos_t qos = _pthread_priority_thread_qos(pp);
struct workqueue *wq = proc_get_wqptr(p);
uint32_t unpaced, upcall_flags = WQ_FLAG_THREAD_NEWSPI;
- int ret = 0;
if (wq == NULL || reqcount <= 0 || reqcount > UINT16_MAX ||
- qos == THREAD_QOS_UNSPECIFIED) {
- ret = EINVAL;
- goto exit;
+ qos == THREAD_QOS_UNSPECIFIED) {
+ return EINVAL;
}
WQ_TRACE_WQ(TRACE_wq_wqops_reqthreads | DBG_FUNC_NONE,
- wq, reqcount, pp, cooperative);
+ wq, reqcount, pp, 0, 0);
workq_threadreq_t req = zalloc(workq_zone_threadreq);
priority_queue_entry_init(&req->tr_entry);
- req->tr_state = WORKQ_TR_STATE_NEW;
+ req->tr_state = TR_STATE_NEW;
+ req->tr_flags = 0;
req->tr_qos = qos;
- workq_tr_flags_t tr_flags = 0;
if (pp & _PTHREAD_PRIORITY_OVERCOMMIT_FLAG) {
- tr_flags |= WORKQ_TR_FLAG_OVERCOMMIT;
+ req->tr_flags |= TR_FLAG_OVERCOMMIT;
upcall_flags |= WQ_FLAG_THREAD_OVERCOMMIT;
}
- if (cooperative) {
- tr_flags |= WORKQ_TR_FLAG_COOPERATIVE;
- upcall_flags |= WQ_FLAG_THREAD_COOPERATIVE;
-
- if (reqcount > 1) {
- ret = ENOTSUP;
- goto free_and_exit;
- }
- }
-
- /* A thread request cannot be both overcommit and cooperative */
- if (workq_tr_is_cooperative(tr_flags) &&
- workq_tr_is_overcommit(tr_flags)) {
- ret = EINVAL;
- goto free_and_exit;
- }
- req->tr_flags = tr_flags;
-
WQ_TRACE_WQ(TRACE_wq_thread_request_initiate | DBG_FUNC_NONE,
- wq, workq_trace_req_id(req), req->tr_qos, reqcount);
+ wq, workq_trace_req_id(req), req->tr_qos, reqcount, 0);
workq_lock_spin(wq);
do {
if (_wq_exiting(wq)) {
- goto unlock_and_exit;
+ goto exiting;
}
/*
@@ -2975,63 +2195,51 @@
* If there aren't enough threads, add one, but re-evaluate everything
* as conditions may now have changed.
*/
- unpaced = reqcount - 1;
-
- if (reqcount > 1) {
- /* We don't handle asking for parallelism on the cooperative
- * workqueue just yet */
- assert(!workq_threadreq_is_cooperative(req));
-
- if (workq_threadreq_is_nonovercommit(req)) {
- unpaced = workq_constrained_allowance(wq, qos, NULL, false, true);
- if (unpaced >= reqcount - 1) {
- unpaced = reqcount - 1;
- }
+ if (reqcount > 1 && (req->tr_flags & TR_FLAG_OVERCOMMIT) == 0) {
+ unpaced = workq_constrained_allowance(wq, qos, NULL, false);
+ if (unpaced >= reqcount - 1) {
+ unpaced = reqcount - 1;
}
+ } else {
+ unpaced = reqcount - 1;
}
/*
* This path does not currently handle custom workloop parameters
* when creating threads for parallelism.
*/
- assert(!(req->tr_flags & WORKQ_TR_FLAG_WL_PARAMS));
+ assert(!(req->tr_flags & TR_FLAG_WL_PARAMS));
/*
* This is a trimmed down version of workq_threadreq_bind_and_unlock()
*/
while (unpaced > 0 && wq->wq_thidlecount) {
- struct uthread *uth;
- bool needs_wakeup;
- uint8_t uu_flags = UT_WORKQ_EARLY_BOUND;
-
- if (workq_tr_is_overcommit(req->tr_flags)) {
- uu_flags |= UT_WORKQ_OVERCOMMIT;
- }
-
- uth = workq_pop_idle_thread(wq, uu_flags, &needs_wakeup);
+ struct uthread *uth = workq_pop_idle_thread(wq);
_wq_thactive_inc(wq, qos);
wq->wq_thscheduled_count[_wq_bucket(qos)]++;
- workq_thread_reset_pri(wq, uth, req, /*unpark*/ true);
+ workq_thread_reset_pri(wq, uth, req);
wq->wq_fulfilled++;
+ uth->uu_workq_flags |= UT_WORKQ_EARLY_BOUND;
+ if ((req->tr_flags & TR_FLAG_OVERCOMMIT) == 0) {
+ uth->uu_workq_flags &= ~UT_WORKQ_OVERCOMMIT;
+ wq->wq_constrained_threads_scheduled++;
+ }
uth->uu_save.uus_workq_park_data.upcall_flags = upcall_flags;
uth->uu_save.uus_workq_park_data.thread_request = req;
- if (needs_wakeup) {
- workq_thread_wakeup(uth);
- }
+ workq_thread_wakeup(uth);
unpaced--;
reqcount--;
}
} while (unpaced && wq->wq_nthreads < wq_max_threads &&
- (workq_add_new_idle_thread(p, wq, workq_unpark_continue,
- false, NULL) == KERN_SUCCESS));
+ workq_add_new_idle_thread(p, wq));
if (_wq_exiting(wq)) {
- goto unlock_and_exit;
- }
-
- req->tr_count = (uint16_t)reqcount;
+ goto exiting;
+ }
+
+ req->tr_count = reqcount;
if (workq_threadreq_enqueue(wq, req)) {
/* This can drop the workqueue lock, and take it again */
workq_schedule_creator(p, wq, WORKQ_THREADREQ_CAN_CREATE_THREADS);
@@ -3039,51 +2247,52 @@
workq_unlock(wq);
return 0;
-unlock_and_exit:
+exiting:
workq_unlock(wq);
-free_and_exit:
zfree(workq_zone_threadreq, req);
-exit:
- return ret;
+ return ECANCELED;
}
bool
-workq_kern_threadreq_initiate(struct proc *p, workq_threadreq_t req,
- struct turnstile *workloop_ts, thread_qos_t qos,
- workq_kern_threadreq_flags_t flags)
+workq_kern_threadreq_initiate(struct proc *p, struct kqrequest *kqr,
+ struct turnstile *workloop_ts, thread_qos_t qos, int flags)
{
struct workqueue *wq = proc_get_wqptr_fast(p);
+ workq_threadreq_t req = &kqr->kqr_req;
struct uthread *uth = NULL;
-
- assert(req->tr_flags & (WORKQ_TR_FLAG_WORKLOOP | WORKQ_TR_FLAG_KEVENT));
- /*
- * Thread bound kqworkloops overlay the priority queue entry with other
- * data (the thread_t and work interval), so should never have their
- * threadreq passed here.
- */
- assert(!(req->tr_flags & WORKQ_TR_FLAG_PERMANENT_BIND));
-
- /*
- * For any new initialization changes done to workqueue thread request below,
- * please also consider if they are relevant to permanently bound thread
- * request. See workq_kern_threadreq_permanent_bind.
- */
- if (req->tr_flags & WORKQ_TR_FLAG_WL_OUTSIDE_QOS) {
+ uint8_t tr_flags = 0;
+
+ if (kqr->kqr_state & KQR_WORKLOOP) {
+ tr_flags = TR_FLAG_WORKLOOP;
+
workq_threadreq_param_t trp = kqueue_threadreq_workloop_param(req);
- qos = thread_workq_qos_for_pri(trp.trp_pri);
- if (qos == THREAD_QOS_UNSPECIFIED) {
- qos = WORKQ_THREAD_QOS_ABOVEUI;
- }
- }
-
- assert(req->tr_state == WORKQ_TR_STATE_IDLE);
+ if (trp.trp_flags & TRP_PRIORITY) {
+ tr_flags |= TR_FLAG_WL_OUTSIDE_QOS;
+ qos = thread_workq_qos_for_pri(trp.trp_pri);
+ if (qos == THREAD_QOS_UNSPECIFIED) {
+ qos = WORKQ_THREAD_QOS_ABOVEUI;
+ }
+ }
+ if (trp.trp_flags) {
+ tr_flags |= TR_FLAG_WL_PARAMS;
+ }
+ } else {
+ tr_flags = TR_FLAG_KEVENT;
+ }
+ if (qos != WORKQ_THREAD_QOS_MANAGER &&
+ (kqr->kqr_state & KQR_THOVERCOMMIT)) {
+ tr_flags |= TR_FLAG_OVERCOMMIT;
+ }
+
+ assert(req->tr_state == TR_STATE_IDLE);
priority_queue_entry_init(&req->tr_entry);
req->tr_count = 1;
- req->tr_state = WORKQ_TR_STATE_NEW;
+ req->tr_state = TR_STATE_NEW;
+ req->tr_flags = tr_flags;
req->tr_qos = qos;
WQ_TRACE_WQ(TRACE_wq_thread_request_initiate | DBG_FUNC_NONE, wq,
- workq_trace_req_id(req), qos, 1);
+ workq_trace_req_id(req), qos, 1, 0);
if (flags & WORKQ_THREADREQ_ATTEMPT_REBIND) {
/*
@@ -3097,64 +2306,40 @@
workq_lock_spin(wq);
if (_wq_exiting(wq)) {
- req->tr_state = WORKQ_TR_STATE_IDLE;
workq_unlock(wq);
return false;
}
if (uth && workq_threadreq_admissible(wq, uth, req)) {
- /* This is the case of the rebind - we were about to park and unbind
- * when more events came so keep the binding.
- */
assert(uth != wq->wq_creator);
-
- if (uth->uu_workq_pri.qos_bucket != req->tr_qos) {
- _wq_thactive_move(wq, uth->uu_workq_pri.qos_bucket, req->tr_qos);
- workq_thread_reset_pri(wq, uth, req, /*unpark*/ false);
- }
- /*
- * We're called from workq_kern_threadreq_initiate()
- * due to an unbind, with the kq req held.
- */
- WQ_TRACE_WQ(TRACE_wq_thread_logical_run | DBG_FUNC_START, wq,
- workq_trace_req_id(req), req->tr_flags, 0);
- wq->wq_fulfilled++;
-
- kqueue_threadreq_bind(p, req, get_machthread(uth), 0);
+ workq_threadreq_bind_and_unlock(p, wq, req, uth);
} else {
if (workloop_ts) {
workq_perform_turnstile_operation_locked(wq, ^{
turnstile_update_inheritor(workloop_ts, wq->wq_turnstile,
- TURNSTILE_IMMEDIATE_UPDATE | TURNSTILE_INHERITOR_TURNSTILE);
+ TURNSTILE_IMMEDIATE_UPDATE | TURNSTILE_INHERITOR_TURNSTILE);
turnstile_update_inheritor_complete(workloop_ts,
- TURNSTILE_INTERLOCK_HELD);
+ TURNSTILE_INTERLOCK_HELD);
});
}
-
- bool reevaluate_creator_thread_group = false;
-#if CONFIG_PREADOPT_TG
- reevaluate_creator_thread_group = (flags & WORKQ_THREADREQ_REEVALUATE_PREADOPT_TG);
-#endif
- /* We enqueued the highest priority item or we may need to reevaluate if
- * the creator needs a thread group pre-adoption */
- if (workq_threadreq_enqueue(wq, req) || reevaluate_creator_thread_group) {
+ if (workq_threadreq_enqueue(wq, req)) {
workq_schedule_creator(p, wq, flags);
}
- }
-
- workq_unlock(wq);
+ workq_unlock(wq);
+ }
return true;
}
void
-workq_kern_threadreq_modify(struct proc *p, workq_threadreq_t req,
- thread_qos_t qos, workq_kern_threadreq_flags_t flags)
+workq_kern_threadreq_modify(struct proc *p, struct kqrequest *kqr,
+ thread_qos_t qos, int flags)
{
struct workqueue *wq = proc_get_wqptr_fast(p);
- bool make_overcommit = false;
-
- if (req->tr_flags & WORKQ_TR_FLAG_WL_OUTSIDE_QOS) {
+ workq_threadreq_t req = &kqr->kqr_req;
+ bool change_overcommit = false;
+
+ if (req->tr_flags & TR_FLAG_WL_OUTSIDE_QOS) {
/* Requests outside-of-QoS shouldn't accept modify operations */
return;
}
@@ -3162,37 +2347,31 @@
workq_lock_spin(wq);
assert(req->tr_qos != WORKQ_THREAD_QOS_MANAGER);
- assert(req->tr_flags & (WORKQ_TR_FLAG_KEVENT | WORKQ_TR_FLAG_WORKLOOP));
-
- if (req->tr_state == WORKQ_TR_STATE_BINDING) {
- kqueue_threadreq_bind(p, req, req->tr_thread, 0);
+ assert(req->tr_flags & (TR_FLAG_KEVENT | TR_FLAG_WORKLOOP));
+
+ if (req->tr_state == TR_STATE_BINDING) {
+ kqueue_threadreq_bind(p, req, req->tr_binding_thread, 0);
workq_unlock(wq);
return;
}
- if (flags & WORKQ_THREADREQ_MAKE_OVERCOMMIT) {
- /* TODO (rokhinip): We come into this code path for kqwl thread
- * requests. kqwl requests cannot be cooperative.
- */
- assert(!workq_threadreq_is_cooperative(req));
-
- make_overcommit = workq_threadreq_is_nonovercommit(req);
- }
-
- if (_wq_exiting(wq) || (req->tr_qos == qos && !make_overcommit)) {
+ change_overcommit = (bool)(kqr->kqr_state & KQR_THOVERCOMMIT) !=
+ (bool)(req->tr_flags & TR_FLAG_OVERCOMMIT);
+
+ if (_wq_exiting(wq) || (req->tr_qos == qos && !change_overcommit)) {
workq_unlock(wq);
return;
}
assert(req->tr_count == 1);
- if (req->tr_state != WORKQ_TR_STATE_QUEUED) {
+ if (req->tr_state != TR_STATE_QUEUED) {
panic("Invalid thread request (%p) state %d", req, req->tr_state);
}
WQ_TRACE_WQ(TRACE_wq_thread_request_modify | DBG_FUNC_NONE, wq,
- workq_trace_req_id(req), qos, 0);
-
- struct priority_queue_sched_max *pq = workq_priority_queue_for_req(wq, req);
+ workq_trace_req_id(req), qos, 0, 0);
+
+ struct priority_queue *pq = workq_priority_queue_for_req(wq, req);
workq_threadreq_t req_max;
/*
@@ -3201,8 +2380,9 @@
* If we dequeue the root item of the constrained priority queue,
* maintain the best constrained request qos invariant.
*/
- if (priority_queue_remove(pq, &req->tr_entry)) {
- if (workq_threadreq_is_nonovercommit(req)) {
+ if (priority_queue_remove(pq, &req->tr_entry,
+ PRIORITY_QUEUE_SCHED_PRI_MAX_HEAP_COMPARE)) {
+ if ((req->tr_flags & TR_FLAG_OVERCOMMIT) == 0) {
_wq_thactive_refresh_best_constrained_req_qos(wq);
}
}
@@ -3213,17 +2393,16 @@
* If the item will not become the root of the priority queue it belongs to,
* then we need to wait in line, just enqueue and return quickly.
*/
- if (__improbable(make_overcommit)) {
- req->tr_flags ^= WORKQ_TR_FLAG_OVERCOMMIT;
+ if (__improbable(change_overcommit)) {
+ req->tr_flags ^= TR_FLAG_OVERCOMMIT;
pq = workq_priority_queue_for_req(wq, req);
}
req->tr_qos = qos;
req_max = priority_queue_max(pq, struct workq_threadreq_s, tr_entry);
if (req_max && req_max->tr_qos >= qos) {
- priority_queue_entry_set_sched_pri(pq, &req->tr_entry,
- workq_priority_for_req(req), false);
- priority_queue_insert(pq, &req->tr_entry);
+ priority_queue_insert(pq, &req->tr_entry, workq_priority_for_req(req),
+ PRIORITY_QUEUE_SCHED_PRI_MAX_HEAP_COMPARE);
workq_unlock(wq);
return;
}
@@ -3233,52 +2412,15 @@
*
* Pretend the thread request is new again:
* - adjust wq_reqcount to not count it anymore.
- * - make its state WORKQ_TR_STATE_NEW (so that workq_threadreq_bind_and_unlock
+ * - make its state TR_STATE_NEW (so that workq_threadreq_bind_and_unlock
* properly attempts a synchronous bind)
*/
wq->wq_reqcount--;
- req->tr_state = WORKQ_TR_STATE_NEW;
-
- /* We enqueued the highest priority item or we may need to reevaluate if
- * the creator needs a thread group pre-adoption if the request got a new TG */
- bool reevaluate_creator_tg = false;
-
-#if CONFIG_PREADOPT_TG
- reevaluate_creator_tg = (flags & WORKQ_THREADREQ_REEVALUATE_PREADOPT_TG);
-#endif
-
- if (workq_threadreq_enqueue(wq, req) || reevaluate_creator_tg) {
+ req->tr_state = TR_STATE_NEW;
+ if (workq_threadreq_enqueue(wq, req)) {
workq_schedule_creator(p, wq, flags);
}
workq_unlock(wq);
-}
-
-void
-workq_kern_bound_thread_reset_pri(workq_threadreq_t req, struct uthread *uth)
-{
- assert(workq_thread_is_permanently_bound(uth));
-
- if (req && (req->tr_flags & WORKQ_TR_FLAG_WL_OUTSIDE_QOS)) {
- /*
- * For requests outside-of-QoS, we set the scheduling policy and
- * absolute priority for the bound thread right at the initialization
- * time. See workq_kern_threadreq_permanent_bind.
- */
- return;
- }
-
- struct workqueue *wq = proc_get_wqptr_fast(current_proc());
- if (req) {
- assert(req->tr_qos != WORKQ_THREAD_QOS_MANAGER);
- workq_thread_reset_pri(wq, uth, req, /*unpark*/ true);
- } else {
- thread_qos_t qos = workq_pri_override(uth->uu_workq_pri);
- if (qos > WORKQ_THREAD_QOS_CLEANUP) {
- workq_thread_reset_pri(wq, uth, NULL, /*unpark*/ true);
- } else {
- uth->uu_save.uus_workq_park_data.qos = qos;
- }
- }
}
void
@@ -3294,27 +2436,28 @@
}
void
-workq_kern_threadreq_update_inheritor(struct proc *p, workq_threadreq_t req,
- thread_t owner, struct turnstile *wl_ts,
- turnstile_update_flags_t flags)
+workq_kern_threadreq_update_inheritor(struct proc *p, struct kqrequest *kqr,
+ thread_t owner, struct turnstile *wl_ts,
+ turnstile_update_flags_t flags)
{
struct workqueue *wq = proc_get_wqptr_fast(p);
+ workq_threadreq_t req = &kqr->kqr_req;
turnstile_inheritor_t inheritor;
assert(req->tr_qos != WORKQ_THREAD_QOS_MANAGER);
- assert(req->tr_flags & WORKQ_TR_FLAG_WORKLOOP);
+ assert(req->tr_flags & TR_FLAG_WORKLOOP);
workq_lock_held(wq);
- if (req->tr_state == WORKQ_TR_STATE_BINDING) {
- kqueue_threadreq_bind(p, req, req->tr_thread,
- KQUEUE_THREADREQ_BIND_NO_INHERITOR_UPDATE);
+ if (req->tr_state == TR_STATE_BINDING) {
+ kqueue_threadreq_bind(p, req, req->tr_binding_thread,
+ KQUEUE_THREADERQ_BIND_NO_INHERITOR_UPDATE);
return;
}
if (_wq_exiting(wq)) {
inheritor = TURNSTILE_INHERITOR_NULL;
} else {
- if (req->tr_state != WORKQ_TR_STATE_QUEUED) {
+ if (req->tr_state != TR_STATE_QUEUED) {
panic("Invalid thread request (%p) state %d", req, req->tr_state);
}
@@ -3332,225 +2475,8 @@
});
}
-/*
- * An entry point for kevent to request a newly created workqueue thread
- * and bind it permanently to the given workqueue thread request.
- *
- * It currently only supports fixed scheduler priority thread requests.
- *
- * The newly created thread counts towards wq_nthreads. This function returns
- * an error if we are above that limit. There is no concept of delayed thread
- * creation for such specially configured kqworkloops.
- *
- * If successful, the newly created thread will be parked in
- * workq_bound_thread_initialize_and_unpark_continue waiting for
- * new incoming events.
- */
-kern_return_t
-workq_kern_threadreq_permanent_bind(struct proc *p, struct workq_threadreq_s *kqr)
-{
- kern_return_t ret = 0;
- thread_t new_thread = NULL;
- struct workqueue *wq = proc_get_wqptr_fast(p);
-
- workq_lock_spin(wq);
-
- if (wq->wq_nthreads >= wq_max_threads) {
- ret = EDOM;
- } else {
- if (kqr->tr_flags & WORKQ_TR_FLAG_WL_OUTSIDE_QOS) {
- workq_threadreq_param_t trp = kqueue_threadreq_workloop_param(kqr);
- /*
- * For requests outside-of-QoS, we fully initialize the thread
- * request here followed by preadopting the scheduling properties
- * on the newly created bound thread.
- */
- thread_qos_t qos = thread_workq_qos_for_pri(trp.trp_pri);
- if (qos == THREAD_QOS_UNSPECIFIED) {
- qos = WORKQ_THREAD_QOS_ABOVEUI;
- }
- kqr->tr_qos = qos;
- }
- kqr->tr_count = 1;
-
- /* workq_lock dropped and retaken around thread creation below. */
- ret = workq_add_new_idle_thread(p, wq,
- workq_bound_thread_initialize_and_unpark_continue,
- true, &new_thread);
- if (ret == KERN_SUCCESS) {
- struct uthread *uth = get_bsdthread_info(new_thread);
- if (kqr->tr_flags & WORKQ_TR_FLAG_WL_OUTSIDE_QOS) {
- workq_thread_reset_pri(wq, uth, kqr, /*unpark*/ true);
- }
- /*
- * The newly created thread goes through a full bind to the kqwl
- * right upon creation.
- * It then falls back to soft bind/unbind upon wakeup/park.
- */
- kqueue_threadreq_bind_prepost(p, kqr, uth);
- uth->uu_workq_flags |= UT_WORKQ_PERMANENT_BIND;
- }
- }
-
- workq_unlock(wq);
-
- if (ret == KERN_SUCCESS) {
- kqueue_threadreq_bind_commit(p, new_thread);
- }
- return ret;
-}
-
-/*
- * Called with kqlock held. It does not need to take the process wide
- * global workq lock -> making it faster.
- */
void
-workq_kern_bound_thread_wakeup(struct workq_threadreq_s *kqr)
-{
- struct uthread *uth = get_bsdthread_info(kqr->tr_thread);
- workq_threadreq_param_t trp = kqueue_threadreq_workloop_param(kqr);
-
- /*
- * See "Locking model for accessing uu_workq_flags" for more information
- * on how access to uu_workq_flags for the bound thread is synchronized.
- */
- assert((uth->uu_workq_flags & (UT_WORKQ_RUNNING | UT_WORKQ_DYING)) == 0);
-
- if (trp.trp_flags & TRP_RELEASED) {
- uth->uu_workq_flags |= UT_WORKQ_DYING;
- } else {
- uth->uu_workq_flags |= UT_WORKQ_RUNNING;
- }
-
- workq_thread_wakeup(uth);
-}
-
-/*
- * Called with kqlock held. Dropped before parking.
- * It does not need to take process wide global workqueue
- * lock -> making it faster.
- */
-__attribute__((noreturn, noinline))
-void
-workq_kern_bound_thread_park(struct workq_threadreq_s *kqr)
-{
- struct uthread *uth = get_bsdthread_info(kqr->tr_thread);
- assert(uth == current_uthread());
-
- /*
- * See "Locking model for accessing uu_workq_flags" for more information
- * on how access to uu_workq_flags for the bound thread is synchronized.
- */
- uth->uu_workq_flags &= ~(UT_WORKQ_RUNNING);
-
- thread_disarm_workqueue_quantum(get_machthread(uth));
-
- /*
- * TODO (pavhad) We could do the reusable userspace stack performance
- * optimization here.
- */
-
- kqworkloop_bound_thread_park_prepost(kqr);
- /* KQ_SLEEP bit is set and kqlock is dropped. */
-
- __assert_only kern_return_t kr;
- kr = thread_set_voucher_name(MACH_PORT_NULL);
- assert(kr == KERN_SUCCESS);
-
- /*
- * Bound threads park (and unpark) with the scheduler callback cleared and
- * not counting towards thactive. If we unpark to do work (as opposed to
- * waking up to thread exit), the scheduler callback is added via
- * workq_setup_and_run, and thactive is incremented in
- * workq_bound_thread_setup_and_run.
- */
- thread_sched_call(get_machthread(uth), NULL);
- proc_t p = current_proc();
- struct workqueue *wq = proc_get_wqptr_fast(p);
- _wq_thactive_dec(wq, uth->uu_workq_pri.qos_bucket);
-
- kqworkloop_bound_thread_park_commit(kqr,
- workq_parked_wait_event(uth), workq_bound_thread_unpark_continue);
-
- __builtin_unreachable();
-}
-
-/*
- * To terminate the permenantly bound workqueue thread. It unbinds itself
- * with the kqwl during uthread_cleanup -> kqueue_threadreq_unbind.
- * It is also when it will release its reference on the kqwl.
- */
-__attribute__((noreturn, noinline))
-void
-workq_kern_bound_thread_terminate(struct workq_threadreq_s *kqr)
-{
- proc_t p = current_proc();
- struct uthread *uth = get_bsdthread_info(kqr->tr_thread);
- uint16_t uu_workq_flags_orig;
-
- assert(uth == current_uthread());
-
- /*
- * See "Locking model for accessing uu_workq_flags" for more information
- * on how access to uu_workq_flags for the bound thread is synchronized.
- */
- kqworkloop_bound_thread_terminate(kqr, &uu_workq_flags_orig);
-
- if (uu_workq_flags_orig & UT_WORKQ_WORK_INTERVAL_JOINED) {
- __assert_only kern_return_t kr;
- kr = kern_work_interval_join(get_machthread(uth), MACH_PORT_NULL);
- /* The bound thread un-joins the work interval and drops its +1 ref. */
- assert(kr == KERN_SUCCESS);
- }
-
- /*
- * Drop the voucher now that we are on our way to termination.
- */
- __assert_only kern_return_t kr;
- kr = thread_set_voucher_name(MACH_PORT_NULL);
- assert(kr == KERN_SUCCESS);
-
- uint32_t upcall_flags = WQ_FLAG_THREAD_NEWSPI;
- upcall_flags |= uth->uu_save.uus_workq_park_data.qos |
- WQ_FLAG_THREAD_PRIO_QOS;
-
- thread_t th = get_machthread(uth);
- vm_map_t vmap = get_task_map(proc_task(p));
-
- uint32_t setup_flags = WQ_SETUP_EXIT_THREAD;
-
- if ((uu_workq_flags_orig & UT_WORKQ_NEW) == 0) {
- upcall_flags |= WQ_FLAG_THREAD_REUSE;
- } else {
- /*
- * The bound thread is exiting before ever having gone to userspace. We
- * need pthread to perform initial setup before calling _pthread_exit.
- * TODO(aaron): We should investigate an optimization both here and in
- * workq_unpark_for_death_and_unlock to never send these threads to
- * userspace, instead processing ASTs and calling destroying the stack.
- */
- setup_flags |= WQ_SETUP_FIRST_USE;
-
- /*
- * Typically we'd set the workq_thport in workq_setup_and_run, but we
- * haven't been through that path yet.
- */
- assert(uth->uu_workq_thport == MACH_PORT_NULL);
- /* convert_thread_to_port_immovable() consumes a reference */
- thread_reference(th);
- /* Convert to immovable thread port, then pin the entry */
- uth->uu_workq_thport = ipc_port_copyout_send_pinned(
- convert_thread_to_port_immovable(th),
- get_task_ipcspace(proc_task(p)));
- }
-
- pthread_functions->workq_setup_thread(p, th, vmap, uth->uu_workq_stackaddr,
- uth->uu_workq_thport, 0, setup_flags, upcall_flags);
- __builtin_unreachable();
-}
-
-void
-workq_kern_threadreq_redrive(struct proc *p, workq_kern_threadreq_flags_t flags)
+workq_kern_threadreq_redrive(struct proc *p, int flags)
{
struct workqueue *wq = proc_get_wqptr_fast(p);
@@ -3559,97 +2485,27 @@
workq_unlock(wq);
}
-/*
- * Always called at AST by the thread on itself
- *
- * Upon quantum expiry, the workqueue subsystem evaluates its state and decides
- * on what the thread should do next. The TSD value is always set by the thread
- * on itself in the kernel and cleared either by userspace when it acks the TSD
- * value and takes action, or by the thread in the kernel when the quantum
- * expires again.
- */
-void
-workq_kern_quantum_expiry_reevaluate(proc_t proc, thread_t thread)
-{
- struct uthread *uth = get_bsdthread_info(thread);
-
- if (uth->uu_workq_flags & UT_WORKQ_DYING) {
- return;
- }
-
- if (!thread_supports_cooperative_workqueue(thread)) {
- panic("Quantum expired for thread that doesn't support cooperative workqueue");
- }
-
- thread_qos_t qos = uth->uu_workq_pri.qos_bucket;
- if (qos == THREAD_QOS_UNSPECIFIED) {
- panic("Thread should not have workq bucket of QoS UN");
- }
-
- assert(thread_has_expired_workqueue_quantum(thread, false));
-
- struct workqueue *wq = proc_get_wqptr(proc);
- assert(wq != NULL);
-
- /*
- * For starters, we're just going to evaluate and see if we need to narrow
- * the pool and tell this thread to park if needed. In the future, we'll
- * evaluate and convey other workqueue state information like needing to
- * pump kevents, etc.
- */
- uint64_t flags = 0;
-
- workq_lock_spin(wq);
-
- if (workq_thread_is_cooperative(uth)) {
- if (!workq_cooperative_allowance(wq, qos, uth, false)) {
- flags |= PTHREAD_WQ_QUANTUM_EXPIRY_NARROW;
- } else {
- /* In the future, when we have kevent hookups for the cooperative
- * pool, we need fancier logic for what userspace should do. But
- * right now, only userspace thread requests exist - so we'll just
- * tell userspace to shuffle work items */
- flags |= PTHREAD_WQ_QUANTUM_EXPIRY_SHUFFLE;
- }
- } else if (workq_thread_is_nonovercommit(uth)) {
- if (!workq_constrained_allowance(wq, qos, uth, false, false)) {
- flags |= PTHREAD_WQ_QUANTUM_EXPIRY_NARROW;
- }
- }
- workq_unlock(wq);
-
- WQ_TRACE(TRACE_wq_quantum_expiry_reevaluate, flags, 0, 0, 0);
-
- kevent_set_workq_quantum_expiry_user_tsd(proc, thread, flags);
-
- /* We have conveyed to userspace about what it needs to do upon quantum
- * expiry, now rearm the workqueue quantum again */
- thread_arm_workqueue_quantum(get_machthread(uth));
-}
-
void
workq_schedule_creator_turnstile_redrive(struct workqueue *wq, bool locked)
{
- if (locked) {
- workq_schedule_creator(NULL, wq, WORKQ_THREADREQ_NONE);
- } else {
- workq_schedule_immediate_thread_creation(wq);
- }
+ if (!locked) workq_lock_spin(wq);
+ workq_schedule_creator(NULL, wq, WORKQ_THREADREQ_CREATOR_SYNC_UPDATE);
+ if (!locked) workq_unlock(wq);
}
static int
workq_thread_return(struct proc *p, struct workq_kernreturn_args *uap,
- struct workqueue *wq)
+ struct workqueue *wq)
{
thread_t th = current_thread();
struct uthread *uth = get_bsdthread_info(th);
- workq_threadreq_t kqr = uth->uu_kqr_bound;
+ struct kqrequest *kqr = uth->uu_kqr_bound;
workq_threadreq_param_t trp = { };
int nevents = uap->affinity, error;
user_addr_t eventlist = uap->item;
if (((thread_get_tag(th) & THREAD_TAG_WORKQUEUE) == 0) ||
- (uth->uu_workq_flags & UT_WORKQ_DYING)) {
+ (uth->uu_workq_flags & UT_WORKQ_DYING)) {
return EINVAL;
}
@@ -3657,43 +2513,24 @@
return EINVAL;
}
- /*
- * Reset signal mask on the workqueue thread to default state,
- * but do not touch any signals that are marked for preservation.
- */
- sigset_t resettable = uth->uu_sigmask & ~p->p_workq_allow_sigmask;
- if (resettable != (sigset_t)~workq_threadmask) {
+ /* reset signal mask on the workqueue thread to default state */
+ if (uth->uu_sigmask != (sigset_t)(~workq_threadmask)) {
proc_lock(p);
- uth->uu_sigmask |= ~workq_threadmask & ~p->p_workq_allow_sigmask;
+ uth->uu_sigmask = ~workq_threadmask;
proc_unlock(p);
}
- if (kqr && kqr->tr_flags & WORKQ_TR_FLAG_WL_PARAMS) {
+ if (kqr && kqr->kqr_req.tr_flags & TR_FLAG_WL_PARAMS) {
/*
* Ensure we store the threadreq param before unbinding
* the kqr from this thread.
*/
- trp = kqueue_threadreq_workloop_param(kqr);
- }
-
- if (kqr && kqr->tr_flags & WORKQ_TR_FLAG_PERMANENT_BIND) {
- goto handle_stack_events;
- }
-
- /*
- * Freeze the base pri while we decide the fate of this thread.
- *
- * Either:
- * - we return to user and kevent_cleanup will have unfrozen the base pri,
- * - or we proceed to workq_select_threadreq_or_park_and_unlock() who will.
- */
- thread_freeze_base_pri(th);
-
-handle_stack_events:
+ trp = kqueue_threadreq_workloop_param(&kqr->kqr_req);
+ }
if (kqr) {
uint32_t upcall_flags = WQ_FLAG_THREAD_NEWSPI | WQ_FLAG_THREAD_REUSE;
- if (kqr->tr_flags & WORKQ_TR_FLAG_WORKLOOP) {
+ if (kqr->kqr_state & KQR_WORKLOOP) {
upcall_flags |= WQ_FLAG_THREAD_WORKLOOP | WQ_FLAG_THREAD_KEVENT;
} else {
upcall_flags |= WQ_FLAG_THREAD_KEVENT;
@@ -3701,23 +2538,21 @@
if (uth->uu_workq_pri.qos_bucket == WORKQ_THREAD_QOS_MANAGER) {
upcall_flags |= WQ_FLAG_THREAD_EVENT_MANAGER;
} else {
- if (workq_thread_is_overcommit(uth)) {
+ if (uth->uu_workq_flags & UT_WORKQ_OVERCOMMIT) {
upcall_flags |= WQ_FLAG_THREAD_OVERCOMMIT;
}
if (uth->uu_workq_flags & UT_WORKQ_OUTSIDE_QOS) {
upcall_flags |= WQ_FLAG_THREAD_OUTSIDEQOS;
} else {
upcall_flags |= uth->uu_workq_pri.qos_req |
- WQ_FLAG_THREAD_PRIO_QOS;
+ WQ_FLAG_THREAD_PRIO_QOS;
}
}
+
error = pthread_functions->workq_handle_stack_events(p, th,
- get_task_map(proc_task(p)), uth->uu_workq_stackaddr,
- uth->uu_workq_thport, eventlist, nevents, upcall_flags);
- if (error) {
- assert(uth->uu_kqr_bound == kqr);
- return error;
- }
+ get_task_map(p->task), uth->uu_workq_stackaddr,
+ uth->uu_workq_thport, eventlist, nevents, upcall_flags);
+ if (error) return error;
// pthread is supposed to pass KEVENT_FLAG_PARKING here
// which should cause the above call to either:
@@ -3727,7 +2562,7 @@
assert(uth->uu_kqr_bound == NULL);
}
- WQ_TRACE_WQ(TRACE_wq_runthread | DBG_FUNC_END, wq, uap->options, 0, 0);
+ WQ_TRACE_WQ(TRACE_wq_runthread | DBG_FUNC_END, wq, uap->options, 0, 0, 0);
thread_sched_call(th, NULL);
thread_will_park_or_terminate(th);
@@ -3736,10 +2571,9 @@
#endif
workq_lock_spin(wq);
- WQ_TRACE_WQ(TRACE_wq_thread_logical_run | DBG_FUNC_END, wq, 0, 0, 0);
+ WQ_TRACE_WQ(TRACE_wq_thread_logical_run | DBG_FUNC_END, wq, 0, 0, 0, 0);
uth->uu_save.uus_workq_park_data.workloop_params = trp.trp_value;
- workq_select_threadreq_or_park_and_unlock(p, wq, uth,
- WQ_SETUP_CLEAR_VOUCHER);
+ workq_select_threadreq_or_park_and_unlock(p, wq, uth);
__builtin_unreachable();
}
@@ -3766,7 +2600,7 @@
* arg3 = kevent support
*/
int offset = arg2;
- if (arg3 & 0x01) {
+ if (arg3 & 0x01){
// If we get here, then userspace has indicated support for kevent delivery.
}
@@ -3778,16 +2612,7 @@
* arg2 = number of threads to start
* arg3 = priority
*/
- error = workq_reqthreads(p, arg2, arg3, false);
- break;
- }
- /* For requesting threads for the cooperative pool */
- case WQOPS_QUEUE_REQTHREADS2: {
- /*
- * arg2 = number of threads to start
- * arg3 = priority
- */
- error = workq_reqthreads(p, arg2, arg3, true);
+ error = workq_reqthreads(p, arg2, arg3);
break;
}
case WQOPS_SET_EVENT_MANAGER_PRIORITY: {
@@ -3808,14 +2633,14 @@
/*
* Normalize the incoming priority so that it is ordered numerically.
*/
- if (_pthread_priority_has_sched_pri(pri)) {
+ if (pri & _PTHREAD_PRIORITY_SCHED_PRI_FLAG) {
pri &= (_PTHREAD_PRIORITY_SCHED_PRI_MASK |
- _PTHREAD_PRIORITY_SCHED_PRI_FLAG);
+ _PTHREAD_PRIORITY_SCHED_PRI_FLAG);
} else {
thread_qos_t qos = _pthread_priority_thread_qos(pri);
int relpri = _pthread_priority_relpri(pri);
if (relpri > 0 || relpri < THREAD_QOS_MIN_TIER_IMPORTANCE ||
- qos == THREAD_QOS_UNSPECIFIED) {
+ qos == THREAD_QOS_UNSPECIFIED) {
error = EINVAL;
break;
}
@@ -3848,7 +2673,7 @@
thread_t th = current_thread();
struct uthread *uth = get_bsdthread_info(th);
if (((thread_get_tag(th) & THREAD_TAG_WORKQUEUE) == 0) ||
- (uth->uu_workq_flags & (UT_WORKQ_DYING | UT_WORKQ_OVERCOMMIT))) {
+ (uth->uu_workq_flags & (UT_WORKQ_DYING|UT_WORKQ_OVERCOMMIT))) {
error = EINVAL;
break;
}
@@ -3859,39 +2684,10 @@
break;
}
workq_lock_spin(wq);
- bool should_narrow = !workq_constrained_allowance(wq, qos, uth, false, false);
+ bool should_narrow = !workq_constrained_allowance(wq, qos, uth, false);
workq_unlock(wq);
*retval = should_narrow;
- break;
- }
- case WQOPS_SETUP_DISPATCH: {
- /*
- * item = pointer to workq_dispatch_config structure
- * arg2 = sizeof(item)
- */
- struct workq_dispatch_config cfg;
- bzero(&cfg, sizeof(cfg));
-
- error = copyin(uap->item, &cfg, MIN(sizeof(cfg), (unsigned long) arg2));
- if (error) {
- break;
- }
-
- if (cfg.wdc_flags & ~WORKQ_DISPATCH_SUPPORTED_FLAGS ||
- cfg.wdc_version < WORKQ_DISPATCH_MIN_SUPPORTED_VERSION) {
- error = ENOTSUP;
- break;
- }
-
- /* Load fields from version 1 */
- p->p_dispatchqueue_serialno_offset = cfg.wdc_queue_serialno_offs;
-
- /* Load fields from version 2 */
- if (cfg.wdc_version >= 2) {
- p->p_dispatchqueue_label_offset = cfg.wdc_queue_label_offs;
- }
-
break;
}
default:
@@ -3899,7 +2695,7 @@
break;
}
- return error;
+ return (error);
}
/*
@@ -3909,39 +2705,15 @@
*/
__attribute__((noreturn, noinline))
static void
-workq_park_and_unlock(proc_t p, struct workqueue *wq, struct uthread *uth,
- uint32_t setup_flags)
+workq_park_and_unlock(proc_t p, struct workqueue *wq, struct uthread *uth)
{
assert(uth == current_uthread());
assert(uth->uu_kqr_bound == NULL);
- workq_push_idle_thread(p, wq, uth, setup_flags); // may not return
+ workq_push_idle_thread(p, wq, uth); // may not return
workq_thread_reset_cpupercent(NULL, uth);
-#if CONFIG_PREADOPT_TG
- /* Clear the preadoption thread group on the thread.
- *
- * Case 1:
- * Creator thread which never picked up a thread request. We set a
- * preadoption thread group on creator threads but if it never picked
- * up a thread request and didn't go to userspace, then the thread will
- * park with a preadoption thread group but no explicitly adopted
- * voucher or work interval.
- *
- * We drop the preadoption thread group here before proceeding to park.
- * Note - we may get preempted when we drop the workq lock below.
- *
- * Case 2:
- * Thread picked up a thread request and bound to it and returned back
- * from userspace and is parking. At this point, preadoption thread
- * group should be NULL since the thread has unbound from the thread
- * request. So this operation should be a no-op.
- */
- thread_set_preadopt_thread_group(get_machthread(uth), NULL);
-#endif
-
- if ((uth->uu_workq_flags & UT_WORKQ_IDLE_CLEANUP) &&
- !(uth->uu_workq_flags & UT_WORKQ_DYING)) {
+ if (uth->uu_workq_flags & UT_WORKQ_IDLE_CLEANUP) {
workq_unlock(wq);
/*
@@ -3949,9 +2721,8 @@
* if it wants us to free the stack before parking.
*/
if (!uth->uu_save.uus_workq_park_data.has_stack) {
- pthread_functions->workq_markfree_threadstack(p,
- get_machthread(uth), get_task_map(proc_task(p)),
- uth->uu_workq_stackaddr);
+ pthread_functions->workq_markfree_threadstack(p, uth->uu_thread,
+ get_task_map(p->task), uth->uu_workq_stackaddr);
}
/*
@@ -3960,10 +2731,6 @@
* the idle list. Then, when we get our importance back we'll be able to
* use this thread from e.g. the kevent call out to deliver a boosting
* message.
- *
- * Note that setting the voucher to NULL will not clear the preadoption
- * thread since this thread could have become the creator again and
- * perhaps acquired a preadoption thread group.
*/
__assert_only kern_return_t kr;
kr = thread_set_voucher_name(MACH_PORT_NULL);
@@ -3971,10 +2738,7 @@
workq_lock_spin(wq);
uth->uu_workq_flags &= ~UT_WORKQ_IDLE_CLEANUP;
- setup_flags &= ~WQ_SETUP_CLEAR_VOUCHER;
- }
-
- WQ_TRACE_WQ(TRACE_wq_thread_logical_run | DBG_FUNC_END, wq, 0, 0, 0);
+ }
if (uth->uu_workq_flags & UT_WORKQ_RUNNING) {
/*
@@ -3983,22 +2747,21 @@
* event their thread_wakeup() was ineffectual. To correct for that,
* we just run the continuation ourselves.
*/
- workq_unpark_select_threadreq_or_park_and_unlock(p, wq, uth, setup_flags);
+ WQ_TRACE_WQ(TRACE_wq_thread_logical_run | DBG_FUNC_END, wq, 0, 0, 0, 0);
+ workq_select_threadreq_or_park_and_unlock(p, wq, uth);
__builtin_unreachable();
}
if (uth->uu_workq_flags & UT_WORKQ_DYING) {
workq_unpark_for_death_and_unlock(p, wq, uth,
- WORKQ_UNPARK_FOR_DEATH_WAS_IDLE, setup_flags);
+ WORKQ_UNPARK_FOR_DEATH_WAS_IDLE);
__builtin_unreachable();
}
- /* Disarm the workqueue quantum since the thread is now idle */
- thread_disarm_workqueue_quantum(get_machthread(uth));
-
- thread_set_pending_block_hint(get_machthread(uth), kThreadWaitParkedWorkQueue);
+ thread_set_pending_block_hint(uth->uu_thread, kThreadWaitParkedWorkQueue);
assert_wait(workq_parked_wait_event(uth), THREAD_INTERRUPTIBLE);
workq_unlock(wq);
+ WQ_TRACE_WQ(TRACE_wq_thread_logical_run | DBG_FUNC_END, wq, 0, 0, 0, 0);
thread_block(workq_unpark_continue);
__builtin_unreachable();
}
@@ -4012,20 +2775,18 @@
* - we are re-using the event manager
*/
return wq->wq_thscheduled_count[_wq_bucket(WORKQ_THREAD_QOS_MANAGER)] == 0 ||
- (uth && uth->uu_workq_pri.qos_bucket == WORKQ_THREAD_QOS_MANAGER);
-}
-
-/* Called with workq lock held. */
+ (uth && uth->uu_workq_pri.qos_bucket == WORKQ_THREAD_QOS_MANAGER);
+}
+
static uint32_t
workq_constrained_allowance(struct workqueue *wq, thread_qos_t at_qos,
- struct uthread *uth, bool may_start_timer, bool record_failed_allowance)
+ struct uthread *uth, bool may_start_timer)
{
assert(at_qos != WORKQ_THREAD_QOS_MANAGER);
- uint32_t allowance_passed = 0;
uint32_t count = 0;
uint32_t max_count = wq->wq_constrained_threads_scheduled;
- if (uth && workq_thread_is_nonovercommit(uth)) {
+ if (uth && (uth->uu_workq_flags & UT_WORKQ_OVERCOMMIT) == 0) {
/*
* don't count the current thread as scheduled
*/
@@ -4034,36 +2795,32 @@
}
if (max_count >= wq_max_constrained_threads) {
WQ_TRACE_WQ(TRACE_wq_constrained_admission | DBG_FUNC_NONE, wq, 1,
- wq->wq_constrained_threads_scheduled,
- wq_max_constrained_threads);
+ wq->wq_constrained_threads_scheduled,
+ wq_max_constrained_threads, 0);
/*
* we need 1 or more constrained threads to return to the kernel before
* we can dispatch additional work
*/
- allowance_passed = 0;
- goto out;
+ return 0;
}
max_count -= wq_max_constrained_threads;
/*
* Compute a metric for many how many threads are active. We find the
- * highest priority request outstanding and then add up the number of active
- * threads in that and all higher-priority buckets. We'll also add any
- * "busy" threads which are not currently active but blocked recently enough
- * that we can't be sure that they won't be unblocked soon and start
- * being active again.
- *
- * We'll then compare this metric to our max concurrency to decide whether
- * to add a new thread.
+ * highest priority request outstanding and then add up the number of
+ * active threads in that and all higher-priority buckets. We'll also add
+ * any "busy" threads which are not active but blocked recently enough that
+ * we can't be sure they've gone idle yet. We'll then compare this metric
+ * to our max concurrency to decide whether to add a new thread.
*/
uint32_t busycount, thactive_count;
thactive_count = _wq_thactive_aggregate_downto_qos(wq, _wq_thactive(wq),
- at_qos, &busycount, NULL);
+ at_qos, &busycount, NULL);
if (uth && uth->uu_workq_pri.qos_bucket != WORKQ_THREAD_QOS_MANAGER &&
- at_qos <= uth->uu_workq_pri.qos_bucket) {
+ at_qos <= uth->uu_workq_pri.qos_bucket) {
/*
* Don't count this thread as currently active, but only if it's not
* a manager thread, as _wq_thactive_aggregate_downto_qos ignores active
@@ -4077,16 +2834,14 @@
if (count > thactive_count + busycount) {
count -= thactive_count + busycount;
WQ_TRACE_WQ(TRACE_wq_constrained_admission | DBG_FUNC_NONE, wq, 2,
- thactive_count, busycount);
- allowance_passed = MIN(count, max_count);
- goto out;
+ thactive_count, busycount, 0);
+ return MIN(count, max_count);
} else {
WQ_TRACE_WQ(TRACE_wq_constrained_admission | DBG_FUNC_NONE, wq, 3,
- thactive_count, busycount);
- allowance_passed = 0;
- }
-
- if (may_start_timer) {
+ thactive_count, busycount, 0);
+ }
+
+ if (busycount && may_start_timer) {
/*
* If this is called from the add timer, we won't have another timer
* fire when the thread exits the "busy" state, so rearm the timer.
@@ -4094,155 +2849,63 @@
workq_schedule_delayed_thread_creation(wq, 0);
}
-out:
- if (record_failed_allowance) {
- wq->wq_exceeded_active_constrained_thread_limit = !allowance_passed;
- }
- return allowance_passed;
+ return 0;
}
static bool
workq_threadreq_admissible(struct workqueue *wq, struct uthread *uth,
- workq_threadreq_t req)
+ workq_threadreq_t req)
{
if (req->tr_qos == WORKQ_THREAD_QOS_MANAGER) {
return workq_may_start_event_mgr_thread(wq, uth);
}
- if (workq_threadreq_is_cooperative(req)) {
- return workq_cooperative_allowance(wq, req->tr_qos, uth, true);
- }
- if (workq_threadreq_is_nonovercommit(req)) {
- return workq_constrained_allowance(wq, req->tr_qos, uth, true, true);
- }
-
+ if ((req->tr_flags & TR_FLAG_OVERCOMMIT) == 0) {
+ return workq_constrained_allowance(wq, req->tr_qos, uth, true);
+ }
return true;
-}
-
-/*
- * Called from the context of selecting thread requests for threads returning
- * from userspace or creator thread
- */
-static workq_threadreq_t
-workq_cooperative_queue_best_req(struct workqueue *wq, struct uthread *uth)
-{
- workq_lock_held(wq);
-
- /*
- * If the current thread is cooperative, we need to exclude it as part of
- * cooperative schedule count since this thread is looking for a new
- * request. Change in the schedule count for cooperative pool therefore
- * requires us to reeevaluate the next best request for it.
- */
- if (uth && workq_thread_is_cooperative(uth)) {
- _wq_cooperative_queue_scheduled_count_dec(wq, uth->uu_workq_pri.qos_req);
-
- (void) _wq_cooperative_queue_refresh_best_req_qos(wq);
-
- _wq_cooperative_queue_scheduled_count_inc(wq, uth->uu_workq_pri.qos_req);
- } else {
- /*
- * The old value that was already precomputed should be safe to use -
- * add an assert that asserts that the best req QoS doesn't change in
- * this case
- */
- assert(_wq_cooperative_queue_refresh_best_req_qos(wq) == false);
- }
-
- thread_qos_t qos = wq->wq_cooperative_queue_best_req_qos;
-
- /* There are no eligible requests in the cooperative pool */
- if (qos == THREAD_QOS_UNSPECIFIED) {
- return NULL;
- }
- assert(qos != WORKQ_THREAD_QOS_ABOVEUI);
- assert(qos != WORKQ_THREAD_QOS_MANAGER);
-
- uint8_t bucket = _wq_bucket(qos);
- assert(!STAILQ_EMPTY(&wq->wq_cooperative_queue[bucket]));
-
- return STAILQ_FIRST(&wq->wq_cooperative_queue[bucket]);
}
static workq_threadreq_t
workq_threadreq_select_for_creator(struct workqueue *wq)
{
- workq_threadreq_t req_qos, req_pri, req_tmp, req_mgr;
+ workq_threadreq_t req_qos, req_pri, req_tmp;
thread_qos_t qos = THREAD_QOS_UNSPECIFIED;
uint8_t pri = 0;
+
+ req_tmp = wq->wq_event_manager_threadreq;
+ if (req_tmp && workq_may_start_event_mgr_thread(wq, NULL)) {
+ return req_tmp;
+ }
/*
* Compute the best priority request, and ignore the turnstile for now
*/
req_pri = priority_queue_max(&wq->wq_special_queue,
- struct workq_threadreq_s, tr_entry);
+ struct workq_threadreq_s, tr_entry);
if (req_pri) {
- pri = (uint8_t)priority_queue_entry_sched_pri(&wq->wq_special_queue,
- &req_pri->tr_entry);
- }
-
- /*
- * Handle the manager thread request. The special queue might yield
- * a higher priority, but the manager always beats the QoS world.
- */
-
- req_mgr = wq->wq_event_manager_threadreq;
- if (req_mgr && workq_may_start_event_mgr_thread(wq, NULL)) {
- uint32_t mgr_pri = wq->wq_event_manager_priority;
-
- if (mgr_pri & _PTHREAD_PRIORITY_SCHED_PRI_FLAG) {
- mgr_pri &= _PTHREAD_PRIORITY_SCHED_PRI_MASK;
- } else {
- mgr_pri = thread_workq_pri_for_qos(
- _pthread_priority_thread_qos(mgr_pri));
- }
-
- return mgr_pri >= pri ? req_mgr : req_pri;
+ pri = priority_queue_entry_key(&wq->wq_special_queue, &req_pri->tr_entry);
}
/*
* Compute the best QoS Request, and check whether it beats the "pri" one
- *
- * Start by comparing the overcommit and the cooperative pool
*/
+
req_qos = priority_queue_max(&wq->wq_overcommit_queue,
- struct workq_threadreq_s, tr_entry);
+ struct workq_threadreq_s, tr_entry);
if (req_qos) {
qos = req_qos->tr_qos;
}
- req_tmp = workq_cooperative_queue_best_req(wq, NULL);
- if (req_tmp && qos <= req_tmp->tr_qos) {
- /*
- * Cooperative TR is better between overcommit and cooperative. Note
- * that if qos is same between overcommit and cooperative, we choose
- * cooperative.
- *
- * Pick cooperative pool if it passes the admissions check
- */
- if (workq_cooperative_allowance(wq, req_tmp->tr_qos, NULL, true)) {
- req_qos = req_tmp;
- qos = req_qos->tr_qos;
- }
- }
-
- /*
- * Compare the best QoS so far - either from overcommit or from cooperative
- * pool - and compare it with the constrained pool
- */
req_tmp = priority_queue_max(&wq->wq_constrained_queue,
- struct workq_threadreq_s, tr_entry);
+ struct workq_threadreq_s, tr_entry);
if (req_tmp && qos < req_tmp->tr_qos) {
- /*
- * Constrained pool is best in QoS between overcommit, cooperative
- * and constrained. Now check how it fairs against the priority case
- */
if (pri && pri >= thread_workq_pri_for_qos(req_tmp->tr_qos)) {
return req_pri;
}
- if (workq_constrained_allowance(wq, req_tmp->tr_qos, NULL, true, true)) {
+ if (workq_constrained_allowance(wq, req_tmp->tr_qos, NULL, true)) {
/*
* If the constrained thread request is the best one and passes
* the admission check, pick it.
@@ -4251,9 +2914,6 @@
}
}
- /*
- * Compare the best of the QoS world with the priority
- */
if (pri && (!qos || pri >= thread_workq_pri_for_qos(qos))) {
return req_pri;
}
@@ -4281,133 +2941,44 @@
return NULL;
}
-/*
- * Returns true if this caused a change in the schedule counts of the
- * cooperative pool
- */
-static bool
-workq_adjust_cooperative_constrained_schedule_counts(struct workqueue *wq,
- struct uthread *uth, thread_qos_t old_thread_qos, workq_tr_flags_t tr_flags)
-{
- workq_lock_held(wq);
-
- /*
- * Row: thread type
- * Column: Request type
- *
- * overcommit non-overcommit cooperative
- * overcommit X case 1 case 2
- * cooperative case 3 case 4 case 5
- * non-overcommit case 6 X case 7
- *
- * Move the thread to the right bucket depending on what state it currently
- * has and what state the thread req it picks, is going to have.
- *
- * Note that the creator thread is an overcommit thread.
- */
- thread_qos_t new_thread_qos = uth->uu_workq_pri.qos_req;
-
- /*
- * Anytime a cooperative bucket's schedule count changes, we need to
- * potentially refresh the next best QoS for that pool when we determine
- * the next request for the creator
- */
- bool cooperative_pool_sched_count_changed = false;
-
- if (workq_thread_is_overcommit(uth)) {
- if (workq_tr_is_nonovercommit(tr_flags)) {
- // Case 1: thread is overcommit, req is non-overcommit
- wq->wq_constrained_threads_scheduled++;
- } else if (workq_tr_is_cooperative(tr_flags)) {
- // Case 2: thread is overcommit, req is cooperative
- _wq_cooperative_queue_scheduled_count_inc(wq, new_thread_qos);
- cooperative_pool_sched_count_changed = true;
- }
- } else if (workq_thread_is_cooperative(uth)) {
- if (workq_tr_is_overcommit(tr_flags)) {
- // Case 3: thread is cooperative, req is overcommit
- _wq_cooperative_queue_scheduled_count_dec(wq, old_thread_qos);
- } else if (workq_tr_is_nonovercommit(tr_flags)) {
- // Case 4: thread is cooperative, req is non-overcommit
- _wq_cooperative_queue_scheduled_count_dec(wq, old_thread_qos);
- wq->wq_constrained_threads_scheduled++;
- } else {
- // Case 5: thread is cooperative, req is also cooperative
- assert(workq_tr_is_cooperative(tr_flags));
- _wq_cooperative_queue_scheduled_count_dec(wq, old_thread_qos);
- _wq_cooperative_queue_scheduled_count_inc(wq, new_thread_qos);
- }
- cooperative_pool_sched_count_changed = true;
- } else {
- if (workq_tr_is_overcommit(tr_flags)) {
- // Case 6: Thread is non-overcommit, req is overcommit
- wq->wq_constrained_threads_scheduled--;
- } else if (workq_tr_is_cooperative(tr_flags)) {
- // Case 7: Thread is non-overcommit, req is cooperative
- wq->wq_constrained_threads_scheduled--;
- _wq_cooperative_queue_scheduled_count_inc(wq, new_thread_qos);
- cooperative_pool_sched_count_changed = true;
- }
- }
-
- return cooperative_pool_sched_count_changed;
-}
-
static workq_threadreq_t
workq_threadreq_select(struct workqueue *wq, struct uthread *uth)
{
- workq_threadreq_t req_qos, req_pri, req_tmp, req_mgr;
+ workq_threadreq_t req_qos, req_pri, req_tmp;
uintptr_t proprietor;
thread_qos_t qos = THREAD_QOS_UNSPECIFIED;
uint8_t pri = 0;
- if (uth == wq->wq_creator) {
- uth = NULL;
+ if (uth == wq->wq_creator) uth = NULL;
+
+ req_tmp = wq->wq_event_manager_threadreq;
+ if (req_tmp && workq_may_start_event_mgr_thread(wq, uth)) {
+ return req_tmp;
}
/*
* Compute the best priority request (special or turnstile)
*/
- pri = (uint8_t)turnstile_workq_proprietor_of_max_turnstile(wq->wq_turnstile,
- &proprietor);
+ pri = turnstile_workq_proprietor_of_max_turnstile(wq->wq_turnstile,
+ &proprietor);
if (pri) {
struct kqworkloop *kqwl = (struct kqworkloop *)proprietor;
- req_pri = &kqwl->kqwl_request;
- if (req_pri->tr_state != WORKQ_TR_STATE_QUEUED) {
+ req_pri = &kqwl->kqwl_request.kqr_req;
+ if (req_pri->tr_state != TR_STATE_QUEUED) {
panic("Invalid thread request (%p) state %d",
- req_pri, req_pri->tr_state);
+ req_pri, req_pri->tr_state);
}
} else {
req_pri = NULL;
}
req_tmp = priority_queue_max(&wq->wq_special_queue,
- struct workq_threadreq_s, tr_entry);
- if (req_tmp && pri < priority_queue_entry_sched_pri(&wq->wq_special_queue,
- &req_tmp->tr_entry)) {
+ struct workq_threadreq_s, tr_entry);
+ if (req_tmp && pri < priority_queue_entry_key(&wq->wq_special_queue,
+ &req_tmp->tr_entry)) {
req_pri = req_tmp;
- pri = (uint8_t)priority_queue_entry_sched_pri(&wq->wq_special_queue,
- &req_tmp->tr_entry);
- }
-
- /*
- * Handle the manager thread request. The special queue might yield
- * a higher priority, but the manager always beats the QoS world.
- */
-
- req_mgr = wq->wq_event_manager_threadreq;
- if (req_mgr && workq_may_start_event_mgr_thread(wq, uth)) {
- uint32_t mgr_pri = wq->wq_event_manager_priority;
-
- if (mgr_pri & _PTHREAD_PRIORITY_SCHED_PRI_FLAG) {
- mgr_pri &= _PTHREAD_PRIORITY_SCHED_PRI_MASK;
- } else {
- mgr_pri = thread_workq_pri_for_qos(
- _pthread_priority_thread_qos(mgr_pri));
- }
-
- return mgr_pri >= pri ? req_mgr : req_pri;
+ pri = priority_queue_entry_key(&wq->wq_special_queue, &req_tmp->tr_entry);
}
/*
@@ -4415,43 +2986,20 @@
*/
req_qos = priority_queue_max(&wq->wq_overcommit_queue,
- struct workq_threadreq_s, tr_entry);
+ struct workq_threadreq_s, tr_entry);
if (req_qos) {
qos = req_qos->tr_qos;
}
- req_tmp = workq_cooperative_queue_best_req(wq, uth);
- if (req_tmp && qos <= req_tmp->tr_qos) {
- /*
- * Cooperative TR is better between overcommit and cooperative. Note
- * that if qos is same between overcommit and cooperative, we choose
- * cooperative.
- *
- * Pick cooperative pool if it passes the admissions check
- */
- if (workq_cooperative_allowance(wq, req_tmp->tr_qos, uth, true)) {
- req_qos = req_tmp;
- qos = req_qos->tr_qos;
- }
- }
-
- /*
- * Compare the best QoS so far - either from overcommit or from cooperative
- * pool - and compare it with the constrained pool
- */
req_tmp = priority_queue_max(&wq->wq_constrained_queue,
- struct workq_threadreq_s, tr_entry);
+ struct workq_threadreq_s, tr_entry);
if (req_tmp && qos < req_tmp->tr_qos) {
- /*
- * Constrained pool is best in QoS between overcommit, cooperative
- * and constrained. Now check how it fairs against the priority case
- */
if (pri && pri >= thread_workq_pri_for_qos(req_tmp->tr_qos)) {
return req_pri;
}
- if (workq_constrained_allowance(wq, req_tmp->tr_qos, uth, true, true)) {
+ if (workq_constrained_allowance(wq, req_tmp->tr_qos, uth, true)) {
/*
* If the constrained thread request is the best one and passes
* the admission check, pick it.
@@ -4489,12 +3037,10 @@
* efficient scheduling and reduced context switches.
*/
static void
-workq_schedule_creator(proc_t p, struct workqueue *wq,
- workq_kern_threadreq_flags_t flags)
+workq_schedule_creator(proc_t p, struct workqueue *wq, int flags)
{
workq_threadreq_t req;
struct uthread *uth;
- bool needs_wakeup;
workq_lock_held(wq);
assert(p || (flags & WORKQ_THREADREQ_CAN_CREATE_THREADS) == 0);
@@ -4503,14 +3049,6 @@
uth = wq->wq_creator;
if (!wq->wq_reqcount) {
- /*
- * There is no thread request left.
- *
- * If there is a creator, leave everything in place, so that it cleans
- * up itself in workq_push_idle_thread().
- *
- * Else, make sure the turnstile state is reset to no inheritor.
- */
if (uth == NULL) {
workq_turnstile_update_inheritor(wq, TURNSTILE_INHERITOR_NULL, 0);
}
@@ -4519,21 +3057,17 @@
req = workq_threadreq_select_for_creator(wq);
if (req == NULL) {
- /*
- * There isn't a thread request that passes the admission check.
- *
- * If there is a creator, do not touch anything, the creator will sort
- * it out when it runs.
- *
- * Else, set the inheritor to "WORKQ" so that the turnstile propagation
- * code calls us if anything changes.
- */
- if (uth == NULL) {
+ if (flags & WORKQ_THREADREQ_CREATOR_SYNC_UPDATE) {
+ assert((flags & WORKQ_THREADREQ_CREATOR_TRANSFER) == 0);
+ /*
+ * turnstile propagation code is reaching out to us,
+ * and we still don't want to do anything, do not recurse.
+ */
+ } else {
workq_turnstile_update_inheritor(wq, wq, TURNSTILE_INHERITOR_WORKQ);
}
return;
}
-
if (uth) {
/*
@@ -4541,265 +3075,51 @@
*/
if (workq_thread_needs_priority_change(req, uth)) {
WQ_TRACE_WQ(TRACE_wq_creator_select | DBG_FUNC_NONE,
- wq, 1, uthread_tid(uth), req->tr_qos);
- workq_thread_reset_pri(wq, uth, req, /*unpark*/ true);
- }
- assert(wq->wq_inheritor == get_machthread(uth));
+ wq, 1, thread_tid(uth->uu_thread), req->tr_qos, 0);
+ workq_thread_reset_pri(wq, uth, req);
+ }
} else if (wq->wq_thidlecount) {
/*
* We need to unpark a creator thread
*/
- wq->wq_creator = uth = workq_pop_idle_thread(wq, UT_WORKQ_OVERCOMMIT,
- &needs_wakeup);
- /* Always reset the priorities on the newly chosen creator */
- workq_thread_reset_pri(wq, uth, req, /*unpark*/ true);
- workq_turnstile_update_inheritor(wq, get_machthread(uth),
- TURNSTILE_INHERITOR_THREAD);
+ wq->wq_creator = uth = workq_pop_idle_thread(wq);
+ if (workq_thread_needs_priority_change(req, uth)) {
+ workq_thread_reset_pri(wq, uth, req);
+ }
+ workq_turnstile_update_inheritor(wq, uth->uu_thread,
+ TURNSTILE_INHERITOR_THREAD);
WQ_TRACE_WQ(TRACE_wq_creator_select | DBG_FUNC_NONE,
- wq, 2, uthread_tid(uth), req->tr_qos);
+ wq, 2, thread_tid(uth->uu_thread), req->tr_qos, 0);
uth->uu_save.uus_workq_park_data.fulfilled_snapshot = wq->wq_fulfilled;
uth->uu_save.uus_workq_park_data.yields = 0;
- if (needs_wakeup) {
- workq_thread_wakeup(uth);
- }
+ workq_thread_wakeup(uth);
} else {
/*
* We need to allocate a thread...
*/
if (__improbable(wq->wq_nthreads >= wq_max_threads)) {
/* out of threads, just go away */
- flags = WORKQ_THREADREQ_NONE;
} else if (flags & WORKQ_THREADREQ_SET_AST_ON_FAILURE) {
act_set_astkevent(current_thread(), AST_KEVENT_REDRIVE_THREADREQ);
} else if (!(flags & WORKQ_THREADREQ_CAN_CREATE_THREADS)) {
/* This can drop the workqueue lock, and take it again */
workq_schedule_immediate_thread_creation(wq);
- } else if ((workq_add_new_idle_thread(p, wq,
- workq_unpark_continue, false, NULL) == KERN_SUCCESS)) {
+ } else if (workq_add_new_idle_thread(p, wq)) {
goto again;
} else {
workq_schedule_delayed_thread_creation(wq, 0);
}
- /*
- * If the current thread is the inheritor:
- *
- * If we set the AST, then the thread will stay the inheritor until
- * either the AST calls workq_kern_threadreq_redrive(), or it parks
- * and calls workq_push_idle_thread().
- *
- * Else, the responsibility of the thread creation is with a thread-call
- * and we need to clear the inheritor.
- */
- if ((flags & WORKQ_THREADREQ_SET_AST_ON_FAILURE) == 0 &&
- wq->wq_inheritor == current_thread()) {
- workq_turnstile_update_inheritor(wq, TURNSTILE_INHERITOR_NULL, 0);
- }
- }
-}
-
-/**
- * Same as workq_unpark_select_threadreq_or_park_and_unlock,
- * but do not allow early binds.
- *
- * Called with the base pri frozen, will unfreeze it.
- */
-__attribute__((noreturn, noinline))
-static void
-workq_select_threadreq_or_park_and_unlock(proc_t p, struct workqueue *wq,
- struct uthread *uth, uint32_t setup_flags)
-{
- workq_threadreq_t req = NULL;
- bool is_creator = (wq->wq_creator == uth);
- bool schedule_creator = false;
-
- if (__improbable(_wq_exiting(wq))) {
- WQ_TRACE_WQ(TRACE_wq_select_threadreq | DBG_FUNC_NONE, wq, 0, 0, 0);
- goto park;
- }
-
- if (wq->wq_reqcount == 0) {
- WQ_TRACE_WQ(TRACE_wq_select_threadreq | DBG_FUNC_NONE, wq, 1, 0, 0);
- goto park;
- }
-
- req = workq_threadreq_select(wq, uth);
- if (__improbable(req == NULL)) {
- WQ_TRACE_WQ(TRACE_wq_select_threadreq | DBG_FUNC_NONE, wq, 2, 0, 0);
- goto park;
- }
-
- struct uu_workq_policy old_pri = uth->uu_workq_pri;
- uint8_t tr_flags = req->tr_flags;
- struct turnstile *req_ts = kqueue_threadreq_get_turnstile(req);
-
- /*
- * Attempt to setup ourselves as the new thing to run, moving all priority
- * pushes to ourselves.
- *
- * If the current thread is the creator, then the fact that we are presently
- * running is proof that we'll do something useful, so keep going.
- *
- * For other cases, peek at the AST to know whether the scheduler wants
- * to preempt us, if yes, park instead, and move the thread request
- * turnstile back to the workqueue.
- */
- if (req_ts) {
- workq_perform_turnstile_operation_locked(wq, ^{
- turnstile_update_inheritor(req_ts, get_machthread(uth),
- TURNSTILE_IMMEDIATE_UPDATE | TURNSTILE_INHERITOR_THREAD);
- turnstile_update_inheritor_complete(req_ts,
- TURNSTILE_INTERLOCK_HELD);
- });
- }
-
- /* accounting changes of aggregate thscheduled_count and thactive which has
- * to be paired with the workq_thread_reset_pri below so that we have
- * uth->uu_workq_pri match with thactive.
- *
- * This is undone when the thread parks */
- if (is_creator) {
- WQ_TRACE_WQ(TRACE_wq_creator_select, wq, 4, 0,
- uth->uu_save.uus_workq_park_data.yields);
- wq->wq_creator = NULL;
- _wq_thactive_inc(wq, req->tr_qos);
- wq->wq_thscheduled_count[_wq_bucket(req->tr_qos)]++;
- } else if (old_pri.qos_bucket != req->tr_qos) {
- _wq_thactive_move(wq, old_pri.qos_bucket, req->tr_qos);
- }
- workq_thread_reset_pri(wq, uth, req, /*unpark*/ true);
-
- /*
- * Make relevant accounting changes for pool specific counts.
- *
- * The schedule counts changing can affect what the next best request
- * for cooperative thread pool is if this request is dequeued.
- */
- bool cooperative_sched_count_changed =
- workq_adjust_cooperative_constrained_schedule_counts(wq, uth,
- old_pri.qos_req, tr_flags);
-
- if (workq_tr_is_overcommit(tr_flags)) {
- workq_thread_set_type(uth, UT_WORKQ_OVERCOMMIT);
- } else if (workq_tr_is_cooperative(tr_flags)) {
- workq_thread_set_type(uth, UT_WORKQ_COOPERATIVE);
- } else {
- workq_thread_set_type(uth, 0);
- }
-
- if (__improbable(thread_unfreeze_base_pri(get_machthread(uth)) && !is_creator)) {
- if (req_ts) {
- workq_perform_turnstile_operation_locked(wq, ^{
- turnstile_update_inheritor(req_ts, wq->wq_turnstile,
- TURNSTILE_IMMEDIATE_UPDATE | TURNSTILE_INHERITOR_TURNSTILE);
- turnstile_update_inheritor_complete(req_ts,
- TURNSTILE_INTERLOCK_HELD);
- });
- }
- WQ_TRACE_WQ(TRACE_wq_select_threadreq | DBG_FUNC_NONE, wq, 3, 0, 0);
-
- /*
- * If a cooperative thread was the one which picked up the manager
- * thread request, we need to reevaluate the cooperative pool before
- * it goes and parks.
- *
- * For every other of thread request that it picks up, the logic in
- * workq_threadreq_select should have done this refresh.
- * See workq_push_idle_thread.
- */
- if (cooperative_sched_count_changed) {
- if (req->tr_qos == WORKQ_THREAD_QOS_MANAGER) {
- _wq_cooperative_queue_refresh_best_req_qos(wq);
- }
- }
- goto park_thawed;
- }
-
- /*
- * We passed all checks, dequeue the request, bind to it, and set it up
- * to return to user.
- */
- WQ_TRACE_WQ(TRACE_wq_thread_logical_run | DBG_FUNC_START, wq,
- workq_trace_req_id(req), tr_flags, 0);
- wq->wq_fulfilled++;
- schedule_creator = workq_threadreq_dequeue(wq, req,
- cooperative_sched_count_changed);
-
- workq_thread_reset_cpupercent(req, uth);
-
- if (tr_flags & (WORKQ_TR_FLAG_KEVENT | WORKQ_TR_FLAG_WORKLOOP)) {
- kqueue_threadreq_bind_prepost(p, req, uth);
- req = NULL;
- } else if (req->tr_count > 0) {
- req = NULL;
- }
-
- if (uth->uu_workq_flags & UT_WORKQ_NEW) {
- uth->uu_workq_flags ^= UT_WORKQ_NEW;
- setup_flags |= WQ_SETUP_FIRST_USE;
- }
-
- /* If one of the following is true, call workq_schedule_creator (which also
- * adjusts priority of existing creator):
- *
- * - We are the creator currently so the wq may need a new creator
- * - The request we're binding to is the highest priority one, existing
- * creator's priority might need to be adjusted to reflect the next
- * highest TR
- */
- if (is_creator || schedule_creator) {
- /* This can drop the workqueue lock, and take it again */
- workq_schedule_creator(p, wq, WORKQ_THREADREQ_CAN_CREATE_THREADS);
- }
-
- workq_unlock(wq);
-
- if (req) {
- zfree(workq_zone_threadreq, req);
- }
-
- /*
- * Run Thread, Run!
- */
- uint32_t upcall_flags = WQ_FLAG_THREAD_NEWSPI;
- if (uth->uu_workq_pri.qos_bucket == WORKQ_THREAD_QOS_MANAGER) {
- upcall_flags |= WQ_FLAG_THREAD_EVENT_MANAGER;
- } else if (workq_tr_is_overcommit(tr_flags)) {
- upcall_flags |= WQ_FLAG_THREAD_OVERCOMMIT;
- } else if (workq_tr_is_cooperative(tr_flags)) {
- upcall_flags |= WQ_FLAG_THREAD_COOPERATIVE;
- }
- if (tr_flags & WORKQ_TR_FLAG_KEVENT) {
- upcall_flags |= WQ_FLAG_THREAD_KEVENT;
- assert((upcall_flags & WQ_FLAG_THREAD_COOPERATIVE) == 0);
- }
-
- if (tr_flags & WORKQ_TR_FLAG_WORKLOOP) {
- upcall_flags |= WQ_FLAG_THREAD_WORKLOOP | WQ_FLAG_THREAD_KEVENT;
- }
- uth->uu_save.uus_workq_park_data.upcall_flags = upcall_flags;
-
- if (tr_flags & (WORKQ_TR_FLAG_KEVENT | WORKQ_TR_FLAG_WORKLOOP)) {
- kqueue_threadreq_bind_commit(p, get_machthread(uth));
- } else {
-#if CONFIG_PREADOPT_TG
- /*
- * The thread may have a preadopt thread group on it already because it
- * got tagged with it as a creator thread. So we need to make sure to
- * clear that since we don't have preadoption for anonymous thread
- * requests
- */
- thread_set_preadopt_thread_group(get_machthread(uth), NULL);
-#endif
- }
-
- workq_setup_and_run(p, uth, setup_flags);
- __builtin_unreachable();
-
-park:
- thread_unfreeze_base_pri(get_machthread(uth));
-park_thawed:
- workq_park_and_unlock(p, wq, uth, setup_flags);
+ if (flags & WORKQ_THREADREQ_CREATOR_TRANSFER) {
+ /*
+ * workq_schedule_creator() failed at creating a thread,
+ * and the responsibility of redriving is now with a thread-call.
+ *
+ * We still need to tell the turnstile the previous creator is gone.
+ */
+ workq_turnstile_update_inheritor(wq, NULL, 0);
+ }
+ }
}
/**
@@ -4815,14 +3135,16 @@
* Either way, the thread request object serviced will be moved to state
* BINDING and attached to the uthread.
*
- * Should be called with the workqueue lock held. Will drop it.
- * Should be called with the base pri not frozen.
+ * Should be called with the workqueue lock held. Will drop it.
*/
__attribute__((noreturn, noinline))
static void
-workq_unpark_select_threadreq_or_park_and_unlock(proc_t p, struct workqueue *wq,
- struct uthread *uth, uint32_t setup_flags)
-{
+workq_select_threadreq_or_park_and_unlock(proc_t p, struct workqueue *wq,
+ struct uthread *uth)
+{
+ uint32_t setup_flags = 0;
+ workq_threadreq_t req;
+
if (uth->uu_workq_flags & UT_WORKQ_EARLY_BOUND) {
if (uth->uu_workq_flags & UT_WORKQ_NEW) {
setup_flags |= WQ_SETUP_FIRST_USE;
@@ -4831,18 +3153,33 @@
/*
* This pointer is possibly freed and only used for tracing purposes.
*/
- workq_threadreq_t req = uth->uu_save.uus_workq_park_data.thread_request;
+ req = uth->uu_save.uus_workq_park_data.thread_request;
workq_unlock(wq);
WQ_TRACE_WQ(TRACE_wq_thread_logical_run | DBG_FUNC_START, wq,
- VM_KERNEL_ADDRHIDE(req), 0, 0);
- (void)req;
-
+ VM_KERNEL_ADDRHIDE(req), 0, 0, 0);
+ goto run;
+ } else if (_wq_exiting(wq)) {
+ WQ_TRACE_WQ(TRACE_wq_select_threadreq | DBG_FUNC_NONE, wq, 0, 0, 0, 0);
+ } else if (wq->wq_reqcount == 0) {
+ WQ_TRACE_WQ(TRACE_wq_select_threadreq | DBG_FUNC_NONE, wq, 1, 0, 0, 0);
+ } else if ((req = workq_threadreq_select(wq, uth)) == NULL) {
+ WQ_TRACE_WQ(TRACE_wq_select_threadreq | DBG_FUNC_NONE, wq, 2, 0, 0, 0);
+ } else {
+ WQ_TRACE_WQ(TRACE_wq_thread_logical_run | DBG_FUNC_START, wq,
+ workq_trace_req_id(req), 0, 0, 0);
+ if (uth->uu_workq_flags & UT_WORKQ_NEW) {
+ uth->uu_workq_flags ^= UT_WORKQ_NEW;
+ setup_flags |= WQ_SETUP_FIRST_USE;
+ }
+ workq_thread_reset_cpupercent(req, uth);
+ workq_threadreq_bind_and_unlock(p, wq, req, uth);
+run:
workq_setup_and_run(p, uth, setup_flags);
__builtin_unreachable();
}
- thread_freeze_base_pri(get_machthread(uth));
- workq_select_threadreq_or_park_and_unlock(p, wq, uth, setup_flags);
+ workq_park_and_unlock(p, wq, uth);
+ __builtin_unreachable();
}
static bool
@@ -4863,17 +3200,17 @@
if (wq->wq_fulfilled - snapshot > conc) {
/* we fulfilled more than NCPU requests since being dispatched */
WQ_TRACE_WQ(TRACE_wq_creator_yield, wq, 1,
- wq->wq_fulfilled, snapshot);
+ wq->wq_fulfilled, snapshot, 0);
return true;
}
- for (uint8_t i = _wq_bucket(qos); i < WORKQ_NUM_QOS_BUCKETS; i++) {
+ for (int i = _wq_bucket(qos); i < WORKQ_NUM_QOS_BUCKETS; i++) {
cnt += wq->wq_thscheduled_count[i];
}
if (conc <= cnt) {
/* We fulfilled requests and have more than NCPU scheduled threads */
WQ_TRACE_WQ(TRACE_wq_creator_yield, wq, 2,
- wq->wq_fulfilled, snapshot);
+ wq->wq_fulfilled, snapshot, 0);
return true;
}
@@ -4881,14 +3218,13 @@
}
/**
- * parked idle thread wakes up
+ * parked thread wakes up
*/
__attribute__((noreturn, noinline))
static void
workq_unpark_continue(void *parameter __unused, wait_result_t wr __unused)
{
- thread_t th = current_thread();
- struct uthread *uth = get_bsdthread_info(th);
+ struct uthread *uth = current_uthread();
proc_t p = current_proc();
struct workqueue *wq = proc_get_wqptr_fast(p);
@@ -4908,7 +3244,7 @@
}
if (__probable(uth->uu_workq_flags & UT_WORKQ_RUNNING)) {
- workq_unpark_select_threadreq_or_park_and_unlock(p, wq, uth, WQ_SETUP_NONE);
+ workq_select_threadreq_or_park_and_unlock(p, wq, uth);
__builtin_unreachable();
}
@@ -4932,7 +3268,7 @@
}
workq_unpark_for_death_and_unlock(p, wq, uth,
- WORKQ_UNPARK_FOR_DEATH_WAS_IDLE, WQ_SETUP_NONE);
+ WORKQ_UNPARK_FOR_DEATH_WAS_IDLE);
__builtin_unreachable();
}
@@ -4940,8 +3276,8 @@
static void
workq_setup_and_run(proc_t p, struct uthread *uth, int setup_flags)
{
- thread_t th = get_machthread(uth);
- vm_map_t vmap = get_task_map(proc_task(p));
+ thread_t th = uth->uu_thread;
+ vm_map_t vmap = get_task_map(p->task);
if (setup_flags & WQ_SETUP_CLEAR_VOUCHER) {
/*
@@ -4953,9 +3289,6 @@
* Those two places are sufficient to ensure we always reset it before
* it goes back out to user space, but be careful to not break that
* guarantee.
- *
- * Note that setting the voucher to NULL will not clear the preadoption
- * thread group on this thread
*/
__assert_only kern_return_t kr;
kr = thread_set_voucher_name(MACH_PORT_NULL);
@@ -4981,24 +3314,14 @@
* anyway.
*/
upcall_flags |= uth->uu_save.uus_workq_park_data.qos |
- WQ_FLAG_THREAD_PRIO_QOS;
+ WQ_FLAG_THREAD_PRIO_QOS;
}
if (uth->uu_workq_thport == MACH_PORT_NULL) {
- /* convert_thread_to_port_immovable() consumes a reference */
+ /* convert_thread_to_port() consumes a reference */
thread_reference(th);
- /* Convert to immovable thread port, then pin the entry */
- uth->uu_workq_thport = ipc_port_copyout_send_pinned(
- convert_thread_to_port_immovable(th),
- get_task_ipcspace(proc_task(p)));
- }
-
- /* Thread has been set up to run, arm its next workqueue quantum or disarm
- * if it is no longer supporting that */
- if (thread_supports_cooperative_workqueue(th)) {
- thread_arm_workqueue_quantum(th);
- } else {
- thread_disarm_workqueue_quantum(th);
+ ipc_port_t port = convert_thread_to_port(th);
+ uth->uu_workq_thport = ipc_port_copyout_send(port, get_task_ipcspace(p->task));
}
/*
@@ -5006,177 +3329,11 @@
* onto the stack, sets up the thread state and then returns to userspace.
*/
WQ_TRACE_WQ(TRACE_wq_runthread | DBG_FUNC_START,
- proc_get_wqptr_fast(p), 0, 0, 0);
-
- if (workq_thread_is_cooperative(uth)) {
- thread_sched_call(th, NULL);
- } else {
- thread_sched_call(th, workq_sched_callback);
- }
-
+ proc_get_wqptr_fast(p), 0, 0, 0, 0);
+ thread_sched_call(th, workq_sched_callback);
pthread_functions->workq_setup_thread(p, th, vmap, uth->uu_workq_stackaddr,
- uth->uu_workq_thport, 0, setup_flags, upcall_flags);
-
- __builtin_unreachable();
-}
-
-/**
- * A wrapper around workq_setup_and_run for permanently bound thread.
- */
-__attribute__((noreturn, noinline))
-static void
-workq_bound_thread_setup_and_run(struct uthread *uth, int setup_flags)
-{
- struct workq_threadreq_s * kqr = uth->uu_kqr_bound;
-
- uint32_t upcall_flags = (WQ_FLAG_THREAD_NEWSPI |
- WQ_FLAG_THREAD_WORKLOOP | WQ_FLAG_THREAD_KEVENT);
- if (workq_tr_is_overcommit(kqr->tr_flags)) {
- workq_thread_set_type(uth, UT_WORKQ_OVERCOMMIT);
- upcall_flags |= WQ_FLAG_THREAD_OVERCOMMIT;
- }
- uth->uu_save.uus_workq_park_data.upcall_flags = upcall_flags;
-
- /*
- * Increment thactive since we've decided this thread should go to
- * userspace. The scheduler callback is set in workq_setup_and_run.
- */
- proc_t p = current_proc();
- struct workqueue *wq = proc_get_wqptr_fast(p);
- _wq_thactive_inc(wq, uth->uu_workq_pri.qos_bucket);
-
- workq_setup_and_run(p, uth, setup_flags);
- __builtin_unreachable();
-}
-
-/**
- * A parked bound thread wakes up for the first time.
- */
-__attribute__((noreturn, noinline))
-static void
-workq_bound_thread_initialize_and_unpark_continue(void *parameter __unused,
- wait_result_t wr)
-{
- /*
- * Locking model for accessing uu_workq_flags :
- *
- * The concurrent access to uu_workq_flags is synchronized with workq lock
- * until a thread gets permanently bound to a kqwl. Post that, kqlock
- * is used for subsequent synchronizations. This gives us a significant
- * benefit by avoiding having to take a process wide workq lock on every
- * wakeup of the bound thread.
- * This flip in locking model is tracked with UT_WORKQ_PERMANENT_BIND flag.
- *
- * There is one more optimization we can perform for when the thread is
- * awakened for running (i.e THREAD_AWAKENED) until it parks.
- * During this window, we know KQ_SLEEP bit is reset so there should not
- * be any concurrent attempts to modify uu_workq_flags by
- * kqworkloop_bound_thread_wakeup because the thread is already "awake".
- * So we can safely access uu_workq_flags within this window without having
- * to take kqlock. This KQ_SLEEP is later set by the bound thread under
- * kqlock on its way to parking.
- */
- struct uthread *uth = get_bsdthread_info(current_thread());
-
- if (__probable(wr == THREAD_AWAKENED)) {
- /* At most one flag. */
- assert((uth->uu_workq_flags & (UT_WORKQ_RUNNING | UT_WORKQ_DYING))
- != (UT_WORKQ_RUNNING | UT_WORKQ_DYING));
-
- assert(workq_thread_is_permanently_bound(uth));
-
- if (uth->uu_workq_flags & UT_WORKQ_RUNNING) {
- assert(uth->uu_workq_flags & UT_WORKQ_NEW);
- uth->uu_workq_flags &= ~UT_WORKQ_NEW;
-
- struct workq_threadreq_s * kqr = uth->uu_kqr_bound;
- if (kqr->tr_work_interval) {
- kern_return_t kr;
- kr = kern_work_interval_explicit_join(get_machthread(uth),
- kqr->tr_work_interval);
- /*
- * The work interval functions requires to be called on the
- * current thread. If we fail here, we record the fact and
- * continue.
- * In the future, we can preflight checking that this join will
- * always be successful when the paird kqwl is configured; but,
- * for now, this should be a rare case (e.g. if you have passed
- * invalid arguments to the join).
- */
- if (kr == KERN_SUCCESS) {
- uth->uu_workq_flags |= UT_WORKQ_WORK_INTERVAL_JOINED;
- /* Thread and kqwl both have +1 ref on the work interval. */
- } else {
- uth->uu_workq_flags |= UT_WORKQ_WORK_INTERVAL_FAILED;
- }
- }
- workq_thread_reset_cpupercent(kqr, uth);
- workq_bound_thread_setup_and_run(uth, WQ_SETUP_FIRST_USE);
- __builtin_unreachable();
- } else {
- /*
- * The permanently bound kqworkloop is getting destroyed so we
- * are woken up to cleanly unbind ourselves from it and terminate.
- * See KQ_WORKLOOP_DESTROY -> workq_kern_bound_thread_wakeup.
- *
- * The actual full unbind happens from
- * uthread_cleanup -> kqueue_threadreq_unbind.
- */
- assert(uth->uu_workq_flags & UT_WORKQ_DYING);
- }
- } else {
- /*
- * The process is getting terminated so we are woken up to die.
- * E.g. SIGKILL'd.
- */
- assert(wr == THREAD_INTERRUPTED);
- /*
- * It is possible we started running as the process is aborted
- * due to termination; but, workq_kern_threadreq_permanent_bind
- * has not had a chance to bind us to the kqwl yet.
- *
- * We synchronize with it using workq lock.
- */
- proc_t p = current_proc();
- struct workqueue *wq = proc_get_wqptr_fast(p);
- workq_lock_spin(wq);
- assert(workq_thread_is_permanently_bound(uth));
- workq_unlock(wq);
-
- /*
- * We do the bind commit ourselves if workq_kern_threadreq_permanent_bind
- * has not done it for us yet so our state is aligned with what the
- * termination path below expects.
- */
- kqueue_threadreq_bind_commit(p, get_machthread(uth));
- }
- workq_kern_bound_thread_terminate(uth->uu_kqr_bound);
- __builtin_unreachable();
-}
-
-/**
- * A parked bound thread wakes up. Not the first time.
- */
-__attribute__((noreturn, noinline))
-static void
-workq_bound_thread_unpark_continue(void *parameter __unused, wait_result_t wr)
-{
- struct uthread *uth = get_bsdthread_info(current_thread());
- assert(workq_thread_is_permanently_bound(uth));
-
- if (__probable(wr == THREAD_AWAKENED)) {
- /* At most one flag. */
- assert((uth->uu_workq_flags & (UT_WORKQ_RUNNING | UT_WORKQ_DYING))
- != (UT_WORKQ_RUNNING | UT_WORKQ_DYING));
- if (uth->uu_workq_flags & UT_WORKQ_RUNNING) {
- workq_bound_thread_setup_and_run(uth, WQ_SETUP_NONE);
- } else {
- assert(uth->uu_workq_flags & UT_WORKQ_DYING);
- }
- } else {
- assert(wr == THREAD_INTERRUPTED);
- }
- workq_kern_bound_thread_terminate(uth->uu_kqr_bound);
+ uth->uu_workq_thport, 0, setup_flags, upcall_flags);
+
__builtin_unreachable();
}
@@ -5187,7 +3344,7 @@
{
struct workqueue *wq = proc_get_wqptr(p);
int error = 0;
- int activecount;
+ int activecount;
if (wq == NULL) {
return EINVAL;
@@ -5207,7 +3364,7 @@
wq_thactive_t act = _wq_thactive(wq);
activecount = _wq_thactive_aggregate_downto_qos(wq, act,
- WORKQ_THREAD_QOS_MIN, NULL, NULL);
+ WORKQ_THREAD_QOS_MIN, NULL, NULL);
if (act & _wq_thactive_offset_for_qos(WORKQ_THREAD_QOS_MANAGER)) {
activecount++;
}
@@ -5224,24 +3381,13 @@
pwqinfo->pwq_state |= WQ_EXCEEDED_TOTAL_THREAD_LIMIT;
}
- uint64_t total_cooperative_threads;
- total_cooperative_threads = workq_num_cooperative_threads_scheduled_total(wq);
- if ((total_cooperative_threads == wq_cooperative_queue_max_size(wq)) &&
- workq_has_cooperative_thread_requests(wq)) {
- pwqinfo->pwq_state |= WQ_EXCEEDED_COOPERATIVE_THREAD_LIMIT;
- }
-
- if (wq->wq_exceeded_active_constrained_thread_limit) {
- pwqinfo->pwq_state |= WQ_EXCEEDED_ACTIVE_CONSTRAINED_THREAD_LIMIT;
- }
-
workq_unlock(wq);
return error;
}
boolean_t
workqueue_get_pwq_exceeded(void *v, boolean_t *exceeded_total,
- boolean_t *exceeded_constrained)
+ boolean_t *exceeded_constrained)
{
proc_t p = v;
struct proc_workqueueinfo pwqinfo;
@@ -5265,22 +3411,16 @@
return TRUE;
}
-uint64_t
-workqueue_get_task_ss_flags_from_pwq_state_kdp(void * v)
+uint32_t
+workqueue_get_pwq_state_kdp(void * v)
{
static_assert((WQ_EXCEEDED_CONSTRAINED_THREAD_LIMIT << 17) ==
- kTaskWqExceededConstrainedThreadLimit);
+ kTaskWqExceededConstrainedThreadLimit);
static_assert((WQ_EXCEEDED_TOTAL_THREAD_LIMIT << 17) ==
- kTaskWqExceededTotalThreadLimit);
+ kTaskWqExceededTotalThreadLimit);
static_assert((WQ_FLAGS_AVAILABLE << 17) == kTaskWqFlagsAvailable);
- static_assert(((uint64_t)WQ_EXCEEDED_COOPERATIVE_THREAD_LIMIT << 34) ==
- (uint64_t)kTaskWqExceededCooperativeThreadLimit);
- static_assert(((uint64_t)WQ_EXCEEDED_ACTIVE_CONSTRAINED_THREAD_LIMIT << 34) ==
- (uint64_t)kTaskWqExceededActiveConstrainedThreadLimit);
static_assert((WQ_FLAGS_AVAILABLE | WQ_EXCEEDED_TOTAL_THREAD_LIMIT |
- WQ_EXCEEDED_CONSTRAINED_THREAD_LIMIT |
- WQ_EXCEEDED_COOPERATIVE_THREAD_LIMIT |
- WQ_EXCEEDED_ACTIVE_CONSTRAINED_THREAD_LIMIT) == 0x1F);
+ WQ_EXCEEDED_CONSTRAINED_THREAD_LIMIT) == 0x7);
if (v == NULL) {
return 0;
@@ -5289,45 +3429,39 @@
proc_t p = v;
struct workqueue *wq = proc_get_wqptr(p);
- if (wq == NULL || workq_lock_is_acquired_kdp(wq)) {
+ if (wq == NULL || workq_lock_spin_is_acquired_kdp(wq)) {
return 0;
}
- uint64_t ss_flags = kTaskWqFlagsAvailable;
+ uint32_t pwq_state = WQ_FLAGS_AVAILABLE;
if (wq->wq_constrained_threads_scheduled >= wq_max_constrained_threads) {
- ss_flags |= kTaskWqExceededConstrainedThreadLimit;
+ pwq_state |= WQ_EXCEEDED_CONSTRAINED_THREAD_LIMIT;
}
if (wq->wq_nthreads >= wq_max_threads) {
- ss_flags |= kTaskWqExceededTotalThreadLimit;
- }
-
- uint64_t total_cooperative_threads;
- total_cooperative_threads = workq_num_cooperative_threads_scheduled_to_qos_internal(wq,
- WORKQ_THREAD_QOS_MIN);
- if ((total_cooperative_threads == wq_cooperative_queue_max_size(wq)) &&
- workq_has_cooperative_thread_requests(wq)) {
- ss_flags |= kTaskWqExceededCooperativeThreadLimit;
- }
-
- if (wq->wq_exceeded_active_constrained_thread_limit) {
- ss_flags |= kTaskWqExceededActiveConstrainedThreadLimit;
- }
-
- return ss_flags;
+ pwq_state |= WQ_EXCEEDED_TOTAL_THREAD_LIMIT;
+ }
+
+ return pwq_state;
}
void
workq_init(void)
{
+ workq_lck_grp_attr = lck_grp_attr_alloc_init();
+ workq_lck_attr = lck_attr_alloc_init();
+ workq_lck_grp = lck_grp_alloc_init("workq", workq_lck_grp_attr);
+
+ workq_zone_workqueue = zinit(sizeof(struct workqueue),
+ 1024 * sizeof(struct workqueue), 8192, "workq.wq");
+ workq_zone_threadreq = zinit(sizeof(struct workq_threadreq_s),
+ 1024 * sizeof(struct workq_threadreq_s), 8192, "workq.threadreq");
+
clock_interval_to_absolutetime_interval(wq_stalled_window.usecs,
- NSEC_PER_USEC, &wq_stalled_window.abstime);
+ NSEC_PER_USEC, &wq_stalled_window.abstime);
clock_interval_to_absolutetime_interval(wq_reduce_pool_window.usecs,
- NSEC_PER_USEC, &wq_reduce_pool_window.abstime);
+ NSEC_PER_USEC, &wq_reduce_pool_window.abstime);
clock_interval_to_absolutetime_interval(wq_max_timer_interval.usecs,
- NSEC_PER_USEC, &wq_max_timer_interval.abstime);
-
- thread_deallocate_daemon_register_queue(&workq_deallocate_queue,
- workq_deallocate_queue_invoke);
-}
+ NSEC_PER_USEC, &wq_max_timer_interval.abstime);
+}