filer: cut metadata-subscription allocation churn (issue #10253) (#10260)

* log_buffer: read only ts_ns during metadata-read binary search

readTs ran a full proto.Unmarshal of each LogEntry just to compare
timestamps during ReadFromBuffer's binary search, allocating fresh
slices for the data and key byte fields on every probe. Under
metadata-subscription fan-out this decode churn dominated allocations
and drove heavy GC, inflating filer RSS well past the live heap.

Scan the wire format for ts_ns (field 1) instead and skip the data/key
payloads without copying them. readTs is now zero-alloc; the binary
search no longer touches the event payloads at all.

* log_buffer: alias event payloads instead of copying them on delivery

The subscribe read loops decoded each LogEntry with a full proto.Unmarshal,
allocating fresh slices for the data and key byte fields on every entry
(protobuf consumeBytesNoZero). When many metadata subscribers each re-read
the in-memory log window under a reconnect storm, that per-entry copy was a
large share of allocation churn -- and for entries a subscriber filters out
by prefix, the copied payload was never even looked at.

Decode with a wire scan that points data/key at sub-slices of the source
buffer instead of copying. The aliased slices are valid only for the
duration of the eachLogEntryFn callback; every current subscriber either
re-decodes the event into its own struct or hands it to a synchronous grpc
Send, so none retain them. Per delivered entry the loop decode drops from
176 B / 2 allocs to zero.

weed/util/log_buffer/log_buffer.go

@@ -2,6 +2,7 @@ package log_buffer
 
 import (
 	"bytes"
+	"encoding/binary"
 	"fmt"
 	"math"
 	"sync"
@@ -936,18 +937,6 @@ var bufferPool = sync.Pool{
 	},
 }
 
-// logEntryPool reduces allocations in readTs which is called frequently during binary search
-var logEntryPool = sync.Pool{
-	New: func() interface{} {
-		return &filer_pb.LogEntry{}
-	},
-}
-
-// resetLogEntry clears a LogEntry for pool reuse
-func resetLogEntry(e *filer_pb.LogEntry) {
-	proto.Reset(e)
-}
-
 func copiedBytes(buf []byte) (copied *bytes.Buffer) {
 	copied = bufferPool.Get().(*bytes.Buffer)
 	copied.Reset()
@@ -970,19 +959,129 @@ func readTs(buf []byte, pos int) (size int, ts int64, err error) {
 
 	entryData := buf[pos+4 : pos+4+size]
 
-	// Use pooled LogEntry to avoid allocation on every call
-	logEntry := logEntryPool.Get().(*filer_pb.LogEntry)
-	defer func() {
-		resetLogEntry(logEntry)
-		logEntryPool.Put(logEntry)
-	}()
-
-	err = proto.Unmarshal(entryData, logEntry)
+	// Read only LogEntry.ts_ns rather than unmarshaling the whole entry. This
+	// runs on every binary-search probe in ReadFromBuffer; a full proto.Unmarshal
+	// there allocates fresh slices for the data/key byte fields on each call, which
+	// dominated allocation churn under metadata-subscription fan-out.
+	ts, err = readTsNs(entryData)
 	if err != nil {
-		// Return error instead of failing fast
-		// This allows caller to handle corruption gracefully
-		return 0, 0, fmt.Errorf("corrupted log buffer: failed to unmarshal LogEntry at pos %d, size %d: %w", pos, size, err)
+		return 0, 0, fmt.Errorf("corrupted log buffer at pos %d, size %d: %w", pos, size, err)
 	}
 
-	return size, logEntry.TsNs, nil
+	return size, ts, nil
+}
+
+// readTsNs scans a marshaled LogEntry and returns its ts_ns (field 1, varint)
+// without decoding the data/key byte fields. Fields serialize in number order,
+// so ts_ns is normally the first tag and this returns after one varint; the full
+// scan keeps it correct for any field order. A missing field 1 means the proto3
+// default, ts_ns == 0.
+func readTsNs(entryData []byte) (tsNs int64, err error) {
+	for i := 0; i < len(entryData); {
+		tag, n := binary.Uvarint(entryData[i:])
+		if n <= 0 {
+			return 0, fmt.Errorf("bad field tag at %d", i)
+		}
+		i += n
+		fieldNum := tag >> 3
+		switch tag & 0x7 { // wire type
+		case 0: // varint
+			v, m := binary.Uvarint(entryData[i:])
+			if m <= 0 {
+				return 0, fmt.Errorf("bad varint for field %d", fieldNum)
+			}
+			i += m
+			if fieldNum == 1 { // ts_ns
+				return int64(v), nil
+			}
+		case 1: // 64-bit
+			i += 8
+		case 2: // length-delimited: read length, skip payload without copying
+			l, m := binary.Uvarint(entryData[i:])
+			if m <= 0 {
+				return 0, fmt.Errorf("bad length for field %d", fieldNum)
+			}
+			i += m
+			if l > uint64(len(entryData)-i) {
+				return 0, fmt.Errorf("field %d length %d overruns buffer", fieldNum, l)
+			}
+			i += int(l)
+		case 5: // 32-bit
+			i += 4
+		default:
+			return 0, fmt.Errorf("unknown wire type for field %d", fieldNum)
+		}
+		if i > len(entryData) {
+			return 0, fmt.Errorf("field %d overruns buffer", fieldNum)
+		}
+	}
+	return 0, nil
+}
+
+// unmarshalLogEntryAliased decodes a marshaled LogEntry into out, pointing the
+// data and key fields at sub-slices of entryData instead of copying them. A full
+// proto.Unmarshal allocates fresh slices for those byte fields on every entry
+// (protobuf consumeBytesNoZero), which dominated allocation churn when many
+// metadata subscribers each re-read the in-memory log window.
+//
+// The aliased data/key are only valid while entryData is, i.e. for the duration
+// of the eachLogEntryFn callback. Callers must copy anything they retain past the
+// callback; all current subscribers do (they re-decode data into their own event
+// or hand it to a synchronous grpc Send).
+func unmarshalLogEntryAliased(entryData []byte, out *filer_pb.LogEntry) error {
+	out.TsNs = 0
+	out.PartitionKeyHash = 0
+	out.Data = nil
+	out.Key = nil
+	out.Offset = 0
+	for i := 0; i < len(entryData); {
+		tag, n := binary.Uvarint(entryData[i:])
+		if n <= 0 {
+			return fmt.Errorf("bad field tag at %d", i)
+		}
+		i += n
+		fieldNum := tag >> 3
+		switch tag & 0x7 { // wire type
+		case 0: // varint: ts_ns / partition_key_hash / offset
+			v, m := binary.Uvarint(entryData[i:])
+			if m <= 0 {
+				return fmt.Errorf("bad varint for field %d", fieldNum)
+			}
+			i += m
+			switch fieldNum {
+			case 1:
+				out.TsNs = int64(v)
+			case 2:
+				out.PartitionKeyHash = int32(v)
+			case 5:
+				out.Offset = int64(v)
+			}
+		case 2: // length-delimited: data / key, aliased not copied
+			l, m := binary.Uvarint(entryData[i:])
+			if m <= 0 {
+				return fmt.Errorf("bad length for field %d", fieldNum)
+			}
+			i += m
+			if l > uint64(len(entryData)-i) {
+				return fmt.Errorf("field %d length %d overruns buffer", fieldNum, l)
+			}
+			switch fieldNum {
+			case 3:
+				out.Data = entryData[i : i+int(l)]
+			case 4:
+				out.Key = entryData[i : i+int(l)]
+			}
+			i += int(l)
+		case 1: // 64-bit
+			i += 8
+		case 5: // 32-bit
+			i += 4
+		default:
+			return fmt.Errorf("unknown wire type for field %d", fieldNum)
+		}
+		if i > len(entryData) {
+			return fmt.Errorf("field %d overruns buffer", fieldNum)
+		}
+	}
+	return nil
 }

weed/util/log_buffer/log_read.go

@@ -6,8 +6,6 @@ import (
 	"fmt"
 	"time"
 
-	"google.golang.org/protobuf/proto"
-
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/util"
@@ -261,7 +259,8 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition
 			entryData := buf[pos+4 : pos+4+int(size)]
 
 			logEntry := &filer_pb.LogEntry{}
-			if err = proto.Unmarshal(entryData, logEntry); err != nil {
+			// data/key alias entryData; valid only within eachLogDataFn below.
+			if err = unmarshalLogEntryAliased(entryData, logEntry); err != nil {
 				glog.Errorf("unexpected unmarshal mq_pb.Message: %v", err)
 				pos += 4 + int(size)
 				continue
@@ -523,7 +522,8 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star
 			entryData := buf[pos+4 : pos+4+int(size)]
 
 			logEntry := &filer_pb.LogEntry{}
-			if err = proto.Unmarshal(entryData, logEntry); err != nil {
+			// data/key alias entryData; valid only within eachLogDataFn below.
+			if err = unmarshalLogEntryAliased(entryData, logEntry); err != nil {
 				glog.Errorf("unexpected unmarshal mq_pb.Message: %v", err)
 				pos += 4 + int(size)
 				continue

weed/util/log_buffer/read_ts_test.go

@@ -0,0 +1,230 @@
+package log_buffer
+
+import (
+	"bytes"
+	"math"
+	"math/rand"
+	"testing"
+
+	"google.golang.org/protobuf/proto"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// readTsNs must return exactly what a full proto.Unmarshal would put in TsNs,
+// for every field combination, without decoding data/key.
+func TestReadTsNsMatchesUnmarshal(t *testing.T) {
+	cases := []*filer_pb.LogEntry{
+		{},                              // all defaults: ts_ns absent on the wire => 0
+		{TsNs: 1},                       // minimal
+		{TsNs: 1783396413597780505},     // realistic nanosecond timestamp
+		{TsNs: math.MaxInt64},           // 9-byte varint
+		{TsNs: -1},                      // negative int64 => 10-byte varint
+		{TsNs: math.MinInt64},           // extreme negative
+		{TsNs: 42, PartitionKeyHash: 7}, // ts followed by another varint field
+		{TsNs: 12345, Data: []byte("some-metadata-event-payload"), Key: []byte("/buckets/pvc-x/1.log")},
+		{TsNs: 99, Data: make([]byte, 64*1024), Key: []byte("k"), Offset: 555}, // large data, all fields
+		{Data: []byte("no-ts"), Key: []byte("k"), Offset: 9},                   // ts_ns == 0 with other fields present
+	}
+	for i, e := range cases {
+		data, err := proto.Marshal(e)
+		if err != nil {
+			t.Fatalf("case %d marshal: %v", i, err)
+		}
+		got, err := readTsNs(data)
+		if err != nil {
+			t.Fatalf("case %d readTsNs: %v", i, err)
+		}
+		if got != e.TsNs {
+			t.Errorf("case %d: readTsNs=%d want %d", i, got, e.TsNs)
+		}
+	}
+}
+
+// Fuzz against proto.Unmarshal with randomized entries.
+func TestReadTsNsFuzz(t *testing.T) {
+	r := rand.New(rand.NewSource(1))
+	for i := 0; i < 5000; i++ {
+		e := &filer_pb.LogEntry{
+			TsNs:             r.Int63() - r.Int63(),
+			PartitionKeyHash: r.Int31(),
+			Offset:           r.Int63(),
+		}
+		if r.Intn(2) == 0 {
+			e.Data = make([]byte, r.Intn(300))
+			r.Read(e.Data)
+		}
+		if r.Intn(2) == 0 {
+			e.Key = make([]byte, r.Intn(64))
+			r.Read(e.Key)
+		}
+		data, err := proto.Marshal(e)
+		if err != nil {
+			t.Fatalf("marshal: %v", err)
+		}
+		var want filer_pb.LogEntry
+		if err := proto.Unmarshal(data, &want); err != nil {
+			t.Fatalf("unmarshal: %v", err)
+		}
+		got, err := readTsNs(data)
+		if err != nil {
+			t.Fatalf("readTsNs iter %d: %v", i, err)
+		}
+		if got != want.TsNs {
+			t.Fatalf("iter %d: readTsNs=%d want %d", i, got, want.TsNs)
+		}
+	}
+}
+
+// readTsNs must not panic and must report an error on truncated/garbage input.
+func TestReadTsNsCorrupted(t *testing.T) {
+	inputs := [][]byte{
+		{0x08},                   // ts_ns tag with no value
+		{0x1a, 0xff, 0xff, 0xff}, // data field, length overruns buffer
+		{0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x7f}, // overlong tag
+	}
+	for i, in := range inputs {
+		if _, err := readTsNs(in); err == nil {
+			t.Errorf("input %d: expected error, got nil", i)
+		}
+	}
+}
+
+// unmarshalLogEntryAliased must produce the same fields as proto.Unmarshal.
+func TestUnmarshalLogEntryAliasedMatchesUnmarshal(t *testing.T) {
+	r := rand.New(rand.NewSource(2))
+	for i := 0; i < 5000; i++ {
+		e := &filer_pb.LogEntry{
+			TsNs:             r.Int63() - r.Int63(),
+			PartitionKeyHash: r.Int31() - r.Int31(),
+			Offset:           r.Int63() - r.Int63(),
+		}
+		if r.Intn(2) == 0 {
+			e.Data = make([]byte, r.Intn(4096))
+			r.Read(e.Data)
+		}
+		if r.Intn(2) == 0 {
+			e.Key = make([]byte, r.Intn(128))
+			r.Read(e.Key)
+		}
+		data, err := proto.Marshal(e)
+		if err != nil {
+			t.Fatalf("marshal: %v", err)
+		}
+		var want, got filer_pb.LogEntry
+		if err := proto.Unmarshal(data, &want); err != nil {
+			t.Fatalf("unmarshal: %v", err)
+		}
+		if err := unmarshalLogEntryAliased(data, &got); err != nil {
+			t.Fatalf("aliased iter %d: %v", i, err)
+		}
+		if got.TsNs != want.TsNs || got.PartitionKeyHash != want.PartitionKeyHash ||
+			got.Offset != want.Offset || !bytes.Equal(got.Data, want.Data) || !bytes.Equal(got.Key, want.Key) {
+			t.Fatalf("iter %d mismatch:\n got=%+v\nwant=%+v", i, &got, &want)
+		}
+	}
+}
+
+// The decoded data/key must point into the source buffer, not a copy.
+func TestUnmarshalLogEntryAliasedAliases(t *testing.T) {
+	data, err := proto.Marshal(&filer_pb.LogEntry{TsNs: 5, Data: []byte("payload"), Key: []byte("path")})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var out filer_pb.LogEntry
+	if err := unmarshalLogEntryAliased(data, &out); err != nil {
+		t.Fatal(err)
+	}
+	if string(out.Data) != "payload" || string(out.Key) != "path" {
+		t.Fatalf("decoded data=%q key=%q", out.Data, out.Key)
+	}
+	for i := range data { // mutate source; an alias must observe it, a copy must not
+		data[i] ^= 0xFF
+	}
+	if string(out.Data) == "payload" || string(out.Key) == "path" {
+		t.Fatal("data/key were copied, expected an alias into the source buffer")
+	}
+}
+
+// sampleEntryData returns a marshaled LogEntry whose Data is a realistic
+// marshaled metadata event, as delivered on the subscribe path.
+func sampleEntryData(tb testing.TB) []byte {
+	event := &filer_pb.SubscribeMetadataResponse{
+		Directory: "/buckets/pvc-bf458866-0413-4cca-99c6-3b7386a80855",
+		EventNotification: &filer_pb.EventNotification{
+			NewEntry: &filer_pb.Entry{
+				Name:       "object-000123",
+				Attributes: &filer_pb.FuseAttributes{FileSize: 4096, Mtime: 1783396413},
+				Chunks: []*filer_pb.FileChunk{
+					{FileId: "3,01637037d6", Size: 4096, Fid: &filer_pb.FileId{VolumeId: 3, FileKey: 23283, Cookie: 2015}},
+				},
+			},
+		},
+	}
+	payload, err := proto.Marshal(event)
+	if err != nil {
+		tb.Fatal(err)
+	}
+	data, err := proto.Marshal(&filer_pb.LogEntry{
+		TsNs: 1783396413597780505, PartitionKeyHash: 12345, Offset: 2013258,
+		Key: []byte("/buckets/pvc-bf458866/object-000123"), Data: payload,
+	})
+	if err != nil {
+		tb.Fatal(err)
+	}
+	return data
+}
+
+// Per-entry decode cost on the delivery path: the old full unmarshal copies the
+// data/key byte fields; the aliased decode does not. The delta is what the fix
+// saves for every entry a subscriber reads (and, for non-matching entries a
+// subscriber filters out, the whole saving with nothing decoded downstream).
+func BenchmarkLogEntryDecodeCopy(b *testing.B) {
+	data := sampleEntryData(b)
+	var out filer_pb.LogEntry
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		if err := proto.Unmarshal(data, &out); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkLogEntryDecodeAlias(b *testing.B) {
+	data := sampleEntryData(b)
+	var out filer_pb.LogEntry
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		if err := unmarshalLogEntryAliased(data, &out); err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func BenchmarkReadTs(b *testing.B) {
+	e := &filer_pb.LogEntry{
+		TsNs:             1783396413597780505,
+		PartitionKeyHash: 12345,
+		Data:             make([]byte, 4*1024), // typical metadata event payload
+		Key:              []byte("/buckets/pvc-bf458866/1.log"),
+		Offset:           2013258,
+	}
+	data, err := proto.Marshal(e)
+	if err != nil {
+		b.Fatal(err)
+	}
+	buf := make([]byte, 4+len(data))
+	util.Uint32toBytes(buf, uint32(len(data)))
+	copy(buf[4:], data)
+
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		if _, _, err := readTs(buf, 0); err != nil {
+			b.Fatal(err)
+		}
+	}
+}