Skip to content

Commit

Permalink
Collator: change deferring behavior when out queue size is big (ton-b…
Browse files Browse the repository at this point in the history
  • Loading branch information
SpyCheese authored Sep 5, 2024
1 parent cb69f30 commit e32a74e
Show file tree
Hide file tree
Showing 5 changed files with 43 additions and 5 deletions.
3 changes: 2 additions & 1 deletion tl/generate/scheme/ton_api.tl
Original file line number Diff line number Diff line change
Expand Up @@ -628,7 +628,8 @@ engine.validator.customOverlaysConfig overlays:(vector engine.validator.customOv
engine.validator.collatorOptions
deferring_enabled:Bool defer_messages_after:int defer_out_queue_size_limit:long
dispatch_phase_2_max_total:int dispatch_phase_3_max_total:int
dispatch_phase_2_max_per_initiator:int dispatch_phase_3_max_per_initiator:int = engine.validator.CollatorOptions;
dispatch_phase_2_max_per_initiator:int dispatch_phase_3_max_per_initiator:int
whitelist:(vector string) prioritylist:(vector string) = engine.validator.CollatorOptions;

---functions---
---types---
Expand Down
Binary file modified tl/generate/scheme/ton_api.tlo
Binary file not shown.
8 changes: 8 additions & 0 deletions validator-engine/validator-engine.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -2531,6 +2531,14 @@ static td::Result<td::Ref<ton::validator::CollatorOptions>> parse_collator_optio
} else {
opts.dispatch_phase_3_max_per_initiator = {};
}
for (const std::string& s : f.whitelist_) {
TRY_RESULT(addr, block::StdAddress::parse(s));
opts.whitelist.emplace(addr.workchain, addr.addr);
}
for (const std::string& s : f.prioritylist_) {
TRY_RESULT(addr, block::StdAddress::parse(s));
opts.prioritylist.emplace(addr.workchain, addr.addr);
}

return ref;
}
Expand Down
32 changes: 28 additions & 4 deletions validator/impl/collator.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -3066,7 +3066,7 @@ int Collator::process_one_new_message(block::NewOutMsg msg, bool enqueue_only, R
bool defer = false;
if (!from_dispatch_queue) {
if (deferring_messages_enabled_ && collator_opts_->deferring_enabled && !is_special && !is_special_account &&
msg.msg_idx != 0) {
!collator_opts_->whitelist.count({src_wc, src_addr}) && msg.msg_idx != 0) {
if (++sender_generated_messages_count_[src_addr] >= collator_opts_->defer_messages_after ||
out_msg_queue_size_ > defer_out_queue_size_limit_) {
defer = true;
Expand Down Expand Up @@ -3697,6 +3697,8 @@ bool Collator::process_dispatch_queue() {
vm::AugmentedDictionary cur_dispatch_queue{dispatch_queue_->get_root(), 256, block::tlb::aug_DispatchQueue};
std::map<std::tuple<WorkchainId, StdSmcAddress, LogicalTime>, size_t> count_per_initiator;
size_t total_count = 0;
auto prioritylist = collator_opts_->prioritylist;
auto prioritylist_iter = prioritylist.begin();
while (!cur_dispatch_queue.is_empty()) {
block_full_ = !block_limit_status_->fits(block::ParamLimits::cl_normal);
if (block_full_) {
Expand All @@ -3713,9 +3715,30 @@ bool Collator::process_dispatch_queue() {
return true;
}
StdSmcAddress src_addr;
auto account_dispatch_queue = block::get_dispatch_queue_min_lt_account(cur_dispatch_queue, src_addr);
td::Ref<vm::CellSlice> account_dispatch_queue;
while (!prioritylist.empty()) {
if (prioritylist_iter == prioritylist.end()) {
prioritylist_iter = prioritylist.begin();
}
auto priority_addr = *prioritylist_iter;
if (priority_addr.first != workchain() || !is_our_address(priority_addr.second)) {
prioritylist_iter = prioritylist.erase(prioritylist_iter);
continue;
}
src_addr = priority_addr.second;
account_dispatch_queue = cur_dispatch_queue.lookup(src_addr);
if (account_dispatch_queue.is_null()) {
prioritylist_iter = prioritylist.erase(prioritylist_iter);
} else {
++prioritylist_iter;
break;
}
}
if (account_dispatch_queue.is_null()) {
return fatal_error("invalid dispatch queue in shard state");
account_dispatch_queue = block::get_dispatch_queue_min_lt_account(cur_dispatch_queue, src_addr);
if (account_dispatch_queue.is_null()) {
return fatal_error("invalid dispatch queue in shard state");
}
}
vm::Dictionary dict{64};
td::uint64 dict_size;
Expand All @@ -3735,7 +3758,8 @@ bool Collator::process_dispatch_queue() {
// Remove message from DispatchQueue
bool ok;
if (iter == 0 ||
(iter == 1 && sender_generated_messages_count_[src_addr] >= collator_opts_->defer_messages_after)) {
(iter == 1 && sender_generated_messages_count_[src_addr] >= collator_opts_->defer_messages_after &&
!collator_opts_->whitelist.count({workchain(), src_addr}))) {
ok = cur_dispatch_queue.lookup_delete(src_addr).not_null();
} else {
dict.lookup_delete(key);
Expand Down
5 changes: 5 additions & 0 deletions validator/validator.h
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,11 @@ struct CollatorOptions : public td::CntObject {
td::uint32 dispatch_phase_3_max_total = 150;
td::uint32 dispatch_phase_2_max_per_initiator = 20;
td::optional<td::uint32> dispatch_phase_3_max_per_initiator; // Default - depends on out msg queue size

// Don't defer messages from these accounts
std::set<std::pair<WorkchainId, StdSmcAddress>> whitelist;
// Prioritize these accounts on each phase of process_dispatch_queue
std::set<std::pair<WorkchainId, StdSmcAddress>> prioritylist;
};

struct ValidatorManagerOptions : public td::CntObject {
Expand Down

0 comments on commit e32a74e

Please sign in to comment.