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

celldb in-memory mode (--celldb-in-memory option)

Browse source
This commit is contained in:
birydrad 2024-09-09 18:08:15 +02:00
parent 420029b056
commit 1723562748
48 changed files with 1966 additions and 201 deletions
Download patch file Download diff file Expand all files Collapse all files

26
common/delay.h
View file

@ -18,6 +18,7 @@
*/
#pragma once
#include "absl/strings/internal/str_format/parser.h"
#include "td/actor/actor.h"
namespace ton {
@ -49,4 +50,29 @@ template <typename T>
void delay_action(T promise, td::Timestamp timeout) {
DelayedAction<T>::create(std::move(promise), timeout);
}
template <typename PromiseT, typename ValueT>
class AsyncApply : public td::actor::Actor {
public:
AsyncApply(PromiseT promise, ValueT value) : promise_(std::move(promise)), value_(std::move(value)){
}
void start_up() override {
promise_(std::move(value_));
stop();
}
static void create(td::Slice name, PromiseT promise, ValueT value ) {
td::actor::create_actor<AsyncApply>(PSLICE() << "async:" << name, std::move(promise), std::move(value)).release();
}
private:
PromiseT promise_;
ValueT value_;
};
template <class PromiseT, class ValueT>
void async_apply(td::Slice name, PromiseT &&promise, ValueT &&value) {
AsyncApply<PromiseT, ValueT>::create(name, std::forward<PromiseT>(promise), std::forward<ValueT>(value));
}
} // namespace ton

3
crypto/CMakeLists.txt
View file

@ -151,6 +151,7 @@ set(TON_DB_SOURCE
vm/db/CellHashTable.h
vm/db/CellStorage.h
vm/db/TonDb.h
vm/db/InMemoryBagOfCellsDb.cpp
)
set(FIFT_SOURCE
@ -533,7 +534,7 @@ target_include_directories(create-state PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT
if (INTERNAL_COMPILE)
target_link_libraries(create-state PUBLIC ton_crypto fift-lib ton_block tonlib git)
else()
if (TONLIB_COMPILE)
if (TONLIB_COMPILE)
target_link_libraries(create-state PUBLIC ton_crypto fift-lib ton_block tonlib git)
else()
target_link_libraries(create-state PUBLIC ton_crypto fift-lib ton_block git)

4
crypto/common/bigint.hpp
View file

@ -2294,11 +2294,11 @@ std::string AnyIntView<Tr>::to_dec_string_destroy_any() {
stack.push_back(divmod_short_any(Tr::max_pow10));
} while (sgn());
char slice[word_bits * 97879 / 325147 + 2];
std::sprintf(slice, "%lld", stack.back());
std::snprintf(slice, sizeof(slice), "%lld", stack.back());
s += slice;
stack.pop_back();
while (stack.size()) {
std::sprintf(slice, "%018lld", stack.back());
std::snprintf(slice, sizeof(slice), "%018lld", stack.back());
s += slice;
stack.pop_back();
}

7
crypto/common/refcnt.cpp
View file

@ -29,6 +29,7 @@ Ref<CntObject> CntObject::clone() const {
namespace detail {
struct SafeDeleter {
public:
thread_local static td::int64 delete_count;
void retire(const CntObject *ptr) {
if (is_active_) {
to_delete_.push_back(ptr);
@ -39,9 +40,11 @@ struct SafeDeleter {
is_active_ = false;
};
delete ptr;
delete_count++;
while (!to_delete_.empty()) {
auto *ptr = to_delete_.back();
to_delete_.pop_back();
delete_count++;
delete ptr;
}
}
@ -50,6 +53,7 @@ struct SafeDeleter {
std::vector<const CntObject *> to_delete_;
bool is_active_{false};
};
thread_local td::int64 SafeDeleter::delete_count{0};
TD_THREAD_LOCAL SafeDeleter *deleter;
void safe_delete(const CntObject *ptr) {
@ -57,4 +61,7 @@ void safe_delete(const CntObject *ptr) {
deleter->retire(ptr);
}
} // namespace detail
int64 ref_get_delete_count() {
return detail::SafeDeleter::delete_count;
}
} // namespace td

1
crypto/common/refcnt.hpp
View file

@ -472,5 +472,6 @@ template <class T>
void swap(Ref<T>& r1, Ref<T>& r2) {
r1.swap(r2);
}
int64 ref_get_delete_count();
} // namespace td

2
crypto/fift/words.cpp
View file

@ -3425,7 +3425,7 @@ void import_cmdline_args(Dictionary& d, std::string arg0, int n, const char* con
cmdline_args->set(std::move(list));
for (int i = 1; i <= n; i++) {
char buffer[14];
sprintf(buffer, "$%d ", i);
snprintf(buffer, sizeof(buffer), "$%d ", i);
d.def_stack_word(buffer, std::bind(interpret_get_fixed_cmdline_arg, _1, i));
}
}

7
crypto/openssl/digest.hpp
View file

@ -48,6 +48,7 @@ struct OpensslEVP_SHA512 {
template <typename H>
class HashCtx {
EVP_MD_CTX *base_ctx{nullptr};
EVP_MD_CTX *ctx{nullptr};
void init();
void clear();
@ -77,16 +78,20 @@ class HashCtx {
template <typename H>
void HashCtx<H>::init() {
ctx = EVP_MD_CTX_create();
base_ctx = EVP_MD_CTX_create();
EVP_DigestInit_ex(base_ctx, H::get_evp(), 0);
reset();
}
template <typename H>
void HashCtx<H>::reset() {
EVP_DigestInit_ex(ctx, H::get_evp(), 0);
EVP_MD_CTX_copy_ex(ctx, base_ctx);
}
template <typename H>
void HashCtx<H>::clear() {
EVP_MD_CTX_destroy(base_ctx);
base_ctx = nullptr;
EVP_MD_CTX_destroy(ctx);
ctx = nullptr;
}

239
crypto/test/test-db.cpp
View file

@ -54,10 +54,14 @@
#include <set>
#include <map>
#include <thread>
#include <openssl/sha.h>
#include "openssl/digest.hpp"
#include "vm/dict.h"
#include <numeric>
namespace vm {
@ -127,12 +131,12 @@ class BenchSha256Low : public td::Benchmark {
void run(int n) override {
int res = 0;
td::Sha256State ctx;
SHA256_CTX ctx;
for (int i = 0; i < n; i++) {
ctx.init();
ctx.feed("abcd");
SHA256_Init(&ctx);
SHA256_Update(&ctx, "abcd", 4);
unsigned char buf[32];
ctx.extract(td::MutableSlice{buf, 32});
SHA256_Final(buf, &ctx);
res += buf[0];
}
td::do_not_optimize_away(res);
@ -158,12 +162,58 @@ class BenchSha256Tdlib : public td::Benchmark {
td::do_not_optimize_away(res);
}
};
template <class F>
void bench_threaded(F &&f) {
class Threaded : public td::Benchmark {
public:
explicit Threaded(F &&f) : f_(std::move(f)), base(f_()) {
}
F f_;
std::decay_t<decltype(f_())> base;
std::string get_description() const override {
return base.get_description() + " threaded";
}
void run(int n) override {
std::atomic<int> task_i{0};
int chunk_size = 1024;
size_t num_threads = 1;
n *= num_threads;
std::vector<td::thread> threads;
for (int i = 0; i < num_threads; i++) {
threads.emplace_back([&]() mutable {
auto bench = f_();
while (true) {
i = task_i.fetch_add(chunk_size, std::memory_order_relaxed);
auto i_end = std::min(n, i + chunk_size);
if (i > n) {
break;
}
bench.run(i_end - i);
}
});
}
for (auto &thread : threads) {
thread.join();
}
};
};
bench(Threaded(std::forward<F>(f)));
}
TEST(Cell, sha_benchmark) {
bench(BenchSha256Tdlib());
bench(BenchSha256Low());
bench(BenchSha256Reuse());
bench(BenchSha256());
}
TEST(Cell, sha_benchmark_threaded) {
bench_threaded([] { return BenchSha256Tdlib(); });
bench_threaded([]() { return BenchSha256Low(); });
bench_threaded([]() { return BenchSha256Reuse(); });
bench_threaded([]() { return BenchSha256(); });
}
std::string serialize_boc(Ref<Cell> cell, int mode = 31) {
CHECK(cell.not_null());
@ -762,16 +812,72 @@ TEST(TonDb, BocMultipleRoots) {
}
};
TEST(TonDb, DynamicBoc) {
TEST(TonDb, InMemoryDynamicBocSimple) {
auto counter = [] {
return td::NamedThreadSafeCounter::get_default().get_counter("DataCell").sum();
};
auto before = counter();
SCOPE_EXIT {
LOG_CHECK(before == counter()) << before << " vs " << counter();;
};
td::Random::Xorshift128plus rnd{123};
auto kv = std::make_shared<td::MemoryKeyValue>();
CellStorer storer(*kv);
auto boc = DynamicBagOfCellsDb::create_in_memory(kv.get(), {});
auto empty_cell = vm::CellBuilder().finalize();
boc->inc(empty_cell);
boc->prepare_commit().ensure();
boc->commit(storer).ensure();
auto got_empty_cell = boc->load_cell(empty_cell->get_hash().as_slice()).move_as_ok();
ASSERT_EQ(empty_cell->get_hash(), got_empty_cell->get_hash());
boc->dec(empty_cell);
auto one_ref_cell = vm::CellBuilder().store_ref(empty_cell).finalize();
boc->inc(one_ref_cell);
boc->prepare_commit().ensure();
boc->commit(storer).ensure();
auto got_one_ref_cell = boc->load_cell(one_ref_cell->get_hash().as_slice()).move_as_ok();
ASSERT_EQ(one_ref_cell->get_hash(), got_one_ref_cell->get_hash());
boc = DynamicBagOfCellsDb::create_in_memory(kv.get(), {});
auto random_ref_cell = gen_random_cell(3, rnd);
boc->inc(random_ref_cell);
boc->prepare_commit().ensure();
boc->commit(storer).ensure();
auto got_random_ref_cell = boc->load_cell(random_ref_cell->get_hash().as_slice()).move_as_ok();
ASSERT_EQ(random_ref_cell->get_hash(), got_random_ref_cell->get_hash());
boc = DynamicBagOfCellsDb::create_in_memory(kv.get(), {});
}
void test_dynamic_boc(std::optional<DynamicBagOfCellsDb::CreateInMemoryOptions> o_in_memory) {
auto counter = [] {
return td::NamedThreadSafeCounter::get_default().get_counter("DataCell").sum();
};
auto before = counter();
SCOPE_EXIT {
LOG_CHECK((o_in_memory && o_in_memory->use_arena) || before == counter()) << before << " vs " << counter();;
};
td::Random::Xorshift128plus rnd{123};
std::string old_root_hash;
std::string old_root_serialization;
auto kv = std::make_shared<td::MemoryKeyValue>();
auto dboc = DynamicBagOfCellsDb::create();
auto create_dboc = [&]() {
if (o_in_memory) {
auto res = DynamicBagOfCellsDb::create_in_memory(kv.get(), *o_in_memory);
auto roots_n = old_root_hash.empty() ? 0 : 1;
ASSERT_EQ(roots_n, res->get_stats().ok().roots_total_count);
return std::move(res);
}
return DynamicBagOfCellsDb::create();
};
auto dboc = create_dboc();
dboc->set_loader(std::make_unique<CellLoader>(kv));
for (int t = 1000; t >= 0; t--) {
if (rnd() % 10 == 0) {
dboc = DynamicBagOfCellsDb::create();
dboc = create_dboc();
}
dboc->set_loader(std::make_unique<CellLoader>(kv));
Ref<Cell> old_root;
@ -795,29 +901,66 @@ TEST(TonDb, DynamicBoc) {
if (t != 0) {
dboc->inc(cell);
}
dboc->prepare_commit();
dboc->prepare_commit().ensure();
{
CellStorer cell_storer(*kv);
dboc->commit(cell_storer);
dboc->commit(cell_storer).ensure();
}
}
ASSERT_EQ(0u, kv->count("").ok());
}
template <class F>
void with_all_boc_options(F &&f) {
LOG(INFO) << "Test dynamic boc";
LOG(INFO) << "\ton disk";
f({});
for (auto use_arena : {false, true}) {
for (auto less_memory : {false, true}) {
LOG(INFO) << "\tuse_arena=" << use_arena << " less_memory=" << less_memory;
f(DynamicBagOfCellsDb::CreateInMemoryOptions{.extra_threads = std::thread::hardware_concurrency(),
.verbose = false,
.use_arena = use_arena,
.use_less_memory_during_creation = less_memory});
}
}
}
TEST(TonDb, DynamicBoc) {
with_all_boc_options(test_dynamic_boc);
};
TEST(TonDb, DynamicBoc2) {
void test_dynamic_boc2(std::optional<DynamicBagOfCellsDb::CreateInMemoryOptions> o_in_memory) {
int VERBOSITY_NAME(boc) = VERBOSITY_NAME(DEBUG) + 10;
td::Random::Xorshift128plus rnd{123};
int total_roots = 10000;
int max_roots = 20;
std::vector<std::string> root_hashes(max_roots);
std::vector<Ref<Cell>> roots(max_roots);
int last_commit_at = 0;
int first_root_id = 0;
int last_root_id = 0;
auto kv = std::make_shared<td::MemoryKeyValue>();
auto dboc = DynamicBagOfCellsDb::create();
auto create_dboc = [&](td::int64 root_n) {
if (o_in_memory) {
auto res = DynamicBagOfCellsDb::create_in_memory(kv.get(), *o_in_memory);
auto stats = res->get_stats().move_as_ok();
ASSERT_EQ(root_n, stats.roots_total_count);
VLOG(boc) << "reset roots_n=" << stats.roots_total_count << " cells_n=" << stats.cells_total_count;
return std::move(res);
}
return DynamicBagOfCellsDb::create();
};
auto dboc = create_dboc(0);
dboc->set_loader(std::make_unique<CellLoader>(kv));
auto counter = [] {
return td::NamedThreadSafeCounter::get_default().get_counter("DataCell").sum();
};
auto before = counter();
SCOPE_EXIT {
LOG_CHECK((o_in_memory && o_in_memory->use_arena) || before == counter()) << before << " vs " << counter();;
};
std::vector<Ref<Cell>> roots(max_roots);
std::vector<std::string> root_hashes(max_roots);
auto add_root = [&](Ref<Cell> root) {
dboc->inc(root);
root_hashes[last_root_id % max_roots] = (root->get_hash().as_slice().str());
@ -825,18 +968,23 @@ TEST(TonDb, DynamicBoc2) {
last_root_id++;
};
auto get_root = [&](int root_id) {
auto get_root = [&](int root_id) -> Ref<Cell> {
VLOG(boc) << " from older root #" << root_id;
auto from_root = roots[root_id % max_roots];
if (from_root.is_null()) {
VLOG(boc) << " from db";
auto from_root_hash = root_hashes[root_id % max_roots];
from_root = dboc->load_cell(from_root_hash).move_as_ok();
if (o_in_memory && (rnd() % 2 == 0)) {
from_root = dboc->load_root(from_root_hash).move_as_ok();
} else {
from_root = dboc->load_cell(from_root_hash).move_as_ok();
}
} else {
VLOG(boc) << "FROM MEMORY";
}
return from_root;
};
std::map<CellHash, int> root_cnt;
auto new_root = [&] {
if (last_root_id == total_roots) {
return;
@ -850,7 +998,9 @@ TEST(TonDb, DynamicBoc2) {
from_root = get_root(rnd.fast(first_root_id, last_root_id - 1));
}
VLOG(boc) << " ...";
add_root(gen_random_cell(rnd.fast(1, 20), from_root, rnd));
auto new_root = gen_random_cell(rnd.fast(1, 20), from_root, rnd);
root_cnt[new_root->get_hash()]++;
add_root(std::move(new_root));
VLOG(boc) << " OK";
};
@ -870,7 +1020,7 @@ TEST(TonDb, DynamicBoc2) {
auto reset = [&] {
VLOG(boc) << "reset";
commit();
dboc = DynamicBagOfCellsDb::create();
dboc = create_dboc(td::int64(root_cnt.size()));
dboc->set_loader(std::make_unique<CellLoader>(kv));
};
@ -879,7 +1029,15 @@ TEST(TonDb, DynamicBoc2) {
if (first_root_id == last_root_id) {
return;
}
dboc->dec(get_root(first_root_id));
auto old_root = get_root(first_root_id);
auto it = root_cnt.find(old_root->get_hash());
it->second--;
CHECK(it->second >= 0);
if (it->second == 0) {
root_cnt.erase(it);
}
dboc->dec(std::move(old_root));
first_root_id++;
VLOG(boc) << " OK";
};
@ -893,6 +1051,10 @@ TEST(TonDb, DynamicBoc2) {
ASSERT_EQ(0u, kv->count("").ok());
}
TEST(TonDb, DynamicBoc2) {
with_all_boc_options(test_dynamic_boc2);
}
template <class BocDeserializerT>
td::Status test_boc_deserializer(std::vector<Ref<Cell>> cells, int mode) {
auto total_data_cells_before = vm::DataCell::get_total_data_cells();
@ -1950,14 +2112,15 @@ TEST(TonDb, BocRespectsUsageCell) {
ASSERT_STREQ(serialization, serialization_of_virtualized_cell);
}
TEST(TonDb, DynamicBocRespectsUsageCell) {
void test_dynamic_boc_respectes_usage_cell(std::optional<vm::DynamicBagOfCellsDb::CreateInMemoryOptions> o_in_memory) {
td::Random::Xorshift128plus rnd(123);
auto cell = vm::gen_random_cell(20, rnd, true);
auto usage_tree = std::make_shared<vm::CellUsageTree>();
auto usage_cell = vm::UsageCell::create(cell, usage_tree->root_ptr());
auto kv = std::make_shared<td::MemoryKeyValue>();
auto dboc = vm::DynamicBagOfCellsDb::create();
auto dboc = o_in_memory ? vm::DynamicBagOfCellsDb::create_in_memory(kv.get(), *o_in_memory)
: vm::DynamicBagOfCellsDb::create();
dboc->set_loader(std::make_unique<vm::CellLoader>(kv));
dboc->inc(usage_cell);
{
@ -1972,6 +2135,42 @@ TEST(TonDb, DynamicBocRespectsUsageCell) {
ASSERT_STREQ(serialization, serialization_of_virtualized_cell);
}
TEST(TonDb, DynamicBocRespectsUsageCell) {
vm::with_all_boc_options(test_dynamic_boc_respectes_usage_cell);
}
TEST(TonDb, LargeBocSerializer) {
td::Random::Xorshift128plus rnd{123};
size_t n = 1000000;
std::vector<td::uint64> data(n);
std::iota(data.begin(), data.end(), 0);
vm::CompactArray arr(data);
auto root = arr.root();
std::string path = "serialization";
td::unlink(path).ignore();
auto fd = td::FileFd::open(path, td::FileFd::Flags::Create | td::FileFd::Flags::Truncate | td::FileFd::Flags::Write)
.move_as_ok();
std_boc_serialize_to_file(root, fd, 31);
fd.close();
auto a = td::read_file_str(path).move_as_ok();
auto kv = std::make_shared<td::MemoryKeyValue>();
auto dboc = vm::DynamicBagOfCellsDb::create();
dboc->set_loader(std::make_unique<vm::CellLoader>(kv));
dboc->inc(root);
dboc->prepare_commit();
vm::CellStorer cell_storer(*kv);
dboc->commit(cell_storer);
dboc->set_loader(std::make_unique<vm::CellLoader>(kv));
td::unlink(path).ignore();
fd = td::FileFd::open(path, td::FileFd::Flags::Create | td::FileFd::Flags::Truncate | td::FileFd::Flags::Write)
.move_as_ok();
std_boc_serialize_to_file_large(dboc->get_cell_db_reader(), root->get_hash(), fd, 31);
fd.close();
auto b = td::read_file_str(path).move_as_ok();
CHECK(a == b);
}
TEST(TonDb, DoNotMakeListsPrunned) {
auto cell = vm::CellBuilder().store_bytes("abc").finalize();
auto is_prunned = [&](const td::Ref<vm::Cell> &cell) { return true; };

2
crypto/vm/atom.cpp
View file

@ -35,7 +35,7 @@ void Atom::print_to(std::ostream& os) const {
std::string Atom::make_name() const {
char buffer[16];
sprintf(buffer, "atom#%d", index_);
snprintf(buffer, sizeof(buffer), "atom#%d", index_);
return buffer;
}

77
crypto/vm/boc.cpp
View file

@ -183,6 +183,9 @@ int BagOfCells::add_root(td::Ref<vm::Cell> add_root) {
// Changes in this function may require corresponding changes in crypto/vm/large-boc-serializer.cpp
td::Status BagOfCells::import_cells() {
if (logger_ptr_) {
logger_ptr_->start_stage("import_cells");
}
cells_clear();
for (auto& root : roots) {
auto res = import_cell(root.cell, 0);
@ -196,6 +199,9 @@ td::Status BagOfCells::import_cells() {
//LOG(INFO) << "[cells: " << cell_count << ", refs: " << int_refs << ", bytes: " << data_bytes
//<< ", internal hashes: " << int_hashes << ", top hashes: " << top_hashes << "]";
CHECK(cell_count != 0);
if (logger_ptr_) {
logger_ptr_->finish_stage(PSLICE() << cell_count << " cells");
}
return td::Status::OK();
}
@ -207,6 +213,9 @@ td::Result<int> BagOfCells::import_cell(td::Ref<vm::Cell> cell, int depth) {
if (cell.is_null()) {
return td::Status::Error("error while importing a cell into a bag of cells: cell is null");
}
if (logger_ptr_) {
TRY_STATUS(logger_ptr_->on_cell_processed());
}
auto it = cells.find(cell->get_hash());
if (it != cells.end()) {
auto pos = it->second;
@ -436,17 +445,19 @@ std::size_t BagOfCells::estimate_serialized_size(int mode) {
return res.ok();
}
BagOfCells& BagOfCells::serialize(int mode) {
td::Status BagOfCells::serialize(int mode) {
std::size_t size_est = estimate_serialized_size(mode);
if (!size_est) {
serialized.clear();
return *this;
return td::Status::OK();
}
serialized.resize(size_est);
if (serialize_to(const_cast<unsigned char*>(serialized.data()), serialized.size(), mode) != size_est) {
TRY_RESULT(size, serialize_to(const_cast<unsigned char*>(serialized.data()), serialized.size(), mode));
if (size != size_est) {
serialized.clear();
return td::Status::Error("serialization failed");
}
return *this;
return td::Status::OK();
}
std::string BagOfCells::serialize_to_string(int mode) {
@ -456,8 +467,8 @@ std::string BagOfCells::serialize_to_string(int mode) {
}
std::string res;
res.resize(size_est, 0);
if (serialize_to(const_cast<unsigned char*>(reinterpret_cast<const unsigned char*>(res.data())), res.size(), mode) ==
res.size()) {
if (serialize_to(const_cast<unsigned char*>(reinterpret_cast<const unsigned char*>(res.data())), res.size(), mode)
.move_as_ok() == res.size()) {
return res;
} else {
return {};
@ -470,8 +481,9 @@ td::Result<td::BufferSlice> BagOfCells::serialize_to_slice(int mode) {
return td::Status::Error("no cells to serialize to this bag of cells");
}
td::BufferSlice res(size_est);
if (serialize_to(const_cast<unsigned char*>(reinterpret_cast<const unsigned char*>(res.data())), res.size(), mode) ==
res.size()) {
TRY_RESULT(size, serialize_to(const_cast<unsigned char*>(reinterpret_cast<const unsigned char*>(res.data())),
res.size(), mode));
if (size == res.size()) {
return std::move(res);
} else {
return td::Status::Error("error while serializing a bag of cells: actual serialized size differs from estimated");
@ -494,14 +506,10 @@ std::string BagOfCells::extract_string() const {
// cell_data:(tot_cells_size * [ uint8 ])
// = BagOfCells;
// Changes in this function may require corresponding changes in crypto/vm/large-boc-serializer.cpp
template<typename WriterT>
std::size_t BagOfCells::serialize_to_impl(WriterT& writer, int mode) {
auto store_ref = [&](unsigned long long value) {
writer.store_uint(value, info.ref_byte_size);
};
auto store_offset = [&](unsigned long long value) {
writer.store_uint(value, info.offset_byte_size);
};
template <typename WriterT>
td::Result<std::size_t> BagOfCells::serialize_to_impl(WriterT& writer, int mode) {
auto store_ref = [&](unsigned long long value) { writer.store_uint(value, info.ref_byte_size); };
auto store_offset = [&](unsigned long long value) { writer.store_uint(value, info.offset_byte_size); };
writer.store_uint(info.magic, 4);
@ -536,6 +544,9 @@ std::size_t BagOfCells::serialize_to_impl(WriterT& writer, int mode) {
DCHECK((unsigned)cell_count == cell_list_.size());
if (info.has_index) {
std::size_t offs = 0;
if (logger_ptr_) {
logger_ptr_->start_stage("generate_index");
}
for (int i = cell_count - 1; i >= 0; --i) {
const Ref<DataCell>& dc = cell_list_[i].dc_ref;
bool with_hash = (mode & Mode::WithIntHashes) && !cell_list_[i].wt;
@ -548,11 +559,20 @@ std::size_t BagOfCells::serialize_to_impl(WriterT& writer, int mode) {
fixed_offset = offs * 2 + cell_list_[i].should_cache;
}
store_offset(fixed_offset);
if (logger_ptr_) {
TRY_STATUS(logger_ptr_->on_cell_processed());
}
}
if (logger_ptr_) {
logger_ptr_->finish_stage("");
}
DCHECK(offs == info.data_size);
}
DCHECK(writer.position() == info.data_offset);
size_t keep_position = writer.position();
if (logger_ptr_) {
logger_ptr_->start_stage("serialize");
}
for (int i = 0; i < cell_count; ++i) {
const auto& dc_info = cell_list_[cell_count - 1 - i];
const Ref<DataCell>& dc = dc_info.dc_ref;
@ -572,6 +592,9 @@ std::size_t BagOfCells::serialize_to_impl(WriterT& writer, int mode) {
// std::cerr << ' ' << k;
}
// std::cerr << std::endl;
if (logger_ptr_) {
TRY_STATUS(logger_ptr_->on_cell_processed());
}
}
writer.chk();
DCHECK(writer.position() - keep_position == info.data_size);
@ -580,11 +603,14 @@ std::size_t BagOfCells::serialize_to_impl(WriterT& writer, int mode) {
unsigned crc = writer.get_crc32();
writer.store_uint(td::bswap32(crc), 4);
}
if (logger_ptr_) {
logger_ptr_->finish_stage(PSLICE() << cell_count << " cells, " << writer.position() << " bytes");
}
DCHECK(writer.empty());
return writer.position();
}
std::size_t BagOfCells::serialize_to(unsigned char* buffer, std::size_t buff_size, int mode) {
td::Result<std::size_t> BagOfCells::serialize_to(unsigned char* buffer, std::size_t buff_size, int mode) {
std::size_t size_est = estimate_serialized_size(mode);
if (!size_est || size_est > buff_size) {
return 0;
@ -599,7 +625,7 @@ td::Status BagOfCells::serialize_to_file(td::FileFd& fd, int mode) {
return td::Status::Error("no cells to serialize to this bag of cells");
}
boc_writers::FileWriter writer{fd, size_est};
size_t s = serialize_to_impl(writer, mode);
TRY_RESULT(s, serialize_to_impl(writer, mode));
TRY_STATUS(writer.finalize());
if (s != size_est) {
return td::Status::Error("error while serializing a bag of cells: actual serialized size differs from estimated");
@ -1001,6 +1027,21 @@ td::Result<td::BufferSlice> std_boc_serialize_multi(std::vector<Ref<Cell>> roots
}
return boc.serialize_to_slice(mode);
}
td::Status std_boc_serialize_to_file(Ref<Cell> root, td::FileFd& fd, int mode,
td::CancellationToken cancellation_token) {
if (root.is_null()) {
return td::Status::Error("cannot serialize a null cell reference into a bag of cells");
}
td::Timer timer;
BagOfCellsLogger logger(std::move(cancellation_token));
BagOfCells boc;
boc.set_logger(&logger);
boc.add_root(std::move(root));
TRY_STATUS(boc.import_cells());
TRY_STATUS(boc.serialize_to_file(fd, mode));
LOG(ERROR) << "serialization took " << timer.elapsed() << "s";
return td::Status::OK();
}
/*
*

55
crypto/vm/boc.h
View file

@ -27,6 +27,8 @@
#include "td/utils/buffer.h"
#include "td/utils/HashMap.h"
#include "td/utils/HashSet.h"
#include "td/utils/Time.h"
#include "td/utils/Timer.h"
#include "td/utils/port/FileFd.h"
namespace vm {
@ -199,6 +201,43 @@ struct CellSerializationInfo {
td::Result<Ref<DataCell>> create_data_cell(td::Slice data, td::Span<Ref<Cell>> refs) const;
};
class BagOfCellsLogger {
public:
BagOfCellsLogger() = default;
explicit BagOfCellsLogger(td::CancellationToken cancellation_token)
: cancellation_token_(std::move(cancellation_token)) {
}
void start_stage(std::string stage) {
log_speed_at_ = td::Timestamp::in(LOG_SPEED_PERIOD);
processed_cells_ = 0;
timer_ = {};
stage_ = std::move(stage);
}
void finish_stage(td::Slice desc) {
LOG(ERROR) << "serializer: " << stage_ << " took " << timer_.elapsed() << "s, " << desc;
}
td::Status on_cell_processed() {
++processed_cells_;
if (processed_cells_ % 1000 == 0) {
TRY_STATUS(cancellation_token_.check());
}
if (log_speed_at_.is_in_past()) {
log_speed_at_ += LOG_SPEED_PERIOD;
LOG(WARNING) << "serializer: " << stage_ << " " << (double)processed_cells_ / LOG_SPEED_PERIOD << " cells/s";
processed_cells_ = 0;
}
return td::Status::OK();
}
private:
std::string stage_;
td::Timer timer_;
td::CancellationToken cancellation_token_;
td::Timestamp log_speed_at_;
size_t processed_cells_ = 0;
static constexpr double LOG_SPEED_PERIOD = 120.0;
};
class BagOfCells {
public:
enum { hash_bytes = vm::Cell::hash_bytes, default_max_roots = 16384 };
@ -283,6 +322,7 @@ class BagOfCells {
const unsigned char* index_ptr{nullptr};
const unsigned char* data_ptr{nullptr};
std::vector<unsigned long long> custom_index;
BagOfCellsLogger* logger_ptr_{nullptr};
public:
void clear();
@ -292,14 +332,17 @@ class BagOfCells {
int add_root(td::Ref<vm::Cell> add_root);
td::Status import_cells() TD_WARN_UNUSED_RESULT;
BagOfCells() = default;
void set_logger(BagOfCellsLogger* logger_ptr) {
logger_ptr_ = logger_ptr;
}
std::size_t estimate_serialized_size(int mode = 0);
BagOfCells& serialize(int mode = 0);
std::string serialize_to_string(int mode = 0);
td::Status serialize(int mode = 0);
td::string serialize_to_string(int mode = 0);
td::Result<td::BufferSlice> serialize_to_slice(int mode = 0);
std::size_t serialize_to(unsigned char* buffer, std::size_t buff_size, int mode = 0);
td::Result<std::size_t> serialize_to(unsigned char* buffer, std::size_t buff_size, int mode = 0);
td::Status serialize_to_file(td::FileFd& fd, int mode = 0);
template<typename WriterT>
std::size_t serialize_to_impl(WriterT& writer, int mode = 0);
template <typename WriterT>
td::Result<std::size_t> serialize_to_impl(WriterT& writer, int mode = 0);
std::string extract_string() const;
td::Result<long long> deserialize(const td::Slice& data, int max_roots = default_max_roots);
@ -345,6 +388,8 @@ td::Result<std::vector<Ref<Cell>>> std_boc_deserialize_multi(td::Slice data,
int max_roots = BagOfCells::default_max_roots);
td::Result<td::BufferSlice> std_boc_serialize_multi(std::vector<Ref<Cell>> root, int mode = 0);
td::Status std_boc_serialize_to_file(Ref<Cell> root, td::FileFd& fd, int mode = 0,
td::CancellationToken cancellation_token = {});
td::Status std_boc_serialize_to_file_large(std::shared_ptr<CellDbReader> reader, Cell::Hash root_hash, td::FileFd& fd,
int mode = 0, td::CancellationToken cancellation_token = {});

28
crypto/vm/cells/Cell.h
View file

@ -19,6 +19,7 @@
#pragma once
#include "common/refcnt.hpp"
#include "common/bitstring.h"
#include "td/utils/HashSet.h"
#include "vm/cells/CellHash.h"
#include "vm/cells/CellTraits.h"
@ -86,4 +87,31 @@ class Cell : public CellTraits {
};
std::ostream& operator<<(std::ostream& os, const Cell& c);
using is_transparent = void; // Pred to use
inline vm::CellHash as_cell_hash(const Ref<Cell>& cell) {
return cell->get_hash();
}
inline vm::CellHash as_cell_hash(td::Slice hash) {
return vm::CellHash::from_slice(hash);
}
inline vm::CellHash as_cell_hash(vm::CellHash hash) {
return hash;
}
struct CellEqF {
using is_transparent = void; // Pred to use
template <class A, class B>
bool operator()(const A& a, const B& b) const {
return as_cell_hash(a) == as_cell_hash(b);
}
};
struct CellHashF {
using is_transparent = void; // Pred to use
using transparent_key_equal = CellEqF;
template <class T>
size_t operator()(const T& value) const {
return cell_hash_slice_hash(as_cell_hash(value).as_slice());
}
};
using CellHashSet = td::HashSet<td::Ref<Cell>, CellHashF, CellEqF>;
} // namespace vm

2
crypto/vm/cells/CellBuilder.cpp
View file

@ -617,7 +617,7 @@ std::string CellBuilder::to_hex() const {
int len = serialize(buff, sizeof(buff));
char hex_buff[Cell::max_serialized_bytes * 2 + 1];
for (int i = 0; i < len; i++) {
sprintf(hex_buff + 2 * i, "%02x", buff[i]);
snprintf(hex_buff + 2 * i, sizeof(hex_buff) - 2 * i, "%02x", buff[i]);
}
return hex_buff;
}

6
crypto/vm/cells/CellHash.h
View file

@ -74,13 +74,17 @@ struct CellHash {
};
} // namespace vm
inline size_t cell_hash_slice_hash(td::Slice hash) {
// use offset 8, because in db keys are grouped by first bytes.
return td::as<size_t>(hash.ubegin() + 8);
}
namespace std {
template <>
struct hash<vm::CellHash> {
typedef vm::CellHash argument_type;
typedef std::size_t result_type;
result_type operator()(argument_type const& s) const noexcept {
return td::as<size_t>(s.as_slice().ubegin());
return cell_hash_slice_hash(s.as_slice());
}
};
} // namespace std

2
crypto/vm/cells/CellSlice.cpp
View file

@ -976,7 +976,7 @@ void CellSlice::dump(std::ostream& os, int level, bool endl) const {
os << "; refs: " << refs_st << ".." << refs_en;
if (level > 2) {
char tmp[64];
std::sprintf(tmp, "; ptr=data+%ld; z=%016llx",
std::snprintf(tmp, sizeof(tmp), "; ptr=data+%ld; z=%016llx",
static_cast<long>(ptr && cell.not_null() ? ptr - cell->get_data() : -1), static_cast<long long>(z));
os << tmp << " (have " << size() << " bits; " << zd << " preloaded)";
}

19
crypto/vm/cells/CellWithStorage.h
View file

@ -20,6 +20,15 @@
namespace vm {
namespace detail {
template <class CellT>
struct DefaultAllocator {
template <class T, class... ArgsT>
std::unique_ptr<CellT> make_unique(ArgsT&&... args) {
return std::make_unique<T>(std::forward<ArgsT>(args)...);
}
};
template <class CellT, size_t Size = 0>
class CellWithArrayStorage : public CellT {
public:
@ -29,14 +38,14 @@ class CellWithArrayStorage : public CellT {
~CellWithArrayStorage() {
CellT::destroy_storage(get_storage());
}
template <class... ArgsT>
static std::unique_ptr<CellT> create(size_t storage_size, ArgsT&&... args) {
template <class Allocator, class... ArgsT>
static auto create(Allocator allocator, size_t storage_size, ArgsT&&... args) {
static_assert(CellT::max_storage_size <= 40 * 8, "");
//size = 128 + 32 + 8;
auto size = (storage_size + 7) / 8;
#define CASE(size) \
case (size): \
return std::make_unique<CellWithArrayStorage<CellT, (size)*8>>(std::forward<ArgsT>(args)...);
return allocator. template make_unique<CellWithArrayStorage<CellT, (size) * 8>>(std::forward<ArgsT>(args)...);
#define CASE2(offset) CASE(offset) CASE(offset + 1)
#define CASE8(offset) CASE2(offset) CASE2(offset + 2) CASE2(offset + 4) CASE2(offset + 6)
#define CASE32(offset) CASE8(offset) CASE8(offset + 8) CASE8(offset + 16) CASE8(offset + 24)
@ -48,6 +57,10 @@ class CellWithArrayStorage : public CellT {
LOG(FATAL) << "TOO BIG " << storage_size;
UNREACHABLE();
}
template <class... ArgsT>
static std::unique_ptr<CellT> create(size_t storage_size, ArgsT&&... args) {
return create(DefaultAllocator<CellT>{}, storage_size, std::forward<ArgsT>(args)...);
}
private:
alignas(alignof(void*)) char storage_[Size];

39
crypto/vm/cells/DataCell.cpp
View file

@ -25,7 +25,44 @@
#include "vm/cells/CellWithStorage.h"
namespace vm {
thread_local bool DataCell::use_arena = false;
namespace {
template <class CellT>
struct ArenaAllocator {
template <class T, class... ArgsT>
std::unique_ptr<CellT> make_unique(ArgsT&&... args) {
auto* ptr = fast_alloc(sizeof(T));
T* obj = new (ptr) T(std::forward<ArgsT>(args)...);
return std::unique_ptr<T>(obj);
}
private:
td::MutableSlice alloc_batch() {
size_t batch_size = 1 << 20;
auto batch = std::make_unique<char[]>(batch_size);
return td::MutableSlice(batch.release(), batch_size);
}
char* fast_alloc(size_t size) {
thread_local td::MutableSlice batch;
auto aligned_size = (size + 7) / 8 * 8;
if (batch.size() < size) {
batch = alloc_batch();
}
auto res = batch.begin();
batch.remove_prefix(aligned_size);
return res;
}
};
}
std::unique_ptr<DataCell> DataCell::create_empty_data_cell(Info info) {
if (use_arena) {
ArenaAllocator<DataCell> allocator;
auto res = detail::CellWithArrayStorage<DataCell>::create(allocator, info.get_storage_size(), info);
// this is dangerous
Ref<DataCell>(res.get()).release();
return res;
}
return detail::CellWithUniquePtrStorage<DataCell>::create(info.get_storage_size(), info);
}
@ -359,7 +396,7 @@ std::string DataCell::to_hex() const {
int len = serialize(buff, sizeof(buff));
char hex_buff[max_serialized_bytes * 2 + 1];
for (int i = 0; i < len; i++) {
sprintf(hex_buff + 2 * i, "%02x", buff[i]);
snprintf(hex_buff + 2 * i, sizeof(hex_buff) - 2 * i, "%02x", buff[i]);
}
return hex_buff;
}

27
crypto/vm/cells/DataCell.h
View file

@ -27,6 +27,9 @@ namespace vm {
class DataCell : public Cell {
public:
// NB: cells created with use_arena=true are never freed
static thread_local bool use_arena;
DataCell(const DataCell& other) = delete;
~DataCell() override;
@ -121,10 +124,6 @@ class DataCell : public Cell {
void destroy_storage(char* storage);
explicit DataCell(Info info);
Cell* get_ref_raw_ptr(unsigned idx) const {
DCHECK(idx < get_refs_cnt());
return info_.get_refs(get_storage())[idx];
}
public:
td::Result<LoadedCell> load_cell() const override {
@ -152,6 +151,20 @@ class DataCell : public Cell {
return Ref<Cell>(get_ref_raw_ptr(idx));
}
Cell* get_ref_raw_ptr(unsigned idx) const {
DCHECK(idx < get_refs_cnt());
return info_.get_refs(get_storage())[idx];
}
Ref<Cell> reset_ref_unsafe(unsigned idx, Ref<Cell> ref, bool check_hash = true) {
CHECK(idx < get_refs_cnt());
auto refs = info_.get_refs(get_storage());
CHECK(!check_hash || refs[idx]->get_hash() == ref->get_hash());
auto res = Ref<Cell>(refs[idx], Ref<Cell>::acquire_t{}); // call destructor
refs[idx] = ref.release();
return res;
}
td::uint32 get_virtualization() const override {
return info_.virtualization_;
}
@ -173,6 +186,9 @@ class DataCell : public Cell {
return ((get_bits() + 23) >> 3) +
(with_hashes ? get_level_mask().get_hashes_count() * (hash_bytes + depth_bytes) : 0);
}
size_t get_storage_size() const {
return info_.get_storage_size();
}
int serialize(unsigned char* buff, int buff_size, bool with_hashes = false) const;
std::string serialize() const;
std::string to_hex() const;
@ -207,6 +223,9 @@ class DataCell : public Cell {
};
std::ostream& operator<<(std::ostream& os, const DataCell& c);
inline CellHash as_cell_hash(const Ref<DataCell>& cell) {
return cell->get_hash();
}
} // namespace vm

12
crypto/vm/cells/PrunnedCell.h
View file

@ -30,18 +30,27 @@ struct PrunnedCellInfo {
template <class ExtraT>
class PrunnedCell : public Cell {
public:
ExtraT& get_extra() {
return extra_;
}
const ExtraT& get_extra() const {
return extra_;
}
static td::Result<Ref<PrunnedCell<ExtraT>>> create(const PrunnedCellInfo& prunned_cell_info, ExtraT&& extra) {
return create(detail::DefaultAllocator<PrunnedCell<ExtraT>>(), prunned_cell_info, std::forward<ExtraT>(extra));
}
template <class AllocatorT>
static td::Result<Ref<PrunnedCell<ExtraT>>> create(AllocatorT allocator, const PrunnedCellInfo& prunned_cell_info,
ExtraT&& extra) {
auto level_mask = prunned_cell_info.level_mask;
if (level_mask.get_level() > max_level) {
return td::Status::Error("Level is too big");
}
Info info(level_mask);
auto prunned_cell =
detail::CellWithUniquePtrStorage<PrunnedCell<ExtraT>>::create(info.get_storage_size(), info, std::move(extra));
detail::CellWithArrayStorage<PrunnedCell<ExtraT>>::create(allocator, info.get_storage_size(), info, std::move(extra));
TRY_STATUS(prunned_cell->init(prunned_cell_info));
return Ref<PrunnedCell<ExtraT>>(prunned_cell.release(), typename Ref<PrunnedCell<ExtraT>>::acquire_t{});
}
@ -51,6 +60,7 @@ class PrunnedCell : public Cell {
}
protected:
static constexpr auto max_storage_size = (max_level + 1) * (hash_bytes + sizeof(td::uint16));
struct Info {
Info(LevelMask level_mask) {
level_mask_ = level_mask.get_mask() & 7;

4
crypto/vm/db/CellHashTable.h
View file

@ -19,7 +19,7 @@
#pragma once
#include "td/utils/Slice.h"
#include "td/utils/HashSet.h"
#include <set>
namespace vm {
@ -73,6 +73,6 @@ class CellHashTable {
}
private:
std::set<InfoT, std::less<>> set_;
td::NodeHashSet<InfoT, typename InfoT::Hash, typename InfoT::Eq> set_;
};
} // namespace vm

23
crypto/vm/db/CellStorage.cpp
View file

@ -153,18 +153,26 @@ CellLoader::CellLoader(std::shared_ptr<KeyValueReader> reader, std::function<voi
td::Result<CellLoader::LoadResult> CellLoader::load(td::Slice hash, bool need_data, ExtCellCreator &ext_cell_creator) {
//LOG(ERROR) << "Storage: load cell " << hash.size() << " " << td::base64_encode(hash);
TD_PERF_COUNTER(cell_load);
LoadResult res;
std::string serialized;
TRY_RESULT(get_status, reader_->get(hash, serialized));
if (get_status != KeyValue::GetStatus::Ok) {
DCHECK(get_status == KeyValue::GetStatus::NotFound);
return res;
return LoadResult{};
}
TRY_RESULT(res, load(hash, serialized, need_data, ext_cell_creator));
if (on_load_callback_) {
on_load_callback_(res);
}
return res;
}
td::Result<CellLoader::LoadResult> CellLoader::load(td::Slice hash, td::Slice value, bool need_data,
ExtCellCreator &ext_cell_creator) {
LoadResult res;
res.status = LoadResult::Ok;
RefcntCellParser refcnt_cell(need_data);
td::TlParser parser(serialized);
td::TlParser parser(value);
refcnt_cell.parse(parser, ext_cell_creator);
TRY_STATUS(parser.get_status());
@ -172,9 +180,6 @@ td::Result<CellLoader::LoadResult> CellLoader::load(td::Slice hash, bool need_da
res.cell_ = std::move(refcnt_cell.cell);
res.stored_boc_ = refcnt_cell.stored_boc_;
//CHECK(res.cell_->get_hash() == hash);
if (on_load_callback_) {
on_load_callback_(res);
}
return res;
}
@ -186,7 +191,11 @@ td::Status CellStorer::erase(td::Slice hash) {
return kv_.erase(hash);
}
std::string CellStorer::serialize_value(td::int32 refcnt, const td::Ref<DataCell> &cell, bool as_boc) {
return td::serialize(RefcntCellStorer(refcnt, cell, as_boc));
}
td::Status CellStorer::set(td::int32 refcnt, const td::Ref<DataCell> &cell, bool as_boc) {
return kv_.set(cell->get_hash().as_slice(), td::serialize(RefcntCellStorer(refcnt, cell, as_boc)));
return kv_.set(cell->get_hash().as_slice(), serialize_value(refcnt, cell, as_boc));
}
} // namespace vm

2
crypto/vm/db/CellStorage.h
View file

@ -49,6 +49,7 @@ class CellLoader {
};
CellLoader(std::shared_ptr<KeyValueReader> reader, std::function<void(const LoadResult &)> on_load_callback = {});
td::Result<LoadResult> load(td::Slice hash, bool need_data, ExtCellCreator &ext_cell_creator);
static td::Result<LoadResult> load(td::Slice hash, td::Slice value, bool need_data, ExtCellCreator &ext_cell_creator);
private:
std::shared_ptr<KeyValueReader> reader_;
@ -60,6 +61,7 @@ class CellStorer {
CellStorer(KeyValue &kv);
td::Status erase(td::Slice hash);
td::Status set(td::int32 refcnt, const td::Ref<DataCell> &cell, bool as_boc);
static std::string serialize_value(td::int32 refcnt, const td::Ref<DataCell> &cell, bool as_boc);
private:
KeyValue &kv_;

20
crypto/vm/db/DynamicBagOfCellsDb.cpp
View file

@ -60,6 +60,20 @@ struct CellInfo {
bool operator<(const CellInfo &other) const {
return key() < other.key();
}
struct Eq {
using is_transparent = void; // Pred to use
bool operator()(const CellInfo &info, const CellInfo &other_info) const { return info.key() == other_info.key();}
bool operator()(const CellInfo &info, td::Slice hash) const { return info.key().as_slice() == hash;}
bool operator()(td::Slice hash, const CellInfo &info) const { return info.key().as_slice() == hash;}
};
struct Hash {
using is_transparent = void; // Pred to use
using transparent_key_equal = Eq;
size_t operator()(td::Slice hash) const { return cell_hash_slice_hash(hash); }
size_t operator()(const CellInfo &info) const { return cell_hash_slice_hash(info.key().as_slice());}
};
};
bool operator<(const CellInfo &a, td::Slice b) {
@ -86,6 +100,12 @@ class DynamicBagOfCellsDbImpl : public DynamicBagOfCellsDb, private ExtCellCreat
TRY_RESULT(loaded_cell, get_cell_info_force(hash).cell->load_cell());
return std::move(loaded_cell.data_cell);
}
td::Result<Ref<DataCell>> load_root(td::Slice hash) override {
return load_cell(hash);
}
td::Result<Ref<DataCell>> load_root_thread_safe(td::Slice hash) const override {
return td::Status::Error("Not implemented");
}
void load_cell_async(td::Slice hash, std::shared_ptr<AsyncExecutor> executor,
td::Promise<Ref<DataCell>> promise) override {
auto info = hash_table_.get_if_exists(hash);

33
crypto/vm/db/DynamicBagOfCellsDb.h
View file

@ -23,6 +23,11 @@
#include "td/utils/Status.h"
#include "td/actor/PromiseFuture.h"
#include <thread>
namespace td {
class KeyValueReader;
}
namespace vm {
class CellLoader;
class CellStorer;
@ -45,12 +50,20 @@ class DynamicBagOfCellsDb {
public:
virtual ~DynamicBagOfCellsDb() = default;
virtual td::Result<Ref<DataCell>> load_cell(td::Slice hash) = 0;
virtual td::Result<Ref<DataCell>> load_root(td::Slice hash) = 0;
virtual td::Result<Ref<DataCell>> load_root_thread_safe(td::Slice hash) const = 0;
struct Stats {
td::int64 roots_total_count{0};
td::int64 cells_total_count{0};
td::int64 cells_total_size{0};
void apply_diff(Stats diff) {
std::vector<std::pair<std::string, std::string>> custom_stats;
void apply_diff(const Stats &diff) {
roots_total_count += diff.roots_total_count;
cells_total_count += diff.cells_total_count;
cells_total_size += diff.cells_total_size;
CHECK(roots_total_count >= 0);
CHECK(cells_total_count >= 0);
CHECK(cells_total_size >= 0);
}
};
virtual void inc(const Ref<Cell> &old_root) = 0;
@ -58,6 +71,9 @@ class DynamicBagOfCellsDb {
virtual td::Status prepare_commit() = 0;
virtual Stats get_stats_diff() = 0;
virtual td::Result<Stats> get_stats() {
return td::Status::Error("Not implemented");
}
virtual td::Status commit(CellStorer &) = 0;
virtual std::shared_ptr<CellDbReader> get_cell_db_reader() = 0;
@ -65,13 +81,24 @@ class DynamicBagOfCellsDb {
virtual td::Status set_loader(std::unique_ptr<CellLoader> loader) = 0;
virtual void set_celldb_compress_depth(td::uint32 value) = 0;
virtual vm::ExtCellCreator& as_ext_cell_creator() = 0;
virtual vm::ExtCellCreator &as_ext_cell_creator() = 0;
static std::unique_ptr<DynamicBagOfCellsDb> create();
struct CreateInMemoryOptions {
size_t extra_threads{std::thread::hardware_concurrency()};
bool verbose{true};
// Allocated DataCels will never be deleted
bool use_arena{false};
// Almost no overhead in memory during creation, but will scan database twice
bool use_less_memory_during_creation{true};
};
static std::unique_ptr<DynamicBagOfCellsDb> create_in_memory(td::KeyValueReader *kv, CreateInMemoryOptions options);
class AsyncExecutor {
public:
virtual ~AsyncExecutor() {}
virtual ~AsyncExecutor() {
}
virtual void execute_async(std::function<void()> f) = 0;
virtual void execute_sync(std::function<void()> f) = 0;
};

982
crypto/vm/db/InMemoryBagOfCellsDb.cpp Normal file
View file

@ -0,0 +1,982 @@
#include "CellStorage.h"
#include "DynamicBagOfCellsDb.h"
#include "td/utils/Timer.h"
#include "td/utils/base64.h"
#include "td/utils/format.h"
#include "td/utils/int_types.h"
#include "td/utils/misc.h"
#include "td/utils/port/Stat.h"
#include "vm/cells/CellHash.h"
#include "vm/cells/CellSlice.h"
#include "vm/cells/DataCell.h"
#include "vm/cells/ExtCell.h"
#include "td/utils/HashMap.h"
#include "td/utils/HashSet.h"
#if TD_PORT_POSIX
#include <sys/mman.h>
#include <unistd.h>
#endif
namespace vm {
namespace {
constexpr bool use_dense_hash_map = true;
template <class F>
void parallel_run(size_t n, F &&run_task, size_t extra_threads_n) {
std::atomic<size_t> next_task_id{0};
auto loop = [&] {
while (true) {
auto task_id = next_task_id++;
if (task_id >= n) {
break;
}
run_task(task_id);
}
};
// NB: it could be important that td::thread is used, not std::thread
std::vector<td::thread> threads;
for (size_t i = 0; i < extra_threads_n; i++) {
threads.emplace_back(loop);
}
loop();
for (auto &thread : threads) {
thread.join();
}
threads.clear();
}
struct UniqueAccess {
struct Release {
void operator()(UniqueAccess *access) const {
if (access) {
access->release();
}
}
};
using Lock = std::unique_ptr<UniqueAccess, Release>;
Lock lock() {
CHECK(!locked_.exchange(true));
return Lock(this);
}
private:
std::atomic<bool> locked_{false};
void release() {
locked_ = false;
}
};
class DefaultPrunnedCellCreator : public ExtCellCreator {
public:
td::Result<Ref<Cell>> ext_cell(Cell::LevelMask level_mask, td::Slice hash, td::Slice depth) override {
TRY_RESULT(cell, PrunnedCell<td::Unit>::create(PrunnedCellInfo{level_mask, hash, depth}, td::Unit{}));
return cell;
}
};
class ArenaPrunnedCellCreator : public ExtCellCreator {
struct ArenaAllocator {
ArenaAllocator() {
// only one instance ever
static UniqueAccess unique_access;
[[maybe_unused]] auto ptr = unique_access.lock().release();
}
std::mutex mutex;
struct Deleter {
static constexpr size_t batch_size = 1 << 24;
#if TD_PORT_POSIX
static std::unique_ptr<char, Deleter> alloc() {
char *ptr = reinterpret_cast<char *>(
mmap(NULL, batch_size, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0));
CHECK(ptr != nullptr);
return std::unique_ptr<char, Deleter>(ptr);
}
void operator()(char *ptr) const {
munmap(ptr, batch_size);
}
#else
static std::unique_ptr<char, Deleter> alloc() {
auto ptr = reinterpret_cast<char *>(malloc(batch_size)));
CHECK(ptr != nullptr);
return std::unique_ptr<char, Deleter>(ptr);
}
void operator()(char *ptr) const {
free(ptr);
}
#endif
};
std::vector<std::unique_ptr<char, Deleter>> arena;
td::uint64 arena_generation{0};
td::MutableSlice alloc_batch() {
auto batch = Deleter::alloc();
auto res = td::MutableSlice(batch.get(), Deleter::batch_size);
std::lock_guard<std::mutex> guard(mutex);
arena.emplace_back(std::move(batch));
return res;
}
char *alloc(size_t size) {
thread_local td::MutableSlice batch;
thread_local td::uint64 batch_generation{0};
auto aligned_size = (size + 7) / 8 * 8;
if (batch.size() < size || batch_generation != arena_generation) {
batch = alloc_batch();
batch_generation = arena_generation;
}
auto res = batch.begin();
batch.remove_prefix(aligned_size);
return res;
}
void clear() {
std::lock_guard<std::mutex> guard(mutex);
arena_generation++;
td::reset_to_empty(arena);
}
};
static ArenaAllocator arena_;
static td::ThreadSafeCounter cells_count_;
public:
struct Counter {
Counter() {
cells_count_.add(1);
}
Counter(Counter &&other) {
cells_count_.add(1);
}
Counter(const Counter &other) {
cells_count_.add(1);
}
~Counter() {
cells_count_.add(-1);
}
};
struct Allocator {
template <class T, class... ArgsT>
std::unique_ptr<PrunnedCell<Counter>> make_unique(ArgsT &&...args) {
auto *ptr = arena_.alloc(sizeof(T));
T *obj = new (ptr) T(std::forward<ArgsT>(args)...);
return std::unique_ptr<T>(obj);
}
};
td::Result<Ref<Cell>> ext_cell(Cell::LevelMask level_mask, td::Slice hash, td::Slice depth) override {
Allocator allocator;
TRY_RESULT(cell, PrunnedCell<Counter>::create(allocator, PrunnedCellInfo{level_mask, hash, depth}, Counter()));
return cell;
}
static td::int64 count() {
return cells_count_.sum();
}
static void clear_arena() {
LOG_CHECK(cells_count_.sum() == 0) << cells_count_.sum();
arena_.clear();
}
};
td::ThreadSafeCounter ArenaPrunnedCellCreator::cells_count_;
ArenaPrunnedCellCreator::ArenaAllocator ArenaPrunnedCellCreator::arena_;
struct CellInfo {
mutable td::int32 db_refcnt{0};
Ref<DataCell> cell;
};
static_assert(sizeof(CellInfo) == 16);
CellHash as_cell_hash(const CellInfo &info) {
return info.cell->get_hash();
}
struct CellInfoHashTableBaseline {
td::HashSet<CellInfo, CellHashF, CellEqF> ht_;
const CellInfo *find(CellHash hash) const {
if (auto it = ht_.find(hash); it != ht_.end()) {
return &*it;
}
return nullptr;
}
void erase(CellHash hash) {
auto it = ht_.find(hash);
CHECK(it != ht_.end());
ht_.erase(it);
}
void insert(CellInfo info) {
ht_.insert(std::move(info));
}
template <class Iterator>
void init_from(Iterator begin, Iterator end) {
ht_ = td::HashSet<CellInfo, CellHashF, CellEqF>(begin, end);
}
size_t size() const {
return ht_.size();
}
auto begin() const {
return ht_.begin();
}
auto end() const {
return ht_.end();
}
size_t bucket_count() const {
return ht_.bucket_count();
}
template <class F>
auto for_each(F &&f) {
for (auto &it : ht_) {
f(it);
}
}
};
struct CellInfoHashTableDense {
size_t dense_ht_size_{0};
size_t dense_ht_buckets_{1};
std::vector<size_t> dense_ht_offsets_{1};
std::vector<CellInfo> dense_ht_values_;
td::HashSet<CellInfo, CellHashF, CellEqF> new_ht_;
size_t dense_choose_bucket(const CellHash &hash) const {
return cell_hash_slice_hash(hash.as_slice()) % dense_ht_buckets_;
}
const CellInfo *dense_find(CellHash hash) const {
auto bucket_i = dense_choose_bucket(hash);
auto begin = dense_ht_values_.begin() + dense_ht_offsets_[bucket_i];
auto end = dense_ht_values_.begin() + dense_ht_offsets_[bucket_i + 1];
for (auto it = begin; it != end; ++it) {
if (it->cell.not_null() && it->cell->get_hash() == hash) {
return &*it;
}
}
return nullptr;
}
CellInfo *dense_find_empty(CellHash hash) {
auto bucket_i = dense_choose_bucket(hash);
auto begin = dense_ht_values_.begin() + dense_ht_offsets_[bucket_i];
auto end = dense_ht_values_.begin() + dense_ht_offsets_[bucket_i + 1];
for (auto it = begin; it != end; ++it) {
if (it->cell.is_null()) {
return &*it;
}
}
return nullptr;
}
const CellInfo *find(CellHash hash) const {
if (auto it = new_ht_.find(hash); it != new_ht_.end()) {
return &*it;
}
if (auto it = dense_find(hash)) {
return it;
}
return nullptr;
}
void erase(CellHash hash) {
if (auto it = new_ht_.find(hash); it != new_ht_.end()) {
new_ht_.erase(it);
return;
}
auto info = dense_find(hash);
CHECK(info && info->db_refcnt > 0);
info->db_refcnt = 0;
const_cast<CellInfo *>(info)->cell = {};
CHECK(dense_ht_size_ > 0);
dense_ht_size_--;
}
void insert(CellInfo info) {
if (auto dest = dense_find_empty(info.cell->get_hash())) {
*dest = std::move(info);
dense_ht_size_++;
return;
}
new_ht_.insert(std::move(info));
}
template <class Iterator>
void init_from(Iterator begin, Iterator end) {
auto size = std::distance(begin, end);
dense_ht_buckets_ = std::max(size_t(1), size_t(size / 8));
std::vector<size_t> offsets(dense_ht_buckets_ + 2);
for (auto it = begin; it != end; ++it) {
auto bucket_i = dense_choose_bucket(it->cell->get_hash());
offsets[bucket_i + 2]++;
}
for (size_t i = 1; i < offsets.size(); i++) {
offsets[i] += offsets[i - 1];
}
dense_ht_values_.resize(size);
for (auto it = begin; it != end; ++it) {
auto bucket_i = dense_choose_bucket(it->cell->get_hash());
dense_ht_values_[offsets[bucket_i + 1]++] = std::move(*it);
}
CHECK(offsets[0] == 0);
CHECK(offsets[offsets.size() - 1] == size);
CHECK(offsets[offsets.size() - 2] == size);
dense_ht_offsets_ = std::move(offsets);
dense_ht_size_ = size;
}
size_t size() const {
return dense_ht_size_ + new_ht_.size();
}
template <class F>
auto for_each(F &&f) {
for (auto &it : dense_ht_values_) {
if (it.cell.not_null()) {
f(it);
}
}
for (auto &it : new_ht_) {
f(it);
}
}
size_t bucket_count() const {
return new_ht_.bucket_count() + dense_ht_values_.size();
}
};
using CellInfoHashTable = std::conditional_t<use_dense_hash_map, CellInfoHashTableDense, CellInfoHashTableBaseline>;
class CellStorage {
struct PrivateTag {};
struct CellBucket;
struct None {
void operator()(CellBucket *bucket) {
}
};
struct CellBucketRef {
UniqueAccess::Lock lock;
std::unique_ptr<CellBucket, None> bucket;
CellBucket &operator*() {
return *bucket;
}
CellBucket *operator->() {
return bucket.get();
}
};
struct CellBucket {
mutable UniqueAccess access_;
CellInfoHashTable infos_;
std::vector<CellInfo> cells_;
std::vector<Ref<DataCell>> roots_;
size_t boc_count_{0};
[[maybe_unused]] char pad3[TD_CONCURRENCY_PAD];
void clear() {
td::reset_to_empty(infos_);
td::reset_to_empty(cells_);
td::reset_to_empty(roots_);
}
CellBucketRef unique_access() const {
auto lock = access_.lock();
return CellBucketRef{.lock = std::move(lock),
.bucket = std::unique_ptr<CellBucket, None>(const_cast<CellBucket *>(this))};
}
};
std::array<CellBucket, 256> buckets_{};
bool inited_{false};
const CellBucket &get_bucket(size_t i) const {
return buckets_.at(i);
}
const CellBucket &get_bucket(const CellHash &hash) const {
return get_bucket(hash.as_array()[0]);
}
mutable UniqueAccess local_access_;
td::HashSet<Ref<DataCell>, CellHashF, CellEqF> local_roots_;
DynamicBagOfCellsDb::Stats stats_;
mutable std::mutex root_mutex_;
td::HashSet<Ref<DataCell>, CellHashF, CellEqF> roots_;
public:
std::optional<CellInfo> get_info(const CellHash &hash) const {
auto lock = local_access_.lock();
auto &bucket = get_bucket(hash);
if (auto info_ptr = bucket.infos_.find(hash)) {
return *info_ptr;
}
return {};
}
DynamicBagOfCellsDb::Stats get_stats() {
auto unique_access = local_access_.lock();
auto stats = stats_;
auto add_stat = [&stats](auto key, auto value) {
stats.custom_stats.emplace_back(std::move(key), PSTRING() << value);
};
if constexpr (use_dense_hash_map) {
size_t dense_ht_capacity = 0;
size_t new_ht_capacity = 0;
size_t dense_ht_size = 0;
size_t new_ht_size = 0;
for_each_bucket(0, [&](auto bucket_id, CellBucket &bucket) {
dense_ht_capacity += bucket.infos_.dense_ht_values_.size();
dense_ht_size += bucket.infos_.dense_ht_size_;
new_ht_capacity += bucket.infos_.new_ht_.bucket_count();
new_ht_size += bucket.infos_.new_ht_.size();
});
auto size = new_ht_size + dense_ht_size;
auto capacity = new_ht_capacity + dense_ht_capacity;
add_stat("ht.capacity", capacity);
add_stat("ht.size", size);
add_stat("ht.load", double(size) / std::max(1.0, double(capacity)));
add_stat("ht.dense_ht_capacity", dense_ht_capacity);
add_stat("ht.dense_ht_size", dense_ht_size);
add_stat("ht.dense_ht_load", double(dense_ht_size) / std::max(1.0, double(dense_ht_capacity)));
add_stat("ht.new_ht_capacity", new_ht_capacity);
add_stat("ht.new_ht_size", new_ht_size);
add_stat("ht.new_ht_load", double(new_ht_size) / std::max(1.0, double(new_ht_capacity)));
} else {
size_t capacity = 0;
size_t size = 0;
for_each_bucket(0, [&](auto bucket_id, CellBucket &bucket) {
capacity += bucket.infos_.bucket_count();
size += bucket.infos_.size();
});
add_stat("ht.capacity", capacity);
add_stat("ht.size", size);
add_stat("ht.load", double(size) / std::max(1.0, double(capacity)));
}
CHECK(stats.roots_total_count == local_roots_.size());
return stats;
}
void apply_stats_diff(DynamicBagOfCellsDb::Stats diff) {
auto unique_access = local_access_.lock();
stats_.apply_diff(diff);
CHECK(stats_.roots_total_count == local_roots_.size());
size_t cells_count{0};
for_each_bucket(0, [&](size_t bucket_id, auto &bucket) { cells_count += bucket.infos_.size(); });
CHECK(stats_.cells_total_count == cells_count);
}
td::Result<Ref<DataCell>> load_cell(const CellHash &hash) const {
auto lock = local_access_.lock();
auto &bucket = get_bucket(hash);
if (auto info_ptr = bucket.infos_.find(hash)) {
return info_ptr->cell;
}
return td::Status::Error("not found");
}
td::Result<Ref<DataCell>> load_root_local(const CellHash &hash) const {
auto lock = local_access_.lock();
if (auto it = local_roots_.find(hash); it != local_roots_.end()) {
return *it;
}
return td::Status::Error("not found");
}
td::Result<Ref<DataCell>> load_root_shared(const CellHash &hash) const {
std::lock_guard<std::mutex> lock(root_mutex_);
if (auto it = roots_.find(hash); it != roots_.end()) {
return *it;
}
return td::Status::Error("not found");
}
void erase(const CellHash &hash) {
auto lock = local_access_.lock();
auto bucket = get_bucket(hash).unique_access();
bucket->infos_.erase(hash);
if (auto local_it = local_roots_.find(hash); local_it != local_roots_.end()) {
local_roots_.erase(local_it);
std::lock_guard<std::mutex> root_lock(root_mutex_);
auto shared_it = roots_.find(hash);
CHECK(shared_it != roots_.end());
roots_.erase(shared_it);
CHECK(stats_.roots_total_count > 0);
stats_.roots_total_count--;
}
}
void add_new_root(Ref<DataCell> cell) {
auto lock = local_access_.lock();
if (local_roots_.insert(cell).second) {
std::lock_guard<std::mutex> lock(root_mutex_);
roots_.insert(std::move(cell));
stats_.roots_total_count++;
}
}
void set(td::int32 refcnt, Ref<DataCell> cell) {
auto lock = local_access_.lock();
//LOG(ERROR) << "setting refcnt to " << refcnt << ", cell " << td::base64_encode(cell->get_hash().as_slice());
auto hash = cell->get_hash();
auto bucket = get_bucket(hash).unique_access();
if (auto info_ptr = bucket->infos_.find(hash)) {
CHECK(info_ptr->cell.get() == cell.get());
info_ptr->db_refcnt = refcnt;
} else {
bucket->infos_.insert({.db_refcnt = refcnt, .cell = std::move(cell)});
}
}
template <class F>
static td::unique_ptr<CellStorage> build(DynamicBagOfCellsDb::CreateInMemoryOptions options,
F &&parallel_scan_cells) {
auto storage = td::make_unique<CellStorage>(PrivateTag{});
storage->do_build(options, parallel_scan_cells);
return storage;
}
~CellStorage() {
clear();
}
CellStorage() = delete;
explicit CellStorage(PrivateTag) {
}
private:
template <class F>
void do_build(DynamicBagOfCellsDb::CreateInMemoryOptions options, F &&parallel_scan_cells) {
auto verbose = options.verbose;
td::Slice P = "loading in-memory cell database: ";
LOG_IF(WARNING, verbose) << P << "start with options use_arena=" << options.use_arena
<< " use_less_memory_during_creation=" << options.use_less_memory_during_creation
<< " use_dense_hash_map=" << use_dense_hash_map;
auto full_timer = td::Timer();
auto lock = local_access_.lock();
CHECK(ArenaPrunnedCellCreator::count() == 0);
ArenaPrunnedCellCreator arena_pc_creator;
DefaultPrunnedCellCreator default_pc_creator;
auto timer = td::Timer();
size_t cell_count{0};
size_t desc_count{0};
if (options.use_less_memory_during_creation) {
auto [new_cell_count, new_desc_count] = parallel_scan_cells(
default_pc_creator, options.use_arena,
[&](td::int32 refcnt, Ref<DataCell> cell) { initial_set_without_refs(refcnt, std::move(cell)); });
cell_count = new_cell_count;
desc_count = new_desc_count;
} else {
auto [new_cell_count, new_desc_count] =
parallel_scan_cells(arena_pc_creator, options.use_arena,
[&](td::int32 refcnt, Ref<DataCell> cell) { initial_set(refcnt, std::move(cell)); });
cell_count = new_cell_count;
desc_count = new_desc_count;
}
LOG_IF(WARNING, verbose) << P << "cells loaded in " << timer.elapsed() << "s, cells_count= " << cell_count
<< " prunned_cells_count=" << ArenaPrunnedCellCreator::count();
timer = td::Timer();
for_each_bucket(options.extra_threads, [&](size_t bucket_id, auto &bucket) { build_hashtable(bucket); });
size_t ht_capacity = 0;
size_t ht_size = 0;
for_each_bucket(0, [&](size_t bucket_id, auto &bucket) {
ht_size += bucket.infos_.size();
ht_capacity += bucket.infos_.bucket_count();
});
double load_factor = double(ht_size) / std::max(double(ht_capacity), 1.0);
LOG_IF(WARNING, verbose) << P << "hashtable created in " << timer.elapsed()
<< "s, hashtables_expected_size=" << td::format::as_size(ht_capacity * sizeof(CellInfo))
<< " load_factor=" << load_factor;
timer = td::Timer();
if (options.use_less_memory_during_creation) {
auto [new_cell_count, new_desc_count] =
parallel_scan_cells(default_pc_creator, false,
[&](td::int32 refcnt, Ref<DataCell> cell) { secondary_set(refcnt, std::move(cell)); });
CHECK(new_cell_count == cell_count);
CHECK(new_desc_count == desc_count);
} else {
for_each_bucket(options.extra_threads, [&](size_t bucket_id, auto &bucket) { reset_refs(bucket); });
}
LOG_IF(WARNING, verbose) << P << "refs rearranged in " << timer.elapsed() << "s";
timer = td::Timer();
using Stats = DynamicBagOfCellsDb::Stats;
std::vector<Stats> bucket_stats(buckets_.size());
std::atomic<size_t> boc_count{0};
for_each_bucket(options.extra_threads, [&](size_t bucket_id, auto &bucket) {
bucket_stats[bucket_id] = validate_bucket_a(bucket, options.use_arena);
boc_count += bucket.boc_count_;
});
for_each_bucket(options.extra_threads, [&](size_t bucket_id, auto &bucket) { validate_bucket_b(bucket); });
stats_ = {};
for (auto &bucket_stat : bucket_stats) {
stats_.apply_diff(bucket_stat);
}
LOG_IF(WARNING, verbose) << P << "refcnt validated in " << timer.elapsed() << "s";
timer = td::Timer();
build_roots();
LOG_IF(WARNING, verbose) << P << "roots hashtable built in " << timer.elapsed() << "s";
ArenaPrunnedCellCreator::clear_arena();
LOG_IF(WARNING, verbose) << P << "arena cleared in " << timer.elapsed();
lock.reset();
auto r_mem_stat = td::mem_stat();
td::MemStat mem_stat;
if (r_mem_stat.is_ok()) {
mem_stat = r_mem_stat.move_as_ok();
}
auto stats = get_stats();
td::StringBuilder sb;
for (auto &[key, value] : stats.custom_stats) {
sb << "\n\t" << key << "=" << value;
}
LOG_IF(ERROR, desc_count != 0 && desc_count != stats.roots_total_count + 1)
<< "desc<> keys count is " << desc_count << " wich is different from roots count " << stats.roots_total_count;
LOG_IF(WARNING, verbose)
<< P << "done in " << full_timer.elapsed() << "\n\troots_count=" << stats.roots_total_count << "\n\t"
<< desc_count << "\n\tcells_count=" << stats.cells_total_count
<< "\n\tcells_size=" << td::format::as_size(stats.cells_total_size) << "\n\tboc_count=" << boc_count.load()
<< sb.as_cslice() << "\n\tdata_cells_size=" << td::format::as_size(sizeof(DataCell) * stats.cells_total_count)
<< "\n\tdata_cell_size=" << sizeof(DataCell) << "\n\texpected_memory_used="
<< td::format::as_size(stats.cells_total_count * (sizeof(DataCell) + sizeof(CellInfo) * 3 / 2) +
stats.cells_total_size)
<< "\n\tbest_possible_memory_used"
<< td::format::as_size(stats.cells_total_count * (sizeof(DataCell) + sizeof(CellInfo)) + stats.cells_total_size)
<< "\n\tmemory_used=" << td::format::as_size(mem_stat.resident_size_)
<< "\n\tpeak_memory_used=" << td::format::as_size(mem_stat.resident_size_peak_);
inited_ = true;
}
template <class F>
void for_each_bucket(size_t extra_threads, F &&f) {
parallel_run(
buckets_.size(), [&](auto task_id) { f(task_id, *get_bucket(task_id).unique_access()); }, extra_threads);
}
void clear() {
auto unique_access = local_access_.lock();
for_each_bucket(td::thread::hardware_concurrency(), [&](size_t bucket_id, auto &bucket) { bucket.clear(); });
local_roots_.clear();
{
auto lock = std::lock_guard<std::mutex>(root_mutex_);
roots_.clear();
}
}
void initial_set(td::int32 refcnt, Ref<DataCell> cell) {
CHECK(!inited_);
auto bucket = get_bucket(cell->get_hash()).unique_access();
bucket->cells_.push_back({.db_refcnt = refcnt, .cell = std::move(cell)});
}
void initial_set_without_refs(td::int32 refcnt, Ref<DataCell> cell_ref) {
CHECK(!inited_);
auto bucket = get_bucket(cell_ref->get_hash()).unique_access();
auto &cell = const_cast<DataCell &>(*cell_ref);
for (unsigned i = 0; i < cell.size_refs(); i++) {
auto to_destroy = cell.reset_ref_unsafe(i, Ref<Cell>(), false);
if (to_destroy->is_loaded()) {
bucket->boc_count_++;
}
}
bucket->cells_.push_back({.db_refcnt = refcnt, .cell = std::move(cell_ref)});
}
void secondary_set(td::int32 refcnt, Ref<DataCell> cell_copy) {
CHECK(!inited_);
auto bucket = get_bucket(cell_copy->get_hash()).unique_access();
auto info = bucket->infos_.find(cell_copy->get_hash());
CHECK(info);
CellSlice cs(NoVm{}, std::move(cell_copy));
auto &cell = const_cast<DataCell &>(*info->cell);
CHECK(cs.size_refs() == cell.size_refs());
for (unsigned i = 0; i < cell.size_refs(); i++) {
auto prunned_cell_hash = cs.fetch_ref()->get_hash();
auto &prunned_cell_bucket = get_bucket(prunned_cell_hash);
auto full_cell_ptr = prunned_cell_bucket.infos_.find(prunned_cell_hash);
CHECK(full_cell_ptr);
auto full_cell = full_cell_ptr->cell;
auto to_destroy = cell.reset_ref_unsafe(i, std::move(full_cell), false);
CHECK(to_destroy.is_null());
}
}
void build_hashtable(CellBucket &bucket) {
bucket.infos_.init_from(bucket.cells_.begin(), bucket.cells_.end());
LOG_CHECK(bucket.infos_.size() == bucket.cells_.size()) << bucket.infos_.size() << " vs " << bucket.cells_.size();
td::reset_to_empty(bucket.cells_);
LOG_CHECK(bucket.cells_.capacity() == 0) << bucket.cells_.capacity();
}
void reset_refs(CellBucket &bucket) {
bucket.infos_.for_each([&](auto &it) {
// This is generally very dangerous, but should be safe here
auto &cell = const_cast<DataCell &>(*it.cell);
for (unsigned i = 0; i < cell.size_refs(); i++) {
auto prunned_cell = cell.get_ref_raw_ptr(i);
auto prunned_cell_hash = prunned_cell->get_hash();
auto &prunned_cell_bucket = get_bucket(prunned_cell_hash);
auto full_cell_ptr = prunned_cell_bucket.infos_.find(prunned_cell_hash);
CHECK(full_cell_ptr);
auto full_cell = full_cell_ptr->cell;
auto to_destroy = cell.reset_ref_unsafe(i, std::move(full_cell));
if (!to_destroy->is_loaded()) {
Ref<PrunnedCell<ArenaPrunnedCellCreator::Counter>> x(std::move(to_destroy));
x->~PrunnedCell<ArenaPrunnedCellCreator::Counter>();
x.release();
} else {
bucket.boc_count_++;
}
}
});
}
DynamicBagOfCellsDb::Stats validate_bucket_a(CellBucket &bucket, bool use_arena) {
DynamicBagOfCellsDb::Stats stats;
bucket.infos_.for_each([&](auto &it) {
int cell_ref_cnt = it.cell->get_refcnt();
CHECK(it.db_refcnt + 1 + use_arena >= cell_ref_cnt);
auto extra_refcnt = it.db_refcnt + 1 + use_arena - cell_ref_cnt;
if (extra_refcnt != 0) {
bucket.roots_.push_back(it.cell);
stats.roots_total_count++;
}
stats.cells_total_count++;
stats.cells_total_size += static_cast<td::int64>(it.cell->get_storage_size());
});
return stats;
}
void validate_bucket_b(CellBucket &bucket) {
// sanity check
bucket.infos_.for_each([&](auto &it) {
CellSlice cs(NoVm{}, it.cell);
while (cs.have_refs()) {
CHECK(cs.fetch_ref().not_null());
}
});
}
void build_roots() {
for (auto &it : buckets_) {
for (auto &root : it.roots_) {
local_roots_.insert(std::move(root));
}
td::reset_to_empty(it.roots_);
}
auto lock = std::lock_guard<std::mutex>(root_mutex_);
roots_ = local_roots_;
}
};
class InMemoryBagOfCellsDb : public DynamicBagOfCellsDb {
public:
explicit InMemoryBagOfCellsDb(td::unique_ptr<CellStorage> storage) : storage_(std::move(storage)) {
}
td::Result<Ref<DataCell>> load_cell(td::Slice hash) override {
return storage_->load_cell(CellHash::from_slice(hash));
}
td::Result<Ref<DataCell>> load_root(td::Slice hash) override {
return storage_->load_root_local(CellHash::from_slice(hash));
}
td::Result<Ref<DataCell>> load_root_thread_safe(td::Slice hash) const override {
return storage_->load_root_shared(CellHash::from_slice(hash));
}
void inc(const Ref<Cell> &cell) override {
if (cell.is_null()) {
return;
}
if (cell->get_virtualization() != 0) {
return;
}
to_inc_.push_back(cell);
}
void dec(const Ref<Cell> &cell) override {
if (cell.is_null()) {
return;
}
if (cell->get_virtualization() != 0) {
return;
}
to_dec_.push_back(cell);
}
td::Status commit(CellStorer &cell_storer) override {
if (!to_inc_.empty() || !to_dec_.empty()) {
TRY_STATUS(prepare_commit());
}
Stats diff;
CHECK(to_dec_.empty());
for (auto &it : info_) {
auto &info = it.second;
if (info.diff_refcnt == 0) {
continue;
}
auto refcnt = td::narrow_cast<td::int32>(static_cast<td::int64>(info.db_refcnt) + info.diff_refcnt);
CHECK(refcnt >= 0);
if (refcnt > 0) {
cell_storer.set(refcnt, info.cell, false);
storage_->set(refcnt, info.cell);
if (info.db_refcnt == 0) {
diff.cells_total_count++;
diff.cells_total_size += static_cast<td::int64>(info.cell->get_storage_size());
}
} else {
cell_storer.erase(info.cell->get_hash().as_slice());
storage_->erase(info.cell->get_hash());
diff.cells_total_count--;
diff.cells_total_size -= static_cast<td::int64>(info.cell->get_storage_size());
}
}
storage_->apply_stats_diff(diff);
info_ = {};
return td::Status::OK();
}
td::Result<Stats> get_stats() override {
return storage_->get_stats();
}
// Not implemented or trivial or deprecated methods
td::Status set_loader(std::unique_ptr<CellLoader> loader) override {
return td::Status::OK();
}
td::Status prepare_commit() override {
CHECK(info_.empty());
for (auto &to_inc : to_inc_) {
auto new_root = do_inc(to_inc);
storage_->add_new_root(std::move(new_root));
}
for (auto &to_dec : to_dec_) {
do_dec(to_dec);
}
to_dec_ = {};
to_inc_ = {};
return td::Status::OK();
}
Stats get_stats_diff() override {
LOG(FATAL) << "Not implemented";
return {};
}
std::shared_ptr<CellDbReader> get_cell_db_reader() override {
return {};
}
void set_celldb_compress_depth(td::uint32 value) override {
LOG(FATAL) << "Not implemented";
}
ExtCellCreator &as_ext_cell_creator() override {
UNREACHABLE();
}
void load_cell_async(td::Slice hash, std::shared_ptr<AsyncExecutor> executor,
td::Promise<Ref<DataCell>> promise) override {
LOG(FATAL) << "Not implemented";
}
private:
td::unique_ptr<CellStorage> storage_;
struct Info {
td::int32 db_refcnt{0};
td::int32 diff_refcnt{0};
Ref<DataCell> cell;
};
td::HashMap<CellHash, Info> info_;
std::unique_ptr<CellLoader> loader_;
std::vector<Ref<Cell>> to_inc_;
std::vector<Ref<Cell>> to_dec_;
Ref<DataCell> do_inc(Ref<Cell> cell) {
auto cell_hash = cell->get_hash();
if (auto it = info_.find(cell_hash); it != info_.end()) {
CHECK(it->second.diff_refcnt != std::numeric_limits<td::int32>::max());
it->second.diff_refcnt++;
return it->second.cell;
}
if (auto o_info = storage_->get_info(cell_hash)) {
info_.emplace(cell_hash, Info{.db_refcnt = o_info->db_refcnt, .diff_refcnt = 1, .cell = o_info->cell});
return std::move(o_info->cell);
}
CellSlice cs(NoVm{}, std::move(cell));
CellBuilder cb;
cb.store_bits(cs.data(), cs.size());
while (cs.have_refs()) {
auto ref = do_inc(cs.fetch_ref());
cb.store_ref(std::move(ref));
}
auto res = cb.finalize(cs.is_special());
CHECK(res->get_hash() == cell_hash);
info_.emplace(cell_hash, Info{.db_refcnt = 0, .diff_refcnt = 1, .cell = res});
return res;
}
void do_dec(Ref<Cell> cell) {
auto cell_hash = cell->get_hash();
auto it = info_.find(cell_hash);
if (it != info_.end()) {
CHECK(it->second.diff_refcnt != std::numeric_limits<td::int32>::min());
--it->second.diff_refcnt;
} else {
auto info = *storage_->get_info(cell_hash);
it = info_.emplace(cell_hash, Info{.db_refcnt = info.db_refcnt, .diff_refcnt = -1, .cell = info.cell}).first;
}
if (it->second.diff_refcnt + it->second.db_refcnt != 0) {
return;
}
CellSlice cs(NoVm{}, std::move(cell));
while (cs.have_refs()) {
do_dec(cs.fetch_ref());
}
}
};
} // namespace
std::unique_ptr<DynamicBagOfCellsDb> DynamicBagOfCellsDb::create_in_memory(td::KeyValueReader *kv,
CreateInMemoryOptions options) {
if (kv == nullptr) {
LOG_IF(WARNING, options.verbose) << "Create empty in-memory cells database (no key value is given)";
auto storage = CellStorage::build(options, [](auto, auto, auto) { return std::make_pair(0, 0); });
return std::make_unique<InMemoryBagOfCellsDb>(std::move(storage));
}
std::vector<std::string> keys;
keys.emplace_back("");
for (td::uint32 c = 1; c <= 0xff; c++) {
keys.emplace_back(1, static_cast<char>(c));
}
keys.emplace_back(33, static_cast<char>(0xff));
auto parallel_scan_cells = [&](ExtCellCreator &pc_creator, bool use_arena,
auto &&f) -> std::pair<td::int64, td::int64> {
std::atomic<td::int64> cell_count{0};
std::atomic<td::int64> desc_count{0};
parallel_run(
keys.size() - 1,
[&](auto task_id) {
td::int64 local_cell_count = 0;
td::int64 local_desc_count = 0;
CHECK(!DataCell::use_arena);
DataCell::use_arena = use_arena;
kv->for_each_in_range(keys.at(task_id), keys.at(task_id + 1), [&](td::Slice key, td::Slice value) {
if (td::begins_with(key, "desc") && key.size() != 32) {
local_desc_count++;
return td::Status::OK();
}
auto r_res = CellLoader::load(key, value.str(), true, pc_creator);
if (r_res.is_error()) {
LOG(ERROR) << r_res.error() << " at " << td::format::escaped(key);
return td::Status::OK();
}
CHECK(key.size() == 32);
CHECK(key.ubegin()[0] == task_id);
auto res = r_res.move_as_ok();
f(res.refcnt(), res.cell());
local_cell_count++;
return td::Status::OK();
}).ensure();
DataCell::use_arena = false;
cell_count += local_cell_count;
desc_count += local_desc_count;
},
options.extra_threads);
return std::make_pair(cell_count.load(), desc_count.load());
};
auto storage = CellStorage::build(options, parallel_scan_cells);
return std::make_unique<InMemoryBagOfCellsDb>(std::move(storage));
}
} // namespace vm

14
crypto/vm/db/TonDb.h
View file

@ -142,6 +142,20 @@ class TonDbTransactionImpl {
friend bool operator<(td::Slice hash, const SmartContractInfo &info) {
return hash < info.hash;
}
struct Eq {
using is_transparent = void; // Pred to use
bool operator()(const SmartContractInfo &info, const SmartContractInfo &other_info) const { return info.hash == other_info.hash;}
bool operator()(const SmartContractInfo &info, td::Slice hash) const { return info.hash == hash;}
bool operator()(td::Slice hash, const SmartContractInfo &info) const { return info.hash == hash;}
};
struct Hash {
using is_transparent = void; // Pred to use
using transparent_key_equal = Eq;
size_t operator()(td::Slice hash) const { return cell_hash_slice_hash(hash); }
size_t operator()(const SmartContractInfo &info) const { return cell_hash_slice_hash(info.hash);}
};
};
CellHashTable<SmartContractInfo> contracts_;

68
crypto/vm/large-boc-serializer.cpp
View file

@ -33,10 +33,12 @@ class LargeBocSerializer {
public:
using Hash = Cell::Hash;
explicit LargeBocSerializer(std::shared_ptr<CellDbReader> reader, td::CancellationToken cancellation_token = {})
: reader(std::move(reader)), cancellation_token(std::move(cancellation_token)) {
explicit LargeBocSerializer(std::shared_ptr<CellDbReader> reader) : reader(std::move(reader)) {
}
void set_logger(BagOfCellsLogger* logger_ptr) {
logger_ptr_ = logger_ptr;
}
void add_root(Hash root);
td::Status import_cells();
td::Status serialize(td::FileFd& fd, int mode);
@ -44,6 +46,7 @@ class LargeBocSerializer {
private:
std::shared_ptr<CellDbReader> reader;
struct CellInfo {
Cell::Hash hash;
std::array<int, 4> ref_idx;
int idx;
unsigned short serialized_size;
@ -67,7 +70,7 @@ class LargeBocSerializer {
return 4;
}
};
std::map<Hash, CellInfo> cells;
td::NodeHashMap<Hash, CellInfo> cells;
std::vector<std::pair<const Hash, CellInfo>*> cell_list;
struct RootInfo {
RootInfo(Hash hash, int idx) : hash(hash), idx(idx) {
@ -85,10 +88,7 @@ class LargeBocSerializer {
int revisit(int cell_idx, int force = 0);
td::uint64 compute_sizes(int mode, int& r_size, int& o_size);
td::CancellationToken cancellation_token;
td::Timestamp log_speed_at_;
size_t processed_cells_ = 0;
static constexpr double LOG_SPEED_PERIOD = 120.0;
BagOfCellsLogger* logger_ptr_{};
};
void LargeBocSerializer::add_root(Hash root) {
@ -96,16 +96,18 @@ void LargeBocSerializer::add_root(Hash root) {
}
td::Status LargeBocSerializer::import_cells() {
td::Timer timer;
log_speed_at_ = td::Timestamp::in(LOG_SPEED_PERIOD);
processed_cells_ = 0;
if (logger_ptr_) {
logger_ptr_->start_stage("import_cells");
}
for (auto& root : roots) {
TRY_RESULT(idx, import_cell(root.hash));
root.idx = idx;
}
reorder_cells();
CHECK(!cell_list.empty());
LOG(ERROR) << "serializer: import_cells took " << timer.elapsed() << "s, " << cell_count << " cells";
if (logger_ptr_) {
logger_ptr_->finish_stage(PSLICE() << cell_count << " cells");
}
return td::Status::OK();
}
@ -113,14 +115,8 @@ td::Result<int> LargeBocSerializer::import_cell(Hash hash, int depth) {
if (depth > Cell::max_depth) {
return td::Status::Error("error while importing a cell into a bag of cells: cell depth too large");
}
++processed_cells_;
if (processed_cells_ % 1000 == 0) {
TRY_STATUS(cancellation_token.check());
}
if (log_speed_at_.is_in_past()) {
log_speed_at_ += LOG_SPEED_PERIOD;
LOG(WARNING) << "serializer: import_cells " << (double)processed_cells_ / LOG_SPEED_PERIOD << " cells/s";
processed_cells_ = 0;
if (logger_ptr_) {
TRY_STATUS(logger_ptr_->on_cell_processed());
}
auto it = cells.find(hash);
if (it != cells.end()) {
@ -306,7 +302,6 @@ td::uint64 LargeBocSerializer::compute_sizes(int mode, int& r_size, int& o_size)
}
td::Status LargeBocSerializer::serialize(td::FileFd& fd, int mode) {
td::Timer timer;
using Mode = BagOfCells::Mode;
BagOfCells::Info info;
if ((mode & Mode::WithCacheBits) && !(mode & Mode::WithIndex)) {
@ -370,6 +365,9 @@ td::Status LargeBocSerializer::serialize(td::FileFd& fd, int mode) {
DCHECK(writer.position() == info.index_offset);
DCHECK((unsigned)cell_count == cell_list.size());
if (info.has_index) {
if (logger_ptr_) {
logger_ptr_->start_stage("generate_index");
}
std::size_t offs = 0;
for (int i = cell_count - 1; i >= 0; --i) {
const auto& dc_info = cell_list[i]->second;
@ -387,13 +385,20 @@ td::Status LargeBocSerializer::serialize(td::FileFd& fd, int mode) {
fixed_offset = offs * 2 + dc_info.should_cache;
}
store_offset(fixed_offset);
if (logger_ptr_) {
TRY_STATUS(logger_ptr_->on_cell_processed());
}
}
DCHECK(offs == info.data_size);
if (logger_ptr_) {
logger_ptr_->finish_stage("");
}
}
DCHECK(writer.position() == info.data_offset);
size_t keep_position = writer.position();
log_speed_at_ = td::Timestamp::in(LOG_SPEED_PERIOD);
processed_cells_ = 0;
if (logger_ptr_) {
logger_ptr_->start_stage("serialize");
}
for (int i = 0; i < cell_count; ++i) {
auto hash = cell_list[cell_count - 1 - i]->first;
const auto& dc_info = cell_list[cell_count - 1 - i]->second;
@ -412,14 +417,8 @@ td::Status LargeBocSerializer::serialize(td::FileFd& fd, int mode) {
DCHECK(k > i && k < cell_count);
store_ref(k);
}
++processed_cells_;
if (processed_cells_ % 1000 == 0) {
TRY_STATUS(cancellation_token.check());
}
if (log_speed_at_.is_in_past()) {
log_speed_at_ += LOG_SPEED_PERIOD;
LOG(WARNING) << "serializer: serialize " << (double)processed_cells_ / LOG_SPEED_PERIOD << " cells/s";
processed_cells_ = 0;
if (logger_ptr_) {
TRY_STATUS(logger_ptr_->on_cell_processed());
}
}
DCHECK(writer.position() - keep_position == info.data_size);
@ -429,8 +428,9 @@ td::Status LargeBocSerializer::serialize(td::FileFd& fd, int mode) {
}
DCHECK(writer.empty());
TRY_STATUS(writer.finalize());
LOG(ERROR) << "serializer: serialize took " << timer.elapsed() << "s, " << cell_count << " cells, "
<< writer.position() << " bytes";
if (logger_ptr_) {
logger_ptr_->finish_stage(PSLICE() << cell_count << " cells, " << writer.position() << " bytes");
}
return td::Status::OK();
}
} // namespace
@ -439,7 +439,9 @@ td::Status std_boc_serialize_to_file_large(std::shared_ptr<CellDbReader> reader,
int mode, td::CancellationToken cancellation_token) {
td::Timer timer;
CHECK(reader != nullptr)
LargeBocSerializer serializer(reader, std::move(cancellation_token));
LargeBocSerializer serializer(reader);
BagOfCellsLogger logger(std::move(cancellation_token));
serializer.set_logger(&logger);
serializer.add_root(root_hash);
TRY_STATUS(serializer.import_cells());
TRY_STATUS(serializer.serialize(fd, mode));

2
tddb/CMakeLists.txt
View file

@ -59,7 +59,7 @@ target_link_libraries(tddb PUBLIC tdutils tdactor tddb_utils)
if (TDDB_USE_ROCKSDB)
target_sources(tddb PRIVATE ${TDDB_ROCKSDB_SOURCE})
target_compile_definitions(tddb PUBLIC -DTDDB_USE_ROCKSDB)
target_link_libraries(tddb PRIVATE rocksdb)
target_link_libraries(tddb PUBLIC rocksdb)
target_include_directories(tddb PRIVATE $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/../third-party/rocksdb/include>)
endif()

3
tddb/td/db/KeyValue.h
View file

@ -32,6 +32,9 @@ class KeyValueReader {
virtual Status for_each(std::function<Status(Slice, Slice)> f) {
return Status::Error("for_each is not supported");
}
virtual Status for_each_in_range (Slice begin, Slice end, std::function<Status(Slice, Slice)> f) {
return td::Status::Error("foreach_range is not supported");
}
};
class PrefixedKeyValueReader : public KeyValueReader {

18
tddb/td/db/MemoryKeyValue.cpp
View file

@ -29,6 +29,24 @@ Result<MemoryKeyValue::GetStatus> MemoryKeyValue::get(Slice key, std::string &va
value = it->second;
return GetStatus::Ok;
}
Status MemoryKeyValue::for_each(std::function<Status(Slice, Slice)> f) {
for (auto &it : map_) {
TRY_STATUS(f(it.first, it.second));
}
return Status::OK();
}
Status MemoryKeyValue::for_each_in_range(Slice begin, Slice end, std::function<Status(Slice, Slice)> f) {
for (auto it = map_.lower_bound(begin); it != map_.end(); it++) {
if (it->first < end) {
TRY_STATUS(f(it->first, it->second));
} else {
break;
}
}
return Status::OK();
}
Status MemoryKeyValue::set(Slice key, Slice value) {
map_[key.str()] = value.str();
return Status::OK();

2
tddb/td/db/MemoryKeyValue.h
View file

@ -25,6 +25,8 @@ namespace td {
class MemoryKeyValue : public KeyValue {
public:
Result<GetStatus> get(Slice key, std::string &value) override;
Status for_each(std::function<Status(Slice, Slice)> f) override;
Status for_each_in_range(Slice begin, Slice end, std::function<Status(Slice, Slice)> f) override;
Status set(Slice key, Slice value) override;
Status erase(Slice key) override;
Result<size_t> count(Slice prefix) override;

34
tddb/td/db/RocksDb.cpp
View file

@ -65,12 +65,16 @@ Result<RocksDb> RocksDb::open(std::string path, RocksDbOptions options) {
rocksdb::Options db_options;
static auto default_cache = rocksdb::NewLRUCache(1 << 30);
if (options.block_cache == nullptr) {
if (!options.no_block_cache && options.block_cache == nullptr) {
options.block_cache = default_cache;
}
rocksdb::BlockBasedTableOptions table_options;
table_options.block_cache = options.block_cache;
if (options.no_block_cache) {
table_options.no_block_cache = true;
} else {
table_options.block_cache = options.block_cache;
}
db_options.table_factory.reset(rocksdb::NewBlockBasedTableFactory(table_options));
db_options.use_direct_reads = options.use_direct_reads;
@ -212,6 +216,32 @@ Status RocksDb::for_each(std::function<Status(Slice, Slice)> f) {
return Status::OK();
}
Status RocksDb::for_each_in_range(Slice begin, Slice end, std::function<Status(Slice, Slice)> f) {
rocksdb::ReadOptions options;
options.snapshot = snapshot_.get();
std::unique_ptr<rocksdb::Iterator> iterator;
if (snapshot_ || !transaction_) {
iterator.reset(db_->NewIterator(options));
} else {
iterator.reset(transaction_->GetIterator(options));
}
auto comparator = rocksdb::BytewiseComparator();
iterator->Seek(to_rocksdb(begin));
for (; iterator->Valid(); iterator->Next()) {
auto key = from_rocksdb(iterator->key());
if (comparator->Compare(to_rocksdb(key), to_rocksdb(end)) >= 0) {
break;
}
auto value = from_rocksdb(iterator->value());
TRY_STATUS(f(key, value));
}
if (!iterator->status().ok()) {
return from_rocksdb(iterator->status());
}
return td::Status::OK();
}
Status RocksDb::begin_write_batch() {
CHECK(!transaction_);
write_batch_ = std::make_unique<rocksdb::WriteBatch>();

5
tddb/td/db/RocksDb.h
View file

@ -23,6 +23,7 @@
#endif
#include "td/db/KeyValue.h"
#include "td/utils/Span.h"
#include "td/utils/Status.h"
#include "td/utils/optional.h"
@ -32,6 +33,8 @@
#include <mutex>
#include <set>
#include <functional>
namespace rocksdb {
class Cache;
class OptimisticTransactionDB;
@ -59,6 +62,7 @@ struct RocksDbOptions {
std::shared_ptr<rocksdb::Cache> block_cache; // Default - one 1G cache for all RocksDb
std::shared_ptr<RocksDbSnapshotStatistics> snapshot_statistics = nullptr;
bool use_direct_reads = false;
bool no_block_cache = false;
};
class RocksDb : public KeyValue {
@ -72,6 +76,7 @@ class RocksDb : public KeyValue {
Status erase(Slice key) override;
Result<size_t> count(Slice prefix) override;
Status for_each(std::function<Status(Slice, Slice)> f) override;
Status for_each_in_range (Slice begin, Slice end, std::function<Status(Slice, Slice)> f) override;
Status begin_write_batch() override;
Status commit_write_batch() override;

1
tdutils/CMakeLists.txt
View file

@ -356,6 +356,7 @@ if (LZ4_FOUND)
message(STATUS "Found LZ4 ${LZ4_LIBRARIES} ${LZ4_INCLUDE_DIRS}")
target_link_libraries(tdutils PRIVATE ${LZ4_LIBRARIES})
target_include_directories(tdutils SYSTEM PRIVATE ${LZ4_INCLUDE_DIRS})
target_link_directories(tdutils PUBLIC ${LZ4_LIBRARY_DIRS})
endif()
if (ABSL_FOUND)

5
tdutils/td/utils/HashMap.h
View file

@ -22,6 +22,7 @@
#if TD_HAVE_ABSL
#include <absl/container/flat_hash_map.h>
#include <absl/container/node_hash_map.h>
#else
#include <unordered_map>
#endif
@ -31,9 +32,13 @@ namespace td {
#if TD_HAVE_ABSL
template <class Key, class Value, class H = Hash<Key>>
using HashMap = absl::flat_hash_map<Key, Value, H>;
template <class Key, class Value, class H = Hash<Key>, class E = std::equal_to<>>
using NodeHashMap = absl::node_hash_map<Key, Value, H, E>;
#else
template <class Key, class Value, class H = Hash<Key>>
using HashMap = std::unordered_map<Key, Value, H>;
template <class Key, class Value, class H = Hash<Key>>
using NodeHashMap = std::unordered_map<Key, Value, H>;
#endif
} // namespace td

13
tdutils/td/utils/HashSet.h
View file

@ -22,6 +22,7 @@
#if TD_HAVE_ABSL
#include <absl/container/flat_hash_set.h>
#include <absl/container/node_hash_set.h>
#else
#include <unordered_set>
#endif
@ -29,11 +30,15 @@
namespace td {
#if TD_HAVE_ABSL
template <class Key, class H = Hash<Key>>
using HashSet = absl::flat_hash_set<Key, H>;
template <class Key, class H = Hash<Key>, class E = std::equal_to<>>
using HashSet = absl::flat_hash_set<Key, H, E>;
template <class Key, class H = Hash<Key>, class E = std::equal_to<>>
using NodeHashSet = absl::node_hash_set<Key, H, E>;
#else
template <class Key, class H = Hash<Key>>
using HashSet = std::unordered_set<Key, H>;
template <class Key, class H = Hash<Key>, class E = std::equal_to<>>
using HashSet = std::unordered_set<Key, H, E>;
template <class Key, class H = Hash<Key>, class E = std::equal_to<>>
using NodeHashSet = HashSet<Key, H, E>;
#endif
} // namespace td

23
tdutils/td/utils/port/Stat.cpp
View file

@ -58,8 +58,7 @@
#define PSAPI_VERSION 1
#endif
#include <psapi.h>
#pragma comment( lib, "psapi.lib" )
#pragma comment(lib, "psapi.lib")
#endif
@ -413,7 +412,7 @@ Result<CpuStat> cpu_stat() {
#endif
}
Result<uint64> get_total_ram() {
Result<TotalMemStat> get_total_mem_stat() {
#if TD_LINUX
TRY_RESULT(fd, FileFd::open("/proc/meminfo", FileFd::Read));
SCOPE_EXIT {
@ -425,8 +424,10 @@ Result<uint64> get_total_ram() {
if (size >= TMEM_SIZE - 1) {
return Status::Error("Failed for read /proc/meminfo");
}
TotalMemStat stat;
mem[size] = 0;
const char* s = mem;
const char *s = mem;
size_t got = 0;
while (*s) {
const char *name_begin = s;
while (*s != 0 && *s != '\n') {
@ -437,18 +438,28 @@ Result<uint64> get_total_ram() {
name_end++;
}
Slice name(name_begin, name_end);
td::uint64 *dest = nullptr;
if (name == "MemTotal") {
dest = &stat.total_ram;
} else if (name == "MemAvailable") {
dest = &stat.available_ram;
}
if (dest != nullptr) {
Slice value(name_end, s);
if (!value.empty() && value[0] == ':') {
value.remove_prefix(1);
}
value = trim(value);
value = split(value).first;
TRY_RESULT_PREFIX(mem, to_integer_safe<uint64>(value), "Invalid value of MemTotal");
TRY_RESULT_PREFIX(mem, to_integer_safe<uint64>(value), PSLICE() << "Invalid value of " << name);
if (mem >= 1ULL << (64 - 10)) {
return Status::Error("Invalid value of MemTotal");
}
return mem * 1024;
*dest = mem * 1024;
got++;
if (got == 2) {
return stat;
}
}
if (*s == 0) {
break;

6
tdutils/td/utils/port/Stat.h
View file

@ -64,6 +64,10 @@ Status update_atime(CSlice path) TD_WARN_UNUSED_RESULT;
#endif
Result<uint64> get_total_ram() TD_WARN_UNUSED_RESULT;
struct TotalMemStat {
uint64 total_ram;
uint64 available_ram;
};
Result<TotalMemStat> get_total_mem_stat() TD_WARN_UNUSED_RESULT;
} // namespace td

46
validator-engine/validator-engine.cpp
View file

@ -1432,7 +1432,11 @@ td::Status ValidatorEngine::load_global_config() {
if (!session_logs_file_.empty()) {
validator_options_.write().set_session_logs_file(session_logs_file_);
}
if (celldb_in_memory_) {
celldb_compress_depth_ = 0;
}
validator_options_.write().set_celldb_compress_depth(celldb_compress_depth_);
validator_options_.write().set_celldb_in_memory(celldb_in_memory_);
validator_options_.write().set_max_open_archive_files(max_open_archive_files_);
validator_options_.write().set_archive_preload_period(archive_preload_period_);
validator_options_.write().set_disable_rocksdb_stats(disable_rocksdb_stats_);
@ -1471,11 +1475,11 @@ td::Status ValidatorEngine::load_global_config() {
}
validator_options_.write().set_hardforks(std::move(h));
auto r_total_ram = td::get_total_ram();
if (r_total_ram.is_error()) {
LOG(ERROR) << "Failed to get total RAM size: " << r_total_ram.move_as_error();
auto r_total_mem_stat = td::get_total_mem_stat();
if (r_total_mem_stat.is_error()) {
LOG(ERROR) << "Failed to get total RAM size: " << r_total_mem_stat.move_as_error();
} else {
td::uint64 total_ram = r_total_ram.move_as_ok();
td::uint64 total_ram = r_total_mem_stat.ok().total_ram;
LOG(WARNING) << "Total RAM = " << td::format::as_size(total_ram);
if (total_ram >= (90ULL << 30)) {
fast_state_serializer_enabled_ = true;
@ -3778,9 +3782,8 @@ void ValidatorEngine::run_control_query(ton::ton_api::engine_validator_addCustom
});
}
void ValidatorEngine::run_control_query(ton::ton_api::engine_validator_delCustomOverlay &query,
td::BufferSlice data, ton::PublicKeyHash src, td::uint32 perm,
td::Promise<td::BufferSlice> promise) {
void ValidatorEngine::run_control_query(ton::ton_api::engine_validator_delCustomOverlay &query, td::BufferSlice data,
ton::PublicKeyHash src, td::uint32 perm, td::Promise<td::BufferSlice> promise) {
if (!(perm & ValidatorEnginePermissions::vep_modify)) {
promise.set_value(create_control_query_error(td::Status::Error(ton::ErrorCode::error, "not authorized")));
return;
@ -3820,8 +3823,8 @@ void ValidatorEngine::run_control_query(ton::ton_api::engine_validator_showCusto
return;
}
promise.set_value(ton::serialize_tl_object<ton::ton_api::engine_validator_customOverlaysConfig>(
custom_overlays_config_, true));
promise.set_value(
ton::serialize_tl_object<ton::ton_api::engine_validator_customOverlaysConfig>(custom_overlays_config_, true));
}
void ValidatorEngine::run_control_query(ton::ton_api::engine_validator_setStateSerializerEnabled &query,
@ -4248,7 +4251,8 @@ int main(int argc, char *argv[]) {
return td::Status::OK();
});
p.add_checked_option(
'\0', "max-archive-fd", "limit for a number of open file descriptirs in archive manager. 0 is unlimited (default)",
'\0', "max-archive-fd",
"limit for a number of open file descriptirs in archive manager. 0 is unlimited (default)",
[&](td::Slice s) -> td::Status {
TRY_RESULT(v, td::to_integer_safe<size_t>(s));
acts.push_back([&x, v]() { td::actor::send_closure(x, &ValidatorEngine::set_max_open_archive_files, v); });
@ -4283,13 +4287,23 @@ int main(int argc, char *argv[]) {
acts.push_back([&x, v]() { td::actor::send_closure(x, &ValidatorEngine::set_celldb_cache_size, v); });
return td::Status::OK();
});
p.add_option('\0', "celldb-direct-io",
"enable direct I/O mode for RocksDb in CellDb (doesn't apply when celldb cache is < 30G)", [&]() {
acts.push_back([&x]() { td::actor::send_closure(x, &ValidatorEngine::set_celldb_direct_io, true); });
});
p.add_option('\0', "celldb-preload-all",
"preload all cells from CellDb on startup (recommended to use with big enough celldb-cache-size and "
"celldb-direct-io)",
[&]() {
acts.push_back([&x]() { td::actor::send_closure(x, &ValidatorEngine::set_celldb_preload_all, true); });
});
p.add_option(
'\0', "celldb-direct-io", "enable direct I/O mode for RocksDb in CellDb (doesn't apply when celldb cache is < 30G)",
[&]() { acts.push_back([&x]() { td::actor::send_closure(x, &ValidatorEngine::set_celldb_direct_io, true); }); });
p.add_option(
'\0', "celldb-preload-all",
"preload all cells from CellDb on startup (recommended to use with big enough celldb-cache-size and celldb-direct-io)",
[&]() { acts.push_back([&x]() { td::actor::send_closure(x, &ValidatorEngine::set_celldb_preload_all, true); }); });
'\0', "celldb-in-memory",
"store all cells in-memory, much faster but requires a lot of RAM. RocksDb is still used as persistent storage",
[&]() {
acts.push_back([&x]() { td::actor::send_closure(x, &ValidatorEngine::set_celldb_in_memory, true); });
});
p.add_checked_option(
'\0', "catchain-max-block-delay", "delay before creating a new catchain block, in seconds (default: 0.4)",
[&](td::Slice s) -> td::Status {

4
validator-engine/validator-engine.hpp
View file

@ -214,6 +214,7 @@ class ValidatorEngine : public td::actor::Actor {
td::optional<td::uint64> celldb_cache_size_ = 1LL << 30;
bool celldb_direct_io_ = false;
bool celldb_preload_all_ = false;
bool celldb_in_memory_ = false;
td::optional<double> catchain_max_block_delay_, catchain_max_block_delay_slow_;
bool read_config_ = false;
bool started_keyring_ = false;
@ -297,6 +298,9 @@ class ValidatorEngine : public td::actor::Actor {
void set_celldb_preload_all(bool value) {
celldb_preload_all_ = value;
}
void set_celldb_in_memory(bool value) {
celldb_in_memory_ = value;
}
void set_catchain_max_block_delay(double value) {
catchain_max_block_delay_ = value;
}

204
validator/db/celldb.cpp
View file

@ -17,10 +17,12 @@
Copyright 2017-2020 Telegram Systems LLP
*/
#include "celldb.hpp"
#include "files-async.hpp"
#include "rootdb.hpp"
#include "td/db/RocksDb.h"
#include "td/utils/filesystem.h"
#include "rocksdb/utilities/optimistic_transaction_db.h"
#include "ton/ton-tl.hpp"
#include "ton/ton-io.hpp"
@ -29,7 +31,6 @@
namespace ton {
namespace validator {
class CellDbAsyncExecutor : public vm::DynamicBagOfCellsDb::AsyncExecutor {
public:
explicit CellDbAsyncExecutor(td::actor::ActorId<CellDbBase> cell_db) : cell_db_(std::move(cell_db)) {
@ -38,11 +39,13 @@ class CellDbAsyncExecutor : public vm::DynamicBagOfCellsDb::AsyncExecutor {
void execute_async(std::function<void()> f) override {
class Runner : public td::actor::Actor {
public:
explicit Runner(std::function<void()> f) : f_(std::move(f)) {}
explicit Runner(std::function<void()> f) : f_(std::move(f)) {
}
void start_up() override {
f_();
stop();
}
private:
std::function<void()> f_;
};
@ -52,6 +55,7 @@ class CellDbAsyncExecutor : public vm::DynamicBagOfCellsDb::AsyncExecutor {
void execute_sync(std::function<void()> f) override {
td::actor::send_closure(cell_db_, &CellDbBase::execute_sync, std::move(f));
}
private:
td::actor::ActorId<CellDbBase> cell_db_;
};
@ -97,13 +101,30 @@ void CellDbIn::start_up() {
LOG(WARNING) << "Set CellDb block cache size to " << td::format::as_size(opts_->get_celldb_cache_size().value());
}
db_options.use_direct_reads = opts_->get_celldb_direct_io();
cell_db_ = std::make_shared<td::RocksDb>(td::RocksDb::open(path_, std::move(db_options)).move_as_ok());
if (opts_->get_celldb_in_memory()) {
td::RocksDbOptions read_db_options;
read_db_options.use_direct_reads = true;
read_db_options.no_block_cache = true;
read_db_options.block_cache = {};
LOG(WARNING) << "Loading all cells in memory (because of --celldb-in-memory)";
td::Timer timer;
auto read_cell_db =
std::make_shared<td::RocksDb>(td::RocksDb::open(path_, std::move(read_db_options)).move_as_ok());
boc_ = vm::DynamicBagOfCellsDb::create_in_memory(read_cell_db.get(), {});
in_memory_load_time_ = timer.elapsed();
td::actor::send_closure(parent_, &CellDb::set_in_memory_boc, boc_);
}
boc_ = vm::DynamicBagOfCellsDb::create();
boc_->set_celldb_compress_depth(opts_->get_celldb_compress_depth());
boc_->set_loader(std::make_unique<vm::CellLoader>(cell_db_->snapshot(), on_load_callback_)).ensure();
td::actor::send_closure(parent_, &CellDb::update_snapshot, cell_db_->snapshot());
auto rocks_db = std::make_shared<td::RocksDb>(td::RocksDb::open(path_, std::move(db_options)).move_as_ok());
rocks_db_ = rocks_db->raw_db();
cell_db_ = std::move(rocks_db);
if (!opts_->get_celldb_in_memory()) {
boc_ = vm::DynamicBagOfCellsDb::create();
boc_->set_celldb_compress_depth(opts_->get_celldb_compress_depth());
boc_->set_loader(std::make_unique<vm::CellLoader>(cell_db_->snapshot(), on_load_callback_)).ensure();
td::actor::send_closure(parent_, &CellDb::update_snapshot, cell_db_->snapshot());
}
alarm_timestamp() = td::Timestamp::in(10.0);
@ -117,27 +138,34 @@ void CellDbIn::start_up() {
if (opts_->get_celldb_preload_all()) {
// Iterate whole DB in a separate thread
delay_action([snapshot = cell_db_->snapshot()]() {
LOG(WARNING) << "CellDb: pre-loading all keys";
td::uint64 total = 0;
td::Timer timer;
auto S = snapshot->for_each([&](td::Slice, td::Slice) {
++total;
if (total % 1000000 == 0) {
LOG(INFO) << "CellDb: iterated " << total << " keys";
}
return td::Status::OK();
});
if (S.is_error()) {
LOG(ERROR) << "CellDb: pre-load failed: " << S.move_as_error();
} else {
LOG(WARNING) << "CellDb: iterated " << total << " keys in " << timer.elapsed() << "s";
}
}, td::Timestamp::now());
delay_action(
[snapshot = cell_db_->snapshot()]() {
LOG(WARNING) << "CellDb: pre-loading all keys";
td::uint64 total = 0;
td::Timer timer;
auto S = snapshot->for_each([&](td::Slice, td::Slice) {
++total;
if (total % 1000000 == 0) {
LOG(INFO) << "CellDb: iterated " << total << " keys";
}
return td::Status::OK();
});
if (S.is_error()) {
LOG(ERROR) << "CellDb: pre-load failed: " << S.move_as_error();
} else {
LOG(WARNING) << "CellDb: iterated " << total << " keys in " << timer.elapsed() << "s";
}
},
td::Timestamp::now());
}
}
void CellDbIn::load_cell(RootHash hash, td::Promise<td::Ref<vm::DataCell>> promise) {
if (opts_->get_celldb_in_memory()) {
auto result = boc_->load_root(hash.as_slice());
async_apply("load_cell_result", std::move(promise), std::move(result));
return;
}
boc_->load_cell_async(hash.as_slice(), async_executor, std::move(promise));
}
@ -182,8 +210,10 @@ void CellDbIn::store_cell(BlockIdExt block_id, td::Ref<vm::Cell> cell, td::Promi
set_block(key_hash, std::move(D));
cell_db_->commit_write_batch().ensure();
boc_->set_loader(std::make_unique<vm::CellLoader>(cell_db_->snapshot(), on_load_callback_)).ensure();
td::actor::send_closure(parent_, &CellDb::update_snapshot, cell_db_->snapshot());
if (!opts_->get_celldb_in_memory()) {
boc_->set_loader(std::make_unique<vm::CellLoader>(cell_db_->snapshot(), on_load_callback_)).ensure();
td::actor::send_closure(parent_, &CellDb::update_snapshot, cell_db_->snapshot());
}
promise.set_result(boc_->load_cell(cell->get_hash().as_slice()));
if (!opts_->get_disable_rocksdb_stats()) {
@ -200,12 +230,50 @@ void CellDbIn::get_last_deleted_mc_state(td::Promise<BlockSeqno> promise) {
promise.set_result(last_deleted_mc_state_);
}
std::vector<std::pair<std::string, std::string>> CellDbIn::prepare_stats() {
TD_PERF_COUNTER(celldb_prepare_stats);
auto r_boc_stats = boc_->get_stats();
if (r_boc_stats.is_ok()) {
cell_db_statistics_.boc_stats_ = r_boc_stats.move_as_ok();
}
cell_db_statistics_.in_memory_load_time_ = in_memory_load_time_;
auto stats = cell_db_statistics_.prepare_stats();
auto add_stat = [&](const auto& key, const auto& value) { stats.emplace_back(key, PSTRING() << value); };
add_stat("started", "true");
auto r_mem_stat = td::mem_stat();
auto r_total_mem_stat = td::get_total_mem_stat();
td::uint64 celldb_size = 0;
bool ok_celldb_size = rocks_db_->GetIntProperty("rocksdb.total-sst-files-size", &celldb_size);
if (celldb_size > 0 && r_mem_stat.is_ok() && r_total_mem_stat.is_ok() && ok_celldb_size) {
auto mem_stat = r_mem_stat.move_as_ok();
auto total_mem_stat = r_total_mem_stat.move_as_ok();
add_stat("rss", td::format::as_size(mem_stat.resident_size_));
add_stat("available_ram", td::format::as_size(total_mem_stat.available_ram));
add_stat("total_ram", td::format::as_size(total_mem_stat.total_ram));
add_stat("actual_ram_to_celldb_ratio", double(total_mem_stat.available_ram) / double(celldb_size));
add_stat("if_restarted_ram_to_celldb_ratio",
double(total_mem_stat.available_ram + mem_stat.resident_size_ - 10 * (td::uint64(1) << 30)) /
double(celldb_size));
add_stat("max_possible_ram_to_celldb_ratio", double(total_mem_stat.total_ram) / double(celldb_size));
}
return stats;
// do not clear statistics, it is needed for flush_db_stats
}
void CellDbIn::flush_db_stats() {
if (opts_->get_disable_rocksdb_stats()) {
return;
}
auto stats = td::RocksDb::statistics_to_string(statistics_) + snapshot_statistics_->to_string() +
cell_db_statistics_.to_string();
auto celldb_stats = prepare_stats();
td::StringBuilder ss;
for (auto& [key, value] : celldb_stats) {
ss << "ton.celldb." << key << " " << value << "\n";
}
auto stats =
td::RocksDb::statistics_to_string(statistics_) + snapshot_statistics_->to_string() + ss.as_cslice().str();
auto to_file_r =
td::FileFd::open(path_ + "/db_stats.txt", td::FileFd::Truncate | td::FileFd::Create | td::FileFd::Write, 0644);
if (to_file_r.is_error()) {
@ -287,7 +355,9 @@ void CellDbIn::gc_cont(BlockHandle handle) {
void CellDbIn::gc_cont2(BlockHandle handle) {
TD_PERF_COUNTER(celldb_gc_cell);
td::PerfWarningTimer timer{"gccell", 0.1};
td::PerfWarningTimer timer_all{"gccell_all", 0.05};
td::PerfWarningTimer timer_get_keys{"gccell_get_keys", 0.05};
auto key_hash = get_key_hash(handle->id());
auto FR = get_block(key_hash);
FR.ensure();
@ -306,22 +376,41 @@ void CellDbIn::gc_cont2(BlockHandle handle) {
P.prev = P.next;
N.next = N.prev;
}
timer_get_keys.reset();
td::PerfWarningTimer timer_boc{"gccell_boc", 0.05};
auto cell = boc_->load_cell(F.root_hash.as_slice()).move_as_ok();
boc_->dec(cell);
boc_->prepare_commit().ensure();
vm::CellStorer stor{*cell_db_};
timer_boc.reset();
td::PerfWarningTimer timer_write_batch{"gccell_write_batch", 0.05};
cell_db_->begin_write_batch().ensure();
boc_->commit(stor).ensure();
cell_db_->erase(get_key(key_hash)).ensure();
set_block(F.prev, std::move(P));
set_block(F.next, std::move(N));
cell_db_->commit_write_batch().ensure();
alarm_timestamp() = td::Timestamp::now();
timer_write_batch.reset();
boc_->set_loader(std::make_unique<vm::CellLoader>(cell_db_->snapshot(), on_load_callback_)).ensure();
td::actor::send_closure(parent_, &CellDb::update_snapshot, cell_db_->snapshot());
td::PerfWarningTimer timer_free_cells{"gccell_free_cells", 0.05};
auto before = td::ref_get_delete_count();
cell = {};
auto after = td::ref_get_delete_count();
if (timer_free_cells.elapsed() > 0.04) {
LOG(ERROR) << "deleted " << after - before << " cells";
}
timer_free_cells.reset();
td::PerfWarningTimer timer_finish{"gccell_finish", 0.05};
if (!opts_->get_celldb_in_memory()) {
boc_->set_loader(std::make_unique<vm::CellLoader>(cell_db_->snapshot(), on_load_callback_)).ensure();
td::actor::send_closure(parent_, &CellDb::update_snapshot, cell_db_->snapshot());
}
DCHECK(get_block(key_hash).is_error());
if (!opts_->get_disable_rocksdb_stats()) {
@ -331,6 +420,8 @@ void CellDbIn::gc_cont2(BlockHandle handle) {
last_deleted_mc_state_ = handle->id().seqno();
}
LOG(DEBUG) << "Deleted state " << handle->id().to_str();
timer_finish.reset();
timer_all.reset();
}
void CellDbIn::skip_gc() {
@ -403,7 +494,7 @@ void CellDbIn::migrate_cells() {
boc_->set_loader(std::make_unique<vm::CellLoader>(*loader)).ensure();
cell_db_->begin_write_batch().ensure();
td::uint32 checked = 0, migrated = 0;
for (auto it = cells_to_migrate_.begin(); it != cells_to_migrate_.end() && checked < 128; ) {
for (auto it = cells_to_migrate_.begin(); it != cells_to_migrate_.end() && checked < 128;) {
++checked;
td::Bits256 hash = *it;
it = cells_to_migrate_.erase(it);
@ -440,7 +531,32 @@ void CellDbIn::migrate_cells() {
}
}
void CellDb::prepare_stats(td::Promise<std::vector<std::pair<std::string, std::string>>> promise) {
promise.set_value(decltype(prepared_stats_)(prepared_stats_));
}
void CellDb::update_stats(td::Result<std::vector<std::pair<std::string, std::string>>> r_stats) {
if (r_stats.is_error()) {
LOG(ERROR) << "error updating stats: " << r_stats.error();
} else {
prepared_stats_ = r_stats.move_as_ok();
}
alarm_timestamp() = td::Timestamp::in(2.0);
}
void CellDb::alarm() {
send_closure(cell_db_, &CellDbIn::prepare_stats, td::promise_send_closure(actor_id(this), &CellDb::update_stats));
}
void CellDb::load_cell(RootHash hash, td::Promise<td::Ref<vm::DataCell>> promise) {
if (in_memory_boc_) {
auto result = in_memory_boc_->load_root_thread_safe(hash.as_slice());
if (result.is_ok()) {
return async_apply("load_cell_result", std::move(promise), std::move(result));
} else {
LOG(ERROR) << "load_root_thread_safe failed - this is suspicious";
}
}
if (!started_) {
td::actor::send_closure(cell_db_, &CellDbIn::load_cell, hash, std::move(promise));
} else {
@ -498,12 +614,24 @@ td::BufferSlice CellDbIn::DbEntry::release() {
return create_serialize_tl_object<ton_api::db_celldb_value>(create_tl_block_id(block_id), prev, next, root_hash);
}
std::string CellDbIn::CellDbStatistics::to_string() {
td::StringBuilder ss;
ss << "ton.celldb.store_cell.micros " << store_cell_time_.to_string() << "\n";
ss << "ton.celldb.gc_cell.micros " << gc_cell_time_.to_string() << "\n";
ss << "ton.celldb.total_time.micros : " << (td::Timestamp::now().at() - stats_start_time_.at()) * 1e6 << "\n";
return ss.as_cslice().str();
std::vector<std::pair<std::string, std::string>> CellDbIn::CellDbStatistics::prepare_stats() {
std::vector<std::pair<std::string, std::string>> stats;
stats.emplace_back("store_cell.micros", PSTRING() << store_cell_time_.to_string());
stats.emplace_back("gc_cell.micros", PSTRING() << gc_cell_time_.to_string());
stats.emplace_back("total_time.micros", PSTRING() << (td::Timestamp::now().at() - stats_start_time_.at()) * 1e6);
stats.emplace_back("in_memory", PSTRING() << bool(in_memory_load_time_));
if (in_memory_load_time_) {
stats.emplace_back("in_memory_load_time", PSTRING() << in_memory_load_time_.value());
}
if (boc_stats_) {
stats.emplace_back("cells_count", PSTRING() << boc_stats_->cells_total_count);
stats.emplace_back("cells_size", PSTRING() << boc_stats_->cells_total_size);
stats.emplace_back("roots_count", PSTRING() << boc_stats_->roots_total_count);
for (auto& [key, value] : boc_stats_->custom_stats) {
stats.emplace_back(key, value);
}
}
return stats;
}
} // namespace validator

30
validator/db/celldb.hpp
View file

@ -31,7 +31,8 @@
namespace rocksdb {
class Statistics;
}
class DB;
} // namespace rocksdb
namespace ton {
@ -58,6 +59,7 @@ class CellDbIn : public CellDbBase {
public:
using KeyHash = td::Bits256;
std::vector<std::pair<std::string, std::string>> prepare_stats();
void load_cell(RootHash hash, td::Promise<td::Ref<vm::DataCell>> promise);
void store_cell(BlockIdExt block_id, td::Ref<vm::Cell> cell, td::Promise<td::Ref<vm::DataCell>> promise);
void get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> promise);
@ -111,13 +113,15 @@ class CellDbIn : public CellDbBase {
std::string path_;
td::Ref<ValidatorManagerOptions> opts_;
std::unique_ptr<vm::DynamicBagOfCellsDb> boc_;
std::shared_ptr<vm::DynamicBagOfCellsDb> boc_;
std::shared_ptr<vm::KeyValue> cell_db_;
std::shared_ptr<rocksdb::DB> rocks_db_;
std::function<void(const vm::CellLoader::LoadResult&)> on_load_callback_;
std::set<td::Bits256> cells_to_migrate_;
td::Timestamp migrate_after_ = td::Timestamp::never();
bool migration_active_ = false;
std::optional<double> in_memory_load_time_;
struct MigrationStats {
td::Timer start_;
@ -133,8 +137,10 @@ class CellDbIn : public CellDbBase {
PercentileStats store_cell_time_;
PercentileStats gc_cell_time_;
td::Timestamp stats_start_time_ = td::Timestamp::now();
std::optional<double> in_memory_load_time_;
std::optional<vm::DynamicBagOfCellsDb::Stats> boc_stats_;
std::string to_string();
std::vector<std::pair<std::string, std::string>> prepare_stats();
void clear() {
*this = CellDbStatistics{};
}
@ -162,12 +168,25 @@ class CellDbIn : public CellDbBase {
class CellDb : public CellDbBase {
public:
void prepare_stats(td::Promise<std::vector<std::pair<std::string, std::string>>> promise);
void load_cell(RootHash hash, td::Promise<td::Ref<vm::DataCell>> promise);
void store_cell(BlockIdExt block_id, td::Ref<vm::Cell> cell, td::Promise<td::Ref<vm::DataCell>> promise);
void update_snapshot(std::unique_ptr<td::KeyValueReader> snapshot) {
CHECK(!opts_->get_celldb_in_memory());
if (!started_) {
alarm();
}
started_ = true;
boc_->set_loader(std::make_unique<vm::CellLoader>(std::move(snapshot), on_load_callback_)).ensure();
}
void set_in_memory_boc(std::shared_ptr<const vm::DynamicBagOfCellsDb> in_memory_boc) {
CHECK(opts_->get_celldb_in_memory());
if (!started_) {
alarm();
}
started_ = true;
in_memory_boc_ = std::move(in_memory_boc);
}
void get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> promise);
void get_last_deleted_mc_state(td::Promise<BlockSeqno> promise);
@ -185,9 +204,14 @@ class CellDb : public CellDbBase {
td::actor::ActorOwn<CellDbIn> cell_db_;
std::unique_ptr<vm::DynamicBagOfCellsDb> boc_;
std::shared_ptr<const vm::DynamicBagOfCellsDb> in_memory_boc_;
bool started_ = false;
std::vector<std::pair<std::string, std::string>> prepared_stats_{{"started", "false"}};
std::function<void(const vm::CellLoader::LoadResult&)> on_load_callback_;
void update_stats(td::Result<std::vector<std::pair<std::string, std::string>>> stats);
void alarm() override;
};
} // namespace validator

1
validator/db/rootdb.cpp
View file

@ -439,6 +439,7 @@ void RootDb::allow_block_gc(BlockIdExt block_id, td::Promise<bool> promise) {
void RootDb::prepare_stats(td::Promise<std::vector<std::pair<std::string, std::string>>> promise) {
auto merger = StatsMerger::create(std::move(promise));
td::actor::send_closure(cell_db_, &CellDb::prepare_stats, merger.make_promise("celldb."));
}
void RootDb::truncate(BlockSeqno seqno, ConstBlockHandle handle, td::Promise<td::Unit> promise) {

25
validator/manager.cpp
View file

@ -2138,8 +2138,8 @@ void ValidatorManagerImpl::update_shards() {
new_next_validator_groups_.emplace(val_group_id, std::move(it->second));
} else {
new_next_validator_groups_.emplace(
val_group_id,
ValidatorGroupEntry{create_validator_group(val_group_id, shard, val_set, key_seqno, opts, started_), shard});
val_group_id, ValidatorGroupEntry{
create_validator_group(val_group_id, shard, val_set, key_seqno, opts, started_), shard});
}
}
}
@ -2808,6 +2808,9 @@ void ValidatorManagerImpl::prepare_stats(td::Promise<std::vector<std::pair<std::
promise.set_value(std::move(vec));
});
}
td::NamedThreadSafeCounter::get_default().for_each([&](auto key, auto value) {
vec.emplace_back("counter." + key, PSTRING() << value);
});
if (!shard_client_.empty()) {
auto P = td::PromiseCreator::lambda([promise = merger.make_promise("")](td::Result<BlockSeqno> R) mutable {
@ -3031,18 +3034,18 @@ void ValidatorManagerImpl::get_block_state_for_litequery(BlockIdExt block_id,
promise.set_result(R.move_as_ok());
return;
}
td::actor::send_closure(manager, &ValidatorManagerImpl::get_block_handle_for_litequery,
block_id, [manager, promise = std::move(promise)](td::Result<ConstBlockHandle> R) mutable {
TRY_RESULT_PROMISE(promise, handle, std::move(R));
td::actor::send_closure_later(manager, &ValidatorManager::get_shard_state_from_db, std::move(handle),
std::move(promise));
});
td::actor::send_closure(manager, &ValidatorManagerImpl::get_block_handle_for_litequery, block_id,
[manager, promise = std::move(promise)](td::Result<ConstBlockHandle> R) mutable {
TRY_RESULT_PROMISE(promise, handle, std::move(R));
td::actor::send_closure_later(manager, &ValidatorManager::get_shard_state_from_db,
std::move(handle), std::move(promise));
});
});
}
}
void ValidatorManagerImpl::get_block_by_lt_for_litequery(AccountIdPrefixFull account, LogicalTime lt,
td::Promise<ConstBlockHandle> promise) {
td::Promise<ConstBlockHandle> promise) {
get_block_by_lt_from_db(
account, lt, [=, SelfId = actor_id(this), promise = std::move(promise)](td::Result<ConstBlockHandle> R) mutable {
if (R.is_ok() && R.ok()->is_applied()) {
@ -3055,7 +3058,7 @@ void ValidatorManagerImpl::get_block_by_lt_for_litequery(AccountIdPrefixFull acc
}
void ValidatorManagerImpl::get_block_by_unix_time_for_litequery(AccountIdPrefixFull account, UnixTime ts,
td::Promise<ConstBlockHandle> promise) {
td::Promise<ConstBlockHandle> promise) {
get_block_by_unix_time_from_db(
account, ts, [=, SelfId = actor_id(this), promise = std::move(promise)](td::Result<ConstBlockHandle> R) mutable {
if (R.is_ok() && R.ok()->is_applied()) {
@ -3068,7 +3071,7 @@ void ValidatorManagerImpl::get_block_by_unix_time_for_litequery(AccountIdPrefixF
}
void ValidatorManagerImpl::get_block_by_seqno_for_litequery(AccountIdPrefixFull account, BlockSeqno seqno,
td::Promise<ConstBlockHandle> promise) {
td::Promise<ConstBlockHandle> promise) {
get_block_by_seqno_from_db(
account, seqno,
[=, SelfId = actor_id(this), promise = std::move(promise)](td::Result<ConstBlockHandle> R) mutable {

30
validator/state-serializer.cpp
View file

@ -22,6 +22,7 @@
#include "ton/ton-io.hpp"
#include "common/delay.h"
#include "td/utils/filesystem.h"
#include "td/utils/HashSet.h"
namespace ton {
@ -227,17 +228,17 @@ void AsyncStateSerializer::got_masterchain_handle(BlockHandle handle) {
class CachedCellDbReader : public vm::CellDbReader {
public:
CachedCellDbReader(std::shared_ptr<vm::CellDbReader> parent,
std::shared_ptr<std::map<td::Bits256, td::Ref<vm::Cell>>> cache)
std::shared_ptr<vm::CellHashSet> cache)
: parent_(std::move(parent)), cache_(std::move(cache)) {
}
td::Result<td::Ref<vm::DataCell>> load_cell(td::Slice hash) override {
++total_reqs_;
DCHECK(hash.size() == 32);
if (cache_) {
auto it = cache_->find(td::Bits256{(const unsigned char*)hash.data()});
auto it = cache_->find(hash);
if (it != cache_->end()) {
++cached_reqs_;
TRY_RESULT(loaded_cell, it->second->load_cell());
TRY_RESULT(loaded_cell, (*it)->load_cell());
return loaded_cell.data_cell;
}
}
@ -248,7 +249,7 @@ class CachedCellDbReader : public vm::CellDbReader {
}
private:
std::shared_ptr<vm::CellDbReader> parent_;
std::shared_ptr<std::map<td::Bits256, td::Ref<vm::Cell>>> cache_;
std::shared_ptr<vm::CellHashSet> cache_;
td::uint64 total_reqs_ = 0;
td::uint64 cached_reqs_ = 0;
@ -272,10 +273,9 @@ void AsyncStateSerializer::PreviousStateCache::prepare_cache(ShardIdFull shard)
td::Timer timer;
LOG(WARNING) << "Preloading previous persistent state for shard " << shard.to_str() << " ("
<< cur_shards.size() << " files)";
std::map<td::Bits256, td::Ref<vm::Cell>> cells;
vm::CellHashSet cells;
std::function<void(td::Ref<vm::Cell>)> dfs = [&](td::Ref<vm::Cell> cell) {
td::Bits256 hash = cell->get_hash().bits();
if (!cells.emplace(hash, cell).second) {
if (!cells.insert(cell).second) {
return;
}
bool is_special;
@ -303,7 +303,7 @@ void AsyncStateSerializer::PreviousStateCache::prepare_cache(ShardIdFull shard)
dfs(r_root.move_as_ok());
}
LOG(WARNING) << "Preloaded previous state: " << cells.size() << " cells in " << timer.elapsed() << "s";
cache = std::make_shared<std::map<td::Bits256, td::Ref<vm::Cell>>>(std::move(cells));
cache = std::make_shared<vm::CellHashSet>(std::move(cells));
}
void AsyncStateSerializer::got_masterchain_state(td::Ref<MasterchainState> state,
@ -322,15 +322,18 @@ void AsyncStateSerializer::got_masterchain_state(td::Ref<MasterchainState> state
shards_.push_back(v->top_block_id());
}
auto write_data = [shard = state->get_shard(), hash = state->root_cell()->get_hash(), cell_db_reader,
auto write_data = [shard = state->get_shard(), root = state->root_cell(), cell_db_reader,
previous_state_cache = previous_state_cache_,
fast_serializer_enabled = opts_->get_fast_state_serializer_enabled(),
cancellation_token = cancellation_token_source_.get_cancellation_token()](td::FileFd& fd) mutable {
if (!cell_db_reader) {
return vm::std_boc_serialize_to_file(root, fd, 31, std::move(cancellation_token));
}
if (fast_serializer_enabled) {
previous_state_cache->prepare_cache(shard);
}
auto new_cell_db_reader = std::make_shared<CachedCellDbReader>(cell_db_reader, previous_state_cache->cache);
auto res = vm::std_boc_serialize_to_file_large(new_cell_db_reader, hash, fd, 31, std::move(cancellation_token));
auto res = vm::std_boc_serialize_to_file_large(new_cell_db_reader, root->get_hash(), fd, 31, std::move(cancellation_token));
new_cell_db_reader->print_stats();
return res;
};
@ -384,15 +387,18 @@ void AsyncStateSerializer::got_shard_state(BlockHandle handle, td::Ref<ShardStat
return;
}
LOG(ERROR) << "serializing shard state " << handle->id().id.to_str();
auto write_data = [shard = state->get_shard(), hash = state->root_cell()->get_hash(), cell_db_reader,
auto write_data = [shard = state->get_shard(), root = state->root_cell(), cell_db_reader,
previous_state_cache = previous_state_cache_,
fast_serializer_enabled = opts_->get_fast_state_serializer_enabled(),
cancellation_token = cancellation_token_source_.get_cancellation_token()](td::FileFd& fd) mutable {
if (!cell_db_reader) {
return vm::std_boc_serialize_to_file(root, fd, 31, std::move(cancellation_token));
}
if (fast_serializer_enabled) {
previous_state_cache->prepare_cache(shard);
}
auto new_cell_db_reader = std::make_shared<CachedCellDbReader>(cell_db_reader, previous_state_cache->cache);
auto res = vm::std_boc_serialize_to_file_large(new_cell_db_reader, hash, fd, 31, std::move(cancellation_token));
auto res = vm::std_boc_serialize_to_file_large(new_cell_db_reader, root->get_hash(), fd, 31, std::move(cancellation_token));
new_cell_db_reader->print_stats();
return res;
};

2
validator/state-serializer.hpp
View file

@ -51,7 +51,7 @@ class AsyncStateSerializer : public td::actor::Actor {
std::vector<BlockIdExt> shards_;
struct PreviousStateCache {
std::vector<std::pair<std::string, ShardIdFull>> state_files;
std::shared_ptr<std::map<td::Bits256, td::Ref<vm::Cell>>> cache;
std::shared_ptr<vm::CellHashSet> cache;
std::vector<ShardIdFull> cur_shards;
void prepare_cache(ShardIdFull shard);

7
validator/validator-options.hpp
View file

@ -138,6 +138,9 @@ struct ValidatorManagerOptionsImpl : public ValidatorManagerOptions {
bool get_celldb_preload_all() const override {
return celldb_preload_all_;
}
bool get_celldb_in_memory() const override {
return celldb_in_memory_;
}
td::optional<double> get_catchain_max_block_delay() const override {
return catchain_max_block_delay_;
}
@ -230,6 +233,9 @@ struct ValidatorManagerOptionsImpl : public ValidatorManagerOptions {
void set_celldb_preload_all(bool value) override {
celldb_preload_all_ = value;
}
void set_celldb_in_memory(bool value) override {
celldb_in_memory_ = value;
}
void set_catchain_max_block_delay(double value) override {
catchain_max_block_delay_ = value;
}
@ -295,6 +301,7 @@ struct ValidatorManagerOptionsImpl : public ValidatorManagerOptions {
td::optional<td::uint64> celldb_cache_size_;
bool celldb_direct_io_ = false;
bool celldb_preload_all_ = false;
bool celldb_in_memory_ = false;
td::optional<double> catchain_max_block_delay_, catchain_max_block_delay_slow_;
bool state_serializer_enabled_ = true;
td::Ref<CollatorOptions> collator_options_{true};

2
validator/validator.h
View file

@ -102,6 +102,7 @@ struct ValidatorManagerOptions : public td::CntObject {
virtual BlockSeqno sync_upto() const = 0;
virtual std::string get_session_logs_file() const = 0;
virtual td::uint32 get_celldb_compress_depth() const = 0;
virtual bool get_celldb_in_memory() const = 0;
virtual size_t get_max_open_archive_files() const = 0;
virtual double get_archive_preload_period() const = 0;
virtual bool get_disable_rocksdb_stats() const = 0;
@ -141,6 +142,7 @@ struct ValidatorManagerOptions : public td::CntObject {
virtual void set_celldb_cache_size(td::uint64 value) = 0;
virtual void set_celldb_direct_io(bool value) = 0;
virtual void set_celldb_preload_all(bool value) = 0;
virtual void set_celldb_in_memory(bool value) = 0;
virtual void set_catchain_max_block_delay(double value) = 0;
virtual void set_catchain_max_block_delay_slow(double value) = 0;
virtual void set_state_serializer_enabled(bool value) = 0;