diff --git a/src/common/test/test_rocksdb_txn_store.cc b/src/common/test/test_rocksdb_txn_store.cc index 72a31cf..86dd243 100644 --- a/src/common/test/test_rocksdb_txn_store.cc +++ b/src/common/test/test_rocksdb_txn_store.cc @@ -16,6 +16,7 @@ #include #include "common/rocksdb_txn_store.h" +#include "common/txn_manager.h" // NOLINTBEGIN(readability-magic-numbers) @@ -110,6 +111,20 @@ TEST_F(TestRocksdbTxnStore, Hset) { ASSERT_EQ(get_value(expected_key), "test_value"); } +TEST_F(TestRocksdbTxnStore, TxnGuard) { + auto txn = create_transaction(); + std::string expected_key = "test_key\1test_field"; + PAIN_TXN(txn.get()) { + auto status = txn->hset("test_key", "test_field", "test_value"); + EXPECT_TRUE(status.ok()) << status.error_str(); + EXPECT_FALSE(key_exists(expected_key)); + return status; + }; + + ASSERT_TRUE(key_exists(expected_key)); + ASSERT_EQ(get_value(expected_key), "test_value"); +} + // 测试 hset 多个字段 TEST_F(TestRocksdbTxnStore, HsetMultipleFields) { auto txn = create_transaction(); diff --git a/src/common/txn_manager.cc b/src/common/txn_manager.cc new file mode 100644 index 0000000..f3a22ac --- /dev/null +++ b/src/common/txn_manager.cc @@ -0,0 +1,7 @@ +#include "common/txn_manager.h" + +namespace pain::common { + +thread_local TxnStore* TxnManager::g_txn_store = nullptr; + +} // namespace pain::common diff --git a/src/common/txn_manager.h b/src/common/txn_manager.h new file mode 100644 index 0000000..d144796 --- /dev/null +++ b/src/common/txn_manager.h @@ -0,0 +1,69 @@ +#pragma once + +#include +#include "common/txn_store.h" + +namespace pain::common { + +class TxnManager { +public: + static const TxnManager& instance() { + static TxnManager s_instance; + return s_instance; + } + + bool in_txn() const { + return g_txn_store != nullptr; + } + + TxnStore* get_txn_store() const { + return g_txn_store; + } + + void begin(TxnStore* txn_store) const { + g_txn_store = txn_store; + } + + Status commit() const { + auto status = g_txn_store->commit(); + g_txn_store = nullptr; + return status; + } + + Status rollback() const { + auto status = g_txn_store->rollback(); + g_txn_store = nullptr; + return status; + } + +private: + static thread_local TxnStore* g_txn_store; +}; + +class TxnGuard { +public: + TxnGuard(TxnStore* txn_store) { + TxnManager::instance().begin(txn_store); + } + + ~TxnGuard() { + if (_status.ok()) { + TxnManager::instance().commit(); + } else { + TxnManager::instance().rollback(); + } + } + + void operator+(std::function cb) { + auto txn_store = TxnManager::instance().get_txn_store(); + _status = cb(txn_store); + } + +private: + Status _status; +}; + +} // namespace pain::common + +#define PAIN_TXN(txn_store) \ + pain::common::TxnGuard(txn_store) + [&]([[maybe_unused]] pain::common::TxnStore * txn) mutable -> pain::Status diff --git a/src/deva/deva.cc b/src/deva/deva.cc index 08a54a9..34e70bb 100644 --- a/src/deva/deva.cc +++ b/src/deva/deva.cc @@ -1,6 +1,7 @@ #include "deva/deva.h" #include #include +#include "common/txn_manager.h" #include "deva/macro.h" #define DEVA_METHOD(name) \ @@ -149,12 +150,44 @@ DEVA_METHOD(SealAndNewChunk) { return Status::OK(); } +Status Deva::set_applied_index(int64_t index) { + if (index <= _applied_index) { + PLOG_WARN(("desc", "index is applied already")("index", index)); + return Status::OK(); + } + auto in_txn = common::TxnManager::instance().in_txn(); + auto this_txn = _store->begin_txn(); + auto txn = in_txn ? common::TxnManager::instance().get_txn_store() : this_txn.get(); + if (txn == nullptr) { + return Status(EIO, "Failed to begin transaction"); + } + auto status = txn->hset(_meta_key, _applied_index_key, std::to_string(index)); + return status; + if (in_txn) { + return Status::OK(); + } + + status = txn->commit(); + return status; +} + Status Deva::save_snapshot(std::string_view path, std::vector* files) { return _store->check_point(path.data(), files); } Status Deva::load_snapshot(std::string_view path) { - return _store->recover(path.data()); + auto status = _store->recover(path.data()); + if (!status.ok()) { + return status; + } + // get applied index + std::string applied_index_str; + status = _store->hget(_meta_key, _applied_index_key, &applied_index_str); + if (!status.ok()) { + return status; + } + _applied_index = std::stoll(applied_index_str); + return Status::OK(); } } // namespace pain::deva diff --git a/src/deva/deva.h b/src/deva/deva.h index a3f9116..26d3de9 100644 --- a/src/deva/deva.h +++ b/src/deva/deva.h @@ -1,11 +1,16 @@ #pragma once +#include #include #include #include "pain/proto/deva_store.pb.h" #include "common/store.h" +#include "common/txn_manager.h" +#include "common/txn_store.h" #include "deva/container.h" #include "deva/namespace.h" +#include "deva/op.h" + #define DEVA_ENTRY(name) \ Status name([[maybe_unused]] int32_t version, \ [[maybe_unused]] const pain::proto::deva::store::name##Request* request, \ @@ -15,7 +20,25 @@ [[maybe_unused]] const pain::proto::deva::store::name##Request* request, \ [[maybe_unused]] pain::proto::deva::store::name##Response* response, \ [[maybe_unused]] int64_t index) { \ - return name(version, request, response, index); \ + if (!need_apply(OpType::k##name)) { \ + return name(version, request, response, index); \ + } \ + if (check_index_is_applied(index)) { \ + PLOG_INFO(("desc", "index is applied already")("index", index)); \ + return Status::OK(); \ + } \ + auto txn = _store->begin_txn(); \ + auto status = Status::OK(); \ + PAIN_TXN(txn.get()) { \ + status = set_applied_index(index); \ + if (!status.ok()) { \ + PLOG_ERROR(("desc", "set applied index failed")("index", index)("error", status.error_str())); \ + return status; \ + } \ + status = name(version, request, response, index); \ + return status; \ + }; \ + return status; \ } namespace pain::deva { @@ -41,12 +64,19 @@ class Deva : public Container { private: Status create(const std::string& path, const UUID& id, FileType type); + Status set_applied_index(int64_t applied_index); + bool check_index_is_applied(int64_t index) const { + return index != 0 && index <= _applied_index; + } private: std::atomic _use_count; common::StorePtr _store; Namespace _namespace; std::unordered_map _file_infos; + const char* _meta_key = "meta"; + const char* _applied_index_key = "applied_index"; + int64_t _applied_index = 0; friend void intrusive_ptr_add_ref(Deva* deva) { ++deva->_use_count; diff --git a/src/deva/namespace.cc b/src/deva/namespace.cc index ca65fef..701c7d9 100644 --- a/src/deva/namespace.cc +++ b/src/deva/namespace.cc @@ -1,6 +1,7 @@ #include "deva/namespace.h" #include #include +#include "common/txn_manager.h" #include "common/txn_store.h" namespace pain::deva { @@ -14,8 +15,10 @@ Status Namespace::load() { } Status Namespace::create(const UUID& parent, const std::string& name, FileType type, const UUID& inode) { - auto txn = _store->begin_txn(); - if (!txn) { + auto in_txn = common::TxnManager::instance().in_txn(); + auto this_txn = _store->begin_txn(); + auto txn = in_txn ? common::TxnManager::instance().get_txn_store() : this_txn.get(); + if (txn == nullptr) { return Status(EIO, "Failed to begin transaction"); } auto rollback = make_scope_exit([&txn]() { @@ -24,6 +27,9 @@ Status Namespace::create(const UUID& parent, const std::string& name, FileType t PLOG_ERROR(("desc", "Failed to rollback")("status", status)); } }); + if (in_txn) { + rollback.release(); + } // find parent std::string dentries_str; proto::DirEntries dentries; @@ -75,7 +81,14 @@ Status Namespace::create(const UUID& parent, const std::string& name, FileType t if (!file_info.SerializeToString(&file_info_str)) { return Status(EBADMSG, "Failed to serialize file info"); } - txn->hset(_inode_key, inode.str(), file_info_str); + status = txn->hset(_inode_key, inode.str(), file_info_str); + if (!status.ok()) { + return status; + } + + if (in_txn) { + return Status::OK(); + } status = txn->commit(); if (!status.ok()) { @@ -86,17 +99,22 @@ Status Namespace::create(const UUID& parent, const std::string& name, FileType t } Status Namespace::remove(const UUID& parent, const std::string& name) { - // find parent - auto txn = _store->begin_txn(); - if (!txn) { + auto in_txn = common::TxnManager::instance().in_txn(); + auto this_txn = _store->begin_txn(); + auto txn = in_txn ? common::TxnManager::instance().get_txn_store() : this_txn.get(); + if (txn == nullptr) { return Status(EIO, "Failed to begin transaction"); } + // find parent auto rollback = make_scope_exit([&txn]() { auto status = txn->rollback(); if (!status.ok()) { PLOG_ERROR(("desc", "Failed to rollback")("status", status)); } }); + if (in_txn) { + rollback.release(); + } // get dentries std::string dentries_str; auto status = txn->hget(_dentry_key, parent.str(), &dentries_str); @@ -122,10 +140,20 @@ Status Namespace::remove(const UUID& parent, const std::string& name) { if (!dentries.SerializeToString(&dentries_str)) { return Status(EBADMSG, "Failed to serialize dentries"); } - txn->hset(_dentry_key, parent.str(), dentries_str); + status = txn->hset(_dentry_key, parent.str(), dentries_str); + if (!status.ok()) { + return status; + } // remove file info - txn->hdel(_inode_key, file_id.str()); + status = txn->hdel(_inode_key, file_id.str()); + if (!status.ok()) { + return status; + } + + if (in_txn) { + return Status::OK(); + } status = txn->commit(); if (!status.ok()) { @@ -138,18 +166,34 @@ Status Namespace::remove(const UUID& parent, const std::string& name) { void Namespace::list(const UUID& parent, std::list* entries) const { entries->clear(); std::string dentries_str; - auto status = _store->hget(_dentry_key, parent.str(), &dentries_str); + auto in_txn = common::TxnManager::instance().in_txn(); + auto this_txn = _store->begin_txn(); + auto txn = in_txn ? common::TxnManager::instance().get_txn_store() : this_txn.get(); + if (txn == nullptr) { + return; + } + auto status = txn->hget(_dentry_key, parent.str(), &dentries_str); if (!status.ok()) { + PLOG_ERROR(("desc", "Failed to get dentries")("status", status)); return; } proto::DirEntries dentries; if (!dentries.ParseFromString(dentries_str)) { + PLOG_ERROR(("desc", "Failed to parse dentries")("status", status)); return; } for (const auto& dentry : dentries.entries()) { entries->emplace_back( UUID(dentry.file_id().high(), dentry.file_id().low()), dentry.name(), static_cast(dentry.type())); } + if (in_txn) { + return; + } + + status = txn->commit(); + if (!status.ok()) { + PLOG_ERROR(("desc", "Failed to commit")("status", status)); + } } // parse path such as /a/b/c to ["a", "b", "c"] @@ -184,9 +228,16 @@ Status Namespace::lookup(const char* path, UUID* inode, FileType* file_type) con UUID parent = _root; *file_type = FileType::kDirectory; + auto in_txn = common::TxnManager::instance().in_txn(); + auto this_txn = _store->begin_txn(); + auto txn = in_txn ? common::TxnManager::instance().get_txn_store() : this_txn.get(); + if (txn == nullptr) { + return Status(EIO, "Failed to begin transaction"); + } + for (const auto& component : components) { std::string dentries_str; - auto status = _store->hget(_dentry_key, parent.str(), &dentries_str); + auto status = txn->hget(_dentry_key, parent.str(), &dentries_str); if (!status.ok()) { return status; } @@ -208,7 +259,15 @@ Status Namespace::lookup(const char* path, UUID* inode, FileType* file_type) con *file_type = static_cast(entry->type()); } *inode = parent; - return Status::OK(); + if (in_txn) { + return Status::OK(); + } + + status = txn->commit(); + if (!status.ok()) { + PLOG_ERROR(("desc", "Failed to commit")("status", status)); + } + return status; } } // namespace pain::deva diff --git a/src/deva/test/test_namespace.cc b/src/deva/test/test_namespace.cc index 806c4c1..7d29600 100644 --- a/src/deva/test/test_namespace.cc +++ b/src/deva/test/test_namespace.cc @@ -5,6 +5,7 @@ #include #include "common/rocksdb_store.h" +#include "common/txn_manager.h" #include "deva/namespace.h" using namespace pain; @@ -253,3 +254,34 @@ TEST_F(TestNamespace, lookup_and_list) { } } } + +TEST_F(TestNamespace, lookup_and_list_in_txn) { + Namespace ns(_store); + UUID a = UUID::from_str_or_die("00000000-0000-0000-0000-000000000001"); + UUID b = UUID::from_str_or_die("00000000-0000-0000-0000-000000000002"); + UUID c = UUID::from_str_or_die("00000000-0000-0000-0000-000000000003"); + UUID d = UUID::from_str_or_die("00000000-0000-0000-0000-000000000004"); + UUID e = UUID::from_str_or_die("00000000-0000-0000-0000-000000000005"); + UUID f = UUID::from_str_or_die("00000000-0000-0000-0000-000000000006"); + auto txn = _store->begin_txn(); + PAIN_TXN(txn.get()) { + auto status = ns.create(ns.root(), "a", FileType::kDirectory, a); + EXPECT_TRUE(status.ok()) << status.error_str(); + status = ns.create(ns.root(), "b", FileType::kDirectory, b); + EXPECT_TRUE(status.ok()) << status.error_str(); + status = ns.create(ns.root(), "c", FileType::kDirectory, c); + EXPECT_TRUE(status.ok()) << status.error_str(); + return status; + }; + + std::list entries; + ns.list(ns.root(), &entries); + ASSERT_EQ(entries.size(), 3) << fmt::format("{}", fmt::join(entries, ", ")); + EXPECT_EQ(entries.front().name, "a"); + entries.pop_front(); + EXPECT_EQ(entries.front().name, "b"); + entries.pop_front(); + EXPECT_EQ(entries.front().name, "c"); + entries.pop_front(); + EXPECT_EQ(entries.size(), 0); +}