From 89943598a0b3c55d268e7dd51bd4199723a20c9d Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Mon, 16 Jun 2025 00:39:31 +0300 Subject: initial faster readfile --- cmd/dcat/main.go | 30 ++++++++++++++++++++----- integrationtests/dtail_test.go | 10 +++++---- internal/io/fs/readfile.go | 39 +++++++++++++++++++-------------- internal/server/handlers/readcommand.go | 3 ++- internal/version/version.go | 2 +- 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!" ) -- cgit v1.2.3