Improve CollatorNode

* Keep track of validator groups
* Pre-generate shard blocks

validator/impl/collator-impl.h

@@ -87,9 +87,9 @@ class Collator final : public td::actor::Actor {
   static constexpr bool shard_splitting_enabled = true;
 
  public:
-  Collator(ShardIdFull shard, bool is_hardfork, BlockIdExt min_masterchain_block_id,
-           std::vector<BlockIdExt> prev, Ref<ValidatorSet> validator_set, Ed25519_PublicKey collator_id,
-           td::actor::ActorId<ValidatorManager> manager, td::Timestamp timeout, td::Promise<BlockCandidate> promise,
+  Collator(ShardIdFull shard, bool is_hardfork, BlockIdExt min_masterchain_block_id, std::vector<BlockIdExt> prev,
+           Ref<ValidatorSet> validator_set, Ed25519_PublicKey collator_id, td::actor::ActorId<ValidatorManager> manager,
+           td::Timestamp timeout, td::Promise<BlockCandidate> promise, td::CancellationToken cancellation_token,
            unsigned mode);
   ~Collator() override = default;
   bool is_busy() const {
@@ -343,6 +343,9 @@ class Collator final : public td::actor::Actor {
   void return_block_candidate(td::Result<td::Unit> saved);
   bool update_last_proc_int_msg(const std::pair<ton::LogicalTime, ton::Bits256>& new_lt_hash);
 
+  td::CancellationToken cancellation_token_;
+  bool check_cancelled();
+
  public:
   static td::uint32 get_skip_externals_queue_size();
 };

validator/impl/collator.cpp

@@ -77,7 +77,7 @@ static inline bool dbg(int c) {
 Collator::Collator(ShardIdFull shard, bool is_hardfork, BlockIdExt min_masterchain_block_id,
                    std::vector<BlockIdExt> prev, td::Ref<ValidatorSet> validator_set, Ed25519_PublicKey collator_id,
                    td::actor::ActorId<ValidatorManager> manager, td::Timestamp timeout,
-                   td::Promise<BlockCandidate> promise, unsigned mode)
+                   td::Promise<BlockCandidate> promise, td::CancellationToken cancellation_token, unsigned mode)
     : shard_(shard)
     , is_hardfork_(is_hardfork)
     , min_mc_block_id{min_masterchain_block_id}
@@ -92,9 +92,11 @@ Collator::Collator(ShardIdFull shard, bool is_hardfork, BlockIdExt min_mastercha
     , medium_timeout_(td::Timestamp::at(timeout.at() - 1.5))
     , main_promise(std::move(promise))
     , mode_(mode)
-    , perf_timer_("collate", 0.1, [manager](double duration) {
-      send_closure(manager, &ValidatorManager::add_perf_timer_stat, "collate", duration);
-    }) {
+    , perf_timer_("collate", 0.1,
+                  [manager](double duration) {
+                    send_closure(manager, &ValidatorManager::add_perf_timer_stat, "collate", duration);
+                  })
+    , cancellation_token_(std::move(cancellation_token)) {
 }
 
 /**
@@ -382,6 +384,9 @@ bool Collator::fatal_error(std::string err_msg, int err_code) {
  */
 void Collator::check_pending() {
   // LOG(DEBUG) << "pending = " << pending;
+  if (!check_cancelled()) {
+    return;
+  }
   if (!pending) {
     step = 2;
     try {
@@ -2423,6 +2428,9 @@ bool Collator::out_msg_queue_cleanup() {
         LOG(WARNING) << "cleaning up outbound queue takes too long, ending";
         break;
       }
+      if (!check_cancelled()) {
+        return false;
+      }
       if (i == queue_parts.size()) {
         i = 0;
       }
@@ -3516,6 +3524,9 @@ bool Collator::process_inbound_internal_messages() {
       LOG(WARNING) << "soft timeout reached, stop processing inbound internal messages";
       break;
     }
+    if (!check_cancelled()) {
+      return false;
+    }
     LOG(DEBUG) << "processing inbound message with (lt,hash)=(" << kv->lt << "," << kv->key.to_hex()
                << ") from neighbor #" << kv->source;
     if (verbosity > 2) {
@@ -3565,6 +3576,9 @@ bool Collator::process_inbound_external_messages() {
       LOG(WARNING) << "medium timeout reached, stop processing inbound external messages";
       break;
     }
+    if (!check_cancelled()) {
+      return false;
+    }
     auto ext_msg = ext_msg_struct.cell;
     ton::Bits256 hash{ext_msg->get_hash().bits()};
     int r = process_external_message(std::move(ext_msg));
@@ -3819,6 +3833,9 @@ bool Collator::process_new_messages(bool enqueue_only) {
       LOG(INFO) << "BLOCK FULL, enqueue all remaining new messages";
       enqueue_only = true;
     }
+    if (!check_cancelled()) {
+      return false;
+    }
     LOG(DEBUG) << "have message with lt=" << msg.lt;
     int res = process_one_new_message(std::move(msg), enqueue_only);
     if (res < 0) {
@@ -5360,6 +5377,18 @@ void Collator::after_get_external_messages(td::Result<std::vector<std::pair<Ref<
   check_pending();
 }
 
+/**
+ * Checks if collation was cancelled via cancellation token
+ *
+ * @returns false if the collation was cancelled, true otherwise
+ */
+bool Collator::check_cancelled() {
+  if (cancellation_token_) {
+    return fatal_error(td::Status::Error(ErrorCode::cancelled, "cancelled"));
+  }
+  return true;
+}
+
 td::uint32 Collator::get_skip_externals_queue_size() {
   return SKIP_EXTERNALS_QUEUE_SIZE;
 }

validator/impl/fabric.cpp

@@ -215,7 +215,7 @@ void run_validate_query(ShardIdFull shard, BlockIdExt min_masterchain_block_id,
 void run_collate_query(ShardIdFull shard, const BlockIdExt& min_masterchain_block_id, std::vector<BlockIdExt> prev,
                        Ed25519_PublicKey creator, td::Ref<ValidatorSet> validator_set,
                        td::actor::ActorId<ValidatorManager> manager, td::Timestamp timeout,
-                       td::Promise<BlockCandidate> promise, unsigned mode) {
+                       td::Promise<BlockCandidate> promise, td::CancellationToken cancellation_token, unsigned mode) {
   BlockSeqno seqno = 0;
   for (auto& p : prev) {
     if (p.seqno() > seqno) {
@@ -224,7 +224,8 @@ void run_collate_query(ShardIdFull shard, const BlockIdExt& min_masterchain_bloc
   }
   td::actor::create_actor<Collator>(PSTRING() << "collate" << shard.to_str() << ":" << (seqno + 1), shard, false,
                                     min_masterchain_block_id, std::move(prev), std::move(validator_set), creator,
-                                    std::move(manager), timeout, std::move(promise), mode)
+                                    std::move(manager), timeout, std::move(promise), std::move(cancellation_token),
+                                    mode)
       .release();
 }
 
@@ -240,7 +241,7 @@ void run_collate_hardfork(ShardIdFull shard, const BlockIdExt& min_masterchain_b
   td::actor::create_actor<Collator>(PSTRING() << "collate" << shard.to_str() << ":" << (seqno + 1), shard, true,
                                     min_masterchain_block_id, std::move(prev), td::Ref<ValidatorSet>{},
                                     Ed25519_PublicKey{Bits256::zero()}, std::move(manager), timeout, std::move(promise),
-                                    0)
+                                    td::CancellationToken{}, 0)
       .release();
 }