Address review

Author: Keno

NEWS.md

@@ -21,6 +21,10 @@ Command-line option changes
 Multi-threading changes
 -----------------------
 
+  - New functions `Threads.atomic_fence_heavy` and `Threads.atoimc_fence_light` provide support for
+    asymmetric atomic fences, speeding up atomic synchronization where one side of the synchronization
+    runs significantly less often than the other ([#60311]).
+
 Build system changes
 --------------------
 

base/atomics.jl

@@ -10,7 +10,7 @@ export
     atomic_add!, atomic_sub!,
     atomic_and!, atomic_nand!, atomic_or!, atomic_xor!,
     atomic_max!, atomic_min!,
-    atomic_fence
+    atomic_fence, atomic_fence_light, atomic_fence_heavy
 
 """
     Threads.Atomic{T}
@@ -334,23 +334,23 @@ atomic_fence() = Core.Intrinsics.atomic_fence(:sequentially_consistent, :system)
 """
     Threads.atomic_fence_light()
 
-This is a read-optimized sequential-consistency memory fence.
-On supported operating systems and architectures, this fence is cheaper
-than `Threads.atomic_fence()`, but synchronizes only with
-[`atomic_fence_heavy`](@ref) calls from other threads.
+Insert the light side of an asymmetric sequential-consistency memory fence.
+Asymmetric memory fences are useful in scenarios where one side of the
+synchronization runs significantly less often than the other side. Use this
+function on the side that runs often and [`atomic_fence_heavy`](@ref) on the
+side that runs rarely.
+
+On supported operating systems and architectures this fence is cheaper than
+`Threads.atomic_fence()`, but synchronizes only with [`atomic_fence_heavy`](@ref)
+calls from other threads.
 """
 atomic_fence_light() = Core.Intrinsics.atomic_fence(:sequentially_consistent, :singlethread)
 
 """
     Threads.atomic_fence_heavy()
 
-This is a write-optimized sequential-consistency memory fence.
-This fence is significantly more expensive than `Threads.atomic_fence`.
-It generally requires a system call and a full interprocessor interrupt
-to all other processors in the system. It synchronizes with both
-[`atomic_fence_light`](@ref) and [`atomic_fence`](@ref) calls from other threads.
-
-For further details, see the Linux `membarrier` syscall or the Windows
-`FlushProcessWriteBuffers` API.
+Insert the heavy side of an asymmetric sequential-consistency memory fence.
+Use this function on the side that runs rarely.
+See [`atomic_fence_light`](@ref) for more details.
 """
 atomic_fence_heavy() = ccall(:jl_membarrier, Cvoid, ())

doc/src/base/multi-threading.md

@@ -50,6 +50,8 @@ Base.Threads.atomic_xor!
 Base.Threads.atomic_max!
 Base.Threads.atomic_min!
 Base.Threads.atomic_fence
+Base.Threads.atomic_fence_heavy
+Base.Threads.atomic_fence_light
 ```
 
 ## ccall using a libuv threadpool (Experimental)

src/runtime_intrinsics.c

@@ -626,13 +626,13 @@ JL_DLLEXPORT jl_value_t *jl_atomic_fence(jl_value_t *order_sym, jl_value_t *sync
 {
     JL_TYPECHK(fence, symbol, order_sym);
     JL_TYPECHK(fence, symbol, syncscope_sym);
+    enum jl_memory_order order = jl_get_atomic_order_checked((jl_sym_t*)order_sym, 1, 1);
     if ((jl_sym_t*)syncscope_sym == jl_singlethread_sym) {
         asm volatile ("" : : : "memory");
         return jl_nothing;
     } else if ((jl_sym_t*)syncscope_sym != jl_system_sym) {
         jl_error("atomic_fence: invalid syncscope");
     }
-    enum jl_memory_order order = jl_get_atomic_order_checked((jl_sym_t*)order_sym, 1, 1);
     if (order > jl_memory_order_monotonic)
         jl_fence();
     return jl_nothing;

src/signals-mach.c

@@ -898,34 +898,32 @@ JL_DLLEXPORT void jl_profile_stop_timer(void)
 // Copyright (c) .NET Foundation and Contributors
 // MIT LICENSE
 JL_DLLEXPORT void jl_membarrier(void) {
-    mach_msg_type_number_t cThreads;
-    thread_act_t *pThreads;
-    kern_return_t machret = task_threads(mach_task_self(), &pThreads, &cThreads);
-    HANDLE_MACH_ERROR("task_threads()", machret);
-
     uintptr_t sp;
     uintptr_t registerValues[128];
+    kern_return_t machret;
 
     // Iterate through each of the threads in the list.
-    for (mach_msg_type_number_t i = 0; i < cThreads; i++)
-    {
+    int nthreads = jl_atomic_load_acquire(&jl_n_threads);
+    for (int tid = 0; tid < nthreads; tid++) {
+        jl_ptls_t ptls2 = jl_atomic_load_relaxed(&jl_all_tls_states)[tid];
+        thread_act_t thread = pthread_mach_thread_np(ptls2->system_id);
         if (__builtin_available (macOS 10.14, iOS 12, tvOS 9, *))
         {
             // Request the threads pointer values to force the thread to emit a memory barrier
             size_t registers = 128;
-            machret = thread_get_register_pointer_values(pThreads[i], &sp, &registers, registerValues);
+            machret = thread_get_register_pointer_values(thread, &sp, &registers, registerValues);
         }
         else
         {
             // fallback implementation for older OS versions
 #if defined(_CPU_X86_64_)
             x86_thread_state64_t threadState;
             mach_msg_type_number_t count = x86_THREAD_STATE64_COUNT;
-            machret = thread_get_state(pThreads[i], x86_THREAD_STATE64, (thread_state_t)&threadState, &count);
+            machret = thread_get_state(thread, x86_THREAD_STATE64, (thread_state_t)&threadState, &count);
 #elif defined(_CPU_AARCH64_)
             arm_thread_state64_t threadState;
             mach_msg_type_number_t count = ARM_THREAD_STATE64_COUNT;
-            machret = thread_get_state(pThreads[i], ARM_THREAD_STATE64, (thread_state_t)&threadState, &count);
+            machret = thread_get_state(thread, ARM_THREAD_STATE64, (thread_state_t)&threadState, &count);
 #else
             #error Unexpected architecture
 #endif
@@ -935,11 +933,5 @@ JL_DLLEXPORT void jl_membarrier(void) {
         {
             HANDLE_MACH_ERROR("thread_get_register_pointer_values()", machret);
         }
-
-        machret = mach_port_deallocate(mach_task_self(), pThreads[i]);
-        HANDLE_MACH_ERROR("mach_port_deallocate()", machret);
     }
-    // Deallocate the thread list now we're done with it.
-    machret = vm_deallocate(mach_task_self(), (vm_address_t)pThreads, cThreads * sizeof(thread_act_t));
-    HANDLE_MACH_ERROR("vm_deallocate()", machret);
 }

src/signals-unix.c

@@ -1275,48 +1275,15 @@ JL_DLLEXPORT int jl_repl_raise_sigtstp(void)
     return raise(SIGTSTP);
 }
 
-// Linux and FreeBSD have compatible membarrier support
-#if defined(_OS_LINUX_) || defined(_OS_FREEBSD_)
-#if defined(_OS_LINUX_)
-# include <sys/syscall.h>
-# if defined(__has_include)
-#   if __has_include(<linux/membarrier.h>)
-#       include <linux/membarrier.h>
-#       define membarrier(...) syscall(__NR_membarrier, __VA_ARGS__)
-#   else
-#     if defined(__NR_membarrier)
-enum membarrier_cmd {
-    MEMBARRIER_CMD_QUERY                        = 0,
-    MEMBARRIER_CMD_PRIVATE_EXPEDITED            = (1 << 3),
-    MEMBARRIER_CMD_REGISTER_PRIVATE_EXPEDITED   = (1 << 4),
-};
-#         define membarrier(...) syscall(__NR_membarrier, __VA_ARGS__)
-#     else
-#         warning "Missing linux kernel headers for membarrier syscall, support disabled"
-#         define membarrier(...) -ENOSYS
-#     endif
-#  endif
-# else
-#   include <linux/membarrier.h>
-# endif
-#elif defined(_OS_FREEBSD_)
-# include <sys/param.h>
-# if __FreeBSD_version >= 1401500
-#   include <sys/membarrier.h>
-# else
-#   define MEMBARRIER_CMD_QUERY                         0x00
-#   define MEMBARRIER_CMD_PRIVATE_EXPEDITED             0x08
-#   define MEMBARRIER_CMD_REGISTER_PRIVATE_EXPEDITED    0x10
-#   define membarrier(...) -ENOSYS
-# endif
-#endif
-
+#if !defined(_OS_DARWIN_)
 // Implementation of the `mprotect` based membarrier fallback.
 // This is a common fallback based on the observation that `mprotect` happens to
 // issue the necessary memory barriers. However, there is no spec that
-// guarantees this behavior, and indeed AArch64 macos does not. However, we
-// only use it as a fallback here for older versions of Linux and FreeBSD where
-// we know that it happens to work.
+// guarantees this behavior, and indeed AArch64 Darwin does not (so we don't use it
+// there). However, we only use it as a fallback here for older versions of
+// Linux and FreeBSD where we know that it happens to work. We also use it as a
+// fallback for unknown Unix systems under the assumption that it will work,
+// but this is not guaranteed.
 static pthread_mutex_t mprotect_barrier_lock = PTHREAD_MUTEX_INITIALIZER;
 static _Atomic(uint64_t) *mprotect_barrier_page = NULL;
 static void jl_init_mprotect_membarrier(void)
@@ -1335,7 +1302,7 @@ static void jl_init_mprotect_membarrier(void)
         }
         result = mlock(mprotect_barrier_page, pagesize);
         if (result != 0) {
-            jl_safe_printf("fatal: failed to mlock barrier page.\n");
+            jl_safe_printf("fatal: failed to mlock barrier page (try increasing RLIMIT_MEMLOCK with `ulimit -l`).\n");
             abort();
         }
     }
@@ -1349,15 +1316,43 @@ static void jl_mprotect_membarrier(void)
     int result = pthread_mutex_lock(&mprotect_barrier_lock);
     assert(result == 0);
     size_t pagesize = jl_getpagesize();
-    result = mprotect(mprotect_barrier_page, pagesize, PROT_NONE);
+    result = mprotect(mprotect_barrier_page, pagesize, PROT_READ | PROT_WRITE);
     jl_atomic_fetch_add_relaxed(mprotect_barrier_page, 1);
     assert(result == 0);
-    result = mprotect(mprotect_barrier_page, pagesize, PROT_READ | PROT_WRITE);
+    result = mprotect(mprotect_barrier_page, pagesize, PROT_NONE);
     assert(result == 0);
     result = pthread_mutex_unlock(&mprotect_barrier_lock);
     assert(result == 0);
     (void)result;
 }
+#endif
+
+// Linux and FreeBSD have compatible membarrier support
+#if defined(_OS_LINUX_) || defined(_OS_FREEBSD_)
+#if defined(_OS_LINUX_)
+#   include <sys/syscall.h>
+#   if defined(__NR_membarrier)
+enum membarrier_cmd {
+    MEMBARRIER_CMD_QUERY                        = 0,
+    MEMBARRIER_CMD_PRIVATE_EXPEDITED            = (1 << 3),
+    MEMBARRIER_CMD_REGISTER_PRIVATE_EXPEDITED   = (1 << 4),
+};
+#    define membarrier(...) syscall(__NR_membarrier, __VA_ARGS__)
+#  else
+#    warning "Missing linux kernel headers for membarrier syscall, support disabled"
+#    define membarrier(...) (errno = ENOSYS, -1)
+#  endif
+#elif defined(_OS_FREEBSD_)
+#  include <sys/param.h>
+#  if __FreeBSD_version >= 1401500
+#    include <sys/membarrier.h>
+#  else
+#    define MEMBARRIER_CMD_QUERY                         0x00
+#    define MEMBARRIER_CMD_PRIVATE_EXPEDITED             0x08
+#    define MEMBARRIER_CMD_REGISTER_PRIVATE_EXPEDITED    0x10
+#    define membarrier(...) (errno = ENOSYS, -1)
+#  endif
+#endif
 
 // Implementation of `jl_membarrier`
 enum membarrier_implementation {
@@ -1391,11 +1386,17 @@ JL_DLLEXPORT void jl_membarrier(void) {
     }
     if (impl == MEMBARRIER_IMPLEMENTATION_SYS_MEMBARRIER) {
         int ret = membarrier(MEMBARRIER_CMD_PRIVATE_EXPEDITED, 0);
-        assert(ret);
+        assert(ret == 0);
         (void)ret;
     } else {
         assert(impl == MEMBARRIER_IMPLEMENTATION_MPROTECT);
         jl_mprotect_membarrier();
     }
 }
+#elif !defined(_OS_DARWIN_)
+JL_DLLEXPORT void jl_membarrier(void) {
+    if (!mprotect_barrier_page)
+        jl_init_mprotect_membarrier();
+    jl_mprotect_membarrier();
+}
 #endif

test/threads_exec.jl

@@ -465,8 +465,8 @@ end
 test_fence()
 
 # Test asymmetric thread fences
-const asymmetric_test_count = 200_000
 struct AsymmetricFenceTestData
+    n::Int
     x::AtomicMemory{Int}
     y::AtomicMemory{Int}
     read_x::AtomicMemory{Int}
@@ -482,23 +482,25 @@ function test_asymmetric_fence(data::AsymmetricFenceTestData, cond1, cond2, thre
     else
         @atomic :monotonic data.y[it] = 1
         Threads.atomic_fence_light()
-        @atomic :monotonic data.read_x[it] = data.x[it]
+        @atomic :monotonic data.read_x[it] = @atomic :monotonic data.x[it]
         notify(cond1)
         wait(cond2)
     end
 end
-function test_asymmetric_fence(data, cond1, cond2, threadid)
-    for i = 1:asymmetric_test_count
+function test_asymmetric_fence(data::AsymmetricFenceTestData, cond1, cond2, threadid)
+    for i = 1:data.n
         test_asymmetric_fence(data, cond1, cond2, threadid, i)
     end
 end
 function test_asymmetric_fence()
+    asymmetric_test_count = 200_000
     cond1 = Threads.Event(true)
     cond2 = Threads.Event(true)
-    data = AsymmetricFenceTestData(AtomicMemory{Int}(undef, asymmetric_test_count),
-                                  AtomicMemory{Int}(undef, asymmetric_test_count),
-                                  AtomicMemory{Int}(undef, asymmetric_test_count),
-                                  AtomicMemory{Int}(undef, asymmetric_test_count))
+    data = AsymmetricFenceTestData(asymmetric_test_count,
+                                   AtomicMemory{Int}(undef, asymmetric_test_count),
+                                   AtomicMemory{Int}(undef, asymmetric_test_count),
+                                   AtomicMemory{Int}(undef, asymmetric_test_count),
+                                   AtomicMemory{Int}(undef, asymmetric_test_count))
     for i = 1:asymmetric_test_count
         @atomic :monotonic data.x[i] = 0
         @atomic :monotonic data.y[i] = 0