external/ton
1
0
Fork 0
mirror of https://github.com/ton-blockchain/ton synced 2025-02-12 11:12:16 +00:00
Code Issues Releases Wiki Activity

Merge pull request #1157 from SpyCheese/testnet-update

Browse source 
Update testnet with safe features
This commit is contained in:
EmelyanenkoK 2024-09-09 19:40:48 +03:00 committed by GitHub
commit e55c132178
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
41 changed files with 1066 additions and 207 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
.github/workflows/docker-ubuntu-branch-image.yml vendored Normal file
View file

@ -0,0 +1,48 @@
name: Docker Ubuntu 22.04 branch image
on:
workflow_dispatch:
push:
branches-ignore:
- master
env:
REGISTRY: ghcr.io
IMAGE_NAME: ${{ github.repository }}
jobs:
build-and-push:
runs-on: ubuntu-22.04
steps:
- name: Check out repository
uses: actions/checkout@v3
with:
submodules: 'recursive'
- name: Set up QEMU
uses: docker/setup-qemu-action@v3
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Login to GitHub Container Registry
uses: docker/login-action@v3
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.repository_owner }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Get tag as branch name
id: tag
run: |
echo "TAG=${GITHUB_REF##*/}" >> $GITHUB_OUTPUT
- name: Build and push
id: docker_build
uses: docker/build-push-action@v6
with:
platforms: linux/amd64
push: true
context: ./
tags: |
${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ steps.tag.outputs.TAG }}

20
adnl/adnl-channel.cpp
View file

@ -112,16 +112,16 @@ void AdnlChannelImpl::send_message(td::uint32 priority, td::actor::ActorId<AdnlN
}
void AdnlChannelImpl::receive(td::IPAddress addr, td::BufferSlice data) {
auto P = td::PromiseCreator::lambda(
[peer = peer_pair_, channel_id = channel_in_id_, addr, id = print_id()](td::Result<AdnlPacket> R) {
if (R.is_error()) {
VLOG(ADNL_WARNING) << id << ": dropping IN message: can not decrypt: " << R.move_as_error();
} else {
auto packet = R.move_as_ok();
packet.set_remote_addr(addr);
td::actor::send_closure(peer, &AdnlPeerPair::receive_packet_from_channel, channel_id, std::move(packet));
}
});
auto P = td::PromiseCreator::lambda([peer = peer_pair_, channel_id = channel_in_id_, addr, id = print_id(),
size = data.size()](td::Result<AdnlPacket> R) {
if (R.is_error()) {
VLOG(ADNL_WARNING) << id << ": dropping IN message: can not decrypt: " << R.move_as_error();
} else {
auto packet = R.move_as_ok();
packet.set_remote_addr(addr);
td::actor::send_closure(peer, &AdnlPeerPair::receive_packet_from_channel, channel_id, std::move(packet), size);
}
});
decrypt(std::move(data), std::move(P));
}

97
adnl/adnl-local-id.cpp
View file

@ -41,20 +41,34 @@ AdnlAddressList AdnlLocalId::get_addr_list() const {
}
void AdnlLocalId::receive(td::IPAddress addr, td::BufferSlice data) {
auto P = td::PromiseCreator::lambda(
[peer_table = peer_table_, dst = short_id_, addr, id = print_id()](td::Result<AdnlPacket> R) {
if (R.is_error()) {
VLOG(ADNL_WARNING) << id << ": dropping IN message: cannot decrypt: " << R.move_as_error();
} else {
auto packet = R.move_as_ok();
packet.set_remote_addr(addr);
td::actor::send_closure(peer_table, &AdnlPeerTable::receive_decrypted_packet, dst, std::move(packet));
}
});
InboundRateLimiter& rate_limiter = inbound_rate_limiter_[addr];
if (!rate_limiter.rate_limiter.take()) {
VLOG(ADNL_NOTICE) << this << ": dropping IN message: rate limit exceeded";
add_dropped_packet_stats(addr);
return;
}
++rate_limiter.currently_decrypting_packets;
auto P = td::PromiseCreator::lambda([SelfId = actor_id(this), peer_table = peer_table_, dst = short_id_, addr,
id = print_id(), size = data.size()](td::Result<AdnlPacket> R) {
td::actor::send_closure(SelfId, &AdnlLocalId::decrypt_packet_done, addr);
if (R.is_error()) {
VLOG(ADNL_WARNING) << id << ": dropping IN message: cannot decrypt: " << R.move_as_error();
} else {
auto packet = R.move_as_ok();
packet.set_remote_addr(addr);
td::actor::send_closure(peer_table, &AdnlPeerTable::receive_decrypted_packet, dst, std::move(packet), size);
}
});
decrypt(std::move(data), std::move(P));
}
void AdnlLocalId::decrypt_packet_done(td::IPAddress addr) {
auto it = inbound_rate_limiter_.find(addr);
CHECK(it != inbound_rate_limiter_.end());
--it->second.currently_decrypting_packets;
add_decrypted_packet_stats(addr);
}
void AdnlLocalId::deliver(AdnlNodeIdShort src, td::BufferSlice data) {
auto s = std::move(data);
for (auto &cb : cb_) {
@ -292,6 +306,67 @@ void AdnlLocalId::update_packet(AdnlPacket packet, bool update_id, bool sign, td
}
}
void AdnlLocalId::get_stats(td::Promise<tl_object_ptr<ton_api::adnl_stats_localId>> promise) {
auto stats = create_tl_object<ton_api::adnl_stats_localId>();
stats->short_id_ = short_id_.bits256_value();
for (auto &[ip, x] : inbound_rate_limiter_) {
if (x.currently_decrypting_packets != 0) {
stats->current_decrypt_.push_back(create_tl_object<ton_api::adnl_stats_ipPackets>(
ip.is_valid() ? PSTRING() << ip.get_ip_str() << ":" << ip.get_port() : "", x.currently_decrypting_packets));
}
}
prepare_packet_stats();
stats->packets_recent_ = packet_stats_prev_.tl();
stats->packets_total_ = packet_stats_total_.tl();
stats->packets_total_->ts_start_ = (double)Adnl::adnl_start_time();
stats->packets_total_->ts_end_ = td::Clocks::system();
promise.set_result(std::move(stats));
}
void AdnlLocalId::add_decrypted_packet_stats(td::IPAddress addr) {
prepare_packet_stats();
++packet_stats_cur_.decrypted_packets[addr];
++packet_stats_total_.decrypted_packets[addr];
}
void AdnlLocalId::add_dropped_packet_stats(td::IPAddress addr) {
prepare_packet_stats();
++packet_stats_cur_.dropped_packets[addr];
++packet_stats_total_.dropped_packets[addr];
}
void AdnlLocalId::prepare_packet_stats() {
double now = td::Clocks::system();
if (now >= packet_stats_cur_.ts_end) {
packet_stats_prev_ = std::move(packet_stats_cur_);
packet_stats_cur_ = {};
auto now_int = (int)td::Clocks::system();
packet_stats_cur_.ts_start = (double)(now_int / 60 * 60);
packet_stats_cur_.ts_end = packet_stats_cur_.ts_start + 60.0;
if (packet_stats_prev_.ts_end < now - 60.0) {
packet_stats_prev_ = {};
packet_stats_prev_.ts_end = packet_stats_cur_.ts_start;
packet_stats_prev_.ts_start = packet_stats_prev_.ts_end - 60.0;
}
}
}
tl_object_ptr<ton_api::adnl_stats_localIdPackets> AdnlLocalId::PacketStats::tl() const {
auto obj = create_tl_object<ton_api::adnl_stats_localIdPackets>();
obj->ts_start_ = ts_start;
obj->ts_end_ = ts_end;
for (const auto &[ip, packets] : decrypted_packets) {
obj->decrypted_packets_.push_back(create_tl_object<ton_api::adnl_stats_ipPackets>(
ip.is_valid() ? PSTRING() << ip.get_ip_str() << ":" << ip.get_port() : "", packets));
}
for (const auto &[ip, packets] : dropped_packets) {
obj->dropped_packets_.push_back(create_tl_object<ton_api::adnl_stats_ipPackets>(
ip.is_valid() ? PSTRING() << ip.get_ip_str() << ":" << ip.get_port() : "", packets));
}
return obj;
}
} // namespace adnl
} // namespace ton

19
adnl/adnl-local-id.h
View file

@ -55,6 +55,7 @@ class AdnlLocalId : public td::actor::Actor {
void deliver(AdnlNodeIdShort src, td::BufferSlice data);
void deliver_query(AdnlNodeIdShort src, td::BufferSlice data, td::Promise<td::BufferSlice> promise);
void receive(td::IPAddress addr, td::BufferSlice data);
void decrypt_packet_done(td::IPAddress addr);
void subscribe(std::string prefix, std::unique_ptr<AdnlPeerTable::Callback> callback);
void unsubscribe(std::string prefix);
@ -77,6 +78,8 @@ class AdnlLocalId : public td::actor::Actor {
void update_packet(AdnlPacket packet, bool update_id, bool sign, td::int32 update_addr_list_if,
td::int32 update_priority_addr_list_if, td::Promise<AdnlPacket> promise);
void get_stats(td::Promise<tl_object_ptr<ton_api::adnl_stats_localId>> promise);
td::uint32 get_mode() {
return mode_;
}
@ -101,6 +104,22 @@ class AdnlLocalId : public td::actor::Actor {
td::uint32 mode_;
struct InboundRateLimiter {
RateLimiter rate_limiter = RateLimiter(75, 0.33);
td::uint64 currently_decrypting_packets = 0;
};
std::map<td::IPAddress, InboundRateLimiter> inbound_rate_limiter_;
struct PacketStats {
double ts_start = 0.0, ts_end = 0.0;
std::map<td::IPAddress, td::uint64> decrypted_packets;
std::map<td::IPAddress, td::uint64> dropped_packets;
tl_object_ptr<ton_api::adnl_stats_localIdPackets> tl() const;
} packet_stats_cur_, packet_stats_prev_, packet_stats_total_;
void add_decrypted_packet_stats(td::IPAddress addr);
void add_dropped_packet_stats(td::IPAddress addr);
void prepare_packet_stats();
void publish_address_list();
};

86
adnl/adnl-peer-table.cpp
View file

@ -84,7 +84,7 @@ void AdnlPeerTableImpl::receive_packet(td::IPAddress addr, AdnlCategoryMask cat_
<< " (len=" << (data.size() + 32) << ")";
}
void AdnlPeerTableImpl::receive_decrypted_packet(AdnlNodeIdShort dst, AdnlPacket packet) {
void AdnlPeerTableImpl::receive_decrypted_packet(AdnlNodeIdShort dst, AdnlPacket packet, td::uint64 serialized_size) {
packet.run_basic_checks().ensure();
if (!packet.inited_from_short()) {
@ -119,7 +119,7 @@ void AdnlPeerTableImpl::receive_decrypted_packet(AdnlNodeIdShort dst, AdnlPacket
return;
}
td::actor::send_closure(it->second, &AdnlPeer::receive_packet, dst, it2->second.mode, it2->second.local_id.get(),
std::move(packet));
std::move(packet), serialized_size);
}
void AdnlPeerTableImpl::add_peer(AdnlNodeIdShort local_id, AdnlNodeIdFull id, AdnlAddressList addr_list) {
@ -385,6 +385,88 @@ void AdnlPeerTableImpl::get_conn_ip_str(AdnlNodeIdShort l_id, AdnlNodeIdShort p_
td::actor::send_closure(it->second, &AdnlPeer::get_conn_ip_str, l_id, std::move(promise));
}
void AdnlPeerTableImpl::get_stats(td::Promise<tl_object_ptr<ton_api::adnl_stats>> promise) {
class Cb : public td::actor::Actor {
public:
explicit Cb(td::Promise<tl_object_ptr<ton_api::adnl_stats>> promise) : promise_(std::move(promise)) {
}
void got_local_id_stats(tl_object_ptr<ton_api::adnl_stats_localId> local_id) {
auto &local_id_stats = local_id_stats_[local_id->short_id_];
if (local_id_stats) {
local_id->peers_ = std::move(local_id_stats->peers_);
}
local_id_stats = std::move(local_id);
dec_pending();
}
void got_peer_stats(std::vector<tl_object_ptr<ton_api::adnl_stats_peerPair>> peer_pairs) {
for (auto &peer_pair : peer_pairs) {
auto &local_id_stats = local_id_stats_[peer_pair->local_id_];
if (local_id_stats == nullptr) {
local_id_stats = create_tl_object<ton_api::adnl_stats_localId>();
local_id_stats->short_id_ = peer_pair->local_id_;
}
local_id_stats->peers_.push_back(std::move(peer_pair));
}
dec_pending();
}
void inc_pending() {
++pending_;
}
void dec_pending() {
CHECK(pending_ > 0);
--pending_;
if (pending_ == 0) {
auto stats = create_tl_object<ton_api::adnl_stats>();
stats->timestamp_ = td::Clocks::system();
for (auto &[id, local_id_stats] : local_id_stats_) {
stats->local_ids_.push_back(std::move(local_id_stats));
}
promise_.set_result(std::move(stats));
stop();
}
}
private:
td::Promise<tl_object_ptr<ton_api::adnl_stats>> promise_;
size_t pending_ = 1;
std::map<td::Bits256, tl_object_ptr<ton_api::adnl_stats_localId>> local_id_stats_;
};
auto callback = td::actor::create_actor<Cb>("adnlstats", std::move(promise)).release();
for (auto &[id, local_id] : local_ids_) {
td::actor::send_closure(callback, &Cb::inc_pending);
td::actor::send_closure(local_id.local_id, &AdnlLocalId::get_stats,
[id = id, callback](td::Result<tl_object_ptr<ton_api::adnl_stats_localId>> R) {
if (R.is_error()) {
VLOG(ADNL_NOTICE)
<< "failed to get stats for local id " << id << " : " << R.move_as_error();
td::actor::send_closure(callback, &Cb::dec_pending);
} else {
td::actor::send_closure(callback, &Cb::got_local_id_stats, R.move_as_ok());
}
});
}
for (auto &[id, peer] : peers_) {
td::actor::send_closure(callback, &Cb::inc_pending);
td::actor::send_closure(
peer, &AdnlPeer::get_stats,
[id = id, callback](td::Result<std::vector<tl_object_ptr<ton_api::adnl_stats_peerPair>>> R) {
if (R.is_error()) {
VLOG(ADNL_NOTICE) << "failed to get stats for peer " << id << " : " << R.move_as_error();
td::actor::send_closure(callback, &Cb::dec_pending);
} else {
td::actor::send_closure(callback, &Cb::got_peer_stats, R.move_as_ok());
}
});
}
td::actor::send_closure(callback, &Cb::dec_pending);
}
} // namespace adnl
} // namespace ton

2
adnl/adnl-peer-table.h
View file

@ -90,7 +90,7 @@ class AdnlPeerTable : public Adnl {
virtual void answer_query(AdnlNodeIdShort src, AdnlNodeIdShort dst, AdnlQueryId query_id, td::BufferSlice data) = 0;
virtual void receive_packet(td::IPAddress addr, AdnlCategoryMask cat_mask, td::BufferSlice data) = 0;
virtual void receive_decrypted_packet(AdnlNodeIdShort dst, AdnlPacket packet) = 0;
virtual void receive_decrypted_packet(AdnlNodeIdShort dst, AdnlPacket packet, td::uint64 serialized_size) = 0;
virtual void send_message_in(AdnlNodeIdShort src, AdnlNodeIdShort dst, AdnlMessage message, td::uint32 flags) = 0;
virtual void register_channel(AdnlChannelIdShort id, AdnlNodeIdShort local_id,

4
adnl/adnl-peer-table.hpp
View file

@ -44,7 +44,7 @@ class AdnlPeerTableImpl : public AdnlPeerTable {
void add_static_nodes_from_config(AdnlNodesList nodes) override;
void receive_packet(td::IPAddress addr, AdnlCategoryMask cat_mask, td::BufferSlice data) override;
void receive_decrypted_packet(AdnlNodeIdShort dst, AdnlPacket data) override;
void receive_decrypted_packet(AdnlNodeIdShort dst, AdnlPacket data, td::uint64 serialized_size) override;
void send_message_in(AdnlNodeIdShort src, AdnlNodeIdShort dst, AdnlMessage message, td::uint32 flags) override;
void send_message(AdnlNodeIdShort src, AdnlNodeIdShort dst, td::BufferSlice data) override {
send_message_ex(src, dst, std::move(data), 0);
@ -108,6 +108,8 @@ class AdnlPeerTableImpl : public AdnlPeerTable {
td::Promise<std::pair<td::actor::ActorOwn<AdnlTunnel>, AdnlAddress>> promise) override;
void get_conn_ip_str(AdnlNodeIdShort l_id, AdnlNodeIdShort p_id, td::Promise<td::string> promise) override;
void get_stats(td::Promise<tl_object_ptr<ton_api::adnl_stats>> promise) override;
struct PrintId {};
PrintId print_id() const {
return PrintId{};

444
adnl/adnl-peer.cpp
View file

@ -26,6 +26,7 @@
#include "td/utils/base64.h"
#include "td/utils/Random.h"
#include "auto/tl/ton_api.h"
#include "td/utils/overloaded.h"
namespace ton {
@ -50,9 +51,13 @@ void AdnlPeerPairImpl::start_up() {
}
void AdnlPeerPairImpl::alarm() {
if (next_dht_query_at_ && next_dht_query_at_.is_in_past()) {
next_dht_query_at_ = td::Timestamp::never();
discover();
if (!disable_dht_query_) {
disable_dht_query_ = true;
if (next_dht_query_at_ && next_dht_query_at_.is_in_past()) {
next_dht_query_at_ = td::Timestamp::never();
discover();
}
alarm_timestamp().relax(next_dht_query_at_);
}
if (next_db_update_at_ && next_db_update_at_.is_in_past()) {
if (received_from_db_ && received_from_static_nodes_ && !peer_id_.empty()) {
@ -68,11 +73,8 @@ void AdnlPeerPairImpl::alarm() {
}
if (retry_send_at_ && retry_send_at_.is_in_past()) {
retry_send_at_ = td::Timestamp::never();
auto messages = std::move(pending_messages_);
pending_messages_.clear();
send_messages_in(std::move(messages), false);
send_messages_from_queue();
}
alarm_timestamp().relax(next_dht_query_at_);
alarm_timestamp().relax(next_db_update_at_);
alarm_timestamp().relax(retry_send_at_);
}
@ -207,18 +209,24 @@ void AdnlPeerPairImpl::receive_packet_checked(AdnlPacket packet) {
}
}
void AdnlPeerPairImpl::receive_packet_from_channel(AdnlChannelIdShort id, AdnlPacket packet) {
void AdnlPeerPairImpl::receive_packet_from_channel(AdnlChannelIdShort id, AdnlPacket packet,
td::uint64 serialized_size) {
add_packet_stats(serialized_size, /* in = */ true, /* channel = */ true);
if (id != channel_in_id_) {
VLOG(ADNL_NOTICE) << this << ": dropping IN message: outdated channel id" << id;
return;
}
if (channel_inited_) {
if (channel_inited_ && !channel_ready_) {
channel_ready_ = true;
if (!out_messages_queue_.empty()) {
td::actor::send_closure(actor_id(this), &AdnlPeerPairImpl::send_messages_from_queue);
}
}
receive_packet_checked(std::move(packet));
}
void AdnlPeerPairImpl::receive_packet(AdnlPacket packet) {
void AdnlPeerPairImpl::receive_packet(AdnlPacket packet, td::uint64 serialized_size) {
add_packet_stats(serialized_size, /* in = */ true, /* channel = */ false);
packet.run_basic_checks().ensure();
if (!encryptor_) {
@ -239,132 +247,132 @@ void AdnlPeerPairImpl::deliver_message(AdnlMessage message) {
message.visit([&](const auto &obj) { this->process_message(obj); });
}
void AdnlPeerPairImpl::send_messages_in(std::vector<OutboundAdnlMessage> messages, bool allow_postpone) {
for (td::int32 idx = 0; idx < 2; idx++) {
std::vector<OutboundAdnlMessage> not_sent;
void AdnlPeerPairImpl::send_messages_from_queue() {
while (!out_messages_queue_.empty() && out_messages_queue_.front().second.is_in_past()) {
out_messages_queue_total_size_ -= out_messages_queue_.front().first.size();
add_expired_msg_stats(out_messages_queue_.front().first.size());
out_messages_queue_.pop();
VLOG(ADNL_NOTICE) << this << ": dropping OUT message: message in queue expired";
}
if (out_messages_queue_.empty()) {
return;
}
auto connR = get_conn(idx == 1);
if (connR.is_error()) {
if (!allow_postpone) {
VLOG(ADNL_NOTICE) << this << ": dropping OUT messages: cannot get conn: " << connR.move_as_error();
return;
}
VLOG(ADNL_INFO) << this << ": delaying OUT messages: cannot get conn: " << connR.move_as_error();
if (!retry_send_at_) {
retry_send_at_.relax(td::Timestamp::in(10.0));
alarm_timestamp().relax(retry_send_at_);
}
for (auto &m : messages) {
pending_messages_.push_back(std::move(m));
}
auto connR = get_conn();
if (connR.is_error()) {
disable_dht_query_ = false;
retry_send_at_.relax(td::Timestamp::in(message_in_queue_ttl_ - 1.0));
alarm_timestamp().relax(retry_send_at_);
VLOG(ADNL_INFO) << this << ": delaying OUT messages: cannot get conn: " << connR.move_as_error();
return;
}
disable_dht_query_ = true;
auto C = connR.move_as_ok();
auto conn = std::move(C.first);
bool is_direct = C.second;
bool first = !skip_init_packet_;
while (!out_messages_queue_.empty()) {
bool try_reinit = try_reinit_at_ && try_reinit_at_.is_in_past();
bool via_channel = channel_ready_ && !try_reinit;
if (!via_channel && !nochannel_rate_limiter_.take()) {
alarm_timestamp().relax(retry_send_at_ = nochannel_rate_limiter_.ready_at());
return;
}
auto C = connR.move_as_ok();
bool is_direct = C.second;
auto conn = std::move(C.first);
if (idx == 1) {
CHECK(is_direct);
if (try_reinit) {
try_reinit_at_ = td::Timestamp::in(td::Random::fast(0.5, 1.5));
}
respond_with_nop_after_ = td::Timestamp::in(td::Random::fast(1.0, 2.0));
size_t s = (via_channel ? channel_packet_header_max_size() : packet_header_max_size());
if (first) {
s += 2 * addr_list_max_size();
}
size_t ptr = 0;
bool first = true;
do {
respond_with_nop_after_ = td::Timestamp::in(td::Random::fast(1.0, 2.0));
bool try_reinit = try_reinit_at_ && try_reinit_at_.is_in_past();
if (try_reinit) {
try_reinit_at_ = td::Timestamp::in(td::Random::fast(0.5, 1.5));
}
bool via_channel = channel_ready_ && !try_reinit;
size_t s = (via_channel ? channel_packet_header_max_size() : packet_header_max_size());
if (first) {
s += 2 * addr_list_max_size();
}
AdnlPacket packet;
packet.set_seqno(++out_seqno_);
packet.set_confirm_seqno(in_seqno_);
AdnlPacket packet;
packet.set_seqno(++out_seqno_);
packet.set_confirm_seqno(in_seqno_);
if (first) {
if (!channel_inited_) {
auto M = adnlmessage::AdnlMessageCreateChannel{channel_pub_, channel_pk_date_};
s += M.size();
packet.add_message(std::move(M));
} else if (!channel_ready_) {
auto M = adnlmessage::AdnlMessageConfirmChannel{channel_pub_, peer_channel_pub_, channel_pk_date_};
s += M.size();
packet.add_message(std::move(M));
}
if (first) {
if (!channel_inited_) {
auto M = adnlmessage::AdnlMessageCreateChannel{channel_pub_, channel_pk_date_};
s += M.size();
packet.add_message(std::move(M));
} else if (!channel_ready_) {
auto M = adnlmessage::AdnlMessageConfirmChannel{channel_pub_, peer_channel_pub_, channel_pk_date_};
s += M.size();
packet.add_message(std::move(M));
}
}
if (!addr_list_.empty()) {
packet.set_received_addr_list_version(addr_list_.version());
}
if (!priority_addr_list_.empty()) {
packet.set_received_priority_addr_list_version(priority_addr_list_.version());
}
if (!addr_list_.empty()) {
packet.set_received_addr_list_version(addr_list_.version());
}
if (!priority_addr_list_.empty()) {
packet.set_received_priority_addr_list_version(priority_addr_list_.version());
}
while (ptr < messages.size()) {
auto &M = messages[ptr];
if (!is_direct && (M.flags() & Adnl::SendFlags::direct_only)) {
not_sent.push_back(std::move(M));
continue;
}
CHECK(M.size() <= get_mtu());
skip_init_packet_ = true;
while (!out_messages_queue_.empty()) {
auto &M = out_messages_queue_.front().first;
if (!is_direct && (M.flags() & Adnl::SendFlags::direct_only)) {
out_messages_queue_total_size_ -= M.size();
out_messages_queue_.pop();
continue;
}
CHECK(M.size() <= get_mtu());
if (s + M.size() <= AdnlNetworkManager::get_mtu()) {
s += M.size();
out_messages_queue_total_size_ -= M.size();
packet.add_message(M.release());
out_messages_queue_.pop();
skip_init_packet_ = false;
} else {
break;
}
}
if (!via_channel) {
packet.set_reinit_date(Adnl::adnl_start_time(), reinit_date_);
packet.set_source(local_id_);
}
if (!first) {
if (!channel_inited_) {
auto M = adnlmessage::AdnlMessageCreateChannel{channel_pub_, channel_pk_date_};
if (s + M.size() <= AdnlNetworkManager::get_mtu()) {
s += M.size();
packet.add_message(M.release());
ptr++;
} else {
break;
packet.add_message(std::move(M));
}
} else if (!channel_ready_) {
auto M = adnlmessage::AdnlMessageConfirmChannel{channel_pub_, peer_channel_pub_, channel_pk_date_};
if (s + M.size() <= AdnlNetworkManager::get_mtu()) {
s += M.size();
packet.add_message(std::move(M));
}
}
if (!via_channel) {
packet.set_reinit_date(Adnl::adnl_start_time(), reinit_date_);
packet.set_source(local_id_);
}
if (!first) {
if (!channel_inited_) {
auto M = adnlmessage::AdnlMessageCreateChannel{channel_pub_, channel_pk_date_};
if (s + M.size() <= AdnlNetworkManager::get_mtu()) {
s += M.size();
packet.add_message(std::move(M));
}
} else if (!channel_ready_) {
auto M = adnlmessage::AdnlMessageConfirmChannel{channel_pub_, peer_channel_pub_, channel_pk_date_};
if (s + M.size() <= AdnlNetworkManager::get_mtu()) {
s += M.size();
packet.add_message(std::move(M));
}
}
}
packet.run_basic_checks().ensure();
auto P = td::PromiseCreator::lambda([SelfId = actor_id(this), conn, id = print_id(),
via_channel](td::Result<AdnlPacket> res) {
if (res.is_error()) {
LOG(ERROR) << id << ": dropping OUT message: error while creating packet: " << res.move_as_error();
} else {
td::actor::send_closure(SelfId, &AdnlPeerPairImpl::send_packet_continue, res.move_as_ok(), conn, via_channel);
}
});
td::actor::send_closure(local_actor_, &AdnlLocalId::update_packet, std::move(packet),
(!channel_ready_ && ack_seqno_ == 0 && in_seqno_ == 0) || try_reinit, !via_channel,
(first || s + addr_list_max_size() <= AdnlNetworkManager::get_mtu())
? (try_reinit ? 0 : peer_recv_addr_list_version_)
: 0x7fffffff,
(first || s + 2 * addr_list_max_size() <= AdnlNetworkManager::get_mtu())
? peer_recv_priority_addr_list_version_
: 0x7fffffff,
std::move(P));
first = false;
} while (ptr < messages.size());
messages = std::move(not_sent);
if (!messages.size()) {
break;
}
packet.run_basic_checks().ensure();
auto P = td::PromiseCreator::lambda([SelfId = actor_id(this), conn, id = print_id(),
via_channel](td::Result<AdnlPacket> res) {
if (res.is_error()) {
LOG(ERROR) << id << ": dropping OUT message: error while creating packet: " << res.move_as_error();
} else {
td::actor::send_closure(SelfId, &AdnlPeerPairImpl::send_packet_continue, res.move_as_ok(), conn, via_channel);
}
});
td::actor::send_closure(local_actor_, &AdnlLocalId::update_packet, std::move(packet),
(!channel_ready_ && ack_seqno_ == 0 && in_seqno_ == 0) || try_reinit, !via_channel,
(first || s + addr_list_max_size() <= AdnlNetworkManager::get_mtu())
? (try_reinit ? 0 : peer_recv_addr_list_version_)
: 0x7fffffff,
(first || s + 2 * addr_list_max_size() <= AdnlNetworkManager::get_mtu())
? peer_recv_priority_addr_list_version_
: 0x7fffffff,
std::move(P));
first = false;
}
}
@ -395,7 +403,11 @@ void AdnlPeerPairImpl::send_messages(std::vector<OutboundAdnlMessage> messages)
}
}
}
send_messages_in(std::move(new_vec), true);
for (auto &m : new_vec) {
out_messages_queue_total_size_ += m.size();
out_messages_queue_.emplace(std::move(m), td::Timestamp::in(message_in_queue_ttl_));
}
send_messages_from_queue();
}
void AdnlPeerPairImpl::send_packet_continue(AdnlPacket packet, td::actor::ActorId<AdnlNetworkConnection> conn,
@ -407,6 +419,7 @@ void AdnlPeerPairImpl::send_packet_continue(AdnlPacket packet, td::actor::ActorI
auto B = serialize_tl_object(packet.tl(), true);
if (via_channel) {
if (channel_ready_) {
add_packet_stats(B.size(), /* in = */ false, /* channel = */ true);
td::actor::send_closure(channel_, &AdnlChannel::send_message, priority_, conn, std::move(B));
} else {
VLOG(ADNL_WARNING) << this << ": dropping OUT message [" << local_id_ << "->" << peer_id_short_
@ -434,6 +447,7 @@ void AdnlPeerPairImpl::send_packet_continue(AdnlPacket packet, td::actor::ActorI
S.remove_prefix(32);
S.copy_from(X.as_slice());
add_packet_stats(B.size(), /* in = */ false, /* channel = */ false);
td::actor::send_closure(conn, &AdnlNetworkConnection::send, local_id_, peer_id_short_, priority_, std::move(enc));
}
@ -520,7 +534,10 @@ void AdnlPeerPairImpl::process_message(const adnlmessage::AdnlMessageConfirmChan
VLOG(ADNL_NOTICE) << this << ": received adnl.message.confirmChannel with old key";
return;
}
channel_ready_ = true;
if (!channel_ready_) {
channel_ready_ = true;
send_messages_from_queue();
}
}
void AdnlPeerPairImpl::process_message(const adnlmessage::AdnlMessageCustom &message) {
@ -674,7 +691,7 @@ void AdnlPeerPairImpl::reinit(td::int32 date) {
}
}
td::Result<std::pair<td::actor::ActorId<AdnlNetworkConnection>, bool>> AdnlPeerPairImpl::get_conn(bool direct_only) {
td::Result<std::pair<td::actor::ActorId<AdnlNetworkConnection>, bool>> AdnlPeerPairImpl::get_conn() {
if (!priority_addr_list_.empty() && priority_addr_list_.expire_at() < td::Clocks::system()) {
priority_addr_list_ = AdnlAddressList{};
priority_conns_.clear();
@ -692,14 +709,18 @@ td::Result<std::pair<td::actor::ActorId<AdnlNetworkConnection>, bool>> AdnlPeerP
}
}
for (auto &conn : priority_conns_) {
if (conn.ready() && (!direct_only || conn.is_direct())) {
return std::make_pair(conn.conn.get(), conn.is_direct());
for (int direct_only = 1; direct_only >= 0; --direct_only) {
for (auto &conn : priority_conns_) {
if (conn.ready() && (!direct_only || conn.is_direct())) {
return std::make_pair(conn.conn.get(), conn.is_direct());
}
}
}
for (auto &conn : conns_) {
if (conn.ready() && (!direct_only || conn.is_direct())) {
return std::make_pair(conn.conn.get(), conn.is_direct());
for (int direct_only = 1; direct_only >= 0; --direct_only) {
for (auto &conn : conns_) {
if (conn.ready() && (!direct_only || conn.is_direct())) {
return std::make_pair(conn.conn.get(), conn.is_direct());
}
}
}
return td::Status::Error(ErrorCode::notready, "no active connections");
@ -787,6 +808,47 @@ void AdnlPeerPairImpl::get_conn_ip_str(td::Promise<td::string> promise) {
promise.set_value("undefined");
}
void AdnlPeerPairImpl::get_stats(td::Promise<tl_object_ptr<ton_api::adnl_stats_peerPair>> promise) {
auto stats = create_tl_object<ton_api::adnl_stats_peerPair>();
stats->local_id_ = local_id_.bits256_value();
stats->peer_id_ = peer_id_short_.bits256_value();
for (const AdnlAddress &addr : addr_list_.addrs()) {
ton_api::downcast_call(*addr->tl(), td::overloaded(
[&](const ton_api::adnl_address_udp &obj) {
stats->ip_str_ = PSTRING() << td::IPAddress::ipv4_to_str(obj.ip_) << ":"
<< obj.port_;
},
[&](const auto &) {}));
if (!stats->ip_str_.empty()) {
break;
}
}
prepare_packet_stats();
stats->last_in_packet_ts_ = last_in_packet_ts_;
stats->last_out_packet_ts_ = last_out_packet_ts_;
stats->packets_total_ = packet_stats_total_.tl();
stats->packets_total_->ts_start_ = started_ts_;
stats->packets_total_->ts_end_ = td::Clocks::system();
stats->packets_recent_ = packet_stats_prev_.tl();
if (channel_ready_) {
stats->channel_status_ = 2;
} else if (channel_inited_) {
stats->channel_status_ = 1;
} else {
stats->channel_status_ = 0;
}
stats->try_reinit_at_ = (try_reinit_at_ ? try_reinit_at_.at_unix() : 0.0);
stats->connection_ready_ =
std::any_of(conns_.begin(), conns_.end(), [](const Conn &conn) { return conn.ready(); }) ||
std::any_of(priority_conns_.begin(), priority_conns_.end(), [](const Conn &conn) { return conn.ready(); });
stats->out_queue_messages_ = out_messages_queue_.size();
stats->out_queue_bytes_ = out_messages_queue_total_size_;
promise.set_result(std::move(stats));
}
void AdnlPeerImpl::update_id(AdnlNodeIdFull id) {
CHECK(id.compute_short_id() == peer_id_short_);
if (!peer_id_.empty()) {
@ -810,10 +872,8 @@ void AdnlPeerPairImpl::Conn::create_conn(td::actor::ActorId<AdnlPeerPairImpl> pe
void AdnlPeerPairImpl::conn_change_state(AdnlConnectionIdShort id, bool ready) {
if (ready) {
if (pending_messages_.size() > 0) {
auto messages = std::move(pending_messages_);
pending_messages_.clear();
send_messages_in(std::move(messages), true);
if (out_messages_queue_.empty()) {
send_messages_from_queue();
}
}
}
@ -835,7 +895,7 @@ td::actor::ActorOwn<AdnlPeer> AdnlPeer::create(td::actor::ActorId<AdnlNetworkMan
}
void AdnlPeerImpl::receive_packet(AdnlNodeIdShort dst, td::uint32 dst_mode, td::actor::ActorId<AdnlLocalId> dst_actor,
AdnlPacket packet) {
AdnlPacket packet, td::uint64 serialized_size) {
if (packet.inited_from()) {
update_id(packet.from());
}
@ -853,7 +913,7 @@ void AdnlPeerImpl::receive_packet(AdnlNodeIdShort dst, td::uint32 dst_mode, td::
}
}
td::actor::send_closure(it->second.get(), &AdnlPeerPair::receive_packet, std::move(packet));
td::actor::send_closure(it->second.get(), &AdnlPeerPair::receive_packet, std::move(packet), serialized_size);
}
void AdnlPeerImpl::send_messages(AdnlNodeIdShort src, td::uint32 src_mode, td::actor::ActorId<AdnlLocalId> src_actor,
@ -933,6 +993,56 @@ void AdnlPeerImpl::update_addr_list(AdnlNodeIdShort local_id, td::uint32 local_m
td::actor::send_closure(it->second, &AdnlPeerPair::update_addr_list, std::move(addr_list));
}
void AdnlPeerImpl::get_stats(td::Promise<std::vector<tl_object_ptr<ton_api::adnl_stats_peerPair>>> promise) {
class Cb : public td::actor::Actor {
public:
explicit Cb(td::Promise<std::vector<tl_object_ptr<ton_api::adnl_stats_peerPair>>> promise)
: promise_(std::move(promise)) {
}
void got_peer_pair_stats(tl_object_ptr<ton_api::adnl_stats_peerPair> peer_pair) {
result_.push_back(std::move(peer_pair));
dec_pending();
}
void inc_pending() {
++pending_;
}
void dec_pending() {
CHECK(pending_ > 0);
--pending_;
if (pending_ == 0) {
promise_.set_result(std::move(result_));
stop();
}
}
private:
td::Promise<std::vector<tl_object_ptr<ton_api::adnl_stats_peerPair>>> promise_;
size_t pending_ = 1;
std::vector<tl_object_ptr<ton_api::adnl_stats_peerPair>> result_;
};
auto callback = td::actor::create_actor<Cb>("adnlpeerstats", std::move(promise)).release();
for (auto &[local_id, peer_pair] : peer_pairs_) {
td::actor::send_closure(callback, &Cb::inc_pending);
td::actor::send_closure(peer_pair, &AdnlPeerPair::get_stats,
[local_id = local_id, peer_id = peer_id_short_,
callback](td::Result<tl_object_ptr<ton_api::adnl_stats_peerPair>> R) {
if (R.is_error()) {
VLOG(ADNL_NOTICE) << "failed to get stats for peer pair " << peer_id << "->" << local_id
<< " : " << R.move_as_error();
td::actor::send_closure(callback, &Cb::dec_pending);
} else {
td::actor::send_closure(callback, &Cb::got_peer_pair_stats, R.move_as_ok());
}
});
}
td::actor::send_closure(callback, &Cb::dec_pending);
}
void AdnlPeerPairImpl::got_data_from_db(td::Result<AdnlDbItem> R) {
received_from_db_ = false;
if (R.is_error()) {
@ -1016,6 +1126,66 @@ void AdnlPeerPairImpl::request_reverse_ping_result(td::Result<td::Unit> R) {
}
}
void AdnlPeerPairImpl::add_packet_stats(td::uint64 bytes, bool in, bool channel) {
prepare_packet_stats();
auto add_stats = [&](PacketStats &stats) {
if (in) {
++stats.in_packets;
stats.in_bytes += bytes;
if (channel) {
++stats.in_packets_channel;
stats.in_bytes_channel += bytes;
}
} else {
++stats.out_packets;
stats.out_bytes += bytes;
if (channel) {
++stats.out_packets_channel;
stats.out_bytes_channel += bytes;
}
}
};
add_stats(packet_stats_cur_);
add_stats(packet_stats_total_);
if (in) {
last_in_packet_ts_ = td::Clocks::system();
} else {
last_out_packet_ts_ = td::Clocks::system();
}
}
void AdnlPeerPairImpl::add_expired_msg_stats(td::uint64 bytes) {
prepare_packet_stats();
auto add_stats = [&](PacketStats &stats) {
++stats.out_expired_messages;
stats.out_expired_bytes += bytes;
};
add_stats(packet_stats_cur_);
add_stats(packet_stats_total_);
}
void AdnlPeerPairImpl::prepare_packet_stats() {
double now = td::Clocks::system();
if (now >= packet_stats_cur_.ts_end) {
packet_stats_prev_ = std::move(packet_stats_cur_);
packet_stats_cur_ = {};
auto now_int = (int)now;
packet_stats_cur_.ts_start = (double)(now_int / 60 * 60);
packet_stats_cur_.ts_end = packet_stats_cur_.ts_start + 60.0;
if (packet_stats_prev_.ts_end < now - 60.0) {
packet_stats_prev_ = {};
packet_stats_prev_.ts_end = packet_stats_cur_.ts_start;
packet_stats_prev_.ts_start = packet_stats_prev_.ts_end - 60.0;
}
}
}
tl_object_ptr<ton_api::adnl_stats_packets> AdnlPeerPairImpl::PacketStats::tl() const {
return create_tl_object<ton_api::adnl_stats_packets>(ts_start, ts_end, in_packets, in_bytes, in_packets_channel,
in_bytes_channel, out_packets, out_bytes, out_packets_channel,
out_bytes_channel, out_expired_messages, out_expired_bytes);
}
} // namespace adnl
} // namespace ton

8
adnl/adnl-peer.h
View file

@ -39,9 +39,9 @@ class AdnlPeer;
class AdnlPeerPair : public td::actor::Actor {
public:
virtual void receive_packet_from_channel(AdnlChannelIdShort id, AdnlPacket packet) = 0;
virtual void receive_packet_from_channel(AdnlChannelIdShort id, AdnlPacket packet, td::uint64 serialized_size) = 0;
virtual void receive_packet_checked(AdnlPacket packet) = 0;
virtual void receive_packet(AdnlPacket packet) = 0;
virtual void receive_packet(AdnlPacket packet, td::uint64 serialized_size) = 0;
virtual void send_messages(std::vector<OutboundAdnlMessage> message) = 0;
inline void send_message(OutboundAdnlMessage message) {
@ -59,6 +59,7 @@ class AdnlPeerPair : public td::actor::Actor {
virtual void update_peer_id(AdnlNodeIdFull id) = 0;
virtual void update_addr_list(AdnlAddressList addr_list) = 0;
virtual void get_conn_ip_str(td::Promise<td::string> promise) = 0;
virtual void get_stats(td::Promise<tl_object_ptr<ton_api::adnl_stats_peerPair>> promise) = 0;
static td::actor::ActorOwn<AdnlPeerPair> create(td::actor::ActorId<AdnlNetworkManager> network_manager,
td::actor::ActorId<AdnlPeerTable> peer_table, td::uint32 local_mode,
@ -71,7 +72,7 @@ class AdnlPeerPair : public td::actor::Actor {
class AdnlPeer : public td::actor::Actor {
public:
virtual void receive_packet(AdnlNodeIdShort dst, td::uint32 dst_mode, td::actor::ActorId<AdnlLocalId> dst_actor,
AdnlPacket message) = 0;
AdnlPacket message, td::uint64 serialized_size) = 0;
virtual void send_messages(AdnlNodeIdShort src, td::uint32 src_mode, td::actor::ActorId<AdnlLocalId> src_actor,
std::vector<OutboundAdnlMessage> messages) = 0;
virtual void send_query(AdnlNodeIdShort src, td::uint32 src_mode, td::actor::ActorId<AdnlLocalId> src_actor,
@ -100,6 +101,7 @@ class AdnlPeer : public td::actor::Actor {
td::actor::ActorId<AdnlLocalId> local_actor, AdnlAddressList addr_list) = 0;
virtual void update_dht_node(td::actor::ActorId<dht::Dht> dht_node) = 0;
virtual void get_conn_ip_str(AdnlNodeIdShort l_id, td::Promise<td::string> promise) = 0;
virtual void get_stats(td::Promise<std::vector<tl_object_ptr<ton_api::adnl_stats_peerPair>>> promise) = 0;
};
} // namespace adnl

41
adnl/adnl-peer.hpp
View file

@ -20,6 +20,7 @@
#include <vector>
#include <map>
#include <queue>
#include "adnl-peer.h"
#include "adnl-peer-table.h"
@ -66,12 +67,12 @@ class AdnlPeerPairImpl : public AdnlPeerPair {
void discover();
void receive_packet_from_channel(AdnlChannelIdShort id, AdnlPacket packet) override;
void receive_packet_from_channel(AdnlChannelIdShort id, AdnlPacket packet, td::uint64 serialized_size) override;
void receive_packet_checked(AdnlPacket packet) override;
void receive_packet(AdnlPacket packet) override;
void receive_packet(AdnlPacket packet, td::uint64 serialized_size) override;
void deliver_message(AdnlMessage message);
void send_messages_in(std::vector<OutboundAdnlMessage> messages, bool allow_postpone);
void send_messages_from_queue();
void send_messages(std::vector<OutboundAdnlMessage> messages) override;
void send_packet_continue(AdnlPacket packet, td::actor::ActorId<AdnlNetworkConnection> conn, bool via_channel);
void send_query(std::string name, td::Promise<td::BufferSlice> promise, td::Timestamp timeout, td::BufferSlice data,
@ -89,6 +90,7 @@ class AdnlPeerPairImpl : public AdnlPeerPair {
void update_peer_id(AdnlNodeIdFull id) override;
void get_conn_ip_str(td::Promise<td::string> promise) override;
void get_stats(td::Promise<tl_object_ptr<ton_api::adnl_stats_peerPair>> promise) override;
void got_data_from_db(td::Result<AdnlDbItem> R);
void got_data_from_static_nodes(td::Result<AdnlNode> R);
@ -124,7 +126,7 @@ class AdnlPeerPairImpl : public AdnlPeerPair {
private:
void respond_with_nop();
void reinit(td::int32 date);
td::Result<std::pair<td::actor::ActorId<AdnlNetworkConnection>, bool>> get_conn(bool direct_only);
td::Result<std::pair<td::actor::ActorId<AdnlNetworkConnection>, bool>> get_conn();
void create_channel(pubkeys::Ed25519 pub, td::uint32 date);
bool received_packet(td::uint64 seqno) const {
@ -183,11 +185,11 @@ class AdnlPeerPairImpl : public AdnlPeerPair {
Conn() {
}
bool ready() {
bool ready() const {
return !conn.empty() && conn.get_actor_unsafe().is_active();
}
bool is_direct() {
bool is_direct() const {
return addr->is_public();
}
@ -195,7 +197,14 @@ class AdnlPeerPairImpl : public AdnlPeerPair {
td::actor::ActorId<Adnl> adnl);
};
std::vector<OutboundAdnlMessage> pending_messages_;
// Messages waiting for connection or for nochannel rate limiter
std::queue<std::pair<OutboundAdnlMessage, td::Timestamp>> out_messages_queue_;
td::uint64 out_messages_queue_total_size_ = 0;
RateLimiter nochannel_rate_limiter_ = RateLimiter(50, 0.5); // max 50, period = 0.5s
td::Timestamp retry_send_at_ = td::Timestamp::never();
bool disable_dht_query_ = false;
bool skip_init_packet_ = false;
double message_in_queue_ttl_ = 10.0;
td::actor::ActorId<AdnlNetworkManager> network_manager_;
td::actor::ActorId<AdnlPeerTable> peer_table_;
@ -254,7 +263,6 @@ class AdnlPeerPairImpl : public AdnlPeerPair {
td::Timestamp next_dht_query_at_ = td::Timestamp::never();
td::Timestamp next_db_update_at_ = td::Timestamp::never();
td::Timestamp retry_send_at_ = td::Timestamp::never();
td::Timestamp last_received_packet_ = td::Timestamp::never();
td::Timestamp try_reinit_at_ = td::Timestamp::never();
@ -262,12 +270,26 @@ class AdnlPeerPairImpl : public AdnlPeerPair {
bool has_reverse_addr_ = false;
td::Timestamp request_reverse_ping_after_ = td::Timestamp::now();
bool request_reverse_ping_active_ = false;
struct PacketStats {
double ts_start = 0.0, ts_end = 0.0;
td::uint64 in_packets = 0, in_bytes = 0, in_packets_channel = 0, in_bytes_channel = 0;
td::uint64 out_packets = 0, out_bytes = 0, out_packets_channel = 0, out_bytes_channel = 0;
td::uint64 out_expired_messages = 0, out_expired_bytes = 0;
tl_object_ptr<ton_api::adnl_stats_packets> tl() const;
} packet_stats_cur_, packet_stats_prev_, packet_stats_total_;
double last_in_packet_ts_ = 0.0, last_out_packet_ts_ = 0.0;
double started_ts_ = td::Clocks::system();
void add_packet_stats(td::uint64 bytes, bool in, bool channel);
void add_expired_msg_stats(td::uint64 bytes);
void prepare_packet_stats();
};
class AdnlPeerImpl : public AdnlPeer {
public:
void receive_packet(AdnlNodeIdShort dst, td::uint32 dst_mode, td::actor::ActorId<AdnlLocalId> dst_actor,
AdnlPacket packet) override;
AdnlPacket packet, td::uint64 serialized_size) override;
void send_messages(AdnlNodeIdShort src, td::uint32 src_mode, td::actor::ActorId<AdnlLocalId> src_actor,
std::vector<OutboundAdnlMessage> messages) override;
void send_query(AdnlNodeIdShort src, td::uint32 src_mode, td::actor::ActorId<AdnlLocalId> src_actor, std::string name,
@ -280,6 +302,7 @@ class AdnlPeerImpl : public AdnlPeer {
AdnlAddressList addr_list) override;
void update_dht_node(td::actor::ActorId<dht::Dht> dht_node) override;
void get_conn_ip_str(AdnlNodeIdShort l_id, td::Promise<td::string> promise) override;
void get_stats(td::Promise<std::vector<tl_object_ptr<ton_api::adnl_stats_peerPair>>> promise) override;
//void check_signature(td::BufferSlice data, td::BufferSlice signature, td::Promise<td::Unit> promise) override;
AdnlPeerImpl(td::actor::ActorId<AdnlNetworkManager> network_manager, td::actor::ActorId<AdnlPeerTable> peer_table,

2
adnl/adnl.h
View file

@ -121,6 +121,8 @@ class Adnl : public AdnlSenderInterface {
virtual void create_tunnel(AdnlNodeIdShort dst, td::uint32 size,
td::Promise<std::pair<td::actor::ActorOwn<AdnlTunnel>, AdnlAddress>> promise) = 0;
virtual void get_stats(td::Promise<tl_object_ptr<ton_api::adnl_stats>> promise) = 0;
static td::actor::ActorOwn<Adnl> create(std::string db, td::actor::ActorId<keyring::Keyring> keyring);
static std::string int_to_bytestring(td::int32 id) {

34
adnl/utils.hpp
View file

@ -40,6 +40,40 @@ inline bool adnl_node_is_older(AdnlNode &a, AdnlNode &b) {
return a.addr_list().version() < b.addr_list().version();
}
class RateLimiter {
public:
explicit RateLimiter(td::uint32 capacity, double period) : capacity_(capacity), period_(period), remaining_(capacity) {
}
bool take() {
while (remaining_ < capacity_ && increment_at_.is_in_past()) {
++remaining_;
increment_at_ += period_;
}
if (remaining_) {
--remaining_;
if (increment_at_.is_in_past()) {
increment_at_ = td::Timestamp::in(period_);
}
return true;
}
return false;
}
td::Timestamp ready_at() const {
if (remaining_) {
return td::Timestamp::now();
}
return increment_at_;
}
private:
td::uint32 capacity_;
double period_;
td::uint32 remaining_;
td::Timestamp increment_at_ = td::Timestamp::never();
};
} // namespace adnl
} // namespace ton

2
catchain/catchain-receiver-source.h
View file

@ -61,7 +61,7 @@ class CatChainReceiverSource {
virtual td::BufferSlice fork_proof() const = 0;
virtual bool fork_is_found() const = 0;
// One block can be sent to one node only a limited number of times to prevent DoS
// One block can be sent to one node in catchain.getDifference only a limited number of times to prevent DoS
virtual bool allow_send_block(CatChainBlockHash hash) = 0;
static td::Result<std::unique_ptr<CatChainReceiverSource>> create(CatChainReceiver *chain, PublicKey pub_key,

14
catchain/catchain-receiver.cpp
View file

@ -368,6 +368,12 @@ void CatChainReceiverImpl::add_block(td::BufferSlice payload, std::vector<CatCha
}
int height = prev->height_ + 1;
auto max_block_height = get_max_block_height(opts_, sources_.size());
if (height > max_block_height) {
VLOG(CATCHAIN_WARNING) << this << ": cannot create block: max height exceeded (" << max_block_height << ")";
active_send_ = false;
return;
}
auto block_data = create_tl_object<ton_api::catchain_block_data>(std::move(prev), std::move(deps_arr));
auto block = create_tl_object<ton_api::catchain_block>(incarnation_, local_idx_, height, std::move(block_data),
td::BufferSlice());
@ -697,12 +703,8 @@ void CatChainReceiverImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::cat
} else {
CatChainReceiverSource *S = get_source_by_adnl_id(src);
CHECK(S != nullptr);
if (S->allow_send_block(it->second->get_hash())) {
promise.set_value(serialize_tl_object(create_tl_object<ton_api::catchain_blockResult>(it->second->export_tl()),
true, it->second->get_payload().as_slice()));
} else {
promise.set_error(td::Status::Error("block was requested too many times"));
}
promise.set_value(serialize_tl_object(create_tl_object<ton_api::catchain_blockResult>(it->second->export_tl()),
true, it->second->get_payload().as_slice()));
}
}

12
test/test-adnl.cpp
View file

@ -225,13 +225,19 @@ int main() {
auto f = td::Clocks::system();
scheduler.run_in_context([&] {
for (td::uint32 i = 1; i <= ton::adnl::Adnl::huge_packet_max_size(); i++) {
// Don't send too many packets
// Channels are disabled, so packet rate is limited
for (td::uint32 i : {1, 2, 3, 4, 100, 500, 900}) {
remaining++;
td::actor::send_closure(adnl, &ton::adnl::Adnl::send_message, src, dst, send_packet(i));
}
for (td::uint32 i = 1024; i <= ton::adnl::Adnl::huge_packet_max_size() /* 1024 * 8 */; i += 1024) {
remaining++;
td::actor::send_closure(adnl, &ton::adnl::Adnl::send_message, src, dst, send_packet(i));
}
});
auto t = td::Timestamp::in(320.0);
auto t = td::Timestamp::in(60.0);
while (scheduler.run(1)) {
if (!remaining) {
break;
@ -241,7 +247,7 @@ int main() {
}
}
LOG(ERROR) << "successfully tested delivering of packets of all sizes. Time=" << (td::Clocks::system() - f);
LOG(ERROR) << "successfully tested delivering of packets of various sizes. Time=" << (td::Clocks::system() - f);
scheduler.run_in_context([&] {
td::actor::send_closure(network_manager, &ton::adnl::TestLoopbackNetworkManager::add_node_id, src, true, true);

26
tl/generate/scheme/ton_api.tl
View file

@ -144,7 +144,26 @@ adnl.message.part hash:int256 total_size:int offset:int data:bytes = adnl.Messag
---types---
adnl.db.node.key local_id:int256 peer_id:int256 = adnl.db.Key;
adnl.db.node.value date:int id:PublicKey addr_list:adnl.addressList priority_addr_list:adnl.addressList = adnl.db.node.Value;
adnl.db.node.value date:int id:PublicKey addr_list:adnl.addressList priority_addr_list:adnl.addressList = adnl.db.node.Value;
adnl.stats.packets ts_start:double ts_end:double
in_packets:long in_bytes:long in_packets_channel:long in_bytes_channel:long
out_packets:long out_bytes:long out_packets_channel:long out_bytes_channel:long
out_expired_messages:long out_expired_bytes:long = adnl.stats.Packets;
adnl.stats.peerPair local_id:int256 peer_id:int256 ip_str:string
packets_recent:adnl.stats.packets packets_total:adnl.stats.packets
last_out_packet_ts:double last_in_packet_ts:double
connection_ready:Bool channel_status:int try_reinit_at:double
out_queue_messages:long out_queue_bytes:long
= adnl.stats.PeerPair;
adnl.stats.ipPackets ip_str:string packets:long = adnl.stats.IpPackets;
adnl.stats.localIdPackets ts_start:double ts_end:double
decrypted_packets:(vector adnl.stats.ipPackets) dropped_packets:(vector adnl.stats.ipPackets) = adnl.stats.LocalIdPackets;
adnl.stats.localId short_id:int256
current_decrypt:(vector adnl.stats.ipPackets)
packets_recent:adnl.stats.localIdPackets packets_total:adnl.stats.localIdPackets
peers:(vector adnl.stats.peerPair) = adnl.stats.LocalId;
adnl.stats timestamp:double local_ids:(vector adnl.stats.localId) = adnl.Stats;
---functions---
@ -622,7 +641,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---
@ -737,6 +757,8 @@ engine.validator.setStateSerializerEnabled enabled:Bool = engine.validator.Succe
engine.validator.setCollatorOptionsJson json:string = engine.validator.Success;
engine.validator.getCollatorOptionsJson = engine.validator.JsonConfig;
engine.validator.getAdnlStats = adnl.Stats;
---types---
storage.pong = storage.Pong;

BIN
tl/generate/scheme/ton_api.tlo
View file

Binary file not shown.

163
validator-engine-console/validator-engine-console-query.cpp
View file

@ -1263,3 +1263,166 @@ td::Status GetCollatorOptionsJsonQuery::receive(td::BufferSlice data) {
td::TerminalIO::out() << "saved config to " << file_name_ << "\n";
return td::Status::OK();
}
td::Status GetAdnlStatsJsonQuery::run() {
TRY_RESULT_ASSIGN(file_name_, tokenizer_.get_token<std::string>());
TRY_STATUS(tokenizer_.check_endl());
return td::Status::OK();
}
td::Status GetAdnlStatsJsonQuery::send() {
auto b =
ton::create_serialize_tl_object<ton::ton_api::engine_validator_getAdnlStats>();
td::actor::send_closure(console_, &ValidatorEngineConsole::envelope_send_query, std::move(b), create_promise());
return td::Status::OK();
}
td::Status GetAdnlStatsJsonQuery::receive(td::BufferSlice data) {
TRY_RESULT_PREFIX(f, ton::fetch_tl_object<ton::ton_api::adnl_stats>(data.as_slice(), true),
"received incorrect answer: ");
auto s = td::json_encode<std::string>(td::ToJson(*f), true);
TRY_STATUS(td::write_file(file_name_, s));
td::TerminalIO::out() << "saved adnl stats to " << file_name_ << "\n";
return td::Status::OK();
}
td::Status GetAdnlStatsQuery::run() {
TRY_STATUS(tokenizer_.check_endl());
return td::Status::OK();
}
td::Status GetAdnlStatsQuery::send() {
auto b =
ton::create_serialize_tl_object<ton::ton_api::engine_validator_getAdnlStats>();
td::actor::send_closure(console_, &ValidatorEngineConsole::envelope_send_query, std::move(b), create_promise());
return td::Status::OK();
}
td::Status GetAdnlStatsQuery::receive(td::BufferSlice data) {
TRY_RESULT_PREFIX(stats, ton::fetch_tl_object<ton::ton_api::adnl_stats>(data.as_slice(), true),
"received incorrect answer: ");
td::StringBuilder sb;
sb << "================================= ADNL STATS =================================\n";
bool first = true;
double now = td::Clocks::system();
for (auto &local_id : stats->local_ids_) {
if (first) {
first = false;
} else {
sb << "\n";
}
sb << "LOCAL ID " << local_id->short_id_ << "\n";
if (!local_id->current_decrypt_.empty()) {
std::sort(local_id->current_decrypt_.begin(), local_id->current_decrypt_.end(),
[](const ton::tl_object_ptr<ton::ton_api::adnl_stats_ipPackets> &a,
const ton::tl_object_ptr<ton::ton_api::adnl_stats_ipPackets> &b) {
return a->packets_ > b->packets_;
});
td::uint64 total = 0;
for (auto &x : local_id->current_decrypt_) {
total += x->packets_;
}
sb << " Packets in decryptor: total=" << total;
for (auto &x : local_id->current_decrypt_) {
sb << " " << (x->ip_str_.empty() ? "unknown" : x->ip_str_) << "=" << x->packets_;
}
sb << "\n";
}
auto print_local_id_packets = [&](const std::string &name,
std::vector<ton::tl_object_ptr<ton::ton_api::adnl_stats_ipPackets>> &vec) {
if (vec.empty()) {
return;
}
std::sort(vec.begin(), vec.end(),
[](const ton::tl_object_ptr<ton::ton_api::adnl_stats_ipPackets> &a,
const ton::tl_object_ptr<ton::ton_api::adnl_stats_ipPackets> &b) {
return a->packets_ > b->packets_;
});
td::uint64 total = 0;
for (auto &x : vec) {
total += x->packets_;
}
sb << " " << name << ": total=" << total;
int cnt = 0;
for (auto &x : vec) {
++cnt;
if (cnt >= 8) {
sb << " ...";
break;
}
sb << " " << (x->ip_str_.empty() ? "unknown" : x->ip_str_) << "=" << x->packets_;
}
sb << "\n";
};
print_local_id_packets("Decrypted packets (recent)", local_id->packets_recent_->decrypted_packets_);
print_local_id_packets("Dropped packets (recent)", local_id->packets_recent_->dropped_packets_);
print_local_id_packets("Decrypted packets (total)", local_id->packets_total_->decrypted_packets_);
print_local_id_packets("Dropped packets (total)", local_id->packets_total_->dropped_packets_);
sb << " PEERS (" << local_id->peers_.size() << "):\n";
std::sort(local_id->peers_.begin(), local_id->peers_.end(),
[](const ton::tl_object_ptr<ton::ton_api::adnl_stats_peerPair> &a,
const ton::tl_object_ptr<ton::ton_api::adnl_stats_peerPair> &b) {
return a->packets_recent_->in_bytes_ + a->packets_recent_->out_bytes_ >
b->packets_recent_->in_bytes_ + b->packets_recent_->out_bytes_;
});
for (auto &peer : local_id->peers_) {
sb << " PEER " << peer->peer_id_ << "\n";
sb << " Address: " << (peer->ip_str_.empty() ? "unknown" : peer->ip_str_) << "\n";
sb << " Connection " << (peer->connection_ready_ ? "ready" : "not ready") << ", ";
switch (peer->channel_status_) {
case 0:
sb << "channel: none\n";
break;
case 1:
sb << "channel: inited\n";
break;
case 2:
sb << "channel: ready\n";
break;
default:
sb << "\n";
}
auto print_packets = [&](const std::string &name,
const ton::tl_object_ptr<ton::ton_api::adnl_stats_packets> &obj) {
if (obj->in_packets_) {
sb << " In (" << name << "): " << obj->in_packets_ << " packets ("
<< td::format::as_size(obj->in_bytes_) << "), channel: " << obj->in_packets_channel_ << " packets ("
<< td::format::as_size(obj->in_bytes_channel_) << ")\n";
}
if (obj->out_packets_) {
sb << " Out (" << name << "): " << obj->out_packets_ << " packets ("
<< td::format::as_size(obj->out_bytes_) << "), channel: " << obj->out_packets_channel_ << " packets ("
<< td::format::as_size(obj->out_bytes_channel_) << ")\n";
}
if (obj->out_expired_messages_) {
sb << " Out expired (" << name << "): " << obj->out_expired_messages_ << " messages ("
<< td::format::as_size(obj->out_expired_bytes_) << ")\n";
}
};
print_packets("recent", peer->packets_recent_);
print_packets("total", peer->packets_total_);
sb << " Last in packet: ";
if (peer->last_in_packet_ts_) {
sb << now - peer->last_in_packet_ts_ << " s ago";
} else {
sb << "never";
}
sb << " Last out packet: ";
if (peer->last_out_packet_ts_) {
sb << now - peer->last_out_packet_ts_ << " s ago";
} else {
sb << "never";
}
sb << "\n";
if (peer->out_queue_messages_) {
sb << " Out message queue: " << peer->out_queue_messages_ << " messages ("
<< td::format::as_size(peer->out_queue_bytes_) << ")\n";
}
}
}
sb << "==============================================================================\n";
td::TerminalIO::out() << sb.as_cslice();
return td::Status::OK();
}

44
validator-engine-console/validator-engine-console-query.h
View file

@ -1292,3 +1292,47 @@ class GetCollatorOptionsJsonQuery : public Query {
private:
std::string file_name_;
};
class GetAdnlStatsJsonQuery : public Query {
public:
GetAdnlStatsJsonQuery(td::actor::ActorId<ValidatorEngineConsole> console, Tokenizer tokenizer)
: Query(console, std::move(tokenizer)) {
}
td::Status run() override;
td::Status send() override;
td::Status receive(td::BufferSlice data) override;
static std::string get_name() {
return "getadnlstatsjson";
}
static std::string get_help() {
return "getadnlstatsjson <filename>\tsave adnl stats to <filename>";
}
std::string name() const override {
return get_name();
}
private:
std::string file_name_;
};
class GetAdnlStatsQuery : public Query {
public:
GetAdnlStatsQuery(td::actor::ActorId<ValidatorEngineConsole> console, Tokenizer tokenizer)
: Query(console, std::move(tokenizer)) {
}
td::Status run() override;
td::Status send() override;
td::Status receive(td::BufferSlice data) override;
static std::string get_name() {
return "getadnlstats";
}
static std::string get_help() {
return "getadnlstats\tdisplay adnl stats";
}
std::string name() const override {
return get_name();
}
private:
std::string file_name_;
};

2
validator-engine-console/validator-engine-console.cpp
View file

@ -150,6 +150,8 @@ void ValidatorEngineConsole::run() {
add_query_runner(std::make_unique<QueryRunnerImpl<SetCollatorOptionsJsonQuery>>());
add_query_runner(std::make_unique<QueryRunnerImpl<ResetCollatorOptionsQuery>>());
add_query_runner(std::make_unique<QueryRunnerImpl<GetCollatorOptionsJsonQuery>>());
add_query_runner(std::make_unique<QueryRunnerImpl<GetAdnlStatsJsonQuery>>());
add_query_runner(std::make_unique<QueryRunnerImpl<GetAdnlStatsQuery>>());
}
bool ValidatorEngineConsole::envelope_send_query(td::BufferSlice query, td::Promise<td::BufferSlice> promise) {

47
validator-engine/validator-engine.cpp
View file

@ -1448,6 +1448,9 @@ td::Status ValidatorEngine::load_global_config() {
if (catchain_max_block_delay_) {
validator_options_.write().set_catchain_max_block_delay(catchain_max_block_delay_.value());
}
if (catchain_max_block_delay_slow_) {
validator_options_.write().set_catchain_max_block_delay_slow(catchain_max_block_delay_slow_.value());
}
std::vector<ton::BlockIdExt> h;
for (auto &x : conf.validator_->hardforks_) {
@ -2528,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;
}
@ -3863,6 +3874,28 @@ void ValidatorEngine::run_control_query(ton::ton_api::engine_validator_getCollat
}
}
void ValidatorEngine::run_control_query(ton::ton_api::engine_validator_getAdnlStats &query, td::BufferSlice data,
ton::PublicKeyHash src, td::uint32 perm, td::Promise<td::BufferSlice> promise) {
if (!(perm & ValidatorEnginePermissions::vep_default)) {
promise.set_value(create_control_query_error(td::Status::Error(ton::ErrorCode::error, "not authorized")));
return;
}
if (adnl_.empty()) {
promise.set_value(create_control_query_error(td::Status::Error(ton::ErrorCode::notready, "not started")));
return;
}
td::actor::send_closure(
adnl_, &ton::adnl::Adnl::get_stats,
[promise = std::move(promise)](td::Result<ton::tl_object_ptr<ton::ton_api::adnl_stats>> R) mutable {
if (R.is_ok()) {
promise.set_value(ton::serialize_tl_object(R.move_as_ok(), true));
} else {
promise.set_value(
create_control_query_error(td::Status::Error(ton::ErrorCode::notready, "failed to get adnl stats")));
}
});
}
void ValidatorEngine::process_control_query(td::uint16 port, ton::adnl::AdnlNodeIdShort src,
ton::adnl::AdnlNodeIdShort dst, td::BufferSlice data,
td::Promise<td::BufferSlice> promise) {
@ -4072,7 +4105,7 @@ int main(int argc, char *argv[]) {
logger_ = td::TsFileLog::create(fname.str()).move_as_ok();
td::log_interface = logger_.get();
});
p.add_checked_option('s', "state-ttl", "state will be gc'd after this time (in seconds) default=3600",
p.add_checked_option('s', "state-ttl", "state will be gc'd after this time (in seconds) default=86400",
[&](td::Slice fname) {
auto v = td::to_double(fname);
if (v <= 0) {
@ -4233,7 +4266,7 @@ int main(int argc, char *argv[]) {
"preload all cells from CellDb on startup (recommended to use with big enough celldb-cache-size and celldb-direct-io)",
[&]() { acts.push_back([&x]() { td::actor::send_closure(x, &ValidatorEngine::set_celldb_preload_all, true); }); });
p.add_checked_option(
'\0', "catchain-max-block-delay", "delay before creating a new catchain block, in seconds (default: 0.5)",
'\0', "catchain-max-block-delay", "delay before creating a new catchain block, in seconds (default: 0.4)",
[&](td::Slice s) -> td::Status {
auto v = td::to_double(s);
if (v < 0) {
@ -4242,6 +4275,16 @@ int main(int argc, char *argv[]) {
acts.push_back([&x, v]() { td::actor::send_closure(x, &ValidatorEngine::set_catchain_max_block_delay, v); });
return td::Status::OK();
});
p.add_checked_option(
'\0', "catchain-max-block-delay-slow", "max extended catchain block delay (for too long rounds), (default: 1.0)",
[&](td::Slice s) -> td::Status {
auto v = td::to_double(s);
if (v < 0) {
return td::Status::Error("catchain-max-block-delay-slow should be non-negative");
}
acts.push_back([&x, v]() { td::actor::send_closure(x, &ValidatorEngine::set_catchain_max_block_delay_slow, v); });
return td::Status::OK();
});
p.add_option(
'\0', "fast-state-serializer",
"faster persistent state serializer, but requires more RAM (enabled automatically on machines with >= 90GB RAM)",

7
validator-engine/validator-engine.hpp
View file

@ -214,7 +214,7 @@ class ValidatorEngine : public td::actor::Actor {
td::optional<td::uint64> celldb_cache_size_ = 1LL << 30;
bool celldb_direct_io_ = false;
bool celldb_preload_all_ = false;
td::optional<double> catchain_max_block_delay_;
td::optional<double> catchain_max_block_delay_, catchain_max_block_delay_slow_;
bool read_config_ = false;
bool started_keyring_ = false;
bool started_ = false;
@ -300,6 +300,9 @@ class ValidatorEngine : public td::actor::Actor {
void set_catchain_max_block_delay(double value) {
catchain_max_block_delay_ = value;
}
void set_catchain_max_block_delay_slow(double value) {
catchain_max_block_delay_slow_ = value;
}
void set_fast_state_serializer_enabled(bool value) {
fast_state_serializer_enabled_ = value;
}
@ -489,6 +492,8 @@ class ValidatorEngine : public td::actor::Actor {
ton::PublicKeyHash src, td::uint32 perm, td::Promise<td::BufferSlice> promise);
void run_control_query(ton::ton_api::engine_validator_getCollatorOptionsJson &query, td::BufferSlice data,
ton::PublicKeyHash src, td::uint32 perm, td::Promise<td::BufferSlice> promise);
void run_control_query(ton::ton_api::engine_validator_getAdnlStats &query, td::BufferSlice data,
ton::PublicKeyHash src, td::uint32 perm, td::Promise<td::BufferSlice> promise);
template <class T>
void run_control_query(T &query, td::BufferSlice data, ton::PublicKeyHash src, td::uint32 perm,
td::Promise<td::BufferSlice> promise) {

8
validator-session/validator-session-state.h
View file

@ -478,6 +478,14 @@ class ValidatorSessionState : public ValidatorSessionDescription::RootObject {
auto get_ts(td::uint32 src_idx) const {
return att_->at(src_idx);
}
td::uint32 cur_attempt_in_round(const ValidatorSessionDescription& desc) const {
td::uint32 first_attempt = cur_round_->get_first_attempt(desc.get_self_idx());
td::uint32 cur_attempt = desc.get_attempt_seqno(desc.get_ts());
if (cur_attempt < first_attempt || first_attempt == 0) {
return 0;
}
return cur_attempt - first_attempt;
}
const SentBlock* choose_block_to_sign(ValidatorSessionDescription& desc, td::uint32 src_idx, bool& found) const;
const SentBlock* get_committed_block(ValidatorSessionDescription& desc, td::uint32 seqno) const;

18
validator-session/validator-session.cpp
View file

@ -814,13 +814,25 @@ void ValidatorSessionImpl::request_new_block(bool now) {
} else {
double lambda = 10.0 / description().get_total_nodes();
double x = -1 / lambda * log(td::Random::fast(1, 999) * 0.001);
if (x > catchain_max_block_delay_) { // default = 0.5
x = catchain_max_block_delay_;
}
x = std::min(x, get_current_max_block_delay()); // default = 0.4
td::actor::send_closure(catchain_, &catchain::CatChain::need_new_block, td::Timestamp::in(x));
}
}
double ValidatorSessionImpl::get_current_max_block_delay() const {
td::uint32 att = real_state_->cur_attempt_in_round(*description_);
td::uint32 att1 = description_->opts().max_round_attempts;
if (att <= att1) {
return catchain_max_block_delay_;
}
td::uint32 att2 = att1 + 4;
if (att >= att2) {
return catchain_max_block_delay_slow_;
}
return catchain_max_block_delay_ +
(catchain_max_block_delay_slow_ - catchain_max_block_delay_) * (double)(att - att1) / (double)(att2 - att1);
}
void ValidatorSessionImpl::on_new_round(td::uint32 round) {
if (round != 0) {
CHECK(cur_round_ < round);

2
validator-session/validator-session.h
View file

@ -109,7 +109,7 @@ class ValidatorSession : public td::actor::Actor {
virtual void get_validator_group_info_for_litequery(
td::uint32 cur_round,
td::Promise<std::vector<tl_object_ptr<lite_api::liteServer_nonfinal_candidateInfo>>> promise) = 0;
virtual void set_catchain_max_block_delay(double value) = 0;
virtual void set_catchain_max_block_delay(double delay, double delay_slow) = 0;
static td::actor::ActorOwn<ValidatorSession> create(
catchain::CatChainSessionId session_id, ValidatorSessionOptions opts, PublicKeyHash local_id,

8
validator-session/validator-session.hpp
View file

@ -91,6 +91,7 @@ class ValidatorSessionImpl : public ValidatorSession {
std::unique_ptr<ValidatorSessionDescription> description_;
double catchain_max_block_delay_ = 0.4;
double catchain_max_block_delay_slow_ = 1.0;
void on_new_round(td::uint32 round);
void on_catchain_started();
@ -150,6 +151,7 @@ class ValidatorSessionImpl : public ValidatorSession {
}
void request_new_block(bool now);
double get_current_max_block_delay() const;
void get_broadcast_p2p(PublicKeyHash node, ValidatorSessionFileHash file_hash,
ValidatorSessionCollatedDataFileHash collated_data_file_hash, PublicKeyHash src,
td::uint32 round, ValidatorSessionRootHash root_hash, td::Promise<td::BufferSlice> promise,
@ -191,8 +193,10 @@ class ValidatorSessionImpl : public ValidatorSession {
void get_validator_group_info_for_litequery(
td::uint32 cur_round,
td::Promise<std::vector<tl_object_ptr<lite_api::liteServer_nonfinal_candidateInfo>>> promise) override;
void set_catchain_max_block_delay(double value) override {
catchain_max_block_delay_ = value;
void set_catchain_max_block_delay(double delay, double delay_slow) override {
catchain_max_block_delay_ = delay;
catchain_max_block_delay_slow_ = delay_slow;
}
void process_blocks(std::vector<catchain::CatChainBlock *> blocks);

13
validator/db/celldb.cpp
View file

@ -188,12 +188,17 @@ void CellDbIn::store_cell(BlockIdExt block_id, td::Ref<vm::Cell> cell, td::Promi
if (!opts_->get_disable_rocksdb_stats()) {
cell_db_statistics_.store_cell_time_.insert(timer.elapsed() * 1e6);
}
LOG(DEBUG) << "Stored state " << block_id.to_str();
}
void CellDbIn::get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> promise) {
promise.set_result(boc_->get_cell_db_reader());
}
void CellDbIn::get_last_deleted_mc_state(td::Promise<BlockSeqno> promise) {
promise.set_result(last_deleted_mc_state_);
}
void CellDbIn::flush_db_stats() {
if (opts_->get_disable_rocksdb_stats()) {
return;
@ -320,6 +325,10 @@ void CellDbIn::gc_cont2(BlockHandle handle) {
if (!opts_->get_disable_rocksdb_stats()) {
cell_db_statistics_.gc_cell_time_.insert(timer.elapsed() * 1e6);
}
if (handle->id().is_masterchain()) {
last_deleted_mc_state_ = handle->id().seqno();
}
LOG(DEBUG) << "Deleted state " << handle->id().to_str();
}
void CellDbIn::skip_gc() {
@ -453,6 +462,10 @@ void CellDb::get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> p
td::actor::send_closure(cell_db_, &CellDbIn::get_cell_db_reader, std::move(promise));
}
void CellDb::get_last_deleted_mc_state(td::Promise<BlockSeqno> promise) {
td::actor::send_closure(cell_db_, &CellDbIn::get_last_deleted_mc_state, std::move(promise));
}
void CellDb::start_up() {
CellDbBase::start_up();
boc_ = vm::DynamicBagOfCellsDb::create();

3
validator/db/celldb.hpp
View file

@ -61,6 +61,7 @@ class CellDbIn : public CellDbBase {
void load_cell(RootHash hash, td::Promise<td::Ref<vm::DataCell>> promise);
void store_cell(BlockIdExt block_id, td::Ref<vm::Cell> cell, td::Promise<td::Ref<vm::DataCell>> promise);
void get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> promise);
void get_last_deleted_mc_state(td::Promise<BlockSeqno> promise);
void migrate_cell(td::Bits256 hash);
@ -143,6 +144,7 @@ class CellDbIn : public CellDbBase {
std::shared_ptr<td::RocksDbSnapshotStatistics> snapshot_statistics_;
CellDbStatistics cell_db_statistics_;
td::Timestamp statistics_flush_at_ = td::Timestamp::never();
BlockSeqno last_deleted_mc_state_ = 0;
public:
class MigrationProxy : public td::actor::Actor {
@ -167,6 +169,7 @@ class CellDb : public CellDbBase {
boc_->set_loader(std::make_unique<vm::CellLoader>(std::move(snapshot), on_load_callback_)).ensure();
}
void get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> promise);
void get_last_deleted_mc_state(td::Promise<BlockSeqno> promise);
CellDb(td::actor::ActorId<RootDb> root_db, std::string path, td::Ref<ValidatorManagerOptions> opts)
: root_db_(root_db), path_(path), opts_(opts) {

4
validator/db/rootdb.cpp
View file

@ -285,6 +285,10 @@ void RootDb::get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> p
td::actor::send_closure(cell_db_, &CellDb::get_cell_db_reader, std::move(promise));
}
void RootDb::get_last_deleted_mc_state(td::Promise<BlockSeqno> promise) {
td::actor::send_closure(cell_db_, &CellDb::get_last_deleted_mc_state, std::move(promise));
}
void RootDb::store_persistent_state_file(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::BufferSlice state,
td::Promise<td::Unit> promise) {
td::actor::send_closure(archive_db_, &ArchiveManager::add_persistent_state, block_id, masterchain_block_id,

1
validator/db/rootdb.hpp
View file

@ -64,6 +64,7 @@ class RootDb : public Db {
td::Promise<td::Ref<ShardState>> promise) override;
void get_block_state(ConstBlockHandle handle, td::Promise<td::Ref<ShardState>> promise) override;
void get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> promise) override;
void get_last_deleted_mc_state(td::Promise<BlockSeqno> promise) override;
void store_block_handle(BlockHandle handle, td::Promise<td::Unit> promise) override;
void get_block_handle(BlockIdExt id, td::Promise<BlockHandle> promise) override;

32
validator/impl/collator.cpp
View file

@ -3064,7 +3064,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;
@ -3695,6 +3695,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_) {
@ -3711,9 +3713,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;
@ -3733,7 +3756,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);

2
validator/impl/liteserver.cpp
View file

@ -870,7 +870,7 @@ void LiteQuery::perform_runSmcMethod(BlockIdExt blkid, WorkchainId workchain, St
vm::FakeVmStateLimits fstate(1000); // limit recursive (de)serialization calls
vm::VmStateInterface::Guard guard(&fstate);
auto cs = vm::load_cell_slice(res.move_as_ok());
if (!(vm::Stack::deserialize_to(cs, stack_, 0) && cs.empty_ext())) {
if (!(vm::Stack::deserialize_to(cs, stack_, 2 /* no continuations */) && cs.empty_ext())) {
fatal_error("parameter list boc cannot be deserialized as a VmStack");
return;
}

2
validator/impl/shard.cpp
View file

@ -44,6 +44,7 @@ ShardStateQ::ShardStateQ(const ShardStateQ& other)
, root(other.root)
, lt(other.lt)
, utime(other.utime)
, global_id_(other.global_id_)
, before_split_(other.before_split_)
, fake_split_(other.fake_split_)
, fake_merge_(other.fake_merge_) {
@ -121,6 +122,7 @@ td::Status ShardStateQ::init() {
}
lt = info.gen_lt;
utime = info.gen_utime;
global_id_ = info.global_id;
before_split_ = info.before_split;
block::ShardId id{info.shard_id};
ton::BlockId hdr_id{ton::ShardIdFull(id), info.seq_no};

4
validator/impl/shard.hpp
View file

@ -38,6 +38,7 @@ class ShardStateQ : virtual public ShardState {
Ref<vm::Cell> root;
LogicalTime lt{0};
UnixTime utime{0};
td::int32 global_id_{0};
bool before_split_{false};
bool fake_split_{false};
bool fake_merge_{false};
@ -81,6 +82,9 @@ class ShardStateQ : virtual public ShardState {
LogicalTime get_logical_time() const override {
return lt;
}
td::int32 get_global_id() const override {
return global_id_;
}
td::optional<BlockIdExt> get_master_ref() const override {
return master_ref;
}

1
validator/interfaces/db.h
View file

@ -52,6 +52,7 @@ class Db : public td::actor::Actor {
td::Promise<td::Ref<ShardState>> promise) = 0;
virtual void get_block_state(ConstBlockHandle handle, td::Promise<td::Ref<ShardState>> promise) = 0;
virtual void get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> promise) = 0;
virtual void get_last_deleted_mc_state(td::Promise<BlockSeqno> promise) = 0;
virtual void store_persistent_state_file(BlockIdExt block_id, BlockIdExt masterchain_block_id, td::BufferSlice state,
td::Promise<td::Unit> promise) = 0;

1
validator/interfaces/shard.h
View file

@ -39,6 +39,7 @@ class ShardState : public td::CntObject {
virtual UnixTime get_unix_time() const = 0;
virtual LogicalTime get_logical_time() const = 0;
virtual td::int32 get_global_id() const = 0;
virtual ShardIdFull get_shard() const = 0;
virtual BlockSeqno get_seqno() const = 0;
virtual BlockIdExt get_block_id() const = 0;

25
validator/manager.cpp
View file

@ -2186,7 +2186,13 @@ void ValidatorManagerImpl::update_shards() {
last_masterchain_state_->get_total_validator_set(0)->export_vector().size());
}
}
} // namespace validator
if (!serializer_.empty()) {
td::actor::send_closure(
serializer_, &AsyncStateSerializer::auto_disable_serializer,
!validator_groups_.empty() && last_masterchain_state_->get_global_id() == -239); // mainnet only
}
}
void ValidatorManagerImpl::written_destroyed_validator_sessions(std::vector<td::actor::ActorId<ValidatorGroup>> list) {
for (auto &v : list) {
@ -2779,6 +2785,23 @@ void ValidatorManagerImpl::prepare_stats(td::Promise<std::vector<std::pair<std::
vec.emplace_back("rotatemasterchainblock", last_rotate_block_id_.to_str());
//vec.emplace_back("shardclientmasterchainseqno", td::to_string(min_confirmed_masterchain_seqno_));
vec.emplace_back("stateserializermasterchainseqno", td::to_string(state_serializer_masterchain_seqno_));
td::actor::send_closure(db_, &Db::get_last_deleted_mc_state,
[promise = merger.make_promise(""),
gc_seqno = gc_masterchain_handle_->id().seqno()](td::Result<BlockSeqno> R) mutable {
TRY_RESULT_PROMISE(promise, seqno, std::move(R));
std::string s;
if (seqno == 0) {
s = "none";
} else if (seqno <= gc_seqno) {
s = PSTRING() << seqno << " (gc_seqno-" << (gc_seqno - seqno) << ")";
} else {
s = PSTRING() << seqno << " (gc_seqno+" << (seqno - gc_seqno) << ")";
}
std::vector<std::pair<std::string, std::string>> vec;
vec.emplace_back("lastgcdmasterchainstate", std::move(s));
promise.set_value(std::move(vec));
});
}
if (!shard_client_.empty()) {

2
validator/state-serializer.cpp
View file

@ -378,6 +378,7 @@ void AsyncStateSerializer::got_shard_handle(BlockHandle handle) {
void AsyncStateSerializer::got_shard_state(BlockHandle handle, td::Ref<ShardState> state,
std::shared_ptr<vm::CellDbReader> cell_db_reader) {
next_idx_++;
if (!opts_->get_state_serializer_enabled() || auto_disabled_) {
success_handler();
return;
@ -406,7 +407,6 @@ void AsyncStateSerializer::got_shard_state(BlockHandle handle, td::Ref<ShardStat
});
td::actor::send_closure(manager_, &ValidatorManager::store_persistent_state_file_gen, handle->id(),
masterchain_handle_->id(), write_data, std::move(P));
next_idx_++;
}
void AsyncStateSerializer::fail_handler(td::Status reason) {

10
validator/validator-group.cpp
View file

@ -348,10 +348,12 @@ void ValidatorGroup::create_session() {
<< ".",
allow_unsafe_self_blocks_resync_);
}
if (opts_->get_catchain_max_block_delay()) {
td::actor::send_closure(session_, &validatorsession::ValidatorSession::set_catchain_max_block_delay,
opts_->get_catchain_max_block_delay().value());
}
double catchain_delay = opts_->get_catchain_max_block_delay() ? opts_->get_catchain_max_block_delay().value() : 0.4;
double catchain_delay_slow =
std::max(catchain_delay,
opts_->get_catchain_max_block_delay_slow() ? opts_->get_catchain_max_block_delay_slow().value() : 1.0);
td::actor::send_closure(session_, &validatorsession::ValidatorSession::set_catchain_max_block_delay, catchain_delay,
catchain_delay_slow);
if (started_) {
td::actor::send_closure(session_, &validatorsession::ValidatorSession::start);
}

8
validator/validator-options.hpp
View file

@ -141,6 +141,9 @@ struct ValidatorManagerOptionsImpl : public ValidatorManagerOptions {
td::optional<double> get_catchain_max_block_delay() const override {
return catchain_max_block_delay_;
}
td::optional<double> get_catchain_max_block_delay_slow() const override {
return catchain_max_block_delay_slow_;
}
bool get_state_serializer_enabled() const override {
return state_serializer_enabled_;
}
@ -230,6 +233,9 @@ struct ValidatorManagerOptionsImpl : public ValidatorManagerOptions {
void set_catchain_max_block_delay(double value) override {
catchain_max_block_delay_ = value;
}
void set_catchain_max_block_delay_slow(double value) override {
catchain_max_block_delay_slow_ = value;
}
void set_state_serializer_enabled(bool value) override {
state_serializer_enabled_ = value;
}
@ -289,7 +295,7 @@ struct ValidatorManagerOptionsImpl : public ValidatorManagerOptions {
td::optional<td::uint64> celldb_cache_size_;
bool celldb_direct_io_ = false;
bool celldb_preload_all_ = false;
td::optional<double> catchain_max_block_delay_;
td::optional<double> catchain_max_block_delay_, catchain_max_block_delay_slow_;
bool state_serializer_enabled_ = true;
td::Ref<CollatorOptions> collator_options_{true};
bool fast_state_serializer_enabled_ = false;

7
validator/validator.h
View file

@ -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 {
@ -105,6 +110,7 @@ struct ValidatorManagerOptions : public td::CntObject {
virtual bool get_celldb_direct_io() const = 0;
virtual bool get_celldb_preload_all() const = 0;
virtual td::optional<double> get_catchain_max_block_delay() const = 0;
virtual td::optional<double> get_catchain_max_block_delay_slow() const = 0;
virtual bool get_state_serializer_enabled() const = 0;
virtual td::Ref<CollatorOptions> get_collator_options() const = 0;
virtual bool get_fast_state_serializer_enabled() const = 0;
@ -136,6 +142,7 @@ struct ValidatorManagerOptions : public td::CntObject {
virtual void set_celldb_direct_io(bool value) = 0;
virtual void set_celldb_preload_all(bool value) = 0;
virtual void set_catchain_max_block_delay(double value) = 0;
virtual void set_catchain_max_block_delay_slow(double value) = 0;
virtual void set_state_serializer_enabled(bool value) = 0;
virtual void set_collator_options(td::Ref<CollatorOptions> value) = 0;
virtual void set_fast_state_serializer_enabled(bool value) = 0;