latch: Extend with wait_for() and arrive_and_wait_for(), i.e. add variants with timeout duration value

2 files changed, 142 insertions, 4 deletions

diff --git a/include/jau/latch.hpp b/include/jau/latch.hpp
index 16b3e9a..ee26393 100644
--- a/include/jau/latch.hpp
+++ b/include/jau/latch.hpp
@@ -49,11 +49,30 @@ namespace jau {
             /** Returns the maximum value of the internal counter supported by the implementation. */
             static constexpr size_t max() noexcept { return std::numeric_limits<size_t>::max(); }
 
+            /**
+             * Initialize instance with given counter.
+             *
+             * Compatible with std::latch.
+             *
+             * @param count_
+             */
             latch(const size_t count_) noexcept
             : count(count_) {}
 
+            /**
+             * No copy constructor nor move constructor.
+             *
+             * Compatible with std::latch.
+             *
+             * @param o
+             */
             latch(const latch& o) = delete;
 
+            /**
+             * Return the current atomic internal counter.
+             *
+             * Extension of std::latch.
+             */
             size_t value() const noexcept { return count; }
 
             /**
@@ -64,6 +83,8 @@ namespace jau {
              *
              * This operation strongly happens-before all calls that are unblocked on this latch.
              *
+             * Compatible with std::latch.
+             *
              * @param n the value by which the internal counter is decreased, defaults to 1
              */
             void count_down(const size_t n = 1) noexcept {
@@ -85,13 +106,19 @@ namespace jau {
 
             /**
              * Returns true only if the internal counter has reached zero.
+             *
+             * Compatible with std::latch.
              */
             bool try_wait() const noexcept {
                 return 0 == count;
             }
 
             /**
-             * Blocks the calling thread until the internal counter reaches 0. If it is zero already, returns immediately.
+             * Blocks the calling thread until the internal counter reaches 0.
+             *
+             * If the internal counter is zero already, returns immediately.
+             *
+             * Compatible with std::latch.
              */
             void wait() const noexcept {
                 if( 0 < count ) {
@@ -105,7 +132,9 @@ namespace jau {
             /**
              * Atomically decrements the internal counter by n and (if necessary) blocks the calling thread until the counter reaches zero.
              *
-             * Equivalent to count_down(n); wait();.
+             * Equivalent to `count_down(n); wait();`.
+             *
+             * Compatible with std::latch.
              *
              * @param n the value by which the internal counter is decreased, defaults to 1
              */
@@ -113,6 +142,91 @@ namespace jau {
                 count_down(n);
                 wait();
             }
+
+            /**
+             * Blocks the calling thread until the internal counter reaches 0 or the given timeout duration has expired.
+             *
+             * If the internal counter is zero already, returns immediately.
+             *
+             * Implementation uses `std::chrono::steady_clock::now()`.
+             *
+             * Extension of std::latch.
+             *
+             * @tparam Rep
+             * @tparam Period
+             * @param timeout_duration maximum time duration to spend waiting
+             * @return true if internal counter has reached zero, otherwise a timeout has occurred.
+             */
+            template<typename Rep, typename Period>
+            bool wait_for(const std::chrono::duration<Rep, Period>& timeout_duration) const noexcept {
+                if( 0 < count ) {
+                    std::unique_lock<std::mutex> lock(mtx_cv);
+                    while( 0 < count ) {
+                        std::chrono::steady_clock::time_point t0 = std::chrono::steady_clock::now();
+                        std::cv_status s = cv.wait_until(lock, t0 + timeout_duration);
+                        if( std::cv_status::timeout == s && 0 < count ) {
+                            return false;
+                        }
+                    }
+                }
+                return true;
+            }
+
+            /**
+             * Blocks the calling thread until the internal counter reaches 0 or the given timeout duration has expired.
+             *
+             * If the internal counter is zero already, returns immediately.
+             *
+             * Implementation uses `std::chrono::steady_clock::now()`.
+             *
+             * Extension of std::latch.
+             *
+             * @param timeout_ms maximum time duration to spend waiting in milliseconds
+             * @return true if internal counter has reached zero, otherwise a timeout has occurred.
+             */
+            bool wait_for(const size_t timeout_ms) const noexcept {
+                return wait_for(std::chrono::milliseconds(timeout_ms));
+            }
+
+            /**
+             * Atomically decrements the internal counter by n and (if necessary) blocks the calling thread until the counter reaches zero
+             * or the given timeout duration has expired.
+             *
+             * Equivalent to `count_down(n); wait(timeout_duration);`.
+             *
+             * Implementation uses `std::chrono::steady_clock::now()`.
+             *
+             * Extension of std::latch.
+             *
+             * @tparam Rep
+             * @tparam Period
+             * @param timeout_duration maximum time duration to spend waiting
+             * @param n the value by which the internal counter is decreased, defaults to 1
+             * @return true if internal counter has reached zero, otherwise a timeout has occurred.
+             */
+            template<typename Rep, typename Period>
+            bool arrive_and_wait_for(const std::chrono::duration<Rep, Period>& timeout_duration, const size_t n = 1) noexcept {
+                count_down(n);
+                return wait_for(timeout_duration);
+            }
+
+            /**
+             * Atomically decrements the internal counter by n and (if necessary) blocks the calling thread until the counter reaches zero
+             * or the given timeout duration has expired.
+             *
+             * Equivalent to `count_down(n); wait(timeout_duration);`.
+             *
+             * Implementation uses `std::chrono::steady_clock::now()`.
+             *
+             * Extension of std::latch.
+             *
+             * @param timeout_ms maximum time duration to spend waiting in milliseconds
+             * @param n the value by which the internal counter is decreased, defaults to 1
+             * @return true if internal counter has reached zero, otherwise a timeout has occurred.
+             */
+            bool arrive_and_wait_for(const size_t timeout_ms, const size_t n = 1) noexcept {
+                return arrive_and_wait_for(std::chrono::milliseconds(timeout_ms), n);
+            }
     };
 
 } /* namespace jau */
diff --git a/test/test_latch01.cpp b/test/test_latch01.cpp
index 0edaae5..d4bd2bb 100644
--- a/test/test_latch01.cpp
+++ b/test/test_latch01.cpp
@@ -48,7 +48,7 @@ class TestLatch01 {
 
   public:
 
-    void test01() {
+    void test01_wait() {
         INFO_STR("\n\ntest01\n");
         const size_t count = 8;
         std::thread tasks[count];
@@ -70,7 +70,31 @@ class TestLatch01 {
             }
         }
     }
+
+    void test02_wait_for() {
+        INFO_STR("\n\ntest02\n");
+        const size_t count = 8;
+        std::thread tasks[count];
+        jau::latch completion(count+1);
+
+        REQUIRE_MSG("not-zero", count+1 == completion.value());
+
+        for(size_t i=0; i<count; i++) {
+            tasks[i] = std::thread(&TestLatch01::something, this, std::ref(completion));
+        }
+        REQUIRE_MSG("complete", true == completion.arrive_and_wait_for(10000 /* timeout_ms */) );
+
+        REQUIRE_MSG("zero", 0 == completion.value());
+        REQUIRE_MSG("8", count == my_counter);
+
+        for(size_t i=0; i<count; i++) {
+            if( tasks[i].joinable() ) {
+                tasks[i].join();
+            }
+        }
+    }
 };
 
-METHOD_AS_TEST_CASE( TestLatch01::test01,      "Test TestLatch01");
+METHOD_AS_TEST_CASE( TestLatch01::test01_wait,      "Test TestLatch01 - test01_wait");
+METHOD_AS_TEST_CASE( TestLatch01::test02_wait_for,  "Test TestLatch01 - test02_wait_for");