unix: rework thread barrier implementation

* dissolve include/uv/pthread-barrier.h

* use libuv mutexes and condition variables, not pthreads's

* drive-by cleanup and simplification enabled by the first two items

Author: bnoordhuis

Makefile.am

@@ -340,8 +340,7 @@ libuv_la_SOURCES += src/unix/aix.c src/unix/aix-common.c
 endif
 
 if ANDROID
-uvinclude_HEADERS += include/uv/android-ifaddrs.h \
-                   include/uv/pthread-barrier.h
+uvinclude_HEADERS += include/uv/android-ifaddrs.h
 libuv_la_SOURCES += src/unix/android-ifaddrs.c \
                     src/unix/pthread-fixes.c
 endif
@@ -361,8 +360,7 @@ libuv_la_SOURCES += src/unix/cygwin.c \
 endif
 
 if DARWIN
-uvinclude_HEADERS += include/uv/darwin.h \
-                   include/uv/pthread-barrier.h
+uvinclude_HEADERS += include/uv/darwin.h
 libuv_la_CFLAGS += -D_DARWIN_USE_64_BIT_INODE=1
 libuv_la_CFLAGS += -D_DARWIN_UNLIMITED_SELECT=1
 libuv_la_SOURCES += src/unix/bsd-ifaddrs.c \
@@ -445,7 +443,6 @@ libuv_la_SOURCES += src/unix/no-proctitle.c \
 endif
 
 if OS390
-uvinclude_HEADERS += include/uv/pthread-barrier.h
 libuv_la_CFLAGS += -D_UNIX03_THREADS \
                    -D_UNIX03_SOURCE \
                    -D_OPEN_SYS_IF_EXT=1 \

include/uv/pthread-barrier.h

@@ -66,10 +66,6 @@
 # include "uv/posix.h"
 #endif
 
-#ifndef PTHREAD_BARRIER_SERIAL_THREAD
-# include "uv/pthread-barrier.h"
-#endif
-
 #ifndef NI_MAXHOST
 # define NI_MAXHOST 1025
 #endif
@@ -136,8 +132,28 @@ typedef pthread_rwlock_t uv_rwlock_t;
 typedef UV_PLATFORM_SEM_T uv_sem_t;
 typedef pthread_cond_t uv_cond_t;
 typedef pthread_key_t uv_key_t;
-typedef pthread_barrier_t uv_barrier_t;
 
+/* Note: guard clauses should match uv_barrier_init's in src/unix/thread.c. */
+#if !defined(PTHREAD_BARRIER_SERIAL_THREAD)
+/* TODO(bnoordhuis) Merge into uv_barrier_t in v2. */
+struct _uv_barrier {
+  uv_mutex_t mutex;
+  uv_cond_t cond;
+  unsigned threshold;
+  unsigned in;
+  unsigned out;
+};
+
+typedef struct {
+  struct _uv_barrier* b;
+# if defined(PTHREAD_BARRIER_SERIAL_THREAD)
+  /* TODO(bnoordhuis) Remove padding in v2. */
+  char pad[sizeof(pthread_barrier_t) - sizeof(struct _uv_barrier*)];
+# endif
+} uv_barrier_t;
+#else
+typedef pthread_barrier_t uv_barrier_t;
+#endif
 
 /* Platform-specific definitions for uv_spawn support. */
 typedef gid_t uv_gid_t;

include/uv/unix.h

@@ -44,108 +44,121 @@
 #undef NANOSEC
 #define NANOSEC ((uint64_t) 1e9)
 
+#if defined(PTHREAD_BARRIER_SERIAL_THREAD)
+STATIC_ASSERT(sizeof(uv_barrier_t) == sizeof(pthread_barrier_t));
+#endif
 
-#if defined(UV__PTHREAD_BARRIER_FALLBACK)
-/* TODO: support barrier_attr */
-int pthread_barrier_init(pthread_barrier_t* barrier,
-                         const void* barrier_attr,
-                         unsigned count) {
+/* Note: guard clauses should match uv_barrier_t's in include/uv/uv-unix.h. */
+#if !defined(PTHREAD_BARRIER_SERIAL_THREAD)
+int uv_barrier_init(uv_barrier_t* barrier, unsigned int count) {
+  struct _uv_barrier* b;
   int rc;
-  _uv_barrier* b;
 
   if (barrier == NULL || count == 0)
-    return EINVAL;
-
-  if (barrier_attr != NULL)
-    return ENOTSUP;
+    return UV_EINVAL;
 
   b = uv__malloc(sizeof(*b));
   if (b == NULL)
-    return ENOMEM;
+    return UV_ENOMEM;
 
   b->in = 0;
   b->out = 0;
   b->threshold = count;
 
-  if ((rc = pthread_mutex_init(&b->mutex, NULL)) != 0)
+  rc = uv_mutex_init(&b->mutex);
+  if (rc != 0)
     goto error2;
-  if ((rc = pthread_cond_init(&b->cond, NULL)) != 0)
+
+  rc = uv_cond_init(&b->cond);
+  if (rc != 0)
     goto error;
 
   barrier->b = b;
   return 0;
 
 error:
-  pthread_mutex_destroy(&b->mutex);
+  uv_mutex_destroy(&b->mutex);
 error2:
   uv__free(b);
   return rc;
 }
 
-int pthread_barrier_wait(pthread_barrier_t* barrier) {
-  int rc;
-  _uv_barrier* b;
+
+int uv_barrier_wait(uv_barrier_t* barrier) {
+  struct _uv_barrier* b;
 
   if (barrier == NULL || barrier->b == NULL)
-    return EINVAL;
+    return UV_EINVAL;
 
   b = barrier->b;
-  /* Lock the mutex*/
-  if ((rc = pthread_mutex_lock(&b->mutex)) != 0)
-    return rc;
+  uv_mutex_lock(&b->mutex);
 
-  /* Increment the count. If this is the first thread to reach the threshold,
-     wake up waiters, unlock the mutex, then return
-     PTHREAD_BARRIER_SERIAL_THREAD. */
   if (++b->in == b->threshold) {
     b->in = 0;
     b->out = b->threshold - 1;
-    rc = pthread_cond_signal(&b->cond);
-    assert(rc == 0);
-
-    pthread_mutex_unlock(&b->mutex);
-    return PTHREAD_BARRIER_SERIAL_THREAD;
+    uv_cond_signal(&b->cond);
+    uv_mutex_unlock(&b->mutex);
+    return 1;  /* This is the first thread to reach the threshold. */
   }
+
   /* Otherwise, wait for other threads until in is set to 0,
      then return 0 to indicate this is not the first thread. */
-  do {
-    if ((rc = pthread_cond_wait(&b->cond, &b->mutex)) != 0)
-      break;
-  } while (b->in != 0);
+  do
+    uv_cond_wait(&b->cond, &b->mutex);
+  while (b->in != 0);
 
   /* mark thread exit */
   b->out--;
-  pthread_cond_signal(&b->cond);
-  pthread_mutex_unlock(&b->mutex);
-  return rc;
+  uv_cond_signal(&b->cond);
+  uv_mutex_unlock(&b->mutex);
+  return 0;
 }
 
-int pthread_barrier_destroy(pthread_barrier_t* barrier) {
-  int rc;
-  _uv_barrier* b;
 
-  if (barrier == NULL || barrier->b == NULL)
-    return EINVAL;
+void uv_barrier_destroy(uv_barrier_t* barrier) {
+  struct _uv_barrier* b;
 
   b = barrier->b;
+  uv_mutex_lock(&b->mutex);
 
-  if ((rc = pthread_mutex_lock(&b->mutex)) != 0)
-    return rc;
-
-  if (b->in > 0 || b->out > 0)
-    rc = EBUSY;
+  assert(b->in == 0);
+  assert(b->out == 0);
 
-  pthread_mutex_unlock(&b->mutex);
+  if (b->in != 0 || b->out != 0)
+    abort();
 
-  if (rc)
-    return rc;
+  uv_mutex_unlock(&b->mutex);
+  uv_mutex_destroy(&b->mutex);
+  uv_cond_destroy(&b->cond);
 
-  pthread_cond_destroy(&b->cond);
-  pthread_mutex_destroy(&b->mutex);
   uv__free(barrier->b);
   barrier->b = NULL;
-  return 0;
 }
+
+#else
+
+int uv_barrier_init(uv_barrier_t* barrier, unsigned int count) {
+  return UV__ERR(pthread_barrier_init(barrier, NULL, count));
+}
+
+
+int uv_barrier_wait(uv_barrier_t* barrier) {
+  int rc;
+
+  rc = pthread_barrier_wait(barrier);
+  if (rc != 0)
+    if (rc != PTHREAD_BARRIER_SERIAL_THREAD)
+      abort();
+
+  return rc == PTHREAD_BARRIER_SERIAL_THREAD;
+}
+
+
+void uv_barrier_destroy(uv_barrier_t* barrier) {
+  if (pthread_barrier_destroy(barrier))
+    abort();
+}
+
 #endif
 
 
@@ -771,25 +784,6 @@ int uv_cond_timedwait(uv_cond_t* cond, uv_mutex_t* mutex, uint64_t timeout) {
 }
 
 
-int uv_barrier_init(uv_barrier_t* barrier, unsigned int count) {
-  return UV__ERR(pthread_barrier_init(barrier, NULL, count));
-}
-
-
-void uv_barrier_destroy(uv_barrier_t* barrier) {
-  if (pthread_barrier_destroy(barrier))
-    abort();
-}
-
-
-int uv_barrier_wait(uv_barrier_t* barrier) {
-  int r = pthread_barrier_wait(barrier);
-  if (r && r != PTHREAD_BARRIER_SERIAL_THREAD)
-    abort();
-  return r == PTHREAD_BARRIER_SERIAL_THREAD;
-}
-
-
 int uv_key_create(uv_key_t* key) {
   return UV__ERR(pthread_key_create(key, NULL));
 }