utils-shell/peco.peco
1
0
Fork 0
mirror of https://github.com/peco/peco.git synced 2026-09-10 07:16:29 -04:00
Code Issues Actions Packages Projects Releases Wiki Activity

use types in pipelines

Browse source
This commit is contained in:
Daisuke Maki 2026-02-16 23:51:36 +09:00
parent c2c812db24
commit f5f6262d82
15 changed files with 194 additions and 326 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

107
buffer.go
View file

@ -126,7 +126,7 @@ func (mb *MemoryBuffer) Done() <-chan struct{} {
return mb.done
}
func (mb *MemoryBuffer) Accept(ctx context.Context, in chan interface{}, _ pipeline.ChanOutput) {
func (mb *MemoryBuffer) Accept(ctx context.Context, in <-chan line.Line, _ pipeline.ChanOutput) {
if pdebug.Enabled {
g := pdebug.Marker("MemoryBuffer.Accept")
defer g.End()
@ -145,62 +145,47 @@ func (mb *MemoryBuffer) Accept(ctx context.Context, in chan interface{}, _ pipel
pdebug.Printf("MemoryBuffer received context done")
}
return
case v := <-in:
switch v := v.(type) {
case error:
if pipeline.IsEndMark(v) {
if pdebug.Enabled {
pdebug.Printf("MemoryBuffer received end mark (read %d lines, %s since starting accept loop)", len(mb.lines), time.Since(start).String())
}
// Flush remaining batch
if len(batch) > 0 {
case v, ok := <-in:
if !ok {
if pdebug.Enabled {
pdebug.Printf("MemoryBuffer input channel closed (read %d lines, %s since starting accept loop)", len(mb.lines)+len(batch), time.Since(start).String())
}
// Flush remaining batch
if len(batch) > 0 {
mb.mutex.Lock()
mb.lines = append(mb.lines, batch...)
mb.mutex.Unlock()
}
return
}
batch = append(batch, v)
// Drain any additional ready values without blocking
drain:
for {
select {
case v2, ok2 := <-in:
if !ok2 {
if pdebug.Enabled {
pdebug.Printf("MemoryBuffer input channel closed (read %d lines, %s since starting accept loop)", len(mb.lines)+len(batch), time.Since(start).String())
}
mb.mutex.Lock()
mb.lines = append(mb.lines, batch...)
mb.mutex.Unlock()
return
}
return
batch = append(batch, v2)
default:
break drain
}
case []line.Line:
batch = append(batch, v...)
mb.mutex.Lock()
mb.lines = append(mb.lines, batch...)
mb.mutex.Unlock()
batch = batch[:0]
case line.Line:
batch = append(batch, v)
// Drain any additional ready values without blocking
drain:
for {
select {
case v2 := <-in:
switch v2 := v2.(type) {
case error:
if pipeline.IsEndMark(v2) {
if pdebug.Enabled {
pdebug.Printf("MemoryBuffer received end mark (read %d lines, %s since starting accept loop)", len(mb.lines)+len(batch), time.Since(start).String())
}
mb.mutex.Lock()
mb.lines = append(mb.lines, batch...)
mb.mutex.Unlock()
return
}
case []line.Line:
batch = append(batch, v2...)
case line.Line:
batch = append(batch, v2)
}
default:
break drain
}
}
// Flush the batch
mb.mutex.Lock()
mb.lines = append(mb.lines, batch...)
mb.mutex.Unlock()
batch = batch[:0]
}
// Flush the batch
mb.mutex.Lock()
mb.lines = append(mb.lines, batch...)
mb.mutex.Unlock()
batch = batch[:0]
}
}
}
@ -390,31 +375,23 @@ func NewMemoryBufferSource(buf *MemoryBuffer) *MemoryBufferSource {
return &MemoryBufferSource{buf: buf}
}
// sourceBatchSize is the number of lines sent per batch from source to
// the filter stage. Larger batches reduce channel operations but increase
// latency to first result. 1024 is a good balance.
const sourceBatchSize = 1024
// Start iterates through the MemoryBuffer's lines and sends them in
// batches to the output channel, implementing pipeline.Source.
// Start iterates through the MemoryBuffer's lines and sends them
// individually to the output channel, implementing pipeline.Source.
// The output channel is closed when all lines have been sent.
func (s *MemoryBufferSource) Start(ctx context.Context, out pipeline.ChanOutput) {
defer out.SendEndMark(ctx, "end of memory buffer source")
defer close(out)
s.buf.mutex.RLock()
lines := s.buf.lines
s.buf.mutex.RUnlock()
for i := 0; i < len(lines); i += sourceBatchSize {
for _, l := range lines {
select {
case <-ctx.Done():
return
default:
}
end := i + sourceBatchSize
if end > len(lines) {
end = len(lines)
}
out.Send(ctx, lines[i:end])
out.Send(ctx, l)
}
}

12
cmd/filterbench/main.go
View file

@ -240,7 +240,7 @@ func benchDirect(name string, f filter.Filter, lines []line.Line, query string)
// Count matches using a channel consumer
matchCount := 0
ch := make(chan interface{}, 4096)
ch := make(chan line.Line, 4096)
done := make(chan struct{})
go func() {
@ -387,16 +387,14 @@ func benchIncremental(name string, f filter.Filter, allLines []line.Line, querie
// collectMatchedLines runs the filter and returns matched lines as a slice.
func collectMatchedLines(f filter.Filter, lines []line.Line, query string) []line.Line {
ctx := f.NewContext(context.Background(), query)
ch := make(chan interface{}, 4096)
ch := make(chan line.Line, 4096)
done := make(chan struct{})
var matched []line.Line
go func() {
defer close(done)
for v := range ch {
if l, ok := v.(line.Line); ok {
matched = append(matched, l)
}
for l := range ch {
matched = append(matched, l)
}
}()
@ -452,7 +450,7 @@ type directFilterProcessor struct {
query string
}
func (p *directFilterProcessor) Accept(ctx context.Context, in chan interface{}, out pipeline.ChanOutput) {
func (p *directFilterProcessor) Accept(ctx context.Context, in <-chan line.Line, out pipeline.ChanOutput) {
peco.AcceptAndFilter(ctx, p.filter, 0, in, out)
}

115
filter.go
View file

@ -25,7 +25,7 @@ func newFilterProcessor(f filter.Filter, q string, bufSize int) *filterProcessor
}
}
func (fp *filterProcessor) Accept(ctx context.Context, in chan interface{}, out pipeline.ChanOutput) {
func (fp *filterProcessor) Accept(ctx context.Context, in <-chan line.Line, out pipeline.ChanOutput) {
acceptAndFilter(ctx, fp.filter, fp.bufSize, in, out)
}
@ -51,7 +51,7 @@ func flusher(ctx context.Context, f filter.Filter, incoming chan []line.Line, do
}
defer close(done)
defer out.SendEndMark(ctx, "end of filter")
defer close(out)
for {
select {
@ -77,7 +77,7 @@ func parallelFlusher(ctx context.Context, f filter.Filter, incoming chan ordered
}
defer close(done)
defer out.SendEndMark(ctx, "end of filter")
defer close(out)
numWorkers := runtime.GOMAXPROCS(0)
if numWorkers < 1 {
@ -114,16 +114,14 @@ func parallelFlusher(ctx context.Context, f filter.Filter, incoming chan ordered
} else {
// Fallback: use channel-based Apply for filters that
// don't implement Collector (e.g. ExternalCmd)
collectCh := make(chan interface{}, len(chunk.lines))
collectCh := make(chan line.Line, len(chunk.lines))
go func(chunk orderedChunk) {
f.Apply(ctx, chunk.lines, pipeline.ChanOutput(collectCh))
close(collectCh)
}(chunk)
matched = make([]line.Line, 0, len(chunk.lines)/2)
for v := range collectCh {
if l, ok := v.(line.Line); ok {
matched = append(matched, l)
}
for l := range collectCh {
matched = append(matched, l)
}
}
@ -199,11 +197,11 @@ func parallelFlusher(ctx context.Context, f filter.Filter, incoming chan ordered
// AcceptAndFilter is the exported entry point for the filter pipeline stage.
// It batches incoming lines and dispatches them to the filter, using parallel
// workers when the filter supports it.
func AcceptAndFilter(ctx context.Context, f filter.Filter, configBufSize int, in chan interface{}, out pipeline.ChanOutput) {
func AcceptAndFilter(ctx context.Context, f filter.Filter, configBufSize int, in <-chan line.Line, out pipeline.ChanOutput) {
acceptAndFilter(ctx, f, configBufSize, in, out)
}
func acceptAndFilter(ctx context.Context, f filter.Filter, configBufSize int, in chan interface{}, out pipeline.ChanOutput) {
func acceptAndFilter(ctx context.Context, f filter.Filter, configBufSize int, in <-chan line.Line, out pipeline.ChanOutput) {
useParallel := f.SupportsParallel() && runtime.GOMAXPROCS(0) > 1
buf := buffer.GetLineListBuf()
@ -223,7 +221,7 @@ func acceptAndFilter(ctx context.Context, f filter.Filter, configBufSize int, in
}
}
func acceptAndFilterSerial(ctx context.Context, f filter.Filter, bufsiz int, buf []line.Line, in chan interface{}, out pipeline.ChanOutput) {
func acceptAndFilterSerial(ctx context.Context, f filter.Filter, bufsiz int, buf []line.Line, in <-chan line.Line, out pipeline.ChanOutput) {
flush := make(chan []line.Line)
flushDone := make(chan struct{})
go flusher(ctx, f, flush, flushDone, out)
@ -248,43 +246,30 @@ func acceptAndFilterSerial(ctx context.Context, f filter.Filter, bufsiz int, buf
flush <- buf
buf = buffer.GetLineListBuf()
}
case v := <-in:
switch v := v.(type) {
case error:
if pipeline.IsEndMark(v) {
if pdebug.Enabled {
pdebug.Printf("filter received end mark (read %d lines, %s since starting accept loop)", lines+len(buf), time.Since(start).String())
}
if len(buf) > 0 {
flush <- buf
}
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
case line.Line:
if pdebug.Enabled {
pdebug.Printf("incoming line")
lines++
}
buf = append(buf, v)
if len(buf) >= bufsiz {
flush <- buf
buf = buffer.GetLineListBuf()
}
case []line.Line:
if pdebug.Enabled {
lines += len(v)
}
buf = append(buf, v...)
if len(buf) >= bufsiz {
flush <- buf
buf = buffer.GetLineListBuf()
}
}
if pdebug.Enabled {
pdebug.Printf("incoming line")
lines++
}
buf = append(buf, v)
if len(buf) >= bufsiz {
flush <- buf
buf = buffer.GetLineListBuf()
}
}
}
}
func acceptAndFilterParallel(ctx context.Context, f filter.Filter, bufsiz int, buf []line.Line, in chan interface{}, out pipeline.ChanOutput) {
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)
@ -311,39 +296,25 @@ func acceptAndFilterParallel(ctx context.Context, f filter.Filter, bufsiz int, b
seq++
buf = buffer.GetLineListBuf()
}
case v := <-in:
switch v := v.(type) {
case error:
if pipeline.IsEndMark(v) {
if pdebug.Enabled {
pdebug.Printf("filter received end mark (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}
}
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 <- orderedChunk{seq: seq, lines: buf}
}
return
case line.Line:
if pdebug.Enabled {
pdebug.Printf("incoming line")
lines++
}
buf = append(buf, v)
if len(buf) >= bufsiz {
flush <- orderedChunk{seq: seq, lines: buf}
seq++
buf = buffer.GetLineListBuf()
}
case []line.Line:
if pdebug.Enabled {
lines += len(v)
}
buf = append(buf, v...)
if len(buf) >= bufsiz {
flush <- orderedChunk{seq: seq, lines: buf}
seq++
buf = buffer.GetLineListBuf()
}
}
if pdebug.Enabled {
pdebug.Printf("incoming line")
lines++
}
buf = append(buf, v)
if len(buf) >= bufsiz {
flush <- orderedChunk{seq: seq, lines: buf}
seq++
buf = buffer.GetLineListBuf()
}
}
}

14
filter/base_test.go
View file

@ -47,15 +47,13 @@ func TestApplyAndApplyCollectConsistency(t *testing.T) {
defer cancel()
// Collect via Apply (channel path)
ch := make(chan interface{}, len(lines)+1)
ch := make(chan line.Line, len(lines)+1)
err := tt.filter.Apply(ctx, lines, pipeline.ChanOutput(ch))
require.NoError(t, err, "Apply should succeed")
close(ch)
var applyResults []string
for v := range ch {
l, ok := v.(line.Line)
require.True(t, ok, "channel value should be line.Line")
for l := range ch {
applyResults = append(applyResults, l.DisplayString())
}
@ -126,17 +124,15 @@ func TestNewContextStoresQuery(t *testing.T) {
ctx := tt.filter.NewContext(context.Background(), "test-query")
// The query should be stored in context — verify by running Apply
// with a line that matches "test-query"
ch := make(chan interface{}, 2)
ch := make(chan line.Line, 2)
lines := makeLines("this is a test-query line")
err := tt.filter.Apply(ctx, lines, pipeline.ChanOutput(ch))
require.NoError(t, err)
close(ch)
var results []line.Line
for v := range ch {
if l, ok := v.(line.Line); ok {
results = append(results, l)
}
for l := range ch {
results = append(results, l)
}
require.Len(t, results, 1, "query stored by NewContext should be used by Apply")
})

2
filter/external.go
View file

@ -30,7 +30,7 @@ func NewExternalCmd(name string, cmd string, args []string, threshold int, idgen
enableSep: enableSep,
idgen: idgen,
name: name,
outCh: pipeline.ChanOutput(make(chan interface{})),
outCh: pipeline.ChanOutput(make(chan line.Line)),
thresholdBufsiz: threshold,
}
}

14
filter/external_test.go
View file

@ -31,13 +31,11 @@ func collectOutput(ctx context.Context, out pipeline.ChanOutput) []line.Line {
select {
case <-ctx.Done():
return results
case v, ok := <-out.OutCh():
case l, ok := <-out.OutCh():
if !ok {
return results
}
if l, ok := v.(line.Line); ok {
results = append(results, l)
}
results = append(results, l)
}
}
}
@ -54,7 +52,7 @@ func TestExternalCmd_CancelCleansUpGoroutine(t *testing.T) {
// it completes quickly. Instead, use "sleep" which blocks for a long time.
ecf := NewExternalCmd("sleep", "sleep", []string{"60"}, 0, idgen, false)
ctx, cancel := context.WithCancel(ecf.NewContext(context.Background(), "test"))
out := pipeline.ChanOutput(make(chan interface{}, 256))
out := pipeline.ChanOutput(make(chan line.Line, 256))
// Record goroutine count before Apply
runtime.GC()
@ -112,7 +110,7 @@ func TestExternalCmdFilter_NullSep(t *testing.T) {
ecf := NewExternalCmd("grep", "grep", []string{"ap"}, 0, idgen, true)
ctx := ecf.NewContext(context.Background(), "ap")
out := pipeline.ChanOutput(make(chan interface{}, 256))
out := pipeline.ChanOutput(make(chan line.Line, 256))
var results []line.Line
done := make(chan struct{})
@ -149,7 +147,7 @@ func TestExternalCmdFilter_NullSep(t *testing.T) {
ecf := NewExternalCmd("grep", "grep", []string{"ap"}, 0, idgen, false)
ctx := ecf.NewContext(context.Background(), "ap")
out := pipeline.ChanOutput(make(chan interface{}, 256))
out := pipeline.ChanOutput(make(chan line.Line, 256))
var results []line.Line
done := make(chan struct{})
@ -184,7 +182,7 @@ func TestExternalCmdFilter_NullSep(t *testing.T) {
ecf := NewExternalCmd("grep", "grep", []string{"-F", "$QUERY"}, 0, idgen, true)
ctx := ecf.NewContext(context.Background(), "dup")
out := pipeline.ChanOutput(make(chan interface{}, 256))
out := pipeline.ChanOutput(make(chan line.Line, 256))
var results []line.Line
done := make(chan struct{})

25
filter/filter_test.go
View file

@ -60,7 +60,7 @@ func testFuzzy(octx context.Context, t *testing.T, filter Filter) {
ctx, cancel := context.WithTimeout(filter.NewContext(octx, v.query), 10*time.Second)
defer cancel()
ch := make(chan interface{}, 1)
ch := make(chan line.Line, 1)
l := line.NewRaw(uint64(i), v.input, false, false)
err := filter.Apply(ctx, []line.Line{l}, pipeline.ChanOutput(ch))
if !assert.NoError(t, err, `filter.Apply should succeed`) {
@ -73,10 +73,6 @@ func testFuzzy(octx context.Context, t *testing.T, filter Filter) {
return
}
if !assert.Implements(t, (*line.Line)(nil), l, "result is a line") {
return
}
t.Logf("%#v", l.(indexer).Indices())
case <-ctx.Done():
if !assert.False(t, v.selected, "did NOT expect to timeout") { // shouldn't happen if we're expecting a result
@ -193,20 +189,17 @@ func testFuzzyLongest(octx context.Context, t *testing.T, filter Filter) {
}
var actual []string
lc := make(chan interface{})
lc := make(chan line.Line)
ec := make(chan error)
go func() {
ec <- filter.Apply(ctx, lines, lc)
ec <- filter.Apply(ctx, lines, pipeline.ChanOutput(lc))
}()
OUTER:
for {
select {
case l := <-lc:
if !assert.Implements(t, (*line.Line)(nil), l, "result is a line") {
return
}
actual = append(actual, l.(line.Line).DisplayString())
actual = append(actual, l.DisplayString())
case err := <-ec:
if !assert.NoError(t, err, `filter.Apply should succeed`) {
return
@ -301,15 +294,13 @@ func collectFilterResults(t *testing.T, f Filter, query string, inputLines []lin
ctx, cancel := context.WithTimeout(f.NewContext(context.Background(), query), 10*time.Second)
defer cancel()
ch := make(chan interface{}, len(inputLines)+1)
ch := make(chan line.Line, len(inputLines)+1)
err := f.Apply(ctx, inputLines, pipeline.ChanOutput(ch))
require.NoError(t, err, "filter.Apply should succeed")
close(ch)
var results []line.Line
for v := range ch {
l, ok := v.(line.Line)
require.True(t, ok, "result should be a line.Line")
for l := range ch {
results = append(results, l)
}
return results
@ -537,10 +528,10 @@ func testFuzzyMatch(octx context.Context, t *testing.T, filter Filter) {
defer cancel()
filter := NewFuzzy(v.sort)
lc := make(chan interface{})
lc := make(chan line.Line)
ec := make(chan error)
go func() {
ec <- filter.Apply(ctx, []line.Line{line.NewRaw(uint64(i), v.input, false, false)}, lc)
ec <- filter.Apply(ctx, []line.Line{line.NewRaw(uint64(i), v.input, false, false)}, pipeline.ChanOutput(lc))
}()
OUTER:

18
filter/parallel_test.go
View file

@ -61,16 +61,14 @@ func TestParallelFilterProducesSameResults(t *testing.T) {
ctx := ft.filter.NewContext(context.Background(), ft.query)
// Run sequentially
seqCh := make(chan interface{}, numLines)
seqCh := make(chan line.Line, numLines)
err := ft.filter.Apply(ctx, lines, pipeline.ChanOutput(seqCh))
require.NoError(t, err)
close(seqCh)
var seqResults []string
for v := range seqCh {
if l, ok := v.(line.Line); ok {
seqResults = append(seqResults, l.DisplayString())
}
for l := range seqCh {
seqResults = append(seqResults, l.DisplayString())
}
// Run on chunks (simulating parallel) - split into multiple chunks
@ -83,15 +81,13 @@ func TestParallelFilterProducesSameResults(t *testing.T) {
}
chunk := lines[start:end]
ch := make(chan interface{}, len(chunk))
ch := make(chan line.Line, len(chunk))
err := ft.filter.Apply(ctx, chunk, pipeline.ChanOutput(ch))
require.NoError(t, err)
close(ch)
for v := range ch {
if l, ok := v.(line.Line); ok {
parResults = append(parResults, l.DisplayString())
}
for l := range ch {
parResults = append(parResults, l.DisplayString())
}
}
@ -119,7 +115,7 @@ func TestParallelFilterContextCancellation(t *testing.T) {
cancel()
}()
ch := make(chan interface{}, numLines)
ch := make(chan line.Line, numLines)
err := f.Apply(ctx, lines, pipeline.ChanOutput(ch))
// Should return context error (cancelled)

4
filter/regexp.go
View file

@ -89,7 +89,7 @@ func NewRegexp() *Regexp {
flags: regexpFlagList(defaultFlags),
quotemeta: false,
name: "Regexp",
outCh: pipeline.ChanOutput(make(chan interface{})),
outCh: pipeline.ChanOutput(make(chan line.Line)),
}
rf.applyFn = rf.applyInternal
return rf
@ -103,7 +103,7 @@ func NewIRegexp() *Regexp {
return rf
}
func (rf *Regexp) OutCh() <-chan interface{} {
func (rf *Regexp) OutCh() <-chan line.Line {
rf.mutex.Lock()
defer rf.mutex.Unlock()
return rf.outCh

39
filter_incremental_test.go
View file

@ -91,25 +91,13 @@ func TestMemoryBufferSource(t *testing.T) {
// Collect lines from source
ctx := context.Background()
out := make(chan interface{}, len(expected)+1) // +1 for end mark
out := make(chan line.Line, len(expected))
go src.Start(ctx, pipeline.ChanOutput(out))
var got []string
for v := range out {
switch v := v.(type) {
case error:
if pipeline.IsEndMark(v) {
goto done
}
case []line.Line:
for _, l := range v {
got = append(got, l.DisplayString())
}
case line.Line:
got = append(got, v.DisplayString())
}
for l := range out {
got = append(got, l.DisplayString())
}
done:
require.Equal(t, expected, got, "MemoryBufferSource should iterate all lines in order")
}
@ -122,7 +110,7 @@ func TestMemoryBufferSourceCancellation(t *testing.T) {
src := NewMemoryBufferSource(mb)
ctx, cancel := context.WithCancel(context.Background())
out := make(chan interface{}, 100)
out := make(chan line.Line, 100)
done := make(chan struct{})
go func() {
src.Start(ctx, pipeline.ChanOutput(out))
@ -133,8 +121,7 @@ func TestMemoryBufferSourceCancellation(t *testing.T) {
cancel()
<-done
// Should have stopped early (not all 10000 lines)
close(out)
// Source closed the channel; drain to count lines sent
count := 0
for range out {
count++
@ -162,16 +149,14 @@ func TestIncrementalFiltering(t *testing.T) {
// First query: "foo"
ctx1 := f.NewContext(context.Background(), "foo")
ch1 := make(chan interface{}, len(allLines))
ch1 := make(chan line.Line, len(allLines))
err := f.Apply(ctx1, allLines, pipeline.ChanOutput(ch1))
require.NoError(t, err)
close(ch1)
var firstResults []line.Line
for v := range ch1 {
if l, ok := v.(line.Line); ok {
firstResults = append(firstResults, l)
}
for l := range ch1 {
firstResults = append(firstResults, l)
}
// Should match: foobar, football, barfoo, foobaz, foobird
@ -179,16 +164,14 @@ func TestIncrementalFiltering(t *testing.T) {
// Second query: "foob" on first results only
ctx2 := f.NewContext(context.Background(), "foob")
ch2 := make(chan interface{}, len(firstResults))
ch2 := make(chan line.Line, len(firstResults))
err = f.Apply(ctx2, firstResults, pipeline.ChanOutput(ch2))
require.NoError(t, err)
close(ch2)
var secondResults []line.Line
for v := range ch2 {
if l, ok := v.(line.Line); ok {
secondResults = append(secondResults, l)
}
for l := range ch2 {
secondResults = append(secondResults, l)
}
// Should match: foobar, foobaz, foobird (not football, not barfoo)

2
issues_test.go
View file

@ -177,6 +177,7 @@ type sliceSource struct {
}
func (s *sliceSource) Start(ctx context.Context, out pipeline.ChanOutput) {
defer close(out)
for _, l := range s.lines {
select {
case <-ctx.Done():
@ -185,7 +186,6 @@ func (s *sliceSource) Start(ctx context.Context, out pipeline.ChanOutput) {
out.Send(ctx, l)
}
}
out.SendEndMark(ctx, "end of sliceSource")
}
func (s *sliceSource) Reset() {}

29
pipeline/interface.go
View file

@ -4,21 +4,14 @@ import (
"sync"
"context"
"github.com/peco/peco/line"
)
// EndMarker is an interface for things that tell us the input
// sequence has ended
type EndMarker interface {
error
EndMark() bool
}
// EndMark is a dummy struct that gets send as an EOL mark of sorts
type EndMark struct{}
type Source interface {
// Start should be able to be called repeatedly, producing the
// same data to be consumed by the chained Acceptors
// same data to be consumed by the chained Acceptors.
// The implementation must close out when done sending.
Start(context.Context, ChanOutput)
Reset()
@ -30,9 +23,10 @@ type Suspender interface {
}
// Acceptor is an object that can accept input, and send to
// an optional output
// an optional output. The implementation must close out (if non-nil)
// when in is exhausted or the context is cancelled.
type Acceptor interface {
Accept(context.Context, chan interface{}, ChanOutput)
Accept(context.Context, <-chan line.Line, ChanOutput)
}
// Destination is a special case Acceptor that has no more Acceptors
@ -52,9 +46,6 @@ type Pipeline struct {
dst Destination
}
type Output interface {
Send(context.Context, interface{}) error
}
// ChanOutput is an alias to `chan interface{}`
type ChanOutput chan interface{}
// ChanOutput is a typed channel for sending line.Line values between
// pipeline stages.
type ChanOutput chan line.Line

46
pipeline/pipeline.go
View file

@ -4,39 +4,22 @@ package pipeline
import (
"context"
"errors"
"fmt"
pdebug "github.com/lestrrat-go/pdebug"
"github.com/peco/peco/line"
)
// EndMark returns true
func (e EndMark) EndMark() bool {
return true
}
// Error returns the error string "end of input"
func (e EndMark) Error() string {
return "end of input"
}
// IsEndMark is an utility function that checks if the given error
// object is an EndMark
func IsEndMark(err error) bool {
var em EndMarker
if errors.As(err, &em) {
return em.EndMark()
}
return false
}
func NilOutput(ctx context.Context) ChanOutput {
ch := make(chan interface{})
ch := make(chan line.Line)
go func() {
for {
select {
case <-ctx.Done():
return
case <-ch:
case _, ok := <-ch:
if !ok {
return
}
}
}
}()
@ -45,14 +28,14 @@ func NilOutput(ctx context.Context) ChanOutput {
}
// OutCh returns the channel that acceptors can listen to
func (oc ChanOutput) OutCh() <-chan interface{} {
func (oc ChanOutput) OutCh() <-chan line.Line {
return oc
}
// Send sends the data `v` through this channel. It blocks until the value
// is sent or the context is cancelled. This avoids the timer allocation
// overhead of the previous implementation while still supporting cancellation.
func (oc ChanOutput) Send(ctx context.Context, v interface{}) (err error) {
func (oc ChanOutput) Send(ctx context.Context, v line.Line) (err error) {
if oc == nil {
return errors.New("nil channel")
}
@ -65,15 +48,6 @@ func (oc ChanOutput) Send(ctx context.Context, v interface{}) (err error) {
}
}
// SendEndMark sends an end mark. If ctx is cancelled, the end mark is
// dropped since all pipeline stages are shutting down via context anyway.
func (oc ChanOutput) SendEndMark(ctx context.Context, s string) error {
if err := oc.Send(ctx, fmt.Errorf("%s: %w", s, EndMark{})); err != nil {
return fmt.Errorf("failed to send end mark: %w", err)
}
return nil
}
// New creates a new Pipeline
func New() *Pipeline {
return &Pipeline{
@ -139,12 +113,12 @@ func (p *Pipeline) Run(ctx context.Context) (err error) {
// Use buffered channels between pipeline stages to allow pipelining
const chanBufSize = 256
prevCh := ChanOutput(make(chan interface{}, chanBufSize))
prevCh := ChanOutput(make(chan line.Line, chanBufSize))
go p.dst.Accept(ctx, prevCh, nil)
for i := len(p.nodes) - 1; i >= 0; i-- {
cur := p.nodes[i]
ch := make(chan interface{}, chanBufSize)
ch := make(chan line.Line, chanBufSize)
go cur.Accept(ctx, ch, prevCh)
prevCh = ChanOutput(ch)
}

59
pipeline/pipeline_test.go
View file

@ -10,6 +10,8 @@ import (
"time"
"context"
"github.com/peco/peco/line"
)
type RegexpFilter struct {
@ -22,38 +24,35 @@ func NewRegexpFilter(rx *regexp.Regexp) *RegexpFilter {
}
}
func (rf *RegexpFilter) Accept(ctx context.Context, in chan interface{}, out ChanOutput) {
func (rf *RegexpFilter) Accept(ctx context.Context, in <-chan line.Line, out ChanOutput) {
defer fmt.Println("END RegexpFilter.Accept")
defer out.SendEndMark(ctx, "end of RegexpFilter")
defer close(out)
for {
select {
case <-ctx.Done():
return
case v := <-in:
if err, ok := v.(error); ok {
if IsEndMark(err) {
return
}
case v, ok := <-in:
if !ok {
return
}
if s, ok := v.(string); ok {
if rf.rx.MatchString(s) {
out.Send(ctx, s)
}
if rf.rx.MatchString(v.DisplayString()) {
out.Send(ctx, v)
}
}
}
}
type LineFeeder struct {
lines []string
lines []line.Line
}
func NewLineFeeder(rdr io.Reader) *LineFeeder {
scan := bufio.NewScanner(rdr)
var lines []string
var lines []line.Line
var id uint64
for scan.Scan() {
lines = append(lines, scan.Text())
lines = append(lines, line.NewRaw(id, scan.Text(), false, false))
id++
}
return &LineFeeder{
lines: lines,
@ -66,14 +65,14 @@ func (f *LineFeeder) Reset() {
func (f *LineFeeder) Start(ctx context.Context, out ChanOutput) {
fmt.Println("START LineFeeder.Start")
defer fmt.Println("END LineFeeder.Start")
defer out.SendEndMark(ctx, "end of LineFeeder")
for _, s := range f.lines {
out.Send(ctx, s)
defer close(out)
for _, l := range f.lines {
out.Send(ctx, l)
}
}
type Receiver struct {
lines []string
lines []line.Line
done chan struct{}
}
@ -92,7 +91,7 @@ func (r *Receiver) Done() <-chan struct{} {
return r.done
}
func (r *Receiver) Accept(ctx context.Context, in chan interface{}, out ChanOutput) {
func (r *Receiver) Accept(ctx context.Context, in <-chan line.Line, out ChanOutput) {
defer fmt.Println("END Receiver.Accept")
defer close(r.done)
@ -100,16 +99,11 @@ func (r *Receiver) Accept(ctx context.Context, in chan interface{}, out ChanOutp
select {
case <-ctx.Done():
return
case v := <-in:
if err, ok := v.(error); ok {
if IsEndMark(err) {
return
}
}
if s, ok := v.(string); ok {
r.lines = append(r.lines, s)
case v, ok := <-in:
if !ok {
return
}
r.lines = append(r.lines, v)
}
}
}
@ -133,5 +127,10 @@ barfoo
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
p.Run(ctx)
t.Logf("%#v", dst.lines)
var got []string
for _, l := range dst.lines {
got = append(got, l.DisplayString())
}
t.Logf("%#v", got)
}

34
source.go
View file

@ -32,7 +32,7 @@ func NewSource(name string, in io.Reader, isInfinite bool, idgen line.IDGenerato
lines: lines,
ready: make(chan struct{}),
setupDone: make(chan struct{}),
ChanOutput: pipeline.ChanOutput(make(chan interface{})),
ChanOutput: pipeline.ChanOutput(make(chan line.Line)),
}
s.Reset()
return s
@ -171,7 +171,7 @@ func (s *Source) Start(ctx context.Context, out pipeline.ChanOutput) {
defer g.End()
defer func() { pdebug.Printf("Source sent %d lines", sent) }()
}
defer out.SendEndMark(ctx, "end of input")
defer close(out)
var resume bool
select {
@ -181,9 +181,8 @@ func (s *Source) Start(ctx context.Context, out pipeline.ChanOutput) {
}
if !resume {
// no fancy resume handling needed. Send lines in batches
// to reduce channel operations.
for i := 0; i < len(s.lines); i += sourceBatchSize {
// no fancy resume handling needed. Send individual lines.
for _, l := range s.lines {
select {
case <-ctx.Done():
if pdebug.Enabled {
@ -192,12 +191,8 @@ func (s *Source) Start(ctx context.Context, out pipeline.ChanOutput) {
return
default:
}
end := i + sourceBatchSize
if end > len(s.lines) {
end = len(s.lines)
}
out.Send(ctx, s.lines[i:end])
sent += end - i
out.Send(ctx, l)
sent++
}
return
}
@ -217,8 +212,8 @@ func (s *Source) Start(ctx context.Context, out pipeline.ChanOutput) {
return
}
// Send available lines in batches
for i := prev; i < upto; i += sourceBatchSize {
// Send available lines individually
for i := prev; i < upto; i++ {
select {
case <-ctx.Done():
if pdebug.Enabled {
@ -227,13 +222,12 @@ func (s *Source) Start(ctx context.Context, out pipeline.ChanOutput) {
return
default:
}
end := i + sourceBatchSize
if end > upto {
end = upto
l, err := s.LineAt(i)
if err != nil {
continue
}
batch := s.linesInRange(i, end)
out.Send(ctx, batch)
sent += len(batch)
out.Send(ctx, l)
sent++
}
// Remember how far we have processed
prev = upto
@ -259,7 +253,7 @@ func (s *Source) Reset() {
g := pdebug.Marker("Source.Reset")
defer g.End()
}
s.ChanOutput = pipeline.ChanOutput(make(chan interface{}))
s.ChanOutput = pipeline.ChanOutput(make(chan line.Line))
}
// Ready returns the "input ready" channel. It will be closed as soon as