add: flow-stream-feature

Author: Dream-Runner-Yu

src/libmodelbox/engine/flowunit_group.cc

@@ -46,6 +46,11 @@ 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);

src/libmodelbox/engine/node.cc

@@ -762,30 +762,48 @@ 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;
+
+  for(auto& ctx: data_ctx_list){
+    
+    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 (!GetOutputNames().empty()) {
+        ret = Send(process_ctx_list);
+        if (!ret) {
+            return ret; 
+        }
+    } else {
+      SetLastError(process_ctx_list);
+    }
+
+    process_ctx_list.clear();
   }
 
   Clean(data_ctx_list);
   return STATUS_SUCCESS;
 }
-
 void Node::SetLastError(
     std::list<std::shared_ptr<FlowUnitDataContext>>& data_ctx_list) {
   for (auto& data_ctx : data_ctx_list) {

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_;