luo980/libuv
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

unix: rework thread barrier implementation

Browse Source 
* 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

PR-URL: https://github.com/libuv/libuv/pull/2019
Reviewed-By: Santiago Gimeno <santiago.gimeno@gmail.com>
This commit is contained in:
Ben Noordhuis 2018-10-08 11:14:41 +02:00
parent daf04e83cb
commit 56702e08bf
4 changed files with 98 additions and 160 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
Makefile.am
View File

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

69
include/uv/pthread-barrier.h
View File

@ -1,69 +0,0 @@
/*
Copyright (c) 2016, Kari Tristan Helgason <kthelgason@gmail.com>
Permission to use, copy, modify, and/or distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
#ifndef _UV_PTHREAD_BARRIER_
#define _UV_PTHREAD_BARRIER_
#include <errno.h>
#include <pthread.h>
#if !defined(__MVS__)
#include <semaphore.h> /* sem_t */
#endif
#define PTHREAD_BARRIER_SERIAL_THREAD 0x12345
#define UV__PTHREAD_BARRIER_FALLBACK 1
/*
* To maintain ABI compatibility with
* libuv v1.x struct is padded according
* to target platform
*/
#if defined(__ANDROID__)
# define UV_BARRIER_STRUCT_PADDING \
sizeof(pthread_mutex_t) + \
sizeof(pthread_cond_t) + \
sizeof(unsigned int) - \
sizeof(void *)
#elif defined(__APPLE__)
# define UV_BARRIER_STRUCT_PADDING \
sizeof(pthread_mutex_t) + \
2 * sizeof(sem_t) + \
2 * sizeof(unsigned int) - \
sizeof(void *)
#else
# define UV_BARRIER_STRUCT_PADDING 0
#endif
typedef struct {
pthread_mutex_t mutex;
pthread_cond_t cond;
unsigned threshold;
unsigned in;
unsigned out;
} _uv_barrier;
typedef struct {
_uv_barrier* b;
char _pad[UV_BARRIER_STRUCT_PADDING];
} pthread_barrier_t;
int pthread_barrier_init(pthread_barrier_t* barrier,
const void* barrier_attr,
unsigned count);
int pthread_barrier_wait(pthread_barrier_t* barrier);
int pthread_barrier_destroy(pthread_barrier_t *barrier);
#endif /* _UV_PTHREAD_BARRIER_ */

26
include/uv/unix.h
View File

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

156
src/unix/thread.c
View File

@ -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;
}
int pthread_barrier_destroy(pthread_barrier_t* barrier) {
int rc;
_uv_barrier* b;
if (barrier == NULL || barrier->b == NULL)
return EINVAL;
b = barrier->b;
if ((rc = pthread_mutex_lock(&b->mutex)) != 0)
return rc;
if (b->in > 0 || b->out > 0)
rc = EBUSY;
pthread_mutex_unlock(&b->mutex);
if (rc)
return rc;
pthread_cond_destroy(&b->cond);
pthread_mutex_destroy(&b->mutex);
uv__free(barrier->b);
barrier->b = NULL;
uv_cond_signal(&b->cond);
uv_mutex_unlock(&b->mutex);
return 0;
}
void uv_barrier_destroy(uv_barrier_t* barrier) {
struct _uv_barrier* b;
b = barrier->b;
uv_mutex_lock(&b->mutex);
assert(b->in == 0);
assert(b->out == 0);
if (b->in != 0 || b->out != 0)
abort();
uv_mutex_unlock(&b->mutex);
uv_mutex_destroy(&b->mutex);
uv_cond_destroy(&b->cond);
uv__free(barrier->b);
barrier->b = NULL;
}
#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));
}