[libc++] refactor semaphore latch barrier in terms of atomic::wait

Author: huixie90

libcxx/include/barrier

@@ -56,8 +56,6 @@ namespace std
 #include <__atomic/memory_order.h>
 #include <__availability>
 #include <__memory/unique_ptr.h>
-#include <__thread/poll_with_backoff.h>
-#include <__thread/timed_backoff_policy.h>
 #include <__utility/move.h>
 #include <cstddef>
 #include <cstdint>
@@ -144,8 +142,7 @@ public:
     return __old_phase;
   }
   _LIBCPP_AVAILABILITY_SYNC _LIBCPP_HIDE_FROM_ABI void wait(arrival_token&& __old_phase) const {
-    auto const __test_fn = [this, __old_phase]() -> bool { return __phase_.load(memory_order_acquire) != __old_phase; };
-    std::__libcpp_thread_poll_with_backoff(__test_fn, __libcpp_timed_backoff_policy());
+    __phase_.wait(__old_phase, std::memory_order_acquire);
   }
   _LIBCPP_AVAILABILITY_SYNC _LIBCPP_HIDE_FROM_ABI void arrive_and_drop() {
     __expected_adjustment_.fetch_sub(1, memory_order_relaxed);
@@ -242,11 +239,11 @@ public:
     return __old & __phase_bit;
   }
   inline _LIBCPP_AVAILABILITY_SYNC _LIBCPP_HIDE_FROM_ABI void wait(arrival_token&& __phase) const {
-    auto const __test_fn = [=]() -> bool {
-      uint64_t const __current = __phase_arrived_expected.load(memory_order_acquire);
-      return ((__current & __phase_bit) != __phase);
-    };
-    __libcpp_thread_poll_with_backoff(__test_fn, __libcpp_timed_backoff_policy());
+    uint64_t __current = __phase_arrived_expected.load(memory_order_acquire);
+    while ((__current & __phase_bit) == __phase) {
+      __phase_arrived_expected.wait(__current, std::memory_order_relaxed);
+      __current = __phase_arrived_expected.load(memory_order_acquire);
+    }
   }
   inline _LIBCPP_AVAILABILITY_SYNC _LIBCPP_HIDE_FROM_ABI void arrive_and_drop() {
     __phase_arrived_expected.fetch_add(__expected_unit, memory_order_relaxed);

libcxx/include/latch

@@ -48,7 +48,6 @@ namespace std
 
 #include <__assert> // all public C++ headers provide the assertion handler
 #include <__atomic/atomic_base.h>
-#include <__atomic/atomic_sync.h>
 #include <__atomic/memory_order.h>
 #include <__availability>
 #include <cstddef>
@@ -99,7 +98,11 @@ public:
   }
   inline _LIBCPP_HIDE_FROM_ABI bool try_wait() const noexcept { return 0 == __a_.load(memory_order_acquire); }
   inline _LIBCPP_AVAILABILITY_SYNC _LIBCPP_HIDE_FROM_ABI void wait() const {
-    __cxx_atomic_wait(&__a_.__a_, [this]() -> bool { return try_wait(); });
+    auto __old = __a_.load(memory_order_acquire);
+    while (__old != 0) {
+      __a_.wait(__old, std::memory_order_relaxed);
+      __old = __a_.load(memory_order_acquire);
+    }
   }
   inline _LIBCPP_AVAILABILITY_SYNC _LIBCPP_HIDE_FROM_ABI void arrive_and_wait(ptrdiff_t __update = 1) {
     _LIBCPP_ASSERT_ARGUMENT_WITHIN_DOMAIN(__update >= 0, "latch::arrive_and_wait called with a negative value");

libcxx/include/semaphore

@@ -95,20 +95,23 @@ public:
     _LIBCPP_ASSERT_ARGUMENT_WITHIN_DOMAIN(
         __update <= _LIBCPP_SEMAPHORE_MAX - __old, "update is greater than the expected value");
 
-    if (__old > 0) {
-      // Nothing to do
-    } else if (__update > 1)
+    if (__old != 0) {
       __a_.notify_all();
-    else
-      __a_.notify_one();
+    }
   }
+
   _LIBCPP_AVAILABILITY_SYNC _LIBCPP_HIDE_FROM_ABI void acquire() {
-    auto const __test_fn = [this]() -> bool {
-      auto __old = __a_.load(memory_order_relaxed);
-      return (__old != 0) && __a_.compare_exchange_strong(__old, __old - 1, memory_order_acquire, memory_order_relaxed);
-    };
-    __cxx_atomic_wait(&__a_.__a_, __test_fn);
+    auto __old = __a_.load(std::memory_order_relaxed);
+    while (true) {
+      if (__old == 0) {
+        __a_.wait(__old, std::memory_order_relaxed);
+        __old = __a_.load(std::memory_order_relaxed);
+      } else if (__a_.compare_exchange_weak(__old, __old - 1, std::memory_order_acquire, std::memory_order_relaxed)) {
+        break;
+      }
+    }
   }
+
   template <class _Rep, class _Period>
   _LIBCPP_AVAILABILITY_SYNC _LIBCPP_HIDE_FROM_ABI bool
   try_acquire_for(chrono::duration<_Rep, _Period> const& __rel_time) {
@@ -117,14 +120,10 @@ public:
     auto const __test_fn = [this]() { return try_acquire(); };
     return std::__libcpp_thread_poll_with_backoff(__test_fn, __libcpp_timed_backoff_policy(), __rel_time);
   }
+
   _LIBCPP_AVAILABILITY_SYNC _LIBCPP_HIDE_FROM_ABI bool try_acquire() {
     auto __old = __a_.load(memory_order_acquire);
-    while (true) {
-      if (__old == 0)
-        return false;
-      if (__a_.compare_exchange_strong(__old, __old - 1, memory_order_acquire, memory_order_relaxed))
-        return true;
-    }
+    return __old != 0 && __a_.compare_exchange_strong(__old, __old - 1, memory_order_acquire, memory_order_relaxed);
   }
 };
 

libcxx/src/thread.cpp

@@ -6,8 +6,6 @@
 //
 //===----------------------------------------------------------------------===//
 
-#include <__thread/poll_with_backoff.h>
-#include <__thread/timed_backoff_policy.h>
 #include <exception>
 #include <future>
 #include <limits>

libcxx/test/std/thread/thread.semaphore/lost_wakeup.pass.cpp

@@ -0,0 +1,64 @@
+//===----------------------------------------------------------------------===//
+//
+// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
+// See https://llvm.org/LICENSE.txt for license information.
+// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+//
+//===----------------------------------------------------------------------===//
+
+// UNSUPPORTED: libcpp-has-no-threads
+// UNSUPPORTED: c++03, c++11, c++14, c++17
+
+// XFAIL: availability-synchronization_library-missing
+
+// This is a regression test for https://llvm.org/PR47013.
+
+// <semaphore>
+
+#include <barrier>
+#include <semaphore>
+#include <thread>
+#include <vector>
+
+#include "make_test_thread.h"
+
+static std::counting_semaphore<> s(0);
+static std::barrier<> b(8 + 1);
+
+void acquire() {
+  for (int i = 0; i < 10'000; ++i) {
+    s.acquire();
+    b.arrive_and_wait();
+  }
+}
+
+void release() {
+  for (int i = 0; i < 10'000; ++i) {
+    s.release(1);
+    s.release(1);
+    s.release(1);
+    s.release(1);
+
+    s.release(1);
+    s.release(1);
+    s.release(1);
+    s.release(1);
+
+    b.arrive_and_wait();
+  }
+}
+
+int main(int, char**) {
+  for (int run = 0; run < 20; ++run) {
+    std::vector<std::thread> threads;
+    for (int i = 0; i < 8; ++i)
+      threads.push_back(support::make_test_thread(acquire));
+
+    threads.push_back(support::make_test_thread(release));
+
+    for (auto& thread : threads)
+      thread.join();
+  }
+
+  return 0;
+}