Improve tweaking for high throughput (ton-blockchain#610)

* Option "--disable-ext-msg-broadcast" * "Get shard out queue size" query * Move disabling ext msg broadcasts from command-line arguments to config * Fix compilation error * Asynchronous store_cell and gc in celldb * Make GC in celldb work evenly over time * Increase timeouts for downloading blocks * Reuse blocks from previous rounds in validator session * Use Rldp2 in FullNode for downloading persistent states and archives * Improve logs in download-archive-slice and download-state * Decrease delay between serializing shards * Make CellDbIn::load_cell synchronous to avoid interfering with store_cell --------- Co-authored-by: SpyCheese <[email protected]>
coalars · Mar 15, 2023 · 47311d6 · 47311d6
1 parent 30c742a
commit 47311d6
Show file tree

Hide file tree

Showing 33 changed files with 712 additions and 163 deletions.
diff --git a/crypto/vm/db/CellHashTable.h b/crypto/vm/db/CellHashTable.h
@@ -43,7 +43,7 @@ class CellHashTable {
   template <class F>
   void for_each(F &&f) {
     for (auto &info : set_) {
-      f(info);
+      f(const_cast<InfoT &>(info));
     }
   }
   template <class F>

diff --git a/crypto/vm/db/CellStorage.cpp b/crypto/vm/db/CellStorage.cpp
@@ -150,6 +150,20 @@ td::Result<CellLoader::LoadResult> CellLoader::load(td::Slice hash, bool need_da
   return res;
 }
 
+td::Result<CellLoader::LoadResult> CellLoader::load_refcnt(td::Slice hash) {
+  LoadResult res;
+  std::string serialized;
+  TRY_RESULT(get_status, reader_->get(hash, serialized));
+  if (get_status != KeyValue::GetStatus::Ok) {
+    DCHECK(get_status == KeyValue::GetStatus::NotFound);
+    return res;
+  }
+  res.status = LoadResult::Ok;
+  td::TlParser parser(serialized);
+  td::parse(res.refcnt_, parser);
+  return res;
+}
+
 CellStorer::CellStorer(KeyValue &kv) : kv_(kv) {
 }
 

diff --git a/crypto/vm/db/CellStorage.h b/crypto/vm/db/CellStorage.h
@@ -48,6 +48,7 @@ class CellLoader {
   };
   CellLoader(std::shared_ptr<KeyValueReader> reader);
   td::Result<LoadResult> load(td::Slice hash, bool need_data, ExtCellCreator &ext_cell_creator);
+  td::Result<LoadResult> load_refcnt(td::Slice hash);  // This only loads refcnt_, cell_ == null
 
  private:
   std::shared_ptr<KeyValueReader> reader_;

diff --git a/crypto/vm/db/DynamicBagOfCellsDb.cpp b/crypto/vm/db/DynamicBagOfCellsDb.cpp
@@ -27,6 +27,9 @@
 #include "td/utils/ThreadSafeCounter.h"
 
 #include "vm/cellslice.h"
+#include <queue>
+#include "td/actor/actor.h"
+#include "common/delay.h"
 
 namespace vm {
 namespace {
@@ -138,8 +141,6 @@ class DynamicBagOfCellsDbImpl : public DynamicBagOfCellsDb, private ExtCellCreat
     if (cell->get_virtualization() != 0) {
       return;
     }
-    //LOG(ERROR) << "INC";
-    //CellSlice(cell, nullptr).print_rec(std::cout);
     to_inc_.push_back(cell);
   }
   void dec(const Ref<Cell> &cell) override {
@@ -149,8 +150,6 @@ class DynamicBagOfCellsDbImpl : public DynamicBagOfCellsDb, private ExtCellCreat
     if (cell->get_virtualization() != 0) {
       return;
     }
-    //LOG(ERROR) << "DEC";
-    //CellSlice(cell, nullptr).print_rec(std::cout);
     to_dec_.push_back(cell);
   }
 
@@ -167,25 +166,20 @@ class DynamicBagOfCellsDbImpl : public DynamicBagOfCellsDb, private ExtCellCreat
     if (is_prepared_for_commit()) {
       return td::Status::OK();
     }
-    //LOG(ERROR) << "dfs_new_cells_in_db";
     for (auto &new_cell : to_inc_) {
       auto &new_cell_info = get_cell_info(new_cell);
       dfs_new_cells_in_db(new_cell_info);
     }
-    //return td::Status::OK();
-    //LOG(ERROR) << "dfs_new_cells";
     for (auto &new_cell : to_inc_) {
       auto &new_cell_info = get_cell_info(new_cell);
       dfs_new_cells(new_cell_info);
     }
 
-    //LOG(ERROR) << "dfs_old_cells";
     for (auto &old_cell : to_dec_) {
       auto &old_cell_info = get_cell_info(old_cell);
       dfs_old_cells(old_cell_info);
     }
 
-    //LOG(ERROR) << "save_diff_prepare";
     save_diff_prepare();
 
     to_inc_.clear();
@@ -363,7 +357,6 @@ class DynamicBagOfCellsDbImpl : public DynamicBagOfCellsDb, private ExtCellCreat
       info.was = true;
       visited_.push_back(&info);
     }
-    //LOG(ERROR) << "dfs new " << td::format::escaped(info.cell->hash());
 
     if (info.was_dfs_new_cells) {
       return;
@@ -384,7 +377,6 @@ class DynamicBagOfCellsDbImpl : public DynamicBagOfCellsDb, private ExtCellCreat
       info.was = true;
       visited_.push_back(&info);
     }
-    //LOG(ERROR) << "dfs old " << td::format::escaped(info.cell->hash());
 
     load_cell(info);
 
@@ -405,7 +397,6 @@ class DynamicBagOfCellsDbImpl : public DynamicBagOfCellsDb, private ExtCellCreat
   }
 
   void save_diff(CellStorer &storer) {
-    //LOG(ERROR) << hash_table_.size();
     for (auto info_ptr : visited_) {
       save_cell(*info_ptr, storer);
     }
@@ -414,7 +405,6 @@ class DynamicBagOfCellsDbImpl : public DynamicBagOfCellsDb, private ExtCellCreat
 
   void save_cell_prepare(CellInfo &info) {
     if (info.refcnt_diff == 0) {
-      //CellSlice(info.cell, nullptr).print_rec(std::cout);
       return;
     }
     load_cell(info);
@@ -450,15 +440,11 @@ class DynamicBagOfCellsDbImpl : public DynamicBagOfCellsDb, private ExtCellCreat
 
     if (info.db_refcnt == 0) {
       CHECK(info.in_db);
-      //LOG(ERROR) << "ERASE";
-      //CellSlice(NoVm(), info.cell).print_rec(std::cout);
       storer.erase(info.cell->get_hash().as_slice());
       info.in_db = false;
       hash_table_.erase(info.cell->get_hash().as_slice());
       guard.dismiss();
     } else {
-      //LOG(ERROR) << "SAVE " << info.db_refcnt;
-      //CellSlice(NoVm(), info.cell).print_rec(std::cout);
       auto loaded_cell = info.cell->load_cell().move_as_ok();
       storer.set(info.db_refcnt, *loaded_cell.data_cell);
       info.in_db = true;
@@ -482,7 +468,6 @@ class DynamicBagOfCellsDbImpl : public DynamicBagOfCellsDb, private ExtCellCreat
     CHECK(cell->is_loaded());
     vm::CellSlice cs(vm::NoVm{}, cell);  // FIXME
     for (unsigned i = 0; i < cs.size_refs(); i++) {
-      //LOG(ERROR) << "---> " << td::format::escaped(cell->ref(i)->hash());
       f(get_cell_info(cs.prefetch_ref(i)));
     }
   }
@@ -573,6 +558,177 @@ class DynamicBagOfCellsDbImpl : public DynamicBagOfCellsDb, private ExtCellCreat
                                               DynamicBocExtCellExtra{cell_db_reader_}));
     return std::move(res);
   }
+
+  struct PrepareCommitAsyncState {
+    size_t remaining_ = 0;
+    std::shared_ptr<AsyncExecutor> executor_;
+    td::Promise<td::Unit> promise_;
+
+    struct CellInfo2 {
+      CellInfo *info;
+      std::vector<CellInfo2 *> parents;
+      unsigned remaining_children = 0;
+      Cell::Hash key() const {
+        return info->key();
+      }
+      bool operator<(const CellInfo2 &other) const {
+        return key() < other.key();
+      }
+
+      friend bool operator<(const CellInfo2 &a, td::Slice b) {
+        return a.key().as_slice() < b;
+      }
+
+      friend bool operator<(td::Slice a, const CellInfo2 &b) {
+        return a < b.key().as_slice();
+      }
+    };
+
+    CellHashTable<CellInfo2> cells_;
+  };
+  std::unique_ptr<PrepareCommitAsyncState> pca_state_;
+
+  void prepare_commit_async(std::shared_ptr<AsyncExecutor> executor, td::Promise<td::Unit> promise) override {
+    if (pca_state_) {
+      promise.set_error(td::Status::Error("Other prepare_commit_async is not finished"));
+      return;
+    }
+    if (is_prepared_for_commit()) {
+      promise.set_result(td::Unit());
+      return;
+    }
+    pca_state_ = std::make_unique<PrepareCommitAsyncState>();
+    pca_state_->executor_ = std::move(executor);
+    pca_state_->promise_ = std::move(promise);
+    for (auto &new_cell : to_inc_) {
+      dfs_new_cells_in_db_async(new_cell);
+    }
+    pca_state_->cells_.for_each([&](PrepareCommitAsyncState::CellInfo2 &info) {
+      ++pca_state_->remaining_;
+      if (info.remaining_children == 0) {
+        pca_load_from_db(&info);
+      }
+    });
+    if (pca_state_->remaining_ == 0) {
+      prepare_commit_async_cont();
+    }
+  }
+
+  void dfs_new_cells_in_db_async(const td::Ref<vm::Cell> &cell, PrepareCommitAsyncState::CellInfo2 *parent = nullptr) {
+    bool exists = true;
+    pca_state_->cells_.apply(cell->get_hash().as_slice(), [&](PrepareCommitAsyncState::CellInfo2 &info) {
+      if (info.info == nullptr) {
+        exists = false;
+        info.info = &get_cell_info(cell);
+      }
+    });
+    auto info = pca_state_->cells_.get_if_exists(cell->get_hash().as_slice());
+    if (parent) {
+      info->parents.push_back(parent);
+      ++parent->remaining_children;
+    }
+    if (exists) {
+      return;
+    }
+    if (cell->is_loaded()) {
+      vm::CellSlice cs(vm::NoVm{}, cell);
+      for (unsigned i = 0; i < cs.size_refs(); i++) {
+        dfs_new_cells_in_db_async(cs.prefetch_ref(i), info);
+      }
+    }
+  }
+
+  void pca_load_from_db(PrepareCommitAsyncState::CellInfo2 *info) {
+    pca_state_->executor_->execute_async(
+        [db = this, info, executor = pca_state_->executor_, loader = *loader_]() mutable {
+          auto res = loader.load_refcnt(info->info->cell->get_hash().as_slice()).move_as_ok();
+          executor->execute_sync([db, info, res = std::move(res)]() {
+            db->pca_set_in_db(info, std::move(res));
+          });
+        });
+  }
+
+  void pca_set_in_db(PrepareCommitAsyncState::CellInfo2 *info, CellLoader::LoadResult result) {
+    info->info->sync_with_db = true;
+    if (result.status == CellLoader::LoadResult::Ok) {
+      info->info->in_db = true;
+      info->info->db_refcnt = result.refcnt();
+    } else {
+      info->info->in_db = false;
+    }
+    for (PrepareCommitAsyncState::CellInfo2 *parent_info : info->parents) {
+      if (parent_info->info->sync_with_db) {
+        continue;
+      }
+      if (!info->info->in_db) {
+        pca_set_in_db(parent_info, {});
+      } else if (--parent_info->remaining_children == 0) {
+        pca_load_from_db(parent_info);
+      }
+    }
+    if (--pca_state_->remaining_ == 0) {
+      prepare_commit_async_cont();
+    }
+  }
+
+  void prepare_commit_async_cont() {
+    for (auto &new_cell : to_inc_) {
+      auto &new_cell_info = get_cell_info(new_cell);
+      dfs_new_cells(new_cell_info);
+    }
+
+    CHECK(pca_state_->remaining_ == 0);
+    for (auto &old_cell : to_dec_) {
+      auto &old_cell_info = get_cell_info(old_cell);
+      dfs_old_cells_async(old_cell_info);
+    }
+    if (pca_state_->remaining_ == 0) {
+      prepare_commit_async_cont2();
+    }
+  }
+
+  void dfs_old_cells_async(CellInfo &info) {
+    if (!info.was) {
+      info.was = true;
+      visited_.push_back(&info);
+      if (!info.sync_with_db) {
+        ++pca_state_->remaining_;
+        load_cell_async(
+            info.cell->get_hash().as_slice(), pca_state_->executor_,
+            [executor = pca_state_->executor_, db = this, info = &info](td::Result<td::Ref<vm::DataCell>> R) {
+              R.ensure();
+              executor->execute_sync([db, info]() {
+                CHECK(info->sync_with_db);
+                db->dfs_old_cells_async(*info);
+                if (--db->pca_state_->remaining_ == 0) {
+                  db->prepare_commit_async_cont2();
+                }
+              });
+            });
+        return;
+      }
+    }
+    info.refcnt_diff--;
+    if (!info.sync_with_db) {
+      return;
+    }
+    auto new_refcnt = info.refcnt_diff + info.db_refcnt;
+    CHECK(new_refcnt >= 0);
+    if (new_refcnt != 0) {
+      return;
+    }
+
+    for_each(info, [this](auto &child_info) { dfs_old_cells_async(child_info); });
+  }
+
+  void prepare_commit_async_cont2() {
+    save_diff_prepare();
+    to_inc_.clear();
+    to_dec_.clear();
+    pca_state_->promise_.set_result(td::Unit());
+    pca_state_ = {};
+  }
+
 };
 }  // namespace
 

diff --git a/crypto/vm/db/DynamicBagOfCellsDb.h b/crypto/vm/db/DynamicBagOfCellsDb.h
@@ -75,6 +75,7 @@ class DynamicBagOfCellsDb {
 
   virtual void load_cell_async(td::Slice hash, std::shared_ptr<AsyncExecutor> executor,
                                td::Promise<Ref<DataCell>> promise) = 0;
+  virtual void prepare_commit_async(std::shared_ptr<AsyncExecutor> executor, td::Promise<td::Unit> promise) = 0;
 };
 
 }  // namespace vm
diff --git a/dht-server/dht-server.cpp b/dht-server/dht-server.cpp
@@ -170,7 +170,7 @@ ton::tl_object_ptr<ton::ton_api::engine_validator_config> Config::tl() const {
   return ton::create_tl_object<ton::ton_api::engine_validator_config>(
       out_port, std::move(addrs_vec), std::move(adnl_vec), std::move(dht_vec), std::move(val_vec),
       ton::PublicKeyHash::zero().tl(), std::move(full_node_slaves_vec), std::move(full_node_masters_vec),
-      std::move(liteserver_vec), std::move(control_vec), std::move(gc_vec));
+      nullptr, std::move(liteserver_vec), std::move(control_vec), std::move(gc_vec));
 }
 
 td::Result<bool> Config::config_add_network_addr(td::IPAddress in_ip, td::IPAddress out_ip,

diff --git a/tdutils/td/utils/OptionParser.cpp b/tdutils/td/utils/OptionParser.cpp
@@ -33,7 +33,7 @@ void OptionParser::set_description(string description) {
 void OptionParser::add_option(Option::Type type, char short_key, Slice long_key, Slice description,
                               std::function<Status(Slice)> callback) {
   for (auto &option : options_) {
-    if (option.short_key == short_key || (!long_key.empty() && long_key == option.long_key)) {
+    if ((short_key != '\0' && option.short_key == short_key) || (!long_key.empty() && long_key == option.long_key)) {
       LOG(ERROR) << "Ignore duplicated option '" << short_key << "' '" << long_key << "'";
     }
   }

diff --git a/tl/generate/scheme/ton_api.tl b/tl/generate/scheme/ton_api.tl
@@ -588,12 +588,14 @@ engine.gc ids:(vector int256) = engine.Gc;
 engine.dht.config dht:(vector engine.dht) gc:engine.gc = engine.dht.Config;
 engine.validator.fullNodeMaster port:int adnl:int256 = engine.validator.FullNodeMaster;
 engine.validator.fullNodeSlave ip:int port:int adnl:PublicKey = engine.validator.FullNodeSlave;
+engine.validator.fullNodeConfig ext_messages_broadcast_disabled:Bool = engine.validator.FullNodeConfig;
 engine.validator.config out_port:int addrs:(vector engine.Addr) adnl:(vector engine.adnl) 
         dht:(vector engine.dht)
         validators:(vector engine.validator) fullnode:int256 fullnodeslaves:(vector engine.validator.fullNodeSlave)
         fullnodemasters:(vector engine.validator.fullNodeMaster)
+        fullnodeconfig:engine.validator.fullNodeConfig
         liteservers:(vector engine.liteServer) control:(vector engine.controlInterface)
-        gc:engine.gc = engine.validator.Config; 
+        gc:engine.gc = engine.validator.Config;
 
 ---functions---
 ---types---
@@ -642,6 +644,8 @@ engine.validator.onePerfTimerStat time:int min:double avg:double max:double = en
 engine.validator.perfTimerStatsByName name:string stats:(vector engine.validator.OnePerfTimerStat) = engine.validator.PerfTimerStatsByName;
 engine.validator.perfTimerStats stats:(vector engine.validator.PerfTimerStatsByName) = engine.validator.PerfTimerStats;
 
+engine.validator.shardOutQueueSize size:int = engine.validator.ShardOutQueueSize;
+
 
 ---functions---
 
@@ -693,6 +697,8 @@ engine.validator.signShardOverlayCertificate workchain:int shard:long signed_key
 engine.validator.importShardOverlayCertificate workchain:int shard:long signed_key:engine.validator.KeyHash cert:overlay.Certificate = engine.validator.Success;
 
 engine.validator.getPerfTimerStats name:string = engine.validator.PerfTimerStats;
+engine.validator.getShardOutQueueSize flags:# block_id:tonNode.blockId dest_wc:flags.0?int dest_shard:flags.0?long = engine.validator.ShardOutQueueSize;
+engine.validator.setExtMessagesBroadcastDisabled disabled:Bool = engine.validator.Success;
 
 ---types---
 

diff --git a/tl/generate/scheme/ton_api.tlo b/tl/generate/scheme/ton_api.tlo