From c048ef4d037705efdd610508ac72d795be6ec748 Mon Sep 17 00:00:00 2001 From: Daisuke Maki Date: Tue, 21 Jun 2016 11:49:35 -0400 Subject: [PATCH] tweak --- filter.go | 22 +++++++++++++++------- 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/filter.go b/filter.go index dad5329..e48d32f 100644 --- a/filter.go +++ b/filter.go @@ -170,17 +170,24 @@ func (rf RegexpFilter) Clone() LineFilter { } } -var bufPool = sync.Pool{ +const filterBufSize = 1000 +var filterBufPool = sync.Pool{ New: func() interface{} { return make([]Line, 0, 1000) }, } func releaseRegexpFilterBuf(l []Line) { - bufPool.Put(l[0:0]) + if l == nil { + return + } + l = l[0:0] + filterBufPool.Put(l) } + func getRegexpFilterBuf() []Line { - return bufPool.Get().([]Line) + l := filterBufPool.Get().([]Line) + return l } func (rf *RegexpFilter) Accept(ctx context.Context, p pipeline.Producer) { @@ -194,15 +201,17 @@ func (rf *RegexpFilter) Accept(ctx context.Context, p pipeline.Producer) { flush := func(buf []Line) { sem <- struct{}{} defer func() { <-sem }() + defer releaseRegexpFilterBuf(buf) + for _, in := range buf { if l, err := rf.filter(in); err == nil { rf.outCh.Send(l) } } - releaseRegexpFilterBuf(buf) } buf := getRegexpFilterBuf() defer func() { releaseRegexpFilterBuf(buf) }() + for { select { case <-ctx.Done(): @@ -219,15 +228,14 @@ func (rf *RegexpFilter) Accept(ctx context.Context, p pipeline.Producer) { } if len(buf) > 0 { flush(buf) + buf = nil } return - } case Line: buf = append(buf, v.(Line)) if len(buf) >= cap(buf) { - lb := buf - go flush(lb) + go flush(buf) buf = getRegexpFilterBuf() } }