Optimization of persistent state serialization (#364)

* Fix double serialization of masterchain; increase sync_blocks_before

* Improve logging in DownloadState

* Write persistent state directly to file instead of a buffer

* Don't keep ref to masterchain state in AsyncStateSerializer

* Sparse state serialization over longer period

Co-authored-by: SpyCheese <mikle98@yandex.ru>

validator/db/archive-manager.cpp

@@ -336,6 +336,28 @@ void ArchiveManager::add_zero_state(BlockIdExt block_id, td::BufferSlice data, t
 
 void ArchiveManager::add_persistent_state(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::BufferSlice data,
                                           td::Promise<td::Unit> promise) {
+  auto create_writer = [&](std::string path, td::Promise<std::string> P) {
+    td::actor::create_actor<db::WriteFile>("writefile", db_root_ + "/archive/tmp/",
+                                           std::move(path), std::move(data), std::move(P))
+        .release();
+  };
+  add_persistent_state_impl(block_id, masterchain_block_id, std::move(promise), std::move(create_writer));
+}
+
+void ArchiveManager::add_persistent_state_gen(BlockIdExt block_id, BlockIdExt masterchain_block_id,
+                                              std::function<td::Status(td::FileFd&)> write_state,
+                                              td::Promise<td::Unit> promise) {
+  auto create_writer = [&](std::string path, td::Promise<std::string> P) {
+    td::actor::create_actor<db::WriteFile>("writefile", db_root_ + "/archive/tmp/",
+                                           std::move(path), std::move(write_state), std::move(P))
+        .release();
+  };
+  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<td::Unit> promise,
+                                               std::function<void(std::string, td::Promise<std::string>)> create_writer) {
   auto id = FileReference{fileref::PersistentState{block_id, masterchain_block_id}};
   auto hash = id.hash();
   if (perm_states_.find(hash) != perm_states_.end()) {
@@ -353,8 +375,7 @@ void ArchiveManager::add_persistent_state(BlockIdExt block_id, BlockIdExt master
           promise.set_value(td::Unit());
         }
       });
-  td::actor::create_actor<db::WriteFile>("writefile", db_root_ + "/archive/tmp/", path, std::move(data), std::move(P))
-      .release();
+  create_writer(std::move(path), std::move(P));
 }
 
 void ArchiveManager::get_zero_state(BlockIdExt block_id, td::Promise<td::BufferSlice> promise) {

validator/db/archive-manager.hpp

@@ -45,6 +45,9 @@ class ArchiveManager : public td::actor::Actor {
   void add_zero_state(BlockIdExt block_id, td::BufferSlice data, td::Promise<td::Unit> promise);
   void add_persistent_state(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::BufferSlice data,
                             td::Promise<td::Unit> promise);
+  void add_persistent_state_gen(BlockIdExt block_id, BlockIdExt masterchain_block_id,
+                                std::function<td::Status(td::FileFd&)> write_state,
+                                td::Promise<td::Unit> promise);
   void get_zero_state(BlockIdExt block_id, td::Promise<td::BufferSlice> promise);
   void get_persistent_state(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::Promise<td::BufferSlice> promise);
   void get_persistent_state_slice(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::int64 offset,
@@ -137,6 +140,8 @@ class ArchiveManager : public td::actor::Actor {
   PackageId get_max_temp_file_desc_idx();
   PackageId get_prev_temp_file_desc_idx(PackageId id);
 
+  void add_persistent_state_impl(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::Promise<td::Unit> promise,
+                                 std::function<void(std::string, td::Promise<std::string>)> create_writer);
   void written_perm_state(FileReferenceShort id);
 
   void persistent_state_gc(FileHash last);

validator/db/files-async.hpp

@@ -52,30 +52,50 @@ class WriteFile : public td::actor::Actor {
     auto res = R.move_as_ok();
     auto file = std::move(res.first);
     auto old_name = res.second;
-    td::uint64 offset = 0;
-    while (data_.size() > 0) {
-      auto R = file.pwrite(data_.as_slice(), offset);
-      auto s = R.move_as_ok();
-      offset += s;
-      data_.confirm_read(s);
+    auto status = write_data_(file);
+    if (!status.is_error()) {
+      status = file.sync();
+    }
+    if (status.is_error()) {
+      td::unlink(old_name);
+      promise_.set_error(std::move(status));
+      stop();
+      return;
     }
-    file.sync().ensure();
     if (new_name_.length() > 0) {
-      td::rename(old_name, new_name_).ensure();
-      promise_.set_value(std::move(new_name_));
+      status = td::rename(old_name, new_name_);
+      if (status.is_error()) {
+        promise_.set_error(std::move(status));
+      } else {
+        promise_.set_value(std::move(new_name_));
+      }
     } else {
       promise_.set_value(std::move(old_name));
     }
     stop();
   }
+  WriteFile(std::string tmp_dir, std::string new_name, std::function<td::Status(td::FileFd&)> write_data,
+            td::Promise<std::string> promise)
+      : tmp_dir_(tmp_dir), new_name_(new_name), write_data_(std::move(write_data)), promise_(std::move(promise)) {
+  }
   WriteFile(std::string tmp_dir, std::string new_name, td::BufferSlice data, td::Promise<std::string> promise)
-      : tmp_dir_(tmp_dir), new_name_(new_name), data_(std::move(data)), promise_(std::move(promise)) {
+      : tmp_dir_(tmp_dir), new_name_(new_name), promise_(std::move(promise)) {
+    write_data_ = [data_ptr = std::make_shared<td::BufferSlice>(std::move(data))] (td::FileFd& fd) {
+      auto data = std::move(*data_ptr);
+      td::uint64 offset = 0;
+      while (data.size() > 0) {
+        TRY_RESULT(s, fd.pwrite(data.as_slice(), offset));
+        offset += s;
+        data.confirm_read(s);
+      }
+      return td::Status::OK();
+    };
   }
 
  private:
   const std::string tmp_dir_;
   std::string new_name_;
-  td::BufferSlice data_;
+  std::function<td::Status(td::FileFd&)> write_data_;
   td::Promise<std::string> promise_;
 };
 

validator/db/rootdb.cpp

@@ -276,6 +276,13 @@ void RootDb::store_persistent_state_file(BlockIdExt block_id, BlockIdExt masterc
                           std::move(state), std::move(promise));
 }
 
+void RootDb::store_persistent_state_file_gen(BlockIdExt block_id, BlockIdExt masterchain_block_id,
+                                             std::function<td::Status(td::FileFd&)> write_data,
+                                             td::Promise<td::Unit> promise) {
+  td::actor::send_closure(archive_db_, &ArchiveManager::add_persistent_state_gen, block_id, masterchain_block_id,
+                          std::move(write_data), std::move(promise));
+}
+
 void RootDb::get_persistent_state_file(BlockIdExt block_id, BlockIdExt masterchain_block_id,
                                        td::Promise<td::BufferSlice> promise) {
   td::actor::send_closure(archive_db_, &ArchiveManager::get_persistent_state, block_id, masterchain_block_id,

validator/db/rootdb.hpp

@@ -69,6 +69,9 @@ class RootDb : public Db {
 
   void store_persistent_state_file(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::BufferSlice state,
                                    td::Promise<td::Unit> promise) override;
+  void store_persistent_state_file_gen(BlockIdExt block_id, BlockIdExt masterchain_block_id,
+                                       std::function<td::Status(td::FileFd&)> write_data,
+                                       td::Promise<td::Unit> promise) override;
   void get_persistent_state_file(BlockIdExt block_id, BlockIdExt masterchain_block_id,
                                  td::Promise<td::BufferSlice> promise) override;
   void get_persistent_state_file_slice(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::int64 offset,

validator/impl/shard.cpp

@@ -26,6 +26,7 @@
 #include "vm/cells/MerkleUpdate.h"
 #include "block/block-parse.h"
 #include "block/block-auto.h"
+#include "td/utils/filesystem.h"
 
 #define LAZY_STATE_DESERIALIZE 1
 
@@ -301,6 +302,30 @@ td::Result<td::BufferSlice> ShardStateQ::serialize() const {
   return st_res.move_as_ok();
 }
 
+td::Status ShardStateQ::serialize_to_file(td::FileFd& fd) const {
+  td::PerfWarningTimer perf_timer_{"serializestate", 0.1};
+  if (!data.is_null()) {
+    auto cur_data = data.clone();
+    while (cur_data.size() > 0) {
+      TRY_RESULT(s, fd.write(cur_data.as_slice()));
+      cur_data.confirm_read(s);
+    }
+    return td::Status::OK();
+  }
+  if (root.is_null()) {
+    return td::Status::Error(-666, "cannot serialize an uninitialized state");
+  }
+  vm::BagOfCells new_boc;
+  new_boc.set_root(root);
+  TRY_STATUS(new_boc.import_cells());
+  auto st_res = new_boc.serialize_to_file(fd, 31);
+  if (st_res.is_error()) {
+    LOG(ERROR) << "cannot serialize a shardchain state";
+    return st_res.move_as_error();
+  }
+  return td::Status::OK();
+}
+
 MasterchainStateQ::MasterchainStateQ(const BlockIdExt& _id, td::BufferSlice _data)
     : MasterchainState(), ShardStateQ(_id, std::move(_data)) {
 }

validator/impl/shard.hpp

@@ -87,6 +87,7 @@ class ShardStateQ : virtual public ShardState {
   td::Result<Ref<ShardState>> merge_with(const ShardState& with) const override;
   td::Result<std::pair<Ref<ShardState>, Ref<ShardState>>> split() const override;
   td::Result<td::BufferSlice> serialize() const override;
+  td::Status serialize_to_file(td::FileFd& fd) const override;
 };
 
 #if TD_MSVC

validator/interfaces/db.h

@@ -53,6 +53,9 @@ class Db : public td::actor::Actor {
 
   virtual void store_persistent_state_file(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::BufferSlice state,
                                            td::Promise<td::Unit> promise) = 0;
+  virtual void store_persistent_state_file_gen(BlockIdExt block_id, BlockIdExt masterchain_block_id,
+                                               std::function<td::Status(td::FileFd&)> write_data,
+                                               td::Promise<td::Unit> promise) = 0;
   virtual void get_persistent_state_file(BlockIdExt block_id, BlockIdExt masterchain_block_id,
                                          td::Promise<td::BufferSlice> promise) = 0;
   virtual void get_persistent_state_file_slice(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::int64 offset,

validator/interfaces/shard.h

@@ -55,6 +55,7 @@ class ShardState : public td::CntObject {
   virtual td::Result<std::pair<td::Ref<ShardState>, td::Ref<ShardState>>> split() const = 0;
 
   virtual td::Result<td::BufferSlice> serialize() const = 0;
+  virtual td::Status serialize_to_file(td::FileFd& fd) const = 0;
 };
 
 class MasterchainState : virtual public ShardState {

validator/interfaces/validator-manager.h

@@ -57,6 +57,9 @@ class ValidatorManager : public ValidatorManagerInterface {
                                td::Promise<td::Ref<ShardState>> promise) = 0;
   virtual void store_persistent_state_file(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::BufferSlice state,
                                            td::Promise<td::Unit> promise) = 0;
+  virtual void store_persistent_state_file_gen(BlockIdExt block_id, BlockIdExt masterchain_block_id,
+                                               std::function<td::Status(td::FileFd&)> write_data,
+                                               td::Promise<td::Unit> promise) = 0;
   virtual void store_zero_state_file(BlockIdExt block_id, td::BufferSlice state, td::Promise<td::Unit> promise) = 0;
   virtual void wait_block_state(BlockHandle handle, td::uint32 priority, td::Timestamp timeout,
                                 td::Promise<td::Ref<ShardState>> promise) = 0;

validator/manager-disk.cpp

@@ -680,6 +680,13 @@ void ValidatorManagerImpl::store_persistent_state_file(BlockIdExt block_id, Bloc
                           std::move(promise));
 }
 
+void ValidatorManagerImpl::store_persistent_state_file_gen(BlockIdExt block_id, BlockIdExt masterchain_block_id,
+                                                           std::function<td::Status(td::FileFd&)> write_data,
+                                                           td::Promise<td::Unit> promise) {
+  td::actor::send_closure(db_, &Db::store_persistent_state_file_gen, block_id, masterchain_block_id,
+                          std::move(write_data), std::move(promise));
+}
+
 void ValidatorManagerImpl::store_zero_state_file(BlockIdExt block_id, td::BufferSlice state,
                                                  td::Promise<td::Unit> promise) {
   td::actor::send_closure(db_, &Db::store_zero_state_file, block_id, std::move(state), std::move(promise));

validator/manager-disk.hpp

@@ -143,6 +143,9 @@ class ValidatorManagerImpl : public ValidatorManager {
                        td::Promise<td::Ref<ShardState>> promise) override;
   void store_persistent_state_file(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::BufferSlice state,
                                    td::Promise<td::Unit> promise) override;
+  void store_persistent_state_file_gen(BlockIdExt block_id, BlockIdExt masterchain_block_id,
+                                       std::function<td::Status(td::FileFd&)> write_data,
+                                       td::Promise<td::Unit> promise) override;
   void store_zero_state_file(BlockIdExt block_id, td::BufferSlice state, td::Promise<td::Unit> promise) override;
   void wait_block_state(BlockHandle handle, td::uint32 priority, td::Timestamp timeout,
                         td::Promise<td::Ref<ShardState>> promise) override;

validator/manager-hardfork.hpp

@@ -169,6 +169,11 @@ class ValidatorManagerImpl : public ValidatorManager {
                                    td::Promise<td::Unit> promise) override {
     UNREACHABLE();
   }
+  void store_persistent_state_file_gen(BlockIdExt block_id, BlockIdExt masterchain_block_id,
+                                       std::function<td::Status(td::FileFd&)> write_data,
+                                       td::Promise<td::Unit> promise) override {
+    UNREACHABLE();
+  }
   void store_zero_state_file(BlockIdExt block_id, td::BufferSlice state, td::Promise<td::Unit> promise) override {
     UNREACHABLE();
   }

validator/manager.cpp

@@ -1051,6 +1051,13 @@ void ValidatorManagerImpl::store_persistent_state_file(BlockIdExt block_id, Bloc
                           std::move(promise));
 }
 
+void ValidatorManagerImpl::store_persistent_state_file_gen(BlockIdExt block_id, BlockIdExt masterchain_block_id,
+                                                           std::function<td::Status(td::FileFd&)> write_data,
+                                                           td::Promise<td::Unit> promise) {
+  td::actor::send_closure(db_, &Db::store_persistent_state_file_gen, block_id, masterchain_block_id,
+                          std::move(write_data), std::move(promise));
+}
+
 void ValidatorManagerImpl::store_zero_state_file(BlockIdExt block_id, td::BufferSlice state,
                                                  td::Promise<td::Unit> promise) {
   td::actor::send_closure(db_, &Db::store_zero_state_file, block_id, std::move(state), std::move(promise));

validator/manager.hpp

@@ -347,6 +347,9 @@ class ValidatorManagerImpl : public ValidatorManager {
                        td::Promise<td::Ref<ShardState>> promise) override;
   void store_persistent_state_file(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::BufferSlice state,
                                    td::Promise<td::Unit> promise) override;
+  void store_persistent_state_file_gen(BlockIdExt block_id, BlockIdExt masterchain_block_id,
+                                       std::function<td::Status(td::FileFd&)> write_data,
+                                       td::Promise<td::Unit> promise) override;
   void store_zero_state_file(BlockIdExt block_id, td::BufferSlice state, td::Promise<td::Unit> promise) override;
   void wait_block_state(BlockHandle handle, td::uint32 priority, td::Timestamp timeout,
                         td::Promise<td::Ref<ShardState>> promise) override;

validator/net/download-state.cpp

@@ -149,6 +149,7 @@ void DownloadState::got_block_state_description(td::BufferSlice data) {
     abort_query(F.move_as_error());
     return;
   }
+  prev_logged_timer_ = td::Timer();
 
   ton_api::downcast_call(
       *F.move_as_ok().get(),
@@ -188,8 +189,12 @@ void DownloadState::got_block_state_part(td::BufferSlice data, td::uint32 reques
   sum_ += data.size();
   parts_.push_back(std::move(data));
 
-  if (sum_ % (1 << 22) == 0) {
-    LOG(DEBUG) << "downloading state " << block_id_ << ": total=" << sum_;
+  double elapsed = prev_logged_timer_.elapsed();
+  if (elapsed > 10.0) {
+    prev_logged_timer_ = td::Timer();
+    LOG(INFO) << "downloading state " << block_id_ << ": total=" << sum_ <<
+        " (" << double(sum_ - prev_logged_sum_) / elapsed << " B/s)";
+    prev_logged_sum_ = sum_;
   }
 
   if (last_part) {

validator/net/download-state.hpp

@@ -72,6 +72,9 @@ class DownloadState : public td::actor::Actor {
   td::BufferSlice state_;
   std::vector<td::BufferSlice> parts_;
   td::uint64 sum_ = 0;
+
+  td::uint64 prev_logged_sum_ = 0;
+  td::Timer prev_logged_timer_;
 };
 
 }  // namespace fullnode

validator/state-serializer.cpp

@@ -122,14 +122,12 @@ 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_)) {
-    if (masterchain_state_.is_null()) {
+    if (!have_masterchain_state_) {
           // block next attempts immediately, but send actual request later
           running_ = true;
           delay_action(
             [SelfId = actor_id(this)]() { td::actor::send_closure(SelfId, &AsyncStateSerializer::request_masterchain_state); },
-            // Masterchain is more important and much lighter than shards
-            // thus lower delay
-            td::Timestamp::in(td::Random::fast(0, 600)));
+            td::Timestamp::in(td::Random::fast(0, 3600)));
       return;
     }
     while (next_idx_ < shards_.size()) {
@@ -140,8 +138,6 @@ void AsyncStateSerializer::next_iteration() {
           running_ = true;
           delay_action(
             [SelfId = actor_id(this), shard = shards_[next_idx_]]() { td::actor::send_closure(SelfId, &AsyncStateSerializer::request_shard_state, shard); },
-            // Shards are less important and heavier than master
-            // thus higher delay
             td::Timestamp::in(td::Random::fast(0, 4 * 3600)));
         return;
       }
@@ -162,7 +158,7 @@ void AsyncStateSerializer::next_iteration() {
   }
   if (masterchain_handle_->inited_next_left()) {
     last_block_id_ = masterchain_handle_->one_next(true);
-    masterchain_state_ = td::Ref<MasterchainState>{};
+    have_masterchain_state_ = false;
     masterchain_handle_ = nullptr;
     saved_to_db_ = false;
     shards_.clear();
@@ -186,25 +182,25 @@ void AsyncStateSerializer::got_masterchain_handle(BlockHandle handle) {
 }
 
 void AsyncStateSerializer::got_masterchain_state(td::Ref<MasterchainState> state) {
-  masterchain_state_ = state;
+  have_masterchain_state_ = true;
   CHECK(next_idx_ == 0);
   CHECK(shards_.size() == 0);
 
-  auto vec = masterchain_state_->get_shards();
-  shards_.push_back(masterchain_handle_->id());
+  auto vec = state->get_shards();
   for (auto &v : vec) {
     shards_.push_back(v->top_block_id());
   }
 
-  auto B = masterchain_state_->serialize();
-  B.ensure();
+  auto write_data = [state] (td::FileFd& fd) {
+    return state->serialize_to_file(fd);
+  };
   auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<td::Unit> R) {
     R.ensure();
     td::actor::send_closure(SelfId, &AsyncStateSerializer::stored_masterchain_state);
   });
 
-  td::actor::send_closure(manager_, &ValidatorManager::store_persistent_state_file, masterchain_handle_->id(),
-                          masterchain_handle_->id(), B.move_as_ok(), std::move(P));
+  td::actor::send_closure(manager_, &ValidatorManager::store_persistent_state_file_gen, masterchain_handle_->id(),
+                          masterchain_handle_->id(), write_data, std::move(P));
 }
 
 void AsyncStateSerializer::stored_masterchain_state() {
@@ -225,13 +221,15 @@ void AsyncStateSerializer::got_shard_handle(BlockHandle handle) {
 }
 
 void AsyncStateSerializer::got_shard_state(BlockHandle handle, td::Ref<ShardState> state) {
-  auto B = state->serialize().move_as_ok();
+  auto write_data = [state] (td::FileFd& fd) {
+    return state->serialize_to_file(fd);
+  };
   auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<td::Unit> R) {
     R.ensure();
     td::actor::send_closure(SelfId, &AsyncStateSerializer::success_handler);
   });
-  td::actor::send_closure(manager_, &ValidatorManager::store_persistent_state_file, handle->id(),
-                          masterchain_handle_->id(), std::move(B), std::move(P));
+  td::actor::send_closure(manager_, &ValidatorManager::store_persistent_state_file_gen, handle->id(),
+                          masterchain_handle_->id(), write_data, std::move(P));
   LOG(INFO) << "storing persistent state for " << masterchain_handle_->id().seqno() << ":" << handle->id().id.shard;
   next_idx_++;
 }

validator/state-serializer.hpp

@@ -43,7 +43,7 @@ class AsyncStateSerializer : public td::actor::Actor {
   td::uint32 next_idx_ = 0;
 
   BlockHandle masterchain_handle_;
-  td::Ref<MasterchainState> masterchain_state_;
+  bool have_masterchain_state_ = false;
 
   std::vector<BlockIdExt> shards_;
 

validator/validator.h

@@ -96,7 +96,7 @@ struct ValidatorManagerOptions : public td::CntObject {
       BlockIdExt zero_block_id, BlockIdExt init_block_id,
       std::function<bool(ShardIdFull, CatchainSeqno, ShardCheckMode)> check_shard = [](ShardIdFull, CatchainSeqno,
                                                                                        ShardCheckMode) { return true; },
-      bool allow_blockchain_init = false, double sync_blocks_before = 300, double block_ttl = 86400 * 7,
+      bool allow_blockchain_init = false, double sync_blocks_before = 86400, double block_ttl = 86400 * 7,
       double state_ttl = 3600, double archive_ttl = 86400 * 365, double key_proof_ttl = 86400 * 3650,
       double max_mempool_num = 999999,
       bool initial_sync_disabled = false);