sanitizer_common: add Semaphore

Semaphore is a portable way to park/unpark threads.
The plan is to use it to implement a portable blocking
mutex in subsequent changes. Semaphore can also be used
to efficiently wait for other things (e.g. we currently
spin to synchronize thread creation and start).

Reviewed By: vitalybuka, melver

Differential Revision: https://reviews.llvm.org/D106071

compiler-rt/lib/sanitizer_common/CMakeLists.txt

@@ -16,6 +16,7 @@ set(SANITIZER_SOURCES_NOTERMINATION
   sanitizer_linux.cpp
   sanitizer_linux_s390.cpp
   sanitizer_mac.cpp
+  sanitizer_mutex.cpp
   sanitizer_netbsd.cpp
   sanitizer_persistent_allocator.cpp
   sanitizer_platform_limits_freebsd.cpp

compiler-rt/lib/sanitizer_common/sanitizer_fuchsia.cpp

@@ -100,6 +100,18 @@ bool SignalContext::IsStackOverflow() const { return false; }
 void SignalContext::DumpAllRegisters(void *context) { UNIMPLEMENTED(); }
 const char *SignalContext::Describe() const { UNIMPLEMENTED(); }
 
+void FutexWait(atomic_uint32_t *p, u32 cmp) {
+  zx_status_t status = _zx_futex_wait(reinterpret_cast<zx_futex_t *>(p), cmp,
+                                      ZX_HANDLE_INVALID, ZX_TIME_INFINITE);
+  if (status != ZX_ERR_BAD_STATE)  // Normal race.
+    CHECK_EQ(status, ZX_OK);
+}
+
+void FutexWake(atomic_uint32_t *p, u32 count) {
+  zx_status_t status = _zx_futex_wake(reinterpret_cast<zx_futex_t *>(p), count);
+  CHECK_EQ(status, ZX_OK);
+}
+
 enum MutexState : int { MtxUnlocked = 0, MtxLocked = 1, MtxSleeping = 2 };
 
 BlockingMutex::BlockingMutex() {

compiler-rt/lib/sanitizer_common/sanitizer_linux.cpp

@@ -639,6 +639,26 @@ char **GetEnviron() {
 }
 
 #if !SANITIZER_SOLARIS
+void FutexWait(atomic_uint32_t *p, u32 cmp) {
+#    if SANITIZER_FREEBSD
+  _umtx_op(p, UMTX_OP_WAIT_UINT, cmp, 0, 0);
+#    elif SANITIZER_NETBSD
+  sched_yield();   /* No userspace futex-like synchronization */
+#    else
+  internal_syscall(SYSCALL(futex), (uptr)p, FUTEX_WAIT_PRIVATE, cmp, 0, 0, 0);
+#    endif
+}
+
+void FutexWake(atomic_uint32_t *p, u32 count) {
+#    if SANITIZER_FREEBSD
+  _umtx_op(p_, UMTX_OP_WAKE, count, 0, 0);
+#    elif SANITIZER_NETBSD
+                   /* No userspace futex-like synchronization */
+#    else
+  internal_syscall(SYSCALL(futex), (uptr)p, FUTEX_WAKE_PRIVATE, count, 0, 0, 0);
+#    endif
+}
+
 enum { MtxUnlocked = 0, MtxLocked = 1, MtxSleeping = 2 };
 
 BlockingMutex::BlockingMutex() {

compiler-rt/lib/sanitizer_common/sanitizer_mac.cpp

@@ -509,6 +509,13 @@ void MprotectMallocZones(void *addr, int prot) {
   }
 }
 
+void FutexWait(atomic_uint32_t *p, u32 cmp) {
+  // FIXME: implement actual blocking.
+  sched_yield();
+}
+
+void FutexWake(atomic_uint32_t *p, u32 count) {}
+
 BlockingMutex::BlockingMutex() {
   internal_memset(this, 0, sizeof(*this));
 }

compiler-rt/lib/sanitizer_common/sanitizer_mutex.cpp

@@ -0,0 +1,39 @@
+//===-- sanitizer_mutex.cpp -----------------------------------------------===//
+//
+// 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
+//
+//===----------------------------------------------------------------------===//
+//
+// This file is shared between AddressSanitizer and ThreadSanitizer
+// run-time libraries.
+//===----------------------------------------------------------------------===//
+
+#include "sanitizer_mutex.h"
+
+#include "sanitizer_common.h"
+
+namespace __sanitizer {
+
+void Semaphore::Wait() {
+  u32 count = atomic_load(&state_, memory_order_relaxed);
+  for (;;) {
+    if (count == 0) {
+      FutexWait(&state_, 0);
+      count = atomic_load(&state_, memory_order_relaxed);
+      continue;
+    }
+    if (atomic_compare_exchange_weak(&state_, &count, count - 1,
+                                     memory_order_acquire))
+      break;
+  }
+}
+
+void Semaphore::Post(u32 count) {
+  CHECK_NE(count, 0);
+  atomic_fetch_add(&state_, count, memory_order_release);
+  FutexWake(&state_, count);
+}
+
+}  // namespace __sanitizer

compiler-rt/lib/sanitizer_common/sanitizer_mutex.h

@@ -69,6 +69,25 @@ class MUTEX SpinMutex : public StaticSpinMutex {
   void operator=(const SpinMutex &) = delete;
 };
 
+// Semaphore provides an OS-dependent way to park/unpark threads.
+// The last thread returned from Wait can destroy the object
+// (destruction-safety).
+class Semaphore {
+ public:
+  constexpr Semaphore() {}
+  Semaphore(const Semaphore &) = delete;
+  void operator=(const Semaphore &) = delete;
+
+  void Wait();
+  void Post(u32 count = 1);
+
+ private:
+  atomic_uint32_t state_ = {0};
+};
+
+void FutexWait(atomic_uint32_t *p, u32 cmp);
+void FutexWake(atomic_uint32_t *p, u32 count);
+
 class MUTEX BlockingMutex {
  public:
   explicit constexpr BlockingMutex(LinkerInitialized)

compiler-rt/lib/sanitizer_common/sanitizer_solaris.cpp

@@ -218,6 +218,13 @@ uptr internal_clock_gettime(__sanitizer_clockid_t clk_id, void *tp) {
 }
 
 // ----------------- sanitizer_common.h
+void FutexWait(atomic_uint32_t *p, u32 cmp) {
+  // FIXME: implement actual blocking.
+  sched_yield();
+}
+
+void FutexWake(atomic_uint32_t *p, u32 count) {}
+
 BlockingMutex::BlockingMutex() {
   CHECK(sizeof(mutex_t) <= sizeof(opaque_storage_));
   internal_memset(this, 0, sizeof(*this));

compiler-rt/lib/sanitizer_common/sanitizer_win.cpp

@@ -813,6 +813,17 @@ uptr GetRSS() {
 void *internal_start_thread(void *(*func)(void *arg), void *arg) { return 0; }
 void internal_join_thread(void *th) { }
 
+void FutexWait(atomic_uint32_t *p, u32 cmp) {
+  WaitOnAddress(p, &cmp, sizeof(cmp), INFINITE);
+}
+
+void FutexWake(atomic_uint32_t *p, u32 count) {
+  if (count == 1)
+    WakeByAddressSingle(p);
+  else
+    WakeByAddressAll(p);
+}
+
 // ---------------------- BlockingMutex ---------------- {{{1
 
 BlockingMutex::BlockingMutex() {

compiler-rt/lib/sanitizer_common/tests/sanitizer_mutex_test.cpp

@@ -133,4 +133,34 @@ TEST(SanitizerCommon, BlockingMutex) {
   check_locked(mtx);
 }
 
+struct SemaphoreData {
+  Semaphore *sem;
+  bool done;
+};
+
+void *SemaphoreThread(void *arg) {
+  auto data = static_cast<SemaphoreData *>(arg);
+  data->sem->Wait();
+  data->done = true;
+  return nullptr;
+}
+
+TEST(SanitizerCommon, Semaphore) {
+  Semaphore sem;
+  sem.Post(1);
+  sem.Wait();
+  sem.Post(3);
+  sem.Wait();
+  sem.Wait();
+  sem.Wait();
+
+  SemaphoreData data = {&sem, false};
+  pthread_t thread;
+  PTHREAD_CREATE(&thread, nullptr, SemaphoreThread, &data);
+  sleep(1);
+  CHECK(!data.done);
+  sem.Post(1);
+  PTHREAD_JOIN(thread, nullptr);
+}
+
 }  // namespace __sanitizer

compiler-rt/lib/tsan/go/build.bat

@@ -33,6 +33,7 @@ type ^
   ..\..\sanitizer_common\sanitizer_termination.cpp ^
   ..\..\sanitizer_common\sanitizer_file.cpp ^
   ..\..\sanitizer_common\sanitizer_symbolizer_report.cpp ^
+  ..\..\sanitizer_common\sanitizer_mutex.cpp ^
   ..\rtl\tsan_external.cpp ^
   > gotsan.cpp
 

compiler-rt/lib/tsan/go/buildgo.sh

@@ -28,6 +28,7 @@ SRCS="
 	../../sanitizer_common/sanitizer_flag_parser.cpp
 	../../sanitizer_common/sanitizer_flags.cpp
 	../../sanitizer_common/sanitizer_libc.cpp
+	../../sanitizer_common/sanitizer_mutex.cpp
 	../../sanitizer_common/sanitizer_persistent_allocator.cpp
 	../../sanitizer_common/sanitizer_printf.cpp
 	../../sanitizer_common/sanitizer_suppressions.cpp