ceph/src/rocksdb/util/repeatable_thread_test.cc

   1 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
   2 //  This source code is licensed under both the GPLv2 (found in the
   3 //  COPYING file in the root directory) and Apache 2.0 License
   4 //  (found in the LICENSE.Apache file in the root directory).
   5
   6 #include <atomic>
   7 #include <memory>
   8
   9 #include "db/db_test_util.h"
  10 #include "util/repeatable_thread.h"
  11 #include "util/sync_point.h"
  12 #include "util/testharness.h"
  13
  14 class RepeatableThreadTest : public testing::Test {
  15  public:
  16   RepeatableThreadTest()
  17       : mock_env_(new rocksdb::MockTimeEnv(rocksdb::Env::Default())) {}
  18
  19  protected:
  20   std::unique_ptr<rocksdb::MockTimeEnv> mock_env_;
  21 };
  22
  23 TEST_F(RepeatableThreadTest, TimedTest) {
  24   constexpr uint64_t kSecond = 1000000;  // 1s = 1000000us
  25   constexpr int kIteration = 3;
  26   rocksdb::Env* env = rocksdb::Env::Default();
  27   rocksdb::port::Mutex mutex;
  28   rocksdb::port::CondVar test_cv(&mutex);
  29   int count = 0;
  30   uint64_t prev_time = env->NowMicros();
  31   rocksdb::RepeatableThread thread(
  32       [&] {
  33         rocksdb::MutexLock l(&mutex);
  34         count++;
  35         uint64_t now = env->NowMicros();
  36         assert(count == 1 || prev_time + 1 * kSecond <= now);
  37         prev_time = now;
  38         if (count >= kIteration) {
  39           test_cv.SignalAll();
  40         }
  41       },
  42       "rt_test", env, 1 * kSecond);
  43   // Wait for execution finish.
  44   {
  45     rocksdb::MutexLock l(&mutex);
  46     while (count < kIteration) {
  47       test_cv.Wait();
  48     }
  49   }
  50
  51   // Test cancel
  52   thread.cancel();
  53 }
  54
  55 TEST_F(RepeatableThreadTest, MockEnvTest) {
  56   constexpr uint64_t kSecond = 1000000;  // 1s = 1000000us
  57   constexpr int kIteration = 3;
  58   mock_env_->set_current_time(0);  // in seconds
  59   std::atomic<int> count{0};
  60
  61 #if defined(OS_MACOSX) && !defined(NDEBUG)
  62   rocksdb::SyncPoint::GetInstance()->DisableProcessing();
  63   rocksdb::SyncPoint::GetInstance()->ClearAllCallBacks();
  64   rocksdb::SyncPoint::GetInstance()->SetCallBack(
  65       "InstrumentedCondVar::TimedWaitInternal", [&](void* arg) {
  66         // Obtain the current (real) time in seconds and add 1000 extra seconds
  67         // to ensure that RepeatableThread::wait invokes TimedWait with a time
  68         // greater than (real) current time. This is to prevent the TimedWait
  69         // function from returning immediately without sleeping and releasing
  70         // the mutex on certain platforms, e.g. OS X. If TimedWait returns
  71         // immediately, the mutex will not be released, and
  72         // RepeatableThread::TEST_WaitForRun never has a chance to execute the
  73         // callback which, in this case, updates the result returned by
  74         // mock_env->NowMicros. Consequently, RepeatableThread::wait cannot
  75         // break out of the loop, causing test to hang. The extra 1000 seconds
  76         // is a best-effort approach because there seems no reliable and
  77         // deterministic way to provide the aforementioned guarantee. By the
  78         // time RepeatableThread::wait is called, it is no guarantee that the
  79         // delay + mock_env->NowMicros will be greater than the current real
  80         // time. However, 1000 seconds should be sufficient in most cases.
  81         uint64_t time_us = *reinterpret_cast<uint64_t*>(arg);
  82         if (time_us < mock_env_->RealNowMicros()) {
  83           *reinterpret_cast<uint64_t*>(arg) = mock_env_->RealNowMicros() + 1000;
  84         }
  85       });
  86   rocksdb::SyncPoint::GetInstance()->EnableProcessing();
  87 #endif  // OS_MACOSX && !NDEBUG
  88
  89   rocksdb::RepeatableThread thread([&] { count++; }, "rt_test", mock_env_.get(),
  90                                    1 * kSecond, 1 * kSecond);
  91   for (int i = 1; i <= kIteration; i++) {
  92     // Bump current time
  93     thread.TEST_WaitForRun([&] { mock_env_->set_current_time(i); });
  94   }
  95   // Test function should be exectued exactly kIteraion times.
  96   ASSERT_EQ(kIteration, count.load());
  97
  98   // Test cancel
  99   thread.cancel();
 100 }
 101
 102 int main(int argc, char** argv) {
 103   ::testing::InitGoogleTest(&argc, argv);
 104
 105   return RUN_ALL_TESTS();
 106 }