Add tests for precision of condition timeouts.

And it turns out that the Windows implementation was returning too early due to rounding error. Fixed.

c++/src/kj/mutex-test.c++

@@ -26,6 +26,8 @@
 #define NOGDI  // NOGDI is needed to make EXPECT_EQ(123u, *lock) compile for some reason
 #endif
 
+#define KJ_MUTEX_TEST 1
+
 #include "mutex.h"
 #include "debug.h"
 #include "thread.h"
@@ -369,6 +371,63 @@ TEST(Mutex, WhenWithTimeout) {
 #endif
 }
 
+TEST(Mutex, WhenWithTimeoutPreciseTiming) {
+  // Test that MutexGuarded::when() with a timeout sleeps for precisely the right amount of time.
+
+  for (uint retryCount = 0; retryCount < 20; retryCount++) {
+    MutexGuarded<uint> value(123);
+
+    auto start = now();
+    uint m = value.when([&value](uint n) {
+      // HACK: Reset the value as a way of testing what happens when the waiting thread is woken
+      //   up but then finds it's not ready yet.
+      value.getWithoutLock() = 123;
+      return n == 321;
+    }, [](uint& n) {
+      return 456;
+    }, 20 * kj::MILLISECONDS);
+
+    KJ_EXPECT(m == 456);
+
+    auto t = now() - start;
+    KJ_EXPECT(t >= 20 * kj::MILLISECONDS);
+    if (t <= 22 * kj::MILLISECONDS) {
+      return;
+    }
+  }
+  KJ_FAIL_ASSERT("time not within expected bounds even after retries");
+}
+
+TEST(Mutex, WhenWithTimeoutPreciseTimingAfterInterrupt) {
+  // Test that MutexGuarded::when() with a timeout sleeps for precisely the right amount of time,
+  // even if the thread is spuriously woken in the middle.
+
+  for (uint retryCount = 0; retryCount < 20; retryCount++) {
+    MutexGuarded<uint> value(123);
+
+    kj::Thread thread([&]() {
+      delay();
+      value.lockExclusive().induceSpuriousWakeupForTest();
+    });
+
+    auto start = now();
+    uint m = value.when([](uint n) {
+      return n == 321;
+    }, [](uint& n) {
+      return 456;
+    }, 20 * kj::MILLISECONDS);
+
+    KJ_EXPECT(m == 456);
+
+    auto t = now() - start;
+    KJ_EXPECT(t >= 20 * kj::MILLISECONDS, t / kj::MILLISECONDS);
+    if (t <= 22 * kj::MILLISECONDS) {
+      return;
+    }
+  }
+  KJ_FAIL_ASSERT("time not within expected bounds even after retries");
+}
+
 TEST(Mutex, Lazy) {
   Lazy<uint> lazy;
   volatile bool initStarted = false;

c++/src/kj/mutex.c++

@@ -346,6 +346,19 @@ void Mutex::lockWhen(Predicate& predicate, Maybe<Duration> timeout) {
   }
 }
 
+void Mutex::induceSpuriousWakeupForTest() {
+  auto nextWaiter = waitersHead;
+  for (;;) {
+    KJ_IF_MAYBE(waiter, nextWaiter) {
+      nextWaiter = waiter->next;
+      syscall(SYS_futex, &waiter->futex, FUTEX_WAKE_PRIVATE, INT_MAX, NULL, NULL, 0);
+    } else {
+      // No more waiters.
+      break;
+    }
+  }
+}
+
 void Once::runOnce(Initializer& init) {
 startOver:
   uint state = UNINITIALIZED;
@@ -500,12 +513,21 @@ void Mutex::lockWhen(Predicate& predicate, Maybe<Duration> timeout) {
   KJ_IF_MAYBE(t, timeout) {
     // Compute initial sleep time.
     sleepMs = *t / kj::MILLISECONDS;
+    if (*t % kj::MILLISECONDS > 0 * kj::SECONDS) {
+      // We guarantee we won't wake up too early.
+      ++sleepMs;
+    }
 
     // Also compute the timeout absolute time in Performance Counter ticks, in case we need to
     // restart the wait later.
     QueryPerformanceFrequency(&frequency);
     QueryPerformanceCounter(&endTime);
-    endTime.QuadPart += *t / kj::MILLISECONDS * frequency.QuadPart / 1000;
+    auto numerator = *t / kj::MILLISECONDS * frequency.QuadPart;
+    endTime.QuadPart += numerator / 1000;
+    if (numerator % 1000 > 0) {
+      // We guarantee we won't wake up too early.
+      ++endTime.QuadPart;
+    }
   } else {
     sleepMs = INFINITE;
   }
@@ -538,8 +560,12 @@ void Mutex::lockWhen(Predicate& predicate, Maybe<Duration> timeout) {
       QueryPerformanceCounter(&now);
 
       if (endTime.QuadPart > now.QuadPart) {
-        uint64_t remaining = endTime.QuadPart - now.QuadPart;
-        sleepMs = remaining * 1000 / frequency.QuadPart;
+        uint64_t numerator = (endTime.QuadPart - now.QuadPart) * 1000;
+        sleepMs = numerator / frequency.QuadPart;
+        if (numerator % frequency.QuadPart > 0) {
+          // We guarantee we won't wake up too early.
+          ++sleepMs;
+        }
       } else {
         // Oops, already timed out.
         return;
@@ -548,6 +574,19 @@ void Mutex::lockWhen(Predicate& predicate, Maybe<Duration> timeout) {
   }
 }
 
+void Mutex::induceSpuriousWakeupForTest() {
+  auto nextWaiter = waitersHead;
+  for (;;) {
+    KJ_IF_MAYBE(waiter, nextWaiter) {
+      nextWaiter = waiter->next;
+      WakeConditionVariable(&coercedCondvar(waiter->condvar));
+    } else {
+      // No more waiters.
+      break;
+    }
+  }
+}
+
 static BOOL WINAPI nullInitializer(PINIT_ONCE initOnce, PVOID parameter, PVOID* context) {
   return true;
 }
@@ -779,6 +818,21 @@ void Mutex::lockWhen(Predicate& predicate, Maybe<Duration> timeout) {
   }
 }
 
+void Mutex::induceSpuriousWakeupForTest() {
+  auto nextWaiter = waitersHead;
+  for (;;) {
+    KJ_IF_MAYBE(waiter, nextWaiter) {
+      nextWaiter = waiter->next;
+      KJ_PTHREAD_CALL(pthread_mutex_lock(&waiter->stupidMutex));
+      KJ_PTHREAD_CALL(pthread_cond_signal(&waiter->condvar));
+      KJ_PTHREAD_CALL(pthread_mutex_unlock(&waiter->stupidMutex));
+    } else {
+      // No more waiters.
+      break;
+    }
+  }
+}
+
 Once::Once(bool startInitialized)
     : state(startInitialized ? INITIALIZED : UNINITIALIZED),
       mutex(PTHREAD_MUTEX_INITIALIZER) {}

c++/src/kj/mutex.h

@@ -79,6 +79,11 @@ public:
   // The mutex is always locked when this returns regardless of whether the timeout expired (and
   // always unlocked if it throws).
 
+  void induceSpuriousWakeupForTest();
+  // Utility method for mutex-test.c++ which causes a spurious thread wakeup on all threads that
+  // are waiting for a lockWhen() condition. Assuming correct implementation, all those threads
+  // should immediately go back to sleep.
+
 private:
 #if KJ_USE_FUTEX
   uint futex;
@@ -246,6 +251,14 @@ private:
   friend class MutexGuarded;
   template <typename U>
   friend class ExternalMutexGuarded;
+
+#if KJ_MUTEX_TEST
+public:
+#endif
+  void induceSpuriousWakeupForTest() { mutex->induceSpuriousWakeupForTest(); }
+  // Utility method for mutex-test.c++ which causes a spurious thread wakeup on all threads that
+  // are waiting for a when() condition. Assuming correct implementation, all those threads should
+  // immediately go back to sleep.
 };
 
 template <typename T>