fix cm2020 check hostResource (#101)

* fix cm2020 check hostResource

* complement log

* psi support retry

* unregister front taskInfo

Author: cyjseagull

cpp/ppc-framework/protocol/DataResource.h

@@ -119,6 +119,10 @@ class DataResource
 
 inline std::string printDataResourceInfo(DataResource::ConstPtr _dataResource)
 {
+    if (!_dataResource)
+    {
+        return "empty";
+    }
     std::ostringstream stringstream;
     stringstream << LOG_KV("dataResource", _dataResource->resourceID());
     if (_dataResource->desc())

cpp/ppc-framework/protocol/PartyResource.h

@@ -67,6 +67,10 @@ using ConstParties = std::vector<PartyResource::ConstPtr>;
 
 inline std::string printPartyInfo(PartyResource::ConstPtr _party)
 {
+    if (!_party)
+    {
+        return "empty";
+    }
     std::ostringstream stringstream;
     stringstream << LOG_KV("partyId", _party->id()) << LOG_KV("partyIndex", _party->partyIndex())
                  << LOG_KV("desc", _party->desc());

cpp/ppc-framework/protocol/Task.h

@@ -143,6 +143,15 @@ class Task
     // decode the task
     virtual void decode(std::string_view _taskData) = 0;
     virtual std::string encode() const = 0;
+
+    virtual bool enableOutputExists() const { return m_enableOutputExists; }
+    virtual void setEnableOutputExists(bool enableOutputExists)
+    {
+        m_enableOutputExists = enableOutputExists;
+    }
+
+protected:
+    bool m_enableOutputExists = false;
 };
 
 class TaskFactory
@@ -160,9 +169,14 @@ class TaskFactory
 inline std::string printTaskInfo(Task::ConstPtr _task)
 {
     std::ostringstream stringstream;
+    if (!_task)
+    {
+        return "empty";
+    }
     stringstream << LOG_KV("id", _task->id())
                  << LOG_KV("type", (ppc::protocol::TaskType)_task->type())
                  << LOG_KV("algorithm", (ppc::protocol::TaskAlgorithmType)_task->algorithm())
+                 << LOG_KV("enableOutputExists", _task->enableOutputExists())
                  << LOG_KV("taskPtr", _task);
     if (_task->selfParty())
     {

cpp/ppc-framework/rpc/RpcStatusInterface.h

@@ -44,7 +44,5 @@ class RpcStatusInterface
     virtual bcos::Error::Ptr insertTask(protocol::Task::Ptr _task) = 0;
     virtual bcos::Error::Ptr updateTaskStatus(protocol::TaskResult::Ptr _taskResult) = 0;
     virtual protocol::TaskResult::Ptr getTaskStatus(const std::string& _taskID) = 0;
-    virtual bcos::Error::Ptr deleteGateway(const std::string& _agencyID) = 0;
-    virtual std::vector<protocol::GatewayInfo> listGateway() = 0;
 };
 }  // namespace ppc::rpc

cpp/wedpr-computing/ppc-pir/src/OtPIRImpl.cpp

@@ -149,10 +149,11 @@ void OtPIRImpl::onReceiveMessage(ppc::front::PPCMessageFace::Ptr _msg)
     }
 }
 
-void OtPIRImpl::onReceivedErrorNotification(const std::string& _taskID)
+void OtPIRImpl::onReceivedErrorNotification(ppc::front::PPCMessageFace::Ptr const& _message)
 {
+    PIR_LOG(WARNING) << LOG_DESC("onReceivedErrorNotification") << printPPCMsg(_message);
     // finish the task while the peer is failed
-    auto taskState = findPendingTask(_taskID);
+    auto taskState = findPendingTask(_message->taskID());
     if (taskState)
     {
         taskState->onPeerNotifyFinish();
@@ -242,7 +243,7 @@ void OtPIRImpl::handleReceivedMessage(const ppc::front::PPCMessageFace::Ptr& _me
             {
             case int(CommonMessageType::ErrorNotification):
             {
-                pir->onReceivedErrorNotification(_message->taskID());
+                pir->onReceivedErrorNotification(_message);
                 break;
             }
             case int(CommonMessageType::PingPeer):
@@ -444,7 +445,8 @@ void OtPIRImpl::asyncRunTask()
                            << LOG_KV("requestAgencyDataset", taskMessage.requestAgencyDataset)
                            << LOG_KV("prefixLength", taskMessage.prefixLength)
                            << LOG_KV("searchId", taskMessage.searchId);
-            auto writer = loadWriter(task->id(), dataResource, m_enableOutputExists);
+            auto writer =
+                loadWriter(task->id(), dataResource, m_taskState->task()->enableOutputExists());
             m_taskState->setWriter(writer);
             runSenderGenerateCipher(taskMessage);
         }

cpp/wedpr-computing/ppc-pir/src/OtPIRImpl.h

@@ -59,7 +59,7 @@ class OtPIRImpl : public std::enable_shared_from_this<OtPIRImpl>,
     // register to the front to get the message related to ot-pir
     void onReceiveMessage(ppc::front::PPCMessageFace::Ptr _message) override;
 
-    void onReceivedErrorNotification(const std::string& _taskID) override;
+    void onReceivedErrorNotification(ppc::front::PPCMessageFace::Ptr const& _message) override;
     void onSelfError(
         const std::string& _taskID, bcos::Error::Ptr _error, bool _noticePeer) override;
 
@@ -150,8 +150,6 @@ class OtPIRImpl : public std::enable_shared_from_this<OtPIRImpl>,
             m_senders.erase(it);
         }
     }
-    // allow the output-path exists, for ut
-    bool m_enableOutputExists = false;
     // 为true时启动时会从配置中加载文件作为匹配源
     bool m_enableMemoryFile = false;
     ppc::protocol::DataResource m_resource;

cpp/wedpr-computing/ppc-pir/tests/FakeOtPIRFactory.h

@@ -46,9 +46,7 @@ class FakeOtPIRImpl : public OtPIRImpl
     using Ptr = std::shared_ptr<FakeOtPIRImpl>;
     FakeOtPIRImpl(OtPIRConfig::Ptr const& _config, unsigned _idleTimeMs = 0)
       : OtPIRImpl(_config, _idleTimeMs)
-    {
-        m_enableOutputExists = true;
-    }
+    {}
     ~FakeOtPIRImpl() override = default;
 };
 

cpp/wedpr-computing/ppc-pir/tests/TestBaseOT.cpp

@@ -194,6 +194,7 @@ void testOTPIRImplFunc(const std::string& _taskID, const std::string& _params, b
     auto senderPIRTask = std::make_shared<JsonTaskImpl>(senderAgencyName);
     senderPIRTask->setId(_taskID);
     senderPIRTask->setParam(_params);
+    senderPIRTask->setEnableOutputExists(true);
     senderPIRTask->setSelf(_senderParty);
     senderPIRTask->addParty(_receiverParty);
     senderPIRTask->setSyncResultToPeer(_syncResults);
@@ -203,6 +204,7 @@ void testOTPIRImplFunc(const std::string& _taskID, const std::string& _params, b
     auto receiverPIRTask = std::make_shared<JsonTaskImpl>(receiverAgencyName);
     receiverPIRTask->setId(_taskID);
     receiverPIRTask->setParam(_params);
+    receiverPIRTask->setEnableOutputExists(true);
     receiverPIRTask->setSelf(_receiverParty);
     receiverPIRTask->addParty(_senderParty);
     receiverPIRTask->setSyncResultToPeer(_syncResults);

cpp/wedpr-computing/ppc-psi/src/cm2020-psi/CM2020PSIImpl.cpp

@@ -152,15 +152,14 @@ void CM2020PSIImpl::asyncRunTask()
         {
             return;
         }
-
+        CM2020_PSI_LOG(INFO) << LOG_DESC("noticePeerToFinish") << printTaskInfo(task);
         psi->noticePeerToFinish(task);
     });
-    // check the memory
-    checkHostResource(m_config->minNeededMemoryGB());
-    addPendingTask(taskState);
-
     try
     {
+        addPendingTask(taskState);
+        // check the memory
+        checkHostResource(m_config->minNeededMemoryGB());
         // prepare reader and writer
         auto dataResource = task->selfParty()->dataResource();
         auto reader = loadReader(task->id(), dataResource, DataSchema::Bytes);
@@ -169,7 +168,7 @@ void CM2020PSIImpl::asyncRunTask()
         auto role = task->selfParty()->partyIndex();
         if (role == uint16_t(PartyType::Client) || task->syncResultToPeer())
         {
-            auto writer = loadWriter(task->id(), dataResource, m_enableOutputExists);
+            auto writer = loadWriter(task->id(), dataResource, task->enableOutputExists());
             taskState->setWriter(writer);
         }
 
@@ -319,10 +318,11 @@ void CM2020PSIImpl::stop()
     CM2020_PSI_LOG(INFO) << LOG_DESC("CM2020-PSI stopped");
 }
 
-void CM2020PSIImpl::onReceivedErrorNotification(const std::string& _taskID)
+void CM2020PSIImpl::onReceivedErrorNotification(ppc::front::PPCMessageFace::Ptr const& _message)
 {
+    CM2020_PSI_LOG(INFO) << LOG_DESC("onReceivedErrorNotification") << printPPCMsg(_message);
     // finish the task while the peer is failed
-    auto taskState = findPendingTask(_taskID);
+    auto taskState = findPendingTask(_message->taskID());
     if (taskState)
     {
         taskState->onPeerNotifyFinish();
@@ -410,7 +410,7 @@ void CM2020PSIImpl::handleReceivedMessage(const ppc::front::PPCMessageFace::Ptr&
             {
             case int(CommonMessageType::ErrorNotification):
             {
-                psi->onReceivedErrorNotification(_message->taskID());
+                psi->onReceivedErrorNotification(_message);
                 break;
             }
             case int(CommonMessageType::PingPeer):

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

@@ -62,7 +62,7 @@ class CM2020PSIImpl : public bcos::Worker,
 
     void start() override;
     void stop() override;
-    void onReceivedErrorNotification(const std::string& _taskID) override;
+    void onReceivedErrorNotification(ppc::front::PPCMessageFace::Ptr const& _message) override;
     void onSelfError(
         const std::string& _taskID, bcos::Error::Ptr _error, bool _noticePeer) override;
 
@@ -148,10 +148,6 @@ class CM2020PSIImpl : public bcos::Worker,
         }
     }
 
-protected:
-    // allow the output-path exists, for ut
-    bool m_enableOutputExists = false;
-
 private:
     void waitSignal()
     {