optimize thread pool

Signed-off-by: liduo <liduo29@huawei.com>

src/binary_file_packer.cpp

@@ -24,14 +24,12 @@ namespace Restool {
 using namespace std;
 
 BinaryFilePacker::BinaryFilePacker(const PackageParser &packageParser, const std::string &moduleName)
-    : packageParser_(packageParser), moduleName_(moduleName), threadPool_(ThreadPool(THREAD_POOL_SIZE))
+    : packageParser_(packageParser), moduleName_(moduleName)
 {
-    threadPool_.Start();
 }
 
 BinaryFilePacker::~BinaryFilePacker()
 {
-    threadPool_.Stop();
 }
 
 void BinaryFilePacker::StopCopy()
@@ -42,7 +40,7 @@ void BinaryFilePacker::StopCopy()
 std::future<uint32_t> BinaryFilePacker::CopyBinaryFileAsync(const std::vector<std::string> &inputs)
 {
     auto func = [this](const vector<string> &inputs) { return this->CopyBinaryFile(inputs); };
-    std::future<uint32_t> res = threadPool_.Enqueue(func, inputs);
+    std::future<uint32_t> res = ThreadPool::GetInstance().Enqueue(func, inputs);
     return res;
 }
 
@@ -126,7 +124,7 @@ uint32_t BinaryFilePacker::CopyBinaryFileImpl(const string &src, const string &d
 
         string path = entry->GetFilePath().GetPath();
         auto copyFunc = [this](const string path, string subPath) { return this->CopySingleFile(path, subPath); };
-        std::future<uint32_t> res = threadPool_.Enqueue(copyFunc, path, subPath);
+        std::future<uint32_t> res = ThreadPool::GetInstance().Enqueue(copyFunc, path, subPath);
         copyResults_.push_back(std::move(res));
     }
     return RESTOOL_SUCCESS;

src/cmd/package_parser.cpp

@@ -15,6 +15,7 @@
 
 #include "cmd/package_parser.h"
 #include <algorithm>
+#include <climits>
 #include "resconfig_parser.h"
 #include "resource_util.h"
 #include "select_compile_parse.h"
@@ -46,6 +47,7 @@ const struct option PackageParser::CMD_OPTS[] = {
     { "target-config", required_argument, nullptr, Option::TARGET_CONFIG},
     { "defined-sysids", required_argument, nullptr, Option::DEFINED_SYSIDS},
     { "compressed-config", required_argument, nullptr, Option::COMPRESSED_CONFIG},
+    { "thread", required_argument, nullptr, Option::THREAD},
     { 0, 0, 0, 0},
 };
 
@@ -410,6 +412,31 @@ uint32_t PackageParser::ParseTargetConfig(const string& argValue)
     return RESTOOL_SUCCESS;
 }
 
+uint32_t PackageParser::ParseThread(const std::string &argValue)
+{
+    if (argValue.empty()) {
+        return RESTOOL_SUCCESS;
+    }
+    char *end;
+    errno = 0;
+    int count = static_cast<int>(strtol(argValue.c_str(), &end, 10));
+    if (end == argValue.c_str() || errno == ERANGE || *end != '\0' || count == INT_MIN || count == INT_MAX) {
+        cerr << "Error: Invalid thread count: " << argValue << endl;
+        return RESTOOL_ERROR;
+    }
+    if (count <= 0) {
+        cerr << "Error: Invalid thread count: " << count << endl;
+        return RESTOOL_ERROR;
+    }
+    threadCount_ = count;
+    return RESTOOL_SUCCESS;
+}
+
+size_t PackageParser::GetThreadCount() const
+{
+    return threadCount_;
+}
+
 const TargetConfig &PackageParser::GetTargetConfigValues() const
 {
     return targetConfig_;
@@ -483,6 +510,7 @@ void PackageParser::InitCommand()
     handles_.emplace(Option::TARGET_CONFIG, bind(&PackageParser::ParseTargetConfig, this, _1));
     handles_.emplace(Option::DEFINED_SYSIDS, bind(&PackageParser::AddSysIdDefined, this, _1));
     handles_.emplace(Option::COMPRESSED_CONFIG, bind(&PackageParser::AddCompressionPath, this, _1));
+    handles_.emplace(Option::THREAD, bind(&PackageParser::ParseThread, this, _1));
 }
 
 uint32_t PackageParser::HandleProcess(int c, const string& argValue)

src/generic_compiler.cpp

@@ -37,12 +37,10 @@ GenericCompiler::~GenericCompiler()
 uint32_t GenericCompiler::CompileFiles(const std::vector<FileInfo> &fileInfos)
 {
     cout << "Info: GenericCompiler::CompileFiles" << endl;
-    ThreadPool pool(THREAD_POOL_SIZE);
-    pool.Start();
     std::vector<std::future<uint32_t>> results;
     for (const auto &fileInfo : fileInfos) {
         auto taskFunc = [this](const FileInfo &fileInfo) { return this->CompileSingleFile(fileInfo); };
-        results.push_back(pool.Enqueue(taskFunc, fileInfo));
+        results.push_back(ThreadPool::GetInstance().Enqueue(taskFunc, fileInfo));
     }
     for (auto &ret : results) {
         if (ret.get() != RESTOOL_SUCCESS) {

src/resconfig_parser.cpp

@@ -92,6 +92,8 @@ void ResConfigParser::InitFileListCommand(HandleBack callback)
         Option::DEFINED_SYSIDS, callback));
     fileListHandles_.emplace("compression", bind(&ResConfigParser::GetString, this, _1,
         Option::COMPRESSED_CONFIG, callback));
+    fileListHandles_.emplace("thread", bind(&ResConfigParser::GetNumber, this, _1,
+        Option::THREAD, callback));
 }
 
 uint32_t ResConfigParser::GetString(const cJSON *node, int c, HandleBack callback)
@@ -163,6 +165,19 @@ uint32_t ResConfigParser::GetBool(const cJSON *node, int c, HandleBack callback)
     }
     return RESTOOL_SUCCESS;
 }
+
+uint32_t ResConfigParser::GetNumber(const cJSON *node, int c, HandleBack callback)
+{
+    if (!node || !cJSON_IsNumber(node)) {
+        cerr << "Error: GetNumber node not number. Option = " << c << endl;
+        return RESTOOL_ERROR;
+    }
+
+    if (callback(c, std::to_string(node->valuedouble)) != RESTOOL_SUCCESS) {
+        return RESTOOL_ERROR;
+    }
+    return RESTOOL_SUCCESS;
+}
 }
 }
 }

src/resource_pack.cpp

@@ -86,6 +86,9 @@ uint32_t ResourcePack::InitResourcePack()
     if (InitModule() != RESTOOL_SUCCESS) {
         return RESTOOL_ERROR;
     }
+    if (ThreadPool::GetInstance().Start(packageParser_.GetThreadCount()) != RESTOOL_SUCCESS) {
+        return RESTOOL_ERROR;
+    };
     return RESTOOL_SUCCESS;
 }
 

src/thread_pool.cpp

@@ -24,18 +24,31 @@ namespace Global {
 namespace Restool {
 using namespace std;
 
-ThreadPool::ThreadPool(size_t threadCount) : threadCount_(threadCount)
+ThreadPool::ThreadPool()
 {}
 
-uint32_t ThreadPool::Start()
+ThreadPool &ThreadPool::GetInstance()
 {
-    if (!workerThreads_.empty() || threadCount_ <= 0) {
-        cerr << "Error: ThreadPool start failed." << endl;
-        return RESTOOL_ERROR;
+    static ThreadPool pool;
+    return pool;
+}
+
+uint32_t ThreadPool::Start(const size_t &threadCount)
+{
+    if (!workerThreads_.empty()) {
+        cout << "Warning: ThreadPool is already started." << endl;
+        return RESTOOL_SUCCESS;
     }
+    size_t hardwareCount = std::thread::hardware_concurrency();
+    cout << "Info: hardware concurrency count is : " << hardwareCount << endl;
+    size_t count = threadCount <= 0 ? (hardwareCount <= 0 ? DEFAULT_POOL_SIZE : hardwareCount) : threadCount;
+    if (count == 1) {
+        count++;
+    }
+    cout << "Info: thread count is : " << count << endl;
     running_ = true;
-    workerThreads_.reserve(threadCount_);
-    for (size_t i = 0; i < threadCount_; ++i) {
+    workerThreads_.reserve(count);
+    for (size_t i = 0; i < count; ++i) {
         workerThreads_.emplace_back([this] { this->WorkInThread(); });
     }
     cout << "Info: thread pool is started" << endl;