repro test case

Author: ajkr

db/db_test_util.h

@@ -579,7 +579,10 @@ class SpecialEnv : public EnvWrapper {
     // To prevent slowdown, this `TimedWait()` is completely synthetic. The mock
     // implementation here pretends the timeout always happened.
     cv->GetMutex()->Unlock();
+    TEST_SYNC_POINT("SpecialEnv::TimedWait:UnlockedPreSleep");
     addon_microseconds_.fetch_add(static_cast<int64_t>(delay_micros));
+    TEST_SYNC_POINT("SpecialEnv::TimedWait:UnlockedPostSleep1");
+    TEST_SYNC_POINT("SpecialEnv::TimedWait:UnlockedPostSleep2");
     cv->GetMutex()->Lock();
     return true;
   }

util/rate_limiter_test.cc

@@ -497,6 +497,63 @@ TEST_F(RateLimiterTest, AutoTuneIncreaseWhenFull) {
   ASSERT_LT(new_bytes_per_sec, orig_bytes_per_sec);
 }
 
+TEST_F(RateLimiterTest, WaitHangingBug) {
+  // At t=0: Threads 0 and 1 request `kBytesPerRefill` bytes at low-pri. One
+  // will be granted immediately and the other will enter `TimedWait()`.
+  //
+  // At t=`kMicrosPerRefill`: Thread 2 requests `kBytesPerRefill` bytes at
+  // low-pri. Thread 2's request enters the queue. To expose the bug scenario,
+  // `SyncPoint`s ensure this happens while the lock is temporarily released in
+  // `TimedWait()`. Before the bug fix, Thread 2's request would then hang in
+  // `Wait()` interminably.
+  const int kBytesPerSecond = 100;
+  const int kMicrosPerSecond = 1000 * 1000;
+  const int kMicrosPerRefill = kMicrosPerSecond;
+  const int kBytesPerRefill =
+      kBytesPerSecond * kMicrosPerRefill / kMicrosPerSecond;
+
+  SpecialEnv special_env(Env::Default(), true /* time_elapse_only_sleep */);
+  std::unique_ptr<RateLimiter> limiter(new GenericRateLimiter(
+      kBytesPerSecond, kMicrosPerRefill, 10 /* fairness */,
+      RateLimiter::Mode::kWritesOnly, special_env.GetSystemClock(),
+      false /* auto_tuned */));
+  std::array<std::thread, 3> request_threads;
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->LoadDependency(
+      {{"RateLimiterTest::WaitHangingBug:InitialRequestsReady",
+        "SpecialEnv::TimedWait:UnlockedPreSleep"},
+       {"SpecialEnv::TimedWait:UnlockedPostSleep1",
+        "RateLimiterTest::WaitHangingBug:TestThreadRequestBegin"},
+       {"RateLimiterTest::WaitHangingBug:TestThreadRequestEnd",
+        "SpecialEnv::TimedWait:UnlockedPostSleep2"}});
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+
+  for (int i = 0; i < 2; i++) {
+    request_threads[i] = std::thread([&]() {
+      limiter->Request(kBytesPerRefill /* bytes */, Env::IOPriority::IO_LOW,
+                       nullptr /* stats */, RateLimiter::OpType::kWrite,
+                       &special_env);
+    });
+  }
+  while (limiter->GetTotalRequests() < 2) {
+  }
+  TEST_SYNC_POINT("RateLimiterTest::WaitHangingBug:InitialRequestsReady");
+
+  TEST_SYNC_POINT("RateLimiterTest::WaitHangingBug:TestThreadRequestBegin");
+  request_threads[2] = std::thread([&]() {
+    limiter->Request(kBytesPerRefill /* bytes */, Env::IOPriority::IO_LOW,
+                     nullptr /* stats */, RateLimiter::OpType::kWrite,
+                     &special_env);
+  });
+  while (limiter->GetTotalRequests() < 3) {
+  }
+  TEST_SYNC_POINT("RateLimiterTest::WaitHangingBug:TestThreadRequestEnd");
+
+  for (int i = 0; i < 3; i++) {
+    request_threads[i].join();
+  }
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {