237b292515
Summary: Previously we store sequence number range of each blob files, and use the sequence number range to check if the file can be possibly visible by a snapshot. But it adds complexity to the code, since the sequence number is only available after a write. (The current implementation get sequence number by calling GetLatestSequenceNumber(), which is wrong.) With the patch, we are not storing sequence number range, and check if snapshot_sequence < obsolete_sequence to decide if the file is visible by a snapshot (previously we check if first_sequence <= snapshot_sequence < obsolete_sequence). Closes https://github.com/facebook/rocksdb/pull/3274 Differential Revision: D6571497 Pulled By: yiwu-arbug fbshipit-source-id: ca06479dc1fcd8782f6525b62b7762cd47d61909
95 lines
3.0 KiB
C++
95 lines
3.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.
|
|
ASSERT_FALSE(Put("key" + ToString(index), "value").ok());
|
|
},
|
|
i));
|
|
}
|
|
for (int i = 0; i < kNumThreads; i++) {
|
|
threads[i].join();
|
|
}
|
|
ASSERT_EQ(1, leader_count);
|
|
// 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();
|
|
}
|