implement pthread_workqueue within libdispatch

Provide an implementation of the pthread_workqueue
functionality (libpwq) as a fully integrated component of
libdispatch.  Integration of the workqueue implementation
into the libdispatch code base simplifies the process of
evolving the APIs between the two layers and thus
prepares for future optimization and enhancements.

The overall design is to augment libdispatch's existing
pthread_root_queue option with periodic user-space monitoring
of the status of a root queue's worker threads and to
oversubscribe the workqueue when not enough workers are
observed to be runnable when the queue has available work.

Initially, the integrated pthread_workqueue is only
enabled by default on Linux. The current monitoring
implementation relies on Linux-specific code to dynamically
estimate the number of runnable worker threads by reading /proc.
Porting the monitoring code to non-Linux platforms would entail
providing similar functionality for those platforms.

Author: dgrove-oss

configure.ac

@@ -320,21 +320,41 @@ AS_IF([test -n "$apple_libpthread_source_path" -a -n "$apple_xnu_source_osfmk_pa
 AC_CHECK_HEADERS([pthread_machdep.h pthread/qos.h])
 
 # pthread_workqueues.
-# Look for own version first, then system version.
-AS_IF([test -f $srcdir/libpwq/configure.ac],
-  [AC_DEFINE(BUILD_OWN_PTHREAD_WORKQUEUES, 1, [Define if building pthread work queues from source])
-   ac_configure_args="--disable-libpwq-install $ac_configure_args"
-   AC_CONFIG_SUBDIRS([libpwq])
-   build_own_pthread_workqueues=true
+# Look for own version first, then for libpwq in our source tree, finally system version.
+AC_ARG_ENABLE([internal-libpwq],
+  [AS_HELP_STRING([--enable-internal-libpwq],
+    [Use libdispatch's own implementation of pthread workqueues.])],,
+  [case $target_os in
+      linux*)
+        enable_internal_libpwq=yes
+	    ;;
+      *)
+        enable_internal_libpwq=no
+   esac]
+)
+AS_IF([test "x$enable_internal_libpwq" = "xyes"],
+  [AC_DEFINE(DISPATCH_USE_INTERNAL_WORKQUEUE, 1, [Use libdispatch's own implementation of pthread_workqueue API])
    AC_DEFINE(HAVE_PTHREAD_WORKQUEUES, 1, [Define if pthread work queues are present])
-   have_pthread_workqueues=true],
-  [build_own_pthread_workqueues=false
-   AC_CHECK_HEADERS([pthread/workqueue_private.h pthread_workqueue.h],
-     [AC_DEFINE(HAVE_PTHREAD_WORKQUEUES, 1, [Define if pthread work queues are present])
+   dispatch_use_internal_workqueue=true
+   have_pthread_workqueues=true
+   build_own_pthread_workqueues=false],
+  [dispatch_use_internal_workqueue=false
+   AS_IF([test -f $srcdir/libpwq/configure.ac],
+     [AC_DEFINE(BUILD_OWN_PTHREAD_WORKQUEUES, 1, [Define if building pthread work queues from source])
+      ac_configure_args="--disable-libpwq-install $ac_configure_args"
+      AC_CONFIG_SUBDIRS([libpwq])
+      build_own_pthread_workqueues=true
+      AC_DEFINE(HAVE_PTHREAD_WORKQUEUES, 1, [Define if pthread work queues are present])
       have_pthread_workqueues=true],
-     [have_pthread_workqueues=false]
-   )]
+     [build_own_pthread_workqueues=false
+      AC_CHECK_HEADERS([pthread/workqueue_private.h pthread_workqueue.h],
+        [AC_DEFINE(HAVE_PTHREAD_WORKQUEUES, 1, [Define if pthread work queues are present])
+         have_pthread_workqueues=true],
+        [have_pthread_workqueues=false]
+      )]
+  )]
 )
+AM_CONDITIONAL(DISPATCH_USE_INTERNAL_WORKQUEUE, $dispatch_use_internal_workqueue)
 AM_CONDITIONAL(BUILD_OWN_PTHREAD_WORKQUEUES, $build_own_pthread_workqueues)
 AM_CONDITIONAL(HAVE_PTHREAD_WORKQUEUES, $have_pthread_workqueues)
 

src/Makefile.am

@@ -9,6 +9,12 @@ else
 lib_LTLIBRARIES=libdispatch.la
 endif
 
+if DISPATCH_USE_INTERNAL_WORKQUEUE
+INTERNAL_WORKQUEUE_SOURCES= 	\
+	event/workqueue.c		\
+	event/workqueue_internal.h
+endif
+
 libdispatch_la_SOURCES=			\
 	allocator.c			\
 	apply.c				\
@@ -60,7 +66,8 @@ libdispatch_la_SOURCES=			\
 	shims/perfmon.h			\
 	shims/time.h			\
 	shims/tsd.h			\
-	shims/yield.h
+	shims/yield.h			\
+	$(INTERNAL_WORKQUEUE_SOURCES)
 
 EXTRA_libdispatch_la_SOURCES=
 EXTRA_libdispatch_la_DEPENDENCIES=
@@ -77,12 +84,18 @@ AM_OBJCFLAGS=$(DISPATCH_CFLAGS) $(CBLOCKS_FLAGS)
 AM_CXXFLAGS=$(PTHREAD_WORKQUEUE_CFLAGS) $(DISPATCH_CFLAGS) $(CXXBLOCKS_FLAGS)
 AM_OBJCXXFLAGS=$(DISPATCH_CFLAGS) $(CXXBLOCKS_FLAGS)
 
+if DISPATCH_USE_INTERNAL_WORKQUEUE
+  PTHREAD_WORKQUEUE_LIBS=
+  PTHREAD_WORKQUEUE_CFLAGS=
+else
 if BUILD_OWN_PTHREAD_WORKQUEUES
   PTHREAD_WORKQUEUE_LIBS=$(top_builddir)/libpwq/libpthread_workqueue.la
   PTHREAD_WORKQUEUE_CFLAGS=-I$(top_srcdir)/libpwq/include
 else
 if HAVE_PTHREAD_WORKQUEUES
   PTHREAD_WORKQUEUE_LIBS=-lpthread_workqueue
+  PTHREAD_WORKQUEUE_CFLAGS=
+endif
 endif
 endif
 

src/apply.c

@@ -159,11 +159,11 @@ static inline void
 _dispatch_apply_f2(dispatch_queue_t dq, dispatch_apply_t da,
 		dispatch_function_t func)
 {
-	uint32_t i = 0;
+	int32_t i = 0;
 	dispatch_continuation_t head = NULL, tail = NULL;
 
 	// The current thread does not need a continuation
-	uint32_t continuation_cnt = da->da_thr_cnt - 1;
+	int32_t continuation_cnt = da->da_thr_cnt - 1;
 
 	dispatch_assert(continuation_cnt);
 
@@ -192,14 +192,14 @@ static void
 _dispatch_apply_redirect(void *ctxt)
 {
 	dispatch_apply_t da = (dispatch_apply_t)ctxt;
-	uint32_t da_width = da->da_thr_cnt - 1;
+	int32_t da_width = da->da_thr_cnt - 1;
 	dispatch_queue_t dq = da->da_dc->dc_data, rq = dq, tq;
 
 	do {
-		uint32_t width = _dispatch_queue_try_reserve_apply_width(rq, da_width);
+		int32_t width = _dispatch_queue_try_reserve_apply_width(rq, da_width);
 
 		if (slowpath(da_width > width)) {
-			uint32_t excess = da_width - width;
+			int32_t excess = da_width - width;
 			for (tq = dq; tq != rq; tq = tq->do_targetq) {
 				_dispatch_queue_relinquish_width(tq, excess);
 			}
@@ -234,7 +234,7 @@ dispatch_apply_f(size_t iterations, dispatch_queue_t dq, void *ctxt,
 	if (slowpath(iterations == 0)) {
 		return;
 	}
-	uint32_t thr_cnt = dispatch_hw_config(active_cpus);
+	int32_t thr_cnt = dispatch_hw_config(active_cpus);
 	dispatch_thread_context_t dtctxt = _dispatch_thread_context_find(_dispatch_apply_key);
 	size_t nested = dtctxt ? dtctxt->dtc_apply_nesting : 0;
 	dispatch_queue_t old_dq = _dispatch_queue_get_current();
@@ -247,7 +247,7 @@ dispatch_apply_f(size_t iterations, dispatch_queue_t dq, void *ctxt,
 				? nested * iterations : DISPATCH_APPLY_MAX;
 	}
 	if (iterations < thr_cnt) {
-		thr_cnt = (uint32_t)iterations;
+		thr_cnt = iterations;
 	}
 	if (slowpath(dq == DISPATCH_APPLY_CURRENT_ROOT_QUEUE)) {
 		dq = old_dq ? old_dq : _dispatch_get_root_queue(