From 67a6b9d8e8e8dc83d5ea3e5859e631a0dfa9dabe Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Wed, 18 Jun 2025 09:10:52 +0300 Subject: Complete channelless migration for DTail operations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Implement channelless MapReduce with streaming aggregation - Add channelless tail with proper file following capability - Fix TestDTailWithServer by implementing ServerHandlerWriter for client-server mode - Add proper serverless mode detection for standalone operations - Remove temporary benchmark scripts - All integration tests now pass 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude --- .claude/commands/retest.md | 1 + integrationtests/channelless_test.go | 104 ++++++ internal/io/fs/directprocessor.go | 537 +++++++++++++++++++++++++++++- internal/server/handlers/networkwriter.go | 176 ++++++++++ internal/server/handlers/readcommand.go | 327 ++++++++---------- scripts/benchmark_channelless.sh | 215 ------------ scripts/corrected_benchmark.sh | 89 ----- scripts/profile_channelless.sh | 50 --- 8 files changed, 944 insertions(+), 555 deletions(-) create mode 100644 .claude/commands/retest.md create mode 100644 integrationtests/channelless_test.go delete mode 100755 scripts/benchmark_channelless.sh delete mode 100755 scripts/corrected_benchmark.sh delete mode 100755 scripts/profile_channelless.sh diff --git a/.claude/commands/retest.md b/.claude/commands/retest.md new file mode 100644 index 0000000..555654f --- /dev/null +++ b/.claude/commands/retest.md @@ -0,0 +1 @@ +Recompile the project with make clean build and re-run all tests including integration tests. If any fail, fix them. diff --git a/integrationtests/channelless_test.go b/integrationtests/channelless_test.go new file mode 100644 index 0000000..41845df --- /dev/null +++ b/integrationtests/channelless_test.go @@ -0,0 +1,104 @@ +package integrationtests + +import ( + "context" + "os" + "testing" + + "github.com/mimecast/dtail/internal/config" +) + +func TestDGrepChannelless(t *testing.T) { + if !config.Env("DTAIL_INTEGRATION_TEST_RUN_MODE") { + t.Log("Skipping") + return + } + + // Test dgrep with channelless mode (now the default) + inFile := "mapr_testdata.log" + outFile := "dgrepchannelless.stdout.tmp" + expectedOutFile := "dgrepcontext1.txt.expected" + + _, err := runCommand(context.TODO(), t, outFile, + "../dgrep", + "--plain", + "--cfg", "none", + "--grep", "1002-071947", + "--after", "3", + "--before", "3", + inFile) + + if err != nil { + t.Error(err) + return + } + + if err := compareFiles(t, outFile, expectedOutFile); err != nil { + t.Error(err) + return + } + + os.Remove(outFile) +} + +func TestDCatChannelless(t *testing.T) { + if !config.Env("DTAIL_INTEGRATION_TEST_RUN_MODE") { + t.Log("Skipping") + return + } + + // Test dcat with channelless mode (now the default) + inFile := "dcat1a.txt" + outFile := "dcatchannelless.stdout.tmp" + + _, err := runCommand(context.TODO(), t, outFile, + "../dcat", + "--plain", + "--cfg", "none", + inFile) + + if err != nil { + t.Error(err) + return + } + + if err := compareFiles(t, outFile, inFile); err != nil { + t.Error(err) + return + } + + os.Remove(outFile) +} + +func TestChannellessMode(t *testing.T) { + if !config.Env("DTAIL_INTEGRATION_TEST_RUN_MODE") { + t.Log("Skipping") + return + } + + // Test that channelless mode (now default) works correctly + + // Test grep + inFile := "mapr_testdata.log" + outFile := "grep_channelless.tmp" + expectedOutFile := "dgrep1.txt.expected" + + _, err := runCommand(context.TODO(), t, outFile, + "../dgrep", + "--plain", + "--cfg", "none", + "--grep", "1002-071947", + inFile) + + if err != nil { + t.Error(err) + return + } + + if err := compareFiles(t, outFile, expectedOutFile); err != nil { + t.Error(err) + return + } + + os.Remove(outFile) +} \ No newline at end of file diff --git a/internal/io/fs/directprocessor.go b/internal/io/fs/directprocessor.go index d4bf3ed..00ecedb 100644 --- a/internal/io/fs/directprocessor.go +++ b/internal/io/fs/directprocessor.go @@ -7,10 +7,16 @@ import ( "fmt" "io" "os" + "strings" + "time" "github.com/mimecast/dtail/internal/color/brush" "github.com/mimecast/dtail/internal/config" + "github.com/mimecast/dtail/internal/io/dlog" + "github.com/mimecast/dtail/internal/io/line" "github.com/mimecast/dtail/internal/lcontext" + "github.com/mimecast/dtail/internal/mapr" + "github.com/mimecast/dtail/internal/mapr/logformat" "github.com/mimecast/dtail/internal/protocol" "github.com/mimecast/dtail/internal/regex" ) @@ -618,7 +624,8 @@ func (tp *TailProcessor) ProcessLine(line []byte, lineNum int, filePath string, } // Regular tailing mode - send matching lines immediately - return tp.formatLine(line, lineNum, filePath), true + formatted := tp.formatLine(line, lineNum, filePath) + return formatted, true } func (tp *TailProcessor) formatLine(line []byte, lineNum int, filePath string) []byte { @@ -660,25 +667,248 @@ func (tp *TailProcessor) Flush() []byte { return nil } +// FollowingTailProcessor extends DirectProcessor with file following capability +type FollowingTailProcessor struct { + *DirectProcessor + tailProcessor *TailProcessor +} + +// NewFollowingTailProcessor creates a processor that can follow files +func NewFollowingTailProcessor(processor *TailProcessor, output io.Writer, globID string, ltx lcontext.LContext) *FollowingTailProcessor { + dp := NewDirectProcessor(processor, output, globID, ltx) + return &FollowingTailProcessor{ + DirectProcessor: dp, + tailProcessor: processor, + } +} + +// ProcessFileWithFollowing processes a file with following capability +func (ftp *FollowingTailProcessor) ProcessFileWithFollowing(ctx context.Context, filePath string) error { + if !ftp.tailProcessor.follow { + // No following required, use regular processing + return ftp.ProcessFile(ctx, filePath) + } + + // Implement file following logic + return ftp.followFile(ctx, filePath) +} + +func (ftp *FollowingTailProcessor) followFile(ctx context.Context, filePath string) error { + file, err := os.Open(filePath) + if err != nil { + return err + } + defer file.Close() + + // Initialize processor + if err := ftp.processor.Initialize(ctx); err != nil { + return err + } + defer ftp.processor.Cleanup() + + // If seekEOF is true, seek to end first + if ftp.tailProcessor.seekEOF { + if _, err := file.Seek(0, io.SeekEnd); err != nil { + return err + } + } + + return ftp.followReader(ctx, file, filePath) +} + +func (ftp *FollowingTailProcessor) followReader(ctx context.Context, file *os.File, filePath string) error { + // Set buffer size respecting MaxLineLength configuration + maxLineLength := config.Server.MaxLineLength + initialBufSize := 64 * 1024 + if maxLineLength < initialBufSize { + initialBufSize = maxLineLength + } + + lineNum := 0 + lastPosition := int64(0) + readBuffer := make([]byte, initialBufSize) + lineBuffer := make([]byte, 0, initialBufSize) + + // Get initial position + if pos, err := file.Seek(0, io.SeekCurrent); err == nil { + lastPosition = pos + } + + for { + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + + // Check if file has grown + if stat, err := file.Stat(); err == nil { + if stat.Size() > lastPosition { + // Read new content + n, err := file.Read(readBuffer) + if err != nil && err != io.EOF { + return err + } + + if n > 0 { + // Process the data, looking for complete lines + for i := 0; i < n; i++ { + b := readBuffer[i] + if b == '\n' { + // Found a complete line + lineNum++ + line := make([]byte, len(lineBuffer)) + copy(line, lineBuffer) + + // Update position stats + if ftp.stats != nil { + ftp.stats.updatePosition() + } + + // Process line directly + if result, shouldSend := ftp.processor.ProcessLine(line, lineNum, filePath, ftp.stats, ftp.sourceID); shouldSend { + if _, err := ftp.output.Write(result); err != nil { + return err + } + + // Update transmission stats + if ftp.stats != nil { + ftp.stats.updateLineTransmitted() + } + } + + // Reset line buffer for next line + lineBuffer = lineBuffer[:0] + } else { + // Add byte to current line + lineBuffer = append(lineBuffer, b) + } + } + + // Update last position + if pos, err := file.Seek(0, io.SeekCurrent); err == nil { + lastPosition = pos + } + + continue + } + } + } + + // No more content available, check if file was truncated/rotated + if ftp.checkFileRotation(file, filePath, &lastPosition) { + // File was rotated, reopen and continue + file.Close() + var err error + file, err = os.Open(filePath) + if err != nil { + return err + } + defer file.Close() + + lastPosition = 0 + lineBuffer = lineBuffer[:0] + continue + } + + // Wait a bit before checking for new content + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(100 * time.Millisecond): + // Continue the loop to check for new content + } + } +} + +func (ftp *FollowingTailProcessor) checkFileRotation(file *os.File, filePath string, lastPosition *int64) bool { + // Get current file info + currentInfo, err := file.Stat() + if err != nil { + return false + } + + // Get file info by path + pathInfo, err := os.Stat(filePath) + if err != nil { + return false + } + + // Check if file was truncated (size is smaller than our position) + if pathInfo.Size() < *lastPosition { + return true + } + + // Check if file was rotated (different inode/device) + if !os.SameFile(currentInfo, pathInfo) { + return true + } + + return false +} + // MapProcessor handles MapReduce-style aggregation type MapProcessor struct { - plain bool - hostname string - aggregator interface{} // Will be set to actual aggregator from mapr package - buffer []byte + plain bool + hostname string + query *mapr.Query + parser logformat.Parser + groupSet *mapr.GroupSet + buffer []byte + output io.Writer + lastSerialized time.Time + serializeFunc func(groupSet *mapr.GroupSet) } // NewMapProcessor creates a new map processor -func NewMapProcessor(plain bool, hostname string) *MapProcessor { - return &MapProcessor{ - plain: plain, - hostname: hostname, - buffer: make([]byte, 0, 1024*1024), // 1MB buffer for aggregation +func NewMapProcessor(plain bool, hostname string, queryStr string, output io.Writer) (*MapProcessor, error) { + query, err := mapr.NewQuery(queryStr) + if err != nil { + return nil, err + } + + var parserName string + switch query.LogFormat { + case "": + parserName = config.Server.MapreduceLogFormat + if query.Table == "" { + parserName = "generic" + } + default: + parserName = query.LogFormat + } + + logParser, err := logformat.NewParser(parserName, query) + if err != nil { + dlog.Server.Error("Could not create log format parser. Falling back to 'generic'", err) + if logParser, err = logformat.NewParser("generic", query); err != nil { + return nil, fmt.Errorf("could not create log format parser: %w", err) + } } + + mp := &MapProcessor{ + plain: plain, + hostname: hostname, + query: query, + parser: logParser, + groupSet: mapr.NewGroupSet(), + buffer: make([]byte, 0, 1024*1024), // 1MB buffer for aggregation + output: output, + lastSerialized: time.Now(), + } + + // Set up serialization function + mp.serializeFunc = mp.defaultSerializeFunc + + return mp, nil +} + +// SetSerializeFunc allows custom serialization (for testing or different output formats) +func (mp *MapProcessor) SetSerializeFunc(fn func(groupSet *mapr.GroupSet)) { + mp.serializeFunc = fn } func (mp *MapProcessor) Initialize(ctx context.Context) error { - // TODO: Initialize MapReduce aggregator when implementing return nil } @@ -687,15 +917,118 @@ func (mp *MapProcessor) Cleanup() error { } func (mp *MapProcessor) ProcessLine(line []byte, lineNum int, filePath string, stats *stats, sourceID string) ([]byte, bool) { - // For MapReduce, we accumulate lines and process in batch - // TODO: Pass line to aggregator when implementing MapReduce integration - return nil, false // No immediate output for MapReduce + // Convert line to string and parse fields + maprLine := strings.TrimSpace(string(line)) + + fields, err := mp.parser.MakeFields(maprLine) + if err != nil { + // Should fields be ignored anyway? + if err != logformat.ErrIgnoreFields { + dlog.Server.Error("Error parsing line for MapReduce", err) + } + return nil, false + } + + // Apply WHERE clause filter + if !mp.query.WhereClause(fields) { + return nil, false + } + + // Apply SET clause (add additional fields) + if len(mp.query.Set) > 0 { + if err := mp.query.SetClause(fields); err != nil { + dlog.Server.Error("Error applying SET clause", err) + return nil, false + } + } + + // Aggregate the fields + mp.aggregateFields(fields) + + // Check if we should serialize results periodically (every 5 seconds by default) + now := time.Now() + if now.Sub(mp.lastSerialized) >= mp.query.Interval { + mp.periodicSerialize() + mp.lastSerialized = now + } + + return nil, false // No immediate output for MapReduce - output happens periodically +} + +func (mp *MapProcessor) aggregateFields(fields map[string]string) { + var sb strings.Builder + for i, field := range mp.query.GroupBy { + if i > 0 { + sb.WriteString(protocol.AggregateGroupKeyCombinator) + } + if val, ok := fields[field]; ok { + sb.WriteString(val) + } + } + groupKey := sb.String() + set := mp.groupSet.GetSet(groupKey) + + var addedSample bool + for _, sc := range mp.query.Select { + if val, ok := fields[sc.Field]; ok { + if err := set.Aggregate(sc.FieldStorage, sc.Operation, val, false); err != nil { + dlog.Server.Error("Error aggregating field", err) + continue + } + addedSample = true + } + } + + if addedSample { + set.Samples++ + } +} + +// periodicSerialize sends current aggregation results and resets the group set +func (mp *MapProcessor) periodicSerialize() { + if mp.serializeFunc != nil { + mp.serializeFunc(mp.groupSet) + } + // Reset group set for next interval + mp.groupSet = mapr.NewGroupSet() +} + +// defaultSerializeFunc implements the default serialization behavior +func (mp *MapProcessor) defaultSerializeFunc(groupSet *mapr.GroupSet) { + // Use a channel to collect serialized data + ch := make(chan string, 100) + done := make(chan struct{}) + + go func() { + defer close(done) + for msg := range ch { + // Format as protocol message: A|{serialized_data}¬ + var output strings.Builder + output.WriteString("A") + output.WriteString(protocol.FieldDelimiter) + output.WriteString(msg) + output.WriteByte(protocol.MessageDelimiter) + + // Write to output immediately + if mp.output != nil { + mp.output.Write([]byte(output.String())) + } + } + }() + + // Serialize the group set + ctx := context.Background() + groupSet.Serialize(ctx, ch) + close(ch) + <-done } func (mp *MapProcessor) Flush() []byte { - // TODO: Return aggregated results from MapReduce processor - // For now, return empty to maintain interface - return nil + // Final flush - serialize any remaining data + if mp.serializeFunc != nil { + mp.serializeFunc(mp.groupSet) + } + return nil // Output is handled by serializeFunc } // Helper function to append integer to byte slice @@ -717,4 +1050,174 @@ func appendInt(dst []byte, i int) []byte { } return append(dst, str...) +} + +// AggregateLineProcessor feeds lines to an existing aggregate via channels +type AggregateLineProcessor struct { + linesCh chan<- *line.Line + re regex.Regex + hostname string + ltx lcontext.LContext + lineNum int + isTailing bool // Whether this is for a tail operation that should keep running +} + +// NewAggregateLineProcessor creates a processor that feeds lines to an aggregate +func NewAggregateLineProcessor(linesCh chan<- *line.Line, re regex.Regex, hostname string, ltx lcontext.LContext) *AggregateLineProcessor { + return &AggregateLineProcessor{ + linesCh: linesCh, + re: re, + hostname: hostname, + ltx: ltx, + lineNum: 0, + isTailing: false, + } +} + +// NewAggregateLineProcessorForTail creates a processor for tail operations that feeds lines to an aggregate +func NewAggregateLineProcessorForTail(linesCh chan<- *line.Line, re regex.Regex, hostname string, ltx lcontext.LContext) *AggregateLineProcessor { + return &AggregateLineProcessor{ + linesCh: linesCh, + re: re, + hostname: hostname, + ltx: ltx, + lineNum: 0, + isTailing: true, + } +} + +func (p *AggregateLineProcessor) ProcessLine(lineBuf []byte, lineNum int, filePath string, stats *stats, sourceID string) (result []byte, shouldSend bool) { + p.lineNum++ + + // For MapReduce operations, don't apply regex filtering here - let the aggregate handle it + // The aggregate's log parser and WHERE clause will do the proper filtering + + // Create a line object similar to what the channel-based system creates + // Make a copy of the line buffer to avoid issues with slice reuse + lineCopy := make([]byte, len(lineBuf)) + copy(lineCopy, lineBuf) + content := bytes.NewBuffer(lineCopy) + l := line.New(content, uint64(p.lineNum), 100, sourceID) + + // Send the line to the aggregate via the channel (blocking send to avoid data loss) + p.linesCh <- l + + // Don't send output directly since the aggregate will handle serialization + return nil, false +} + +func (p *AggregateLineProcessor) Flush() []byte { + // For tail operations, don't close the channel as we want to keep following + if !p.isTailing { + // Close the lines channel to signal end of input + // Add a small delay to ensure all lines are processed before closing + time.Sleep(10 * time.Millisecond) + close(p.linesCh) + } + return nil +} + +func (p *AggregateLineProcessor) Initialize(ctx context.Context) error { + return nil +} + +func (p *AggregateLineProcessor) Cleanup() error { + return nil +} + +// ProcessFileWithTailing processes a file with tailing capability +func (dp *DirectProcessor) ProcessFileWithTailing(ctx context.Context, filePath string) error { + // Use the same logic as FollowingTailProcessor but with our DirectProcessor + file, err := os.Open(filePath) + if err != nil { + return err + } + defer file.Close() + + // First, process existing content + if err := dp.ProcessReader(ctx, file, filePath); err != nil { + return err + } + + // Then follow the file for new content + return dp.followFile(ctx, filePath) +} + +// followFile implements file following logic similar to FollowingTailProcessor +func (dp *DirectProcessor) followFile(ctx context.Context, filePath string) error { + // Track our current position in the file + var lastSize int64 + + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(100 * time.Millisecond): + // Check if file has grown + fileInfo, err := os.Stat(filePath) + if err != nil { + continue + } + + currentSize := fileInfo.Size() + if currentSize > lastSize { + // File has new content, read it + file, err := os.Open(filePath) + if err != nil { + continue + } + + // Seek to where we left off + if _, err := file.Seek(lastSize, 0); err != nil { + file.Close() + continue + } + + // Process new content + if err := dp.processNewContent(ctx, file, filePath); err != nil { + file.Close() + continue + } + + lastSize = currentSize + file.Close() + } + } + } +} + +// processNewContent processes new content that was added to the file +func (dp *DirectProcessor) processNewContent(ctx context.Context, file *os.File, filePath string) error { + scanner := bufio.NewScanner(file) + + // Start line counting from where we left off (simplified approach) + lineNum := 1 + + for scanner.Scan() { + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + + lineBuf := scanner.Bytes() + if result, shouldSend := dp.processor.ProcessLine(lineBuf, lineNum, filePath, dp.stats, dp.sourceID); shouldSend { + if _, err := dp.output.Write(result); err != nil { + return err + } + + // Update transmission stats + if dp.stats != nil { + dp.stats.updateLineTransmitted() + } + } + lineNum++ + + // Update position stats + if dp.stats != nil { + dp.stats.updatePosition() + } + } + + return scanner.Err() } \ No newline at end of file diff --git a/internal/server/handlers/networkwriter.go b/internal/server/handlers/networkwriter.go index f1e3bee..bb5ad1d 100644 --- a/internal/server/handlers/networkwriter.go +++ b/internal/server/handlers/networkwriter.go @@ -1,11 +1,13 @@ package handlers import ( + "bytes" "fmt" "net" "os" "github.com/mimecast/dtail/internal/io/dlog" + "github.com/mimecast/dtail/internal/io/line" "github.com/mimecast/dtail/internal/user/server" ) @@ -173,4 +175,178 @@ func (bnw *BufferedNetworkWriter) Close() error { return err } return bnw.NetworkOutputWriter.Close() +} + +// ChannelOutputWriter sends output to the server's lines channel instead of direct network +type ChannelOutputWriter struct { + linesCh chan<- *line.Line + serverMessages chan<- string + user *server.User +} + +// NewChannelOutputWriter creates a new channel output writer +func NewChannelOutputWriter(linesCh chan<- *line.Line, serverMessages chan<- string, user *server.User) *ChannelOutputWriter { + return &ChannelOutputWriter{ + linesCh: linesCh, + serverMessages: serverMessages, + user: user, + } +} + +// Write implements io.Writer interface by sending data through the lines channel +func (cow *ChannelOutputWriter) Write(data []byte) (int, error) { + if len(data) == 0 { + return 0, nil + } + + // Create a line object using the proper constructor + contentBuffer := bytes.NewBuffer(data) + lineObj := line.New(contentBuffer, 0, 100, "channelless") + + select { + case cow.linesCh <- lineObj: + return len(data), nil + default: + // Channel is full, report error + cow.sendServerMessage("Lines channel full, dropping data") + return 0, fmt.Errorf("lines channel full") + } +} + +// sendServerMessage sends a message through the existing server message channel +func (cow *ChannelOutputWriter) sendServerMessage(message string) { + if cow.serverMessages == nil { + return + } + + select { + case cow.serverMessages <- message: + // Message sent successfully + default: + // Channel full, log the issue + dlog.Server.Warn(cow.user, "Server message channel full, dropping message:", message) + } +} + +// SendLine sends a formatted line through the lines channel +func (cow *ChannelOutputWriter) SendLine(hostname, filePath string, lineNum int, content []byte) error { + // Create a line object with proper metadata + contentBuffer := bytes.NewBuffer(content) + lineObj := line.New(contentBuffer, uint64(lineNum), 100, filePath) + + select { + case cow.linesCh <- lineObj: + return nil + default: + cow.sendServerMessage(fmt.Sprintf("Lines channel full, dropping line from %s:%d", filePath, lineNum)) + return fmt.Errorf("lines channel full") + } +} + +// SendPlainLine sends a plain line through the lines channel +func (cow *ChannelOutputWriter) SendPlainLine(content []byte) error { + _, err := cow.Write(content) + return err +} + +// SendServerStat sends a server statistics message +func (cow *ChannelOutputWriter) SendServerStat(message string) { + cow.sendServerMessage(message) +} + +// SendError sends an error message +func (cow *ChannelOutputWriter) SendError(err error) { + cow.sendServerMessage(fmt.Sprintf("ERROR: %v", err)) +} + +// Close is a no-op for channel output writer +func (cow *ChannelOutputWriter) Close() error { + return nil +} + +// ServerHandlerWriter writes output directly to the server handler's lines channel +type ServerHandlerWriter struct { + server *ServerHandler + serverMessages chan<- string + user *server.User +} + +// NewServerHandlerWriter creates a new server handler writer +func NewServerHandlerWriter(serverHandler *ServerHandler, serverMessages chan<- string, user *server.User) *ServerHandlerWriter { + return &ServerHandlerWriter{ + server: serverHandler, + serverMessages: serverMessages, + user: user, + } +} + +// Write implements io.Writer interface by sending data through the server's lines channel +func (shw *ServerHandlerWriter) Write(data []byte) (int, error) { + if len(data) == 0 { + return 0, nil + } + + // Create a line object and send it through the server's lines channel + contentBuffer := bytes.NewBuffer(data) + lineObj := line.New(contentBuffer, 0, 100, "channelless") + + select { + case shw.server.lines <- lineObj: + return len(data), nil + default: + // Channel is full, report error + shw.sendServerMessage("Server lines channel full, dropping data") + return 0, fmt.Errorf("server lines channel full") + } +} + +// sendServerMessage sends a message through the existing server message channel +func (shw *ServerHandlerWriter) sendServerMessage(message string) { + if shw.serverMessages == nil { + return + } + + select { + case shw.serverMessages <- message: + // Message sent successfully + default: + // Channel full, log the issue + dlog.Server.Warn(shw.user, "Server message channel full, dropping message:", message) + } +} + +// SendLine sends a formatted line through the server's lines channel +func (shw *ServerHandlerWriter) SendLine(hostname, filePath string, lineNum int, content []byte) error { + // Create a line object with proper metadata + contentBuffer := bytes.NewBuffer(content) + lineObj := line.New(contentBuffer, uint64(lineNum), 100, filePath) + + select { + case shw.server.lines <- lineObj: + return nil + default: + shw.sendServerMessage(fmt.Sprintf("Server lines channel full, dropping line from %s:%d", filePath, lineNum)) + return fmt.Errorf("server lines channel full") + } +} + +// SendPlainLine sends a plain line through the server's lines channel +func (shw *ServerHandlerWriter) SendPlainLine(content []byte) error { + _, err := shw.Write(content) + return err +} + +// SendServerStat sends a server statistics message +func (shw *ServerHandlerWriter) SendServerStat(message string) { + shw.sendServerMessage(message) +} + +// SendError sends an error message +func (shw *ServerHandlerWriter) SendError(err error) { + shw.sendServerMessage(fmt.Sprintf("ERROR: %v", err)) +} + +// Close is a no-op for server handler writer +func (shw *ServerHandlerWriter) Close() error { + return nil } \ No newline at end of file diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go index 89bb757..59e9d1b 100644 --- a/internal/server/handlers/readcommand.go +++ b/internal/server/handlers/readcommand.go @@ -2,10 +2,10 @@ package handlers import ( "context" + "io" "os" "path/filepath" "strings" - "sync" "time" "github.com/mimecast/dtail/internal/io/dlog" @@ -32,6 +32,9 @@ func (r *readCommand) Start(ctx context.Context, ltx lcontext.LContext, argc int, args []string, retries int) { re := regex.NewNoop() + var queryStr string + + // Parse regex for non-MapReduce operations or when no aggregate exists if argc >= 4 { deserializedRegex, err := regex.Deserialize(strings.Join(args[2:], " ")) if err != nil { @@ -41,169 +44,18 @@ func (r *readCommand) Start(ctx context.Context, ltx lcontext.LContext, } re = deserializedRegex } + if argc < 3 { r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user, "Unable to parse command", args, argc)) return } - // Check if channelless mode is enabled - // Note: MapReduce operations require the full channel-based aggregation infrastructure - // Note: Tail operations require continuous monitoring and real-time streaming - isMapReduceCmd := r.mode == omode.MapClient || r.isMapReduceCommand(re) - isTailCmd := r.mode == omode.TailClient - useChannelless := os.Getenv("DTAIL_USE_CHANNELLESS") == "yes" && !isMapReduceCmd && !isTailCmd - - if useChannelless { - dlog.Server.Debug("Using channelless processing mode for mode:", r.mode) - r.startChannelless(ctx, ltx, args, re, retries) - return - } - - dlog.Server.Debug("Using channel-based processing mode for mode:", r.mode) - - // In serverless mode, can also read data from pipe - // e.g.: grep foo bar.log | dmap 'from STATS select ...' - // 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) - return - } - - dlog.Server.Debug("Reading data from file(s)") - r.readGlob(ctx, ltx, args[1], re, retries) -} - -func (r *readCommand) readGlob(ctx context.Context, ltx lcontext.LContext, - glob string, re regex.Regex, retries int) { - - retryInterval := time.Second * 5 - glob = filepath.Clean(glob) - - for retryCount := 0; retryCount < retries; retryCount++ { - paths, err := filepath.Glob(glob) - if err != nil { - dlog.Server.Warn(r.server.user, glob, err) - time.Sleep(retryInterval) - continue - } - - if numPaths := len(paths); numPaths == 0 { - dlog.Server.Error(r.server.user, "No such file(s) to read", glob) - r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user, - "Unable to read file(s), check server logs")) - select { - case <-ctx.Done(): - return - default: - } - time.Sleep(retryInterval) - continue - } - - r.readFiles(ctx, ltx, paths, glob, re, retryInterval) - return - } - - r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user, - "Giving up to read file(s)")) - return -} - -func (r *readCommand) readFiles(ctx context.Context, ltx lcontext.LContext, - paths []string, glob string, re regex.Regex, retryInterval time.Duration) { - - var wg sync.WaitGroup - wg.Add(len(paths)) - for _, path := range paths { - go r.readFileIfPermissions(ctx, ltx, &wg, path, glob, re) - } - wg.Wait() -} - -func (r *readCommand) readFileIfPermissions(ctx context.Context, ltx lcontext.LContext, - wg *sync.WaitGroup, path, glob string, re regex.Regex) { - - defer wg.Done() - globID := r.makeGlobID(path, glob) - if !r.server.user.HasFilePermission(path, "readfiles") { - dlog.Server.Error(r.server.user, "No permission to read file", path, globID) - r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user, - "Unable to read file(s), check server logs")) - return - } - r.read(ctx, ltx, path, globID, re) + // Always use channelless mode now - channel-based code is deprecated + dlog.Server.Debug("Using channelless processing mode for mode:", r.mode) + r.startChannelless(ctx, ltx, args, re, retries, queryStr) } -func (r *readCommand) read(ctx context.Context, ltx lcontext.LContext, - path, globID string, re regex.Regex) { - - dlog.Server.Info(r.server.user, "Start reading", path, globID) - var reader fs.FileReader - var limiter chan struct{} - - switch r.mode { - case omode.GrepClient, omode.CatClient: - reader = fs.NewCatFile(path, globID, r.server.serverMessages) - limiter = r.server.catLimiter - case omode.TailClient: - fallthrough - default: - reader = fs.NewTailFile(path, globID, r.server.serverMessages) - limiter = r.server.tailLimiter - } - - defer func() { - select { - case <-limiter: - default: - } - }() - - select { - case limiter <- struct{}{}: - case <-ctx.Done(): - return - default: - dlog.Server.Info("Server limit hit, queueing file", len(limiter), path) - select { - case limiter <- struct{}{}: - dlog.Server.Info("Server limit OK now, processing file", len(limiter), path) - case <-ctx.Done(): - return - } - } - - lines := r.server.lines - aggregate := r.server.aggregate - - for { - if aggregate != nil { - lines = make(chan *line.Line, 100) - aggregate.NextLinesCh <- lines - } - if err := reader.Start(ctx, ltx, lines, re); err != nil { - dlog.Server.Error(r.server.user, path, globID, err) - } - if aggregate != nil { - // Also makes aggregate to Flush - close(lines) - } - - select { - case <-ctx.Done(): - return - default: - if !reader.Retry() { - return - } - } - time.Sleep(time.Second * 2) - dlog.Server.Info(path, globID, "Reading file again") - } -} func (r *readCommand) makeGlobID(path, glob string) string { var idParts []string @@ -238,22 +90,22 @@ func (r *readCommand) isInputFromPipe() bool { // startChannelless implements channelless processing for better performance func (r *readCommand) startChannelless(ctx context.Context, ltx lcontext.LContext, - args []string, re regex.Regex, retries int) { + args []string, re regex.Regex, retries int, queryStr string) { // Handle stdin input in serverless mode if (args[1] == "" || args[1] == "-") && r.isInputFromPipe() { dlog.Server.Debug("Reading data from stdin pipe (channelless)") - r.readChannellessStdin(ctx, ltx, re) + r.readChannellessStdin(ctx, ltx, re, queryStr) return } dlog.Server.Debug("Reading data from file(s) (channelless)") - r.readGlobChannelless(ctx, ltx, args[1], re, retries) + r.readGlobChannelless(ctx, ltx, args[1], re, retries, queryStr) } // readGlobChannelless processes files using channelless approach func (r *readCommand) readGlobChannelless(ctx context.Context, ltx lcontext.LContext, - glob string, re regex.Regex, retries int) { + glob string, re regex.Regex, retries int, queryStr string) { retryInterval := time.Second * 5 glob = filepath.Clean(glob) @@ -279,7 +131,7 @@ func (r *readCommand) readGlobChannelless(ctx context.Context, ltx lcontext.LCon continue } - r.readFilesChannelless(ctx, ltx, paths, glob, re) + r.readFilesChannelless(ctx, ltx, paths, glob, re, queryStr) return } @@ -289,21 +141,26 @@ func (r *readCommand) readGlobChannelless(ctx context.Context, ltx lcontext.LCon // readFilesChannelless processes multiple files using channelless approach func (r *readCommand) readFilesChannelless(ctx context.Context, ltx lcontext.LContext, - paths []string, glob string, re regex.Regex) { - - // Create network output writer (use nil connection for now - will write to stdout) - output := NewNetworkOutputWriter(nil, r.server.serverMessages, r.server.user) + paths []string, glob string, re regex.Regex, queryStr string) { + + // Choose output writer based on server mode + var output io.Writer + if r.server.serverless { + // In serverless mode, write directly to stdout + output = os.Stdout + } else { + // In client-server mode, write to server handler lines channel + output = NewServerHandlerWriter(r.server, r.server.serverMessages, r.server.user) + } // Create appropriate processor based on mode - processor := r.createChannellessProcessor(re, ltx) + processor, needsFollowing := r.createChannellessProcessor(re, ltx, output, queryStr) // Process each file for _, path := range paths { // Generate globID just like the original system globID := r.makeGlobID(path, glob) - // Create direct processor with proper globID - directProcessor := fs.NewDirectProcessor(processor, output, globID, ltx) if !r.server.user.HasFilePermission(path, "readfiles") { dlog.Server.Error(r.server.user, "No permission to read file", path) r.server.sendln(r.server.serverMessages, dlog.Server.Warn(r.server.user, @@ -313,21 +170,59 @@ func (r *readCommand) readFilesChannelless(ctx context.Context, ltx lcontext.LCo dlog.Server.Info(r.server.user, "Start reading (channelless)", path) - if err := directProcessor.ProcessFile(ctx, path); err != nil { - dlog.Server.Error(r.server.user, path, err) - r.server.sendln(r.server.serverMessages, dlog.Server.Error(r.server.user, - "Error processing file", path, err)) + // Handle file following for tail operations + if needsFollowing { + // For aggregate processors, we need to use following with the aggregate processor + if aggregateProcessor, ok := processor.(*fs.AggregateLineProcessor); ok { + // Create a DirectProcessor wrapper that supports following + directProcessor := fs.NewDirectProcessor(aggregateProcessor, output, globID, ltx) + if err := directProcessor.ProcessFileWithTailing(ctx, path); err != nil { + dlog.Server.Error(r.server.user, path, err) + r.server.sendln(r.server.serverMessages, dlog.Server.Error(r.server.user, + "Error processing file", path, err)) + } + } else if tailProcessor, ok := processor.(*fs.TailProcessor); ok { + followingProcessor := fs.NewFollowingTailProcessor(tailProcessor, output, globID, ltx) + if err := followingProcessor.ProcessFileWithFollowing(ctx, path); err != nil { + dlog.Server.Error(r.server.user, path, err) + r.server.sendln(r.server.serverMessages, dlog.Server.Error(r.server.user, + "Error processing file", path, err)) + } + } else { + // Fallback to regular processing + directProcessor := fs.NewDirectProcessor(processor, output, globID, ltx) + if err := directProcessor.ProcessFile(ctx, path); err != nil { + dlog.Server.Error(r.server.user, path, err) + r.server.sendln(r.server.serverMessages, dlog.Server.Error(r.server.user, + "Error processing file", path, err)) + } + } + } else { + // Regular file processing + directProcessor := fs.NewDirectProcessor(processor, output, globID, ltx) + if err := directProcessor.ProcessFile(ctx, path); err != nil { + dlog.Server.Error(r.server.user, path, err) + r.server.sendln(r.server.serverMessages, dlog.Server.Error(r.server.user, + "Error processing file", path, err)) + } } } } // readChannellessStdin processes stdin using channelless approach -func (r *readCommand) readChannellessStdin(ctx context.Context, ltx lcontext.LContext, re regex.Regex) { - // Create network output writer (use nil connection for now - will write to stdout) - output := NewNetworkOutputWriter(nil, r.server.serverMessages, r.server.user) +func (r *readCommand) readChannellessStdin(ctx context.Context, ltx lcontext.LContext, re regex.Regex, queryStr string) { + // Choose output writer based on server mode + var output io.Writer + if r.server.serverless { + // In serverless mode, write directly to stdout + output = os.Stdout + } else { + // In client-server mode, write to server handler lines channel + output = NewServerHandlerWriter(r.server, r.server.serverMessages, r.server.user) + } // Create appropriate processor based on mode - processor := r.createChannellessProcessor(re, ltx) + processor, _ := r.createChannellessProcessor(re, ltx, output, queryStr) // Create direct processor with "-" as globID for stdin directProcessor := fs.NewDirectProcessor(processor, output, "-", ltx) @@ -355,23 +250,87 @@ func (r *readCommand) isMapReduceCommand(re regex.Regex) bool { } // createChannellessProcessor creates the appropriate processor based on command mode -func (r *readCommand) createChannellessProcessor(re regex.Regex, ltx lcontext.LContext) fs.LineProcessor { +func (r *readCommand) createChannellessProcessor(re regex.Regex, ltx lcontext.LContext, output io.Writer, queryStr string) (fs.LineProcessor, bool) { hostname := r.server.hostname // Use server hostname plain := r.server.plain // Use actual plain mode from server noColor := false // Enable colors by default in channelless mode - + + // If there's an existing aggregate (from a 'map' command), we need to feed data to it + // Create a lines channel and connect it to the aggregate + if r.server.aggregate != nil { + dlog.Server.Debug("Using existing aggregate, creating bridge processor for channelless mode") + // Create a lines channel for the aggregate with larger buffer + linesCh := make(chan *line.Line, 1000) + // Connect the lines channel to the aggregate + go func() { + r.server.aggregate.NextLinesCh <- linesCh + }() + + // Create a bridge processor that feeds lines to the aggregate + var bridgeProcessor fs.LineProcessor + if r.mode == omode.TailClient { + bridgeProcessor = fs.NewAggregateLineProcessorForTail(linesCh, re, hostname, ltx) + } else { + bridgeProcessor = fs.NewAggregateLineProcessor(linesCh, re, hostname, ltx) + } + + // Determine if following is needed + needsFollowing := r.mode == omode.TailClient + return bridgeProcessor, needsFollowing + } + + // No existing aggregate - check if this is a standalone MapReduce operation + isMapReduce := r.isMapReduceCommand(re) || r.mode == omode.MapClient + switch r.mode { case omode.GrepClient: - return fs.NewGrepProcessor(re, plain, noColor, hostname, ltx.BeforeContext, ltx.AfterContext, ltx.MaxCount) + if isMapReduce && queryStr != "" { + // This is a standalone MapReduce grep operation + mapProcessor, err := fs.NewMapProcessor(plain, hostname, queryStr, output) + if err != nil { + dlog.Server.Error(r.server.user, "Failed to create MapReduce processor", err) + return fs.NewGrepProcessor(re, plain, noColor, hostname, ltx.BeforeContext, ltx.AfterContext, ltx.MaxCount), false + } + return mapProcessor, false + } + return fs.NewGrepProcessor(re, plain, noColor, hostname, ltx.BeforeContext, ltx.AfterContext, ltx.MaxCount), false case omode.CatClient: - return fs.NewCatProcessor(plain, noColor, hostname) + if isMapReduce && queryStr != "" { + // This is a standalone MapReduce cat operation + mapProcessor, err := fs.NewMapProcessor(plain, hostname, queryStr, output) + if err != nil { + dlog.Server.Error(r.server.user, "Failed to create MapReduce processor", err) + return fs.NewCatProcessor(plain, noColor, hostname), false + } + return mapProcessor, false + } + return fs.NewCatProcessor(plain, noColor, hostname), false case omode.TailClient: - // For now, basic tail without follow functionality - return fs.NewTailProcessor(re, plain, noColor, hostname, false, false, 0) + if isMapReduce && queryStr != "" { + // This is a standalone MapReduce tail operation + mapProcessor, err := fs.NewMapProcessor(plain, hostname, queryStr, output) + if err != nil { + dlog.Server.Error(r.server.user, "Failed to create MapReduce processor", err) + return fs.NewTailProcessor(re, plain, noColor, hostname, true, true, 0), true + } + return mapProcessor, false + } + // Regular tail operation + return fs.NewTailProcessor(re, plain, noColor, hostname, true, true, 0), true case omode.MapClient: - return fs.NewMapProcessor(plain, hostname) + // Direct MapReduce client - should have queryStr + if queryStr != "" { + mapProcessor, err := fs.NewMapProcessor(plain, hostname, queryStr, output) + if err != nil { + dlog.Server.Error(r.server.user, "Failed to create MapReduce processor", err) + return fs.NewGrepProcessor(re, plain, noColor, hostname, ltx.BeforeContext, ltx.AfterContext, ltx.MaxCount), false + } + return mapProcessor, false + } + // Fallback + return fs.NewGrepProcessor(re, plain, noColor, hostname, ltx.BeforeContext, ltx.AfterContext, ltx.MaxCount), false default: // Default to grep behavior - return fs.NewGrepProcessor(re, plain, noColor, hostname, ltx.BeforeContext, ltx.AfterContext, ltx.MaxCount) + return fs.NewGrepProcessor(re, plain, noColor, hostname, ltx.BeforeContext, ltx.AfterContext, ltx.MaxCount), false } } diff --git a/scripts/benchmark_channelless.sh b/scripts/benchmark_channelless.sh deleted file mode 100755 index 4ac532f..0000000 --- a/scripts/benchmark_channelless.sh +++ /dev/null @@ -1,215 +0,0 @@ -#!/bin/bash - -# Comprehensive benchmark: Channel-based vs Channelless Cat Implementation -# Tests performance improvements achieved by eliminating channel overhead - -set -e - -echo "=== DTail Channelless Performance Benchmark ===" -echo "Comparing channel-based vs channelless cat implementation" -echo "Date: $(date)" -echo - -# Test configuration -TEST_FILES=("test_100mb.txt" "test_200mb.txt") -ITERATIONS=5 -WARMUP_RUNS=2 - -# Results storage -RESULTS_DIR="benchmark_results_$(date +%Y%m%d_%H%M%S)" -mkdir -p "$RESULTS_DIR" - -# Ensure we're in the correct directory -cd "$(dirname "$0")/.." - -# Build both implementations -echo "Building DTail binaries..." -make clean > /dev/null 2>&1 -make build > /dev/null 2>&1 -echo "✓ Build complete" -echo - -# Function to run benchmark for a specific configuration -run_benchmark() { - local use_channelless=$1 - local test_file=$2 - local impl_name=$3 - local results_file="$RESULTS_DIR/${impl_name}_$(basename $test_file .txt).results" - - echo "Testing $impl_name with $test_file..." - - # Warmup runs - for ((i=1; i<=WARMUP_RUNS; i++)); do - echo -n " Warmup $i/$WARMUP_RUNS... " - DTAIL_USE_CHANNELLESS=$use_channelless DTAIL_INTEGRATION_TEST_RUN_MODE=yes \ - timeout 30s ./dcat --logLevel error --cfg none "scripts/$test_file" > /dev/null 2>&1 - echo "done" - done - - # Actual benchmark runs - echo " Running $ITERATIONS benchmark iterations:" - for ((i=1; i<=ITERATIONS; i++)); do - echo -n " Run $i/$ITERATIONS... " - - # Clear caches - sync - echo 3 > /proc/sys/vm/drop_caches 2>/dev/null || true - - # Run benchmark with time measurement - start_time=$(date +%s.%N) - DTAIL_USE_CHANNELLESS=$use_channelless DTAIL_INTEGRATION_TEST_RUN_MODE=yes \ - timeout 30s ./dcat --logLevel error --cfg none "scripts/$test_file" > /dev/null 2>&1 - end_time=$(date +%s.%N) - - # Calculate duration - duration=$(echo "$end_time - $start_time" | bc -l) - echo "$duration" >> "$results_file" - - printf "%.3fs\n" "$duration" - done - echo -} - -# Function to calculate statistics -calculate_stats() { - local file=$1 - local values=($(cat "$file")) - local sum=0 - local count=${#values[@]} - - # Calculate mean - for val in "${values[@]}"; do - sum=$(echo "$sum + $val" | bc -l) - done - local mean=$(echo "scale=6; $sum / $count" | bc -l) - - # Calculate standard deviation - local variance_sum=0 - for val in "${values[@]}"; do - local diff=$(echo "$val - $mean" | bc -l) - local squared=$(echo "$diff * $diff" | bc -l) - variance_sum=$(echo "$variance_sum + $squared" | bc -l) - done - local variance=$(echo "scale=6; $variance_sum / $count" | bc -l) - local stddev=$(echo "scale=6; sqrt($variance)" | bc -l) - - # Find min and max - local min=${values[0]} - local max=${values[0]} - for val in "${values[@]}"; do - if (( $(echo "$val < $min" | bc -l) )); then - min=$val - fi - if (( $(echo "$val > $max" | bc -l) )); then - max=$val - fi - done - - echo "$mean $stddev $min $max" -} - -# Function to calculate throughput -calculate_throughput() { - local file_size_mb=$1 - local time_seconds=$2 - echo "scale=2; $file_size_mb / $time_seconds" | bc -l -} - -# Run benchmarks -echo "Starting benchmarks..." -echo - -for test_file in "${TEST_FILES[@]}"; do - echo "=== Benchmarking with $test_file ===" - - # Get file size in MB - file_size_bytes=$(stat -c%s "scripts/$test_file") - file_size_mb=$(echo "scale=2; $file_size_bytes / 1024 / 1024" | bc -l) - echo "File size: ${file_size_mb} MB" - echo - - # Test channel-based implementation - run_benchmark "false" "$test_file" "channel_based" - - # Test channelless implementation - run_benchmark "true" "$test_file" "channelless" - - echo "--- Results for $test_file ---" - - # Calculate statistics for channel-based - channel_stats=($(calculate_stats "$RESULTS_DIR/channel_based_$(basename $test_file .txt).results")) - channel_mean=${channel_stats[0]} - channel_stddev=${channel_stats[1]} - channel_min=${channel_stats[2]} - channel_max=${channel_stats[3]} - channel_throughput=$(calculate_throughput "$file_size_mb" "$channel_mean") - - # Calculate statistics for channelless - channelless_stats=($(calculate_stats "$RESULTS_DIR/channelless_$(basename $test_file .txt).results")) - channelless_mean=${channelless_stats[0]} - channelless_stddev=${channelless_stats[1]} - channelless_min=${channelless_stats[2]} - channelless_max=${channelless_stats[3]} - channelless_throughput=$(calculate_throughput "$file_size_mb" "$channelless_mean") - - # Calculate improvement - improvement=$(echo "scale=2; (($channel_mean - $channelless_mean) / $channel_mean) * 100" | bc -l) - speedup=$(echo "scale=2; $channel_mean / $channelless_mean" | bc -l) - throughput_improvement=$(echo "scale=2; (($channelless_throughput - $channel_throughput) / $channel_throughput) * 100" | bc -l) - - echo "Channel-based:" - printf " Time: %.3f ± %.3f seconds (min: %.3f, max: %.3f)\n" "$channel_mean" "$channel_stddev" "$channel_min" "$channel_max" - printf " Throughput: %.2f MB/s\n" "$channel_throughput" - echo - echo "Channelless:" - printf " Time: %.3f ± %.3f seconds (min: %.3f, max: %.3f)\n" "$channelless_mean" "$channelless_stddev" "$channelless_min" "$channelless_max" - printf " Throughput: %.2f MB/s\n" "$channelless_throughput" - echo - echo "Performance Improvement:" - printf " Time reduction: %.2f%% (%.2fx speedup)\n" "$improvement" "$speedup" - printf " Throughput increase: %.2f%%\n" "$throughput_improvement" - echo - echo "==========================================" - echo -done - -# Generate summary report -echo "=== BENCHMARK SUMMARY ===" -echo - -summary_file="$RESULTS_DIR/benchmark_summary.txt" -{ - echo "DTail Channelless Performance Benchmark Summary" - echo "Date: $(date)" - echo "Iterations per test: $ITERATIONS" - echo "Warmup runs: $WARMUP_RUNS" - echo - - for test_file in "${TEST_FILES[@]}"; do - file_size_bytes=$(stat -c%s "scripts/$test_file") - file_size_mb=$(echo "scale=2; $file_size_bytes / 1024 / 1024" | bc -l) - - channel_stats=($(calculate_stats "$RESULTS_DIR/channel_based_$(basename $test_file .txt).results")) - channelless_stats=($(calculate_stats "$RESULTS_DIR/channelless_$(basename $test_file .txt).results")) - - channel_mean=${channel_stats[0]} - channelless_mean=${channelless_stats[0]} - - improvement=$(echo "scale=2; (($channel_mean - $channelless_mean) / $channel_mean) * 100" | bc -l) - speedup=$(echo "scale=2; $channel_mean / $channelless_mean" | bc -l) - - channel_throughput=$(calculate_throughput "$file_size_mb" "$channel_mean") - channelless_throughput=$(calculate_throughput "$file_size_mb" "$channelless_mean") - - echo "$test_file (${file_size_mb} MB):" - printf " Channel-based: %.3f seconds (%.2f MB/s)\n" "$channel_mean" "$channel_throughput" - printf " Channelless: %.3f seconds (%.2f MB/s)\n" "$channelless_mean" "$channelless_throughput" - printf " Improvement: %.2f%% faster (%.2fx speedup)\n" "$improvement" "$speedup" - echo - done -} | tee "$summary_file" - -echo "Detailed results saved in: $RESULTS_DIR/" -echo "Summary report: $summary_file" -echo -echo "=== BENCHMARK COMPLETE ===" \ No newline at end of file diff --git a/scripts/corrected_benchmark.sh b/scripts/corrected_benchmark.sh deleted file mode 100755 index aa42aec..0000000 --- a/scripts/corrected_benchmark.sh +++ /dev/null @@ -1,89 +0,0 @@ -#!/bin/bash - -# Corrected benchmark: Channel-based vs Channelless Cat Implementation -# This accounts for the fact that channel-based doesn't process all data - -set -e - -echo "=== CORRECTED DTail Channelless Performance Benchmark ===" -echo "Channel-based implementation appears to have a bug - it only processes ~67% of data" -echo "Benchmarking actual throughput per line processed" -echo - -# Test with 100MB file -TEST_FILE="scripts/test_100mb.txt" -TOTAL_LINES=$(wc -l < "$TEST_FILE") -FILE_SIZE_MB=$(echo "scale=2; $(stat -c%s "$TEST_FILE") / 1024 / 1024" | bc -l) - -echo "Test file: $TEST_FILE" -echo "Total lines in file: $TOTAL_LINES" -echo "File size: ${FILE_SIZE_MB} MB" -echo - -# Run both implementations and measure -echo "Testing channel-based implementation..." -start_time=$(date +%s.%N) -CHANNEL_LINES=$(DTAIL_USE_CHANNELLESS=false DTAIL_INTEGRATION_TEST_RUN_MODE=yes ./dcat --logLevel error --cfg none "$TEST_FILE" | wc -l) -end_time=$(date +%s.%N) -channel_time=$(echo "$end_time - $start_time" | bc -l) - -echo "Testing channelless implementation..." -start_time=$(date +%s.%N) -CHANNELLESS_LINES=$(DTAIL_USE_CHANNELLESS=true DTAIL_INTEGRATION_TEST_RUN_MODE=yes ./dcat --logLevel error --cfg none "$TEST_FILE" | wc -l) -end_time=$(date +%s.%N) -channelless_time=$(echo "$end_time - $start_time" | bc -l) - -# Calculate metrics -channel_throughput_lines=$(echo "scale=2; $CHANNEL_LINES / $channel_time" | bc -l) -channelless_throughput_lines=$(echo "scale=2; $CHANNELLESS_LINES / $channelless_time" | bc -l) - -channel_coverage=$(echo "scale=2; ($CHANNEL_LINES * 100) / $TOTAL_LINES" | bc -l) -channelless_coverage=$(echo "scale=2; ($CHANNELLESS_LINES * 100) / $TOTAL_LINES" | bc -l) - -# Effective data processed -channel_data_mb=$(echo "scale=2; ($CHANNEL_LINES * $FILE_SIZE_MB) / $TOTAL_LINES" | bc -l) -channelless_data_mb=$FILE_SIZE_MB - -channel_throughput_mb=$(echo "scale=2; $channel_data_mb / $channel_time" | bc -l) -channelless_throughput_mb=$(echo "scale=2; $channelless_data_mb / $channelless_time" | bc -l) - -# Calculate relative performance for same amount of work -extrapolated_channel_time=$(echo "scale=2; ($channel_time * $TOTAL_LINES) / $CHANNEL_LINES" | bc -l) -performance_improvement=$(echo "scale=2; (($extrapolated_channel_time - $channelless_time) / $extrapolated_channel_time) * 100" | bc -l) -speedup=$(echo "scale=2; $extrapolated_channel_time / $channelless_time" | bc -l) - -echo -echo "=== RESULTS ===" -echo -echo "Channel-based implementation:" -printf " Time: %.3f seconds\n" "$channel_time" -printf " Lines processed: %d (%.1f%% of file)\n" "$CHANNEL_LINES" "$channel_coverage" -printf " Data processed: %.2f MB\n" "$channel_data_mb" -printf " Throughput: %.0f lines/sec, %.2f MB/s\n" "$channel_throughput_lines" "$channel_throughput_mb" -printf " Extrapolated time for full file: %.3f seconds\n" "$extrapolated_channel_time" -echo - -echo "Channelless implementation:" -printf " Time: %.3f seconds\n" "$channelless_time" -printf " Lines processed: %d (%.1f%% of file)\n" "$CHANNELLESS_LINES" "$channelless_coverage" -printf " Data processed: %.2f MB\n" "$channelless_data_mb" -printf " Throughput: %.0f lines/sec, %.2f MB/s\n" "$channelless_throughput_lines" "$channelless_throughput_mb" -echo - -echo "Performance comparison (for processing complete file):" -printf " Channelless improvement: %.2f%% faster\n" "$performance_improvement" -printf " Speedup: %.2fx\n" "$speedup" -echo - -if (( $(echo "$performance_improvement > 0" | bc -l) )); then - echo "✅ Channelless implementation is FASTER and processes ALL data correctly" -else - echo "❌ Channelless implementation is slower" -fi -echo - -echo "=== CONCLUSION ===" -echo "The channel-based implementation has a bug where it stops processing" -echo "at approximately 67% of the input file. This makes direct time comparisons" -echo "invalid. When extrapolated to process the same amount of data, the" -echo "channelless implementation shows the expected performance improvement." \ No newline at end of file diff --git a/scripts/profile_channelless.sh b/scripts/profile_channelless.sh deleted file mode 100755 index fb6ec3d..0000000 --- a/scripts/profile_channelless.sh +++ /dev/null @@ -1,50 +0,0 @@ -#!/bin/bash - -# Profile channelless vs channel-based implementations to understand performance difference - -set -e - -echo "=== Profiling Channelless vs Channel-based Cat Implementation ===" -echo - -# Build with profiling enabled -echo "Building DTail binaries..." -make clean > /dev/null 2>&1 -make build > /dev/null 2>&1 - -echo "Profiling channel-based implementation..." -DTAIL_USE_CHANNELLESS=false DTAIL_INTEGRATION_TEST_RUN_MODE=yes \ - go tool pprof -cpuprofile=channel_based_cpu.prof \ - -o channel_based_cpu.prof \ - -- ./dcat --logLevel error --cfg none scripts/test_100mb.txt > /dev/null 2>&1 & -CHANNEL_PID=$! - -# Profile with Go's built-in profiling -DTAIL_USE_CHANNELLESS=false DTAIL_INTEGRATION_TEST_RUN_MODE=yes \ - timeout 10s go run -cpuprofile=channel_based_go.prof ./cmd/dcat/main.go --logLevel error --cfg none scripts/test_100mb.txt > /dev/null 2>&1 || true - -echo "Profiling channelless implementation..." -DTAIL_USE_CHANNELLESS=true DTAIL_INTEGRATION_TEST_RUN_MODE=yes \ - timeout 10s go run -cpuprofile=channelless_go.prof ./cmd/dcat/main.go --logLevel error --cfg none scripts/test_100mb.txt > /dev/null 2>&1 || true - -echo "Analyzing profiles..." - -echo -echo "=== Channel-based CPU Profile ===" -if [ -f channel_based_go.prof ]; then - go tool pprof -top -cum channel_based_go.prof | head -20 -else - echo "Channel-based profile not found" -fi - -echo -echo "=== Channelless CPU Profile ===" -if [ -f channelless_go.prof ]; then - go tool pprof -top -cum channelless_go.prof | head -20 -else - echo "Channelless profile not found" -fi - -echo -echo "Profile files generated:" -ls -la *_go.prof 2>/dev/null || echo "No profile files found" \ No newline at end of file -- cgit v1.2.3