Merge pull request swiftlang#206 from dgrove-oss/internal-pwq-impl

implement pthread_workqueue within libdispatch

Author: MadCoder

.gitmodules

@@ -1,3 +0,0 @@
-[submodule "libpwq"]
-	path = libpwq
-	url = https://github.com/mheily/libpwq.git

Makefile.am

@@ -4,17 +4,12 @@
 
 ACLOCAL_AMFLAGS = -I m4
 
-if BUILD_OWN_PTHREAD_WORKQUEUES
-  MAYBE_PTHREAD_WORKQUEUES = libpwq
-endif
-
 if BUILD_TESTS
   MAYBE_TESTS = tests
 endif
 
 SUBDIRS=						\
 	dispatch					\
-	$(MAYBE_PTHREAD_WORKQUEUES)	\
 	man							\
 	os							\
 	private						\

configure.ac

@@ -320,22 +320,31 @@ 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, than see if there is a 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])
+   dispatch_use_internal_workqueue=true
    have_pthread_workqueues=true],
-  [build_own_pthread_workqueues=false
+  [dispatch_use_internal_workqueue=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(BUILD_OWN_PTHREAD_WORKQUEUES, $build_own_pthread_workqueues)
+AM_CONDITIONAL(DISPATCH_USE_INTERNAL_WORKQUEUE, $dispatch_use_internal_workqueue)
 AM_CONDITIONAL(HAVE_PTHREAD_WORKQUEUES, $have_pthread_workqueues)
 
 AC_CHECK_HEADERS([libproc_internal.h], [], [], [#include <mach/mach.h>])

libpwq

@@ -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,13 @@ AM_OBJCFLAGS=$(DISPATCH_CFLAGS) $(CBLOCKS_FLAGS)
 AM_CXXFLAGS=$(PTHREAD_WORKQUEUE_CFLAGS) $(DISPATCH_CFLAGS) $(CXXBLOCKS_FLAGS)
 AM_OBJCXXFLAGS=$(DISPATCH_CFLAGS) $(CXXBLOCKS_FLAGS)
 
-if BUILD_OWN_PTHREAD_WORKQUEUES
-  PTHREAD_WORKQUEUE_LIBS=$(top_builddir)/libpwq/libpthread_workqueue.la
-  PTHREAD_WORKQUEUE_CFLAGS=-I$(top_srcdir)/libpwq/include
+if DISPATCH_USE_INTERNAL_WORKQUEUE
+  PTHREAD_WORKQUEUE_LIBS=
+  PTHREAD_WORKQUEUE_CFLAGS=
 else
 if HAVE_PTHREAD_WORKQUEUES
   PTHREAD_WORKQUEUE_LIBS=-lpthread_workqueue
+  PTHREAD_WORKQUEUE_CFLAGS=
 endif
 endif
 

src/Makefile.am

@@ -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(