diff options
Diffstat (limited to 'internal')
| -rw-r--r-- | internal/clients/baseclient.go | 19 | ||||
| -rw-r--r-- | internal/clients/stats.go | 15 | ||||
| -rw-r--r-- | internal/io/fs/chunkedreader.go | 63 |
3 files changed, 84 insertions, 13 deletions
diff --git a/internal/clients/baseclient.go b/internal/clients/baseclient.go index 3025f72..29a9cfc 100644 --- a/internal/clients/baseclient.go +++ b/internal/clients/baseclient.go @@ -15,6 +15,9 @@ import ( gossh "golang.org/x/crypto/ssh" ) +// Reusable timer for retry delays - PBO optimization +var retryTimer = time.NewTimer(2 * time.Second) + // This is the main client data structure. type baseClient struct { config.Args @@ -124,8 +127,20 @@ func (c *baseClient) startConnection(ctx context.Context, i int, default: } - // Yes, we want to retry. - time.Sleep(time.Second * 2) + // Yes, we want to retry using reusable timer - PBO optimization + if !retryTimer.Stop() { + // Drain timer channel if it fired + select { + case <-retryTimer.C: + default: + } + } + retryTimer.Reset(2 * time.Second) + select { + case <-retryTimer.C: + case <-ctx.Done(): + return + } dlog.Client.Debug(conn.Server(), "Reconnecting") conn = c.makeConnection(conn.Server(), c.sshAuthMethods, c.hostKeyCallback) c.connections[i] = conn diff --git a/internal/clients/stats.go b/internal/clients/stats.go index 2da3cf7..9a17899 100644 --- a/internal/clients/stats.go +++ b/internal/clients/stats.go @@ -14,6 +14,9 @@ import ( "github.com/mimecast/dtail/internal/protocol" ) +// Reusable timer to reduce allocations - PBO optimization +var statsTimer = time.NewTimer(3 * time.Second) + // Used to collect and display various client stats. type stats struct { // Total amount servers to connect to. @@ -44,11 +47,21 @@ func (s *stats) Start(ctx context.Context, throttleCh <-chan struct{}, var force bool var messages []string + // Reset the reusable timer to reduce allocations - PBO optimization + if !statsTimer.Stop() { + // Drain timer channel if it fired + select { + case <-statsTimer.C: + default: + } + } + statsTimer.Reset(3 * time.Second) + select { case message := <-statsCh: messages = append(messages, message) force = true - case <-time.After(time.Second * 3): + case <-statsTimer.C: case <-ctx.Done(): return } diff --git a/internal/io/fs/chunkedreader.go b/internal/io/fs/chunkedreader.go index 5f16c12..ab78ba1 100644 --- a/internal/io/fs/chunkedreader.go +++ b/internal/io/fs/chunkedreader.go @@ -10,6 +10,9 @@ import ( "github.com/mimecast/dtail/internal/io/pool" ) +// Reusable timer to reduce allocations - PBO optimization +var sharedTimer = time.NewTimer(10 * time.Millisecond) + // ChunkedReader reads data in large chunks and processes it line by line // This replaces the byte-by-byte reading approach for better performance type ChunkedReader struct { @@ -18,6 +21,9 @@ type ChunkedReader struct { remaining []byte // Partial line from previous chunk chunkSize int eof bool + // PBO optimization: Pre-allocate line buffer to reduce allocations + lineBuffer []byte + lineLen int } // NewChunkedReader creates a new chunked reader with the specified chunk size @@ -26,9 +32,11 @@ func NewChunkedReader(reader io.Reader, chunkSize int) *ChunkedReader { chunkSize = 64 * 1024 // Default 64KB chunks } return &ChunkedReader{ - reader: reader, - buffer: make([]byte, chunkSize), - chunkSize: chunkSize, + reader: reader, + buffer: make([]byte, chunkSize), + chunkSize: chunkSize, + // PBO optimization: Pre-allocate line buffer + lineBuffer: make([]byte, 0, 8192), // 8KB initial capacity } } @@ -60,11 +68,19 @@ func (cr *ChunkedReader) ProcessLines(ctx context.Context, rawLines chan *bytes. return nil } else { // In tailing mode - EOF means wait and try again - // Use shorter polling interval for better responsiveness to rapid writes + // Use shared timer to reduce allocations - PBO optimization + if !sharedTimer.Stop() { + // Drain timer channel if it fired + select { + case <-sharedTimer.C: + default: + } + } + sharedTimer.Reset(10 * time.Millisecond) select { case <-ctx.Done(): return ctx.Err() - case <-time.After(10 * time.Millisecond): + case <-sharedTimer.C: // Continue reading after brief pause continue } @@ -97,32 +113,54 @@ func (cr *ChunkedReader) ProcessLines(ctx context.Context, rawLines chan *bytes. return nil } - // Process data and extract complete lines + // Process data and extract complete lines - PBO optimized + // Reset line buffer for this chunk + cr.lineBuffer = cr.lineBuffer[:0] + cr.lineLen = 0 + for _, b := range cr.remaining { - message.WriteByte(b) + // Use pre-allocated buffer to reduce byte-by-byte WriteByte calls + if cr.lineLen < len(cr.lineBuffer) { + cr.lineBuffer[cr.lineLen] = b + } else { + cr.lineBuffer = append(cr.lineBuffer, b) + } + cr.lineLen++ switch b { case '\n': - // Send the complete line + // Send the complete line using Write for bulk operation + message.Write(cr.lineBuffer[:cr.lineLen]) select { case rawLines <- message: message = pool.BytesBuffer.Get().(*bytes.Buffer) warnedAboutLongLine = false + // Reset line buffer for next line + cr.lineLen = 0 case <-ctx.Done(): return ctx.Err() } default: // Check line length limit - if message.Len() >= maxLineLength { + if cr.lineLen >= maxLineLength { if !warnedAboutLongLine { serverMessages <- dlog.Common.Warn(filePath, "Long log line, splitting into multiple lines") + "\n" warnedAboutLongLine = true } - message.WriteByte('\n') + // Add newline to current buffer and send + if cr.lineLen < len(cr.lineBuffer) { + cr.lineBuffer[cr.lineLen] = '\n' + } else { + cr.lineBuffer = append(cr.lineBuffer, '\n') + } + cr.lineLen++ + message.Write(cr.lineBuffer[:cr.lineLen]) select { case rawLines <- message: message = pool.BytesBuffer.Get().(*bytes.Buffer) + // Reset line buffer for next line + cr.lineLen = 0 case <-ctx.Done(): return ctx.Err() } @@ -130,6 +168,11 @@ func (cr *ChunkedReader) ProcessLines(ctx context.Context, rawLines chan *bytes. } } + // If we have remaining data in line buffer, add it to message + if cr.lineLen > 0 { + message.Write(cr.lineBuffer[:cr.lineLen]) + } + // Clear the remaining buffer - any partial line is now in the message buffer cr.remaining = nil } |