Fix potential race in dispatcher wakeup

src/hydra-queue-runner/dispatcher.cc

@@ -36,11 +36,14 @@ void State::dispatcher()
         /* Sleep until we're woken up (either because a runnable build
            is added, or because a build finishes). */
         {
-            std::unique_lock<std::mutex> lock(dispatcherMutex);
-            printMsg(lvlDebug, format("dispatcher sleeping for %1%s") %
-                std::chrono::duration_cast<std::chrono::seconds>(sleepUntil - std::chrono::system_clock::now()).count());
-            dispatcherWakeup.wait_until(lock, sleepUntil);
+            auto dispatcherWakeup_(dispatcherWakeup.lock());
+            if (!*dispatcherWakeup_) {
+                printMsg(lvlDebug, format("dispatcher sleeping for %1%s") %
+                    std::chrono::duration_cast<std::chrono::seconds>(sleepUntil - std::chrono::system_clock::now()).count());
+                dispatcherWakeup_.wait_until(dispatcherWakeupCV, sleepUntil);
+            }
             nrDispatcherWakeups++;
+            *dispatcherWakeup_ = false;
         }
     }
 
@@ -167,6 +170,9 @@ system_time State::doDispatch()
 
 void State::wakeDispatcher()
 {
-    { std::lock_guard<std::mutex> lock(dispatcherMutex); } // barrier
-    dispatcherWakeup.notify_one();
+    {
+        auto dispatcherWakeup_(dispatcherWakeup.lock());
+        *dispatcherWakeup_ = true;
+    }
+    dispatcherWakeupCV.notify_one();
 }

src/hydra-queue-runner/state.hh

@@ -202,8 +202,8 @@ private:
     Sync<Runnable> runnable;
 
     /* CV for waking up the dispatcher. */
-    std::condition_variable dispatcherWakeup;
-    std::mutex dispatcherMutex;
+    Sync<bool> dispatcherWakeup;
+    std::condition_variable_any dispatcherWakeupCV;
 
     /* PostgreSQL connection pool. */
     Pool<Connection> dbPool;

src/hydra-queue-runner/sync.hh

@@ -60,6 +60,14 @@ public:
             assert(s);
             return cv.wait_for(s->mutex, duration, pred);
         }
+
+        template<class Clock, class Duration>
+        std::cv_status wait_until(std::condition_variable_any & cv,
+            const std::chrono::time_point<Clock, Duration> & duration)
+        {
+            assert(s);
+            return cv.wait_until(s->mutex, duration);
+        }
     };
 
     Lock lock() { return Lock(this); }