Use load-acquire + store-release on synchronization variables Synchronization using relaxed atomics + fences instead of LA/SR violates C11/C++11 memory model and cause failures under thread sanitizer
diff --git a/src/pthreads.c b/src/pthreads.c index a7e4619..2d945a0 100644 --- a/src/pthreads.c +++ b/src/pthreads.c
@@ -80,12 +80,12 @@ static void checkin_worker_thread(struct pthreadpool* threadpool) { #if PTHREADPOOL_USE_FUTEX if (pthreadpool_decrement_fetch_relaxed_size_t(&threadpool->active_threads) == 0) { - pthreadpool_store_relaxed_uint32_t(&threadpool->has_active_threads, 0); + pthreadpool_store_release_uint32_t(&threadpool->has_active_threads, 0); futex_wake_all(&threadpool->has_active_threads); } #else pthread_mutex_lock(&threadpool->completion_mutex); - if (pthreadpool_decrement_fetch_relaxed_size_t(&threadpool->active_threads) == 0) { + if (pthreadpool_decrement_fetch_release_size_t(&threadpool->active_threads) == 0) { pthread_cond_signal(&threadpool->completion_condvar); } pthread_mutex_unlock(&threadpool->completion_mutex); @@ -95,12 +95,12 @@ static void wait_worker_threads(struct pthreadpool* threadpool) { /* Initial check */ #if PTHREADPOOL_USE_FUTEX - uint32_t has_active_threads = pthreadpool_load_relaxed_uint32_t(&threadpool->has_active_threads); + uint32_t has_active_threads = pthreadpool_load_acquire_uint32_t(&threadpool->has_active_threads); if (has_active_threads == 0) { return; } #else - size_t active_threads = pthreadpool_load_relaxed_size_t(&threadpool->active_threads); + size_t active_threads = pthreadpool_load_acquire_size_t(&threadpool->active_threads); if (active_threads == 0) { return; } @@ -112,12 +112,12 @@ pthreadpool_fence_acquire(); #if PTHREADPOOL_USE_FUTEX - has_active_threads = pthreadpool_load_relaxed_uint32_t(&threadpool->has_active_threads); + has_active_threads = pthreadpool_load_acquire_uint32_t(&threadpool->has_active_threads); if (has_active_threads == 0) { return; } #else - active_threads = pthreadpool_load_relaxed_size_t(&threadpool->active_threads); + active_threads = pthreadpool_load_acquire_size_t(&threadpool->active_threads); if (active_threads == 0) { return; } @@ -126,12 +126,12 @@ /* Fall-back to mutex/futex wait */ #if PTHREADPOOL_USE_FUTEX - while ((has_active_threads = pthreadpool_load_relaxed_uint32_t(&threadpool->has_active_threads)) != 0) { + while ((has_active_threads = pthreadpool_load_acquire_uint32_t(&threadpool->has_active_threads)) != 0) { futex_wait(&threadpool->has_active_threads, 1); } #else pthread_mutex_lock(&threadpool->completion_mutex); - while (pthreadpool_load_relaxed_size_t(&threadpool->active_threads) != 0) { + while (pthreadpool_load_acquire_size_t(&threadpool->active_threads) != 0) { pthread_cond_wait(&threadpool->completion_condvar, &threadpool->completion_mutex); }; pthread_mutex_unlock(&threadpool->completion_mutex); @@ -143,7 +143,7 @@ uint32_t last_command, uint32_t last_flags) { - uint32_t command = pthreadpool_load_relaxed_uint32_t(&threadpool->command); + uint32_t command = pthreadpool_load_acquire_uint32_t(&threadpool->command); if (command != last_command) { return command; } @@ -154,7 +154,7 @@ /* This fence serves as a sleep instruction */ pthreadpool_fence_acquire(); - command = pthreadpool_load_relaxed_uint32_t(&threadpool->command); + command = pthreadpool_load_acquire_uint32_t(&threadpool->command); if (command != last_command) { return command; } @@ -165,13 +165,13 @@ #if PTHREADPOOL_USE_FUTEX do { futex_wait(&threadpool->command, last_command); - command = pthreadpool_load_relaxed_uint32_t(&threadpool->command); + command = pthreadpool_load_acquire_uint32_t(&threadpool->command); } while (command == last_command); #else /* Lock the command mutex */ pthread_mutex_lock(&threadpool->command_mutex); /* Read the command */ - while ((command = pthreadpool_load_relaxed_uint32_t(&threadpool->command)) == last_command) { + while ((command = pthreadpool_load_acquire_uint32_t(&threadpool->command)) == last_command) { /* Wait for new command */ pthread_cond_wait(&threadpool->command_condvar, &threadpool->command_mutex); } @@ -280,7 +280,7 @@ #if PTHREADPOOL_USE_FUTEX pthreadpool_store_relaxed_uint32_t(&threadpool->has_active_threads, 1); #endif - pthreadpool_store_release_size_t(&threadpool->active_threads, threads_count - 1 /* caller thread */); + pthreadpool_store_relaxed_size_t(&threadpool->active_threads, threads_count - 1 /* caller thread */); /* Caller thread serves as worker #0. Thus, we create system threads starting with worker #1. */ for (size_t tid = 1; tid < threads_count; tid++) {
diff --git a/src/threadpool-atomics.h b/src/threadpool-atomics.h index e2fbe6a..2b46c9a 100644 --- a/src/threadpool-atomics.h +++ b/src/threadpool-atomics.h
@@ -43,6 +43,18 @@ return __c11_atomic_load(address, __ATOMIC_RELAXED); } + static inline uint32_t pthreadpool_load_acquire_uint32_t( + pthreadpool_atomic_uint32_t* address) + { + return __c11_atomic_load(address, __ATOMIC_ACQUIRE); + } + + static inline size_t pthreadpool_load_acquire_size_t( + pthreadpool_atomic_size_t* address) + { + return __c11_atomic_load(address, __ATOMIC_ACQUIRE); + } + static inline void pthreadpool_store_relaxed_uint32_t( pthreadpool_atomic_uint32_t* address, uint32_t value) @@ -84,6 +96,12 @@ return __c11_atomic_fetch_sub(address, 1, __ATOMIC_RELAXED) - 1; } + static inline size_t pthreadpool_decrement_fetch_release_size_t( + pthreadpool_atomic_size_t* address) + { + return __c11_atomic_fetch_sub(address, 1, __ATOMIC_RELEASE) - 1; + } + static inline bool pthreadpool_try_decrement_relaxed_size_t( pthreadpool_atomic_size_t* value) { @@ -128,6 +146,24 @@ return *address; } + static inline uint32_t pthreadpool_load_acquire_uint32_t( + pthreadpool_atomic_uint32_t* address) + { + /* x86-64 loads always have acquire semantics; use only a compiler barrier */ + const uint32_t value = *address; + _ReadBarrier(); + return value; + } + + static inline size_t pthreadpool_load_acquire_size_t( + pthreadpool_atomic_size_t* address) + { + /* x86-64 loads always have acquire semantics; use only a compiler barrier */ + const size_t value = *address; + _ReadBarrier(); + return value; + } + static inline void pthreadpool_store_relaxed_uint32_t( pthreadpool_atomic_uint32_t* address, uint32_t value) @@ -173,6 +209,12 @@ return (size_t) _InterlockedDecrement64((volatile __int64*) address); } + static inline size_t pthreadpool_decrement_fetch_release_size_t( + pthreadpool_atomic_size_t* address) + { + return (size_t) _InterlockedDecrement64((volatile __int64*) address); + } + static inline bool pthreadpool_try_decrement_relaxed_size_t( pthreadpool_atomic_size_t* value) { @@ -221,6 +263,24 @@ return *address; } + static inline uint32_t pthreadpool_load_acquire_uint32_t( + pthreadpool_atomic_uint32_t* address) + { + /* x86 loads always have acquire semantics; use only a compiler barrier */ + const uint32_t value = *address; + _ReadBarrier(); + return value; + } + + static inline size_t pthreadpool_load_acquire_size_t( + pthreadpool_atomic_size_t* address) + { + /* x86 loads always have acquire semantics; use only a compiler barrier */ + const size_t value = *address; + _ReadBarrier(); + return value; + } + static inline void pthreadpool_store_relaxed_uint32_t( pthreadpool_atomic_uint32_t* address, uint32_t value) @@ -246,7 +306,8 @@ pthreadpool_atomic_uint32_t* address, uint32_t value) { - /* x86 stores always have release semantics */ + /* x86 stores always have release semantics; use only a compiler barrier */ + _WriteBarrier(); *address = value; } @@ -254,7 +315,8 @@ pthreadpool_atomic_size_t* address, size_t value) { - /* x86 stores always have release semantics */ + /* x86 stores always have release semantics; use only a compiler barrier */ + _WriteBarrier(); *address = value; } @@ -264,6 +326,12 @@ return (size_t) _InterlockedDecrement((volatile long*) address); } + static inline size_t pthreadpool_decrement_fetch_release_size_t( + pthreadpool_atomic_size_t* address) + { + return (size_t) _InterlockedDecrement((volatile long*) address); + } + static inline bool pthreadpool_try_decrement_relaxed_size_t( pthreadpool_atomic_size_t* value) { @@ -310,6 +378,18 @@ return (void*) __iso_volatile_load64((const volatile __int64*) address); } + static inline uint32_t pthreadpool_load_acquire_uint32_t( + pthreadpool_atomic_uint32_t* address) + { + return (uint32_t) __ldar32((volatile unsigned __int32*) address); + } + + static inline size_t pthreadpool_load_acquire_size_t( + pthreadpool_atomic_size_t* address) + { + return (size_t) __ldar64((volatile unsigned __int64*) address); + } + static inline void pthreadpool_store_relaxed_uint32_t( pthreadpool_atomic_uint32_t* address, uint32_t value) @@ -353,6 +433,12 @@ return (size_t) _InterlockedDecrement64_nf((volatile __int64*) address); } + static inline size_t pthreadpool_decrement_fetch_release_size_t( + pthreadpool_atomic_size_t* address) + { + return (size_t) _InterlockedDecrement64_rel((volatile __int64*) address); + } + static inline bool pthreadpool_try_decrement_relaxed_size_t( pthreadpool_atomic_size_t* value) { @@ -401,6 +487,24 @@ return (void*) __iso_volatile_load32((const volatile __int32*) address); } + static inline uint32_t pthreadpool_load_acquire_uint32_t( + pthreadpool_atomic_uint32_t* address) + { + const uint32_t value = (uint32_t) __iso_volatile_load32((const volatile __int32*) address); + __dmb(_ARM_BARRIER_ISH); + _ReadBarrier(); + return value; + } + + static inline size_t pthreadpool_load_acquire_size_t( + pthreadpool_atomic_size_t* address) + { + const size_t value = (size_t) __iso_volatile_load32((const volatile __int32*) address); + __dmb(_ARM_BARRIER_ISH); + _ReadBarrier(); + return value; + } + static inline void pthreadpool_store_relaxed_uint32_t( pthreadpool_atomic_uint32_t* address, uint32_t value) @@ -446,6 +550,12 @@ return (size_t) _InterlockedDecrement_nf((volatile long*) address); } + static inline size_t pthreadpool_decrement_fetch_release_size_t( + pthreadpool_atomic_size_t* address) + { + return (size_t) _InterlockedDecrement_rel((volatile long*) address); + } + static inline bool pthreadpool_try_decrement_relaxed_size_t( pthreadpool_atomic_size_t* value) { @@ -496,6 +606,18 @@ return atomic_load_explicit(address, memory_order_relaxed); } + static inline uint32_t pthreadpool_load_acquire_uint32_t( + pthreadpool_atomic_uint32_t* address) + { + return atomic_load_explicit(address, memory_order_acquire); + } + + static inline size_t pthreadpool_load_acquire_size_t( + pthreadpool_atomic_size_t* address) + { + return atomic_load_explicit(address, memory_order_acquire); + } + static inline void pthreadpool_store_relaxed_uint32_t( pthreadpool_atomic_uint32_t* address, uint32_t value) @@ -537,6 +659,12 @@ return atomic_fetch_sub_explicit(address, 1, memory_order_relaxed) - 1; } + static inline size_t pthreadpool_decrement_fetch_release_size_t( + pthreadpool_atomic_size_t* address) + { + return atomic_fetch_sub_explicit(address, 1, memory_order_release) - 1; + } + static inline bool pthreadpool_try_decrement_relaxed_size_t( pthreadpool_atomic_size_t* value) {
diff --git a/src/windows.c b/src/windows.c index e68b250..19e534f 100644 --- a/src/windows.c +++ b/src/windows.c
@@ -21,14 +21,14 @@ static void checkin_worker_thread(struct pthreadpool* threadpool, uint32_t event_index) { - if (pthreadpool_decrement_fetch_relaxed_size_t(&threadpool->active_threads) == 0) { + if (pthreadpool_decrement_fetch_release_size_t(&threadpool->active_threads) == 0) { SetEvent(threadpool->completion_event[event_index]); } } static void wait_worker_threads(struct pthreadpool* threadpool, uint32_t event_index) { /* Initial check */ - size_t active_threads = pthreadpool_load_relaxed_size_t(&threadpool->active_threads); + size_t active_threads = pthreadpool_load_acquire_size_t(&threadpool->active_threads); if (active_threads == 0) { return; } @@ -38,7 +38,7 @@ /* This fence serves as a sleep instruction */ pthreadpool_fence_acquire(); - active_threads = pthreadpool_load_relaxed_size_t(&threadpool->active_threads); + active_threads = pthreadpool_load_acquire_size_t(&threadpool->active_threads); if (active_threads == 0) { return; } @@ -55,7 +55,7 @@ uint32_t last_command, uint32_t last_flags) { - uint32_t command = pthreadpool_load_relaxed_uint32_t(&threadpool->command); + uint32_t command = pthreadpool_load_acquire_uint32_t(&threadpool->command); if (command != last_command) { return command; } @@ -66,7 +66,7 @@ /* This fence serves as a sleep instruction */ pthreadpool_fence_acquire(); - command = pthreadpool_load_relaxed_uint32_t(&threadpool->command); + command = pthreadpool_load_acquire_uint32_t(&threadpool->command); if (command != last_command) { return command; } @@ -170,7 +170,7 @@ NULL /* name */); } - pthreadpool_store_release_size_t(&threadpool->active_threads, threads_count - 1 /* caller thread */); + pthreadpool_store_relaxed_size_t(&threadpool->active_threads, threads_count - 1 /* caller thread */); /* Caller thread serves as worker #0. Thus, we create system threads starting with worker #1. */ for (size_t tid = 1; tid < threads_count; tid++) {