diff --git a/adnl/adnl-channel.cpp b/adnl/adnl-channel.cpp index 5c8229ca4..4da9d2eed 100644 --- a/adnl/adnl-channel.cpp +++ b/adnl/adnl-channel.cpp @@ -112,16 +112,16 @@ void AdnlChannelImpl::send_message(td::uint32 priority, td::actor::ActorId 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 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)); } diff --git a/adnl/adnl-local-id.cpp b/adnl/adnl-local-id.cpp index b48182763..d72fc7bcc 100644 --- a/adnl/adnl-local-id.cpp +++ b/adnl/adnl-local-id.cpp @@ -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 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 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> promise) { + auto stats = create_tl_object(); + 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( + 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 AdnlLocalId::PacketStats::tl() const { + auto obj = create_tl_object(); + 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( + 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( + ip.is_valid() ? PSTRING() << ip.get_ip_str() << ":" << ip.get_port() : "", packets)); + } + return obj; +} + + } // namespace adnl } // namespace ton diff --git a/adnl/adnl-local-id.h b/adnl/adnl-local-id.h index c9ecfff16..be9d79d2c 100644 --- a/adnl/adnl-local-id.h +++ b/adnl/adnl-local-id.h @@ -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 promise); void receive(td::IPAddress addr, td::BufferSlice data); + void decrypt_packet_done(td::IPAddress addr); void subscribe(std::string prefix, std::unique_ptr 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 promise); + void get_stats(td::Promise> 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 inbound_rate_limiter_; + struct PacketStats { + double ts_start = 0.0, ts_end = 0.0; + std::map decrypted_packets; + std::map dropped_packets; + + tl_object_ptr 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(); }; diff --git a/adnl/adnl-peer-table.cpp b/adnl/adnl-peer-table.cpp index 548915159..d885623a9 100644 --- a/adnl/adnl-peer-table.cpp +++ b/adnl/adnl-peer-table.cpp @@ -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> promise) { + class Cb : public td::actor::Actor { + public: + explicit Cb(td::Promise> promise) : promise_(std::move(promise)) { + } + + void got_local_id_stats(tl_object_ptr 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> 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(); + 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(); + 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> promise_; + size_t pending_ = 1; + + std::map> local_id_stats_; + }; + auto callback = td::actor::create_actor("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> 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>> 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 diff --git a/adnl/adnl-peer-table.h b/adnl/adnl-peer-table.h index cb7da6135..055f32ac1 100644 --- a/adnl/adnl-peer-table.h +++ b/adnl/adnl-peer-table.h @@ -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, diff --git a/adnl/adnl-peer-table.hpp b/adnl/adnl-peer-table.hpp index 1c30b84c7..12f64fcb2 100644 --- a/adnl/adnl-peer-table.hpp +++ b/adnl/adnl-peer-table.hpp @@ -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, AdnlAddress>> promise) override; void get_conn_ip_str(AdnlNodeIdShort l_id, AdnlNodeIdShort p_id, td::Promise promise) override; + void get_stats(td::Promise> promise) override; + struct PrintId {}; PrintId print_id() const { return PrintId{}; diff --git a/adnl/adnl-peer.cpp b/adnl/adnl-peer.cpp index d82486fed..2b0077a8e 100644 --- a/adnl/adnl-peer.cpp +++ b/adnl/adnl-peer.cpp @@ -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 messages, bool allow_postpone) { - for (td::int32 idx = 0; idx < 2; idx++) { - std::vector 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 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_); + size_t s = (via_channel ? channel_packet_header_max_size() : packet_header_max_size()); + if (first) { + s += 2 * addr_list_max_size(); + } - 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)); - } + 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 (!addr_list_.empty()) { - packet.set_received_addr_list_version(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()); + } + + 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; } - if (!priority_addr_list_.empty()) { - packet.set_received_priority_addr_list_version(priority_addr_list_.version()); + 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_); + } - 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; + 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)); } - CHECK(M.size() <= get_mtu()); + } 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(M.release()); - ptr++; - } else { - break; + 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 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); } - - packet.run_basic_checks().ensure(); - auto P = td::PromiseCreator::lambda([SelfId = actor_id(this), conn, id = print_id(), - via_channel](td::Result 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; - } + }); + + 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 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 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, bool>> AdnlPeerPairImpl::get_conn(bool direct_only) { +td::Result, 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, 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 promise) { promise.set_value("undefined"); } +void AdnlPeerPairImpl::get_stats(td::Promise> promise) { + auto stats = create_tl_object(); + 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 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::create(td::actor::ActorId 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 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>> promise) { + class Cb : public td::actor::Actor { + public: + explicit Cb(td::Promise>> promise) + : promise_(std::move(promise)) { + } + + void got_peer_pair_stats(tl_object_ptr 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>> promise_; + size_t pending_ = 1; + std::vector> result_; + }; + auto callback = td::actor::create_actor("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> 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 R) { received_from_db_ = false; if (R.is_error()) { @@ -1016,6 +1126,66 @@ void AdnlPeerPairImpl::request_reverse_ping_result(td::Result 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 AdnlPeerPairImpl::PacketStats::tl() const { + return create_tl_object(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 diff --git a/adnl/adnl-peer.h b/adnl/adnl-peer.h index 8488e82ee..b7d6adc0f 100644 --- a/adnl/adnl-peer.h +++ b/adnl/adnl-peer.h @@ -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 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 promise) = 0; + virtual void get_stats(td::Promise> promise) = 0; static td::actor::ActorOwn create(td::actor::ActorId network_manager, td::actor::ActorId 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 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 src_actor, std::vector messages) = 0; virtual void send_query(AdnlNodeIdShort src, td::uint32 src_mode, td::actor::ActorId src_actor, @@ -100,6 +101,7 @@ class AdnlPeer : public td::actor::Actor { td::actor::ActorId local_actor, AdnlAddressList addr_list) = 0; virtual void update_dht_node(td::actor::ActorId dht_node) = 0; virtual void get_conn_ip_str(AdnlNodeIdShort l_id, td::Promise promise) = 0; + virtual void get_stats(td::Promise>> promise) = 0; }; } // namespace adnl diff --git a/adnl/adnl-peer.hpp b/adnl/adnl-peer.hpp index 40c9eb088..d25a24cf4 100644 --- a/adnl/adnl-peer.hpp +++ b/adnl/adnl-peer.hpp @@ -20,6 +20,7 @@ #include #include +#include #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 messages, bool allow_postpone); + void send_messages_from_queue(); void send_messages(std::vector messages) override; void send_packet_continue(AdnlPacket packet, td::actor::ActorId conn, bool via_channel); void send_query(std::string name, td::Promise 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 promise) override; + void get_stats(td::Promise> promise) override; void got_data_from_db(td::Result R); void got_data_from_static_nodes(td::Result R); @@ -124,7 +126,7 @@ class AdnlPeerPairImpl : public AdnlPeerPair { private: void respond_with_nop(); void reinit(td::int32 date); - td::Result, bool>> get_conn(bool direct_only); + td::Result, 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); }; - std::vector pending_messages_; + // Messages waiting for connection or for nochannel rate limiter + std::queue> 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 network_manager_; td::actor::ActorId 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 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 dst_actor, - AdnlPacket packet) override; + AdnlPacket packet, td::uint64 serialized_size) override; void send_messages(AdnlNodeIdShort src, td::uint32 src_mode, td::actor::ActorId src_actor, std::vector messages) override; void send_query(AdnlNodeIdShort src, td::uint32 src_mode, td::actor::ActorId 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_node) override; void get_conn_ip_str(AdnlNodeIdShort l_id, td::Promise promise) override; + void get_stats(td::Promise>> promise) override; //void check_signature(td::BufferSlice data, td::BufferSlice signature, td::Promise promise) override; AdnlPeerImpl(td::actor::ActorId network_manager, td::actor::ActorId peer_table, diff --git a/adnl/adnl.h b/adnl/adnl.h index a1c39d5e4..a276e0c21 100644 --- a/adnl/adnl.h +++ b/adnl/adnl.h @@ -121,6 +121,8 @@ class Adnl : public AdnlSenderInterface { virtual void create_tunnel(AdnlNodeIdShort dst, td::uint32 size, td::Promise, AdnlAddress>> promise) = 0; + virtual void get_stats(td::Promise> promise) = 0; + static td::actor::ActorOwn create(std::string db, td::actor::ActorId keyring); static std::string int_to_bytestring(td::int32 id) { diff --git a/adnl/utils.hpp b/adnl/utils.hpp index 50aec2efb..18d3f2074 100644 --- a/adnl/utils.hpp +++ b/adnl/utils.hpp @@ -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 diff --git a/tl/generate/scheme/ton_api.tl b/tl/generate/scheme/ton_api.tl index bf919b0fd..e15e18612 100644 --- a/tl/generate/scheme/ton_api.tl +++ b/tl/generate/scheme/ton_api.tl @@ -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--- @@ -723,6 +742,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; diff --git a/tl/generate/scheme/ton_api.tlo b/tl/generate/scheme/ton_api.tlo index 337dd071e..3d39734ef 100644 Binary files a/tl/generate/scheme/ton_api.tlo and b/tl/generate/scheme/ton_api.tlo differ diff --git a/validator-engine-console/validator-engine-console-query.cpp b/validator-engine-console/validator-engine-console-query.cpp index 41721ab96..372fa8121 100644 --- a/validator-engine-console/validator-engine-console-query.cpp +++ b/validator-engine-console/validator-engine-console-query.cpp @@ -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()); + TRY_STATUS(tokenizer_.check_endl()); + return td::Status::OK(); +} + +td::Status GetAdnlStatsJsonQuery::send() { + auto b = + ton::create_serialize_tl_object(); + 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(data.as_slice(), true), + "received incorrect answer: "); + auto s = td::json_encode(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(); + 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(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 &a, + const ton::tl_object_ptr &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> &vec) { + if (vec.empty()) { + return; + } + std::sort(vec.begin(), vec.end(), + [](const ton::tl_object_ptr &a, + const ton::tl_object_ptr &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 &a, + const ton::tl_object_ptr &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 &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(); +} diff --git a/validator-engine-console/validator-engine-console-query.h b/validator-engine-console/validator-engine-console-query.h index 08ac1572a..6314d6199 100644 --- a/validator-engine-console/validator-engine-console-query.h +++ b/validator-engine-console/validator-engine-console-query.h @@ -1292,3 +1292,47 @@ class GetCollatorOptionsJsonQuery : public Query { private: std::string file_name_; }; + +class GetAdnlStatsJsonQuery : public Query { + public: + GetAdnlStatsJsonQuery(td::actor::ActorId 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 \tsave adnl stats to "; + } + std::string name() const override { + return get_name(); + } + + private: + std::string file_name_; +}; + +class GetAdnlStatsQuery : public Query { + public: + GetAdnlStatsQuery(td::actor::ActorId 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_; +}; diff --git a/validator-engine-console/validator-engine-console.cpp b/validator-engine-console/validator-engine-console.cpp index d8a230801..1ec0f3803 100644 --- a/validator-engine-console/validator-engine-console.cpp +++ b/validator-engine-console/validator-engine-console.cpp @@ -150,6 +150,8 @@ void ValidatorEngineConsole::run() { add_query_runner(std::make_unique>()); add_query_runner(std::make_unique>()); add_query_runner(std::make_unique>()); + add_query_runner(std::make_unique>()); + add_query_runner(std::make_unique>()); } bool ValidatorEngineConsole::envelope_send_query(td::BufferSlice query, td::Promise promise) { diff --git a/validator-engine/validator-engine.cpp b/validator-engine/validator-engine.cpp index 11f1a9a28..31f79275e 100644 --- a/validator-engine/validator-engine.cpp +++ b/validator-engine/validator-engine.cpp @@ -3866,6 +3866,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 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> 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 promise) { diff --git a/validator-engine/validator-engine.hpp b/validator-engine/validator-engine.hpp index 285e075f8..9b5d39791 100644 --- a/validator-engine/validator-engine.hpp +++ b/validator-engine/validator-engine.hpp @@ -492,6 +492,8 @@ class ValidatorEngine : public td::actor::Actor { ton::PublicKeyHash src, td::uint32 perm, td::Promise promise); void run_control_query(ton::ton_api::engine_validator_getCollatorOptionsJson &query, td::BufferSlice data, ton::PublicKeyHash src, td::uint32 perm, td::Promise promise); + void run_control_query(ton::ton_api::engine_validator_getAdnlStats &query, td::BufferSlice data, + ton::PublicKeyHash src, td::uint32 perm, td::Promise promise); template void run_control_query(T &query, td::BufferSlice data, ton::PublicKeyHash src, td::uint32 perm, td::Promise promise) {