kjuulh/dagger
Archived
1
0
Fork 0
Code Issues 1 Pull Requests 1 Projects Releases Wiki Activity

Merge pull request #54 from blocklayerhq/concurrency

Browse Source 
Fix locking around Fill and Lookup
This commit is contained in:
Andrea Luzzardi 2021-01-20 14:30:36 -08:00 committed by GitHub
commit 1d64422320
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 261 additions and 260 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
dagger/compiler.go
View File

@ -18,7 +18,7 @@ import (
// (we call it compiler to avoid confusion with dagger runtime)
// Use this instead of cue.Runtime
type Compiler struct {
sync.Mutex
sync.RWMutex
cue.Runtime
spec *Spec
}
@ -27,20 +27,19 @@ func (cc *Compiler) Cue() *cue.Runtime {
return &(cc.Runtime)
}
func (cc *Compiler) Spec() (*Spec, error) {
func (cc *Compiler) Spec() *Spec {
if cc.spec != nil {
return cc.spec, nil
return cc.spec
}
v, err := cc.Compile("spec.cue", DaggerSpec)
if err != nil {
return nil, err
panic(err)
}
spec, err := v.Spec()
cc.spec, err = v.Spec()
if err != nil {
return nil, err
panic(err)
}
cc.spec = spec
return spec, nil
return cc.spec
}
// Compile an empty struct

12
dagger/env.go
View File

@ -3,7 +3,6 @@ package dagger
import (
"context"
"os"
"sync"
"cuelang.org/go/cue"
cueflow "cuelang.org/go/tools/flow"
@ -144,14 +143,9 @@ func (env *Env) Walk(ctx context.Context, fn EnvWalkFunc) (*Value, error) {
return nil, err
}
l := sync.Mutex{}
// Cueflow config
flowCfg := &cueflow.Config{
UpdateFunc: func(c *cueflow.Controller, t *cueflow.Task) error {
l.Lock()
defer l.Unlock()
if t == nil {
return nil
}
@ -180,9 +174,6 @@ func (env *Env) Walk(ctx context.Context, fn EnvWalkFunc) (*Value, error) {
}
// Cueflow match func
flowMatchFn := func(v cue.Value) (cueflow.Runner, error) {
l.Lock()
defer l.Unlock()
lg := lg.
With().
Str("path", v.Path().String()).
@ -200,9 +191,6 @@ func (env *Env) Walk(ctx context.Context, fn EnvWalkFunc) (*Value, error) {
return nil, err
}
return cueflow.RunnerFunc(func(t *cueflow.Task) error {
l.Lock()
defer l.Unlock()
return fn(ctx, c, t)
}), nil
}

13
dagger/op.go
View File

@ -202,6 +202,12 @@ func (op *Op) Export(ctx context.Context, fs FS, out Fillable) (FS, error) {
}
switch format {
case "string":
log.
Ctx(ctx).
Debug().
Bytes("contents", contents).
Msg("exporting string")
if err := out.Fill(string(contents)); err != nil {
return fs, err
}
@ -210,6 +216,13 @@ func (op *Op) Export(ctx context.Context, fs FS, out Fillable) (FS, error) {
if err := json.Unmarshal(contents, &o); err != nil {
return fs, err
}
log.
Ctx(ctx).
Debug().
Interface("contents", o).
Msg("exporting json")
if err := out.Fill(o); err != nil {
return fs, err
}

23
dagger/spec.go
View File

@ -1,7 +1,6 @@
package dagger
import (
"cuelang.org/go/cue"
cueerrors "cuelang.org/go/cue/errors"
"github.com/pkg/errors"
)
@ -12,28 +11,14 @@ type Spec struct {
}
// eg. Validate(op, "#Op")
func (s Spec) Validate(v *Value, defpath string) (err error) {
// Expand cue errors to get full details
// FIXME: there is probably a cleaner way to do this.
defer func() {
if err != nil {
err = errors.New(cueerrors.Details(err, nil))
}
}()
func (s Spec) Validate(v *Value, defpath string) error {
// Lookup def by name, eg. "#Script" or "#Copy"
// See dagger/spec.cue
def := s.root.Get(defpath)
if err := def.Validate(); err != nil {
return err
}
merged := def.Unwrap().Fill(v.Value)
if err := merged.Err(); err != nil {
return err
}
if err := merged.Validate(cue.Final()); err != nil {
return err
if err := def.Fill(v); err != nil {
return errors.New(cueerrors.Details(err, nil))
}
return nil
}

460
dagger/value.go
View File

@ -9,51 +9,261 @@ import (
"github.com/moby/buildkit/client/llb"
)
// Polyfill for cue.Value.
// Value is a wrapper around cue.Value.
// Use instead of cue.Value and cue.Instance
type Value struct {
// FIXME: don't embed, cleaner API
cue.Value
val cue.Value
cc *Compiler
inst *cue.Instance
}
func (v *Value) Lock() {
if v.cc == nil {
return
func (v *Value) CueInst() *cue.Instance {
return v.inst
}
func (v *Value) Compiler() *Compiler {
return v.cc
}
func (v *Value) Wrap(v2 cue.Value) *Value {
return wrapValue(v2, v.inst, v.cc)
}
func wrapValue(v cue.Value, inst *cue.Instance, cc *Compiler) *Value {
return &Value{
val: v,
cc: cc,
inst: inst,
}
}
// Fill is a concurrency safe wrapper around cue.Value.Fill()
// This is the only method which changes the value in-place.
func (v *Value) Fill(x interface{}) error {
v.cc.Lock()
defer v.cc.Unlock()
// If calling Fill() with a Value, we want to use the underlying
// cue.Value to fill.
if val, ok := x.(*Value); ok {
v.val = v.val.Fill(val.val)
} else {
v.val = v.val.Fill(x)
}
return v.Validate()
}
func (v *Value) Unlock() {
if v.cc == nil {
return
}
v.cc.Unlock()
}
func (v *Value) Lookup(path ...string) *Value {
v.Lock()
defer v.Unlock()
return v.Wrap(v.Unwrap().LookupPath(cueStringsToCuePath(path...)))
}
// LookupPath is a concurrency safe wrapper around cue.Value.LookupPath
func (v *Value) LookupPath(p cue.Path) *Value {
v.Lock()
defer v.Unlock()
return v.Wrap(v.Unwrap().LookupPath(p))
v.cc.RLock()
defer v.cc.RUnlock()
return v.Wrap(v.val.LookupPath(p))
}
// FIXME: deprecated by Get()
func (v *Value) LookupTarget(target string) *Value {
return v.LookupPath(cue.ParsePath(target))
// Lookup is a helper function to lookup by path parts.
func (v *Value) Lookup(path ...string) *Value {
return v.LookupPath(cueStringsToCuePath(path...))
}
// Get is a helper function to lookup by path string
func (v *Value) Get(target string) *Value {
return v.LookupPath(cue.ParsePath(target))
}
// Proxy function to the underlying cue.Value
func (v *Value) Len() cue.Value {
return v.val.Len()
}
// Proxy function to the underlying cue.Value
func (v *Value) List() (cue.Iterator, error) {
return v.val.List()
}
// Proxy function to the underlying cue.Value
func (v *Value) Fields() (*cue.Iterator, error) {
return v.val.Fields()
}
// Proxy function to the underlying cue.Value
func (v *Value) Struct() (*cue.Struct, error) {
return v.val.Struct()
}
// Proxy function to the underlying cue.Value
func (v *Value) Exists() bool {
return v.val.Exists()
}
// Proxy function to the underlying cue.Value
func (v *Value) String() (string, error) {
return v.val.String()
}
// Proxy function to the underlying cue.Value
func (v *Value) Path() cue.Path {
return v.val.Path()
}
// Proxy function to the underlying cue.Value
func (v *Value) Decode(x interface{}) error {
return v.val.Decode(x)
}
func (v *Value) RangeList(fn func(int, *Value) error) error {
it, err := v.List()
if err != nil {
return err
}
i := 0
for it.Next() {
if err := fn(i, v.Wrap(it.Value())); err != nil {
return err
}
i++
}
return nil
}
func (v *Value) RangeStruct(fn func(string, *Value) error) error {
it, err := v.Fields()
if err != nil {
return err
}
for it.Next() {
if err := fn(it.Label(), v.Wrap(it.Value())); err != nil {
return err
}
}
return nil
}
// FIXME: receive string path?
func (v *Value) Merge(x interface{}, path ...string) (*Value, error) {
if xval, ok := x.(*Value); ok {
if xval.Compiler() != v.Compiler() {
return nil, fmt.Errorf("can't merge values from different compilers")
}
x = xval.val
}
v.cc.Lock()
result := v.Wrap(v.val.Fill(x, path...))
v.cc.Unlock()
return result, result.Validate()
}
func (v *Value) MergePath(x interface{}, p cue.Path) (*Value, error) {
// FIXME: array indexes and defs are not supported,
// they will be silently converted to regular fields.
// eg. `foo.#bar[0]` will become `foo["#bar"]["0"]`
return v.Merge(x, cuePathToStrings(p)...)
}
func (v *Value) MergeTarget(x interface{}, target string) (*Value, error) {
return v.MergePath(x, cue.ParsePath(target))
}
// Recursive concreteness check.
// Return false if v is not concrete, or contains any
// non-concrete fields or items.
func (v *Value) IsConcreteR() bool {
// FIXME: use Value.Walk
if it, err := v.Fields(); err == nil {
for it.Next() {
w := v.Wrap(it.Value())
if !w.IsConcreteR() {
return false
}
}
return true
}
if it, err := v.List(); err == nil {
for it.Next() {
w := v.Wrap(it.Value())
if !w.IsConcreteR() {
return false
}
}
return true
}
dv, _ := v.val.Default()
return v.val.IsConcrete() || dv.IsConcrete()
}
// Export concrete values to JSON. ignoring non-concrete values.
// Contrast with cue.Value.MarshalJSON which requires all values
// to be concrete.
func (v *Value) JSON() JSON {
var out JSON
v.val.Walk(
func(v cue.Value) bool {
b, err := v.MarshalJSON()
if err == nil {
newOut, err := out.Set(b, cuePathToStrings(v.Path())...)
if err == nil {
out = newOut
}
return false
}
return true
},
nil,
)
return out
}
func (v *Value) SaveJSON(fs FS, filename string) FS {
return fs.Change(func(st llb.State) llb.State {
return st.File(
llb.Mkfile(filename, 0600, v.JSON()),
)
})
}
func (v *Value) Save(fs FS, filename string) (FS, error) {
src, err := v.Source()
if err != nil {
return fs, err
}
return fs.Change(func(st llb.State) llb.State {
return st.File(
llb.Mkfile(filename, 0600, src),
)
}), nil
}
func (v *Value) Validate(defs ...string) error {
if err := v.val.Validate(); err != nil {
return err
}
if len(defs) == 0 {
return nil
}
spec := v.Compiler().Spec()
for _, def := range defs {
if err := spec.Validate(v, def); err != nil {
return err
}
}
return nil
}
func (v *Value) Source() ([]byte, error) {
return cueformat.Node(v.val.Eval().Syntax())
}
func (v *Value) IsEmptyStruct() bool {
if st, err := v.Struct(); err == nil {
if st.Len() == 0 {
return true
}
}
return false
}
// Component returns the component value if v is a valid dagger component or an error otherwise.
// If no '#dagger' annotation is present, os.ErrNotExist
// is returned.
@ -111,11 +321,8 @@ func (v *Value) ScriptOrComponent() (interface{}, error) {
func (v *Value) Op() (*Op, error) {
// Merge #Op definition from spec to get default values
spec, err := v.Compiler().Spec()
if err != nil {
return nil, err
}
v, err = spec.Get("#Op").Merge(v)
spec := v.Compiler().Spec()
v, err := spec.Get("#Op").Merge(v)
if err != nil {
return nil, err
}
@ -143,194 +350,3 @@ func (v *Value) Spec() (*Spec, error) {
root: v,
}, nil
}
// FIXME: receive string path?
func (v *Value) Merge(x interface{}, path ...string) (*Value, error) {
if xval, ok := x.(*Value); ok {
if xval.Compiler() != v.Compiler() {
return nil, fmt.Errorf("can't merge values from different compilers")
}
x = xval.Unwrap()
}
result := v.Wrap(v.Unwrap().Fill(x, path...))
return result, result.Validate()
}
func (v *Value) MergePath(x interface{}, p cue.Path) (*Value, error) {
// FIXME: array indexes and defs are not supported,
// they will be silently converted to regular fields.
// eg. `foo.#bar[0]` will become `foo["#bar"]["0"]`
return v.Merge(x, cuePathToStrings(p)...)
}
func (v *Value) MergeTarget(x interface{}, target string) (*Value, error) {
return v.MergePath(x, cue.ParsePath(target))
}
func (v *Value) RangeList(fn func(int, *Value) error) error {
it, err := v.List()
if err != nil {
return err
}
i := 0
for it.Next() {
if err := fn(i, v.Wrap(it.Value())); err != nil {
return err
}
i++
}
return nil
}
func (v *Value) RangeStruct(fn func(string, *Value) error) error {
it, err := v.Fields()
if err != nil {
return err
}
for it.Next() {
if err := fn(it.Label(), v.Wrap(it.Value())); err != nil {
return err
}
}
return nil
}
// Recursive concreteness check.
// Return false if v is not concrete, or contains any
// non-concrete fields or items.
func (v *Value) IsConcreteR() bool {
// FIXME: use Value.Walk
if it, err := v.Fields(); err == nil {
for it.Next() {
w := v.Wrap(it.Value())
if !w.IsConcreteR() {
return false
}
}
return true
}
if it, err := v.List(); err == nil {
for it.Next() {
w := v.Wrap(it.Value())
if !w.IsConcreteR() {
return false
}
}
return true
}
dv, _ := v.Default()
return v.IsConcrete() || dv.IsConcrete()
}
// Export concrete values to JSON. ignoring non-concrete values.
// Contrast with cue.Value.MarshalJSON which requires all values
// to be concrete.
func (v *Value) JSON() JSON {
v.Lock()
defer v.Unlock()
var out JSON
v.Walk(
func(v cue.Value) bool {
b, err := v.MarshalJSON()
if err == nil {
newOut, err := out.Set(b, cuePathToStrings(v.Path())...)
if err == nil {
out = newOut
}
return false
}
return true
},
nil,
)
return out
}
func (v *Value) SaveJSON(fs FS, filename string) FS {
return fs.Change(func(st llb.State) llb.State {
return st.File(
llb.Mkfile(filename, 0600, v.JSON()),
)
})
}
func (v *Value) Save(fs FS, filename string) (FS, error) {
src, err := v.Source()
if err != nil {
return fs, err
}
return fs.Change(func(st llb.State) llb.State {
return st.File(
llb.Mkfile(filename, 0600, src),
)
}), nil
}
func (v *Value) Validate(defs ...string) error {
if err := v.Unwrap().Validate(); err != nil {
return err
}
if len(defs) == 0 {
return nil
}
spec, err := v.Compiler().Spec()
if err != nil {
return err
}
for _, def := range defs {
if err := spec.Validate(v, def); err != nil {
return err
}
}
return nil
}
// Value implements Fillable.
// This is the only method which changes the value in-place.
// FIXME this co-exists awkwardly with the rest of Value.
func (v *Value) Fill(x interface{}) error {
v.Value = v.Value.Fill(x)
return v.Validate()
}
func (v *Value) Source() ([]byte, error) {
v.Lock()
defer v.Unlock()
return cueformat.Node(v.Eval().Syntax())
}
func (v *Value) IsEmptyStruct() bool {
if st, err := v.Struct(); err == nil {
if st.Len() == 0 {
return true
}
}
return false
}
func (v *Value) CueInst() *cue.Instance {
return v.inst
}
func (v *Value) Compiler() *Compiler {
// if v.cc == nil {
// return &Compiler{}
// }
return v.cc
}
func (v *Value) Wrap(v2 cue.Value) *Value {
return wrapValue(v2, v.inst, v.cc)
}
func (v *Value) Unwrap() cue.Value {
return v.Value
}
func wrapValue(v cue.Value, inst *cue.Instance, cc *Compiler) *Value {
return &Value{
Value: v,
cc: cc,
inst: inst,
}
}

2
examples/tests/fetch-container/exist/main.cue
View File

@ -31,7 +31,7 @@ busybox4: {
#dagger: compute: [
{
do: "fetch-container"
ref: "busyboxa@sha256:e2af53705b841ace3ab3a44998663d4251d33ee8a9acaf71b66df4ae01c3bbe7"
ref: "busybox@sha256:e2af53705b841ace3ab3a44998663d4251d33ee8a9acaf71b66df4ae01c3bbe7"
},
]
}