improved First parallel timings

internal/internalpipe/first.go

@@ -113,15 +113,24 @@ func first[T any](limit, grtCnt int, fn func(i int) (*T, bool)) (f *T) {
 
 			done := res.ctx.Done()
 			for j := lf; j < rg; j++ {
-				select {
-				case <-done:
-					return
-				default:
+				// 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)
 						return
 					}
+				} else {
+					select {
+					case <-done:
+						return
+					default:
+						val, skipped := fn(j)
+						if !skipped {
+							res.setVal(val, stepCnt)
+							return
+						}
+					}
 				}
 			}
 			res.stepDone(stepCnt)

perf/perf_test.go

@@ -222,3 +222,53 @@ func BenchmarkAnyFor(b *testing.B) {
 		}
 	}
 }
+
+func BenchmarkFirst(b *testing.B) {
+	b.StopTimer()
+	input := make([]int, 1_000_000)
+	for i := 0; i < len(input); i++ {
+		input[i] = i
+	}
+	b.StartTimer()
+
+	for j := 0; j < b.N; j++ {
+		pipe := pipe.Slice(input).Filter(func(x *int) bool { return *x > 5_000_00 })
+		result := pipe.First()
+		_ = result
+	}
+}
+
+func BenchmarkFirstParallel(b *testing.B) {
+	b.StopTimer()
+	input := make([]int, 1_000_000)
+	for i := 0; i < len(input); i++ {
+		input[i] = i
+	}
+	b.StartTimer()
+
+	for j := 0; j < b.N; j++ {
+		pipe := pipe.Slice(input).
+			Parallel(uint16(runtime.NumCPU())).
+			Filter(func(x *int) bool { return *x > 5_000_00 })
+		result := pipe.First()
+		_ = result
+	}
+}
+
+func BenchmarkFirstFor(b *testing.B) {
+	b.StopTimer()
+	input := make([]int, 1_000_000)
+	for i := 0; i < len(input); i++ {
+		input[i] = i
+	}
+	b.StartTimer()
+
+	for j := 0; j < b.N; j++ {
+		for i := 0; i < len(input); i++ {
+			if input[i] > 5_000_00 {
+				_ = input[i]
+				break
+			}
+		}
+	}
+}