storage: share the volume strings a cluster repeats (#10665)

* storage: share the volume strings a cluster repeats Decoding a heartbeat allocates a fresh string for the collection, disk type and remote backend of every volume, and a master holding a million volumes then holds a million copies of the same handful of names. Not the remote storage key, which is unique per volume: interning that would fill the table rather than share anything. 800k volumes registered from a heartbeat that has actually been over the wire: 227 -> 211 B/volume, and 238 -> 214 when the volumes are tiered, since the backend name shares too. * storage: hold the interned strings rather than let them be collected unique.Make clears its entries by weak reference, and its canonical value does not survive a collection even while a caller still holds the string it handed back -- so a volume reported later would get a second copy of a name the rest of the cluster already shares. With only changed volumes reported, most are interned once and never again, so that is the common case rather than a corner. The table therefore only grows, which is why it stays restricted to values drawn from a small set. Ten thousand collections keep a few hundred kilobytes.
2026-08-16 04:06:44 +00:00 · 2026-08-08 23:56:09 -07:00
parent 923d0bd20c
commit 38db7e1493
2 changed files with 199 additions and 5 deletions
@@ -3,6 +3,7 @@ package storage
 import (
 	"fmt"
 	"sort"
+	"sync"

 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
@@ -32,7 +33,7 @@ func NewVolumeInfo(m *master_pb.VolumeInformationMessage) (vi VolumeInfo, err er
 	vi = VolumeInfo{
 		Id:                needle.VolumeId(m.Id),
 		Size:              m.Size,
-		Collection:        m.Collection,
+		Collection:        internVolumeString(m.Collection),
 		FileCount:         int(m.FileCount),
 		DeleteCount:       int(m.DeleteCount),
 		DeletedByteCount:  m.DeletedByteCount,
@@ -40,9 +41,9 @@ func NewVolumeInfo(m *master_pb.VolumeInformationMessage) (vi VolumeInfo, err er
 		Version:           needle.Version(m.Version),
 		CompactRevision:   m.CompactRevision,
 		ModifiedAtSecond:  m.ModifiedAtSecond,
-		RemoteStorageName: m.RemoteStorageName,
+		RemoteStorageName: internVolumeString(m.RemoteStorageName),
 		RemoteStorageKey:  m.RemoteStorageKey,
-		DiskType:          m.DiskType,
+		DiskType:          internVolumeString(m.DiskType),
 		DiskId:            m.DiskId,
 	}
 	rp, e := super_block.NewReplicaPlacementFromByte(byte(m.ReplicaPlacement))
@@ -57,7 +58,7 @@ func NewVolumeInfo(m *master_pb.VolumeInformationMessage) (vi VolumeInfo, err er
 func NewVolumeInfoFromShort(m *master_pb.VolumeShortInformationMessage) (vi VolumeInfo, err error) {
 	vi = VolumeInfo{
 		Id:         needle.VolumeId(m.Id),
-		Collection: m.Collection,
+		Collection: internVolumeString(m.Collection),
 		Version:    needle.Version(m.Version),
 	}
 	rp, e := super_block.NewReplicaPlacementFromByte(byte(m.ReplicaPlacement))
@@ -66,10 +67,50 @@ func NewVolumeInfoFromShort(m *master_pb.VolumeShortInformationMessage) (vi Volu
 	}
 	vi.ReplicaPlacement = rp
 	vi.Ttl = needle.LoadTTLFromUint32(m.Ttl)
-	vi.DiskType = m.DiskType
+	vi.DiskType = internVolumeString(m.DiskType)
 	return vi, nil
 }

+// internedVolumeStrings holds one copy of each value a cluster repeats across
+// its volumes. It only ever grows, which is why it must stay restricted to
+// values drawn from a small set: collection, disk type, remote backend. A
+// cluster with ten thousand collections keeps a few hundred kilobytes here.
+//
+// unique.Make would clear entries by weak reference, but its canonical value
+// does not survive a collection even while a caller still holds the string it
+// returned, so a later volume would get a second copy. Holding them is the
+// point.
+var (
+	internedVolumeStringsLock sync.RWMutex
+	internedVolumeStrings     = make(map[string]string)
+)
+
+// internVolumeString shares one copy of a repeated value. Decoding a heartbeat
+// allocates a fresh string for each, so a master holding a million volumes
+// otherwise holds a million copies of the same handful of names.
+//
+// Never for something unique per volume, such as a remote storage key: that
+// would fill the table rather than share anything.
+func internVolumeString(s string) string {
+	if s == "" {
+		return ""
+	}
+	internedVolumeStringsLock.RLock()
+	shared, found := internedVolumeStrings[s]
+	internedVolumeStringsLock.RUnlock()
+	if found {
+		return shared
+	}
+
+	internedVolumeStringsLock.Lock()
+	defer internedVolumeStringsLock.Unlock()
+	if shared, found = internedVolumeStrings[s]; found {
+		return shared
+	}
+	internedVolumeStrings[s] = s
+	return s
+}
+
 func (vi VolumeInfo) IsRemote() bool {
 	return vi.RemoteStorageName != ""
 }
@@ -0,0 +1,153 @@
+package storage
+
+import (
+	"runtime"
+	"sync"
+	"testing"
+	"unsafe"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+)
+
+func stringData(s string) uintptr {
+	if s == "" {
+		return 0
+	}
+	return uintptr(unsafe.Pointer(unsafe.StringData(s)))
+}
+
+// Every heartbeat decodes a fresh string for values a cluster repeats across
+// all its volumes, so a master holding a million volumes would otherwise hold a
+// million copies of the same handful of names.
+func TestRepeatedVolumeStringsAreShared(t *testing.T) {
+	message := func() *master_pb.VolumeInformationMessage {
+		return &master_pb.VolumeInformationMessage{
+			Id: 1, Version: 3,
+			// Built from bytes so each is a distinct allocation, as decoding is.
+			Collection:        string([]byte("somecollection")),
+			DiskType:          string([]byte("ssd")),
+			RemoteStorageName: string([]byte("s3cold")),
+			RemoteStorageKey:  string([]byte("seaweed/somecollection/1.dat")),
+		}
+	}
+
+	first, err := NewVolumeInfo(message())
+	if err != nil {
+		t.Fatal(err)
+	}
+	second, err := NewVolumeInfo(message())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, tc := range []struct {
+		name  string
+		a, b  string
+		share bool
+	}{
+		{"Collection", first.Collection, second.Collection, true},
+		{"DiskType", first.DiskType, second.DiskType, true},
+		{"RemoteStorageName", first.RemoteStorageName, second.RemoteStorageName, true},
+		// Unique per volume: interning it would fill the table rather than
+		// share anything.
+		{"RemoteStorageKey", first.RemoteStorageKey, second.RemoteStorageKey, false},
+	} {
+		if tc.a != tc.b {
+			t.Fatalf("%s: values differ, %q vs %q", tc.name, tc.a, tc.b)
+		}
+		if shared := stringData(tc.a) == stringData(tc.b); shared != tc.share {
+			t.Errorf("%s: shared=%v, want %v", tc.name, shared, tc.share)
+		}
+	}
+}
+
+func TestShortVolumeInfoSharesTheSameStrings(t *testing.T) {
+	message := func() *master_pb.VolumeShortInformationMessage {
+		return &master_pb.VolumeShortInformationMessage{
+			Id: 1, Version: 3,
+			Collection: string([]byte("somecollection")),
+			DiskType:   string([]byte("ssd")),
+		}
+	}
+	first, err := NewVolumeInfoFromShort(message())
+	if err != nil {
+		t.Fatal(err)
+	}
+	second, err := NewVolumeInfoFromShort(message())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if stringData(first.Collection) != stringData(second.Collection) {
+		t.Error("collection is not shared between volumes reported as a delta")
+	}
+	if stringData(first.DiskType) != stringData(second.DiskType) {
+		t.Error("disk type is not shared between volumes reported as a delta")
+	}
+}
+
+func TestEmptyVolumeStringsStayEmpty(t *testing.T) {
+	vi, err := NewVolumeInfo(&master_pb.VolumeInformationMessage{Id: 1, Version: 3})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if vi.Collection != "" || vi.DiskType != "" || vi.RemoteStorageName != "" {
+		t.Errorf("expected empty strings to survive, got %+v", vi)
+	}
+	if vi.IsRemote() {
+		t.Error("a volume with no remote backend reads as remote")
+	}
+}
+
+// Sharing has to survive collection: volumes are interned when they are first
+// reported, and in a cluster sending only what changed most are never reported
+// again. A table that let its entries be collected would hand the next volume
+// a second copy of a name the rest of the cluster already shares.
+func TestVolumeStringsStaySharedAcrossCollection(t *testing.T) {
+	first, err := NewVolumeInfo(&master_pb.VolumeInformationMessage{
+		Id: 1, Version: 3, Collection: string([]byte("somecollection")),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	original := stringData(first.Collection)
+
+	for i := 0; i < 5; i++ {
+		runtime.GC()
+	}
+
+	later, err := NewVolumeInfo(&master_pb.VolumeInformationMessage{
+		Id: 2, Version: 3, Collection: string([]byte("somecollection")),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if stringData(later.Collection) != original {
+		t.Error("a volume reported after a collection got its own copy of the name")
+	}
+	runtime.KeepAlive(first)
+}
+
+func TestInterningIsSafeUnderConcurrentReports(t *testing.T) {
+	var wg sync.WaitGroup
+	shared := make([]uintptr, 16)
+	for i := range shared {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			vi, err := NewVolumeInfo(&master_pb.VolumeInformationMessage{
+				Id: uint32(i), Version: 3, Collection: string([]byte("concurrent")),
+			})
+			if err != nil {
+				t.Error(err)
+				return
+			}
+			shared[i] = stringData(vi.Collection)
+		}(i)
+	}
+	wg.Wait()
+	for i, got := range shared {
+		if got != shared[0] {
+			t.Fatalf("report %d got its own copy of the name", i)
+		}
+	}
+}