fix(usenet): optimize memory by using lazy buffer allocation and pooling in BufferedPipe

Author: drondeseries

internal/utils/bufpipe.go

@@ -8,25 +8,36 @@ import (
 
 var (
 	ErrClosedPipe = errors.New("bufpipe: closed pipe")
+	// Global pool for 1MB buffers to reduce GC pressure and memory spikes
+	bufferPool = sync.Pool{
+		New: func() any {
+			b := make([]byte, 1024*1024)
+			return &b
+		},
+	}
 )
 
 // BufferedPipe is a thread-safe pipe with an internal buffer.
 // It allows the writer to continue writing as long as there is space in the buffer,
 // and the reader to read available data.
+// Memory optimization: The buffer is only allocated from a pool on the first Write call.
 type BufferedPipe struct {
-	buf    []byte
+	buf    *[]byte
 	head   int
 	tail   int
 	closed bool
 	err    error
 	cond   *sync.Cond
 	mu     sync.Mutex
+	
+	// Track if buffer was returned to pool
+	released bool
 }
 
+// NewBufferedPipe creates a new buffered pipe. The size parameter is currently
+// ignored to maintain compatibility with the fixed-size buffer pool (1MB).
 func NewBufferedPipe(size int) *BufferedPipe {
-	p := &BufferedPipe{
-		buf: make([]byte, size),
-	}
+	p := &BufferedPipe{}
 	p.cond = sync.NewCond(&p.mu)
 	return p
 }
@@ -40,27 +51,33 @@ func (p *BufferedPipe) Write(data []byte) (n int, err error) {
 			return n, ErrClosedPipe
 		}
 
+		// Lazy allocation: only get a buffer from the pool when we actually start writing
+		if p.buf == nil {
+			p.buf = bufferPool.Get().(*[]byte)
+		}
+
 		free := p.freeSpace()
 		if free == 0 {
 			p.cond.Wait()
 			continue
 		}
 
+		buf := *p.buf
 		chunkSize := len(data) - n
 		if chunkSize > free {
 			chunkSize = free
 		}
 
-		// Calculate copy length up to the end of the buffer
+		// Calculate copy length up to the end of the buffer (linear part)
 		end := p.tail + chunkSize
-		if end > len(p.buf) {
-			end = len(p.buf)
+		if end > len(buf) {
+			end = len(buf)
 		}
 
 		copyLen := end - p.tail
-		copy(p.buf[p.tail:], data[n:n+copyLen])
+		copy(buf[p.tail:], data[n:n+copyLen])
 
-		p.tail = (p.tail + copyLen) % len(p.buf)
+		p.tail = (p.tail + copyLen) % len(buf)
 		n += copyLen
 
 		p.cond.Broadcast()
@@ -89,20 +106,28 @@ func (p *BufferedPipe) Read(data []byte) (n int, err error) {
 		n = available
 	}
 
-	// Calculate copy length up to the end of the buffer
+	buf := *p.buf
+	// Calculate copy length up to the end of the buffer (linear part)
 	end := p.head + n
-	if end > len(p.buf) {
-		end = len(p.buf)
+	if end > len(buf) {
+		end = len(buf)
 	}
 
 	copyLen := end - p.head
-	copy(data, p.buf[p.head:end])
+	copy(data, buf[p.head:end])
 
+	// If there's more to read (wrapped around), copy from the beginning
 	if n > copyLen {
-		copy(data[copyLen:], p.buf[:n-copyLen])
+		copy(data[copyLen:], buf[:n-copyLen])
+	}
+
+	p.head = (p.head + n) % len(buf)
+	
+	// If we just read the last byte and the pipe is closed, we can release the buffer
+	if p.closed && p.availableData() == 0 {
+		p.releaseBuffer()
 	}
 
-	p.head = (p.head + n) % len(p.buf)
 	p.cond.Broadcast()
 
 	return n, nil
@@ -120,24 +145,45 @@ func (p *BufferedPipe) CloseWithError(err error) error {
 	}
 	p.closed = true
 	p.err = err
+	
+	// If buffer is empty or was never allocated, we can release it now
+	if p.availableData() == 0 {
+		p.releaseBuffer()
+	}
+
 	p.cond.Broadcast()
 	return nil
 }
 
+func (p *BufferedPipe) releaseBuffer() {
+	if p.buf != nil && !p.released {
+		bufferPool.Put(p.buf)
+		p.buf = nil
+		p.released = true
+	}
+}
+
 func (p *BufferedPipe) availableData() int {
+	if p.buf == nil {
+		return 0
+	}
+	bufLen := len(*p.buf)
 	if p.tail >= p.head {
 		return p.tail - p.head
 	}
-	return len(p.buf) - p.head + p.tail
+	return bufLen - p.head + p.tail
 }
 
 func (p *BufferedPipe) freeSpace() int {
-	return len(p.buf) - p.availableData() - 1
+	if p.buf == nil {
+		return 1024*1024 - 1 // Default size minus 1 for head/tail boundary
+	}
+	return len(*p.buf) - p.availableData() - 1
 }
 
 func (p *BufferedPipe) getErr() error {
 	if p.err == nil {
 		return io.EOF
 	}
 	return p.err
-}
+}