diff --git a/filter.go b/filter.go index 901de05..26168a4 100644 --- a/filter.go +++ b/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() } } diff --git a/filter_incremental_test.go b/filter_incremental_test.go index 596aca0..21f55e9 100644 --- a/filter_incremental_test.go +++ b/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