max_executor_thread_num && flow-stream-feature

Author: Dream-Runner-Yu

src/libmodelbox/base/include/modelbox/base/executor.h

@@ -33,6 +33,8 @@ class Executor {
 
   virtual ~Executor();
 
+  void SetThreadCount(int thread_count);
+
   template <typename func, typename... ts>
   auto Run(func &&fun, int32_t priority, ts &&...params)
       -> std::future<typename std::result_of<func(ts...)>::type> {

src/libmodelbox/engine/flowunit_data_executor.cc

@@ -34,6 +34,10 @@ Executor::Executor(int thread_count) {
 
 Executor::~Executor() { thread_pool_ = nullptr; }
 
+void Executor::SetThreadCount(int thread_count) {
+  thread_pool_->SetThreadSize(thread_count);
+}
+
 FlowUnitExecContext::FlowUnitExecContext(
     std::shared_ptr<FlowUnitDataContext> data_ctx)
     : data_ctx_(std::move(data_ctx)) {}

src/libmodelbox/engine/flowunit_group.cc

@@ -46,6 +46,8 @@ void FlowUnitGroup::InitTrace() {
   }
 }
 
+uint32_t FlowUnitGroup::GetBatchSize() const { return batch_size_; }
+
 std::shared_ptr<TraceSlice> FlowUnitGroup::StartTrace(
     FUExecContextList &exec_ctx_list) {
   std::call_once(trace_init_flag_, &FlowUnitGroup::InitTrace, this);
@@ -388,7 +390,7 @@ Status FlowUnitGroup::Open(const CreateExternalDataFunc &create_func) {
 
     return STATUS_OK;
   };
-  
+
   ThreadPool pool(std::thread::hardware_concurrency());
   pool.SetName(unit_name_ + "-Open");
   std::vector<std::future<Status>> result;

src/libmodelbox/engine/flowunit_manager.cc

@@ -59,6 +59,14 @@ Status FlowUnitManager::Initialize(
   SetDeviceManager(std::move(device_mgr));
   Status status;
   status = InitFlowUnitFactory(driver);
+
+  if (config != nullptr) {
+    max_executor_thread_num_ =
+        config->GetUint32("graph.max_executor_thread_num", 0);
+  } else {
+    max_executor_thread_num_ = 0;
+  }
+
   if (status != STATUS_SUCCESS) {
     return status;
   }
@@ -407,6 +415,12 @@ std::shared_ptr<FlowUnit> FlowUnitManager::CreateSingleFlowUnit(
     return nullptr;
   }
 
+  if (max_executor_thread_num_ > 0) {
+    MBLOG_INFO << "find the parameter max_executor_thread_num in the config: "
+               << max_executor_thread_num_;
+    device->GetDeviceExecutor()->SetThreadCount(max_executor_thread_num_);
+  }
+
   flowunit->SetBindDevice(device);
   std::vector<FlowUnitInput> &in_list = flowunit_desc->GetFlowUnitInput();
   for (auto &in_item : in_list) {

src/libmodelbox/engine/node.cc

@@ -763,23 +763,43 @@ void Node::CleanDataContext() {
 
 Status Node::Run(RunType type) {
   std::list<std::shared_ptr<FlowUnitDataContext>> data_ctx_list;
+  size_t process_count = 0;
   auto ret = Recv(type, data_ctx_list);
-  if (!ret) {
-    return ret;
-  }
 
-  ret = Process(data_ctx_list);
   if (!ret) {
     return ret;
   }
 
-  if (!GetOutputNames().empty()) {
-    ret = Send(data_ctx_list);
+  std::list<std::shared_ptr<FlowUnitDataContext>> process_ctx_list;
+
+  auto output_names_is_empty = GetOutputNames().empty();
+
+  for (auto& ctx : data_ctx_list) {
+    // process data according to batch size
+    process_count++;
+    process_ctx_list.push_back(ctx);
+
+    if (process_ctx_list.size() < flowunit_group_->GetBatchSize()) {
+      if (process_count < data_ctx_list.size()) {
+        continue;
+      }
+    }
+
+    ret = Process(process_ctx_list);
     if (!ret) {
       return ret;
     }
-  } else {
-    SetLastError(data_ctx_list);
+
+    if (!output_names_is_empty) {
+      ret = Send(process_ctx_list);
+      if (!ret) {
+        return ret;
+      }
+    } else {
+      SetLastError(process_ctx_list);
+    }
+
+    process_ctx_list.clear();
   }
 
   Clean(data_ctx_list);

src/libmodelbox/include/modelbox/flowunit.h

@@ -612,6 +612,8 @@ class FlowUnitManager {
 
   std::shared_ptr<DeviceManager> GetDeviceManager();
 
+  int max_executor_thread_num_;
+
  private:
   Status CheckParams(const std::string &unit_name, const std::string &unit_type,
                      const std::string &unit_device_id);

src/libmodelbox/include/modelbox/flowunit_group.h

@@ -64,6 +64,8 @@ class FlowUnitGroup {
 
   Status Close();
 
+  uint32_t GetBatchSize() const;
+
  private:
   std::weak_ptr<Node> node_;
   uint32_t batch_size_;