move checkHostResource to psi module for notifing peers (#100)

* fix mysql gone

* add try-except for model-controller

* upgrade setuptools to 70.0.0

* change minNeedMemoryGB to 1

* clear the generated psi output file if run-failed

* move checkHostResource to psi module for notifing peers

* register-component in case of scheduler model-nodes in other wedpr-envs

* add components information to the serviceMeta

* query status from the cache firstly

Author: cyjseagull

cpp/ppc-framework/io/DataResourceLoader.h

@@ -50,5 +50,11 @@ class DataResourceLoader
 
     virtual void checkResourceExists(ppc::protocol::DataResourceDesc::ConstPtr _desc,
         ppc::storage::FileStorage::Ptr const& _storage = nullptr) = 0;
+
+    virtual void deleteResource(ppc::protocol::DataResourceDesc::ConstPtr _desc,
+        ppc::storage::FileStorage::Ptr const& _storage = nullptr) = 0;
+
+    virtual void renameResource(ppc::protocol::DataResourceDesc::ConstPtr _desc,
+        std::string const& _new_path, ppc::storage::FileStorage::Ptr const& _storage = nullptr) = 0;
 };
 }  // namespace ppc::io

cpp/tools/build_ppc.sh

@@ -411,6 +411,8 @@ generate_node_config_ini() {
     ; data_location = data
     ; task_timeout_minutes = 180
     ; thread_count = 8
+    ; GB, reject task if available memory is less than min_needed_memory
+    ; min_needed_memory = 1
 
 [transport]
    ; the endpoint information
@@ -458,8 +460,6 @@ generate_node_config_ini() {
     ; ssl connection switch, if disable the ssl connection, default: false
     ;disable_ssl = true
     ;disable_db = true
-    ; GB, reject task if available memory is less than min_needed_memory
-    ; min_needed_memory = 5
 
 [cert]
     ; directory the certificates located in

cpp/tools/build_wedpr_cem.sh

@@ -351,8 +351,6 @@ generate_config_ini() {
     sm_ssl=${sm_mode}
     ; ssl connection switch, if disable the ssl connection, default: false
     ;disable_ssl = true
-    ; GB, reject task if available memory is less than min_needed_memory
-    ; min_needed_memory = 5
 
 [hdfs_storage]
     ; the hdfs configuration

cpp/tools/ppc-builder/src/tpl/config.ini.node

@@ -9,6 +9,8 @@
     ; data_location = data
     ; task_timeout_minutes = 180
     ; thread_count = 8
+    ; GB, reject task if available memory is less than min_needed_memory
+    ; min_needed_memory = 1
 
 [crypto]
     sm_crypto = false
@@ -23,8 +25,6 @@
     ; ssl connection switch, if disable the ssl connection, default: true
     disable_ssl = true
     ; disable_db = true
-    ; GB, reject task if available memory is less than min_needed_memory
-    ; min_needed_memory = 5
 
 [cert]
     ; directory the certificates located in

cpp/wedpr-computing/ppc-psi/src/Common.h

@@ -18,6 +18,8 @@
  * @date 2022-12-26
  */
 #pragma once
+#include "ppc-framework/Common.h"
+#include "ppc-tools/src/common/MemInfo.h"
 #include <cstdint>
 #include <sstream>
 #define PSI_LOG(LEVEL) BCOS_LOG(LEVEL) << LOG_BADGE("PSI")
@@ -149,4 +151,17 @@ inline std::ostream& operator<<(std::ostream& _out, CacheState const& _state)
     }
     return _out;
 }
+
+inline void checkHostResource(uint64_t minNeededMemoryGB)
+{
+#ifdef __linux__
+    if (!ppc::tools::hasAvailableMem(minNeededMemoryGB))
+    {
+        BOOST_THROW_EXCEPTION(WeDPRException() << bcos::errinfo_comment(
+                                  "Lack of memory, please try again later. minNeededMemoryGB: " +
+                                  std::to_string(minNeededMemoryGB)));
+    }
+#endif
+}
+
 }  // namespace ppc::psi

cpp/wedpr-computing/ppc-psi/src/PSIConfig.h

@@ -37,26 +37,31 @@ class PSIConfig
 public:
     using Ptr = std::shared_ptr<PSIConfig>;
     PSIConfig(ppc::protocol::TaskAlgorithmType _algorithmType,
-        ppc::io::DataResourceLoader::Ptr _dataResourceLoader)
-      : m_algorithmType(_algorithmType), m_dataResourceLoader(std::move(_dataResourceLoader))
+        ppc::io::DataResourceLoader::Ptr _dataResourceLoader, uint32_t minNeededMemoryGB = 1)
+      : m_algorithmType(_algorithmType),
+        m_dataResourceLoader(std::move(_dataResourceLoader)),
+        m_minNeededMemoryGB(minNeededMemoryGB)
     {}
 
     PSIConfig(ppc::protocol::TaskAlgorithmType _algorithmType, const std::string& _selfParty,
         ppc::front::FrontInterface::Ptr _front,
         ppc::front::PPCMessageFaceFactory::Ptr _ppcMsgFactory,
-        ppc::io::DataResourceLoader::Ptr _dataResourceLoader, int _holdingMessageMinutes)
+        ppc::io::DataResourceLoader::Ptr _dataResourceLoader, int _holdingMessageMinutes,
+        uint32_t minNeededMemoryGB = 1)
       : m_algorithmType(_algorithmType),
         m_selfParty(_selfParty),
         m_front(std::move(_front)),
         m_ppcMsgFactory(std::move(_ppcMsgFactory)),
         m_dataResourceLoader(std::move(_dataResourceLoader)),
         m_networkTimeout(_holdingMessageMinutes * 60 * 1000),
-        m_taskExpireTime(m_networkTimeout)
+        m_taskExpireTime(m_networkTimeout),
+        m_minNeededMemoryGB(minNeededMemoryGB)
     {
         PSI_LOG(INFO) << LOG_DESC("create PSIConfig") << LOG_KV("algorithmType", m_algorithmType)
                       << LOG_KV("holdingMessageMinutes", _holdingMessageMinutes)
                       << LOG_KV("networkTimeout", m_networkTimeout)
-                      << LOG_KV("taskExpireTime", m_taskExpireTime);
+                      << LOG_KV("taskExpireTime", m_taskExpireTime)
+                      << LOG_KV("minNeededMemoryGB", m_minNeededMemoryGB);
     }
 
     virtual ~PSIConfig() = default;
@@ -128,6 +133,8 @@ class PSIConfig
 
     std::vector<std::string> agencyList() const { return m_front->agencies(); }
 
+    uint32_t minNeededMemoryGB() const { return m_minNeededMemoryGB; }
+
 protected:
     ppc::front::PPCMessageFace::Ptr generatePPCMsg(
         std::string const& _taskID, PSIMessageInterface::Ptr const& _msg, uint32_t _seq)
@@ -161,5 +168,7 @@ class PSIConfig
 
     // the task-expire time
     int m_taskExpireTime = 10000;
+
+    uint32_t m_minNeededMemoryGB = 1;
 };
 }  // namespace ppc::psi

cpp/wedpr-computing/ppc-psi/src/bs-ecdh-psi/BsEcdhPSIFactory.h

@@ -34,10 +34,11 @@ class BsEcdhPSIFactory
     virtual ~BsEcdhPSIFactory() = default;
 
     virtual BsEcdhPSIImpl::Ptr buildBsEcdhPSI(bcos::ThreadPool::Ptr _threadPool,
-        ppc::io::DataResourceLoader::Ptr _dataResourceLoader, uint32_t _timeoutMinutes)
+        ppc::io::DataResourceLoader::Ptr _dataResourceLoader, uint32_t _timeoutMinutes,
+        uint32_t minNeededMemoryGB = 1)
     {
-        auto config = std::make_shared<PSIConfig>(
-            ppc::protocol::TaskAlgorithmType::BS_ECDH_PSI, std::move(_dataResourceLoader));
+        auto config = std::make_shared<PSIConfig>(ppc::protocol::TaskAlgorithmType::BS_ECDH_PSI,
+            std::move(_dataResourceLoader), minNeededMemoryGB);
         return std::make_shared<BsEcdhPSIImpl>(
             std::move(config), std::move(_threadPool), _timeoutMinutes);
     }

cpp/wedpr-computing/ppc-psi/src/bs-ecdh-psi/BsEcdhPSIImpl.cpp

@@ -147,7 +147,8 @@ void BsEcdhPSIImpl::asyncRunTask(
         // init task state
         auto taskState = std::make_shared<BsEcdhTaskState>(
             _request->taskID, TaskStatus::PENDING, m_timeoutMinutes);
-
+        // check the memory
+        checkHostResource(m_config->minNeededMemoryGB());
         // create ecdh cache
         auto cache = std::make_shared<BsEcdhCache>(
             _request->taskID, m_threadPool, m_taskGuarder, _request->dataResource,

cpp/wedpr-computing/ppc-psi/src/cm2020-psi/CM2020PSIConfig.h

@@ -41,11 +41,11 @@ class CM2020PSIConfig : public PSIConfig
     CM2020PSIConfig(std::string const& _selfPartyID, ppc::front::FrontInterface::Ptr _front,
         ppc::crypto::CryptoBox::Ptr _cryptoBox, bcos::ThreadPool::Ptr _threadPool,
         ppc::io::DataResourceLoader::Ptr _dataResourceLoader, int _holdingMessageMinutes,
-        uint16_t _parallelism = 3,
+        uint32_t minNeededMemoryGB = 1, uint16_t _parallelism = 3,
         const front::PPCMessageFactory::Ptr& _msgFactory =
             std::make_shared<front::PPCMessageFactory>())
       : PSIConfig(ppc::protocol::TaskAlgorithmType::CM_PSI_2PC, _selfPartyID, std::move(_front),
-            _msgFactory, std::move(_dataResourceLoader), _holdingMessageMinutes),
+            _msgFactory, std::move(_dataResourceLoader), _holdingMessageMinutes, minNeededMemoryGB),
         m_cryptoBox(std::move(_cryptoBox)),
         m_threadPool(std::move(_threadPool)),
         m_parallelism(_parallelism)

cpp/wedpr-computing/ppc-psi/src/cm2020-psi/CM2020PSIFactory.h

@@ -34,11 +34,11 @@ class CM2020PSIFactory
     virtual CM2020PSIImpl::Ptr buildCM2020PSI(std::string const& _selfParty,
         ppc::front::FrontInterface::Ptr _front, ppc::crypto::CryptoBox::Ptr _cryptoBox,
         bcos::ThreadPool::Ptr _threadPool, ppc::io::DataResourceLoader::Ptr _dataResourceLoader,
-        int _holdingMessageMinutes, uint16_t _parallelism)
+        int _holdingMessageMinutes, uint16_t _parallelism, uint32_t minNeededMemoryGB = 1)
     {
         auto config = std::make_shared<CM2020PSIConfig>(_selfParty, std::move(_front),
             std::move(_cryptoBox), std::move(_threadPool), std::move(_dataResourceLoader),
-            _holdingMessageMinutes, _parallelism);
+            _holdingMessageMinutes, minNeededMemoryGB, _parallelism);
         return std::make_shared<CM2020PSIImpl>(config);
     }
 };