implement pthread_workqueue within libdispatch

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

Initially, the integrated pthread_workqueue is only
enabled by default on Linux. Most of the internals are
built on pthread APIs and thus should be fairly portable.
However, Linux-specific code is used to dynamically estimate
the number of runnable worker threads by reading /proc.
Porting the thread pool management code to non-Linux
platforms would entail providing similar functionality
for those platforms (or otherwise restructuring the manager).

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_workqueue API.])],,
+  [case $target_os in
+      linux*)
+        enable_internal_libpwq=yes
+	    ;;
+      *)
+        enable_internal_libpwq=no
+   esac]
+)
+AS_IF([test "x$enable_internal_libpwq" = "xyes"],
+  [AC_DEFINE(HAVE_INTERNAL_PTHREAD_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])
+   have_internal_pthread_workqueues=true
+   have_pthread_workqueues=true
+   build_own_pthread_workqueues=false],
+  [have_internal_pthread_workqueues=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(HAVE_INTERNAL_PTHREAD_WORKQUEUES, $have_internal_pthread_workqueues)
 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 HAVE_INTERNAL_PTHREAD_WORKQUEUES
+INTERNAL_WORKQUEUE_SOURCES= 	\
+	workqueue/workqueue.c		\
+	workqueue/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,6 +84,7 @@ AM_OBJCFLAGS=$(DISPATCH_CFLAGS) $(CBLOCKS_FLAGS)
 AM_CXXFLAGS=$(PTHREAD_WORKQUEUE_CFLAGS) $(DISPATCH_CFLAGS) $(CXXBLOCKS_FLAGS)
 AM_OBJCXXFLAGS=$(DISPATCH_CFLAGS) $(CXXBLOCKS_FLAGS)
 
+if !HAVE_INTERNAL_PTHREAD_WORKQUEUES
 if BUILD_OWN_PTHREAD_WORKQUEUES
   PTHREAD_WORKQUEUE_LIBS=$(top_builddir)/libpwq/libpthread_workqueue.la
   PTHREAD_WORKQUEUE_CFLAGS=-I$(top_srcdir)/libpwq/include
@@ -85,6 +93,7 @@ if HAVE_PTHREAD_WORKQUEUES
   PTHREAD_WORKQUEUE_LIBS=-lpthread_workqueue
 endif
 endif
+endif
 
 if BUILD_OWN_BLOCKS_RUNTIME
 libdispatch_la_SOURCES+= BlocksRuntime/data.c BlocksRuntime/runtime.c

src/queue.c

@@ -32,11 +32,12 @@
 #endif
 #if HAVE_PTHREAD_WORKQUEUES && (!HAVE_PTHREAD_WORKQUEUE_QOS || DISPATCH_DEBUG) && \
 		!HAVE_PTHREAD_WORKQUEUE_SETDISPATCH_NP && \
+		!HAVE_INTERNAL_PTHREAD_WORKQUEUE && \
 		!defined(DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK)
 #define DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK 1
 #endif
 #if HAVE_PTHREAD_WORKQUEUES && DISPATCH_USE_PTHREAD_POOL && \
-		!DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK
+	!(DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK || HAVE_INTERNAL_PTHREAD_WORKQUEUE)
 #define pthread_workqueue_t void*
 #endif
 
@@ -66,7 +67,7 @@ static void _dispatch_worker_thread2(int priority, int options, void *context);
 #endif
 #if DISPATCH_USE_PTHREAD_POOL
 static void *_dispatch_worker_thread(void *context);
-static int _dispatch_pthread_sigmask(int how, sigset_t *set, sigset_t *oset);
+int _dispatch_pthread_sigmask(int how, sigset_t *set, sigset_t *oset);
 #endif
 
 #if DISPATCH_COCOA_COMPAT
@@ -152,7 +153,7 @@ struct dispatch_root_queue_context_s {
 #if HAVE_PTHREAD_WORKQUEUES
 			qos_class_t dgq_qos;
 			int dgq_wq_priority, dgq_wq_options;
-#if DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK || DISPATCH_USE_PTHREAD_POOL
+#if DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK || DISPATCH_USE_PTHREAD_POOL || HAVE_INTERNAL_PTHREAD_WORKQUEUE
 			pthread_workqueue_t dgq_kworkqueue;
 #endif
 #endif // HAVE_PTHREAD_WORKQUEUES
@@ -635,7 +636,7 @@ _dispatch_root_queues_init_workq(int *wq_supported)
 		result = !r;
 	}
 #endif // HAVE_PTHREAD_WORKQUEUE_SETDISPATCH_NP
-#if DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK || DISPATCH_USE_PTHREAD_POOL
+#if DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK || DISPATCH_USE_PTHREAD_POOL || HAVE_INTERNAL_PTHREAD_WORKQUEUE
 	if (!result) {
 #if DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK
 		pthread_workqueue_attr_t pwq_attr;
@@ -663,6 +664,15 @@ _dispatch_root_queues_init_workq(int *wq_supported)
 				result = result || dispatch_assume(pwq);
 			}
 #endif // DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK
+#if HAVE_INTERNAL_PTHREAD_WORKQUEUE
+			if (!disable_wq && qc->dgq_wq_priority != WORKQ_PRIO_INVALID) {
+				int priority = qc->dgq_wq_priority;
+				int overcommit = qc->dgq_wq_options & WORKQ_ADDTHREADS_OPTION_OVERCOMMIT;
+				r = dispatch_workqueue_get_wq(&pwq, priority, overcommit);
+				(void)dispatch_assume_zero(r);
+				result = result || dispatch_assume(pwq);
+			}
+#endif // HAVE_INTERNAL_PTHREAD_WORKQUEUE
 			qc->dgq_kworkqueue = pwq ? pwq : (void*)(~0ul);
 		}
 #if DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK
@@ -672,7 +682,7 @@ _dispatch_root_queues_init_workq(int *wq_supported)
 		}
 #endif
 	}
-#endif // DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK || DISPATCH_ENABLE_THREAD_POOL
+#endif // DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK || DISPATCH_ENABLE_THREAD_POOL || HAVE_INTERNAL_PTHREAD_WORKQUEUE
 #endif // HAVE_PTHREAD_WORKQUEUES
 	return result;
 }
@@ -3972,6 +3982,16 @@ _dispatch_global_queue_poke_slow(dispatch_queue_t dq, unsigned int n)
 			return;
 		}
 #endif // DISPATCH_USE_LEGACY_WORKQUEUE_FALLBACK
+#if HAVE_INTERNAL_PTHREAD_WORKQUEUE
+		if (qc->dgq_kworkqueue) {
+			do {
+				r = dispatch_workqueue_additem_np(qc->dgq_kworkqueue,
+						_dispatch_worker_thread4, dq);
+				(void)dispatch_assume_zero(r);
+			} while (--i);
+			return;
+		}
+#endif // HAVE_INTERNAL_PTHREAD_WORKQUEUE
 #if HAVE_PTHREAD_WORKQUEUE_QOS
 		r = _pthread_workqueue_addthreads((int)i,
 				_dispatch_priority_to_pp(dq->dq_priority));

src/shims.h

@@ -42,8 +42,12 @@
 #if __has_include(<pthread/workqueue_private.h>)
 #include <pthread/workqueue_private.h>
 #else
+#if HAVE_INTERNAL_PTHREAD_WORKQUEUE
+#include <workqueue/workqueue_internal.h>
+#else
 #include <pthread_workqueue.h>
 #endif
+#endif
 #ifndef WORKQ_FEATURE_MAINTENANCE
 #define WORKQ_FEATURE_MAINTENANCE 0x10
 #endif