add storage option check

cyjseagull · cyjseagull · commit 6026a71bb67a · 2024-11-14T20:07:53.000+08:00
diff --git a/cpp/ppc-framework/protocol/Protocol.h b/cpp/ppc-framework/protocol/Protocol.h
@@ -23,7 +23,7 @@
 #define WIN32_LEAN_AND_MEAN
 #include <windows.h>
 #endif
-
+#include "ppc-framework/Common.h"
 #include <bcos-utilities/Log.h>
 #include <map>
 #include <memory>
@@ -402,6 +402,35 @@ struct SQLConnectionOption
     std::string user;
     std::string password;
     std::string database;
+    void check() const
+    {
+        if (host.size() == 0)
+        {
+            BOOST_THROW_EXCEPTION(WeDPRException() << bcos::errinfo_comment(
+                                      "Invalid SQL option: Must set the host!"));
+        }
+        if (user.size() == 0)
+        {
+            BOOST_THROW_EXCEPTION(WeDPRException() << bcos::errinfo_comment(
+                                      "Invalid SQL option: Must set the user!"));
+        }
+        if (password.size() == 0)
+        {
+            BOOST_THROW_EXCEPTION(WeDPRException() << bcos::errinfo_comment(
+                                      "Invalid SQL option: Must set the password!"));
+        }
+        if (database.size() == 0)
+        {
+            BOOST_THROW_EXCEPTION(WeDPRException() << bcos::errinfo_comment(
+                                      "Invalid SQL option: Must set the database!"));
+        }
+        if (port == 0 || port > 65535)
+        {
+            BOOST_THROW_EXCEPTION(WeDPRException() << bcos::errinfo_comment(
+                                      "Invalid SQL Option, Must set valid port!"));
+        }
+    }
+
     inline std::string desc() const
     {
         std::stringstream oss;
@@ -423,6 +452,25 @@ struct FileStorageConnectionOption
     bool replaceDataNodeOnFailure = false;
     // the default connection-timeout for the hdfs is 1000ms
     uint16_t connectionTimeout = 1000;
+
+    void check() const
+    {
+        if (nameNode.size() == 0)
+        {
+            BOOST_THROW_EXCEPTION(WeDPRException() << bcos::errinfo_comment(
+                                      "Invalid HDFS Option, Must set the nameNode!"));
+        }
+        if (userName.size() == 0)
+        {
+            BOOST_THROW_EXCEPTION(WeDPRException() << bcos::errinfo_comment(
+                                      "Invalid HDFS Option, Must set the userName!"));
+        }
+        if (nameNodePort == 0 || nameNodePort > 65535)
+        {
+            BOOST_THROW_EXCEPTION(WeDPRException() << bcos::errinfo_comment(
+                                      "Invalid HDFS Option, Must set valid namenodeport!"));
+        }
+    }
     inline std::string desc() const
     {
         std::stringstream oss;
diff --git a/cpp/wedpr-computing/ppc-psi/src/ecdh-multi-psi/EcdhMultiCache.cpp b/cpp/wedpr-computing/ppc-psi/src/ecdh-multi-psi/EcdhMultiCache.cpp
@@ -266,5 +266,5 @@ void CalculatorCache::setIntersectionCipher(std::map<uint32_t, bcos::bytes>&& _c
     m_intersectionCipher = std::move(_cipherData);
     m_receiveIntersection = true;
     ECDH_MULTI_LOG(INFO) << LOG_DESC("setIntersectionCipher")
-                         << LOG_KV("dataSize", _cipherData.size()) << printCacheState();
+                         << LOG_KV("dataSize", m_intersectionCipher.size()) << printCacheState();
 }
diff --git a/cpp/wedpr-computing/ppc-psi/src/ecdh-multi-psi/core/EcdhMultiPSICalculator.cpp b/cpp/wedpr-computing/ppc-psi/src/ecdh-multi-psi/core/EcdhMultiPSICalculator.cpp
@@ -167,7 +167,6 @@ void EcdhMultiPSICalculator::loadAndEncrypt(std::string _taskID, bcos::bytes _ra
                     seq);
                 dataOffset += dataBatch->size();
             }
-            dataBatch->release();
         } while (!m_taskState->sqlReader());
     }
     catch (std::exception& e)
diff --git a/cpp/wedpr-computing/ppc-psi/src/ecdh-multi-psi/core/EcdhMultiPSIMaster.cpp b/cpp/wedpr-computing/ppc-psi/src/ecdh-multi-psi/core/EcdhMultiPSIMaster.cpp
@@ -98,6 +98,8 @@ void EcdhMultiPSIMaster::encAndSendIntersectionData()
     for (auto& calcultor : m_calculatorParties)
     {
         ECDH_MASTER_LOG(INFO) << LOG_DESC("send intersection cipher to calculator")
+                              << LOG_KV("taskID", m_taskState->task()->id())
+                              << LOG_KV("intersectionSize", encryptedData.size())
                               << LOG_KV("target", calcultor.first);
         m_config->generateAndSendPPCMessage(calcultor.first, m_taskID, message,
             [self = weak_from_this()](bcos::Error::Ptr&& _error) {
diff --git a/cpp/wedpr-helper/ppc-tools/src/config/PPCConfig.cpp b/cpp/wedpr-helper/ppc-tools/src/config/PPCConfig.cpp
@@ -559,7 +559,7 @@ void PPCConfig::loadHDFSConfig(boost::property_tree::ptree const& _pt)
     // the name node port
     option->nameNodePort = _pt.get<int>("hdfs_storage.name_node_port", 8020);
     // the user
-    option->userName = _pt.get<std::string>("hdfs_storage.user", "root");
+    option->userName = _pt.get<std::string>("hdfs_storage.user", "");
     // checkNonEmptyField("hdfs_storage.user", option->userName);
     // the token
     option->token = _pt.get<std::string>("hdfs_storage.token", "");
diff --git a/cpp/wedpr-storage/ppc-io/src/DataResourceLoaderImpl.cpp b/cpp/wedpr-storage/ppc-io/src/DataResourceLoaderImpl.cpp
@@ -46,19 +46,40 @@ DataResourceLoaderImpl::DataResourceLoaderImpl(
     m_sqlStorageFactory(_sqlStorageFactory),
     m_fileStorageFactory(_fileStorageFactory),
     m_remoteStorageFactory(_remoteStorageFactory)
+{}
+
+void DataResourceLoaderImpl::lazyLoadHdfsStorage()
 {
-    if (m_fileStorageConnectionOpt)
+    if (!m_fileStorageConnectionOpt)
     {
-        m_hdfsStorage = m_fileStorageFactory->createFileStorage(
-            DataResourceType::HDFS, m_fileStorageConnectionOpt);
+        return;
     }
-    if (m_sqlConnectionOpt)
+    bcos::Guard l(x_hdfsStorage);
+    if (m_hdfsStorage)
     {
-        m_sqlStorage =
-            m_sqlStorageFactory->createSQLStorage(DataResourceType::MySQL, m_sqlConnectionOpt);
+        return;
     }
+    m_hdfsStorage =
+        m_fileStorageFactory->createFileStorage(DataResourceType::HDFS, m_fileStorageConnectionOpt);
+    IO_LOG(INFO) << LOG_DESC("lazyLoadHdfsStorage") << m_fileStorageConnectionOpt->desc();
+    return;
 }
 
+void DataResourceLoaderImpl::lazyLoadSqlStorage()
+{
+    if (!m_sqlConnectionOpt)
+    {
+        return;
+    }
+    bcos::Guard l(x_sqlStorage);
+    if (m_sqlStorage)
+    {
+        return;
+    }
+    m_sqlStorage =
+        m_sqlStorageFactory->createSQLStorage(DataResourceType::MySQL, m_sqlConnectionOpt);
+    IO_LOG(INFO) << LOG_DESC("lazyLoadSqlStorage") << m_sqlConnectionOpt->desc();
+}
 
 LineReader::Ptr DataResourceLoaderImpl::loadReader(DataResourceDesc::ConstPtr _desc,
     DataSchema _schema, bool _parseByColumn, FileStorage::Ptr const& _fileStorage)
@@ -104,6 +125,7 @@ LineReader::Ptr DataResourceLoaderImpl::loadSQLResource(
     }
     else if (m_sqlConnectionOpt)
     {
+        lazyLoadSqlStorage();
         storage = m_sqlStorage;
     }
     else
@@ -133,6 +155,7 @@ LineReader::Ptr DataResourceLoaderImpl::loadHDFSResource(
         }
         else if (m_fileStorageConnectionOpt)
         {
+            lazyLoadHdfsStorage();
             storage = m_hdfsStorage;
         }
         else
@@ -170,6 +193,7 @@ void DataResourceLoaderImpl::deleteResource(
             }
             else if (m_fileStorageConnectionOpt)
             {
+                lazyLoadHdfsStorage();
                 storage = m_hdfsStorage;
             }
             else
@@ -219,6 +243,7 @@ void DataResourceLoaderImpl::renameResource(ppc::protocol::DataResourceDesc::Con
             }
             else if (m_fileStorageConnectionOpt)
             {
+                lazyLoadHdfsStorage();
                 storage = m_hdfsStorage;
             }
             else
@@ -277,6 +302,7 @@ void DataResourceLoaderImpl::checkResourceExists(
             }
             else if (m_fileStorageConnectionOpt)
             {
+                lazyLoadHdfsStorage();
                 storage = m_hdfsStorage;
             }
             else
@@ -330,6 +356,7 @@ LineWriter::Ptr DataResourceLoaderImpl::loadWriter(
             }
             else if (m_fileStorageConnectionOpt)
             {
+                lazyLoadHdfsStorage();
                 storage = m_hdfsStorage;
             }
             else
diff --git a/cpp/wedpr-storage/ppc-io/src/DataResourceLoaderImpl.h b/cpp/wedpr-storage/ppc-io/src/DataResourceLoaderImpl.h
@@ -63,13 +63,19 @@ class DataResourceLoaderImpl : public DataResourceLoader
     virtual LineReader::Ptr loadHDFSResource(ppc::protocol::DataResourceDesc::ConstPtr _desc,
         ppc::storage::FileStorage::Ptr const& _fileStorage);
 
+    void lazyLoadHdfsStorage();
+    void lazyLoadSqlStorage();
+
 private:
     // the sql storage
     ppc::protocol::SQLConnectionOption::Ptr m_sqlConnectionOpt;
     ppc::storage::SQLStorage::Ptr m_sqlStorage;
+    bcos::Mutex x_sqlStorage;
+
     // the hdfs storage
     ppc::protocol::FileStorageConnectionOption::Ptr m_fileStorageConnectionOpt;
     ppc::storage::FileStorage::Ptr m_hdfsStorage;
+    bcos::Mutex x_hdfsStorage;
 
     ppc::protocol::RemoteStorageConnectionOption::Ptr m_remoteStorageConnectionOpt;
 
diff --git a/cpp/wedpr-storage/ppc-storage/src/FileStorageFactoryImpl.h b/cpp/wedpr-storage/ppc-storage/src/FileStorageFactoryImpl.h
@@ -36,6 +36,7 @@ class FileStorageFactoryImpl : public FileStorageFactory
         {
         case ppc::protocol::DataResourceType::HDFS:
         {
+            _option->check();
             return std::make_shared<HDFSStorage>(_option);
         }
         default:
diff --git a/cpp/wedpr-storage/ppc-storage/src/SQLStorageFactoryImpl.h b/cpp/wedpr-storage/ppc-storage/src/SQLStorageFactoryImpl.h
@@ -39,6 +39,7 @@ class SQLStorageFactoryImpl : public SQLStorageFactory
         {
         case ppc::protocol::DataResourceType::MySQL:
         {
+            _option->check();
             return std::make_shared<MySQLStorage>(_option);
         }
         default:

Original file line number	Diff line number	Diff line change
`@@ -266,5 +266,5 @@ void CalculatorCache::setIntersectionCipher(std::map<uint32_t, bcos::bytes>&& _c`
`266`	`266`	`m_intersectionCipher = std::move(_cipherData);`
`267`	`267`	`m_receiveIntersection = true;`
`268`	`268`	`ECDH_MULTI_LOG(INFO) << LOG_DESC("setIntersectionCipher")`
`269`		`- << LOG_KV("dataSize", _cipherData.size()) << printCacheState();`
	`269`	`+ << LOG_KV("dataSize", m_intersectionCipher.size()) << printCacheState();`
`270`	`270`	`}`
Original file line number	Diff line number	Diff line change
`@@ -167,7 +167,6 @@ void EcdhMultiPSICalculator::loadAndEncrypt(std::string _taskID, bcos::bytes _ra`
`167`	`167`	`seq);`
`168`	`168`	`dataOffset += dataBatch->size();`
`169`	`169`	`}`
`170`		`- dataBatch->release();`
`171`	`170`	`} while (!m_taskState->sqlReader());`
`172`	`171`	`}`
`173`	`172`	`catch (std::exception& e)`
Original file line number	Diff line number	Diff line change
`@@ -98,6 +98,8 @@ void EcdhMultiPSIMaster::encAndSendIntersectionData()`
`98`	`98`	`for (auto& calcultor : m_calculatorParties)`
`99`	`99`	`{`
`100`	`100`	`ECDH_MASTER_LOG(INFO) << LOG_DESC("send intersection cipher to calculator")`
	`101`	`+ << LOG_KV("taskID", m_taskState->task()->id())`
	`102`	`+ << LOG_KV("intersectionSize", encryptedData.size())`
`101`	`103`	`<< LOG_KV("target", calcultor.first);`
`102`	`104`	`m_config->generateAndSendPPCMessage(calcultor.first, m_taskID, message,`
`103`	`105`	`[self = weak_from_this()](bcos::Error::Ptr&& _error) {`
Original file line number	Diff line number	Diff line change
`@@ -46,19 +46,40 @@ DataResourceLoaderImpl::DataResourceLoaderImpl(`
`46`	`46`	`m_sqlStorageFactory(_sqlStorageFactory),`
`47`	`47`	`m_fileStorageFactory(_fileStorageFactory),`
`48`	`48`	`m_remoteStorageFactory(_remoteStorageFactory)`
	`49`	`+{}`
	`50`	`+`
	`51`	`+void DataResourceLoaderImpl::lazyLoadHdfsStorage()`
`49`	`52`	`{`
`50`		`- if (m_fileStorageConnectionOpt)`
	`53`	`+ if (!m_fileStorageConnectionOpt)`
`51`	`54`	`{`
`52`		`- m_hdfsStorage = m_fileStorageFactory->createFileStorage(`
`53`		`- DataResourceType::HDFS, m_fileStorageConnectionOpt);`
	`55`	`+ return;`
`54`	`56`	`}`
`55`		`- if (m_sqlConnectionOpt)`
	`57`	`+ bcos::Guard l(x_hdfsStorage);`
	`58`	`+ if (m_hdfsStorage)`
`56`	`59`	`{`
`57`		`- m_sqlStorage =`
`58`		`- m_sqlStorageFactory->createSQLStorage(DataResourceType::MySQL, m_sqlConnectionOpt);`
	`60`	`+ return;`
`59`	`61`	`}`
	`62`	`+ m_hdfsStorage =`
	`63`	`+ m_fileStorageFactory->createFileStorage(DataResourceType::HDFS, m_fileStorageConnectionOpt);`
	`64`	`+ IO_LOG(INFO) << LOG_DESC("lazyLoadHdfsStorage") << m_fileStorageConnectionOpt->desc();`
	`65`	`+ return;`
`60`	`66`	`}`
`61`	`67`
	`68`	`+void DataResourceLoaderImpl::lazyLoadSqlStorage()`
	`69`	`+{`
	`70`	`+ if (!m_sqlConnectionOpt)`
	`71`	`+ {`
	`72`	`+ return;`
	`73`	`+ }`
	`74`	`+ bcos::Guard l(x_sqlStorage);`
	`75`	`+ if (m_sqlStorage)`
	`76`	`+ {`
	`77`	`+ return;`
	`78`	`+ }`
	`79`	`+ m_sqlStorage =`
	`80`	`+ m_sqlStorageFactory->createSQLStorage(DataResourceType::MySQL, m_sqlConnectionOpt);`
	`81`	`+ IO_LOG(INFO) << LOG_DESC("lazyLoadSqlStorage") << m_sqlConnectionOpt->desc();`
	`82`	`+}`
`62`	`83`
`63`	`84`	`LineReader::Ptr DataResourceLoaderImpl::loadReader(DataResourceDesc::ConstPtr _desc,`
`64`	`85`	`DataSchema _schema, bool _parseByColumn, FileStorage::Ptr const& _fileStorage)`
`@@ -104,6 +125,7 @@ LineReader::Ptr DataResourceLoaderImpl::loadSQLResource(`
`104`	`125`	`}`
`105`	`126`	`else if (m_sqlConnectionOpt)`
`106`	`127`	`{`
	`128`	`+ lazyLoadSqlStorage();`
`107`	`129`	`storage = m_sqlStorage;`
`108`	`130`	`}`
`109`	`131`	`else`
`@@ -133,6 +155,7 @@ LineReader::Ptr DataResourceLoaderImpl::loadHDFSResource(`
`133`	`155`	`}`
`134`	`156`	`else if (m_fileStorageConnectionOpt)`
`135`	`157`	`{`
	`158`	`+ lazyLoadHdfsStorage();`
`136`	`159`	`storage = m_hdfsStorage;`
`137`	`160`	`}`
`138`	`161`	`else`
`@@ -170,6 +193,7 @@ void DataResourceLoaderImpl::deleteResource(`
`170`	`193`	`}`
`171`	`194`	`else if (m_fileStorageConnectionOpt)`
`172`	`195`	`{`
	`196`	`+ lazyLoadHdfsStorage();`
`173`	`197`	`storage = m_hdfsStorage;`
`174`	`198`	`}`
`175`	`199`	`else`
`@@ -219,6 +243,7 @@ void DataResourceLoaderImpl::renameResource(ppc::protocol::DataResourceDesc::Con`
`219`	`243`	`}`
`220`	`244`	`else if (m_fileStorageConnectionOpt)`
`221`	`245`	`{`
	`246`	`+ lazyLoadHdfsStorage();`
`222`	`247`	`storage = m_hdfsStorage;`
`223`	`248`	`}`
`224`	`249`	`else`
`@@ -277,6 +302,7 @@ void DataResourceLoaderImpl::checkResourceExists(`
`277`	`302`	`}`
`278`	`303`	`else if (m_fileStorageConnectionOpt)`
`279`	`304`	`{`
	`305`	`+ lazyLoadHdfsStorage();`
`280`	`306`	`storage = m_hdfsStorage;`
`281`	`307`	`}`
`282`	`308`	`else`
`@@ -330,6 +356,7 @@ LineWriter::Ptr DataResourceLoaderImpl::loadWriter(`
`330`	`356`	`}`
`331`	`357`	`else if (m_fileStorageConnectionOpt)`
`332`	`358`	`{`
	`359`	`+ lazyLoadHdfsStorage();`
`333`	`360`	`storage = m_hdfsStorage;`
`334`	`361`	`}`
`335`	`362`	`else`
Original file line number	Diff line number	Diff line change
`@@ -36,6 +36,7 @@ class FileStorageFactoryImpl : public FileStorageFactory`
`36`	`36`	`{`
`37`	`37`	`case ppc::protocol::DataResourceType::HDFS:`
`38`	`38`	`{`
	`39`	`+ _option->check();`
`39`	`40`	`return std::make_shared<HDFSStorage>(_option);`
`40`	`41`	`}`
`41`	`42`	`default:`
Original file line number	Diff line number	Diff line change
`@@ -39,6 +39,7 @@ class SQLStorageFactoryImpl : public SQLStorageFactory`
`39`	`39`	`{`
`40`	`40`	`case ppc::protocol::DataResourceType::MySQL:`
`41`	`41`	`{`
	`42`	`+ _option->check();`
`42`	`43`	`return std::make_shared<MySQLStorage>(_option);`
`43`	`44`	`}`
`44`	`45`	`default:`