psi support retry

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/wedpr-computing/ppc-pir/src/OtPIRImpl.cpp

@@ -445,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

@@ -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

@@ -168,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);
         }
 

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

@@ -148,10 +148,6 @@ class CM2020PSIImpl : public bcos::Worker,
         }
     }
 
-protected:
-    // allow the output-path exists, for ut
-    bool m_enableOutputExists = false;
-
 private:
     void waitSignal()
     {

cpp/wedpr-computing/ppc-psi/src/ecdh-conn-psi/EcdhConnPSIImpl.cpp

@@ -55,7 +55,7 @@ void EcdhConnPSIImpl::asyncRunTask(
     if (role == uint16_t(PartyType::Client))
     {
         ECDH_CONN_LOG(INFO) << LOG_DESC("Client do the Task") << LOG_KV("taskID", _task->id());
-        auto writer = loadWriter(_task->id(), dataResource, m_enableOutputExists);
+        auto writer = loadWriter(_task->id(), dataResource, _task->enableOutputExists());
         ecdhTaskState->setWriter(writer);
         auto client = std::make_shared<EcdhConnPSIClient>(m_config, ecdhTaskState);
         addClient(client);
@@ -66,7 +66,7 @@ void EcdhConnPSIImpl::asyncRunTask(
         ECDH_CONN_LOG(INFO) << LOG_DESC("Server do the Task") << LOG_KV("taskID", _task->id());
         if (_task->syncResultToPeer())
         {
-            auto writer = loadWriter(_task->id(), dataResource, m_enableOutputExists);
+            auto writer = loadWriter(_task->id(), dataResource, _task->enableOutputExists());
             ecdhTaskState->setWriter(writer);
         }
         auto server = std::make_shared<EcdhConnPSIServer>(m_config, ecdhTaskState);