andreacavalli/rocksdb
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Have Transactions use WriteBatch::RollbackToSavePoint

Browse Source 
Summary:
Clean up transactions to use the new RollbackToSavePoint api in WriteBatchWithIndex.

Note, this diff depends on Pessimistic Transactions diff and ManagedSnapshot diff (D40869 and D43293).

Test Plan: unit tests

Reviewers: rven, yhchiang, kradhakrishnan, spetrunia, sdong

Reviewed By: sdong

Subscribers: dhruba, leveldb

Differential Revision: https://reviews.facebook.net/D43371
This commit is contained in:
agiardullo 2015-07-31 15:18:27 -07:00
parent 0db807ec28
commit c3466eab07
9 changed files with 87 additions and 291 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
include/rocksdb/utilities/transaction.h
View File

@ -93,12 +93,10 @@ class Transaction {
virtual void SetSavePoint() = 0; virtual void SetSavePoint() = 0;
// Undo all operations in this transaction (Put, Merge, Delete, PutLogData) // Undo all operations in this transaction (Put, Merge, Delete, PutLogData)
// since the // since the most recent call to SetSavePoint() and removes the most recent
// most recent call to SetSavePoint() and removes the most recent
// SetSavePoint(). // SetSavePoint().
// If there is no previous call to SetSavePoint(), behaves the same as // If there is no previous call to SetSavePoint(), returns Status::NotFound()
// Rollback() virtual Status RollbackToSavePoint() = 0;
virtual void RollbackToSavePoint() = 0;
// This function is similar to DB::Get() except it will also read pending // This function is similar to DB::Get() except it will also read pending
// changes in this transaction. // changes in this transaction.

86
utilities/transactions/optimistic_transaction_impl.cc
View File

@ -30,30 +30,24 @@ OptimisticTransactionImpl::OptimisticTransactionImpl(
: txn_db_(txn_db), : txn_db_(txn_db),
db_(txn_db->GetBaseDB()), db_(txn_db->GetBaseDB()),
write_options_(write_options), write_options_(write_options),
snapshot_(nullptr),
cmp_(txn_options.cmp), cmp_(txn_options.cmp),
write_batch_(new WriteBatchWithIndex(txn_options.cmp, 0, true)) { write_batch_(new WriteBatchWithIndex(txn_options.cmp, 0, true)) {
if (txn_options.set_snapshot) { if (txn_options.set_snapshot) {
SetSnapshot(); SetSnapshot();
} else {
start_sequence_number_ = db_->GetLatestSequenceNumber();
} }
} }
OptimisticTransactionImpl::~OptimisticTransactionImpl() { OptimisticTransactionImpl::~OptimisticTransactionImpl() {
}
void OptimisticTransactionImpl::Cleanup() {
tracked_keys_.clear(); tracked_keys_.clear();
if (snapshot_ != nullptr) { save_points_.reset(nullptr);
db_->ReleaseSnapshot(snapshot_); write_batch_->Clear();
}
} }
void OptimisticTransactionImpl::SetSnapshot() { void OptimisticTransactionImpl::SetSnapshot() {
if (snapshot_ != nullptr) { snapshot_.reset(new ManagedSnapshot(db_));
db_->ReleaseSnapshot(snapshot_);
}
snapshot_ = db_->GetSnapshot();
start_sequence_number_ = snapshot_->GetSequenceNumber();
} }
Status OptimisticTransactionImpl::Commit() { Status OptimisticTransactionImpl::Commit() {
@ -73,66 +67,38 @@ Status OptimisticTransactionImpl::Commit() {
write_options_, write_batch_->GetWriteBatch(), &callback); write_options_, write_batch_->GetWriteBatch(), &callback);
if (s.ok()) { if (s.ok()) {
tracked_keys_.clear(); Cleanup();
write_batch_->Clear();
num_entries_ = 0;
} }
return s; return s;
} }
void OptimisticTransactionImpl::Rollback() { void OptimisticTransactionImpl::Rollback() {
tracked_keys_.clear(); Cleanup();
write_batch_->Clear();
num_entries_ = 0;
} }
void OptimisticTransactionImpl::SetSavePoint() { void OptimisticTransactionImpl::SetSavePoint() {
if (num_entries_ > 0) { if (save_points_ == nullptr) {
// If transaction is empty, no need to record anything. save_points_.reset(new std::stack<std::shared_ptr<ManagedSnapshot>>());
if (save_points_ == nullptr) {
save_points_.reset(new std::stack<size_t>());
}
save_points_->push(num_entries_);
} }
save_points_->push(snapshot_);
write_batch_->SetSavePoint();
} }
void OptimisticTransactionImpl::RollbackToSavePoint() { Status OptimisticTransactionImpl::RollbackToSavePoint() {
size_t savepoint_entries = 0;
if (save_points_ != nullptr && save_points_->size() > 0) { if (save_points_ != nullptr && save_points_->size() > 0) {
savepoint_entries = save_points_->top(); // Restore saved snapshot
snapshot_ = save_points_->top();
save_points_->pop(); save_points_->pop();
}
assert(savepoint_entries <= num_entries_); // Rollback batch
Status s = write_batch_->RollbackToSavePoint();
assert(s.ok());
if (savepoint_entries == num_entries_) { return s;
// No changes to rollback
} else if (savepoint_entries == 0) {
// Rollback everything
Rollback();
} else { } else {
DBImpl* db_impl = dynamic_cast<DBImpl*>(db_->GetRootDB()); assert(write_batch_->RollbackToSavePoint().IsNotFound());
assert(db_impl); return Status::NotFound();
WriteBatchWithIndex* new_batch = new WriteBatchWithIndex(cmp_, 0, true);
Status s = TransactionUtil::CopyFirstN(
savepoint_entries, write_batch_.get(), new_batch, db_impl);
if (!s.ok()) {
// TODO: Should we change this function to return a Status or should we
// somehow make it
// so RollbackToSavePoint() can never fail??
// Consider moving this functionality into WriteBatchWithIndex
fprintf(stderr, "STATUS: %s \n", s.ToString().c_str());
delete new_batch;
} else {
write_batch_.reset(new_batch);
}
num_entries_ = savepoint_entries;
} }
} }
@ -143,7 +109,7 @@ void OptimisticTransactionImpl::RecordOperation(
SequenceNumber seq; SequenceNumber seq;
if (snapshot_) { if (snapshot_) {
seq = start_sequence_number_; seq = snapshot_->snapshot()->GetSequenceNumber();
} else { } else {
seq = db_->GetLatestSequenceNumber(); seq = db_->GetLatestSequenceNumber();
} }
@ -261,7 +227,6 @@ Status OptimisticTransactionImpl::Put(ColumnFamilyHandle* column_family,
RecordOperation(column_family, key); RecordOperation(column_family, key);
write_batch_->Put(column_family, key, value); write_batch_->Put(column_family, key, value);
num_entries_++;
return Status::OK(); return Status::OK();
} }
@ -272,7 +237,6 @@ Status OptimisticTransactionImpl::Put(ColumnFamilyHandle* column_family,
RecordOperation(column_family, key); RecordOperation(column_family, key);
write_batch_->Put(column_family, key, value); write_batch_->Put(column_family, key, value);
num_entries_++;
return Status::OK(); return Status::OK();
} }
@ -307,7 +271,6 @@ Status OptimisticTransactionImpl::Delete(ColumnFamilyHandle* column_family,
Status OptimisticTransactionImpl::PutUntracked( Status OptimisticTransactionImpl::PutUntracked(
ColumnFamilyHandle* column_family, const Slice& key, const Slice& value) { ColumnFamilyHandle* column_family, const Slice& key, const Slice& value) {
write_batch_->Put(column_family, key, value); write_batch_->Put(column_family, key, value);
num_entries_++;
return Status::OK(); return Status::OK();
} }
@ -316,7 +279,6 @@ Status OptimisticTransactionImpl::PutUntracked(
ColumnFamilyHandle* column_family, const SliceParts& key, ColumnFamilyHandle* column_family, const SliceParts& key,
const SliceParts& value) { const SliceParts& value) {
write_batch_->Put(column_family, key, value); write_batch_->Put(column_family, key, value);
num_entries_++;
return Status::OK(); return Status::OK();
} }
@ -324,7 +286,6 @@ Status OptimisticTransactionImpl::PutUntracked(
Status OptimisticTransactionImpl::MergeUntracked( Status OptimisticTransactionImpl::MergeUntracked(
ColumnFamilyHandle* column_family, const Slice& key, const Slice& value) { ColumnFamilyHandle* column_family, const Slice& key, const Slice& value) {
write_batch_->Merge(column_family, key, value); write_batch_->Merge(column_family, key, value);
num_entries_++;
return Status::OK(); return Status::OK();
} }
@ -332,7 +293,6 @@ Status OptimisticTransactionImpl::MergeUntracked(
Status OptimisticTransactionImpl::DeleteUntracked( Status OptimisticTransactionImpl::DeleteUntracked(
ColumnFamilyHandle* column_family, const Slice& key) { ColumnFamilyHandle* column_family, const Slice& key) {
write_batch_->Delete(column_family, key); write_batch_->Delete(column_family, key);
num_entries_++;
return Status::OK(); return Status::OK();
} }
@ -340,14 +300,12 @@ Status OptimisticTransactionImpl::DeleteUntracked(
Status OptimisticTransactionImpl::DeleteUntracked( Status OptimisticTransactionImpl::DeleteUntracked(
ColumnFamilyHandle* column_family, const SliceParts& key) { ColumnFamilyHandle* column_family, const SliceParts& key) {
write_batch_->Delete(column_family, key); write_batch_->Delete(column_family, key);
num_entries_++;
return Status::OK(); return Status::OK();
} }
void OptimisticTransactionImpl::PutLogData(const Slice& blob) { void OptimisticTransactionImpl::PutLogData(const Slice& blob) {
write_batch_->PutLogData(blob); write_batch_->PutLogData(blob);
num_entries_++;
} }
WriteBatchWithIndex* OptimisticTransactionImpl::GetWriteBatch() { WriteBatchWithIndex* OptimisticTransactionImpl::GetWriteBatch() {

22
utilities/transactions/optimistic_transaction_impl.h
View File

@ -15,6 +15,7 @@
#include "db/write_callback.h" #include "db/write_callback.h"
#include "rocksdb/db.h" #include "rocksdb/db.h"
#include "rocksdb/slice.h" #include "rocksdb/slice.h"
#include "rocksdb/snapshot.h"
#include "rocksdb/status.h" #include "rocksdb/status.h"
#include "rocksdb/types.h" #include "rocksdb/types.h"
#include "rocksdb/utilities/transaction.h" #include "rocksdb/utilities/transaction.h"
@ -38,7 +39,7 @@ class OptimisticTransactionImpl : public Transaction {
void SetSavePoint() override; void SetSavePoint() override;
void RollbackToSavePoint() override; Status RollbackToSavePoint() override;
Status Get(const ReadOptions& options, ColumnFamilyHandle* column_family, Status Get(const ReadOptions& options, ColumnFamilyHandle* column_family,
const Slice& key, std::string* value) override; const Slice& key, std::string* value) override;
@ -147,7 +148,9 @@ class OptimisticTransactionImpl : public Transaction {
const TransactionKeyMap* GetTrackedKeys() const { return &tracked_keys_; } const TransactionKeyMap* GetTrackedKeys() const { return &tracked_keys_; }
const Snapshot* GetSnapshot() const override { return snapshot_; } const Snapshot* GetSnapshot() const override {
return snapshot_ ? snapshot_->snapshot() : nullptr;
}
void SetSnapshot() override; void SetSnapshot() override;
@ -157,8 +160,7 @@ class OptimisticTransactionImpl : public Transaction {
OptimisticTransactionDB* const txn_db_; OptimisticTransactionDB* const txn_db_;
DB* db_; DB* db_;
const WriteOptions write_options_; const WriteOptions write_options_;
const Snapshot* snapshot_; std::shared_ptr<ManagedSnapshot> snapshot_;
SequenceNumber start_sequence_number_;
const Comparator* cmp_; const Comparator* cmp_;
std::unique_ptr<WriteBatchWithIndex> write_batch_; std::unique_ptr<WriteBatchWithIndex> write_batch_;
@ -169,13 +171,9 @@ class OptimisticTransactionImpl : public Transaction {
// not changed since this sequence number. // not changed since this sequence number.
TransactionKeyMap tracked_keys_; TransactionKeyMap tracked_keys_;
// Records the number of entries currently in the WriteBatch including calls // Stack of the Snapshot saved at each save point. Saved snapshots may be
// to // nullptr if there was no snapshot at the time SetSavePoint() was called.
// Put, Merge, Delete, and PutLogData() std::unique_ptr<std::stack<std::shared_ptr<ManagedSnapshot>>> save_points_;
size_t num_entries_ = 0;
// Stack of number of entries in write_batch at each save point
std::unique_ptr<std::stack<size_t>> save_points_;
friend class OptimisticTransactionCallback; friend class OptimisticTransactionCallback;
@ -190,6 +188,8 @@ class OptimisticTransactionImpl : public Transaction {
void RecordOperation(ColumnFamilyHandle* column_family, void RecordOperation(ColumnFamilyHandle* column_family,
const SliceParts& key); const SliceParts& key);
void Cleanup();
// No copying allowed // No copying allowed
OptimisticTransactionImpl(const OptimisticTransactionImpl&); OptimisticTransactionImpl(const OptimisticTransactionImpl&);
void operator=(const OptimisticTransactionImpl&); void operator=(const OptimisticTransactionImpl&);

17
utilities/transactions/optimistic_transaction_test.cc
View File

@ -955,12 +955,14 @@ TEST_F(OptimisticTransactionTest, SavepointTest) {
Transaction* txn = txn_db->BeginTransaction(write_options); Transaction* txn = txn_db->BeginTransaction(write_options);
ASSERT_TRUE(txn); ASSERT_TRUE(txn);
txn->RollbackToSavePoint(); s = txn->RollbackToSavePoint();
ASSERT_TRUE(s.IsNotFound());
txn->SetSavePoint(); // 1 txn->SetSavePoint(); // 1
txn->RollbackToSavePoint(); // Rollback to beginning of txn ASSERT_OK(txn->RollbackToSavePoint()); // Rollback to beginning of txn
txn->RollbackToSavePoint(); s = txn->RollbackToSavePoint();
ASSERT_TRUE(s.IsNotFound());
s = txn->Put("B", "b"); s = txn->Put("B", "b");
ASSERT_OK(s); ASSERT_OK(s);
@ -996,7 +998,7 @@ TEST_F(OptimisticTransactionTest, SavepointTest) {
s = txn->Put("D", "d"); s = txn->Put("D", "d");
ASSERT_OK(s); ASSERT_OK(s);
txn->RollbackToSavePoint(); // Rollback to 2 ASSERT_OK(txn->RollbackToSavePoint()); // Rollback to 2
s = txn->Get(read_options, "A", &value); s = txn->Get(read_options, "A", &value);
ASSERT_OK(s); ASSERT_OK(s);
@ -1019,7 +1021,10 @@ TEST_F(OptimisticTransactionTest, SavepointTest) {
s = txn->Put("E", "e"); s = txn->Put("E", "e");
ASSERT_OK(s); ASSERT_OK(s);
txn->RollbackToSavePoint(); // Rollback to beginning of txn // Rollback to beginning of txn
s = txn->RollbackToSavePoint();
ASSERT_TRUE(s.IsNotFound());
txn->Rollback();
s = txn->Get(read_options, "A", &value); s = txn->Get(read_options, "A", &value);
ASSERT_TRUE(s.IsNotFound()); ASSERT_TRUE(s.IsNotFound());
@ -1065,7 +1070,7 @@ TEST_F(OptimisticTransactionTest, SavepointTest) {
s = txn->Get(read_options, "B", &value); s = txn->Get(read_options, "B", &value);
ASSERT_TRUE(s.IsNotFound()); ASSERT_TRUE(s.IsNotFound());
txn->RollbackToSavePoint(); // Rollback to 3 ASSERT_OK(txn->RollbackToSavePoint()); // Rollback to 3
s = txn->Get(read_options, "F", &value); s = txn->Get(read_options, "F", &value);
ASSERT_OK(s); ASSERT_OK(s);

85
utilities/transactions/transaction_impl.cc
View File

@ -16,6 +16,7 @@
#include "db/db_impl.h" #include "db/db_impl.h"
#include "rocksdb/comparator.h" #include "rocksdb/comparator.h"
#include "rocksdb/db.h" #include "rocksdb/db.h"
#include "rocksdb/snapshot.h"
#include "rocksdb/status.h" #include "rocksdb/status.h"
#include "rocksdb/utilities/transaction_db.h" #include "rocksdb/utilities/transaction_db.h"
#include "util/string_util.h" #include "util/string_util.h"
@ -39,7 +40,6 @@ TransactionImpl::TransactionImpl(TransactionDB* txn_db,
txn_db_impl_(nullptr), txn_db_impl_(nullptr),
txn_id_(GenTxnID()), txn_id_(GenTxnID()),
write_options_(write_options), write_options_(write_options),
snapshot_(nullptr),
cmp_(GetColumnFamilyUserComparator(txn_db->DefaultColumnFamily())), cmp_(GetColumnFamilyUserComparator(txn_db->DefaultColumnFamily())),
write_batch_(new WriteBatchWithIndex(cmp_, 0, true)), write_batch_(new WriteBatchWithIndex(cmp_, 0, true)),
start_time_( start_time_(
@ -62,24 +62,15 @@ TransactionImpl::TransactionImpl(TransactionDB* txn_db,
} }
TransactionImpl::~TransactionImpl() { TransactionImpl::~TransactionImpl() {
Cleanup(); txn_db_impl_->UnLock(this, &tracked_keys_);
if (snapshot_ != nullptr) {
db_->ReleaseSnapshot(snapshot_);
}
} }
void TransactionImpl::SetSnapshot() { void TransactionImpl::SetSnapshot() {
if (snapshot_ != nullptr) { snapshot_.reset(new ManagedSnapshot(db_));
db_->ReleaseSnapshot(snapshot_);
}
snapshot_ = db_->GetSnapshot();
} }
void TransactionImpl::Cleanup() { void TransactionImpl::Cleanup() {
write_batch_->Clear(); write_batch_->Clear();
num_entries_ = 0;
txn_db_impl_->UnLock(this, &tracked_keys_); txn_db_impl_->UnLock(this, &tracked_keys_);
tracked_keys_.clear(); tracked_keys_.clear();
save_points_.reset(nullptr); save_points_.reset(nullptr);
@ -145,53 +136,27 @@ Status TransactionImpl::DoCommit(WriteBatch* batch) {
void TransactionImpl::Rollback() { Cleanup(); } void TransactionImpl::Rollback() { Cleanup(); }
void TransactionImpl::SetSavePoint() { void TransactionImpl::SetSavePoint() {
if (num_entries_ > 0) { if (save_points_ == nullptr) {
// If transaction is empty, no need to record anything. save_points_.reset(new std::stack<std::shared_ptr<ManagedSnapshot>>());
if (save_points_ == nullptr) {
save_points_.reset(new std::stack<size_t>());
}
save_points_->push(num_entries_);
} }
save_points_->push(snapshot_);
write_batch_->SetSavePoint();
} }
void TransactionImpl::RollbackToSavePoint() { Status TransactionImpl::RollbackToSavePoint() {
size_t savepoint_entries = 0;
if (save_points_ != nullptr && save_points_->size() > 0) { if (save_points_ != nullptr && save_points_->size() > 0) {
savepoint_entries = save_points_->top(); // Restore saved snapshot
snapshot_ = save_points_->top();
save_points_->pop(); save_points_->pop();
}
assert(savepoint_entries <= num_entries_); // Rollback batch
Status s = write_batch_->RollbackToSavePoint();
assert(s.ok());
if (savepoint_entries == num_entries_) { return s;
// No changes to rollback
} else if (savepoint_entries == 0) {
// Rollback everything
Rollback();
} else { } else {
assert(dynamic_cast<DBImpl*>(db_->GetBaseDB()) != nullptr); assert(write_batch_->RollbackToSavePoint().IsNotFound());
auto db_impl = reinterpret_cast<DBImpl*>(db_->GetBaseDB()); return Status::NotFound();
WriteBatchWithIndex* new_batch = new WriteBatchWithIndex(cmp_, 0, true);
Status s = TransactionUtil::CopyFirstN(
savepoint_entries, write_batch_.get(), new_batch, db_impl);
if (!s.ok()) {
// TODO: Should we change this function to return a Status or should we
// somehow make it so RollbackToSavePoint() can never fail?? Not easy to
// handle the case where a client accesses a column family that's been
// dropped.
// After chatting with Siying, I'm going to send a diff that adds
// savepoint support in WriteBatchWithIndex and let reviewers decide which
// approach is cleaner.
fprintf(stderr, "STATUS: %s \n", s.ToString().c_str());
delete new_batch;
} else {
write_batch_.reset(new_batch);
}
num_entries_ = savepoint_entries;
} }
} }
@ -331,7 +296,8 @@ Status TransactionImpl::TryLock(ColumnFamilyHandle* column_family,
// If the key has been previous validated at a sequence number earlier // If the key has been previous validated at a sequence number earlier
// than the curent snapshot's sequence number, we already know it has not // than the curent snapshot's sequence number, we already know it has not
// been modified. // been modified.
bool already_validated = iter->second <= snapshot_->GetSequenceNumber(); SequenceNumber seq = snapshot_->snapshot()->GetSequenceNumber();
bool already_validated = iter->second <= seq;
if (!already_validated) { if (!already_validated) {
s = CheckKeySequence(column_family, key); s = CheckKeySequence(column_family, key);
@ -339,7 +305,7 @@ Status TransactionImpl::TryLock(ColumnFamilyHandle* column_family,
if (s.ok()) { if (s.ok()) {
// Record that there have been no writes to this key after this // Record that there have been no writes to this key after this
// sequence. // sequence.
iter->second = snapshot_->GetSequenceNumber(); iter->second = seq;
} else { } else {
// Failed to validate key // Failed to validate key
if (!previously_locked) { if (!previously_locked) {
@ -369,7 +335,7 @@ Status TransactionImpl::CheckKeySequence(ColumnFamilyHandle* column_family,
result = TransactionUtil::CheckKeyForConflicts( result = TransactionUtil::CheckKeyForConflicts(
db_impl, cfh, key.ToString(), db_impl, cfh, key.ToString(),
snapshot_->GetSequenceNumber()); snapshot_->snapshot()->GetSequenceNumber());
} }
return result; return result;
@ -457,7 +423,6 @@ Status TransactionImpl::Put(ColumnFamilyHandle* column_family, const Slice& key,
if (s.ok()) { if (s.ok()) {
write_batch_->Put(column_family, key, value); write_batch_->Put(column_family, key, value);
num_entries_++;
} }
return s; return s;
@ -469,7 +434,6 @@ Status TransactionImpl::Put(ColumnFamilyHandle* column_family,
if (s.ok()) { if (s.ok()) {
write_batch_->Put(column_family, key, value); write_batch_->Put(column_family, key, value);
num_entries_++;
} }
return s; return s;
@ -481,7 +445,6 @@ Status TransactionImpl::Merge(ColumnFamilyHandle* column_family,
if (s.ok()) { if (s.ok()) {
write_batch_->Merge(column_family, key, value); write_batch_->Merge(column_family, key, value);
num_entries_++;
} }
return s; return s;
@ -493,7 +456,6 @@ Status TransactionImpl::Delete(ColumnFamilyHandle* column_family,
if (s.ok()) { if (s.ok()) {
write_batch_->Delete(column_family, key); write_batch_->Delete(column_family, key);
num_entries_++;
} }
return s; return s;
@ -505,7 +467,6 @@ Status TransactionImpl::Delete(ColumnFamilyHandle* column_family,
if (s.ok()) { if (s.ok()) {
write_batch_->Delete(column_family, key); write_batch_->Delete(column_family, key);
num_entries_++;
} }
return s; return s;
@ -525,7 +486,6 @@ Status TransactionImpl::PutUntracked(ColumnFamilyHandle* column_family,
if (s.ok()) { if (s.ok()) {
write_batch_->Put(column_family, key, value); write_batch_->Put(column_family, key, value);
num_entries_++;
} }
return s; return s;
@ -539,7 +499,6 @@ Status TransactionImpl::PutUntracked(ColumnFamilyHandle* column_family,
if (s.ok()) { if (s.ok()) {
write_batch_->Put(column_family, key, value); write_batch_->Put(column_family, key, value);
num_entries_++;
} }
return s; return s;
@ -552,7 +511,6 @@ Status TransactionImpl::MergeUntracked(ColumnFamilyHandle* column_family,
if (s.ok()) { if (s.ok()) {
write_batch_->Merge(column_family, key, value); write_batch_->Merge(column_family, key, value);
num_entries_++;
} }
return s; return s;
@ -565,7 +523,6 @@ Status TransactionImpl::DeleteUntracked(ColumnFamilyHandle* column_family,
if (s.ok()) { if (s.ok()) {
write_batch_->Delete(column_family, key); write_batch_->Delete(column_family, key);
num_entries_++;
} }
return s; return s;
@ -578,7 +535,6 @@ Status TransactionImpl::DeleteUntracked(ColumnFamilyHandle* column_family,
if (s.ok()) { if (s.ok()) {
write_batch_->Delete(column_family, key); write_batch_->Delete(column_family, key);
num_entries_++;
} }
return s; return s;
@ -586,7 +542,6 @@ Status TransactionImpl::DeleteUntracked(ColumnFamilyHandle* column_family,
void TransactionImpl::PutLogData(const Slice& blob) { void TransactionImpl::PutLogData(const Slice& blob) {
write_batch_->PutLogData(blob); write_batch_->PutLogData(blob);
num_entries_++;
} }
WriteBatchWithIndex* TransactionImpl::GetWriteBatch() { WriteBatchWithIndex* TransactionImpl::GetWriteBatch() {

18
utilities/transactions/transaction_impl.h
View File

@ -16,6 +16,7 @@
#include "db/write_callback.h" #include "db/write_callback.h"
#include "rocksdb/db.h" #include "rocksdb/db.h"
#include "rocksdb/slice.h" #include "rocksdb/slice.h"
#include "rocksdb/snapshot.h"
#include "rocksdb/status.h" #include "rocksdb/status.h"
#include "rocksdb/types.h" #include "rocksdb/types.h"
#include "rocksdb/utilities/transaction.h" #include "rocksdb/utilities/transaction.h"
@ -44,7 +45,7 @@ class TransactionImpl : public Transaction {
void SetSavePoint() override; void SetSavePoint() override;
void RollbackToSavePoint() override; Status RollbackToSavePoint() override;
Status Get(const ReadOptions& options, ColumnFamilyHandle* column_family, Status Get(const ReadOptions& options, ColumnFamilyHandle* column_family,
const Slice& key, std::string* value) override; const Slice& key, std::string* value) override;
@ -151,7 +152,9 @@ class TransactionImpl : public Transaction {
void PutLogData(const Slice& blob) override; void PutLogData(const Slice& blob) override;
const Snapshot* GetSnapshot() const override { return snapshot_; } const Snapshot* GetSnapshot() const override {
return snapshot_ ? snapshot_->snapshot() : nullptr;
}
void SetSnapshot() override; void SetSnapshot() override;
@ -190,7 +193,7 @@ class TransactionImpl : public Transaction {
// If snapshot_ is set, all keys that locked must also have not been written // If snapshot_ is set, all keys that locked must also have not been written
// since this snapshot // since this snapshot
const Snapshot* snapshot_; std::shared_ptr<ManagedSnapshot> snapshot_;
const Comparator* cmp_; const Comparator* cmp_;
@ -214,12 +217,9 @@ class TransactionImpl : public Transaction {
// stored. // stored.
TransactionKeyMap tracked_keys_; TransactionKeyMap tracked_keys_;
// Records the number of entries currently in the WriteBatch include calls to // Stack of the Snapshot saved at each save point. Saved snapshots may be
// PutLogData() // nullptr if there was no snapshot at the time SetSavePoint() was called.
size_t num_entries_ = 0; std::unique_ptr<std::stack<std::shared_ptr<ManagedSnapshot>>> save_points_;
// Stack of number of entries in write_batch at each save point
std::unique_ptr<std::stack<size_t>> save_points_;
Status TryLock(ColumnFamilyHandle* column_family, const Slice& key, Status TryLock(ColumnFamilyHandle* column_family, const Slice& key,
bool check_snapshot = true); bool check_snapshot = true);

17
utilities/transactions/transaction_test.cc
View File

@ -1299,12 +1299,14 @@ TEST_F(TransactionTest, SavepointTest) {
Transaction* txn = db->BeginTransaction(write_options); Transaction* txn = db->BeginTransaction(write_options);
ASSERT_TRUE(txn); ASSERT_TRUE(txn);
txn->RollbackToSavePoint(); s = txn->RollbackToSavePoint();
ASSERT_TRUE(s.IsNotFound());
txn->SetSavePoint(); // 1 txn->SetSavePoint(); // 1
txn->RollbackToSavePoint(); // Rollback to beginning of txn ASSERT_OK(txn->RollbackToSavePoint()); // Rollback to beginning of txn
txn->RollbackToSavePoint(); s = txn->RollbackToSavePoint();
ASSERT_TRUE(s.IsNotFound());
s = txn->Put("B", "b"); s = txn->Put("B", "b");
ASSERT_OK(s); ASSERT_OK(s);
@ -1340,7 +1342,7 @@ TEST_F(TransactionTest, SavepointTest) {
s = txn->Put("D", "d"); s = txn->Put("D", "d");
ASSERT_OK(s); ASSERT_OK(s);
txn->RollbackToSavePoint(); // Rollback to 2 ASSERT_OK(txn->RollbackToSavePoint()); // Rollback to 2
s = txn->Get(read_options, "A", &value); s = txn->Get(read_options, "A", &value);
ASSERT_OK(s); ASSERT_OK(s);
@ -1363,7 +1365,10 @@ TEST_F(TransactionTest, SavepointTest) {
s = txn->Put("E", "e"); s = txn->Put("E", "e");
ASSERT_OK(s); ASSERT_OK(s);
txn->RollbackToSavePoint(); // Rollback to beginning of txn // Rollback to beginning of txn
s = txn->RollbackToSavePoint();
ASSERT_TRUE(s.IsNotFound());
txn->Rollback();
s = txn->Get(read_options, "A", &value); s = txn->Get(read_options, "A", &value);
ASSERT_TRUE(s.IsNotFound()); ASSERT_TRUE(s.IsNotFound());
@ -1409,7 +1414,7 @@ TEST_F(TransactionTest, SavepointTest) {
s = txn->Get(read_options, "B", &value); s = txn->Get(read_options, "B", &value);
ASSERT_TRUE(s.IsNotFound()); ASSERT_TRUE(s.IsNotFound());
txn->RollbackToSavePoint(); // Rollback to 3 ASSERT_OK(txn->RollbackToSavePoint()); // Rollback to 3
s = txn->Get(read_options, "F", &value); s = txn->Get(read_options, "F", &value);
ASSERT_OK(s); ASSERT_OK(s);

119
utilities/transactions/transaction_util.cc
View File

@ -141,125 +141,6 @@ Status TransactionUtil::CheckKeysForConflicts(DBImpl* db_impl,
return result; return result;
} }
Status TransactionUtil::CopyFirstN(size_t num, WriteBatchWithIndex* batch,
WriteBatchWithIndex* new_batch,
DBImpl* db_impl) {
// Handler for iterating through batch and copying entries to new_batch
class Handler : public WriteBatch::Handler {
public:
WriteBatchWithIndex* batch;
const size_t limit;
DBImpl* db_impl;
size_t seen = 0;
std::unordered_map<uint32_t, SuperVersion*> super_versions;
std::unordered_map<uint32_t, ColumnFamilyHandle*> handles;
Handler(WriteBatchWithIndex* dest, size_t new_limit, DBImpl* db)
: batch(dest), limit(new_limit), db_impl(db) {}
~Handler() {
for (auto& iter : super_versions) {
db_impl->ReturnAndCleanupSuperVersionUnlocked(iter.first, iter.second);
}
}
Status GetColumnFamily(uint32_t column_family_id,
ColumnFamilyHandle** cfh) {
// Need to look up ColumnFamilyHandle for this column family id. Since
// doing this requires grabbing a mutex, lets only do it once per column
// family and cache it.
// In order to ensure that the ColumnFamilyHandle is still valid, we need
// to hold the superversion.
const auto& iter = handles.find(column_family_id);
if (iter == handles.end()) {
// Don't have ColumnFamilyHandle cached, look it up from the db.
SuperVersion* sv =
db_impl->GetAndRefSuperVersionUnlocked(column_family_id);
if (sv == nullptr) {
return Status::InvalidArgument(
"Could not find column family for ID " +
ToString(column_family_id));
}
super_versions.insert({column_family_id, sv});
*cfh = db_impl->GetColumnFamilyHandleUnlocked(column_family_id);
if (*cfh == nullptr) {
return Status::InvalidArgument(
"Could not find column family handle for ID " +
ToString(column_family_id));
}
handles.insert({column_family_id, *cfh});
} else {
*cfh = iter->second;
}
return Status::OK();
}
virtual Status PutCF(uint32_t column_family_id, const Slice& key,
const Slice& value) override {
if (seen >= limit) {
// Found the first N entries, return Aborted to stop the Iteration.
return Status::Aborted();
}
ColumnFamilyHandle* cfh = nullptr;
Status s = GetColumnFamily(column_family_id, &cfh);
if (s.ok()) {
batch->Put(cfh, key, value);
}
seen++;
return s;
}
virtual Status MergeCF(uint32_t column_family_id, const Slice& key,
const Slice& value) override {
if (seen >= limit) {
// Found the first N entries, return Aborted to stop the Iteration.
return Status::Aborted();
}
ColumnFamilyHandle* cfh = nullptr;
Status s = GetColumnFamily(column_family_id, &cfh);
if (s.ok()) {
batch->Merge(cfh, key, value);
}
seen++;
return s;
}
virtual Status DeleteCF(uint32_t column_family_id,
const Slice& key) override {
if (seen >= limit) {
// Found the first N entries, return Aborted to stop the Iteration.
return Status::Aborted();
}
ColumnFamilyHandle* cfh = nullptr;
Status s = GetColumnFamily(column_family_id, &cfh);
if (s.ok()) {
batch->Delete(cfh, key);
}
seen++;
return s;
}
virtual void LogData(const Slice& blob) override {
if (seen < limit) {
batch->PutLogData(blob);
}
seen++;
}
};
// Iterating on this handler will add all keys in this batch into a new batch
// up to
// the limit.
Handler handler(new_batch, num, db_impl);
Status s = batch->GetWriteBatch()->Iterate(&handler);
if (s.IsAborted()) {
// Handler returns Aborted when it is done copying to stop the iteration.
s = Status::OK();
}
return s;
}
} // namespace rocksdb } // namespace rocksdb

6
utilities/transactions/transaction_util.h
View File

@ -48,12 +48,6 @@ class TransactionUtil {
// mutex is held. // mutex is held.
static Status CheckKeysForConflicts(DBImpl* db_impl, TransactionKeyMap* keys); static Status CheckKeysForConflicts(DBImpl* db_impl, TransactionKeyMap* keys);
// Copies the first num entries from batch into new_batch (including Put,
// Merge, Delete, and PutLogData).
// Returns non-OK on error.
static Status CopyFirstN(size_t num, WriteBatchWithIndex* batch,
WriteBatchWithIndex* new_batch, DBImpl* db_impl);
private: private:
static Status CheckKey(DBImpl* db_impl, SuperVersion* sv, static Status CheckKey(DBImpl* db_impl, SuperVersion* sv,
SequenceNumber earliest_seq, SequenceNumber key_seq, SequenceNumber earliest_seq, SequenceNumber key_seq,