Refactor v3 Paths to parse YAML using TranslatePipeline.

Fix goroutine resource leak in `datamodel/low/v3/path_item.go`.
pb33f · Jul 27, 2023 · f6396aa · f6396aa
1 parent 0ce74b5
commit f6396aa
Show file tree

Hide file tree

Showing 2 changed files with 123 additions and 71 deletions.
diff --git a/datamodel/low/v3/path_item.go b/datamodel/low/v3/path_item.go
@@ -4,6 +4,7 @@
 package v3
 
 import (
+	"context"
 	"crypto/sha256"
 	"fmt"
 	"sort"
@@ -272,16 +273,25 @@ func (p *PathItem) Build(root *yaml.Node, idx *index.SpecIndex) error {
 	// now we need to build out the operation, we will do this asynchronously for speed.
 	opBuildChan := make(chan bool)
 	opErrorChan := make(chan error)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
 
 	buildOpFunc := func(op low.NodeReference[*Operation], ch chan<- bool, errCh chan<- error, ref string) {
 		er := op.Value.Build(op.ValueNode, idx)
 		if ref != "" {
 			op.Value.Reference.Reference = ref
 		}
 		if er != nil {
-			errCh <- er
+			select {
+			case errCh <- er:
+			case <-ctx.Done():
+			}
+			return
+		}
+		select {
+		case ch <- true:
+		case <-ctx.Done():
 		}
-		ch <- true
 	}
 
 	if len(ops) <= 0 {
@@ -298,12 +308,15 @@ func (p *PathItem) Build(root *yaml.Node, idx *index.SpecIndex) error {
 
 	n := 0
 	total := len(ops)
+FORLOOP1:
 	for n < total {
 		select {
 		case buildError := <-opErrorChan:
 			return buildError
 		case <-opBuildChan:
 			n++
+		case <-ctx.Done():
+			break FORLOOP1
 		}
 	}
 

diff --git a/datamodel/low/v3/paths.go b/datamodel/low/v3/paths.go
@@ -4,11 +4,14 @@
 package v3
 
 import (
+	"context"
 	"crypto/sha256"
 	"fmt"
 	"sort"
 	"strings"
+	"sync"
 
+	"github.com/pb33f/libopenapi/datamodel"
 	"github.com/pb33f/libopenapi/datamodel/low"
 	"github.com/pb33f/libopenapi/index"
 	"github.com/pb33f/libopenapi/utils"
@@ -63,90 +66,126 @@ func (p *Paths) Build(root *yaml.Node, idx *index.SpecIndex) error {
 	utils.CheckForMergeNodes(root)
 	p.Reference = new(low.Reference)
 	p.Extensions = low.ExtractExtensions(root)
-	skip := false
-	var currentNode *yaml.Node
 
-	pathsMap := make(map[low.KeyReference[string]]low.ValueReference[*PathItem])
-
-	// build each new path, in a new thread.
+	// Translate YAML nodes to pathsMap using `TranslatePipeline`.
 	type pathBuildResult struct {
 		k low.KeyReference[string]
 		v low.ValueReference[*PathItem]
 	}
+	type nodeItem struct {
+		currentNode *yaml.Node
+		pathNode    *yaml.Node
+	}
+	pathsMap := make(map[low.KeyReference[string]]low.ValueReference[*PathItem])
+	in := make(chan nodeItem)
+	out := make(chan pathBuildResult)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var wg sync.WaitGroup
+	wg.Add(2) // input and output goroutines.
+
+	// TranslatePipeline input.
+	go func() {
+		defer func() {
+			close(in)
+			wg.Done()
+		}()
+		skip := false
+		var currentNode *yaml.Node
+		for i, pathNode := range root.Content {
+			if strings.HasPrefix(strings.ToLower(pathNode.Value), "x-") {
+				skip = true
+				continue
+			}
+			if skip {
+				skip = false
+				continue
+			}
+			if i%2 == 0 {
+				currentNode = pathNode
+				continue
+			}
 
-	bChan := make(chan pathBuildResult)
-	eChan := make(chan error)
-	buildPathItem := func(cNode, pNode *yaml.Node, b chan<- pathBuildResult, e chan<- error) {
-		if ok, _, _ := utils.IsNodeRefValue(pNode); ok {
-			r, err := low.LocateRefNode(pNode, idx)
-			if r != nil {
-				pNode = r
-				if r.Tag == "" {
-					// If it's a node from file, tag is empty
-					// If it's a reference we need to extract actual operation node
-					pNode = r.Content[0]
+			select {
+			case in <- nodeItem{
+				currentNode: currentNode,
+				pathNode:    pathNode,
+			}:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	// TranslatePipeline output.
+	go func() {
+		defer func() {
+			cancel()
+			wg.Done()
+		}()
+		for {
+			select {
+			case result, ok := <-out:
+				if !ok {
+					return
 				}
+				pathsMap[result.k] = result.v
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	err := datamodel.TranslatePipeline[nodeItem, pathBuildResult](in, out,
+		func(value nodeItem) (pathBuildResult, error) {
+			pNode := value.pathNode
+			cNode := value.currentNode
+
+			if ok, _, _ := utils.IsNodeRefValue(pNode); ok {
+				r, err := low.LocateRefNode(pNode, idx)
+				if r != nil {
+					pNode = r
+					if r.Tag == "" {
+						// If it's a node from file, tag is empty
+						// If it's a reference we need to extract actual operation node
+						pNode = r.Content[0]
+					}
 
-				if err != nil {
-					if !idx.AllowCircularReferenceResolving() {
-						e <- fmt.Errorf("path item build failed: %s", err.Error())
-						return
+					if err != nil {
+						if !idx.AllowCircularReferenceResolving() {
+							return pathBuildResult{}, fmt.Errorf("path item build failed: %s", err.Error())
+						}
 					}
+				} else {
+					return pathBuildResult{}, fmt.Errorf("path item build failed: cannot find reference: %s at line %d, col %d",
+						pNode.Content[1].Value, pNode.Content[1].Line, pNode.Content[1].Column)
 				}
-			} else {
-				e <- fmt.Errorf("path item build failed: cannot find reference: %s at line %d, col %d",
-					pNode.Content[1].Value, pNode.Content[1].Line, pNode.Content[1].Column)
-				return
 			}
-		}
 
-		path := new(PathItem)
-		_ = low.BuildModel(pNode, path)
-		err := path.Build(pNode, idx)
-		if err != nil {
-			e <- err
-			return
-		}
-		b <- pathBuildResult{
-			k: low.KeyReference[string]{
-				Value:   cNode.Value,
-				KeyNode: cNode,
-			},
-			v: low.ValueReference[*PathItem]{
-				Value:     path,
-				ValueNode: pNode,
-			},
-		}
-	}
+			path := new(PathItem)
+			_ = low.BuildModel(pNode, path)
+			err := path.Build(pNode, idx)
+			if err != nil {
+				return pathBuildResult{}, err
+			}
 
-	pathCount := 0
-	for i, pathNode := range root.Content {
-		if strings.HasPrefix(strings.ToLower(pathNode.Value), "x-") {
-			skip = true
-			continue
-		}
-		if skip {
-			skip = false
-			continue
-		}
-		if i%2 == 0 {
-			currentNode = pathNode
-			continue
-		}
-		pathCount++
-		go buildPathItem(currentNode, pathNode, bChan, eChan)
+			return pathBuildResult{
+				k: low.KeyReference[string]{
+					Value:   cNode.Value,
+					KeyNode: cNode,
+				},
+				v: low.ValueReference[*PathItem]{
+					Value:     path,
+					ValueNode: pNode,
+				},
+			}, nil
+		},
+	)
+	wg.Wait()
+	if err != nil {
+		return err
 	}
 
-	completedItems := 0
-	for completedItems < pathCount {
-		select {
-		case err := <-eChan:
-			return err
-		case res := <-bChan:
-			completedItems++
-			pathsMap[res.k] = res.v
-		}
-	}
 	p.PathItems = pathsMap
 	return nil
 }