Reduce the goroutine in flowgraph to 2

internal/datanode/data_sync_service.go

@@ -81,7 +81,7 @@ type nodeConfig struct {
 	serverID     UniqueID
 }
 
-// start the flow graph in datasyncservice
+// start the flow graph in dataSyncService
 func (dsService *dataSyncService) start() {
 	if dsService.fg != nil {
 		log.Info("dataSyncService starting flow graph", zap.Int64("collectionID", dsService.collectionID),
@@ -357,6 +357,7 @@ func getServiceWithChannel(initCtx context.Context, node *DataNode, info *datapb
 	ds.flushManager = flushManager
 
 	// init flowgraph
+
 	fg := flowgraph.NewTimeTickedFlowGraph(node.ctx)
 	dmStreamNode, err := newDmInputNode(initCtx, node.dispClient, info.GetVchan().GetSeekPosition(), config)
 	if err != nil {

internal/datanode/event_manager.go

@@ -84,7 +84,7 @@ func (node *DataNode) StartWatchChannels(ctx context.Context) {
 	}
 }
 
-// checkWatchedList list all nodes under [prefix]/channel/{node_id} and make sure all nodeds are watched
+// checkWatchedList list all nodes under [prefix]/channel/{node_id} and make sure all nodes are watched
 // serves the corner case for etcd connection lost and missing some events
 func (node *DataNode) checkWatchedList() error {
 	// REF MEP#7 watch path should be [prefix]/channel/{node_id}/{channel_name}

internal/datanode/flow_graph_dd_node.go

@@ -276,7 +276,9 @@ func (ddn *ddNode) isDropped(segID UniqueID) bool {
 	return false
 }
 
-func (ddn *ddNode) Close() {}
+func (ddn *ddNode) Close() {
+	log.Info("Flowgraph DD Node closing")
+}
 
 func newDDNode(ctx context.Context, collID UniqueID, vChannelName string, droppedSegmentIDs []UniqueID,
 	sealedSegments []*datapb.SegmentInfo, growingSegments []*datapb.SegmentInfo, compactor *compactionExecutor,

internal/datanode/flow_graph_delete_node.go

@@ -146,11 +146,19 @@ func (dn *deleteNode) Operate(in []Msg) []Msg {
 		dn.flushManager.notifyAllFlushed()
 		log.Info("DeleteNode notifies BackgroundGC to release vchannel", zap.String("vChannelName", dn.channelName))
 		dn.clearSignal <- dn.channelName
+		log.Info("DeleteNode notifies BackgroundGC to release vchannel done", zap.String("vChannelName", dn.channelName))
 	}
 
 	for _, sp := range spans {
 		sp.End()
 	}
+	if !dn.IsValidInMsg(in) {
+		log.Info("lxg input is invalid", zap.String("vChannelName", dn.channelName))
+	}
+	if isCloseMsg(in) {
+		log.Info("lxg debug in is close msg", zap.String("vChannelName", dn.channelName))
+	}
+	log.Info("lxg test delete node operate ok")
 	return in
 }
 
@@ -211,3 +219,10 @@ func newDeleteNode(ctx context.Context, fm flushManager, manager *DeltaBufferMan
 		clearSignal:      sig,
 	}, nil
 }
+
+func isCloseMsg(msgs []Msg) bool {
+	if len(msgs) == 1 {
+		return msgs[0].IsClose()
+	}
+	return false
+}

internal/util/flowgraph/flow_graph.go

@@ -30,20 +30,22 @@ import (
 // TimeTickedFlowGraph flowgraph with input from tt msg stream
 type TimeTickedFlowGraph struct {
 	nodeCtx         map[NodeName]*nodeCtx
+	nodeCtxManager  *nodeCtxManager
 	stopOnce        sync.Once
 	startOnce       sync.Once
 	closeWg         *sync.WaitGroup
 	closeGracefully *atomic.Bool
 }
 
-// AddNode add Node into flowgraph
+// AddNode add Node into flowgraph and fill nodeCtxManager
 func (fg *TimeTickedFlowGraph) AddNode(node Node) {
 	nodeCtx := nodeCtx{
-		node:    node,
-		closeCh: make(chan struct{}),
-		closeWg: fg.closeWg,
+		node: node,
 	}
 	fg.nodeCtx[node.Name()] = &nodeCtx
+	if node.IsInputNode() {
+		fg.nodeCtxManager = NewNodeCtxManager(&nodeCtx, fg.closeWg)
+	}
 }
 
 // SetEdges set directed edges from in nodes to out nodes
@@ -79,8 +81,9 @@ func (fg *TimeTickedFlowGraph) SetEdges(nodeName string, out []string) error {
 func (fg *TimeTickedFlowGraph) Start() {
 	fg.startOnce.Do(func() {
 		for _, v := range fg.nodeCtx {
-			v.Start()
+			v.node.Start()
 		}
+		fg.nodeCtxManager.Start()
 	})
 }
 
@@ -120,6 +123,7 @@ func (fg *TimeTickedFlowGraph) Close() {
 func NewTimeTickedFlowGraph(ctx context.Context) *TimeTickedFlowGraph {
 	flowGraph := TimeTickedFlowGraph{
 		nodeCtx:         make(map[string]*nodeCtx),
+		nodeCtxManager:  &nodeCtxManager{},
 		closeWg:         &sync.WaitGroup{},
 		closeGracefully: atomic.NewBool(CloseImmediately),
 	}

internal/util/flowgraph/flow_graph_test.go

@@ -35,7 +35,7 @@ import (
 // nodeD: count c = b + 2
 
 type nodeA struct {
-	BaseNode
+	InputNode
 	inputChan chan float64
 	a         float64
 }
@@ -47,7 +47,7 @@ type nodeB struct {
 
 type nodeC struct {
 	BaseNode
-	d          float64
+	c          float64
 	outputChan chan float64
 }
 
@@ -117,8 +117,10 @@ func createExampleFlowGraph() (*TimeTickedFlowGraph, chan float64, chan float64,
 	fg := NewTimeTickedFlowGraph(ctx)
 
 	var a Node = &nodeA{
-		BaseNode: BaseNode{
-			maxQueueLength: MaxQueueLength,
+		InputNode: InputNode{
+			BaseNode: BaseNode{
+				maxQueueLength: MaxQueueLength,
+			},
 		},
 		inputChan: inputChan,
 	}
@@ -175,8 +177,10 @@ func TestTimeTickedFlowGraph_AddNode(t *testing.T) {
 	fg := NewTimeTickedFlowGraph(context.TODO())
 
 	var a Node = &nodeA{
-		BaseNode: BaseNode{
-			maxQueueLength: MaxQueueLength,
+		InputNode: InputNode{
+			BaseNode: BaseNode{
+				maxQueueLength: MaxQueueLength,
+			},
 		},
 		inputChan: inputChan,
 	}

internal/util/flowgraph/input_node.go

@@ -19,7 +19,6 @@ package flowgraph
 import (
 	"context"
 	"fmt"
-	"sync"
 
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/trace"
@@ -49,7 +48,6 @@ type InputNode struct {
 	collectionID int64
 	dataType     string
 
-	closeOnce       sync.Once
 	closeGracefully *atomic.Bool
 }