Skip to content

Commit

Permalink
Merge pull request #820 from SpyCheese/queue_clearance
Browse files Browse the repository at this point in the history
Improve Collator::opt_msg_queue_cleanup, increase collator timeout
  • Loading branch information
EmelyanenkoK authored Dec 6, 2023
2 parents 6fa8161 + cbbd4f7 commit 027338c
Show file tree
Hide file tree
Showing 4 changed files with 40 additions and 22 deletions.
20 changes: 12 additions & 8 deletions crypto/vm/dict.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand Down Expand Up @@ -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;
}
Expand All @@ -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) {
Expand Down
4 changes: 2 additions & 2 deletions crypto/vm/dict.h
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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,
Expand Down
36 changes: 25 additions & 11 deletions validator/impl/collator.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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))
Expand Down Expand Up @@ -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;
Expand All @@ -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_;
Expand All @@ -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();
Expand Down
2 changes: 1 addition & 1 deletion validator/validator-group.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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));
});
Expand Down

0 comments on commit 027338c

Please sign in to comment.