lang: Core language and GAPI changes

These changes help plumb things in more easily for the lambdas work.

gapi/gapi.go

@@ -119,11 +119,13 @@ type GAPI interface {
 
 	// Next returns a stream of switch events. The engine will run Graph()
 	// to build a new graph after every Next event.
+	// TODO: add context for shutting down to the input and change Close to Cleanup
 	Next() chan Next
 
 	// Close shuts down the GAPI. It asks the GAPI to close, and must cause
 	// Next() to unblock even if is currently blocked and waiting to send a
 	// new event.
+	// TODO: change Close to Cleanup
 	Close() error
 }
 

lang/gapi/gapi.go

@@ -20,6 +20,7 @@ package gapi
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"strings"
 	"sync"
@@ -61,7 +62,11 @@ func init() {
 type GAPI struct {
 	InputURI string // input URI of code file system to run
 
-	lang *lang.Lang // lang struct
+	lang   *lang.Lang // lang struct
+	wgRun  *sync.WaitGroup
+	ctx    context.Context
+	cancel func()
+	reterr error
 
 	// this data struct is only available *after* Init, so as a result, it
 	// can not be used inside the Cli(...) method.
@@ -472,13 +477,27 @@ func (obj *GAPI) LangInit() error {
 	if err := obj.lang.Init(); err != nil {
 		return errwrap.Wrapf(err, "can't init the lang")
 	}
+
+	// XXX: I'm certain I've probably got a deadlock or race somewhere here
+	// or in lib/main.go so we'll fix it with an API fixup and rewrite soon
+	obj.wgRun = &sync.WaitGroup{}
+	obj.ctx, obj.cancel = context.WithCancel(context.Background())
+	obj.wgRun.Add(1)
+	go func() {
+		defer obj.wgRun.Done()
+		obj.reterr = obj.lang.Run(obj.ctx)
+	}()
+
 	return nil
 }
 
 // LangClose is a wrapper around the lang Close method.
 func (obj *GAPI) LangClose() error {
 	if obj.lang != nil {
-		err := obj.lang.Close()
+		obj.cancel()
+		obj.wgRun.Wait()
+		err := obj.lang.Cleanup()
+		err = errwrap.Append(err, obj.reterr)             // from obj.lang.Run
 		obj.lang = nil                                    // clear it to avoid double closing
 		return errwrap.Wrapf(err, "can't close the lang") // nil passthrough
 	}
@@ -520,7 +539,7 @@ func (obj *GAPI) Next() chan gapi.Next {
 		startChan := make(chan struct{}) // start signal
 		close(startChan)                 // kick it off!
 
-		streamChan := make(chan error)
+		streamChan := make(<-chan error)
 		//defer obj.LangClose() // close any old lang
 
 		var ok bool
@@ -545,6 +564,7 @@ func (obj *GAPI) Next() chan gapi.Next {
 			obj.data.Logf("generating new graph...")
 
 			// skip this to pass through the err if present
+			// XXX: redo this old garbage code
 			if langSwap && err == nil {
 				obj.data.Logf("swap!")
 				// run up to these three but fail on err

lang/lang.go

@@ -21,13 +21,14 @@ package lang
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"sync"
 
 	"github.com/purpleidea/mgmt/engine"
 	"github.com/purpleidea/mgmt/lang/ast"
-	"github.com/purpleidea/mgmt/lang/funcs"
 	_ "github.com/purpleidea/mgmt/lang/funcs/core" // import so the funcs register
+	"github.com/purpleidea/mgmt/lang/funcs/dage"
 	"github.com/purpleidea/mgmt/lang/funcs/vars"
 	"github.com/purpleidea/mgmt/lang/inputs"
 	"github.com/purpleidea/mgmt/lang/interfaces"
@@ -62,30 +63,20 @@ type Lang struct {
 	Logf     func(format string, v ...interface{})
 
 	ast   interfaces.Stmt // store main prog AST here
-	funcs *funcs.Engine   // function event engine
+	funcs *dage.Engine    // function event engine
+	graph *pgraph.Graph   // function graph
 
-	loadedChan chan struct{} // loaded signal
-
-	streamChan chan error // signals a new graph can be created or problem
+	streamChan <-chan error // signals a new graph can be created or problem
 	//streamBurst bool // should we try and be bursty with the stream events?
 
-	closeChan chan struct{} // close signal
-	wg        *sync.WaitGroup
+	wg *sync.WaitGroup
 }
 
-// Init initializes the lang struct, and starts up the initial data sources.
+// Init initializes the lang struct, and starts up the initial input parsing.
 // NOTE: The trick is that we need to get the list of funcs to watch AND start
 // watching them, *before* we pull their values, that way we'll know if they
 // changed from the values we wanted.
 func (obj *Lang) Init() error {
-	obj.loadedChan = make(chan struct{})
-	obj.streamChan = make(chan error)
-	obj.closeChan = make(chan struct{})
-	obj.wg = &sync.WaitGroup{}
-
-	once := &sync.Once{}
-	loadedSignal := func() { close(obj.loadedChan) } // only run once!
-
 	if obj.Debug {
 		obj.Logf("input: %s", obj.Input)
 		tree, err := util.FsTree(obj.Fs, "/") // should look like gapi
@@ -215,114 +206,120 @@ func (obj *Lang) Init() error {
 	obj.Logf("building function graph...")
 	// we assume that for some given code, the list of funcs doesn't change
 	// iow, we don't support variable, variables or absurd things like that
-	graph, err := obj.ast.Graph() // build the graph of functions
+	obj.graph = &pgraph.Graph{Name: "functionGraph"}
+	env := make(map[string]interfaces.Func)
+	for k, v := range scope.Variables {
+		g, builtinFunc, err := v.Graph(nil)
+		if err != nil {
+			return errwrap.Wrapf(err, "calling Graph on builtins")
+		}
+		obj.graph.AddGraph(g)
+		env[k] = builtinFunc
+	}
+	g, err := obj.ast.Graph() // build the graph of functions
 	if err != nil {
 		return errwrap.Wrapf(err, "could not generate function graph")
 	}
+	obj.graph.AddGraph(g)
 
 	if obj.Debug {
-		obj.Logf("function graph: %+v", graph)
-		graph.Logf(obj.Logf) // log graph output with this logger...
+		obj.Logf("function graph: %+v", obj.graph)
+		obj.graph.Logf(obj.Logf) // log graph output with this logger...
+		//if err := obj.graph.ExecGraphviz("/tmp/graphviz.dot"); err != nil {
+		//	return errwrap.Wrapf(err, "writing graph failed")
+		//}
 	}
 
-	if graph.NumVertices() == 0 { // no funcs to load!
-		// send only one signal since we won't ever send after this!
-		obj.Logf("static graph found")
-		obj.wg.Add(1)
-		go func() {
-			defer obj.wg.Done()
-			defer close(obj.streamChan) // no more events are coming!
-			close(obj.loadedChan)       // signal
-			select {
-			case obj.streamChan <- nil: // send one signal
-				// pass
-			case <-obj.closeChan:
-				return
-			}
-		}()
-		return nil // exit early, no funcs to load!
-	}
-
-	obj.funcs = &funcs.Engine{
-		Graph:    graph, // not the same as the output graph!
+	obj.funcs = &dage.Engine{
+		Name:     "lang", // TODO: arbitrary name for now
 		Hostname: obj.Hostname,
 		World:    obj.World,
 		Debug:    obj.Debug,
 		Logf: func(format string, v ...interface{}) {
 			obj.Logf("funcs: "+format, v...)
 		},
-		Glitch: false, // FIXME: verify this functionality is perfect!
 	}
 
 	obj.Logf("function engine initializing...")
-	if err := obj.funcs.Init(); err != nil {
+	if err := obj.funcs.Setup(); err != nil {
 		return errwrap.Wrapf(err, "init error with func engine")
 	}
 
-	obj.Logf("function engine validating...")
-	if err := obj.funcs.Validate(); err != nil {
-		return errwrap.Wrapf(err, "validate error with func engine")
-	}
+	obj.streamChan = obj.funcs.Stream() // after obj.funcs.Setup runs
+
+	return nil
+}
+
+// Run kicks off the function engine. Use the context to shut it down.
+func (obj *Lang) Run(ctx context.Context) (reterr error) {
+	wg := &sync.WaitGroup{}
+	defer wg.Wait()
+
+	runCtx, cancel := context.WithCancel(context.Background()) // Don't inherit from parent
+	defer cancel()
+
+	//obj.Logf("function engine validating...")
+	//if err := obj.funcs.Validate(); err != nil {
+	//	return errwrap.Wrapf(err, "validate error with func engine")
+	//}
 
 	obj.Logf("function engine starting...")
-	// On failure, we expect the caller to run Close() to shutdown all of
-	// the currently initialized (and running) funcs... This is needed if
-	// we successfully ran `Run` but isn't needed only for Init/Validate.
-	if err := obj.funcs.Run(); err != nil {
-		return errwrap.Wrapf(err, "run error with func engine")
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := obj.funcs.Run(runCtx); err == nil {
+			reterr = errwrap.Append(reterr, err)
+		}
+		// Run() should only error if not a dag I think...
+	}()
+
+	<-obj.funcs.Started() // wait for startup (will not block forever)
+
+	// Sanity checks for graph size.
+	if count := obj.funcs.NumVertices(); count != 0 {
+		return fmt.Errorf("expected empty graph on start, got %d vertices", count)
 	}
+	defer func() {
+		if count := obj.funcs.NumVertices(); count != 0 {
+			err := fmt.Errorf("expected empty graph on exit, got %d vertices", count)
+			reterr = errwrap.Append(reterr, err)
+		}
+	}()
+	defer wg.Wait()
+	defer cancel() // now cancel Run only after Reverse and Free are done!
+
+	txn := obj.funcs.Txn()
+	defer txn.Free() // remember to call Free()
+	txn.AddGraph(obj.graph)
+	if err := txn.Commit(); err != nil {
+		return errwrap.Wrapf(err, "error adding to function graph engine")
+	}
+	defer func() {
+		if err := txn.Reverse(); err != nil { // should remove everything we added
+			reterr = errwrap.Append(reterr, err)
+		}
+	}()
 
 	// wait for some activity
 	obj.Logf("stream...")
-	stream := obj.funcs.Stream()
-	obj.wg.Add(1)
-	go func() {
-		obj.Logf("loop...")
-		defer obj.wg.Done()
-		defer close(obj.streamChan) // no more events are coming!
-		for {
-			var err error
-			var ok bool
-			select {
-			case err, ok = <-stream:
-				if !ok {
-					obj.Logf("stream closed")
-					return
-				}
-				if err == nil {
-					// only do this once, on the first event
-					once.Do(loadedSignal) // signal
-				}
 
-			case <-obj.closeChan:
-				return
-			}
+	select {
+	case <-ctx.Done():
+	}
 
-			select {
-			case obj.streamChan <- err: // send
-				if err != nil {
-					obj.Logf("stream error: %+v", err)
-					return
-				}
-
-			case <-obj.closeChan:
-				return
-			}
-		}
-	}()
 	return nil
 }
 
 // Stream returns a channel of graph change requests or errors. These are
 // usually sent when a func output changes.
-func (obj *Lang) Stream() chan error {
+func (obj *Lang) Stream() <-chan error {
 	return obj.streamChan
 }
 
 // Interpret runs the interpreter and returns a graph and corresponding error.
 func (obj *Lang) Interpret() (*pgraph.Graph, error) {
 	select {
-	case <-obj.loadedChan: // funcs are now loaded!
+	case <-obj.funcs.Loaded(): // funcs are now loaded!
 		// pass
 	default:
 		// if this is hit, someone probably called this too early!
@@ -332,37 +329,9 @@ func (obj *Lang) Interpret() (*pgraph.Graph, error) {
 
 	obj.Logf("running interpret...")
 	table := obj.funcs.Table() // map[pgraph.Vertex]types.Value
-	fn := func(n interfaces.Node) error {
-		expr, ok := n.(interfaces.Expr)
-		if !ok {
-			return nil
-		}
-		v, ok := expr.(pgraph.Vertex)
-		if !ok {
-			panic("programming error in interfaces.Expr -> pgraph.Vertex lookup")
-		}
-		val, exists := table[v]
-		if !exists {
-			fmt.Printf("XXX: missing value in table is pointer: %p\n", v)
-			return nil // XXX: workaround for now...
-			//return fmt.Errorf("missing value in table for: %s", v)
-		}
-		return expr.SetValue(val) // set the value
-	}
-	obj.funcs.Lock() // XXX: apparently there are races between SetValue and reading obj.V values...
-	if err := obj.ast.Apply(fn); err != nil {
-		if obj.Debug {
-			for k, v := range table {
-				obj.Logf("table: key: %+v ; value: %+v", k, v)
-			}
-		}
-		obj.funcs.Unlock()
-		return nil, err
-	}
-	obj.funcs.Unlock()
 
 	// this call returns the graph
-	graph, err := interpret.Interpret(obj.ast)
+	graph, err := interpret.Interpret(obj.ast, table)
 	if err != nil {
 		return nil, errwrap.Wrapf(err, "could not interpret")
 	}
@@ -370,14 +339,7 @@ func (obj *Lang) Interpret() (*pgraph.Graph, error) {
 	return graph, nil // return a graph
 }
 
-// Close shuts down the lang struct and causes all the funcs to shutdown. It
-// must be called when finished after any successful Init ran.
-func (obj *Lang) Close() error {
-	var err error
-	if obj.funcs != nil {
-		err = obj.funcs.Close()
-	}
-	close(obj.closeChan)
-	obj.wg.Wait()
-	return err
+// Cleanup cleans up and frees memory and resources after everything is done.
+func (obj *Lang) Cleanup() error {
+	return obj.funcs.Cleanup()
 }

lang/lang_test.go

@@ -20,7 +20,9 @@
 package lang
 
 import (
+	"context"
 	"fmt"
+	"sync"
 	"testing"
 
 	"github.com/purpleidea/mgmt/engine"
@@ -90,7 +92,7 @@ func edgeCmpFn(e1, e2 pgraph.Edge) (bool, error) {
 	return e1.String() == e2.String(), nil
 }
 
-func runInterpret(t *testing.T, code string) (*pgraph.Graph, error) {
+func runInterpret(t *testing.T, code string) (_ *pgraph.Graph, reterr error) {
 	logf := func(format string, v ...interface{}) {
 		t.Logf("test: lang: "+format, v...)
 	}
@@ -123,31 +125,41 @@ func runInterpret(t *testing.T, code string) (*pgraph.Graph, error) {
 	if err := lang.Init(); err != nil {
 		return nil, errwrap.Wrapf(err, "init failed")
 	}
-	closeFn := func() error {
-		return errwrap.Wrapf(lang.Close(), "close failed")
-	}
+	defer lang.Cleanup()
+
+	wg := &sync.WaitGroup{}
+	defer wg.Wait()
+
+	ctx, cancel := context.WithCancel(context.Background()) // TODO: get it from parent
+	defer cancel()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := lang.Run(ctx); err != nil {
+			reterr = errwrap.Append(reterr, err)
+		}
+	}()
+	defer cancel() // shutdown the Run
 
 	// we only wait for the first event, instead of the continuous stream
 	select {
 	case err, ok := <-lang.Stream():
 		if !ok {
-			return nil, errwrap.Wrapf(closeFn(), "stream closed without event")
+			return nil, fmt.Errorf("stream closed without event")
 		}
 		if err != nil {
-			return nil, errwrap.Wrapf(err, "stream failed, close: %+v", closeFn())
+			return nil, errwrap.Wrapf(err, "stream failed")
 		}
 	}
 
 	// run artificially without the entire GAPI loop
 	graph, err := lang.Interpret()
 	if err != nil {
-		err := errwrap.Wrapf(err, "interpret failed")
-		e := closeFn()
-		err = errwrap.Append(err, e) // list of errors
-		return nil, err
+		return nil, errwrap.Wrapf(err, "interpret failed")
 	}
 
-	return graph, closeFn()
+	return graph, nil
 }
 
 // TODO: empty code is not currently allowed, should we allow it?

lang/unification_test.go

@@ -451,60 +451,6 @@ func TestUnification1(t *testing.T) {
 			},
 		})
 	}
-	{
-		//$v = 42
-		//$x = template("hello", $v) # redirect var for harder unification
-		//test "t1" {
-		//	anotherstr => $x,
-		//}
-		innerFunc := &ast.ExprCall{
-			Name: "template",
-			Args: []interfaces.Expr{
-				&ast.ExprStr{
-					V: "hello", // whatever...
-				},
-				&ast.ExprVar{
-					Name: "v",
-				},
-			},
-		}
-		stmt := &ast.StmtProg{
-			Body: []interfaces.Stmt{
-				&ast.StmtBind{
-					Ident: "v",
-					Value: &ast.ExprInt{
-						V: 42,
-					},
-				},
-				&ast.StmtBind{
-					Ident: "x",
-					Value: innerFunc,
-				},
-				&ast.StmtRes{
-					Kind: "test",
-					Name: &ast.ExprStr{
-						V: "t1",
-					},
-					Contents: []ast.StmtResContents{
-						&ast.StmtResField{
-							Field: "anotherstr",
-							Value: &ast.ExprVar{
-								Name: "x",
-							},
-						},
-					},
-				},
-			},
-		}
-		testCases = append(testCases, test{
-			name: "complex template",
-			ast:  stmt,
-			fail: false,
-			expect: map[interfaces.Expr]*types.Type{
-				innerFunc: types.NewType("str"),
-			},
-		})
-	}
 	{
 		// import "datetime"
 		//test "t1" {