rx/SharedAtomic: fix win32 IPC implementation, wait & notify_one is only supported apis

rx/src/SharedAtomic.cpp

@@ -143,6 +143,27 @@ int shared_atomic32::notify_n(int count) const {
 #include <cmath>
 #include <windows.h>
 
+static auto g_events = [] {
+  std::array<HANDLE, 256> result;
+  SECURITY_ATTRIBUTES securityAttr{
+      .nLength = sizeof(SECURITY_ATTRIBUTES),
+      .lpSecurityDescriptor = nullptr,
+      .bInheritHandle = true,
+  };
+
+  for (auto &handle : result) {
+    handle = CreateEvent(&securityAttr, false, false, nullptr);
+  }
+  return result;
+}();
+
+static HANDLE getEventFor(const shared_atomic32 *atomic) {
+  auto hash = (2654404609 * std::bit_cast<std::uintptr_t>(atomic)) >> 26;
+  hash ^= hash >> 16;
+  hash ^= hash >> 8;
+  return g_events[hash & 0xff];
+}
+
 std::errc shared_atomic32::wait_impl(std::uint32_t oldValue,
                                      std::chrono::microseconds usec_timeout) {
 
@@ -151,57 +172,48 @@ std::errc shared_atomic32::wait_impl(std::uint32_t oldValue,
   bool unblock = (!useTimeout || usec_timeout.count() > 1000) &&
                  g_scopedUnblock != nullptr;
 
+  auto event = getEventFor(this);
+
   if (unblock) {
     if (!g_scopedUnblock(true)) {
       return std::errc::interrupted;
     }
   }
 
-  BOOL result = WaitOnAddress(
+  if (load(std::memory_order::relaxed) != oldValue) {
-      this, &oldValue, sizeof(std::uint32_t),
+    return {};
-      useTimeout
-          ? std::chrono::duration_cast<std::chrono::milliseconds>(usec_timeout)
-                .count()
-          : INFINITY);
-
-  DWORD error = 0;
-  if (!result) {
-    error = GetLastError();
-  } else {
-    if (load(std::memory_order::relaxed) == oldValue) {
-      error = ERROR_ALERTED; // dummy error
-    }
   }
 
-  if (unblock) {
+  auto timeoutMs = INFINITE;
-    if (!g_scopedUnblock(false)) {
-      if (result != TRUE) {
-        return std::errc::interrupted;
-      }
 
-      return {};
+  if (useTimeout) {
-    }
+    timeoutMs =
+        std::chrono::duration_cast<std::chrono::milliseconds>(usec_timeout)
+            .count();
   }
 
-  if (error == ERROR_TIMEOUT) {
+  auto result = WaitForSingleObject(event, timeoutMs);
+
+  bool unblockInterrupted = unblock && !g_scopedUnblock(false);
+
+  if (result == WAIT_OBJECT_0) {
+    return {};
+  }
+
+  if (result == WAIT_TIMEOUT) {
     return std::errc::timed_out;
   }
 
+  if (unblockInterrupted) {
+    return std::errc::interrupted;
+  }
+
   return std::errc::resource_unavailable_try_again;
 }
 
 int shared_atomic32::notify_n(int count) const {
-  if (count == 1) {
+  SetEvent(getEventFor(this));
-    WakeByAddressSingle(const_cast<shared_atomic32 *>(this));
+  return 1;
-  } else if (count == std::numeric_limits<int>::max()) {
-    WakeByAddressAll(const_cast<shared_atomic32 *>(this));
-  } else {
-    for (int i = 0; i < count; ++i) {
-      WakeByAddressSingle(const_cast<shared_atomic32 *>(this));
-    }
-  }
-
-  return 1; // FIXME
 }
 #else
 #error Unimplemented atomic for this platform