update install_deps

Author: cyjseagull

cpp/wedpr-computing/ppc-psi/src/ecdh-multi-psi/EcdhMultiCache.cpp

@@ -27,7 +27,7 @@ using namespace bcos;
 void MasterCache::addCalculatorCipher(std::string _peerId,
     std::map<uint32_t, bcos::bytes>&& _cipherData, uint32_t seq, uint32_t dataBatchCount)
 {
-    bcos::WriteGuard lock(x_calculatorCipher);
+    bcos::WriteGuard l(x_calculatorCipher);
     m_calculatorCipher.insert(_cipherData.begin(), _cipherData.end());
     m_calculatorCipherSeqs.insert(seq);
     if (dataBatchCount)
@@ -36,35 +36,42 @@ void MasterCache::addCalculatorCipher(std::string _peerId,
     }
     ECDH_MULTI_LOG(INFO) << LOG_DESC(
                                 "addCalculatorCipher: master receive cipher data from calculator")
-                         << LOG_KV("calculator", _peerId)
-                         << LOG_KV("task", printCacheState(m_taskState))
+                         << LOG_KV("calculator", _peerId) << LOG_KV("task", printCacheState())
                          << LOG_KV("receivedSize", m_calculatorCipherSeqs.size())
                          << LOG_KV("dataBatchCount", m_calculatorDataBatchCount);
     if (m_calculatorDataBatchCount > 0 &&
         m_calculatorCipherSeqs.size() == m_calculatorDataBatchCount)
     {
         ECDH_MULTI_LOG(INFO) << LOG_DESC("The master receive all cipher data from the calculator")
                              << LOG_KV("calculatorId", _peerId)
-                             << LOG_KV("task", printCacheState(m_taskState));
+                             << LOG_KV("task", printCacheState());
         m_finishedPartners.insert(_peerId);
     }
 }
 
 void MasterCache::addPartnerCipher(std::string _peerId, std::vector<bcos::bytes>&& _cipherData,
-    uint32_t seq, uint32_t needSendTimes)
+    uint32_t seq, uint32_t parternerDataCount)
 {
     bcos::WriteGuard lock(x_partnerToCipher);
     if (!m_partnerToCipher.count(_peerId))
     {
-        m_partnerToCipher.insert(std::make_pair(_peerId, std::set()));
+        m_partnerToCipher.insert(std::make_pair(_peerId, std::set<bcos::bytes>()));
     }
     m_partnerToCipher[_peerId].insert(_cipherData.begin(), _cipherData.end());
     m_partnerCipherSeqs[_peerId].insert(seq);
     ECDH_MULTI_LOG(INFO) << LOG_DESC("addPartnerCipher") << LOG_KV("partner", _peerId)
                          << LOG_KV("seqSize", m_partnerCipherSeqs.at(_peerId).size())
-                         << LOG_KV("task", printCacheState(m_taskState));
-
-    if (m_partnerCipherSeqs[_peerId].size() == needSendTimes)
+                         << LOG_KV("task", printCacheState());
+    if (parternerDataCount > 0)
+    {
+        m_parternerDataCount.insert(std::make_pair(_peerId, parternerDataCount));
+    }
+    if (!m_parternerDataCount.count(_peerId))
+    {
+        return;
+    }
+    auto expectedCount = m_parternerDataCount.at(_peerId);
+    if (m_partnerCipherSeqs[_peerId].size() == expectedCount)
     {
         m_finishedPartners.insert(_peerId);
     }
@@ -77,10 +84,9 @@ bool MasterCache::tryToIntersection()
     {
         return false;
     }
-    m_state = CacheState::IntersectionProgressing;
+    m_cacheState = CacheState::IntersectionProgressing;
 
-    ECDH_MULTI_LOG(INFO) << LOG_DESC("tryToIntersection ")
-                         << LOG_KV("task", printCacheState(m_taskState));
+    ECDH_MULTI_LOG(INFO) << LOG_DESC("tryToIntersection ") << LOG_KV("task", printCacheState());
     auto startT = utcSteadyTime();
     // iterator the calculator cipher to obtain intersection
     for (auto&& it : m_calculatorCipher)
@@ -95,74 +101,98 @@ bool MasterCache::tryToIntersection()
                 break;
             }
         }
-        if (intersection)
+        if (insersected)
         {
-            m_intersecCipher.insert(std::make_pair(it.first, std::move(it.second)));
+            m_intersecCipher.emplace_back(std::move(it.second));
+            m_intersecCipherIndex.emplace_back(it.first);
         }
     }
-    m_state = CacheState::Intersectioned;
+    m_cacheState = CacheState::Intersectioned;
     ECDH_MULTI_LOG(INFO) << LOG_DESC("tryToIntersection success")
-                         << LOG_KV("task", printCacheState(m_taskState))
+                         << LOG_KV("task", printCacheState())
                          << LOG_KV("timecost", (utcSteadyTime() - startT));
     return true;
 }
 
-std::vector<bcos::bytes> CalculatorCache::encryptIntersection(bcos::bytes const& randomKey)
+std::vector<std::pair<uint64_t, bcos::bytes>> MasterCache::encryptIntersection(
+    bcos::bytes const& randomKey)
 {
     std::vector<std::pair<uint64_t, bcos::bytes>> cipherData(m_intersecCipher.size());
-    tbb::parallel_for_each(
-        m_intersecCipher.begin(), m_intersecCipher.end(), [&](auto const& _pair) {
-            auto value = _pair.second;
-            auto cipherValue = m_config->eccCrypto()->ecMultiply(value, randomKey);
-            cipherData[i] = std::make_pair(_pair.first, cipherValue);
+    tbb::parallel_for(
+        tbb::blocked_range<size_t>(0U, m_intersecCipher.size()), [&](auto const& range) {
+            for (auto i = range.begin(); i < range.end(); i++)
+            {
+                auto cipherValue =
+                    m_config->eccCrypto()->ecMultiply(m_intersecCipher[i], randomKey);
+                cipherData[i] = std::make_pair(m_intersecCipherIndex[i], cipherValue);
+            }
         });
     return cipherData;
 }
 
-bcos::bytes CalculatorCache::getPlainDataByIndex(uint64_t index) {}
+bcos::bytes CalculatorCache::getPlainDataByIndex(uint64_t index)
+{
+    uint64_t startIndex = 0;
+    uint64_t endIndex = 0;
+    for (auto const& it : m_plainData)
+    {
+        endIndex += it->size();
+        if (index >= startIndex && index < endIndex)
+        {
+            return it->getBytes((index - startIndex));
+        }
+        startIndex += it->size();
+    }
+    return bcos::bytes();
+}
 
-void CalculatorCache::tryToFinalize()
+bool CalculatorCache::tryToFinalize()
 {
     if (!shouldFinalize())
     {
-        return;
+        return false;
     }
     auto startT = utcSteadyTime();
     ECDH_MULTI_LOG(INFO) << LOG_DESC("tryToFinalize: compute intersection")
                          << printTaskInfo(m_taskState->task());
-    m_state = CacheState::Finalizing;
+    m_cacheState = CacheState::Finalizing;
     // find the intersection
     for (auto const& it : m_intersectionCipher)
     {
         if (m_masterCipher.count(it.second))
         {
-            m_intersectionResult.emplace_back(getPlainDataByIndex(it.first));
+            auto ret = getPlainDataByIndex(it.first);
+            if (ret.size() > 0)
+            {
+                m_intersectionResult.emplace_back(ret);
+            }
         }
     }
-    m_state = CacheState::Finalized;
+    m_cacheState = CacheState::Finalized;
     ECDH_MULTI_LOG(INFO) << LOG_DESC("tryToFinalize:  compute intersection success")
                          << printTaskInfo(m_taskState->task())
-                         << LOG_KV("intersectionSize", m_intersectionResult.size());
-    << LOG_KV("timecost", (utcSteadyTime() - startT));
+                         << LOG_KV("intersectionSize", m_intersectionResult.size())
+                         << LOG_KV("timecost", (utcSteadyTime() - startT));
 
     ECDH_MULTI_LOG(INFO) << LOG_DESC("tryToFinalize: syncIntersections")
                          << printTaskInfo(m_taskState->task());
-    m_state = CacheState::Syncing;
+    m_cacheState = CacheState::Syncing;
     syncIntersections();
-    m_state = CacheState::Synced;
+    m_cacheState = CacheState::Synced;
 
-    m_state = CacheState::StoreProgressing;
+    m_cacheState = CacheState::StoreProgressing;
     m_taskState->storePSIResult(m_config->dataResourceLoader(), m_intersectionResult);
-    m_state = CacheState::Stored;
+    m_cacheState = CacheState::Stored;
     ECDH_MULTI_LOG(INFO) << LOG_DESC("tryToFinalize: syncIntersections and store success")
                          << printTaskInfo(m_taskState->task());
+    return true;
 }
 
 void CalculatorCache::syncIntersections()
 {
     ECDH_MULTI_LOG(INFO) << LOG_DESC("syncIntersections") << printTaskInfo(m_taskState->task());
     auto peers = m_taskState->task()->getAllPeerParties();
-    auto taskID = m_taskState->task()->taskID();
+    auto taskID = m_taskState->task()->id();
     // notify task result
     if (!m_syncResult)
     {
@@ -178,8 +208,9 @@ void CalculatorCache::syncIntersections()
                     if (_error && _error->errorCode() != 0)
                     {
                         ECDH_MULTI_LOG(WARNING)
-                            << LOG_DESC("sync task result to peer failed") << LOG_KV("peer", peer)
-                            << LOG_KV("taskID", taskID) << LOG_KV("code", _error->errorCode())
+                            << LOG_DESC("sync task result to peer failed")
+                            << LOG_KV("peer", peer.first) << LOG_KV("taskID", taskID)
+                            << LOG_KV("code", _error->errorCode())
                             << LOG_KV("msg", _error->errorMessage());
                         return;
                     }
@@ -197,12 +228,12 @@ void CalculatorCache::syncIntersections()
     for (auto& peer : peers)
     {
         m_config->generateAndSendPPCMessage(
-            _peer.first, taskID, message,
+            peer.first, taskID, message,
             [taskID, peer](bcos::Error::Ptr&& _error) {
                 if (_error && _error->errorCode() != 0)
                 {
                     ECDH_MULTI_LOG(WARNING)
-                        << LOG_DESC("sync psi result to peer failed") << LOG_KV("peer", peer)
+                        << LOG_DESC("sync psi result to peer failed") << LOG_KV("peer", peer.first)
                         << LOG_KV("taskID", taskID) << LOG_KV("code", _error->errorCode())
                         << LOG_KV("msg", _error->errorMessage());
                     return;

cpp/wedpr-computing/ppc-psi/src/ecdh-multi-psi/EcdhMultiCache.h

@@ -19,10 +19,12 @@
  */
 #pragma once
 #include "Common.h"
+#include "EcdhMultiPSIConfig.h"
 #include "ppc-psi/src/Common.h"
 #include "ppc-psi/src/psi-framework/TaskState.h"
 #include <gperftools/malloc_extension.h>
 #include <memory>
+#include <sstream>
 
 namespace ppc::psi
 {
@@ -31,7 +33,9 @@ class MasterCache
 {
 public:
     using Ptr = std::shared_ptr<MasterCache>;
-    MasterCache(TaskState::Ptr const& taskState) : m_taskState(taskState) {}
+    MasterCache(TaskState::Ptr const& taskState, EcdhMultiPSIConfig::Ptr const& config)
+      : m_taskState(taskState), m_config(config)
+    {}
     virtual ~MasterCache()
     {
         m_intersecCipher.clear();
@@ -40,42 +44,36 @@ class MasterCache
         m_partnerToCipher.clear();
         m_calculatorCipherSeqs.clear();
         m_partnerCipherSeqs.clear();
-        std::map<uint32_t, bcos::bytes>().swap(m_intersecCipher);
+        std::vector<bcos::bytes>().swap(m_intersecCipher);
         std::set<std::string>().swap(m_finishedPartners);
         std::map<uint32_t, bcos::bytes>().swap(m_calculatorCipher);
-        std::map<std::string, std::vector<bcos::bytes>>().swap(m_partnerToCipher);
+        std::map<std::string, std::set<bcos::bytes>>().swap(m_partnerToCipher);
         std::set<uint32_t>().swap(m_calculatorCipherSeqs);
         std::map<std::string, std::set<uint32_t>>().swap(m_partnerCipherSeqs);
         MallocExtension::instance()->ReleaseFreeMemory();
         ECDH_MULTI_LOG(INFO) << LOG_DESC("the master cipher datacache destroyed ")
-                             << LOG_KV("taskID", m_taskID);
+                             << LOG_KV("taskID", m_taskState->task()->id());
     }
 
     void addCalculatorCipher(std::string _peerId, std::map<uint32_t, bcos::bytes>&& _cipherData,
         uint32_t seq, uint32_t dataBatchCount);
 
     void addPartnerCipher(std::string _peerId, std::vector<bcos::bytes>&& _cipherData, uint32_t seq,
-        uint32_t needSendTimes);
-
-    // Note: the m_intersecCipher will not been changed once generated
-    std::map<uint32_t, bcos::bytes> const& intersecCipher() { return m_intersecCipher; }
+        uint32_t parternerDataCount);
 
     bool tryToIntersection();
 
     std::string printCacheState()
     {
         std::ostringstream stringstream;
-        stringstream << LOG_KV("taskID", m_taskState->task()->taskID())
-                     << LOG_KV("CacheState", CacheState)
-                     << LOG_KV("intersectionSize", m_intersecCipher.size());
-        if (_msg->header()->hasOptionalField())
-        {
-            stringstream << printOptionalField(_msg->header()->optionalField());
-        }
+        stringstream << LOG_KV("taskID", m_taskState->task()->id())
+                     << LOG_KV("CacheState", m_cacheState)
+                     << LOG_KV("intersectionSize", m_intersecCipher.size())
+                     << printTaskInfo(m_taskState->task());
         return stringstream.str();
     }
 
-    std::vector<bcos::bytes> encryptIntersection(bcos::bytes const& randomKey);
+    std::vector<std::pair<uint64_t, bcos::bytes>> encryptIntersection(bcos::bytes const& randomKey);
 
 private:
     bool shouldIntersection()
@@ -102,11 +100,14 @@ class MasterCache
 
 private:
     TaskState::Ptr m_taskState;
+    EcdhMultiPSIConfig::Ptr m_config;
     CacheState m_cacheState = CacheState::Evaluating;
 
     // the intersection cipher data of the master
     // calculator data index ==> cipher
-    std::map<uint32_t, bcos::bytes> m_intersecCipher;
+    std::vector<bcos::bytes> m_intersecCipher;
+    std::vector<uint32_t> m_intersecCipherIndex;
+
     std::set<std::string> m_finishedPartners;
     // the cipher data from calculator to master
     std::map<uint32_t, bcos::bytes> m_calculatorCipher;
@@ -119,6 +120,7 @@ class MasterCache
     bcos::SharedMutex x_partnerToCipher;
     // partnerId=>received partner seqs
     std::map<std::string, std::set<uint32_t>> m_partnerCipherSeqs;
+    std::map<std::string, uint32_t> m_parternerDataCount;
 
     bcos::Mutex m_mutex;
 };
@@ -127,8 +129,9 @@ class CalculatorCache
 {
 public:
     using Ptr = std::shared_ptr<CalculatorCache>;
-    CalculatorCache(TaskState::Ptr const& taskState, bool syncResult)
-      : m_taskState(taskState), m_syncResult(syncResult)
+    CalculatorCache(
+        TaskState::Ptr const& taskState, bool syncResult, EcdhMultiPSIConfig::Ptr const& config)
+      : m_taskState(taskState), m_syncResult(syncResult), m_config(config)
     {}
     virtual ~CalculatorCache()
     {
@@ -137,42 +140,39 @@ class CalculatorCache
         m_receivedMasterCipher.clear();
         m_masterCipher.clear();
         m_intersectionCipher.clear();
-        m_calculatorIntersectionCipherDataFinalMap.clear();
         std::set<uint32_t>().swap(m_CipherDataFromCalculatorSubSeq);
         std::set<uint32_t>().swap(m_calculatorIntersectionSubSeq);
         std::set<uint32_t>().swap(m_receivedMasterCipher);
-        std::vector<bcos::bytes>().swap(m_masterCipher);
+        std::set<bcos::bytes>().swap(m_masterCipher);
         std::map<uint32_t, bcos::bytes>().swap(m_intersectionCipher);
-        std::map<uint32_t, bcos::bytes>().swap(m_calculatorIntersectionCipherDataFinalMap);
         MallocExtension::instance()->ReleaseFreeMemory();
         ECDH_MULTI_LOG(INFO) << LOG_DESC("the calculator cipher datacache destroyed")
-                             << LOG_KV("taskID", m_taskID);
+                             << LOG_KV("taskID", m_taskState->task()->id());
     }
 
-    void tryToFinalize();
+    bool tryToFinalize();
 
     bool appendMasterCipher(
         std::vector<bcos::bytes>&& _cipherData, uint32_t seq, uint32_t dataBatchSize);
 
     void setIntersectionCipher(std::map<uint32_t, bcos::bytes>&& _cipherData);
 
-    std::map<uint32_t, bcos::bytes> const& calculatorIntersectionCipherDataFinalMap()
-    {
-        return m_calculatorIntersectionCipherDataFinalMap;
-    }
-
-
     void appendPlainData(ppc::io::DataBatch::Ptr const& data)
     {
         bcos::WriteGuard l(x_plainData);
-        m_plainData->emplace_back(data);
+        m_plainData.emplace_back(data);
     }
 
 private:
+    bcos::bytes getPlainDataByIndex(uint64_t index);
     bool shouldFinalize()
     {
         // only can finalize in Evaluating state
-        if (m_state != CacheState::Evaluating)
+        if (m_cacheState != CacheState::Evaluating)
+        {
+            return false;
+        }
+        if (m_receivedMasterCipher.size() == 0)
         {
             return false;
         }
@@ -183,8 +183,9 @@ class CalculatorCache
 
 private:
     TaskState::Ptr m_taskState;
-    CacheState m_cacheState = CacheState::Evaluating;
     bool m_syncResult;
+    EcdhMultiPSIConfig::Ptr m_config;
+    CacheState m_cacheState = CacheState::Evaluating;
 
     std::vector<ppc::io::DataBatch::Ptr> m_plainData;
     bcos::SharedMutex x_plainData;