Flat file block storage (#2180)

Signed-off-by: Konstantin Munichev <toobwn@gmail.com>

irohad/ametsuchi/CMakeLists.txt

@@ -20,6 +20,8 @@ add_library(ametsuchi
     impl/tx_presence_cache_impl.cpp
     impl/in_memory_block_storage.cpp
     impl/in_memory_block_storage_factory.cpp
+    impl/flat_file_block_storage.cpp
+    impl/flat_file_block_storage_factory.cpp
     )
 
 target_link_libraries(ametsuchi

irohad/ametsuchi/block_storage_factory.hpp

@@ -12,7 +12,6 @@
 
 namespace iroha {
   namespace ametsuchi {
-
     /**
      * Creates a block storage
      */

irohad/ametsuchi/impl/flat_file/flat_file.cpp

@@ -18,6 +18,7 @@
 
 using namespace iroha::ametsuchi;
 using Identifier = FlatFile::Identifier;
+using BlockIdCollectionType = FlatFile::BlockIdCollectionType;
 
 // ----------| public API |----------
 
@@ -27,29 +28,45 @@ std::string FlatFile::id_to_name(Identifier id) {
   return os.str();
 }
 
+boost::optional<Identifier> FlatFile::name_to_id(const std::string &name) {
+  if (name.size() != FlatFile::DIGIT_CAPACITY) {
+    return boost::none;
+  }
+  try {
+    auto id = std::stoul(name);
+    return boost::make_optional<Identifier>(id);
+  } catch (const std::exception &e) {
+    return boost::none;
+  }
+}
+
 boost::optional<std::unique_ptr<FlatFile>> FlatFile::create(
-    const std::string &path,
-    logger::LoggerPtr log) {
+    const std::string &path, logger::LoggerPtr log) {
   boost::system::error_code err;
   if (not boost::filesystem::is_directory(path, err)
       and not boost::filesystem::create_directory(path, err)) {
     log->error("Cannot create storage dir: {}\n{}", path, err.message());
     return boost::none;
   }
 
-  auto res = FlatFile::check_consistency(path, log);
-  return std::make_unique<FlatFile>(*res, path, private_tag{}, std::move(log));
+  BlockIdCollectionType files_found;
+  for (auto it = boost::filesystem::directory_iterator{path};
+       it != boost::filesystem::directory_iterator{};
+       ++it) {
+    if (auto id = FlatFile::name_to_id(it->path().filename().string())) {
+      files_found.insert(*id);
+    } else {
+      boost::filesystem::remove(it->path());
+    }
+  }
+
+  return std::make_unique<FlatFile>(
+      path, std::move(files_found), private_tag{}, std::move(log));
 }
 
 bool FlatFile::add(Identifier id, const Bytes &block) {
   // TODO(x3medima17): Change bool to generic Result return type
 
-  if (id != current_id_ + 1) {
-    log_->warn("Cannot append non-consecutive block");
-    return false;
-  }
-
-  auto next_id = id;
   const auto file_name = boost::filesystem::path{dump_dir_} / id_to_name(id);
 
   // Write block to binary file
@@ -71,8 +88,7 @@ bool FlatFile::add(Identifier id, const Bytes &block) {
   file.write(reinterpret_cast<const char *>(block.data()),
              block.size() * val_size);
 
-  // Update internals, release lock
-  current_id_ = next_id;
+  available_blocks_.insert(id);
   return true;
 }
 
@@ -100,50 +116,24 @@ std::string FlatFile::directory() const {
 }
 
 Identifier FlatFile::last_id() const {
-  return current_id_.load();
+  return (available_blocks_.empty()) ? 0 : *available_blocks_.rbegin();
 }
 
 void FlatFile::dropAll() {
   iroha::remove_dir_contents(dump_dir_, log_);
-  auto res = FlatFile::check_consistency(dump_dir_, log_);
-  current_id_.store(*res);
+  available_blocks_.clear();
+}
+
+const BlockIdCollectionType &FlatFile::blockIdentifiers() const {
+  return available_blocks_;
 }
 
 // ----------| private API |----------
 
-FlatFile::FlatFile(Identifier current_id,
-                   const std::string &path,
+FlatFile::FlatFile(std::string path,
+                   BlockIdCollectionType existing_files,
                    FlatFile::private_tag,
                    logger::LoggerPtr log)
-    : dump_dir_(path), log_{std::move(log)} {
-  current_id_.store(current_id);
-}
-
-boost::optional<Identifier> FlatFile::check_consistency(
-    const std::string &dump_dir, logger::LoggerPtr log) {
-  if (dump_dir.empty()) {
-    log->error("check_consistency({}), not directory", dump_dir);
-    return boost::none;
-  }
-
-  auto const files = [&dump_dir] {
-    std::vector<boost::filesystem::path> ps;
-    std::copy(boost::filesystem::directory_iterator{dump_dir},
-              boost::filesystem::directory_iterator{},
-              std::back_inserter(ps));
-    std::sort(ps.begin(), ps.end(), std::less<boost::filesystem::path>());
-    return ps;
-  }();
-
-  auto const missing = boost::range::find_if(
-      files | boost::adaptors::indexed(1), [](const auto &it) {
-        return FlatFile::id_to_name(it.index()) != it.value().filename();
-      });
-
-  std::for_each(
-      missing.get(), files.cend(), [](const boost::filesystem::path &p) {
-        boost::filesystem::remove(p);
-      });
-
-  return missing.get() - files.cbegin();
-}
+    : dump_dir_(std::move(path)),
+      available_blocks_(std::move(existing_files)),
+      log_{std::move(log)} {}

irohad/ametsuchi/impl/flat_file/flat_file.hpp

@@ -8,8 +8,8 @@
 
 #include "ametsuchi/key_value_storage.hpp"
 
-#include <atomic>
 #include <memory>
+#include <set>
 
 #include "logger/logger_fwd.hpp"
 
@@ -29,6 +29,8 @@ namespace iroha {
      public:
       // ----------| public API |----------
 
+      using BlockIdCollectionType = std::set<Identifier>;
+
       static const uint32_t DIGIT_CAPACITY = 16;
 
       /**
@@ -45,6 +47,13 @@ namespace iroha {
        */
       static std::string id_to_name(Identifier id);
 
+      /**
+       * Converts aligned string (see above) to number.
+       * @param name - name to convert
+       * @return id or boost::none
+       */
+      static boost::optional<Identifier> name_to_id(const std::string &name);
+
       /**
        * Create storage in paths
        * @param path - target path for creating
@@ -62,18 +71,12 @@ namespace iroha {
 
       Identifier last_id() const override;
 
+      void dropAll() override;
+
       /**
-       * Checking consistency of storage for provided folder
-       * If some block in the middle is missing all blocks following it are
-       * deleted
-       * @param dump_dir - folder of storage
-       * @param log - log for local messages
-       * @return - last available identifier
+       * @return collection of available block ids
        */
-      static boost::optional<Identifier> check_consistency(
-          const std::string &dump_dir, logger::LoggerPtr log);
-
-      void dropAll() override;
+      const BlockIdCollectionType &blockIdentifiers() const;
 
       // ----------| modify operations |----------
 
@@ -88,29 +91,24 @@ namespace iroha {
       // ----------| private API |----------
 
       /**
-       * Create storage in path with respect to last key
-       * @param last_id - maximal key written in storage
+       * Create storage in path
        * @param path - folder of storage
+       * @param existing_files - collection of existing files names
        * @param log to print progress
        */
-      FlatFile(Identifier last_id,
-               const std::string &path,
+      FlatFile(std::string path,
+               BlockIdCollectionType existing_files,
                FlatFile::private_tag,
                logger::LoggerPtr log);
 
      private:
-      // ----------| private fields |----------
-
-      /**
-       * Last written key
-       */
-      std::atomic<Identifier> current_id_;
-
       /**
        * Folder of storage
        */
       const std::string dump_dir_;
 
+      BlockIdCollectionType available_blocks_;
+
       logger::LoggerPtr log_;
 
      public:

irohad/ametsuchi/impl/flat_file_block_storage.cpp

@@ -0,0 +1,85 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "ametsuchi/impl/flat_file_block_storage.hpp"
+
+#include <boost/filesystem.hpp>
+
+#include "backend/protobuf/block.hpp"
+#include "common/byteutils.hpp"
+#include "logger/logger.hpp"
+
+using namespace iroha::ametsuchi;
+
+FlatFileBlockStorage::FlatFileBlockStorage(
+    std::unique_ptr<FlatFile> flat_file,
+    std::shared_ptr<shared_model::interface::BlockJsonConverter> json_converter,
+    logger::LoggerPtr log)
+    : flat_file_storage_(std::move(flat_file)),
+      json_converter_(std::move(json_converter)),
+      log_(std::move(log)) {}
+
+FlatFileBlockStorage::~FlatFileBlockStorage() {
+  log_->info("Remove {} temp directory", flat_file_storage_->directory());
+  boost::filesystem::remove_all(flat_file_storage_->directory());
+}
+
+bool FlatFileBlockStorage::insert(
+    std::shared_ptr<const shared_model::interface::Block> block) {
+  return json_converter_->serialize(*block).match(
+      [&](const expected::Value<std::string> &block_json) {
+        return flat_file_storage_->add(block->height(),
+                                       stringToBytes(block_json.value));
+      },
+      [this](const auto &error) {
+        log_->warn("Error while block serialization: {}", error.error);
+        return false;
+      });
+}
+
+bool FlatFileBlockStorage::insert(const shared_model::interface::Block &block) {
+  return insert(clone(block));
+}
+
+boost::optional<std::shared_ptr<const shared_model::interface::Block>>
+FlatFileBlockStorage::fetch(
+    shared_model::interface::types::HeightType height) const {
+  auto storage_block = flat_file_storage_->get(height);
+  if (not storage_block) {
+    return boost::none;
+  }
+
+  return json_converter_->deserialize(bytesToString(*storage_block))
+      .match(
+          [&](expected::Value<std::unique_ptr<shared_model::interface::Block>>
+                  &block) {
+            return boost::make_optional<
+                std::shared_ptr<const shared_model::interface::Block>>(
+                std::move(block.value));
+          },
+          [&](expected::Error<std::string> &error)
+              -> boost::optional<
+                  std::shared_ptr<const shared_model::interface::Block>> {
+            log_->warn("Error while block deserialization: {}", error.error);
+            return boost::none;
+          });
+}
+
+size_t FlatFileBlockStorage::size() const {
+  return flat_file_storage_->blockIdentifiers().size();
+}
+
+void FlatFileBlockStorage::clear() {
+  flat_file_storage_->dropAll();
+}
+
+void FlatFileBlockStorage::forEach(
+    iroha::ametsuchi::BlockStorage::FunctionType function) const {
+  for (auto block_id : flat_file_storage_->blockIdentifiers()) {
+    auto block = fetch(block_id);
+    BOOST_ASSERT(block);
+    function(*block);
+  }
+}

irohad/ametsuchi/impl/flat_file_block_storage.hpp

@@ -0,0 +1,50 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#ifndef IROHA_FLAT_FILE_BLOCK_STORAGE_HPP
+#define IROHA_FLAT_FILE_BLOCK_STORAGE_HPP
+
+#include "ametsuchi/block_storage.hpp"
+
+#include "ametsuchi/impl/flat_file/flat_file.hpp"
+#include "interfaces/iroha_internal/block_json_converter.hpp"
+#include "logger/logger_fwd.hpp"
+
+namespace iroha {
+  namespace ametsuchi {
+    class FlatFileBlockStorage : public BlockStorage {
+     public:
+      FlatFileBlockStorage(
+          std::unique_ptr<FlatFile> flat_file,
+          std::shared_ptr<shared_model::interface::BlockJsonConverter>
+              json_converter,
+          logger::LoggerPtr log);
+
+      ~FlatFileBlockStorage() override;
+
+      bool insert(
+          std::shared_ptr<const shared_model::interface::Block> block) override;
+
+      bool insert(const shared_model::interface::Block &block) override;
+
+      boost::optional<std::shared_ptr<const shared_model::interface::Block>>
+      fetch(shared_model::interface::types::HeightType height) const override;
+
+      size_t size() const override;
+
+      void clear() override;
+
+      void forEach(FunctionType function) const override;
+
+     private:
+      std::unique_ptr<FlatFile> flat_file_storage_;
+      std::shared_ptr<shared_model::interface::BlockJsonConverter>
+          json_converter_;
+      logger::LoggerPtr log_;
+    };
+  }  // namespace ametsuchi
+}  // namespace iroha
+
+#endif  // IROHA_FLAT_FILE_BLOCK_STORAGE_HPP