[Runtime][Pipeline Executor] multiple threads management and the data forwarding notification mechanism. (apache#10234)

* [Runtime][Pipeline Executor] multiple threads management and the
data forwarding notification mechanism.

In this patch we create working threads for each runtime of pipeline.
the threads would be terminated once the runtime class gets destroyed.

We also add a notification mechanism derived from the 'binding configuration'
of the runtime to forward the data notification.

* address review comments.

* address review comments.

* fix typo.

* fix typo.

* trigger build.

* address review comments.

* address review comments.

* address review comments.

* address review comments.

Author: huajsj

python/tvm/contrib/pipeline_executor.py

@@ -117,22 +117,18 @@ def __init__(self, module):
         # Get the packed functions from the pipeline executor.
         self._get_params_group_pipeline_map = self.module["get_params_group_pipeline_map"]
         self._run = self.module["run"]
-        self._stop = self.module["stop"]
         self._set_param = self.module["set_param"]
         self._set_input = self.module["set_input"]
         self._get_input = self.module["get_input"]
         self._get_output = self.module["get_output"]
         self._get_num_outputs = self.module["get_num_outputs"]
         self._get_input_pipeline_map = self.module["get_input_pipeline_map"]
+        self._get_pipe_execute_count = self.module["get_execute_count"]
 
     def run(self, sync=False):
         """Run the pipeline executor."""
         self._run(sync)
 
-    def stop(self):
-        """Stop the pipeline executor."""
-        self._stop()
-
     def get_input_pipeline_map(self, name):
         """Using the "name" to get the corresponding subgraph index and also get the "input name"
         of the corresponding subgraph interface.
@@ -213,6 +209,16 @@ def get_output(self):
         """
         return self._get_output()
 
+    @property
+    def num_executing_pipeline(self):
+        """Getting the count of running pipeline.
+        Returns
+        -------
+        count : int
+            The count of running pipeline.
+        """
+        return self._get_pipe_execute_count()
+
     @property
     def num_outputs(self):
         """Get the number of outputs.

src/runtime/pipeline/pipeline_executor.cc

@@ -79,8 +79,9 @@ PackedFunc PipelineExecutor::GetFunction(const std::string& name,
         [sptr_to_self, this](TVMArgs args, TVMRetValue* rv) { *rv = this->GetOutput(); });
   } else if (name == "run") {
     return PackedFunc([sptr_to_self, this](TVMArgs args, TVMRetValue* rv) { this->Run(args[0]); });
-  } else if (name == "stop") {
-    return PackedFunc([sptr_to_self, this](TVMArgs args, TVMRetValue* rv) { this->Stop(); });
+  } else if (name == "get_execute_count") {
+    return PackedFunc(
+        [sptr_to_self, this](TVMArgs args, TVMRetValue* rv) { *rv = this->GetExecutionCount(); });
   } else {
     LOG(FATAL) << "Unknown packed function: " << name;
     return PackedFunc();
@@ -96,7 +97,6 @@ PackedFunc PipelineExecutor::GetFunction(const std::string& name,
 void PipelineExecutor::SetInput(std::string input_name, DLTensor* data_in) {
   std::pair<int, int> indexs = this->GetInputIndex(input_name);
   if (indexs.first < 0 || indexs.first >= static_cast<int>(runtimes_.size())) {
-    this->Stop();
     LOG(FATAL) << "input name " << input_name << " not found.";
   }
   runtimes_[indexs.first]->SetInput(indexs.second, data_in);
@@ -109,7 +109,6 @@ void PipelineExecutor::SetInput(std::string input_name, DLTensor* data_in) {
 NDArray PipelineExecutor::GetInput(std::string input_name) {
   std::pair<int, int> indexs = this->GetInputIndex(input_name);
   if (indexs.first < 0 || indexs.first >= static_cast<int>(runtimes_.size())) {
-    this->Stop();
     LOG(FATAL) << "input name " << input_name << " not found.";
   }
   return runtimes_[indexs.first]->GetInput(indexs.second);
@@ -153,11 +152,6 @@ void PipelineExecutor::Run(bool serialized_mode) {
  * \brief return A list of global output data.
  */
 Array<NDArray> PipelineExecutor::GetOutput(void) { return pipeline_scheduler_.PipelineGetOutput(); }
-/*!
- * \brief Stop the pipeline executor.
- */
-void PipelineExecutor::Stop() { pipeline_scheduler_.PipelineStop(); }
-
 /*!
  * \brief Use the mod_config information to create a graph runtime list.
  * \param mod_config The config information that generates by the export library function call.
@@ -242,6 +236,10 @@ std::pair<int, int> PipelineExecutor::GetInputIndex(const std::string& name) {
   auto gruntime = runtimes_[index.first];
   return std::make_pair(index.first, gruntime->GetInputIndex(index.second));
 }
+/*!
+ * \brief Getting the count of running pipeline.
+ */
+int PipelineExecutor::GetExecutionCount() { return runtimes_.back()->GetExecutionCount(); }
 /*!
  * \brief Initialize the pipeline executor with a list of modules to be pipelined
  *  and config in JSON format.

src/runtime/pipeline/pipeline_executor.h

@@ -95,6 +95,10 @@ class TVM_DLL PipelineExecutor : public ModuleNode {
    * \return Return input data.
    */
   NDArray GetInput(std::string input_name);
+  /*!
+   * \brief Getting the count of running pipeline.
+   */
+  int GetExecutionCount();
   /*!
    * \brief Use the parameters group name to get the specific backend runtime then use
    *  the param_key_name to set param data for the said backend runtime.
@@ -114,10 +118,6 @@ class TVM_DLL PipelineExecutor : public ModuleNode {
    * \param serialized_mode Whether run the pipeline executor in serialized mode.
    */
   void Run(bool serialized_mode);
-  /*!
-   * \brief Stop the pipeline executor.
-   */
-  void Stop();
   /*!
    * \brief Get a list output data.
    * \return A list of output data.

src/runtime/pipeline/pipeline_scheduler.cc

@@ -32,9 +32,10 @@ std::vector<std::shared_ptr<BackendRuntime>> PipelineScheduler::PipelineInit(
     const std::vector<Module>& modules, const ConfigPipelineExecution& pipeline_config) {
   std::vector<std::shared_ptr<BackendRuntime>> runtimes;
   graph_modules_ = modules;
+  // Creating a list of runtimes.
   for (size_t i = 0; i < graph_modules_.size(); i++) {
-    auto runItem = std::make_shared<BackendRuntime>(graph_modules_[i], i);
-    runtimes.push_back(runItem);
+    auto run_item = std::make_shared<BackendRuntime>(graph_modules_[i], i);
+    runtimes.push_back(run_item);
   }
   // Creating a list of NDArray in order to storage the outputs data.
   auto global_output_map = pipeline_config.GetGlobalConfigOutputBindings();
@@ -46,6 +47,10 @@ std::vector<std::shared_ptr<BackendRuntime>> PipelineScheduler::PipelineInit(
     NDArray output = runtimes[output_pair.first]->CreateFromOutput(output_pair.second);
     output_arrays_.push_back(output);
   }
+  // Initializing and then running the worker thread.
+  for (auto runtime : runtimes) {
+    runtime->InitializePipeline(pipeline_config, &runtimes);
+  }
   return runtimes;
 }
 /*!
@@ -101,18 +106,11 @@ void PipelineScheduler::PipelineRunSequential(
 void PipelineScheduler::PipelineRun(const std::vector<std::shared_ptr<BackendRuntime>>& runtimes,
                                     ConfigPipelineExecution pipeline_config, bool sequential_mode) {
   if (!sequential_mode) {
-    // TODO(huajsj) remove this check after all of pipeline features in.
-    LOG(FATAL) << "Currently only supports sequential mode.";
+    runtimes.front()->RunPipeline();
   } else {
     PipelineRunSequential(runtimes, pipeline_config);
   }
 }
-/*!
- * \brief Stop the pipeline exection.
- */
-void PipelineScheduler::PipelineStop() {
-  // TODO(huajsj) Add stop logic.
-}
 /*!
  * \brief Get a list of output.
  */

src/runtime/pipeline/pipeline_scheduler.h

@@ -58,10 +58,6 @@ class PipelineScheduler {
    */
   void PipelineRunSequential(const std::vector<std::shared_ptr<BackendRuntime>>& runtimes,
                              ConfigPipelineExecution pipeline_config);
-  /*!
-   * \brief Stop the pipeline exection.
-   */
-  void PipelineStop();
   /*!
    * \brief Get a list of outputs.
    */