Author: Yuta Mukai Date: 2026-09-05T09:12:01+02:00 New Revision: ce68f28703200d5cf0f80521a09b23c880efdbde
URL: https://github.com/llvm/llvm-project/commit/ce68f28703200d5cf0f80521a09b23c880efdbde DIFF: https://github.com/llvm/llvm-project/commit/ce68f28703200d5cf0f80521a09b23c880efdbde.diff LOG: [OpenMP][libomp] Fix dist barrier arrival synchronization (#213845) Make distributedBarrier::stillNeed atomic and use release/acquire ordering for distributed barrier gather arrival flags. The old volatile stillNeed flag did not synchronize an arriving thread's pre-barrier writes with the thread that observed its arrival. On weakly ordered architectures, a group leader could observe stillNeed == 0 before the arriving thread's pre-barrier writes were visible. This allowed another thread to pass the barrier and read stale data written before the barrier. Use release stores when publishing stillNeed == 0. Keep the spin loops on relaxed loads, then perform one acquire fence after all expected zero values have been observed. This connects the arriving threads' pre-barrier writes to the observer through the standard release/acquire happens-before chain, without using acquire loads on every poll. The same pattern is used when a group leader publishes its own stillNeed == 0 after the in-group gather completes. The existing go-flag release/acquire path then carries the ordering to threads leaving the barrier. Initialization and next-iteration reset stores remain relaxed because they do not publish pre-barrier data. On x86, this does not change the generated instruction sequence for the affected stillNeed polling/arrival code, since TSO already provides the needed ordering for ordinary loads and stores. Reproducer: I used the following stress test to reproduce the distributed-barrier visibility issue on AArch64 (Grace). `test.c`: ```c #include <stdio.h> #include <stdlib.h> #include <string.h> #include <omp.h> int main() { int nt; #pragma omp parallel #pragma omp master nt = omp_get_num_threads(); int n = 1024; int *p = aligned_alloc(64, nt * n * sizeof(int)); memset(p, 0, nt * n * sizeof(int)); #pragma omp parallel { int t = omp_get_thread_num(); int *pt = p + t * n; int *pt2 = p + ((t + 1) % nt) * n; for (int it = 1; it <= 100000; it++) { for (int i = 0; i < n; i++) pt[i] = it; #pragma omp barrier for (int i = 0; i < n; i++) { if (pt2[i] != it) { printf("thread #%d observes pt2[%d] != %d\n", t, i, it); abort(); } } #pragma omp barrier } } printf("ok\n"); return 0; } ``` `noise.c`: ```c // background memory-bandwidth noise #include <stdlib.h> #include <omp.h> int main() { long N = 1024L*1024; #pragma omp parallel { long *b = malloc(sizeof(long)*N); for (long j=0; j<1000; j++) for (long i = 0; i < N; i += 8) b[i] += 1; } return 0; } ``` Build and run: ```bash clang test.c -O -fopenmp -o test clang noise.c -O -fopenmp -o noise ./noise & KMP_PLAIN_BARRIER_PATTERN=dist,dist \ KMP_FORKJOIN_BARRIER_PATTERN=dist,dist \ KMP_REDUCTION_BARRIER_PATTERN=dist,dist \ OMP_NUM_THREADS=10 \ ./test ``` With the old runtime, this can fail as follows: ```text thread #5 observes pt2[112] != 343 Aborted (core dumped) ``` The same test passes with the patched runtime. (cherry picked from commit 0fa6182e280d36dc73472f9a79dcb1b82d548893) Added: Modified: openmp/runtime/src/kmp_barrier.cpp openmp/runtime/src/kmp_barrier.h Removed: ################################################################################ diff --git a/openmp/runtime/src/kmp_barrier.cpp b/openmp/runtime/src/kmp_barrier.cpp index 4d7989d1ce5eb..5f86c3121392f 100644 --- a/openmp/runtime/src/kmp_barrier.cpp +++ b/openmp/runtime/src/kmp_barrier.cpp @@ -171,7 +171,7 @@ kmp_uint64 distributedBarrier::go_release() { void distributedBarrier::go_reset() { for (size_t j = 0; j < max_threads; ++j) { for (size_t i = 0; i < distributedBarrier::MAX_ITERS; ++i) { - flags[i][j].stillNeed = 1; + flags[i][j].stillNeed.store(1, std::memory_order_relaxed); } go[j].go.store(0); iter[j].iter = 0; @@ -188,7 +188,7 @@ void distributedBarrier::init(size_t nthr) { for (size_t i = 0; i < max_threads; i++) { for (size_t j = 0; j < distributedBarrier::MAX_ITERS; j++) { - flags[j][i].stillNeed = 1; + flags[j][i].stillNeed.store(1, std::memory_order_relaxed); } go[i].go.store(0); iter[i].iter = 0; @@ -291,8 +291,11 @@ static void __kmp_dist_barrier_gather( threads_pending = 0; // Check all the flags every time to avoid branch misspredict for (size_t thr = group_start; thr < group_end; thr++) { - // Each thread uses a diff erent cache line - threads_pending += b->flags[my_current_iter][thr].stillNeed; + // Each thread uses a diff erent cache line. Use relaxed loads while + // polling; the acquire is performed once after the loop observes that + // all threads have arrived. + threads_pending += b->flags[my_current_iter][thr].stillNeed.load( + std::memory_order_relaxed); } // Execute tasks here if (__kmp_tasking_mode != tskm_immediate_exec) { @@ -320,6 +323,9 @@ static void __kmp_dist_barrier_gather( this_thr->th.th_reap_state = KMP_NOT_SAFE_TO_REAP; } } while (threads_pending > 0); + // Acquire: now that all monitored stillNeed=0 stores are observed, make the + // arrived threads' pre-barrier writes (incl. reduce_data) visible here. + std::atomic_thread_fence(std::memory_order_acquire); if (reduce) { // Perform reduction if needed OMPT_REDUCTION_DECL(this_thr, gtid); @@ -333,15 +339,18 @@ static void __kmp_dist_barrier_gather( } // Set flag for next iteration - b->flags[my_next_iter][tid].stillNeed = 1; + b->flags[my_next_iter][tid].stillNeed.store(1, std::memory_order_relaxed); // Each thread uses a diff erent cache line; resets stillNeed to 0 to - // indicate it has reached the barrier - b->flags[my_current_iter][tid].stillNeed = 0; + // indicate it has reached the barrier. Release so that this thread's + // pre-barrier writes are visible to whoever observes the 0. + b->flags[my_current_iter][tid].stillNeed.store(0, + std::memory_order_release); do { // wait for all group leaders threads_pending = 0; for (size_t thr = 0; thr < nproc; thr += b->threads_per_group) { - threads_pending += b->flags[my_current_iter][thr].stillNeed; + threads_pending += b->flags[my_current_iter][thr].stillNeed.load( + std::memory_order_relaxed); } // Execute tasks here if (__kmp_tasking_mode != tskm_immediate_exec) { @@ -369,6 +378,8 @@ static void __kmp_dist_barrier_gather( this_thr->th.th_reap_state = KMP_NOT_SAFE_TO_REAP; } } while (threads_pending > 0); + // Acquire: pair with the group leaders' releasing stillNeed=0 stores. + std::atomic_thread_fence(std::memory_order_acquire); if (reduce) { // Perform reduction if needed if (KMP_MASTER_TID(tid)) { // Master reduces over group leaders @@ -384,10 +395,12 @@ static void __kmp_dist_barrier_gather( } } else { // Set flag for next iteration - b->flags[my_next_iter][tid].stillNeed = 1; + b->flags[my_next_iter][tid].stillNeed.store(1, std::memory_order_relaxed); // Each thread uses a diff erent cache line; resets stillNeed to 0 to - // indicate it has reached the barrier - b->flags[my_current_iter][tid].stillNeed = 0; + // indicate it has reached the barrier. Release so that this thread's + // pre-barrier writes are visible to whoever observes the 0. + b->flags[my_current_iter][tid].stillNeed.store(0, + std::memory_order_release); } KMP_MFENCE(); diff --git a/openmp/runtime/src/kmp_barrier.h b/openmp/runtime/src/kmp_barrier.h index ce6100acc008e..40d558087fc53 100644 --- a/openmp/runtime/src/kmp_barrier.h +++ b/openmp/runtime/src/kmp_barrier.h @@ -58,7 +58,7 @@ static inline void *KMP_ALIGNED_ALLOCATE(size_t size, size_t alignment) { class distributedBarrier { struct flags_s { - kmp_uint32 volatile KMP_FOURLINE_ALIGN_CACHE stillNeed; + std::atomic<kmp_uint32> KMP_FOURLINE_ALIGN_CACHE stillNeed; }; struct go_s { _______________________________________________ llvm-branch-commits mailing list [email protected] https://lists.llvm.org/cgi-bin/mailman/listinfo/llvm-branch-commits
