Merge pull request #1779 from ton-blockchain/optimize-collate

Optimize optimistic collation

Author: EmelyanenkoK

blockchain-explorer/blockchain-explorer-query.cpp

@@ -525,8 +525,8 @@ HttpQueryBlockSearch::HttpQueryBlockSearch(std::map<std::string, std::string> op
   }
   if (opts.count("utime") == 1) {
     try {
-      seqno_ = static_cast<td::uint32>(std::stoull(opts["utime"]));
-      mode_ = 1;
+      utime_ = static_cast<td::uint32>(std::stoull(opts["utime"]));
+      mode_ = 4;
     } catch (...) {
       error_ = td::Status::Error("cannot parse utime");
       return;

crypto/vm/boc-compression.cpp

@@ -283,7 +283,7 @@ td::Result<td::BufferSlice> boc_compress_improved_structure_lz4(const std::vecto
         continue;
 
       int delta = rank[boc_graph[node][j]] - i - 2; // Always >= 0 because of above check
-      size_t required_bits = 1 + (31 ^ __builtin_clz(node_count - i - 3));
+      size_t required_bits = 1 + (31 ^ td::count_leading_zeroes32(node_count - i - 3));
 
       if (required_bits < 8 - (result.size() + 1) % 8 + 1) {
         append_uint(result, delta, required_bits);
@@ -494,7 +494,7 @@ td::Result<std::vector<td::Ref<vm::Cell>>> boc_decompress_improved_structure_lz4
     for (int j = 0; j < cell_refs_cnt[i]; ++j) {
       if (!boc_graph[i][j]) {
         size_t pref_size = (orig_size - bit_reader.size());
-        size_t required_bits = 1 + (31 ^ __builtin_clz(node_count - i - 3));
+        size_t required_bits = 1 + (31 ^ td::count_leading_zeroes32(node_count - i - 3));
 
         if (required_bits < 8 - (pref_size + 1) % 8 + 1) {
           TRY_RESULT_ASSIGN(boc_graph[i][j], read_uint(bit_reader, required_bits));

tl/generate/scheme/ton_api.tl

@@ -962,10 +962,6 @@ validatorStats.blockStats
     ext_msgs:validatorStats.blockStats.extMsgsStats transactions:int shard_configuration:(vector tonNode.blockIdExt)
     old_out_msg_queue_size:long new_out_msg_queue_size:long msg_queue_cleaned:int
     neighbors:(vector validatorStats.blockStats.neighborStats) = validatorStats.BlockStats;
-validatorStats.collateWorkTimeStats
-    total:double optimistic_apply:double queue_cleanup:double prelim_storage_stat:double trx_tvm:double trx_storage_stat:double
-    trx_other:double final_storage_stat:double create_block:double create_collated_data:double create_block_candidate:double
-    = validatorStats.CollateWorkTimeStats;
 validatorStats.storageStatCacheStats
     small_cnt:long small_cells:long hit_cnt:long hit_cells:long miss_cnt:long miss_cells:long = validatorStats.StorageStatCacheStats;
 
@@ -974,22 +970,20 @@ validatorStats.collatedBlock
     bytes:int collated_data_bytes:int attempt:int
     self:int256 is_validator:Bool
     total_time:double work_time:double cpu_work_time:double time_stats:string
-    work_time_real_stats:validatorStats.collateWorkTimeStats
-    work_time_cpu_stats:validatorStats.collateWorkTimeStats
+    work_time_real_stats:string
+    work_time_cpu_stats:string
     block_limits:validatorStats.blockLimitsStatus
     block_stats:validatorStats.blockStats
     storage_stat_cache:validatorStats.storageStatCacheStats = validatorSession.stats.CollatedBlock;
 
-validatorStats.validateWorkTimeStats
-    total:double trx_tvm:double trx_storage_stat:double trx_other:double = validatorStats.ValidateWorkTimeStats;
 validatorStats.validatedBlock
     block_id:tonNode.blockIdExt collated_data_hash:int256 validated_at:double
     self:int256
     valid:Bool comment:string
     bytes:int collated_data_bytes:int
-    total_time:double work_time:double cpu_work_time:double
-    work_time_real_stats:validatorStats.validateWorkTimeStats
-    work_time_cpu_stats:validatorStats.validateWorkTimeStats
+    total_time:double work_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;
 
 validatorStats.newValidatorGroup.node id:int256 pubkey:PublicKey adnl_id:int256 weight:long = validatorStats.newValidatorGroup.Node;
@@ -1008,7 +1002,7 @@ validatorStats.collatorNodeResponse self:int256 validator_id:int256 timestamp:do
 collatorNode.candidate source:PublicKey id:tonNode.blockIdExt data:bytes collated_data:bytes = collatorNode.Candidate;
 collatorNode.compressedCandidate flags:# source:PublicKey id:tonNode.blockIdExt decompressed_size:int data:bytes = collatorNode.Candidate;
 collatorNode.compressedCandidateV2 flags:# source:PublicKey id:tonNode.blockIdExt data:bytes = collatorNode.Candidate;
-collatorNode.pong flags:# = collatorNode.Pong;
+collatorNode.pong#5bbf0521 flags:# version:flags.0?int = collatorNode.Pong;
 collatorNode.error code:int message:string = collatorNode.Error;
 
 shardBlockVerifier.subscribed flags:# = shardBlockVerifier.Subscribed;
@@ -1017,6 +1011,9 @@ shardBlockVerifier.confirmBlocks blocks:(vector tonNode.blockIdExt) = shardBlock
 ---functions---
 collatorNode.generateBlock shard:tonNode.shardId cc_seqno:int prev_blocks:(vector tonNode.blockIdExt)
     creator:int256 round:int first_block_round:int priority:int = collatorNode.Candidate;
+collatorNode.generateBlockOptimistic shard:tonNode.shardId cc_seqno:int prev_blocks:(vector tonNode.blockIdExt)
+    creator:int256 round:int first_block_round:int priority:int = collatorNode.Candidate;
+collatorNode.requestBlockCallback flags:# block_id:tonNode.BlockIdExt = collatorNode.Candidate;
 collatorNode.ping flags:# = collatorNode.Pong;
 
 shardBlockVerifier.subscribe shard:tonNode.shardId flags:# = shardBlockVerifier.Subscribed;

tl/generate/scheme/ton_api.tlo

@@ -411,6 +411,8 @@ struct Ed25519_PrivateKey {
 
 struct Ed25519_PublicKey {
   Bits256 _pubkey;
+  Ed25519_PublicKey() : _pubkey(td::Bits256::zero()) {
+  }
   explicit Ed25519_PublicKey(const Bits256& x) : _pubkey(x) {
   }
   explicit Ed25519_PublicKey(const td::ConstBitPtr x) : _pubkey(x) {

ton/ton-types.h

@@ -26,4 +26,4 @@ target_include_directories(validatorsession PUBLIC
   $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}>/.. 
   ${OPENSSL_INCLUDE_DIR}
 )
-target_link_libraries(validatorsession PRIVATE tdutils tdactor adnl rldp tl_api dht tdfec overlay catchain)
+target_link_libraries(validatorsession PRIVATE tdutils tdactor adnl rldp2 tl_api dht tdfec overlay catchain)

validator-session/CMakeLists.txt

@@ -230,7 +230,7 @@ struct NewValidatorGroupStats {
   std::vector<BlockIdExt> prev;
   td::uint32 self_idx = 0;
   PublicKeyHash self = PublicKeyHash::zero();
-  std::vector<Node> nodes;
+  std::vector<Node> nodes{};
 
   tl_object_ptr<ton_api::validatorStats_newValidatorGroup> tl() const {
     std::vector<tl_object_ptr<ton_api::tonNode_blockIdExt>> prev_arr;
@@ -257,7 +257,7 @@ struct EndValidatorGroupStats {
   ValidatorSessionId session_id = ValidatorSessionId::zero();
   double timestamp = -1.0;
   PublicKeyHash self = PublicKeyHash::zero();
-  std::vector<Node> nodes;
+  std::vector<Node> nodes{};
 
   tl_object_ptr<ton_api::validatorStats_endValidatorGroup> tl() const {
     std::vector<tl_object_ptr<ton_api::validatorStats_endValidatorGroup_node>> nodes_arr;

validator-session/validator-session-types.h

@@ -294,13 +294,24 @@ void ValidatorSessionImpl::process_broadcast(PublicKeyHash src, td::BufferSlice
                                       << "]: duplicate optimistic broadcast for round " << block_round;
       return;
     }
+    int priority = description().get_node_priority(src_idx, block_round);
+    if (priority < 0) {
+      VLOG(VALIDATOR_SESSION_WARNING) << this << "[node " << src << "][broadcast " << sha256_bits256(data.as_slice())
+                                      << "]: node is not a producer in round " << block_round;
+      return;
+    }
     if (block_round > cur_round_) {
       OptimisticBroadcast &optimistic_broadcast = optimistic_broadcasts_[{block_round, src_idx}];
       optimistic_broadcast.candidate = std::move(candidate);
       optimistic_broadcast.prev_candidate_id = optimistic_prev_candidate;
       optimistic_broadcast.broadcast_info = broadcast_info;
       VLOG(VALIDATOR_SESSION_WARNING) << this << ": received optimistic broadcast " << block_id << " from " << src
                                       << ", round " << block_round;
+      validate_optimistic_broadcast(
+          BlockSourceInfo{description().get_source_public_key(src_idx),
+                          BlockCandidatePriority{block_round, block_round, priority}},
+          optimistic_broadcast.candidate->root_hash_, optimistic_broadcast.candidate->data_.clone(),
+          optimistic_broadcast.candidate->collated_data_.clone(), optimistic_broadcast.prev_candidate_id);
       return;
     }
     if (SentBlock::get_block_id(real_state_->get_committed_block(description(), block_round - 1)) !=
@@ -336,7 +347,7 @@ void ValidatorSessionImpl::process_received_block(td::uint32 block_round, Public
       }
     }
     stat->deserialize_time = info.deserialize_time;
-    stat->serialized_size = info.serialized_size;
+    stat->serialized_size = (int)info.serialized_size;
     stat->block_id.root_hash = candidate->root_hash_;
     stat->block_id.file_hash = info.file_hash;
     stat->collated_data_hash = info.collated_data_hash;
@@ -361,6 +372,12 @@ void ValidatorSessionImpl::process_received_block(td::uint32 block_round, Public
   }
 
   blocks_[block_id] = std::move(candidate);
+  if (auto it = block_waiters_.find(block_id); it != block_waiters_.end()) {
+    for (auto &promise : it->second) {
+      promise.set_result(td::Unit());
+    }
+    block_waiters_.erase(it);
+  }
 
   VLOG(VALIDATOR_SESSION_WARNING) << this << ": received broadcast " << block_id;
   if (block_round != cur_round_) {
@@ -381,6 +398,37 @@ void ValidatorSessionImpl::process_received_block(td::uint32 block_round, Public
   }
 }
 
+void ValidatorSessionImpl::validate_optimistic_broadcast(BlockSourceInfo source_info,
+                                                         ValidatorSessionRootHash root_hash, td::BufferSlice data,
+                                                         td::BufferSlice collated_data,
+                                                         ValidatorSessionCandidateId prev_candidate_id) {
+  if (source_info.priority.round <= cur_round_) {
+    VLOG(VALIDATOR_SESSION_DEBUG) << this << ": validate optimistic broadcast from "
+                                  << source_info.source.compute_short_id() << " : too old";
+    return;
+  }
+  auto it = blocks_.find(prev_candidate_id);
+  if (it == blocks_.end()) {
+    VLOG(VALIDATOR_SESSION_DEBUG) << this << ": validate optimistic broadcast from "
+                                  << source_info.source.compute_short_id() << " : wait for prev block";
+    block_waiters_[prev_candidate_id].push_back(
+        [=, SelfId = actor_id(this), data = std::move(data),
+         collated_data = std::move(collated_data)](td::Result<td::Unit> R) mutable {
+          if (R.is_ok()) {
+            td::actor::send_closure(SelfId, &ValidatorSessionImpl::validate_optimistic_broadcast, source_info,
+                                    root_hash, std::move(data), std::move(collated_data), prev_candidate_id);
+          }
+        });
+    return;
+  }
+  VLOG(VALIDATOR_SESSION_DEBUG) << this << ": validate optimistic broadcast from "
+                                << source_info.source.compute_short_id();
+  callback_->on_optimistic_candidate(
+      source_info, root_hash, std::move(data), std::move(collated_data),
+      description().get_source_public_key(description().get_source_idx(PublicKeyHash{it->second->src_})),
+      it->second->root_hash_, it->second->data_.clone(), it->second->collated_data_.clone());
+}
+
 void ValidatorSessionImpl::process_message(PublicKeyHash src, td::BufferSlice data) {
 }
 
@@ -473,7 +521,7 @@ void ValidatorSessionImpl::candidate_decision_ok(td::uint32 round, ValidatorSess
     stat->block_status = ValidatorSessionStats::status_approved;
     stat->comment = PSTRING() << "ts=" << ok_from;
     stat->validation_time = validation_time;
-    stat->gen_utime = (double)ok_from;
+    stat->gen_utime = (int)ok_from;
     stat->validated_at = td::Clocks::system();
     stat->validation_cached = validation_cached;
   }
@@ -528,7 +576,11 @@ void ValidatorSessionImpl::generated_block(td::uint32 round, GeneratedCandidate
     stat->block_status = ValidatorSessionStats::status_received;
     stat->collation_time = collation_time;
     stat->collated_at = td::Clocks::system();
-    stat->got_block_at = td::Clocks::system();
+    if (auto it = sent_candidates_.find(block_id); it != sent_candidates_.end()) {
+      stat->got_block_at = it->second.sent_at;
+    } else {
+      stat->got_block_at = td::Clocks::system();
+    }
     stat->got_block_by = ValidatorSessionStats::recv_collated;
     stat->collation_cached = c.is_cached;
     stat->self_collated = c.self_collated;
@@ -548,6 +600,12 @@ void ValidatorSessionImpl::generated_block(td::uint32 round, GeneratedCandidate
   blocks_[block_id] = create_tl_object<ton_api::validatorSession_candidate>(
       local_id().tl(), round, c.candidate.id.root_hash, std::move(c.candidate.data),
       std::move(c.candidate.collated_data));
+  if (auto it = block_waiters_.find(block_id); it != block_waiters_.end()) {
+    for (auto &promise : it->second) {
+      promise.set_result(td::Unit());
+    }
+    block_waiters_.erase(it);
+  }
   pending_generate_ = false;
   generated_ = true;
   generated_block_ = block_id;
@@ -738,7 +796,6 @@ void ValidatorSessionImpl::try_approve_block(const SentBlock *block) {
                   VLOG(VALIDATOR_SESSION_WARNING) << print_id << ": failed to get candidate " << hash << " from " << id
                                                   << ": " << R.move_as_error();
                 } else {
-                  LOG(ERROR) << "QQQQQ Got block " << R.ok().size();
                   td::actor::send_closure(SelfId, &ValidatorSessionImpl::process_broadcast, src_id, R.move_as_ok(),
                                           candidate_id, false, false);
                 }
@@ -1082,7 +1139,7 @@ ValidatorSessionImpl::ValidatorSessionImpl(catchain::CatChainSessionId session_i
                                            PublicKeyHash local_id, std::vector<ValidatorSessionNode> nodes,
                                            std::unique_ptr<Callback> callback,
                                            td::actor::ActorId<keyring::Keyring> keyring,
-                                           td::actor::ActorId<adnl::Adnl> adnl, td::actor::ActorId<rldp::Rldp> rldp,
+                                           td::actor::ActorId<adnl::Adnl> adnl, td::actor::ActorId<rldp2::Rldp> rldp,
                                            td::actor::ActorId<overlay::Overlays> overlays, std::string db_root,
                                            std::string db_suffix, bool allow_unsafe_self_blocks_resync)
     : unique_hash_(session_id)
@@ -1206,7 +1263,6 @@ void ValidatorSessionImpl::start_up() {
   virtual_state_ = real_state_;
 
   check_all();
-  td::actor::send_closure(rldp_, &rldp::Rldp::add_id, description().get_source_adnl_id(local_idx()));
 }
 
 void ValidatorSessionImpl::stats_init() {
@@ -1365,25 +1421,49 @@ void ValidatorSessionImpl::process_approve(td::uint32 node_id, td::uint32 round,
   bool is_approved_66pct = stat->approved_66pct_at > 0.0;
 
   if (allow_optimistic_generation_ && !was_approved_66pct && is_approved_66pct && cur_round_ == round &&
-      description().get_node_priority(local_idx(), round + 1) == 0 && blocks_.contains(candidate_id)) {
-    auto &block = blocks_[candidate_id];
-    if (cur_round_ == first_block_round_ &&
-        description().get_node_priority(description().get_source_idx(PublicKeyHash{block->src_}), cur_round_) == 0) {
-      callback_->generate_block_optimistic(BlockSourceInfo{description().get_source_public_key(local_idx()),
-                                                           BlockCandidatePriority{round + 1, round + 1, 0}},
-                                           block->data_.clone(), block->root_hash_, stat->block_id.file_hash,
-                                           [=, SelfId = actor_id(this)](td::Result<GeneratedCandidate> R) {
-                                             if (R.is_error()) {
-                                               return;
-                                             }
-                                             td::actor::send_closure(
-                                                 SelfId, &ValidatorSessionImpl::generated_optimistic_candidate,
-                                                 round + 1, R.move_as_ok(), candidate_id);
-                                           });
+      cur_round_ == first_block_round_ && description().get_node_priority(local_idx(), round + 1) == 0 &&
+      blocks_.contains(candidate_id) &&
+      description().get_node_priority(description().get_source_idx(stat->validator_id), cur_round_) == 0) {
+    if (blocks_.contains(candidate_id)) {
+      generate_block_optimistic(round, candidate_id);
+    } else {
+      block_waiters_[candidate_id].push_back([=, SelfId = actor_id(this)](td::Result<td::Unit> R) {
+        if (R.is_ok()) {
+          td::actor::send_closure(SelfId, &ValidatorSessionImpl::generate_block_optimistic, round, candidate_id);
+        }
+      });
     }
   }
 }
 
+void ValidatorSessionImpl::generate_block_optimistic(td::uint32 cur_round,
+                                                     ValidatorSessionCandidateId prev_candidate_id) {
+  if (cur_round != cur_round_) {
+    return;
+  }
+  auto it = blocks_.find(prev_candidate_id);
+  if (it == blocks_.end()) {
+    return;
+  }
+  auto &block = it->second;
+  auto stat = stats_get_candidate_stat_by_id(cur_round, prev_candidate_id);
+  if (!stat) {
+    return;
+  }
+  callback_->generate_block_optimistic(BlockSourceInfo{description().get_source_public_key(local_idx()),
+                                                       BlockCandidatePriority{cur_round + 1, cur_round + 1, 0}},
+                                       block->data_.clone(), block->root_hash_, stat->block_id.file_hash,
+                                       [=, SelfId = actor_id(this)](td::Result<GeneratedCandidate> R) {
+                                         if (R.is_error()) {
+                                           LOG(DEBUG) << "Optimistic generation error: " << R.move_as_error();
+                                           return;
+                                         }
+                                         td::actor::send_closure(SelfId,
+                                                                 &ValidatorSessionImpl::generated_optimistic_candidate,
+                                                                 cur_round + 1, R.move_as_ok(), prev_candidate_id);
+                                       });
+}
+
 void ValidatorSessionImpl::generated_optimistic_candidate(td::uint32 round, GeneratedCandidate candidate,
                                                           ValidatorSessionCandidateId prev_candidate) {
   if (cur_round_ > round) {
@@ -1396,7 +1476,7 @@ td::actor::ActorOwn<ValidatorSession> ValidatorSession::create(
     catchain::CatChainSessionId session_id, ValidatorSessionOptions opts, PublicKeyHash local_id,
     std::vector<ValidatorSessionNode> nodes, std::unique_ptr<Callback> callback,
     td::actor::ActorId<keyring::Keyring> keyring, td::actor::ActorId<adnl::Adnl> adnl,
-    td::actor::ActorId<rldp::Rldp> rldp, td::actor::ActorId<overlay::Overlays> overlays, std::string db_root,
+    td::actor::ActorId<rldp2::Rldp> rldp, td::actor::ActorId<overlay::Overlays> overlays, std::string db_root,
     std::string db_suffix, bool allow_unsafe_self_blocks_resync) {
   return td::actor::create_actor<ValidatorSessionImpl>("session", session_id, std::move(opts), local_id,
                                                        std::move(nodes), std::move(callback), keyring, adnl, rldp,