Parallel apply block in ArchiveImporterLocal

Author: SpyCheese

crypto/common/refcnt.hpp

@@ -285,14 +285,14 @@ class Ref {
   Ref& operator=(Ref<S>&& r);
   const typename RefValue<T>::Type* operator->() const {
     if (!ptr) {
-      CHECK(ptr && "deferencing null Ref");
+      LOG_CHECK(ptr) << "dereferencing null Ref<" << typeid(T).name() << ">";
       throw NullRef{};
     }
     return RefValue<T>::make_const_ptr(ptr);
   }
   const typename RefValue<T>::Type& operator*() const {
     if (!ptr) {
-      CHECK(ptr && "deferencing null Ref");
+      LOG_CHECK(ptr) << "dereferencing null Ref<" << typeid(T).name() << ">";
       throw NullRef{};
     }
     return RefValue<T>::make_const_ref(ptr);

validator/db/archive-manager.cpp

@@ -657,9 +657,13 @@ void ArchiveManager::load_package(PackageId id) {
 
 td::actor::ActorOwn<ArchiveSlice> ArchiveManager::create_archive_slice(const PackageId &id,
                                                                        td::uint32 shard_split_depth) {
-  return td::actor::create_actor<ArchiveSlice>(
+  auto actor = td::actor::create_actor<ArchiveSlice>(
       PSTRING() << "slice." << (id.temp ? "temp." : (id.key ? "key." : "")) << id.id, id.id, id.key, id.temp, false,
       shard_split_depth, db_root_, archive_lru_.get(), statistics_);
+  if (async_mode_) {
+    td::actor::send_closure(actor, &ArchiveSlice::set_async_mode, true, [](td::Result<td::Unit>) {});
+  }
+  return actor;
 }
 
 const ArchiveManager::FileDescription *ArchiveManager::get_file_desc(ShardIdFull shard, PackageId id, BlockSeqno seqno,
@@ -1199,23 +1203,8 @@ void ArchiveManager::get_archive_slice(td::uint64 archive_id, td::uint64 offset,
   td::actor::send_closure(F->file_actor_id(), &ArchiveSlice::get_slice, archive_id, offset, limit, std::move(promise));
 }
 
-void ArchiveManager::commit_transaction() {
-  if (!async_mode_ || huge_transaction_size_++ >= 100) {
-    index_->commit_transaction().ensure();
-    if (async_mode_) {
-      huge_transaction_size_ = 0;
-      huge_transaction_started_ = false;
-    }
-  }
-}
-
 void ArchiveManager::set_async_mode(bool mode, td::Promise<td::Unit> promise) {
   async_mode_ = mode;
-  if (!async_mode_ && huge_transaction_started_) {
-    index_->commit_transaction().ensure();
-    huge_transaction_size_ = 0;
-    huge_transaction_started_ = false;
-  }
 
   td::MultiPromise mp;
   auto ig = mp.init_guard();

validator/db/archive-manager.hpp

@@ -76,7 +76,6 @@ class ArchiveManager : public td::actor::Actor {
   void start_up() override;
   void alarm() override;
 
-  void commit_transaction();
   void set_async_mode(bool mode, td::Promise<td::Unit> promise);
 
   void set_current_shard_split_depth(td::uint32 value) {
@@ -183,8 +182,6 @@ class ArchiveManager : public td::actor::Actor {
   td::actor::ActorOwn<ArchiveLru> archive_lru_;
   BlockSeqno finalized_up_to_{0};
   bool async_mode_ = false;
-  bool huge_transaction_started_ = false;
-  td::uint32 huge_transaction_size_ = 0;
   td::uint32 cur_shard_split_depth_ = 0;
 
   DbStatistics statistics_;

validator/db/archive-slice.cpp

@@ -26,6 +26,9 @@
 #include "files-async.hpp"
 #include "db-utils.h"
 #include "td/utils/PathView.h"
+#include "td/utils/Random.h"
+
+#include <csignal>
 
 namespace ton {
 
@@ -107,6 +110,10 @@ std::string DbStatistics::to_string_and_reset() {
   return ss.str();
 }
 
+void PackageWriter::tear_down() {
+  sync_now();
+}
+
 void PackageWriter::append(std::string filename, td::BufferSlice data,
                            td::Promise<std::pair<td::uint64, td::uint64>> promise) {
   td::uint64 offset, size;
@@ -119,14 +126,45 @@ void PackageWriter::append(std::string filename, td::BufferSlice data,
       return;
     }
     start = td::Timestamp::now();
-    offset = p->append(std::move(filename), std::move(data), !async_mode_);
+    offset = p->append(std::move(filename), std::move(data), false);
     end = td::Timestamp::now();
     size = p->size();
   }
   if (statistics_) {
     statistics_->record_write((end.at() - start.at()) * 1e6, data_size);
   }
-  promise.set_value(std::pair<td::uint64, td::uint64>{offset, size});
+  sync([=, promise = std::move(promise)](td::Result<td::Unit>) mutable {
+    promise.set_value(std::pair<td::uint64, td::uint64>{offset, size});
+  });
+}
+
+void PackageWriter::sync(td::Promise<td::Unit> promise) {
+  if (!async_mode_) {
+    sync_now();
+    promise.set_value(td::Unit{});
+    return;
+  }
+  sync_waiters_.push_back(std::move(promise));
+  if (!sync_pending_) {
+    sync_pending_ = true;
+    td::actor::send_closure_later(actor_id(this), &PackageWriter::sync_now);
+  }
+}
+
+void PackageWriter::sync_now() {
+  auto p = package_.lock();
+  if (p) {
+    p->sync();
+  }
+  sync_pending_ = false;
+  auto waiters = std::move(sync_waiters_);
+  sync_waiters_.clear();
+  if (!waiters.empty()) {
+    LOG(DEBUG) << "Writer huge transaction size = " << waiters.size();
+  }
+  for (auto &promise: waiters) {
+    promise.set_value(td::Unit{});
+  }
 }
 
 class PackageReader : public td::actor::Actor {
@@ -168,6 +206,10 @@ static std::string package_info_to_str(BlockSeqno seqno, ShardIdFull shard_prefi
   return PSTRING() << seqno << "." << shard_prefix.workchain << ":" << shard_to_str(shard_prefix.shard);
 }
 
+void ArchiveSlice::tear_down() {
+  commit_transaction_now();
+}
+
 void ArchiveSlice::add_handle(BlockHandle handle, td::Promise<td::Unit> promise) {
   if (destroyed_) {
     promise.set_error(td::Status::Error(ErrorCode::notready, "package already gc'd"));
@@ -239,16 +281,17 @@ void ArchiveSlice::add_handle(BlockHandle handle, td::Promise<td::Unit> promise)
     kv_->set(shard_key.as_slice(), shard_value.as_slice()).ensure();
   }
   kv_->set(get_db_key_block_info(handle->id()), handle->serialize().as_slice()).ensure();
-  commit_transaction();
-
-  handle->flushed_upto(version);
-  handle->set_handle_moved_to_archive();
+  commit_transaction([=, this, promise = std::move(promise)](td::Result<td::Unit> R) mutable {
+    R.ensure();
+    handle->flushed_upto(version);
+    handle->set_handle_moved_to_archive();
 
-  if (handle->need_flush()) {
-    update_handle(std::move(handle), std::move(promise));
-  } else {
-    promise.set_value(td::Unit());
-  }
+    if (handle->need_flush()) {
+      update_handle(std::move(handle), std::move(promise));
+    } else {
+      promise.set_value(td::Unit());
+    }
+  });
 }
 
 void ArchiveSlice::update_handle(BlockHandle handle, td::Promise<td::Unit> promise) {
@@ -269,12 +312,13 @@ void ArchiveSlice::update_handle(BlockHandle handle, td::Promise<td::Unit> promi
     kv_->set(get_db_key_block_info(handle->id()), handle->serialize().as_slice()).ensure();
     handle->flushed_upto(version);
   } while (handle->need_flush());
-  commit_transaction();
-  if (!temp_) {
-    handle->set_handle_moved_to_archive();
-  }
-
-  promise.set_value(td::Unit());
+  commit_transaction([=, this, promise = std::move(promise)](td::Result<td::Unit> R) mutable {
+    R.ensure();
+    if (!temp_) {
+      handle->set_handle_moved_to_archive();
+    }
+    promise.set_value(td::Unit());
+  });
 }
 
 void ArchiveSlice::add_file(BlockHandle handle, FileReference ref_id, td::BufferSlice data,
@@ -326,7 +370,7 @@ void ArchiveSlice::add_file_cont(size_t idx, FileReference ref_id, td::uint64 of
     kv_->set("status", td::to_string(size)).ensure();
     kv_->set(ref_id.hash().to_hex(), td::to_string(offset)).ensure();
   }
-  commit_transaction();
+  commit_transaction(std::move(promise));
   promise.set_value(td::Unit());
 }
 
@@ -712,6 +756,7 @@ void ArchiveSlice::do_close() {
   if (destroyed_) {
     return;
   }
+  commit_transaction_now();
   CHECK(status_ != st_closed && active_queries_ == 0);
   LOG(DEBUG) << "Closing archive slice " << db_path_;
   status_ = st_closed;
@@ -741,30 +786,48 @@ void ArchiveSlice::end_async_query() {
 }
 
 void ArchiveSlice::begin_transaction() {
-  if (!async_mode_ || !huge_transaction_started_) {
-    kv_->begin_transaction().ensure();
-    if (async_mode_) {
-      huge_transaction_started_ = true;
-    }
+  if (transaction_started_) {
+    return;
   }
+  transaction_started_ = true;
+  kv_->begin_transaction().ensure();
 }
 
-void ArchiveSlice::commit_transaction() {
-  if (!async_mode_ || huge_transaction_size_++ >= 100) {
-    kv_->commit_transaction().ensure();
-    if (async_mode_) {
-      huge_transaction_size_ = 0;
-      huge_transaction_started_ = false;
-    }
+void ArchiveSlice::commit_transaction(td::Promise<td::Unit> promise) {
+  if (!transaction_started_ || !async_mode_) {
+    commit_transaction_now();
+    promise.set_value(td::Unit{});
+    return;
+  }
+  transaction_commit_waiters_.push_back(std::move(promise));
+  if (!transaction_commit_pending_) {
+    transaction_commit_pending_ = true;
+    td::actor::send_closure_later(actor_id(this), &ArchiveSlice::commit_transaction_now);
+  }
+}
+
+void ArchiveSlice::commit_transaction_now() {
+  if (!transaction_started_) {
+    return;
+  }
+  kv_->commit_transaction().ensure();
+  transaction_started_ = false;
+  transaction_commit_pending_ = false;
+  auto waiters = std::move(transaction_commit_waiters_);
+  transaction_commit_waiters_.clear();
+  if (!waiters.empty()) {
+    LOG(DEBUG) << "Huge transaction size = " << waiters.size();
+  }
+  for (auto &promise: waiters) {
+    promise.set_value(td::Unit{});
   }
 }
 
+
 void ArchiveSlice::set_async_mode(bool mode, td::Promise<td::Unit> promise) {
   async_mode_ = mode;
-  if (!async_mode_ && huge_transaction_started_ && kv_) {
-    kv_->commit_transaction().ensure();
-    huge_transaction_size_ = 0;
-    huge_transaction_started_ = false;
+  if (!async_mode_) {
+    commit_transaction_now();
   }
 
   td::MultiPromise mp;
@@ -820,7 +883,7 @@ td::Result<ArchiveSlice::PackageInfo *> ArchiveSlice::choose_package(BlockSeqno
     if (shard_separated_) {
       kv_->set(PSTRING() << "info." << v, package_info_to_str(masterchain_seqno, shard_prefix)).ensure();
     }
-    commit_transaction();
+    commit_transaction_now();
     add_package(masterchain_seqno, shard_prefix, 0, default_package_version());
     return &packages_[v];
   } else {
@@ -874,6 +937,7 @@ void destroy_db(std::string name, td::uint32 attempt, td::Promise<td::Unit> prom
 }  // namespace
 
 void ArchiveSlice::destroy(td::Promise<td::Unit> promise) {
+  commit_transaction_now();
   before_query();
   destroyed_ = true;
 
@@ -1081,7 +1145,8 @@ void ArchiveSlice::truncate(BlockSeqno masterchain_seqno, ConstBlockHandle, td::
     package->writer.reset();
     td::unlink(package->path).ensure();
     td::rename(package->path + ".new", package->path).ensure();
-    package->writer = td::actor::create_actor<PackageWriter>("writer", new_package, async_mode_);
+    package->writer = td::actor::create_actor<PackageWriter>(
+        PSTRING() << "writer." << td::PathView(package->path).file_name(), new_package, async_mode_);
   }
 
   std::vector<PackageInfo> new_packages_info;

validator/db/archive-slice.hpp

@@ -74,22 +74,25 @@ class PackageWriter : public td::actor::Actor {
       : package_(std::move(package)), async_mode_(async_mode), statistics_(std::move(statistics)) {
   }
 
+  void tear_down() override;
   void append(std::string filename, td::BufferSlice data, td::Promise<std::pair<td::uint64, td::uint64>> promise);
   void set_async_mode(bool mode, td::Promise<td::Unit> promise) {
     async_mode_ = mode;
     if (!async_mode_) {
-      auto p = package_.lock();
-      if (p) {
-        p->sync();
-      }
+      sync_now();
     }
     promise.set_value(td::Unit());
   }
 
  private:
   std::weak_ptr<Package> package_;
   bool async_mode_ = false;
+  bool sync_pending_ = false;
+  std::vector<td::Promise<td::Unit>> sync_waiters_;
   std::shared_ptr<PackageStatistics> statistics_;
+
+  void sync(td::Promise<td::Unit> promise);
+  void sync_now();
 };
 
 class ArchiveLru;
@@ -99,6 +102,8 @@ class ArchiveSlice : public td::actor::Actor {
   ArchiveSlice(td::uint32 archive_id, bool key_blocks_only, bool temp, bool finalized, td::uint32 shard_split_depth,
                std::string db_root, td::actor::ActorId<ArchiveLru> archive_lru, DbStatistics statistics = {});
 
+  void tear_down() override;
+
   void get_archive_id(BlockSeqno masterchain_seqno, ShardIdFull shard_prefix, td::Promise<td::uint64> promise);
 
   void add_handle(BlockHandle handle, td::Promise<td::Unit> promise);
@@ -137,7 +142,8 @@ class ArchiveSlice : public td::actor::Actor {
   void end_async_query();
 
   void begin_transaction();
-  void commit_transaction();
+  void commit_transaction(td::Promise<td::Unit> promise);
+  void commit_transaction_now();
 
   void add_file_cont(size_t idx, FileReference ref_id, td::uint64 offset, td::uint64 size,
                      td::Promise<td::Unit> promise);
@@ -157,9 +163,10 @@ class ArchiveSlice : public td::actor::Actor {
 
   bool destroyed_ = false;
   bool async_mode_ = false;
-  bool huge_transaction_started_ = false;
+  bool transaction_started_ = false;
+  std::vector<td::Promise<td::Unit>> transaction_commit_waiters_;
+  bool transaction_commit_pending_ = false;
   bool sliced_mode_{false};
-  td::uint32 huge_transaction_size_ = 0;
   td::uint32 slice_size_{100};
   bool shard_separated_{false};
   td::uint32 shard_split_depth_ = 0;