From 2209dd4c0ce67ebc9146829675eee57c6937e42f Mon Sep 17 00:00:00 2001 From: Tony Holdstock-Brown Date: Sun, 7 Jan 2024 21:13:14 -0800 Subject: [PATCH] Make tree lookups concurrent --- expr.go | 79 ++++++---- expr_test.go | 2 +- go.mod | 3 +- go.sum | 2 + vendor/golang.org/x/sync/LICENSE | 27 ++++ vendor/golang.org/x/sync/PATENTS | 22 +++ vendor/golang.org/x/sync/errgroup/errgroup.go | 135 ++++++++++++++++++ vendor/golang.org/x/sync/errgroup/go120.go | 13 ++ .../golang.org/x/sync/errgroup/pre_go120.go | 14 ++ vendor/modules.txt | 3 + 10 files changed, 270 insertions(+), 30 deletions(-) create mode 100644 vendor/golang.org/x/sync/LICENSE create mode 100644 vendor/golang.org/x/sync/PATENTS create mode 100644 vendor/golang.org/x/sync/errgroup/errgroup.go create mode 100644 vendor/golang.org/x/sync/errgroup/go120.go create mode 100644 vendor/golang.org/x/sync/errgroup/pre_go120.go diff --git a/expr.go b/expr.go index 69e8b62..1f1b938 100644 --- a/expr.go +++ b/expr.go @@ -9,6 +9,7 @@ import ( "github.com/google/cel-go/common/operators" "github.com/ohler55/ojg/jp" + "golang.org/x/sync/errgroup" ) var ( @@ -201,8 +202,16 @@ func (a *aggregator) AggregateMatch(ctx context.Context, data map[string]any) ([ counts := map[groupID]int{} // Store all expression parts per group ID for returning. found := map[groupID][]ExpressionPart{} + // protect the above locks with a map. + lock := &sync.Mutex{} + // run lookups concurrently. + eg := errgroup.Group{} add := func(all []ExpressionPart) { + // This is called concurrently, so don't mess with maps in goroutines + lock.Lock() + defer lock.Unlock() + for _, eval := range all { counts[eval.GroupID] += 1 if _, ok := found[eval.GroupID]; !ok { @@ -218,40 +227,54 @@ func (a *aggregator) AggregateMatch(ctx context.Context, data map[string]any) ([ // // TODO: we should iterate through the expression in a top-down order, ensuring that if // any of the top groups fail to match we quit early. - for path, tree := range a.artIdents { - x, err := jp.ParseString(path) - if err != nil { - return nil, err - } - res := x.Get(data) - if len(res) != 1 { - continue - } + for n, item := range a.artIdents { + tree := item + path := n + eg.Go(func() error { + x, err := jp.ParseString(path) + if err != nil { + return err + } + res := x.Get(data) + if len(res) != 1 { + return nil + } - cast, ok := res[0].(string) - if !ok { - // This isn't a string, so we can't compare within the radix tree. - continue - } + cast, ok := res[0].(string) + if !ok { + // This isn't a string, so we can't compare within the radix tree. + return nil + } - add(tree.Search(ctx, path, cast)) + add(tree.Search(ctx, path, cast)) + return nil + }) } // Match on nulls. - for path, tree := range a.nullLookups { - x, err := jp.ParseString(path) - if err != nil { - return nil, err - } + for n, item := range a.nullLookups { + tree := item + path := n + eg.Go(func() error { + x, err := jp.ParseString(path) + if err != nil { + return err + } - res := x.Get(data) - if len(res) == 0 { - // This isn't present, which matches null in our overloads. Set the - // value to nil. - res = []any{nil} - } - // This matches null, nil (as null), and any non-null items. - add(tree.Search(ctx, path, res[0])) + res := x.Get(data) + if len(res) == 0 { + // This isn't present, which matches null in our overloads. Set the + // value to nil. + res = []any{nil} + } + // This matches null, nil (as null), and any non-null items. + add(tree.Search(ctx, path, res[0])) + return nil + }) + } + + if err := eg.Wait(); err != nil { + return nil, err } for k, count := range counts { diff --git a/expr_test.go b/expr_test.go index 2981f73..903c255 100644 --- a/expr_test.go +++ b/expr_test.go @@ -481,7 +481,7 @@ func TestEmptyExpressions(t *testing.T) { }) } -func TestNull(t *testing.T) { +func TestEvaluate_Null(t *testing.T) { ctx := context.Background() parser, err := newParser() require.NoError(t, err) diff --git a/go.mod b/go.mod index 33be73f..93632d9 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,8 @@ require ( github.com/ohler55/ojg v1.21.0 github.com/plar/go-adaptive-radix-tree v1.0.5 github.com/stretchr/testify v1.8.4 + golang.org/x/sync v0.6.0 + google.golang.org/protobuf v1.31.0 ) require ( @@ -19,6 +21,5 @@ require ( golang.org/x/text v0.9.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20230803162519-f966b187b2e5 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20230803162519-f966b187b2e5 // indirect - google.golang.org/protobuf v1.31.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index f8dc05c..8533854 100644 --- a/go.sum +++ b/go.sum @@ -34,6 +34,8 @@ github.com/wsxiaoys/terminal v0.0.0-20160513160801-0940f3fc43a0 h1:3UeQBvD0TFrlV github.com/wsxiaoys/terminal v0.0.0-20160513160801-0940f3fc43a0/go.mod h1:IXCdmsXIht47RaVFLEdVnh1t+pgYtTAhQGj73kz+2DM= golang.org/x/exp v0.0.0-20230515195305-f3d0a9c9a5cc h1:mCRnTeVUjcrhlRmO0VK8a6k6Rrf6TF9htwo2pJVSjIU= golang.org/x/exp v0.0.0-20230515195305-f3d0a9c9a5cc/go.mod h1:V1LtkGg67GoY2N1AnLN78QLrzxkLyJw7RJb1gzOOz9w= +golang.org/x/sync v0.6.0 h1:5BMeUDZ7vkXGfEr1x9B4bRcTH4lpkTkpdh0T/J+qjbQ= +golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE= golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= diff --git a/vendor/golang.org/x/sync/LICENSE b/vendor/golang.org/x/sync/LICENSE new file mode 100644 index 0000000..6a66aea --- /dev/null +++ b/vendor/golang.org/x/sync/LICENSE @@ -0,0 +1,27 @@ +Copyright (c) 2009 The Go Authors. All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are +met: + + * Redistributions of source code must retain the above copyright +notice, this list of conditions and the following disclaimer. + * Redistributions in binary form must reproduce the above +copyright notice, this list of conditions and the following disclaimer +in the documentation and/or other materials provided with the +distribution. + * Neither the name of Google Inc. nor the names of its +contributors may be used to endorse or promote products derived from +this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/vendor/golang.org/x/sync/PATENTS b/vendor/golang.org/x/sync/PATENTS new file mode 100644 index 0000000..7330990 --- /dev/null +++ b/vendor/golang.org/x/sync/PATENTS @@ -0,0 +1,22 @@ +Additional IP Rights Grant (Patents) + +"This implementation" means the copyrightable works distributed by +Google as part of the Go project. + +Google hereby grants to You a perpetual, worldwide, non-exclusive, +no-charge, royalty-free, irrevocable (except as stated in this section) +patent license to make, have made, use, offer to sell, sell, import, +transfer and otherwise run, modify and propagate the contents of this +implementation of Go, where such license applies only to those patent +claims, both currently owned or controlled by Google and acquired in +the future, licensable by Google that are necessarily infringed by this +implementation of Go. This grant does not include claims that would be +infringed only as a consequence of further modification of this +implementation. If you or your agent or exclusive licensee institute or +order or agree to the institution of patent litigation against any +entity (including a cross-claim or counterclaim in a lawsuit) alleging +that this implementation of Go or any code incorporated within this +implementation of Go constitutes direct or contributory patent +infringement, or inducement of patent infringement, then any patent +rights granted to you under this License for this implementation of Go +shall terminate as of the date such litigation is filed. diff --git a/vendor/golang.org/x/sync/errgroup/errgroup.go b/vendor/golang.org/x/sync/errgroup/errgroup.go new file mode 100644 index 0000000..948a3ee --- /dev/null +++ b/vendor/golang.org/x/sync/errgroup/errgroup.go @@ -0,0 +1,135 @@ +// Copyright 2016 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +// Package errgroup provides synchronization, error propagation, and Context +// cancelation for groups of goroutines working on subtasks of a common task. +// +// [errgroup.Group] is related to [sync.WaitGroup] but adds handling of tasks +// returning errors. +package errgroup + +import ( + "context" + "fmt" + "sync" +) + +type token struct{} + +// A Group is a collection of goroutines working on subtasks that are part of +// the same overall task. +// +// A zero Group is valid, has no limit on the number of active goroutines, +// and does not cancel on error. +type Group struct { + cancel func(error) + + wg sync.WaitGroup + + sem chan token + + errOnce sync.Once + err error +} + +func (g *Group) done() { + if g.sem != nil { + <-g.sem + } + g.wg.Done() +} + +// WithContext returns a new Group and an associated Context derived from ctx. +// +// The derived Context is canceled the first time a function passed to Go +// returns a non-nil error or the first time Wait returns, whichever occurs +// first. +func WithContext(ctx context.Context) (*Group, context.Context) { + ctx, cancel := withCancelCause(ctx) + return &Group{cancel: cancel}, ctx +} + +// Wait blocks until all function calls from the Go method have returned, then +// returns the first non-nil error (if any) from them. +func (g *Group) Wait() error { + g.wg.Wait() + if g.cancel != nil { + g.cancel(g.err) + } + return g.err +} + +// Go calls the given function in a new goroutine. +// It blocks until the new goroutine can be added without the number of +// active goroutines in the group exceeding the configured limit. +// +// The first call to return a non-nil error cancels the group's context, if the +// group was created by calling WithContext. The error will be returned by Wait. +func (g *Group) Go(f func() error) { + if g.sem != nil { + g.sem <- token{} + } + + g.wg.Add(1) + go func() { + defer g.done() + + if err := f(); err != nil { + g.errOnce.Do(func() { + g.err = err + if g.cancel != nil { + g.cancel(g.err) + } + }) + } + }() +} + +// TryGo calls the given function in a new goroutine only if the number of +// active goroutines in the group is currently below the configured limit. +// +// The return value reports whether the goroutine was started. +func (g *Group) TryGo(f func() error) bool { + if g.sem != nil { + select { + case g.sem <- token{}: + // Note: this allows barging iff channels in general allow barging. + default: + return false + } + } + + g.wg.Add(1) + go func() { + defer g.done() + + if err := f(); err != nil { + g.errOnce.Do(func() { + g.err = err + if g.cancel != nil { + g.cancel(g.err) + } + }) + } + }() + return true +} + +// SetLimit limits the number of active goroutines in this group to at most n. +// A negative value indicates no limit. +// +// Any subsequent call to the Go method will block until it can add an active +// goroutine without exceeding the configured limit. +// +// The limit must not be modified while any goroutines in the group are active. +func (g *Group) SetLimit(n int) { + if n < 0 { + g.sem = nil + return + } + if len(g.sem) != 0 { + panic(fmt.Errorf("errgroup: modify limit while %v goroutines in the group are still active", len(g.sem))) + } + g.sem = make(chan token, n) +} diff --git a/vendor/golang.org/x/sync/errgroup/go120.go b/vendor/golang.org/x/sync/errgroup/go120.go new file mode 100644 index 0000000..f93c740 --- /dev/null +++ b/vendor/golang.org/x/sync/errgroup/go120.go @@ -0,0 +1,13 @@ +// Copyright 2023 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +//go:build go1.20 + +package errgroup + +import "context" + +func withCancelCause(parent context.Context) (context.Context, func(error)) { + return context.WithCancelCause(parent) +} diff --git a/vendor/golang.org/x/sync/errgroup/pre_go120.go b/vendor/golang.org/x/sync/errgroup/pre_go120.go new file mode 100644 index 0000000..88ce334 --- /dev/null +++ b/vendor/golang.org/x/sync/errgroup/pre_go120.go @@ -0,0 +1,14 @@ +// Copyright 2023 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +//go:build !go1.20 + +package errgroup + +import "context" + +func withCancelCause(parent context.Context) (context.Context, func(error)) { + ctx, cancel := context.WithCancel(parent) + return ctx, func(error) { cancel() } +} diff --git a/vendor/modules.txt b/vendor/modules.txt index 1397f8c..d791e94 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -52,6 +52,9 @@ github.com/stretchr/testify/require ## explicit; go 1.20 golang.org/x/exp/constraints golang.org/x/exp/slices +# golang.org/x/sync v0.6.0 +## explicit; go 1.18 +golang.org/x/sync/errgroup # golang.org/x/text v0.9.0 ## explicit; go 1.17 golang.org/x/text/transform