From 9982a77afd72d13560dbd322bd6ca2dc2353cadb Mon Sep 17 00:00:00 2001 From: SpyCheese Date: Wed, 17 Jul 2024 15:06:07 +0300 Subject: [PATCH] Improve state serializer Use previous persistent state to speed up reading --- crypto/vm/cells/DataCell.cpp | 3 + tdutils/td/utils/filesystem.cpp | 23 +++-- validator/db/archive-manager.cpp | 106 ++++++++++++++------ validator/db/archive-manager.hpp | 10 +- validator/db/rootdb.cpp | 6 ++ validator/db/rootdb.hpp | 2 + validator/full-node-shard.cpp | 10 +- validator/full-node-shard.hpp | 4 - validator/interfaces/db.h | 2 + validator/manager-disk.hpp | 4 + validator/manager-hardfork.hpp | 4 + validator/manager.cpp | 9 ++ validator/manager.hpp | 2 + validator/state-serializer.cpp | 166 ++++++++++++++++++++++++++----- validator/state-serializer.hpp | 13 +++ validator/validator.h | 2 + 16 files changed, 286 insertions(+), 80 deletions(-) diff --git a/crypto/vm/cells/DataCell.cpp b/crypto/vm/cells/DataCell.cpp index cccb11dce..8aac575bf 100644 --- a/crypto/vm/cells/DataCell.cpp +++ b/crypto/vm/cells/DataCell.cpp @@ -113,6 +113,9 @@ td::Result> DataCell::create(td::ConstBitPtr data, unsigned bits, if (bits != 8 + hash_bytes * 8) { return td::Status::Error("Not enouch data for a Library special cell"); } + if (!refs.empty()) { + return td::Status::Error("Library special cell has a cell reference"); + } break; } diff --git a/tdutils/td/utils/filesystem.cpp b/tdutils/td/utils/filesystem.cpp index 562a42816..b84b6b3f8 100644 --- a/tdutils/td/utils/filesystem.cpp +++ b/tdutils/td/utils/filesystem.cpp @@ -68,9 +68,14 @@ Result read_file_impl(CSlice path, int64 size, int64 offset) { return Status::Error("Failed to read file: invalid size"); } auto content = create_empty(narrow_cast(size)); - TRY_RESULT(got_size, from_file.pread(as_mutable_slice(content), offset)); - if (got_size != static_cast(size)) { - return Status::Error("Failed to read file"); + MutableSlice slice = as_mutable_slice(content); + while (!slice.empty()) { + TRY_RESULT(got_size, from_file.pread(slice, offset)); + if (got_size == 0) { + return Status::Error("Failed to read file"); + } + offset += got_size; + slice.remove_prefix(got_size); } from_file.close(); return std::move(content); @@ -103,9 +108,15 @@ Status write_file(CSlice to, Slice data, WriteFileOptions options) { TRY_STATUS(to_file.lock(FileFd::LockFlags::Write, to.str(), 10)); TRY_STATUS(to_file.truncate_to_current_position(0)); } - TRY_RESULT(written, to_file.write(data)); - if (written != size) { - return Status::Error(PSLICE() << "Failed to write file: written " << written << " bytes instead of " << size); + size_t total_written = 0; + while (!data.empty()) { + TRY_RESULT(written, to_file.write(data)); + if (written == 0) { + return Status::Error(PSLICE() << "Failed to write file: written " << total_written << " bytes instead of " + << size); + } + total_written += written; + data.remove_prefix(written); } if (options.need_sync) { TRY_STATUS(to_file.sync()); diff --git a/validator/db/archive-manager.cpp b/validator/db/archive-manager.cpp index b87d04f78..14d3ec469 100644 --- a/validator/db/archive-manager.cpp +++ b/validator/db/archive-manager.cpp @@ -310,14 +310,17 @@ void ArchiveManager::get_file(ConstBlockHandle handle, FileReference ref_id, td: get_file_short_cont(std::move(ref_id), get_max_temp_file_desc_idx(), std::move(promise)); } -void ArchiveManager::written_perm_state(FileReferenceShort id) { - perm_states_.emplace(id.hash(), id); +void ArchiveManager::register_perm_state(FileReferenceShort id) { + BlockSeqno masterchain_seqno = 0; + id.ref().visit(td::overloaded( + [&](const fileref::PersistentStateShort &x) { masterchain_seqno = x.masterchain_seqno; }, [&](const auto &) {})); + perm_states_[{masterchain_seqno, id.hash()}] = id; } void ArchiveManager::add_zero_state(BlockIdExt block_id, td::BufferSlice data, td::Promise promise) { auto id = FileReference{fileref::ZeroState{block_id}}; auto hash = id.hash(); - if (perm_states_.find(hash) != perm_states_.end()) { + if (perm_states_.find({0, hash}) != perm_states_.end()) { promise.set_value(td::Unit()); return; } @@ -328,7 +331,7 @@ void ArchiveManager::add_zero_state(BlockIdExt block_id, td::BufferSlice data, t if (R.is_error()) { promise.set_error(R.move_as_error()); } else { - td::actor::send_closure(SelfId, &ArchiveManager::written_perm_state, id); + td::actor::send_closure(SelfId, &ArchiveManager::register_perm_state, id); promise.set_value(td::Unit()); } }); @@ -357,12 +360,13 @@ void ArchiveManager::add_persistent_state_gen(BlockIdExt block_id, BlockIdExt ma add_persistent_state_impl(block_id, masterchain_block_id, std::move(promise), std::move(create_writer)); } -void ArchiveManager::add_persistent_state_impl(BlockIdExt block_id, BlockIdExt masterchain_block_id, - td::Promise promise, - std::function)> create_writer) { +void ArchiveManager::add_persistent_state_impl( + BlockIdExt block_id, BlockIdExt masterchain_block_id, td::Promise promise, + std::function)> create_writer) { auto id = FileReference{fileref::PersistentState{block_id, masterchain_block_id}}; + BlockSeqno masterchain_seqno = masterchain_block_id.seqno(); auto hash = id.hash(); - if (perm_states_.find(hash) != perm_states_.end()) { + if (perm_states_.find({masterchain_seqno, hash}) != perm_states_.end()) { promise.set_value(td::Unit()); return; } @@ -373,7 +377,7 @@ void ArchiveManager::add_persistent_state_impl(BlockIdExt block_id, BlockIdExt m if (R.is_error()) { promise.set_error(R.move_as_error()); } else { - td::actor::send_closure(SelfId, &ArchiveManager::written_perm_state, id); + td::actor::send_closure(SelfId, &ArchiveManager::register_perm_state, id); promise.set_value(td::Unit()); } }); @@ -383,7 +387,7 @@ void ArchiveManager::add_persistent_state_impl(BlockIdExt block_id, BlockIdExt m void ArchiveManager::get_zero_state(BlockIdExt block_id, td::Promise promise) { auto id = FileReference{fileref::ZeroState{block_id}}; auto hash = id.hash(); - if (perm_states_.find(hash) == perm_states_.end()) { + if (perm_states_.find({0, hash}) == perm_states_.end()) { promise.set_error(td::Status::Error(ErrorCode::notready, "zerostate not in db")); return; } @@ -395,18 +399,38 @@ void ArchiveManager::get_zero_state(BlockIdExt block_id, td::Promise promise) { auto id = FileReference{fileref::ZeroState{block_id}}; auto hash = id.hash(); - if (perm_states_.find(hash) == perm_states_.end()) { + if (perm_states_.find({0, hash}) == perm_states_.end()) { promise.set_result(false); return; } promise.set_result(true); } +void ArchiveManager::get_previous_persistent_state_files( + BlockSeqno cur_mc_seqno, td::Promise>> promise) { + auto it = perm_states_.lower_bound({cur_mc_seqno, FileHash::zero()}); + if (it == perm_states_.begin()) { + promise.set_value({}); + return; + } + --it; + BlockSeqno mc_seqno = it->first.first; + std::vector> files; + while (it->first.first == mc_seqno) { + files.emplace_back(db_root_ + "/archive/states/" + it->second.filename_short(), it->second.shard()); + if (it == perm_states_.begin()) { + break; + } + --it; + } + promise.set_value(std::move(files)); +} + void ArchiveManager::get_persistent_state(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::Promise promise) { auto id = FileReference{fileref::PersistentState{block_id, masterchain_block_id}}; auto hash = id.hash(); - if (perm_states_.find(hash) == perm_states_.end()) { + if (perm_states_.find({masterchain_block_id.seqno(), hash}) == perm_states_.end()) { promise.set_error(td::Status::Error(ErrorCode::notready, "state file not in db")); return; } @@ -419,7 +443,7 @@ void ArchiveManager::get_persistent_state_slice(BlockIdExt block_id, BlockIdExt td::int64 max_size, td::Promise promise) { auto id = FileReference{fileref::PersistentState{block_id, masterchain_block_id}}; auto hash = id.hash(); - if (perm_states_.find(hash) == perm_states_.end()) { + if (perm_states_.find({masterchain_block_id.seqno(), hash}) == perm_states_.end()) { promise.set_error(td::Status::Error(ErrorCode::notready, "state file not in db")); return; } @@ -432,7 +456,7 @@ void ArchiveManager::check_persistent_state(BlockIdExt block_id, BlockIdExt mast td::Promise promise) { auto id = FileReference{fileref::PersistentState{block_id, masterchain_block_id}}; auto hash = id.hash(); - if (perm_states_.find(hash) == perm_states_.end()) { + if (perm_states_.find({masterchain_block_id.seqno(), hash}) == perm_states_.end()) { promise.set_result(false); return; } @@ -884,13 +908,11 @@ void ArchiveManager::start_up() { R = FileReferenceShort::create(newfname); R.ensure(); } - auto f = R.move_as_ok(); - auto hash = f.hash(); - perm_states_[hash] = std::move(f); + register_perm_state(R.move_as_ok()); } }).ensure(); - persistent_state_gc(FileHash::zero()); + persistent_state_gc({0, FileHash::zero()}); double open_since = td::Clocks::system() - opts_->get_archive_preload_period(); for (auto it = files_.rbegin(); it != files_.rend(); ++it) { @@ -976,11 +998,12 @@ void ArchiveManager::run_gc(UnixTime mc_ts, UnixTime gc_ts, UnixTime archive_ttl } } -void ArchiveManager::persistent_state_gc(FileHash last) { - if (perm_states_.size() == 0) { +void ArchiveManager::persistent_state_gc(std::pair last) { + if (perm_states_.empty()) { delay_action( - [hash = FileHash::zero(), SelfId = actor_id(this)]() { - td::actor::send_closure(SelfId, &ArchiveManager::persistent_state_gc, hash); + [SelfId = actor_id(this)]() { + td::actor::send_closure(SelfId, &ArchiveManager::persistent_state_gc, + std::pair{0, FileHash::zero()}); }, td::Timestamp::in(1.0)); return; @@ -993,12 +1016,12 @@ void ArchiveManager::persistent_state_gc(FileHash last) { it = perm_states_.begin(); } + auto key = it->first; auto &F = it->second; - auto hash = F.hash(); int res = 0; BlockSeqno seqno = 0; - F.ref().visit(td::overloaded([&](const fileref::ZeroStateShort &x) { res = 1; }, + F.ref().visit(td::overloaded([&](const fileref::ZeroStateShort &) { res = 1; }, [&](const fileref::PersistentStateShort &x) { res = 0; seqno = x.masterchain_seqno; @@ -1010,24 +1033,41 @@ void ArchiveManager::persistent_state_gc(FileHash last) { perm_states_.erase(it); } if (res != 0) { - delay_action([hash, SelfId = actor_id( - this)]() { td::actor::send_closure(SelfId, &ArchiveManager::persistent_state_gc, hash); }, + delay_action([key, SelfId = actor_id( + this)]() { td::actor::send_closure(SelfId, &ArchiveManager::persistent_state_gc, key); }, + td::Timestamp::in(1.0)); + return; + } + CHECK(seqno == key.first); + + // Do not delete the most recent fully serialized state + bool allow_delete = false; + auto it2 = perm_states_.lower_bound({seqno + 1, FileHash::zero()}); + if (it2 != perm_states_.end()) { + it2 = perm_states_.lower_bound({it2->first.first + 1, FileHash::zero()}); + if (it2 != perm_states_.end()) { + allow_delete = true; + } + } + if (!allow_delete) { + delay_action([key, SelfId = actor_id( + this)]() { td::actor::send_closure(SelfId, &ArchiveManager::persistent_state_gc, key); }, td::Timestamp::in(1.0)); return; } - auto P = td::PromiseCreator::lambda([SelfId = actor_id(this), hash](td::Result R) { + auto P = td::PromiseCreator::lambda([SelfId = actor_id(this), key](td::Result R) { if (R.is_error()) { - td::actor::send_closure(SelfId, &ArchiveManager::got_gc_masterchain_handle, nullptr, hash); + td::actor::send_closure(SelfId, &ArchiveManager::got_gc_masterchain_handle, nullptr, key); } else { - td::actor::send_closure(SelfId, &ArchiveManager::got_gc_masterchain_handle, R.move_as_ok(), hash); + td::actor::send_closure(SelfId, &ArchiveManager::got_gc_masterchain_handle, R.move_as_ok(), key); } }); get_block_by_seqno(AccountIdPrefixFull{masterchainId, 0}, seqno, std::move(P)); } -void ArchiveManager::got_gc_masterchain_handle(ConstBlockHandle handle, FileHash hash) { +void ArchiveManager::got_gc_masterchain_handle(ConstBlockHandle handle, std::pair key) { bool to_del = false; if (!handle || !handle->inited_unix_time() || !handle->unix_time()) { to_del = true; @@ -1035,15 +1075,15 @@ void ArchiveManager::got_gc_masterchain_handle(ConstBlockHandle handle, FileHash auto ttl = ValidatorManager::persistent_state_ttl(handle->unix_time()); to_del = ttl < td::Clocks::system(); } - auto it = perm_states_.find(hash); + auto it = perm_states_.find(key); CHECK(it != perm_states_.end()); auto &F = it->second; if (to_del) { td::unlink(db_root_ + "/archive/states/" + F.filename_short()).ignore(); perm_states_.erase(it); } - delay_action([hash, SelfId = actor_id( - this)]() { td::actor::send_closure(SelfId, &ArchiveManager::persistent_state_gc, hash); }, + delay_action([key, SelfId = actor_id( + this)]() { td::actor::send_closure(SelfId, &ArchiveManager::persistent_state_gc, key); }, td::Timestamp::in(1.0)); } diff --git a/validator/db/archive-manager.hpp b/validator/db/archive-manager.hpp index a1ed97022..622969ec5 100644 --- a/validator/db/archive-manager.hpp +++ b/validator/db/archive-manager.hpp @@ -54,6 +54,8 @@ class ArchiveManager : public td::actor::Actor { td::int64 max_size, td::Promise promise); void check_persistent_state(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::Promise promise); void check_zero_state(BlockIdExt block_id, td::Promise promise); + void get_previous_persistent_state_files(BlockSeqno cur_mc_seqno, + td::Promise>> promise); void truncate(BlockSeqno masterchain_seqno, ConstBlockHandle handle, td::Promise promise); //void truncate_continue(BlockSeqno masterchain_seqno, td::Promise promise); @@ -180,7 +182,7 @@ class ArchiveManager : public td::actor::Actor { return p.key ? key_files_ : p.temp ? temp_files_ : files_; } - std::map perm_states_; + std::map, FileReferenceShort> perm_states_; // Mc block seqno, hash -> state void load_package(PackageId seqno); void delete_package(PackageId seqno, td::Promise promise); @@ -207,10 +209,10 @@ class ArchiveManager : public td::actor::Actor { void add_persistent_state_impl(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::Promise promise, std::function)> create_writer); - void written_perm_state(FileReferenceShort id); + void register_perm_state(FileReferenceShort id); - void persistent_state_gc(FileHash last); - void got_gc_masterchain_handle(ConstBlockHandle handle, FileHash hash); + void persistent_state_gc(std::pair last); + void got_gc_masterchain_handle(ConstBlockHandle handle, std::pair key); std::string db_root_; td::Ref opts_; diff --git a/validator/db/rootdb.cpp b/validator/db/rootdb.cpp index ff9abae68..93dcfc91f 100644 --- a/validator/db/rootdb.cpp +++ b/validator/db/rootdb.cpp @@ -317,6 +317,12 @@ void RootDb::check_zero_state_file_exists(BlockIdExt block_id, td::Promise td::actor::send_closure(archive_db_, &ArchiveManager::check_zero_state, block_id, std::move(promise)); } +void RootDb::get_previous_persistent_state_files( + BlockSeqno cur_mc_seqno, td::Promise>> promise) { + td::actor::send_closure(archive_db_, &ArchiveManager::get_previous_persistent_state_files, cur_mc_seqno, + std::move(promise)); +} + void RootDb::store_block_handle(BlockHandle handle, td::Promise promise) { td::actor::send_closure(archive_db_, &ArchiveManager::update_handle, std::move(handle), std::move(promise)); } diff --git a/validator/db/rootdb.hpp b/validator/db/rootdb.hpp index 97b9550b8..45044e4f8 100644 --- a/validator/db/rootdb.hpp +++ b/validator/db/rootdb.hpp @@ -84,6 +84,8 @@ class RootDb : public Db { void store_zero_state_file(BlockIdExt block_id, td::BufferSlice state, td::Promise promise) override; void get_zero_state_file(BlockIdExt block_id, td::Promise promise) override; void check_zero_state_file_exists(BlockIdExt block_id, td::Promise promise) override; + void get_previous_persistent_state_files( + BlockSeqno cur_mc_seqno, td::Promise>> promise) override; void try_get_static_file(FileHash file_hash, td::Promise promise) override; diff --git a/validator/full-node-shard.cpp b/validator/full-node-shard.cpp index fa0c3e62a..fbdbbfd7a 100644 --- a/validator/full-node-shard.cpp +++ b/validator/full-node-shard.cpp @@ -832,8 +832,7 @@ void FullNodeShardImpl::download_persistent_state(BlockIdExt id, BlockIdExt mast auto &b = choose_neighbour(); td::actor::create_actor(PSTRING() << "downloadstatereq" << id.id.to_str(), id, masterchain_block_id, adnl_id_, overlay_id_, b.adnl_id, priority, timeout, validator_manager_, - b.use_rldp2() ? (td::actor::ActorId)rldp2_ : rldp_, - overlays_, adnl_, client_, std::move(promise)) + rldp2_, overlays_, adnl_, client_, std::move(promise)) .release(); } @@ -867,10 +866,9 @@ void FullNodeShardImpl::get_next_key_blocks(BlockIdExt block_id, td::Timestamp t void FullNodeShardImpl::download_archive(BlockSeqno masterchain_seqno, std::string tmp_dir, td::Timestamp timeout, td::Promise promise) { auto &b = choose_neighbour(); - td::actor::create_actor( - "archive", masterchain_seqno, std::move(tmp_dir), adnl_id_, overlay_id_, b.adnl_id, timeout, validator_manager_, - b.use_rldp2() ? (td::actor::ActorId)rldp2_ : rldp_, overlays_, adnl_, client_, - create_neighbour_promise(b, std::move(promise))) + td::actor::create_actor("archive", masterchain_seqno, std::move(tmp_dir), adnl_id_, overlay_id_, + b.adnl_id, timeout, validator_manager_, rldp2_, overlays_, adnl_, + client_, create_neighbour_promise(b, std::move(promise))) .release(); } diff --git a/validator/full-node-shard.hpp b/validator/full-node-shard.hpp index cec7c6494..a7cf89ac5 100644 --- a/validator/full-node-shard.hpp +++ b/validator/full-node-shard.hpp @@ -46,10 +46,6 @@ struct Neighbour { void query_failed(); void update_roundtrip(double t); - bool use_rldp2() const { - return std::make_pair(proto_version, capabilities) >= std::make_pair(2, 2); - } - static Neighbour zero; }; diff --git a/validator/interfaces/db.h b/validator/interfaces/db.h index 84ea2b366..8bbf7f31f 100644 --- a/validator/interfaces/db.h +++ b/validator/interfaces/db.h @@ -66,6 +66,8 @@ class Db : public td::actor::Actor { virtual void store_zero_state_file(BlockIdExt block_id, td::BufferSlice state, td::Promise promise) = 0; virtual void get_zero_state_file(BlockIdExt block_id, td::Promise promise) = 0; virtual void check_zero_state_file_exists(BlockIdExt block_id, td::Promise promise) = 0; + virtual void get_previous_persistent_state_files( + BlockSeqno cur_mc_seqno, td::Promise>> promise) = 0; virtual void try_get_static_file(FileHash file_hash, td::Promise promise) = 0; diff --git a/validator/manager-disk.hpp b/validator/manager-disk.hpp index 4000acd05..a77be2725 100644 --- a/validator/manager-disk.hpp +++ b/validator/manager-disk.hpp @@ -116,6 +116,10 @@ class ValidatorManagerImpl : public ValidatorManager { td::int64 max_length, td::Promise promise) override { UNREACHABLE(); } + void get_previous_persistent_state_files( + BlockSeqno cur_mc_seqno, td::Promise>> promise) override { + UNREACHABLE(); + } void get_block_proof(BlockHandle handle, td::Promise promise) override; void get_block_proof_link(BlockHandle block_id, td::Promise promise) override { UNREACHABLE(); diff --git a/validator/manager-hardfork.hpp b/validator/manager-hardfork.hpp index 1d1e48837..e7175b77b 100644 --- a/validator/manager-hardfork.hpp +++ b/validator/manager-hardfork.hpp @@ -139,6 +139,10 @@ class ValidatorManagerImpl : public ValidatorManager { td::int64 max_length, td::Promise promise) override { UNREACHABLE(); } + void get_previous_persistent_state_files( + BlockSeqno cur_mc_seqno, td::Promise>> promise) override { + UNREACHABLE(); + } void get_block_proof(BlockHandle handle, td::Promise promise) override; void get_block_proof_link(BlockHandle block_id, td::Promise promise) override; void get_key_block_proof(BlockIdExt block_id, td::Promise promise) override; diff --git a/validator/manager.cpp b/validator/manager.cpp index 3dbca852d..1e94ff4c9 100644 --- a/validator/manager.cpp +++ b/validator/manager.cpp @@ -308,6 +308,11 @@ void ValidatorManagerImpl::get_persistent_state_slice(BlockIdExt block_id, Block std::move(promise)); } +void ValidatorManagerImpl::get_previous_persistent_state_files( + BlockSeqno cur_mc_seqno, td::Promise>> promise) { + td::actor::send_closure(db_, &Db::get_previous_persistent_state_files, cur_mc_seqno, std::move(promise)); +} + void ValidatorManagerImpl::get_block_proof(BlockHandle handle, td::Promise promise) { auto P = td::PromiseCreator::lambda([promise = std::move(promise)](td::Result> R) mutable { if (R.is_error()) { @@ -1714,6 +1719,8 @@ void ValidatorManagerImpl::read_gc_list(std::vector list) { serializer_ = td::actor::create_actor("serializer", last_key_block_handle_->id(), opts_, actor_id(this)); + td::actor::send_closure(serializer_, &AsyncStateSerializer::update_last_known_key_block_ts, + last_key_block_handle_->unix_time()); if (last_masterchain_block_handle_->inited_next_left()) { auto b = last_masterchain_block_handle_->one_next(true); @@ -1903,6 +1910,8 @@ void ValidatorManagerImpl::new_masterchain_block() { last_known_key_block_handle_ = last_key_block_handle_; callback_->new_key_block(last_key_block_handle_); } + td::actor::send_closure(serializer_, &AsyncStateSerializer::update_last_known_key_block_ts, + last_key_block_handle_->unix_time()); } update_shards(); diff --git a/validator/manager.hpp b/validator/manager.hpp index 42a3531aa..12354c634 100644 --- a/validator/manager.hpp +++ b/validator/manager.hpp @@ -371,6 +371,8 @@ class ValidatorManagerImpl : public ValidatorManager { td::Promise promise) override; void get_persistent_state_slice(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::int64 offset, td::int64 max_length, td::Promise promise) override; + void get_previous_persistent_state_files( + BlockSeqno cur_mc_seqno, td::Promise>> promise) override; void get_block_proof(BlockHandle handle, td::Promise promise) override; void get_block_proof_link(BlockHandle block_id, td::Promise promise) override; void get_key_block_proof(BlockIdExt block_id, td::Promise promise) override; diff --git a/validator/state-serializer.cpp b/validator/state-serializer.cpp index 4f10d959e..ece1be989 100644 --- a/validator/state-serializer.cpp +++ b/validator/state-serializer.cpp @@ -21,6 +21,7 @@ #include "adnl/utils.hpp" #include "ton/ton-io.hpp" #include "common/delay.h" +#include "td/utils/filesystem.h" namespace ton { @@ -84,6 +85,20 @@ void AsyncStateSerializer::alarm() { td::actor::send_closure(manager_, &ValidatorManager::get_top_masterchain_block, std::move(P)); } +void AsyncStateSerializer::request_previous_state_files() { + td::actor::send_closure( + manager_, &ValidatorManager::get_previous_persistent_state_files, masterchain_handle_->id().seqno(), + [SelfId = actor_id(this)](td::Result>> R) { + R.ensure(); + td::actor::send_closure(SelfId, &AsyncStateSerializer::got_previous_state_files, R.move_as_ok()); + }); +} + +void AsyncStateSerializer::got_previous_state_files(std::vector> files) { + previous_state_files_ = std::move(files); + request_masterchain_state(); +} + void AsyncStateSerializer::request_masterchain_state() { auto P = td::PromiseCreator::lambda([SelfId = actor_id(this), manager = manager_](td::Result> R) { if (R.is_error()) { @@ -133,37 +148,43 @@ void AsyncStateSerializer::next_iteration() { } CHECK(masterchain_handle_->id() == last_block_id_); if (attempt_ < max_attempt() && last_key_block_id_.id.seqno < last_block_id_.id.seqno && - need_serialize(masterchain_handle_) && opts_->get_state_serializer_enabled()) { - if (!have_masterchain_state_) { - LOG(ERROR) << "started serializing persistent state for " << masterchain_handle_->id().id.to_str(); - // block next attempts immediately, but send actual request later - running_ = true; - double delay = td::Random::fast(0, 3600); - LOG(WARNING) << "serializer delay = " << delay << "s"; - delay_action([SelfId = actor_id( - this)]() { td::actor::send_closure(SelfId, &AsyncStateSerializer::request_masterchain_state); }, - td::Timestamp::in(delay)); - return; - } - while (next_idx_ < shards_.size()) { - if (!need_monitor(shards_[next_idx_].shard_full())) { - next_idx_++; - } else { + need_serialize(masterchain_handle_)) { + if (!have_masterchain_state_ && !opts_->get_state_serializer_enabled()) { + LOG(ERROR) << "skipping serializing persistent state for " << masterchain_handle_->id().id.to_str() + << ": serializer is disabled"; + } else if (!have_masterchain_state_ && have_newer_persistent_state(masterchain_handle_->unix_time())) { + LOG(ERROR) << "skipping serializing persistent state for " << masterchain_handle_->id().id.to_str() + << ": newer key block with ts=" << last_known_key_block_ts_ << " exists"; + } else { + if (!have_masterchain_state_) { + LOG(ERROR) << "started serializing persistent state for " << masterchain_handle_->id().id.to_str(); // block next attempts immediately, but send actual request later running_ = true; - double delay = td::Random::fast(0, 1800); + double delay = td::Random::fast(0, 3600); LOG(WARNING) << "serializer delay = " << delay << "s"; delay_action( - [SelfId = actor_id(this), shard = shards_[next_idx_]]() { - td::actor::send_closure(SelfId, &AsyncStateSerializer::request_shard_state, shard); + [SelfId = actor_id(this)]() { + td::actor::send_closure(SelfId, &AsyncStateSerializer::request_previous_state_files); }, td::Timestamp::in(delay)); return; } + while (next_idx_ < shards_.size()) { + if (!need_monitor(shards_[next_idx_].shard_full())) { + next_idx_++; + } else { + running_ = true; + request_shard_state(shards_[next_idx_]); + return; + } + } + LOG(ERROR) << "finished serializing persistent state for " << masterchain_handle_->id().id.to_str(); } - LOG(ERROR) << "finished serializing persistent state for " << masterchain_handle_->id().id.to_str(); last_key_block_ts_ = masterchain_handle_->unix_time(); last_key_block_id_ = masterchain_handle_->id(); + previous_state_files_ = {}; + previous_state_cache_ = {}; + previous_state_cur_shards_ = {}; } if (!saved_to_db_) { running_ = true; @@ -177,9 +198,6 @@ void AsyncStateSerializer::next_iteration() { return; } if (masterchain_handle_->inited_next_left()) { - if (need_serialize(masterchain_handle_) && !opts_->get_state_serializer_enabled()) { - LOG(ERROR) << "skipping serializing persistent state for " << masterchain_handle_->id().id.to_str(); - } last_block_id_ = masterchain_handle_->one_next(true); have_masterchain_state_ = false; masterchain_handle_ = nullptr; @@ -204,6 +222,88 @@ void AsyncStateSerializer::got_masterchain_handle(BlockHandle handle) { next_iteration(); } +class CachedCellDbReader : public vm::CellDbReader { + public: + CachedCellDbReader(std::shared_ptr parent, + std::shared_ptr>> cache) + : parent_(std::move(parent)), cache_(std::move(cache)) { + } + td::Result> load_cell(td::Slice hash) override { + ++total_reqs_; + DCHECK(hash.size() == 32); + if (cache_) { + auto it = cache_->find(td::Bits256{(const unsigned char*)hash.data()}); + if (it != cache_->end()) { + ++cached_reqs_; + TRY_RESULT(loaded_cell, it->second->load_cell()); + return loaded_cell.data_cell; + } + } + return parent_->load_cell(hash); + } + void print_stats() const { + LOG(WARNING) << "CachedCellDbReader stats : " << total_reqs_ << " reads, " << cached_reqs_ << " cached"; + } + private: + std::shared_ptr parent_; + std::shared_ptr>> cache_; + + td::uint64 total_reqs_ = 0; + td::uint64 cached_reqs_ = 0; +}; + +void AsyncStateSerializer::prepare_previous_state_cache(ShardIdFull shard) { + std::vector prev_shards; + for (const auto& [_, prev_shard] : previous_state_files_) { + if (shard_intersects(shard, prev_shard)) { + prev_shards.push_back(prev_shard); + } + } + if (prev_shards == previous_state_cur_shards_) { + return; + } + previous_state_cur_shards_ = std::move(prev_shards); + previous_state_cache_ = {}; + if (previous_state_cur_shards_.empty()) { + return; + } + td::Timer timer; + LOG(WARNING) << "Preloading previous persistent state for shard " << shard.to_str() << " (" + << previous_state_cur_shards_.size() << " files)"; + std::map> cells; + std::function)> dfs = [&](td::Ref cell) { + td::Bits256 hash = cell->get_hash().bits(); + if (!cells.emplace(hash, cell).second) { + return; + } + bool is_special; + vm::CellSlice cs = vm::load_cell_slice_special(cell, is_special); + for (unsigned i = 0; i < cs.size_refs(); ++i) { + dfs(cs.prefetch_ref(i)); + } + }; + for (const auto& [file, prev_shard] : previous_state_files_) { + if (!shard_intersects(shard, prev_shard)) { + continue; + } + auto r_data = td::read_file(file); + if (r_data.is_error()) { + LOG(INFO) << "Reading " << file << " : " << r_data.move_as_error(); + continue; + } + LOG(INFO) << "Reading " << file << " : " << td::format::as_size(r_data.ok().size()); + auto r_root = vm::std_boc_deserialize(r_data.move_as_ok()); + if (r_root.is_error()) { + LOG(WARNING) << "Deserialize error : " << r_root.move_as_error(); + continue; + } + r_data = {}; + dfs(r_root.move_as_ok()); + } + LOG(WARNING) << "Preloaded previous state: " << cells.size() << " cells in " << timer.elapsed() << "s"; + previous_state_cache_ = std::make_shared>>(std::move(cells)); +} + void AsyncStateSerializer::got_masterchain_state(td::Ref state, std::shared_ptr cell_db_reader) { if (!opts_->get_state_serializer_enabled()) { @@ -211,6 +311,8 @@ void AsyncStateSerializer::got_masterchain_state(td::Ref state return; } LOG(ERROR) << "serializing masterchain state " << masterchain_handle_->id().id.to_str(); + prepare_previous_state_cache(state->get_shard()); + auto new_cell_db_reader = std::make_shared(cell_db_reader, previous_state_cache_); have_masterchain_state_ = true; CHECK(next_idx_ == 0); CHECK(shards_.size() == 0); @@ -220,9 +322,11 @@ void AsyncStateSerializer::got_masterchain_state(td::Ref state shards_.push_back(v->top_block_id()); } - auto write_data = [hash = state->root_cell()->get_hash(), cell_db_reader, + auto write_data = [hash = state->root_cell()->get_hash(), cell_db_reader = new_cell_db_reader, cancellation_token = cancellation_token_source_.get_cancellation_token()](td::FileFd& fd) mutable { - return vm::std_boc_serialize_to_file_large(cell_db_reader, hash, fd, 31, std::move(cancellation_token)); + auto res = vm::std_boc_serialize_to_file_large(cell_db_reader, hash, fd, 31, std::move(cancellation_token)); + cell_db_reader->print_stats(); + return res; }; auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result R) { if (R.is_error() && R.error().code() == cancelled) { @@ -273,9 +377,13 @@ void AsyncStateSerializer::got_shard_state(BlockHandle handle, td::Refid().id.to_str(); - auto write_data = [hash = state->root_cell()->get_hash(), cell_db_reader, + prepare_previous_state_cache(state->get_shard()); + auto new_cell_db_reader = std::make_shared(cell_db_reader, previous_state_cache_); + auto write_data = [hash = state->root_cell()->get_hash(), cell_db_reader = new_cell_db_reader, cancellation_token = cancellation_token_source_.get_cancellation_token()](td::FileFd& fd) mutable { - return vm::std_boc_serialize_to_file_large(cell_db_reader, hash, fd, 31, std::move(cancellation_token)); + auto res = vm::std_boc_serialize_to_file_large(cell_db_reader, hash, fd, 31, std::move(cancellation_token)); + cell_db_reader->print_stats(); + return res; }; auto P = td::PromiseCreator::lambda([SelfId = actor_id(this), handle](td::Result R) { if (R.is_error() && R.error().code() == cancelled) { @@ -329,6 +437,10 @@ bool AsyncStateSerializer::need_serialize(BlockHandle handle) { ValidatorManager::persistent_state_ttl(handle->unix_time()) > (UnixTime)td::Clocks::system(); } +bool AsyncStateSerializer::have_newer_persistent_state(UnixTime cur_ts) { + return cur_ts / (1 << 17) < last_known_key_block_ts_ / (1 << 17); +} + } // namespace validator } // namespace ton diff --git a/validator/state-serializer.hpp b/validator/state-serializer.hpp index 0bee70315..6d966f930 100644 --- a/validator/state-serializer.hpp +++ b/validator/state-serializer.hpp @@ -38,6 +38,7 @@ class AsyncStateSerializer : public td::actor::Actor { td::Ref opts_; td::CancellationTokenSource cancellation_token_source_; + UnixTime last_known_key_block_ts_ = 0; td::actor::ActorId manager_; @@ -47,6 +48,11 @@ class AsyncStateSerializer : public td::actor::Actor { bool have_masterchain_state_ = false; std::vector shards_; + std::vector> previous_state_files_; + std::shared_ptr>> previous_state_cache_; + std::vector previous_state_cur_shards_; + + void prepare_previous_state_cache(ShardIdFull shard); public: AsyncStateSerializer(BlockIdExt block_id, td::Ref opts, @@ -60,12 +66,15 @@ class AsyncStateSerializer : public td::actor::Actor { bool need_serialize(BlockHandle handle); bool need_monitor(ShardIdFull shard); + bool have_newer_persistent_state(UnixTime cur_ts); void alarm() override; void start_up() override; void got_self_state(AsyncSerializerState state); void got_init_handle(BlockHandle handle); + void request_previous_state_files(); + void got_previous_state_files(std::vector> files); void request_masterchain_state(); void request_shard_state(BlockIdExt shard); @@ -81,6 +90,10 @@ class AsyncStateSerializer : public td::actor::Actor { promise.set_result(last_block_id_.id.seqno); } + void update_last_known_key_block_ts(UnixTime ts) { + last_known_key_block_ts_ = std::max(last_known_key_block_ts_, ts); + } + void saved_to_db() { saved_to_db_ = true; running_ = false; diff --git a/validator/validator.h b/validator/validator.h index 5bbb66475..81f13487e 100644 --- a/validator/validator.h +++ b/validator/validator.h @@ -217,6 +217,8 @@ class ValidatorManagerInterface : public td::actor::Actor { td::Promise promise) = 0; virtual void get_persistent_state_slice(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::int64 offset, td::int64 max_length, td::Promise promise) = 0; + virtual void get_previous_persistent_state_files( + BlockSeqno cur_mc_seqno, td::Promise>> promise) = 0; virtual void get_block_proof(BlockHandle handle, td::Promise promise) = 0; virtual void get_block_proof_link(BlockHandle handle, td::Promise promise) = 0; virtual void get_block_handle(BlockIdExt block_id, bool force, td::Promise promise) = 0;