diff --git a/validator/full-node-shard.cpp b/validator/full-node-shard.cpp index c351653f4..7d33a1950 100644 --- a/validator/full-node-shard.cpp +++ b/validator/full-node-shard.cpp @@ -673,7 +673,7 @@ void FullNodeShardImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::tonNod void FullNodeShardImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::tonNode_getOutMsgQueueProof &query, td::Promise promise) { std::vector blocks; - for (const auto& x : query.blocks_) { + for (const auto &x : query.blocks_) { BlockIdExt id = create_block_id(x); if (!id.is_valid_ext()) { promise.set_error(td::Status::Error("invalid block_id")); @@ -691,18 +691,14 @@ void FullNodeShardImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::tonNod return; } block::ImportedMsgQueueLimits limits{(td::uint32)query.limits_->max_bytes_, (td::uint32)query.limits_->max_msgs_}; - if (limits.max_bytes > (1 << 24)) { + if (limits.max_msgs > 512) { + promise.set_error(td::Status::Error("max_msgs is too big")); + return; + } + if (limits.max_bytes > (1 << 21)) { promise.set_error(td::Status::Error("max_bytes is too big")); return; } - auto P = td::PromiseCreator::lambda( - [promise = std::move(promise)](td::Result> R) mutable { - if (R.is_error()) { - promise.set_result(create_serialize_tl_object()); - } else { - promise.set_result(serialize_tl_object(R.move_as_ok(), true)); - } - }); FLOG(DEBUG) { sb << "Got query getOutMsgQueueProof to shard " << dst_shard.to_str() << " from blocks"; for (const BlockIdExt &id : blocks) { @@ -710,9 +706,24 @@ void FullNodeShardImpl::process_query(adnl::AdnlNodeIdShort src, ton_api::tonNod } sb << " from " << src; }; - td::actor::create_actor("buildqueueproof", dst_shard, std::move(blocks), limits, - validator_manager_, std::move(P)) - .release(); + td::actor::send_closure( + full_node_, &FullNode::get_out_msg_queue_query_token, + [=, manager = validator_manager_, blocks = std::move(blocks), + promise = std::move(promise)](td::Result> R) mutable { + TRY_RESULT_PROMISE(promise, token, std::move(R)); + auto P = + td::PromiseCreator::lambda([promise = std::move(promise), token = std::move(token)]( + td::Result> R) mutable { + if (R.is_error()) { + promise.set_result(create_serialize_tl_object()); + } else { + promise.set_result(serialize_tl_object(R.move_as_ok(), true)); + } + }); + td::actor::create_actor("buildqueueproof", dst_shard, std::move(blocks), limits, manager, + std::move(P)) + .release(); + }); } void FullNodeShardImpl::receive_query(adnl::AdnlNodeIdShort src, td::BufferSlice query, diff --git a/validator/full-node.cpp b/validator/full-node.cpp index bb0377ff9..658cb34e3 100644 --- a/validator/full-node.cpp +++ b/validator/full-node.cpp @@ -21,6 +21,7 @@ #include "td/actor/MultiPromise.h" #include "full-node.h" #include "common/delay.h" +#include "impl/out-msg-queue-proof.hpp" #include "td/utils/Random.h" #include "ton/ton-tl.hpp" @@ -575,6 +576,11 @@ void FullNodeImpl::process_block_candidate_broadcast(BlockIdExt block_id, Catcha std::move(data)); } +void FullNodeImpl::get_out_msg_queue_query_token(td::Promise> promise) { + td::actor::send_closure(out_msg_queue_query_token_manager_, &TokenManager::get_token, 1, 0, td::Timestamp::in(10.0), + std::move(promise)); +} + void FullNodeImpl::set_validator_telemetry_filename(std::string value) { validator_telemetry_filename_ = std::move(value); update_validator_telemetry_collector(); diff --git a/validator/full-node.h b/validator/full-node.h index 73ecbd720..fdb1bf3b0 100644 --- a/validator/full-node.h +++ b/validator/full-node.h @@ -91,6 +91,7 @@ class FullNode : public td::actor::Actor { virtual void process_block_broadcast(BlockBroadcast broadcast) = 0; virtual void process_block_candidate_broadcast(BlockIdExt block_id, CatchainSeqno cc_seqno, td::uint32 validator_set_hash, td::BufferSlice data) = 0; + virtual void get_out_msg_queue_query_token(td::Promise> promise) = 0; virtual void set_validator_telemetry_filename(std::string value) = 0; diff --git a/validator/full-node.hpp b/validator/full-node.hpp index 00a125c7d..0ea6fa0b3 100644 --- a/validator/full-node.hpp +++ b/validator/full-node.hpp @@ -28,6 +28,7 @@ #include #include #include +#include namespace ton { @@ -90,6 +91,7 @@ class FullNodeImpl : public FullNode { void process_block_broadcast(BlockBroadcast broadcast) override; void process_block_candidate_broadcast(BlockIdExt block_id, CatchainSeqno cc_seqno, td::uint32 validator_set_hash, td::BufferSlice data) override; + void get_out_msg_queue_query_token(td::Promise> promise) override; void set_validator_telemetry_filename(std::string value) override; @@ -163,6 +165,9 @@ class FullNodeImpl : public FullNode { PublicKeyHash validator_telemetry_collector_key_ = PublicKeyHash::zero(); void update_validator_telemetry_collector(); + + td::actor::ActorOwn out_msg_queue_query_token_manager_ = + td::actor::create_actor("tokens", /* max_tokens = */ 1); }; } // namespace fullnode diff --git a/validator/manager-disk.hpp b/validator/manager-disk.hpp index 76a95aa71..cd06bf555 100644 --- a/validator/manager-disk.hpp +++ b/validator/manager-disk.hpp @@ -288,7 +288,7 @@ class ValidatorManagerImpl : public ValidatorManager { void try_get_static_file(FileHash file_hash, td::Promise promise) override; void get_download_token(size_t download_size, td::uint32 priority, td::Timestamp timeout, - td::Promise> promise) override { + td::Promise> promise) override { promise.set_error(td::Status::Error(ErrorCode::error, "download disabled")); } diff --git a/validator/manager-hardfork.hpp b/validator/manager-hardfork.hpp index f55b4bc0c..0b8b9e736 100644 --- a/validator/manager-hardfork.hpp +++ b/validator/manager-hardfork.hpp @@ -362,7 +362,7 @@ class ValidatorManagerImpl : public ValidatorManager { void try_get_static_file(FileHash file_hash, td::Promise promise) override; void get_download_token(size_t download_size, td::uint32 priority, td::Timestamp timeout, - td::Promise> promise) override { + td::Promise> promise) override { promise.set_error(td::Status::Error(ErrorCode::error, "download disabled")); } diff --git a/validator/manager.hpp b/validator/manager.hpp index 26235853c..519cab12a 100644 --- a/validator/manager.hpp +++ b/validator/manager.hpp @@ -527,8 +527,8 @@ class ValidatorManagerImpl : public ValidatorManager { void try_get_static_file(FileHash file_hash, td::Promise promise) override; void get_download_token(size_t download_size, td::uint32 priority, td::Timestamp timeout, - td::Promise> promise) override { - td::actor::send_closure(token_manager_, &TokenManager::get_download_token, download_size, priority, timeout, + td::Promise> promise) override { + td::actor::send_closure(token_manager_, &TokenManager::get_token, download_size, priority, timeout, std::move(promise)); } diff --git a/validator/net/download-block-new.cpp b/validator/net/download-block-new.cpp index e9a193b46..37580cef1 100644 --- a/validator/net/download-block-new.cpp +++ b/validator/net/download-block-new.cpp @@ -144,7 +144,7 @@ void DownloadBlockNew::got_block_handle(BlockHandle handle) { return; } - auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result> R) { + auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result> R) { if (R.is_error()) { td::actor::send_closure(SelfId, &DownloadBlockNew::abort_query, R.move_as_error_prefix("failed to get download token: ")); @@ -156,7 +156,7 @@ void DownloadBlockNew::got_block_handle(BlockHandle handle) { std::move(P)); } -void DownloadBlockNew::got_download_token(std::unique_ptr token) { +void DownloadBlockNew::got_download_token(std::unique_ptr token) { token_ = std::move(token); if (download_from_.is_zero() && client_.empty()) { diff --git a/validator/net/download-block-new.hpp b/validator/net/download-block-new.hpp index d2a0e1363..ecd062ee3 100644 --- a/validator/net/download-block-new.hpp +++ b/validator/net/download-block-new.hpp @@ -49,7 +49,7 @@ class DownloadBlockNew : public td::actor::Actor { void start_up() override; void got_block_handle(BlockHandle handle); - void got_download_token(std::unique_ptr token); + void got_download_token(std::unique_ptr token); void got_node_to_download(adnl::AdnlNodeIdShort node); void got_data(td::BufferSlice data); void got_data_from_db(td::BufferSlice data); @@ -79,7 +79,7 @@ class DownloadBlockNew : public td::actor::Actor { bool allow_partial_proof_ = false; - std::unique_ptr token_; + std::unique_ptr token_; }; } // namespace fullnode diff --git a/validator/net/download-block.cpp b/validator/net/download-block.cpp index 9ca84be29..c60955ed2 100644 --- a/validator/net/download-block.cpp +++ b/validator/net/download-block.cpp @@ -128,7 +128,7 @@ void DownloadBlock::got_block_handle(BlockHandle handle) { return; } - auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result> R) { + auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result> R) { if (R.is_error()) { td::actor::send_closure(SelfId, &DownloadBlock::abort_query, R.move_as_error_prefix("failed to get download token: ")); @@ -140,7 +140,7 @@ void DownloadBlock::got_block_handle(BlockHandle handle) { std::move(P)); } -void DownloadBlock::got_download_token(std::unique_ptr token) { +void DownloadBlock::got_download_token(std::unique_ptr token) { token_ = std::move(token); if (download_from_.is_zero() && !short_ && client_.empty()) { diff --git a/validator/net/download-block.hpp b/validator/net/download-block.hpp index b1847d586..2e2a715bf 100644 --- a/validator/net/download-block.hpp +++ b/validator/net/download-block.hpp @@ -49,7 +49,7 @@ class DownloadBlock : public td::actor::Actor { void start_up() override; void got_block_handle(BlockHandle handle); - void got_download_token(std::unique_ptr token); + void got_download_token(std::unique_ptr token); void got_node_to_download(adnl::AdnlNodeIdShort node); void got_block_proof_description(td::BufferSlice proof_description); void got_block_proof(td::BufferSlice data); @@ -86,7 +86,7 @@ class DownloadBlock : public td::actor::Actor { bool allow_partial_proof_ = false; - std::unique_ptr token_; + std::unique_ptr token_; }; } // namespace fullnode diff --git a/validator/net/download-proof.cpp b/validator/net/download-proof.cpp index 2ff95b885..784ecac20 100644 --- a/validator/net/download-proof.cpp +++ b/validator/net/download-proof.cpp @@ -107,7 +107,7 @@ void DownloadProof::start_up() { } void DownloadProof::checked_db() { - auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result> R) { + auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result> R) { if (R.is_error()) { td::actor::send_closure(SelfId, &DownloadProof::abort_query, R.move_as_error_prefix("failed to get download token: ")); @@ -119,7 +119,7 @@ void DownloadProof::checked_db() { std::move(P)); } -void DownloadProof::got_download_token(std::unique_ptr token) { +void DownloadProof::got_download_token(std::unique_ptr token) { token_ = std::move(token); if (download_from_.is_zero() && client_.empty()) { diff --git a/validator/net/download-proof.hpp b/validator/net/download-proof.hpp index 0739dcaf1..9caf9c3a1 100644 --- a/validator/net/download-proof.hpp +++ b/validator/net/download-proof.hpp @@ -45,7 +45,7 @@ class DownloadProof : public td::actor::Actor { void start_up() override; void checked_db(); - void got_download_token(std::unique_ptr token); + void got_download_token(std::unique_ptr token); void got_node_to_download(adnl::AdnlNodeIdShort node); void got_block_proof_description(td::BufferSlice proof_description); void got_block_proof(td::BufferSlice data); @@ -72,7 +72,7 @@ class DownloadProof : public td::actor::Actor { td::BufferSlice data_; - std::unique_ptr token_; + std::unique_ptr token_; }; } // namespace fullnode diff --git a/validator/net/get-next-key-blocks.cpp b/validator/net/get-next-key-blocks.cpp index 3354b0057..2c12e495b 100644 --- a/validator/net/get-next-key-blocks.cpp +++ b/validator/net/get-next-key-blocks.cpp @@ -84,7 +84,7 @@ void GetNextKeyBlocks::finish_query() { void GetNextKeyBlocks::start_up() { alarm_timestamp() = timeout_; - auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result> R) { + auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result> R) { if (R.is_error()) { td::actor::send_closure(SelfId, &GetNextKeyBlocks::abort_query, R.move_as_error_prefix("failed to get download token: ")); @@ -96,7 +96,7 @@ void GetNextKeyBlocks::start_up() { std::move(P)); } -void GetNextKeyBlocks::got_download_token(std::unique_ptr token) { +void GetNextKeyBlocks::got_download_token(std::unique_ptr token) { token_ = std::move(token); if (download_from_.is_zero() && client_.empty()) { diff --git a/validator/net/get-next-key-blocks.hpp b/validator/net/get-next-key-blocks.hpp index 074289e21..14c040bd2 100644 --- a/validator/net/get-next-key-blocks.hpp +++ b/validator/net/get-next-key-blocks.hpp @@ -44,7 +44,7 @@ class GetNextKeyBlocks : public td::actor::Actor { void finish_query(); void start_up() override; - void got_download_token(std::unique_ptr token); + void got_download_token(std::unique_ptr token); void got_node_to_download(adnl::AdnlNodeIdShort node); void send_request(); void got_result(td::BufferSlice res); @@ -75,7 +75,7 @@ class GetNextKeyBlocks : public td::actor::Actor { std::vector pending_; std::vector res_; - std::unique_ptr token_; + std::unique_ptr token_; }; } // namespace fullnode diff --git a/validator/token-manager.cpp b/validator/token-manager.cpp index 0bc4a9c60..8242f9219 100644 --- a/validator/token-manager.cpp +++ b/validator/token-manager.cpp @@ -22,23 +22,23 @@ namespace ton { namespace validator { -void TokenManager::get_download_token(size_t download_size, td::uint32 priority, td::Timestamp timeout, - td::Promise> promise) { +void TokenManager::get_token(size_t size, td::uint32 priority, td::Timestamp timeout, + td::Promise> promise) { if (free_priority_tokens_ > 0 && priority > 0) { --free_priority_tokens_; - promise.set_value(gen_token(download_size, priority)); + promise.set_value(gen_token(size, priority)); return; } if (free_tokens_ > 0) { --free_tokens_; - promise.set_value(gen_token(download_size, priority)); + promise.set_value(gen_token(size, priority)); return; } - pending_.emplace(PendingPromiseKey{download_size, priority, seqno_++}, PendingPromise{timeout, std::move(promise)}); + pending_.emplace(PendingPromiseKey{size, priority, seqno_++}, PendingPromise{timeout, std::move(promise)}); } -void TokenManager::download_token_cleared(size_t download_size, td::uint32 priority) { +void TokenManager::token_cleared(size_t size, td::uint32 priority) { (priority ? free_priority_tokens_ : free_tokens_)++; if (free_priority_tokens_ > max_priority_tokens_) { free_priority_tokens_--; @@ -47,7 +47,7 @@ void TokenManager::download_token_cleared(size_t download_size, td::uint32 prior for (auto it = pending_.begin(); it != pending_.end();) { if (it->first.priority && (free_tokens_ || free_priority_tokens_)) { - it->second.promise.set_value(gen_token(download_size, priority)); + it->second.promise.set_value(gen_token(size, priority)); auto it2 = it++; pending_.erase(it2); if (free_priority_tokens_ > 0) { @@ -56,7 +56,7 @@ void TokenManager::download_token_cleared(size_t download_size, td::uint32 prior free_tokens_--; } } else if (!it->first.priority && free_tokens_) { - it->second.promise.set_value(gen_token(download_size, priority)); + it->second.promise.set_value(gen_token(size, priority)); auto it2 = it++; pending_.erase(it2); free_tokens_--; @@ -69,7 +69,7 @@ void TokenManager::download_token_cleared(size_t download_size, td::uint32 prior void TokenManager::alarm() { for (auto it = pending_.begin(); it != pending_.end();) { if (it->second.timeout.is_in_past()) { - it->second.promise.set_error(td::Status::Error(ErrorCode::timeout, "timeout in wait download token")); + it->second.promise.set_error(td::Status::Error(ErrorCode::timeout, "timeout in wait token")); it = pending_.erase(it); } else { it++; @@ -77,23 +77,23 @@ void TokenManager::alarm() { } } -std::unique_ptr TokenManager::gen_token(size_t download_size, td::uint32 priority) { - class Token : public DownloadToken { +std::unique_ptr TokenManager::gen_token(size_t size, td::uint32 priority) { + class TokenImpl : public ActionToken { public: - Token(size_t download_size, td::uint32 priority, td::actor::ActorId manager) - : download_size_(download_size), priority_(priority), manager_(manager) { + TokenImpl(size_t size, td::uint32 priority, td::actor::ActorId manager) + : size_(size), priority_(priority), manager_(manager) { } - ~Token() override { - td::actor::send_closure(manager_, &TokenManager::download_token_cleared, download_size_, priority_); + ~TokenImpl() override { + td::actor::send_closure(manager_, &TokenManager::token_cleared, size_, priority_); } private: - size_t download_size_; + size_t size_; td::uint32 priority_; td::actor::ActorId manager_; }; - return std::make_unique(download_size, priority, actor_id(this)); + return std::make_unique(size, priority, actor_id(this)); } } // namespace validator diff --git a/validator/token-manager.h b/validator/token-manager.h index 0d75710f9..0fd0126a3 100644 --- a/validator/token-manager.h +++ b/validator/token-manager.h @@ -31,16 +31,19 @@ class TokenManager : public td::actor::Actor { public: TokenManager() { } + explicit TokenManager(td::uint32 max_tokens) + : free_tokens_(max_tokens), free_priority_tokens_(max_tokens), max_priority_tokens_(max_tokens) { + } void alarm() override; - void get_download_token(size_t download_size, td::uint32 priority, td::Timestamp timeout, - td::Promise> promise); - void download_token_cleared(size_t download_size, td::uint32 priority); + void get_token(size_t size, td::uint32 priority, td::Timestamp timeout, + td::Promise> promise); + void token_cleared(size_t size, td::uint32 priority); private: - std::unique_ptr gen_token(size_t download_size, td::uint32 priority); + std::unique_ptr gen_token(size_t size, td::uint32 priority); struct PendingPromiseKey { - size_t download_size; + size_t size; td::uint32 priority; td::uint64 seqno; @@ -50,7 +53,7 @@ class TokenManager : public td::actor::Actor { }; struct PendingPromise { td::Timestamp timeout; - td::Promise> promise; + td::Promise> promise; }; td::uint64 seqno_ = 0; std::map pending_; diff --git a/validator/validator.h b/validator/validator.h index e6aff97cc..73065aa98 100644 --- a/validator/validator.h +++ b/validator/validator.h @@ -42,9 +42,9 @@ namespace ton { namespace validator { -class DownloadToken { +class ActionToken { public: - virtual ~DownloadToken() = default; + virtual ~ActionToken() = default; }; struct PerfTimerStats { @@ -252,7 +252,7 @@ class ValidatorManagerInterface : public td::actor::Actor { virtual void add_ext_server_port(td::uint16 port) = 0; virtual void get_download_token(size_t download_size, td::uint32 priority, td::Timestamp timeout, - td::Promise> promise) = 0; + td::Promise> promise) = 0; virtual void get_block_data_from_db(ConstBlockHandle handle, td::Promise> promise) = 0; virtual void get_block_data_from_db_short(BlockIdExt block_id, td::Promise> promise) = 0;