From d5e8b68ad2cafd0bcf578d419440a44beff32fd2 Mon Sep 17 00:00:00 2001 From: Somefive Date: Thu, 10 Mar 2022 16:43:22 +0800 Subject: [PATCH] Fix: rework parallel execution for ApplyComponents and ResourceKeeper Dispatch (#3407) * Fix: applyComponents lock and rework parallel execution Signed-off-by: Somefive * Fix: rename Signed-off-by: Somefive --- pkg/resourcekeeper/dispatch.go | 33 ++---- pkg/utils/errors/list.go | 14 +++ pkg/utils/parallel/parallel.go | 127 +++++++++++++++++++++++ pkg/utils/parallel/parallel_test.go | 101 ++++++++++++++++++ pkg/workflow/providers/oam/apply.go | 51 ++++----- pkg/workflow/providers/oam/apply_test.go | 23 ++++ 6 files changed, 295 insertions(+), 54 deletions(-) create mode 100644 pkg/utils/parallel/parallel.go create mode 100644 pkg/utils/parallel/parallel_test.go diff --git a/pkg/resourcekeeper/dispatch.go b/pkg/resourcekeeper/dispatch.go index 688b712d8..843dabded 100644 --- a/pkg/resourcekeeper/dispatch.go +++ b/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)) } diff --git a/pkg/utils/errors/list.go b/pkg/utils/errors/list.go index b6598092e..100bca99a 100644 --- a/pkg/utils/errors/list.go +++ b/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 +} diff --git a/pkg/utils/parallel/parallel.go b/pkg/utils/parallel/parallel.go new file mode 100644 index 000000000..001110aaf --- /dev/null +++ b/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() +} diff --git a/pkg/utils/parallel/parallel_test.go b/pkg/utils/parallel/parallel_test.go new file mode 100644 index 000000000..21d4df119 --- /dev/null +++ b/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) +} diff --git a/pkg/workflow/providers/oam/apply.go b/pkg/workflow/providers/oam/apply.go index 1ea54bc12..de5e3248e 100644 --- a/pkg/workflow/providers/oam/apply.go +++ b/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 diff --git a/pkg/workflow/providers/oam/apply_test.go b/pkg/workflow/providers/oam/apply_test.go index 4d4310d93..73f893a92 100644 --- a/pkg/workflow/providers/oam/apply_test.go +++ b/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) +}