优化ioswitch数据结构

common/pkgs/ioswitch2/parser/parser.go

@@ -572,8 +572,8 @@ func omitSegmentSplitJoin(ctx *ParseContext) bool {
 			}
 
 			if dstNode == nil {
-				dstNode = out.To().Get(0).Node
-			} else if dstNode != out.To().Get(0).Node {
+				dstNode = out.To().Get(0)
+			} else if dstNode != out.To().Get(0) {
 				return true
 			}
 		}
@@ -598,7 +598,7 @@ func omitSegmentSplitJoin(ctx *ParseContext) bool {
 		// F->Split->Join->T 变换为：F->T
 		splitInput := splitNode.InputStreams().Get(0)
 		for _, to := range joinNode.Joined().To().RawArray() {
-			splitInput.StreamTo(to.Node, to.SlotIndex)
+			splitInput.StreamTo(to, to.InputStreams().IndexOf(joinNode.Joined()))
 		}
 		splitInput.StreamNotTo(splitNode, 0)
 
@@ -694,8 +694,8 @@ func omitSplitJoin(ctx *ParseContext) bool {
 			}
 
 			if dstNode == nil {
-				dstNode = out.To().Get(0).Node
-			} else if dstNode != out.To().Get(0).Node {
+				dstNode = out.To().Get(0)
+			} else if dstNode != out.To().Get(0) {
 				return true
 			}
 		}
@@ -720,7 +720,7 @@ func omitSplitJoin(ctx *ParseContext) bool {
 		// F->Split->Join->T 变换为：F->T
 		splitInput := splitNode.InputStreams().Get(0)
 		for _, to := range joinNode.Joined().To().RawArray() {
-			splitInput.StreamTo(to.Node, to.SlotIndex)
+			splitInput.StreamTo(to, to.InputStreams().IndexOf(joinNode.Joined()))
 		}
 		splitInput.StreamNotTo(splitNode, 0)
 
@@ -748,13 +748,13 @@ func pin(ctx *ParseContext) bool {
 		var toEnv *dag.NodeEnv
 		for _, out := range node.OutputStreams().RawArray() {
 			for _, to := range out.To().RawArray() {
-				if to.Node.Env().Type == dag.EnvUnknown {
+				if to.Env().Type == dag.EnvUnknown {
 					continue
 				}
 
 				if toEnv == nil {
-					toEnv = to.Node.Env()
-				} else if !toEnv.Equals(to.Node.Env()) {
+					toEnv = to.Env()
+				} else if !toEnv.Equals(to.Env()) {
 					toEnv = nil
 					break
 				}
@@ -773,13 +773,13 @@ func pin(ctx *ParseContext) bool {
 		// 否则根据输入流的始发地来固定
 		var fromEnv *dag.NodeEnv
 		for _, in := range node.InputStreams().RawArray() {
-			if in.From().Node.Env().Type == dag.EnvUnknown {
+			if in.From().Env().Type == dag.EnvUnknown {
 				continue
 			}
 
 			if fromEnv == nil {
-				fromEnv = in.From().Node.Env()
-			} else if !fromEnv.Equals(in.From().Node.Env()) {
+				fromEnv = in.From().Env()
+			} else if !fromEnv.Equals(in.From().Env()) {
 				fromEnv = nil
 				break
 			}
@@ -853,7 +853,7 @@ func generateRange(ctx *ParseContext) {
 		if toStrIdx.IsRaw() {
 			n := ctx.DAG.NewRange()
 			toInput := toNode.Input()
-			*n.Env() = *toInput.Var.From().Node.Env()
+			*n.Env() = *toInput.Var.From().Env()
 			rnged := n.RangeStream(toInput.Var, exec.Range{
 				Offset: toRng.Offset - ctx.StreamRange.Offset,
 				Length: toRng.Length,
@@ -869,7 +869,7 @@ func generateRange(ctx *ParseContext) {
 
 			n := ctx.DAG.NewRange()
 			toInput := toNode.Input()
-			*n.Env() = *toInput.Var.From().Node.Env()
+			*n.Env() = *toInput.Var.From().Env()
 			rnged := n.RangeStream(toInput.Var, exec.Range{
 				Offset: toRng.Offset - blkStart,
 				Length: toRng.Length,
@@ -903,32 +903,32 @@ func generateRange(ctx *ParseContext) {
 // 生成Clone指令
 func generateClone(ctx *ParseContext) {
 	ctx.DAG.Walk(func(node dag.Node) bool {
-		for _, out := range node.OutputStreams().RawArray() {
-			if out.To().Len() <= 1 {
+		for _, outVar := range node.OutputStreams().RawArray() {
+			if outVar.To().Len() <= 1 {
 				continue
 			}
 
 			c := ctx.DAG.NewCloneStream()
 			*c.Env() = *node.Env()
-			for _, to := range out.To().RawArray() {
-				c.NewOutput().StreamTo(to.Node, to.SlotIndex)
+			for _, to := range outVar.To().RawArray() {
+				c.NewOutput().StreamTo(to, to.InputStreams().IndexOf(outVar))
 			}
-			out.To().Resize(0)
-			c.SetInput(out)
+			outVar.To().Resize(0)
+			c.SetInput(outVar)
 		}
 
-		for _, out := range node.OutputValues().RawArray() {
-			if out.To().Len() <= 1 {
+		for _, outVar := range node.OutputValues().RawArray() {
+			if outVar.To().Len() <= 1 {
 				continue
 			}
 
 			t := ctx.DAG.NewCloneValue()
 			*t.Env() = *node.Env()
-			for _, to := range out.To().RawArray() {
-				t.NewOutput().ValueTo(to.Node, to.SlotIndex)
+			for _, to := range outVar.To().RawArray() {
+				t.NewOutput().ValueTo(to, to.InputValues().IndexOf(outVar))
 			}
-			out.To().Resize(0)
-			t.SetInput(out)
+			outVar.To().Resize(0)
+			t.SetInput(outVar)
 		}
 
 		return true

common/pkgs/ioswitchlrc/parser/passes.go

@@ -142,13 +142,13 @@ func pin(ctx *GenerateContext) bool {
 		var toEnv *dag.NodeEnv
 		for _, out := range node.OutputStreams().RawArray() {
 			for _, to := range out.To().RawArray() {
-				if to.Node.Env().Type == dag.EnvUnknown {
+				if to.Env().Type == dag.EnvUnknown {
 					continue
 				}
 
 				if toEnv == nil {
-					toEnv = to.Node.Env()
-				} else if !toEnv.Equals(to.Node.Env()) {
+					toEnv = to.Env()
+				} else if !toEnv.Equals(to.Env()) {
 					toEnv = nil
 					break
 				}
@@ -167,13 +167,13 @@ func pin(ctx *GenerateContext) bool {
 		// 否则根据输入流的始发地来固定
 		var fromEnv *dag.NodeEnv
 		for _, in := range node.InputStreams().RawArray() {
-			if in.From().Node.Env().Type == dag.EnvUnknown {
+			if in.From().Env().Type == dag.EnvUnknown {
 				continue
 			}
 
 			if fromEnv == nil {
-				fromEnv = in.From().Node.Env()
-			} else if !fromEnv.Equals(in.From().Node.Env()) {
+				fromEnv = in.From().Env()
+			} else if !fromEnv.Equals(in.From().Env()) {
 				fromEnv = nil
 				break
 			}
@@ -233,7 +233,7 @@ func generateRange(ctx *GenerateContext) {
 		if toDataIdx == -1 {
 			n := ctx.DAG.NewRange()
 			toInput := toNode.Input()
-			*n.Env() = *toInput.Var.From().Node.Env()
+			*n.Env() = *toInput.Var.From().Env()
 			rnged := n.RangeStream(toInput.Var, exec.Range{
 				Offset: toRng.Offset - ctx.StreamRange.Offset,
 				Length: toRng.Length,
@@ -249,7 +249,7 @@ func generateRange(ctx *GenerateContext) {
 
 			n := ctx.DAG.NewRange()
 			toInput := toNode.Input()
-			*n.Env() = *toInput.Var.From().Node.Env()
+			*n.Env() = *toInput.Var.From().Env()
 			rnged := n.RangeStream(toInput.Var, exec.Range{
 				Offset: toRng.Offset - blkStart,
 				Length: toRng.Length,
@@ -263,32 +263,32 @@ func generateRange(ctx *GenerateContext) {
 // 生成Clone指令
 func generateClone(ctx *GenerateContext) {
 	ctx.DAG.Walk(func(node dag.Node) bool {
-		for _, out := range node.OutputStreams().RawArray() {
-			if out.To().Len() <= 1 {
+		for _, outVar := range node.OutputStreams().RawArray() {
+			if outVar.To().Len() <= 1 {
 				continue
 			}
 
 			t := ctx.DAG.NewCloneStream()
 			*t.Env() = *node.Env()
-			for _, to := range out.To().RawArray() {
-				t.NewOutput().StreamTo(to.Node, to.SlotIndex)
+			for _, to := range outVar.To().RawArray() {
+				t.NewOutput().StreamTo(to, to.InputStreams().IndexOf(outVar))
 			}
-			out.To().Resize(0)
-			t.SetInput(out)
+			outVar.To().Resize(0)
+			t.SetInput(outVar)
 		}
 
-		for _, out := range node.OutputValues().RawArray() {
-			if out.To().Len() <= 1 {
+		for _, outVar := range node.OutputValues().RawArray() {
+			if outVar.To().Len() <= 1 {
 				continue
 			}
 
 			t := ctx.DAG.NewCloneValue()
 			*t.Env() = *node.Env()
-			for _, to := range out.To().RawArray() {
-				t.NewOutput().ValueTo(to.Node, to.SlotIndex)
+			for _, to := range outVar.To().RawArray() {
+				t.NewOutput().ValueTo(to, to.InputValues().IndexOf(outVar))
 			}
-			out.To().Resize(0)
-			t.SetInput(out)
+			outVar.To().Resize(0)
+			t.SetInput(outVar)
 		}
 
 		return true