diff --git a/CMakeLists.txt b/CMakeLists.txt index 58163f69a..1fb05a921 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -166,6 +166,7 @@ set(uv_sources src/random.c src/strscpy.c src/strtok.c + src/thread-common.c src/threadpool.c src/timer.c src/uv-common.c diff --git a/Makefile.am b/Makefile.am index c2c9970d7..be2661580 100644 --- a/Makefile.am +++ b/Makefile.am @@ -38,6 +38,7 @@ libuv_la_SOURCES = src/fs-poll.c \ src/random.c \ src/strscpy.c \ src/strscpy.h \ + src/thread-common.c \ src/threadpool.c \ src/timer.c \ src/uv-data-getter-setters.c \ diff --git a/include/uv/win.h b/include/uv/win.h index 3c3fa0164..1dee8ac5c 100644 --- a/include/uv/win.h +++ b/include/uv/win.h @@ -274,11 +274,12 @@ typedef struct { } uv_rwlock_t; typedef struct { - unsigned int n; - unsigned int count; + unsigned threshold; + unsigned in; uv_mutex_t mutex; - uv_sem_t turnstile1; - uv_sem_t turnstile2; + /* TODO: in v2 make this a uv_cond_t, without unused_ */ + CONDITION_VARIABLE cond; + unsigned out; } uv_barrier_t; typedef struct { diff --git a/src/thread-common.c b/src/thread-common.c new file mode 100644 index 000000000..c67c0a7dd --- /dev/null +++ b/src/thread-common.c @@ -0,0 +1,175 @@ +/* Copyright libuv project contributors. All rights reserved. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + */ + +#include "uv.h" +#include "uv-common.h" + +#include +#ifndef _WIN32 +#include +#endif + +#if defined(PTHREAD_BARRIER_SERIAL_THREAD) +STATIC_ASSERT(sizeof(uv_barrier_t) == sizeof(pthread_barrier_t)); +#endif + +/* Note: guard clauses should match uv_barrier_t's in include/uv/unix.h. */ +#if defined(_AIX) || \ + defined(__OpenBSD__) || \ + !defined(PTHREAD_BARRIER_SERIAL_THREAD) +int uv_barrier_init(uv_barrier_t* barrier, unsigned int count) { + int rc; +#ifdef _WIN32 + uv_barrier_t* b; + b = barrier; + + if (barrier == NULL || count == 0) + return UV_EINVAL; +#else + struct _uv_barrier* b; + + if (barrier == NULL || count == 0) + return UV_EINVAL; + + b = uv__malloc(sizeof(*b)); + if (b == NULL) + return UV_ENOMEM; +#endif + + b->in = 0; + b->out = 0; + b->threshold = count; + + rc = uv_mutex_init(&b->mutex); + if (rc != 0) + goto error2; + + /* TODO(vjnash): remove these uv_cond_t casts in v2. */ + rc = uv_cond_init((uv_cond_t*) &b->cond); + if (rc != 0) + goto error; + +#ifndef _WIN32 + barrier->b = b; +#endif + return 0; + +error: + uv_mutex_destroy(&b->mutex); +error2: +#ifndef _WIN32 + uv__free(b); +#endif + return rc; +} + + +int uv_barrier_wait(uv_barrier_t* barrier) { + int last; +#ifdef _WIN32 + uv_barrier_t* b; + b = barrier; +#else + struct _uv_barrier* b; + + if (barrier == NULL || barrier->b == NULL) + return UV_EINVAL; + + b = barrier->b; +#endif + + uv_mutex_lock(&b->mutex); + + while (b->out != 0) + uv_cond_wait((uv_cond_t*) &b->cond, &b->mutex); + + if (++b->in == b->threshold) { + b->in = 0; + b->out = b->threshold; + uv_cond_broadcast((uv_cond_t*) &b->cond); + } else { + do + uv_cond_wait((uv_cond_t*) &b->cond, &b->mutex); + while (b->in != 0); + } + + last = (--b->out == 0); + if (last) + uv_cond_broadcast((uv_cond_t*) &b->cond); + + uv_mutex_unlock(&b->mutex); + return last; +} + + +void uv_barrier_destroy(uv_barrier_t* barrier) { +#ifdef _WIN32 + uv_barrier_t* b; + b = barrier; +#else + struct _uv_barrier* b; + b = barrier->b; +#endif + + uv_mutex_lock(&b->mutex); + + assert(b->in == 0); + while (b->out != 0) + uv_cond_wait((uv_cond_t*) &b->cond, &b->mutex); + + if (b->in != 0) + abort(); + + uv_mutex_unlock(&b->mutex); + uv_mutex_destroy(&b->mutex); + uv_cond_destroy((uv_cond_t*) &b->cond); + +#ifndef _WIN32 + uv__free(barrier->b); + barrier->b = NULL; +#endif +} + +#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 diff --git a/src/unix/thread.c b/src/unix/thread.c index 8e3652f10..4d6f4b823 100644 --- a/src/unix/thread.c +++ b/src/unix/thread.c @@ -55,124 +55,6 @@ #undef NANOSEC #define NANOSEC ((uint64_t) 1e9) -#if defined(PTHREAD_BARRIER_SERIAL_THREAD) -STATIC_ASSERT(sizeof(uv_barrier_t) == sizeof(pthread_barrier_t)); -#endif - -/* Note: guard clauses should match uv_barrier_t's in include/uv/unix.h. */ -#if defined(_AIX) || \ - defined(__OpenBSD__) || \ - !defined(PTHREAD_BARRIER_SERIAL_THREAD) -int uv_barrier_init(uv_barrier_t* barrier, unsigned int count) { - struct _uv_barrier* b; - int rc; - - if (barrier == NULL || count == 0) - return UV_EINVAL; - - b = uv__malloc(sizeof(*b)); - if (b == NULL) - return UV_ENOMEM; - - b->in = 0; - b->out = 0; - b->threshold = count; - - rc = uv_mutex_init(&b->mutex); - if (rc != 0) - goto error2; - - rc = uv_cond_init(&b->cond); - if (rc != 0) - goto error; - - barrier->b = b; - return 0; - -error: - uv_mutex_destroy(&b->mutex); -error2: - uv__free(b); - return rc; -} - - -int uv_barrier_wait(uv_barrier_t* barrier) { - struct _uv_barrier* b; - int last; - - if (barrier == NULL || barrier->b == NULL) - return UV_EINVAL; - - b = barrier->b; - uv_mutex_lock(&b->mutex); - - if (++b->in == b->threshold) { - b->in = 0; - b->out = b->threshold; - uv_cond_signal(&b->cond); - } else { - do - uv_cond_wait(&b->cond, &b->mutex); - while (b->in != 0); - } - - last = (--b->out == 0); - uv_cond_signal(&b->cond); - - uv_mutex_unlock(&b->mutex); - return last; -} - - -void uv_barrier_destroy(uv_barrier_t* barrier) { - struct _uv_barrier* b; - - b = barrier->b; - uv_mutex_lock(&b->mutex); - - assert(b->in == 0); - while (b->out != 0) - uv_cond_wait(&b->cond, &b->mutex); - - if (b->in != 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 - - /* Musl's PTHREAD_STACK_MIN is 2 KB on all architectures, which is * too small to safely receive signals on. * @@ -696,7 +578,7 @@ static void uv__custom_sem_post(uv_sem_t* sem_) { uv_mutex_lock(&sem->mutex); sem->value++; if (sem->value == 1) - uv_cond_signal(&sem->cond); + uv_cond_signal(&sem->cond); /* Release one to replace us. */ uv_mutex_unlock(&sem->mutex); } diff --git a/src/win/thread.c b/src/win/thread.c index 5e86999c2..57c25e8f5 100644 --- a/src/win/thread.c +++ b/src/win/thread.c @@ -449,6 +449,7 @@ void uv_cond_wait(uv_cond_t* cond, uv_mutex_t* mutex) { abort(); } + int uv_cond_timedwait(uv_cond_t* cond, uv_mutex_t* mutex, uint64_t timeout) { if (SleepConditionVariableCS(&cond->cond_var, mutex, (DWORD)(timeout / 1e6))) return 0; @@ -458,69 +459,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) { - int err; - - barrier->n = count; - barrier->count = 0; - - err = uv_mutex_init(&barrier->mutex); - if (err) - return err; - - err = uv_sem_init(&barrier->turnstile1, 0); - if (err) - goto error2; - - err = uv_sem_init(&barrier->turnstile2, 1); - if (err) - goto error; - - return 0; - -error: - uv_sem_destroy(&barrier->turnstile1); -error2: - uv_mutex_destroy(&barrier->mutex); - return err; - -} - - -void uv_barrier_destroy(uv_barrier_t* barrier) { - uv_sem_destroy(&barrier->turnstile2); - uv_sem_destroy(&barrier->turnstile1); - uv_mutex_destroy(&barrier->mutex); -} - - -int uv_barrier_wait(uv_barrier_t* barrier) { - int serial_thread; - - uv_mutex_lock(&barrier->mutex); - if (++barrier->count == barrier->n) { - uv_sem_wait(&barrier->turnstile2); - uv_sem_post(&barrier->turnstile1); - } - uv_mutex_unlock(&barrier->mutex); - - uv_sem_wait(&barrier->turnstile1); - uv_sem_post(&barrier->turnstile1); - - uv_mutex_lock(&barrier->mutex); - serial_thread = (--barrier->count == 0); - if (serial_thread) { - uv_sem_wait(&barrier->turnstile1); - uv_sem_post(&barrier->turnstile2); - } - uv_mutex_unlock(&barrier->mutex); - - uv_sem_wait(&barrier->turnstile2); - uv_sem_post(&barrier->turnstile2); - return serial_thread; -} - - int uv_key_create(uv_key_t* key) { key->tls_index = TlsAlloc(); if (key->tls_index == TLS_OUT_OF_INDEXES) diff --git a/test/test-barrier.c b/test/test-barrier.c index 89858db57..c780f0cf2 100644 --- a/test/test-barrier.c +++ b/test/test-barrier.c @@ -27,20 +27,22 @@ typedef struct { uv_barrier_t barrier; - int delay; - volatile int posted; - int main_barrier_wait_rval; - int worker_barrier_wait_rval; + unsigned delay; + unsigned niter; + unsigned main_barrier_wait_rval; + unsigned worker_barrier_wait_rval; } worker_config; static void worker(void* arg) { worker_config* c = arg; + unsigned i; if (c->delay) uv_sleep(c->delay); - c->worker_barrier_wait_rval = uv_barrier_wait(&c->barrier); + for (i = 0; i < c->niter; i++) + c->worker_barrier_wait_rval += uv_barrier_wait(&c->barrier); } @@ -49,17 +51,18 @@ TEST_IMPL(barrier_1) { worker_config wc; memset(&wc, 0, sizeof(wc)); + wc.niter = 1; - ASSERT(0 == uv_barrier_init(&wc.barrier, 2)); - ASSERT(0 == uv_thread_create(&thread, worker, &wc)); + ASSERT_EQ(0, uv_barrier_init(&wc.barrier, 2)); + ASSERT_EQ(0, uv_thread_create(&thread, worker, &wc)); uv_sleep(100); wc.main_barrier_wait_rval = uv_barrier_wait(&wc.barrier); - ASSERT(0 == uv_thread_join(&thread)); + ASSERT_EQ(0, uv_thread_join(&thread)); uv_barrier_destroy(&wc.barrier); - ASSERT(1 == (wc.main_barrier_wait_rval ^ wc.worker_barrier_wait_rval)); + ASSERT_EQ(1, (wc.main_barrier_wait_rval ^ wc.worker_barrier_wait_rval)); return 0; } @@ -71,16 +74,17 @@ TEST_IMPL(barrier_2) { memset(&wc, 0, sizeof(wc)); wc.delay = 100; + wc.niter = 1; - ASSERT(0 == uv_barrier_init(&wc.barrier, 2)); - ASSERT(0 == uv_thread_create(&thread, worker, &wc)); + ASSERT_EQ(0, uv_barrier_init(&wc.barrier, 2)); + ASSERT_EQ(0, uv_thread_create(&thread, worker, &wc)); wc.main_barrier_wait_rval = uv_barrier_wait(&wc.barrier); - ASSERT(0 == uv_thread_join(&thread)); + ASSERT_EQ(0, uv_thread_join(&thread)); uv_barrier_destroy(&wc.barrier); - ASSERT(1 == (wc.main_barrier_wait_rval ^ wc.worker_barrier_wait_rval)); + ASSERT_EQ(1, (wc.main_barrier_wait_rval ^ wc.worker_barrier_wait_rval)); return 0; } @@ -89,26 +93,32 @@ TEST_IMPL(barrier_2) { TEST_IMPL(barrier_3) { uv_thread_t thread; worker_config wc; + unsigned i; memset(&wc, 0, sizeof(wc)); + wc.niter = 5; - ASSERT(0 == uv_barrier_init(&wc.barrier, 2)); - ASSERT(0 == uv_thread_create(&thread, worker, &wc)); + ASSERT_EQ(0, uv_barrier_init(&wc.barrier, 2)); + ASSERT_EQ(0, uv_thread_create(&thread, worker, &wc)); - wc.main_barrier_wait_rval = uv_barrier_wait(&wc.barrier); + for (i = 0; i < wc.niter; i++) + wc.main_barrier_wait_rval += uv_barrier_wait(&wc.barrier); - ASSERT(0 == uv_thread_join(&thread)); + ASSERT_EQ(0, uv_thread_join(&thread)); uv_barrier_destroy(&wc.barrier); - ASSERT(1 == (wc.main_barrier_wait_rval ^ wc.worker_barrier_wait_rval)); + ASSERT_EQ(wc.niter, wc.main_barrier_wait_rval + wc.worker_barrier_wait_rval); return 0; } static void serial_worker(void* data) { uv_barrier_t* barrier; + unsigned i; barrier = data; + for (i = 0; i < 5; i++) + uv_barrier_wait(barrier); if (uv_barrier_wait(barrier) > 0) uv_barrier_destroy(barrier); @@ -123,16 +133,18 @@ TEST_IMPL(barrier_serial_thread) { uv_barrier_t barrier; unsigned i; - ASSERT(0 == uv_barrier_init(&barrier, ARRAY_SIZE(threads) + 1)); + ASSERT_EQ(0, uv_barrier_init(&barrier, ARRAY_SIZE(threads) + 1)); for (i = 0; i < ARRAY_SIZE(threads); ++i) - ASSERT(0 == uv_thread_create(&threads[i], serial_worker, &barrier)); + ASSERT_EQ(0, uv_thread_create(&threads[i], serial_worker, &barrier)); + for (i = 0; i < 5; i++) + uv_barrier_wait(&barrier); if (uv_barrier_wait(&barrier) > 0) uv_barrier_destroy(&barrier); for (i = 0; i < ARRAY_SIZE(threads); ++i) - ASSERT(0 == uv_thread_join(&threads[i])); + ASSERT_EQ(0, uv_thread_join(&threads[i])); return 0; } @@ -141,8 +153,8 @@ TEST_IMPL(barrier_serial_thread) { TEST_IMPL(barrier_serial_thread_single) { uv_barrier_t barrier; - ASSERT(0 == uv_barrier_init(&barrier, 1)); - ASSERT(0 < uv_barrier_wait(&barrier)); + ASSERT_EQ(0, uv_barrier_init(&barrier, 1)); + ASSERT_LT(0, uv_barrier_wait(&barrier)); uv_barrier_destroy(&barrier); return 0; }