From a1a23b66c8fb8689bdc8814e8b2230fcffda6b5e Mon Sep 17 00:00:00 2001 From: James Shubin Date: Mon, 25 Sep 2023 17:14:21 -0400 Subject: [PATCH] lang: funcs: structs: Core function struct modifications MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit These changes are needed for the lambdas implementation. Co-authored-by: Samuel Gélineau --- lang/funcs/structs/call.go | 217 +++++++++++++-------- lang/funcs/structs/channel_based_sink.go | 129 ++++++++++++ lang/funcs/structs/channel_based_source.go | 103 ++++++++++ lang/funcs/structs/const.go | 8 +- lang/funcs/structs/function.go | 206 ------------------- lang/funcs/structs/util.go | 74 +++++++ lang/funcs/structs/var.go | 133 ------------- 7 files changed, 446 insertions(+), 424 deletions(-) create mode 100644 lang/funcs/structs/channel_based_sink.go create mode 100644 lang/funcs/structs/channel_based_source.go delete mode 100644 lang/funcs/structs/function.go create mode 100644 lang/funcs/structs/util.go delete mode 100644 lang/funcs/structs/var.go diff --git a/lang/funcs/structs/call.go b/lang/funcs/structs/call.go index 80285000..46ed4edd 100644 --- a/lang/funcs/structs/call.go +++ b/lang/funcs/structs/call.go @@ -23,30 +23,38 @@ import ( "github.com/purpleidea/mgmt/lang/interfaces" "github.com/purpleidea/mgmt/lang/types" + "github.com/purpleidea/mgmt/lang/types/full" "github.com/purpleidea/mgmt/util/errwrap" ) const ( // CallFuncName is the unique name identifier for this function. CallFuncName = "call" + + // CallFuncArgNameFunction is the name for the edge which connects the + // input function to CallFunc. + CallFuncArgNameFunction = "fn" ) -// CallFunc is a function that takes in a function and all the args, and passes -// through the results of running the function call. +// CallFunc receives a function from upstream, but not the arguments. Instead, +// the Funcs which emit those arguments must be specified at construction time. +// The arguments are connected to the received FuncValues in such a way that +// CallFunc emits the result of applying the function to the arguments. type CallFunc struct { - Type *types.Type // this is the type of the var's value that we hold - FuncType *types.Type - Edge string // name of the edge used (typically starts with: `call:`) - //Func interfaces.Func // this isn't actually used in the Stream :/ - //Fn *types.FuncValue // pass in the actual function instead of Edge + Type *types.Type // the type of the result of applying the function + FuncType *types.Type // the type of the function + EdgeName string // name of the edge used - // Indexed specifies that args are accessed by index instead of name. - // This is currently unused. - Indexed bool + ArgVertices []interfaces.Func - init *interfaces.Init - last types.Value // last value received to use for diff - result types.Value // last calculated output + init *interfaces.Init + + lastFuncValue *full.FuncValue // remember the last function value + + // outputChan is an initially-nil channel from which we receive output + // lists from the subgraph. This channel is reset when the subgraph is + // recreated. + outputChan chan types.Value } // String returns a simple name for this function. This is needed so this struct @@ -64,7 +72,7 @@ func (obj *CallFunc) Validate() error { return fmt.Errorf("must specify a func type") } // TODO: maybe we can remove this if we use this for core functions... - if obj.Edge == "" { + if obj.EdgeName == "" { return fmt.Errorf("must specify an edge name") } typ := obj.FuncType @@ -72,6 +80,9 @@ func (obj *CallFunc) Validate() error { if err := obj.Type.Cmp(typ.Out); err != nil { return errwrap.Wrapf(err, "call expr type must match func out type") } + if len(obj.ArgVertices) != len(typ.Ord) { + return fmt.Errorf("number of arg Funcs must match number of func args in the type") + } return nil } @@ -79,25 +90,8 @@ func (obj *CallFunc) Validate() error { // Info returns some static info about itself. func (obj *CallFunc) Info() *interfaces.Info { var typ *types.Type - if obj.Type != nil { // don't panic if called speculatively - typ = &types.Type{ - Kind: types.KindFunc, // function type - Map: make(map[string]*types.Type), - Ord: []string{}, - Out: obj.Type, // this is the output type for the expression - } - - sig := obj.FuncType - if obj.Edge != "" { - typ.Map[obj.Edge] = sig // we get a function in - typ.Ord = append(typ.Ord, obj.Edge) - } - - // add any incoming args - for _, key := range sig.Ord { // sig.Out, not sig! - typ.Map[key] = sig.Map[key] - typ.Ord = append(typ.Ord, key) - } + if obj.Type != nil && obj.FuncType != nil { // don't panic if called speculatively + typ = types.NewType(fmt.Sprintf("func(%s %s) %s", obj.EdgeName, obj.FuncType, obj.Type)) } return &interfaces.Info{ @@ -111,6 +105,7 @@ func (obj *CallFunc) Info() *interfaces.Info { // Init runs some startup code for this composite function. func (obj *CallFunc) Init(init *interfaces.Init) error { obj.init = init + obj.lastFuncValue = nil return nil } @@ -119,65 +114,119 @@ func (obj *CallFunc) Init(init *interfaces.Init) error { // on the changing inputs to that value. func (obj *CallFunc) Stream(ctx context.Context) error { defer close(obj.init.Output) // the sender closes + + obj.outputChan = nil + + defer func() { + obj.init.Txn.Reverse() + }() + + canReceiveMoreFuncValues := true + canReceiveMoreOutputValues := true for { - select { - case input, ok := <-obj.init.Input: - if !ok { - return nil // can't output any more - } - //if err := input.Type().Cmp(obj.Info().Sig.Input); err != nil { - // return errwrap.Wrapf(err, "wrong function input") - //} - if obj.last != nil && input.Cmp(obj.last) == nil { - continue // value didn't change, skip it - } - obj.last = input // store for next - st := input.(*types.StructValue) // must be! - - // get the function - fn, exists := st.Lookup(obj.Edge) - if !exists { - return fmt.Errorf("missing expected input argument `%s`", obj.Edge) - } - - // get the arguments to call the function - args := []types.Value{} - typ := obj.FuncType - for ix, key := range typ.Ord { // sig! - if obj.Indexed { - key = fmt.Sprintf("%d", ix) - } - value, exists := st.Lookup(key) - // TODO: replace with: - //value, exists := st.Lookup(fmt.Sprintf("arg:%s", key)) - if !exists { - return fmt.Errorf("missing expected input argument `%s`", key) - } - args = append(args, value) - } - - // actually call it - result, err := fn.(*types.FuncValue).Call(args) - if err != nil { - return errwrap.Wrapf(err, "error calling function") - } - - // skip sending an update... - if obj.result != nil && result.Cmp(obj.result) == nil { - continue // result didn't change - } - obj.result = result // store new result - - case <-ctx.Done(): + if !canReceiveMoreFuncValues && !canReceiveMoreOutputValues { + // break return nil } select { - case obj.init.Output <- obj.result: // send - // pass + case input, ok := <-obj.init.Input: + if !ok { + obj.init.Input = nil // block looping back here + canReceiveMoreFuncValues = false + continue + } + + value, exists := input.Struct()[obj.EdgeName] + if !exists { + return fmt.Errorf("programming error, can't find edge") + } + + newFuncValue, ok := value.(*full.FuncValue) + if !ok { + return fmt.Errorf("programming error, can't convert to *FuncValue") + } + + // It's important to have this compare step to avoid + // redundant graph replacements which slow things down, + // but also cause the engine to lock, which can preempt + // the process scheduler, which can cause duplicate or + // unnecessary re-sending of values here, which causes + // the whole process to repeat ad-nauseum. + if newFuncValue == obj.lastFuncValue { + continue + } + // If we have a new function, then we need to replace + // the subgraph with a new one that uses the new + // function. + obj.lastFuncValue = newFuncValue + + if err := obj.replaceSubGraph(newFuncValue); err != nil { + return errwrap.Wrapf(err, "could not replace subgraph") + } + canReceiveMoreOutputValues = true + continue + + case outputValue, ok := <-obj.outputChan: + // send the new output value downstream + if !ok { + obj.outputChan = nil + canReceiveMoreOutputValues = false + continue + } + + // send to the output + select { + case obj.init.Output <- outputValue: + case <-ctx.Done(): + return nil + } + case <-ctx.Done(): return nil } } } + +func (obj *CallFunc) replaceSubGraph(newFuncValue *full.FuncValue) error { + // Create a subgraph which looks as follows. Most of the nodes are + // elided because we don't know which nodes the FuncValues will create. + // + // digraph { + // ArgVertices[0] -> ... + // ArgVertices[1] -> ... + // ArgVertices[2] -> ... + // + // outputFunc -> "subgraphOutput" + // } + + // delete the old subgraph + if err := obj.init.Txn.Reverse(); err != nil { + return errwrap.Wrapf(err, "could not Reverse") + } + + // create the new subgraph + // This passed in Txn has AddVertex, AddEdge, and possibly AddGraph + // methods called on it. Nothing else. It will _not_ call Commit or + // Reverse. It adds to the graph, and our Commit and Reverse operations + // are the ones that actually make the change. + outputFunc, err := newFuncValue.Call(obj.init.Txn, obj.ArgVertices) + if err != nil { + return errwrap.Wrapf(err, "could not call newFuncValue.Call()") + } + + obj.outputChan = make(chan types.Value) + edgeName := ChannelBasedSinkFuncArgName + subgraphOutput := &ChannelBasedSinkFunc{ + Name: "subgraphOutput", + Target: obj, + EdgeName: edgeName, + Chan: obj.outputChan, + Type: obj.Type, + } + edge := &interfaces.FuncEdge{Args: []string{edgeName}} + obj.init.Txn.AddVertex(subgraphOutput) + obj.init.Txn.AddEdge(outputFunc, subgraphOutput, edge) + return obj.init.Txn.Commit() +} diff --git a/lang/funcs/structs/channel_based_sink.go b/lang/funcs/structs/channel_based_sink.go new file mode 100644 index 00000000..4013c2e0 --- /dev/null +++ b/lang/funcs/structs/channel_based_sink.go @@ -0,0 +1,129 @@ +// Mgmt +// Copyright (C) 2013-2023+ James Shubin and the project contributors +// Written by James Shubin and the project contributors +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +package structs + +import ( + "context" + "fmt" + + "github.com/purpleidea/mgmt/lang/interfaces" + "github.com/purpleidea/mgmt/lang/types" +) + +const ( + // ChannelBasedSinkFuncArgName is the name for the edge which connects + // the input value to ChannelBasedSinkFunc. + ChannelBasedSinkFuncArgName = "channelBasedSinkFuncArg" +) + +// ChannelBasedSinkFunc is a Func which receives values from upstream nodes and +// emits them to a golang channel. +type ChannelBasedSinkFunc struct { + Name string + EdgeName string + Target interfaces.Func // for drawing dashed edges in the Graphviz visualization + + Chan chan types.Value + Type *types.Type + + init *interfaces.Init + last types.Value // last value received to use for diff +} + +// String returns a simple name for this function. This is needed so this struct +// can satisfy the pgraph.Vertex interface. +func (obj *ChannelBasedSinkFunc) String() string { + return obj.Name +} + +// ArgGen returns the Nth arg name for this function. +func (obj *ChannelBasedSinkFunc) ArgGen(index int) (string, error) { + if index != 0 { + return "", fmt.Errorf("the ChannelBasedSinkFunc only has one argument") + } + return obj.EdgeName, nil +} + +// Validate makes sure we've built our struct properly. It is usually unused for +// normal functions that users can use directly. +func (obj *ChannelBasedSinkFunc) Validate() error { + if obj.Chan == nil { + return fmt.Errorf("the Chan was not set") + } + return nil +} + +// Info returns some static info about itself. +func (obj *ChannelBasedSinkFunc) Info() *interfaces.Info { + return &interfaces.Info{ + Pure: false, + Memo: false, + Sig: types.NewType(fmt.Sprintf("func(%s %s) %s", obj.EdgeName, obj.Type, obj.Type)), + Err: obj.Validate(), + } +} + +// Init runs some startup code for this function. +func (obj *ChannelBasedSinkFunc) Init(init *interfaces.Init) error { + obj.init = init + return nil +} + +// Stream returns the changing values that this func has over time. +func (obj *ChannelBasedSinkFunc) Stream(ctx context.Context) error { + defer close(obj.init.Output) // the sender closes + defer close(obj.Chan) // the sender closes + + for { + select { + case input, ok := <-obj.init.Input: + if !ok { + return nil // can't output any more + } + + value, exists := input.Struct()[obj.EdgeName] + if !exists { + return fmt.Errorf("programming error, can't find edge") + } + + if obj.last != nil && value.Cmp(obj.last) == nil { + continue // value didn't change, skip it + } + obj.last = value // store so we can send after this select + + case <-ctx.Done(): + return nil + } + + select { + case obj.Chan <- obj.last: // send + case <-ctx.Done(): + return nil + } + + // Also send the value downstream. If we don't, then when we + // close the Output channel, the function engine is going to + // complain that we closed that channel without sending it any + // value. + select { + case obj.init.Output <- obj.last: // send + case <-ctx.Done(): + return nil + } + } +} diff --git a/lang/funcs/structs/channel_based_source.go b/lang/funcs/structs/channel_based_source.go new file mode 100644 index 00000000..411968d2 --- /dev/null +++ b/lang/funcs/structs/channel_based_source.go @@ -0,0 +1,103 @@ +// Mgmt +// Copyright (C) 2013-2023+ James Shubin and the project contributors +// Written by James Shubin and the project contributors +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +package structs + +import ( + "context" + "fmt" + + "github.com/purpleidea/mgmt/lang/interfaces" + "github.com/purpleidea/mgmt/lang/types" +) + +// ChannelBasedSourceFunc is a Func which receives values from a golang channel +// and emits them to the downstream nodes. +type ChannelBasedSourceFunc struct { + Name string + Source interfaces.Func // for drawing dashed edges in the Graphviz visualization + + Chan chan types.Value + Type *types.Type + + init *interfaces.Init + last types.Value // last value received to use for diff +} + +// String returns a simple name for this function. This is needed so this struct +// can satisfy the pgraph.Vertex interface. +func (obj *ChannelBasedSourceFunc) String() string { + return "ChannelBasedSourceFunc" +} + +// ArgGen returns the Nth arg name for this function. +func (obj *ChannelBasedSourceFunc) ArgGen(index int) (string, error) { + return "", fmt.Errorf("the ChannelBasedSourceFunc doesn't have any arguments") +} + +// Validate makes sure we've built our struct properly. It is usually unused for +// normal functions that users can use directly. +func (obj *ChannelBasedSourceFunc) Validate() error { + if obj.Chan == nil { + return fmt.Errorf("the Chan was not set") + } + return nil +} + +// Info returns some static info about itself. +func (obj *ChannelBasedSourceFunc) Info() *interfaces.Info { + return &interfaces.Info{ + Pure: false, + Memo: false, + Sig: types.NewType(fmt.Sprintf("func() %s", obj.Type)), + Err: obj.Validate(), + } +} + +// Init runs some startup code for this function. +func (obj *ChannelBasedSourceFunc) Init(init *interfaces.Init) error { + obj.init = init + return nil +} + +// Stream returns the changing values that this func has over time. +func (obj *ChannelBasedSourceFunc) Stream(ctx context.Context) error { + defer close(obj.init.Output) // the sender closes + + for { + select { + case input, ok := <-obj.Chan: + if !ok { + return nil // can't output any more + } + + //if obj.last != nil && input.Cmp(obj.last) == nil { + // continue // value didn't change, skip it + //} + obj.last = input // store so we can send after this select + + case <-ctx.Done(): + return nil + } + + select { + case obj.init.Output <- obj.last: // send + case <-ctx.Done(): + return nil + } + } +} diff --git a/lang/funcs/structs/const.go b/lang/funcs/structs/const.go index 8f18c6ab..4a718109 100644 --- a/lang/funcs/structs/const.go +++ b/lang/funcs/structs/const.go @@ -32,7 +32,8 @@ const ( // ConstFunc is a function that returns the constant value passed to Value. type ConstFunc struct { - Value types.Value + Value types.Value + NameHint string init *interfaces.Init } @@ -40,7 +41,12 @@ type ConstFunc struct { // String returns a simple name for this function. This is needed so this struct // can satisfy the pgraph.Vertex interface. func (obj *ConstFunc) String() string { + if obj.NameHint != "" { + return obj.NameHint + } return ConstFuncName + //return fmt.Sprintf("%s: %s(%s)", ConstFuncName, obj.Value.Type().String(), obj.Value.String()) + //return fmt.Sprintf("%s(%s)", obj.Value.Type().String(), obj.Value.String()) } // Validate makes sure we've built our struct properly. diff --git a/lang/funcs/structs/function.go b/lang/funcs/structs/function.go deleted file mode 100644 index 330fa296..00000000 --- a/lang/funcs/structs/function.go +++ /dev/null @@ -1,206 +0,0 @@ -// Mgmt -// Copyright (C) 2013-2023+ James Shubin and the project contributors -// Written by James Shubin and the project contributors -// -// This program is free software: you can redistribute it and/or modify -// it under the terms of the GNU General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// This program is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU General Public License for more details. -// -// You should have received a copy of the GNU General Public License -// along with this program. If not, see . - -package structs - -import ( - "context" - "fmt" - - "github.com/purpleidea/mgmt/lang/funcs" - "github.com/purpleidea/mgmt/lang/interfaces" - "github.com/purpleidea/mgmt/lang/types" - "github.com/purpleidea/mgmt/util/errwrap" -) - -const ( - // FunctionFuncName is the unique name identifier for this function. - FunctionFuncName = "function" -) - -// FunctionFunc is a function that passes through the function body it receives. -type FunctionFunc struct { - Type *types.Type // this is the type of the function that we hold - Edge string // name of the edge used (typically "body") - Func interfaces.Func - Fn *types.FuncValue - - init *interfaces.Init - last types.Value // last value received to use for diff - result types.Value // last calculated output -} - -// String returns a simple name for this function. This is needed so this struct -// can satisfy the pgraph.Vertex interface. -func (obj *FunctionFunc) String() string { - return FunctionFuncName -} - -// fn returns the function that wraps the Func interface if that API is used. -func (obj *FunctionFunc) fn() (*types.FuncValue, error) { - fn := func(args []types.Value) (types.Value, error) { - // FIXME: can we run a recursive engine - // to support running non-pure functions? - if !obj.Func.Info().Pure { - return nil, fmt.Errorf("only pure functions can be used by value") - } - - // XXX: this might not be needed anymore... - return funcs.PureFuncExec(obj.Func, args) - } - - result := types.NewFunc(obj.Type) // new func - if err := result.Set(fn); err != nil { - return nil, errwrap.Wrapf(err, "can't build func from built-in") - } - - return result, nil -} - -// Validate makes sure we've built our struct properly. -func (obj *FunctionFunc) Validate() error { - if obj.Type == nil { - return fmt.Errorf("must specify a type") - } - if obj.Type.Kind != types.KindFunc { - return fmt.Errorf("can't use type `%s`", obj.Type.String()) - } - if obj.Edge == "" && obj.Func == nil && obj.Fn == nil { - return fmt.Errorf("must specify an Edge, Func, or Fn") - } - - if obj.Fn != nil && obj.Fn.Type() != obj.Type { - return fmt.Errorf("type of Fn did not match input Type") - } - - return nil -} - -// Info returns some static info about itself. -func (obj *FunctionFunc) Info() *interfaces.Info { - var typ *types.Type - if obj.Type != nil { // don't panic if called speculatively - typ = &types.Type{ - Kind: types.KindFunc, // function type - Map: make(map[string]*types.Type), - Ord: []string{}, - Out: obj.Type, // after the function is called it's this... - } - - // type of body is what we'd get by running the function (what's inside) - if obj.Edge != "" { - typ.Map[obj.Edge] = obj.Type.Out - typ.Ord = append(typ.Ord, obj.Edge) - } - } - - pure := true // assume true - if obj.Func != nil { - pure = obj.Func.Info().Pure - } - - return &interfaces.Info{ - Pure: pure, // TODO: can we guarantee this? - Memo: false, // TODO: ??? - Sig: typ, - Err: obj.Validate(), - } -} - -// Init runs some startup code for this composite function. -func (obj *FunctionFunc) Init(init *interfaces.Init) error { - obj.init = init - return nil -} - -// Stream takes an input struct in the format as described in the Func and Graph -// methods of the Expr, and returns the actual expected value as a stream based -// on the changing inputs to that value. -func (obj *FunctionFunc) Stream(ctx context.Context) error { - defer close(obj.init.Output) // the sender closes - for { - select { - case input, ok := <-obj.init.Input: - if !ok { - if obj.Edge != "" { // then it's not a built-in - return nil // can't output any more - } - - var result *types.FuncValue - - if obj.Fn != nil { - result = obj.Fn - } else { - var err error - result, err = obj.fn() - if err != nil { - return err - } - } - - // if we never had input args, send the function - select { - case obj.init.Output <- result: // send - // pass - case <-ctx.Done(): - return nil - } - - return nil - } - //if err := input.Type().Cmp(obj.Info().Sig.Input); err != nil { - // return errwrap.Wrapf(err, "wrong function input") - //} - if obj.last != nil && input.Cmp(obj.last) == nil { - continue // value didn't change, skip it - } - obj.last = input // store for next - - var result types.Value - - st := input.(*types.StructValue) // must be! - value, exists := st.Lookup(obj.Edge) // single argName - if !exists { - return fmt.Errorf("missing expected input argument `%s`", obj.Edge) - } - - result = obj.Type.New() // new func - fn := func([]types.Value) (types.Value, error) { - return value, nil - } - if err := result.(*types.FuncValue).Set(fn); err != nil { - return errwrap.Wrapf(err, "can't build func with body") - } - - // skip sending an update... - if obj.result != nil && result.Cmp(obj.result) == nil { - continue // result didn't change - } - obj.result = result // store new result - - case <-ctx.Done(): - return nil - } - - select { - case obj.init.Output <- obj.result: // send - // pass - case <-ctx.Done(): - return nil - } - } -} diff --git a/lang/funcs/structs/util.go b/lang/funcs/structs/util.go new file mode 100644 index 00000000..d5b98c38 --- /dev/null +++ b/lang/funcs/structs/util.go @@ -0,0 +1,74 @@ +// Mgmt +// Copyright (C) 2013-2023+ James Shubin and the project contributors +// Written by James Shubin and the project contributors +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. +// +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +package structs + +import ( + "github.com/purpleidea/mgmt/lang/funcs/simple" + "github.com/purpleidea/mgmt/lang/interfaces" + "github.com/purpleidea/mgmt/lang/types" + "github.com/purpleidea/mgmt/lang/types/full" +) + +// In the following set of conversion functions, a "constant" Func is a node +// with in-degree zero which always outputs the same function value, while a +// "direct" Func is a node with one upstream node for each of the function's +// arguments. + +// FuncValueToConstFunc transforms a *full.FuncValue into an interfaces.Func +// which is implemented by &ConstFunc{}. +func FuncValueToConstFunc(fv *full.FuncValue) interfaces.Func { + return &ConstFunc{ + Value: fv, + NameHint: "FuncValue", + } +} + +// SimpleFnToDirectFunc transforms a name and *types.FuncValue into an +// interfaces.Func which is implemented by &simple.WrappedFunc{}. +func SimpleFnToDirectFunc(name string, fv *types.FuncValue) interfaces.Func { + return &simple.WrappedFunc{ + Name: name, + Fn: fv, + } +} + +// SimpleFnToFuncValue transforms a name and *types.FuncValue into a +// *full.FuncValue. +func SimpleFnToFuncValue(name string, fv *types.FuncValue) *full.FuncValue { + return &full.FuncValue{ + V: func(txn interfaces.Txn, args []interfaces.Func) (interfaces.Func, error) { + wrappedFunc := SimpleFnToDirectFunc(name, fv) + txn.AddVertex(wrappedFunc) + for i, arg := range args { + argName := fv.T.Ord[i] + txn.AddEdge(arg, wrappedFunc, &interfaces.FuncEdge{ + Args: []string{argName}, + }) + } + return wrappedFunc, nil + }, + T: fv.T, + } +} + +// SimpleFnToConstFunc transforms a name and *types.FuncValue into an +// interfaces.Func which is implemented by FuncValueToConstFunc and +// SimpleFnToFuncValue. +func SimpleFnToConstFunc(name string, fv *types.FuncValue) interfaces.Func { + return FuncValueToConstFunc(SimpleFnToFuncValue(name, fv)) +} diff --git a/lang/funcs/structs/var.go b/lang/funcs/structs/var.go deleted file mode 100644 index 8fef1330..00000000 --- a/lang/funcs/structs/var.go +++ /dev/null @@ -1,133 +0,0 @@ -// Mgmt -// Copyright (C) 2013-2023+ James Shubin and the project contributors -// Written by James Shubin and the project contributors -// -// This program is free software: you can redistribute it and/or modify -// it under the terms of the GNU General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// This program is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU General Public License for more details. -// -// You should have received a copy of the GNU General Public License -// along with this program. If not, see . - -package structs - -import ( - "context" - "fmt" - - "github.com/purpleidea/mgmt/lang/interfaces" - "github.com/purpleidea/mgmt/lang/types" - //"github.com/purpleidea/mgmt/util/errwrap" -) - -const ( - // VarFuncName is the unique name identifier for this function. - VarFuncName = "var" -) - -// VarFunc is a function that passes through a function that came from a bind -// lookup. It exists so that the reactive function engine type checks correctly. -type VarFunc struct { - Type *types.Type // this is the type of the var's value that we hold - Edge string // name of the edge used - //Func interfaces.Func // this isn't actually used in the Stream :/ - - init *interfaces.Init - last types.Value // last value received to use for diff - result types.Value // last calculated output -} - -// String returns a simple name for this function. This is needed so this struct -// can satisfy the pgraph.Vertex interface. -func (obj *VarFunc) String() string { - return VarFuncName -} - -// Validate makes sure we've built our struct properly. -func (obj *VarFunc) Validate() error { - if obj.Type == nil { - return fmt.Errorf("must specify a type") - } - if obj.Edge == "" { - return fmt.Errorf("must specify an edge name") - } - return nil -} - -// Info returns some static info about itself. -func (obj *VarFunc) Info() *interfaces.Info { - var typ *types.Type - if obj.Type != nil { // don't panic if called speculatively - typ = &types.Type{ - Kind: types.KindFunc, // function type - Map: map[string]*types.Type{obj.Edge: obj.Type}, - Ord: []string{obj.Edge}, - Out: obj.Type, // this is the output type for the expression - } - } - - return &interfaces.Info{ - Pure: true, - Memo: false, // TODO: ??? - Sig: typ, - Err: obj.Validate(), - } -} - -// Init runs some startup code for this composite function. -func (obj *VarFunc) Init(init *interfaces.Init) error { - obj.init = init - return nil -} - -// Stream takes an input struct in the format as described in the Func and Graph -// methods of the Expr, and returns the actual expected value as a stream based -// on the changing inputs to that value. -func (obj *VarFunc) Stream(ctx context.Context) error { - defer close(obj.init.Output) // the sender closes - for { - select { - case input, ok := <-obj.init.Input: - if !ok { - return nil // can't output any more - } - //if err := input.Type().Cmp(obj.Info().Sig.Input); err != nil { - // return errwrap.Wrapf(err, "wrong function input") - //} - if obj.last != nil && input.Cmp(obj.last) == nil { - continue // value didn't change, skip it - } - obj.last = input // store for next - - var result types.Value - st := input.(*types.StructValue) // must be! - value, exists := st.Lookup(obj.Edge) - if !exists { - return fmt.Errorf("missing expected input argument `%s`", obj.Edge) - } - result = value - - // skip sending an update... - if obj.result != nil && result.Cmp(obj.result) == nil { - continue // result didn't change - } - obj.result = result // store new result - - case <-ctx.Done(): - return nil - } - - select { - case obj.init.Output <- obj.result: // send - // pass - case <-ctx.Done(): - return nil - } - } -}