Merge pull request #19 from koss-null/1.0.9

v1.0.9

HUGE PERFORMANCE BOOST: kill per-element allocation in Map (destination-pointer generator)

Internal GeneratorFn now writes into a caller-provided slot
(func(i int, dst *T) bool) instead of returning a fresh *T. Stages reuse one
scratch per worker, so Map stops allocating a heap box per element.

1M elements: Map 1,000,005 -> 6 allocs/op (96 -> 70 ms);
MapParallel 1,000,022 -> 25 allocs/op (43.5 -> 32.5 ms).

**No exported signature changed.** Behaviour: First/Any now return a pointer to a
copy (not into the source);

Author: koss-null

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.
 

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{} {

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)
 }

internal/internalpipe/any.go

@@ -1,35 +1,36 @@
 package internalpipe
 
 import (
+	"runtime"
 	"sync"
-	"time"
 )
 
+// hugeLenStep is the base chunk size used when the pipe length is unknown.
 const hugeLenStep = 1 << 15
 
-func anySingleThread[T any](limit int, fn GeneratorFn[T]) *T {
-	var obj *T
-	var skipped bool
+// 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
 	for i := 0; i < limit; i++ {
-		if obj, skipped = fn(i); !skipped {
-			return obj
+		if !fn(i, &obj) {
+			return &obj
 		}
 	}
 	return nil
 }
 
 // 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 +54,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 +85,17 @@ 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
+				var obj T
 				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 !p.Fn(j, &obj) {
+						setObj(&obj)
+						return
 					}
 				}
 			}(i, i+step)

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,

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++