Add option to open DB in secondary mode

tddb/CMakeLists.txt

@@ -37,6 +37,9 @@ set(TDDB_SOURCE
 set(TDDB_ROCKSDB_SOURCE
   td/db/RocksDb.cpp
   td/db/RocksDb.h
+
+  td/db/RocksDbSecondary.cpp
+  td/db/RocksDbSecondary.h
 )
 
 set(TDDB_TEST_SOURCE

tddb/td/db/RocksDbSecondary.cpp

@@ -0,0 +1,198 @@
+/*
+    This file is part of TON Blockchain Library.
+    TON Blockchain Library is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Lesser General Public License as published by
+    the Free Software Foundation, either version 2 of the License, or
+    (at your option) any later version.
+    TON Blockchain Library is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Lesser General Public License for more details.
+    You should have received a copy of the GNU Lesser General Public License
+    along with TON Blockchain Library.  If not, see <http://www.gnu.org/licenses/>.
+    Copyright 2017-2020 Telegram Systems LLP
+*/
+#include "td/db/RocksDbSecondary.h"
+
+#include "rocksdb/db.h"
+#include "rocksdb/table.h"
+#include "rocksdb/statistics.h"
+#include "rocksdb/write_batch.h"
+#include "rocksdb/utilities/transaction.h"
+
+namespace td {
+namespace {
+static Status from_rocksdb(rocksdb::Status status) {
+  if (status.ok()) {
+    return Status::OK();
+  }
+  return Status::Error(status.ToString());
+}
+static Slice from_rocksdb(rocksdb::Slice slice) {
+  return Slice(slice.data(), slice.size());
+}
+static rocksdb::Slice to_rocksdb(Slice slice) {
+  return rocksdb::Slice(slice.data(), slice.size());
+}
+}  // namespace
+
+Status RocksDbSecondary::destroy(Slice path) {
+  return from_rocksdb(rocksdb::DestroyDB(path.str(), {}));
+}
+
+RocksDbSecondary::RocksDbSecondary(RocksDbSecondary &&) = default;
+RocksDbSecondary &RocksDbSecondary::operator=(RocksDbSecondary &&) = default;
+
+RocksDbSecondary::~RocksDbSecondary() {
+  if (!db_) {
+    return;
+  }
+  end_snapshot().ensure();
+}
+
+RocksDbSecondary RocksDbSecondary::clone() const {
+  return RocksDbSecondary{db_, statistics_};
+}
+
+Result<RocksDbSecondary> RocksDbSecondary::open(std::string path) {
+  rocksdb::DB *db;
+  auto statistics = rocksdb::CreateDBStatistics();
+  {
+    rocksdb::Options options;
+
+    static auto cache = rocksdb::NewLRUCache(1 << 30);
+
+    rocksdb::BlockBasedTableOptions table_options;
+    table_options.block_cache = cache;
+    options.table_factory.reset(rocksdb::NewBlockBasedTableFactory(table_options));
+
+    options.manual_wal_flush = true;
+    options.create_if_missing = true;
+    options.max_background_compactions = 4;
+    options.max_background_flushes = 2;
+    options.bytes_per_sync = 1 << 20;
+    options.writable_file_max_buffer_size = 2 << 14;
+    options.keep_log_file_num = 1;
+    options.statistics = statistics;
+    rocksdb::ColumnFamilyOptions cf_options(options);
+    std::vector<rocksdb::ColumnFamilyDescriptor> column_families;
+    column_families.push_back(rocksdb::ColumnFamilyDescriptor(rocksdb::kDefaultColumnFamilyName, cf_options));
+    std::vector<rocksdb::ColumnFamilyHandle *> handles;
+    TRY_STATUS(from_rocksdb(
+      rocksdb::DB::OpenAsSecondary(options, path, path + "/secondary", column_families, &handles, &db)));
+    CHECK(handles.size() == 1);
+    // i can delete the handle since DBImpl is always holding a reference to
+    // default column family
+    delete handles[0];
+  }
+  TRY_STATUS(from_rocksdb(db->TryCatchUpWithPrimary()));
+  return RocksDbSecondary(std::shared_ptr<rocksdb::DB>(db), std::move(statistics));
+}
+
+Status RocksDbSecondary::try_catch_up_with_primary() {
+  return from_rocksdb(db_->TryCatchUpWithPrimary());
+}
+
+std::unique_ptr<KeyValueReader> RocksDbSecondary::snapshot() {
+  auto res = std::make_unique<RocksDbSecondary>(clone());
+  res->begin_snapshot().ensure();
+  return std::move(res);
+}
+
+std::string RocksDbSecondary::stats() const {
+  std::string out;
+  db_->GetProperty("rocksdb.stats", &out);
+  //db_->GetProperty("rocksdb.cur-size-all-mem-tables", &out);
+  return out;
+}
+
+Result<RocksDbSecondary::GetStatus> RocksDbSecondary::get(Slice key, std::string &value) {
+  rocksdb::Status status;
+  if (snapshot_) {
+    rocksdb::ReadOptions options;
+    options.snapshot = snapshot_.get();
+    status = db_->Get(options, to_rocksdb(key), &value);
+  } else {
+    status = db_->Get({}, to_rocksdb(key), &value);
+  }
+  if (status.ok()) {
+    return GetStatus::Ok;
+  }
+  if (status.code() == rocksdb::Status::kNotFound) {
+    return GetStatus::NotFound;
+  }
+  return from_rocksdb(status);
+}
+
+Status RocksDbSecondary::set(Slice key, Slice value) {
+  CHECK(false)
+}
+
+Status RocksDbSecondary::erase(Slice key) {
+  CHECK(false)
+}
+
+Result<size_t> RocksDbSecondary::count(Slice prefix) {
+  rocksdb::ReadOptions options;
+  options.snapshot = snapshot_.get();
+  std::unique_ptr<rocksdb::Iterator> iterator;
+
+  iterator.reset(db_->NewIterator(options));
+
+  size_t res = 0;
+  for (iterator->Seek(to_rocksdb(prefix)); iterator->Valid(); iterator->Next()) {
+    if (from_rocksdb(iterator->key()).truncate(prefix.size()) != prefix) {
+      break;
+    }
+    res++;
+  }
+  if (!iterator->status().ok()) {
+    return from_rocksdb(iterator->status());
+  }
+  return res;
+}
+
+Status RocksDbSecondary::begin_write_batch() {
+  CHECK(false)
+}
+
+Status RocksDbSecondary::begin_transaction() {
+  CHECK(false)
+}
+
+Status RocksDbSecondary::commit_write_batch() {
+  CHECK(false)
+}
+
+Status RocksDbSecondary::commit_transaction() {
+  CHECK(false)
+}
+
+Status RocksDbSecondary::abort_write_batch() {
+  CHECK(false)
+}
+
+Status RocksDbSecondary::abort_transaction() {
+  CHECK(false)
+}
+
+Status RocksDbSecondary::flush() {
+  CHECK(false)
+}
+
+Status RocksDbSecondary::begin_snapshot() {
+  snapshot_.reset(db_->GetSnapshot());
+  return td::Status::OK();
+}
+
+Status RocksDbSecondary::end_snapshot() {
+  if (snapshot_) {
+    db_->ReleaseSnapshot(snapshot_.release());
+  }
+  return td::Status::OK();
+}
+
+RocksDbSecondary::RocksDbSecondary(std::shared_ptr<rocksdb::DB> db, std::shared_ptr<rocksdb::Statistics> statistics)
+    : db_(std::move(db)), statistics_(std::move(statistics)) {
+}
+}  // namespace td

tddb/td/db/RocksDbSecondary.h

@@ -0,0 +1,85 @@
+/*
+    This file is part of TON Blockchain Library.
+    TON Blockchain Library is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Lesser General Public License as published by
+    the Free Software Foundation, either version 2 of the License, or
+    (at your option) any later version.
+    TON Blockchain Library is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Lesser General Public License for more details.
+    You should have received a copy of the GNU Lesser General Public License
+    along with TON Blockchain Library.  If not, see <http://www.gnu.org/licenses/>.
+    Copyright 2017-2020 Telegram Systems LLP
+*/
+#pragma once
+
+#if !TDDB_USE_ROCKSDB
+#error "RocksDb is not supported"
+#endif
+
+#include "td/db/KeyValue.h"
+#include "td/utils/Status.h"
+
+namespace rocksdb {
+class DB;
+class Transaction;
+class WriteBatch;
+class Snapshot;
+class Statistics;
+}  // namespace rocksdb
+
+namespace td {
+class RocksDbSecondary : public KeyValue {
+ public:
+  static Status destroy(Slice path);
+  RocksDbSecondary clone() const;
+  static Result<RocksDbSecondary> open(std::string path);
+
+  Status try_catch_up_with_primary();
+
+  Result<GetStatus> get(Slice key, std::string &value) override;
+  Status set(Slice key, Slice value) override;
+  Status erase(Slice key) override;
+  Result<size_t> count(Slice prefix) override;
+
+  Status begin_write_batch() override;
+  Status commit_write_batch() override;
+  Status abort_write_batch() override;
+
+  Status begin_transaction() override;
+  Status commit_transaction() override;
+  Status abort_transaction() override;
+  Status flush() override;
+
+  Status begin_snapshot();
+  Status end_snapshot();
+
+  std::unique_ptr<KeyValueReader> snapshot() override;
+  std::string stats() const override;
+
+  RocksDbSecondary(RocksDbSecondary &&);
+  RocksDbSecondary &operator=(RocksDbSecondary &&);
+  ~RocksDbSecondary();
+
+  std::shared_ptr<rocksdb::DB> raw_db() const {
+    return db_;
+  };
+
+ private:
+  std::shared_ptr<rocksdb::DB> db_;
+  std::shared_ptr<rocksdb::Statistics> statistics_;
+
+  class UnreachableDeleter {
+   public:
+    template <class T>
+    void operator()(T *) {
+      UNREACHABLE();
+    }
+  };
+  std::unique_ptr<const rocksdb::Snapshot, UnreachableDeleter> snapshot_;
+
+  explicit RocksDbSecondary(std::shared_ptr<rocksdb::DB> db,
+                   std::shared_ptr<rocksdb::Statistics> statistics);
+};
+}  // namespace td