Add condition variable wrapper (#17)

Co-authored-by: Jonatan Kłosko <jonatanklosko@gmail.com>

Author: brodeuralexis

README.md

@@ -524,10 +524,14 @@ can be called from multiple Erlang processes simultaneously, leading to race
 conditions. While C++ provides synchronization mechanisms, these are unknown to
 Erlang and cannot take advantage of tools like *lock checker* or *lcnt*.
 
-Fine provides analogues to `std::mutex` and `std::shared_mutex`, respectively
-called `fine::Mutex` and `fine::SharedMutex`. Those are compatible with the
-standard mutex wrappers, such as `std::unique_lock` and `std::shared_lock`.
-For example:
+Fine provides analogues to `std::mutex`, `std::shared_mutex`, and
+`std::condition_variable`, respectively called `fine::Mutex`,
+`fine::SharedMutex`, and `fine::ConditionVariable`. All of their implementations 
+are provided in the `<fine/sync.h>` header.
+
+
+`fine::Mutex` and `fine::SharedMutex` are compatible with `std::unique_lock`
+and `std::shared_lock`. For example:
 
 ```c++
 #include <fine/sync.hpp>
@@ -571,6 +575,33 @@ const char* my_object__name(struct my_object*);
 
 fine::SharedMutex my_object_rwlock("my_lib", "my_object", my_object__name(my_object));
 ```
+
+Fine also provides `fine::ConditionVariable` with an API similar to
+`std::condition_variable`:
+
+```c++
+bool notified = false;
+fine::Mutex mutex;
+fine::ConditionVariable cond;
+
+std::thread t1([&]() {
+  auto lock = std::unique_lock{mutex};
+  cond.wait(lock, []() { return notified; });
+});
+
+std::thread t2([&]() {
+  {
+    auto lock = std::unique_lock{mutex};
+    notified = true;
+  }
+  
+  cond.notify_all();
+});
+
+t2.join();
+t1.join();
+```
+
 ## Allocators
 
 For compatibility with the STL, fine supports stateless allocators when

c_include/fine/sync.hpp

@@ -194,6 +194,112 @@ class SharedMutex final {
   };
   std::unique_ptr<ErlNifRWLock, Deleter> m_handle;
 };
+
+// Condition variable. Used when threads must wait for a specific
+// condition to appear before continuing execution. Condition
+// variables must be used with associated mutexes.
+class ConditionVariable final {
+public:
+  // Creates a condition variable.
+  ConditionVariable() : m_handle{enif_cond_create(nullptr)} {
+    if (!m_handle) {
+      throw std::runtime_error("failed to create cond");
+    }
+  }
+
+  // Creates a ConditionVariable from an ErlNifCond handle.
+  explicit ConditionVariable(ErlNifCond *handle) : m_handle{handle} {}
+
+  // Creates a condition variable.
+  //
+  // `name` is a string identifying the created condition variable. It is used
+  // to identify the condition variable in planned future debug functionality.
+  explicit ConditionVariable(const char *name)
+      : m_handle{enif_cond_create(const_cast<char *>(name))} {
+    if (!m_handle) {
+      throw std::runtime_error("failed to create cond");
+    }
+  }
+
+  // Creates a condition variable.
+  //
+  // `name` is a string identifying the created condition variable. It is used
+  // to identify the condition variable in planned future debug functionality.
+  explicit ConditionVariable(const std::string &name)
+      : m_handle{enif_cond_create(const_cast<char *>(name.c_str()))} {
+    if (!m_handle) {
+      throw std::runtime_error("failed to create cond");
+    }
+  }
+
+  // Converts this ConditionVariable to a ErlNifConditionVariable handle.
+  //
+  // Ownership still belongs to this instance.
+  operator ErlNifCond *() const & noexcept { return m_handle.get(); }
+
+  // Releases ownership of the ErlNifCond handle to the caller.
+  //
+  // This operation is only possible by:
+  // ```
+  // static_cast<ErlNifCond*>(std::move(rwlock))
+  // ```
+  explicit operator ErlNifCond *() && noexcept { return m_handle.release(); }
+
+  // Broadcasts on this condition variable. That is, if other threads are
+  // waiting on the condition variable being broadcast on, all of them are
+  // woken.
+  //
+  // This function is thread-safe.
+  void notify_all() noexcept { enif_cond_broadcast(m_handle.get()); }
+
+  // Signals on a condition variable. That is, if other threads are waiting on
+  // the condition variable being signaled, one of them is woken.
+  //
+  // This function is thread-safe.
+  void notify_one() noexcept { enif_cond_signal(m_handle.get()); }
+
+  // Prefer the use of `wait(std::unique_lock<Mutex>&, Predicate)` over this
+  // function.
+  //
+  // Waits on a condition variable. The calling thread is blocked until another
+  // thread wakes it by signaling or broadcasting on the condition variable.
+  // Before the calling thread is blocked, it unlocks the mutex passed as
+  // argument. When the calling thread is woken, it locks the same mutex before
+  // returning. That is, the mutex currently must be locked by the calling
+  // thread when calling this function.
+  //
+  // `wait` can return even if no one has signaled or broadcast on the condition
+  // variable. Code calling `wait` is always to be prepared for `wait` returning
+  // even if the condition that the thread was waiting for has not occurred.
+  // That is, when returning from `wait`, always check if the condition has
+  // occurred, and if not call `wait` again.
+  //
+  // This function is thread-safe.
+  void wait(std::unique_lock<Mutex> &lock) noexcept {
+    enif_cond_wait(m_handle.get(), *lock.mutex());
+  }
+
+  // Waits on a condition variable. The calling thread is blocked until another
+  // thread wakes it by signaling or broadcasting on the condition variable.
+  // Before the calling thread is blocked, it unlocks the mutex passed as
+  // argument. When the calling thread is woken, it locks the same mutex before
+  // returning. That is, the mutex currently must be locked by the calling
+  // thread when calling this function.
+  //
+  // This function is thread-safe.
+  template <typename Predicate>
+  void wait(std::unique_lock<Mutex> &lock, Predicate pred) {
+    while (!pred()) {
+      enif_cond_wait(m_handle.get(), *lock.mutex());
+    }
+  }
+
+private:
+  struct Deleter {
+    void operator()(ErlNifCond *handle) noexcept { enif_cond_destroy(handle); }
+  };
+  std::unique_ptr<ErlNifCond, Deleter> m_handle;
+};
 } // namespace fine
 
 #endif

test/c_src/finest.cpp

@@ -1,3 +1,4 @@
+#include <atomic>
 #include <cstring>
 #include <exception>
 #include <functional>
@@ -422,6 +423,64 @@ std::nullopt_t shared_mutex_shared_lock_test(ErlNifEnv *) {
 }
 FINE_NIF(shared_mutex_shared_lock_test, 0);
 
+class ResetEvent {
+public:
+  explicit ResetEvent(const bool signaled = false) noexcept
+      : m_signaled{signaled} {}
+
+  void set() {
+    {
+      auto lock = std::unique_lock{m_mutex};
+      m_signaled = true;
+    }
+
+    m_cond.notify_one();
+  }
+
+  void reset() {
+    auto lock = std::unique_lock{m_mutex};
+    m_signaled = false;
+  }
+
+  void wait() {
+    auto lock = std::unique_lock{m_mutex};
+    m_cond.wait(lock, [&] { return m_signaled; });
+    m_signaled = false;
+  }
+
+private:
+  bool m_signaled;
+  fine::Mutex m_mutex;
+  fine::ConditionVariable m_cond;
+};
+
+std::nullopt_t condition_variable_test(ErlNifEnv *) {
+  ResetEvent event{true};
+  event.reset();
+
+  std::thread wait_thread_1([&] {
+    event.wait();
+    event.set();
+  });
+  std::thread wait_thread_2([&] {
+    event.wait();
+    event.set();
+  });
+  std::thread wait_thread_3([&] {
+    event.wait();
+    event.set();
+  });
+
+  event.set();
+
+  wait_thread_1.join();
+  wait_thread_2.join();
+  wait_thread_3.join();
+
+  return std::nullopt;
+}
+FINE_NIF(condition_variable_test, 0);
+
 bool compare_eq(ErlNifEnv *, fine::Term lhs, fine::Term rhs) noexcept {
   return lhs == rhs;
 }

test/lib/finest/nif.ex

@@ -70,6 +70,8 @@ defmodule Finest.NIF do
   def shared_mutex_unique_lock_test(), do: err!()
   def shared_mutex_shared_lock_test(), do: err!()
 
+  def condition_variable_test(), do: err!()
+
   def compare_eq(_lhs, _rhs), do: err!()
   def compare_ne(_lhs, _rhs), do: err!()
   def compare_lt(_lhs, _rhs), do: err!()

test/test/finest_test.exs

@@ -451,6 +451,12 @@ defmodule FinestTest do
     end
   end
 
+  describe "condition_variable" do
+    test "condition_variable" do
+      NIF.condition_variable_test()
+    end
+  end
+
   describe "comparison" do
     test "equal" do
       refute NIF.compare_eq(64, 42)