external/ton
1
0
Fork 0
mirror of https://github.com/ton-blockchain/ton synced 2025-03-09 15:40:10 +00:00
Code Issues Releases Wiki Activity

Pause and resume upload in storage-daemon (#567)

Browse source
This commit is contained in:
SpyCheese 2022-12-23 11:05:29 +03:00 • committed by GitHub
parent 898d1ff8b2
commit 48d48e595e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 155 additions and 68 deletions
Download patch file Download diff file Expand all files Collapse all files

43
storage/NodeActor.cpp
View file

@ -32,24 +32,27 @@
namespace ton {
NodeActor::NodeActor(PeerId self_id, Torrent torrent, td::unique_ptr<Callback> callback,
td::unique_ptr<NodeCallback> node_callback, std::shared_ptr<db::DbType> db, bool should_download)
: self_id_(self_id)
, torrent_(std::move(torrent))
, callback_(std::move(callback))
, node_callback_(std::move(node_callback))
, db_(std::move(db))
, should_download_(should_download) {
}
NodeActor::NodeActor(PeerId self_id, ton::Torrent torrent, td::unique_ptr<Callback> callback,
td::unique_ptr<NodeCallback> node_callback, std::shared_ptr<db::DbType> db, bool should_download,
DbInitialData db_initial_data)
bool should_upload)
: self_id_(self_id)
, torrent_(std::move(torrent))
, callback_(std::move(callback))
, node_callback_(std::move(node_callback))
, db_(std::move(db))
, should_download_(should_download)
, should_upload_(should_upload) {
}
NodeActor::NodeActor(PeerId self_id, ton::Torrent torrent, td::unique_ptr<Callback> callback,
td::unique_ptr<NodeCallback> node_callback, std::shared_ptr<db::DbType> db, bool should_download,
bool should_upload, DbInitialData db_initial_data)
: self_id_(self_id)
, torrent_(std::move(torrent))
, callback_(std::move(callback))
, node_callback_(std::move(node_callback))
, db_(std::move(db))
, should_download_(should_download)
, should_upload_(should_upload)
, pending_set_file_priority_(std::move(db_initial_data.priorities))
, pieces_in_db_(std::move(db_initial_data.pieces_in_db)) {
}
@ -199,7 +202,7 @@ void NodeActor::loop_will_upload() {
}
for (auto &it : peers_) {
auto will_upload = peers_set.count(it.first) > 0;
auto will_upload = peers_set.count(it.first) > 0 && should_upload_;
auto &state = it.second.state;
auto node_state = state->node_state_.load();
if (node_state.will_upload != will_upload) {
@ -399,6 +402,15 @@ void NodeActor::set_should_download(bool should_download) {
yield();
}
void NodeActor::set_should_upload(bool should_upload) {
if (should_upload == should_upload_) {
return;
}
should_upload_ = should_upload;
db_store_torrent();
yield();
}
void NodeActor::load_from(td::optional<TorrentMeta> meta, std::string files_path, td::Promise<td::Unit> promise) {
auto S = [&]() -> td::Status {
if (meta) {
@ -573,7 +585,7 @@ void NodeActor::loop_peer(const PeerId &peer_id, Peer &peer) {
for (td::uint32 part_id : state->peer_queries_.read()) {
should_notify_peer = true;
auto res = [&]() -> td::Result<PeerState::Part> {
if (!node_state.will_upload) {
if (!node_state.will_upload || !should_upload_) {
return td::Status::Error("Won't upload");
}
TRY_RESULT(proof, torrent_.get_piece_proof(part_id));
@ -685,6 +697,7 @@ void NodeActor::db_store_torrent() {
}
auto obj = create_tl_object<ton_api::storage_db_torrent>();
obj->active_download_ = should_download_;
obj->active_upload_ = should_upload_;
obj->root_dir_ = torrent_.get_root_dir();
db_->set(create_hash_tl_object<ton_api::storage_db_key_torrent>(torrent_.get_hash()), serialize_tl_object(obj, true),
[](td::Result<td::Unit> R) {
@ -837,6 +850,7 @@ void NodeActor::load_from_db(std::shared_ptr<db::DbType> db, td::Bits256 hash, t
void got_torrent(tl_object_ptr<ton_api::storage_db_torrent> obj) {
root_dir_ = std::move(obj->root_dir_);
active_download_ = obj->active_download_;
active_upload_ = obj->active_upload_;
db_->get(create_hash_tl_object<ton_api::storage_db_key_torrentMeta>(hash_),
[SelfId = actor_id(this)](td::Result<db::DbType::GetResult> R) {
if (R.is_error()) {
@ -962,7 +976,7 @@ void NodeActor::load_from_db(std::shared_ptr<db::DbType> db, td::Bits256 hash, t
data.pieces_in_db = std::move(pieces_in_db_);
finish(td::actor::create_actor<NodeActor>("Node", 1, torrent_.unwrap(), std::move(callback_),
std::move(node_callback_), std::move(db_), active_download_,
std::move(data)));
active_upload_, std::move(data)));
}
private:
@ -974,6 +988,7 @@ void NodeActor::load_from_db(std::shared_ptr<db::DbType> db, td::Bits256 hash, t
std::string root_dir_;
bool active_download_{false};
bool active_upload_{false};
td::optional<Torrent> torrent_;
std::vector<PendingSetFilePriority> priorities_;
std::set<td::uint64> pieces_in_db_;

12
storage/NodeActor.h
View file

@ -63,26 +63,29 @@ class NodeActor : public td::actor::Actor {
};
NodeActor(PeerId self_id, ton::Torrent torrent, td::unique_ptr<Callback> callback,
td::unique_ptr<NodeCallback> node_callback, std::shared_ptr<db::DbType> db, bool should_download = true);
td::unique_ptr<NodeCallback> node_callback, std::shared_ptr<db::DbType> db, bool should_download = true,
bool should_upload = true);
NodeActor(PeerId self_id, ton::Torrent torrent, td::unique_ptr<Callback> callback,
td::unique_ptr<NodeCallback> node_callback, std::shared_ptr<db::DbType> db, bool should_download,
DbInitialData db_initial_data);
bool should_upload, DbInitialData db_initial_data);
void start_peer(PeerId peer_id, td::Promise<td::actor::ActorId<PeerActor>> promise);
struct NodeState {
Torrent &torrent;
bool active_download;
bool active_upload;
double download_speed;
double upload_speed;
const std::vector<td::uint8> &file_priority;
};
void with_torrent(td::Promise<NodeState> promise) {
promise.set_value(
NodeState{torrent_, should_download_, download_speed_.speed(), upload_speed_.speed(), file_priority_});
promise.set_value(NodeState{torrent_, should_download_, should_upload_, download_speed_.speed(),
upload_speed_.speed(), file_priority_});
}
std::string get_stats_str();
void set_should_download(bool should_download);
void set_should_upload(bool should_upload);
void set_all_files_priority(td::uint8 priority, td::Promise<bool> promise);
void set_file_priority_by_idx(size_t i, td::uint8 priority, td::Promise<bool> promise);
@ -107,6 +110,7 @@ class NodeActor : public td::actor::Actor {
td::unique_ptr<NodeCallback> node_callback_;
std::shared_ptr<db::DbType> db_;
bool should_download_{false};
bool should_upload_{false};
class Notifier : public td::actor::Actor {
public:

23
storage/storage-daemon/StorageManager.cpp
View file

@ -126,13 +126,14 @@ td::unique_ptr<NodeActor::Callback> StorageManager::create_callback(
return td::make_unique<Callback>(actor_id(this), hash, std::move(closing_state));
}
void StorageManager::add_torrent(Torrent torrent, bool start_download, td::Promise<td::Unit> promise) {
TRY_STATUS_PROMISE(promise, add_torrent_impl(std::move(torrent), start_download));
void StorageManager::add_torrent(Torrent torrent, bool start_download, bool allow_upload,
td::Promise<td::Unit> promise) {
TRY_STATUS_PROMISE(promise, add_torrent_impl(std::move(torrent), start_download, allow_upload));
db_store_torrent_list();
promise.set_result(td::Unit());
}
td::Status StorageManager::add_torrent_impl(Torrent torrent, bool start_download) {
td::Status StorageManager::add_torrent_impl(Torrent torrent, bool start_download, bool allow_upload) {
td::Bits256 hash = torrent.get_hash();
if (torrents_.count(hash)) {
return td::Status::Error("Cannot add torrent: duplicate hash");
@ -145,25 +146,25 @@ td::Status StorageManager::add_torrent_impl(Torrent torrent, bool start_download
LOG(INFO) << "Added torrent " << hash.to_hex() << " , root_dir = " << torrent.get_root_dir();
entry.actor =
td::actor::create_actor<NodeActor>("Node", 1, std::move(torrent), create_callback(hash, entry.closing_state),
std::move(context), db_, start_download);
std::move(context), db_, start_download, allow_upload);
return td::Status::OK();
}
void StorageManager::add_torrent_by_meta(TorrentMeta meta, std::string root_dir, bool start_download,
void StorageManager::add_torrent_by_meta(TorrentMeta meta, std::string root_dir, bool start_download, bool allow_upload,
td::Promise<td::Unit> promise) {
td::Bits256 hash(meta.info.get_hash());
Torrent::Options options;
options.root_dir = root_dir.empty() ? db_root_ + "/torrent-files/" + hash.to_hex() : root_dir;
TRY_RESULT_PROMISE(promise, torrent, Torrent::open(std::move(options), std::move(meta)));
add_torrent(std::move(torrent), start_download, std::move(promise));
add_torrent(std::move(torrent), start_download, allow_upload, std::move(promise));
}
void StorageManager::add_torrent_by_hash(td::Bits256 hash, std::string root_dir, bool start_download,
void StorageManager::add_torrent_by_hash(td::Bits256 hash, std::string root_dir, bool start_download, bool allow_upload,
td::Promise<td::Unit> promise) {
Torrent::Options options;
options.root_dir = root_dir.empty() ? db_root_ + "/torrent-files/" + hash.to_hex() : root_dir;
TRY_RESULT_PROMISE(promise, torrent, Torrent::open(std::move(options), hash));
add_torrent(std::move(torrent), start_download, std::move(promise));
add_torrent(std::move(torrent), start_download, allow_upload, std::move(promise));
}
void StorageManager::set_active_download(td::Bits256 hash, bool active, td::Promise<td::Unit> promise) {
@ -172,6 +173,12 @@ void StorageManager::set_active_download(td::Bits256 hash, bool active, td::Prom
promise.set_result(td::Unit());
}
void StorageManager::set_active_upload(td::Bits256 hash, bool active, td::Promise<td::Unit> promise) {
TRY_RESULT_PROMISE(promise, entry, get_torrent(hash));
td::actor::send_closure(entry->actor, &NodeActor::set_should_upload, active);
promise.set_result(td::Unit());
}
void StorageManager::with_torrent(td::Bits256 hash, td::Promise<NodeActor::NodeState> promise) {
TRY_RESULT_PROMISE(promise, entry, get_torrent(hash));
td::actor::send_closure(entry->actor, &NodeActor::with_torrent, std::move(promise));

9
storage/storage-daemon/StorageManager.h
View file

@ -39,11 +39,12 @@ class StorageManager : public td::actor::Actor {
void start_up() override;
void add_torrent(Torrent torrent, bool start_download, td::Promise<td::Unit> promise);
void add_torrent_by_meta(TorrentMeta meta, std::string root_dir, bool start_download, td::Promise<td::Unit> promise);
void add_torrent_by_hash(td::Bits256 hash, std::string root_dir, bool start_download, td::Promise<td::Unit> promise);
void add_torrent(Torrent torrent, bool start_download, bool allow_upload, td::Promise<td::Unit> promise);
void add_torrent_by_meta(TorrentMeta meta, std::string root_dir, bool start_download, bool allow_upload, td::Promise<td::Unit> promise);
void add_torrent_by_hash(td::Bits256 hash, std::string root_dir, bool start_download, bool allow_upload, td::Promise<td::Unit> promise);
void set_active_download(td::Bits256 hash, bool active, td::Promise<td::Unit> promise);
void set_active_upload(td::Bits256 hash, bool active, td::Promise<td::Unit> promise);
void with_torrent(td::Bits256 hash, td::Promise<NodeActor::NodeState> promise);
void get_all_torrents(td::Promise<std::vector<td::Bits256>> promise);
@ -85,7 +86,7 @@ class StorageManager : public td::actor::Actor {
std::map<td::Bits256, TorrentEntry> torrents_;
td::Status add_torrent_impl(Torrent torrent, bool start_download);
td::Status add_torrent_impl(Torrent torrent, bool start_download, bool allow_upload);
td::Result<TorrentEntry*> get_torrent(td::Bits256 hash) {
auto it = torrents_.find(hash);

30
storage/storage-daemon/StorageProvider.cpp
View file

@ -155,7 +155,7 @@ void StorageProvider::start_up() {
init_new_storage_contract(address, contract);
break;
case StorageContract::st_downloaded:
check_contract_active(address);
after_contract_downloaded(address);
break;
case StorageContract::st_active:
contract.check_next_proof_at = td::Timestamp::now();
@ -382,7 +382,7 @@ void StorageProvider::db_update_microchunk_tree(const ContractAddress& address)
void StorageProvider::init_new_storage_contract(ContractAddress address, StorageContract& contract) {
CHECK(contract.state == StorageContract::st_downloading);
td::actor::send_closure(storage_manager_, &StorageManager::add_torrent_by_hash, contract.torrent_hash, "", false,
[](td::Result<td::Unit> R) {
false, [](td::Result<td::Unit> R) {
// Ignore errors: error can mean that the torrent already exists, other errors will be caught later
if (R.is_error()) {
LOG(DEBUG) << "Add torrent: " << R.move_as_error();
@ -449,11 +449,25 @@ void StorageProvider::downloaded_torrent(ContractAddress address, MicrochunkTree
contract.microchunk_tree = std::make_shared<MicrochunkTree>(std::move(microchunk_tree));
db_update_microchunk_tree(address);
db_update_storage_contract(address, false);
check_contract_active(address);
after_contract_downloaded(address);
}
void StorageProvider::check_contract_active(ContractAddress address, td::Timestamp retry_until,
void StorageProvider::after_contract_downloaded(ContractAddress address, td::Timestamp retry_until,
td::Timestamp retry_false_until) {
auto it = contracts_.find(address);
if (it == contracts_.end()) {
LOG(WARNING) << "Contract " << address.to_string() << " does not exist anymore";
return;
}
auto& contract = it->second;
td::actor::send_closure(storage_manager_, &StorageManager::set_active_upload, contract.torrent_hash, true,
[SelfId = actor_id(this), address](td::Result<td::Unit> R) {
if (R.is_error()) {
LOG(ERROR) << "Set active upload: " << R.move_as_error();
return;
}
LOG(DEBUG) << "Set active upload: OK";
});
get_storage_contract_data(address, tonlib_client_,
[=, SelfId = actor_id(this)](td::Result<StorageContractData> R) mutable {
if (R.is_error()) {
@ -461,7 +475,7 @@ void StorageProvider::check_contract_active(ContractAddress address, td::Timesta
if (retry_until && retry_until.is_in_past()) {
delay_action(
[=]() {
td::actor::send_closure(SelfId, &StorageProvider::check_contract_active,
td::actor::send_closure(SelfId, &StorageProvider::after_contract_downloaded,
address, retry_until, retry_false_until);
},
td::Timestamp::in(5.0));
@ -473,8 +487,8 @@ void StorageProvider::check_contract_active(ContractAddress address, td::Timesta
} else if (retry_false_until && retry_false_until.is_in_past()) {
delay_action(
[=]() {
td::actor::send_closure(SelfId, &StorageProvider::check_contract_active, address,
retry_until, retry_false_until);
td::actor::send_closure(SelfId, &StorageProvider::after_contract_downloaded,
address, retry_until, retry_false_until);
},
td::Timestamp::in(5.0));
} else {
@ -497,7 +511,7 @@ void StorageProvider::activate_contract_cont(ContractAddress address) {
td::Timestamp::in(10.0));
return;
}
td::actor::send_closure(SelfId, &StorageProvider::check_contract_active, address, td::Timestamp::in(60.0),
td::actor::send_closure(SelfId, &StorageProvider::after_contract_downloaded, address, td::Timestamp::in(60.0),
td::Timestamp::in(40.0));
});
}

2
storage/storage-daemon/StorageProvider.h
View file

@ -109,7 +109,7 @@ class StorageProvider : public td::actor::Actor {
void on_new_storage_contract_cont(ContractAddress address, StorageContractData data, td::Promise<td::Unit> promise);
void init_new_storage_contract(ContractAddress address, StorageContract& contract);
void downloaded_torrent(ContractAddress address, MicrochunkTree microchunk_tree);
void check_contract_active(ContractAddress address, td::Timestamp retry_until = td::Timestamp::in(30.0),
void after_contract_downloaded(ContractAddress address, td::Timestamp retry_until = td::Timestamp::in(30.0),
td::Timestamp retry_false_until = td::Timestamp::never());
void activate_contract_cont(ContractAddress address);
void activated_storage_contract(ContractAddress address);

75
storage/storage-daemon/storage-daemon-cli.cpp
View file

@ -298,6 +298,7 @@ class StorageDaemonCli : public td::actor::Actor {
} else if (tokens[0] == "create") {
std::string path;
bool found_path = false;
bool upload = true;
std::string description;
bool json = false;
for (size_t i = 1; i < tokens.size(); ++i) {
@ -310,6 +311,10 @@ class StorageDaemonCli : public td::actor::Actor {
description = tokens[i];
continue;
}
if (tokens[i] == "--no-upload") {
upload = false;
continue;
}
if (tokens[i] == "--json") {
json = true;
continue;
@ -325,11 +330,12 @@ class StorageDaemonCli : public td::actor::Actor {
if (!found_path) {
return td::Status::Error("Unexpected EOLN");
}
return execute_create(std::move(path), std::move(description), json);
return execute_create(std::move(path), std::move(description), upload, json);
} else if (tokens[0] == "add-by-hash" || tokens[0] == "add-by-meta") {
td::optional<std::string> param;
std::string root_dir;
bool paused = false;
bool upload = true;
bool json = false;
td::optional<std::vector<std::string>> partial;
for (size_t i = 1; i < tokens.size(); ++i) {
@ -346,6 +352,10 @@ class StorageDaemonCli : public td::actor::Actor {
paused = true;
continue;
}
if (tokens[i] == "--no-upload") {
upload = false;
continue;
}
if (tokens[i] == "--json") {
json = true;
continue;
@ -366,9 +376,9 @@ class StorageDaemonCli : public td::actor::Actor {
}
if (tokens[0] == "add-by-hash") {
TRY_RESULT(hash, parse_hash(param.value()));
return execute_add_by_hash(hash, std::move(root_dir), paused, std::move(partial), json);
return execute_add_by_hash(hash, std::move(root_dir), paused, upload, std::move(partial), json);
} else {
return execute_add_by_meta(param.value(), std::move(root_dir), paused, std::move(partial), json);
return execute_add_by_meta(param.value(), std::move(root_dir), paused, upload, std::move(partial), json);
}
} else if (tokens[0] == "list") {
bool with_hashes = false;
@ -441,6 +451,12 @@ class StorageDaemonCli : public td::actor::Actor {
}
TRY_RESULT(hash, parse_torrent(tokens[1]));
return execute_set_active_download(hash, tokens[0] == "download-resume");
} else if (tokens[0] == "upload-pause" || tokens[0] == "upload-resume") {
if (tokens.size() != 2) {
return td::Status::Error("Expected bag");
}
TRY_RESULT(hash, parse_torrent(tokens[1]));
return execute_set_active_upload(hash, tokens[0] == "upload-resume");
} else if (tokens[0] == "priority-all") {
if (tokens.size() != 3) {
return td::Status::Error("Expected bag and priority");
@ -731,19 +747,21 @@ class StorageDaemonCli : public td::actor::Actor {
td::Status execute_help() {
td::TerminalIO::out() << "help\tPrint this help\n";
td::TerminalIO::out() << "create [-d description] [--json] <file/dir>\tCreate bag of files from <file/dir>\n";
td::TerminalIO::out() << "\t-d - Description will be stored in torrent info.\n";
td::TerminalIO::out() << "\t--json\tOutput in json\n";
td::TerminalIO::out()
<< "add-by-hash <bag-id> [-d root_dir] [--paused] [--json] [--partial file1 file2 ...]\tAdd bag "
"with given BagID (in hex)\n";
<< "create [-d description] [--no-upload] [--json] <file/dir>\tCreate bag of files from <file/dir>\n";
td::TerminalIO::out() << "\t-d\tDescription will be stored in torrent info\n";
td::TerminalIO::out() << "\t--no-upload\tDon't share bag with peers\n";
td::TerminalIO::out() << "\t--json\tOutput in json\n";
td::TerminalIO::out() << "add-by-hash <bag-id> [-d root_dir] [--paused] [--no-upload] [--json] [--partial file1 "
"file2 ...]\tAdd bag with given BagID (in hex)\n";
td::TerminalIO::out() << "\t-d\tTarget directory, default is an internal directory of storage-daemon\n";
td::TerminalIO::out() << "\t--paused\tDon't start download immediately\n";
td::TerminalIO::out() << "\t--no-upload\tDon't share bag with peers\n";
td::TerminalIO::out()
<< "\t--partial\tEverything after this flag is a list of filenames. Only these files will be downloaded.\n";
td::TerminalIO::out() << "\t--json\tOutput in json\n";
td::TerminalIO::out() << "add-by-meta <meta> [-d root_dir] [--paused] [--json] [--partial file1 file2 ...]\tLoad "
"meta from file and add bag\n";
td::TerminalIO::out() << "add-by-meta <meta> [-d root_dir] [--paused] [--no-upload] [--json] [--partial file1 "
"file2 ...]\tLoad meta from file and add bag\n";
td::TerminalIO::out() << "\tFlags are the same as in add-by-hash\n";
td::TerminalIO::out() << "list [--hashes] [--json]\tPrint list of bags\n";
td::TerminalIO::out() << "\t--hashes\tPrint full BagID\n";
@ -828,9 +846,9 @@ class StorageDaemonCli : public td::actor::Actor {
return td::Status::OK();
}
td::Status execute_create(std::string path, std::string description, bool json) {
td::Status execute_create(std::string path, std::string description, bool upload, bool json) {
TRY_RESULT_PREFIX_ASSIGN(path, td::realpath(path), "Invalid path: ");
auto query = create_tl_object<ton_api::storage_daemon_createTorrent>(path, description);
auto query = create_tl_object<ton_api::storage_daemon_createTorrent>(path, description, upload);
send_query(std::move(query),
[=, SelfId = actor_id(this)](td::Result<tl_object_ptr<ton_api::storage_daemon_torrentFull>> R) {
if (R.is_error()) {
@ -847,7 +865,7 @@ class StorageDaemonCli : public td::actor::Actor {
return td::Status::OK();
}
td::Status execute_add_by_hash(td::Bits256 hash, std::string root_dir, bool paused,
td::Status execute_add_by_hash(td::Bits256 hash, std::string root_dir, bool paused, bool upload,
td::optional<std::vector<std::string>> partial, bool json) {
if (!root_dir.empty()) {
TRY_STATUS_PREFIX(td::mkpath(root_dir), "Failed to create directory: ");
@ -861,8 +879,8 @@ class StorageDaemonCli : public td::actor::Actor {
priorities.push_back(create_tl_object<ton_api::storage_priorityAction_name>(std::move(f), 1));
}
}
auto query =
create_tl_object<ton_api::storage_daemon_addByHash>(hash, std::move(root_dir), !paused, std::move(priorities));
auto query = create_tl_object<ton_api::storage_daemon_addByHash>(hash, std::move(root_dir), !paused, upload,
std::move(priorities));
send_query(std::move(query),
[=, SelfId = actor_id(this)](td::Result<tl_object_ptr<ton_api::storage_daemon_torrentFull>> R) {
if (R.is_error()) {
@ -879,7 +897,7 @@ class StorageDaemonCli : public td::actor::Actor {
return td::Status::OK();
}
td::Status execute_add_by_meta(std::string meta_file, std::string root_dir, bool paused,
td::Status execute_add_by_meta(std::string meta_file, std::string root_dir, bool paused, bool upload,
td::optional<std::vector<std::string>> partial, bool json) {
TRY_RESULT_PREFIX(meta, td::read_file(meta_file), "Failed to read meta: ");
if (!root_dir.empty()) {
@ -895,7 +913,7 @@ class StorageDaemonCli : public td::actor::Actor {
}
}
auto query = create_tl_object<ton_api::storage_daemon_addByMeta>(std::move(meta), std::move(root_dir), !paused,
std::move(priorities));
upload, std::move(priorities));
send_query(std::move(query),
[=, SelfId = actor_id(this)](td::Result<tl_object_ptr<ton_api::storage_daemon_torrentFull>> R) {
if (R.is_error()) {
@ -1020,6 +1038,19 @@ class StorageDaemonCli : public td::actor::Actor {
return td::Status::OK();
}
td::Status execute_set_active_upload(td::Bits256 hash, bool active) {
auto query = create_tl_object<ton_api::storage_daemon_setActiveUpload>(hash, active);
send_query(std::move(query),
[SelfId = actor_id(this)](td::Result<tl_object_ptr<ton_api::storage_daemon_success>> R) {
if (R.is_error()) {
return;
}
td::TerminalIO::out() << "Success\n";
td::actor::send_closure(SelfId, &StorageDaemonCli::command_finished, td::Status::OK());
});
return td::Status::OK();
}
td::Status execute_set_priority_all(td::Bits256 hash, td::uint8 priority) {
auto query = create_tl_object<ton_api::storage_daemon_setFilePriorityAll>(hash, priority);
send_query(std::move(query),
@ -1601,7 +1632,12 @@ class StorageDaemonCli : public td::actor::Actor {
} else {
td::TerminalIO::out() << "Download paused\n";
}
td::TerminalIO::out() << "Upload speed: " << td::format::as_size((td::uint64)obj.torrent_->upload_speed_) << "/s\n";
if (obj.torrent_->active_upload_) {
td::TerminalIO::out() << "Upload speed: " << td::format::as_size((td::uint64)obj.torrent_->upload_speed_)
<< "/s\n";
} else {
td::TerminalIO::out() << "Upload paused\n";
}
td::TerminalIO::out() << "Root dir: " << obj.torrent_->root_dir_ << "\n";
if (obj.torrent_->flags_ & 2) { // header ready
td::TerminalIO::out() << obj.files_.size() << " files:\n";
@ -1639,7 +1675,7 @@ class StorageDaemonCli : public td::actor::Actor {
});
td::TerminalIO::out() << obj.torrents_.size() << " bags\n";
std::vector<std::vector<std::string>> table;
table.push_back({"#####", "BagID", "Description", "Downloaded", "Total", "Speed"});
table.push_back({"#####", "BagID", "Description", "Downloaded", "Total", "Download", "Upload"});
for (const auto& torrent : obj.torrents_) {
std::vector<std::string> row;
row.push_back(std::to_string(hash_to_id_[torrent->hash_]));
@ -1683,6 +1719,7 @@ class StorageDaemonCli : public td::actor::Actor {
row.push_back(downloaded_size.append("/").append(included_size));
row.push_back(total_size);
row.push_back(status);
row.push_back(torrent->active_upload_ ? size_to_str((td::uint64)torrent->upload_speed_) + "/s" : "Paused");
table.push_back(std::move(row));
}
print_table(table, {2});

16
storage/storage-daemon/storage-daemon.cpp
View file

@ -291,7 +291,7 @@ class StorageDaemon : public td::actor::Actor {
options.description = std::move(query.description_);
TRY_RESULT_PROMISE(promise, torrent, Torrent::Creator::create_from_path(std::move(options), query.path_));
td::Bits256 hash = torrent.get_hash();
td::actor::send_closure(manager, &StorageManager::add_torrent, std::move(torrent), false,
td::actor::send_closure(manager, &StorageManager::add_torrent, std::move(torrent), false, query.allow_upload_,
[manager, hash, promise = std::move(promise)](td::Result<td::Unit> R) mutable {
if (R.is_error()) {
promise.set_error(R.move_as_error());
@ -308,6 +308,7 @@ class StorageDaemon : public td::actor::Actor {
bool start_download_now = query.start_download_ && query.priorities_.empty();
td::actor::send_closure(
manager_, &StorageManager::add_torrent_by_hash, hash, std::move(query.root_dir_), start_download_now,
query.allow_upload_,
query_add_torrent_cont(hash, query.start_download_, std::move(query.priorities_), std::move(promise)));
}
@ -317,6 +318,7 @@ class StorageDaemon : public td::actor::Actor {
bool start_download_now = query.start_download_ && query.priorities_.empty();
td::actor::send_closure(
manager_, &StorageManager::add_torrent_by_meta, std::move(meta), std::move(query.root_dir_), start_download_now,
query.allow_upload_,
query_add_torrent_cont(hash, query.start_download_, std::move(query.priorities_), std::move(promise)));
}
@ -362,6 +364,12 @@ class StorageDaemon : public td::actor::Actor {
promise.wrap([](td::Unit &&) { return create_serialize_tl_object<ton_api::storage_daemon_success>(); }));
}
void run_control_query(ton_api::storage_daemon_setActiveUpload &query, td::Promise<td::BufferSlice> promise) {
td::actor::send_closure(
manager_, &StorageManager::set_active_upload, query.hash_, query.active_,
promise.wrap([](td::Unit &&) { return create_serialize_tl_object<ton_api::storage_daemon_success>(); }));
}
void run_control_query(ton_api::storage_daemon_getTorrents &query, td::Promise<td::BufferSlice> promise) {
td::actor::send_closure(
manager_, &StorageManager::get_all_torrents,
@ -644,9 +652,8 @@ class StorageDaemon : public td::actor::Actor {
promise.set_error(td::Status::Error("No storage provider"));
return;
}
td::actor::send_closure(provider_, &StorageProvider::get_params, promise.wrap([](ProviderParams params) {
return serialize_tl_object(params.tl(), true);
}));
td::actor::send_closure(provider_, &StorageProvider::get_params,
promise.wrap([](ProviderParams params) { return serialize_tl_object(params.tl(), true); }));
}
void run_control_query(ton_api::storage_daemon_setProviderParams &query, td::Promise<td::BufferSlice> promise) {
@ -785,6 +792,7 @@ class StorageDaemon : public td::actor::Actor {
auto obj = create_tl_object<ton_api::storage_daemon_torrent>();
fill_torrent_info_short(state.torrent, *obj);
obj->active_download_ = state.active_download;
obj->active_upload_ = state.active_upload;
obj->download_speed_ = state.download_speed;
obj->upload_speed_ = state.upload_speed;
promise.set_result(std::move(obj));

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

@ -758,7 +758,7 @@ storage.db.key.piecesInDb hash:int256 = storage.db.key.PiecesInDb;
storage.db.key.pieceInDb hash:int256 idx:long = storage.db.key.PieceInDb;
storage.db.torrentList torrents:(vector int256) = storage.db.TorrentList;
storage.db.torrent root_dir:string active_download:Bool = storage.db.TorrentShort;
storage.db.torrent root_dir:string active_download:Bool active_upload:Bool = storage.db.TorrentShort;
storage.db.priorities actions:(vector storage.PriorityAction) = storage.db.Priorities;
storage.db.piecesInDb pieces:(vector long) = storage.db.PiecesInDb;
@ -794,7 +794,7 @@ storage.daemon.torrent
total_size:flags.0?long description:flags.0?string
files_count:flags.1?long included_size:flags.1?long dir_name:flags.1?string
downloaded_size:long
root_dir:string active_download:Bool completed:Bool
root_dir:string active_download:Bool active_upload:Bool completed:Bool
download_speed:double upload_speed:double
fatal_error:flags.2?string
= storage.daemon.Torrent;
@ -829,10 +829,11 @@ storage.daemon.providerAddress address:string = storage.daemon.ProviderAddress;
---functions---
storage.daemon.setVerbosity verbosity:int = storage.daemon.Success;
storage.daemon.createTorrent path:string description:string = storage.daemon.TorrentFull;
storage.daemon.addByHash hash:int256 root_dir:string start_download:Bool priorities:(vector storage.PriorityAction) = storage.daemon.TorrentFull;
storage.daemon.addByMeta meta:bytes root_dir:string start_download:Bool priorities:(vector storage.PriorityAction) = storage.daemon.TorrentFull;
storage.daemon.createTorrent path:string description:string allow_upload:Bool = storage.daemon.TorrentFull;
storage.daemon.addByHash hash:int256 root_dir:string start_download:Bool allow_upload:Bool priorities:(vector storage.PriorityAction) = storage.daemon.TorrentFull;
storage.daemon.addByMeta meta:bytes root_dir:string start_download:Bool allow_upload:Bool priorities:(vector storage.PriorityAction) = storage.daemon.TorrentFull;
storage.daemon.setActiveDownload hash:int256 active:Bool = storage.daemon.Success;
storage.daemon.setActiveUpload hash:int256 active:Bool = storage.daemon.Success;
storage.daemon.getTorrents = storage.daemon.TorrentList;
storage.daemon.getTorrentFull hash:int256 = storage.daemon.TorrentFull;

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

Binary file not shown.