runner config add param: workers_num

include/api/model_parallel_runner.h

@@ -24,6 +24,7 @@
 namespace mindspore {
 struct RunnerConfig {
   std::shared_ptr<Context> context = nullptr;
+  int workers_num = 0;
 };
 
 /// \brief The ModelParallelRunner class is used to define a MindSpore ModelParallelRunner, facilitating Model

mindspore/lite/src/cxx_api/model_pool/model_pool.cc

@@ -20,7 +20,7 @@
 #include "include/lite_types.h"
 #include "src/common/config_file.h"
 #include "src/runtime/inner_allocator.h"
-#include "src/common//file_utils.h"
+#include "src/common/file_utils.h"
 #include "src/pack_weight_manager.h"
 namespace mindspore {
 namespace {
@@ -89,8 +89,10 @@ std::shared_ptr<Context> ModelPool::InitContext(const std::shared_ptr<RunnerConf
     }
     if (device->GetDeviceType() == kGPU) {
       num_models_ = 1;
-    } else {
+    } else if (runner_config->workers_num == 0) {
       num_models_ = GetCoreNum() / static_cast<int>(model_context->GetThreadNum());
+    } else {
+      num_models_ = runner_config->workers_num;
     }
   } else {
     MS_LOG(DEBUG) << "use default config.";
@@ -172,12 +174,20 @@ Status ModelPool::Init(const std::string &model_path, const std::shared_ptr<Runn
     return kLiteError;
   }
   size_t size = 0;
+  if (graph_buf_ != nullptr) {
+    delete[] graph_buf_;
+    graph_buf_ = nullptr;
+  }
   graph_buf_ = lite::ReadFile(model_path.c_str(), &size);
   if (graph_buf_ == nullptr) {
     MS_LOG(ERROR) << "read file failed.";
     return kLiteError;
   }
-  lite::PackWeightManager::GetInstance()->InitWeightManagerByBuf(graph_buf_);
+  auto ret = lite::PackWeightManager::GetInstance()->InitWeightManagerByBuf(graph_buf_);
+  if (ret != kSuccess) {
+    MS_LOG(ERROR) << "InitWeightManagerByBuf failed.";
+    return kLiteError;
+  }
   std::shared_ptr<ModelThread> model_thread = nullptr;
   for (size_t i = 0; i < num_models_; i++) {
     model_thread = std::make_shared<ModelThread>();

mindspore/lite/src/pack_weight_manager.cc

@@ -26,16 +26,17 @@ PackWeightManager *PackWeightManager::GetInstance() {
   return &instance;
 }
 
-void PackWeightManager::InitWeightManagerByBuf(const char *model_buf) {
-  MS_CHECK_TRUE_RET_VOID(model_buf != nullptr);
+STATUS PackWeightManager::InitWeightManagerByBuf(const char *model_buf) {
+  MS_CHECK_TRUE_RET(model_buf != nullptr, RET_ERROR);
   if (buf_model_weight_.find(model_buf) == buf_model_weight_.end()) {
     auto *model_const_weight = new (std::nothrow) ModelConstWeight();
     if (model_const_weight == nullptr) {
       MS_LOG(ERROR) << "model_const_weight is nullptr.";
-      return;
+      return RET_ERROR;
     }
     buf_model_weight_[model_buf] = model_const_weight;
   }
+  return RET_OK;
 }
 
 void PackWeightManager::InitWeightManagerByPath(const std::string &model_path, const char *model_buf) {

mindspore/lite/src/pack_weight_manager.h

@@ -45,7 +45,7 @@ class PackWeightManager {
   virtual ~PackWeightManager();
 
   void InitWeightManagerByPath(const std::string &model_path, const char *model_buf);
-  void InitWeightManagerByBuf(const char *model_buf);
+  STATUS InitWeightManagerByBuf(const char *model_buf);
   void DeleteSavedModelPtr(LiteModel *delete_model);
   STATUS StoreLiteModel(const char *model_buf, const Model *model);
   void *GetTensorData(const LiteModel *model, const SchemaTensorWrapper *origin_tensor, size_t tensor_index);

mindspore/lite/tools/benchmark/benchmark_base.h

@@ -139,11 +139,11 @@ class MS_API BenchmarkFlags : public virtual FlagParser {
     AddFlag(&BenchmarkFlags::cosine_distance_threshold_, "cosineDistanceThreshold", "cosine distance threshold", -1.1);
     AddFlag(&BenchmarkFlags::resize_dims_in_, "inputShapes",
             "Shape of input data, the format should be NHWC. e.g. 1,32,32,32:1,1,32,32,1", "");
-#ifdef SERVER_INFERENCE
     AddFlag(&BenchmarkFlags::enable_parallel_predict_, "enableParallelPredict", "Enable model parallel : true | false",
             false);
-    AddFlag(&BenchmarkFlags::num_require_, "numRequire", "require num", 1);
-#endif
+    AddFlag(&BenchmarkFlags::parallel_request_num_, "parallelRequestNum", "parallel request num of parallel predict",
+            1);
+    AddFlag(&BenchmarkFlags::workers_num_, "workersNum", "works num of parallel predict", 2);
 #ifdef ENABLE_OPENGL_TEXTURE
     AddFlag(&BenchmarkFlags::enable_gl_texture_, "enableGLTexture", "Enable GlTexture2D", false);
 #endif
@@ -157,10 +157,9 @@ class MS_API BenchmarkFlags : public virtual FlagParser {
 
  public:
   // common
-#ifdef SERVER_INFERENCE
   bool enable_parallel_predict_ = false;
-  int num_require_ = 1;
-#endif
+  int parallel_request_num_ = 1;
+  int workers_num_ = 2;
   std::string model_file_;
   std::string in_data_file_;
   std::string config_file_;

mindspore/lite/tools/benchmark/benchmark_unified_api.cc

@@ -946,6 +946,7 @@ int BenchmarkUnifiedApi::RunModelPool(std::shared_ptr<mindspore::Context> contex
   ModelParallelRunner model_pool;
   auto runner_config = std::make_shared<RunnerConfig>();
   runner_config->context = context;
+  runner_config->workers_num = flags_->workers_num_;
   auto model_init_start = GetTimeUs();
   auto ret = model_pool.Init(flags_->model_file_, runner_config);
   if (ret != kSuccess) {
@@ -959,7 +960,7 @@ int BenchmarkUnifiedApi::RunModelPool(std::shared_ptr<mindspore::Context> contex
     MS_LOG(ERROR) << "model pool input is empty.";
     return RET_ERROR;
   }
-  for (int i = 0; i < flags_->num_require_ + flags_->warm_up_loop_count_; i++) {
+  for (int i = 0; i < flags_->parallel_request_num_ + flags_->warm_up_loop_count_; i++) {
     auto status = LoadInput();
     if (status != RET_OK) {
       MS_LOG(ERROR) << "Generate input data error";
@@ -990,7 +991,7 @@ int BenchmarkUnifiedApi::RunModelPool(std::shared_ptr<mindspore::Context> contex
       MS_LOG(ERROR) << "model pool predict failed.";
     }
     auto predict_end = GetTimeUs();
-    std::cout << "run predict time: " << (predict_end - predict_start) / kFloatMSEC << " ms\n";
+    std::cout << "per predict time: " << (predict_end - predict_start) / kFloatMSEC << " ms\n";
     if (!flags_->benchmark_data_file_.empty()) {
       auto status = CompareOutputForModelPool(&output);
       if (status != RET_OK) {
@@ -1005,11 +1006,11 @@ int BenchmarkUnifiedApi::RunModelPool(std::shared_ptr<mindspore::Context> contex
   for (auto &warm_up_thread : model_thread_warm_up) {
     warm_up_thread.join();
   }
-  std::cout << "================ end warm up ================";
+  std::cout << "================ end warm up ================\n";
   auto all_start = GetTimeUs();
   for (int loop_count_num = 0; loop_count_num < flags_->loop_count_; loop_count_num++) {
     std::vector<std::thread> model_thread_run;
-    for (int i = 0; i < flags_->num_require_; i++) {
+    for (int i = 0; i < flags_->parallel_request_num_; i++) {
       model_thread_run.push_back(std::thread(model_pool_run, i + flags_->warm_up_loop_count_));
     }
     for (auto &run_thread : model_thread_run) {
@@ -1018,8 +1019,8 @@ int BenchmarkUnifiedApi::RunModelPool(std::shared_ptr<mindspore::Context> contex
   }
   auto all_end = GetTimeUs();
   std::cout << "=================================" << std::endl;
-  std::cout << "model pool init time: " << (model_init_end - model_init_start) / kFloatMSEC << " ms\n";
-  std::cout << "model pool all run time: " << (all_end - all_start) / kFloatMSEC / flags_->loop_count_ << " ms\n";
+  std::cout << "parallel predict init time: " << (model_init_end - model_init_start) / kFloatMSEC << " ms\n";
+  std::cout << "parallel predict all run time: " << (all_end - all_start) / kFloatMSEC / flags_->loop_count_ << " ms\n";
   std::cout << "=================================" << std::endl;
   return RET_OK;
 }