From 04cd510ddccf553afee717d5443d7dfc1f4eb3ae Mon Sep 17 00:00:00 2001 From: Somefive Date: Wed, 31 May 2023 14:01:30 +0800 Subject: [PATCH] Chore: use package function to replace parallel function (#6052) --- pkg/auth/privileges.go | 14 +-- pkg/resourcekeeper/dispatch.go | 7 +- pkg/utils/parallel/parallel.go | 132 ---------------------------- pkg/utils/parallel/parallel_test.go | 101 --------------------- 4 files changed, 10 insertions(+), 244 deletions(-) delete mode 100644 pkg/utils/parallel/parallel.go delete mode 100644 pkg/utils/parallel/parallel_test.go diff --git a/pkg/auth/privileges.go b/pkg/auth/privileges.go index 45c291e1d..7e91c20f0 100644 --- a/pkg/auth/privileges.go +++ b/pkg/auth/privileges.go @@ -25,6 +25,7 @@ import ( "sync" "github.com/gosuri/uitable/util/wordwrap" + velaslices "github.com/kubevela/pkg/util/slices" "github.com/xlab/treeprint" rbacv1 "k8s.io/api/rbac/v1" "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions" @@ -38,7 +39,6 @@ import ( "github.com/oam-dev/kubevela/pkg/multicluster" "github.com/oam-dev/kubevela/pkg/utils" velaerrors "github.com/oam-dev/kubevela/pkg/utils/errors" - "github.com/oam-dev/kubevela/pkg/utils/parallel" ) // PrivilegeInfo describes one privilege in Kubernetes. Either one ClusterRole or @@ -82,15 +82,15 @@ type RoleBindingRef authObjRef // ListPrivileges retrieve privilege information in specified clusters func ListPrivileges(ctx context.Context, cli client.Client, clusters []string, identity *Identity) (map[string][]PrivilegeInfo, error) { var m sync.Map - errs := parallel.Run(func(cluster string) error { + errs := velaslices.ParMap(clusters, func(cluster string) error { info, err := listPrivilegesInCluster(ctx, cli, cluster, identity) if err != nil { return err } m.Store(cluster, info) return nil - }, clusters, parallel.DefaultParallelism) - if err := velaerrors.AggregateErrors(errs.([]error)); err != nil { + }) + if err := velaerrors.AggregateErrors(errs); err != nil { return nil, err } privilegesMap := make(map[string][]PrivilegeInfo) @@ -147,7 +147,7 @@ func listPrivilegesInCluster(ctx context.Context, cli client.Client, cluster str infos = append(infos, PrivilegeInfo{RoleRef: roleRef, RoleBindingRefs: roleBindingRefs}) } var m sync.Map - errs := parallel.Run(func(info PrivilegeInfo) error { + errs := velaslices.ParMap(infos, func(info PrivilegeInfo) error { key := types.NamespacedName{Namespace: info.RoleRef.Namespace, Name: info.RoleRef.Name} var rules []rbacv1.PolicyRule if info.RoleRef.Kind == "Role" { @@ -165,8 +165,8 @@ func listPrivilegesInCluster(ctx context.Context, cli client.Client, cluster str } m.Store(authObjRef(info.RoleRef).FullName(), rules) return nil - }, infos, parallel.DefaultParallelism) - if err := velaerrors.AggregateErrors(errs.([]error)); err != nil { + }) + if err := velaerrors.AggregateErrors(errs); err != nil { return nil, err } for i, info := range infos { diff --git a/pkg/resourcekeeper/dispatch.go b/pkg/resourcekeeper/dispatch.go index 8c51414ea..16e49d05f 100644 --- a/pkg/resourcekeeper/dispatch.go +++ b/pkg/resourcekeeper/dispatch.go @@ -33,7 +33,6 @@ import ( "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 @@ -145,7 +144,7 @@ func (h *resourceKeeper) record(ctx context.Context, manifests []*unstructured.U } func (h *resourceKeeper) dispatch(ctx context.Context, manifests []*unstructured.Unstructured, applyOpts []apply.ApplyOption) error { - errs := parallel.Run(func(manifest *unstructured.Unstructured) error { + errs := velaslices.ParMap(manifests, func(manifest *unstructured.Unstructured) error { applyCtx := multicluster.ContextWithClusterName(ctx, oam.GetCluster(manifest)) applyCtx = auth.ContextWithUserInfo(applyCtx, h.app) ao := applyOpts @@ -166,6 +165,6 @@ func (h *resourceKeeper) dispatch(ctx context.Context, manifests []*unstructured return errors.Wrapf(err, "failed to apply once policy for application %s,%s", h.app.Name, err.Error()) } return h.applicator.Apply(applyCtx, manifest, ao...) - }, manifests, MaxDispatchConcurrent) - return velaerrors.AggregateErrors(errs.([]error)) + }, velaslices.Parallelism(MaxDispatchConcurrent)) + return velaerrors.AggregateErrors(errs) } diff --git a/pkg/utils/parallel/parallel.go b/pkg/utils/parallel/parallel.go deleted file mode 100644 index e3b4d05d7..000000000 --- a/pkg/utils/parallel/parallel.go +++ /dev/null @@ -1,132 +0,0 @@ -/* -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" -) - -const ( - // DefaultParallelism default parallelism - DefaultParallelism int = 5 -) - -// 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 deleted file mode 100644 index 21d4df119..000000000 --- a/pkg/utils/parallel/parallel_test.go +++ /dev/null @@ -1,101 +0,0 @@ -/* -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) -}