Improve Collator::opt_msg_queue_cleanup, increase collator timeout

SpyCheese · Dec 6, 2023 · cbbd4f7 · cbbd4f7
1 parent 405e5ae
commit cbbd4f7
Show file tree

Hide file tree

Showing 4 changed files with 40 additions and 22 deletions.
diff --git a/crypto/vm/dict.cpp b/crypto/vm/dict.cpp
@@ -21,6 +21,7 @@
 #include "vm/cellslice.h"
 #include "vm/stack.hpp"
 #include "common/bitstring.h"
+#include "td/utils/Random.h"
 
 #include "td/utils/bits.h"
 
@@ -2007,7 +2008,7 @@ bool DictionaryFixed::combine_with(DictionaryFixed& dict2) {
 
 bool DictionaryFixed::dict_check_for_each(Ref<Cell> dict, td::BitPtr key_buffer, int n, int total_key_len,
                                           const DictionaryFixed::foreach_func_t& foreach_func,
-                                          bool invert_first) const {
+                                          bool invert_first, bool shuffle) const {
   if (dict.is_null()) {
     return true;
   }
@@ -2026,26 +2027,29 @@ bool DictionaryFixed::dict_check_for_each(Ref<Cell> dict, td::BitPtr key_buffer,
   key_buffer += l + 1;
   if (l) {
     invert_first = false;
-  } else if (invert_first) {
+  }
+  bool invert = shuffle ? td::Random::fast(0, 1) == 1: invert_first;
+  if (invert) {
     std::swap(c1, c2);
   }
-  key_buffer[-1] = invert_first;
+  key_buffer[-1] = invert;
   // recursive check_foreach applied to both children
-  if (!dict_check_for_each(std::move(c1), key_buffer, n - l - 1, total_key_len, foreach_func)) {
+  if (!dict_check_for_each(std::move(c1), key_buffer, n - l - 1, total_key_len, foreach_func, false, shuffle)) {
     return false;
   }
-  key_buffer[-1] = !invert_first;
-  return dict_check_for_each(std::move(c2), key_buffer, n - l - 1, total_key_len, foreach_func);
+  key_buffer[-1] = !invert;
+  return dict_check_for_each(std::move(c2), key_buffer, n - l - 1, total_key_len, foreach_func, false, shuffle);
 }
 
-bool DictionaryFixed::check_for_each(const foreach_func_t& foreach_func, bool invert_first) {
+bool DictionaryFixed::check_for_each(const foreach_func_t& foreach_func, bool invert_first, bool shuffle) {
   force_validate();
   if (is_empty()) {
     return true;
   }
   int key_len = get_key_bits();
   unsigned char key_buffer[max_key_bytes];
-  return dict_check_for_each(get_root_cell(), td::BitPtr{key_buffer}, key_len, key_len, foreach_func, invert_first);
+  return dict_check_for_each(get_root_cell(), td::BitPtr{key_buffer}, key_len, key_len, foreach_func, invert_first,
+                             shuffle);
 }
 
 static inline bool set_bit(td::BitPtr ptr, bool value = true) {

diff --git a/crypto/vm/dict.h b/crypto/vm/dict.h
@@ -223,7 +223,7 @@ class DictionaryFixed : public DictionaryBase {
   int get_common_prefix(td::BitPtr buffer, unsigned buffer_len);
   bool cut_prefix_subdict(td::ConstBitPtr prefix, int prefix_len, bool remove_prefix = false);
   Ref<vm::Cell> extract_prefix_subdict_root(td::ConstBitPtr prefix, int prefix_len, bool remove_prefix = false);
-  bool check_for_each(const foreach_func_t& foreach_func, bool invert_first = false);
+  bool check_for_each(const foreach_func_t& foreach_func, bool invert_first = false, bool shuffle = false);
   int filter(filter_func_t check);
   bool combine_with(DictionaryFixed& dict2, const combine_func_t& combine_func, int mode = 0);
   bool combine_with(DictionaryFixed& dict2, const simple_combine_func_t& simple_combine_func, int mode = 0);
@@ -292,7 +292,7 @@ class DictionaryFixed : public DictionaryBase {
   std::pair<Ref<Cell>, bool> extract_prefix_subdict_internal(Ref<Cell> dict, td::ConstBitPtr prefix, int prefix_len,
                                                              bool remove_prefix = false) const;
   bool dict_check_for_each(Ref<Cell> dict, td::BitPtr key_buffer, int n, int total_key_len,
-                           const foreach_func_t& foreach_func, bool invert_first = false) const;
+                           const foreach_func_t& foreach_func, bool invert_first = false, bool shuffle = false) const;
   std::pair<Ref<Cell>, int> dict_filter(Ref<Cell> dict, td::BitPtr key, int n, const filter_func_t& check_leaf,
                                         int& skip_rest) const;
   Ref<Cell> dict_combine_with(Ref<Cell> dict1, Ref<Cell> dict2, td::BitPtr key_buffer, int n, int total_key_len,

diff --git a/validator/impl/collator.cpp b/validator/impl/collator.cpp
@@ -81,8 +81,8 @@ Collator::Collator(ShardIdFull shard, bool is_hardfork, UnixTime min_ts, BlockId
     , validator_set_(std::move(validator_set))
     , manager(manager)
     , timeout(timeout)
-    // default timeout is 10 seconds, declared in validator/validator-group.cpp:generate_block_candidate:run_collate_query
-    , queue_cleanup_timeout_(td::Timestamp::at(timeout.at() - 5.0))
+    // default timeout is 25 seconds, declared in validator/validator-group.cpp:generate_block_candidate:run_collate_query
+    , queue_cleanup_timeout_(td::Timestamp::at(timeout.at() - 10.0))
     , soft_timeout_(td::Timestamp::at(timeout.at() - 3.0))
     , medium_timeout_(td::Timestamp::at(timeout.at() - 1.5))
     , main_promise(std::move(promise))
@@ -2175,18 +2175,28 @@ bool Collator::out_msg_queue_cleanup() {
     }
   }
 
-  auto res = out_msg_queue_->filter([&](vm::CellSlice& cs, td::ConstBitPtr key, int n) -> int {
+  auto queue_root = out_msg_queue_->get_root_cell();
+  if (queue_root.is_null()) {
+    LOG(DEBUG) << "out_msg_queue is empty";
+    return true;
+  }
+  auto old_out_msg_queue = std::make_unique<vm::AugmentedDictionary>(queue_root, 352, block::tlb::aug_OutMsgQueue);
+
+  int deleted = 0;
+  bool fail = false;
+  old_out_msg_queue->check_for_each([&](Ref<vm::CellSlice> value, td::ConstBitPtr key, int n) -> bool {
     assert(n == 352);
+    vm::CellSlice& cs = value.write();
     // LOG(DEBUG) << "key is " << key.to_hex(n);
     if (queue_cleanup_timeout_.is_in_past(td::Timestamp::now())) {
       LOG(WARNING) << "cleaning up outbound queue takes too long, ending";
       outq_cleanup_partial_ = true;
-      return (1 << 30) + 1;  // retain all remaining outbound queue entries including this one without processing
+      return false;  // retain all remaining outbound queue entries including this one without processing
     }
     if (block_full_) {
       LOG(WARNING) << "BLOCK FULL while cleaning up outbound queue, cleanup completed only partially";
       outq_cleanup_partial_ = true;
-      return (1 << 30) + 1;  // retain all remaining outbound queue entries including this one without processing
+      return false;  // retain all remaining outbound queue entries including this one without processing
     }
     block::EnqueuedMsgDescr enq_msg_descr;
     unsigned long long created_lt;
@@ -2195,7 +2205,8 @@ bool Collator::out_msg_queue_cleanup() {
           && enq_msg_descr.check_key(key)      // check key
           && enq_msg_descr.lt_ == created_lt)) {
       LOG(ERROR) << "cannot unpack EnqueuedMsg with key " << key.to_hex(n);
-      return -1;
+      fail = true;
+      return false;
     }
     LOG(DEBUG) << "scanning outbound message with (lt,hash)=(" << enq_msg_descr.lt_ << ","
                << enq_msg_descr.hash_.to_hex() << ") enqueued_lt=" << enq_msg_descr.enqueued_lt_;
@@ -2213,20 +2224,23 @@ bool Collator::out_msg_queue_cleanup() {
     if (delivered) {
       LOG(DEBUG) << "outbound message with (lt,hash)=(" << enq_msg_descr.lt_ << "," << enq_msg_descr.hash_.to_hex()
                  << ") enqueued_lt=" << enq_msg_descr.enqueued_lt_ << " has been already delivered, dequeueing";
+      ++deleted;
+      out_msg_queue_->lookup_delete_with_extra(key, n);
       if (!dequeue_message(std::move(enq_msg_descr.msg_env_), deliver_lt)) {
         fatal_error(PSTRING() << "cannot dequeue outbound message with (lt,hash)=(" << enq_msg_descr.lt_ << ","
                               << enq_msg_descr.hash_.to_hex() << ") by inserting a msg_export_deq record");
-        return -1;
+        fail = true;
+        return false;
       }
       register_out_msg_queue_op();
       if (!block_limit_status_->fits(block::ParamLimits::cl_normal)) {
         block_full_ = true;
       }
     }
-    return !delivered;
-  });
-  LOG(DEBUG) << "deleted " << res << " messages from out_msg_queue";
-  if (res < 0) {
+    return true;
+  }, false, true /* random order */);
+  LOG(INFO) << "deleted " << deleted << " messages from out_msg_queue";
+  if (fail) {
     return fatal_error("error scanning/updating OutMsgQueue");
   }
   auto rt = out_msg_queue_->get_root();

diff --git a/validator/validator-group.cpp b/validator/validator-group.cpp
@@ -46,7 +46,7 @@ void ValidatorGroup::generate_block_candidate(td::uint32 round_id, td::Promise<B
   cached_collated_block_->promises.push_back(std::move(promise));
   run_collate_query(
       shard_, min_ts_, min_masterchain_block_id_, prev_block_ids_,
-      Ed25519_PublicKey{local_id_full_.ed25519_value().raw()}, validator_set_, manager_, td::Timestamp::in(10.0),
+      Ed25519_PublicKey{local_id_full_.ed25519_value().raw()}, validator_set_, manager_, td::Timestamp::in(25.0),
       [SelfId = actor_id(this), cache = cached_collated_block_](td::Result<BlockCandidate> R) {
         td::actor::send_closure(SelfId, &ValidatorGroup::generated_block_candidate, std::move(cache), std::move(R));
       });