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