DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Improve TypedValue clone to go parallel after certain threshold

Browse Source
This commit is contained in:
Tomaž Jerman
2024-01-31 15:14:56 +01:00
parent 52ac5698c8
commit ad6494e866
2 changed files with 185 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/pkg/expr/expr_types.go
+102
View File
@@ -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)),
}
server/pkg/expr/expr_types_test.go
+83
View File
@@ -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)
}