mirror of
https://github.com/ton-blockchain/ton
synced 2025-03-09 15:40:10 +00:00
231 lines
7.4 KiB
C++
231 lines
7.4 KiB
C++
/*
|
|
This file is part of TON Blockchain Library.
|
|
|
|
TON Blockchain Library is free software: you can redistribute it and/or modify
|
|
it under the terms of the GNU Lesser General Public License as published by
|
|
the Free Software Foundation, either version 2 of the License, or
|
|
(at your option) any later version.
|
|
|
|
TON Blockchain Library is distributed in the hope that it will be useful,
|
|
but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
GNU Lesser General Public License for more details.
|
|
|
|
You should have received a copy of the GNU Lesser General Public License
|
|
along with TON Blockchain Library. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
Copyright 2017-2020 Telegram Systems LLP
|
|
*/
|
|
#include "output-queue-merger.h"
|
|
|
|
namespace block {
|
|
|
|
/*
|
|
*
|
|
* OUTPUT QUEUE MERGER
|
|
*
|
|
*/
|
|
|
|
bool OutputQueueMerger::MsgKeyValue::operator<(const MsgKeyValue& other) const {
|
|
return lt < other.lt ||
|
|
(lt == other.lt && td::bitstring::bits_memcmp(key.cbits() + 96, other.key.cbits() + 96, 256) < 0);
|
|
}
|
|
|
|
bool OutputQueueMerger::MsgKeyValue::less(const std::unique_ptr<MsgKeyValue>& he1,
|
|
const std::unique_ptr<MsgKeyValue>& he2) {
|
|
return *he1 < *he2;
|
|
}
|
|
|
|
bool OutputQueueMerger::MsgKeyValue::greater(const std::unique_ptr<MsgKeyValue>& he1,
|
|
const std::unique_ptr<MsgKeyValue>& he2) {
|
|
return *he2 < *he1;
|
|
}
|
|
|
|
OutputQueueMerger::MsgKeyValue::MsgKeyValue(td::ConstBitPtr key_pfx, int key_pfx_len, int _src, Ref<vm::Cell> node)
|
|
: source(_src) {
|
|
unpack_node(key_pfx, key_pfx_len, std::move(node));
|
|
}
|
|
|
|
OutputQueueMerger::MsgKeyValue::MsgKeyValue(int _src, Ref<vm::Cell> node) : source(_src) {
|
|
unpack_node(td::ConstBitPtr{nullptr}, 0, std::move(node));
|
|
}
|
|
|
|
bool OutputQueueMerger::MsgKeyValue::invalidate() {
|
|
msg.clear();
|
|
lt = 0;
|
|
source = -1;
|
|
return false;
|
|
}
|
|
|
|
ton::LogicalTime OutputQueueMerger::MsgKeyValue::get_node_lt(Ref<vm::Cell> node, int key_pfx_len) {
|
|
if (node.is_null() || (unsigned)key_pfx_len > (unsigned)max_key_len) {
|
|
return std::numeric_limits<td::uint64>::max();
|
|
}
|
|
vm::dict::LabelParser label{std::move(node), max_key_len - key_pfx_len, vm::dict::LabelParser::chk_size};
|
|
if (!label.is_valid()) {
|
|
return std::numeric_limits<td::uint64>::max();
|
|
}
|
|
label.skip_label();
|
|
return label.remainder->prefetch_ulong(64);
|
|
}
|
|
|
|
bool OutputQueueMerger::MsgKeyValue::unpack_node(td::ConstBitPtr key_pfx, int key_pfx_len, Ref<vm::Cell> node) {
|
|
if (node.is_null() || (unsigned)key_pfx_len >= (unsigned)max_key_len) {
|
|
return invalidate();
|
|
}
|
|
if (!key_pfx.is_null()) {
|
|
td::bitstring::bits_memcpy(key.bits(), key_pfx, key_pfx_len);
|
|
}
|
|
vm::dict::LabelParser label{std::move(node), max_key_len - key_pfx_len, vm::dict::LabelParser::chk_size};
|
|
if (!label.is_valid()) {
|
|
return invalidate();
|
|
}
|
|
label.extract_label_to(key.bits() + key_pfx_len);
|
|
key_len = key_pfx_len + label.l_bits;
|
|
msg = std::move(label.remainder);
|
|
if (!msg.write().fetch_uint_to(64, lt)) {
|
|
return invalidate();
|
|
}
|
|
if (is_fork() && msg->size_ext() != 0x20000) {
|
|
return invalidate();
|
|
}
|
|
return true;
|
|
}
|
|
|
|
bool OutputQueueMerger::MsgKeyValue::replace_with_child(bool child_idx) {
|
|
if (!is_fork() || msg.is_null() || msg->size_ext() != 0x20000) {
|
|
return false;
|
|
}
|
|
key[key_len] = child_idx;
|
|
return unpack_node(td::ConstBitPtr{nullptr}, key_len + 1, msg->prefetch_ref(child_idx));
|
|
}
|
|
|
|
bool OutputQueueMerger::MsgKeyValue::replace_by_prefix(td::ConstBitPtr req_pfx, int req_pfx_len) {
|
|
do {
|
|
if (td::bitstring::bits_memcmp(req_pfx, key.cbits(), std::min(req_pfx_len, key_len))) {
|
|
return false;
|
|
}
|
|
if (key_len >= req_pfx_len) {
|
|
return true;
|
|
}
|
|
} while (replace_with_child(req_pfx[key_len]));
|
|
return false;
|
|
}
|
|
|
|
bool OutputQueueMerger::MsgKeyValue::split(MsgKeyValue& second) {
|
|
if (!is_fork() || msg.is_null()) {
|
|
return false;
|
|
}
|
|
unsigned long long keep_lt = lt;
|
|
unsigned long long left_lt = get_node_lt(msg->prefetch_ref(0), key_len + 1);
|
|
bool sw = (left_lt == lt);
|
|
second.source = source;
|
|
key[key_len] = sw;
|
|
if (!second.unpack_node(key.cbits(), key_len + 1, msg->prefetch_ref(sw))) {
|
|
return false;
|
|
}
|
|
key[key_len] = 1 - sw;
|
|
if (!unpack_node(td::ConstBitPtr{nullptr}, key_len + 1, msg->prefetch_ref(1 - sw))) {
|
|
return false;
|
|
}
|
|
if (lt != keep_lt || second.lt < keep_lt) {
|
|
return false;
|
|
}
|
|
return true;
|
|
}
|
|
|
|
void OutputQueueMerger::add_root(int src, Ref<vm::Cell> outmsg_root, td::int32 msg_limit) {
|
|
if (outmsg_root.is_null()) {
|
|
return;
|
|
}
|
|
//block::gen::HashmapAug{352, block::gen::t_EnqueuedMsg, block::gen::t_uint64}.print_ref(std::cerr, outmsg_root);
|
|
auto kv = std::make_unique<MsgKeyValue>(src, std::move(outmsg_root));
|
|
if (kv->replace_by_prefix(common_pfx.cbits(), common_pfx_len)) {
|
|
heap.push_back(std::move(kv));
|
|
}
|
|
if ((int)src_remaining_msgs_.size() < src + 1) {
|
|
src_remaining_msgs_.resize(src + 1);
|
|
}
|
|
src_remaining_msgs_[src] = msg_limit;
|
|
}
|
|
|
|
OutputQueueMerger::OutputQueueMerger(ton::ShardIdFull queue_for, std::vector<OutputQueueMerger::Neighbor> neighbors)
|
|
: eof(false), failed(false) {
|
|
common_pfx.bits().store_int(queue_for.workchain, 32);
|
|
int l = queue_for.pfx_len();
|
|
td::bitstring::bits_store_long_top(common_pfx.bits() + 32, queue_for.shard, l);
|
|
common_pfx_len = 32 + l;
|
|
int i = 0;
|
|
for (Neighbor& neighbor : neighbors) {
|
|
if (!neighbor.disabled_) {
|
|
LOG(DEBUG) << "adding " << (neighbor.outmsg_root_.is_null() ? "" : "non-") << "empty output queue for neighbor #"
|
|
<< i << " (" << neighbor.block_id_.to_str() << ")";
|
|
add_root(i++, neighbor.outmsg_root_, neighbor.msg_limit_);
|
|
} else {
|
|
LOG(DEBUG) << "skipping output queue for disabled neighbor #" << i;
|
|
i++;
|
|
}
|
|
}
|
|
std::make_heap(heap.begin(), heap.end(), MsgKeyValue::greater);
|
|
eof = heap.empty();
|
|
if (!eof) {
|
|
load();
|
|
}
|
|
}
|
|
|
|
OutputQueueMerger::MsgKeyValue* OutputQueueMerger::cur() {
|
|
return eof ? nullptr : msg_list.at(pos).get();
|
|
}
|
|
|
|
std::unique_ptr<OutputQueueMerger::MsgKeyValue> OutputQueueMerger::extract_cur() {
|
|
return eof ? std::unique_ptr<MsgKeyValue>{} : std::move(msg_list.at(pos));
|
|
}
|
|
|
|
bool OutputQueueMerger::next() {
|
|
if (eof) {
|
|
return false;
|
|
} else if (++pos < msg_list.size() || load()) {
|
|
return true;
|
|
} else {
|
|
eof = true;
|
|
return false;
|
|
}
|
|
}
|
|
|
|
bool OutputQueueMerger::load() {
|
|
if (heap.empty() || failed) {
|
|
return false;
|
|
}
|
|
unsigned long long lt = heap[0]->lt;
|
|
std::size_t orig_size = msg_list.size();
|
|
do {
|
|
while (heap[0]->is_fork()) {
|
|
auto other = std::make_unique<MsgKeyValue>();
|
|
if (!heap[0]->split(*other)) {
|
|
failed = true;
|
|
return false;
|
|
}
|
|
heap.push_back(std::move(other));
|
|
std::push_heap(heap.begin(), heap.end(), MsgKeyValue::greater);
|
|
}
|
|
assert(heap[0]->lt == lt);
|
|
std::pop_heap(heap.begin(), heap.end(), MsgKeyValue::greater);
|
|
msg_list.push_back(std::move(heap.back()));
|
|
heap.pop_back();
|
|
} while (!heap.empty() && heap[0]->lt <= lt);
|
|
std::sort(msg_list.begin() + orig_size, msg_list.end(), MsgKeyValue::less);
|
|
for (size_t i = orig_size; i < msg_list.size(); ++i) {
|
|
td::int32 &remaining = src_remaining_msgs_[msg_list[i]->source];
|
|
if (remaining != -1) {
|
|
if (remaining == 0) {
|
|
limit_exceeded = true;
|
|
} else {
|
|
--remaining;
|
|
}
|
|
}
|
|
msg_list[i]->limit_exceeded = limit_exceeded;
|
|
}
|
|
return msg_list.size() > orig_size;
|
|
}
|
|
|
|
} // namespace block
|