Batch sending source -> filter

buffer.go

@@ -149,6 +149,12 @@ func (mb *MemoryBuffer) Accept(ctx context.Context, in chan interface{}, _ pipel
 					}
 					return
 				}
+			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)
 
@@ -168,6 +174,8 @@ func (mb *MemoryBuffer) Accept(ctx context.Context, in chan interface{}, _ pipel
 								mb.mutex.Unlock()
 								return
 							}
+						case []line.Line:
+							batch = append(batch, v2...)
 						case line.Line:
 							batch = append(batch, v2)
 						}
@@ -227,8 +235,13 @@ func NewMemoryBufferSource(buf *MemoryBuffer) *MemoryBufferSource {
 	return &MemoryBufferSource{buf: buf}
 }
 
-// Start iterates through the MemoryBuffer's lines and sends them to the
-// output channel, implementing pipeline.Source.
+// 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.
 func (s *MemoryBufferSource) Start(ctx context.Context, out pipeline.ChanOutput) {
 	defer out.SendEndMark(ctx, "end of memory buffer source")
 
@@ -236,13 +249,17 @@ func (s *MemoryBufferSource) Start(ctx context.Context, out pipeline.ChanOutput)
 	lines := s.buf.lines
 	s.buf.mutex.RUnlock()
 
-	for _, l := range lines {
+	for i := 0; i < len(lines); i += sourceBatchSize {
 		select {
 		case <-ctx.Done():
 			return
 		default:
-			out.Send(ctx, l)
 		}
+		end := i + sourceBatchSize
+		if end > len(lines) {
+			end = len(lines)
+		}
+		out.Send(ctx, lines[i:end])
 	}
 }
 

filter.go

@@ -269,6 +269,15 @@ func acceptAndFilterSerial(ctx context.Context, f filter.Filter, bufsiz int, buf
 					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()
+				}
 			}
 		}
 	}
@@ -324,6 +333,16 @@ func acceptAndFilterParallel(ctx context.Context, f filter.Filter, bufsiz int, b
 					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()
+				}
 			}
 		}
 	}

filter_incremental_test.go

@@ -70,6 +70,10 @@ func TestMemoryBufferSource(t *testing.T) {
 			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())
 		}

source.go

@@ -180,8 +180,9 @@ func (s *Source) Start(ctx context.Context, out pipeline.ChanOutput) {
 	}
 
 	if !resume {
-		// no fancy resume handling needed. just go
-		for _, l := range s.lines {
+		// no fancy resume handling needed. Send lines in batches
+		// to reduce channel operations.
+		for i := 0; i < len(s.lines); i += sourceBatchSize {
 			select {
 			case <-ctx.Done():
 				if pdebug.Enabled {
@@ -189,9 +190,13 @@ func (s *Source) Start(ctx context.Context, out pipeline.ChanOutput) {
 				}
 				return
 			default:
-				out.Send(ctx, l)
-				sent++
 			}
+			end := i + sourceBatchSize
+			if end > len(s.lines) {
+				end = len(s.lines)
+			}
+			out.Send(ctx, s.lines[i:end])
+			sent += end - i
 		}
 		return
 	}
@@ -211,7 +216,8 @@ func (s *Source) Start(ctx context.Context, out pipeline.ChanOutput) {
 			return
 		}
 
-		for i := prev; i < upto; i++ {
+		// Send available lines in batches
+		for i := prev; i < upto; i += sourceBatchSize {
 			select {
 			case <-ctx.Done():
 				if pdebug.Enabled {
@@ -219,10 +225,14 @@ func (s *Source) Start(ctx context.Context, out pipeline.ChanOutput) {
 				}
 				return
 			default:
-				l, _ := s.LineAt(i)
-				out.Send(ctx, l)
-				sent++
 			}
+			end := i + sourceBatchSize
+			if end > upto {
+				end = upto
+			}
+			batch := s.linesInRange(i, end)
+			out.Send(ctx, batch)
+			sent += len(batch)
 		}
 		// Remember how far we have processed
 		prev = upto