From a6d88a12087d8852bc7de5d2c05d438af5c2fd65 Mon Sep 17 00:00:00 2001
From: mutouyun <orz@orzz.org>
Date: Tue, 12 Feb 2019 23:28:57 +0800
Subject: [PATCH] ReleaseSemaphore mayn't wake WaitForSingleObject up soon

---
 src/platform/detail.h     |  3 ++-
 src/platform/waiter_win.h | 28 +++++++++++++++-------------
 test/test_ipc.cpp         | 19 ++++++++++++-------
 3 files changed, 29 insertions(+), 21 deletions(-)

diff --git a/src/platform/detail.h b/src/platform/detail.h
index 5a859bf..54312f8 100644
--- a/src/platform/detail.h
+++ b/src/platform/detail.h
@@ -8,6 +8,7 @@
 #include <atomic>
 
 #include "def.h"
+#include "export.h"
 
 // pre-defined
 
@@ -82,7 +83,7 @@ constexpr auto shared_lock(T&& lc) {
 
 #endif/*__cplusplus < 201703L*/
 
-std::size_t calc_unique_id();
+IPC_EXPORT std::size_t calc_unique_id();
 
 template <typename F, typename D>
 constexpr decltype(auto) static_switch(std::size_t /*i*/, std::index_sequence<>, F&& /*f*/, D&& def) {
diff --git a/src/platform/waiter_win.h b/src/platform/waiter_win.h
index 1e9ed57..8715ace 100644
--- a/src/platform/waiter_win.h
+++ b/src/platform/waiter_win.h
@@ -2,11 +2,7 @@
 
 #include <Windows.h>
 
-#include <algorithm>
-#include <iterator>
 #include <atomic>
-#include <array>
-#include <tuple>
 
 #include "rw_lock.h"
 #include "pool_alloc.h"
@@ -51,19 +47,25 @@ public:
 
     void notify(handle_t h) {
         if (h == invalid()) return;
-        IPC_UNUSED_ auto guard = ipc::detail::unique_lock(lock_);
-        if (counter_ == 0) return;
-        -- counter_;
-        ::ReleaseSemaphore(h, 1, NULL);
+        {
+            IPC_UNUSED_ auto guard = ipc::detail::unique_lock(lock_);
+            if (counter_ == 0) return;
+            -- counter_;
+            ::ReleaseSemaphore(h, 1, NULL);
+        }
+        ::Sleep(1);
     }
 
     void broadcast(handle_t h) {
         if (h == invalid()) return;
-        IPC_UNUSED_ auto guard = ipc::detail::unique_lock(lock_);
-        if (counter_ == 0) return;
-        long all_count = counter_;
-        counter_ = 0;
-        ::ReleaseSemaphore(h, all_count, NULL);
+        {
+            IPC_UNUSED_ auto guard = ipc::detail::unique_lock(lock_);
+            if (counter_ == 0) return;
+            long all_count = counter_;
+            counter_ = 0;
+            ::ReleaseSemaphore(h, all_count, NULL);
+        }
+        ::Sleep(1);
     }
 };
 
diff --git a/test/test_ipc.cpp b/test/test_ipc.cpp
index 7295bab..f4c20e2 100644
--- a/test/test_ipc.cpp
+++ b/test/test_ipc.cpp
@@ -368,7 +368,7 @@ void Unit::test_route_rtt() {
         for (std::size_t i = 0;; ++i) {
             auto dd = cc.recv();
             if (dd.size() < 2) return;
-//            std::cout << "recving: " << i << "-[" << dd.size() << "]" << std::endl;
+            //std::cout << "recv: " << i << "-[" << dd.size() << "]" << std::endl;
             while (!cr.send(ipc::buff_t('a'))) {
                 std::this_thread::yield();
             }
@@ -383,8 +383,8 @@ void Unit::test_route_rtt() {
         }
         sw.start();
         for (std::size_t i = 0; i < LoopCount; ++i) {
-//            std::cout << "sending: " << i << "-[" << datas__[i].size() << "]" << std::endl;
             cc.send(datas__[i]);
+            //std::cout << "sent: " << i << "-[" << datas__[i].size() << "]" << std::endl;
             /*auto dd = */cr.recv();
 //            if (dd.size() != 1 || dd[0] != 'a') {
 //                QVERIFY(false);
@@ -434,14 +434,17 @@ void Unit::test_channel_rtt() {
 
     std::thread t1 {[&] {
         ipc::channel cc { "my-ipc-channel" };
+        bool recv_2 = false;
         for (std::size_t i = 0;; ++i) {
             auto dd = cc.recv();
             if (dd.size() < 2) return;
-            //std::cout << "recving: " << i << "-[" << dd.size() << "]" << std::endl;
-            while (!cc.send(ipc::buff_t('a'))) {
-                cc.wait_for_recv(1);
+            //if (i % 1000 == 0) {
+            //    std::cout << "recv: " << i << "-[" << dd.size() << "]" << std::endl;
+            //}
+            while (!recv_2) {
+                recv_2 = cc.wait_for_recv(2);
             }
-            //std::cout << "sent ack." << std::endl;
+            cc.send(ipc::buff_t('a'));
         }
     }};
 
@@ -450,7 +453,9 @@ void Unit::test_channel_rtt() {
         cc.wait_for_recv(1);
         sw.start();
         for (std::size_t i = 0; i < LoopCount; ++i) {
-            //std::cout << "sending: " << i << "-[" << datas__[i].size() << "]" << std::endl;
+            //if (i % 1000 == 0) {
+            //    std::cout << "send: " << i << "-[" << datas__[i].size() << "]" << std::endl;
+            //}
             cc.send(datas__[i]);
             /*auto dd = */cc.recv();
             //if (dd.size() != 1 || dd.data<char>()[0] != 'a') {