psi support retry

cyjseagull · cyjseagull · commit d9a66211a62d · 2024-11-25T19:07:46.000+08:00
diff --git a/cpp/ppc-framework/protocol/Task.h b/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
diff --git a/cpp/wedpr-computing/ppc-pir/src/OtPIRImpl.cpp b/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);
         }
diff --git a/cpp/wedpr-computing/ppc-pir/src/OtPIRImpl.h b/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;
diff --git a/cpp/wedpr-computing/ppc-pir/tests/FakeOtPIRFactory.h b/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;
 };
 
diff --git a/cpp/wedpr-computing/ppc-pir/tests/TestBaseOT.cpp b/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);
diff --git a/cpp/wedpr-computing/ppc-psi/src/cm2020-psi/CM2020PSIImpl.cpp b/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);
         }
 
diff --git a/cpp/wedpr-computing/ppc-psi/src/cm2020-psi/CM2020PSIImpl.h b/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()
     {
diff --git a/cpp/wedpr-computing/ppc-psi/src/ecdh-conn-psi/EcdhConnPSIImpl.cpp b/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);
diff --git a/cpp/wedpr-computing/ppc-psi/src/ecdh-conn-psi/EcdhConnPSIImpl.h b/cpp/wedpr-computing/ppc-psi/src/ecdh-conn-psi/EcdhConnPSIImpl.h
@@ -66,9 +66,6 @@ class EcdhConnPSIImpl : public std::enable_shared_from_this<EcdhConnPSIImpl>,
         const std::string& _taskID, bcos::Error::Ptr _error, bool _noticePeer) override;
     void executeWorker() override;
 
-protected:
-    bool m_enableOutputExists = false;
-
     void handlerPSIReceiveMessage(PSIConnMessage::Ptr _msg);
     void onHandShakeRequestHandler(const std::string& _taskId, const bcos::bytes& _msg);
     void onHandShakeResponseHandler(const std::string& _taskId, const bcos::bytes& _msg);
diff --git a/cpp/wedpr-computing/ppc-psi/src/ecdh-multi-psi/EcdhMultiPSIImpl.cpp b/cpp/wedpr-computing/ppc-psi/src/ecdh-multi-psi/EcdhMultiPSIImpl.cpp
@@ -155,7 +155,7 @@ void EcdhMultiPSIImpl::asyncRunTask(
                              << LOG_KV("roleId", role);
         if (role == uint16_t(PartiesType::Calculator))
         {
-            auto writer = loadWriter(_task->id(), dataResource, m_enableOutputExists);
+            auto writer = loadWriter(_task->id(), dataResource, _task->enableOutputExists());
             taskState->setWriter(writer);
             ECDH_MULTI_LOG(INFO) << LOG_DESC("Calculator do the Task")
                                  << LOG_KV("taskID", _task->id());
@@ -170,7 +170,7 @@ void EcdhMultiPSIImpl::asyncRunTask(
             if (_task->syncResultToPeer() && std::find(receivers.begin(), receivers.end(),
                                                  m_config->selfParty()) != receivers.end())
             {
-                auto writer = loadWriter(_task->id(), dataResource, m_enableOutputExists);
+                auto writer = loadWriter(_task->id(), dataResource, _task->enableOutputExists());
                 taskState->setWriter(writer);
             }
             auto partner = std::make_shared<EcdhMultiPSIPartner>(m_config, taskState);
@@ -183,7 +183,7 @@ void EcdhMultiPSIImpl::asyncRunTask(
             if (_task->syncResultToPeer() && std::find(receivers.begin(), receivers.end(),
                                                  m_config->selfParty()) != receivers.end())
             {
-                auto writer = loadWriter(_task->id(), dataResource, m_enableOutputExists);
+                auto writer = loadWriter(_task->id(), dataResource, _task->enableOutputExists());
                 taskState->setWriter(writer);
             }
             auto master = std::make_shared<EcdhMultiPSIMaster>(m_config, taskState);
diff --git a/cpp/wedpr-computing/ppc-psi/src/ecdh-multi-psi/EcdhMultiPSIImpl.h b/cpp/wedpr-computing/ppc-psi/src/ecdh-multi-psi/EcdhMultiPSIImpl.h
@@ -47,7 +47,6 @@ class EcdhMultiPSIImpl : public std::enable_shared_from_this<EcdhMultiPSIImpl>,
 
 
 protected:
-    bool m_enableOutputExists = false;
     virtual void onReceiveRandomA(PSIMessageInterface::Ptr _msg);
     virtual void onReceiveCalCipher(PSIMessageInterface::Ptr _msg);
     virtual void handlerPSIReceiveMessage(PSIMessageInterface::Ptr _msg);
diff --git a/cpp/wedpr-computing/ppc-psi/src/ecdh-psi/EcdhPSIImpl.cpp b/cpp/wedpr-computing/ppc-psi/src/ecdh-psi/EcdhPSIImpl.cpp
@@ -207,7 +207,7 @@ bool EcdhPSIImpl::initTaskState(TaskState::Ptr const& _taskState)
         if (!server)
         {
             m_cache->insertServerCipherCache(task->id(), _taskState);
-            if (!m_enableOutputExists)
+            if (!_taskState->task()->enableOutputExists())
             {
                 // Note: if the output-resource already exists, will throw exception
                 m_config->dataResourceLoader()->checkResourceExists(dataResource->outputDesc());
diff --git a/cpp/wedpr-computing/ppc-psi/src/ecdh-psi/EcdhPSIImpl.h b/cpp/wedpr-computing/ppc-psi/src/ecdh-psi/EcdhPSIImpl.h
@@ -85,8 +85,5 @@ class EcdhPSIImpl : public PSIFramework, public std::enable_shared_from_this<Ecd
 protected:
     EcdhPSIConfig::Ptr m_config;
     EcdhCache::Ptr m_cache;
-
-    // allow the output-path exists, for ut
-    bool m_enableOutputExists = false;
 };
 }  // namespace ppc::psi
diff --git a/cpp/wedpr-computing/ppc-psi/src/labeled-psi/LabeledPSIImpl.cpp b/cpp/wedpr-computing/ppc-psi/src/labeled-psi/LabeledPSIImpl.cpp
@@ -554,7 +554,7 @@ void LabeledPSIImpl::saveSenderCache(const ppc::protocol::Task::ConstPtr& _task)
         auto dataResource = _task->selfParty()->dataResource();
 
         LineWriter::Ptr writer;
-        if (!m_enableOutputExists)
+        if (!_task->enableOutputExists())
         {
             // Note: if the output-resource already exists, will throw exception
             m_config->dataResourceLoader()->checkResourceExists(dataResource->outputDesc());
diff --git a/cpp/wedpr-computing/ppc-psi/src/labeled-psi/LabeledPSIImpl.h b/cpp/wedpr-computing/ppc-psi/src/labeled-psi/LabeledPSIImpl.h
@@ -116,9 +116,6 @@ class LabeledPSIImpl : public bcos::Worker,
         }
     }
 
-protected:
-    // allow the output-path exists, for ut
-    bool m_enableOutputExists = false;
 
 private:
     void waitSignal()
diff --git a/cpp/wedpr-computing/ppc-psi/src/ra2018-psi/RA2018PSIImpl.cpp b/cpp/wedpr-computing/ppc-psi/src/ra2018-psi/RA2018PSIImpl.cpp
@@ -646,7 +646,7 @@ void RA2018PSIImpl::runClientPSI(TaskState::Ptr const& _taskState)
     {
         return;
     }
-    if (!m_enableOutputExists)
+    if (!_taskState->task()->enableOutputExists())
     {
         // Note: if the output-resource already exists, will throw exception
         m_config->dataResourceLoader()->checkResourceExists(dataResource->outputDesc());
diff --git a/cpp/wedpr-computing/ppc-psi/src/ra2018-psi/RA2018PSIImpl.h b/cpp/wedpr-computing/ppc-psi/src/ra2018-psi/RA2018PSIImpl.h
@@ -179,8 +179,5 @@ class RA2018PSIImpl : public PSIFramework, public std::enable_shared_from_this<R
     bcos::ThreadPool::Ptr m_worker;
     // the flag means that response to the sdk once handling the task or after the task completed
     bool m_waitResult;
-
-    // allow the output-path exists, for ut
-    bool m_enableOutputExists = false;
 };
 }  // namespace ppc::psi
diff --git a/cpp/wedpr-computing/ppc-psi/tests/cm2020-psi/FakeCM2020PSIFactory.h b/cpp/wedpr-computing/ppc-psi/tests/cm2020-psi/FakeCM2020PSIFactory.h
@@ -43,9 +43,7 @@ class FakeCM2020PSIImpl : public CM2020PSIImpl
     using Ptr = std::shared_ptr<FakeCM2020PSIImpl>;
     FakeCM2020PSIImpl(CM2020PSIConfig::Ptr const& _config, unsigned _idleTimeMs = 0)
       : CM2020PSIImpl(_config, _idleTimeMs)
-    {
-        m_enableOutputExists = true;
-    }
+    {}
     ~FakeCM2020PSIImpl() override = default;
 };
 
diff --git a/cpp/wedpr-computing/ppc-psi/tests/cm2020-psi/TestCM2020Impl.cpp b/cpp/wedpr-computing/ppc-psi/tests/cm2020-psi/TestCM2020Impl.cpp
@@ -153,6 +153,7 @@ void testCM2020PSIImplFunc(const std::string& _taskID, const std::string& _param
     senderPSITask->setId(_taskID);
     senderPSITask->setParam(_params);
     senderPSITask->setSelf(_senderParty);
+    senderPSITask->setEnableOutputExists(true);
     senderPSITask->addParty(_receiverParty);
     senderPSITask->setSyncResultToPeer(_syncResults);
     senderPSITask->setAlgorithm((uint8_t)TaskAlgorithmType::CM_PSI_2PC);
@@ -161,6 +162,7 @@ void testCM2020PSIImplFunc(const std::string& _taskID, const std::string& _param
     receiverPSITask->setId(_taskID);
     receiverPSITask->setParam(_params);
     receiverPSITask->setSelf(_receiverParty);
+    receiverPSITask->setEnableOutputExists(true);
     receiverPSITask->addParty(_senderParty);
     receiverPSITask->setSyncResultToPeer(_syncResults);
     receiverPSITask->setAlgorithm((uint8_t)TaskAlgorithmType::CM_PSI_2PC);
diff --git a/cpp/wedpr-computing/ppc-psi/tests/labeled-psi/FakeLabeledPSIFactory.h b/cpp/wedpr-computing/ppc-psi/tests/labeled-psi/FakeLabeledPSIFactory.h
@@ -43,9 +43,7 @@ class FakeLabeledPSIImpl : public LabeledPSIImpl
     using Ptr = std::shared_ptr<FakeLabeledPSIImpl>;
     FakeLabeledPSIImpl(LabeledPSIConfig::Ptr const& _config, unsigned _idleTimeMs = 0)
       : LabeledPSIImpl(_config, _idleTimeMs)
-    {
-        m_enableOutputExists = true;
-    }
+    {}
     ~FakeLabeledPSIImpl() override = default;
 };
 
diff --git a/cpp/wedpr-computing/ppc-psi/tests/labeled-psi/TestLabeledPSIImpl.cpp b/cpp/wedpr-computing/ppc-psi/tests/labeled-psi/TestLabeledPSIImpl.cpp
@@ -59,6 +59,7 @@ void runSetup(LabeledPSIImpl::Ptr _sender)
 
     auto senderPSITask = std::make_shared<JsonTaskImpl>(senderAgencyName);
     senderPSITask->setId("0x00000000");
+    senderPSITask->setEnableOutputExists(true);
     senderPSITask->setParam(R"(["setup_sender_db","32"])");
     senderPSITask->setSelf(senderParty);
     senderPSITask->setAlgorithm((uint8_t)TaskAlgorithmType::LABELED_PSI_2PC);
@@ -95,6 +96,7 @@ void saveCache(LabeledPSIImpl::Ptr _sender)
 
     auto senderPSITask = std::make_shared<JsonTaskImpl>(senderAgencyName);
     senderPSITask->setId("0x00000012");
+    senderPSITask->setEnableOutputExists(true);
     senderPSITask->setParam(R"(["save_sender_cache"])");
     senderPSITask->setSelf(senderParty);
     senderPSITask->setAlgorithm((uint8_t)TaskAlgorithmType::LABELED_PSI_2PC);
@@ -234,13 +236,15 @@ void testLabeledPSIImplFunc(const std::string& _taskID, const std::string& _para
     // trigger the psi task
     auto senderPSITask = std::make_shared<JsonTaskImpl>(senderAgencyName);
     senderPSITask->setId(_taskID);
+    senderPSITask->setEnableOutputExists(true);
     senderPSITask->setParam(_params);
     senderPSITask->setSelf(_senderParty);
     senderPSITask->setAlgorithm((uint8_t)TaskAlgorithmType::LABELED_PSI_2PC);
     senderPSITask->addParty(_receiverParty);
 
     auto receiverPSITask = std::make_shared<JsonTaskImpl>(receiverAgencyName);
     receiverPSITask->setId(_taskID);
+    receiverPSITask->setEnableOutputExists(true);
     receiverPSITask->setSelf(_receiverParty);
     receiverPSITask->setAlgorithm((uint8_t)TaskAlgorithmType::LABELED_PSI_2PC);
     receiverPSITask->addParty(_senderParty);
diff --git a/cpp/wedpr-computing/ppc-psi/tests/ra2018-psi/TestEcdhPSIImpl.cpp b/cpp/wedpr-computing/ppc-psi/tests/ra2018-psi/TestEcdhPSIImpl.cpp
@@ -87,6 +87,8 @@ void testEcdhImplFunc(int64_t _dataBatchSize, std::string const& _serverPSIDataS
     auto clientPSITask = std::make_shared<JsonTaskImpl>(clientAgencyName);
     std::string taskID = "runPSI";
     clientPSITask->setId(taskID);
+    clientPSITask->setEnableOutputExists(true);
+    clientPSITask->setEnableOutputExists(true);
     clientPSITask->setType((int8_t)TaskType::PSI);
     clientPSITask->setAlgorithm((int8_t)TaskAlgorithmType::ECDH_PSI_2PC);
     clientPSITask->setSelf(clientParty);
@@ -103,6 +105,8 @@ void testEcdhImplFunc(int64_t _dataBatchSize, std::string const& _serverPSIDataS
     {
         serverPSITask->setId(taskID);
     }
+    serverPSITask->setEnableOutputExists(true);
+    serverPSITask->setEnableOutputExists(true);
     serverPSITask->setType((int8_t)TaskType::PSI);
     serverPSITask->setAlgorithm((int8_t)TaskAlgorithmType::ECDH_PSI_2PC);
     serverPSITask->addParty(clientParty);
diff --git a/cpp/wedpr-computing/ppc-psi/tests/ra2018-psi/TestRA2018Impl.cpp b/cpp/wedpr-computing/ppc-psi/tests/ra2018-psi/TestRA2018Impl.cpp
@@ -105,6 +105,7 @@ void testRA2018PSIImplFunc(int _dataBatchSize, CuckoofilterOption::Ptr option,
     auto offlineFullEvaluateTask = std::make_shared<JsonTaskImpl>(serverAgencyName);
     offlineFullEvaluateTask->setId("offlineFullEvaluate");
     offlineFullEvaluateTask->setSelf(serverParty);
+    offlineFullEvaluateTask->setEnableOutputExists(true);
     // insert operation
     std::string param = "[\"data_preprocessing\", 0]";
     offlineFullEvaluateTask->setParam(param);
@@ -127,6 +128,7 @@ void testRA2018PSIImplFunc(int _dataBatchSize, CuckoofilterOption::Ptr option,
     auto clientPSITask = std::make_shared<JsonTaskImpl>(clientAgencyName);
     std::string taskID = "runPSI";
     clientPSITask->setId(taskID);
+    clientPSITask->setEnableOutputExists(true);
     clientPSITask->setSelf(clientParty);
     clientPSITask->addParty(serverParty);
     param = "[\"ra2018_psi\"]";
@@ -135,6 +137,7 @@ void testRA2018PSIImplFunc(int _dataBatchSize, CuckoofilterOption::Ptr option,
     // the server task
     auto serverPSITask = std::make_shared<JsonTaskImpl>(serverAgencyName);
     serverPSITask->setSelf(serverParty);
+    serverPSITask->setEnableOutputExists(true);
     serverPSITask->setId(taskID);
     serverPSITask->setParam(param);
     serverPSITask->addParty(clientParty);
diff --git a/cpp/wedpr-computing/ppc-psi/tests/ra2018-psi/mock/EcdhPSIFixture.h b/cpp/wedpr-computing/ppc-psi/tests/ra2018-psi/mock/EcdhPSIFixture.h
@@ -49,7 +49,6 @@ class FakeEcdhPSIImpl : public EcdhPSIImpl
     {
         // set the m_started flag to be true
         m_started = true;
-        m_enableOutputExists = true;
         m_taskSyncTimer->registerTimeoutHandler([this]() { syncTaskInfo(); });
         m_taskSyncTimer->start();
     }
diff --git a/cpp/wedpr-computing/ppc-psi/tests/ra2018-psi/mock/RA2018PSIFixture.h b/cpp/wedpr-computing/ppc-psi/tests/ra2018-psi/mock/RA2018PSIFixture.h
@@ -51,7 +51,6 @@ class FakeRA2018Impl : public RA2018PSIImpl
     {
         // set the m_started flag to be true
         m_started = true;
-        m_enableOutputExists = true;
         m_taskSyncTimer->registerTimeoutHandler([this]() { syncTaskInfo(); });
         m_taskSyncTimer->start();
     }
diff --git a/cpp/wedpr-protocol/protocol/src/JsonTaskImpl.cpp b/cpp/wedpr-protocol/protocol/src/JsonTaskImpl.cpp
@@ -54,6 +54,10 @@ void JsonTaskImpl::decodeJsonValue(Json::Value const& root)
     {
         BOOST_THROW_EXCEPTION(InvalidParameter() << errinfo_comment("Must specify the taskType"));
     }
+    if (root.isMember("enableOutputExists"))
+    {
+        m_enableOutputExists = root["enableOutputExists"].asBool();
+    }
     checkNull(root["type"], "taskType");
     m_type = root["type"].asUInt();
     // the taskAlgorithm
@@ -309,6 +313,7 @@ std::string JsonTaskImpl::encode() const
     // sync-result or not
     taskInfo["syncResult"] = m_syncResultToPeer;
     taskInfo["lowBandwidth"] = m_lowBandwidth;
+    taskInfo["enableOutputExists"] = m_enableOutputExists;
 
     Json::Value receiverList;
     for (auto const& it : m_receiverLists)
diff --git a/cpp/wedpr-transport/ppc-rpc/src/RpcMemory.cpp b/cpp/wedpr-transport/ppc-rpc/src/RpcMemory.cpp
@@ -72,9 +72,15 @@ void RpcMemory::cleanTask()
 bcos::Error::Ptr RpcMemory::insertTask(protocol::Task::Ptr _task)
 {
     WriteGuard l(x_tasks);
-    if (m_tasks.find(_task->id()) != m_tasks.end())
+    auto it = m_tasks.find(_task->id());
+    if (it != m_tasks.end())
     {
-        return std::make_shared<bcos::Error>(PPCRetCode::WRITE_RPC_STATUS_ERROR, "task exists");
+        auto taskResult = it->second.second;
+        // the task already exists case
+        if (!taskResult || taskResult->status() == toString(TaskStatus::RUNNING))
+        {
+            return std::make_shared<bcos::Error>(PPCRetCode::WRITE_RPC_STATUS_ERROR, "task exists");
+        }
     }
     auto taskResult = std::make_shared<TaskResult>(_task->id());
     taskResult->setStatus(toString(TaskStatus::RUNNING));

Original file line number	Diff line number	Diff line change
`@@ -150,8 +150,6 @@ class OtPIRImpl : public std::enable_shared_from_this<OtPIRImpl>,`
`150`	`150`	`m_senders.erase(it);`
`151`	`151`	`}`
`152`	`152`	`}`
`153`		`- // allow the output-path exists, for ut`
`154`		`- bool m_enableOutputExists = false;`
`155`	`153`	`// 为true时启动时会从配置中加载文件作为匹配源`
`156`	`154`	`bool m_enableMemoryFile = false;`
`157`	`155`	`ppc::protocol::DataResource m_resource;`
Original file line number	Diff line number	Diff line change
`@@ -168,7 +168,7 @@ void CM2020PSIImpl::asyncRunTask()`
`168`	`168`	`auto role = task->selfParty()->partyIndex();`
`169`	`169`	`if (role == uint16_t(PartyType::Client) \|\| task->syncResultToPeer())`
`170`	`170`	`{`
`171`		`- auto writer = loadWriter(task->id(), dataResource, m_enableOutputExists);`
	`171`	`+ auto writer = loadWriter(task->id(), dataResource, task->enableOutputExists());`
`172`	`172`	`taskState->setWriter(writer);`
`173`	`173`	`}`
`174`	`174`
Original file line number	Diff line number	Diff line change
`@@ -148,10 +148,6 @@ class CM2020PSIImpl : public bcos::Worker,`
`148`	`148`	`}`
`149`	`149`	`}`
`150`	`150`
`151`		`-protected:`
`152`		`- // allow the output-path exists, for ut`
`153`		`- bool m_enableOutputExists = false;`
`154`		`-`
`155`	`151`	`private:`
`156`	`152`	`void waitSignal()`
`157`	`153`	`{`
Original file line number	Diff line number	Diff line change
`@@ -55,7 +55,7 @@ void EcdhConnPSIImpl::asyncRunTask(`
`55`	`55`	`if (role == uint16_t(PartyType::Client))`
`56`	`56`	`{`
`57`	`57`	`ECDH_CONN_LOG(INFO) << LOG_DESC("Client do the Task") << LOG_KV("taskID", _task->id());`
`58`		`- auto writer = loadWriter(_task->id(), dataResource, m_enableOutputExists);`
	`58`	`+ auto writer = loadWriter(_task->id(), dataResource, _task->enableOutputExists());`
`59`	`59`	`ecdhTaskState->setWriter(writer);`
`60`	`60`	`auto client = std::make_shared<EcdhConnPSIClient>(m_config, ecdhTaskState);`
`61`	`61`	`addClient(client);`
`@@ -66,7 +66,7 @@ void EcdhConnPSIImpl::asyncRunTask(`
`66`	`66`	`ECDH_CONN_LOG(INFO) << LOG_DESC("Server do the Task") << LOG_KV("taskID", _task->id());`
`67`	`67`	`if (_task->syncResultToPeer())`
`68`	`68`	`{`
`69`		`- auto writer = loadWriter(_task->id(), dataResource, m_enableOutputExists);`
	`69`	`+ auto writer = loadWriter(_task->id(), dataResource, _task->enableOutputExists());`
`70`	`70`	`ecdhTaskState->setWriter(writer);`
`71`	`71`	`}`
`72`	`72`	`auto server = std::make_shared<EcdhConnPSIServer>(m_config, ecdhTaskState);`
Original file line number	Diff line number	Diff line change
`@@ -155,7 +155,7 @@ void EcdhMultiPSIImpl::asyncRunTask(`
`155`	`155`	`<< LOG_KV("roleId", role);`
`156`	`156`	`if (role == uint16_t(PartiesType::Calculator))`
`157`	`157`	`{`
`158`		`- auto writer = loadWriter(_task->id(), dataResource, m_enableOutputExists);`
	`158`	`+ auto writer = loadWriter(_task->id(), dataResource, _task->enableOutputExists());`
`159`	`159`	`taskState->setWriter(writer);`
`160`	`160`	`ECDH_MULTI_LOG(INFO) << LOG_DESC("Calculator do the Task")`
`161`	`161`	`<< LOG_KV("taskID", _task->id());`
`@@ -170,7 +170,7 @@ void EcdhMultiPSIImpl::asyncRunTask(`
`170`	`170`	`if (_task->syncResultToPeer() && std::find(receivers.begin(), receivers.end(),`
`171`	`171`	`m_config->selfParty()) != receivers.end())`
`172`	`172`	`{`
`173`		`- auto writer = loadWriter(_task->id(), dataResource, m_enableOutputExists);`
	`173`	`+ auto writer = loadWriter(_task->id(), dataResource, _task->enableOutputExists());`
`174`	`174`	`taskState->setWriter(writer);`
`175`	`175`	`}`
`176`	`176`	`auto partner = std::make_shared<EcdhMultiPSIPartner>(m_config, taskState);`
`@@ -183,7 +183,7 @@ void EcdhMultiPSIImpl::asyncRunTask(`
`183`	`183`	`if (_task->syncResultToPeer() && std::find(receivers.begin(), receivers.end(),`
`184`	`184`	`m_config->selfParty()) != receivers.end())`
`185`	`185`	`{`
`186`		`- auto writer = loadWriter(_task->id(), dataResource, m_enableOutputExists);`
	`186`	`+ auto writer = loadWriter(_task->id(), dataResource, _task->enableOutputExists());`
`187`	`187`	`taskState->setWriter(writer);`
`188`	`188`	`}`
`189`	`189`	`auto master = std::make_shared<EcdhMultiPSIMaster>(m_config, taskState);`