[event] Workqueue monitoring for OpenBSD.

Simply speaking, the global concurrent queue has a thread per cpu and
schedules functions or blocks on one of these threads. If these threads
are all busy, without workqueue monitoring, we cannot schedule a new
function or block until one of those threads returns. If thread(s) have
not returned however because they are asleep (say), then we could
overcommit the global concurrent queue by scheduling more work.

The Linux implementation checks each registered thread ID to see if the
corresponding thread is in a runnable state according to procfs every
second.

On some platforms without procfs, this same information is obtainable
essentially via the relevant sysctls. Libraries may exist to facilitate
issuing the correct sysctls (c.f. `kvm_getprocs`) but these were not
used here so to avoid complications with the right invocations when
linking.

A unit test is also added to test the overcommit functionality. In this
test, we tie up all cpus to prevent the scheduled threads from returning
and then attempt to run one more task. If the overcommit monitoring is
working as intended, then Dispatch will detect all the threads are
simply asleep and will attempt to overcommit. If not, the test will
notice the function we attempted to overcommit did not run and fail the
test.

Author: 3405691582

src/event/workqueue.c

@@ -242,6 +242,52 @@ _dispatch_workq_count_runnable_workers(dispatch_workq_monitor_t mon)
 
 	_dispatch_unfair_lock_unlock(&mon->registered_tid_lock);
 }
+#elif defined(__OpenBSD__)
+#include <sys/param.h>
+#include <sys/sysctl.h>
+#include <sys/proc.h>
+
+static void
+_dispatch_workq_count_runnable_workers(dispatch_workq_monitor_t mon)
+{
+	struct kinfo_proc kp[WORKQ_MAX_TRACKED_TIDS] = {0};
+	size_t size, len;
+	int mib[] = {CTL_KERN, KERN_PROC, KERN_PROC_PID, (int)getpid(), (int)sizeof(struct kinfo_proc), 0};
+	if (sysctl(mib, 6, NULL, &size, NULL, 0) < 0) {
+		_dispatch_debug("workq: Failed to sysctl1");
+		return;
+	}
+
+	size = size > sizeof(kp)? sizeof(kp): size;
+	len = size / sizeof(struct kinfo_proc);
+	mib[5] = (int)len;
+	if (sysctl(mib, 6, kp, &size, NULL, 0) < 0) {
+		_dispatch_debug("workq: Failed to sysctl2");
+		return;
+	}
+
+	int running_count = 0;
+
+	_dispatch_unfair_lock_lock(&mon->registered_tid_lock);
+
+	for (int i = 0; i < mon->num_registered_tids; i++) {
+		dispatch_tid tid = mon->registered_tids[i];
+		for (size_t j = 0; j < len; j++) {
+			if ((dispatch_tid)kp[j].p_tid != tid) {
+				continue;
+			}
+
+			if (kp[j].p_stat == SRUN || kp[j].p_stat == SIDL || kp[j].p_stat == SONPROC) {
+				running_count++;
+				break;
+			}
+		}
+	}
+
+	mon->num_runnable = running_count;
+
+	_dispatch_unfair_lock_unlock(&mon->registered_tid_lock);
+}
 #else
 #error must define _dispatch_workq_count_runnable_workers
 #endif

src/event/workqueue_internal.h

@@ -30,7 +30,7 @@
 void _dispatch_workq_worker_register(dispatch_queue_global_t root_q);
 void _dispatch_workq_worker_unregister(dispatch_queue_global_t root_q);
 
-#if defined(__linux__) || defined(_WIN32)
+#if defined(__linux__) || defined(_WIN32) || defined(__OpenBSD__)
 #define HAVE_DISPATCH_WORKQ_MONITORING 1
 #else
 #define HAVE_DISPATCH_WORKQ_MONITORING 0

tests/CMakeLists.txt

@@ -183,6 +183,12 @@ set_tests_properties(dispatch_io_pipe_close PROPERTIES TIMEOUT 5)
 add_unit_test(dispatch_c99 SOURCES dispatch_c99.c)
 add_unit_test(dispatch_plusplus SOURCES dispatch_plusplus.cpp)
 
+if (DISPATCH_USE_INTERNAL_WORKQUEUE)
+  add_unit_test(dispatch_workqueue
+                SOURCES
+                  dispatch_workqueue.c)
+endif()
+
 # test-specific link options
 if(WIN32)
   target_link_libraries(dispatch_io_muxed PRIVATE WS2_32)

tests/dispatch_workqueue.c

@@ -0,0 +1,72 @@
+#include <bsdtests.h>
+#include "dispatch_test.h"
+
+struct test_context {
+	uint32_t ncpu;
+	int flag;
+};
+
+static void
+timeout(void *context)
+{
+	struct test_context *ctx = (struct test_context *)context;
+	sleep(2);  // Give the monitor the best chance of firing.
+	test_int32_format(ctx->flag, 1, "flag");
+	test_stop();
+}
+
+static void
+raise_flag(void *context)
+{
+	struct test_context *ctx = (struct test_context *)context;
+	ctx->flag++;
+}
+
+static void
+spin(void *context)
+{
+	struct test_context *ctx = (struct test_context *)context;
+	sleep(ctx->ncpu * 2);
+}
+
+static uint32_t
+activecpu(void)
+{
+        uint32_t activecpu;
+#if defined(__linux__) || defined(__OpenBSD__)
+        activecpu = (uint32_t)sysconf(_SC_NPROCESSORS_ONLN);
+#elif defined(_WIN32)
+        SYSTEM_INFO si;
+        GetSystemInfo(&si);
+        activecpu = si.dwNumberOfProcessors;
+#else
+        size_t s = sizeof(activecpu);
+        sysctlbyname("hw.activecpu", &activecpu, &s, NULL, 0);
+#endif
+	return activecpu;
+}
+
+int
+main(void)
+{
+	uint32_t ncpu = activecpu();
+
+	dispatch_test_start("Dispatch workqueue");
+
+	dispatch_queue_t global = dispatch_get_global_queue(0, 0);
+	test_ptr_notnull("dispatch_get_global_queue", global);
+
+	struct test_context ctx = {ncpu, 0};
+	dispatch_async_f(global, &ctx, timeout);
+
+	for(int i = 0; i < (int)ncpu - 1; i++) {
+		dispatch_async_f(global, &ctx, spin);
+	}
+
+	// All cpus are tied up at this point. Workqueue
+	// should execute this function by overcommit.
+	dispatch_async_f(global, &ctx, raise_flag);
+
+	dispatch_main();
+	return 0;
+}