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 #ifndef ROCKSDB_LITE
7
8 #include "utilities/transactions/optimistic_transaction.h"
9
10 #include <string>
11
12 #include "db/column_family.h"
13 #include "db/db_impl/db_impl.h"
14 #include "rocksdb/comparator.h"
15 #include "rocksdb/db.h"
16 #include "rocksdb/status.h"
17 #include "rocksdb/utilities/optimistic_transaction_db.h"
18 #include "util/cast_util.h"
19 #include "util/string_util.h"
20 #include "utilities/transactions/transaction_util.h"
21 #include "utilities/transactions/optimistic_transaction.h"
22 #include "utilities/transactions/optimistic_transaction_db_impl.h"
23
24 namespace ROCKSDB_NAMESPACE {
25
26 struct WriteOptions;
27
OptimisticTransaction(OptimisticTransactionDB * txn_db,const WriteOptions & write_options,const OptimisticTransactionOptions & txn_options)28 OptimisticTransaction::OptimisticTransaction(
29 OptimisticTransactionDB* txn_db, const WriteOptions& write_options,
30 const OptimisticTransactionOptions& txn_options)
31 : TransactionBaseImpl(txn_db->GetBaseDB(), write_options), txn_db_(txn_db) {
32 Initialize(txn_options);
33 }
34
Initialize(const OptimisticTransactionOptions & txn_options)35 void OptimisticTransaction::Initialize(
36 const OptimisticTransactionOptions& txn_options) {
37 if (txn_options.set_snapshot) {
38 SetSnapshot();
39 }
40 }
41
Reinitialize(OptimisticTransactionDB * txn_db,const WriteOptions & write_options,const OptimisticTransactionOptions & txn_options)42 void OptimisticTransaction::Reinitialize(
43 OptimisticTransactionDB* txn_db, const WriteOptions& write_options,
44 const OptimisticTransactionOptions& txn_options) {
45 TransactionBaseImpl::Reinitialize(txn_db->GetBaseDB(), write_options);
46 Initialize(txn_options);
47 }
48
~OptimisticTransaction()49 OptimisticTransaction::~OptimisticTransaction() {}
50
Clear()51 void OptimisticTransaction::Clear() { TransactionBaseImpl::Clear(); }
52
Prepare()53 Status OptimisticTransaction::Prepare() {
54 return Status::InvalidArgument(
55 "Two phase commit not supported for optimistic transactions.");
56 }
57
Commit()58 Status OptimisticTransaction::Commit() {
59 auto txn_db_impl = static_cast_with_check<OptimisticTransactionDBImpl,
60 OptimisticTransactionDB>(txn_db_);
61 assert(txn_db_impl);
62 switch (txn_db_impl->GetValidatePolicy()) {
63 case OccValidationPolicy::kValidateParallel:
64 return CommitWithParallelValidate();
65 case OccValidationPolicy::kValidateSerial:
66 return CommitWithSerialValidate();
67 default:
68 assert(0);
69 }
70 // unreachable, just void compiler complain
71 return Status::OK();
72 }
73
CommitWithSerialValidate()74 Status OptimisticTransaction::CommitWithSerialValidate() {
75 // Set up callback which will call CheckTransactionForConflicts() to
76 // check whether this transaction is safe to be committed.
77 OptimisticTransactionCallback callback(this);
78
79 DBImpl* db_impl = static_cast_with_check<DBImpl, DB>(db_->GetRootDB());
80
81 Status s = db_impl->WriteWithCallback(
82 write_options_, GetWriteBatch()->GetWriteBatch(), &callback);
83
84 if (s.ok()) {
85 Clear();
86 }
87
88 return s;
89 }
90
CommitWithParallelValidate()91 Status OptimisticTransaction::CommitWithParallelValidate() {
92 auto txn_db_impl = static_cast_with_check<OptimisticTransactionDBImpl,
93 OptimisticTransactionDB>(txn_db_);
94 assert(txn_db_impl);
95 DBImpl* db_impl = static_cast_with_check<DBImpl, DB>(db_->GetRootDB());
96 assert(db_impl);
97 const size_t space = txn_db_impl->GetLockBucketsSize();
98 std::set<size_t> lk_idxes;
99 std::vector<std::unique_lock<std::mutex>> lks;
100 for (auto& cfit : GetTrackedKeys()) {
101 for (auto& keyit : cfit.second) {
102 lk_idxes.insert(fastrange64(GetSliceNPHash64(keyit.first), space));
103 }
104 }
105 // NOTE: in a single txn, all bucket-locks are taken in ascending order.
106 // In this way, txns from different threads all obey this rule so that
107 // deadlock can be avoided.
108 for (auto v : lk_idxes) {
109 lks.emplace_back(txn_db_impl->LockBucket(v));
110 }
111
112 Status s = TransactionUtil::CheckKeysForConflicts(db_impl, GetTrackedKeys(),
113 true /* cache_only */);
114 if (!s.ok()) {
115 return s;
116 }
117
118 s = db_impl->Write(write_options_, GetWriteBatch()->GetWriteBatch());
119 if (s.ok()) {
120 Clear();
121 }
122
123 return s;
124 }
125
Rollback()126 Status OptimisticTransaction::Rollback() {
127 Clear();
128 return Status::OK();
129 }
130
131 // Record this key so that we can check it for conflicts at commit time.
132 //
133 // 'exclusive' is unused for OptimisticTransaction.
TryLock(ColumnFamilyHandle * column_family,const Slice & key,bool read_only,bool exclusive,const bool do_validate,const bool assume_tracked)134 Status OptimisticTransaction::TryLock(ColumnFamilyHandle* column_family,
135 const Slice& key, bool read_only,
136 bool exclusive, const bool do_validate,
137 const bool assume_tracked) {
138 assert(!assume_tracked); // not supported
139 (void)assume_tracked;
140 if (!do_validate) {
141 return Status::OK();
142 }
143 uint32_t cfh_id = GetColumnFamilyID(column_family);
144
145 SetSnapshotIfNeeded();
146
147 SequenceNumber seq;
148 if (snapshot_) {
149 seq = snapshot_->GetSequenceNumber();
150 } else {
151 seq = db_->GetLatestSequenceNumber();
152 }
153
154 std::string key_str = key.ToString();
155
156 TrackKey(cfh_id, key_str, seq, read_only, exclusive);
157
158 // Always return OK. Confilct checking will happen at commit time.
159 return Status::OK();
160 }
161
162 // Returns OK if it is safe to commit this transaction. Returns Status::Busy
163 // if there are read or write conflicts that would prevent us from committing OR
164 // if we can not determine whether there would be any such conflicts.
165 //
166 // Should only be called on writer thread in order to avoid any race conditions
167 // in detecting write conflicts.
CheckTransactionForConflicts(DB * db)168 Status OptimisticTransaction::CheckTransactionForConflicts(DB* db) {
169 Status result;
170
171 auto db_impl = static_cast_with_check<DBImpl, DB>(db);
172
173 // Since we are on the write thread and do not want to block other writers,
174 // we will do a cache-only conflict check. This can result in TryAgain
175 // getting returned if there is not sufficient memtable history to check
176 // for conflicts.
177 return TransactionUtil::CheckKeysForConflicts(db_impl, GetTrackedKeys(),
178 true /* cache_only */);
179 }
180
SetName(const TransactionName &)181 Status OptimisticTransaction::SetName(const TransactionName& /* unused */) {
182 return Status::InvalidArgument("Optimistic transactions cannot be named.");
183 }
184
185 } // namespace ROCKSDB_NAMESPACE
186
187 #endif // ROCKSDB_LITE
188