Estimator (#663)

* Add estimator

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

Author: ortyomka

core/miner/main/main.cpp

@@ -45,6 +45,7 @@
 #include "sector_storage/fetch_handler.hpp"
 #include "sector_storage/impl/manager_impl.hpp"
 #include "sector_storage/impl/scheduler_impl.hpp"
+#include "sector_storage/impl/worker_estimator_impl.hpp"
 #include "sector_storage/stores/impl/index_impl.hpp"
 #include "sector_storage/stores/impl/local_store.hpp"
 #include "sector_storage/stores/impl/remote_store.hpp"
@@ -95,6 +96,7 @@ namespace fc {
     boost::optional<RegisteredSealProof> seal_type;
     std::vector<Address> precommit_control;
     int api_port{};
+    uint64_t estimator_window{};
 
     /** Path to presealed sectors */
     boost::optional<boost::filesystem::path> preseal_path;
@@ -113,8 +115,9 @@ namespace fc {
       const std::shared_ptr<storage::PersistentBufferMap> &ds) {
     OUTCOME_TRY(file, common::readFile(path));
     OUTCOME_TRY(j_file, codec::json::parse(gsl::make_span(file)));
-    OUTCOME_TRY(
-        psm, codec::json::decode<std::map<std::string, miner::types::Miner>>(j_file));
+    OUTCOME_TRY(psm,
+                codec::json::decode<std::map<std::string, miner::types::Miner>>(
+                    j_file));
 
     const auto it_psm = psm.find(encodeToString(maddr));
     if (it_psm == psm.end()) {
@@ -155,6 +158,8 @@ namespace fc {
     option("owner", po::value(&config.owner));
     option("worker", po::value(&config.worker));
     option("sector-size", po::value(&raw.sector_size));
+    option("estimator-window",
+           po::value(&config.estimator_window)->default_value(10));
     option("precommit-control", po::value(&config.precommit_control));
     option("pre-sealed-sectors",
            po::value(&config.preseal_path),
@@ -425,7 +430,11 @@ namespace fc {
     IoThread io_thread2;
     OUTCOME_TRY(wscheduler,
                 sector_storage::SchedulerImpl::newScheduler(
-                    io_thread2.io, prefixed("scheduler_works/")));
+                    io_thread2.io,
+                    prefixed("scheduler_works/"),
+                    std::make_shared<sector_storage::EstimatorImpl>(
+                        config.estimator_window)));
+
     IoThread io_thread3;
 
     {

core/sector_storage/CMakeLists.txt

@@ -42,6 +42,14 @@ target_link_libraries(remote_worker
         rpc
         )
 
+add_library(estimator
+        impl/worker_estimator_impl.cpp
+        )
+
+target_link_libraries(estimator
+        Boost::boost
+        )
+
 add_library(scheduler
         impl/scheduler_impl.cpp
         )
@@ -51,6 +59,7 @@ target_link_libraries(scheduler
         resources
         logger
         worker
+        estimator
         Boost::thread
         )
 

core/sector_storage/impl/scheduler_impl.cpp

@@ -16,16 +16,20 @@ namespace fc::sector_storage {
 
   outcome::result<std::shared_ptr<SchedulerImpl>> SchedulerImpl::newScheduler(
       std::shared_ptr<boost::asio::io_context> io_context,
-      std::shared_ptr<BufferMap> datastore) {
+      std::shared_ptr<BufferMap> datastore,
+      std::shared_ptr<Estimator> estimator) {
     struct make_unique_enabler : public SchedulerImpl {
       make_unique_enabler(std::shared_ptr<boost::asio::io_context> io_context,
-                          std::shared_ptr<BufferMap> datastore)
-          : SchedulerImpl{std::move(io_context), std::move(datastore)} {};
+                          std::shared_ptr<BufferMap> datastore,
+                          std::shared_ptr<Estimator> estimator)
+          : SchedulerImpl{std::move(io_context),
+                          std::move(datastore),
+                          std::move(estimator)} {};
     };
 
     std::shared_ptr<SchedulerImpl> scheduler =
-        std::make_shared<make_unique_enabler>(std::move(io_context),
-                                              std::move(datastore));
+        std::make_shared<make_unique_enabler>(
+            std::move(io_context), std::move(datastore), std::move(estimator));
 
     OUTCOME_TRY(scheduler->resetWorks());
 
@@ -34,8 +38,10 @@ namespace fc::sector_storage {
 
   SchedulerImpl::SchedulerImpl(
       std::shared_ptr<boost::asio::io_context> io_context,
-      std::shared_ptr<BufferMap> datastore)
+      std::shared_ptr<BufferMap> datastore,
+      std::shared_ptr<Estimator> estimator)
       : current_worker_id_(0),
+        estimator_(std::move(estimator)),
         call_kv_(std::move(datastore)),
         io_(std::move(io_context)),
         logger_(common::createLogger("scheduler")) {}
@@ -143,7 +149,7 @@ namespace fc::sector_storage {
     if (current_worker_id_ == std::numeric_limits<uint64_t>::max()) {
       current_worker_id_ = 0;  // TODO(ortyomka): maybe better mechanism
     }
-    WorkerID wid = current_worker_id_++;
+    WorkerId wid = current_worker_id_++;
     workers_.insert({wid, std::move(worker)});
     lock.unlock();
 
@@ -154,7 +160,7 @@ namespace fc::sector_storage {
       const std::shared_ptr<TaskRequest> &request) {
     std::lock_guard<std::mutex> lock(workers_lock_);
 
-    std::vector<WorkerID> acceptable;
+    std::vector<WorkerId> acceptable;
     uint64_t tried = 0;
 
     for (const auto &[wid, worker] : workers_) {
@@ -179,27 +185,47 @@ namespace fc::sector_storage {
 
     if (!acceptable.empty()) {
       bool does_error_occurs = false;
-      std::stable_sort(acceptable.begin(),
-                       acceptable.end(),
-                       [&](WorkerID lhs, WorkerID rhs) {
-                         auto maybe_res = request->sel->is_preferred(
-                             request->task_type, workers_[lhs], workers_[rhs]);
-
-                         if (maybe_res.has_error()) {
-                           logger_->error("selecting best worker: "
-                                          + maybe_res.error().message());
-                           does_error_occurs = true;
-                           return false;
-                         }
-
-                         return maybe_res.value();
-                       });
+      std::stable_sort(
+          acceptable.begin(),
+          acceptable.end(),
+          [&](WorkerId lhs, WorkerId rhs) {
+            const auto l_time = estimator_->getTime(lhs, request->task_type);
+            const auto r_time = estimator_->getTime(rhs, request->task_type);
+
+            // if time is available, then compare them
+            if (l_time and r_time) {
+              return l_time < r_time;
+            }
+
+            // if some workers don't have data about time, then we prefer worker
+            // without time, to give a chance to prove yourself
+            if (l_time) {
+              return false;
+            }
+
+            if (r_time) {
+              return true;
+            }
+
+            // if both workers without time, then compare with selector
+            auto maybe_res = request->sel->is_preferred(
+                request->task_type, workers_[lhs], workers_[rhs]);
+
+            if (maybe_res.has_error()) {
+              logger_->error("selecting best worker: "
+                             + maybe_res.error().message());
+              does_error_occurs = true;
+              return false;
+            }
+
+            return maybe_res.value();
+          });
 
       if (does_error_occurs) {
         return SchedulerErrors::kCannotSelectWorker;
       }
 
-      WorkerID wid = acceptable[0];
+      WorkerId wid = acceptable[0];
 
       assignWorker(wid, workers_[wid], request);
 
@@ -214,7 +240,7 @@ namespace fc::sector_storage {
   }
 
   void SchedulerImpl::assignWorker(
-      WorkerID wid,
+      WorkerId wid,
       const std::shared_ptr<WorkerHandle> &worker,
       const std::shared_ptr<TaskRequest> &request) {
     worker->preparing.add(worker->info.resources, request->need_resources);
@@ -242,8 +268,18 @@ namespace fc::sector_storage {
                          + maybe_call_id.error().message());
           return clear();
         }
-        ReturnCb new_cb = [request, clear = std::move(clear)](
+        estimator_->startWork(wid, request->task_type, maybe_call_id.value());
+        ReturnCb new_cb = [call_id{maybe_call_id.value()},
+                           estimator{estimator_},
+                           request,
+                           clear = std::move(clear)](
                               outcome::result<CallResult> result) -> void {
+          if (result.has_value()) {
+            estimator->finishWork(call_id);
+          } else {
+            estimator->abortWork(call_id);
+          }
+
           request->cb(std::move(result));
 
           return clear();
@@ -287,7 +323,7 @@ namespace fc::sector_storage {
     });
   }
 
-  void SchedulerImpl::freeWorker(WorkerID wid) {
+  void SchedulerImpl::freeWorker(WorkerId wid) {
     std::shared_ptr<WorkerHandle> worker;
     {
       std::lock_guard<std::mutex> lock(workers_lock_);

core/sector_storage/impl/scheduler_impl.hpp

@@ -13,23 +13,23 @@
 #include <unordered_map>
 #include <utility>
 #include "primitives/resources/resources.hpp"
+#include "sector_storage/worker_estimator.hpp"
 #include "storage/buffer_map.hpp"
 
 namespace fc::sector_storage {
-  using storage::BufferMap;
-  using WorkerID = uint64_t;
   using primitives::Resources;
+  using storage::BufferMap;
 
   struct TaskRequest {
     inline TaskRequest(const SectorRef &sector,
-                       TaskType task_type,
+                       const TaskType &task_type,
                        uint64_t priority,
                        std::shared_ptr<WorkerSelector> sel,
                        WorkerAction prepare,
                        WorkerAction work,
                        ReturnCb cb)
         : sector(sector),
-          task_type(std::move(task_type)),
+          task_type(task_type),
           priority(priority),
           sel(std::move(sel)),
           prepare(std::move(prepare)),
@@ -65,7 +65,8 @@ namespace fc::sector_storage {
    public:
     static outcome::result<std::shared_ptr<SchedulerImpl>> newScheduler(
         std::shared_ptr<boost::asio::io_context> io_context,
-        std::shared_ptr<BufferMap> datastore);
+        std::shared_ptr<BufferMap> datastore,
+        std::shared_ptr<Estimator> estimator);
 
     outcome::result<void> schedule(
         const SectorRef &sector,
@@ -84,22 +85,25 @@ namespace fc::sector_storage {
 
    private:
     explicit SchedulerImpl(std::shared_ptr<boost::asio::io_context> io_context,
-                           std::shared_ptr<BufferMap> datastore);
+                           std::shared_ptr<BufferMap> datastore,
+                           std::shared_ptr<Estimator> estimator);
 
     outcome::result<void> resetWorks();
 
     outcome::result<bool> maybeScheduleRequest(
         const std::shared_ptr<TaskRequest> &request);
 
-    void assignWorker(WorkerID wid,
+    void assignWorker(WorkerId wid,
                       const std::shared_ptr<WorkerHandle> &worker,
                       const std::shared_ptr<TaskRequest> &request);
 
-    void freeWorker(WorkerID wid);
+    void freeWorker(WorkerId wid);
 
     std::mutex workers_lock_;
-    WorkerID current_worker_id_;
-    std::unordered_map<WorkerID, std::shared_ptr<WorkerHandle>> workers_;
+    WorkerId current_worker_id_;
+    std::unordered_map<WorkerId, std::shared_ptr<WorkerHandle>> workers_;
+
+    std::shared_ptr<Estimator> estimator_;
 
     std::mutex cbs_lock_;
     std::map<CallId, ReturnCb> callbacks_;

core/sector_storage/impl/worker_estimator_impl.cpp

@@ -0,0 +1,69 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "sector_storage/impl/worker_estimator_impl.hpp"
+
+namespace fc::sector_storage {
+
+  EstimatorImpl::EstimatorImpl(uint64_t window_size)
+      : window_size_(window_size) {}
+
+  void EstimatorImpl::startWork(WorkerId worker_id,
+                                TaskType type,
+                                CallId call_id) {
+    const auto start = std::chrono::steady_clock::now();
+    std::lock_guard locker(mutex_);
+    active_works_[call_id] = ActiveWork{type, worker_id, start};
+  }
+
+  void EstimatorImpl::finishWork(CallId call_id) {
+    const auto finish = std::chrono::steady_clock::now();
+
+    std::lock_guard locker(mutex_);
+    const auto it = active_works_.find(call_id);
+    if (it == active_works_.end()) return;
+    const auto &work{it->second};
+
+    auto &task_map{workers_data_[work.worker]};
+    auto it2 = task_map.find(work.type);
+    if (it2 == task_map.end()) {
+      std::tie(it2, std::ignore) =
+          task_map.try_emplace(work.type, CallsData(window_size_));
+    }
+    it2->second.addData(std::chrono::duration_cast<std::chrono::milliseconds>(
+                            finish - work.start)
+                            .count());
+
+    active_works_.erase(it);
+  }
+
+  void EstimatorImpl::abortWork(CallId call_id) {
+    std::lock_guard locker(mutex_);
+    const auto it = active_works_.find(call_id);
+    if (it == active_works_.end()) return;
+
+    // we also can calculate how much works were aborted. It can be not worker's
+    // guilt, but if we want to have this statistic we can.
+
+    active_works_.erase(it);
+  }
+
+  boost::optional<double> EstimatorImpl::getTime(WorkerId id,
+                                                 TaskType type) const {
+    std::shared_lock locker(mutex_);
+
+    const auto tasks_it = workers_data_.find(id);
+    if (tasks_it == workers_data_.end()) {
+      return boost::none;
+    }
+
+    const auto it = tasks_it->second.find(type);
+    if (it == tasks_it->second.end()) {
+      return boost::none;
+    }
+
+    return it->second.getAverage();
+  }
+}  // namespace fc::sector_storage