route log buffer flush copies through the shared slab pool (#10336)

* route log buffer flush copies through the shared slab pool

Each flush copied the sealed window into a bytes.Buffer drawn from a
package-local sync.Pool. GC drains that pool, so once collections run
often (e.g. under GOMEMLIMIT) most flushes miss and Write grows a fresh
window-sized array, the dominant bytes.growSlice source under write load.

Copy into a size-classed slab from weed/util/mem instead. Slabs are
reused process-wide and returned to their exact size class after the
flush, so variable-sized flushes across many partitions stop churning
mismatched buffers.

* guard nil slab in flush releaseMemory

mem.Free(nil) resolves to the smallest slot pool and stores a zero-cap
slice, so a later mem.Allocate could return nil and panic. The current
call sites never pass nil, but the guard keeps a double release harmless.

weed/util/log_buffer/log_buffer.go

@@ -15,6 +15,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
 	"github.com/seaweedfs/seaweedfs/weed/util"
+	"github.com/seaweedfs/seaweedfs/weed/util/mem"
 )
 
 const BufferSize = 8 * 1024 * 1024
@@ -34,7 +35,7 @@ var (
 type dataToFlush struct {
 	startTime time.Time
 	stopTime  time.Time
-	data      *bytes.Buffer
+	data      []byte // slab from mem.Allocate; returned via mem.Free after flush
 	minOffset int64
 	maxOffset int64
 	done      chan struct{} // Signal when flush completes
@@ -541,7 +542,7 @@ func (logBuffer *LogBuffer) loopFlush() {
 		if d == nil {
 			break // shutdown sentinel
 		}
-		logBuffer.flushFn(logBuffer, d.startTime, d.stopTime, d.data.Bytes(), d.minOffset, d.maxOffset)
+		logBuffer.flushFn(logBuffer, d.startTime, d.stopTime, d.data, d.minOffset, d.maxOffset)
 		d.releaseMemory()
 		// local logbuffer is different from aggregate logbuffer here
 		if d.maxOffset >= 0 {
@@ -711,8 +712,13 @@ func (logBuffer *LogBuffer) SetLastFlushTsNs(ts int64) {
 }
 
 func (d *dataToFlush) releaseMemory() {
-	d.data.Reset()
-	bufferPool.Put(d.data)
+	// Guard nil: mem.Free(nil) would put a zero-cap slice into the smallest slot
+	// pool, so a later Allocate could hand back nil and panic. Also makes a double
+	// release harmless.
+	if d.data != nil {
+		mem.Free(d.data)
+		d.data = nil
+	}
 }
 
 // ReadFromBuffer returns the in-memory log data at lastReadPosition. isPooled
@@ -958,11 +964,14 @@ var bufferPool = sync.Pool{
 	},
 }
 
-func copiedBytes(buf []byte) (copied *bytes.Buffer) {
-	copied = bufferPool.Get().(*bytes.Buffer)
-	copied.Reset()
-	copied.Write(buf)
-	return
+// copiedBytes returns a private copy of buf backed by the shared, size-classed
+// slab pool. The caller owns it until mem.Free (see dataToFlush.releaseMemory),
+// so the live buffer never outlives the flush. Routing through mem reuses slabs
+// across the process instead of reallocating a window-sized array per flush.
+func copiedBytes(buf []byte) []byte {
+	copied := mem.Allocate(len(buf))
+	copy(copied, buf)
+	return copied
 }
 
 // sharedBufferView wraps the sealed window's shared snapshot from pos onward.