From 45153cd057ffc5b39dbb088883281ac52a2f96db Mon Sep 17 00:00:00 2001 From: Dima Kossovich Date: Mon, 12 Oct 2026 17:34:02 +0300 Subject: [PATCH 1/3] performance tweaks --- internal/algo/parallel/qsort/qsort.go | 34 ++++++------- internal/algo/parallel/qsort/qsort_test.go | 15 +++--- internal/internalpipe/any.go | 58 ++++++++-------------- internal/internalpipe/do.go | 25 +++++++++- internal/internalpipe/pipe.go | 6 ++- internal/internalpipe/reduce.go | 47 +++++++++++++++++- 6 files changed, 115 insertions(+), 70 deletions(-) diff --git a/internal/algo/parallel/qsort/qsort.go b/internal/algo/parallel/qsort/qsort.go index 1827eb5..145ae30 100644 --- a/internal/algo/parallel/qsort/qsort.go +++ b/internal/algo/parallel/qsort/qsort.go @@ -62,19 +62,13 @@ func qsort[T any]( func partition[T any](data []T, lf, rg int, less func(*T, *T) bool) int { // small arrays are not sorted with this method midIdx := (rg-lf)/2 + lf - med, medIdx := median([3]T{data[lf], data[midIdx], data[rg]}, less) - switch medIdx { - case 0: - data[lf], data[midIdx] = data[midIdx], data[lf] - case 2: - data[rg], data[midIdx] = data[midIdx], data[rg] - } + med := median([3]T{data[lf], data[midIdx], data[rg]}, less) for lf <= rg { - for less(&data[lf], med) { + for less(&data[lf], &med) { lf++ } - for less(med, &data[rg]) { + for less(&med, &data[rg]) { rg-- } if lf >= rg { @@ -88,37 +82,37 @@ func partition[T any](data []T, lf, rg int, less func(*T, *T) bool) int { return rg } -// median returns the median of 3 elements; it's a bit ugly bit effective -func median[T any](elems [3]T, less func(*T, *T) bool) (*T, int16) { +// median returns the median value of 3 elements. +func median[T any](elems [3]T, less func(*T, *T) bool) T { if less(&elems[1], &elems[0]) && less(&elems[0], &elems[2]) { - return &elems[0], 0 + return elems[0] } if less(&elems[2], &elems[0]) && less(&elems[0], &elems[1]) { - return &elems[0], 0 + return elems[0] } if less(&elems[0], &elems[1]) && less(&elems[1], &elems[2]) { - return &elems[1], 1 + return elems[1] } if less(&elems[2], &elems[1]) && less(&elems[1], &elems[0]) { - return &elems[1], 1 + return elems[1] } if less(&elems[0], &elems[2]) && less(&elems[2], &elems[1]) { - return &elems[2], 2 + return elems[2] } if less(&elems[1], &elems[2]) && less(&elems[2], &elems[0]) { - return &elems[2], 2 + return elems[2] } // two elements are equal if !less(&elems[0], &elems[1]) && !less(&elems[1], &elems[0]) { - return &elems[0], 0 + return elems[0] } if !less(&elems[1], &elems[2]) && !less(&elems[2], &elems[1]) { - return &elems[1], 1 + return elems[1] } - return &elems[2], 2 + return elems[2] } func genTickets(n int) chan struct{} { diff --git a/internal/algo/parallel/qsort/qsort_test.go b/internal/algo/parallel/qsort/qsort_test.go index 627e4d1..d315407 100644 --- a/internal/algo/parallel/qsort/qsort_test.go +++ b/internal/algo/parallel/qsort/qsort_test.go @@ -156,15 +156,12 @@ func Test_sort_same(t *testing.T) { } func Test_median(t *testing.T) { - val, idx := median([...]int{1, 1, 3}, func(x, y *int) bool { return *x < *y }) - require.Equal(t, 1, *val) - require.Equal(t, int16(0), idx) + val := median([...]int{1, 1, 3}, func(x, y *int) bool { return *x < *y }) + require.Equal(t, 1, val) - val, idx = median([...]int{0, 1, 1}, func(x, y *int) bool { return *x < *y }) - require.Equal(t, 1, *val) - require.Equal(t, int16(1), idx) + val = median([...]int{0, 1, 1}, func(x, y *int) bool { return *x < *y }) + require.Equal(t, 1, val) - val, idx = median([...]int{1, 2, 1}, func(x, y *int) bool { return *x < *y }) - require.Equal(t, 1, *val) - require.Equal(t, int16(2), idx) + val = median([...]int{1, 2, 1}, func(x, y *int) bool { return *x < *y }) + require.Equal(t, 1, val) } diff --git a/internal/internalpipe/any.go b/internal/internalpipe/any.go index 52d37ed..96d58f1 100644 --- a/internal/internalpipe/any.go +++ b/internal/internalpipe/any.go @@ -1,12 +1,17 @@ package internalpipe import ( + "runtime" "sync" - "time" ) +// hugeLenStep is the base chunk size used when the pipe length is unknown. const hugeLenStep = 1 << 15 +// anyResCheckInterval controls how often a worker re-reads the shared result +// flag while scanning a chunk, so it does not take a lock on every element. +const anyResCheckInterval = 64 + func anySingleThread[T any](limit int, fn GeneratorFn[T]) *T { var obj *T var skipped bool @@ -21,15 +26,13 @@ func anySingleThread[T any](limit int, fn GeneratorFn[T]) *T { // Any returns a pointer to a random element in the pipe or nil if none left. func (p Pipe[T]) Any() *T { - const mutexUpdateCoef = 18 - limit := p.limit() if p.GoroutinesCnt == 1 { return anySingleThread(limit, p.Fn) } bounded := p.lenSet() || p.limitSet() - step := hugeLenStep + step := hugeLenStep * max(1, runtime.GOMAXPROCS(0)) if bounded { step = max(divUp(limit, p.GoroutinesCnt), 1) } @@ -53,6 +56,12 @@ func (p Pipe[T]) Any() *T { } mx.Unlock() } + getResSet := func() bool { + mx.Lock() + rs := resSet + mx.Unlock() + return rs + } go func() { // i >= 0 is for an int owerflow case @@ -78,43 +87,16 @@ func (p Pipe[T]) Any() *T { rg = min(rg, limit) } - var avgFnTime time.Duration - var avgUpdResSetTime time.Duration - resSetUpdCnt := int64(0) - beforeLastResSetUpd := 0 - - getResSet := func() bool { - start := time.Now() - mx.Lock() - rs := resSet - mx.Unlock() - avgUpdResSetTime = time.Duration( - (int64(time.Since(start)) + int64(avgUpdResSetTime)*(resSetUpdCnt)) / (resSetUpdCnt + 1), - ) - resSetUpdCnt++ - beforeLastResSetUpd = 0 - return rs + if getResSet() { + return } - rs := getResSet() - cnt := 0 for j := lf; j < rg; j++ { - beforeLastResSetUpd++ - if j != lf && - avgFnTime != 0 && - int64(beforeLastResSetUpd) > (mutexUpdateCoef*int64(avgUpdResSetTime)/int64(avgFnTime)) { - rs = getResSet() - cnt++ + if j != lf && (j-lf)%anyResCheckInterval == 0 && getResSet() { + return } - if !rs { - start := time.Now() - obj, skipped := p.Fn(j) - if !skipped { - setObj(obj) - return - } - avgFnTime = time.Duration( - (int64(time.Since(start)) + int64(avgFnTime)*int64(j-lf)) / int64(j-lf+1), - ) + if obj, skipped := p.Fn(j); !skipped { + setObj(obj) + return } } }(i, i+step) diff --git a/internal/internalpipe/do.go b/internal/internalpipe/do.go index e07cfbe..f846fdc 100644 --- a/internal/internalpipe/do.go +++ b/internal/internalpipe/do.go @@ -51,9 +51,32 @@ func (p *Pipe[T]) do(needResult bool) ([]T, int) { defer p.y.Handle() } + limit := p.limit() + + // A single worker keeps the natural order, so the result can be produced + // in one pass with no intermediate buffer. + if p.GoroutinesCnt == 1 { + if !needResult { + sCnt := 0 + for i := 0; i < limit; i++ { + if _, skipped := p.Fn(i); skipped { + sCnt++ + } + } + return nil, limit - sCnt + } + + res := make([]T, 0, limit) + for i := 0; i < limit; i++ { + if obj, skipped := p.Fn(i); !skipped { + res = append(res, *obj) + } + } + return res, len(res) + } + var ( eval []ev[T] - limit = p.limit() step = max(divUp(limit, p.GoroutinesCnt), 1) wg sync.WaitGroup skipCnt atomic.Int64 diff --git a/internal/internalpipe/pipe.go b/internal/internalpipe/pipe.go index 3a168b3..e2447ba 100644 --- a/internal/internalpipe/pipe.go +++ b/internal/internalpipe/pipe.go @@ -95,7 +95,11 @@ func max[T constraints.Ordered](a, b T) T { } func divUp(a, b int) int { - return int(math.Ceil(float64(a) / float64(b))) + q := a / b + if a%b != 0 { + q++ + } + return q } func genTickets(n int) chan struct{} { diff --git a/internal/internalpipe/reduce.go b/internal/internalpipe/reduce.go index 09d6daf..b79aa11 100644 --- a/internal/internalpipe/reduce.go +++ b/internal/internalpipe/reduce.go @@ -1,10 +1,55 @@ package internalpipe +import "math" + type AccumFn[T any] func(*T, *T) T // Reduce applies the result of a function to each element one-by-one: f(p[n], f(p[n-1], f(p[n-2, ...]))). func (p Pipe[T]) Reduce(fn AccumFn[T]) *T { - data := p.Do() + if p.GoroutinesCnt > 1 { + // Fold over the parallel evaluation result. + return reduceSlice(p.Do(), fn) + } + + var ( + res T + set bool + ) + fold := func(obj *T) { + if set { + res = fn(&res, obj) + } else { + res = *obj + set = true + } + } + + if p.limitSet() { + for i, got := 0, 0; got < p.ValLim; i++ { + if obj, skipped := p.Fn(i); !skipped { + fold(obj) + got++ + } + if i == math.MaxInt { + panic(panicLimitExceededMsg) + } + } + } else { + limit := p.limit() + for i := 0; i < limit; i++ { + if obj, skipped := p.Fn(i); !skipped { + fold(obj) + } + } + } + + if !set { + return nil + } + return &res +} + +func reduceSlice[T any](data []T, fn AccumFn[T]) *T { switch len(data) { case 0: return nil From b1387a42727d3636a9d79a188dfc08385b6b7456 Mon Sep 17 00:00:00 2001 From: Dima Kossovich Date: Mon, 12 Oct 2026 17:53:37 +0300 Subject: [PATCH 2/3] do improved dummy --- internal/internalpipe/do.go | 57 +++++++++++++++++++++++-------------- 1 file changed, 36 insertions(+), 21 deletions(-) diff --git a/internal/internalpipe/do.go b/internal/internalpipe/do.go index f846fdc..5b7582d 100644 --- a/internal/internalpipe/do.go +++ b/internal/internalpipe/do.go @@ -6,11 +6,6 @@ import ( "sync/atomic" ) -type ev[T any] struct { - skipped bool - obj *T -} - // Do evaluates all the pipeline and returns the result slice. func (p Pipe[T]) Do() []T { if p.limitSet() { @@ -52,6 +47,7 @@ func (p *Pipe[T]) do(needResult bool) ([]T, int) { } limit := p.limit() + fn := p.Fn // A single worker keeps the natural order, so the result can be produced // in one pass with no intermediate buffer. @@ -59,7 +55,7 @@ func (p *Pipe[T]) do(needResult bool) ([]T, int) { if !needResult { sCnt := 0 for i := 0; i < limit; i++ { - if _, skipped := p.Fn(i); skipped { + if _, skipped := fn(i); skipped { sCnt++ } } @@ -68,21 +64,30 @@ func (p *Pipe[T]) do(needResult bool) ([]T, int) { res := make([]T, 0, limit) for i := 0; i < limit; i++ { - if obj, skipped := p.Fn(i); !skipped { + if obj, skipped := fn(i); !skipped { res = append(res, *obj) } } return res, len(res) } + // Parallel path. Every worker writes its results straight into the shared + // result buffer at the element's own index and marks dropped positions in a + // zeroed bitmap. This avoids keeping one alive *T per element (a Map boxes + // every value), so the per-element allocations become collectable instead of + // surviving until the gather. When nothing is skipped — the common Map case — + // the buffer already is the answer and the serial gather pass disappears + // entirely; otherwise the result is compacted in place. var ( - eval []ev[T] + res []T + skipped []bool step = max(divUp(limit, p.GoroutinesCnt), 1) wg sync.WaitGroup skipCnt atomic.Int64 ) - if needResult && limit > 0 { - eval = make([]ev[T], limit) + if needResult { + res = make([]T, limit) + skipped = make([]bool, limit) } tickets := genTickets(p.GoroutinesCnt) for i := 0; i > -1 && i < limit; i += step { @@ -95,15 +100,16 @@ func (p *Pipe[T]) do(needResult bool) ([]T, int) { rg = min(rg, limit) var sCnt int64 for j := lf; j < rg; j++ { - obj, skipped := p.Fn(j) - if skipped { + obj, sk := fn(j) + if sk { sCnt++ + if needResult { + skipped[j] = true + } + continue } if needResult { - eval[j] = ev[T]{ - obj: obj, - skipped: skipped, - } + res[j] = *obj } } skipCnt.Add(sCnt) @@ -113,11 +119,20 @@ func (p *Pipe[T]) do(needResult bool) ([]T, int) { } wg.Wait() - res := make([]T, 0, limit-int(skipCnt.Load())) - for i := range eval { - if !eval[i].skipped { - res = append(res, *eval[i].obj) + skips := int(skipCnt.Load()) + if !needResult { + return nil, limit - skips + } + if skips == 0 { + return res, limit + } + + w := 0 + for i := range res { + if !skipped[i] { + res[w] = res[i] + w++ } } - return res, limit - int(skipCnt.Load()) + return res[:w], w } From 78fdaf6ac3edeff7c5e286c1ce969e138c59564a Mon Sep 17 00:00:00 2001 From: Dima Kossovich Date: Mon, 12 Oct 2026 18:27:00 +0300 Subject: [PATCH 3/3] improved performance of Map and Filter (and others) significantly, minor internal behaviour changes --- README.md | 2 +- internal/internalpipe/any.go | 13 +++--- internal/internalpipe/constructor.go | 34 +++++++++----- internal/internalpipe/constructor_test.go | 11 ++--- internal/internalpipe/do.go | 19 ++++---- internal/internalpipe/do_test.go | 20 +++++---- internal/internalpipe/erase.go | 13 +++--- internal/internalpipe/filter.go | 11 ++--- internal/internalpipe/filter_test.go | 25 ++++++----- internal/internalpipe/first.go | 23 +++++----- internal/internalpipe/first_test.go | 47 ++++++++++--------- internal/internalpipe/map.go | 10 ++--- internal/internalpipe/map_test.go | 20 +++++---- internal/internalpipe/mapfilter.go | 11 ++--- internal/internalpipe/mapfilterer_test.go | 35 +++++++++------ internal/internalpipe/pipe.go | 7 ++- internal/internalpipe/pipe_test.go | 55 ++++++++++++++--------- internal/internalpipe/promices.go | 6 +-- internal/internalpipe/reduce.go | 9 ++-- internal/internalpipe/reduce_test.go | 40 ++++++++++------- internal/internalpipe/sort.go | 7 +-- internal/internalpipe/sum.go | 37 ++++++++------- internal/internalpipe/sum_test.go | 52 ++++++++++++--------- pkg/pipe/collectors.go | 28 +++++++----- pkg/pipe/prefixpipe.go | 52 ++++++++++++--------- 25 files changed, 338 insertions(+), 249 deletions(-) diff --git a/README.md b/README.md index 327a06a..229d085 100644 --- a/README.md +++ b/README.md @@ -155,7 +155,7 @@ These two only exist on `PiperNoLen` and are how you turn an unknown-length pipe Sometimes your pipeline needs to lug around different types at different steps. `Erase` gives up the compile-time type so you can work with `any`, and `Collect` puts a type back on. -- :frog: `Erase() Piper[any]` (or `PiperNoLen[any]`) -- every element becomes `any` (backed by a pointer to the original value). +- :frog: `Erase() Piper[any]` (or `PiperNoLen[any]`) -- every element becomes `any` (each backed by a `*T`, a pointer to the value). - :frog: `pipe.Collect[DstT](Piper[any]) Piper[DstT]` -- pull a `Piper[any]` back into a concrete `Piper[DstT]`. - :frog: `pipe.CollectNL[DstT](PiperNoLen[any]) PiperNoLen[DstT]` -- the same for a pipe with no known length. diff --git a/internal/internalpipe/any.go b/internal/internalpipe/any.go index 96d58f1..ac7487d 100644 --- a/internal/internalpipe/any.go +++ b/internal/internalpipe/any.go @@ -13,12 +13,10 @@ const hugeLenStep = 1 << 15 const anyResCheckInterval = 64 func anySingleThread[T any](limit int, fn GeneratorFn[T]) *T { - var obj *T - var skipped bool - + var obj T for i := 0; i < limit; i++ { - if obj, skipped = fn(i); !skipped { - return obj + if !fn(i, &obj) { + return &obj } } return nil @@ -90,12 +88,13 @@ func (p Pipe[T]) Any() *T { if getResSet() { return } + var obj T for j := lf; j < rg; j++ { if j != lf && (j-lf)%anyResCheckInterval == 0 && getResSet() { return } - if obj, skipped := p.Fn(j); !skipped { - setObj(obj) + if !p.Fn(j, &obj) { + setObj(&obj) return } } diff --git a/internal/internalpipe/constructor.go b/internal/internalpipe/constructor.go index cf1d84c..232b260 100644 --- a/internal/internalpipe/constructor.go +++ b/internal/internalpipe/constructor.go @@ -13,11 +13,12 @@ const ( func Slice[T any](dt []T) Pipe[T] { return Pipe[T]{ - Fn: func(i int) (*T, bool) { + Fn: func(i int, dst *T) bool { if i >= len(dt) { - return nil, true + return true } - return &dt[i], false + *dst = dt[i] + return false }, Len: len(dt), ValLim: notSet, @@ -27,9 +28,13 @@ func Slice[T any](dt []T) Pipe[T] { func Func[T any](fn func(i int) (T, bool)) Pipe[T] { return Pipe[T]{ - Fn: func(i int) (*T, bool) { + Fn: func(i int, dst *T) bool { obj, exist := fn(i) - return &obj, !exist + if !exist { + return true + } + *dst = obj + return false }, Len: notSet, ValLim: notSet, @@ -39,9 +44,13 @@ func Func[T any](fn func(i int) (T, bool)) Pipe[T] { func FuncP[T any](fn func(i int) (*T, bool)) Pipe[T] { return Pipe[T]{ - Fn: func(i int) (*T, bool) { + Fn: func(i int, dst *T) bool { obj, exist := fn(i) - return obj, !exist + if !exist { + return true + } + *dst = *obj + return false }, Len: notSet, ValLim: notSet, @@ -82,9 +91,10 @@ func Range[T constraints.Integer | constraints.Float](start, finish, step T) Pip } } return Pipe[T]{ - Fn: func(i int) (*T, bool) { + Fn: func(i int, dst *T) bool { val := start + T(i)*step - return &val, pred(val) + *dst = val + return pred(val) }, Len: rangeLen(start, finish, step, pred), ValLim: notSet, @@ -103,9 +113,9 @@ func Repeat[T any](x T, n int) Pipe[T] { } return Pipe[T]{ - Fn: func(i int) (*T, bool) { - cp := x - return &cp, i >= n + Fn: func(i int, dst *T) bool { + *dst = x + return i >= n }, Len: n, ValLim: notSet, diff --git a/internal/internalpipe/constructor_test.go b/internal/internalpipe/constructor_test.go index 936930d..23f01e4 100644 --- a/internal/internalpipe/constructor_test.go +++ b/internal/internalpipe/constructor_test.go @@ -15,10 +15,10 @@ func Test_Slice(t *testing.T) { require.Equal(t, p.Len, 10) require.Equal(t, p.ValLim, notSet) require.Equal(t, p.GoroutinesCnt, defaultParallelWrks) - r, _ := p.Fn(5) - require.Equal(t, *r, a[5]) - _, skipped := p.Fn(15) - require.True(t, skipped) + var r int + require.True(t, p.Fn(15, &r)) + require.False(t, p.Fn(5, &r)) + require.Equal(t, a[5], r) } func Test_FuncP(t *testing.T) { @@ -163,8 +163,9 @@ func Test_Range_FloatLen(t *testing.T) { p := Range(c.start, c.finish, c.step) yielded := 0 + var tmp float64 for i := 0; ; i++ { - if _, skipped := p.Fn(i); skipped { + if p.Fn(i, &tmp) { break } yielded++ diff --git a/internal/internalpipe/do.go b/internal/internalpipe/do.go index 5b7582d..9f6bb9c 100644 --- a/internal/internalpipe/do.go +++ b/internal/internalpipe/do.go @@ -27,10 +27,10 @@ func (p *Pipe[T]) doToLimit() []T { } res := make([]T, 0, p.ValLim) + var v T for i := 0; len(res) < p.ValLim; i++ { - obj, skipped := p.Fn(i) - if !skipped { - res = append(res, *obj) + if !p.Fn(i, &v) { + res = append(res, v) } if i == math.MaxInt { @@ -52,10 +52,11 @@ func (p *Pipe[T]) do(needResult bool) ([]T, int) { // A single worker keeps the natural order, so the result can be produced // in one pass with no intermediate buffer. if p.GoroutinesCnt == 1 { + var v T if !needResult { sCnt := 0 for i := 0; i < limit; i++ { - if _, skipped := fn(i); skipped { + if fn(i, &v) { sCnt++ } } @@ -64,8 +65,8 @@ func (p *Pipe[T]) do(needResult bool) ([]T, int) { res := make([]T, 0, limit) for i := 0; i < limit; i++ { - if obj, skipped := fn(i); !skipped { - res = append(res, *obj) + if !fn(i, &v) { + res = append(res, v) } } return res, len(res) @@ -99,9 +100,9 @@ func (p *Pipe[T]) do(needResult bool) ([]T, int) { } rg = min(rg, limit) var sCnt int64 + var v T for j := lf; j < rg; j++ { - obj, sk := fn(j) - if sk { + if fn(j, &v) { sCnt++ if needResult { skipped[j] = true @@ -109,7 +110,7 @@ func (p *Pipe[T]) do(needResult bool) ([]T, int) { continue } if needResult { - res[j] = *obj + res[j] = v } } skipCnt.Add(sCnt) diff --git a/internal/internalpipe/do_test.go b/internal/internalpipe/do_test.go index 4e0502c..2cb1fd8 100644 --- a/internal/internalpipe/do_test.go +++ b/internal/internalpipe/do_test.go @@ -18,8 +18,9 @@ func Test_Do(t *testing.T) { t.Run("single thread lim set", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 100_000, ValLim: -1, @@ -38,8 +39,9 @@ func Test_Do(t *testing.T) { t.Run("seven thread lim set", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 100_000, ValLim: -1, @@ -58,8 +60,9 @@ func Test_Do(t *testing.T) { t.Run("single thread ValLim set", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: len(exp), @@ -78,8 +81,9 @@ func Test_Do(t *testing.T) { t.Run("seven thread ValLim set", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: len(exp), diff --git a/internal/internalpipe/erase.go b/internal/internalpipe/erase.go index b344ec2..3bf4c9c 100644 --- a/internal/internalpipe/erase.go +++ b/internal/internalpipe/erase.go @@ -2,12 +2,15 @@ package internalpipe func (p Pipe[T]) Erase() Pipe[any] { return Pipe[any]{ - Fn: func(i int) (*any, bool) { - if obj, skipped := p.Fn(i); !skipped { - anyObj := any(obj) - return &anyObj, false + Fn: func(i int, dst *any) bool { + var obj T + if p.Fn(i, &obj) { + return true } - return nil, true + // Keep the historical behaviour: the erased value stays a *T, + // so Collect can assert it back to *DstT. + *dst = any(&obj) + return false }, Len: p.Len, ValLim: p.ValLim, diff --git a/internal/internalpipe/filter.go b/internal/internalpipe/filter.go index e8d4e7f..4722913 100644 --- a/internal/internalpipe/filter.go +++ b/internal/internalpipe/filter.go @@ -3,14 +3,11 @@ package internalpipe // Filter leaves only items with true predicate fn. func (p Pipe[T]) Filter(fn func(*T) bool) Pipe[T] { return Pipe[T]{ - Fn: func(i int) (*T, bool) { - if obj, skipped := p.Fn(i); !skipped { - if !fn(obj) { - return nil, true - } - return obj, false + Fn: func(i int, dst *T) bool { + if p.Fn(i, dst) { + return true } - return nil, true + return !fn(dst) }, Len: p.Len, ValLim: p.ValLim, diff --git a/internal/internalpipe/filter_test.go b/internal/internalpipe/filter_test.go index 2a60725..06920a3 100644 --- a/internal/internalpipe/filter_test.go +++ b/internal/internalpipe/filter_test.go @@ -13,8 +13,9 @@ func Test_Filter(t *testing.T) { t.Run("single thread even numbers filter", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 100_000, ValLim: -1, @@ -32,8 +33,9 @@ func Test_Filter(t *testing.T) { }) t.Run("seven thread even numbers filter", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 100_000, ValLim: -1, @@ -52,8 +54,9 @@ func Test_Filter(t *testing.T) { t.Run("single thread even numbers empty res filter", func(t *testing.T) { pts := pointer.Ref(7) p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return pts, false + Fn: func(i int, dst *int) bool { + *dst = *pts + return false }, Len: 100_000, ValLim: -1, @@ -68,8 +71,9 @@ func Test_Filter(t *testing.T) { t.Run("seven thread even numbers empty res filter", func(t *testing.T) { pts := pointer.Ref(7) p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return pts, false + Fn: func(i int, dst *int) bool { + *dst = *pts + return false }, Len: 100_000, ValLim: -1, @@ -84,8 +88,9 @@ func Test_Filter(t *testing.T) { t.Run("seven thread even numbers empty res double filter", func(t *testing.T) { pts := pointer.Ref(7) p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return pts, false + Fn: func(i int, dst *int) bool { + *dst = *pts + return false }, Len: 100_000, ValLim: -1, diff --git a/internal/internalpipe/first.go b/internal/internalpipe/first.go index 83f4198..22a9c05 100644 --- a/internal/internalpipe/first.go +++ b/internal/internalpipe/first.go @@ -14,13 +14,11 @@ func (p Pipe[T]) First() *T { return first(limit, p.GoroutinesCnt, p.Fn) } -func firstSingleThread[T any](limit int, fn func(i int) (*T, bool)) *T { - var obj *T - var skipped bool +func firstSingleThread[T any](limit int, fn GeneratorFn[T]) *T { + var obj T for i := 0; i < limit; i++ { - obj, skipped = fn(i) - if !skipped { - return obj + if !fn(i, &obj) { + return &obj } } return nil @@ -102,7 +100,7 @@ func (r *firstResult[T]) stepDone(step int) { } } -func first[T any](limit, grtCnt int, fn func(i int) (*T, bool)) (f *T) { +func first[T any](limit, grtCnt int, fn func(i int, dst *T) bool) (f *T) { if limit == 0 { return } @@ -129,12 +127,12 @@ func first[T any](limit, grtCnt int, fn func(i int) (*T, bool)) (f *T) { rg = min(rg, limit) done := res.ctx.Done() + var val T for j := lf; j < rg; j++ { // FIXME: this code is ugly but saves about 30% of time on locks if j%2 != 0 { - val, skipped := fn(j) - if !skipped { - res.setVal(val, stepCnt) + if !fn(j, &val) { + res.setVal(&val, stepCnt) return } } else { @@ -142,9 +140,8 @@ func first[T any](limit, grtCnt int, fn func(i int) (*T, bool)) (f *T) { case <-done: return default: - val, skipped := fn(j) - if !skipped { - res.setVal(val, stepCnt) + if !fn(j, &val) { + res.setVal(&val, stepCnt) return } } diff --git a/internal/internalpipe/first_test.go b/internal/internalpipe/first_test.go index 28f3f41..60f161d 100644 --- a/internal/internalpipe/first_test.go +++ b/internal/internalpipe/first_test.go @@ -23,8 +23,9 @@ func TestFirst(t *testing.T) { t.Parallel() p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &s[i], i <= 900_000 + Fn: func(i int, dst *int) bool { + *dst = s[i] + return i <= 900_000 }, Len: len(s), ValLim: -1, @@ -37,8 +38,9 @@ func TestFirst(t *testing.T) { t.Parallel() p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &s[i], true + Fn: func(i int, dst *int) bool { + *dst = s[i] + return true }, Len: len(s), ValLim: -1, @@ -51,8 +53,9 @@ func TestFirst(t *testing.T) { t.Parallel() p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &s[i], i <= 900_000 + Fn: func(i int, dst *int) bool { + *dst = s[i] + return i <= 900_000 }, Len: len(s), ValLim: -1, @@ -66,8 +69,9 @@ func TestFirst(t *testing.T) { t.Parallel() p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &s[i], s[i] < 1000 + Fn: func(i int, dst *int) bool { + *dst = s[i] + return s[i] < 1000 }, Len: len(s), ValLim: -1, @@ -81,8 +85,9 @@ func TestFirst(t *testing.T) { t.Parallel() p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &s[i], i <= 900_000 + Fn: func(i int, dst *int) bool { + *dst = s[i] + return i <= 900_000 }, Len: len(s), ValLim: -1, @@ -96,8 +101,8 @@ func TestFirst(t *testing.T) { t.Parallel() p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return nil, true + Fn: func(i int, dst *int) bool { + return true }, Len: len(s), ValLim: -1, @@ -110,8 +115,8 @@ func TestFirst(t *testing.T) { t.Parallel() p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return nil, true + Fn: func(i int, dst *int) bool { + return true }, Len: 0, ValLim: 0, @@ -148,8 +153,8 @@ func TestFirst_UnevenChunks(t *testing.T) { for _, c := range []struct{ length, goroutines int }{{10, 7}, {5, 4}, {7, 3}, {13, 5}} { c := c p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return nil, true + Fn: func(i int, dst *int) bool { + return true }, Len: c.length, ValLim: -1, @@ -166,8 +171,9 @@ func TestFirst_UnevenChunks(t *testing.T) { for _, target := range []int{0, 1, 6, 9, 500_000, 999_999} { target := target p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &s[i], i != target + Fn: func(i int, dst *int) bool { + *dst = s[i] + return i != target }, Len: len(s), ValLim: -1, @@ -182,8 +188,9 @@ func TestFirst_UnevenChunks(t *testing.T) { s := genSlice(1_000_000) const border = 100_000 - fn := func(i int) (*int, bool) { - return &s[i], s[i] < border + fn := func(i int, dst *int) bool { + *dst = s[i] + return s[i] < border } want := *Pipe[int]{Fn: fn, Len: len(s), ValLim: -1, GoroutinesCnt: 1}.First() got := firstWithin(t, Pipe[int]{Fn: fn, Len: len(s), ValLim: -1, GoroutinesCnt: 7}) diff --git a/internal/internalpipe/map.go b/internal/internalpipe/map.go index e3341d1..869dc4e 100644 --- a/internal/internalpipe/map.go +++ b/internal/internalpipe/map.go @@ -4,12 +4,12 @@ package internalpipe // returns the slice where each element is n[i] = f(p[i]). func (p Pipe[T]) Map(fn func(T) T) Pipe[T] { return Pipe[T]{ - Fn: func(i int) (*T, bool) { - if obj, skipped := p.Fn(i); !skipped { - res := fn(*obj) - return &res, false + Fn: func(i int, dst *T) bool { + if p.Fn(i, dst) { + return true } - return nil, true + *dst = fn(*dst) + return false }, Len: p.Len, ValLim: p.ValLim, diff --git a/internal/internalpipe/map_test.go b/internal/internalpipe/map_test.go index 2504adb..c78519a 100644 --- a/internal/internalpipe/map_test.go +++ b/internal/internalpipe/map_test.go @@ -16,8 +16,9 @@ func Test_Map(t *testing.T) { t.Run("single thread lim set", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 100_000, ValLim: -1, @@ -36,8 +37,9 @@ func Test_Map(t *testing.T) { t.Run("seven thread lim set", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 100_000, ValLim: -1, @@ -56,8 +58,9 @@ func Test_Map(t *testing.T) { t.Run("single thread ValLim set", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: len(exp), @@ -76,8 +79,9 @@ func Test_Map(t *testing.T) { t.Run("seven thread ValLim set", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: len(exp), diff --git a/internal/internalpipe/mapfilter.go b/internal/internalpipe/mapfilter.go index 9908cde..6ee6939 100644 --- a/internal/internalpipe/mapfilter.go +++ b/internal/internalpipe/mapfilter.go @@ -5,12 +5,13 @@ package internalpipe // returns the slice where each element is n[i] = f(p[i]) if it is not skipped. func (p Pipe[T]) MapFilter(fn func(T) (T, bool)) Pipe[T] { return Pipe[T]{ - Fn: func(i int) (*T, bool) { - if obj, skipped := p.Fn(i); !skipped { - res, take := fn(*obj) - return &res, !take + Fn: func(i int, dst *T) bool { + if p.Fn(i, dst) { + return true } - return nil, true + res, take := fn(*dst) + *dst = res + return !take }, Len: p.Len, ValLim: p.ValLim, diff --git a/internal/internalpipe/mapfilterer_test.go b/internal/internalpipe/mapfilterer_test.go index 3187ab4..d3bd28a 100644 --- a/internal/internalpipe/mapfilterer_test.go +++ b/internal/internalpipe/mapfilterer_test.go @@ -18,8 +18,9 @@ func Test_MapFilter(t *testing.T) { t.Run("single thread lim set", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 100_000, ValLim: -1, @@ -36,8 +37,9 @@ func Test_MapFilter(t *testing.T) { t.Run("seven thread lim set", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 100_000, ValLim: -1, @@ -54,8 +56,9 @@ func Test_MapFilter(t *testing.T) { t.Run("single thread ValLim set", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: len(exp), @@ -72,8 +75,9 @@ func Test_MapFilter(t *testing.T) { t.Run("seven thread ValLim set", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: len(exp), @@ -90,8 +94,9 @@ func Test_MapFilter(t *testing.T) { t.Run("seven thread ValLim multiple calls", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: len(exp), @@ -109,8 +114,9 @@ func Test_MapFilter(t *testing.T) { t.Run("single thread lim set empty", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 100_000, ValLim: -1, @@ -123,8 +129,9 @@ func Test_MapFilter(t *testing.T) { t.Run("seven thread lim set empty", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 100_000, ValLim: -1, diff --git a/internal/internalpipe/pipe.go b/internal/internalpipe/pipe.go index e2447ba..c0d53f9 100644 --- a/internal/internalpipe/pipe.go +++ b/internal/internalpipe/pipe.go @@ -10,7 +10,12 @@ const ( panicLimitExceededMsg = "the limit have been exceeded, but the result is not calculated" ) -type GeneratorFn[T any] func(int) (*T, bool) +// GeneratorFn fills *dst with the element at position i and reports whether +// that position was skipped (in which case dst is left untouched). The caller +// hands in a reusable destination, so no stage ever has to heap-allocate a box +// per element — the single biggest source of allocation in a Map- or +// Filter-heavy pipeline. +type GeneratorFn[T any] func(i int, dst *T) bool type Pipe[T any] struct { Fn GeneratorFn[T] diff --git a/internal/internalpipe/pipe_test.go b/internal/internalpipe/pipe_test.go index 6013689..7114722 100644 --- a/internal/internalpipe/pipe_test.go +++ b/internal/internalpipe/pipe_test.go @@ -65,8 +65,9 @@ func Test_do(t *testing.T) { t.Parallel() p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 10000, ValLim: -1, @@ -82,8 +83,9 @@ func Test_Parallel(t *testing.T) { t.Parallel() p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 10000, ValLim: -1, @@ -95,8 +97,9 @@ func Test_Parallel(t *testing.T) { require.Equal(t, 10, p.GoroutinesCnt) p = Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 10000, ValLim: -1, @@ -112,8 +115,9 @@ func Test_Take(t *testing.T) { t.Parallel() p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: -1, @@ -127,8 +131,9 @@ func Test_Take(t *testing.T) { require.Equal(t, 10, p.limit()) p = Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: -1, @@ -146,8 +151,9 @@ func Test_Gen(t *testing.T) { t.Parallel() p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: -1, @@ -161,8 +167,9 @@ func Test_Gen(t *testing.T) { require.Equal(t, 10, p.limit()) p = Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: -1, @@ -180,8 +187,9 @@ func Test_Count(t *testing.T) { t.Parallel() p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 10000, ValLim: -1, @@ -190,8 +198,9 @@ func Test_Count(t *testing.T) { require.Equal(t, 10000, p.Count()) p = Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 10000, ValLim: -1, @@ -200,8 +209,9 @@ func Test_Count(t *testing.T) { require.Equal(t, 10000, p.Count()) p = Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: 10000, @@ -210,8 +220,9 @@ func Test_Count(t *testing.T) { require.Equal(t, 10000, p.Count()) p = Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: 10000, diff --git a/internal/internalpipe/promices.go b/internal/internalpipe/promices.go index b1ab909..559ba50 100644 --- a/internal/internalpipe/promices.go +++ b/internal/internalpipe/promices.go @@ -7,11 +7,11 @@ func (p Pipe[T]) Promices() []func() (T, bool) { for i := 0; i < limit; i++ { cpi := i proms[i] = func() (T, bool) { - obj, skipped := p.Fn(cpi) - if skipped { + var obj T + if p.Fn(cpi, &obj) { return empty, false } - return *obj, true + return obj, true } } return proms diff --git a/internal/internalpipe/reduce.go b/internal/internalpipe/reduce.go index b79aa11..6c08244 100644 --- a/internal/internalpipe/reduce.go +++ b/internal/internalpipe/reduce.go @@ -24,10 +24,11 @@ func (p Pipe[T]) Reduce(fn AccumFn[T]) *T { } } + var v T if p.limitSet() { for i, got := 0, 0; got < p.ValLim; i++ { - if obj, skipped := p.Fn(i); !skipped { - fold(obj) + if !p.Fn(i, &v) { + fold(&v) got++ } if i == math.MaxInt { @@ -37,8 +38,8 @@ func (p Pipe[T]) Reduce(fn AccumFn[T]) *T { } else { limit := p.limit() for i := 0; i < limit; i++ { - if obj, skipped := p.Fn(i); !skipped { - fold(obj) + if !p.Fn(i, &v) { + fold(&v) } } } diff --git a/internal/internalpipe/reduce_test.go b/internal/internalpipe/reduce_test.go index 09c1aff..9b88464 100644 --- a/internal/internalpipe/reduce_test.go +++ b/internal/internalpipe/reduce_test.go @@ -11,8 +11,9 @@ func Test_Reduce(t *testing.T) { t.Run("single thread lim set", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 100_000, ValLim: -1, @@ -24,8 +25,9 @@ func Test_Reduce(t *testing.T) { t.Run("seven thread lim set", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: 100_000, ValLim: -1, @@ -37,8 +39,9 @@ func Test_Reduce(t *testing.T) { t.Run("single thread ValLim set", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: 100_000, @@ -50,8 +53,9 @@ func Test_Reduce(t *testing.T) { t.Run("seven thread ValLim set", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: 100_000, @@ -64,9 +68,10 @@ func Test_Reduce(t *testing.T) { t.Run("single thread single value", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { + Fn: func(i int, dst *int) bool { i++ - return &i, false + *dst = i + return false }, Len: -1, ValLim: 1, @@ -79,8 +84,9 @@ func Test_Reduce(t *testing.T) { t.Run("single thread zero value", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: 0, @@ -93,9 +99,10 @@ func Test_Reduce(t *testing.T) { t.Run("seven thread single value", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { + Fn: func(i int, dst *int) bool { i++ - return &i, false + *dst = i + return false }, Len: -1, ValLim: 1, @@ -108,8 +115,9 @@ func Test_Reduce(t *testing.T) { t.Run("seven thread zero value", func(t *testing.T) { p := Pipe[int]{ - Fn: func(i int) (*int, bool) { - return &i, false + Fn: func(i int, dst *int) bool { + *dst = i + return false }, Len: -1, ValLim: 0, diff --git a/internal/internalpipe/sort.go b/internal/internalpipe/sort.go index 8a0e052..821f6ae 100644 --- a/internal/internalpipe/sort.go +++ b/internal/internalpipe/sort.go @@ -12,7 +12,7 @@ func (p Pipe[T]) Sort(less func(*T, *T) bool) Pipe[T] { var sorted []T return Pipe[T]{ - Fn: func(i int) (*T, bool) { + Fn: func(i int, dst *T) bool { once.Do(func() { data := p.Do() if len(data) == 0 { @@ -21,9 +21,10 @@ func (p Pipe[T]) Sort(less func(*T, *T) bool) Pipe[T] { sorted = qsort.Sort(data, less, p.GoroutinesCnt) }) if i >= len(sorted) { - return nil, true + return true } - return &sorted[i], false + *dst = sorted[i] + return false }, Len: p.Len, ValLim: p.ValLim, diff --git a/internal/internalpipe/sum.go b/internal/internalpipe/sum.go index c65e4c7..cef7fb7 100644 --- a/internal/internalpipe/sum.go +++ b/internal/internalpipe/sum.go @@ -4,23 +4,22 @@ import "sync" func sumSingleThread[T any](length int, plus AccumFn[T], fn GeneratorFn[T]) T { var res T - var obj *T - var skipped bool + var obj T i := 0 for ; i < length; i++ { - obj, skipped = fn(i) - if !skipped { - res = *obj - i++ - break + if fn(i, &obj) { + continue } + res = obj + i++ + break } for ; i < length; i++ { - obj, skipped = fn(i) - if !skipped { - res = plus(&res, obj) + if fn(i, &obj) { + continue } + res = plus(&res, &obj) } return res } @@ -54,16 +53,16 @@ func (p Pipe[T]) Sum(plus AccumFn[T]) T { go func(i, lf, rg int) { var res T var ok bool - var obj *T - var skipped bool + var obj T for j := lf; j < rg; j++ { - if obj, skipped = p.Fn(j); !skipped { - if ok { - res = plus(&res, obj) - } else { - res = *obj - ok = true - } + if p.Fn(j, &obj) { + continue + } + if ok { + res = plus(&res, &obj) + } else { + res = obj + ok = true } } parts[i] = partial{val: res, ok: ok} diff --git a/internal/internalpipe/sum_test.go b/internal/internalpipe/sum_test.go index 9f95645..cdc67e9 100644 --- a/internal/internalpipe/sum_test.go +++ b/internal/internalpipe/sum_test.go @@ -4,8 +4,6 @@ import ( "testing" "github.com/stretchr/testify/require" - - "github.com/koss-null/funcfrog/internal/primitive/pointer" ) func Test_Sum(t *testing.T) { @@ -14,8 +12,9 @@ func Test_Sum(t *testing.T) { t.Run("Single thread happy 100k sum", func(t *testing.T) { p := Pipe[float64]{ - Fn: func(i int) (*float64, bool) { - return &a100k[i], false + Fn: func(i int, dst *float64) bool { + *dst = a100k[i] + return false }, Len: len(a100k), ValLim: -1, @@ -31,8 +30,9 @@ func Test_Sum(t *testing.T) { t.Run("Quadro thread happy 100k sum", func(t *testing.T) { p := Pipe[float64]{ - Fn: func(i int) (*float64, bool) { - return &a100k[i], false + Fn: func(i int, dst *float64) bool { + *dst = a100k[i] + return false }, Len: len(a100k), ValLim: -1, @@ -48,8 +48,9 @@ func Test_Sum(t *testing.T) { t.Run("Single thread empty", func(t *testing.T) { p := Pipe[float64]{ - Fn: func(i int) (*float64, bool) { - return pointer.Ref(1.), false + Fn: func(i int, dst *float64) bool { + *dst = 1. + return false }, Len: 0, ValLim: -1, @@ -65,8 +66,9 @@ func Test_Sum(t *testing.T) { t.Run("Quadro thread empty", func(t *testing.T) { p := Pipe[float64]{ - Fn: func(i int) (*float64, bool) { - return pointer.Ref(1.), false + Fn: func(i int, dst *float64) bool { + *dst = 1. + return false }, Len: 0, ValLim: -1, @@ -82,8 +84,9 @@ func Test_Sum(t *testing.T) { t.Run("Single thread 1 elem", func(t *testing.T) { p := Pipe[float64]{ - Fn: func(i int) (*float64, bool) { - return pointer.Ref(100500.), i != 0 + Fn: func(i int, dst *float64) bool { + *dst = 100500. + return i != 0 }, Len: 1, ValLim: -1, @@ -99,8 +102,9 @@ func Test_Sum(t *testing.T) { t.Run("Quadro thread 1 elem", func(t *testing.T) { p := Pipe[float64]{ - Fn: func(i int) (*float64, bool) { - return pointer.Ref(100500.), i != 0 + Fn: func(i int, dst *float64) bool { + *dst = 100500. + return i != 0 }, Len: 1, ValLim: -1, @@ -117,8 +121,9 @@ func Test_Sum(t *testing.T) { t.Run("Single thread multiply", func(t *testing.T) { vals := []float64{1, 2, 3, 4} p := Pipe[float64]{ - Fn: func(i int) (*float64, bool) { - return &vals[i], false + Fn: func(i int, dst *float64) bool { + *dst = vals[i] + return false }, Len: len(vals), ValLim: -1, @@ -134,8 +139,9 @@ func Test_Sum(t *testing.T) { t.Run("Quadro thread multiply", func(t *testing.T) { vals := []float64{1, 2, 3, 4} p := Pipe[float64]{ - Fn: func(i int) (*float64, bool) { - return &vals[i], false + Fn: func(i int, dst *float64) bool { + *dst = vals[i] + return false }, Len: len(vals), ValLim: -1, @@ -151,8 +157,9 @@ func Test_Sum(t *testing.T) { t.Run("Single thread min", func(t *testing.T) { vals := []float64{4, 3, 7, 1, 9} p := Pipe[float64]{ - Fn: func(i int) (*float64, bool) { - return &vals[i], false + Fn: func(i int, dst *float64) bool { + *dst = vals[i] + return false }, Len: len(vals), ValLim: -1, @@ -171,8 +178,9 @@ func Test_Sum(t *testing.T) { t.Run("Quadro thread min", func(t *testing.T) { vals := []float64{4, 3, 7, 1, 9} p := Pipe[float64]{ - Fn: func(i int) (*float64, bool) { - return &vals[i], false + Fn: func(i int, dst *float64) bool { + *dst = vals[i] + return false }, Len: len(vals), ValLim: -1, diff --git a/pkg/pipe/collectors.go b/pkg/pipe/collectors.go index 8fdab1e..db75d6c 100644 --- a/pkg/pipe/collectors.go +++ b/pkg/pipe/collectors.go @@ -6,12 +6,16 @@ import "github.com/koss-null/funcfrog/internal/internalpipe" func Collect[DstT any](p Piper[any]) Piper[DstT] { pp := any(p).(entrails[any]).Entrails() return &Pipe[DstT]{internalpipe.Pipe[DstT]{ - Fn: func(i int) (*DstT, bool) { - if obj, skipped := pp.Fn(i); !skipped { - dst, ok := (*obj).(*DstT) - return dst, !ok + Fn: func(i int, dst *DstT) bool { + var obj any + if pp.Fn(i, &obj) { + return true } - return nil, true + if p, ok := obj.(*DstT); ok { + *dst = *p + return false + } + return true }, Len: pp.Len, ValLim: pp.ValLim, @@ -23,12 +27,16 @@ func Collect[DstT any](p Piper[any]) Piper[DstT] { func CollectNL[DstT any](p PiperNoLen[any]) PiperNoLen[DstT] { pp := any(p).(entrails[any]).Entrails() return &PipeNL[DstT]{internalpipe.Pipe[DstT]{ - Fn: func(i int) (*DstT, bool) { - if obj, skipped := pp.Fn(i); !skipped { - dst, ok := (*obj).(*DstT) - return dst, !ok + Fn: func(i int, dst *DstT) bool { + var obj any + if pp.Fn(i, &obj) { + return true + } + if p, ok := obj.(*DstT); ok { + *dst = *p + return false } - return nil, true + return true }, Len: pp.Len, ValLim: pp.ValLim, diff --git a/pkg/pipe/prefixpipe.go b/pkg/pipe/prefixpipe.go index 373f8ff..57eb8cb 100644 --- a/pkg/pipe/prefixpipe.go +++ b/pkg/pipe/prefixpipe.go @@ -13,12 +13,13 @@ func Map[SrcT any, DstT any]( ) Piper[DstT] { pp := any(p).(entrails[SrcT]).Entrails() return &Pipe[DstT]{internalpipe.Pipe[DstT]{ - Fn: func(i int) (*DstT, bool) { - if obj, skipped := pp.Fn(i); !skipped { - dst := fn(*obj) - return &dst, false + Fn: func(i int, dst *DstT) bool { + var src SrcT + if pp.Fn(i, &src) { + return true } - return nil, true + *dst = fn(src) + return false }, Len: pp.Len, ValLim: pp.ValLim, @@ -33,12 +34,13 @@ func MapNL[SrcT, DstT any]( ) PiperNoLen[DstT] { pp := any(p).(entrails[SrcT]).Entrails() return &PipeNL[DstT]{internalpipe.Pipe[DstT]{ - Fn: func(i int) (*DstT, bool) { - if obj, skipped := pp.Fn(i); !skipped { - dst := fn(*obj) - return &dst, false + Fn: func(i int, dst *DstT) bool { + var src SrcT + if pp.Fn(i, &src) { + return true } - return nil, true + *dst = fn(src) + return false }, Len: pp.Len, ValLim: pp.ValLim, @@ -54,12 +56,17 @@ func MapFilter[SrcT, DstT any]( ) Piper[DstT] { pp := any(p).(entrails[SrcT]).Entrails() return &Pipe[DstT]{internalpipe.Pipe[DstT]{ - Fn: func(i int) (*DstT, bool) { - if obj, skipped := pp.Fn(i); !skipped { - dst, exist := fn(*obj) - return &dst, !exist + Fn: func(i int, dst *DstT) bool { + var src SrcT + if pp.Fn(i, &src) { + return true } - return nil, true + v, exist := fn(src) + if !exist { + return true + } + *dst = v + return false }, Len: pp.Len, ValLim: pp.ValLim, @@ -75,12 +82,17 @@ func MapFilterNL[SrcT, DstT any]( ) PiperNoLen[DstT] { pp := any(p).(entrails[SrcT]).Entrails() return &PipeNL[DstT]{internalpipe.Pipe[DstT]{ - Fn: func(i int) (*DstT, bool) { - if obj, skipped := pp.Fn(i); !skipped { - dst, exist := fn(*obj) - return &dst, !exist + Fn: func(i int, dst *DstT) bool { + var src SrcT + if pp.Fn(i, &src) { + return true + } + v, exist := fn(src) + if !exist { + return true } - return nil, true + *dst = v + return false }, Len: pp.Len, ValLim: pp.ValLim,