model parallel runner add new api: init by buf

include/api/model_parallel_runner.h

@@ -92,6 +92,16 @@ class MS_API ModelParallelRunner {
   /// \return Status.
   inline Status Init(const std::string &model_path, const std::shared_ptr<RunnerConfig> &runner_config = nullptr);
 
+  /// \brief build a model parallel runner from model buffer so that it can run on a device. Only valid for Lite.
+  ///
+  /// \param[in] model_data Define the buffer read from a model file.
+  /// \param[in] data_size Define bytes number of model buffer.
+  /// \param[in] runner_config Define the config used to store options during model pool init.
+  ///
+  /// \return Status.
+  Status Init(const void *model_data, const size_t data_size,
+              const std::shared_ptr<RunnerConfig> &runner_config = nullptr);
+
   /// \brief Obtains all input tensors information of the model.
   ///
   /// \return The vector that includes all input tensors.

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

@@ -68,7 +68,28 @@ Status ModelParallelRunner::Init(const std::vector<char> &model_path,
     MS_LOG(ERROR) << "model pool is nullptr.";
     return kLiteNullptr;
   }
-  auto status = model_pool_->Init(CharToString(model_path), runner_config);
+  auto status = model_pool_->InitByPath(CharToString(model_path), runner_config);
+  if (status != kSuccess) {
+    MS_LOG(ERROR) << "model runner init failed.";
+    return kLiteError;
+  }
+  return status;
+}
+
+Status ModelParallelRunner::Init(const void *model_data, size_t data_size,
+                                 const std::shared_ptr<RunnerConfig> &runner_config) {
+#ifdef USE_GLOG
+  mindspore::mindspore_log_init();
+#endif
+  if (!PlatformInstructionSetSupportCheck()) {
+    return kLiteNotSupport;
+  }
+  model_pool_ = std::make_shared<ModelPool>();
+  if (model_pool_ == nullptr) {
+    MS_LOG(ERROR) << "model pool is nullptr.";
+    return kLiteNullptr;
+  }
+  auto status = model_pool_->InitByBuf(static_cast<const char *>(model_data), data_size, runner_config);
   if (status != kSuccess) {
     MS_LOG(ERROR) << "model runner init failed.";
     return kLiteError;

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

@@ -843,14 +843,7 @@ Status ModelPool::InitAdvancedStrategy(const char *model_buf, size_t size, int b
   return kSuccess;
 }
 
-Status ModelPool::Init(const std::string &model_path, const std::shared_ptr<RunnerConfig> &runner_config) {
-  size_t size = 0;
-  auto model_buf = lite::ReadFile(model_path.c_str(), &size);
-  if (model_buf == nullptr) {
-    MS_LOG(ERROR) << "read ms model failed, model path: " << model_path;
-    return kLiteNullptr;
-  }
-
+Status ModelPool::Init(const char *model_buf, size_t size, const std::shared_ptr<RunnerConfig> &runner_config) {
   auto status = InitNumaParameter(runner_config);
   if (status != kSuccess) {
     MS_LOG(ERROR) << "Init numa parameter failed.";
@@ -886,6 +879,27 @@ Status ModelPool::Init(const std::string &model_path, const std::shared_ptr<Runn
   for (size_t i = 0; i < kNumMaxTaskQueueSize; i++) {
     free_tasks_id_.push(i);
   }
+  return kSuccess;
+}
+
+Status ModelPool::InitByBuf(const char *model_data, size_t size, const std::shared_ptr<RunnerConfig> &runner_config) {
+  return Init(model_data, size, runner_config);
+}
+
+Status ModelPool::InitByPath(const std::string &model_path, const std::shared_ptr<RunnerConfig> &runner_config) {
+  size_t size = 0;
+  auto model_buf = lite::ReadFile(model_path.c_str(), &size);
+  if (model_buf == nullptr) {
+    MS_LOG(ERROR) << "read ms model failed, model path: " << model_path;
+    return kLiteNullptr;
+  }
+  auto status = Init(model_buf, size, runner_config);
+  if (status != kSuccess) {
+    MS_LOG(ERROR) << "init failed.";
+    delete[] model_buf;
+    model_buf = nullptr;
+    return kLiteError;
+  }
   if (model_buf != nullptr) {
     delete[] model_buf;
     model_buf = nullptr;

mindspore/lite/src/extendrt/cxx_api/model_pool/model_pool.h

@@ -55,7 +55,11 @@ class ModelPool {
 
   ~ModelPool();
 
-  Status Init(const std::string &model_path, const std::shared_ptr<RunnerConfig> &runner_config = nullptr);
+  Status InitByPath(const std::string &model_path, const std::shared_ptr<RunnerConfig> &runner_config = nullptr);
+
+  Status InitByBuf(const char *model_data, size_t size, const std::shared_ptr<RunnerConfig> &runner_config = nullptr);
+
+  Status Init(const char *model_buf, size_t size, const std::shared_ptr<RunnerConfig> &runner_config);
 
   Status UpdateConfig(const std::string &section, const std::pair<std::string, std::string> &config);
 

mindspore/lite/test/ut/src/api/model_parallel_runner_test.cc

@@ -23,6 +23,7 @@ namespace {
 const char in_data_path[] = "./mobilenetv2.ms.bin";
 const char model_path[] = "./mobilenetv2.ms";
 const size_t kInputDataSize = 1 * 224 * 224 * 3 * sizeof(float);
+const size_t kOutputDataSize = 1 * 1001 * sizeof(float);
 
 void SetInputTensorData(std::vector<MSTensor> *inputs) {
   ASSERT_EQ(inputs->size(), 1);
@@ -191,4 +192,52 @@ TEST_F(ModelParallelRunnerTest, RunnerPredict) {
     tensor.SetData(nullptr);
   }
 }
+
+TEST_F(ModelParallelRunnerTest, RunnerInitByBuf) {
+  auto config = std::make_shared<RunnerConfig>();
+  ASSERT_NE(nullptr, config);
+
+  auto context = std::make_shared<Context>();
+  ASSERT_NE(nullptr, context);
+  auto &device_list = context->MutableDeviceInfo();
+  auto device_info = std::make_shared<mindspore::CPUDeviceInfo>();
+  ASSERT_NE(nullptr, device_info);
+  device_list.push_back(device_info);
+  ASSERT_EQ(device_list.size(), 1);
+  config->SetContext(context);
+  config->SetWorkersNum(2);
+  ModelParallelRunner runner;
+
+  size_t size = 0;
+  auto model_buf = lite::ReadFile(model_path, &size);
+  ASSERT_NE(nullptr, model_buf);
+  auto status = runner.Init(model_buf, size, config);
+  delete[] model_buf;  // after init, users can release buf data
+  ASSERT_EQ(status, kSuccess);
+  auto inputs = runner.GetInputs();
+  SetInputTensorData(&inputs);
+  std::vector<MSTensor> outputs;
+  for (auto &tensor : outputs) {
+    auto tensor_size = tensor.DataSize();
+    ASSERT_NE(0, tensor_size);
+    ASSERT_EQ(tensor_size, kOutputDataSize);
+    auto data = malloc(tensor_size);
+    ASSERT_NE(nullptr, data);
+    tensor.SetShape({1, 1001});
+    tensor.SetData(data);
+  }
+  status = runner.Predict(inputs, &outputs);
+  ASSERT_EQ(status, kSuccess);
+  // free user data
+  for (auto &tensor : inputs) {
+    char *input_data = static_cast<char *>(tensor.MutableData());
+    delete[] input_data;
+    tensor.SetData(nullptr);
+  }
+  for (auto &tensor : outputs) {
+    auto *output_data = tensor.MutableData();
+    free(output_data);
+    tensor.SetData(nullptr);
+  }
+}
 }  // namespace mindspore