[dxvk] Rework SetEventOnCompletion

* Create the waiter thread on demand
* Don't wake up the waiter thread every 10ms
  when no events are queued
* Wait on caller thread when hEvent = null

src/d3d11/d3d11_fence.cpp

@@ -81,13 +81,13 @@ namespace dxvk {
   HRESULT STDMETHODCALLTYPE D3D11Fence::SetEventOnCompletion(
           UINT64              Value,
           HANDLE              hEvent) {
-    // TODO in case of rewinds, the stored value may be higher.
+    if (hEvent) {
-    // For shared fences, calling vkWaitSemaphores here could alleviate the issue.
+      m_fence->enqueueWait(Value, [hEvent] {
-
+        SetEvent(hEvent);
-    m_fence->enqueueWait(Value, [hEvent] {
+      });
-      SetEvent(hEvent);
+    } else {
-    });
+      m_fence->wait(Value);
-
+    }
     return S_OK;
   }
 

src/dxvk/dxvk_fence.cpp

@@ -56,28 +56,51 @@ namespace dxvk {
         Logger::warn(str::format("Importing semaphores of type ", info.sharedType, " not supported by device"));
       }
     }
-
-    m_thread = dxvk::thread([this] { run(); });
   }
 
 
   DxvkFence::~DxvkFence() {
-    m_stop.store(true);
+    if (m_thread.joinable()) {
-    m_thread.join();
+      {
-
+        std::unique_lock<dxvk::mutex> lock(m_mutex);
+        m_running = false;
+        m_condVar.notify_one();
+      }
+      m_thread.join();
+    }
     m_vkd->vkDestroySemaphore(m_vkd->device(), m_semaphore, nullptr);
   }
 
 
   void DxvkFence::enqueueWait(uint64_t value, DxvkFenceEvent&& event) {
-    std::unique_lock<dxvk::mutex> lock(m_mutex);
+    if (value > getValue()) {
-
+      std::unique_lock<dxvk::mutex> lock(m_mutex);
-    if (value > m_lastValue.load())
       m_queue.emplace(value, std::move(event));
-    else
+
+      if (!m_running) {
+        m_running = true;
+        m_thread = dxvk::thread([this] { run(); });
+      } else {
+        m_condVar.notify_one();
+      }
+      lock.unlock();
+    } else {
       event();
+    }
+  }
+
+  void DxvkFence::wait(uint64_t value) {
+    VkSemaphoreWaitInfo waitInfo = { VK_STRUCTURE_TYPE_SEMAPHORE_WAIT_INFO };
+    waitInfo.semaphoreCount = 1;
+    waitInfo.pSemaphores = &m_semaphore;
+    waitInfo.pValues = &value;
+    VkResult vr = m_vkd->vkWaitSemaphores(
+      m_vkd->device(), &waitInfo, ~0ull);
+
+    if (vr != VK_SUCCESS) {
+      Logger::err(str::format("Failed to wait for semaphore: ", vr));
+    }
   }
-  
 
   void DxvkFence::run() {
     uint64_t value = 0ull;
@@ -87,9 +110,11 @@ namespace dxvk {
     waitInfo.pSemaphores = &m_semaphore;
     waitInfo.pValues = &value;
 
-    while (!m_stop.load()) {
+    while (true) {
       std::unique_lock<dxvk::mutex> lock(m_mutex);
 
+      m_condVar.wait(lock, [&]() { return !m_queue.empty() || !m_running; });
+
       // Query actual semaphore value and start from there, so that
       // we can skip over large increments in the semaphore value
       VkResult vr = m_vkd->vkGetSemaphoreCounterValue(m_vkd->device(), m_semaphore, &value);
@@ -99,8 +124,6 @@ namespace dxvk {
         return;
       }
 
-      m_lastValue.store(value);
-
       // Signal all enqueued events whose value is not greater than
       // the current semaphore value
       while (!m_queue.empty() && m_queue.top().value <= value) {
@@ -108,9 +131,12 @@ namespace dxvk {
         m_queue.pop();
       }
 
-      if (m_stop)
+      if (!m_running)
         return;
 
+      if (m_queue.empty())
+        continue;
+
       lock.unlock();
 
       // Wait for the semaphore to be singaled again and update state.
@@ -130,6 +156,15 @@ namespace dxvk {
     }
   }
 
+  uint64_t DxvkFence::getValue() {
+    uint64_t value = 0;
+    VkResult vr = m_vkd->vkGetSemaphoreCounterValue(m_vkd->device(), m_semaphore, &value);
+    if (vr != VK_SUCCESS) {
+      Logger::err(str::format("Failed to query semaphore value: ", vr));
+    }
+    return value;
+  }
+
   HANDLE DxvkFence::sharedHandle() const {
     if (m_info.sharedType == VK_EXTERNAL_SEMAPHORE_HANDLE_TYPE_FLAG_BITS_MAX_ENUM)
       return INVALID_HANDLE_VALUE;

src/dxvk/dxvk_fence.h

@@ -70,9 +70,7 @@ namespace dxvk {
      * \brief Retrieves current semaphore value
      * \returns Current semaphore value
      */
-    uint64_t getValue() {
+    uint64_t getValue();
-      return m_lastValue.load();
-    }
 
     /**
      * \brief Enqueues semaphore wait
@@ -90,6 +88,15 @@ namespace dxvk {
      */
     HANDLE sharedHandle() const;
 
+    /*
+     * \brief Waits for the given value
+     *
+     * Blocks the calling thread until
+     * the fence reaches the given value.
+     * \param [in] value Value to wait for
+    */
+    void wait(uint64_t value);
+
   private:
 
     struct QueueItem {
@@ -113,10 +120,10 @@ namespace dxvk {
     VkSemaphore                     m_semaphore;
 
     std::priority_queue<QueueItem>  m_queue;
-    std::atomic<uint64_t>           m_lastValue = { 0ull  };
+    bool                            m_running    = false;
-    std::atomic<bool>               m_stop      = { false };
 
     dxvk::mutex                     m_mutex;
+    dxvk::condition_variable        m_condVar;
     dxvk::thread                    m_thread;
 
     void run();