1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
#ifndef ROCKSDB_LITE
#include "utilities/transactions/transaction_db_mutex_impl.h"
#include <chrono>
#include <condition_variable>
#include <functional>
#include <mutex>
#include "rocksdb/utilities/transaction_db_mutex.h"
namespace ROCKSDB_NAMESPACE {
class TransactionDBMutexImpl : public TransactionDBMutex {
public:
TransactionDBMutexImpl() {}
~TransactionDBMutexImpl() override {}
Status Lock() override;
Status TryLockFor(int64_t timeout_time) override;
void UnLock() override { mutex_.unlock(); }
friend class TransactionDBCondVarImpl;
private:
std::mutex mutex_;
};
class TransactionDBCondVarImpl : public TransactionDBCondVar {
public:
TransactionDBCondVarImpl() {}
~TransactionDBCondVarImpl() override {}
Status Wait(std::shared_ptr<TransactionDBMutex> mutex) override;
Status WaitFor(std::shared_ptr<TransactionDBMutex> mutex,
int64_t timeout_time) override;
void Notify() override { cv_.notify_one(); }
void NotifyAll() override { cv_.notify_all(); }
private:
std::condition_variable cv_;
};
std::shared_ptr<TransactionDBMutex>
TransactionDBMutexFactoryImpl::AllocateMutex() {
return std::shared_ptr<TransactionDBMutex>(new TransactionDBMutexImpl());
}
std::shared_ptr<TransactionDBCondVar>
TransactionDBMutexFactoryImpl::AllocateCondVar() {
return std::shared_ptr<TransactionDBCondVar>(new TransactionDBCondVarImpl());
}
Status TransactionDBMutexImpl::Lock() {
mutex_.lock();
return Status::OK();
}
Status TransactionDBMutexImpl::TryLockFor(int64_t timeout_time) {
bool locked = true;
if (timeout_time == 0) {
locked = mutex_.try_lock();
} else {
// Previously, this code used a std::timed_mutex. However, this was changed
// due to known bugs in gcc versions < 4.9.
// https://gcc.gnu.org/bugzilla/show_bug.cgi?id=54562
//
// Since this mutex isn't held for long and only a single mutex is ever
// held at a time, it is reasonable to ignore the lock timeout_time here
// and only check it when waiting on the condition_variable.
mutex_.lock();
}
if (!locked) {
// timeout acquiring mutex
return Status::TimedOut(Status::SubCode::kMutexTimeout);
}
return Status::OK();
}
Status TransactionDBCondVarImpl::Wait(
std::shared_ptr<TransactionDBMutex> mutex) {
auto mutex_impl = reinterpret_cast<TransactionDBMutexImpl*>(mutex.get());
std::unique_lock<std::mutex> lock(mutex_impl->mutex_, std::adopt_lock);
cv_.wait(lock);
// Make sure unique_lock doesn't unlock mutex when it destructs
lock.release();
return Status::OK();
}
Status TransactionDBCondVarImpl::WaitFor(
std::shared_ptr<TransactionDBMutex> mutex, int64_t timeout_time) {
Status s;
auto mutex_impl = reinterpret_cast<TransactionDBMutexImpl*>(mutex.get());
std::unique_lock<std::mutex> lock(mutex_impl->mutex_, std::adopt_lock);
if (timeout_time < 0) {
// If timeout is negative, do not use a timeout
cv_.wait(lock);
} else {
auto duration = std::chrono::microseconds(timeout_time);
auto cv_status = cv_.wait_for(lock, duration);
// Check if the wait stopped due to timing out.
if (cv_status == std::cv_status::timeout) {
s = Status::TimedOut(Status::SubCode::kMutexTimeout);
}
}
// Make sure unique_lock doesn't unlock mutex when it destructs
lock.release();
// CV was signaled, or we spuriously woke up (but didn't time out)
return s;
}
} // namespace ROCKSDB_NAMESPACE
#endif // ROCKSDB_LITE
|
