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

Introduce FaultInjectionTestFS to test fault File system instead of Env (#6414)

Browse Source 
Summary:
In the current code base, we can use FaultInjectionTestEnv to simulate the env issue such as file write/read errors, which are used in most of the test. The PR https://github.com/facebook/rocksdb/issues/5761 introduce the File System as a new Env API. This PR implement the FaultInjectionTestFS, which can be used to simulate when File System has issues such as IO error. user can specify any IOStatus error as input, such that FS corresponding actions will return certain error to the caller.

A set of ErrorHandlerFSTests are introduced for testing
Pull Request resolved: https://github.com/facebook/rocksdb/pull/6414

Test Plan: pass make asan_check, pass error_handler_fs_test.

Differential Revision: D20252421

Pulled By: zhichao-cao

fbshipit-source-id: e922038f8ce7e6d1da329fd0bba7283c4b779a21
This commit is contained in:
Zhichao Cao 2020-03-04 12:30:34 -08:00 committed by Facebook Github Bot
parent 8bbd76edbf
commit e62fe50634
8 changed files with 840 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CMakeLists.txt
View File

@ -967,7 +967,7 @@ if(WITH_TESTS)
db/db_write_test.cc
db/dbformat_test.cc
db/deletefile_test.cc
db/error_handler_test.cc
db/error_handler_fs_test.cc
db/obsolete_files_test.cc
db/external_sst_file_basic_test.cc
db/external_sst_file_test.cc
@ -1084,6 +1084,7 @@ if(WITH_TESTS)
monitoring/thread_status_updater_debug.cc
table/mock_table.cc
test_util/fault_injection_test_env.cc
test_util/fault_injection_test_fs.cc
utilities/cassandra/test_utils.cc
)
enable_testing()

4
Makefile
View File

@ -492,7 +492,7 @@ TESTS = \
db_table_properties_test \
db_statistics_test \
db_write_test \
error_handler_test \
error_handler_fs_test \
autovector_test \
blob_db_test \
cleanable_test \
@ -1366,7 +1366,7 @@ db_statistics_test: db/db_statistics_test.o db/db_test_util.o $(LIBOBJECTS) $(TE
db_write_test: db/db_write_test.o db/db_test_util.o $(LIBOBJECTS) $(TESTHARNESS)
$(AM_LINK)
error_handler_test: db/error_handler_test.o db/db_test_util.o $(LIBOBJECTS) $(TESTHARNESS)
error_handler_fs_test: db/error_handler_fs_test.o db/db_test_util.o $(LIBOBJECTS) $(TESTHARNESS)
$(AM_LINK)
external_sst_file_basic_test: db/external_sst_file_basic_test.o db/db_test_util.o $(LIBOBJECTS) $(TESTHARNESS)

5
TARGETS
View File

@ -368,6 +368,7 @@ cpp_library(
"db/db_test_util.cc",
"table/mock_table.cc",
"test_util/fault_injection_test_env.cc",
"test_util/fault_injection_test_fs.cc",
"test_util/testharness.cc",
"test_util/testutil.cc",
"tools/block_cache_analyzer/block_cache_trace_analyzer.cc",
@ -969,8 +970,8 @@ ROCKS_TESTS = [
[],
],
[
"error_handler_test",
"db/error_handler_test.cc",
"error_handler_fs_test",
"db/error_handler_fs_test.cc",
"serial",
[],
[],

285
db/error_handler_test.cc → db/error_handler_fs_test.cc
View File

@ -10,18 +10,20 @@
#include "db/db_test_util.h"
#include "port/stack_trace.h"
#include "rocksdb/io_status.h"
#include "rocksdb/perf_context.h"
#include "rocksdb/sst_file_manager.h"
#include "test_util/fault_injection_test_env.h"
#include "test_util/fault_injection_test_fs.h"
#if !defined(ROCKSDB_LITE)
#include "test_util/sync_point.h"
#endif
namespace ROCKSDB_NAMESPACE {
class DBErrorHandlingTest : public DBTestBase {
class DBErrorHandlingFSTest : public DBTestBase {
public:
DBErrorHandlingTest() : DBTestBase("/db_error_handling_test") {}
DBErrorHandlingFSTest() : DBTestBase("/db_error_handling_fs_test") {}
std::string GetManifestNameFromLiveFiles() {
std::vector<std::string> live_files;
@ -39,21 +41,24 @@ class DBErrorHandlingTest : public DBTestBase {
}
};
class DBErrorHandlingEnv : public EnvWrapper {
public:
DBErrorHandlingEnv() : EnvWrapper(Env::Default()),
trig_no_space(false), trig_io_error(false) {}
class DBErrorHandlingFS : public FileSystemWrapper {
public:
DBErrorHandlingFS()
: FileSystemWrapper(FileSystem::Default().get()),
trig_no_space(false),
trig_io_error(false) {}
void SetTrigNoSpace() {trig_no_space = true;}
void SetTrigIoError() {trig_io_error = true;}
private:
bool trig_no_space;
bool trig_io_error;
void SetTrigNoSpace() { trig_no_space = true; }
void SetTrigIoError() { trig_io_error = true; }
private:
bool trig_no_space;
bool trig_io_error;
};
class ErrorHandlerListener : public EventListener {
class ErrorHandlerFSListener : public EventListener {
public:
ErrorHandlerListener()
ErrorHandlerFSListener()
: mutex_(),
cv_(&mutex_),
no_auto_recovery_(false),
@ -61,7 +66,7 @@ class ErrorHandlerListener : public EventListener {
file_creation_started_(false),
override_bg_error_(false),
file_count_(0),
fault_env_(nullptr) {}
fault_fs_(nullptr) {}
void OnTableFileCreationStarted(
const TableFileCreationBriefInfo& /*ti*/) override {
@ -69,16 +74,15 @@ class ErrorHandlerListener : public EventListener {
file_creation_started_ = true;
if (file_count_ > 0) {
if (--file_count_ == 0) {
fault_env_->SetFilesystemActive(false, file_creation_error_);
file_creation_error_ = Status::OK();
fault_fs_->SetFilesystemActive(false, file_creation_error_);
file_creation_error_ = IOStatus::OK();
}
}
cv_.SignalAll();
}
void OnErrorRecoveryBegin(BackgroundErrorReason /*reason*/,
Status /*bg_error*/,
bool* auto_recovery) override {
Status /*bg_error*/, bool* auto_recovery) override {
if (*auto_recovery && no_auto_recovery_) {
*auto_recovery = false;
}
@ -125,11 +129,11 @@ class ErrorHandlerListener : public EventListener {
override_bg_error_ = true;
}
void InjectFileCreationError(FaultInjectionTestEnv* env, int file_count,
Status s) {
fault_env_ = env;
void InjectFileCreationError(FaultInjectionTestFS* fs, int file_count,
IOStatus io_s) {
fault_fs_ = fs;
file_count_ = file_count;
file_creation_error_ = s;
file_creation_error_ = io_s;
}
private:
@ -140,18 +144,19 @@ class ErrorHandlerListener : public EventListener {
bool file_creation_started_;
bool override_bg_error_;
int file_count_;
Status file_creation_error_;
IOStatus file_creation_error_;
Status bg_error_;
FaultInjectionTestEnv* fault_env_;
FaultInjectionTestFS* fault_fs_;
};
TEST_F(DBErrorHandlingTest, FLushWriteError) {
std::unique_ptr<FaultInjectionTestEnv> fault_env(
new FaultInjectionTestEnv(Env::Default()));
std::shared_ptr<ErrorHandlerListener> listener(new ErrorHandlerListener());
TEST_F(DBErrorHandlingFSTest, FLushWriteError) {
FaultInjectionTestFS* fault_fs =
new FaultInjectionTestFS(FileSystem::Default().get());
std::shared_ptr<ErrorHandlerFSListener> listener(
new ErrorHandlerFSListener());
Options options = GetDefaultOptions();
options.file_system.reset(fault_fs);
options.create_if_missing = true;
options.env = fault_env.get();
options.listeners.emplace_back(listener);
Status s;
@ -159,15 +164,14 @@ TEST_F(DBErrorHandlingTest, FLushWriteError) {
DestroyAndReopen(options);
Put(Key(0), "val");
SyncPoint::GetInstance()->SetCallBack(
"FlushJob::Start", [&](void *) {
fault_env->SetFilesystemActive(false, Status::NoSpace("Out of space"));
SyncPoint::GetInstance()->SetCallBack("FlushJob::Start", [&](void*) {
fault_fs->SetFilesystemActive(false, IOStatus::NoSpace("Out of space"));
});
SyncPoint::GetInstance()->EnableProcessing();
s = Flush();
ASSERT_EQ(s.severity(), ROCKSDB_NAMESPACE::Status::Severity::kHardError);
SyncPoint::GetInstance()->DisableProcessing();
fault_env->SetFilesystemActive(true);
fault_fs->SetFilesystemActive(true);
s = dbfull()->Resume();
ASSERT_EQ(s, Status::OK());
@ -176,13 +180,14 @@ TEST_F(DBErrorHandlingTest, FLushWriteError) {
Destroy(options);
}
TEST_F(DBErrorHandlingTest, ManifestWriteError) {
std::unique_ptr<FaultInjectionTestEnv> fault_env(
new FaultInjectionTestEnv(Env::Default()));
std::shared_ptr<ErrorHandlerListener> listener(new ErrorHandlerListener());
TEST_F(DBErrorHandlingFSTest, ManifestWriteError) {
FaultInjectionTestFS* fault_fs =
new FaultInjectionTestFS(FileSystem::Default().get());
std::shared_ptr<ErrorHandlerFSListener> listener(
new ErrorHandlerFSListener());
Options options = GetDefaultOptions();
options.file_system.reset(fault_fs);
options.create_if_missing = true;
options.env = fault_env.get();
options.listeners.emplace_back(listener);
Status s;
std::string old_manifest;
@ -196,15 +201,15 @@ TEST_F(DBErrorHandlingTest, ManifestWriteError) {
Flush();
Put(Key(1), "val");
SyncPoint::GetInstance()->SetCallBack(
"VersionSet::LogAndApply:WriteManifest", [&](void *) {
fault_env->SetFilesystemActive(false, Status::NoSpace("Out of space"));
});
"VersionSet::LogAndApply:WriteManifest", [&](void*) {
fault_fs->SetFilesystemActive(false, IOStatus::NoSpace("Out of space"));
});
SyncPoint::GetInstance()->EnableProcessing();
s = Flush();
ASSERT_EQ(s.severity(), ROCKSDB_NAMESPACE::Status::Severity::kHardError);
SyncPoint::GetInstance()->ClearAllCallBacks();
SyncPoint::GetInstance()->DisableProcessing();
fault_env->SetFilesystemActive(true);
fault_fs->SetFilesystemActive(true);
s = dbfull()->Resume();
ASSERT_EQ(s, Status::OK());
@ -217,13 +222,14 @@ TEST_F(DBErrorHandlingTest, ManifestWriteError) {
Close();
}
TEST_F(DBErrorHandlingTest, DoubleManifestWriteError) {
std::unique_ptr<FaultInjectionTestEnv> fault_env(
new FaultInjectionTestEnv(Env::Default()));
std::shared_ptr<ErrorHandlerListener> listener(new ErrorHandlerListener());
TEST_F(DBErrorHandlingFSTest, DoubleManifestWriteError) {
FaultInjectionTestFS* fault_fs =
new FaultInjectionTestFS(FileSystem::Default().get());
std::shared_ptr<ErrorHandlerFSListener> listener(
new ErrorHandlerFSListener());
Options options = GetDefaultOptions();
options.file_system.reset(fault_fs);
options.create_if_missing = true;
options.env = fault_env.get();
options.listeners.emplace_back(listener);
Status s;
std::string old_manifest;
@ -237,18 +243,18 @@ TEST_F(DBErrorHandlingTest, DoubleManifestWriteError) {
Flush();
Put(Key(1), "val");
SyncPoint::GetInstance()->SetCallBack(
"VersionSet::LogAndApply:WriteManifest", [&](void *) {
fault_env->SetFilesystemActive(false, Status::NoSpace("Out of space"));
});
"VersionSet::LogAndApply:WriteManifest", [&](void*) {
fault_fs->SetFilesystemActive(false, IOStatus::NoSpace("Out of space"));
});
SyncPoint::GetInstance()->EnableProcessing();
s = Flush();
ASSERT_EQ(s.severity(), ROCKSDB_NAMESPACE::Status::Severity::kHardError);
fault_env->SetFilesystemActive(true);
fault_fs->SetFilesystemActive(true);
// This Resume() will attempt to create a new manifest file and fail again
s = dbfull()->Resume();
ASSERT_EQ(s.severity(), ROCKSDB_NAMESPACE::Status::Severity::kHardError);
fault_env->SetFilesystemActive(true);
fault_fs->SetFilesystemActive(true);
SyncPoint::GetInstance()->ClearAllCallBacks();
SyncPoint::GetInstance()->DisableProcessing();
@ -265,15 +271,16 @@ TEST_F(DBErrorHandlingTest, DoubleManifestWriteError) {
Close();
}
TEST_F(DBErrorHandlingTest, CompactionManifestWriteError) {
std::unique_ptr<FaultInjectionTestEnv> fault_env(
new FaultInjectionTestEnv(Env::Default()));
std::shared_ptr<ErrorHandlerListener> listener(new ErrorHandlerListener());
TEST_F(DBErrorHandlingFSTest, CompactionManifestWriteError) {
FaultInjectionTestFS* fault_fs =
new FaultInjectionTestFS(FileSystem::Default().get());
std::shared_ptr<ErrorHandlerFSListener> listener(
new ErrorHandlerFSListener());
Options options = GetDefaultOptions();
options.file_system.reset(fault_fs);
options.create_if_missing = true;
options.level0_file_num_compaction_trigger = 2;
options.listeners.emplace_back(listener);
options.env = fault_env.get();
Status s;
std::string old_manifest;
std::string new_manifest;
@ -304,8 +311,8 @@ TEST_F(DBErrorHandlingTest, CompactionManifestWriteError) {
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
"VersionSet::LogAndApply:WriteManifest", [&](void*) {
if (fail_manifest.load()) {
fault_env->SetFilesystemActive(false,
Status::NoSpace("Out of space"));
fault_fs->SetFilesystemActive(false,
IOStatus::NoSpace("Out of space"));
}
});
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
@ -318,7 +325,7 @@ TEST_F(DBErrorHandlingTest, CompactionManifestWriteError) {
TEST_SYNC_POINT("CompactionManifestWriteError:0");
// Clear all errors so when the compaction is retried, it will succeed
fault_env->SetFilesystemActive(true);
fault_fs->SetFilesystemActive(true);
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
TEST_SYNC_POINT("CompactionManifestWriteError:1");
TEST_SYNC_POINT("CompactionManifestWriteError:2");
@ -336,15 +343,16 @@ TEST_F(DBErrorHandlingTest, CompactionManifestWriteError) {
Close();
}
TEST_F(DBErrorHandlingTest, CompactionWriteError) {
std::unique_ptr<FaultInjectionTestEnv> fault_env(
new FaultInjectionTestEnv(Env::Default()));
std::shared_ptr<ErrorHandlerListener> listener(new ErrorHandlerListener());
TEST_F(DBErrorHandlingFSTest, CompactionWriteError) {
FaultInjectionTestFS* fault_fs =
new FaultInjectionTestFS(FileSystem::Default().get());
std::shared_ptr<ErrorHandlerFSListener> listener(
new ErrorHandlerFSListener());
Options options = GetDefaultOptions();
options.file_system.reset(fault_fs);
options.create_if_missing = true;
options.level0_file_num_compaction_trigger = 2;
options.listeners.emplace_back(listener);
options.env = fault_env.get();
Status s;
DestroyAndReopen(options);
@ -354,15 +362,14 @@ TEST_F(DBErrorHandlingTest, CompactionWriteError) {
ASSERT_EQ(s, Status::OK());
listener->OverrideBGError(
Status(Status::NoSpace(), Status::Severity::kHardError)
);
Status(Status::NoSpace(), Status::Severity::kHardError));
listener->EnableAutoRecovery(false);
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->LoadDependency(
{{"DBImpl::FlushMemTable:FlushMemTableFinished",
"BackgroundCallCompaction:0"}});
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
"BackgroundCallCompaction:0", [&](void*) {
fault_env->SetFilesystemActive(false, Status::NoSpace("Out of space"));
fault_fs->SetFilesystemActive(false, IOStatus::NoSpace("Out of space"));
});
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
@ -373,19 +380,19 @@ TEST_F(DBErrorHandlingTest, CompactionWriteError) {
s = dbfull()->TEST_WaitForCompact();
ASSERT_EQ(s.severity(), ROCKSDB_NAMESPACE::Status::Severity::kHardError);
fault_env->SetFilesystemActive(true);
fault_fs->SetFilesystemActive(true);
s = dbfull()->Resume();
ASSERT_EQ(s, Status::OK());
Destroy(options);
}
TEST_F(DBErrorHandlingTest, CorruptionError) {
std::unique_ptr<FaultInjectionTestEnv> fault_env(
new FaultInjectionTestEnv(Env::Default()));
TEST_F(DBErrorHandlingFSTest, CorruptionError) {
FaultInjectionTestFS* fault_fs =
new FaultInjectionTestFS(FileSystem::Default().get());
Options options = GetDefaultOptions();
options.file_system.reset(fault_fs);
options.create_if_missing = true;
options.level0_file_num_compaction_trigger = 2;
options.env = fault_env.get();
Status s;
DestroyAndReopen(options);
@ -399,7 +406,8 @@ TEST_F(DBErrorHandlingTest, CorruptionError) {
"BackgroundCallCompaction:0"}});
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
"BackgroundCallCompaction:0", [&](void*) {
fault_env->SetFilesystemActive(false, Status::Corruption("Corruption"));
fault_fs->SetFilesystemActive(false,
IOStatus::Corruption("Corruption"));
});
ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
@ -411,19 +419,20 @@ TEST_F(DBErrorHandlingTest, CorruptionError) {
ASSERT_EQ(s.severity(),
ROCKSDB_NAMESPACE::Status::Severity::kUnrecoverableError);
fault_env->SetFilesystemActive(true);
fault_fs->SetFilesystemActive(true);
s = dbfull()->Resume();
ASSERT_NE(s, Status::OK());
Destroy(options);
}
TEST_F(DBErrorHandlingTest, AutoRecoverFlushError) {
std::unique_ptr<FaultInjectionTestEnv> fault_env(
new FaultInjectionTestEnv(Env::Default()));
std::shared_ptr<ErrorHandlerListener> listener(new ErrorHandlerListener());
TEST_F(DBErrorHandlingFSTest, AutoRecoverFlushError) {
FaultInjectionTestFS* fault_fs =
new FaultInjectionTestFS(FileSystem::Default().get());
std::shared_ptr<ErrorHandlerFSListener> listener(
new ErrorHandlerFSListener());
Options options = GetDefaultOptions();
options.file_system.reset(fault_fs);
options.create_if_missing = true;
options.env = fault_env.get();
options.listeners.emplace_back(listener);
Status s;
@ -432,13 +441,13 @@ TEST_F(DBErrorHandlingTest, AutoRecoverFlushError) {
Put(Key(0), "val");
SyncPoint::GetInstance()->SetCallBack("FlushJob::Start", [&](void*) {
fault_env->SetFilesystemActive(false, Status::NoSpace("Out of space"));
fault_fs->SetFilesystemActive(false, IOStatus::NoSpace("Out of space"));
});
SyncPoint::GetInstance()->EnableProcessing();
s = Flush();
ASSERT_EQ(s.severity(), ROCKSDB_NAMESPACE::Status::Severity::kHardError);
SyncPoint::GetInstance()->DisableProcessing();
fault_env->SetFilesystemActive(true);
fault_fs->SetFilesystemActive(true);
ASSERT_EQ(listener->WaitForRecovery(5000000), true);
s = Put(Key(1), "val");
@ -450,13 +459,14 @@ TEST_F(DBErrorHandlingTest, AutoRecoverFlushError) {
Destroy(options);
}
TEST_F(DBErrorHandlingTest, FailRecoverFlushError) {
std::unique_ptr<FaultInjectionTestEnv> fault_env(
new FaultInjectionTestEnv(Env::Default()));
std::shared_ptr<ErrorHandlerListener> listener(new ErrorHandlerListener());
TEST_F(DBErrorHandlingFSTest, FailRecoverFlushError) {
FaultInjectionTestFS* fault_fs =
new FaultInjectionTestFS(FileSystem::Default().get());
std::shared_ptr<ErrorHandlerFSListener> listener(
new ErrorHandlerFSListener());
Options options = GetDefaultOptions();
options.file_system.reset(fault_fs);
options.create_if_missing = true;
options.env = fault_env.get();
options.listeners.emplace_back(listener);
Status s;
@ -465,7 +475,7 @@ TEST_F(DBErrorHandlingTest, FailRecoverFlushError) {
Put(Key(0), "val");
SyncPoint::GetInstance()->SetCallBack("FlushJob::Start", [&](void*) {
fault_env->SetFilesystemActive(false, Status::NoSpace("Out of space"));
fault_fs->SetFilesystemActive(false, IOStatus::NoSpace("Out of space"));
});
SyncPoint::GetInstance()->EnableProcessing();
s = Flush();
@ -476,14 +486,15 @@ TEST_F(DBErrorHandlingTest, FailRecoverFlushError) {
DestroyDB(dbname_, options);
}
TEST_F(DBErrorHandlingTest, WALWriteError) {
std::unique_ptr<FaultInjectionTestEnv> fault_env(
new FaultInjectionTestEnv(Env::Default()));
std::shared_ptr<ErrorHandlerListener> listener(new ErrorHandlerListener());
TEST_F(DBErrorHandlingFSTest, WALWriteError) {
FaultInjectionTestFS* fault_fs =
new FaultInjectionTestFS(FileSystem::Default().get());
std::shared_ptr<ErrorHandlerFSListener> listener(
new ErrorHandlerFSListener());
Options options = GetDefaultOptions();
options.file_system.reset(fault_fs);
options.create_if_missing = true;
options.writable_file_max_buffer_size = 32768;
options.env = fault_env.get();
options.listeners.emplace_back(listener);
Status s;
Random rnd(301);
@ -494,7 +505,7 @@ TEST_F(DBErrorHandlingTest, WALWriteError) {
{
WriteBatch batch;
for (auto i = 0; i<100; ++i) {
for (auto i = 0; i < 100; ++i) {
batch.Put(Key(i), RandomString(&rnd, 1024));
}
@ -507,16 +518,18 @@ TEST_F(DBErrorHandlingTest, WALWriteError) {
WriteBatch batch;
int write_error = 0;
for (auto i = 100; i<199; ++i) {
for (auto i = 100; i < 199; ++i) {
batch.Put(Key(i), RandomString(&rnd, 1024));
}
SyncPoint::GetInstance()->SetCallBack("WritableFileWriter::Append:BeforePrepareWrite", [&](void*) {
write_error++;
if (write_error > 2) {
fault_env->SetFilesystemActive(false, Status::NoSpace("Out of space"));
}
});
SyncPoint::GetInstance()->SetCallBack(
"WritableFileWriter::Append:BeforePrepareWrite", [&](void*) {
write_error++;
if (write_error > 2) {
fault_fs->SetFilesystemActive(false,
IOStatus::NoSpace("Out of space"));
}
});
SyncPoint::GetInstance()->EnableProcessing();
WriteOptions wopts;
wopts.sync = true;
@ -524,9 +537,9 @@ TEST_F(DBErrorHandlingTest, WALWriteError) {
ASSERT_EQ(s, s.NoSpace());
}
SyncPoint::GetInstance()->DisableProcessing();
fault_env->SetFilesystemActive(true);
fault_fs->SetFilesystemActive(true);
ASSERT_EQ(listener->WaitForRecovery(5000000), true);
for (auto i=0; i<199; ++i) {
for (auto i = 0; i < 199; ++i) {
if (i < 100) {
ASSERT_NE(Get(Key(i)), "NOT_FOUND");
} else {
@ -534,7 +547,7 @@ TEST_F(DBErrorHandlingTest, WALWriteError) {
}
}
Reopen(options);
for (auto i=0; i<199; ++i) {
for (auto i = 0; i < 199; ++i) {
if (i < 100) {
ASSERT_NE(Get(Key(i)), "NOT_FOUND");
} else {
@ -544,14 +557,15 @@ TEST_F(DBErrorHandlingTest, WALWriteError) {
Close();
}
TEST_F(DBErrorHandlingTest, MultiCFWALWriteError) {
std::unique_ptr<FaultInjectionTestEnv> fault_env(
new FaultInjectionTestEnv(Env::Default()));
std::shared_ptr<ErrorHandlerListener> listener(new ErrorHandlerListener());
TEST_F(DBErrorHandlingFSTest, MultiCFWALWriteError) {
FaultInjectionTestFS* fault_fs =
new FaultInjectionTestFS(FileSystem::Default().get());
std::shared_ptr<ErrorHandlerFSListener> listener(
new ErrorHandlerFSListener());
Options options = GetDefaultOptions();
options.file_system.reset(fault_fs);
options.create_if_missing = true;
options.writable_file_max_buffer_size = 32768;
options.env = fault_env.get();
options.listeners.emplace_back(listener);
Status s;
Random rnd(301);
@ -586,8 +600,8 @@ TEST_F(DBErrorHandlingTest, MultiCFWALWriteError) {
"WritableFileWriter::Append:BeforePrepareWrite", [&](void*) {
write_error++;
if (write_error > 2) {
fault_env->SetFilesystemActive(false,
Status::NoSpace("Out of space"));
fault_fs->SetFilesystemActive(false,
IOStatus::NoSpace("Out of space"));
}
});
SyncPoint::GetInstance()->EnableProcessing();
@ -597,7 +611,7 @@ TEST_F(DBErrorHandlingTest, MultiCFWALWriteError) {
ASSERT_EQ(s, s.NoSpace());
}
SyncPoint::GetInstance()->DisableProcessing();
fault_env->SetFilesystemActive(true);
fault_fs->SetFilesystemActive(true);
ASSERT_EQ(listener->WaitForRecovery(5000000), true);
for (auto i = 1; i < 4; ++i) {
@ -627,24 +641,25 @@ TEST_F(DBErrorHandlingTest, MultiCFWALWriteError) {
Close();
}
TEST_F(DBErrorHandlingTest, MultiDBCompactionError) {
TEST_F(DBErrorHandlingFSTest, MultiDBCompactionError) {
FaultInjectionTestEnv* def_env = new FaultInjectionTestEnv(Env::Default());
std::vector<std::unique_ptr<FaultInjectionTestEnv>> fault_env;
std::vector<FaultInjectionTestFS*> fault_fs;
std::vector<Options> options;
std::vector<std::shared_ptr<ErrorHandlerListener>> listener;
std::vector<std::shared_ptr<ErrorHandlerFSListener>> listener;
std::vector<DB*> db;
std::shared_ptr<SstFileManager> sfm(NewSstFileManager(def_env));
int kNumDbInstances = 3;
Random rnd(301);
for (auto i = 0; i < kNumDbInstances; ++i) {
listener.emplace_back(new ErrorHandlerListener());
listener.emplace_back(new ErrorHandlerFSListener());
options.emplace_back(GetDefaultOptions());
fault_env.emplace_back(new FaultInjectionTestEnv(Env::Default()));
fault_fs.emplace_back(
new FaultInjectionTestFS(FileSystem::Default().get()));
options[i].create_if_missing = true;
options[i].level0_file_num_compaction_trigger = 2;
options[i].writable_file_max_buffer_size = 32768;
options[i].env = fault_env[i].get();
options[i].file_system.reset(fault_fs[i]);
options[i].listeners.emplace_back(listener[i]);
options[i].sst_file_manager = sfm;
DB* dbptr;
@ -652,8 +667,8 @@ TEST_F(DBErrorHandlingTest, MultiDBCompactionError) {
listener[i]->EnableAutoRecovery();
// Setup for returning error for the 3rd SST, which would be level 1
listener[i]->InjectFileCreationError(fault_env[i].get(), 3,
Status::NoSpace("Out of space"));
listener[i]->InjectFileCreationError(fault_fs[i], 3,
IOStatus::NoSpace("Out of space"));
snprintf(buf, sizeof(buf), "_%d", i);
DestroyDB(dbname_ + std::string(buf), options[i]);
ASSERT_EQ(DB::Open(options[i], dbname_ + std::string(buf), &dbptr),
@ -692,7 +707,7 @@ TEST_F(DBErrorHandlingTest, MultiDBCompactionError) {
for (auto i = 0; i < kNumDbInstances; ++i) {
Status s = static_cast<DBImpl*>(db[i])->TEST_WaitForCompact(true);
ASSERT_EQ(s.severity(), Status::Severity::kSoftError);
fault_env[i]->SetFilesystemActive(true);
fault_fs[i]->SetFilesystemActive(true);
}
def_env->SetFilesystemActive(true);
@ -713,7 +728,7 @@ TEST_F(DBErrorHandlingTest, MultiDBCompactionError) {
char buf[16];
snprintf(buf, sizeof(buf), "_%d", i);
delete db[i];
fault_env[i]->SetFilesystemActive(true);
fault_fs[i]->SetFilesystemActive(true);
if (getenv("KEEP_DB")) {
printf("DB is still at %s%s\n", dbname_.c_str(), buf);
} else {
@ -725,24 +740,25 @@ TEST_F(DBErrorHandlingTest, MultiDBCompactionError) {
delete def_env;
}
TEST_F(DBErrorHandlingTest, MultiDBVariousErrors) {
TEST_F(DBErrorHandlingFSTest, MultiDBVariousErrors) {
FaultInjectionTestEnv* def_env = new FaultInjectionTestEnv(Env::Default());
std::vector<std::unique_ptr<FaultInjectionTestEnv>> fault_env;
std::vector<FaultInjectionTestFS*> fault_fs;
std::vector<Options> options;
std::vector<std::shared_ptr<ErrorHandlerListener>> listener;
std::vector<std::shared_ptr<ErrorHandlerFSListener>> listener;
std::vector<DB*> db;
std::shared_ptr<SstFileManager> sfm(NewSstFileManager(def_env));
int kNumDbInstances = 3;
Random rnd(301);
for (auto i = 0; i < kNumDbInstances; ++i) {
listener.emplace_back(new ErrorHandlerListener());
listener.emplace_back(new ErrorHandlerFSListener());
options.emplace_back(GetDefaultOptions());
fault_env.emplace_back(new FaultInjectionTestEnv(Env::Default()));
fault_fs.emplace_back(
new FaultInjectionTestFS(FileSystem::Default().get()));
options[i].create_if_missing = true;
options[i].level0_file_num_compaction_trigger = 2;
options[i].writable_file_max_buffer_size = 32768;
options[i].env = fault_env[i].get();
options[i].file_system.reset(fault_fs[i]);
options[i].listeners.emplace_back(listener[i]);
options[i].sst_file_manager = sfm;
DB* dbptr;
@ -752,14 +768,14 @@ TEST_F(DBErrorHandlingTest, MultiDBVariousErrors) {
switch (i) {
case 0:
// Setup for returning error for the 3rd SST, which would be level 1
listener[i]->InjectFileCreationError(fault_env[i].get(), 3,
Status::NoSpace("Out of space"));
listener[i]->InjectFileCreationError(fault_fs[i], 3,
IOStatus::NoSpace("Out of space"));
break;
case 1:
// Setup for returning error after the 1st SST, which would result
// in a hard error
listener[i]->InjectFileCreationError(fault_env[i].get(), 2,
Status::NoSpace("Out of space"));
listener[i]->InjectFileCreationError(fault_fs[i], 2,
IOStatus::NoSpace("Out of space"));
break;
default:
break;
@ -816,7 +832,7 @@ TEST_F(DBErrorHandlingTest, MultiDBVariousErrors) {
ASSERT_EQ(s, Status::OK());
break;
}
fault_env[i]->SetFilesystemActive(true);
fault_fs[i]->SetFilesystemActive(true);
}
def_env->SetFilesystemActive(true);
@ -840,7 +856,7 @@ TEST_F(DBErrorHandlingTest, MultiDBVariousErrors) {
for (auto i = 0; i < kNumDbInstances; ++i) {
char buf[16];
snprintf(buf, sizeof(buf), "_%d", i);
fault_env[i]->SetFilesystemActive(true);
fault_fs[i]->SetFilesystemActive(true);
delete db[i];
if (getenv("KEEP_DB")) {
printf("DB is still at %s%s\n", dbname_.c_str(), buf);
@ -851,7 +867,6 @@ TEST_F(DBErrorHandlingTest, MultiDBVariousErrors) {
options.clear();
delete def_env;
}
} // namespace ROCKSDB_NAMESPACE
int main(int argc, char** argv) {

2
include/rocksdb/io_status.h
View File

@ -209,9 +209,7 @@ inline IOStatus& IOStatus::operator=(IOStatus&& s)
subcode_ = std::move(s.subcode_);
s.subcode_ = kNone;
retryable_ = s.retryable_;
retryable_ = false;
data_loss_ = s.data_loss_;
data_loss_ = false;
scope_ = s.scope_;
scope_ = kIOErrorScopeFileSystem;
delete[] state_;

3
src.mk
View File

@ -263,6 +263,7 @@ ANALYZER_LIB_SOURCES = \
MOCK_LIB_SOURCES = \
table/mock_table.cc \
test_util/fault_injection_test_fs.cc \
test_util/fault_injection_test_env.cc
BENCH_LIB_SOURCES = \
@ -340,7 +341,7 @@ MAIN_SOURCES = \
db/dbformat_test.cc \
db/deletefile_test.cc \
db/env_timed_test.cc \
db/error_handler_test.cc \
db/error_handler_fs_test.cc \
db/external_sst_file_basic_test.cc \
db/external_sst_file_test.cc \
db/fault_injection_test.cc \

430
test_util/fault_injection_test_fs.cc Normal file
View File

@ -0,0 +1,430 @@
// 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).
//
// Copyright 2014 The LevelDB Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file. See the AUTHORS file for names of contributors.
// This test uses a custom FileSystem to keep track of the state of a file
// system the last "Sync". The data being written is cached in a "buffer".
// Only when "Sync" is called, the data will be persistent. It can similate
// file data loss (or entire files) not protected by a "Sync". For any of the
// FileSystem related operations, by specify the "IOStatus Error", a specific
// error can be returned when file system is not activated.
#include "test_util/fault_injection_test_fs.h"
#include <functional>
#include <utility>
namespace ROCKSDB_NAMESPACE {
// Assume a filename, and not a directory name like "/foo/bar/"
std::string TestFSGetDirName(const std::string filename) {
size_t found = filename.find_last_of("/\\");
if (found == std::string::npos) {
return "";
} else {
return filename.substr(0, found);
}
}
// Trim the tailing "/" in the end of `str`
std::string TestFSTrimDirname(const std::string& str) {
size_t found = str.find_last_not_of("/");
if (found == std::string::npos) {
return str;
}
return str.substr(0, found + 1);
}
// Return pair <parent directory name, file name> of a full path.
std::pair<std::string, std::string> TestFSGetDirAndName(
const std::string& name) {
std::string dirname = TestFSGetDirName(name);
std::string fname = name.substr(dirname.size() + 1);
return std::make_pair(dirname, fname);
}
IOStatus FSFileState::DropUnsyncedData() {
buffer_.resize(0);
return IOStatus::OK();
}
IOStatus FSFileState::DropRandomUnsyncedData(Random* rand) {
int range = static_cast<int>(buffer_.size());
size_t truncated_size = static_cast<size_t>(rand->Uniform(range));
buffer_.resize(truncated_size);
return IOStatus::OK();
}
IOStatus TestFSDirectory::Fsync(const IOOptions& options, IODebugContext* dbg) {
if (!fs_->IsFilesystemActive()) {
return fs_->GetError();
}
fs_->SyncDir(dirname_);
return dir_->Fsync(options, dbg);
}
TestFSWritableFile::TestFSWritableFile(const std::string& fname,
std::unique_ptr<FSWritableFile>&& f,
FaultInjectionTestFS* fs)
: state_(fname),
target_(std::move(f)),
writable_file_opened_(true),
fs_(fs) {
assert(target_ != nullptr);
state_.pos_ = 0;
}
TestFSWritableFile::~TestFSWritableFile() {
if (writable_file_opened_) {
Close(IOOptions(), nullptr);
}
}
IOStatus TestFSWritableFile::Append(const Slice& data, const IOOptions&,
IODebugContext*) {
MutexLock l(&mutex_);
if (!fs_->IsFilesystemActive()) {
return fs_->GetError();
}
state_.buffer_.append(data.data(), data.size());
state_.pos_ += data.size();
fs_->WritableFileAppended(state_);
return IOStatus::OK();
}
IOStatus TestFSWritableFile::Close(const IOOptions& options,
IODebugContext* dbg) {
if (!fs_->IsFilesystemActive()) {
return fs_->GetError();
}
writable_file_opened_ = false;
IOStatus io_s;
io_s = target_->Append(state_.buffer_, options, dbg);
if (io_s.ok()) {
state_.buffer_.resize(0);
target_->Sync(options, dbg);
io_s = target_->Close(options, dbg);
}
if (io_s.ok()) {
fs_->WritableFileClosed(state_);
}
return io_s;
}
IOStatus TestFSWritableFile::Flush(const IOOptions&, IODebugContext*) {
if (!fs_->IsFilesystemActive()) {
return fs_->GetError();
}
IOStatus io_s;
if (io_s.ok() && fs_->IsFilesystemActive()) {
state_.pos_at_last_flush_ = state_.pos_;
}
return io_s;
}
IOStatus TestFSWritableFile::Sync(const IOOptions& options,
IODebugContext* dbg) {
if (!fs_->IsFilesystemActive()) {
return fs_->GetError();
}
IOStatus io_s = target_->Append(state_.buffer_, options, dbg);
state_.buffer_.resize(0);
target_->Sync(options, dbg);
state_.pos_at_last_sync_ = state_.pos_;
fs_->WritableFileSynced(state_);
return io_s;
}
TestFSRandomRWFile::TestFSRandomRWFile(const std::string& /*fname*/,
std::unique_ptr<FSRandomRWFile>&& f,
FaultInjectionTestFS* fs)
: target_(std::move(f)), file_opened_(true), fs_(fs) {
assert(target_ != nullptr);
}
TestFSRandomRWFile::~TestFSRandomRWFile() {
if (file_opened_) {
Close(IOOptions(), nullptr);
}
}
IOStatus TestFSRandomRWFile::Write(uint64_t offset, const Slice& data,
const IOOptions& options,
IODebugContext* dbg) {
if (!fs_->IsFilesystemActive()) {
return fs_->GetError();
}
return target_->Write(offset, data, options, dbg);
}
IOStatus TestFSRandomRWFile::Read(uint64_t offset, size_t n,
const IOOptions& options, Slice* result,
char* scratch, IODebugContext* dbg) const {
if (!fs_->IsFilesystemActive()) {
return fs_->GetError();
}
return target_->Read(offset, n, options, result, scratch, dbg);
}
IOStatus TestFSRandomRWFile::Close(const IOOptions& options,
IODebugContext* dbg) {
if (!fs_->IsFilesystemActive()) {
return fs_->GetError();
}
file_opened_ = false;
return target_->Close(options, dbg);
}
IOStatus TestFSRandomRWFile::Flush(const IOOptions& options,
IODebugContext* dbg) {
if (!fs_->IsFilesystemActive()) {
return fs_->GetError();
}
return target_->Flush(options, dbg);
}
IOStatus TestFSRandomRWFile::Sync(const IOOptions& options,
IODebugContext* dbg) {
if (!fs_->IsFilesystemActive()) {
return fs_->GetError();
}
return target_->Sync(options, dbg);
}
IOStatus FaultInjectionTestFS::NewDirectory(
const std::string& name, const IOOptions& options,
std::unique_ptr<FSDirectory>* result, IODebugContext* dbg) {
std::unique_ptr<FSDirectory> r;
IOStatus io_s = target()->NewDirectory(name, options, &r, dbg);
assert(io_s.ok());
if (!io_s.ok()) {
return io_s;
}
result->reset(
new TestFSDirectory(this, TestFSTrimDirname(name), r.release()));
return IOStatus::OK();
}
IOStatus FaultInjectionTestFS::NewWritableFile(
const std::string& fname, const FileOptions& file_opts,
std::unique_ptr<FSWritableFile>* result, IODebugContext* dbg) {
if (!IsFilesystemActive()) {
return GetError();
}
// Not allow overwriting files
IOStatus io_s = target()->FileExists(fname, IOOptions(), dbg);
if (io_s.ok()) {
return IOStatus::Corruption("File already exists.");
} else if (!io_s.IsNotFound()) {
assert(io_s.IsIOError());
return io_s;
}
io_s = target()->NewWritableFile(fname, file_opts, result, dbg);
if (io_s.ok()) {
result->reset(new TestFSWritableFile(fname, std::move(*result), this));
// WritableFileWriter* file is opened
// again then it will be truncated - so forget our saved state.
UntrackFile(fname);
MutexLock l(&mutex_);
open_files_.insert(fname);
auto dir_and_name = TestFSGetDirAndName(fname);
auto& list = dir_to_new_files_since_last_sync_[dir_and_name.first];
list.insert(dir_and_name.second);
}
return io_s;
}
IOStatus FaultInjectionTestFS::ReopenWritableFile(
const std::string& fname, const FileOptions& file_opts,
std::unique_ptr<FSWritableFile>* result, IODebugContext* dbg) {
if (!IsFilesystemActive()) {
return GetError();
}
IOStatus io_s = target()->ReopenWritableFile(fname, file_opts, result, dbg);
if (io_s.ok()) {
result->reset(new TestFSWritableFile(fname, std::move(*result), this));
// WritableFileWriter* file is opened
// again then it will be truncated - so forget our saved state.
UntrackFile(fname);
MutexLock l(&mutex_);
open_files_.insert(fname);
auto dir_and_name = TestFSGetDirAndName(fname);
auto& list = dir_to_new_files_since_last_sync_[dir_and_name.first];
list.insert(dir_and_name.second);
}
return io_s;
}
IOStatus FaultInjectionTestFS::NewRandomRWFile(
const std::string& fname, const FileOptions& file_opts,
std::unique_ptr<FSRandomRWFile>* result, IODebugContext* dbg) {
if (!IsFilesystemActive()) {
return GetError();
}
IOStatus io_s = target()->NewRandomRWFile(fname, file_opts, result, dbg);
if (io_s.ok()) {
result->reset(new TestFSRandomRWFile(fname, std::move(*result), this));
// WritableFileWriter* file is opened
// again then it will be truncated - so forget our saved state.
UntrackFile(fname);
MutexLock l(&mutex_);
open_files_.insert(fname);
auto dir_and_name = TestFSGetDirAndName(fname);
auto& list = dir_to_new_files_since_last_sync_[dir_and_name.first];
list.insert(dir_and_name.second);
}
return io_s;
}
IOStatus FaultInjectionTestFS::NewRandomAccessFile(
const std::string& fname, const FileOptions& file_opts,
std::unique_ptr<FSRandomAccessFile>* result, IODebugContext* dbg) {
if (!IsFilesystemActive()) {
return GetError();
}
return target()->NewRandomAccessFile(fname, file_opts, result, dbg);
}
IOStatus FaultInjectionTestFS::DeleteFile(const std::string& f,
const IOOptions& options,
IODebugContext* dbg) {
if (!IsFilesystemActive()) {
return GetError();
}
IOStatus io_s = FileSystemWrapper::DeleteFile(f, options, dbg);
if (!io_s.ok()) {
fprintf(stderr, "Cannot delete file %s: %s\n", f.c_str(),
io_s.ToString().c_str());
}
if (io_s.ok()) {
UntrackFile(f);
}
return io_s;
}
IOStatus FaultInjectionTestFS::RenameFile(const std::string& s,
const std::string& t,
const IOOptions& options,
IODebugContext* dbg) {
if (!IsFilesystemActive()) {
return GetError();
}
IOStatus io_s = FileSystemWrapper::RenameFile(s, t, options, dbg);
if (io_s.ok()) {
MutexLock l(&mutex_);
if (db_file_state_.find(s) != db_file_state_.end()) {
db_file_state_[t] = db_file_state_[s];
db_file_state_.erase(s);
}
auto sdn = TestFSGetDirAndName(s);
auto tdn = TestFSGetDirAndName(t);
if (dir_to_new_files_since_last_sync_[sdn.first].erase(sdn.second) != 0) {
auto& tlist = dir_to_new_files_since_last_sync_[tdn.first];
assert(tlist.find(tdn.second) == tlist.end());
tlist.insert(tdn.second);
}
}
return io_s;
}
void FaultInjectionTestFS::WritableFileClosed(const FSFileState& state) {
MutexLock l(&mutex_);
if (open_files_.find(state.filename_) != open_files_.end()) {
db_file_state_[state.filename_] = state;
open_files_.erase(state.filename_);
}
}
void FaultInjectionTestFS::WritableFileSynced(const FSFileState& state) {
MutexLock l(&mutex_);
if (open_files_.find(state.filename_) != open_files_.end()) {
if (db_file_state_.find(state.filename_) == db_file_state_.end()) {
db_file_state_.insert(std::make_pair(state.filename_, state));
} else {
db_file_state_[state.filename_] = state;
}
}
}
void FaultInjectionTestFS::WritableFileAppended(const FSFileState& state) {
MutexLock l(&mutex_);
if (open_files_.find(state.filename_) != open_files_.end()) {
if (db_file_state_.find(state.filename_) == db_file_state_.end()) {
db_file_state_.insert(std::make_pair(state.filename_, state));
} else {
db_file_state_[state.filename_] = state;
}
}
}
IOStatus FaultInjectionTestFS::DropUnsyncedFileData() {
IOStatus io_s;
MutexLock l(&mutex_);
for (std::map<std::string, FSFileState>::iterator it = db_file_state_.begin();
io_s.ok() && it != db_file_state_.end(); ++it) {
FSFileState& fs_state = it->second;
if (!fs_state.IsFullySynced()) {
io_s = fs_state.DropUnsyncedData();
}
}
return io_s;
}
IOStatus FaultInjectionTestFS::DropRandomUnsyncedFileData(Random* rnd) {
IOStatus io_s;
MutexLock l(&mutex_);
for (std::map<std::string, FSFileState>::iterator it = db_file_state_.begin();
io_s.ok() && it != db_file_state_.end(); ++it) {
FSFileState& fs_state = it->second;
if (!fs_state.IsFullySynced()) {
io_s = fs_state.DropRandomUnsyncedData(rnd);
}
}
return io_s;
}
IOStatus FaultInjectionTestFS::DeleteFilesCreatedAfterLastDirSync(
const IOOptions& options, IODebugContext* dbg) {
// Because DeleteFile access this container make a copy to avoid deadlock
std::map<std::string, std::set<std::string>> map_copy;
{
MutexLock l(&mutex_);
map_copy.insert(dir_to_new_files_since_last_sync_.begin(),
dir_to_new_files_since_last_sync_.end());
}
for (auto& pair : map_copy) {
for (std::string name : pair.second) {
IOStatus io_s = DeleteFile(pair.first + "/" + name, options, dbg);
if (!io_s.ok()) {
return io_s;
}
}
}
return IOStatus::OK();
}
void FaultInjectionTestFS::ResetState() {
MutexLock l(&mutex_);
db_file_state_.clear();
dir_to_new_files_since_last_sync_.clear();
SetFilesystemActiveNoLock(true);
}
void FaultInjectionTestFS::UntrackFile(const std::string& f) {
MutexLock l(&mutex_);
auto dir_and_name = TestFSGetDirAndName(f);
dir_to_new_files_since_last_sync_[dir_and_name.first].erase(
dir_and_name.second);
db_file_state_.erase(f);
open_files_.erase(f);
}
} // namespace ROCKSDB_NAMESPACE

251
test_util/fault_injection_test_fs.h Normal file
View File

@ -0,0 +1,251 @@
// 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).
//
// Copyright 2014 The LevelDB Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file. See the AUTHORS file for names of contributors.
// This test uses a custom FileSystem to keep track of the state of a file
// system the last "Sync". The data being written is cached in a "buffer".
// Only when "Sync" is called, the data will be persistent. It can similate
// file data loss (or entire files) not protected by a "Sync". For any of the
// FileSystem related operations, by specify the "IOStatus Error", a specific
// error can be returned when file system is not activated.
#pragma once
#include <map>
#include <set>
#include <string>
#include "db/version_set.h"
#include "env/mock_env.h"
#include "file/filename.h"
#include "include/rocksdb/file_system.h"
#include "rocksdb/db.h"
#include "rocksdb/env.h"
#include "util/mutexlock.h"
#include "util/random.h"
namespace ROCKSDB_NAMESPACE {
class TestFSWritableFile;
class FaultInjectionTestFS;
struct FSFileState {
std::string filename_;
ssize_t pos_;
ssize_t pos_at_last_sync_;
ssize_t pos_at_last_flush_;
std::string buffer_;
explicit FSFileState(const std::string& filename)
: filename_(filename),
pos_(-1),
pos_at_last_sync_(-1),
pos_at_last_flush_(-1) {}
FSFileState() : pos_(-1), pos_at_last_sync_(-1), pos_at_last_flush_(-1) {}
bool IsFullySynced() const { return pos_ <= 0 || pos_ == pos_at_last_sync_; }
IOStatus DropUnsyncedData();
IOStatus DropRandomUnsyncedData(Random* rand);
};
// A wrapper around WritableFileWriter* file
// is written to or sync'ed.
class TestFSWritableFile : public FSWritableFile {
public:
explicit TestFSWritableFile(const std::string& fname,
std::unique_ptr<FSWritableFile>&& f,
FaultInjectionTestFS* fs);
virtual ~TestFSWritableFile();
virtual IOStatus Append(const Slice& data, const IOOptions&,
IODebugContext*) override;
virtual IOStatus Truncate(uint64_t size, const IOOptions& options,
IODebugContext* dbg) override {
return target_->Truncate(size, options, dbg);
}
virtual IOStatus Close(const IOOptions& options,
IODebugContext* dbg) override;
virtual IOStatus Flush(const IOOptions&, IODebugContext*) override;
virtual IOStatus Sync(const IOOptions& options, IODebugContext* dbg) override;
virtual bool IsSyncThreadSafe() const override { return true; }
virtual IOStatus PositionedAppend(const Slice& data, uint64_t offset,
const IOOptions& options,
IODebugContext* dbg) override {
return target_->PositionedAppend(data, offset, options, dbg);
}
virtual bool use_direct_io() const override {
return target_->use_direct_io();
};
private:
FSFileState state_;
std::unique_ptr<FSWritableFile> target_;
bool writable_file_opened_;
FaultInjectionTestFS* fs_;
port::Mutex mutex_;
};
// A wrapper around WritableFileWriter* file
// is written to or sync'ed.
class TestFSRandomRWFile : public FSRandomRWFile {
public:
explicit TestFSRandomRWFile(const std::string& fname,
std::unique_ptr<FSRandomRWFile>&& f,
FaultInjectionTestFS* fs);
virtual ~TestFSRandomRWFile();
IOStatus Write(uint64_t offset, const Slice& data, const IOOptions& options,
IODebugContext* dbg) override;
IOStatus Read(uint64_t offset, size_t n, const IOOptions& options,
Slice* result, char* scratch,
IODebugContext* dbg) const override;
IOStatus Close(const IOOptions& options, IODebugContext* dbg) override;
IOStatus Flush(const IOOptions& options, IODebugContext* dbg) override;
IOStatus Sync(const IOOptions& options, IODebugContext* dbg) override;
size_t GetRequiredBufferAlignment() const override {
return target_->GetRequiredBufferAlignment();
}
bool use_direct_io() const override { return target_->use_direct_io(); };
private:
std::unique_ptr<FSRandomRWFile> target_;
bool file_opened_;
FaultInjectionTestFS* fs_;
};
class TestFSDirectory : public FSDirectory {
public:
explicit TestFSDirectory(FaultInjectionTestFS* fs, std::string dirname,
FSDirectory* dir)
: fs_(fs), dirname_(dirname), dir_(dir) {}
~TestFSDirectory() {}
virtual IOStatus Fsync(const IOOptions& options,
IODebugContext* dbg) override;
private:
FaultInjectionTestFS* fs_;
std::string dirname_;
std::unique_ptr<FSDirectory> dir_;
};
class FaultInjectionTestFS : public FileSystemWrapper {
public:
explicit FaultInjectionTestFS(FileSystem* base)
: FileSystemWrapper(base), filesystem_active_(true) {}
virtual ~FaultInjectionTestFS() {}
const char* Name() const override { return "FaultInjectionTestFS"; }
IOStatus NewDirectory(const std::string& name, const IOOptions& options,
std::unique_ptr<FSDirectory>* result,
IODebugContext* dbg) override;
IOStatus NewWritableFile(const std::string& fname,
const FileOptions& file_opts,
std::unique_ptr<FSWritableFile>* result,
IODebugContext* dbg) override;
IOStatus ReopenWritableFile(const std::string& fname,
const FileOptions& file_opts,
std::unique_ptr<FSWritableFile>* result,
IODebugContext* dbg) override;
IOStatus NewRandomRWFile(const std::string& fname,
const FileOptions& file_opts,
std::unique_ptr<FSRandomRWFile>* result,
IODebugContext* dbg) override;
IOStatus NewRandomAccessFile(const std::string& fname,
const FileOptions& file_opts,
std::unique_ptr<FSRandomAccessFile>* result,
IODebugContext* dbg) override;
virtual IOStatus DeleteFile(const std::string& f, const IOOptions& options,
IODebugContext* dbg) override;
virtual IOStatus RenameFile(const std::string& s, const std::string& t,
const IOOptions& options,
IODebugContext* dbg) override;
// Undef to eliminate clash on Windows
#undef GetFreeSpace
virtual IOStatus GetFreeSpace(const std::string& path,
const IOOptions& options, uint64_t* disk_free,
IODebugContext* dbg) override {
if (!IsFilesystemActive() && error_ == IOStatus::NoSpace()) {
*disk_free = 0;
return IOStatus::OK();
} else {
return target()->GetFreeSpace(path, options, disk_free, dbg);
}
}
void WritableFileClosed(const FSFileState& state);
void WritableFileSynced(const FSFileState& state);
void WritableFileAppended(const FSFileState& state);
IOStatus DropUnsyncedFileData();
IOStatus DropRandomUnsyncedFileData(Random* rnd);
IOStatus DeleteFilesCreatedAfterLastDirSync(const IOOptions& options,
IODebugContext* dbg);
void ResetState();
void UntrackFile(const std::string& f);
void SyncDir(const std::string& dirname) {
MutexLock l(&mutex_);
dir_to_new_files_since_last_sync_.erase(dirname);
}
// Setting the filesystem to inactive is the test equivalent to simulating a
// system reset. Setting to inactive will freeze our saved filesystem state so
// that it will stop being recorded. It can then be reset back to the state at
// the time of the reset.
bool IsFilesystemActive() {
MutexLock l(&mutex_);
return filesystem_active_;
}
void SetFilesystemActiveNoLock(
bool active, IOStatus error = IOStatus::Corruption("Not active")) {
filesystem_active_ = active;
if (!active) {
error_ = error;
}
}
void SetFilesystemActive(
bool active, IOStatus error = IOStatus::Corruption("Not active")) {
MutexLock l(&mutex_);
SetFilesystemActiveNoLock(active, error);
}
void AssertNoOpenFile() { assert(open_files_.empty()); }
IOStatus GetError() { return error_; }
void SetFileSystemIOError(IOStatus io_error) {
MutexLock l(&mutex_);
error_ = io_error;
}
private:
port::Mutex mutex_;
std::map<std::string, FSFileState> db_file_state_;
std::set<std::string> open_files_;
std::unordered_map<std::string, std::set<std::string>>
dir_to_new_files_since_last_sync_;
bool filesystem_active_; // Record flushes, syncs, writes
IOStatus error_;
};
} // namespace ROCKSDB_NAMESPACE