impl quit_waiting

Author: mutouyun

src/libipc/ipc.cpp

@@ -282,9 +282,9 @@ struct conn_info_head {
     }
 
     void quit_waiting() {
-        // cc_waiter_.quit_waiting();
-        // wt_waiter_.quit_waiting();
-        // rd_waiter_.quit_waiting();
+        cc_waiter_.quit_waiting();
+        wt_waiter_.quit_waiting();
+        rd_waiter_.quit_waiting();
     }
 
     auto acc() {

src/libipc/waiter.h

@@ -3,6 +3,7 @@
 #include <utility>
 #include <string>
 #include <mutex>
+#include <atomic>
 
 #include "libipc/def.h"
 #include "libipc/mutex.h"
@@ -15,6 +16,7 @@ namespace detail {
 class waiter {
     ipc::sync::condition cond_;
     ipc::sync::mutex     lock_;
+    std::atomic<bool>    quit_ {false};
 
 public:
     waiter() = default;
@@ -31,6 +33,7 @@ class waiter {
     }
 
     bool open(char const *name) noexcept {
+        quit_.store(false, std::memory_order_relaxed);
         if (!cond_.open((std::string{"_waiter_cond_"} + name).c_str())) {
             return false;
         }
@@ -49,7 +52,10 @@ class waiter {
     template <typename F>
     bool wait_if(F &&pred, std::uint64_t tm = ipc::invalid_value) noexcept {
         IPC_UNUSED_ std::lock_guard<ipc::sync::mutex> guard {lock_};
-        while (std::forward<F>(pred)()) {
+        while ([this, &pred] {
+                    return !quit_.load(std::memory_order_relaxed)
+                        && std::forward<F>(pred)();
+                }()) {
             if (!cond_.wait(lock_, tm)) return false;
         }
         return true;
@@ -65,7 +71,9 @@ class waiter {
         return cond_.broadcast();
     }
 
-    void quit_waiting() {
+    bool quit_waiting() {
+        quit_.store(true, std::memory_order_release);
+        return broadcast();
     }
 };
 

test/test_waiter.cpp

@@ -35,6 +35,34 @@ TEST(Waiter, broadcast) {
 }
 
 TEST(Waiter, quit_waiting) {
+    ipc::detail::waiter waiter;
+    EXPECT_TRUE(waiter.open("test-ipc-waiter"));
+
+    std::thread t1 {
+        [&waiter] {
+            EXPECT_TRUE(waiter.wait_if([] { return true; }));
+        }
+    };
+
+    bool quit = false;
+    std::thread t2 {
+        [&quit] {
+            ipc::detail::waiter waiter {"test-ipc-waiter"};
+            EXPECT_TRUE(waiter.wait_if([&quit] { return !quit; }));
+        }
+    };
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    EXPECT_TRUE(waiter.quit_waiting());
+    t1.join();
+    ASSERT_TRUE(t2.joinable());
+
+    EXPECT_TRUE(waiter.open("test-ipc-waiter"));
+    std::cout << "nofify quit...\n";
+    quit = true;
+    EXPECT_TRUE(waiter.notify());
+    t2.join();
+    std::cout << "quit... \n";
 }
 
 } // internal-linkage