Fixes for miner (#599)

* Miner Fixes

Signed-off-by: ortyomka <[email protected]>

Author: ortyomka

cmake/dependencies.cmake

@@ -5,9 +5,6 @@
 hunter_add_package(GTest)
 find_package(GTest CONFIG REQUIRED)
 
-hunter_add_package(libarchive)
-find_package(libarchive CONFIG REQUIRED)
-
 # https://docs.hunter.sh/en/latest/packages/pkg/Boost.html
 hunter_add_package(Boost COMPONENTS date_time filesystem iostreams random program_options thread)
 find_package(Boost CONFIG REQUIRED date_time filesystem iostreams random program_options thread)
@@ -72,5 +69,8 @@ find_package(RapidJSON CONFIG REQUIRED)
 hunter_add_package(jwt-cpp)
 find_package(jwt-cpp CONFIG REQUIRED)
 
+hunter_add_package(libarchive)
+find_package(libarchive CONFIG REQUIRED)
+
 hunter_add_package(prometheus-cpp)
 find_package(prometheus-cpp CONFIG REQUIRED)

core/common/put_in_function.hpp

@@ -0,0 +1,25 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+namespace fc::common {
+
+  /**
+   * Class to put move-only lambdas in std::function interface
+   */
+  template <typename T>
+  class PutInFunction {
+   public:
+    explicit PutInFunction(T &&function)
+        : move_function_{std::make_shared<T>(std::move(function))} {};
+
+    template <typename... Args>
+    auto operator()(Args &&...args) const {
+      return move_function_->operator()(std::forward<Args>(args)...);
+    }
+
+   private:
+    std::shared_ptr<T> move_function_;
+  };
+}  // namespace fc::common

core/common/vector_cow.hpp

@@ -0,0 +1,69 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <gsl/span>
+#include <variant>
+
+namespace fc {
+
+  template <typename T>
+  class VectorCoW {
+   public:
+    using VectorType = std::vector<T>;
+    using SpanType = gsl::span<const T>;
+
+    VectorCoW() = default;
+    // NOLINTNEXTLINE(google-explicit-constructor)
+    VectorCoW(VectorType &&vector) : variant_{std::move(vector)} {}
+    VectorCoW(const VectorType &vector) = delete;
+    // NOLINTNEXTLINE(google-explicit-constructor)
+    VectorCoW(const SpanType &span) : variant_{span} {}
+    VectorCoW(const VectorCoW<T> &) = delete;
+    VectorCoW(VectorCoW<T> &&) = default;
+    VectorCoW<T> &operator=(const VectorCoW<T> &) = delete;
+    VectorCoW<T> &operator=(VectorCoW<T> &&) = default;
+    ~VectorCoW() = default;
+
+    bool owned() const {
+      return variant_.index() == 1;
+    }
+
+    SpanType span() const {
+      if (!owned()) {
+        return std::get<SpanType>(variant_);
+      }
+      return SpanType{std::get<VectorType>(variant_)};
+    }
+
+    size_t size() const {
+      return span().size();
+    }
+
+    bool empty() const {
+      return span().empty();
+    }
+
+    // get mutable container reference, copy once if span
+    VectorType &mut() {
+      if (!owned()) {
+        const auto span = std::get<SpanType>(variant_);
+        variant_.template emplace<VectorType>(span.begin(), span.end());
+      }
+      return std::get<VectorType>(variant_);
+    }
+
+    // move container away, copy once if span
+    VectorType into() {
+      auto vector{std::move(mut())};
+      variant_.template emplace<SpanType>();
+      return vector;
+    }
+
+   private:
+    std::variant<SpanType, VectorType> variant_;
+  };
+}  // namespace fc

core/markets/storage/provider/impl/provider_impl.cpp

@@ -681,10 +681,7 @@ namespace fc::markets::storage::provider {
                                 deal_context->deal->is_fast_retrieval});
     FSM_HALT_ON_ERROR(
         maybe_piece_location, "Unable to locate piece", deal_context);
-    FSM_HALT_ON_ERROR(
-        recordPieceInfo(deal_context->deal, maybe_piece_location.value()),
-        "Record piece failed",
-        deal_context);
+    deal_context->maybe_piece_location = maybe_piece_location.value();
     // TODO(a.chernyshov): add piece retry
     FSM_SEND(deal_context, ProviderEvent::ProviderEventDealHandedOff);
   }
@@ -700,6 +697,17 @@ namespace fc::markets::storage::provider {
   }
 
   FSM_HANDLE_DEFINITION(StorageProviderImpl::onProviderEventDealActivated) {
+    if (not deal_context->maybe_piece_location.has_value()) {
+      (deal_context)->deal->message = "Unknown piece location";
+      FSM_SEND((deal_context), ProviderEvent::ProviderEventFailed);
+      return;
+    }
+
+    FSM_HALT_ON_ERROR(
+        recordPieceInfo(deal_context->deal,
+                        deal_context->maybe_piece_location.value()),
+        "Record piece failed",
+        deal_context);
     // TODO(a.chernyshov): wait expiration
   }
 

core/markets/storage/provider/impl/provider_impl.hpp

@@ -90,6 +90,7 @@ namespace fc::markets::storage::provider {
     struct DealContext {
       std::shared_ptr<MinerDeal> deal;
       std::string protocol;
+      boost::optional<PieceLocation> maybe_piece_location{};
     };
 
     using ProviderTransition =
@@ -116,7 +117,6 @@ namespace fc::markets::storage::provider {
               if (request) {
                 if (auto asker{stored_ask.lock()}) {
                   if (auto ask{asker->getAsk(request.value().miner)}) {
-
                     return stream->write(AskResponseType{{ask.value()}},
                                          [stream](auto) { stream->close(); });
                   }

core/miner/storage_fsm/impl/sealing_impl.cpp

@@ -251,12 +251,13 @@ namespace fc::mining {
       auto sector_ref = minerSector(seal_proof_type, piece_location.sector);
 
       if (sector_and_padding.padding.size != 0) {
-        OUTCOME_TRY(
-            sealer_->addPieceSync(sector_ref,
-                                  unsealed_sector.piece_sizes,
-                                  sector_and_padding.padding.size.unpadded(),
-                                  PieceData::makeNull(),
-                                  kDealSectorPriority));
+        OUTCOME_TRY(sealer_->addPieceSync(
+            sector_ref,
+            VectorCoW(gsl::span<const UnpaddedPieceSize>(
+                unsealed_sector.piece_sizes)),
+            sector_and_padding.padding.size.unpadded(),
+            PieceData::makeNull(),
+            kDealSectorPriority));
 
         unsealed_sector.stored += sector_and_padding.padding.size;
 
@@ -283,7 +284,9 @@ namespace fc::mining {
       logger_->info("Add piece to sector {}", piece_location.sector);
       OUTCOME_TRY(piece_info,
                   sealer_->addPieceSync(sector_ref,
-                                        unsealed_sector.piece_sizes,
+                                        VectorCoW(
+                                            gsl::span<const UnpaddedPieceSize>(
+                                                unsealed_sector.piece_sizes)),
                                         size,
                                         std::move(piece_data),
                                         kDealSectorPriority));
@@ -972,7 +975,7 @@ namespace fc::mining {
 
     sealer_->addPiece(
         sector,
-        existing_piece_sizes,
+        VectorCoW(std::move(existing_piece_sizes)),
         filler,
         PieceData::makeNull(),
         [fill = std::move(result), cb](const auto &maybe_error) -> void {

core/sector_storage/impl/local_worker.cpp

@@ -18,11 +18,13 @@
 #include <thread>
 #include <utility>
 #include "api/storage_miner/storage_api.hpp"
+#include "common/put_in_function.hpp"
 #include "primitives/rle_bitset/runs_utils.hpp"
 #include "primitives/sector_file/sector_file.hpp"
 #include "sector_storage/stores/store_error.hpp"
 
 namespace fc::sector_storage {
+  using common::PutInFunction;
   using primitives::piece::PaddedByteIndex;
   using primitives::piece::PaddedPieceSize;
   using primitives::sector_file::SectorFile;
@@ -270,20 +272,22 @@ namespace fc::sector_storage {
 
   outcome::result<CallId> LocalWorker::addPiece(
       const SectorRef &sector,
-      gsl::span<const UnpaddedPieceSize> piece_sizes,
+      VectorCoW<UnpaddedPieceSize> piece_sizes,
       const UnpaddedPieceSize &new_piece_size,
       proofs::PieceData piece_data) {
     return asyncCall(
         sector,
         return_->ReturnAddPiece,
-        [=, piece_data{std::make_shared<PieceData>(std::move(piece_data))}](
-            Self self) -> outcome::result<PieceInfo> {
+        PutInFunction([=,
+                       exist_sizes = std::move(piece_sizes),
+                       data = std::move(piece_data)](
+                          const Self &self) -> outcome::result<PieceInfo> {
           OUTCOME_TRY(max_size,
                       primitives::sector::getSectorSize(sector.proof_type));
 
           UnpaddedPieceSize offset;
 
-          for (const auto &piece_size : piece_sizes) {
+          for (const auto &piece_size : exist_sizes.span()) {
             offset += piece_size;
           }
 
@@ -299,7 +303,7 @@ namespace fc::sector_storage {
             }
           });
 
-          if (piece_sizes.empty()) {
+          if (exist_sizes.empty()) {
             OUTCOME_TRYA(acquire_response,
                          self->acquireSector(sector,
                                              SectorFileType::FTNone,
@@ -322,13 +326,13 @@ namespace fc::sector_storage {
           }
 
           OUTCOME_TRY(piece_info,
-                      staged_file->write(*piece_data,
+                      staged_file->write(data,
                                          offset.padded(),
                                          new_piece_size.padded(),
                                          sector.proof_type));
 
           return piece_info.value();
-        });
+        }));
   }
 
   outcome::result<CallId> LocalWorker::sealPreCommit1(
@@ -461,11 +465,12 @@ namespace fc::sector_storage {
   }
 
   outcome::result<CallId> LocalWorker::finalizeSector(
-      const SectorRef &sector, const gsl::span<const Range> &keep_unsealed) {
+      const SectorRef &sector, std::vector<Range> keep_unsealed) {
     return asyncCall(
         sector,
         return_->ReturnFinalizeSector,
-        [=](Self self) -> outcome::result<void> {
+        [=, keep_unsealed{std::move(keep_unsealed)}](
+            Self self) -> outcome::result<void> {
           OUTCOME_TRY(size,
                       primitives::sector::getSectorSize(sector.proof_type));
           {
@@ -796,17 +801,16 @@ namespace fc::sector_storage {
                 }
               });
 
-              OUTCOME_TRY(self->proofs_->unsealRange(
-                  sector.proof_type,
-                  response.paths.cache,
-                  sealed,
-                  PieceData(fds[1]),
-                  sector.id.sector,
-                  sector.id.miner,
-                  randomness,
-                  unsealed_cid,
-                  primitives::piece::paddedIndex(range.offset),
-                  range.size));
+              OUTCOME_TRY(self->proofs_->unsealRange(sector.proof_type,
+                                                     response.paths.cache,
+                                                     sealed,
+                                                     PieceData(fds[1]),
+                                                     sector.id.sector,
+                                                     sector.id.miner,
+                                                     randomness,
+                                                     unsealed_cid,
+                                                     range.offset,
+                                                     range.size));
             }
 
             for (auto &t : threads) {

core/sector_storage/impl/local_worker.hpp

@@ -29,11 +29,10 @@ namespace fc::sector_storage {
                 std::shared_ptr<proofs::ProofEngine> proofs =
                     std::make_shared<proofs::ProofEngineImpl>());
 
-    outcome::result<CallId> addPiece(
-        const SectorRef &sector,
-        gsl::span<const UnpaddedPieceSize> piece_sizes,
-        const UnpaddedPieceSize &new_piece_size,
-        PieceData piece_data) override;
+    outcome::result<CallId> addPiece(const SectorRef &sector,
+                                     VectorCoW<UnpaddedPieceSize> piece_sizes,
+                                     const UnpaddedPieceSize &new_piece_size,
+                                     PieceData piece_data) override;
 
     outcome::result<CallId> sealPreCommit1(
         const SectorRef &sector,
@@ -54,8 +53,7 @@ namespace fc::sector_storage {
         const SectorRef &sector, const Commit1Output &commit_1_output) override;
 
     outcome::result<CallId> finalizeSector(
-        const SectorRef &sector,
-        const gsl::span<const Range> &keep_unsealed) override;
+        const SectorRef &sector, std::vector<Range> keep_unsealed) override;
 
     outcome::result<CallId> replicaUpdate(
         const SectorRef &sector, const std::vector<PieceInfo> &pieces) override;