modify interface of sync.condition

Author: mutouyun

include/libipc/condition.h

@@ -27,8 +27,8 @@ class IPC_EXPORT condition {
     void close() noexcept;
 
     bool wait(ipc::sync::mutex &mtx, std::uint64_t tm = ipc::invalid_value) noexcept;
-    bool notify() noexcept;
-    bool broadcast() noexcept;
+    bool notify(ipc::sync::mutex &mtx) noexcept;
+    bool broadcast(ipc::sync::mutex &mtx) noexcept;
 
 private:
     class condition_;

src/libipc/platform/linux/condition.h

@@ -115,7 +115,7 @@ class condition {
         return true;
     }
 
-    bool notify() noexcept {
+    bool notify(ipc::sync::mutex &) noexcept {
         if (!valid()) return false;
         int eno;
         if ((eno = ::pthread_cond_signal(cond_)) != 0) {
@@ -125,7 +125,7 @@ class condition {
         return true;
     }
 
-    bool broadcast() noexcept {
+    bool broadcast(ipc::sync::mutex &) noexcept {
         if (!valid()) return false;
         int eno;
         if ((eno = ::pthread_cond_broadcast(cond_)) != 0) {

src/libipc/platform/win/condition.h

@@ -89,7 +89,7 @@ class condition {
         return rs && rl;
     }
 
-    bool notify() noexcept {
+    bool notify(ipc::sync::mutex &) noexcept {
         if (!valid()) return false;
         auto &cnt = counter();
         if (!lock_.lock()) return false;
@@ -101,7 +101,7 @@ class condition {
         return lock_.unlock() && ret;
     }
 
-    bool broadcast() noexcept {
+    bool broadcast(ipc::sync::mutex &) noexcept {
         if (!valid()) return false;
         auto &cnt = counter();
         if (!lock_.lock()) return false;

src/libipc/sync/condition.cpp

@@ -58,12 +58,12 @@ bool condition::wait(ipc::sync::mutex &mtx, std::uint64_t tm) noexcept {
     return impl(p_)->cond_.wait(mtx, tm);
 }
 
-bool condition::notify() noexcept {
-    return impl(p_)->cond_.notify();
+bool condition::notify(ipc::sync::mutex &mtx) noexcept {
+    return impl(p_)->cond_.notify(mtx);
 }
 
-bool condition::broadcast() noexcept {
-    return impl(p_)->cond_.broadcast();
+bool condition::broadcast(ipc::sync::mutex &mtx) noexcept {
+    return impl(p_)->cond_.broadcast(mtx);
 }
 
 } // namespace sync

src/libipc/waiter.h

@@ -63,12 +63,12 @@ class waiter {
 
     bool notify() noexcept {
         std::lock_guard<ipc::sync::mutex>{lock_}; // barrier
-        return cond_.notify();
+        return cond_.notify(lock_);
     }
 
     bool broadcast() noexcept {
         std::lock_guard<ipc::sync::mutex>{lock_}; // barrier
-        return cond_.broadcast();
+        return cond_.broadcast(lock_);
     }
 
     bool quit_waiting() {

test/profiler/README.md

@@ -113,7 +113,7 @@ TEST(Sync, Condition) {
     auto job = [&que](int num) {
         ipc::sync::condition cond {"test-cond"};
         ipc::sync::mutex lock {"test-mutex"};
-        for (;;) {
+        for (int i = 0; i < 10; ++i) {
             int val = 0;
             {
                 std::lock_guard<ipc::sync::mutex> guard {lock};
@@ -123,6 +123,19 @@ TEST(Sync, Condition) {
                 val = que.front();
                 que.pop_front();
             }
+            EXPECT_NE(val, 0);
+            std::printf("test-cond-%d: %d\n", num, val);
+        }
+        for (;;) {
+            int val = 0;
+            {
+                std::lock_guard<ipc::sync::mutex> guard {lock};
+                while (que.empty()) {
+                    EXPECT_TRUE(cond.wait(lock, 1000));
+                }
+                val = que.front();
+                que.pop_front();
+            }
             if (val == 0) {
                 std::printf("test-cond-%d: exit.\n", num);
                 return;
@@ -139,25 +152,25 @@ TEST(Sync, Condition) {
         {
             std::lock_guard<ipc::sync::mutex> guard {lock};
             que.push_back(i);
+            ASSERT_TRUE(cond.notify(lock));
         }
-        cond.notify();
         std::this_thread::sleep_for(std::chrono::milliseconds(20));
     }
     for (int i = 1; i < 100; ++i) {
         {
             std::lock_guard<ipc::sync::mutex> guard {lock};
             que.push_back(i);
+            ASSERT_TRUE(cond.broadcast(lock));
         }
-        cond.broadcast();
         std::this_thread::sleep_for(std::chrono::milliseconds(20));
     }
     {
         std::lock_guard<ipc::sync::mutex> guard {lock};
         for (int i = 0; i < (int)test_conds.size(); ++i) {
             que.push_back(0);
         }
+        ASSERT_TRUE(cond.broadcast(lock));
     }
-    cond.broadcast();
 
     for (auto &t : test_conds) t.join();
 }
@@ -171,7 +184,7 @@ TEST(Sync, ConditionRobust) {
     printf("XXXXXXXXXXXXXXXXXXXXXXXXXXXXXX 2\n");
     ipc::sync::mutex lock {"test-mutex"};
     printf("XXXXXXXXXXXXXXXXXXXXXXXXXXXXXX 3\n");
-    lock.lock();
+    ASSERT_TRUE(lock.lock());
     std::thread unlock {[] {
         printf("WWWWWWWWWWWWWWWWWWWWWWWWWWWWWW 1\n");
         ipc::sync::condition cond {"test-cond"};
@@ -183,13 +196,13 @@ TEST(Sync, ConditionRobust) {
         }
         std::this_thread::sleep_for(std::chrono::seconds(1));
         printf("WWWWWWWWWWWWWWWWWWWWWWWWWWWWWW 4\n");
-        cond.broadcast();
+        ASSERT_TRUE(cond.broadcast(lock));
         printf("WWWWWWWWWWWWWWWWWWWWWWWWWWWWWW 5\n");
     }};
     printf("XXXXXXXXXXXXXXXXXXXXXXXXXXXXXX 4\n");
-    cond.wait(lock);
+    ASSERT_TRUE(cond.wait(lock));
     printf("XXXXXXXXXXXXXXXXXXXXXXXXXXXXXX 5\n");
-    lock.unlock();
+    ASSERT_TRUE(lock.unlock());
     printf("XXXXXXXXXXXXXXXXXXXXXXXXXXXXXX 6\n");
     unlock.join();
 }