[Impeller] Support static thread safety analysis with condition variables.

impeller/base/base_unittests.cc

@@ -76,5 +76,72 @@ TEST(StringsTest, CanSPrintF) {
   ASSERT_EQ(SPrintF("%sx%.2f", "Hello", 12.122222), "Hellox12.12");
 }
 
+struct CVTest {
+  Mutex mutex;
+  ConditionVariable cv;
+  uint32_t rando_ivar IPLR_GUARDED_BY(mutex) = 0;
+};
+
+TEST(ConditionVariableTest, WaitUntil) {
+  CVTest test;
+  // test.rando_ivar = 12; // <--- Static analysis error
+  for (size_t i = 0; i < 2; ++i) {
+    test.mutex.Lock();  // <--- Static analysis error without this.
+    auto result = test.cv.WaitUntil(
+        test.mutex,
+        std::chrono::high_resolution_clock::now() +
+            std::chrono::milliseconds{10},
+        [&]() IPLR_REQUIRES(test.mutex) {
+          test.rando_ivar = 12;  // <-- Static analysics error without the
+                                 // IPLR_REQUIRES on the pred.
+          return false;
+        });
+    test.mutex.Unlock();
+    ASSERT_FALSE(result);
+  }
+  Lock lock(test.mutex);  // <--- Static analysis error without this.
+  // The predicate never returns true. So return has to be due to a non-spurious
+  // wake.
+  ASSERT_EQ(test.rando_ivar, 12u);
+}
+
+TEST(ConditionVariableTest, WaitFor) {
+  CVTest test;
+  // test.rando_ivar = 12; // <--- Static analysis error
+  for (size_t i = 0; i < 2; ++i) {
+    test.mutex.Lock();  // <--- Static analysis error without this.
+    auto result = test.cv.WaitFor(
+        test.mutex, std::chrono::milliseconds{10},
+        [&]() IPLR_REQUIRES(test.mutex) {
+          test.rando_ivar = 12;  // <-- Static analysics error without the
+                                 // IPLR_REQUIRES on the pred.
+          return false;
+        });
+    test.mutex.Unlock();
+    ASSERT_FALSE(result);
+  }
+  Lock lock(test.mutex);  // <--- Static analysis error without this.
+  // The predicate never returns true. So return has to be due to a non-spurious
+  // wake.
+  ASSERT_EQ(test.rando_ivar, 12u);
+}
+
+TEST(ConditionVariableTest, WaitForever) {
+  CVTest test;
+  // test.rando_ivar = 12; // <--- Static analysis error
+  for (size_t i = 0; i < 2; ++i) {
+    test.mutex.Lock();  // <--- Static analysis error without this.
+    test.cv.Wait(test.mutex, [&]() IPLR_REQUIRES(test.mutex) {
+      test.rando_ivar = 12;  // <-- Static analysics error without
+                             // the IPLR_REQUIRES on the pred.
+      return true;
+    });
+    test.mutex.Unlock();
+  }
+  Lock lock(test.mutex);  // <--- Static analysis error without this.
+  // The wake only happens when the predicate returns true.
+  ASSERT_EQ(test.rando_ivar, 12u);
+}
+
 }  // namespace testing
 }  // namespace impeller

impeller/base/thread.h

@@ -4,6 +4,9 @@
 
 #pragma once
 
+#include <chrono>
+#include <condition_variable>
+#include <functional>
 #include <memory>
 #include <mutex>
 #include <thread>
@@ -14,6 +17,8 @@
 
 namespace impeller {
 
+class ConditionVariable;
+
 class IPLR_CAPABILITY("mutex") Mutex {
  public:
   Mutex() = default;
@@ -25,6 +30,8 @@ class IPLR_CAPABILITY("mutex") Mutex {
   void Unlock() IPLR_RELEASE() { mutex_.unlock(); }
 
  private:
+  friend class ConditionVariable;
+
   std::mutex mutex_;
 
   Mutex(const Mutex&) = delete;
@@ -124,4 +131,134 @@ class IPLR_SCOPED_CAPABILITY WriterLock {
   WriterLock& operator=(WriterLock&&) = delete;
 };
 
+//------------------------------------------------------------------------------
+/// @brief      A condition variable exactly similar to the one in libcxx with
+///             two major differences:
+///
+///             * On the Wait, WaitFor, and WaitUntil calls, static analysis
+///               annotation are respected.
+///             * There is no ability to wait on a condition variable and also
+///               be susceptible to spurious wakes. This is because the
+///               predicate is mandatory.
+///
+class ConditionVariable {
+ public:
+  ConditionVariable() = default;
+
+  ~ConditionVariable() = default;
+
+  ConditionVariable(const ConditionVariable&) = delete;
+
+  ConditionVariable& operator=(const ConditionVariable&) = delete;
+
+  void NotifyOne() { cv_.notify_one(); }
+
+  void NotifyAll() { cv_.notify_all(); }
+
+  using Predicate = std::function<bool()>;
+
+  //----------------------------------------------------------------------------
+  /// @brief      Atomically unlocks the mutex and waits on the condition
+  ///             variable up to a specified time point. Spurious wakes may
+  ///             happen before the time point is reached. In such cases the
+  ///             predicate is invoked and it must return `false` for the wait
+  ///             to continue. The predicate will be invoked with the mutex
+  ///             locked.
+  ///
+  /// @note       Since the predicate is invoked with the mutex locked, if it
+  ///             accesses other guarded resources, the predicate itself must be
+  ///             decorated with the IPLR_REQUIRES directive. For instance,
+  ///
+  ///             ```c++
+  ///                [] () IPLR_REQUIRES(mutex) {
+  ///                return my_guarded_resource.should_stop_waiting;
+  ///              }
+  ///              ```
+  ///
+  /// @param      mutex                The mutex.
+  /// @param[in]  time_point           The time point to wait to.
+  /// @param[in]  should_stop_waiting  The predicate invoked on spurious wakes.
+  ///                                  Must return false for the wait to
+  ///                                  continue.
+  ///
+  /// @tparam     Clock                The clock type.
+  /// @tparam     Duration             The duration type.
+  ///
+  /// @return     The value of the predicate at the end of the wait.
+  ///
+  template <class Clock, class Duration>
+  bool WaitUntil(Mutex& mutex,
+                 const std::chrono::time_point<Clock, Duration>& time_point,
+                 const Predicate& should_stop_waiting) IPLR_REQUIRES(mutex) {
+    std::unique_lock lock(mutex.mutex_, std::adopt_lock);
+    return cv_.wait_until(lock, time_point, should_stop_waiting);
+  }
+
+  //----------------------------------------------------------------------------
+  /// @brief      Atomically unlocks the mutex and waits on the condition
+  ///             variable for a designated duration. Spurious wakes may happen
+  ///             before the time point is reached. In such cases the predicate
+  ///             is invoked and it must return `false` for the wait to
+  ///             continue. The predicate will be invoked with the mutex locked.
+  ///
+  /// @note       Since the predicate is invoked with the mutex locked, if it
+  ///             accesses other guarded resources, the predicate itself must be
+  ///             decorated with the IPLR_REQUIRES directive. For instance,
+  ///
+  ///             ```c++
+  ///                [] () IPLR_REQUIRES(mutex) {
+  ///                return my_guarded_resource.should_stop_waiting;
+  ///              }
+  ///              ```
+  ///
+  /// @param      mutex                The mutex.
+  /// @param[in]  duration             The duration to wait for.
+  /// @param[in]  should_stop_waiting  The predicate invoked on spurious wakes.
+  ///                                  Must return false for the wait to
+  ///                                  continue.
+  ///
+  /// @tparam     Representation       The duration representation type.
+  /// @tparam     Period               The duration period type.
+  ///
+  /// @return     The value of the predicate at the end of the wait.
+  ///
+  template <class Representation, class Period>
+  bool WaitFor(Mutex& mutex,
+               const std::chrono::duration<Representation, Period>& duration,
+               const Predicate& should_stop_waiting) IPLR_REQUIRES(mutex) {
+    return WaitUntil(mutex, std::chrono::steady_clock::now() + duration,
+                     should_stop_waiting);
+  }
+
+  //----------------------------------------------------------------------------
+  /// @brief      Atomically unlocks the mutex and waits on the condition
+  ///             variable indefinitely till the predicate determines that the
+  ///             wait must end. Spurious wakes may happen before the time point
+  ///             is reached. In such cases the predicate is invoked and it must
+  ///             return `false` for the wait to continue. The predicate will be
+  ///             invoked with the mutex locked.
+  ///
+  /// @note       Since the predicate is invoked with the mutex locked, if it
+  ///             accesses other guarded resources, the predicate itself must be
+  ///             decorated with the IPLR_REQUIRES directive. For instance,
+  ///
+  ///             ```c++
+  ///                [] () IPLR_REQUIRES(mutex) {
+  ///                return my_guarded_resource.should_stop_waiting;
+  ///              }
+  ///              ```
+  ///
+  /// @param      mutex                The mutex
+  /// @param[in]  should_stop_waiting  The should stop waiting
+  ///
+  void Wait(Mutex& mutex, const Predicate& should_stop_waiting)
+      IPLR_REQUIRES(mutex) {
+    std::unique_lock lock(mutex.mutex_, std::adopt_lock);
+    cv_.wait(lock, should_stop_waiting);
+  }
+
+ private:
+  std::condition_variable cv_;
+};
+
 }  // namespace impeller