tidy up to prepare patch for review

Author: dgrove-oss

src/event/workqueue.c

@@ -41,40 +41,40 @@
 
 #pragma mark static data for management subsystem
 
-/*
- * The number of runnable worker threads as observed by
- * the scheduling management subsystem.
- */
-static volatile int32_t _dispatch_workq_runnable_workers;
-
-/*
- * The desired number of runnable worker threads
- * for a workqueue (assuming sufficient work).
- */
-static int32_t _dispatch_workq_target_runnable_workers;
-
 #if DISPATCH_ENABLE_PWQ_KEXT
 /* Are we using user-level or kext based management? */
 static bool _dispatch_workq_kext_active;
 #endif
 
 /*
- * State for the user-level pool management.
+ * State for the user-level monitoring of a workqueue.
  */
 typedef struct dispatch_workq_manager_s {
+	/* The observed number of runnable registered workers */
+	int32_t runnable_workers;
+
+	/* The desired number of runnable registered workers */
+	int32_t target_runnable_workers;
+
 	/*
-	 * Tracking of registered workers; all updates and reads
-	 * are performed while holding the lock.
-	 */
+     * Tracking of registered workers; all accesses must hold lock.
+     * Invariant: registered_workers[0]...registered_workers[num_registered_workers-1]
+     *   contain the pids of the workers that we are managing.
+     */
 	dispatch_unfair_lock_s registered_worker_lock;
-	int num_registered_workers;
 	pid_t *registered_workers;
-
+	int num_registered_workers;
 } dispatch_workq_manager_s, *dispatch_workq_manager_t;
 
 static dispatch_workq_manager_s _dispatch_workq_manager;
 
-#pragma mark backdoors into queue.c internals
+#pragma mark inline helper functions
+
+DISPATCH_INLINE
+dispatch_workq_manager_t
+_dispatch_workq_get_default_manager() {
+	return &_dispatch_workq_manager;
+}
 
 DISPATCH_INLINE
 dispatch_queue_t
@@ -114,17 +114,18 @@ dispatch_workq_worker_register(dispatch_queue_t root_q)
 		return true;
 	}
 #endif
+	dispatch_workq_manager_t mgr = _dispatch_workq_get_default_manager();
 	bool rc;
 	int tid = syscall(SYS_gettid);
-	_dispatch_unfair_lock_lock(&_dispatch_workq_manager.registered_worker_lock);
-	if (_dispatch_workq_manager.num_registered_workers < WORKQ_MAX_TRACKED_WORKERS-1) {
-		int worker_id = _dispatch_workq_manager.num_registered_workers++;
-		_dispatch_workq_manager.registered_workers[worker_id] = tid;
+	_dispatch_unfair_lock_lock(&mgr->registered_worker_lock);
+	if (mgr->num_registered_workers < WORKQ_MAX_TRACKED_WORKERS-1) {
+		int worker_id = mgr->num_registered_workers++;
+		mgr->registered_workers[worker_id] = tid;
 		rc = true;
 	} else {
 		rc = false;
 	}
-	_dispatch_unfair_lock_unlock(&_dispatch_workq_manager.registered_worker_lock);
+	_dispatch_unfair_lock_unlock(&mgr->registered_worker_lock);
 
 	return rc;
 }
@@ -143,18 +144,19 @@ dispatch_workq_worker_unregister(dispatch_queue_t root_q)
 		return;
 	}
 #endif
+	dispatch_workq_manager_t mgr = _dispatch_workq_get_default_manager();
 	int tid = syscall(SYS_gettid);
-	_dispatch_unfair_lock_lock(&_dispatch_workq_manager.registered_worker_lock);
-	for (int i=0; i<_dispatch_workq_manager.num_registered_workers; i++) {
-		if (_dispatch_workq_manager.registered_workers[i] == tid) {
-			int last = _dispatch_workq_manager.num_registered_workers - 1;
-			_dispatch_workq_manager.registered_workers[i] = _dispatch_workq_manager.registered_workers[last];
-			_dispatch_workq_manager.registered_workers[last] = 0;
-			_dispatch_workq_manager.num_registered_workers--;
+	_dispatch_unfair_lock_lock(&mgr->registered_worker_lock);
+	for (int i=0; i<mgr->num_registered_workers; i++) {
+		if (mgr->registered_workers[i] == tid) {
+			int last = mgr->num_registered_workers - 1;
+			mgr->registered_workers[i] = mgr->registered_workers[last];
+			mgr->registered_workers[last] = 0;
+			mgr->num_registered_workers--;
 			break;
 		}
 	}
-	_dispatch_unfair_lock_unlock(&_dispatch_workq_manager.registered_worker_lock);
+	_dispatch_unfair_lock_unlock(&mgr->registered_worker_lock);
 }
 
 
@@ -163,19 +165,19 @@ dispatch_workq_worker_unregister(dispatch_queue_t root_q)
  * to get a count of the number of them that are actually runnable.
  * See the proc(5) man page for the format of the contents of /proc/[pid]/stat
  */
-static int
-_dispatch_workq_count_runnable_workers(void)
+static void
+_dispatch_workq_count_runnable_workers(dispatch_workq_manager_t mgr)
 {
 	char path[128];
 	char buf[4096];
 	int running_count = 0;
 
 	memset(buf, 0, sizeof(buf));
 
-	_dispatch_unfair_lock_lock(&_dispatch_workq_manager.registered_worker_lock);
+	_dispatch_unfair_lock_lock(&mgr->registered_worker_lock);
 
-	for (int i=0; i<_dispatch_workq_manager.num_registered_workers; i++) {
-		pid_t worker_pid = _dispatch_workq_manager.registered_workers[i];
+	for (int i=0; i<mgr->num_registered_workers; i++) {
+		pid_t worker_pid = mgr->registered_workers[i];
 		int fd;
 		size_t bytes_read = -1;
 
@@ -188,10 +190,10 @@ _dispatch_workq_count_runnable_workers(void)
 			// Must mean worker exited uncleanly (without executing _dispatch_worker_unregister())
 			// Clean up by removing pid and decrementing number of registered workers
 			_dispatch_debug("workq: Unable to open /proc/%d/stat; removing worker from monitoring list", worker_pid);
-			int last = _dispatch_workq_manager.num_registered_workers-1;
-			_dispatch_workq_manager.registered_workers[i] = _dispatch_workq_manager.registered_workers[last];
-			_dispatch_workq_manager.registered_workers[last] = 0;
-			_dispatch_workq_manager.num_registered_workers--;
+			int last = mgr->num_registered_workers-1;
+			mgr->registered_workers[i] = mgr->registered_workers[last];
+			mgr->registered_workers[last] = 0;
+			mgr->num_registered_workers--;
 		} else {
 			bytes_read = read(fd, buf, sizeof(buf));
 			(void)close(fd);
@@ -213,31 +215,33 @@ _dispatch_workq_count_runnable_workers(void)
 		}
 	}
 
-	_dispatch_unfair_lock_unlock(&_dispatch_workq_manager.registered_worker_lock);
+	mgr->runnable_workers = running_count;
 
-	return running_count;
+	_dispatch_unfair_lock_unlock(&mgr->registered_worker_lock);
 }
 
 static void
 _dispatch_workq_monitor_thread_pool(void *context DISPATCH_UNUSED)
 {
+	dispatch_workq_manager_t mgr = _dispatch_workq_get_default_manager();
 	dispatch_queue_t dq = _dispatch_workq_get_default_root_queue();
 	bool work_available = _dispatch_workq_root_queue_has_work(dq);
 	if (work_available) {
-		_dispatch_workq_runnable_workers = _dispatch_workq_count_runnable_workers();
+		_dispatch_workq_count_runnable_workers(mgr);
 
 		_dispatch_debug("workq: %s is non-empty and has %d runnable workers\n",
-				dq->dq_label, _dispatch_workq_runnable_workers);
+				dq->dq_label, mgr->runnable_workers);
 
-		if (_dispatch_workq_runnable_workers < _dispatch_workq_target_runnable_workers) {
-			int32_t count = _dispatch_pthread_root_queue_size(dq);
-			int32_t allowed_over = WORKQ_OVERSUBSCRIBE_FACTOR * _dispatch_workq_target_runnable_workers;
+		if (mgr->runnable_workers < mgr->target_runnable_workers) {
+			int32_t count = _dispatch_pthread_root_queue_thread_pool_size(dq);
+			int32_t allowed_over = WORKQ_OVERSUBSCRIBE_FACTOR * mgr->target_runnable_workers;
+			allowed_over = MIN(allowed_over, WORKQ_MAX_TRACKED_WORKERS - mgr->target_runnable_workers);
 			if (count + allowed_over > 0) {
 				_dispatch_debug("workq: %s has count %d; requesting 1 additional worker",
 						dq->dq_label, count);
 				_dispatch_pthread_root_queue_oversubscribe(dq, 1);
 			} else {
-				_dispatch_debug("workq: %s is already oversubscribed by %d; taking no action",
+				_dispatch_debug("workq: %s already oversubscribed by %d; taking no action",
 						dq->dq_label, -count);
 			}
 		}
@@ -257,10 +261,9 @@ _dispatch_workq_init_once(void *context DISPATCH_UNUSED)
 		return;
 	}
 #endif
-	_dispatch_workq_manager.registered_workers =
-			_dispatch_calloc(WORKQ_MAX_TRACKED_WORKERS, sizeof(pid_t));
-
-	_dispatch_workq_target_runnable_workers = dispatch_hw_config(active_cpus);
+	dispatch_workq_manager_t mgr = _dispatch_workq_get_default_manager();
+	mgr->registered_workers = _dispatch_calloc(WORKQ_MAX_TRACKED_WORKERS, sizeof(pid_t));
+	mgr->target_runnable_workers = dispatch_hw_config(active_cpus);
 
 	// Create monitoring timer that will periodically run on dispatch_mgr_q
 	dispatch_source_t ds = dispatch_source_create(DISPATCH_SOURCE_TYPE_TIMER, 0, 0, &_dispatch_mgr_q);

src/queue.c

@@ -2078,7 +2078,7 @@ dispatch_pthread_root_queue_copy_current(void)
 }
 
 int32_t
-_dispatch_pthread_root_queue_size(dispatch_queue_t dq) {
+_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, ordered);
 }
@@ -3958,7 +3958,7 @@ _dispatch_runloop_queue_poke(dispatch_queue_t dq, dispatch_qos_t qos,
 #if DISPATCH_USE_PTHREAD_POOL
 DISPATCH_INLINE
 void
-_dispatch_spawn_pthread_root_queue_workers(dispatch_queue_t dq,
+_dispatch_pthread_root_queue_spawn_workers(dispatch_queue_t dq,
 		pthread_attr_t *attr, int32_t n)
 {
 	pthread_t tid, *pthr = &tid;
@@ -3984,7 +3984,7 @@ _dispatch_pthread_root_queue_oversubscribe(dispatch_queue_t dq, int n)
 
 	(void)os_atomic_sub2o(qc, dgq_thread_pool_size, n, release);
 
-	_dispatch_spawn_pthread_root_queue_workers(dq, attr, n);
+	_dispatch_pthread_root_queue_spawn_workers(dq, attr, n);
 }
 #endif
 
@@ -4062,7 +4062,7 @@ _dispatch_global_queue_poke_slow(dispatch_queue_t dq, unsigned int n)
 		pthr = _dispatch_mgr_root_queue_init();
 	}
 #endif
-	_dispatch_spawn_pthread_root_queue_workers(dq, attr, j);
+	_dispatch_pthread_root_queue_spawn_workers(dq, attr, j);
 #endif // DISPATCH_USE_PTHREAD_POOL
 }
 

src/queue_internal.h

@@ -593,7 +593,7 @@ 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_size(dispatch_queue_t dq);
+int32_t _dispatch_pthread_root_queue_thread_pool_size(dispatch_queue_t dq);
 void _dispatch_pthread_root_queue_oversubscribe(dispatch_queue_t dq, int n);
 #else
 static inline void _dispatch_mgr_priority_init(void) {}