Fix: rework parallel execution for ApplyComponents and ResourceKeeper Dispatch (#3407)

* Fix: applyComponents lock and rework parallel execution

Signed-off-by: Somefive <yd219913@alibaba-inc.com>

* Fix: rename

Signed-off-by: Somefive <yd219913@alibaba-inc.com>

pkg/resourcekeeper/dispatch.go

@@ -18,16 +18,16 @@ package resourcekeeper
 
 import (
 	"context"
-	"sync"
 
 	"github.com/pkg/errors"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
-	kerrors "k8s.io/apimachinery/pkg/util/errors"
 
 	"github.com/oam-dev/kubevela/pkg/multicluster"
 	"github.com/oam-dev/kubevela/pkg/oam"
 	"github.com/oam-dev/kubevela/pkg/resourcetracker"
 	"github.com/oam-dev/kubevela/pkg/utils/apply"
+	velaerrors "github.com/oam-dev/kubevela/pkg/utils/errors"
+	"github.com/oam-dev/kubevela/pkg/utils/parallel"
 )
 
 // MaxDispatchConcurrent is the max dispatch concurrent number
@@ -118,29 +118,10 @@ func (h *resourceKeeper) record(ctx context.Context, manifests []*unstructured.U
 }
 
 func (h *resourceKeeper) dispatch(ctx context.Context, manifests []*unstructured.Unstructured) error {
-	var errs []error
-	var l sync.Mutex
-	var wg sync.WaitGroup
-
-	ch := make(chan struct{}, MaxDispatchConcurrent)
 	applyOpts := []apply.ApplyOption{apply.MustBeControlledByApp(h.app), apply.NotUpdateRenderHashEqual()}
-
-	for i := 0; i < len(manifests); i++ {
-		ch <- struct{}{}
-		wg.Add(1)
-		go func(index int) {
-			defer wg.Done()
-			manifest := manifests[index]
-			applyCtx := multicluster.ContextWithClusterName(ctx, oam.GetCluster(manifest))
-			err := h.applicator.Apply(applyCtx, manifest, applyOpts...)
-			if err != nil {
-				l.Lock()
-				errs = append(errs, err)
-				l.Unlock()
-			}
-			<-ch
-		}(i)
-	}
-	wg.Wait()
-	return kerrors.NewAggregate(errs)
+	errs := parallel.Run(func(manifest *unstructured.Unstructured) error {
+		applyCtx := multicluster.ContextWithClusterName(ctx, oam.GetCluster(manifest))
+		return h.applicator.Apply(applyCtx, manifest, applyOpts...)
+	}, manifests, MaxDispatchConcurrent)
+	return velaerrors.AggregateErrors(errs.([]error))
 }

pkg/utils/errors/list.go

@@ -44,3 +44,17 @@ func (e ErrorList) HasError() bool {
 	}
 	return len(e) > 0
 }
+
+// AggregateErrors aggregate errors into ErrorList and filter nil, if no error found, return nil
+func AggregateErrors(errs []error) error {
+	var es ErrorList
+	for _, err := range errs {
+		if err != nil {
+			es = append(es, err)
+		}
+	}
+	if es.HasError() {
+		return es
+	}
+	return nil
+}

pkg/utils/parallel/parallel.go

@@ -0,0 +1,127 @@
+/*
+Copyright 2021 The KubeVela Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package parallel
+
+import (
+	"reflect"
+)
+
+// ParInput input for parallel execution
+type ParInput interface{}
+
+// ParOutput output for parallel execution
+type ParOutput interface{}
+
+type orderedParOutput struct {
+	ParOutput
+	index int
+}
+
+// RunBatch parallel execute handler function on parInputs, with maximum concurrency as parallelism
+func RunBatch(handler func(ParInput) ParOutput, parInputs []ParInput, parallelism int) []ParOutput {
+	outs := make(chan orderedParOutput)
+	pool := make(chan struct{}, parallelism)
+	for _idx, _input := range parInputs {
+		go func(idx int, input ParInput) {
+			pool <- struct{}{}
+			output := handler(input)
+			outs <- orderedParOutput{ParOutput: output, index: idx}
+			<-pool
+		}(_idx, _input)
+	}
+	outputs := make([]ParOutput, len(parInputs))
+	for range parInputs {
+		out := <-outs
+		outputs[out.index] = out.ParOutput
+	}
+	close(outs)
+	close(pool)
+	return outputs
+}
+
+// Run execute handler on parInputs, with automatic type conversion and maximum concurrency as parallelism
+// Examples:
+// > out := Run(func(x int) int { return x*x }, []int{1,2,3,4,5}, 5)
+// < out: []int{1,4,19,16,25}
+// > out := Run(func(x int, y string) (string, bool) { return y, x%2==0 }, [][]interface{}{{1,"n"},{2,"y"}}, 2)
+// < out: [][]interface{{"n",false},{"y",true}}
+func Run(handler interface{}, parInputs interface{}, parallelism int) interface{} {
+	parInputsVal := reflect.ValueOf(parInputs)
+	elemLen := parInputsVal.Len()
+	_parInputVal := reflect.MakeSlice(reflect.TypeOf([]ParInput{}), elemLen, elemLen)
+	for i := 0; i < elemLen; i++ {
+		v := parInputsVal.Index(i)
+		if v.IsValid() {
+			_parInputVal.Index(i).Set(v)
+		}
+	}
+
+	handleFunc := reflect.ValueOf(handler)
+	handleFuncTyp := reflect.TypeOf(handler)
+	nParams, nReturns := handleFuncTyp.NumIn(), handleFuncTyp.NumOut()
+	parOutputTyp := reflect.TypeOf([]ParOutput{}).Elem()
+	_handler := reflect.MakeFunc(reflect.TypeOf(func(ParInput) ParOutput { return nil }), func(args []reflect.Value) (results []reflect.Value) {
+		in := make([]reflect.Value, nParams)
+		_inputVal := args[0].Elem()
+		if nParams > 1 {
+			for i := 0; i < nParams; i++ {
+				in[i] = _inputVal.Index(i).Elem()
+			}
+		} else if nParams == 1 {
+			in[0] = _inputVal
+		}
+		for i := 0; i < nParams; i++ {
+			if !in[i].IsValid() {
+				in[i] = reflect.New(handleFuncTyp.In(i)).Elem()
+			}
+		}
+		out := handleFunc.Call(in)
+
+		_outputVal := reflect.New(parOutputTyp).Elem()
+		if nReturns > 1 {
+			ret := reflect.MakeSlice(reflect.TypeOf([]interface{}{}), nReturns, nReturns)
+			for i := 0; i < nReturns; i++ {
+				if out[i].IsValid() {
+					ret.Index(i).Set(out[i])
+				}
+			}
+			_outputVal.Set(ret)
+		} else if nReturns == 1 {
+			if out[0].IsValid() {
+				_outputVal.Set(out[0])
+			}
+		}
+		return []reflect.Value{_outputVal}
+	})
+	outs := RunBatch(_handler.Interface().(func(ParInput) ParOutput), _parInputVal.Interface().([]ParInput), parallelism)
+	if nReturns == 0 {
+		return nil
+	}
+	var outputs reflect.Value
+	if nReturns == 1 {
+		outputs = reflect.MakeSlice(reflect.SliceOf(handleFuncTyp.Out(0)), elemLen, elemLen)
+	} else {
+		outputs = reflect.MakeSlice(reflect.TypeOf([]interface{}{}), elemLen, elemLen)
+	}
+	for i := 0; i < elemLen; i++ {
+		v := reflect.ValueOf(outs[i])
+		if v.IsValid() {
+			outputs.Index(i).Set(v)
+		}
+	}
+	return outputs.Interface()
+}

pkg/utils/parallel/parallel_test.go

@@ -0,0 +1,101 @@
+/*
+Copyright 2021 The KubeVela Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package parallel
+
+import (
+	"math"
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/utils/pointer"
+)
+
+func TestParExecNto1(t *testing.T) {
+	var inputs [][]interface{}
+	size := 100
+	parallelism := 20
+	for i := 0; i < size; i++ {
+		if i%2 == 0 {
+			inputs = append(inputs, []interface{}{i, i + 1, math.Sqrt(float64(i)), nil})
+		} else {
+			inputs = append(inputs, []interface{}{i, i + 1, math.Sqrt(float64(i)), pointer.Int(i)})
+		}
+	}
+	outs := Run(func(a int, b int, c float64, d *int) *float64 {
+		time.Sleep(time.Duration(rand.Intn(200)+25) * time.Millisecond)
+		if d == nil {
+			return nil
+		}
+		return pointer.Float64(float64(a*b) + c + float64(*d))
+	}, inputs, parallelism)
+	outputs, ok := outs.([]*float64)
+	r := require.New(t)
+	r.True(ok)
+	r.Equal(size, len(outputs))
+	for i, j := range outputs {
+		if i%2 == 0 {
+			r.Nil(j)
+		} else {
+			r.NotNil(j)
+			r.Equal(float64(i*(i+1))+math.Sqrt(float64(i))+float64(i), *j)
+		}
+	}
+}
+
+func TestParExec0toN(t *testing.T) {
+	var inputs [][]interface{}
+	size := 100
+	parallelism := 20
+	for i := 0; i < size; i++ {
+		inputs = append(inputs, nil)
+	}
+	outs := Run(func() (bool, string) {
+		time.Sleep(time.Duration(rand.Intn(200)+25) * time.Millisecond)
+		return false, "ok"
+	}, inputs, parallelism)
+	outputs, ok := outs.([]interface{})
+	r := require.New(t)
+	r.True(ok)
+	r.Equal(size, len(outputs))
+	for _, _j := range outputs {
+		j, ok := _j.([]interface{})
+		r.True(ok)
+		r.Equal(2, len(j))
+		j0, ok := j[0].(bool)
+		r.True(ok)
+		r.False(j0)
+		j1, ok := j[1].(string)
+		r.True(ok)
+		r.Equal("ok", j1)
+	}
+}
+
+func TestParExec1to0(t *testing.T) {
+	var inputs []struct{ key int }
+	size := 100
+	parallelism := 20
+	for i := 0; i < size; i++ {
+		inputs = append(inputs, struct{ key int }{key: i})
+	}
+	outs := Run(func(obj struct{ key int }) {
+		time.Sleep(time.Duration(rand.Intn(50)+size-obj.key) * time.Millisecond)
+	}, inputs, parallelism)
+	r := require.New(t)
+	r.Nil(outs)
+}

pkg/workflow/providers/oam/apply.go

@@ -33,7 +33,8 @@ import (
 	"github.com/oam-dev/kubevela/pkg/cue/model/sets"
 	"github.com/oam-dev/kubevela/pkg/cue/model/value"
 	"github.com/oam-dev/kubevela/pkg/oam"
-	errors2 "github.com/oam-dev/kubevela/pkg/utils/errors"
+	velaerrors "github.com/oam-dev/kubevela/pkg/utils/errors"
+	"github.com/oam-dev/kubevela/pkg/utils/parallel"
 	wfContext "github.com/oam-dev/kubevela/pkg/workflow/context"
 	"github.com/oam-dev/kubevela/pkg/workflow/providers"
 	wfTypes "github.com/oam-dev/kubevela/pkg/workflow/types"
@@ -60,7 +61,7 @@ type provider struct {
 
 // RenderComponent render component
 func (p *provider) RenderComponent(ctx wfContext.Context, v *value.Value, act wfTypes.Action) error {
-	comp, patcher, clusterName, overrideNamespace, env, err := lookUpValues(v)
+	comp, patcher, clusterName, overrideNamespace, env, err := lookUpValues(v, nil)
 	if err != nil {
 		return err
 	}
@@ -88,7 +89,7 @@ func (p *provider) RenderComponent(ctx wfContext.Context, v *value.Value, act wf
 }
 
 func (p *provider) applyComponent(_ wfContext.Context, v *value.Value, act wfTypes.Action, mu *sync.Mutex) error {
-	comp, patcher, clusterName, overrideNamespace, env, err := lookUpValues(v)
+	comp, patcher, clusterName, overrideNamespace, env, err := lookUpValues(v, mu)
 	if err != nil {
 		return err
 	}
@@ -147,37 +148,31 @@ func (p *provider) ApplyComponents(ctx wfContext.Context, v *value.Value, act wf
 	if parallelism <= 0 {
 		return errors.Errorf("parallelism cannot be smaller than 1")
 	}
+	// prepare parallel execution args
 	mu := &sync.Mutex{}
-	var wg sync.WaitGroup
-	ch := make(chan struct{}, parallelism)
-	var errs errors2.ErrorList
-	err = components.StepByFields(func(name string, in *value.Value) (bool, error) {
-		ch <- struct{}{}
-		wg.Add(1)
-		go func(_name string, _in *value.Value) {
-			defer func() {
-				wg.Done()
-				<-ch
-			}()
-			if err := p.applyComponent(ctx, _in, act, mu); err != nil {
-				mu.Lock()
-				errs = append(errs, errors.Wrapf(err, "failed to apply component %s", _name))
-				mu.Unlock()
-			}
-		}(name, in)
+	var parInputs [][]interface{}
+	if err = components.StepByFields(func(name string, in *value.Value) (bool, error) {
+		parInputs = append(parInputs, []interface{}{name, ctx, in, act, mu})
 		return false, nil
-	})
-	wg.Wait()
-	if err != nil {
+	}); err != nil {
 		return errors.Wrapf(err, "failed to looping over components")
 	}
-	if errs.HasError() {
-		return errs
-	}
-	return nil
+	// parallel execution
+	outputs := parallel.Run(func(name string, ctx wfContext.Context, v *value.Value, act wfTypes.Action, mu *sync.Mutex) error {
+		if err := p.applyComponent(ctx, v, act, mu); err != nil {
+			return errors.Wrapf(err, "failed to apply component %s", name)
+		}
+		return nil
+	}, parInputs, int(parallelism))
+	// aggregate errors
+	return velaerrors.AggregateErrors(outputs.([]error))
 }
 
-func lookUpValues(v *value.Value) (*common.ApplicationComponent, *value.Value, string, string, string, error) {
+func lookUpValues(v *value.Value, mu *sync.Mutex) (*common.ApplicationComponent, *value.Value, string, string, string, error) {
+	if mu != nil {
+		mu.Lock()
+		defer mu.Unlock()
+	}
 	compSettings, err := v.LookupValue("value")
 	if err != nil {
 		return nil, nil, "", "", "", err

pkg/workflow/providers/oam/apply_test.go

@@ -17,12 +17,15 @@ limitations under the License.
 package oam
 
 import (
+	"fmt"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/require"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/rand"
 
 	"github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1"
 	"github.com/oam-dev/kubevela/pkg/appfile"
@@ -182,6 +185,21 @@ func TestApplyComponents(t *testing.T) {
 	}
 }
 
+func TestApplyComponentsHard(t *testing.T) {
+	r := require.New(t)
+	input := `comp0:{value:{name:"comp0"}}`
+	for i := 1; i < 1000; i++ {
+		input += fmt.Sprintf(`,comp%d:{value:{name:"comp%d"}}`, i, i)
+	}
+	input = fmt.Sprintf(`{components:{%s},parallelism:50}`, input)
+	p := &provider{apply: delayedComponentApplyForTest}
+	act := &mock.Action{}
+	v, err := value.NewValue("", nil, "")
+	r.NoError(err)
+	r.NoError(v.FillRaw(input))
+	r.NoError(p.ApplyComponents(nil, v, act))
+}
+
 func TestLoadComponent(t *testing.T) {
 	r := require.New(t)
 	p := &provider{
@@ -369,3 +387,8 @@ func simpleComponentApplyForTest(comp common.ApplicationComponent, _ *value.Valu
 	traits := []*unstructured.Unstructured{trait}
 	return workload, traits, testHealthy, nil
 }
+
+func delayedComponentApplyForTest(comp common.ApplicationComponent, v *value.Value, x string, y string, z string) (*unstructured.Unstructured, []*unstructured.Unstructured, bool, error) {
+	time.Sleep(time.Duration(rand.Intn(200)+25) * time.Millisecond)
+	return simpleComponentApplyForTest(comp, v, x, y, z)
+}