Implement parallelism in validator engine (#1812)

* Apply saved patches for parallel checking of accounts' transactions

* Cleanup unneeded code

* Fix account_dict update logic

* Implement thread-safe scan_account_libraries

* Implement thread-safe deferring messages

* Implement thread-safe timer

* Implement thread-safe storage stat cache

* Implement thread-safe work time stat

* Cleanup threadpool

* Add option to control parallel validation of accounts

* Final touches

* Integrate with td::actor system

* Implement synchronous `CheckAccountTxs` execution

* Implement parallel `CheckAccountTxs` execution

* Cleanup unneeded code

* Remove unneeded lock-free structure

* Final polishing

* Fix ValidateQuery actor dying before all CheckAccountTxs actors finish

* Catch VM errors inside CheckAccountTxs actor

* Prevent ValidateQuery actor dying by timeout before CheckAccountTxs

* Better naming for CheckAccountTxs actors

* Fixes after merge

* Fix timer measurements

* More accurate validation stats

* More accurate error handling

* Remove unneeded alarm in CheckAccountTxs actor

* Fix incorrect error reporting

* Final fix of error handling

* Fix error handling (again)

* Rename option

* Add fixes from review

* Codestyle

* Remove debug code

Author: avevad

tl/generate/scheme/ton_api.tl

@@ -980,10 +980,11 @@ validatorStats.validatedBlock
     self:int256
     valid:Bool comment:string
     bytes:int collated_data_bytes:int
-    total_time:double work_time:double cpu_work_time:double time_stats:string
+    total_time:double work_time:double actual_time:double cpu_work_time:double time_stats:string
     work_time_real_stats:string
     work_time_cpu_stats:string
-    storage_stat_cache:validatorStats.storageStatCacheStats = validatorStats.ValidatedBlock;
+    storage_stat_cache:validatorStats.storageStatCacheStats
+    parallel_accounts_validation:Bool = validatorStats.ValidatedBlock;
 
 validatorStats.newValidatorGroup.node id:int256 pubkey:PublicKey adnl_id:int256 weight:long = validatorStats.newValidatorGroup.Node;
 validatorStats.newValidatorGroup session_id:int256 shard:tonNode.shardId cc_seqno:int

validator-engine/validator-engine.cpp

@@ -1638,6 +1638,7 @@ td::Status ValidatorEngine::load_global_config() {
   }
   validator_options_.write().set_hardforks(std::move(h));
   validator_options_.write().set_catchain_broadcast_speed_multiplier(broadcast_speed_multiplier_catchain_);
+  validator_options_.write().set_parallel_validation(parallel_validation_);
 
   for (auto &id : config_.collator_node_whitelist) {
     validator_options_.write().set_collator_node_whitelisted_validator(id, true);
@@ -5576,6 +5577,9 @@ int main(int argc, char *argv[]) {
                          }
                          return td::Status::OK();
                        });
+  p.add_option('\0', "parallel-validation", "parallel validation over different accounts", [&]() {
+    acts.push_back([&x]() { td::actor::send_closure(x, &ValidatorEngine::set_parallel_validation, true); });
+  });
   auto S = p.run(argc, argv);
   if (S.is_error()) {
     LOG(ERROR) << "failed to parse options: " << S.move_as_error();

validator-engine/validator-engine.hpp

@@ -257,6 +257,7 @@ class ValidatorEngine : public td::actor::Actor {
   td::optional<ton::BlockSeqno> sync_shards_upto_;
   ton::adnl::AdnlNodeIdShort shard_block_retainer_adnl_id_ = ton::adnl::AdnlNodeIdShort::zero();
   bool shard_block_retainer_adnl_id_fullnode_ = false;
+  bool parallel_validation_ = false;
   double initial_sync_delay_ = 60.0;
 
   std::set<ton::CatchainSeqno> unsafe_catchains_;
@@ -382,6 +383,9 @@ class ValidatorEngine : public td::actor::Actor {
   void set_shard_block_retainer_adnl_id_fullnode() {
     shard_block_retainer_adnl_id_fullnode_ = true;
   }
+  void set_parallel_validation(bool value) {
+    parallel_validation_ = value;
+  }
   void set_initial_sync_delay(double value) {
     initial_sync_delay_ = value;
   }

validator/fabric.h

@@ -54,6 +54,8 @@ struct ValidateParams {
 
   // Optional - used for validation of optimistic candidates
   Ref<BlockData> optimistic_prev_block = {};
+
+  bool parallel_validation = false;
 };
 
 td::actor::ActorOwn<Db> create_db_actor(td::actor::ActorId<ValidatorManager> manager, std::string db_root_,

validator/impl/validate-query.cpp

@@ -18,7 +18,9 @@
 */
 #pragma once
 
+#include <atomic>
 #include <map>
+#include <optional>
 #include <string>
 #include <vector>
 
@@ -103,9 +105,9 @@ inline ErrorCtxSet ErrorCtx::set_guard(std::vector<std::string> str_list) {
 /*
  *
  * must write candidate to disk, if accepted
- * can reject block only if it is invalid (i.e. in case of 
+ * can reject block only if it is invalid (i.e. in case of
  * internal errors must retry or crash)
- * only exception: block can be rejected, if it is known from 
+ * only exception: block can be rejected, if it is known from
  * masterchain, that it will not be part of shardchain finalized
  * state
  *
@@ -150,6 +152,13 @@ class ValidateQuery : public td::actor::Actor {
   bool prev_key_block_exists_{false};
   bool debug_checks_{false};
   bool outq_cleanup_partial_{false};
+  bool parallel_accounts_validation_{false};
+  bool parallel_accounts_validation_pending_{false};
+  bool check_account_failed_{false};
+  td::RealCpuTimer parallel_work_timer_{/*is_paused=*/true};
+  std::optional<td::Status> check_account_fatal_error_ = std::nullopt;
+  std::optional<std::string> check_account_reject_error_ = std::nullopt;
+  std::optional<td::BufferSlice> check_account_reject_reason_ = std::nullopt;
   BlockSeqno prev_key_seqno_{~0u};
   int stage_{0};
   td::BitArray<64> shard_pfx_;
@@ -201,8 +210,7 @@ class ValidateQuery : public td::actor::Actor {
   ton::BlockIdExt prev_key_block_;
   ton::LogicalTime prev_key_block_lt_;
   std::unique_ptr<block::BlockLimits> block_limits_;
-  std::unique_ptr<block::BlockLimitStatus> block_limit_status_;
-  td::uint64 total_gas_used_{0}, total_special_gas_used_{0};
+  mutable std::atomic_uint64_t total_gas_used_{0}, total_special_gas_used_{0};
 
   LogicalTime start_lt_, end_lt_;
   UnixTime prev_now_{~0u}, now_{~0u};
@@ -330,6 +338,7 @@ class ValidateQuery : public td::actor::Actor {
   bool unpack_block_candidate();
   bool extract_collated_data_from(Ref<vm::Cell> croot, int idx);
   bool extract_collated_data();
+  bool check_account_failures();
   bool try_validate();
   bool compute_prev_state();
   bool compute_next_state();
@@ -395,13 +404,56 @@ class ValidateQuery : public td::actor::Actor {
                                        td::Bits256& msg_hash);
   bool check_in_queue();
   bool check_delivered_dequeued();
-  std::unique_ptr<block::Account> make_account_from(td::ConstBitPtr addr, Ref<vm::CellSlice> account);
-  std::unique_ptr<block::Account> unpack_account(td::ConstBitPtr addr);
-  bool check_one_transaction(block::Account& account, LogicalTime lt, Ref<vm::Cell> trans_root, bool is_first,
-                             bool is_last);
-  bool check_account_transactions(const StdSmcAddress& acc_addr, Ref<vm::CellSlice> acc_tr);
+
+  class CheckAccountTxs : public Actor {
+   public:
+    struct Context {
+      std::vector<std::tuple<Bits256, LogicalTime, LogicalTime>> msg_proc_lt{};
+      block::CurrencyCollection total_burned{0};
+      std::vector<std::tuple<Bits256, Bits256, bool>> lib_publishers{};
+      bool defer_all_messages = false;
+      std::vector<std::pair<td::Ref<vm::Cell>, td::uint32>> storage_stat_cache_update{};
+      ValidationStats::WorkTimeStats work_time{};
+
+      std::optional<td::Status> fatal_error;
+      std::optional<std::string> reject_error;
+      std::optional<td::BufferSlice> reject_reason;
+    };
+
+    CheckAccountTxs(const ValidateQuery& vq, td::actor::ActorId<ValidateQuery> vq_id, StdSmcAddress address,
+                    Ref<vm::CellSlice> acc_tr, Context ctx);
+
+    bool try_check();
+    Context extract_context();
+
+   private:
+    void start_up() override;
+
+    void abort_query(td::Status error);
+    bool reject_query(std::string error, td::BufferSlice reason = {});
+    bool reject_query(std::string err_msg, td::Status error, td::BufferSlice reason = {});
+    bool fatal_error(td::Status error);
+    bool fatal_error(std::string err_msg, int err_code = -666);
+
+    std::unique_ptr<block::Account> make_account_from(td::ConstBitPtr addr, Ref<vm::CellSlice> account);
+    std::unique_ptr<block::Account> unpack_account(td::ConstBitPtr addr);
+    bool check_one_transaction(block::Account& account, LogicalTime lt, Ref<vm::Cell> trans_root, bool is_first,
+                               bool is_last);
+    bool scan_account_libraries(Ref<vm::Cell> orig_libs, Ref<vm::Cell> final_libs, const td::Bits256& addr);
+
+    const ValidateQuery& vq_;
+    td::actor::ActorId<ValidateQuery> vq_id_;
+    StdSmcAddress address_;
+    Ref<vm::CellSlice> acc_tr_;
+    Context ctx_;
+  };
+  friend CheckAccountTxs;
+
+  CheckAccountTxs::Context load_check_account_transactions_context(const StdSmcAddress& address);
+  void save_account_transactions_context(const StdSmcAddress& address, CheckAccountTxs::Context ctx);
+
+  void after_check_account_finished(StdSmcAddress address, CheckAccountTxs::Context context);
   bool check_transactions();
-  bool scan_account_libraries(Ref<vm::Cell> orig_libs, Ref<vm::Cell> final_libs, const td::Bits256& addr);
   bool check_all_ticktock_processed();
   bool check_message_processing_order();
   bool check_special_message(Ref<vm::Cell> in_msg_root, const block::CurrencyCollection& amount,

validator/impl/validate-query.hpp

@@ -57,9 +57,21 @@ struct AsyncSerializerState {
 };
 
 struct StorageStatCacheStats {
-  td::uint64 small_cnt = 0, small_cells = 0;
-  td::uint64 hit_cnt = 0, hit_cells = 0;
-  td::uint64 miss_cnt = 0, miss_cells = 0;
+  std::atomic<td::uint64> small_cnt = 0, small_cells = 0;
+  std::atomic<td::uint64> hit_cnt = 0, hit_cells = 0;
+  std::atomic<td::uint64> miss_cnt = 0, miss_cells = 0;
+
+  StorageStatCacheStats() {
+  }
+
+  StorageStatCacheStats(const StorageStatCacheStats& other)
+      : small_cnt(other.small_cnt.load())
+      , small_cells(other.small_cells.load())
+      , hit_cnt(other.hit_cnt.load())
+      , hit_cells(other.hit_cells.load())
+      , miss_cnt(other.miss_cnt.load())
+      , miss_cells(other.miss_cells.load()) {
+  }
 
   tl_object_ptr<ton_api::validatorStats_storageStatCacheStats> tl() const {
     return create_tl_object<ton_api::validatorStats_storageStatCacheStats>(small_cnt, small_cells, hit_cnt, hit_cells,
@@ -180,6 +192,8 @@ struct ValidationStats {
   td::uint32 actual_bytes = 0, actual_collated_data_bytes = 0;
   double total_time = 0.0;
   std::string time_stats;
+  double actual_time = 0.0;
+  bool parallel_accounts_validation = false;
 
   struct WorkTimeStats {
     td::RealCpuTimer::Time total;
@@ -195,13 +209,14 @@ struct ValidationStats {
     }
   };
   WorkTimeStats work_time;
-  StorageStatCacheStats storage_stat_cache;
+  mutable StorageStatCacheStats storage_stat_cache;
 
   tl_object_ptr<ton_api::validatorStats_validatedBlock> tl() const {
     return create_tl_object<ton_api::validatorStats_validatedBlock>(
         create_tl_block_id(block_id), collated_data_hash, validated_at, self.bits256_value(), valid, comment,
-        actual_bytes, actual_collated_data_bytes, total_time, work_time.total.real, work_time.total.cpu, time_stats,
-        work_time.to_str(false), work_time.to_str(true), storage_stat_cache.tl());
+        actual_bytes, actual_collated_data_bytes, total_time, actual_time, work_time.total.real, work_time.total.cpu,
+        time_stats, work_time.to_str(false), work_time.to_str(true), storage_stat_cache.tl(),
+        parallel_accounts_validation);
   }
 };
 

validator/interfaces/validator-manager.h

@@ -248,7 +248,8 @@ void ValidatorGroup::validate_block_candidate(validatorsession::BlockSourceInfo
                                     .prev = std::move(prev),
                                     .validator_set = validator_set_,
                                     .local_validator_id = local_id_,
-                                    .optimistic_prev_block = optimistic_prev_block_data},
+                                    .optimistic_prev_block = optimistic_prev_block_data,
+                                    .parallel_validation = opts_.get()->get_parallel_validation()},
                      manager_, td::Timestamp::in(15.0), std::move(P));
 }
 

validator/validator-group.cpp

@@ -173,6 +173,9 @@ struct ValidatorManagerOptionsImpl : public ValidatorManagerOptions {
   td::Ref<ShardBlockVerifierConfig> get_shard_block_verifier_config() const override {
     return shard_block_verifier_config_;
   }
+  bool get_parallel_validation() const override {
+    return parallel_validation;
+  }
 
   void set_zero_block_id(BlockIdExt block_id) override {
     zero_block_id_ = block_id;
@@ -295,6 +298,10 @@ struct ValidatorManagerOptionsImpl : public ValidatorManagerOptions {
     shard_block_verifier_config_ = std::move(config);
   }
 
+  void set_parallel_validation(bool value) override {
+    parallel_validation = value;
+  }
+
   ValidatorManagerOptionsImpl* make_copy() const override {
     return new ValidatorManagerOptionsImpl(*this);
   }
@@ -352,6 +359,7 @@ struct ValidatorManagerOptionsImpl : public ValidatorManagerOptions {
   std::set<adnl::AdnlNodeIdShort> collator_node_whitelist_;
   bool collator_node_whitelist_enabled_ = false;
   td::Ref<ShardBlockVerifierConfig> shard_block_verifier_config_{true};
+  bool parallel_validation = false;
 };
 
 }  // namespace validator

validator/validator-options.hpp

@@ -144,6 +144,7 @@ struct ValidatorManagerOptions : public td::CntObject {
   virtual td::optional<double> get_catchain_max_block_delay_slow() const = 0;
   virtual bool get_state_serializer_enabled() const = 0;
   virtual td::Ref<CollatorOptions> get_collator_options() const = 0;
+  virtual bool get_parallel_validation() const = 0;
   virtual double get_catchain_broadcast_speed_multiplier() const = 0;
   virtual bool get_permanent_celldb() const = 0;
   virtual td::Ref<CollatorsList> get_collators_list() const = 0;
@@ -188,6 +189,7 @@ struct ValidatorManagerOptions : public td::CntObject {
   virtual void set_collator_node_whitelisted_validator(adnl::AdnlNodeIdShort id, bool add) = 0;
   virtual void set_collator_node_whitelist_enabled(bool enabled) = 0;
   virtual void set_shard_block_verifier_config(td::Ref<ShardBlockVerifierConfig> config) = 0;
+  virtual void set_parallel_validation(bool value) = 0;
 
   static td::Ref<ValidatorManagerOptions> create(BlockIdExt zero_block_id, BlockIdExt init_block_id,
                                                  bool allow_blockchain_init = false, double sync_blocks_before = 3600,