Skip to content

Commit

Permalink
small db changes
Browse files Browse the repository at this point in the history
  • Loading branch information
ton committed Apr 7, 2020
1 parent cea5281 commit 148a5e0
Show file tree
Hide file tree
Showing 4 changed files with 49 additions and 17 deletions.
9 changes: 5 additions & 4 deletions validator/db/archive-manager.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -216,7 +216,7 @@ void ArchiveManager::get_file_short(FileReference ref_id, td::Promise<td::Buffer
td::actor::send_closure(SelfId, &ArchiveManager::get_temp_file_short, std::move(ref_id), std::move(promise));
}
});
td::actor::send_closure(f->file_actor_id(), &ArchiveSlice::get_file, ref_id, std::move(P));
td::actor::send_closure(f->file_actor_id(), &ArchiveSlice::get_file, nullptr, ref_id, std::move(P));
return;
}
}
Expand All @@ -239,7 +239,7 @@ void ArchiveManager::get_key_block_proof(FileReference ref_id, td::Promise<td::B
if (search_in_key) {
auto f = get_file_desc_by_seqno(block_id.shard_full(), block_id.seqno(), true);
if (f) {
td::actor::send_closure(f->file_actor_id(), &ArchiveSlice::get_file, ref_id, std::move(promise));
td::actor::send_closure(f->file_actor_id(), &ArchiveSlice::get_file, nullptr, ref_id, std::move(promise));
} else {
promise.set_error(td::Status::Error(ErrorCode::notready, "key proof not in db"));
}
Expand Down Expand Up @@ -267,14 +267,15 @@ void ArchiveManager::get_file_short_cont(FileReference ref_id, PackageId idx, td
td::actor::send_closure(SelfId, &ArchiveManager::get_file_short_cont, std::move(ref_id), idx, std::move(promise));
}
});
td::actor::send_closure(f->file_actor_id(), &ArchiveSlice::get_file, std::move(ref_id), std::move(P));
td::actor::send_closure(f->file_actor_id(), &ArchiveSlice::get_file, nullptr, std::move(ref_id), std::move(P));
}

void ArchiveManager::get_file(ConstBlockHandle handle, FileReference ref_id, td::Promise<td::BufferSlice> promise) {
if (handle->moved_to_archive()) {
auto f = get_file_desc(handle->id().shard_full(), get_package_id(handle->masterchain_ref_block()), 0, 0, 0, false);
if (f) {
td::actor::send_closure(f->file_actor_id(), &ArchiveSlice::get_file, std::move(ref_id), std::move(promise));
td::actor::send_closure(f->file_actor_id(), &ArchiveSlice::get_file, std::move(handle), std::move(ref_id),
std::move(promise));
return;
}
}
Expand Down
37 changes: 27 additions & 10 deletions validator/db/archive-slice.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,8 @@ void ArchiveSlice::add_file(BlockHandle handle, FileReference ref_id, td::Buffer
promise.set_error(td::Status::Error(ErrorCode::notready, "package already gc'd"));
return;
}
auto &p = choose_package(
handle ? handle->id().is_masterchain() ? handle->id().seqno() : handle->masterchain_ref_block() : 0);
std::string value;
auto R = kv_->get(ref_id.hash().to_hex(), value);
R.ensure();
Expand All @@ -186,7 +188,7 @@ void ArchiveSlice::add_file(BlockHandle handle, FileReference ref_id, td::Buffer
std::move(promise));
});

td::actor::send_closure(writer_, &PackageWriter::append, ref_id.filename(), std::move(data), std::move(P));
td::actor::send_closure(p.writer, &PackageWriter::append, ref_id.filename(), std::move(data), std::move(P));
}

void ArchiveSlice::add_file_cont(FileReference ref_id, td::uint64 offset, td::uint64 size,
Expand Down Expand Up @@ -246,11 +248,13 @@ void ArchiveSlice::get_temp_handle(BlockIdExt block_id, td::Promise<ConstBlockHa
promise.set_value(std::move(handle));
}

void ArchiveSlice::get_file(FileReference ref_id, td::Promise<td::BufferSlice> promise) {
void ArchiveSlice::get_file(ConstBlockHandle handle, FileReference ref_id, td::Promise<td::BufferSlice> promise) {
if (destroyed_) {
promise.set_error(td::Status::Error(ErrorCode::notready, "package already gc'd"));
return;
}
auto &p = choose_package(
handle ? handle->id().is_masterchain() ? handle->id().seqno() : handle->masterchain_ref_block() : 0);
std::string value;
auto R = kv_->get(ref_id.hash().to_hex(), value);
R.ensure();
Expand All @@ -267,7 +271,7 @@ void ArchiveSlice::get_file(FileReference ref_id, td::Promise<td::BufferSlice> p
promise.set_value(std::move(R.move_as_ok().second));
}
});
td::actor::create_actor<PackageReader>("reader", package_, offset, std::move(P)).release();
td::actor::create_actor<PackageReader>("reader", p.package, offset, std::move(P)).release();
}

void ArchiveSlice::get_block_common(AccountIdPrefixFull account_id,
Expand Down Expand Up @@ -426,7 +430,7 @@ void ArchiveSlice::start_up() {
<< "': " << R.move_as_error();
return;
}
package_ = std::make_shared<Package>(R.move_as_ok());
auto pack = std::make_shared<Package>(R.move_as_ok());
kv_ = std::make_shared<td::RocksDb>(td::RocksDb::open(prefix_ + ".index").move_as_ok());

std::string value;
Expand All @@ -435,12 +439,13 @@ void ArchiveSlice::start_up() {

if (R2.move_as_ok() == td::KeyValue::GetStatus::Ok) {
auto len = td::to_integer<td::uint64>(value);
package_->truncate(len);
pack->truncate(len);
} else {
package_->truncate(0);
pack->truncate(0);
}

writer_ = td::actor::create_actor<PackageWriter>("writer", package_);
auto writer = td::actor::create_actor<PackageWriter>("writer", pack);
packages_.emplace_back(std::move(pack), std::move(writer), prefix_ + ".pack", 0);
}

void ArchiveSlice::begin_transaction() {
Expand Down Expand Up @@ -470,13 +475,26 @@ void ArchiveSlice::set_async_mode(bool mode, td::Promise<td::Unit> promise) {
huge_transaction_started_ = false;
}

td::actor::send_closure(writer_, &PackageWriter::set_async_mode, mode, std::move(promise));
td::MultiPromise mp;
auto ig = mp.init_guard();
ig.add_promise(std::move(promise));

for (auto &p : packages_) {
td::actor::send_closure(p.writer, &PackageWriter::set_async_mode, mode, std::move(promise));
}
}

ArchiveSlice::ArchiveSlice(td::uint32 archive_id, bool key_blocks_only, bool temp, std::string prefix)
: archive_id_(archive_id), key_blocks_only_(key_blocks_only), temp_(temp), prefix_(std::move(prefix)) {
}

ArchiveSlice::PackageInfo &ArchiveSlice::choose_package(BlockSeqno masterchain_seqno) {
if (temp_ || key_blocks_only_) {
return packages_[0];
}
return packages_[0];
}

namespace {

void destroy_db(std::string name, td::uint32 attempt, td::Promise<td::Unit> promise) {
Expand All @@ -502,8 +520,7 @@ void ArchiveSlice::destroy(td::Promise<td::Unit> promise) {
ig.add_promise(std::move(promise));
destroyed_ = true;

writer_.reset();
package_ = nullptr;
packages_.clear();
kv_ = nullptr;

td::unlink(prefix_ + ".pack").ensure();
Expand Down
18 changes: 15 additions & 3 deletions validator/db/archive-slice.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ class ArchiveSlice : public td::actor::Actor {
void add_file(BlockHandle handle, FileReference ref_id, td::BufferSlice data, td::Promise<td::Unit> promise);
void get_handle(BlockIdExt block_id, td::Promise<BlockHandle> promise);
void get_temp_handle(BlockIdExt block_id, td::Promise<ConstBlockHandle> promise);
void get_file(FileReference ref_id, td::Promise<td::BufferSlice> promise);
void get_file(ConstBlockHandle handle, FileReference ref_id, td::Promise<td::BufferSlice> promise);

/* from LTDB */
void get_block_by_unix_time(AccountIdPrefixFull account_id, UnixTime ts, td::Promise<ConstBlockHandle> promise);
Expand Down Expand Up @@ -99,9 +99,21 @@ class ArchiveSlice : public td::actor::Actor {
td::uint32 huge_transaction_size_ = 0;

std::string prefix_;
std::shared_ptr<Package> package_;
std::shared_ptr<td::KeyValue> kv_;
td::actor::ActorOwn<PackageWriter> writer_;

struct PackageInfo {
PackageInfo(std::shared_ptr<Package> package, td::actor::ActorOwn<PackageWriter> writer, std::string path,
td::uint32 idx)
: package(std::move(package)), writer(std ::move(writer)), path(std::move(path)), idx(idx) {
}
std::shared_ptr<Package> package;
td::actor::ActorOwn<PackageWriter> writer;
std::string path;
td::uint32 idx;
};
std::vector<PackageInfo> packages_;

PackageInfo &choose_package(BlockSeqno masterchain_seqno);
};

} // namespace validator
Expand Down
2 changes: 2 additions & 0 deletions validator/manager.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1445,6 +1445,8 @@ void ValidatorManagerImpl::started(ValidatorManagerInitResult R) {

last_key_block_handle_ = std::move(R.last_key_block_handle_);
last_known_key_block_handle_ = last_key_block_handle_;

CHECK(last_masterchain_block_handle_->is_applied());
callback_->new_key_block(last_key_block_handle_);

gc_masterchain_handle_ = std::move(R.gc_handle);
Expand Down

0 comments on commit 148a5e0

Please sign in to comment.