Skip to content

Commit

Permalink
Merge branch 'testnet' into accelerator
Browse files Browse the repository at this point in the history
  • Loading branch information
SpyCheese committed Dec 4, 2024
2 parents 4826f76 + 7bc50e6 commit 09c4488
Show file tree
Hide file tree
Showing 18 changed files with 85 additions and 59 deletions.
37 changes: 24 additions & 13 deletions validator/full-node-shard.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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<td::BufferSlice> promise) {
std::vector<BlockIdExt> 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"));
Expand All @@ -691,28 +691,39 @@ 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<tl_object_ptr<ton_api::tonNode_outMsgQueueProof>> R) mutable {
if (R.is_error()) {
promise.set_result(create_serialize_tl_object<ton_api::tonNode_outMsgQueueProofEmpty>());
} 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) {
sb << " " << id.id.to_str();
}
sb << " from " << src;
};
td::actor::create_actor<BuildOutMsgQueueProof>("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<std::unique_ptr<ActionToken>> 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<tl_object_ptr<ton_api::tonNode_outMsgQueueProof>> R) mutable {
if (R.is_error()) {
promise.set_result(create_serialize_tl_object<ton_api::tonNode_outMsgQueueProofEmpty>());
} else {
promise.set_result(serialize_tl_object(R.move_as_ok(), true));
}
});
td::actor::create_actor<BuildOutMsgQueueProof>("buildqueueproof", dst_shard, std::move(blocks), limits, manager,
std::move(P))
.release();
});
}

void FullNodeShardImpl::receive_query(adnl::AdnlNodeIdShort src, td::BufferSlice query,
Expand Down
6 changes: 6 additions & 0 deletions validator/full-node.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand Down Expand Up @@ -644,6 +645,11 @@ void FullNodeImpl::process_block_candidate_broadcast(BlockIdExt block_id, Catcha
std::move(data));
}

void FullNodeImpl::get_out_msg_queue_query_token(td::Promise<std::unique_ptr<ActionToken>> 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();
Expand Down
1 change: 1 addition & 0 deletions validator/full-node.h
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,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<std::unique_ptr<ActionToken>> promise) = 0;

virtual void set_validator_telemetry_filename(std::string value) = 0;

Expand Down
5 changes: 5 additions & 0 deletions validator/full-node.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
#include <map>
#include <set>
#include <queue>
#include <token-manager.h>

namespace ton {

Expand Down Expand Up @@ -94,6 +95,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<std::unique_ptr<ActionToken>> promise) override;

void set_validator_telemetry_filename(std::string value) override;

Expand Down Expand Up @@ -179,6 +181,9 @@ class FullNodeImpl : public FullNode {
PublicKeyHash validator_telemetry_collector_key_ = PublicKeyHash::zero();

void update_validator_telemetry_collector();

td::actor::ActorOwn<TokenManager> out_msg_queue_query_token_manager_ =
td::actor::create_actor<TokenManager>("tokens", /* max_tokens = */ 1);
};

} // namespace fullnode
Expand Down
2 changes: 1 addition & 1 deletion validator/manager-disk.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -292,7 +292,7 @@ class ValidatorManagerImpl : public ValidatorManager {
void try_get_static_file(FileHash file_hash, td::Promise<td::BufferSlice> promise) override;

void get_download_token(size_t download_size, td::uint32 priority, td::Timestamp timeout,
td::Promise<std::unique_ptr<DownloadToken>> promise) override {
td::Promise<std::unique_ptr<ActionToken>> promise) override {
promise.set_error(td::Status::Error(ErrorCode::error, "download disabled"));
}

Expand Down
2 changes: 1 addition & 1 deletion validator/manager-hardfork.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -366,7 +366,7 @@ class ValidatorManagerImpl : public ValidatorManager {
void try_get_static_file(FileHash file_hash, td::Promise<td::BufferSlice> promise) override;

void get_download_token(size_t download_size, td::uint32 priority, td::Timestamp timeout,
td::Promise<std::unique_ptr<DownloadToken>> promise) override {
td::Promise<std::unique_ptr<ActionToken>> promise) override {
promise.set_error(td::Status::Error(ErrorCode::error, "download disabled"));
}

Expand Down
4 changes: 2 additions & 2 deletions validator/manager.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -543,8 +543,8 @@ class ValidatorManagerImpl : public ValidatorManager {
void try_get_static_file(FileHash file_hash, td::Promise<td::BufferSlice> promise) override;

void get_download_token(size_t download_size, td::uint32 priority, td::Timestamp timeout,
td::Promise<std::unique_ptr<DownloadToken>> promise) override {
td::actor::send_closure(token_manager_, &TokenManager::get_download_token, download_size, priority, timeout,
td::Promise<std::unique_ptr<ActionToken>> promise) override {
td::actor::send_closure(token_manager_, &TokenManager::get_token, download_size, priority, timeout,
std::move(promise));
}

Expand Down
4 changes: 2 additions & 2 deletions validator/net/download-block-new.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -144,7 +144,7 @@ void DownloadBlockNew::got_block_handle(BlockHandle handle) {
return;
}

auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<std::unique_ptr<DownloadToken>> R) {
auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<std::unique_ptr<ActionToken>> R) {
if (R.is_error()) {
td::actor::send_closure(SelfId, &DownloadBlockNew::abort_query,
R.move_as_error_prefix("failed to get download token: "));
Expand All @@ -156,7 +156,7 @@ void DownloadBlockNew::got_block_handle(BlockHandle handle) {
std::move(P));
}

void DownloadBlockNew::got_download_token(std::unique_ptr<DownloadToken> token) {
void DownloadBlockNew::got_download_token(std::unique_ptr<ActionToken> token) {
token_ = std::move(token);

if (download_from_.is_zero() && client_.empty()) {
Expand Down
4 changes: 2 additions & 2 deletions validator/net/download-block-new.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -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<DownloadToken> token);
void got_download_token(std::unique_ptr<ActionToken> token);
void got_node_to_download(adnl::AdnlNodeIdShort node);
void got_data(td::BufferSlice data);
void got_data_from_db(td::BufferSlice data);
Expand Down Expand Up @@ -79,7 +79,7 @@ class DownloadBlockNew : public td::actor::Actor {

bool allow_partial_proof_ = false;

std::unique_ptr<DownloadToken> token_;
std::unique_ptr<ActionToken> token_;
};

} // namespace fullnode
Expand Down
4 changes: 2 additions & 2 deletions validator/net/download-block.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,7 @@ void DownloadBlock::got_block_handle(BlockHandle handle) {
return;
}

auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<std::unique_ptr<DownloadToken>> R) {
auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<std::unique_ptr<ActionToken>> R) {
if (R.is_error()) {
td::actor::send_closure(SelfId, &DownloadBlock::abort_query,
R.move_as_error_prefix("failed to get download token: "));
Expand All @@ -140,7 +140,7 @@ void DownloadBlock::got_block_handle(BlockHandle handle) {
std::move(P));
}

void DownloadBlock::got_download_token(std::unique_ptr<DownloadToken> token) {
void DownloadBlock::got_download_token(std::unique_ptr<ActionToken> token) {
token_ = std::move(token);

if (download_from_.is_zero() && !short_ && client_.empty()) {
Expand Down
4 changes: 2 additions & 2 deletions validator/net/download-block.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -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<DownloadToken> token);
void got_download_token(std::unique_ptr<ActionToken> 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);
Expand Down Expand Up @@ -86,7 +86,7 @@ class DownloadBlock : public td::actor::Actor {

bool allow_partial_proof_ = false;

std::unique_ptr<DownloadToken> token_;
std::unique_ptr<ActionToken> token_;
};

} // namespace fullnode
Expand Down
4 changes: 2 additions & 2 deletions validator/net/download-proof.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -107,7 +107,7 @@ void DownloadProof::start_up() {
}

void DownloadProof::checked_db() {
auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<std::unique_ptr<DownloadToken>> R) {
auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<std::unique_ptr<ActionToken>> R) {
if (R.is_error()) {
td::actor::send_closure(SelfId, &DownloadProof::abort_query,
R.move_as_error_prefix("failed to get download token: "));
Expand All @@ -119,7 +119,7 @@ void DownloadProof::checked_db() {
std::move(P));
}

void DownloadProof::got_download_token(std::unique_ptr<DownloadToken> token) {
void DownloadProof::got_download_token(std::unique_ptr<ActionToken> token) {
token_ = std::move(token);

if (download_from_.is_zero() && client_.empty()) {
Expand Down
4 changes: 2 additions & 2 deletions validator/net/download-proof.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ class DownloadProof : public td::actor::Actor {

void start_up() override;
void checked_db();
void got_download_token(std::unique_ptr<DownloadToken> token);
void got_download_token(std::unique_ptr<ActionToken> 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);
Expand All @@ -72,7 +72,7 @@ class DownloadProof : public td::actor::Actor {

td::BufferSlice data_;

std::unique_ptr<DownloadToken> token_;
std::unique_ptr<ActionToken> token_;
};

} // namespace fullnode
Expand Down
4 changes: 2 additions & 2 deletions validator/net/get-next-key-blocks.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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<std::unique_ptr<DownloadToken>> R) {
auto P = td::PromiseCreator::lambda([SelfId = actor_id(this)](td::Result<std::unique_ptr<ActionToken>> R) {
if (R.is_error()) {
td::actor::send_closure(SelfId, &GetNextKeyBlocks::abort_query,
R.move_as_error_prefix("failed to get download token: "));
Expand All @@ -96,7 +96,7 @@ void GetNextKeyBlocks::start_up() {
std::move(P));
}

void GetNextKeyBlocks::got_download_token(std::unique_ptr<DownloadToken> token) {
void GetNextKeyBlocks::got_download_token(std::unique_ptr<ActionToken> token) {
token_ = std::move(token);

if (download_from_.is_zero() && client_.empty()) {
Expand Down
4 changes: 2 additions & 2 deletions validator/net/get-next-key-blocks.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ class GetNextKeyBlocks : public td::actor::Actor {
void finish_query();

void start_up() override;
void got_download_token(std::unique_ptr<DownloadToken> token);
void got_download_token(std::unique_ptr<ActionToken> token);
void got_node_to_download(adnl::AdnlNodeIdShort node);
void send_request();
void got_result(td::BufferSlice res);
Expand Down Expand Up @@ -75,7 +75,7 @@ class GetNextKeyBlocks : public td::actor::Actor {
std::vector<BlockIdExt> pending_;
std::vector<BlockIdExt> res_;

std::unique_ptr<DownloadToken> token_;
std::unique_ptr<ActionToken> token_;
};

} // namespace fullnode
Expand Down
34 changes: 17 additions & 17 deletions validator/token-manager.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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<std::unique_ptr<DownloadToken>> promise) {
void TokenManager::get_token(size_t size, td::uint32 priority, td::Timestamp timeout,
td::Promise<std::unique_ptr<ActionToken>> 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_--;
Expand All @@ -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) {
Expand All @@ -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_--;
Expand All @@ -69,31 +69,31 @@ 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++;
}
}
}

std::unique_ptr<DownloadToken> TokenManager::gen_token(size_t download_size, td::uint32 priority) {
class Token : public DownloadToken {
std::unique_ptr<ActionToken> 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<TokenManager> manager)
: download_size_(download_size), priority_(priority), manager_(manager) {
TokenImpl(size_t size, td::uint32 priority, td::actor::ActorId<TokenManager> 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<TokenManager> manager_;
};

return std::make_unique<Token>(download_size, priority, actor_id(this));
return std::make_unique<TokenImpl>(size, priority, actor_id(this));
}

} // namespace validator
Expand Down
Loading

0 comments on commit 09c4488

Please sign in to comment.