cue: tools/flow: high memory usage

What version of CUE are you using (cue version)?

v0.4.3

Does this issue reproduce with the latest release?

Yes.

What did you do?

go build -o main
./main
go tool pprof heap
./main -disjunctions=false
go tool pprof heap

-- go.mod --
module dagger.io/dagger/cue-memory-use

go 1.18

require cuelang.org/go v0.4.3

require (
	github.com/cockroachdb/apd/v2 v2.0.1 // indirect
	github.com/google/uuid v1.2.0 // indirect
	github.com/mpvl/unique v0.0.0-20150818121801-cbe035fff7de // indirect
	github.com/pkg/errors v0.8.1 // indirect
	golang.org/x/net v0.0.0-20200226121028-0de0cce0169b // indirect
	golang.org/x/text v0.3.7 // indirect
	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
)
-- go.sum --
cuelang.org/go v0.4.3 h1:W3oBBjDTm7+IZfCKZAmC8uDG0eYfJL4Pp/xbbCMKaVo=
cuelang.org/go v0.4.3/go.mod h1:7805vR9H+VoBNdWFdI7jyDR3QLUPp4+naHfbcgp55HI=
github.com/cockroachdb/apd v1.1.0 h1:3LFP3629v+1aKXU5Q37mxmRxX/pIu1nijXydLShEq5I=
github.com/cockroachdb/apd/v2 v2.0.1 h1:y1Rh3tEU89D+7Tgbw+lp52T6p/GJLpDmNvr10UWqLTE=
github.com/cockroachdb/apd/v2 v2.0.1/go.mod h1:DDxRlzC2lo3/vSlmSoS7JkqbbrARPuFOGr0B9pvN3Gw=
github.com/google/uuid v1.2.0 h1:qJYtXnJRWmpe7m/3XlyhrsLrEURqHRM2kxzoxXqyUDs=
github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/mpvl/unique v0.0.0-20150818121801-cbe035fff7de h1:D5x39vF5KCwKQaw+OC9ZPiLVHXz3UFw2+psEX+gYcto=
github.com/mpvl/unique v0.0.0-20150818121801-cbe035fff7de/go.mod h1:kJun4WP5gFuHZgRjZUWWuH1DTxCtxbHDOIJsudS8jzY=
github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/net v0.0.0-20200226121028-0de0cce0169b h1:0mm1VjtFUOIlE1SbDlwjYaDxZVDP2S5ou6y0gSgXHu8=
golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-- main.go --
package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"runtime/pprof"
	"time"

	"cuelang.org/go/cue"
	"cuelang.org/go/cue/ast"
	"cuelang.org/go/cue/cuecontext"
	"cuelang.org/go/cue/format"
	"cuelang.org/go/cue/token"
	"cuelang.org/go/tools/flow"
)

func genTask(name, prev int) *ast.Field {
	var prevTask ast.Expr
	if prev >= 0 {
		prevTask = ast.NewStruct(
			ast.NewIdent("input"),
			ast.NewSel(ast.NewIdent(fmt.Sprintf("task%d", prev)), "output"),
		)
	} else {
		prevTask = ast.NewStruct(ast.NewIdent("input"), ast.NewString("world"))
	}
	return &ast.Field{
		Label: ast.NewIdent(fmt.Sprintf("task%d", name)),
		Value: ast.NewBinExpr(token.AND,
			ast.NewIdent("#MyDef"),
			prevTask,
		),
	}
}

func genFile(disjunctions bool) ast.File {
	var schema ast.Expr
	if disjunctions {
		schema = ast.NewBinExpr(token.OR,
			ast.NewIdent("string"),
			ast.NewIdent("bool"),
			ast.NewIdent("int"),
			ast.NewStruct(
				ast.NewIdent("a"), ast.NewIdent("string"),
				ast.NewIdent("b"), ast.NewIdent("int"),
			),
		)
	} else {
		schema = ast.NewIdent("string")
	}
	return ast.File{
		Filename: "example.cue",
		Decls: []ast.Decl{
			&ast.Field{
				Label: ast.NewIdent("#MyDef"),
				Value: ast.NewStruct(
					&ast.Field{
						Label: ast.NewIdent("input"),
						Value: schema,
					},
					&ast.Field{
						Label: ast.NewIdent("output"),
						Value: schema,
					},
				),
			},
		},
	}
}

func main() {
	r := cuecontext.New()
	file := genFile(true)
	fmt.Println("genTasks: Starting at ", time.Now())
	for i := 0; i < 1000; i++ {
		file.Decls = append(file.Decls, genTask(i, i-1))
	}
	fmt.Println("genTasks: Ended at ", time.Now())

	b, _ := format.Node(&file, format.Simplify())
	fmt.Println(string(b))
	val := r.BuildFile(&file)
	if err := val.Err(); err != nil {
		log.Fatal(err)
	}
	fmt.Println("Starting at ", time.Now())
	controller := flow.New(nil, val, ioTaskFunc)
	if err := controller.Run(context.Background()); err != nil {
		log.Fatal(err)
	}

	fmt.Println("Taking heap dump")
	outputFile, _ := os.Create("heap")
	err := pprof.WriteHeapProfile(outputFile)
	if err != nil {
		panic(err)
	}
	fmt.Println("Ended at ", time.Now())
}

func ioTaskFunc(v cue.Value) (flow.Runner, error) {
	inputPath := cue.ParsePath("input")

	input := v.LookupPath(inputPath)
	if !input.Exists() {
		return nil, nil
	}

	return flow.RunnerFunc(func(t *flow.Task) error {
		inputVal, err := t.Value().LookupPath(inputPath).String()
		if err != nil {
			return fmt.Errorf("input not of type string")
		}

		outputVal := inputVal

		return t.Fill(map[string]string{
			"output": outputVal,
		})
	}), nil
}

What did you expect to see?

Light memory usage. This tools/flow run is literally copying input -> output.

What did you see instead?

4-5GB of memory usage over the course of this run.

The executable will dump pprof files (named heap*).

There is a fairly different memory usage pattern for -disjunctions (2.5x more).

Dependencies

• #1871