rework counting of pending thread creations

use poke with negative floor to drive oversubscription

Author: dgrove-oss

src/event/workqueue.c

@@ -88,10 +88,9 @@ _dispatch_workq_worker_register(dispatch_queue_t root_q, int priority)
 	dispatch_assert(mon->dq == root_q);
 	dispatch_tid tid = _dispatch_thread_getspecific(tid);
 	_dispatch_unfair_lock_lock(&mon->registered_tid_lock);
-	if (mon->num_registered_tids < WORKQ_MAX_TRACKED_TIDS-1) {
-		int worker_id = mon->num_registered_tids++;
-		mon->registered_tids[worker_id] = tid;
-	}
+	dispatch_assert(mon->num_registered_tids < WORKQ_MAX_TRACKED_TIDS-1);
+	int worker_id = mon->num_registered_tids++;
+	mon->registered_tids[worker_id] = tid;
 	_dispatch_unfair_lock_unlock(&mon->registered_tid_lock);
 }
 
@@ -187,36 +186,29 @@ _dispatch_workq_monitor_pools(void *context DISPATCH_UNUSED)
 		}
 
 		_dispatch_workq_count_runnable_workers(mon);
-		int32_t count = _dispatch_pthread_root_queue_thread_pool_size(dq);
-		_dispatch_debug("workq: %s is non-empty with pool_size %d (%d runnable)",
-				dq->dq_label, count, mon->num_runnable);
-
-		if (mon->num_runnable < mon->target_runnable) {
-			// If we are below target there are two cases to consider:
-			//   (a) We are below target, but some workers are still runnable.
-			//       We want to oversubscribe to hit the target, but this
-			//       may be transitory so only go up to a small multiple
-			//       of threads per core.
-			//   (b) We are below target, and no worker is runnable.
-			//       It is likely the program is stalled. Therefore treat
-			//       this as if dq was an overcommit queue and create
-			//       another worker unless we have already hit the hard
-			//       limit on the maximum number of workers for dq.
-			int32_t limit_a = WORKQ_OVERSUBSCRIBE_FACTOR * mon->target_runnable;
-			int32_t limit_b = WORKQ_MAX_TRACKED_TIDS - mon->target_runnable;
-			int32_t limit =
-					mon->num_runnable == 0 ? limit_b : MIN(limit_a, limit_b);
-			if (count + limit > 0) {
-				bool r = _dispatch_pthread_root_queue_oversubscribe(dq, 1);
-				if (r) {
-					_dispatch_debug("workq: requested overscribe worker for %s", dq->dq_label);
-				} else {
-					_dispatch_debug("workq: still pending worker create on %s", dq->dq_label);
-				}
-			} else {
-				_dispatch_debug("workq: %s already over by %d; taking no action",
-						dq->dq_label, -count);
-			}
+		_dispatch_debug("workq: %s has %d runnable wokers (target is %d)",
+				dq->dq_label, mon->num_runnable, mon->target_runnable);
+
+		if (mon->num_runnable == 0) {
+			// We are below target, and no worker is runnable.
+			// It is likely the program is stalled. Therefore treat
+			// this as if dq were an overcommit queue and call poke
+			// with the limit being the maximum number of workers for dq.
+			int32_t floor = mon->target_runnable - WORKQ_MAX_TRACKED_TIDS;
+			_dispatch_debug("workq: %s has no runnable workers; poking with floor %d",
+					dq->dq_label, floor);
+			_dispatch_global_queue_poke(dq, 1, floor);
+		} else if (mon->num_runnable < mon->target_runnable) {
+			// We are below target, but some workers are still runnable.
+			// We want to oversubscribe to hit the desired load target.
+			// However, this under-utilization may be transitory so set the
+			// floor as a small multiple of threads per core.
+			int32_t floor = (1 - WORKQ_OVERSUBSCRIBE_FACTOR) * mon->target_runnable;
+			int32_t floor2 = mon->target_runnable - WORKQ_MAX_TRACKED_TIDS;
+			floor = MAX(floor, floor2);
+			_dispatch_debug("workq: %s under utilization target; poking with floor %d",
+					dq->dq_label, floor);
+			_dispatch_global_queue_poke(dq, 1, floor);
 		}
 	}
 }

src/inline_internal.h

@@ -1479,7 +1479,7 @@ _dispatch_root_queue_push_inline(dispatch_queue_t dq, dispatch_object_t _head,
 	struct dispatch_object_s *head = _head._do, *tail = _tail._do;
 	if (unlikely(_dispatch_queue_push_update_tail_list(dq, head, tail))) {
 		_dispatch_queue_push_update_head(dq, head);
-		return _dispatch_global_queue_poke(dq, n);
+		return _dispatch_global_queue_poke(dq, n, 0);
 	}
 }
 

src/queue.c

@@ -2077,13 +2077,6 @@ dispatch_pthread_root_queue_copy_current(void)
 	return (dispatch_queue_t)_os_object_retain_with_resurrect(dq->_as_os_obj);
 }
 
-int32_t
-_dispatch_pthread_root_queue_thread_pool_size(dispatch_queue_t dq)
-{
-	dispatch_root_queue_context_t qc = dq->do_ctxt;
-	return os_atomic_load2o(qc, dgq_thread_pool_size, relaxed);
-}
-
 #endif // DISPATCH_ENABLE_PTHREAD_ROOT_QUEUES
 
 void
@@ -3956,50 +3949,12 @@ _dispatch_runloop_queue_poke(dispatch_queue_t dq, dispatch_qos_t qos,
 }
 #endif
 
-#if DISPATCH_USE_PTHREAD_POOL
-static void
-_dispatch_pthread_root_queue_spawn_workers(dispatch_queue_t dq,
-		pthread_attr_t *attr, int n)
-{
-	pthread_t tid, *pthr = &tid;
-	int r;
-
-	do {
-		_dispatch_retain(dq);
-		while ((r = pthread_create(pthr, attr, _dispatch_worker_thread, dq))) {
-			if (r != EAGAIN) {
-				(void)dispatch_assume_zero(r);
-			}
-			_dispatch_temporary_resource_shortage();
-		}
-	} while (--n);
-}
-
-bool
-_dispatch_pthread_root_queue_oversubscribe(dispatch_queue_t dq, int n)
-{
-	dispatch_root_queue_context_t qc = dq->do_ctxt;
-	dispatch_pthread_root_queue_context_t pqc = qc->dgq_ctxt;
-	pthread_attr_t *attr = &pqc->dpq_thread_attr;
-
-	if (!os_atomic_cmpxchg2o(qc, dgq_pending, 0, n, relaxed)) {
-		return false;
-	}
-
-	(void)os_atomic_sub2o(qc, dgq_thread_pool_size, n, release);
-
-	_dispatch_pthread_root_queue_spawn_workers(dq, attr, n);
-
-	return true;
-}
-#endif
-
 DISPATCH_NOINLINE
 static void
-_dispatch_global_queue_poke_slow(dispatch_queue_t dq, int n)
+_dispatch_global_queue_poke_slow(dispatch_queue_t dq, int n, int floor)
 {
 	dispatch_root_queue_context_t qc = dq->do_ctxt;
-	int32_t i = n;
+	int remaining = n;
 	int r = ENOSYS;
 
 	_dispatch_root_queues_init();
@@ -4019,16 +3974,16 @@ _dispatch_global_queue_poke_slow(dispatch_queue_t dq, int n)
 				r = pthread_workqueue_additem_np(qc->dgq_kworkqueue,
 						_dispatch_worker_thread4, dq, &wh, &gen_cnt);
 				(void)dispatch_assume_zero(r);
-			} while (--i);
+			} while (--remaining);
 			return;
 		}
 #endif // DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK
 #if HAVE_PTHREAD_WORKQUEUE_QOS
-		r = _pthread_workqueue_addthreads((int)i,
+		r = _pthread_workqueue_addthreads(remaining,
 				_dispatch_priority_to_pp(dq->dq_priority));
 #elif HAVE_PTHREAD_WORKQUEUE_SETDISPATCH_NP
 		r = pthread_workqueue_addthreads_np(qc->dgq_wq_priority,
-				qc->dgq_wq_options, (int)i);
+				qc->dgq_wq_options, remaining);
 #endif
 		(void)dispatch_assume_zero(r);
 		return;
@@ -4038,28 +3993,43 @@ _dispatch_global_queue_poke_slow(dispatch_queue_t dq, int n)
 	dispatch_pthread_root_queue_context_t pqc = qc->dgq_ctxt;
 	if (fastpath(pqc->dpq_thread_mediator.do_vtable)) {
 		while (dispatch_semaphore_signal(&pqc->dpq_thread_mediator)) {
-			if (!--i) {
+			_dispatch_root_queue_debug("signaled sleeping worker for "
+					"global queue: %p", dq);
+			if (!--remaining) {
 				return;
 			}
 		}
 	}
-	int32_t j, t_count;
+
+	bool overcommit = dq->dq_priority & DISPATCH_PRIORITY_FLAG_OVERCOMMIT;
+	if (overcommit) {
+		os_atomic_add2o(qc, dgq_pending, remaining, relaxed);
+	} else {
+		if (!os_atomic_cmpxchg2o(qc, dgq_pending, 0, remaining, relaxed)) {
+			_dispatch_root_queue_debug("worker thread request still pending for "
+					"global queue: %p", dq);
+			return;
+		}
+	}
+
+	int32_t can_request, t_count;
 	// seq_cst with atomic store to tail <rdar://problem/16932833>
 	t_count = os_atomic_load2o(qc, dgq_thread_pool_size, ordered);
 	do {
-#if HAVE_INTERNAL_PTHREAD_WORKQUEUE
-		bool pool_full = t_count <= 0; // oversubscription encoded by negative dgq_thread_pool_size
-#else
-		bool pool_full = t_count == 0;
-#endif
-		if (pool_full) {
+		can_request = t_count < floor ? 0 : t_count - floor;
+		if (remaining > can_request) {
+			_dispatch_root_queue_debug("pthread pool reducing request from %d to %d",
+					remaining, can_request);
+			os_atomic_sub2o(qc, dgq_pending, remaining - can_request, relaxed);
+			remaining = can_request;
+		}
+		if (remaining == 0) {
 			_dispatch_root_queue_debug("pthread pool is full for root queue: "
 					"%p", dq);
 			return;
 		}
-		j = i > t_count ? t_count : i;
 	} while (!os_atomic_cmpxchgvw2o(qc, dgq_thread_pool_size, t_count,
-			t_count - j, &t_count, acquire));
+			t_count - remaining, &t_count, acquire));
 
 	pthread_attr_t *attr = &pqc->dpq_thread_attr;
 	pthread_t tid, *pthr = &tid;
@@ -4068,30 +4038,41 @@ _dispatch_global_queue_poke_slow(dispatch_queue_t dq, int n)
 		pthr = _dispatch_mgr_root_queue_init();
 	}
 #endif
-	_dispatch_pthread_root_queue_spawn_workers(dq, attr, j);
+	do {
+		_dispatch_retain(dq);
+		while ((r = pthread_create(pthr, attr, _dispatch_worker_thread, dq))) {
+			if (r != EAGAIN) {
+				(void)dispatch_assume_zero(r);
+			}
+			_dispatch_temporary_resource_shortage();
+		}
+	} while (--remaining);
 #endif // DISPATCH_USE_PTHREAD_POOL
 }
 
 DISPATCH_NOINLINE
 void
-_dispatch_global_queue_poke(dispatch_queue_t dq, int n)
+_dispatch_global_queue_poke(dispatch_queue_t dq, int n, int floor)
 {
 	if (!_dispatch_queue_class_probe(dq)) {
 		return;
 	}
 #if HAVE_PTHREAD_WORKQUEUES
 	dispatch_root_queue_context_t qc = dq->do_ctxt;
 	if (
+#if HAVE_INTERNAL_PTHREAD_WORKQUEUE
+		false && /* counting done in poke_slow for this config */
+#endif
 #if DISPATCH_USE_PTHREAD_POOL
-			(qc->dgq_kworkqueue != (void*)(~0ul)) &&
+		(qc->dgq_kworkqueue != (void*)(~0ul)) &&
 #endif
-			!os_atomic_cmpxchg2o(qc, dgq_pending, 0, n, relaxed)) {
-		_dispatch_root_queue_debug("worker thread request still pending for "
+		!os_atomic_cmpxchg2o(qc, dgq_pending, 0, n, relaxed)) {
+			_dispatch_root_queue_debug("worker thread request still pending for "
 				"global queue: %p", dq);
 		return;
 	}
 #endif // HAVE_PTHREAD_WORKQUEUES
-	return _dispatch_global_queue_poke_slow(dq, n);
+	return _dispatch_global_queue_poke_slow(dq, n, floor);
 }
 
 #pragma mark -
@@ -5296,7 +5277,7 @@ _dispatch_root_queue_drain_one_slow(dispatch_queue_t dq)
 		(void)os_atomic_dec2o(qc, dgq_pending, relaxed);
 	}
 	if (!available) {
-		_dispatch_global_queue_poke(dq, 1);
+		_dispatch_global_queue_poke(dq, 1, 0);
 	}
 	return available;
 }
@@ -5363,7 +5344,7 @@ _dispatch_root_queue_drain_one(dispatch_queue_t dq)
 	}
 
 	os_atomic_store2o(dq, dq_items_head, next, relaxed);
-	_dispatch_global_queue_poke(dq, 1);
+	_dispatch_global_queue_poke(dq, 1, 0);
 out:
 	return head;
 }
@@ -5506,6 +5487,9 @@ _dispatch_worker_thread(void *context)
 	dispatch_root_queue_context_t qc = dq->do_ctxt;
 	dispatch_pthread_root_queue_context_t pqc = qc->dgq_ctxt;
 
+	int pending = (int)os_atomic_dec2o(qc, dgq_pending, relaxed);
+	dispatch_assert(pending >= 0);
+
 	if (pqc->dpq_observer_hooks.queue_will_execute) {
 		_dispatch_set_pthread_root_queue_observer_hooks(
 				&pqc->dpq_observer_hooks);
@@ -5525,7 +5509,9 @@ _dispatch_worker_thread(void *context)
 
 #if HAVE_INTERNAL_PTHREAD_WORKQUEUE
 	bool overcommit = qc->dgq_wq_options & WORKQ_ADDTHREADS_OPTION_OVERCOMMIT;
-	if (!overcommit) {
+	bool manager = dq == &_dispatch_mgr_root_queue;
+	bool monitored = !(overcommit || manager);
+	if (monitored) {
 		_dispatch_workq_worker_register(dq, qc->dgq_wq_priority);
 	}
 #endif
@@ -5539,12 +5525,12 @@ _dispatch_worker_thread(void *context)
 			dispatch_time(0, timeout)) == 0);
 
 #if HAVE_INTERNAL_PTHREAD_WORKQUEUE
-	if (!overcommit) {
+	if (monitored) {
 		_dispatch_workq_worker_unregister(dq,  qc->dgq_wq_priority);
 	}
 #endif
 	(void)os_atomic_inc2o(qc, dgq_thread_pool_size, release);
-	_dispatch_global_queue_poke(dq, 1);
+	_dispatch_global_queue_poke(dq, 1, 0);
 	_dispatch_release(dq);
 
 	return NULL;

src/queue_internal.h

@@ -563,7 +563,7 @@ void _dispatch_queue_resume(dispatch_queue_t dq, bool activate);
 void _dispatch_queue_finalize_activation(dispatch_queue_t dq);
 void _dispatch_queue_invoke(dispatch_queue_t dq,
 		dispatch_invoke_context_t dic, dispatch_invoke_flags_t flags);
-void _dispatch_global_queue_poke(dispatch_queue_t dq, int n);
+void _dispatch_global_queue_poke(dispatch_queue_t dq, int n, int floor);
 void _dispatch_queue_push(dispatch_queue_t dq, dispatch_object_t dou,
 		dispatch_qos_t qos);
 void _dispatch_try_lock_transfer_or_wakeup(dispatch_queue_t dq);
@@ -593,8 +593,6 @@ void _dispatch_runloop_queue_dispose(dispatch_queue_t dq);
 void _dispatch_mgr_queue_drain(void);
 #if DISPATCH_USE_MGR_THREAD && DISPATCH_ENABLE_PTHREAD_ROOT_QUEUES
 void _dispatch_mgr_priority_init(void);
-int32_t _dispatch_pthread_root_queue_thread_pool_size(dispatch_queue_t dq);
-bool _dispatch_pthread_root_queue_oversubscribe(dispatch_queue_t dq, int n);
 #else
 static inline void _dispatch_mgr_priority_init(void) {}
 #endif