initial faster readfile

5 files changed, 57 insertions, 27 deletions

diff --git a/cmd/dcat/main.go b/cmd/dcat/main.go
index 98da089..71e2f1a 100644
--- a/cmd/dcat/main.go
+++ b/cmd/dcat/main.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"flag"
 	"os"
+	"runtime/pprof"
 	"sync"
 
 	"net/http"
@@ -23,7 +24,8 @@ import (
 func main() {
 	var args config.Args
 	var displayVersion bool
-	var pprof string
+	var pprofAddr string
+	var cpuprofile string
 
 	userName := user.Name()
 
@@ -44,7 +46,8 @@ func main() {
 	flag.StringVar(&args.ServersStr, "servers", "", "Remote servers to connect")
 	flag.StringVar(&args.UserName, "user", userName, "Your system user name")
 	flag.StringVar(&args.What, "files", "", "File(s) to read")
-	flag.StringVar(&pprof, "pprof", "", "Start PProf server this address")
+	flag.StringVar(&pprofAddr, "pprof", "", "Start PProf server this address")
+	flag.StringVar(&cpuprofile, "cpuprofile", "", "Write CPU profile to file")
 
 	flag.Parse()
 	config.Setup(source.Client, &args, flag.Args())
@@ -53,14 +56,26 @@ func main() {
 		version.PrintAndExit()
 	}
 
+	if cpuprofile != "" {
+		f, err := os.Create(cpuprofile)
+		if err != nil {
+			panic(err)
+		}
+		defer f.Close()
+		if err := pprof.StartCPUProfile(f); err != nil {
+			panic(err)
+		}
+		defer pprof.StopCPUProfile()
+	}
+
 	ctx, cancel := context.WithCancel(context.Background())
 	var wg sync.WaitGroup
 	wg.Add(1)
 	dlog.Start(ctx, &wg, source.Client)
 
-	if pprof != "" {
-		go http.ListenAndServe(pprof, nil)
-		dlog.Client.Info("Started PProf", pprof)
+	if pprofAddr != "" {
+		go http.ListenAndServe(pprofAddr, nil)
+		dlog.Client.Info("Started PProf", pprofAddr)
 	}
 
 	client, err := clients.NewCatClient(args)
@@ -72,5 +87,10 @@ func main() {
 	cancel()
 
 	wg.Wait()
+	
+	if cpuprofile != "" {
+		pprof.StopCPUProfile()
+	}
+	
 	os.Exit(status)
 }
diff --git a/integrationtests/dtail_test.go b/integrationtests/dtail_test.go
index 64a32f1..9e7f281 100644
--- a/integrationtests/dtail_test.go
+++ b/integrationtests/dtail_test.go
@@ -74,9 +74,11 @@ func TestDTailWithServer(t *testing.T) {
 		var circular int
 		for {
 			select {
-			case <-time.After(time.Second):
+			case <-time.After(time.Second * 2):
 				fd.WriteString(time.Now().String())
-				fd.WriteString(fmt.Sprintf(" - Hello %s\n", greetings[circular]))
+				message := fmt.Sprintf(" - Hello %s\n", greetings[circular])
+				fd.WriteString(message)
+				t.Logf("Wrote '%s' into file", message)
 				circular = (circular + 1) % len(greetings)
 			case <-ctx.Done():
 				return
@@ -120,8 +122,8 @@ func TestDTailWithServer(t *testing.T) {
 	for i, g := range greetingsRecv {
 		index := (i + offset) % len(greetings)
 		if greetings[index] != g {
-			t.Error(fmt.Sprintf("Expected '%s' but got '%s' at '%v' vs '%v'\n",
-				g, greetings[index], greetings, greetingsRecv))
+			t.Error(fmt.Sprintf("Expected '%s' but got '%s' at '%v(%d)' vs '%v(%d)'\n",
+				g, greetings[index], greetings, len(greetings), greetingsRecv, len(greetingsRecv)))
 			return
 		}
 	}
diff --git a/internal/io/fs/readfile.go b/internal/io/fs/readfile.go
index 669f99f..3dafd9b 100644
--- a/internal/io/fs/readfile.go
+++ b/internal/io/fs/readfile.go
@@ -177,28 +177,35 @@ func (f *readFile) makeCompressedFileReader(fd *os.File) (reader *bufio.Reader,
 func (f *readFile) read(ctx context.Context, fd *os.File, reader *bufio.Reader,
 	rawLines chan *bytes.Buffer, truncate <-chan struct{}) error {
 
-	var offset uint64
-	message := pool.BytesBuffer.Get().(*bytes.Buffer)
-
+	// Use chunked reader for better performance
+	chunkedReader := NewChunkedReader(reader, 64*1024) // 64KB chunks
+	
+	// Create a goroutine to handle truncate signals
+	go func() {
+		select {
+		case <-truncate:
+			// Handle truncation by attempting to seek to end of file
+			if fd != nil {
+				fd.Seek(0, io.SeekEnd)
+			}
+		case <-ctx.Done():
+			return
+		}
+	}()
+	
+	// Process lines using chunked reader
 	for {
-		b, err := reader.ReadByte()
+		err := chunkedReader.ProcessLines(ctx, rawLines, config.Server.MaxLineLength, 
+			f.filePath, f.serverMessages, f.seekEOF)
 		if err != nil {
-			status, err := f.handleReadError(ctx, err, fd, rawLines, truncate, message)
-			if abortReading == status {
-				return err
+			if err == io.EOF || err == context.Canceled {
+				return nil
 			}
+			// Handle read errors similar to original implementation
 			time.Sleep(time.Millisecond * 100)
 			continue
 		}
-
-		offset++
-		message.WriteByte(b)
-
-		status, newMessage := f.handleReadByte(ctx, b, rawLines, message)
-		if status == abortReading {
-			return nil
-		}
-		message = newMessage
+		return nil
 	}
 }
 
diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go
index 44ba9e4..5f7da86 100644
--- a/internal/server/handlers/readcommand.go
+++ b/internal/server/handlers/readcommand.go
@@ -49,7 +49,8 @@ func (r *readCommand) Start(ctx context.Context, ltx lcontext.LContext,
 
 	// In serverless mode, can also read data from pipe
 	// e.g.: grep foo bar.log | dmap 'from STATS select ...'
-	if r.isInputFromPipe() {
+	// Only read from stdin if no file is specified AND input is from pipe
+	if (args[1] == "" || args[1] == "-") && r.isInputFromPipe() {
 		dlog.Server.Debug("Reading data from stdin pipe")
 		// Empty file path and globID "-" represents reading from the stdin pipe.
 		r.read(ctx, ltx, "", "-", re)
diff --git a/internal/version/version.go b/internal/version/version.go
index 15ea50f..20a269f 100644
--- a/internal/version/version.go
+++ b/internal/version/version.go
@@ -13,7 +13,7 @@ const (
 	// Name of DTail.
 	Name string = "DTail"
 	// Version of DTail.
-	Version string = "4.3.0"
+	Version string = "4.3.1-develop"
 	// Additional information for DTail
 	Additional string = "Have a lot of fun!"
 )