From ad6494e8662416f8d19b546781251da372a8d73b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Toma=C5=BE=20Jerman?= Date: Wed, 31 Jan 2024 15:14:56 +0100 Subject: [PATCH] Improve TypedValue clone to go parallel after certain threshold --- server/pkg/expr/expr_types.go | 102 +++++++++++++++++++++++++++++ server/pkg/expr/expr_types_test.go | 83 +++++++++++++++++++++++ 2 files changed, 185 insertions(+) diff --git a/server/pkg/expr/expr_types.go b/server/pkg/expr/expr_types.go index f64079b3e..7f4b8251b 100644 --- a/server/pkg/expr/expr_types.go +++ b/server/pkg/expr/expr_types.go @@ -7,10 +7,12 @@ import ( "fmt" "io" "io/ioutil" + "math" "net/http" "net/url" "reflect" "strings" + "sync" "time" "github.com/PaesslerAG/gval" @@ -39,6 +41,10 @@ type ( } ) +const ( + cloneParallelItemThreshold = 200 +) + func KvFunctions() []gval.Language { return []gval.Language{ gval.Function("set", set), @@ -53,6 +59,14 @@ func (v *Vars) Clone() (out TypedValue, err error) { return EmptyVars(), nil } + if len(v.value) > cloneParallelItemThreshold { + return v.cloneParallel() + } + + return v.clone() +} + +func (v *Vars) clone() (out TypedValue, err error) { aux := &Vars{ value: make(map[string]TypedValue, len(v.value)), } @@ -70,6 +84,51 @@ func (v *Vars) Clone() (out TypedValue, err error) { return aux, nil } +func (v *Vars) cloneParallel() (out TypedValue, err error) { + keys := make([]string, 0, len(v.value)) + for k := range v.value { + keys = append(keys, k) + } + + auxValues := make([]TypedValue, len(v.value)) + errors := make([]error, len(v.value)) + + wg := sync.WaitGroup{} + for i := 0; i < len(keys); i += cloneParallelItemThreshold { + wg.Add(1) + go func(i int) { + defer wg.Done() + + for j, k := range keys[i:int(math.Min(float64(i+cloneParallelItemThreshold), float64(len(keys))-1))] { + aux, err := v.value[k].Clone() + if err != nil { + errors[i+j] = err + } + + auxValues[i+j] = aux + } + }(i) + } + + wg.Wait() + + for _, err := range errors { + if err != nil { + return nil, err + } + } + + aux := &Vars{ + value: make(map[string]TypedValue, len(v.value)), + } + + for i, k := range keys { + aux.value[k] = auxValues[i] + } + + return aux, nil +} + func EmptyKV() *KV { return &KV{value: make(map[string]string)} } @@ -901,6 +960,49 @@ func (v *Any) Clone() (out TypedValue, err error) { } func (v *Array) Clone() (out TypedValue, err error) { + if len(v.value) > cloneParallelItemThreshold { + return v.cloneParallel() + } + + return v.clone() +} + +func (v *Array) clone() (_ TypedValue, err error) { + errors := make([]error, len(v.value)) + wg := sync.WaitGroup{} + + out := &Array{ + value: make([]TypedValue, len(v.value)), + } + + for i := 0; i < len(v.value); i += cloneParallelItemThreshold { + wg.Add(1) + go func(i int) { + defer wg.Done() + + for j, v := range v.value[i:int(math.Min(float64(i+cloneParallelItemThreshold), float64(len(v.value))-1))] { + aux, err := v.Clone() + if err != nil { + errors[i+j] = err + } + + out.value[i+j] = aux + } + }(i) + } + + wg.Wait() + + for _, err := range errors { + if err != nil { + return nil, err + } + } + + return out, nil +} + +func (v *Array) cloneParallel() (out TypedValue, err error) { aux := &Array{ value: make([]TypedValue, len(v.value)), } diff --git a/server/pkg/expr/expr_types_test.go b/server/pkg/expr/expr_types_test.go index f42ae7de7..c74bb94c6 100644 --- a/server/pkg/expr/expr_types_test.go +++ b/server/pkg/expr/expr_types_test.go @@ -550,3 +550,86 @@ func TestArrayDecode(t *testing.T) { req.Len(foo.Strings, 2) req.Len(foo.Values, 2) } + +// Original +// goos: darwin +// goarch: arm64 +// pkg: github.com/cortezaproject/corteza/server/pkg/expr +// BenchmarkVarsUnref_10_10-12 2623 459222 ns/op 370399 B/op 4886 allocs/op +// BenchmarkVarsUnref_100_100-12 25 46813440 ns/op 42108291 B/op 453755 allocs/op +// BenchmarkVarsUnref_1000_1000-12 1 4572438792 ns/op 4004143648 B/op 45064443 allocs/op + +// Reworked, all seq. +// goos: darwin +// goarch: arm64 +// pkg: github.com/cortezaproject/corteza/server/pkg/expr +// BenchmarkVarsUnref_10_10-12 37764 31204 ns/op 90607 B/op 859 allocs/op +// BenchmarkVarsUnref_100_100-12 303 3630116 ns/op 9128270 B/op 80430 allocs/op +// BenchmarkVarsUnref_1000_1000-12 3 372980819 ns/op 898107160 B/op 8004031 allocs/op + +// Reworked, seq threshold to parallel +// goos: darwin +// goarch: arm64 +// pkg: github.com/cortezaproject/corteza/server/pkg/expr +// BenchmarkVarsUnref_10_10-12 35852 33396 ns/op 90606 B/op 859 allocs/op +// BenchmarkVarsUnref_100_100-12 320 3692905 ns/op 9128270 B/op 80430 allocs/op +// BenchmarkVarsUnref_1000_1000-12 7 175036667 ns/op 946114581 B/op 8003063 allocs/op + +func benchmarkVarsUnref(b *testing.B, rootCount, nestedCount int) { + nest := &Vars{ + value: map[string]TypedValue{}, + } + + for i := 0; i < nestedCount; i++ { + nest.value[fmt.Sprintf("nest_value_%d", i)] = &Vars{ + value: map[string]TypedValue{ + "nested": &Vars{ + value: map[string]TypedValue{ + "nested": &Vars{ + value: map[string]TypedValue{}, + }, + }, + }, + }, + } + } + + v := &Vars{ + value: map[string]TypedValue{ + "a11": Must(NewVars(map[string]any{})), + "a22": Must(NewVars(map[string]any{})), + "a33": Must(NewVars(map[string]any{})), + "a44": Must(NewVars(map[string]any{})), + "a55": Must(NewVars(map[string]any{})), + "a66": Must(NewVars(map[string]any{})), + "a77": Must(NewVars(map[string]any{})), + "a88": Must(NewVars(map[string]any{})), + "a99": Must(NewVars(map[string]any{})), + "a1010": Must(NewVars(map[string]any{})), + "a1111": Must(NewVars(map[string]any{})), + "a1212": Must(NewVars(map[string]any{})), + "a1313": Must(NewVars(map[string]any{})), + }, + } + + for i := 0; i < rootCount; i++ { + v.value[fmt.Sprintf("nest_value_%d", i)] = nest + } + + b.ResetTimer() + for i := 0; i < b.N; i++ { + v.Clone() + } +} + +func BenchmarkVarsUnref_10_10(b *testing.B) { + benchmarkVarsUnref(b, 10, 10) +} + +func BenchmarkVarsUnref_100_100(b *testing.B) { + benchmarkVarsUnref(b, 100, 100) +} + +func BenchmarkVarsUnref_1000_1000(b *testing.B) { + benchmarkVarsUnref(b, 1000, 1000) +}