Threading: Add lightweight Mutex/ConditionVariable wrappers

src/common-tests/CMakeLists.txt

@@ -19,6 +19,7 @@ add_executable(common-tests
   small_string_tests.cpp
   string_pool_tests.cpp
   string_tests.cpp
+  threading_tests.cpp
 )
 
 target_include_directories(common-tests PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/..")

src/common-tests/common-tests.vcxproj

@@ -16,6 +16,7 @@
     <ClCompile Include="small_string_tests.cpp" />
     <ClCompile Include="string_pool_tests.cpp" />
     <ClCompile Include="string_tests.cpp" />
+    <ClCompile Include="threading_tests.cpp" />
     <ClCompile Include="gsvector_yuvtorgb_test.cpp" />
   </ItemGroup>
   <ItemGroup>

src/common-tests/common-tests.vcxproj.filters

@@ -16,5 +16,6 @@
     <ClCompile Include="lru_cache_tests.cpp" />
     <ClCompile Include="string_pool_tests.cpp" />
     <ClCompile Include="misc_tests.cpp" />
+    <ClCompile Include="threading_tests.cpp" />
   </ItemGroup>
 </Project>

src/common-tests/threading_tests.cpp

@@ -0,0 +1,157 @@
+// SPDX-FileCopyrightText: 2019-2026 Connor McLaughlin <stenzek@gmail.com>
+// SPDX-License-Identifier: CC-BY-NC-ND-4.0
+
+#include "common/threading.h"
+
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <mutex>
+#include <thread>
+#include <type_traits>
+#include <vector>
+
+static_assert(!std::is_copy_constructible_v<Threading::Mutex>);
+static_assert(!std::is_copy_assignable_v<Threading::Mutex>);
+static_assert(!std::is_copy_constructible_v<Threading::ConditionVariable>);
+static_assert(!std::is_copy_assignable_v<Threading::ConditionVariable>);
+
+TEST(ThreadingMutex, TryLock)
+{
+  Threading::Mutex mutex;
+  EXPECT_TRUE(mutex.try_lock());
+  EXPECT_FALSE(mutex.try_lock());
+  mutex.unlock();
+  EXPECT_TRUE(mutex.try_lock());
+  mutex.unlock();
+}
+
+TEST(ThreadingMutex, StandardLockWrappers)
+{
+  Threading::Mutex mutex;
+
+  {
+    const std::lock_guard lock(mutex);
+    EXPECT_FALSE(mutex.try_lock());
+  }
+
+  {
+    std::unique_lock lock(mutex);
+    EXPECT_TRUE(lock.owns_lock());
+    EXPECT_FALSE(mutex.try_lock());
+  }
+
+  EXPECT_TRUE(mutex.try_lock());
+  mutex.unlock();
+}
+
+TEST(ThreadingMutex, MutualExclusion)
+{
+  static constexpr u32 NUM_THREADS = 4;
+  static constexpr u32 NUM_INCREMENTS = 10000;
+
+  Threading::Mutex mutex;
+  u32 value = 0;
+  std::vector<std::thread> threads;
+  threads.reserve(NUM_THREADS);
+  for (u32 i = 0; i < NUM_THREADS; i++)
+  {
+    threads.emplace_back([&mutex, &value]() {
+      for (u32 j = 0; j < NUM_INCREMENTS; j++)
+      {
+        const std::lock_guard lock(mutex);
+        value++;
+      }
+    });
+  }
+
+  for (std::thread& thread : threads)
+    thread.join();
+
+  EXPECT_EQ(value, NUM_THREADS * NUM_INCREMENTS);
+}
+
+TEST(ThreadingConditionVariable, NotifyOneAndPredicateWait)
+{
+  Threading::Mutex mutex;
+  Threading::ConditionVariable condition;
+  Threading::KernelSemaphore waiting;
+  bool wake = false;
+  bool woke = false;
+
+  std::thread thread([&]() {
+    std::unique_lock lock(mutex);
+    waiting.Post();
+    condition.wait(lock, [&wake]() { return wake; });
+    woke = true;
+  });
+
+  waiting.Wait();
+  {
+    const std::lock_guard lock(mutex);
+    EXPECT_FALSE(woke);
+    wake = true;
+  }
+  condition.notify_one();
+  thread.join();
+  EXPECT_TRUE(woke);
+}
+
+TEST(ThreadingConditionVariable, NotifyAll)
+{
+  static constexpr u32 NUM_THREADS = 4;
+
+  Threading::Mutex mutex;
+  Threading::ConditionVariable condition;
+  Threading::KernelSemaphore waiting;
+  bool wake = false;
+  u32 num_woken = 0;
+  std::vector<std::thread> threads;
+  threads.reserve(NUM_THREADS);
+  for (u32 i = 0; i < NUM_THREADS; i++)
+  {
+    threads.emplace_back([&]() {
+      std::unique_lock lock(mutex);
+      waiting.Post();
+      condition.wait(lock, [&wake]() { return wake; });
+      num_woken++;
+    });
+  }
+
+  for (u32 i = 0; i < NUM_THREADS; i++)
+    waiting.Wait();
+
+  {
+    const std::lock_guard lock(mutex);
+    wake = true;
+  }
+  condition.notify_all();
+
+  for (std::thread& thread : threads)
+    thread.join();
+
+  EXPECT_EQ(num_woken, NUM_THREADS);
+}
+
+TEST(ThreadingKernelSemaphore, PostWaitAndTryWait)
+{
+  Threading::KernelSemaphore semaphore;
+  EXPECT_FALSE(semaphore.TryWait());
+  semaphore.Post();
+  EXPECT_TRUE(semaphore.TryWait());
+  EXPECT_FALSE(semaphore.TryWait());
+
+  Threading::KernelSemaphore started;
+  std::atomic_bool finished = false;
+  std::thread thread([&]() {
+    started.Post();
+    semaphore.Wait();
+    finished.store(true, std::memory_order_release);
+  });
+
+  started.Wait();
+  EXPECT_FALSE(finished.load(std::memory_order_acquire));
+  semaphore.Post();
+  thread.join();
+  EXPECT_TRUE(finished.load(std::memory_order_acquire));
+}

src/common/threading.cpp

@@ -22,7 +22,11 @@
 #include "windows_headers.h"
 #include <process.h>
 #else
+#include <cerrno>
 #include <pthread.h>
+#if !defined(__APPLE__)
+#include <semaphore.h>
+#endif
 #include <unistd.h>
 #if defined(__linux__)
 #include <sched.h>
@@ -49,6 +53,20 @@
 
 LOG_CHANNEL(Threading);
 
+#ifdef _WIN32
+#define NATIVE_MUTEX_PTR(storage) static_cast<SRWLOCK*>(static_cast<void*>(&(storage)))
+#define NATIVE_CONDITION_VARIABLE_PTR(storage) static_cast<CONDITION_VARIABLE*>(static_cast<void*>(&(storage)))
+#else
+#define NATIVE_MUTEX_PTR(storage) static_cast<pthread_mutex_t*>(static_cast<void*>(&(storage)))
+#define NATIVE_CONDITION_VARIABLE_PTR(storage) static_cast<pthread_cond_t*>(static_cast<void*>(&(storage)))
+#endif
+
+#if defined(__APPLE__)
+#define NATIVE_SEMAPHORE(storage) (*static_cast<semaphore_t*>(static_cast<void*>(&(storage))))
+#elif !defined(_WIN32)
+#define NATIVE_SEMAPHORE(storage) static_cast<sem_t*>(static_cast<void*>(&(storage)))
+#endif
+
 #ifdef _WIN32
 union FileTimeU64Union
 {
@@ -549,6 +567,129 @@ Threading::ThreadHandle& Threading::Thread::operator=(Thread&& thread)
   return *this;
 }
 
+#ifdef _WIN32
+
+// Trick we do to avoid the constructor/destructor.
+static_assert(sizeof(SRWLOCK) == sizeof(void*));
+static_assert(alignof(SRWLOCK) == alignof(void*));
+static_assert(
+  []() {
+    const SRWLOCK ensure_srwlock_init_is_zero = SRWLOCK_INIT;
+    return (ensure_srwlock_init_is_zero.Ptr == nullptr);
+  }(),
+  "SRWLOCK_INIT is equivalent to null");
+
+#else // _WIN32
+
+Threading::Mutex::Mutex()
+{
+  static_assert(sizeof(m_data) == sizeof(pthread_mutex_t));
+  static_assert(alignof(Mutex) >= alignof(pthread_mutex_t));
+  [[maybe_unused]] const int result = pthread_mutex_init(NATIVE_MUTEX_PTR(m_data), nullptr);
+  DebugAssert(result == 0);
+}
+
+Threading::Mutex::~Mutex()
+{
+  [[maybe_unused]] const int result = pthread_mutex_destroy(NATIVE_MUTEX_PTR(m_data));
+  DebugAssert(result == 0);
+}
+
+#endif // _WIN32
+
+void Threading::Mutex::lock()
+{
+#ifdef _WIN32
+  AcquireSRWLockExclusive(NATIVE_MUTEX_PTR(m_data));
+#else
+  [[maybe_unused]] const int result = pthread_mutex_lock(NATIVE_MUTEX_PTR(m_data));
+  DebugAssert(result == 0);
+#endif
+}
+
+bool Threading::Mutex::try_lock()
+{
+#ifdef _WIN32
+  return TryAcquireSRWLockExclusive(NATIVE_MUTEX_PTR(m_data)) != FALSE;
+#else
+  const int result = pthread_mutex_trylock(NATIVE_MUTEX_PTR(m_data));
+  DebugAssert(result == 0 || result == EBUSY);
+  return result == 0;
+#endif
+}
+
+void Threading::Mutex::unlock()
+{
+#ifdef _WIN32
+  ReleaseSRWLockExclusive(NATIVE_MUTEX_PTR(m_data));
+#else
+  [[maybe_unused]] const int result = pthread_mutex_unlock(NATIVE_MUTEX_PTR(m_data));
+  DebugAssert(result == 0);
+#endif
+}
+
+#ifndef _WIN32
+
+Threading::ConditionVariable::ConditionVariable()
+{
+  static_assert(sizeof(m_data) == sizeof(pthread_cond_t));
+  static_assert(alignof(ConditionVariable) >= alignof(pthread_cond_t));
+  [[maybe_unused]] const int result = pthread_cond_init(NATIVE_CONDITION_VARIABLE_PTR(m_data), nullptr);
+  DebugAssert(result == 0);
+}
+
+Threading::ConditionVariable::~ConditionVariable()
+{
+  [[maybe_unused]] const int result = pthread_cond_destroy(NATIVE_CONDITION_VARIABLE_PTR(m_data));
+  DebugAssert(result == 0);
+}
+
+#else // _WIN32
+
+static_assert(sizeof(CONDITION_VARIABLE) == sizeof(void*));
+static_assert(alignof(CONDITION_VARIABLE) == alignof(void*));
+static_assert(
+  []() {
+    const CONDITION_VARIABLE ensure_condition_variable_init_is_zero = CONDITION_VARIABLE_INIT;
+    return (ensure_condition_variable_init_is_zero.Ptr == nullptr);
+  }(),
+  "CONDITION_VARIABLE_INIT is equivalent to null");
+
+#endif // _WIN32
+
+void Threading::ConditionVariable::notify_one()
+{
+#ifdef _WIN32
+  WakeConditionVariable(NATIVE_CONDITION_VARIABLE_PTR(m_data));
+#else
+  [[maybe_unused]] const int result = pthread_cond_signal(NATIVE_CONDITION_VARIABLE_PTR(m_data));
+  DebugAssert(result == 0);
+#endif
+}
+
+void Threading::ConditionVariable::notify_all()
+{
+#ifdef _WIN32
+  WakeAllConditionVariable(NATIVE_CONDITION_VARIABLE_PTR(m_data));
+#else
+  [[maybe_unused]] const int result = pthread_cond_broadcast(NATIVE_CONDITION_VARIABLE_PTR(m_data));
+  DebugAssert(result == 0);
+#endif
+}
+
+void Threading::ConditionVariable::Wait(Mutex& mutex)
+{
+#ifdef _WIN32
+  [[maybe_unused]] const BOOL result =
+    SleepConditionVariableSRW(NATIVE_CONDITION_VARIABLE_PTR(m_data), NATIVE_MUTEX_PTR(mutex.m_data), INFINITE, 0);
+  DebugAssert(result);
+#else
+  [[maybe_unused]] const int result =
+    pthread_cond_wait(NATIVE_CONDITION_VARIABLE_PTR(m_data), NATIVE_MUTEX_PTR(mutex.m_data));
+  DebugAssert(result == 0);
+#endif
+}
+
 u64 Threading::GetThreadCpuTime()
 {
 #if defined(_WIN32) && !defined(_M_ARM64)
@@ -673,15 +814,21 @@ void Threading::SetNameOfCurrentThread(const char* name)
 Threading::KernelSemaphore::KernelSemaphore()
 {
 #ifdef _WIN32
-  m_sema = CreateSemaphore(nullptr, 0, LONG_MAX, nullptr);
-  if (m_sema == NULL) [[unlikely]]
+  static_assert(sizeof(m_data) == sizeof(HANDLE));
+  static_assert(alignof(KernelSemaphore) >= alignof(HANDLE));
+  m_data = CreateSemaphore(nullptr, 0, LONG_MAX, nullptr);
+  if (m_data == nullptr) [[unlikely]]
     Panic("CreateSemaphore() failed");
 #elif defined(__APPLE__)
-  const kern_return_t kr = semaphore_create(mach_task_self(), &m_sema, SYNC_POLICY_FIFO, 0);
+  static_assert(sizeof(m_data) >= sizeof(semaphore_t));
+  static_assert(alignof(KernelSemaphore) >= alignof(semaphore_t));
+  const kern_return_t kr = semaphore_create(mach_task_self(), &NATIVE_SEMAPHORE(m_data), SYNC_POLICY_FIFO, 0);
   if (kr != KERN_SUCCESS) [[unlikely]]
     Panic("CreateSemaphore() failed");
 #else
-  if (sem_init(&m_sema, false, 0) != 0) [[unlikely]]
+  static_assert(sizeof(m_data) == sizeof(sem_t));
+  static_assert(alignof(KernelSemaphore) >= alignof(sem_t));
+  if (sem_init(NATIVE_SEMAPHORE(m_data), false, 0) != 0) [[unlikely]]
     Panic("sem_init() failed");
 #endif
 }
@@ -689,35 +836,35 @@ Threading::KernelSemaphore::KernelSemaphore()
 Threading::KernelSemaphore::~KernelSemaphore()
 {
 #ifdef _WIN32
-  CloseHandle(m_sema);
+  CloseHandle(m_data);
 #elif defined(__APPLE__)
-  semaphore_destroy(mach_task_self(), m_sema);
+  semaphore_destroy(mach_task_self(), NATIVE_SEMAPHORE(m_data));
 #else
-  sem_destroy(&m_sema);
+  sem_destroy(NATIVE_SEMAPHORE(m_data));
 #endif
 }
 
 void Threading::KernelSemaphore::Post()
 {
 #ifdef _WIN32
-  ReleaseSemaphore(m_sema, 1, nullptr);
+  ReleaseSemaphore(m_data, 1, nullptr);
 #elif defined(__APPLE__)
-  semaphore_signal(m_sema);
+  semaphore_signal(NATIVE_SEMAPHORE(m_data));
 #else
-  sem_post(&m_sema);
+  sem_post(NATIVE_SEMAPHORE(m_data));
 #endif
 }
 
 void Threading::KernelSemaphore::Wait()
 {
 #ifdef _WIN32
-  WaitForSingleObject(m_sema, INFINITE);
+  WaitForSingleObject(m_data, INFINITE);
 #elif defined(__APPLE__)
-  semaphore_wait(m_sema);
+  semaphore_wait(NATIVE_SEMAPHORE(m_data));
 #else
   do
   {
-    if (sem_wait(&m_sema) == 0) [[likely]]
+    if (sem_wait(NATIVE_SEMAPHORE(m_data)) == 0) [[likely]]
       return;
   } while (errno == EINTR);
 #endif
@@ -726,12 +873,18 @@ void Threading::KernelSemaphore::Wait()
 bool Threading::KernelSemaphore::TryWait()
 {
 #ifdef _WIN32
-  return WaitForSingleObject(m_sema, 0) == WAIT_OBJECT_0;
+  return WaitForSingleObject(m_data, 0) == WAIT_OBJECT_0;
 #elif defined(__APPLE__)
   mach_timespec_t time = {};
-  kern_return_t res = semaphore_timedwait(m_sema, time);
+  kern_return_t res = semaphore_timedwait(NATIVE_SEMAPHORE(m_data), time);
   return (res != KERN_OPERATION_TIMED_OUT);
 #else
-  return sem_trywait(&m_sema) == 0;
+  return sem_trywait(NATIVE_SEMAPHORE(m_data)) == 0;
 #endif
 }
+
+#undef NATIVE_MUTEX_PTR
+#undef NATIVE_CONDITION_VARIABLE_PTR
+#if !defined(_WIN32)
+#undef NATIVE_SEMAPHORE
+#endif

src/common/threading.h

@@ -5,12 +5,6 @@
 
 #include "types.h"
 
-#if defined(__APPLE__)
-#include <mach/semaphore.h>
-#elif !defined(_WIN32)
-#include <semaphore.h>
-#endif
-
 #include <atomic>
 #include <functional>
 
@@ -119,6 +113,107 @@ protected:
 #endif
 };
 
+// --------------------------------------------------------------------------------------
+//  Mutex
+// --------------------------------------------------------------------------------------
+// A lightweight replacement for std::mutex. The native object is stored inline to avoid
+// the oversized standard library representation on Windows, without exposing platform
+// headers to users of this header.
+//
+class Mutex
+{
+public:
+#ifdef _WIN32
+  Mutex() = default;
+#else
+  Mutex();
+  ~Mutex();
+#endif
+
+  Mutex(const Mutex&) = delete;
+  Mutex& operator=(const Mutex&) = delete;
+
+  void lock();
+  bool try_lock();
+  void unlock();
+
+private:
+  friend class ConditionVariable;
+
+#if defined(_WIN32)
+  void* m_data = nullptr;
+#elif defined(__APPLE__)
+  static constexpr u32 NATIVE_STORAGE_SIZE = 64;
+#elif defined(__ANDROID__)
+  static constexpr u32 NATIVE_STORAGE_SIZE = (sizeof(void*) == 8) ? 40 : 4;
+#elif defined(__linux__) && defined(CPU_ARCH_ARM64)
+  static constexpr u32 NATIVE_STORAGE_SIZE = 48;
+#elif defined(__linux__)
+  static constexpr u32 NATIVE_STORAGE_SIZE = (sizeof(void*) == 8) ? 40 : 24;
+#else
+#error Unsupported platform.
+#endif
+
+#if !defined(_WIN32)
+  alignas(void*) u8 m_data[NATIVE_STORAGE_SIZE];
+#endif
+};
+
+// --------------------------------------------------------------------------------------
+//  ConditionVariable
+// --------------------------------------------------------------------------------------
+// A lightweight replacement for the subset of std::condition_variable used by the
+// project. Spurious wakeups are permitted, matching std::condition_variable.
+//
+class ConditionVariable
+{
+public:
+#ifdef _WIN32
+  ConditionVariable() = default;
+#else
+  ConditionVariable();
+  ~ConditionVariable();
+#endif
+
+  ConditionVariable(const ConditionVariable&) = delete;
+  ConditionVariable& operator=(const ConditionVariable&) = delete;
+
+  void notify_one();
+  void notify_all();
+
+  template<typename LockType>
+  void wait(LockType& lock)
+  {
+    Wait(*lock.mutex());
+  }
+
+  template<typename LockType, typename Predicate>
+  void wait(LockType& lock, Predicate predicate)
+  {
+    while (!predicate())
+      wait(lock);
+  }
+
+private:
+  void Wait(Mutex& mutex);
+
+#if defined(_WIN32)
+  void* m_data = nullptr;
+#elif defined(__APPLE__) || (defined(__linux__) && !defined(__ANDROID__))
+  static constexpr u32 NATIVE_STORAGE_SIZE = 48;
+  static constexpr u32 NATIVE_STORAGE_ALIGNMENT = 8;
+#elif defined(__ANDROID__)
+  static constexpr u32 NATIVE_STORAGE_SIZE = (sizeof(void*) == 8) ? 48 : 4;
+  static constexpr u32 NATIVE_STORAGE_ALIGNMENT = alignof(void*);
+#else
+#error Unsupported platform.
+#endif
+
+#if !defined(_WIN32)
+  alignas(NATIVE_STORAGE_ALIGNMENT) u8 m_data[NATIVE_STORAGE_SIZE];
+#endif
+};
+
 /// A semaphore that requires a system call to wake/sleep.
 class KernelSemaphore
 {
@@ -134,13 +229,19 @@ public:
   bool TryWait();
 
 private:
-#if defined(_WIN32)
-  void* m_sema;
-#elif defined(__APPLE__)
-  semaphore_t m_sema;
+#if defined(_WIN32) || defined(__APPLE__)
+  void* m_data = nullptr;
+#elif defined(__ANDROID__)
+  static constexpr u32 NATIVE_STORAGE_SIZE = (sizeof(void*) == 8) ? 16 : 4;
+#elif defined(__linux__)
+  static constexpr u32 NATIVE_STORAGE_SIZE = (sizeof(void*) == 8) ? 32 : 16;
 #else
-  sem_t m_sema;
+#error Unsupported platform.
+#endif
+
+#if !defined(_WIN32) && !defined(__APPLE__)
+  alignas(void*) u8 m_data[NATIVE_STORAGE_SIZE] = {};
 #endif
 };
 
-} // namespace Threading
+} // namespace Threading