gre-ilya/libuv
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

barrier: wait for prior out before next in (#3905)

Browse Source 
This code would previously get confused between rounds of the barrier
being called and a thread might incorrectly get stuck (deadlock) if the
next round started before that thread had exited the current round.

Avoid that by not starting the next round in++ before out-- has reached
zero indicating that all threads have left the prior round.

And fix it that on Windows by replacing the implementation with the one
from unix. There are some awkward platform-specific redirection here
with an extra malloc that is not needed on Win32, but that will be fixed
in libuv v2.

Fixes: https://github.com/libuv/libuv/issue/3872
This commit is contained in:
Jameson Nash 2023-03-06 17:57:25 -05:00 committed by GitHub
parent 460accf9be
commit 964f79f7c8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 219 additions and 209 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CMakeLists.txt
View File

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

1
Makefile.am
View File

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

9
include/uv/win.h
View File

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

175
src/thread-common.c Normal file
View File

@ -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 <stdlib.h>
#ifndef _WIN32
#include <pthread.h>
#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

120
src/unix/thread.c
View File

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

64
src/win/thread.c
View File

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

58
test/test-barrier.c
View File

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