duplicate worker fix

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

Author: Elestrias

core/api/rpc/wsc.hpp

@@ -58,15 +58,11 @@ namespace fc::api::rpc {
       visit(api, [&](auto &m) { _setup(*this, m); });
     }
     struct ClientData {
-      ClientData(std::string host,
-                 std::string port,
-                 std::string target,
-                 std::string token)
-          : host(host), port(port), target(target), token(token){};
-      ClientData() = default;
-
-      std::string host, port, target, token;
-    }client_data;
+      std::string host;
+      std::string port;
+      std::string target;
+      std::string token;
+    } client_data;
 
     void reconnect(int counter, std::chrono::milliseconds wait);
 
@@ -76,7 +72,9 @@ namespace fc::api::rpc {
     io_context io;
     io_context &io2;
     boost::asio::executor_work_guard<io_context::executor_type> work_guard;
-    boost::optional<boost::beast::websocket::stream<boost::asio::ip::tcp::socket>> socket;
+    boost::optional<
+        boost::beast::websocket::stream<boost::asio::ip::tcp::socket>>
+        socket;
     boost::beast::flat_buffer buffer;
     std::mutex mutex;
     uint64_t next_req{};

core/primitives/sector_file/sector_file.hpp

@@ -83,11 +83,11 @@ namespace fc::primitives::sector_file {
   struct SectorPaths {
    public:
     SectorId id;
-    std::string unsealed;
-    std::string sealed;
-    std::string cache;
-    std::string update;
-    std::string update_cache;
+    std::string unsealed{};
+    std::string sealed{};
+    std::string cache{};
+    std::string update{};
+    std::string update_cache{};
 
     void setPathByType(const SectorFileType &file_type,
                        const std::string &path);

core/primitives/types.hpp

@@ -39,7 +39,7 @@ namespace fc::primitives {
   struct FsStat {
     uint64_t capacity = 0;
     uint64_t available = 0;
-    uint64_t fs_available = 0; // Available to use for sector storage
+    uint64_t fs_available = 0;  // Available to use for sector storage
     uint64_t reserved = 0;
     uint64_t max = 0;
     uint64_t used = 0;
@@ -91,6 +91,14 @@ namespace fc::primitives {
     std::vector<std::string> gpus;
   };
 
+  inline bool operator==(const WorkerResources &lhs,
+                         const WorkerResources &rhs) {
+    return (lhs.physical_memory == rhs.physical_memory
+            && lhs.swap_memory == rhs.swap_memory
+            && lhs.reserved_memory == rhs.reserved_memory
+            && lhs.cpus == rhs.cpus && lhs.gpus == rhs.gpus);
+  }
+
   struct WorkerInfo {
     std::string hostname;
     WorkerResources resources;

core/sector_storage/impl/local_worker.cpp

@@ -904,7 +904,7 @@ namespace fc::sector_storage {
 
     return call_id;
   }
-  void LocalWorker::ping(std::function<void(bool resp)> cb) {
+  void LocalWorker::ping(std::function<void(const bool resp)> cb) {
     cb(true);
   }
 }  // namespace fc::sector_storage

core/sector_storage/impl/local_worker.hpp

@@ -99,7 +99,7 @@ namespace fc::sector_storage {
     outcome::result<std::vector<primitives::StoragePath>> getAccessiblePaths()
         override;
 
-    void ping(std::function<void(bool resp)> cb) override;
+    void ping(std::function<void(const bool resp)> cb) override;
 
    private:
     template <typename W, typename R>

core/sector_storage/impl/manager_impl.cpp

@@ -260,7 +260,6 @@ namespace fc::sector_storage {
 
     worker_handler->worker = std::move(worker);
     worker_handler->info = std::move(info);
-
     scheduler_->newWorker(std::move(worker_handler));
 
     return outcome::success();

core/sector_storage/impl/remote_worker.cpp

@@ -255,7 +255,7 @@ namespace fc::sector_storage {
     return api_.Fetch(sector, file_type, path_type, mode);
   }
 
-  void RemoteWorker::ping(std::function<void(bool resp)> cb) {
+  void RemoteWorker::ping(std::function<void(const bool resp)> cb) {
     api_.Version([=](auto res) { cb(!res.has_error()); });
   }
 }  // namespace fc::sector_storage

core/sector_storage/impl/remote_worker.hpp

@@ -39,7 +39,7 @@ namespace fc::sector_storage {
         const SectorRef &sector,
         const PreCommit1Output &pre_commit_1_output) override;
 
-    void ping(std::function<void(bool resp)> cb) override;
+    void ping(std::function<void(const bool resp)> cb) override;
 
     outcome::result<CallId> sealCommit1(const SectorRef &sector,
                                         const SealRandomness &ticket,

core/sector_storage/impl/scheduler_impl.cpp

@@ -137,6 +137,11 @@ namespace fc::sector_storage {
 
   void SchedulerImpl::newWorker(std::unique_ptr<WorkerHandle> worker) {
     std::unique_lock<std::mutex> lock(workers_lock_);
+    for(const auto &[key, value] : workers_){
+      if(*value == *worker){
+        return;
+      }
+    }
     if (current_worker_id_ == std::numeric_limits<uint64_t>::max()) {
       current_worker_id_ = 0;  // TODO(ortyomka): maybe better mechanism
     }
@@ -200,7 +205,7 @@ namespace fc::sector_storage {
       std::future<WorkerID> wid_future = wid_promise.get_future();
       auto done = std::make_shared<std::atomic_bool>();
       for (const auto &cur : acceptable) {
-        workers_[cur]->worker->ping([&wid_promise, done, cur](bool resp) {
+        workers_[cur]->worker->ping([&wid_promise, done, cur](const bool resp) {
           if (resp && !done->exchange(true)) {
             wid_promise.set_value(cur);
           }

core/sector_storage/selector.hpp

@@ -24,6 +24,11 @@ namespace fc::sector_storage {
     ActiveResources active;
   };
 
+  inline bool operator==(const WorkerHandle &lhs, const WorkerHandle &rhs) {
+    return lhs.info.hostname == rhs.info.hostname
+           && lhs.info.resources == rhs.info.resources;
+  }
+
   class WorkerSelector {
    public:
     virtual ~WorkerSelector() = default;

core/sector_storage/worker.hpp

@@ -143,7 +143,7 @@ namespace fc::sector_storage {
     virtual outcome::result<std::vector<primitives::StoragePath>>
     getAccessiblePaths() = 0;
 
-    virtual void ping(std::function<void(bool resp)> cb) = 0;
+    virtual void ping(std::function<void(const bool resp)> cb) = 0;
   };
 
   enum class CallErrorCode : uint64_t {

test/testutil/mocks/sector_storage/worker_mock.hpp

@@ -101,6 +101,6 @@ namespace fc::sector_storage {
                                          gsl::span<const UnpaddedPieceSize>,
                                          const UnpaddedPieceSize &,
                                          int));
-    MOCK_METHOD1(ping, void(std::function<void(const bool &)>));
+    MOCK_METHOD1(ping, void(std::function<void(const bool)>));
   };
 }  // namespace fc::sector_storage