mirror of
https://github.com/facebook/rocksdb.git
synced 2026-07-07 14:47:40 +08:00
Compare commits
11 Commits
f2c0eb41ef
...
v3.12.1
| Author | SHA1 | Date | |
|---|---|---|---|
| c03f2462a8 | |||
| 9de6ee2820 | |||
| d99227db23 | |||
| 28d3aa3667 | |||
| 79c2075b53 | |||
| 0640c28723 | |||
| 718309432b | |||
| bc672aefb3 | |||
| b565228e1e | |||
| dc3cdfede9 | |||
| 8856b22edd |
@@ -1,10 +1,16 @@
|
||||
# Rocksdb Change Log
|
||||
|
||||
## 3.12.1 (7/16/2015)
|
||||
* Fix data loss after DB recovery by not allowing flush/compaction to be scheduled until DB opened
|
||||
|
||||
## 3.12.0 (7/2/2015)
|
||||
### New Features
|
||||
* Added experimental support for optimistic transactions. See include/rocksdb/utilities/optimistic_transaction.h for more info.
|
||||
* Added a new way to report QPS from db_bench (check out --report_file and --report_interval_seconds)
|
||||
* Added a cache for individual rows. See DBOptions::row_cache for more info.
|
||||
* Several new features on EventListener (see include/rocksdb/listener.h):
|
||||
- OnCompationCompleted() now returns per-compaciton job statistics, defined in include/rocksdb/compaction_job_stats.h.
|
||||
- Added OnTableFileCreated() and OnTableFileDeleted().
|
||||
|
||||
### Public API changes
|
||||
* EventListener::OnFlushCompleted() now passes FlushJobInfo instead of a list of parameters.
|
||||
|
||||
@@ -59,6 +59,7 @@ ColumnFamilyHandleImpl::~ColumnFamilyHandleImpl() {
|
||||
if (job_context.HaveSomethingToDelete()) {
|
||||
db_->PurgeObsoleteFiles(job_context);
|
||||
}
|
||||
job_context.Clean();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -76,11 +76,10 @@ class CompactionJobTest : public testing::Test {
|
||||
largest = internal_key;
|
||||
largest_seqno = sequence_number;
|
||||
}
|
||||
std::pair<std::string, std::string> key_value(
|
||||
{bottommost_internal_key.Encode().ToString(), value});
|
||||
contents.insert(key_value);
|
||||
contents.insert({internal_key.Encode().ToString(), value});
|
||||
if (i == 1 || k < kKeysPerFile / 2) {
|
||||
expected_results.insert(key_value);
|
||||
expected_results.insert(
|
||||
{bottommost_internal_key.Encode().ToString(), value});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -97,7 +96,7 @@ class CompactionJobTest : public testing::Test {
|
||||
mutable_cf_options_, &edit, &mutex_);
|
||||
mutex_.Unlock();
|
||||
}
|
||||
versions_->SetLastSequence(sequence_number);
|
||||
versions_->SetLastSequence(sequence_number + 1);
|
||||
return expected_results;
|
||||
}
|
||||
|
||||
|
||||
+31
-21
@@ -527,6 +527,25 @@ void DBImpl::FindObsoleteFiles(JobContext* job_context, bool force,
|
||||
versions_->GetObsoleteFiles(&job_context->sst_delete_files,
|
||||
job_context->min_pending_output);
|
||||
|
||||
uint64_t min_log_number = versions_->MinLogNumber();
|
||||
if (!alive_log_files_.empty()) {
|
||||
// find newly obsoleted log files
|
||||
while (alive_log_files_.begin()->number < min_log_number) {
|
||||
auto& earliest = *alive_log_files_.begin();
|
||||
job_context->log_delete_files.push_back(earliest.number);
|
||||
total_log_size_ -= earliest.size;
|
||||
alive_log_files_.pop_front();
|
||||
// Current log should always stay alive since it can't have
|
||||
// number < MinLogNumber().
|
||||
assert(alive_log_files_.size());
|
||||
}
|
||||
}
|
||||
|
||||
// We're just cleaning up for DB::Write().
|
||||
assert(job_context->logs_to_free.empty());
|
||||
job_context->logs_to_free = logs_to_free_;
|
||||
logs_to_free_.clear();
|
||||
|
||||
// store the current filenum, lognum, etc
|
||||
job_context->manifest_file_number = versions_->manifest_file_number();
|
||||
job_context->pending_manifest_file_number =
|
||||
@@ -1309,17 +1328,6 @@ Status DBImpl::FlushMemTableToOutputFile(
|
||||
VersionStorageInfo::LevelSummaryStorage tmp;
|
||||
LogToBuffer(log_buffer, "[%s] Level summary: %s\n", cfd->GetName().c_str(),
|
||||
cfd->current()->storage_info()->LevelSummary(&tmp));
|
||||
|
||||
if (disable_delete_obsolete_files_ == 0) {
|
||||
// add to deletion state
|
||||
while (alive_log_files_.size() &&
|
||||
alive_log_files_.begin()->number < versions_->MinLogNumber()) {
|
||||
const auto& earliest = *alive_log_files_.begin();
|
||||
job_context->log_delete_files.push_back(earliest.number);
|
||||
total_log_size_ -= earliest.size;
|
||||
alive_log_files_.pop_front();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (!s.ok() && !s.IsShutdownInProgress() && db_options_.paranoid_checks &&
|
||||
@@ -2048,6 +2056,10 @@ Status DBImpl::WaitForFlushMemTable(ColumnFamilyData* cfd) {
|
||||
|
||||
void DBImpl::MaybeScheduleFlushOrCompaction() {
|
||||
mutex_.AssertHeld();
|
||||
if (!opened_successfully_) {
|
||||
// Compaction may introduce data race to DB open
|
||||
return;
|
||||
}
|
||||
if (bg_work_gate_closed_) {
|
||||
// gate closed for background work
|
||||
return;
|
||||
@@ -2145,7 +2157,9 @@ void DBImpl::RecordFlushIOStats() {
|
||||
|
||||
void DBImpl::BGWorkFlush(void* db) {
|
||||
IOSTATS_SET_THREAD_POOL_ID(Env::Priority::HIGH);
|
||||
TEST_SYNC_POINT("DBImpl::BGWorkFlush");
|
||||
reinterpret_cast<DBImpl*>(db)->BackgroundCallFlush();
|
||||
TEST_SYNC_POINT("DBImpl::BGWorkFlush:done");
|
||||
}
|
||||
|
||||
void DBImpl::BGWorkCompaction(void* db) {
|
||||
@@ -2238,10 +2252,6 @@ void DBImpl::BackgroundCallFlush() {
|
||||
|
||||
ReleaseFileNumberFromPendingOutputs(pending_outputs_inserted_elem);
|
||||
|
||||
// We're just cleaning up for DB::Write()
|
||||
job_context.logs_to_free = logs_to_free_;
|
||||
logs_to_free_.clear();
|
||||
|
||||
// If flush failed, we want to delete all temporary files that we might have
|
||||
// created. Thus, we force full scan in FindObsoleteFiles()
|
||||
FindObsoleteFiles(&job_context, !s.ok() && !s.IsShutdownInProgress());
|
||||
@@ -2308,10 +2318,6 @@ void DBImpl::BackgroundCallCompaction() {
|
||||
|
||||
ReleaseFileNumberFromPendingOutputs(pending_outputs_inserted_elem);
|
||||
|
||||
// We're just cleaning up for DB::Write()
|
||||
job_context.logs_to_free = logs_to_free_;
|
||||
logs_to_free_.clear();
|
||||
|
||||
// If compaction failed, we want to delete all temporary files that we might
|
||||
// have created (they might not be all recorded in job_context in case of a
|
||||
// failure). Thus, we force full scan in FindObsoleteFiles()
|
||||
@@ -2597,6 +2603,7 @@ Status DBImpl::BackgroundCompaction(bool* madeProgress, JobContext* job_context,
|
||||
|
||||
mutex_.Unlock();
|
||||
status = compaction_job.Run();
|
||||
TEST_SYNC_POINT("DBImpl::BackgroundCompaction:NonTrivial:AfterRun");
|
||||
mutex_.Lock();
|
||||
|
||||
compaction_job.Install(&status, *c->mutable_cf_options(), &mutex_);
|
||||
@@ -4353,11 +4360,14 @@ Status DB::Open(const DBOptions& db_options, const std::string& dbname,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
TEST_SYNC_POINT("DBImpl::Open:Opened");
|
||||
if (s.ok()) {
|
||||
impl->opened_successfully_ = true;
|
||||
impl->MaybeScheduleFlushOrCompaction();
|
||||
}
|
||||
impl->mutex_.Unlock();
|
||||
|
||||
if (s.ok()) {
|
||||
impl->opened_successfully_ = true;
|
||||
Log(InfoLogLevel::INFO_LEVEL, impl->db_options_.info_log, "DB pointer %p",
|
||||
impl);
|
||||
*dbptr = impl;
|
||||
|
||||
@@ -290,6 +290,8 @@ class DBImpl : public DB {
|
||||
|
||||
size_t TEST_LogsToFreeSize();
|
||||
|
||||
uint64_t TEST_LogfileNumber();
|
||||
|
||||
#endif // ROCKSDB_LITE
|
||||
|
||||
// Returns the list of live files in 'live' and the list
|
||||
|
||||
@@ -148,5 +148,10 @@ size_t DBImpl::TEST_LogsToFreeSize() {
|
||||
return logs_to_free_.size();
|
||||
}
|
||||
|
||||
uint64_t DBImpl::TEST_LogfileNumber() {
|
||||
InstrumentedMutexLock l(&mutex_);
|
||||
return logfile_number_;
|
||||
}
|
||||
|
||||
} // namespace rocksdb
|
||||
#endif // ROCKSDB_LITE
|
||||
|
||||
+2
-5
@@ -350,9 +350,6 @@ void DBIter::MergeValuesNewToOld() {
|
||||
void DBIter::Prev() {
|
||||
assert(valid_);
|
||||
if (direction_ == kForward) {
|
||||
if (!iter_->Valid()) {
|
||||
iter_->SeekToLast();
|
||||
}
|
||||
FindPrevUserKey();
|
||||
direction_ = kReverse;
|
||||
}
|
||||
@@ -556,7 +553,7 @@ void DBIter::FindNextUserKey() {
|
||||
ParsedInternalKey ikey;
|
||||
FindParseableKey(&ikey, kForward);
|
||||
while (iter_->Valid() &&
|
||||
user_comparator_->Compare(ikey.user_key, saved_key_.GetKey()) <= 0) {
|
||||
user_comparator_->Compare(ikey.user_key, saved_key_.GetKey()) != 0) {
|
||||
iter_->Next();
|
||||
FindParseableKey(&ikey, kForward);
|
||||
}
|
||||
@@ -571,7 +568,7 @@ void DBIter::FindPrevUserKey() {
|
||||
ParsedInternalKey ikey;
|
||||
FindParseableKey(&ikey, kReverse);
|
||||
while (iter_->Valid() &&
|
||||
user_comparator_->Compare(ikey.user_key, saved_key_.GetKey()) >= 0) {
|
||||
user_comparator_->Compare(ikey.user_key, saved_key_.GetKey()) == 0) {
|
||||
if (num_skipped >= max_skip_) {
|
||||
num_skipped = 0;
|
||||
IterKey last_key;
|
||||
|
||||
+6
-2
@@ -1668,7 +1668,9 @@ TEST_F(DBIteratorTest, DBIterator8) {
|
||||
ASSERT_EQ(db_iter->value().ToString(), "0");
|
||||
}
|
||||
|
||||
TEST_F(DBIteratorTest, DBIterator9) {
|
||||
// TODO(3.13): fix the issue of Seek() then Prev() which might not necessary
|
||||
// return the biggest element smaller than the seek key.
|
||||
TEST_F(DBIteratorTest, DISABLED_DBIterator9) {
|
||||
Options options;
|
||||
options.merge_operator = MergeOperators::CreateFromStringId("stringappend");
|
||||
{
|
||||
@@ -1716,7 +1718,9 @@ TEST_F(DBIteratorTest, DBIterator9) {
|
||||
}
|
||||
}
|
||||
|
||||
TEST_F(DBIteratorTest, DBIterator10) {
|
||||
// TODO(3.13): fix the issue of Seek() then Prev() which might not necessary
|
||||
// return the biggest element smaller than the seek key.
|
||||
TEST_F(DBIteratorTest, DISABLED_DBIterator10) {
|
||||
Options options;
|
||||
|
||||
TestIterator* internal_iter = new TestIterator(BytewiseComparator());
|
||||
|
||||
+41
-4
@@ -7579,7 +7579,9 @@ TEST_F(DBTest, DropWrites) {
|
||||
ASSERT_LT(CountFiles(), num_files + 3);
|
||||
|
||||
// Check that compaction attempts slept after errors
|
||||
ASSERT_TRUE(env_->sleep_counter_.WaitFor(5));
|
||||
// TODO @krad: Figure out why ASSERT_EQ 5 keeps failing in certain compiler
|
||||
// versions
|
||||
ASSERT_GE(env_->sleep_counter_.Read(), 4);
|
||||
} while (ChangeCompactOptions());
|
||||
}
|
||||
|
||||
@@ -8540,7 +8542,6 @@ TEST_F(DBTest, TransactionLogIterator) {
|
||||
} while (ChangeCompactOptions());
|
||||
}
|
||||
|
||||
#ifndef NDEBUG // sync point is not included with DNDEBUG build
|
||||
TEST_F(DBTest, TransactionLogIteratorRace) {
|
||||
static const int LOG_ITERATOR_RACE_TEST_COUNT = 2;
|
||||
static const char* sync_points[LOG_ITERATOR_RACE_TEST_COUNT][4] = {
|
||||
@@ -8595,7 +8596,6 @@ TEST_F(DBTest, TransactionLogIteratorRace) {
|
||||
} while (ChangeCompactOptions());
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
TEST_F(DBTest, TransactionLogIteratorStallAtLastRecord) {
|
||||
do {
|
||||
@@ -12672,6 +12672,7 @@ TEST_F(DBTest, DontDeletePendingOutputs) {
|
||||
dbfull()->FindObsoleteFiles(&job_context, true /*force*/);
|
||||
dbfull()->TEST_UnlockMutex();
|
||||
dbfull()->PurgeObsoleteFiles(job_context);
|
||||
job_context.Clean();
|
||||
};
|
||||
|
||||
env_->table_write_callback_ = &purge_obsolete_files_function;
|
||||
@@ -14113,7 +14114,9 @@ TEST_F(DBTest, RowCache) {
|
||||
ASSERT_EQ(TestGetTickerCount(options, ROW_CACHE_MISS), 1);
|
||||
}
|
||||
|
||||
TEST_F(DBTest, PrevAfterMerge) {
|
||||
// TODO(3.13): fix the issue of Seek() + Prev() which might not necessary
|
||||
// return the biggest key which is smaller than the seek key.
|
||||
TEST_F(DBTest, DISABLED_PrevAfterMerge) {
|
||||
Options options;
|
||||
options.create_if_missing = true;
|
||||
options.merge_operator = MergeOperators::CreatePutOperator();
|
||||
@@ -14136,6 +14139,40 @@ TEST_F(DBTest, PrevAfterMerge) {
|
||||
ASSERT_EQ("1", it->key().ToString());
|
||||
}
|
||||
|
||||
TEST_F(DBTest, DeletingOldWalAfterDrop) {
|
||||
rocksdb::SyncPoint::GetInstance()->LoadDependency(
|
||||
{ { "Test:AllowFlushes", "DBImpl::BGWorkFlush" },
|
||||
{ "DBImpl::BGWorkFlush:done", "Test:WaitForFlush"} });
|
||||
rocksdb::SyncPoint::GetInstance()->ClearTrace();
|
||||
|
||||
rocksdb::SyncPoint::GetInstance()->DisableProcessing();
|
||||
Options options = CurrentOptions();
|
||||
options.max_total_wal_size = 8192;
|
||||
options.compression = kNoCompression;
|
||||
options.write_buffer_size = 1 << 20;
|
||||
options.level0_file_num_compaction_trigger = (1<<30);
|
||||
options.level0_slowdown_writes_trigger = (1<<30);
|
||||
options.level0_stop_writes_trigger = (1<<30);
|
||||
options.disable_auto_compactions = true;
|
||||
DestroyAndReopen(options);
|
||||
rocksdb::SyncPoint::GetInstance()->EnableProcessing();
|
||||
|
||||
CreateColumnFamilies({"cf1", "cf2"}, options);
|
||||
ASSERT_OK(Put(0, "key1", DummyString(8192)));
|
||||
ASSERT_OK(Put(0, "key2", DummyString(8192)));
|
||||
// the oldest wal should now be getting_flushed
|
||||
ASSERT_OK(db_->DropColumnFamily(handles_[0]));
|
||||
// all flushes should now do nothing because their CF is dropped
|
||||
TEST_SYNC_POINT("Test:AllowFlushes");
|
||||
TEST_SYNC_POINT("Test:WaitForFlush");
|
||||
uint64_t lognum1 = dbfull()->TEST_LogfileNumber();
|
||||
ASSERT_OK(Put(1, "key3", DummyString(8192)));
|
||||
ASSERT_OK(Put(1, "key4", DummyString(8192)));
|
||||
// new wal should have been created
|
||||
uint64_t lognum2 = dbfull()->TEST_LogfileNumber();
|
||||
EXPECT_GT(lognum2, lognum1);
|
||||
}
|
||||
|
||||
} // namespace rocksdb
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
|
||||
+86
-16
@@ -25,6 +25,7 @@
|
||||
#include "util/logging.h"
|
||||
#include "util/mock_env.h"
|
||||
#include "util/mutexlock.h"
|
||||
#include "util/sync_point.h"
|
||||
#include "util/testharness.h"
|
||||
#include "util/testutil.h"
|
||||
|
||||
@@ -185,6 +186,13 @@ class FaultInjectionTestEnv : public EnvWrapper {
|
||||
Status NewWritableFile(const std::string& fname,
|
||||
unique_ptr<WritableFile>* result,
|
||||
const EnvOptions& soptions) override {
|
||||
if (!IsFilesystemActive()) {
|
||||
return Status::Corruption("Not Active");
|
||||
}
|
||||
// Not allow overwriting files
|
||||
if (target()->FileExists(fname)) {
|
||||
return Status::Corruption("File already exists.");
|
||||
}
|
||||
Status s = target()->NewWritableFile(fname, result, soptions);
|
||||
if (s.ok()) {
|
||||
result->reset(new TestWritableFile(fname, std::move(*result), this));
|
||||
@@ -201,6 +209,9 @@ class FaultInjectionTestEnv : public EnvWrapper {
|
||||
}
|
||||
|
||||
virtual Status DeleteFile(const std::string& f) override {
|
||||
if (!IsFilesystemActive()) {
|
||||
return Status::Corruption("Not Active");
|
||||
}
|
||||
Status s = EnvWrapper::DeleteFile(f);
|
||||
if (!s.ok()) {
|
||||
fprintf(stderr, "Cannot delete file %s: %s\n", f.c_str(),
|
||||
@@ -215,6 +226,9 @@ class FaultInjectionTestEnv : public EnvWrapper {
|
||||
|
||||
virtual Status RenameFile(const std::string& s,
|
||||
const std::string& t) override {
|
||||
if (!IsFilesystemActive()) {
|
||||
return Status::Corruption("Not Active");
|
||||
}
|
||||
Status ret = EnvWrapper::RenameFile(s, t);
|
||||
|
||||
if (ret.ok()) {
|
||||
@@ -373,8 +387,11 @@ TestWritableFile::~TestWritableFile() {
|
||||
}
|
||||
|
||||
Status TestWritableFile::Append(const Slice& data) {
|
||||
if (!env_->IsFilesystemActive()) {
|
||||
return Status::Corruption("Not Active");
|
||||
}
|
||||
Status s = target_->Append(data);
|
||||
if (s.ok() && env_->IsFilesystemActive()) {
|
||||
if (s.ok()) {
|
||||
state_.pos_ += data.size();
|
||||
}
|
||||
return s;
|
||||
@@ -544,33 +561,34 @@ class FaultInjectionTest : public testing::Test {
|
||||
ASSERT_OK(s);
|
||||
}
|
||||
|
||||
void Build(const WriteOptions& write_options, int start_idx, int num_vals) {
|
||||
void Build(const WriteOptions& write_options, int start_idx, int num_vals,
|
||||
bool sequential = true) {
|
||||
std::string key_space, value_space;
|
||||
WriteBatch batch;
|
||||
for (int i = start_idx; i < start_idx + num_vals; i++) {
|
||||
Slice key = Key(i, &key_space);
|
||||
Slice key = Key(sequential, i, &key_space);
|
||||
batch.Clear();
|
||||
batch.Put(key, Value(i, &value_space));
|
||||
ASSERT_OK(db_->Write(write_options, &batch));
|
||||
}
|
||||
}
|
||||
|
||||
Status ReadValue(int i, std::string* val) const {
|
||||
Status ReadValue(int i, std::string* val, bool sequential) const {
|
||||
std::string key_space, value_space;
|
||||
Slice key = Key(i, &key_space);
|
||||
Slice key = Key(sequential, i, &key_space);
|
||||
Value(i, &value_space);
|
||||
ReadOptions options;
|
||||
return db_->Get(options, key, val);
|
||||
}
|
||||
|
||||
Status Verify(int start_idx, int num_vals,
|
||||
ExpectedVerifResult expected) const {
|
||||
Status Verify(int start_idx, int num_vals, ExpectedVerifResult expected,
|
||||
bool seqeuntial = true) const {
|
||||
std::string val;
|
||||
std::string value_space;
|
||||
Status s;
|
||||
for (int i = start_idx; i < start_idx + num_vals && s.ok(); i++) {
|
||||
Value(i, &value_space);
|
||||
s = ReadValue(i, &val);
|
||||
s = ReadValue(i, &val, seqeuntial);
|
||||
if (s.ok()) {
|
||||
EXPECT_EQ(value_space, val);
|
||||
}
|
||||
@@ -590,9 +608,16 @@ class FaultInjectionTest : public testing::Test {
|
||||
}
|
||||
|
||||
// Return the ith key
|
||||
Slice Key(int i, std::string* storage) const {
|
||||
Slice Key(bool seqeuntial, int i, std::string* storage) const {
|
||||
int num = i;
|
||||
if (!seqeuntial) {
|
||||
// random transfer
|
||||
const int m = 0x5bd1e995;
|
||||
num *= m;
|
||||
num ^= num << 24;
|
||||
}
|
||||
char buf[100];
|
||||
snprintf(buf, sizeof(buf), "%016d", i);
|
||||
snprintf(buf, sizeof(buf), "%016d", num);
|
||||
storage->assign(buf, strlen(buf));
|
||||
return Slice(*storage);
|
||||
}
|
||||
@@ -772,14 +797,14 @@ TEST_F(FaultInjectionTest, DISABLED_WriteOptionSyncTest) {
|
||||
write_options.sync = false;
|
||||
|
||||
std::string key_space, value_space;
|
||||
ASSERT_OK(
|
||||
db_->Put(write_options, Key(1, &key_space), Value(1, &value_space)));
|
||||
ASSERT_OK(db_->Put(write_options, Key(true, 1, &key_space),
|
||||
Value(1, &value_space)));
|
||||
FlushOptions flush_options;
|
||||
flush_options.wait = false;
|
||||
ASSERT_OK(db_->Flush(flush_options));
|
||||
write_options.sync = true;
|
||||
ASSERT_OK(
|
||||
db_->Put(write_options, Key(2, &key_space), Value(2, &value_space)));
|
||||
ASSERT_OK(db_->Put(write_options, Key(true, 2, &key_space),
|
||||
Value(2, &value_space)));
|
||||
|
||||
env_->SetFilesystemActive(false);
|
||||
NoWriteTestReopenWithFault(kResetDropAndDeleteUnsynced);
|
||||
@@ -788,14 +813,59 @@ TEST_F(FaultInjectionTest, DISABLED_WriteOptionSyncTest) {
|
||||
ASSERT_OK(OpenDB());
|
||||
std::string val;
|
||||
Value(2, &value_space);
|
||||
ASSERT_OK(ReadValue(2, &val));
|
||||
ASSERT_OK(ReadValue(2, &val, true));
|
||||
ASSERT_EQ(value_space, val);
|
||||
|
||||
Value(1, &value_space);
|
||||
ASSERT_OK(ReadValue(1, &val));
|
||||
ASSERT_OK(ReadValue(1, &val, true));
|
||||
ASSERT_EQ(value_space, val);
|
||||
}
|
||||
|
||||
TEST_F(FaultInjectionTest, UninstalledCompaction) {
|
||||
options_.target_file_size_base = 32 * 1024;
|
||||
options_.write_buffer_size = 100 << 10; // 100KB
|
||||
options_.level0_file_num_compaction_trigger = 6;
|
||||
options_.level0_stop_writes_trigger = 1 << 10;
|
||||
options_.level0_slowdown_writes_trigger = 1 << 10;
|
||||
options_.max_background_compactions = 1;
|
||||
OpenDB();
|
||||
|
||||
rocksdb::SyncPoint::GetInstance()->LoadDependency({
|
||||
{"FaultInjectionTest::FaultTest:0", "DBImpl::BGWorkCompaction"},
|
||||
{"CompactionJob::Run():End", "FaultInjectionTest::FaultTest:1"},
|
||||
{"FaultInjectionTest::FaultTest:2",
|
||||
"DBImpl::BackgroundCompaction:NonTrivial:AfterRun"},
|
||||
});
|
||||
rocksdb::SyncPoint::GetInstance()->EnableProcessing();
|
||||
|
||||
int kNumKeys = 1000;
|
||||
Build(WriteOptions(), 0, kNumKeys, false);
|
||||
FlushOptions flush_options;
|
||||
flush_options.wait = true;
|
||||
db_->Flush(flush_options);
|
||||
ASSERT_OK(db_->Put(WriteOptions(), "", ""));
|
||||
TEST_SYNC_POINT("FaultInjectionTest::FaultTest:0");
|
||||
TEST_SYNC_POINT("FaultInjectionTest::FaultTest:1");
|
||||
env_->SetFilesystemActive(false);
|
||||
TEST_SYNC_POINT("FaultInjectionTest::FaultTest:2");
|
||||
CloseDB();
|
||||
rocksdb::SyncPoint::GetInstance()->DisableProcessing();
|
||||
ResetDBState(kResetDropUnsyncedData);
|
||||
|
||||
std::atomic<bool> opened(false);
|
||||
rocksdb::SyncPoint::GetInstance()->SetCallBack(
|
||||
"DBImpl::Open:Opened", [&](void* arg) { opened.store(true); });
|
||||
rocksdb::SyncPoint::GetInstance()->SetCallBack(
|
||||
"DBImpl::BGWorkCompaction",
|
||||
[&](void* arg) { ASSERT_TRUE(opened.load()); });
|
||||
rocksdb::SyncPoint::GetInstance()->EnableProcessing();
|
||||
ASSERT_OK(OpenDB());
|
||||
static_cast<DBImpl*>(db_)->TEST_WaitForCompact();
|
||||
ASSERT_OK(Verify(0, kNumKeys, FaultInjectionTest::kValExpectFound, false));
|
||||
ASSERT_OK(db_->Put(WriteOptions(), "", ""));
|
||||
rocksdb::SyncPoint::GetInstance()->DisableProcessing();
|
||||
}
|
||||
|
||||
} // namespace rocksdb
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
|
||||
@@ -169,6 +169,7 @@ void ForwardIterator::Cleanup(bool release_sv) {
|
||||
if (job_context.HaveSomethingToDelete()) {
|
||||
db_->PurgeObsoleteFiles(job_context);
|
||||
}
|
||||
job_context.Clean();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -83,6 +83,10 @@ struct JobContext {
|
||||
new_superversion = create_superversion ? new SuperVersion() : nullptr;
|
||||
}
|
||||
|
||||
// For non-empty JobContext Clean() has to be called at least once before
|
||||
// before destruction (see asserts in ~JobContext()). Should be called with
|
||||
// unlocked DB mutex. Destructor doesn't call Clean() to avoid accidentally
|
||||
// doing potentially slow Clean() with locked DB mutex.
|
||||
void Clean() {
|
||||
// free pending memtables
|
||||
for (auto m : memtables_to_free) {
|
||||
@@ -109,6 +113,7 @@ struct JobContext {
|
||||
assert(memtables_to_free.size() == 0);
|
||||
assert(superversions_to_free.size() == 0);
|
||||
assert(new_superversion == nullptr);
|
||||
assert(logs_to_free.size() == 0);
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
+3
-1
@@ -612,7 +612,9 @@ class VersionSet {
|
||||
uint64_t MinLogNumber() const {
|
||||
uint64_t min_log_num = std::numeric_limits<uint64_t>::max();
|
||||
for (auto cfd : *column_family_set_) {
|
||||
if (min_log_num > cfd->GetLogNumber()) {
|
||||
// It's safe to ignore dropped column families here:
|
||||
// cfd->IsDropped() becomes true after the drop is persisted in MANIFEST.
|
||||
if (min_log_num > cfd->GetLogNumber() && !cfd->IsDropped()) {
|
||||
min_log_num = cfd->GetLogNumber();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6,7 +6,7 @@
|
||||
|
||||
#define ROCKSDB_MAJOR 3
|
||||
#define ROCKSDB_MINOR 12
|
||||
#define ROCKSDB_PATCH 0
|
||||
#define ROCKSDB_PATCH 1
|
||||
|
||||
// Do not use these. We made the mistake of declaring macros starting with
|
||||
// double underscore. Now we have to live with our choice. We'll deprecate these
|
||||
|
||||
@@ -38,7 +38,6 @@
|
||||
|
||||
namespace rocksdb {
|
||||
|
||||
namespace {
|
||||
class BackupRateLimiter {
|
||||
public:
|
||||
BackupRateLimiter(Env* env, uint64_t max_bytes_per_second,
|
||||
@@ -79,7 +78,6 @@ class BackupRateLimiter {
|
||||
uint64_t bytes_since_start_;
|
||||
static const uint64_t kMicrosInSecond = 1000 * 1000LL;
|
||||
};
|
||||
} // namespace
|
||||
|
||||
void BackupStatistics::IncrementNumberSuccessBackup() {
|
||||
number_success_backup++;
|
||||
@@ -444,26 +442,6 @@ BackupEngineImpl::BackupEngineImpl(Env* db_env,
|
||||
copy_file_buffer_size_(kDefaultCopyFileBufferSize),
|
||||
read_only_(read_only) {
|
||||
|
||||
// set up threads perform copies from files_to_copy_ in the background
|
||||
for (int t = 0; t < options_.max_background_operations; t++) {
|
||||
threads_.emplace_back([&]() {
|
||||
CopyWorkItem work_item;
|
||||
while (files_to_copy_.read(work_item)) {
|
||||
CopyResult result;
|
||||
result.status = CopyFile(work_item.src_path,
|
||||
work_item.dst_path,
|
||||
work_item.src_env,
|
||||
work_item.dst_env,
|
||||
work_item.sync,
|
||||
work_item.rate_limiter,
|
||||
&result.size,
|
||||
&result.checksum_value,
|
||||
work_item.size_limit);
|
||||
work_item.result.set_value(std::move(result));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
if (read_only_) {
|
||||
Log(options_.info_log, "Starting read_only backup engine");
|
||||
}
|
||||
@@ -583,6 +561,27 @@ BackupEngineImpl::BackupEngineImpl(Env* db_env,
|
||||
if (!read_only_) {
|
||||
PutLatestBackupFileContents(latest_backup_id_); // Ignore errors
|
||||
}
|
||||
|
||||
// set up threads perform copies from files_to_copy_ in the background
|
||||
for (int t = 0; t < options_.max_background_operations; t++) {
|
||||
threads_.emplace_back([&]() {
|
||||
CopyWorkItem work_item;
|
||||
while (files_to_copy_.read(work_item)) {
|
||||
CopyResult result;
|
||||
result.status = CopyFile(work_item.src_path,
|
||||
work_item.dst_path,
|
||||
work_item.src_env,
|
||||
work_item.dst_env,
|
||||
work_item.sync,
|
||||
work_item.rate_limiter,
|
||||
&result.size,
|
||||
&result.checksum_value,
|
||||
work_item.size_limit);
|
||||
work_item.result.set_value(std::move(result));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
Log(options_.info_log, "Initialized BackupEngine");
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user