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

Merge branch 'master' into columnfamilies

Browse Source 
Conflicts:
	utilities/backupable/backupable_db.cc
This commit is contained in:
Igor Canadi 2014-03-17 17:57:14 -07:00
commit 64904b39a0
5 changed files with 150 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
db/memtable.cc
View File

@ -78,21 +78,23 @@ bool MemTable::ShouldFlushNow() const {
auto allocated_memory = auto allocated_memory =
table_->ApproximateMemoryUsage() + arena_.MemoryAllocatedBytes(); table_->ApproximateMemoryUsage() + arena_.MemoryAllocatedBytes();
if (allocated_memory + kArenaBlockSize * kAllowOverAllocationRatio < // if we can still allocate one more block without exceeding the
kWriteBufferSize) { // over-allocation ratio, then we should not flush.
if (allocated_memory + kArenaBlockSize <
kWriteBufferSize + kArenaBlockSize * kAllowOverAllocationRatio) {
return false; return false;
} }
// if user keeps adding entries that exceeds kWriteBufferSize, we need to // if user keeps adding entries that exceeds kWriteBufferSize, we need to
// flush // flush earlier even though we still have much available memory left.
// earlier even though we still have much available memory left. if (allocated_memory >
if (allocated_memory > kWriteBufferSize * (1 + kAllowOverAllocationRatio)) { kWriteBufferSize + kArenaBlockSize * kAllowOverAllocationRatio) {
return true; return true;
} }
// In this code path, Arena has already allocated its "last block", which // In this code path, Arena has already allocated its "last block", which
// means the total allocatedmemory size is either: // means the total allocatedmemory size is either:
// (1) "moderately" over allocated the memory (no more than `0.4 * arena // (1) "moderately" over allocated the memory (no more than `0.6 * arena
// block size`. Or, // block size`. Or,
// (2) the allocated memory is less than write buffer size, but we'll stop // (2) the allocated memory is less than write buffer size, but we'll stop
// here since if we allocate a new arena block, we'll over allocate too much // here since if we allocate a new arena block, we'll over allocate too much

43
include/utilities/backupable_db.h
View File

@ -55,19 +55,39 @@ struct BackupableDBOptions {
// Default: false // Default: false
bool destroy_old_data; bool destroy_old_data;
// If false, we won't backup log files. This option can be useful for backing
// up in-memory databases where log file are persisted, but table files are in
// memory.
// Default: true
bool backup_log_files;
void Dump(Logger* logger) const; void Dump(Logger* logger) const;
explicit BackupableDBOptions(const std::string& _backup_dir, explicit BackupableDBOptions(const std::string& _backup_dir,
Env* _backup_env = nullptr, Env* _backup_env = nullptr,
bool _share_table_files = true, bool _share_table_files = true,
Logger* _info_log = nullptr, bool _sync = true, Logger* _info_log = nullptr, bool _sync = true,
bool _destroy_old_data = false) bool _destroy_old_data = false,
bool _backup_log_files = true)
: backup_dir(_backup_dir), : backup_dir(_backup_dir),
backup_env(_backup_env), backup_env(_backup_env),
share_table_files(_share_table_files), share_table_files(_share_table_files),
info_log(_info_log), info_log(_info_log),
sync(_sync), sync(_sync),
destroy_old_data(_destroy_old_data) {} destroy_old_data(_destroy_old_data),
backup_log_files(_backup_log_files) {}
};
struct RestoreOptions {
// If true, restore won't overwrite the existing log files in wal_dir. It will
// also move all log files from archive directory to wal_dir. Use this option
// in combination with BackupableDBOptions::backup_log_files = false for
// persisting in-memory databases.
// Default: false
bool keep_log_files;
explicit RestoreOptions(bool _keep_log_files = false)
: keep_log_files(_keep_log_files) {}
}; };
typedef uint32_t BackupID; typedef uint32_t BackupID;
@ -96,11 +116,12 @@ class BackupEngine {
virtual void StopBackup() = 0; virtual void StopBackup() = 0;
virtual void GetBackupInfo(std::vector<BackupInfo>* backup_info) = 0; virtual void GetBackupInfo(std::vector<BackupInfo>* backup_info) = 0;
virtual Status RestoreDBFromBackup(BackupID backup_id, virtual Status RestoreDBFromBackup(
const std::string& db_dir, BackupID backup_id, const std::string& db_dir, const std::string& wal_dir,
const std::string& wal_dir) = 0; const RestoreOptions& restore_options = RestoreOptions()) = 0;
virtual Status RestoreDBFromLatestBackup(const std::string& db_dir, virtual Status RestoreDBFromLatestBackup(
const std::string& wal_dir) = 0; const std::string& db_dir, const std::string& wal_dir,
const RestoreOptions& restore_options = RestoreOptions()) = 0;
}; };
// Stack your DB with BackupableDB to be able to backup the DB // Stack your DB with BackupableDB to be able to backup the DB
@ -156,11 +177,15 @@ class RestoreBackupableDB {
// If you want to create new backup, you will first have to delete backups 4 // If you want to create new backup, you will first have to delete backups 4
// and 5. // and 5.
Status RestoreDBFromBackup(BackupID backup_id, const std::string& db_dir, Status RestoreDBFromBackup(BackupID backup_id, const std::string& db_dir,
const std::string& wal_dir); const std::string& wal_dir,
const RestoreOptions& restore_options =
RestoreOptions());
// restore from the latest backup // restore from the latest backup
Status RestoreDBFromLatestBackup(const std::string& db_dir, Status RestoreDBFromLatestBackup(const std::string& db_dir,
const std::string& wal_dir); const std::string& wal_dir,
const RestoreOptions& restore_options =
RestoreOptions());
// deletes old backups, keeping latest num_backups_to_keep alive // deletes old backups, keeping latest num_backups_to_keep alive
Status PurgeOldBackups(uint32_t num_backups_to_keep); Status PurgeOldBackups(uint32_t num_backups_to_keep);
// deletes a specific backup // deletes a specific backup

8
tools/db_stress.cc
View File

@ -979,13 +979,14 @@ class StressTest {
} }
int count = 0; int count = 0;
while (iters[0]->Valid()) { while (iters[0]->Valid() && iters[0]->key().starts_with(prefix_slices[0])) {
count++; count++;
std::string values[10]; std::string values[10];
// get list of all values for this iteration // get list of all values for this iteration
for (int i = 0; i < 10; i++) { for (int i = 0; i < 10; i++) {
// no iterator should finish before the first one // no iterator should finish before the first one
assert(iters[i]->Valid()); assert(iters[i]->Valid() &&
iters[i]->key().starts_with(prefix_slices[i]));
values[i] = iters[i]->value().ToString(); values[i] = iters[i]->value().ToString();
char expected_first = (prefixes[i])[0]; char expected_first = (prefixes[i])[0];
@ -1013,7 +1014,8 @@ class StressTest {
// cleanup iterators and snapshot // cleanup iterators and snapshot
for (int i = 0; i < 10; i++) { for (int i = 0; i < 10; i++) {
// if the first iterator finished, they should have all finished // if the first iterator finished, they should have all finished
assert(!iters[i]->Valid()); assert(!iters[i]->Valid() ||
!iters[i]->key().starts_with(prefix_slices[i]));
assert(iters[i]->status().ok()); assert(iters[i]->status().ok());
delete iters[i]; delete iters[i];
} }

106
utilities/backupable/backupable_db.cc
View File

@ -35,6 +35,8 @@ void BackupableDBOptions::Dump(Logger* logger) const {
Log(logger, " Options.sync: %d", static_cast<int>(sync)); Log(logger, " Options.sync: %d", static_cast<int>(sync));
Log(logger, " Options.destroy_old_data: %d", Log(logger, " Options.destroy_old_data: %d",
static_cast<int>(destroy_old_data)); static_cast<int>(destroy_old_data));
Log(logger, " Options.backup_log_files: %d",
static_cast<int>(backup_log_files));
} }
// -------- BackupEngineImpl class --------- // -------- BackupEngineImpl class ---------
@ -50,14 +52,21 @@ class BackupEngineImpl : public BackupEngine {
} }
void GetBackupInfo(std::vector<BackupInfo>* backup_info); void GetBackupInfo(std::vector<BackupInfo>* backup_info);
Status RestoreDBFromBackup(BackupID backup_id, const std::string &db_dir, Status RestoreDBFromBackup(BackupID backup_id, const std::string& db_dir,
const std::string &wal_dir); const std::string& wal_dir,
Status RestoreDBFromLatestBackup(const std::string &db_dir, const RestoreOptions& restore_options =
const std::string &wal_dir) { RestoreOptions());
return RestoreDBFromBackup(latest_backup_id_, db_dir, wal_dir); Status RestoreDBFromLatestBackup(const std::string& db_dir,
const std::string& wal_dir,
const RestoreOptions& restore_options =
RestoreOptions()) {
return RestoreDBFromBackup(latest_backup_id_, db_dir, wal_dir,
restore_options);
} }
private: private:
void DeleteChildren(const std::string& dir, uint32_t file_type_filter = 0);
struct FileInfo { struct FileInfo {
FileInfo(const std::string& fname, uint64_t sz, uint32_t checksum) FileInfo(const std::string& fname, uint64_t sz, uint32_t checksum)
: refs(0), filename(fname), size(sz), checksum_value(checksum) {} : refs(0), filename(fname), size(sz), checksum_value(checksum) {}
@ -314,7 +323,8 @@ Status BackupEngineImpl::CreateNewBackup(DB* db, bool flush_before_backup) {
// this will return live_files prefixed with "/" // this will return live_files prefixed with "/"
s = db->GetLiveFiles(live_files, &manifest_file_size, flush_before_backup); s = db->GetLiveFiles(live_files, &manifest_file_size, flush_before_backup);
} }
if (s.ok()) { // if we didn't flush before backup, we need to also get WAL files
if (s.ok() && !flush_before_backup && options_.backup_log_files) {
// returns file names prefixed with "/" // returns file names prefixed with "/"
s = db->GetSortedWalFiles(live_wal_files); s = db->GetSortedWalFiles(live_wal_files);
} }
@ -468,9 +478,9 @@ void BackupEngineImpl::GetBackupInfo(std::vector<BackupInfo>* backup_info) {
} }
} }
Status BackupEngineImpl::RestoreDBFromBackup(BackupID backup_id, Status BackupEngineImpl::RestoreDBFromBackup(
const std::string& db_dir, BackupID backup_id, const std::string& db_dir, const std::string& wal_dir,
const std::string& wal_dir) { const RestoreOptions& restore_options) {
auto backup_itr = backups_.find(backup_id); auto backup_itr = backups_.find(backup_id);
if (backup_itr == backups_.end()) { if (backup_itr == backups_.end()) {
return Status::NotFound("Backup not found"); return Status::NotFound("Backup not found");
@ -481,25 +491,40 @@ Status BackupEngineImpl::RestoreDBFromBackup(BackupID backup_id,
} }
Log(options_.info_log, "Restoring backup id %u\n", backup_id); Log(options_.info_log, "Restoring backup id %u\n", backup_id);
Log(options_.info_log, "keep_log_files: %d\n",
static_cast<int>(restore_options.keep_log_files));
// just in case. Ignore errors // just in case. Ignore errors
db_env_->CreateDirIfMissing(db_dir); db_env_->CreateDirIfMissing(db_dir);
db_env_->CreateDirIfMissing(wal_dir); db_env_->CreateDirIfMissing(wal_dir);
// delete log files that might have been already in wal_dir. if (restore_options.keep_log_files) {
// This is important since they might get replayed to the restored DB, // delete files in db_dir, but keep all the log files
// which will then differ from the backuped DB DeleteChildren(db_dir, 1 << kLogFile);
std::vector<std::string> delete_children; // move all the files from archive dir to wal_dir
db_env_->GetChildren(wal_dir, &delete_children); // ignore errors std::string archive_dir = ArchivalDirectory(wal_dir);
for (auto f : delete_children) { std::vector<std::string> archive_files;
db_env_->DeleteFile(wal_dir + "/" + f); // ignore errors db_env_->GetChildren(archive_dir, &archive_files); // ignore errors
} for (const auto& f : archive_files) {
// Also delete all the db_dir children. This is not so important uint64_t number;
// because obsolete files will be deleted by DBImpl::PurgeObsoleteFiles() FileType type;
delete_children.clear(); bool ok = ParseFileName(f, &number, &type);
db_env_->GetChildren(db_dir, &delete_children); // ignore errors if (ok && type == kLogFile) {
for (auto f : delete_children) { Log(options_.info_log, "Moving log file from archive/ to wal_dir: %s",
db_env_->DeleteFile(db_dir + "/" + f); // ignore errors f.c_str());
Status s =
db_env_->RenameFile(archive_dir + "/" + f, wal_dir + "/" + f);
if (!s.ok()) {
// if we can't move log file from archive_dir to wal_dir,
// we should fail, since it might mean data loss
return s;
}
}
}
} else {
DeleteChildren(wal_dir);
DeleteChildren(ArchivalDirectory(wal_dir));
DeleteChildren(db_dir);
} }
Status s; Status s;
@ -760,6 +785,23 @@ Status BackupEngineImpl::CalculateChecksum(const std::string& src, Env* src_env,
return s; return s;
} }
void BackupEngineImpl::DeleteChildren(const std::string& dir,
uint32_t file_type_filter) {
std::vector<std::string> children;
db_env_->GetChildren(dir, &children); // ignore errors
for (const auto& f : children) {
uint64_t number;
FileType type;
bool ok = ParseFileName(f, &number, &type);
if (ok && (file_type_filter & (1 << type))) {
// don't delete this file
continue;
}
db_env_->DeleteFile(dir + "/" + f); // ignore errors
}
}
void BackupEngineImpl::GarbageCollection(bool full_scan) { void BackupEngineImpl::GarbageCollection(bool full_scan) {
Log(options_.info_log, "Starting garbage collection"); Log(options_.info_log, "Starting garbage collection");
std::vector<std::string> to_delete; std::vector<std::string> to_delete;
@ -1042,16 +1084,18 @@ RestoreBackupableDB::GetBackupInfo(std::vector<BackupInfo>* backup_info) {
backup_engine_->GetBackupInfo(backup_info); backup_engine_->GetBackupInfo(backup_info);
} }
Status RestoreBackupableDB::RestoreDBFromBackup(BackupID backup_id, Status RestoreBackupableDB::RestoreDBFromBackup(
const std::string& db_dir, BackupID backup_id, const std::string& db_dir, const std::string& wal_dir,
const std::string& wal_dir) { const RestoreOptions& restore_options) {
return backup_engine_->RestoreDBFromBackup(backup_id, db_dir, wal_dir); return backup_engine_->RestoreDBFromBackup(backup_id, db_dir, wal_dir,
restore_options);
} }
Status Status RestoreBackupableDB::RestoreDBFromLatestBackup(
RestoreBackupableDB::RestoreDBFromLatestBackup(const std::string& db_dir, const std::string& db_dir, const std::string& wal_dir,
const std::string& wal_dir) { const RestoreOptions& restore_options) {
return backup_engine_->RestoreDBFromLatestBackup(db_dir, wal_dir); return backup_engine_->RestoreDBFromLatestBackup(db_dir, wal_dir,
restore_options);
} }
Status RestoreBackupableDB::PurgeOldBackups(uint32_t num_backups_to_keep) { Status RestoreBackupableDB::PurgeOldBackups(uint32_t num_backups_to_keep) {

31
utilities/backupable/backupable_db_test.cc
View File

@ -402,16 +402,20 @@ class BackupableDBTest {
// if backup_id == 0, it means restore from latest // if backup_id == 0, it means restore from latest
// if end == 0, don't check AssertEmpty // if end == 0, don't check AssertEmpty
void AssertBackupConsistency(BackupID backup_id, uint32_t start_exist, void AssertBackupConsistency(BackupID backup_id, uint32_t start_exist,
uint32_t end_exist, uint32_t end = 0) { uint32_t end_exist, uint32_t end = 0,
bool keep_log_files = false) {
RestoreOptions restore_options(keep_log_files);
bool opened_restore = false; bool opened_restore = false;
if (restore_db_.get() == nullptr) { if (restore_db_.get() == nullptr) {
opened_restore = true; opened_restore = true;
OpenRestoreDB(); OpenRestoreDB();
} }
if (backup_id > 0) { if (backup_id > 0) {
ASSERT_OK(restore_db_->RestoreDBFromBackup(backup_id, dbname_, dbname_)); ASSERT_OK(restore_db_->RestoreDBFromBackup(backup_id, dbname_, dbname_,
restore_options));
} else { } else {
ASSERT_OK(restore_db_->RestoreDBFromLatestBackup(dbname_, dbname_)); ASSERT_OK(restore_db_->RestoreDBFromLatestBackup(dbname_, dbname_,
restore_options));
} }
DB* db = OpenDB(); DB* db = OpenDB();
AssertExists(db, start_exist, end_exist); AssertExists(db, start_exist, end_exist);
@ -795,6 +799,27 @@ TEST(BackupableDBTest, DeleteTmpFiles) {
ASSERT_EQ(false, file_manager_->FileExists(private_tmp_dir)); ASSERT_EQ(false, file_manager_->FileExists(private_tmp_dir));
} }
TEST(BackupableDBTest, KeepLogFiles) {
// basically infinite
backupable_options_->backup_log_files = false;
options_.WAL_ttl_seconds = 24 * 60 * 60;
OpenBackupableDB(true);
FillDB(db_.get(), 0, 100);
ASSERT_OK(db_->Flush(FlushOptions()));
FillDB(db_.get(), 100, 200);
ASSERT_OK(db_->CreateNewBackup(false));
FillDB(db_.get(), 200, 300);
ASSERT_OK(db_->Flush(FlushOptions()));
FillDB(db_.get(), 300, 400);
ASSERT_OK(db_->Flush(FlushOptions()));
FillDB(db_.get(), 400, 500);
ASSERT_OK(db_->Flush(FlushOptions()));
CloseBackupableDB();
// all data should be there if we call with keep_log_files = true
AssertBackupConsistency(0, 0, 500, 600, true);
}
} // anon namespace } // anon namespace
} // namespace rocksdb } // namespace rocksdb