Improve ingestion performance (by around 40%)

Summary
    fzf --sync --bind load:accept < 27M-lines ran
      1.16 ± 0.01 times faster than fzf-41b3511 --sync --bind load:accept < 27M-lines
      1.44 ± 0.01 times faster than fzf-0.48.1 --sync --bind load:accept < 27M-lines

CHANGELOG.md

@@ -3,7 +3,7 @@ CHANGELOG
 
 0.49.0
 ------
-- Ingestion performance improved by around 20%
+- Ingestion performance improved by around 40%
 - Added two environment variables exported to the child processes
     - `FZF_PREVIEW_LABEL`
     - `FZF_BORDER_LABEL`

src/reader.go

@@ -1,13 +1,12 @@
 package fzf
 
 import (
-	"bufio"
+	"bytes"
 	"context"
 	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
-	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -112,86 +111,82 @@ func (r *Reader) ReadSource(root string, opts walkerOpts, ignores []string) {
 }
 
 func (r *Reader) feed(src io.Reader) {
-	readerSlabSize, ae := strconv.Atoi(os.Getenv("SLAB_KB"))
+	/*
-	if ae != nil {
+		readerSlabSize, ae := strconv.Atoi(os.Getenv("SLAB_KB"))
-		readerSlabSize = 128 * 1024
+		if ae != nil {
-	} else {
+			readerSlabSize = 128 * 1024
-		readerSlabSize *= 1024
+		} else {
-	}
+			readerSlabSize *= 1024
-	readerBufferSize, be := strconv.Atoi(os.Getenv("BUF_KB"))
+		}
-	if be != nil {
+		readerBufferSize, be := strconv.Atoi(os.Getenv("BUF_KB"))
-		readerBufferSize = 64 * 1024
+		if be != nil {
-	} else {
+			readerBufferSize = 64 * 1024
-		readerBufferSize *= 1024
+		} else {
-	}
+			readerBufferSize *= 1024
+		}
+	*/
 
-	slab := make([]byte, readerSlabSize)
-	pointer := 0
 	delim := byte('\n')
 	if r.delimNil {
 		delim = '\000'
 	}
-	reader := bufio.NewReaderSize(src, readerBufferSize)
-
-	// We do not put a slice longer than 10% of the slab to reduce fragmentation
-	maxBytes := readerBufferSize / 10
 
+	slab := make([]byte, readerSlabSize)
+	leftover := []byte{}
+	var err error
 	for {
-		var frags [][]byte
+		n := 0
-		fragsLen := 0
+		scope := slab[:util.Min(len(slab), readerBufferSize)]
-		for {
+		for i := 0; i < 100; i++ {
-			bytea, err := reader.ReadSlice(delim)
+			n, err = src.Read(scope)
-			if err == bufio.ErrBufferFull {
+			if n > 0 || err != nil {
-				// Could not find the delimiter in the reader buffer.
+				break
-				// Need to collect the fragments and merge them later.
+			}
-				frags = append(frags, bytea)
+		}
-				fragsLen += len(bytea)
-			} else {
-				byteaLen := len(bytea)
-				if err == nil {
-					// No errors. Found the delimiter.
-					if util.IsWindows() && byteaLen >= 2 && bytea[byteaLen-2] == byte('\r') {
-						bytea = bytea[:byteaLen-2]
-						byteaLen -= 2
-					} else {
-						bytea = bytea[:byteaLen-1]
-						byteaLen--
-					}
-				}
 
-				itemLen := fragsLen + byteaLen
+		// We're not making any progress after 100 tries. Stop.
-				pointer += itemLen
+		if n == 0 && err == nil {
-				var slice []byte
+			break
-				if itemLen <= maxBytes { // We can use the slab
+		}
-					// Allocate a new slab if it doesn't fit
-					if pointer > readerSlabSize {
-						slab = make([]byte, readerSlabSize)
-						pointer = itemLen
-					}
-					slice = slab[pointer-itemLen : pointer]
-				} else { // We can't use the slab because the item is too large
-					slice = make([]byte, itemLen)
-				}
 
-				if len(frags) > 0 {
+		buf := slab[:n]
-					// Collect the fragments
+		slab = slab[n:]
-					n := 0
+
-					for _, frag := range frags {
+		for len(buf) > 0 {
-						n += copy(slice[n:], frag)
+			if i := bytes.IndexByte(buf, delim); i >= 0 {
-					}
+				// Found the delimiter
-					copy(slice[n:], bytea)
+				slice := buf[:i+1]
-				} else if byteaLen > 0 {
+				buf = buf[i+1:]
-					copy(slice, bytea)
+				if util.IsWindows() && len(slice) >= 2 && slice[len(slice)-2] == byte('\r') {
+					slice = slice[:len(slice)-2]
+				} else {
+					slice = slice[:len(slice)-1]
 				}
-				if (err == nil || itemLen > 0) && r.pusher(slice) {
+				if len(leftover) > 0 {
-					atomic.StoreInt32(&r.event, int32(EvtReadNew))
+					slice = append(leftover, slice...)
+					leftover = []byte{}
 				}
-				if err != nil {
+				if (err == nil || len(slice) > 0) && r.pusher(slice) {
-					return
+					atomic.StoreInt32(&r.event, int32(EvtReadNew))
 				}
+			} else {
+				// Could not find the delimiter in the buffer
+				leftover = append(leftover, buf...)
 				break
 			}
 		}
+
+		if err == io.EOF {
+			leftover = append(leftover, buf...)
+			break
+		}
+
+		if len(slab) == 0 {
+			slab = make([]byte, readerSlabSize)
+		}
+	}
+	if len(leftover) > 0 && r.pusher(leftover) {
+		atomic.StoreInt32(&r.event, int32(EvtReadNew))
 	}
 }