mirror of
https://github.com/kubevela/kubevela.git
synced 2026-08-18 03:56:36 +00:00
Chore: use package function to replace parallel function (#6052)
This commit is contained in:
Somefive
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
Reference in New Issue
Block a user