From 38db7e149316a869390734c85b083e86ef2f4f19 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 8 Aug 2026 23:56:09 -0700 Subject: [PATCH] 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. --- weed/storage/volume_info.go | 51 +++++++- weed/storage/volume_info_intern_test.go | 153 ++++++++++++++++++++++++ 2 files changed, 199 insertions(+), 5 deletions(-) create mode 100644 weed/storage/volume_info_intern_test.go diff --git a/weed/storage/volume_info.go b/weed/storage/volume_info.go index f3f7b2cad..e993c9c58 100644 --- a/weed/storage/volume_info.go +++ b/weed/storage/volume_info.go @@ -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 != "" } diff --git a/weed/storage/volume_info_intern_test.go b/weed/storage/volume_info_intern_test.go new file mode 100644 index 000000000..880dc1a8b --- /dev/null +++ b/weed/storage/volume_info_intern_test.go @@ -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) + } + } +}