context local cache

dynaml/exec.go

@@ -8,7 +8,6 @@ import (
 	"os/exec"
 	"strconv"
 	"strings"
-	"sync"
 
 	"github.com/mandelsoft/spiff/legacy/candiedyaml"
 
@@ -17,6 +16,8 @@ import (
 )
 
 func func_exec(cached bool, arguments []interface{}, binding Binding) (interface{}, EvaluationInfo, bool) {
+	var cache ExecCache
+
 	info := DefaultInfo()
 
 	if len(arguments) < 1 {
@@ -25,6 +26,9 @@ func func_exec(cached bool, arguments []interface{}, binding Binding) (interface
 	if !binding.GetState().OSAccessAllowed() {
 		return info.DenyOSOperation("exec")
 	}
+	if cached {
+		cache = binding.GetState().GetExecCache()
+	}
 	args := []string{}
 	wopt := WriteOpts{}
 	debug.Debug("exec: found %d arguments for call\n", len(arguments))
@@ -52,7 +56,7 @@ func func_exec(cached bool, arguments []interface{}, binding Binding) (interface
 			args = append(args, v)
 		}
 	}
-	result, err := cachedExecute(cached, nil, args)
+	result, err := cachedExecute(cache, nil, args)
 	if err != nil {
 		return info.Error("execution '%s' failed", args[0])
 	}
@@ -122,14 +126,11 @@ func getArg(key interface{}, value interface{}, wopt WriteOpts, allowyaml bool)
 	}
 }
 
-var cache = make(map[string][]byte)
-var lock sync.Mutex
-
 type Bytes interface {
 	Bytes() []byte
 }
 
-func cachedExecute(cached bool, content *string, args []string) ([]byte, error) {
+func cachedExecute(cache ExecCache, content *string, args []string) ([]byte, error) {
 	h := md5.New()
 	if content != nil {
 		h.Write([]byte(*content))
@@ -138,10 +139,10 @@ func cachedExecute(cached bool, content *string, args []string) ([]byte, error)
 		h.Write([]byte(arg))
 	}
 	hash := fmt.Sprintf("%x", h.Sum(nil))
-	if cached {
-		lock.Lock()
-		defer lock.Unlock()
-		result := cache[hash]
+	if cache != nil {
+		cache.Lock()
+		defer cache.Unlock()
+		result := cache.Get(hash)
 		if result != nil {
 			debug.Debug("exec: reusing cache %s for %v\n", hash, args)
 			return result, nil
@@ -158,8 +159,8 @@ func cachedExecute(cached bool, content *string, args []string) ([]byte, error)
 		fmt.Fprintf(os.Stderr, "exec: calling %v\n", args)
 		fmt.Fprintf(os.Stderr, "  error: %v\n", stderr)
 	}
-	if cached {
-		cache[hash] = result
+	if cache != nil {
+		cache.Set(hash, result)
 	}
 	return result, err
 }

dynaml/expression.go

@@ -17,6 +17,14 @@ type SourceProvider interface {
 	SourceName() string
 }
 
+type ExecCache interface {
+	Lock()
+	Unlock()
+	Get(key string) []byte
+	Set(key string, content []byte)
+	Clear()
+}
+
 type State interface {
 	GetTempName(data []byte) (string, error)
 	GetFileContent(file string, cached bool) ([]byte, error)
@@ -26,6 +34,7 @@ type State interface {
 	FileSystem() vfs.VFS
 	GetRegistry() Registry
 	GetFeatures() features.FeatureFlags
+	GetExecCache() ExecCache
 	InterpolationEnabled() bool
 	ControlEnabled() bool
 	SetTag(name string, node yaml.Node, path []string, scope TagScope) error

dynaml/pipe.go

@@ -6,6 +6,7 @@ import (
 )
 
 func func_pipe(cached bool, arguments []interface{}, binding Binding) (interface{}, EvaluationInfo, bool) {
+	var cache ExecCache
 	info := DefaultInfo()
 
 	if len(arguments) <= 2 {
@@ -14,6 +15,9 @@ func func_pipe(cached bool, arguments []interface{}, binding Binding) (interface
 	if !binding.GetState().OSAccessAllowed() {
 		return info.DenyOSOperation("pipe")
 	}
+	if cached {
+		cache = binding.GetState().GetExecCache()
+	}
 	args := []string{}
 	wopt := WriteOpts{}
 	debug.Debug("pipe: found %d arguments for call\n", len(arguments))
@@ -49,7 +53,7 @@ func func_pipe(cached bool, arguments []interface{}, binding Binding) (interface
 			args = append(args, v)
 		}
 	}
-	result, err := cachedExecute(cached, &args[0], args[1:])
+	result, err := cachedExecute(cache, &args[0], args[1:])
 	if err != nil {
 		return info.Error("execution '%s' failed", args[1])
 	}

flow/state.go

@@ -11,6 +11,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 
 	"github.com/mandelsoft/vfs/pkg/osfs"
 	"github.com/mandelsoft/vfs/pkg/vfs"
@@ -24,12 +25,40 @@ import (
 const MODE_FILE_ACCESS = 1 // support file system access
 const MODE_OS_ACCESS = 2   // support os commands like pipe and exec
 
+type execCache struct {
+	cache map[string][]byte
+	lock  sync.Mutex
+}
+
+func (c *execCache) Lock() {
+	c.lock.Lock()
+}
+
+func (c *execCache) Unlock() {
+	c.lock.Unlock()
+}
+
+func (c *execCache) Clear() {
+	c.cache = make(map[string][]byte)
+}
+
+func (c *execCache) Get(key string) []byte {
+	return c.cache[key]
+}
+
+func (c *execCache) Set(key string, content []byte) {
+	c.cache[key] = content
+}
+
+var _ dynaml.ExecCache = &execCache{}
+
 type State struct {
 	files      map[string]string // content hash to temp file name
 	fileCache  map[string][]byte // file content cache
 	key        string            // default encryption key
 	mode       int
-	fileSystem vfs.VFS // virtual filesystem to use for filesystem based operations
+	exec_cache dynaml.ExecCache // execution cache
+	fileSystem vfs.VFS          // virtual filesystem to use for filesystem based operations
 	registry   dynaml.Registry
 	features   features.FeatureFlags
 	tags       map[string]*dynaml.TagInfo
@@ -54,6 +83,7 @@ func NewState(key string, mode int, optfs ...vfs.FileSystem) *State {
 		fileCache:  map[string][]byte{},
 		key:        key,
 		mode:       mode,
+		exec_cache: &execCache{cache: make(map[string][]byte)},
 		fileSystem: vfs.New(fs),
 		docno:      1,
 		features:   features.Features(),
@@ -133,6 +163,10 @@ func (s *State) GetEncryptionKey() string {
 	return s.key
 }
 
+func (s *State) GetExecCache() dynaml.ExecCache {
+	return s.exec_cache
+}
+
 func (s *State) GetTempName(data []byte) (string, error) {
 	if !s.FileAccessAllowed() {
 		return "", fmt.Errorf("tempname: no OS operations supported in this execution environment")