From a3fb275fc96dc4b6e52bbc15eee493d75fca6e66 Mon Sep 17 00:00:00 2001 From: allen <1007729991@qq.com> Date: Sun, 14 Sep 2025 23:09:47 +0800 Subject: [PATCH] extract rsm to common Signed-off-by: allen <1007729991@qq.com> --- src/common/BUILD | 3 +- src/common/rsm/bridge.h | 39 ++++++ src/{deva => common/rsm}/container.h | 6 +- src/common/rsm/container_op.cc | 13 ++ src/{deva => common/rsm}/container_op.h | 22 +-- src/{deva => common/rsm}/op.cc | 8 +- src/common/rsm/op.h | 76 ++++++++++ src/common/rsm/op_factory.h | 14 ++ src/common/rsm/rsm.cc | 175 ++++++++++++++++++++++++ src/common/rsm/rsm.h | 77 +++++++++++ src/deva/bridge.h | 28 +--- src/deva/container_op.cc | 50 ------- src/deva/deva.h | 12 +- src/deva/deva_op_factory.h | 71 ++++++++++ src/deva/deva_service_impl.cc | 2 +- src/deva/deva_service_impl.h | 6 +- src/deva/mock/deva_machine.h | 6 +- src/deva/op.h | 87 ------------ src/deva/rsm.cc | 162 +--------------------- src/deva/rsm.h | 71 +--------- 20 files changed, 511 insertions(+), 417 deletions(-) create mode 100644 src/common/rsm/bridge.h rename src/{deva => common/rsm}/container.h (84%) create mode 100644 src/common/rsm/container_op.cc rename src/{deva => common/rsm}/container_op.h (88%) rename src/{deva => common/rsm}/op.cc (91%) create mode 100644 src/common/rsm/op.h create mode 100644 src/common/rsm/op_factory.h create mode 100644 src/common/rsm/rsm.cc create mode 100644 src/common/rsm/rsm.h delete mode 100644 src/deva/container_op.cc create mode 100644 src/deva/deva_op_factory.h delete mode 100644 src/deva/op.h diff --git a/src/common/BUILD b/src/common/BUILD index 9649789..f4a9ac5 100644 --- a/src/common/BUILD +++ b/src/common/BUILD @@ -2,7 +2,7 @@ load('//:copts.bzl', 'PAIN_COPTS', 'PAIN_LINKOPTS', 'PAIN_TEST_COPTS') cc_library( name = "pain_common", - srcs = glob(["*.h", "*.cc"]), + srcs = glob(["*.h", "*.cc", "rsm/*.h", "rsm/*.cc"]), copts = PAIN_COPTS, linkopts = PAIN_LINKOPTS, deps = [ @@ -12,6 +12,7 @@ cc_library( "@boost.smart_ptr", "@rocksdb", "@braft", + "@magic_enum", ], visibility = ["//visibility:public"], ) diff --git a/src/common/rsm/bridge.h b/src/common/rsm/bridge.h new file mode 100644 index 0000000..2f1a602 --- /dev/null +++ b/src/common/rsm/bridge.h @@ -0,0 +1,39 @@ +#pragma once + +#include +#include +#include +#include +#include "common/rsm/container_op.h" +#include "common/rsm/op.h" +#include "common/rsm/rsm.h" + +namespace pain::common { + +template +void bridge(uint32_t op_type, + int32_t version, + common::RsmPtr rsm, + const Request& request, + Response* response, + std::move_only_function cb) { + auto op = new common::ContainerOp( + version, op_type, rsm, request, response, [cb = std::move(cb)](Status status) mutable { + cb(std::move(status)); + }); + op->apply(); +} + +// Future style +template +Future bridge(uint32_t op_type, int32_t version, RsmPtr rsm, const Request& request, Response* response) { + Promise promise; + auto future = promise.get_future(); + bridge( + op_type, version, rsm, request, response, [promise = std::move(promise)](Status status) mutable { + promise.set_value(std::move(status)); + }); + return future; +} + +} // namespace pain::common diff --git a/src/deva/container.h b/src/common/rsm/container.h similarity index 84% rename from src/deva/container.h rename to src/common/rsm/container.h index 5998a88..06d4851 100644 --- a/src/deva/container.h +++ b/src/common/rsm/container.h @@ -5,14 +5,16 @@ #include #include #include +#include "common/rsm/op_factory.h" -namespace pain::deva { +namespace pain::common { class Container { public: virtual ~Container() = default; virtual Status save_snapshot(std::string_view path, std::vector* files) = 0; virtual Status load_snapshot(std::string_view path) = 0; + virtual OpFactory* op_factory() = 0; private: std::atomic _use_count = 0; @@ -30,4 +32,4 @@ class Container { using ContainerPtr = boost::intrusive_ptr; -} // namespace pain::deva +} // namespace pain::common diff --git a/src/common/rsm/container_op.cc b/src/common/rsm/container_op.cc new file mode 100644 index 0000000..d15686e --- /dev/null +++ b/src/common/rsm/container_op.cc @@ -0,0 +1,13 @@ +#include "common/rsm/container_op.h" +#include "common/rsm/op_factory.h" + +namespace pain::common { + +OpPtr decode(int32_t version, uint32_t op_type, IOBuf* buf, RsmPtr rsm) { + auto op_factory = rsm->container()->op_factory(); + auto op = op_factory->create(op_type, version, rsm); + op->decode(buf); + return op; +} + +} // namespace pain::common diff --git a/src/deva/container_op.h b/src/common/rsm/container_op.h similarity index 88% rename from src/deva/container_op.h rename to src/common/rsm/container_op.h index 86ab517..07fa548 100644 --- a/src/deva/container_op.h +++ b/src/common/rsm/container_op.h @@ -1,14 +1,14 @@ #pragma once #include +#include #include #include #include -#include "deva/macro.h" -#include "deva/op.h" -#include "deva/rsm.h" +#include "common/rsm/op.h" +#include "common/rsm/rsm.h" -namespace pain::deva { +namespace pain::common { class OpClosure : public braft::Closure { public: @@ -34,14 +34,14 @@ class OpClosure : public braft::Closure { std::shared_ptr _span; }; -OpPtr decode(int32_t version, OpType op_type, IOBuf* buf, RsmPtr rsm); +OpPtr decode(int32_t version, uint32_t op_type, IOBuf* buf, RsmPtr rsm); template class ContainerOp : public Op { public: using OnFinish = std::move_only_function; ContainerOp(int32_t version, - OpType type, + uint32_t type, RsmPtr rsm, Request request, Response* response = nullptr, @@ -57,18 +57,18 @@ class ContainerOp : public Op { } } - OpType type() const override { + uint32_t type() const override { return _type; } void apply() override { - SPAN(span); + SPAN("common", span); PLOG_DEBUG(("desc", "apply op")("type", _type)("version", _version)); if (need_apply(_type)) { braft::Task task; IOBuf buf; OpPtr self(this); - pain::deva::encode(_version, self, &buf); + pain::common::encode(_version, self, &buf); task.data = &buf; task.done = new OpClosure(self, span); task.expected_term = -1; @@ -109,7 +109,7 @@ class ContainerOp : public Op { protected: int32_t _version; - OpType _type; + uint32_t _type; RsmPtr _rsm; Request _request; Response* _response; @@ -117,4 +117,4 @@ class ContainerOp : public Op { OnFinish _finish; }; -} // namespace pain::deva +} // namespace pain::common diff --git a/src/deva/op.cc b/src/common/rsm/op.cc similarity index 91% rename from src/deva/op.cc rename to src/common/rsm/op.cc index 84b50e8..6ca9aae 100644 --- a/src/deva/op.cc +++ b/src/common/rsm/op.cc @@ -1,10 +1,10 @@ -#include "deva/op.h" +#include "common/rsm/op.h" #include #include #include "butil/iobuf.h" #include "butil/time.h" -namespace pain::deva { +namespace pain::common { void encode(int32_t version, OpPtr op, IOBuf* buf) { OpMeta op_meta = {}; @@ -19,7 +19,7 @@ void encode(int32_t version, OpPtr op, IOBuf* buf) { buf->append(meta); } -OpPtr decode(IOBuf* buf, std::move_only_function decode) { +OpPtr decode(IOBuf* buf, std::move_only_function decode) { OpMeta op_meta = {}; static_assert(sizeof(op_meta) == 64, "OpMeta size must be 64byte"); // NOLINT(readability-magic-numbers) uint32_t op_size = 0; @@ -35,4 +35,4 @@ OpPtr decode(IOBuf* buf, std::move_only_function return op; } -} // namespace pain::deva +} // namespace pain::common diff --git a/src/common/rsm/op.h b/src/common/rsm/op.h new file mode 100644 index 0000000..23bdd07 --- /dev/null +++ b/src/common/rsm/op.h @@ -0,0 +1,76 @@ +#pragma once +#include +#include +#include +#include +#include +#include + +#define DEFINE_RSM_OP(code, name, need_apply) k##name = ((code << 1) | (need_apply ? 1 : 0)) + +namespace pain::common { + +// enum class OpType : uint32_t { +// kInvalid = 0, +// // DevaOp: 1 ~ 100 +// DEFINE_DEVA_OP(1, CreateFile, true), +// DEFINE_DEVA_OP(2, CreateDir, true), +// DEFINE_DEVA_OP(3, RemoveFile, true), +// DEFINE_DEVA_OP(4, SealFile, true), +// DEFINE_DEVA_OP(5, CreateChunk, true), +// DEFINE_DEVA_OP(6, CheckInChunk, true), +// DEFINE_DEVA_OP(7, SealChunk, true), +// DEFINE_DEVA_OP(8, SealAndNewChunk, true), +// DEFINE_DEVA_OP(9, ReadDir, false), +// DEFINE_DEVA_OP(10, GetFileInfo, false), +// DEFINE_DEVA_OP(20, ManusyaHeartbeat, false), +// DEFINE_DEVA_OP(21, ListManusya, false), +// DEFINE_DEVA_OP(100, MaxDevaOp, true), +// }; + +inline constexpr bool need_apply(uint32_t type) { + return (type & 1) == 1; +} + +struct OpMeta { + int32_t version; // op version + uint32_t type; + uint64_t timestamp; + uint32_t size; + char reserved[40]; // NOLINT(readability-magic-numbers) +}; + +static_assert(sizeof(OpMeta) == 64, "OpMeta size must be 64byte"); // NOLINT(readability-magic-numbers) + +class Op; +using OpPtr = boost::intrusive_ptr; +class Op { +public: + Op() = default; + virtual ~Op() = default; + virtual uint32_t type() const = 0; + virtual void apply() = 0; + virtual void on_apply(int64_t index) = 0; + virtual void on_finish(Status status) = 0; + virtual void encode(IOBuf* buf) = 0; + virtual void decode(IOBuf* buf) = 0; + +private: + std::atomic _use_count = 0; + friend void intrusive_ptr_add_ref(Op* op) { + ++op->_use_count; + } + friend void intrusive_ptr_release(Op* op) { + if (op->_use_count.fetch_sub(1) == 1) { + delete op; + } + } +}; + +class Rsm; +using RsmPtr = boost::intrusive_ptr; + +void encode(int32_t version, OpPtr op, IOBuf* buf); +OpPtr decode(IOBuf* buf, std::move_only_function decode); + +} // namespace pain::common diff --git a/src/common/rsm/op_factory.h b/src/common/rsm/op_factory.h new file mode 100644 index 0000000..d9e68b0 --- /dev/null +++ b/src/common/rsm/op_factory.h @@ -0,0 +1,14 @@ +#pragma once + +#include "common/rsm/op.h" + +namespace pain::common { + +class OpFactory { +public: + OpFactory() = default; + virtual ~OpFactory() = default; + virtual OpPtr create(uint32_t op_type, int32_t version, RsmPtr rsm) = 0; +}; + +} // namespace pain::common diff --git a/src/common/rsm/rsm.cc b/src/common/rsm/rsm.cc new file mode 100644 index 0000000..991e22a --- /dev/null +++ b/src/common/rsm/rsm.cc @@ -0,0 +1,175 @@ +#include "common/rsm/rsm.h" + +#include // braft::Node braft::StateMachine +#include // braft::SnapshotWriter +#include // braft::AsyncClosureGuard +#include // brpc::Controller +#include // brpc::Server +#include // butil::NetToHost32 +#include // open +#include // DEFINE_* +#include +#include // O_CREAT +#include "common/rsm/container.h" +#include "common/rsm/container_op.h" +#include "common/rsm/op.h" + +namespace pain::common { + +Rsm::Rsm(const butil::EndPoint& address, + const std::string& group, + const braft::NodeOptions& node_options, + ContainerPtr container) : + _address(address), + _group(group), + _node_options(node_options), + _node(nullptr), + _leader_term(-1), + _container(container) { + _node_options.fsm = this; +} +Rsm::~Rsm() { + PLOG_INFO(("desc", "destructor rsm") // + ("group", _group) // + ("address", butil::endpoint2str(_address).c_str())); + delete _node; +} + +int Rsm::start() { + PLOG_INFO(("desc", "start rsm") // + ("group", _group) // + ("address", butil::endpoint2str(_address).c_str())); + braft::Node* node = new braft::Node(_group, braft::PeerId(_address)); + if (node->init(_node_options) != 0) { + LOG(ERROR) << "Fail to init raft node"; + delete node; + return -1; + } + _node = node; + return 0; +} + +bool Rsm::is_leader() const { + if (_node == nullptr) { + return false; + } + return _node->is_leader(); +} + +void Rsm::shutdown() { + if (_node != nullptr) { + _node->shutdown(nullptr); + } +} + +void Rsm::join() { + if (_node != nullptr) { + _node->join(); + } +} + +void Rsm::apply(const braft::Task& task) { + if (_node != nullptr) { + _node->apply(task); + } +} + +void Rsm::on_apply(braft::Iterator& iter) { + for (; iter.valid(); iter.next()) { + braft::AsyncClosureGuard closure_guard(iter.done()); + butil::IOBuf data; + off_t offset = 0; + if (iter.done() != nullptr) { + // Run at closure_guard destructed + auto c = static_cast(iter.done()); + c->set_index(iter.index()); + } else { + butil::IOBuf saved_log = iter.data(); + // clang-format off + auto op = decode(&saved_log, [rsm = RsmPtr(this)](int32_t version, uint32_t op_type, IOBuf* buf) { + return decode(version, op_type, buf, rsm); + }); + // clang-format on + op->on_apply(iter.index()); + } + + LOG(INFO) << "Write " << data.size() << " bytes" + << " from offset=" << offset << " at log_index=" << iter.index(); + } +} + +struct SnapshotArg { + braft::SnapshotWriter* writer; + braft::Closure* done; +}; + +void* Rsm::save_snapshot(void* arg) { + PLOG_INFO(("desc", "save_snapshot")); + SnapshotArg* sa = (SnapshotArg*)arg; + std::unique_ptr arg_guard(sa); + brpc::ClosureGuard done_guard(sa->done); + std::string snapshot_path = sa->writer->get_path(); + std::vector files; + auto status = sa->container->save_snapshot(snapshot_path, &files); + if (!status.ok()) { + sa->done->status() = status; + LOG(ERROR) << "Fail to save snapshot to " << snapshot_path; + return nullptr; + } + for (const auto& file : files) { + PLOG_INFO(("desc", "add_file to snapshot")("file", file)); + sa->writer->add_file(file); + } + PLOG_INFO(("desc", "save_snapshot done")); + return nullptr; +} + +void Rsm::on_snapshot_save(braft::SnapshotWriter* writer, braft::Closure* done) { + PLOG_INFO(("desc", "on_snapshot_save")); + SnapshotArg* arg = new SnapshotArg; + arg->writer = writer; + arg->done = done; + arg->container = _container; + bthread_t tid = 0; + bthread_start_urgent(&tid, nullptr, save_snapshot, arg); +} + +// NOLINTNEXTLINE +int Rsm::on_snapshot_load(braft::SnapshotReader* reader) { + PLOG_INFO(("desc", "on_snapshot_load")); + CHECK(!is_leader()) << "Leader is not supposed to load snapshot"; + auto path = reader->get_path(); + auto status = _container->load_snapshot(path); + if (!status.ok()) { + PLOG_ERROR(("desc", "fail to load snapshot from ")("path", path)("error", status.error_str())); + return -1; + } + return 0; +} + +void Rsm::on_leader_start(int64_t term) { + _leader_term.store(term, butil::memory_order_release); + LOG(INFO) << "Node becomes leader"; +} +void Rsm::on_leader_stop(const butil::Status& status) { + _leader_term.store(-1, butil::memory_order_release); + LOG(INFO) << "Node stepped down : " << status; +} + +void Rsm::on_shutdown() { + LOG(INFO) << "This node is down"; +} +void Rsm::on_error(const ::braft::Error& e) { + LOG(ERROR) << "Met raft error " << e; +} +void Rsm::on_configuration_committed(const ::braft::Configuration& conf) { + LOG(INFO) << "Configuration of this group is " << conf; +} +void Rsm::on_stop_following(const ::braft::LeaderChangeContext& ctx) { + LOG(INFO) << "Node stops following " << ctx; +} +void Rsm::on_start_following(const ::braft::LeaderChangeContext& ctx) { + LOG(INFO) << "Node start following " << ctx; +} + +} // namespace pain::common diff --git a/src/common/rsm/rsm.h b/src/common/rsm/rsm.h new file mode 100644 index 0000000..5ae0c25 --- /dev/null +++ b/src/common/rsm/rsm.h @@ -0,0 +1,77 @@ +#pragma once + +#include // braft::Node braft::StateMachine +#include // braft::SnapshotWriter +#include +#include "common/rsm/container.h" + +namespace pain::common { + +class Rsm; +using RsmPtr = boost::intrusive_ptr; +class Rsm : public braft::StateMachine { +public: + Rsm(const butil::EndPoint& address, + const std::string& group, + const braft::NodeOptions& node_options, + ContainerPtr container); + ~Rsm(); + + int start(); + + bool is_leader() const; + + void shutdown(); + + void join(); + + void apply(const braft::Task& task); + void on_apply(braft::Iterator& iter) override; + + struct SnapshotArg { + braft::SnapshotWriter* writer; + braft::Closure* done; + ContainerPtr container; + }; + + static void* save_snapshot(void* arg); + + void on_snapshot_save(braft::SnapshotWriter* writer, braft::Closure* done) override; + + int on_snapshot_load(braft::SnapshotReader* reader) override; + + void on_leader_start(int64_t term) override; + void on_leader_stop(const butil::Status& status) override; + + void on_shutdown() override; + void on_error(const ::braft::Error& e) override; + void on_configuration_committed(const ::braft::Configuration& conf) override; + void on_stop_following(const ::braft::LeaderChangeContext& ctx) override; + void on_start_following(const ::braft::LeaderChangeContext& ctx) override; + + ContainerPtr container() { + return _container; + } + +private: + butil::EndPoint _address; + std::string _group; + braft::NodeOptions _node_options; + braft::Node* volatile _node; + butil::atomic _leader_term; + std::atomic _use_count = {0}; + + friend void intrusive_ptr_add_ref(Rsm* rsm) { + ++rsm->_use_count; + } + + friend void intrusive_ptr_release(Rsm* rsm) { + if (rsm->_use_count.fetch_sub(1) == 1) { + delete rsm; + } + } + + ContainerPtr _container; +}; + +} // namespace pain::common diff --git a/src/deva/bridge.h b/src/deva/bridge.h index 20f2fb4..f9e5c7f 100644 --- a/src/deva/bridge.h +++ b/src/deva/bridge.h @@ -1,39 +1,23 @@ #pragma once -#include -#include -#include -#include -#include "deva/container_op.h" -#include "deva/op.h" -#include "deva/rsm.h" +#include "common/rsm/bridge.h" +#include "deva/deva_op_factory.h" namespace pain::deva { template void bridge(int32_t version, - RsmPtr rsm, + common::RsmPtr rsm, const Request& request, Response* response, std::move_only_function cb) { - // TODO: get rsm by partition id - auto op = new ContainerOp( - version, OpType, rsm, request, response, [cb = std::move(cb)](Status status) mutable { - cb(std::move(status)); - }); - op->apply(); + common::bridge(static_cast(OpType), version, rsm, request, response, std::move(cb)); } // Future style template -Future bridge(int32_t version, RsmPtr rsm, const Request& request, Response* response) { - Promise promise; - auto future = promise.get_future(); - bridge( - version, rsm, request, response, [promise = std::move(promise)](Status status) mutable { - promise.set_value(std::move(status)); - }); - return future; +Future bridge(int32_t version, common::RsmPtr rsm, const Request& request, Response* response) { + return common::bridge(static_cast(OpType), version, rsm, request, response); } } // namespace pain::deva diff --git a/src/deva/container_op.cc b/src/deva/container_op.cc deleted file mode 100644 index f3f9136..0000000 --- a/src/deva/container_op.cc +++ /dev/null @@ -1,50 +0,0 @@ -#include "deva/container_op.h" -#include -#include -#include "pain/proto/deva_store.pb.h" -#include "deva/deva.h" -#include "deva/op.h" - -namespace pain::deva { - -template -OpPtr create(int32_t version, RsmPtr rsm) { - Request request; - OpPtr op = nullptr; - - if constexpr (OpType < OpType::kMaxDevaOp) { - op = new ContainerOp(version, OpType, rsm, request, nullptr); - } - return op; -} - -#define BRANCH(name) \ - case OpType::k##name: \ - return create(version, \ - rsm); \ - break; - -OpPtr decode(int32_t version, OpType op_type, IOBuf* buf, RsmPtr rsm) { - auto op = [](int32_t version, OpType op_type, RsmPtr rsm) -> OpPtr { - switch (op_type) { - BRANCH(CreateFile) - BRANCH(CreateDir) - BRANCH(ReadDir) - BRANCH(RemoveFile) - BRANCH(SealFile) - BRANCH(CreateChunk) - BRANCH(CheckInChunk) - BRANCH(SealChunk) - BRANCH(SealAndNewChunk) - default: - BOOST_ASSERT_MSG(false, fmt::format("unknown op type: {}", op_type).c_str()); - } - return nullptr; - }(version, op_type, rsm); - op->decode(buf); - return op; -} - -#undef BRANCH - -} // namespace pain::deva diff --git a/src/deva/deva.h b/src/deva/deva.h index 4c901a3..a839da5 100644 --- a/src/deva/deva.h +++ b/src/deva/deva.h @@ -4,13 +4,13 @@ #include #include #include "pain/proto/deva_store.pb.h" +#include "common/rsm/container.h" #include "common/store.h" #include "common/txn_manager.h" #include "common/txn_store.h" -#include "deva/container.h" +#include "deva/deva_op_factory.h" #include "deva/manusya_descriptor.h" #include "deva/namespace.h" -#include "deva/op.h" #define DEVA_ENTRY(name) \ Status name([[maybe_unused]] int32_t version, \ @@ -21,7 +21,7 @@ [[maybe_unused]] const pain::proto::deva::store::name##Request* request, \ [[maybe_unused]] pain::proto::deva::store::name##Response* response, \ [[maybe_unused]] int64_t index) { \ - if (!need_apply(OpType::k##name)) { \ + if (!common::need_apply(static_cast(OpType::k##name))) { \ return name(version, request, response, index); \ } \ if (check_index_is_applied(index)) { \ @@ -46,7 +46,7 @@ namespace pain::deva { class Deva; using DevaPtr = boost::intrusive_ptr; -class Deva : public Container { +class Deva : public common::Container { public: Deva(common::StorePtr store) : _store(store), _namespace(store) {} @@ -65,6 +65,10 @@ class Deva : public Container { Status save_snapshot(std::string_view path, std::vector* files) override; Status load_snapshot(std::string_view path) override; + common::OpFactory* op_factory() override { + static DevaOpFactory s_op_factory; + return &s_op_factory; + } private: Status create(const std::string& path, const ObjectId& id, FileType type); diff --git a/src/deva/deva_op_factory.h b/src/deva/deva_op_factory.h new file mode 100644 index 0000000..e024618 --- /dev/null +++ b/src/deva/deva_op_factory.h @@ -0,0 +1,71 @@ +#pragma once + +#include "pain/proto/deva_store.pb.h" +#include "common/rsm/container_op.h" +#include "common/rsm/op_factory.h" + +#define BRANCH(name) \ + case OpType::k##name: \ + return create_op( \ + version, rsm); \ + break; + +namespace pain::deva { +class Deva; + +using OpPtr = common::OpPtr; +using RsmPtr = common::RsmPtr; + +template +using ContainerOp = common::ContainerOp; + +enum class OpType : uint32_t { + kInvalid = 0, + // DevaOp: 1 ~ 100 + DEFINE_RSM_OP(1, CreateFile, true), + DEFINE_RSM_OP(2, CreateDir, true), + DEFINE_RSM_OP(3, RemoveFile, true), + DEFINE_RSM_OP(4, SealFile, true), + DEFINE_RSM_OP(5, CreateChunk, true), + DEFINE_RSM_OP(6, CheckInChunk, true), + DEFINE_RSM_OP(7, SealChunk, true), + DEFINE_RSM_OP(8, SealAndNewChunk, true), + DEFINE_RSM_OP(9, ReadDir, false), + DEFINE_RSM_OP(10, GetFileInfo, false), + DEFINE_RSM_OP(20, ManusyaHeartbeat, false), + DEFINE_RSM_OP(21, ListManusya, false), + DEFINE_RSM_OP(100, MaxDevaOp, true), +}; + +template +OpPtr create_op(int32_t version, RsmPtr rsm) { + Request request; + OpPtr op = nullptr; + + if constexpr (OpType < OpType::kMaxDevaOp) { + op = new ContainerOp(version, static_cast(OpType), rsm, request, nullptr); + } + return op; +} + +class DevaOpFactory : public common::OpFactory { +public: + OpPtr create(uint32_t op_type, int32_t version, RsmPtr rsm) override { + switch (static_cast(op_type)) { + BRANCH(CreateFile) + BRANCH(CreateDir) + BRANCH(ReadDir) + BRANCH(RemoveFile) + BRANCH(SealFile) + BRANCH(CreateChunk) + BRANCH(CheckInChunk) + BRANCH(SealChunk) + BRANCH(SealAndNewChunk) + default: + BOOST_ASSERT_MSG(false, fmt::format("unknown op type: {}", op_type).c_str()); + } + return nullptr; + } +}; + +} // namespace pain::deva diff --git a/src/deva/deva_service_impl.cc b/src/deva/deva_service_impl.cc index 5a28c1e..b2707b7 100644 --- a/src/deva/deva_service_impl.cc +++ b/src/deva/deva_service_impl.cc @@ -16,7 +16,7 @@ namespace pain::deva { -DevaServiceImpl::DevaServiceImpl(RsmPtr rsm) : _rsm(rsm) {} +DevaServiceImpl::DevaServiceImpl(common::RsmPtr rsm) : _rsm(rsm) {} DEVA_SERVICE_METHOD(OpenFile) { brpc::ClosureGuard done_guard(done); diff --git a/src/deva/deva_service_impl.h b/src/deva/deva_service_impl.h index 925e470..3f7ae73 100644 --- a/src/deva/deva_service_impl.h +++ b/src/deva/deva_service_impl.h @@ -1,7 +1,7 @@ #pragma once #include "pain/proto/deva.pb.h" -#include "deva/rsm.h" +#include "common/rsm/rsm.h" #include @@ -15,7 +15,7 @@ namespace pain::deva { class DevaServiceImpl : public pain::proto::deva::DevaService { public: - DevaServiceImpl(RsmPtr rsm); + DevaServiceImpl(common::RsmPtr rsm); ~DevaServiceImpl() override = default; DEVA_SERVICE_METHOD(OpenFile); DEVA_SERVICE_METHOD(CloseFile); @@ -31,7 +31,7 @@ class DevaServiceImpl : public pain::proto::deva::DevaService { DEVA_SERVICE_METHOD(ListManusya); private: - RsmPtr _rsm; + common::RsmPtr _rsm; std::atomic _partition_id = 0; }; diff --git a/src/deva/mock/deva_machine.h b/src/deva/mock/deva_machine.h index 387cec3..d4aa9b4 100644 --- a/src/deva/mock/deva_machine.h +++ b/src/deva/mock/deva_machine.h @@ -2,9 +2,9 @@ #include #include #include "common/rocksdb_store.h" +#include "common/rsm/rsm.h" #include "deva/deva.h" #include "deva/deva_service_impl.h" -#include "deva/rsm.h" namespace pain::deva::mock { @@ -36,7 +36,7 @@ class DevaMachine { common::RocksdbStorePtr store; auto status = common::RocksdbStore::open(rocksdb_path.c_str(), &store); BOOST_ASSERT_MSG(status.ok(), "Fail to open rocksdb store"); - _rsm = new Rsm(addr, group, node_options, new Deva(store)); + _rsm = new common::Rsm(addr, group, node_options, new Deva(store)); } Status start() { @@ -71,7 +71,7 @@ class DevaMachine { std::string _address; std::string _node_conf; brpc::Server _server; - RsmPtr _rsm; + common::RsmPtr _rsm; }; } // namespace pain::deva::mock diff --git a/src/deva/op.h b/src/deva/op.h deleted file mode 100644 index 0eb652d..0000000 --- a/src/deva/op.h +++ /dev/null @@ -1,87 +0,0 @@ -#pragma once -#include -#include -#include -#include -#include -#include - -#define DEFINE_DEVA_OP(code, name, need_apply) k##name = ((code << 1) | (need_apply ? 1 : 0)) - -namespace pain::deva { - -enum class OpType : uint32_t { - kInvalid = 0, - // DevaOp: 1 ~ 100 - DEFINE_DEVA_OP(1, CreateFile, true), - DEFINE_DEVA_OP(2, CreateDir, true), - DEFINE_DEVA_OP(3, RemoveFile, true), - DEFINE_DEVA_OP(4, SealFile, true), - DEFINE_DEVA_OP(5, CreateChunk, true), - DEFINE_DEVA_OP(6, CheckInChunk, true), - DEFINE_DEVA_OP(7, SealChunk, true), - DEFINE_DEVA_OP(8, SealAndNewChunk, true), - DEFINE_DEVA_OP(9, ReadDir, false), - DEFINE_DEVA_OP(10, GetFileInfo, false), - DEFINE_DEVA_OP(20, ManusyaHeartbeat, false), - DEFINE_DEVA_OP(21, ListManusya, false), - DEFINE_DEVA_OP(100, MaxDevaOp, true), -}; - -inline constexpr bool need_apply(OpType type) { - return (static_cast(type) & 1) == 1; -} - -struct OpMeta { - int32_t version; // op version - OpType type; - uint64_t timestamp; - uint32_t size; - char reserved[40]; // NOLINT(readability-magic-numbers) -}; - -static_assert(sizeof(OpMeta) == 64, "OpMeta size must be 64byte"); // NOLINT(readability-magic-numbers) - -class Op; -using OpPtr = boost::intrusive_ptr; -class Op { -public: - Op() = default; - virtual ~Op() = default; - virtual OpType type() const = 0; - virtual void apply() = 0; - virtual void on_apply(int64_t index) = 0; - virtual void on_finish(Status status) = 0; - virtual void encode(IOBuf* buf) = 0; - virtual void decode(IOBuf* buf) = 0; - -private: - std::atomic _use_count = 0; - friend void intrusive_ptr_add_ref(Op* op) { - ++op->_use_count; - } - friend void intrusive_ptr_release(Op* op) { - if (op->_use_count.fetch_sub(1) == 1) { - delete op; - } - } -}; - -class Rsm; -using RsmPtr = boost::intrusive_ptr; - -void encode(int32_t version, OpPtr op, IOBuf* buf); -OpPtr decode(IOBuf* buf, std::move_only_function decode); - -} // namespace pain::deva - -template <> -struct fmt::formatter : public fmt::formatter { - template - auto format(pain::deva::OpType type, FormatContext& ctx) const { - std::string_view name = magic_enum::enum_name(type); - return fmt::formatter::format(name, ctx); - } -}; - -#undef DEFINE_DEVA_OP diff --git a/src/deva/rsm.cc b/src/deva/rsm.cc index b5cd4a6..3e9c775 100644 --- a/src/deva/rsm.cc +++ b/src/deva/rsm.cc @@ -11,8 +11,6 @@ #include #include // O_CREAT #include "common/rocksdb_store.h" -#include "deva/container.h" -#include "deva/container_op.h" #include "deva/deva.h" DEFINE_bool(rsm_check_term, true, "Check if the leader changed to another term"); @@ -26,163 +24,7 @@ DEFINE_string(rsm_listen_address, "127.0.0.1:8001", "Listen address of deva"); namespace pain::deva { -Rsm::Rsm(const butil::EndPoint& address, - const std::string& group, - const braft::NodeOptions& node_options, - ContainerPtr container) : - _address(address), - _group(group), - _node_options(node_options), - _node(nullptr), - _leader_term(-1), - _container(container) { - _node_options.fsm = this; -} -Rsm::~Rsm() { - PLOG_INFO(("desc", "destructor rsm") // - ("group", _group) // - ("address", butil::endpoint2str(_address).c_str())); - delete _node; -} - -int Rsm::start() { - PLOG_INFO(("desc", "start rsm") // - ("group", _group) // - ("address", butil::endpoint2str(_address).c_str())); - braft::Node* node = new braft::Node(_group, braft::PeerId(_address)); - if (node->init(_node_options) != 0) { - LOG(ERROR) << "Fail to init raft node"; - delete node; - return -1; - } - _node = node; - return 0; -} - -bool Rsm::is_leader() const { - if (_node == nullptr) { - return false; - } - return _node->is_leader(); -} - -void Rsm::shutdown() { - if (_node != nullptr) { - _node->shutdown(nullptr); - } -} - -void Rsm::join() { - if (_node != nullptr) { - _node->join(); - } -} - -void Rsm::apply(const braft::Task& task) { - if (_node != nullptr) { - _node->apply(task); - } -} - -void Rsm::on_apply(braft::Iterator& iter) { - for (; iter.valid(); iter.next()) { - braft::AsyncClosureGuard closure_guard(iter.done()); - butil::IOBuf data; - off_t offset = 0; - if (iter.done() != nullptr) { - // Run at closure_guard destructed - auto c = static_cast(iter.done()); - c->set_index(iter.index()); - } else { - butil::IOBuf saved_log = iter.data(); - // clang-format off - auto op = decode(&saved_log, [rsm = RsmPtr(this)](int32_t version, OpType op_type, IOBuf* buf) { - return decode(version, op_type, buf, rsm); - }); - // clang-format on - op->on_apply(iter.index()); - } - - LOG(INFO) << "Write " << data.size() << " bytes" - << " from offset=" << offset << " at log_index=" << iter.index(); - } -} - -struct SnapshotArg { - braft::SnapshotWriter* writer; - braft::Closure* done; -}; - -void* Rsm::save_snapshot(void* arg) { - PLOG_INFO(("desc", "save_snapshot")); - SnapshotArg* sa = (SnapshotArg*)arg; - std::unique_ptr arg_guard(sa); - brpc::ClosureGuard done_guard(sa->done); - std::string snapshot_path = sa->writer->get_path(); - std::vector files; - auto status = sa->container->save_snapshot(snapshot_path, &files); - if (!status.ok()) { - sa->done->status() = status; - LOG(ERROR) << "Fail to save snapshot to " << snapshot_path; - return nullptr; - } - for (const auto& file : files) { - PLOG_INFO(("desc", "add_file to snapshot")("file", file)); - sa->writer->add_file(file); - } - PLOG_INFO(("desc", "save_snapshot done")); - return nullptr; -} - -void Rsm::on_snapshot_save(braft::SnapshotWriter* writer, braft::Closure* done) { - PLOG_INFO(("desc", "on_snapshot_save")); - SnapshotArg* arg = new SnapshotArg; - arg->writer = writer; - arg->done = done; - arg->container = _container; - bthread_t tid = 0; - bthread_start_urgent(&tid, nullptr, save_snapshot, arg); -} - -// NOLINTNEXTLINE -int Rsm::on_snapshot_load(braft::SnapshotReader* reader) { - PLOG_INFO(("desc", "on_snapshot_load")); - CHECK(!is_leader()) << "Leader is not supposed to load snapshot"; - auto path = reader->get_path(); - auto status = _container->load_snapshot(path); - if (!status.ok()) { - PLOG_ERROR(("desc", "fail to load snapshot from ")("path", path)("error", status.error_str())); - return -1; - } - return 0; -} - -void Rsm::on_leader_start(int64_t term) { - _leader_term.store(term, butil::memory_order_release); - LOG(INFO) << "Node becomes leader"; -} -void Rsm::on_leader_stop(const butil::Status& status) { - _leader_term.store(-1, butil::memory_order_release); - LOG(INFO) << "Node stepped down : " << status; -} - -void Rsm::on_shutdown() { - LOG(INFO) << "This node is down"; -} -void Rsm::on_error(const ::braft::Error& e) { - LOG(ERROR) << "Met raft error " << e; -} -void Rsm::on_configuration_committed(const ::braft::Configuration& conf) { - LOG(INFO) << "Configuration of this group is " << conf; -} -void Rsm::on_stop_following(const ::braft::LeaderChangeContext& ctx) { - LOG(INFO) << "Node stops following " << ctx; -} -void Rsm::on_start_following(const ::braft::LeaderChangeContext& ctx) { - LOG(INFO) << "Node start following " << ctx; -} - -RsmPtr default_rsm() { +common::RsmPtr default_rsm() { std::string data_path = FLAGS_rsm_data_path + "/data"; butil::EndPoint addr; std::string group = "default"; @@ -209,7 +51,7 @@ RsmPtr default_rsm() { common::RocksdbStorePtr store; auto status = common::RocksdbStore::open(rocksdb_path.c_str(), &store); BOOST_ASSERT_MSG(status.ok(), "Fail to open rocksdb store"); - static RsmPtr s_rsm = new Rsm(addr, group, node_options, new Deva(store)); + static common::RsmPtr s_rsm = new common::Rsm(addr, group, node_options, new Deva(store)); return s_rsm; } diff --git a/src/deva/rsm.h b/src/deva/rsm.h index 9879978..277ca4f 100644 --- a/src/deva/rsm.h +++ b/src/deva/rsm.h @@ -3,77 +3,10 @@ #include // braft::Node braft::StateMachine #include // braft::SnapshotWriter #include -#include "deva/container.h" +#include "common/rsm/rsm.h" namespace pain::deva { -class Rsm; -using RsmPtr = boost::intrusive_ptr; -class Rsm : public braft::StateMachine { -public: - Rsm(const butil::EndPoint& address, - const std::string& group, - const braft::NodeOptions& node_options, - ContainerPtr container); - ~Rsm(); - - int start(); - - bool is_leader() const; - - void shutdown(); - - void join(); - - void apply(const braft::Task& task); - void on_apply(braft::Iterator& iter) override; - - struct SnapshotArg { - braft::SnapshotWriter* writer; - braft::Closure* done; - ContainerPtr container; - }; - - static void* save_snapshot(void* arg); - - void on_snapshot_save(braft::SnapshotWriter* writer, braft::Closure* done) override; - - int on_snapshot_load(braft::SnapshotReader* reader) override; - - void on_leader_start(int64_t term) override; - void on_leader_stop(const butil::Status& status) override; - - void on_shutdown() override; - void on_error(const ::braft::Error& e) override; - void on_configuration_committed(const ::braft::Configuration& conf) override; - void on_stop_following(const ::braft::LeaderChangeContext& ctx) override; - void on_start_following(const ::braft::LeaderChangeContext& ctx) override; - - ContainerPtr container() { - return _container; - } - -private: - butil::EndPoint _address; - std::string _group; - braft::NodeOptions _node_options; - braft::Node* volatile _node; - butil::atomic _leader_term; - std::atomic _use_count = {0}; - - friend void intrusive_ptr_add_ref(Rsm* rsm) { - ++rsm->_use_count; - } - - friend void intrusive_ptr_release(Rsm* rsm) { - if (rsm->_use_count.fetch_sub(1) == 1) { - delete rsm; - } - } - - ContainerPtr _container; -}; - -RsmPtr default_rsm(); +common::RsmPtr default_rsm(); } // namespace pain::deva