b4fa51dfaf
Summary: There is a potential failure case in DBImpl::SwitchMemtable() that is not handled properly. The call to cur_log_writer->WriteBuffer() can fail due to an IO error. In that case, we need to call SetBGError() in order set the background error since the WriteBuffer() failure may result in data loss. Also, the asserts for !new_mem and !new_log are incorrect, as those would have been allocated by the time this failure is detected. Pull Request resolved: https://github.com/facebook/rocksdb/pull/5072 Differential Revision: D14461384 Pulled By: anand1976 fbshipit-source-id: fb59bce9d61378f37d2dfcd28c0b704b0f43c3cf
181 lines
6.0 KiB
C++
181 lines
6.0 KiB
C++
// 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).
|
|
|
|
#include <atomic>
|
|
#include <memory>
|
|
#include <thread>
|
|
#include <vector>
|
|
#include "db/db_test_util.h"
|
|
#include "db/write_batch_internal.h"
|
|
#include "db/write_thread.h"
|
|
#include "port/port.h"
|
|
#include "port/stack_trace.h"
|
|
#include "util/fault_injection_test_env.h"
|
|
#include "util/string_util.h"
|
|
#include "util/sync_point.h"
|
|
|
|
namespace rocksdb {
|
|
|
|
// Test variations of WriteImpl.
|
|
class DBWriteTest : public DBTestBase, public testing::WithParamInterface<int> {
|
|
public:
|
|
DBWriteTest() : DBTestBase("/db_write_test") {}
|
|
|
|
Options GetOptions() { return DBTestBase::GetOptions(GetParam()); }
|
|
|
|
void Open() { DBTestBase::Reopen(GetOptions()); }
|
|
};
|
|
|
|
// It is invalid to do sync write while disabling WAL.
|
|
TEST_P(DBWriteTest, SyncAndDisableWAL) {
|
|
WriteOptions write_options;
|
|
write_options.sync = true;
|
|
write_options.disableWAL = true;
|
|
ASSERT_TRUE(dbfull()->Put(write_options, "foo", "bar").IsInvalidArgument());
|
|
WriteBatch batch;
|
|
ASSERT_OK(batch.Put("foo", "bar"));
|
|
ASSERT_TRUE(dbfull()->Write(write_options, &batch).IsInvalidArgument());
|
|
}
|
|
|
|
TEST_P(DBWriteTest, IOErrorOnWALWritePropagateToWriteThreadFollower) {
|
|
constexpr int kNumThreads = 5;
|
|
std::unique_ptr<FaultInjectionTestEnv> mock_env(
|
|
new FaultInjectionTestEnv(Env::Default()));
|
|
Options options = GetOptions();
|
|
options.env = mock_env.get();
|
|
Reopen(options);
|
|
std::atomic<int> ready_count{0};
|
|
std::atomic<int> leader_count{0};
|
|
std::vector<port::Thread> threads;
|
|
mock_env->SetFilesystemActive(false);
|
|
|
|
// Wait until all threads linked to write threads, to make sure
|
|
// all threads join the same batch group.
|
|
SyncPoint::GetInstance()->SetCallBack(
|
|
"WriteThread::JoinBatchGroup:Wait", [&](void* arg) {
|
|
ready_count++;
|
|
auto* w = reinterpret_cast<WriteThread::Writer*>(arg);
|
|
if (w->state == WriteThread::STATE_GROUP_LEADER) {
|
|
leader_count++;
|
|
while (ready_count < kNumThreads) {
|
|
// busy waiting
|
|
}
|
|
}
|
|
});
|
|
SyncPoint::GetInstance()->EnableProcessing();
|
|
for (int i = 0; i < kNumThreads; i++) {
|
|
threads.push_back(port::Thread(
|
|
[&](int index) {
|
|
// All threads should fail.
|
|
auto res = Put("key" + ToString(index), "value");
|
|
if (options.manual_wal_flush) {
|
|
ASSERT_TRUE(res.ok());
|
|
// we should see fs error when we do the flush
|
|
|
|
// TSAN reports a false alarm for lock-order-inversion but Open and
|
|
// FlushWAL are not run concurrently. Disabling this until TSAN is
|
|
// fixed.
|
|
// res = dbfull()->FlushWAL(false);
|
|
// ASSERT_FALSE(res.ok());
|
|
} else {
|
|
ASSERT_FALSE(res.ok());
|
|
}
|
|
},
|
|
i));
|
|
}
|
|
for (int i = 0; i < kNumThreads; i++) {
|
|
threads[i].join();
|
|
}
|
|
ASSERT_EQ(1, leader_count);
|
|
// Close before mock_env destruct.
|
|
Close();
|
|
}
|
|
|
|
TEST_P(DBWriteTest, ManualWalFlushInEffect) {
|
|
Options options = GetOptions();
|
|
Reopen(options);
|
|
// try the 1st WAL created during open
|
|
ASSERT_TRUE(Put("key" + ToString(0), "value").ok());
|
|
ASSERT_TRUE(options.manual_wal_flush != dbfull()->TEST_WALBufferIsEmpty());
|
|
ASSERT_TRUE(dbfull()->FlushWAL(false).ok());
|
|
ASSERT_TRUE(dbfull()->TEST_WALBufferIsEmpty());
|
|
// try the 2nd wal created during SwitchWAL
|
|
dbfull()->TEST_SwitchWAL();
|
|
ASSERT_TRUE(Put("key" + ToString(0), "value").ok());
|
|
ASSERT_TRUE(options.manual_wal_flush != dbfull()->TEST_WALBufferIsEmpty());
|
|
ASSERT_TRUE(dbfull()->FlushWAL(false).ok());
|
|
ASSERT_TRUE(dbfull()->TEST_WALBufferIsEmpty());
|
|
}
|
|
|
|
TEST_P(DBWriteTest, IOErrorOnWALWriteTriggersReadOnlyMode) {
|
|
std::unique_ptr<FaultInjectionTestEnv> mock_env(
|
|
new FaultInjectionTestEnv(Env::Default()));
|
|
Options options = GetOptions();
|
|
options.env = mock_env.get();
|
|
Reopen(options);
|
|
for (int i = 0; i < 2; i++) {
|
|
// Forcibly fail WAL write for the first Put only. Subsequent Puts should
|
|
// fail due to read-only mode
|
|
mock_env->SetFilesystemActive(i != 0);
|
|
auto res = Put("key" + ToString(i), "value");
|
|
// TSAN reports a false alarm for lock-order-inversion but Open and
|
|
// FlushWAL are not run concurrently. Disabling this until TSAN is
|
|
// fixed.
|
|
/*
|
|
if (options.manual_wal_flush && i == 0) {
|
|
// even with manual_wal_flush the 2nd Put should return error because of
|
|
// the read-only mode
|
|
ASSERT_TRUE(res.ok());
|
|
// we should see fs error when we do the flush
|
|
res = dbfull()->FlushWAL(false);
|
|
}
|
|
*/
|
|
if (!options.manual_wal_flush) {
|
|
ASSERT_FALSE(res.ok());
|
|
}
|
|
}
|
|
// Close before mock_env destruct.
|
|
Close();
|
|
}
|
|
|
|
TEST_P(DBWriteTest, IOErrorOnSwitchMemtable) {
|
|
Random rnd(301);
|
|
std::unique_ptr<FaultInjectionTestEnv> mock_env(
|
|
new FaultInjectionTestEnv(Env::Default()));
|
|
Options options = GetOptions();
|
|
options.env = mock_env.get();
|
|
options.writable_file_max_buffer_size = 4 * 1024 * 1024;
|
|
options.write_buffer_size = 3 * 512 * 1024;
|
|
options.wal_bytes_per_sync = 256 * 1024;
|
|
options.manual_wal_flush = true;
|
|
Reopen(options);
|
|
mock_env->SetFilesystemActive(false, Status::IOError("Not active"));
|
|
Status s;
|
|
for (int i = 0; i < 4 * 512; ++i) {
|
|
s = Put(Key(i), RandomString(&rnd, 1024));
|
|
if (!s.ok()) {
|
|
break;
|
|
}
|
|
}
|
|
ASSERT_EQ(s.severity(), Status::Severity::kFatalError);
|
|
|
|
mock_env->SetFilesystemActive(true);
|
|
// Close before mock_env destruct.
|
|
Close();
|
|
}
|
|
|
|
INSTANTIATE_TEST_CASE_P(DBWriteTestInstance, DBWriteTest,
|
|
testing::Values(DBTestBase::kDefault,
|
|
DBTestBase::kConcurrentWALWrites,
|
|
DBTestBase::kPipelinedWrite));
|
|
|
|
} // namespace rocksdb
|
|
|
|
int main(int argc, char** argv) {
|
|
rocksdb::port::InstallStackTraceHandler();
|
|
::testing::InitGoogleTest(&argc, argv);
|
|
return RUN_ALL_TESTS();
|
|
}
|