Merge pull request #676 from peco/refactor-dedup-batch-loop

Batch flush refactoring

filter.go

@@ -225,62 +225,34 @@ func acceptAndFilterSerial(ctx context.Context, f filter.Filter, bufsiz int, buf
 	flush := make(chan []line.Line)
 	flushDone := make(chan struct{})
 	go flusher(ctx, f, flush, flushDone, out)
-
 	defer func() { <-flushDone }()
 	defer close(flush)
 
-	flushTicker := time.NewTicker(50 * time.Millisecond)
-	defer flushTicker.Stop()
-
-	start := time.Now()
-	lines := 0
-	for {
-		select {
-		case <-ctx.Done():
-			if pdebug.Enabled {
-				pdebug.Printf("filter received done")
-			}
-			return
-		case <-flushTicker.C:
-			if len(buf) > 0 {
-				flush <- buf
-				buf = buffer.GetLineListBuf()
-			}
-		case v, ok := <-in:
-			if !ok {
-				if pdebug.Enabled {
-					pdebug.Printf("filter input closed (read %d lines, %s since starting accept loop)", lines+len(buf), time.Since(start).String())
-				}
-				if len(buf) > 0 {
-					flush <- buf
-				}
-				return
-			}
-			if pdebug.Enabled {
-				pdebug.Printf("incoming line")
-				lines++
-			}
-			buf = append(buf, v)
-			if len(buf) >= bufsiz {
-				flush <- buf
-				buf = buffer.GetLineListBuf()
-			}
-		}
-	}
+	batchAndFlush(ctx, bufsiz, buf, in, flush, func(b []line.Line) []line.Line { return b })
 }
 
 func acceptAndFilterParallel(ctx context.Context, f filter.Filter, bufsiz int, buf []line.Line, in <-chan line.Line, out pipeline.ChanOutput) {
 	flush := make(chan orderedChunk)
 	flushDone := make(chan struct{})
 	go parallelFlusher(ctx, f, flush, flushDone, out)
-
 	defer func() { <-flushDone }()
 	defer close(flush)
 
+	seq := 0
+	batchAndFlush(ctx, bufsiz, buf, in, flush, func(b []line.Line) orderedChunk {
+		chunk := orderedChunk{seq: seq, lines: b}
+		seq++
+		return chunk
+	})
+}
+
+// batchAndFlush reads lines from in, batches them into slices of up to bufsiz,
+// and sends each batch to flushCh via the wrap function. Batches are flushed
+// when full or every 50ms, whichever comes first.
+func batchAndFlush[T any](ctx context.Context, bufsiz int, buf []line.Line, in <-chan line.Line, flushCh chan T, wrap func([]line.Line) T) {
 	flushTicker := time.NewTicker(50 * time.Millisecond)
 	defer flushTicker.Stop()
 
-	seq := 0
 	start := time.Now()
 	lines := 0
 	for {
@@ -292,8 +264,7 @@ func acceptAndFilterParallel(ctx context.Context, f filter.Filter, bufsiz int, b
 			return
 		case <-flushTicker.C:
 			if len(buf) > 0 {
-				flush <- orderedChunk{seq: seq, lines: buf}
-				seq++
+				flushCh <- wrap(buf)
 				buf = buffer.GetLineListBuf()
 			}
 		case v, ok := <-in:
@@ -302,7 +273,7 @@ func acceptAndFilterParallel(ctx context.Context, f filter.Filter, bufsiz int, b
 					pdebug.Printf("filter input closed (read %d lines, %s since starting accept loop)", lines+len(buf), time.Since(start).String())
 				}
 				if len(buf) > 0 {
-					flush <- orderedChunk{seq: seq, lines: buf}
+					flushCh <- wrap(buf)
 				}
 				return
 			}
@@ -312,8 +283,7 @@ func acceptAndFilterParallel(ctx context.Context, f filter.Filter, bufsiz int, b
 			}
 			buf = append(buf, v)
 			if len(buf) >= bufsiz {
-				flush <- orderedChunk{seq: seq, lines: buf}
-				seq++
+				flushCh <- wrap(buf)
 				buf = buffer.GetLineListBuf()
 			}
 		}

filter_incremental_test.go

@@ -3,6 +3,7 @@ package peco
 import (
 	"context"
 	"fmt"
+	"runtime"
 	"testing"
 
 	"github.com/peco/peco/filter"
@@ -131,6 +132,83 @@ func TestMemoryBufferSourceCancellation(t *testing.T) {
 	require.LessOrEqual(t, count, 10000)
 }
 
+// TestAcceptAndFilterSerial exercises the serial (non-parallel) batching path
+// through AcceptAndFilter. It sends lines through a channel and verifies
+// that the filter is applied correctly and all matching lines appear in the output.
+func TestAcceptAndFilterSerial(t *testing.T) {
+	f := filter.NewFuzzy(true) // Fuzzy with sortLongest=true does NOT support parallel
+	require.False(t, f.SupportsParallel(), "Fuzzy with sortLongest should not support parallel")
+
+	inputLines := []line.Line{
+		line.NewRaw(0, "alpha bravo", false, false),
+		line.NewRaw(1, "charlie delta", false, false),
+		line.NewRaw(2, "alpha charlie", false, false),
+		line.NewRaw(3, "echo foxtrot", false, false),
+		line.NewRaw(4, "alpha delta", false, false),
+	}
+
+	ctx := f.NewContext(context.Background(), "alpha")
+	in := make(chan line.Line, len(inputLines))
+	for _, l := range inputLines {
+		in <- l
+	}
+	close(in)
+
+	out := make(chan line.Line, len(inputLines))
+	AcceptAndFilter(ctx, f, 0, in, pipeline.ChanOutput(out))
+
+	var got []string
+	for l := range out {
+		got = append(got, l.DisplayString())
+	}
+
+	// Fuzzy with sortLongest sorts by: longer match > earlier match > shorter line.
+	// All three match "alpha" at position 0 with length 5, so the tiebreaker is
+	// line length (shorter first). "alpha bravo" and "alpha delta" tie at 11 chars,
+	// so sort.SliceStable preserves their input order.
+	require.Equal(t, []string{"alpha bravo", "alpha delta", "alpha charlie"}, got)
+}
+
+// TestAcceptAndFilterParallel exercises the parallel batching path.
+// IgnoreCase (a Regexp filter) supports parallel execution.
+func TestAcceptAndFilterParallel(t *testing.T) {
+	if runtime.GOMAXPROCS(0) < 2 {
+		t.Skip("parallel path requires GOMAXPROCS >= 2")
+	}
+
+	f := filter.NewIgnoreCase()
+	require.True(t, f.SupportsParallel(), "IgnoreCase should support parallel")
+
+	inputLines := []line.Line{
+		line.NewRaw(0, "foobar test", false, false),
+		line.NewRaw(1, "football game", false, false),
+		line.NewRaw(2, "something else", false, false),
+		line.NewRaw(3, "foobaz entry", false, false),
+		line.NewRaw(4, "barfoo other", false, false),
+		line.NewRaw(5, "the foobird flies", false, false),
+		line.NewRaw(6, "no match here", false, false),
+	}
+
+	ctx := f.NewContext(context.Background(), "foo")
+	in := make(chan line.Line, len(inputLines))
+	for _, l := range inputLines {
+		in <- l
+	}
+	close(in)
+
+	out := make(chan line.Line, len(inputLines))
+	AcceptAndFilter(ctx, f, 0, in, pipeline.ChanOutput(out))
+
+	var got []string
+	for l := range out {
+		got = append(got, l.DisplayString())
+	}
+
+	// Parallel preserves input order via ordered chunks — verify exact order
+	expected := []string{"foobar test", "football game", "foobaz entry", "barfoo other", "the foobird flies"}
+	require.Equal(t, expected, got)
+}
+
 func TestIncrementalFiltering(t *testing.T) {
 	// Test that filtering "foo" then "foob" produces correct results
 	// and the second query runs on fewer lines