mirror of
https://github.com/ton-blockchain/ton
synced 2025-02-15 04:32:21 +00:00
Merge branch 'testnet' into accelerator
This commit is contained in:
SpyCheese
commit
36e04971eb
15 changed files with 633 additions and 132 deletions
|
|
@ -21,6 +21,12 @@ jobs:
|
|||
sudo apt-get update
|
||||
sudo apt-get install -y build-essential git cmake ninja-build zlib1g-dev libsecp256k1-dev libmicrohttpd-dev libsodium-dev liblz4-dev libjemalloc-dev
|
||||
|
||||
- if: matrix.os == 'ubuntu-20.04'
|
||||
run: |
|
||||
sudo apt install -y manpages-dev software-properties-common
|
||||
sudo add-apt-repository ppa:ubuntu-toolchain-r/test
|
||||
sudo apt update && sudo apt install gcc-11 g++-11
|
||||
|
||||
- if: matrix.os != 'ubuntu-24.04'
|
||||
run: |
|
||||
wget https://apt.llvm.org/llvm.sh
|
||||
|
|
|
|||
|
|
@ -88,7 +88,7 @@ if [ ! -d "openssl_3" ]; then
|
|||
cd openssl_3
|
||||
opensslPath=`pwd`
|
||||
git checkout openssl-3.1.4
|
||||
./config -static
|
||||
./config
|
||||
make build_libs -j12
|
||||
test $? -eq 0 || { echo "Can't compile openssl_3"; exit 1; }
|
||||
cd ..
|
||||
|
|
|
|||
|
|
@ -81,7 +81,7 @@ if [ ! -d "openssl_3" ]; then
|
|||
cd openssl_3
|
||||
opensslPath=`pwd`
|
||||
git checkout openssl-3.1.4
|
||||
./config -static
|
||||
./config
|
||||
make build_libs -j12
|
||||
test $? -eq 0 || { echo "Can't compile openssl_3"; exit 1; }
|
||||
cd ..
|
||||
|
|
|
|||
|
|
@ -747,8 +747,8 @@ td::uint64 BlockLimitStatus::estimate_block_size(const vm::NewCellStorageStat::S
|
|||
if (extra) {
|
||||
sum += *extra;
|
||||
}
|
||||
return 2000 + (sum.bits >> 3) + sum.cells * 12 + sum.internal_refs * 3 + sum.external_refs * 40 + accounts * 200 +
|
||||
transactions * 200 + (extra ? 200 : 0) + extra_out_msgs * 300 + public_library_diff * 700;
|
||||
return 2000 + (sum.bits >> 3) + sum.cells * 12 + sum.internal_refs * 3 + sum.external_refs * 40 + transactions * 200 +
|
||||
(extra ? 200 : 0) + extra_out_msgs * 300 + public_library_diff * 700;
|
||||
}
|
||||
|
||||
int BlockLimitStatus::classify() const {
|
||||
|
|
|
|||
|
|
@ -61,10 +61,81 @@
|
|||
#include "openssl/digest.hpp"
|
||||
#include "vm/dict.h"
|
||||
|
||||
#include <condition_variable>
|
||||
#include <latch>
|
||||
#include <numeric>
|
||||
#include <optional>
|
||||
#include <queue>
|
||||
|
||||
namespace vm {
|
||||
class ThreadExecutor : public DynamicBagOfCellsDb::AsyncExecutor {
|
||||
public:
|
||||
explicit ThreadExecutor(size_t threads_n) {
|
||||
for (size_t i = 0; i < threads_n; ++i) {
|
||||
threads_.emplace_back([this]() {
|
||||
while (true) {
|
||||
auto task = pop_task();
|
||||
if (!task) {
|
||||
break;
|
||||
}
|
||||
CHECK(generation_.load() % 2 == 1);
|
||||
task();
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
~ThreadExecutor() override {
|
||||
for (size_t i = 0; i < threads_.size(); ++i) {
|
||||
push_task({});
|
||||
}
|
||||
for (auto &t : threads_) {
|
||||
t.join();
|
||||
}
|
||||
}
|
||||
|
||||
void execute_async(std::function<void()> f) override {
|
||||
push_task(std::move(f));
|
||||
}
|
||||
|
||||
void execute_sync(std::function<void()> f) override {
|
||||
auto x = generation_.load();
|
||||
std::scoped_lock lock(sync_mutex_);
|
||||
CHECK(x == generation_);
|
||||
CHECK(generation_.load() % 2 == 1);
|
||||
f();
|
||||
CHECK(generation_.load() % 2 == 1);
|
||||
}
|
||||
void inc_generation() {
|
||||
generation_.fetch_add(1);
|
||||
}
|
||||
|
||||
private:
|
||||
std::atomic<size_t> generation_{0};
|
||||
std::queue<std::pair<std::function<void()>, size_t>> queue_;
|
||||
std::mutex queue_mutex_;
|
||||
std::condition_variable cv_;
|
||||
std::mutex sync_mutex_;
|
||||
std::vector<td::thread> threads_;
|
||||
|
||||
std::function<void()> pop_task() {
|
||||
std::unique_lock lock(queue_mutex_);
|
||||
cv_.wait(lock, [&] { return !queue_.empty(); });
|
||||
CHECK(!queue_.empty());
|
||||
auto task = std::move(queue_.front());
|
||||
queue_.pop();
|
||||
CHECK(task.second == generation_);
|
||||
return task.first;
|
||||
}
|
||||
|
||||
void push_task(std::function<void()> task) {
|
||||
{
|
||||
std::scoped_lock lock(queue_mutex_);
|
||||
queue_.emplace(std::move(task), generation_.load());
|
||||
}
|
||||
cv_.notify_one();
|
||||
}
|
||||
};
|
||||
|
||||
std::vector<int> do_get_serialization_modes() {
|
||||
std::vector<int> res;
|
||||
|
|
@ -890,25 +961,91 @@ TEST(TonDb, InMemoryDynamicBocSimple) {
|
|||
boc = DynamicBagOfCellsDb::create_in_memory(kv.get(), {});
|
||||
}
|
||||
|
||||
void test_dynamic_boc(std::optional<DynamicBagOfCellsDb::CreateInMemoryOptions> o_in_memory) {
|
||||
int VERBOSITY_NAME(boc) = VERBOSITY_NAME(DEBUG) + 10;
|
||||
|
||||
struct BocOptions {
|
||||
std::shared_ptr<ThreadExecutor> async_executor;
|
||||
std::optional<DynamicBagOfCellsDb::CreateInMemoryOptions> o_in_memory;
|
||||
td::uint64 seed{123};
|
||||
|
||||
auto create_dboc(td::KeyValueReader *kv, std::optional<td::int64> o_root_n) {
|
||||
if (o_in_memory) {
|
||||
auto res = DynamicBagOfCellsDb::create_in_memory(kv, *o_in_memory);
|
||||
auto stats = res->get_stats().move_as_ok();
|
||||
if (o_root_n) {
|
||||
ASSERT_EQ(*o_root_n, stats.roots_total_count);
|
||||
}
|
||||
VLOG(boc) << "reset roots_n=" << stats.roots_total_count << " cells_n=" << stats.cells_total_count;
|
||||
return res;
|
||||
}
|
||||
return DynamicBagOfCellsDb::create();
|
||||
};
|
||||
void prepare_commit(DynamicBagOfCellsDb &dboc) {
|
||||
if (async_executor) {
|
||||
async_executor->inc_generation();
|
||||
std::latch latch(1);
|
||||
td::Result<td::Unit> res;
|
||||
async_executor->execute_sync([&] {
|
||||
dboc.prepare_commit_async(async_executor, [&](auto r) {
|
||||
res = std::move(r);
|
||||
latch.count_down();
|
||||
});
|
||||
});
|
||||
latch.wait();
|
||||
async_executor->execute_sync([&] {});
|
||||
async_executor->inc_generation();
|
||||
} else {
|
||||
dboc.prepare_commit();
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
template <class F>
|
||||
void with_all_boc_options(F &&f, size_t tests_n = 500) {
|
||||
LOG(INFO) << "Test dynamic boc";
|
||||
auto counter = [] { return td::NamedThreadSafeCounter::get_default().get_counter("DataCell").sum(); };
|
||||
auto run = [&](BocOptions options) {
|
||||
LOG(INFO) << "\t" << (options.o_in_memory ? "in memory" : "on disk") << (options.async_executor ? " async" : "");
|
||||
if (options.o_in_memory) {
|
||||
LOG(INFO) << "\t\tuse_arena=" << options.o_in_memory->use_arena
|
||||
<< " less_memory=" << options.o_in_memory->use_less_memory_during_creation;
|
||||
}
|
||||
for (td::uint32 i = 0; i < tests_n; i++) {
|
||||
auto before = counter();
|
||||
options.seed = i == 0 ? 123 : i;
|
||||
f(options);
|
||||
auto after = counter();
|
||||
LOG_CHECK((options.o_in_memory && options.o_in_memory->use_arena) || before == after)
|
||||
<< before << " vs " << after;
|
||||
}
|
||||
};
|
||||
run({.async_executor = std::make_shared<ThreadExecutor>(4)});
|
||||
run({});
|
||||
for (auto use_arena : {false, true}) {
|
||||
for (auto less_memory : {false, true}) {
|
||||
run({.o_in_memory =
|
||||
DynamicBagOfCellsDb::CreateInMemoryOptions{.extra_threads = std::thread::hardware_concurrency(),
|
||||
.verbose = false,
|
||||
.use_arena = use_arena,
|
||||
.use_less_memory_during_creation = less_memory}});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void test_dynamic_boc(BocOptions options) {
|
||||
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();
|
||||
;
|
||||
LOG_CHECK((options.o_in_memory && options.o_in_memory->use_arena) || before == counter())
|
||||
<< before << " vs " << counter();
|
||||
};
|
||||
td::Random::Xorshift128plus rnd{123};
|
||||
td::Random::Xorshift128plus rnd{options.seed};
|
||||
std::string old_root_hash;
|
||||
std::string old_root_serialization;
|
||||
auto kv = std::make_shared<td::MemoryKeyValue>();
|
||||
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 res;
|
||||
}
|
||||
return DynamicBagOfCellsDb::create();
|
||||
return options.create_dboc(kv.get(), roots_n);
|
||||
};
|
||||
auto dboc = create_dboc();
|
||||
dboc->set_loader(std::make_unique<CellLoader>(kv));
|
||||
|
|
@ -947,51 +1084,28 @@ void test_dynamic_boc(std::optional<DynamicBagOfCellsDb::CreateInMemoryOptions>
|
|||
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);
|
||||
with_all_boc_options(test_dynamic_boc, 1);
|
||||
};
|
||||
|
||||
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;
|
||||
void test_dynamic_boc2(BocOptions options) {
|
||||
td::Random::Xorshift128plus rnd{options.seed};
|
||||
|
||||
int total_roots = rnd.fast(1, !rnd.fast(0, 10) * 100 + 10);
|
||||
int max_roots = rnd.fast(1, 20);
|
||||
int last_commit_at = 0;
|
||||
int first_root_id = 0;
|
||||
int last_root_id = 0;
|
||||
auto kv = std::make_shared<td::MemoryKeyValue>();
|
||||
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 res;
|
||||
}
|
||||
return DynamicBagOfCellsDb::create();
|
||||
};
|
||||
auto create_dboc = [&](td::int64 root_n) { return options.create_dboc(kv.get(), root_n); };
|
||||
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();
|
||||
SCOPE_EXIT{
|
||||
// LOG_CHECK((options.o_in_memory && options.o_in_memory->use_arena) || before == counter())
|
||||
// << before << " vs " << counter();
|
||||
};
|
||||
|
||||
std::vector<Ref<Cell>> roots(max_roots);
|
||||
|
|
@ -1009,7 +1123,7 @@ void test_dynamic_boc2(std::optional<DynamicBagOfCellsDb::CreateInMemoryOptions>
|
|||
if (from_root.is_null()) {
|
||||
VLOG(boc) << " from db";
|
||||
auto from_root_hash = root_hashes[root_id % max_roots];
|
||||
if (o_in_memory && (rnd() % 2 == 0)) {
|
||||
if (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();
|
||||
|
|
@ -1041,7 +1155,8 @@ void test_dynamic_boc2(std::optional<DynamicBagOfCellsDb::CreateInMemoryOptions>
|
|||
|
||||
auto commit = [&] {
|
||||
VLOG(boc) << "commit";
|
||||
dboc->prepare_commit();
|
||||
//rnd.fast(0, 1);
|
||||
options.prepare_commit(*dboc);
|
||||
{
|
||||
CellStorer cell_storer(*kv);
|
||||
dboc->commit(cell_storer);
|
||||
|
|
@ -2147,18 +2262,18 @@ TEST(TonDb, BocRespectsUsageCell) {
|
|||
ASSERT_STREQ(serialization, serialization_of_virtualized_cell);
|
||||
}
|
||||
|
||||
void test_dynamic_boc_respectes_usage_cell(std::optional<vm::DynamicBagOfCellsDb::CreateInMemoryOptions> o_in_memory) {
|
||||
td::Random::Xorshift128plus rnd(123);
|
||||
void test_dynamic_boc_respectes_usage_cell(vm::BocOptions options) {
|
||||
td::Random::Xorshift128plus rnd(options.seed);
|
||||
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 = o_in_memory ? vm::DynamicBagOfCellsDb::create_in_memory(kv.get(), *o_in_memory)
|
||||
: vm::DynamicBagOfCellsDb::create();
|
||||
auto dboc = options.create_dboc(kv.get(), {});
|
||||
dboc->set_loader(std::make_unique<vm::CellLoader>(kv));
|
||||
dboc->inc(usage_cell);
|
||||
{
|
||||
options.prepare_commit(*dboc);
|
||||
vm::CellStorer cell_storer(*kv);
|
||||
dboc->commit(cell_storer);
|
||||
}
|
||||
|
|
@ -2171,7 +2286,7 @@ void test_dynamic_boc_respectes_usage_cell(std::optional<vm::DynamicBagOfCellsDb
|
|||
}
|
||||
|
||||
TEST(TonDb, DynamicBocRespectsUsageCell) {
|
||||
vm::with_all_boc_options(test_dynamic_boc_respectes_usage_cell);
|
||||
vm::with_all_boc_options(test_dynamic_boc_respectes_usage_cell, 20);
|
||||
}
|
||||
|
||||
TEST(TonDb, LargeBocSerializer) {
|
||||
|
|
|
|||
|
|
@ -43,7 +43,7 @@ class CellHashTable {
|
|||
template <class F>
|
||||
void for_each(F &&f) {
|
||||
for (auto &info : set_) {
|
||||
f(info);
|
||||
f(const_cast<InfoT &>(info));
|
||||
}
|
||||
}
|
||||
template <class F>
|
||||
|
|
|
|||
|
|
@ -184,6 +184,24 @@ td::Result<CellLoader::LoadResult> CellLoader::load(td::Slice hash, td::Slice va
|
|||
return res;
|
||||
}
|
||||
|
||||
td::Result<CellLoader::LoadResult> CellLoader::load_refcnt(td::Slice hash) {
|
||||
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;
|
||||
}
|
||||
res.status = LoadResult::Ok;
|
||||
td::TlParser parser(serialized);
|
||||
td::parse(res.refcnt_, parser);
|
||||
if (res.refcnt_ == -1) {
|
||||
parse(res.refcnt_, parser);
|
||||
}
|
||||
TRY_STATUS(parser.get_status());
|
||||
return res;
|
||||
}
|
||||
|
||||
CellStorer::CellStorer(KeyValue &kv) : kv_(kv) {
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -50,6 +50,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);
|
||||
td::Result<LoadResult> load_refcnt(td::Slice hash); // This only loads refcnt_, cell_ == null
|
||||
|
||||
private:
|
||||
std::shared_ptr<KeyValueReader> reader_;
|
||||
|
|
|
|||
|
|
@ -27,6 +27,9 @@
|
|||
#include "td/utils/ThreadSafeCounter.h"
|
||||
|
||||
#include "vm/cellslice.h"
|
||||
#include <queue>
|
||||
#include "td/actor/actor.h"
|
||||
#include "common/delay.h"
|
||||
|
||||
namespace vm {
|
||||
namespace {
|
||||
|
|
@ -180,6 +183,9 @@ class DynamicBagOfCellsDbImpl : public DynamicBagOfCellsDb, private ExtCellCreat
|
|||
}
|
||||
|
||||
td::Status prepare_commit() override {
|
||||
if (pca_state_) {
|
||||
return td::Status::Error("prepare_commit_async is not finished");
|
||||
}
|
||||
if (is_prepared_for_commit()) {
|
||||
return td::Status::OK();
|
||||
}
|
||||
|
|
@ -585,6 +591,221 @@ class DynamicBagOfCellsDbImpl : public DynamicBagOfCellsDb, private ExtCellCreat
|
|||
DynamicBocExtCellExtra{cell_db_reader_}));
|
||||
return std::move(res);
|
||||
}
|
||||
|
||||
struct PrepareCommitAsyncState {
|
||||
size_t remaining_ = 0;
|
||||
std::shared_ptr<AsyncExecutor> executor_;
|
||||
td::Promise<td::Unit> promise_;
|
||||
|
||||
struct CellInfo2 {
|
||||
CellInfo *info{};
|
||||
std::vector<CellInfo2 *> parents;
|
||||
unsigned remaining_children = 0;
|
||||
Cell::Hash key() const {
|
||||
return info->key();
|
||||
}
|
||||
bool operator<(const CellInfo2 &other) const {
|
||||
return key() < other.key();
|
||||
}
|
||||
|
||||
friend bool operator<(const CellInfo2 &a, td::Slice b) {
|
||||
return a.key().as_slice() < b;
|
||||
}
|
||||
|
||||
friend bool operator<(td::Slice a, const CellInfo2 &b) {
|
||||
return a < b.key().as_slice();
|
||||
}
|
||||
|
||||
struct Eq {
|
||||
using is_transparent = void; // Pred to use
|
||||
bool operator()(const CellInfo2 &info, const CellInfo2 &other_info) const {
|
||||
return info.key() == other_info.key();
|
||||
}
|
||||
bool operator()(const CellInfo2 &info, td::Slice hash) const {
|
||||
return info.key().as_slice() == hash;
|
||||
}
|
||||
bool operator()(td::Slice hash, const CellInfo2 &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 CellInfo2 &info) const {
|
||||
return cell_hash_slice_hash(info.key().as_slice());
|
||||
}
|
||||
};
|
||||
};
|
||||
|
||||
CellHashTable<CellInfo2> cells_;
|
||||
|
||||
std::queue<CellInfo2*> load_queue_;
|
||||
td::uint32 active_load_ = 0;
|
||||
td::uint32 max_parallel_load_ = 4;
|
||||
};
|
||||
std::unique_ptr<PrepareCommitAsyncState> pca_state_;
|
||||
|
||||
void prepare_commit_async(std::shared_ptr<AsyncExecutor> executor, td::Promise<td::Unit> promise) override {
|
||||
hash_table_ = {};
|
||||
if (pca_state_) {
|
||||
promise.set_error(td::Status::Error("Other prepare_commit_async is not finished"));
|
||||
return;
|
||||
}
|
||||
if (is_prepared_for_commit()) {
|
||||
promise.set_result(td::Unit());
|
||||
return;
|
||||
}
|
||||
pca_state_ = std::make_unique<PrepareCommitAsyncState>();
|
||||
pca_state_->executor_ = std::move(executor);
|
||||
pca_state_->promise_ = std::move(promise);
|
||||
for (auto &new_cell : to_inc_) {
|
||||
dfs_new_cells_in_db_async(new_cell);
|
||||
}
|
||||
pca_state_->cells_.for_each([&](PrepareCommitAsyncState::CellInfo2 &info) {
|
||||
++pca_state_->remaining_;
|
||||
if (info.remaining_children == 0) {
|
||||
pca_load_from_db(&info);
|
||||
}
|
||||
});
|
||||
if (pca_state_->remaining_ == 0) {
|
||||
prepare_commit_async_cont();
|
||||
}
|
||||
}
|
||||
|
||||
void dfs_new_cells_in_db_async(const td::Ref<vm::Cell> &cell, PrepareCommitAsyncState::CellInfo2 *parent = nullptr) {
|
||||
bool exists = true;
|
||||
pca_state_->cells_.apply(cell->get_hash().as_slice(), [&](PrepareCommitAsyncState::CellInfo2 &info) {
|
||||
if (info.info == nullptr) {
|
||||
exists = false;
|
||||
info.info = &get_cell_info(cell);
|
||||
}
|
||||
});
|
||||
auto info = pca_state_->cells_.get_if_exists(cell->get_hash().as_slice());
|
||||
if (parent) {
|
||||
info->parents.push_back(parent);
|
||||
++parent->remaining_children;
|
||||
}
|
||||
if (exists) {
|
||||
return;
|
||||
}
|
||||
if (cell->is_loaded()) {
|
||||
vm::CellSlice cs(vm::NoVm{}, cell);
|
||||
for (unsigned i = 0; i < cs.size_refs(); i++) {
|
||||
dfs_new_cells_in_db_async(cs.prefetch_ref(i), info);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void pca_load_from_db(PrepareCommitAsyncState::CellInfo2 *info) {
|
||||
if (pca_state_->active_load_ >= pca_state_->max_parallel_load_) {
|
||||
pca_state_->load_queue_.push(info);
|
||||
return;
|
||||
}
|
||||
++pca_state_->active_load_;
|
||||
pca_state_->executor_->execute_async(
|
||||
[db = this, info, executor = pca_state_->executor_, loader = *loader_]() mutable {
|
||||
auto res = loader.load_refcnt(info->info->cell->get_hash().as_slice()).move_as_ok();
|
||||
executor->execute_sync([db, info, res = std::move(res)]() {
|
||||
--db->pca_state_->active_load_;
|
||||
db->pca_process_load_queue();
|
||||
db->pca_set_in_db(info, std::move(res));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
void pca_process_load_queue() {
|
||||
while (pca_state_->active_load_ < pca_state_->max_parallel_load_ && !pca_state_->load_queue_.empty()) {
|
||||
PrepareCommitAsyncState::CellInfo2 *info = pca_state_->load_queue_.front();
|
||||
pca_state_->load_queue_.pop();
|
||||
pca_load_from_db(info);
|
||||
}
|
||||
}
|
||||
|
||||
void pca_set_in_db(PrepareCommitAsyncState::CellInfo2 *info, CellLoader::LoadResult result) {
|
||||
info->info->sync_with_db = true;
|
||||
if (result.status == CellLoader::LoadResult::Ok) {
|
||||
info->info->in_db = true;
|
||||
info->info->db_refcnt = result.refcnt();
|
||||
} else {
|
||||
info->info->in_db = false;
|
||||
}
|
||||
for (PrepareCommitAsyncState::CellInfo2 *parent_info : info->parents) {
|
||||
if (parent_info->info->sync_with_db) {
|
||||
continue;
|
||||
}
|
||||
if (!info->info->in_db) {
|
||||
pca_set_in_db(parent_info, {});
|
||||
} else if (--parent_info->remaining_children == 0) {
|
||||
pca_load_from_db(parent_info);
|
||||
}
|
||||
}
|
||||
CHECK(pca_state_->remaining_ != 0);
|
||||
if (--pca_state_->remaining_ == 0) {
|
||||
prepare_commit_async_cont();
|
||||
}
|
||||
}
|
||||
|
||||
void prepare_commit_async_cont() {
|
||||
for (auto &new_cell : to_inc_) {
|
||||
auto &new_cell_info = get_cell_info(new_cell);
|
||||
dfs_new_cells(new_cell_info);
|
||||
}
|
||||
|
||||
CHECK(pca_state_->remaining_ == 0);
|
||||
for (auto &old_cell : to_dec_) {
|
||||
auto &old_cell_info = get_cell_info(old_cell);
|
||||
dfs_old_cells_async(old_cell_info);
|
||||
}
|
||||
if (pca_state_->remaining_ == 0) {
|
||||
prepare_commit_async_cont2();
|
||||
}
|
||||
}
|
||||
|
||||
void dfs_old_cells_async(CellInfo &info) {
|
||||
if (!info.was) {
|
||||
info.was = true;
|
||||
visited_.push_back(&info);
|
||||
if (!info.sync_with_db) {
|
||||
++pca_state_->remaining_;
|
||||
load_cell_async(
|
||||
info.cell->get_hash().as_slice(), pca_state_->executor_,
|
||||
[executor = pca_state_->executor_, db = this, info = &info](td::Result<td::Ref<vm::DataCell>> R) {
|
||||
R.ensure();
|
||||
executor->execute_sync([db, info]() {
|
||||
CHECK(info->sync_with_db);
|
||||
db->dfs_old_cells_async(*info);
|
||||
if (--db->pca_state_->remaining_ == 0) {
|
||||
db->prepare_commit_async_cont2();
|
||||
}
|
||||
});
|
||||
});
|
||||
return;
|
||||
}
|
||||
}
|
||||
info.refcnt_diff--;
|
||||
if (!info.sync_with_db) {
|
||||
return;
|
||||
}
|
||||
auto new_refcnt = info.refcnt_diff + info.db_refcnt;
|
||||
CHECK(new_refcnt >= 0);
|
||||
if (new_refcnt != 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
for_each(info, [this](auto &child_info) { dfs_old_cells_async(child_info); });
|
||||
}
|
||||
|
||||
void prepare_commit_async_cont2() {
|
||||
save_diff_prepare();
|
||||
to_inc_.clear();
|
||||
to_dec_.clear();
|
||||
pca_state_->promise_.set_result(td::Unit());
|
||||
pca_state_ = {};
|
||||
}
|
||||
|
||||
};
|
||||
} // namespace
|
||||
|
||||
|
|
|
|||
|
|
@ -105,6 +105,7 @@ class DynamicBagOfCellsDb {
|
|||
|
||||
virtual void load_cell_async(td::Slice hash, std::shared_ptr<AsyncExecutor> executor,
|
||||
td::Promise<Ref<DataCell>> promise) = 0;
|
||||
virtual void prepare_commit_async(std::shared_ptr<AsyncExecutor> executor, td::Promise<td::Unit> promise) = 0;
|
||||
};
|
||||
|
||||
} // namespace vm
|
||||
|
|
|
|||
|
|
@ -848,6 +848,10 @@ class InMemoryBagOfCellsDb : public DynamicBagOfCellsDb {
|
|||
to_inc_ = {};
|
||||
return td::Status::OK();
|
||||
}
|
||||
void prepare_commit_async(std::shared_ptr<AsyncExecutor> executor, td::Promise<td::Unit> promise) override {
|
||||
TRY_STATUS_PROMISE(promise, prepare_commit());
|
||||
promise.set_value(td::Unit());
|
||||
}
|
||||
Stats get_stats_diff() override {
|
||||
LOG(FATAL) << "Not implemented";
|
||||
return {};
|
||||
|
|
|
|||
|
|
@ -161,16 +161,33 @@ void CellDbIn::start_up() {
|
|||
}
|
||||
|
||||
void CellDbIn::load_cell(RootHash hash, td::Promise<td::Ref<vm::DataCell>> promise) {
|
||||
if (db_busy_) {
|
||||
action_queue_.push([self = this, hash, promise = std::move(promise)](td::Result<td::Unit> R) mutable {
|
||||
R.ensure();
|
||||
self->load_cell(hash, std::move(promise));
|
||||
});
|
||||
return;
|
||||
}
|
||||
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));
|
||||
auto cell = boc_->load_cell(hash.as_slice());
|
||||
delay_action(
|
||||
[cell = std::move(cell), promise = std::move(promise)]() mutable { promise.set_result(std::move(cell)); },
|
||||
td::Timestamp::now());
|
||||
}
|
||||
|
||||
void CellDbIn::store_cell(BlockIdExt block_id, td::Ref<vm::Cell> cell, td::Promise<td::Ref<vm::DataCell>> promise) {
|
||||
TD_PERF_COUNTER(celldb_store_cell);
|
||||
if (db_busy_) {
|
||||
action_queue_.push(
|
||||
[self = this, block_id, cell = std::move(cell), promise = std::move(promise)](td::Result<td::Unit> R) mutable {
|
||||
R.ensure();
|
||||
self->store_cell(block_id, std::move(cell), std::move(promise));
|
||||
});
|
||||
return;
|
||||
}
|
||||
td::PerfWarningTimer timer{"storecell", 0.1};
|
||||
auto key_hash = get_key_hash(block_id);
|
||||
auto R = get_block(key_hash);
|
||||
|
|
@ -180,6 +197,16 @@ void CellDbIn::store_cell(BlockIdExt block_id, td::Ref<vm::Cell> cell, td::Promi
|
|||
return;
|
||||
}
|
||||
|
||||
boc_->inc(cell);
|
||||
db_busy_ = true;
|
||||
boc_->prepare_commit_async(async_executor, [=, this, SelfId = actor_id(this), timer = std::move(timer),
|
||||
timer_prepare = td::Timer{}, promise = std::move(promise),
|
||||
cell = std::move(cell)](td::Result<td::Unit> Res) mutable {
|
||||
Res.ensure();
|
||||
timer_prepare.pause();
|
||||
td::actor::send_lambda(
|
||||
SelfId, [=, this, timer = std::move(timer), promise = std::move(promise), cell = std::move(cell)]() mutable {
|
||||
TD_PERF_COUNTER(celldb_store_cell);
|
||||
auto empty = get_empty_key_hash();
|
||||
auto ER = get_block(empty);
|
||||
ER.ensure();
|
||||
|
|
@ -199,16 +226,15 @@ void CellDbIn::store_cell(BlockIdExt block_id, td::Ref<vm::Cell> cell, td::Promi
|
|||
E.next = key_hash;
|
||||
P.prev = key_hash;
|
||||
}
|
||||
|
||||
boc_->inc(cell);
|
||||
boc_->prepare_commit().ensure();
|
||||
vm::CellStorer stor{*cell_db_.get()};
|
||||
td::Timer timer_write;
|
||||
vm::CellStorer stor{*cell_db_};
|
||||
cell_db_->begin_write_batch().ensure();
|
||||
boc_->commit(stor).ensure();
|
||||
set_block(empty, std::move(E));
|
||||
set_block(get_empty_key_hash(), std::move(E));
|
||||
set_block(D.prev, std::move(P));
|
||||
set_block(key_hash, std::move(D));
|
||||
cell_db_->commit_write_batch().ensure();
|
||||
timer_write.pause();
|
||||
|
||||
if (!opts_->get_celldb_in_memory()) {
|
||||
boc_->set_loader(std::make_unique<vm::CellLoader>(cell_db_->snapshot(), on_load_callback_)).ensure();
|
||||
|
|
@ -218,11 +244,24 @@ void CellDbIn::store_cell(BlockIdExt block_id, td::Ref<vm::Cell> cell, td::Promi
|
|||
promise.set_result(boc_->load_cell(cell->get_hash().as_slice()));
|
||||
if (!opts_->get_disable_rocksdb_stats()) {
|
||||
cell_db_statistics_.store_cell_time_.insert(timer.elapsed() * 1e6);
|
||||
cell_db_statistics_.store_cell_prepare_time_.insert(timer_prepare.elapsed() * 1e6);
|
||||
cell_db_statistics_.store_cell_write_time_.insert(timer_write.elapsed() * 1e6);
|
||||
}
|
||||
LOG(DEBUG) << "Stored state " << block_id.to_str();
|
||||
release_db();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
void CellDbIn::get_cell_db_reader(td::Promise<std::shared_ptr<vm::CellDbReader>> promise) {
|
||||
if (db_busy_) {
|
||||
action_queue_.push(
|
||||
[self = this, promise = std::move(promise)](td::Result<td::Unit> R) mutable {
|
||||
R.ensure();
|
||||
self->get_cell_db_reader(std::move(promise));
|
||||
});
|
||||
return;
|
||||
}
|
||||
promise.set_result(boc_->get_cell_db_reader());
|
||||
}
|
||||
|
||||
|
|
@ -262,6 +301,13 @@ void CellDbIn::flush_db_stats() {
|
|||
if (opts_->get_disable_rocksdb_stats()) {
|
||||
return;
|
||||
}
|
||||
if (db_busy_) {
|
||||
action_queue_.push([self = this](td::Result<td::Unit> R) mutable {
|
||||
R.ensure();
|
||||
self->flush_db_stats();
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
auto celldb_stats = prepare_stats();
|
||||
td::StringBuilder ss;
|
||||
|
|
@ -350,7 +396,14 @@ void CellDbIn::gc_cont(BlockHandle handle) {
|
|||
}
|
||||
|
||||
void CellDbIn::gc_cont2(BlockHandle handle) {
|
||||
TD_PERF_COUNTER(celldb_gc_cell);
|
||||
if (db_busy_) {
|
||||
action_queue_.push([self = this, handle = std::move(handle)](td::Result<td::Unit> R) mutable {
|
||||
R.ensure();
|
||||
self->gc_cont2(handle);
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
td::PerfWarningTimer timer{"gccell", 0.1};
|
||||
td::PerfWarningTimer timer_all{"gccell_all", 0.05};
|
||||
|
||||
|
|
@ -379,7 +432,16 @@ void CellDbIn::gc_cont2(BlockHandle handle) {
|
|||
auto cell = boc_->load_cell(F.root_hash.as_slice()).move_as_ok();
|
||||
|
||||
boc_->dec(cell);
|
||||
boc_->prepare_commit().ensure();
|
||||
db_busy_ = true;
|
||||
boc_->prepare_commit_async(
|
||||
async_executor, [this, SelfId = actor_id(this), timer_boc = std::move(timer_boc), F = std::move(F), key_hash,
|
||||
P = std::move(P), N = std::move(N), cell = std::move(cell), timer = std::move(timer),
|
||||
timer_all = std::move(timer_all), handle](td::Result<td::Unit> R) mutable {
|
||||
R.ensure();
|
||||
td::actor::send_lambda(SelfId, [this, timer_boc = std::move(timer_boc), F = std::move(F), key_hash,
|
||||
P = std::move(P), N = std::move(N), cell = std::move(cell),
|
||||
timer = std::move(timer), timer_all = std::move(timer_all), handle]() mutable {
|
||||
TD_PERF_COUNTER(celldb_gc_cell);
|
||||
vm::CellStorer stor{*cell_db_};
|
||||
timer_boc.reset();
|
||||
|
||||
|
|
@ -419,6 +481,9 @@ void CellDbIn::gc_cont2(BlockHandle handle) {
|
|||
LOG(DEBUG) << "Deleted state " << handle->id().to_str();
|
||||
timer_finish.reset();
|
||||
timer_all.reset();
|
||||
release_db();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
void CellDbIn::skip_gc() {
|
||||
|
|
@ -478,6 +543,13 @@ void CellDbIn::migrate_cell(td::Bits256 hash) {
|
|||
|
||||
void CellDbIn::migrate_cells() {
|
||||
migrate_after_ = td::Timestamp::never();
|
||||
if (db_busy_) {
|
||||
action_queue_.push([self = this](td::Result<td::Unit> R) mutable {
|
||||
R.ensure();
|
||||
self->migrate_cells();
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (cells_to_migrate_.empty()) {
|
||||
migration_active_ = false;
|
||||
return;
|
||||
|
|
@ -610,6 +682,8 @@ td::BufferSlice CellDbIn::DbEntry::release() {
|
|||
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("store_cell.prepare.micros", PSTRING() << store_cell_prepare_time_.to_string());
|
||||
stats.emplace_back("store_cell.write.micros", PSTRING() << store_cell_write_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_));
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@
|
|||
#include "td/db/RocksDb.h"
|
||||
|
||||
#include <optional>
|
||||
#include <queue>
|
||||
|
||||
namespace rocksdb {
|
||||
class Statistics;
|
||||
|
|
@ -136,6 +137,8 @@ class CellDbIn : public CellDbBase {
|
|||
|
||||
struct CellDbStatistics {
|
||||
PercentileStats store_cell_time_;
|
||||
PercentileStats store_cell_prepare_time_;
|
||||
PercentileStats store_cell_write_time_;
|
||||
PercentileStats gc_cell_time_;
|
||||
td::Timestamp stats_start_time_ = td::Timestamp::now();
|
||||
std::optional<double> in_memory_load_time_;
|
||||
|
|
@ -153,6 +156,18 @@ class CellDbIn : public CellDbBase {
|
|||
td::Timestamp statistics_flush_at_ = td::Timestamp::never();
|
||||
BlockSeqno last_deleted_mc_state_ = 0;
|
||||
|
||||
bool db_busy_ = false;
|
||||
std::queue<td::Promise<td::Unit>> action_queue_;
|
||||
|
||||
void release_db() {
|
||||
db_busy_ = false;
|
||||
while (!db_busy_ && !action_queue_.empty()) {
|
||||
auto action = std::move(action_queue_.front());
|
||||
action_queue_.pop();
|
||||
action.set_value(td::Unit());
|
||||
}
|
||||
}
|
||||
|
||||
public:
|
||||
class MigrationProxy : public td::actor::Actor {
|
||||
public:
|
||||
|
|
|
|||
|
|
@ -227,6 +227,10 @@ class Collator final : public td::actor::Actor {
|
|||
td::uint64 defer_out_queue_size_limit_;
|
||||
td::uint64 hard_defer_out_queue_size_limit_;
|
||||
|
||||
std::unique_ptr<vm::AugmentedDictionary> account_dict_estimator_;
|
||||
std::set<td::Bits256> account_dict_estimator_added_accounts_;
|
||||
unsigned account_dict_ops_{0};
|
||||
|
||||
bool msg_metadata_enabled_ = false;
|
||||
bool deferring_messages_enabled_ = false;
|
||||
bool store_out_msg_queue_size_ = false;
|
||||
|
|
@ -333,6 +337,7 @@ class Collator final : public td::actor::Actor {
|
|||
bool insert_out_msg(Ref<vm::Cell> out_msg, td::ConstBitPtr msg_hash);
|
||||
bool register_out_msg_queue_op(bool force = false);
|
||||
bool register_dispatch_queue_op(bool force = false);
|
||||
bool update_account_dict_estimation(const block::transaction::Transaction& trans);
|
||||
bool update_min_mc_seqno(ton::BlockSeqno some_mc_seqno);
|
||||
bool combine_account_transactions();
|
||||
bool update_public_libraries();
|
||||
|
|
|
|||
|
|
@ -1137,7 +1137,7 @@ bool Collator::split_last_state(block::ShardState& ss) {
|
|||
/**
|
||||
* Imports the shard state data into the Collator object.
|
||||
*
|
||||
* SETS: account_dict, shard_libraries_, mc_state_extra
|
||||
* SETS: account_dict = account_dict_estimator_, shard_libraries_, mc_state_extra
|
||||
* total_balance_ = old_total_balance_, total_validator_fees_
|
||||
* SETS: overload_history_, underload_history_
|
||||
* SETS: prev_state_utime_, prev_state_lt_, prev_vert_seqno_
|
||||
|
|
@ -1149,6 +1149,7 @@ bool Collator::split_last_state(block::ShardState& ss) {
|
|||
*/
|
||||
bool Collator::import_shard_state_data(block::ShardState& ss) {
|
||||
account_dict = std::move(ss.account_dict_);
|
||||
account_dict_estimator_ = std::make_unique<vm::AugmentedDictionary>(*account_dict);
|
||||
shard_libraries_ = std::move(ss.shard_libraries_);
|
||||
mc_state_extra_ = std::move(ss.mc_state_extra_);
|
||||
overload_history_ = ss.overload_history_;
|
||||
|
|
@ -2857,6 +2858,9 @@ bool Collator::create_ticktock_transaction(const ton::StdSmcAddress& smc_addr, t
|
|||
return fatal_error(
|
||||
td::Status::Error(-666, std::string{"cannot commit new transaction for smart contract "} + smc_addr.to_hex()));
|
||||
}
|
||||
if (!update_account_dict_estimation(*trans)) {
|
||||
return fatal_error(-666, "cannot update account dict size estimation");
|
||||
}
|
||||
update_max_lt(acc->last_trans_end_lt_);
|
||||
block::MsgMetadata new_msg_metadata{0, acc->workchain, acc->addr, trans->start_lt};
|
||||
register_new_msgs(*trans, std::move(new_msg_metadata));
|
||||
|
|
@ -2950,6 +2954,10 @@ Ref<vm::Cell> Collator::create_ordinary_transaction(Ref<vm::Cell> msg_root,
|
|||
fatal_error("cannot commit new transaction for smart contract "s + addr.to_hex());
|
||||
return {};
|
||||
}
|
||||
if (!update_account_dict_estimation(*trans)) {
|
||||
fatal_error("cannot update account dict size estimation");
|
||||
return {};
|
||||
}
|
||||
|
||||
td::optional<block::MsgMetadata> new_msg_metadata;
|
||||
if (external || is_special_tx) {
|
||||
|
|
@ -5175,6 +5183,39 @@ bool Collator::register_dispatch_queue_op(bool force) {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Update size estimation for the account dictionary.
|
||||
* This is required to count the depth of the ShardAccounts dictionary in the block size estimation.
|
||||
* account_dict_estimator_ is used for block limits only.
|
||||
*
|
||||
* @param trans Newly-created transaction.
|
||||
*
|
||||
* @returns True on success, false otherwise.
|
||||
*/
|
||||
bool Collator::update_account_dict_estimation(const block::transaction::Transaction& trans) {
|
||||
const block::Account& acc = trans.account;
|
||||
if (acc.orig_total_state->get_hash() != acc.total_state->get_hash() &&
|
||||
account_dict_estimator_added_accounts_.insert(acc.addr).second) {
|
||||
// see combine_account_transactions
|
||||
if (acc.status == block::Account::acc_nonexist) {
|
||||
account_dict_estimator_->lookup_delete(acc.addr);
|
||||
} else {
|
||||
vm::CellBuilder cb;
|
||||
if (!(cb.store_ref_bool(acc.total_state) // account_descr$_ account:^Account
|
||||
&& cb.store_bits_bool(acc.last_trans_hash_) // last_trans_hash:bits256
|
||||
&& cb.store_long_bool(acc.last_trans_lt_, 64) // last_trans_lt:uint64
|
||||
&& account_dict_estimator_->set_builder(acc.addr, cb))) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
++account_dict_ops_;
|
||||
if (!(account_dict_ops_ & 15)) {
|
||||
return block_limit_status_->add_proof(account_dict_estimator_->get_root_cell());
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new shard state and the Merkle update.
|
||||
*
|
||||
|
|
|
|||
Loading…
Reference in a new issue