From eed3c27d15640d60bb71671fc8bf31fa675a867b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 27 Aug 2026 22:25:15 -0700 Subject: [PATCH] volume: cut the memory a server holding millions of volumes still uses (#10999) * volume: stop the .vif guard depending on which entry the scan handed over A volume has both an .idx and a .vif, and loadExistingVolume skipped a .vif next to an .ecx as EC shard metadata. That was only ever correct because os.ReadDir sorted .idx ahead of .vif: an interrupted encode, where the .idx is still there, has to reach validateEcVolume to be reclaimed. Ask for the .idx instead of trusting the order. Claude-Session: https://claude.ai/code/session_01NWpFUwAJcR2KUENrhLc9Sy * volume: walk volume directories in batches instead of listing them whole os.ReadDir builds, and sorts, a slice of every entry before the caller sees the first one. A disk holding millions of volumes has a .dat, .idx and .vif per volume, so each startup scan costs hundreds of MB of peak heap that the runtime is slow to hand back -- and there are several of them before the first volume loads. Walk in batches instead, and keep only the entries each scan acts on: loadAllEcShards now sorts and stats the shard and index files alone rather than every file on the disk. Claude-Session: https://claude.ai/code/session_01NWpFUwAJcR2KUENrhLc9Sy * volume: skip the sibling-.dat scan when no EC volume is loaded pruneIncompleteEcWithSiblingDat only ever prunes EC volumes that are loaded, but it first walks every disk and keys a map by every .dat on the server. On a store with no EC volumes at all that is millions of map entries built to answer no question. Claude-Session: https://claude.ai/code/session_01NWpFUwAJcR2KUENrhLc9Sy * volume: stop keeping a departure message for every volume The report state held a VolumeShortInformationMessage per volume copy so a departure could be named, but almost no volume ever departs. Hold a handle to the identity instead -- volumes share very few distinct ones -- and build the message on the way out. Measured over a populated report state: 195 -> 83 bytes per volume. Claude-Session: https://claude.ai/code/session_01NWpFUwAJcR2KUENrhLc9Sy * rust volume: stop keeping a whole volume message per volume held The send loop kept a VolumeInformationMessage for every volume just to notice mounts and unmounts, and rebuilt the map from scratch on every beat. Keep the identity a delta names, which is what the Go report state keeps for the same reason. Claude-Session: https://claude.ai/code/session_01NWpFUwAJcR2KUENrhLc9Sy * rust volume: keep only the EC files the shard scan acts on load_all_ec_shards named every file on the disk twice -- once in the dedup set and once in the sorted vector -- before deciding it only wanted .ec?? and .ecx. Filter while reading instead. Mirrors the same change in loadAllEcShards. Claude-Session: https://claude.ai/code/session_01NWpFUwAJcR2KUENrhLc9Sy * volume: share the strings every .vif repeats A tiered volume's .vif names its replication and its backend, and every decode allocates a fresh copy, so a server holding millions of them holds millions of copies of the same handful of names. Route them through the interning table the volume info decode already uses. The remote key names one volume and is left alone. Claude-Session: https://claude.ai/code/session_01NWpFUwAJcR2KUENrhLc9Sy --- seaweed-volume/src/server/heartbeat.rs | 78 ++++++++++----- seaweed-volume/src/storage/disk_location.rs | 29 +++--- weed/storage/dir_scan.go | 40 ++++++++ weed/storage/dir_scan_test.go | 70 +++++++++++++ weed/storage/disk_location.go | 99 +++++++++++-------- weed/storage/disk_location_ec.go | 57 ++++++----- weed/storage/disk_location_ec_test.go | 17 +--- weed/storage/disk_location_scan_order_test.go | 42 ++++++++ weed/storage/store_ec_reconcile.go | 80 +++++++++------ weed/storage/store_volume_report.go | 67 +++++++++---- weed/storage/store_volume_report_test.go | 27 +++++ weed/storage/volume_tier.go | 15 +++ weed/storage/volume_tier_intern_test.go | 59 +++++++++++ 13 files changed, 515 insertions(+), 165 deletions(-) create mode 100644 weed/storage/dir_scan.go create mode 100644 weed/storage/dir_scan_test.go create mode 100644 weed/storage/disk_location_scan_order_test.go create mode 100644 weed/storage/volume_tier_intern_test.go diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index f0a23c314..d7a2e1667 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -398,8 +398,7 @@ async fn do_heartbeat( // Keep track of what we sent, to generate delta updates let (initial_hb, initial_volumes) = collect_heartbeat_with_snapshot(config, state); - let mut last_volumes: HashMap = - initial_volumes.iter().map(|v| (v.id, v.clone())).collect(); + let mut last_volumes: HashMap = volume_identities(&initial_volumes); let mut last_ec_shards = { let store = state.store.read().unwrap(); collect_ec_shard_delta_messages(&store) @@ -466,8 +465,7 @@ async fn do_heartbeat( if changed { let (adjusted_hb, adjusted_volumes) = collect_heartbeat_with_snapshot(config, state); - last_volumes = - adjusted_volumes.iter().map(|v| (v.id, v.clone())).collect(); + last_volumes = volume_identities(&adjusted_volumes); last_ec_shards = { let store = state.store.read().unwrap(); collect_ec_shard_delta_messages(&store) @@ -501,7 +499,7 @@ async fn do_heartbeat( s.maybe_adjust_volume_max(); } let (current_hb, current_volumes) = collect_heartbeat_with_snapshot(config, state); - last_volumes = current_volumes.iter().map(|v| (v.id, v.clone())).collect(); + last_volumes = volume_identities(¤t_volumes); last_ec_shards = { let store = state.store.read().unwrap(); collect_ec_shard_delta_messages(&store) @@ -530,7 +528,7 @@ async fn do_heartbeat( return Ok(None); } let held_volumes = collect_volume_snapshot(config, state); - let current_volumes: HashMap = held_volumes.iter().map(|v| (v.id, v.clone())).collect(); + let current_volumes = volume_identities(&held_volumes); let current_ec_shards = { let store = state.store.read().unwrap(); collect_ec_shard_delta_messages(&store) @@ -541,29 +539,13 @@ async fn do_heartbeat( for (id, vol) in ¤t_volumes { if !last_volumes.contains_key(id) { - new_vols.push(master_pb::VolumeShortInformationMessage { - id: *id, - collection: vol.collection.clone(), - version: vol.version, - replica_placement: vol.replica_placement, - ttl: vol.ttl, - disk_type: vol.disk_type.clone(), - disk_id: vol.disk_id, - }); + new_vols.push(vol.to_short_message(*id)); } } for (id, vol) in &last_volumes { if !current_volumes.contains_key(id) { - del_vols.push(master_pb::VolumeShortInformationMessage { - id: *id, - collection: vol.collection.clone(), - version: vol.version, - replica_placement: vol.replica_placement, - ttl: vol.ttl, - disk_type: vol.disk_type.clone(), - disk_id: vol.disk_id, - }); + del_vols.push(vol.to_short_message(*id)); } } @@ -744,6 +726,54 @@ fn parse_bool_property(value: Option<&String>) -> bool { .unwrap_or(true) } +/// What a mount or unmount delta has to name, which is far less than the +/// information message the heartbeat carries. A server holding millions of +/// volumes cannot keep a whole message for each just to notice one leave; the +/// Go report state keeps the same fields for the same reason. +#[derive(Clone)] +struct VolumeIdentity { + collection: String, + disk_type: String, + version: u32, + replica_placement: u32, + ttl: u32, + disk_id: u32, +} + +impl VolumeIdentity { + fn of(v: &master_pb::VolumeInformationMessage) -> Self { + Self { + collection: v.collection.clone(), + disk_type: v.disk_type.clone(), + version: v.version, + replica_placement: v.replica_placement, + ttl: v.ttl, + disk_id: v.disk_id, + } + } + + fn to_short_message(&self, id: u32) -> master_pb::VolumeShortInformationMessage { + master_pb::VolumeShortInformationMessage { + id, + collection: self.collection.clone(), + version: self.version, + replica_placement: self.replica_placement, + ttl: self.ttl, + disk_type: self.disk_type.clone(), + disk_id: self.disk_id, + } + } +} + +fn volume_identities( + volumes: &[master_pb::VolumeInformationMessage], +) -> HashMap { + volumes + .iter() + .map(|v| (v.id, VolumeIdentity::of(v))) + .collect() +} + /// Collect volume information into a Heartbeat message. fn collect_heartbeat_with_snapshot( config: &HeartbeatConfig, diff --git a/seaweed-volume/src/storage/disk_location.rs b/seaweed-volume/src/storage/disk_location.rs index 1f10d7eb6..5e39b361c 100644 --- a/seaweed-volume/src/storage/disk_location.rs +++ b/seaweed-volume/src/storage/disk_location.rs @@ -921,27 +921,32 @@ impl DiskLocation { // double-count for those filenames. let mut seen: HashSet = HashSet::new(); let mut entries: Vec = Vec::new(); - for ent in fs::read_dir(&self.directory)? { - let ent = ent?; - if ent.file_type().map(|ft| ft.is_dir()).unwrap_or(false) { - continue; - } - let name = ent.file_name().to_string_lossy().into_owned(); - if seen.insert(name.clone()) { - entries.push(name); - } - } - if self.idx_directory != self.directory { - for ent in fs::read_dir(&self.idx_directory)? { + // Keep only the shard and index files this scan acts on: a disk of + // regular volumes has millions of .dat/.idx/.vif names that would + // otherwise each cost a String here and a slot in the sort below. + let mut collect = |dir: &str| -> io::Result<()> { + for ent in fs::read_dir(dir)? { let ent = ent?; if ent.file_type().map(|ft| ft.is_dir()).unwrap_or(false) { continue; } let name = ent.file_name().to_string_lossy().into_owned(); + let Some(dot) = name.rfind('.') else { + continue; + }; + let ext = &name[dot..]; + if parse_ec_shard_extension(ext).is_none() && ext != ".ecx" { + continue; + } if seen.insert(name.clone()) { entries.push(name); } } + Ok(()) + }; + collect(&self.directory)?; + if self.idx_directory != self.directory { + collect(&self.idx_directory)?; } entries.sort(); diff --git a/weed/storage/dir_scan.go b/weed/storage/dir_scan.go new file mode 100644 index 000000000..9be14fe61 --- /dev/null +++ b/weed/storage/dir_scan.go @@ -0,0 +1,40 @@ +package storage + +import ( + "errors" + "io" + "os" +) + +// dirScanBatch bounds how many entries a directory walk holds at once. A disk +// holding millions of volumes has a file per volume for each of .dat, .idx and +// .vif, and os.ReadDir builds — and sorts — a slice of all of them before the +// caller sees the first entry. Every startup scan then costs hundreds of MB of +// peak heap that the runtime is slow to hand back, which is most of the gap +// between a volume server's live heap and its resident set. +const dirScanBatch = 1024 + +// eachDirEntry calls visit for every entry in dir, in whatever order the +// filesystem returns them, and stops early once visit returns false. Callers +// that need a defined order sort the few entries they keep. +func eachDirEntry(dir string, visit func(entry os.DirEntry) bool) error { + f, err := os.Open(dir) + if err != nil { + return err + } + defer f.Close() + for { + entries, readErr := f.ReadDir(dirScanBatch) + for _, entry := range entries { + if !visit(entry) { + return nil + } + } + if readErr != nil { + if errors.Is(readErr, io.EOF) { + return nil + } + return readErr + } + } +} diff --git a/weed/storage/dir_scan_test.go b/weed/storage/dir_scan_test.go new file mode 100644 index 000000000..7045714e7 --- /dev/null +++ b/weed/storage/dir_scan_test.go @@ -0,0 +1,70 @@ +package storage + +import ( + "fmt" + "os" + "path/filepath" + "testing" +) + +// The walk reads in batches, so it has to keep going past the first one and +// stop at the end without repeating or dropping an entry. +func TestEachDirEntrySeesEveryEntry(t *testing.T) { + dir := t.TempDir() + want := make(map[string]bool, dirScanBatch*2+3) + for i := 0; i < dirScanBatch*2+3; i++ { + name := fmt.Sprintf("%d.dat", i) + if err := os.WriteFile(filepath.Join(dir, name), nil, 0644); err != nil { + t.Fatal(err) + } + want[name] = true + } + + seen := make(map[string]bool, len(want)) + if err := eachDirEntry(dir, func(entry os.DirEntry) bool { + if seen[entry.Name()] { + t.Errorf("entry %s visited twice", entry.Name()) + } + seen[entry.Name()] = true + return true + }); err != nil { + t.Fatal(err) + } + if len(seen) != len(want) { + t.Fatalf("walked %d entries, want %d", len(seen), len(want)) + } + for name := range want { + if !seen[name] { + t.Errorf("entry %s was never visited", name) + } + } +} + +func TestEachDirEntryStopsWhenAsked(t *testing.T) { + dir := t.TempDir() + for i := 0; i < dirScanBatch+10; i++ { + if err := os.WriteFile(filepath.Join(dir, fmt.Sprintf("%d.dat", i)), nil, 0644); err != nil { + t.Fatal(err) + } + } + + visited := 0 + if err := eachDirEntry(dir, func(entry os.DirEntry) bool { + visited++ + return visited < 5 + }); err != nil { + t.Fatal(err) + } + if visited != 5 { + t.Errorf("walk visited %d entries after being asked to stop at 5", visited) + } +} + +func TestEachDirEntryReportsAMissingDirectory(t *testing.T) { + if err := eachDirEntry(filepath.Join(t.TempDir(), "absent"), func(os.DirEntry) bool { + t.Error("visited an entry of a directory that does not exist") + return true + }); err == nil { + t.Error("walking a missing directory reported no error") + } +} diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go index d8b83a0ca..59f121f14 100644 --- a/weed/storage/disk_location.go +++ b/weed/storage/disk_location.go @@ -202,11 +202,7 @@ func vifIsEcVolume(vifPath string) bool { return err == nil && vi.GetEcShardConfig() != nil } -func (l *DiskLocation) loadExistingVolume(dirEntry os.DirEntry, needleMapKind NeedleMapKind, skipIfEcVolumesExists bool, ldbTimeout int64, diskId uint32) bool { - basename := dirEntry.Name() - if dirEntry.IsDir() { - return false - } +func (l *DiskLocation) loadExistingVolume(basename string, needleMapKind NeedleMapKind, skipIfEcVolumesExists bool, ldbTimeout int64, diskId uint32) bool { volumeName := getValidVolumeName(basename) if volumeName == "" { return false @@ -227,9 +223,13 @@ func (l *DiskLocation) loadExistingVolume(dirEntry os.DirEntry, needleMapKind Ne return false } - // .vif next to .ecx is EC shard metadata, not a regular volume. - // Without this guard NewVolume below would create a phantom empty .dat. - if strings.HasSuffix(basename, ".vif") && l.hasEcxFile(volumeName) { + // A .vif next to an .ecx with no .idx beside it is EC shard metadata, not a + // regular volume. Without this guard NewVolume below would create a phantom + // empty .dat. Ask for the .idx rather than trust which of a volume's two + // entries the scan handed over: an .idx next to the .ecx is an interrupted + // encode, and validateEcVolume below is what decides that one. + if strings.HasSuffix(basename, ".vif") && l.hasEcxFile(volumeName) && + !util.FileExists(l.Directory+"/"+volumeName+".idx") { glog.V(1).Infof("loadExistingVolume: skipping .vif-only entry for volume %d (collection=%q); .ecx present", vid, collection) return false } @@ -307,20 +307,32 @@ func (l *DiskLocation) loadExistingVolume(dirEntry os.DirEntry, needleMapKind Ne func (l *DiskLocation) concurrentLoadingVolumes(needleMapKind NeedleMapKind, concurrency int, ldbTimeout int64, diskId uint32) { - task_queue := make(chan os.DirEntry, 10*concurrency) + // Read the directory to its end before the workers start writing into it: + // loading a volume creates .sdx, .vif and .ldb files in the same directory, + // and a stream left open across those writes is not guaranteed to hand back + // every entry it has not reached yet. Only the names are kept, one per + // volume, which is what the dedup here always held. + foundVolumeNames := make(map[string]string) + if err := eachDirEntry(l.Directory, func(entry os.DirEntry) bool { + if entry.IsDir() { + return true + } + volumeName := getValidVolumeName(entry.Name()) + if volumeName == "" { + return true + } + if _, found := foundVolumeNames[volumeName]; !found { + foundVolumeNames[volumeName] = entry.Name() + } + return true + }); err != nil { + glog.Warningf("scan volume directory %s: %v", l.Directory, err) + } + + task_queue := make(chan string, 10*concurrency) go func() { - foundVolumeNames := make(map[string]bool) - if dirEntries, err := os.ReadDir(l.Directory); err == nil { - for _, entry := range dirEntries { - volumeName := getValidVolumeName(entry.Name()) - if volumeName == "" { - continue - } - if _, found := foundVolumeNames[volumeName]; !found { - foundVolumeNames[volumeName] = true - task_queue <- entry - } - } + for _, basename := range foundVolumeNames { + task_queue <- basename } close(task_queue) }() @@ -330,8 +342,8 @@ func (l *DiskLocation) concurrentLoadingVolumes(needleMapKind NeedleMapKind, con wg.Add(1) go func() { defer wg.Done() - for fi := range task_queue { - _ = l.loadExistingVolume(fi, needleMapKind, true, ldbTimeout, diskId) + for basename := range task_queue { + _ = l.loadExistingVolume(basename, needleMapKind, true, ldbTimeout, diskId) } }() } @@ -385,23 +397,22 @@ func (l *DiskLocation) reconcileCompactStates() { } pending := make(map[volKey]bool) collect := func(dir string) { - entries, err := os.ReadDir(dir) - if err != nil { - return - } - for _, entry := range entries { + if err := eachDirEntry(dir, func(entry os.DirEntry) bool { if entry.IsDir() { - continue + return true } name := entry.Name() if !strings.HasSuffix(name, ".cpc") && !strings.HasSuffix(name, ".cpd") && !strings.HasSuffix(name, ".cpx") { - continue + return true } collection, vid, err := parseCollectionVolumeId(name[:len(name)-4]) if err != nil { - continue + return true } pending[volKey{collection, vid}] = true + return true + }); err != nil { + glog.Warningf("scan %s for interrupted compactions: %v", dir, err) } } collect(l.Directory) @@ -499,7 +510,7 @@ func (l *DiskLocation) deleteVolumeById(vid needle.VolumeId, onlyEmpty bool, kee func (l *DiskLocation) LoadVolume(diskId uint32, vid needle.VolumeId, needleMapKind NeedleMapKind) bool { if fileInfo, found := l.LocateVolume(vid); found { - return l.loadExistingVolume(fileInfo, needleMapKind, false, 0, diskId) + return l.loadExistingVolume(fileInfo.Name(), needleMapKind, false, 0, diskId) } return false } @@ -637,19 +648,21 @@ func (l *DiskLocation) Close() { } func (l *DiskLocation) LocateVolume(vid needle.VolumeId) (os.DirEntry, bool) { - // println("LocateVolume", vid, "on", l.Directory) - if dirEntries, err := os.ReadDir(l.Directory); err == nil { - for _, entry := range dirEntries { - // println("checking", entry.Name(), "...") - volId, _, err := volumeIdFromFileName(entry.Name()) - // println("volId", volId, "err", err) - if vid == volId && err == nil { - return entry, true - } + var found os.DirEntry + if err := eachDirEntry(l.Directory, func(entry os.DirEntry) bool { + if entry.IsDir() { + return true } + volId, _, err := volumeIdFromFileName(entry.Name()) + if vid == volId && err == nil { + found = entry + return false + } + return true + }); err != nil { + glog.Warningf("locate volume %d in %s: %v", vid, l.Directory, err) } - - return nil, false + return found, found != nil } func (l *DiskLocation) UnUsedSpace(volumeSizeLimit uint64) (unUsedSpace uint64) { diff --git a/weed/storage/disk_location_ec.go b/weed/storage/disk_location_ec.go index c306fcd11..218abc8d6 100644 --- a/weed/storage/disk_location_ec.go +++ b/weed/storage/disk_location_ec.go @@ -218,19 +218,41 @@ const staleZeroShardAge = time.Hour func (l *DiskLocation) loadAllEcShards(onShardLoad func(collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, ecVolume *erasure_coding.EcVolume)) (err error) { - dirEntries, err := os.ReadDir(l.Directory) - if err != nil { + // Keep only the shard and index files this scan acts on: a disk of regular + // volumes has millions of .dat/.idx/.vif entries that would otherwise each + // cost a slot in the sorted slice below and a stat() for its size. + type ecDirEntry struct { + name string + size int64 + } + var dirEntries []ecDirEntry + collect := func(dir string) error { + return eachDirEntry(dir, func(entry os.DirEntry) bool { + if entry.IsDir() { + return true + } + ext := path.Ext(entry.Name()) + if !re.MatchString(ext) && ext != ".ecx" { + return true + } + info, infoErr := entry.Info() + if infoErr != nil { + return true + } + dirEntries = append(dirEntries, ecDirEntry{name: entry.Name(), size: info.Size()}) + return true + }) + } + if err := collect(l.Directory); err != nil { return fmt.Errorf("load all ec shards in dir %s: %v", l.Directory, err) } if l.IdxDirectory != l.Directory { - indexDirEntries, err := os.ReadDir(l.IdxDirectory) - if err != nil { + if err := collect(l.IdxDirectory); err != nil { return fmt.Errorf("load all ec shards in dir %s: %v", l.IdxDirectory, err) } - dirEntries = append(dirEntries, indexDirEntries...) } - slices.SortFunc(dirEntries, func(a, b os.DirEntry) int { - return strings.Compare(a.Name(), b.Name()) + slices.SortFunc(dirEntries, func(a, b ecDirEntry) int { + return strings.Compare(a.name, b.name) }) var sameVolumeShards []string @@ -245,11 +267,8 @@ func (l *DiskLocation) loadAllEcShards(onShardLoad func(collection string, vid n } for _, fileInfo := range dirEntries { - if fileInfo.IsDir() { - continue - } - ext := path.Ext(fileInfo.Name()) - name := fileInfo.Name() + name := fileInfo.name + ext := path.Ext(name) baseName := name[:len(name)-len(ext)] collection, volumeId, err := parseCollectionVolumeId(baseName) @@ -257,12 +276,6 @@ func (l *DiskLocation) loadAllEcShards(onShardLoad func(collection string, vid n continue } - info, err := fileInfo.Info() - - if err != nil { - continue - } - // A zero-sized shard file is residue of a failed operation (never // loaded, but its presence poisons later rebuilds, which select // inputs from the directory). Delete it once it is old enough that @@ -272,7 +285,7 @@ func (l *DiskLocation) loadAllEcShards(onShardLoad func(collection string, vid n // and a candidate path can be different files with one name: each // candidate's own age decides, and a same-named fresh file (possibly // an in-flight copy's just-created one) always survives. - if re.MatchString(ext) && info.Size() == 0 { + if re.MatchString(ext) && fileInfo.size == 0 { for _, dir := range []string{l.Directory, l.IdxDirectory} { p := path.Join(dir, name) fi, statErr := os.Stat(p) @@ -293,14 +306,14 @@ func (l *DiskLocation) loadAllEcShards(onShardLoad func(collection string, vid n // 0 byte files should be only appearing erroneously for ec data files // so we ignore them - if re.MatchString(ext) && info.Size() > 0 { + if re.MatchString(ext) && fileInfo.size > 0 { // Group shards by both collection and volumeId to avoid mixing collections if prevVolumeId == 0 || (volumeId == prevVolumeId && collection == prevCollection) { - sameVolumeShards = append(sameVolumeShards, fileInfo.Name()) + sameVolumeShards = append(sameVolumeShards, name) } else { // Before starting a new group, check if previous group had orphaned shards l.checkOrphanedShards(sameVolumeShards, prevCollection, prevVolumeId) - sameVolumeShards = []string{fileInfo.Name()} + sameVolumeShards = []string{name} } prevVolumeId = volumeId prevCollection = collection diff --git a/weed/storage/disk_location_ec_test.go b/weed/storage/disk_location_ec_test.go index 6fcc7b591..93401b28b 100644 --- a/weed/storage/disk_location_ec_test.go +++ b/weed/storage/disk_location_ec_test.go @@ -720,22 +720,7 @@ func TestLoadExistingVolumeSkipsVifWhenEcxPresent(t *testing.T) { t.Fatalf("write .ecx: %v", err) } - entries, err := os.ReadDir(dataDir) - if err != nil { - t.Fatalf("read dir: %v", err) - } - var vifEntry os.DirEntry - for _, e := range entries { - if filepath.Ext(e.Name()) == ".vif" { - vifEntry = e - break - } - } - if vifEntry == nil { - t.Fatalf(".vif entry missing from dir listing") - } - - loaded := diskLocation.loadExistingVolume(vifEntry, NeedleMapInMemory, false, 0, 0) + loaded := diskLocation.loadExistingVolume(filepath.Base(vifPath), NeedleMapInMemory, false, 0, 0) if loaded { t.Fatalf("loadExistingVolume should refuse to load a .vif-only entry when .ecx is present (volume %d)", vid) } diff --git a/weed/storage/disk_location_scan_order_test.go b/weed/storage/disk_location_scan_order_test.go new file mode 100644 index 000000000..10724fa1e --- /dev/null +++ b/weed/storage/disk_location_scan_order_test.go @@ -0,0 +1,42 @@ +package storage + +import ( + "os" + "path/filepath" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +// A volume has both an .idx and a .vif, and the scan hands over whichever the +// filesystem returned first. Which one it was must not decide whether the +// volume loads -- it used to, through the .vif guard, and only a sorted +// listing kept the .idx in front. +func TestScanOrderDoesNotDecideWhetherAVolumeLoads(t *testing.T) { + loaded := make(map[string]bool) + for _, ext := range []string{".idx", ".vif"} { + store := newTestStore(t, 1) + location := store.Locations[0] + mountTestVolume(t, location, 9, "") + location.UnloadVolume(needle.VolumeId(9)) + // A .dat with data in it and a stray .ecx with no shards beside it: an + // interrupted encode, which the EC validation below the guard reclaims. + if err := os.Truncate(filepath.Join(location.Directory, "9.dat"), 4096); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(location.Directory, "9.ecx"), make([]byte, 20), 0644); err != nil { + t.Fatal(err) + } + + location.loadExistingVolume("9"+ext, NeedleMapInMemory, true, 0, 0) + _, found := location.FindVolume(needle.VolumeId(9)) + loaded[ext] = found + } + + if loaded[".idx"] != loaded[".vif"] { + t.Errorf("scanning the .idx loaded=%v but scanning the .vif loaded=%v", loaded[".idx"], loaded[".vif"]) + } + if !loaded[".idx"] { + t.Error("a volume with an .idx beside its .vif did not load") + } +} diff --git a/weed/storage/store_ec_reconcile.go b/weed/storage/store_ec_reconcile.go index 3a430bdbf..c85778470 100644 --- a/weed/storage/store_ec_reconcile.go +++ b/weed/storage/store_ec_reconcile.go @@ -3,6 +3,7 @@ package storage import ( "os" "path" + "slices" "strconv" "strings" @@ -154,17 +155,13 @@ func (s *Store) indexEcxOwners() map[ecKeyForReconcile]ecxOwnerInfo { continue } seen[scan] = true - entries, err := os.ReadDir(scan) - if err != nil { - continue - } - for _, entry := range entries { + if err := eachDirEntry(scan, func(entry os.DirEntry) bool { if entry.IsDir() { - continue + return true } name := entry.Name() if !strings.HasSuffix(name, ".ecx") { - continue + return true } // A 0-byte .ecx is a corrupt stub from a failed copy and // not a credible owner — skip it so the scan keeps looking @@ -175,17 +172,20 @@ func (s *Store) indexEcxOwners() map[ecKeyForReconcile]ecxOwnerInfo { // shards unloaded even when a valid index exists nearby. info, statErr := entry.Info() if statErr != nil || info.Size() == 0 { - continue + return true } base := name[:len(name)-len(".ecx")] collection, vid, err := parseCollectionVolumeId(base) if err != nil { - continue + return true } key := ecKeyForReconcile{collection: collection, vid: vid} if _, exists := owners[key]; !exists { owners[key] = ecxOwnerInfo{location: loc, idxDir: scan} } + return true + }); err != nil { + glog.Warningf("scan %s for .ecx owners: %v", scan, err) } } } @@ -243,10 +243,29 @@ func (s *Store) countEcShardsNodeWide(collection string, vid needle.VolumeId) in return len(seen) } +// hasEcVolumes reports whether any disk on this store has an EC volume loaded. +func (s *Store) hasEcVolumes() bool { + for _, loc := range s.Locations { + loc.ecVolumesLock.RLock() + count := len(loc.ecVolumes) + loc.ecVolumesLock.RUnlock() + if count > 0 { + return true + } + } + return false +} + func (s *Store) pruneIncompleteEcWithSiblingDat() { if len(s.Locations) < 2 { return } + // Only loaded EC volumes are ever pruned, so a store holding none has + // nothing to decide — and indexDatOwners below would otherwise walk every + // disk and key a map by every .dat on the server to answer no question. + if !s.hasEcVolumes() { + return + } datOwners := s.indexDatOwners() if len(datOwners) == 0 { @@ -344,31 +363,30 @@ func (s *Store) pruneIncompleteEcWithSiblingDat() { func (s *Store) indexDatOwners() map[ecKeyForReconcile]datOwnerInfo { owners := make(map[ecKeyForReconcile]datOwnerInfo) for _, loc := range s.Locations { - entries, err := os.ReadDir(loc.Directory) - if err != nil { - continue - } - for _, entry := range entries { + if err := eachDirEntry(loc.Directory, func(entry os.DirEntry) bool { if entry.IsDir() { - continue + return true } name := entry.Name() if !strings.HasSuffix(name, ".dat") { - continue + return true } base := name[:len(name)-len(".dat")] collection, vid, err := parseCollectionVolumeId(base) if err != nil { - continue + return true } info, err := entry.Info() if err != nil { - continue + return true } key := ecKeyForReconcile{collection: collection, vid: vid} if _, exists := owners[key]; !exists { owners[key] = datOwnerInfo{location: loc, size: info.Size()} } + return true + }); err != nil { + glog.Warningf("scan %s for .dat owners: %v", loc.Directory, err) } } return owners @@ -382,38 +400,42 @@ func (s *Store) indexDatOwners() map[ecKeyForReconcile]datOwnerInfo { // Zero-byte shard files are ignored — loadAllEcShards already treats them // as cleanup-worthy noise and we want the same shape here. func (l *DiskLocation) collectOrphanEcShards() map[ecKeyForReconcile][]string { - entries, err := os.ReadDir(l.Directory) - if err != nil { - return nil - } orphans := make(map[ecKeyForReconcile][]string) - for _, entry := range entries { + if err := eachDirEntry(l.Directory, func(entry os.DirEntry) bool { if entry.IsDir() { - continue + return true } name := entry.Name() ext := path.Ext(name) if !re.MatchString(ext) { - continue + return true } info, err := entry.Info() if err != nil || info.Size() == 0 { - continue + return true } shardId, err := strconv.ParseInt(ext[3:], 10, 64) if err != nil || shardId < 0 || shardId > 255 { - continue + return true } base := name[:len(name)-len(ext)] collection, vid, err := parseCollectionVolumeId(base) if err != nil { - continue + return true } if _, loaded := l.FindEcShard(vid, erasure_coding.ShardId(shardId)); loaded { - continue + return true } key := ecKeyForReconcile{collection: collection, vid: vid} orphans[key] = append(orphans[key], name) + return true + }); err != nil { + return nil + } + // os.ReadDir used to hand these back sorted; the shard lists are logged + // and mounted in order, so keep them so. + for _, shards := range orphans { + slices.Sort(shards) } return orphans } diff --git a/weed/storage/store_volume_report.go b/weed/storage/store_volume_report.go index 70e8f8a94..d74ef72ab 100644 --- a/weed/storage/store_volume_report.go +++ b/weed/storage/store_volume_report.go @@ -2,6 +2,7 @@ package storage import ( "sync" + "unique" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" ) @@ -14,13 +15,32 @@ type volumeReportKey struct { volumeId uint32 } +// reportedIdentity is what naming a departure needs beyond the volume and disk +// ids the key already holds. Volumes share very few distinct values here — one +// per collection, placement, version, ttl and disk type in use — so entries +// hold a handle to a shared copy, which keeps them small enough that a server +// with millions of volumes is not paying for identity per volume. +// +// The handle is what keeps the shared copy alive, and dropping the last one +// clears the entry. Handing out the value and letting the handle go, as +// internVolumeString warns against, would have the next volume make a second +// copy. +type reportedIdentity struct { + collection string + diskType string + replicaPlacement uint32 + version uint32 + ttl uint32 +} + // reportedVolume is what the master was told about one volume copy: the hash // that detects change, the heartbeat pass that last found the copy held, and -// enough identity to name the volume if it departs. +// the identity that names the volume if it departs. The departure message +// itself is built on the way out, since almost no volume ever leaves. type reportedVolume struct { - hash uint64 - pass uint64 - short *master_pb.VolumeShortInformationMessage + hash uint64 + pass uint64 + identity unique.Handle[reportedIdentity] } // volumeReportState remembers what the master was last told about each volume, @@ -88,9 +108,7 @@ func (s *volumeReportState) record(m *master_pb.VolumeInformationMessage, hash u if previous, known := s.lastReported[key]; known { changed := previous.hash != hash if changed { - // Only a departure hands a short message out, and that entry leaves - // the map in the same step, so the one held here is read by no one. - fillShortInformation(previous.short, m) + previous.identity = identityOf(m) } previous.hash, previous.pass = hash, pass s.lastReported[key] = previous @@ -99,20 +117,31 @@ func (s *volumeReportState) record(m *master_pb.VolumeInformationMessage, hash u if s.lastReported == nil { s.lastReported = make(map[volumeReportKey]reportedVolume) } - short := &master_pb.VolumeShortInformationMessage{} - fillShortInformation(short, m) - s.lastReported[key] = reportedVolume{hash: hash, pass: pass, short: short} + s.lastReported[key] = reportedVolume{hash: hash, pass: pass, identity: identityOf(m)} return true } -func fillShortInformation(short *master_pb.VolumeShortInformationMessage, m *master_pb.VolumeInformationMessage) { - short.Id = m.Id - short.Collection = m.Collection - short.ReplicaPlacement = m.ReplicaPlacement - short.Version = m.Version - short.Ttl = m.Ttl - short.DiskType = m.DiskType - short.DiskId = m.DiskId +func identityOf(m *master_pb.VolumeInformationMessage) unique.Handle[reportedIdentity] { + return unique.Make(reportedIdentity{ + collection: m.Collection, + diskType: m.DiskType, + replicaPlacement: m.ReplicaPlacement, + version: m.Version, + ttl: m.Ttl, + }) +} + +func (held reportedVolume) toShortInformation(key volumeReportKey) *master_pb.VolumeShortInformationMessage { + identity := held.identity.Value() + return &master_pb.VolumeShortInformationMessage{ + Id: key.volumeId, + Collection: identity.collection, + ReplicaPlacement: identity.replicaPlacement, + Version: identity.version, + Ttl: identity.ttl, + DiskType: identity.diskType, + DiskId: key.diskId, + } } // commit closes the heartbeat. Copies this pass did not find are forgotten, so @@ -151,7 +180,7 @@ func (s *volumeReportState) commit(pass uint64, generation uint64, full bool) [] var gone []*master_pb.VolumeShortInformationMessage for _, key := range goneKeys { if !full && goneIds[key.volumeId] { - gone = append(gone, s.lastReported[key].short) + gone = append(gone, s.lastReported[key].toShortInformation(key)) } delete(s.lastReported, key) } diff --git a/weed/storage/store_volume_report_test.go b/weed/storage/store_volume_report_test.go index 9a2d80c86..6725c3384 100644 --- a/weed/storage/store_volume_report_test.go +++ b/weed/storage/store_volume_report_test.go @@ -274,3 +274,30 @@ func TestOverlappingHeartbeatsNameNoDepartures(t *testing.T) { } } } + +// The departure message is built from what the entry held, not kept ready as a +// message per volume, so everything the master matches on has to survive the +// round trip through the entry. +func TestDepartureCarriesTheVolumeIdentity(t *testing.T) { + store := newTestStore(t, 1) + v := mountTestVolume(t, store.Locations[0], 7, "pictures") + v.SuperBlock.ReplicaPlacement = &super_block.ReplicaPlacement{SameRackCount: 1} + v.SuperBlock.Ttl, _ = needle.ReadTTL("5m") + v.diskId = 3 + store.ResetVolumeReporting() + store.AcceptVolumeChanges() + reported := store.CollectHeartbeat().Volumes[0] + + store.Locations[0].UnloadVolume(needle.VolumeId(7)) + heartbeat := store.CollectHeartbeat() + if len(heartbeat.DeletedVolumes) != 1 { + t.Fatalf("expected volume 7 to be named as departed, got %v", heartbeat.DeletedVolumes) + } + departed := heartbeat.DeletedVolumes[0] + if departed.Id != reported.Id || departed.DiskId != reported.DiskId || + departed.Collection != reported.Collection || departed.DiskType != reported.DiskType || + departed.ReplicaPlacement != reported.ReplicaPlacement || + departed.Version != reported.Version || departed.Ttl != reported.Ttl { + t.Errorf("departure named %v, want it to match the volume as reported %v", departed, reported) + } +} diff --git a/weed/storage/volume_tier.go b/weed/storage/volume_tier.go index 1f0214b9e..204bb2076 100644 --- a/weed/storage/volume_tier.go +++ b/weed/storage/volume_tier.go @@ -22,6 +22,7 @@ func (v *Volume) maybeLoadVolumeInfo() (found bool) { var hasRemoteFile bool v.volumeInfo, hasRemoteFile, found, err = volume_info.MaybeLoadVolumeInfo(v.FileName(".vif")) v.hasRemoteFile.Store(hasRemoteFile) + internVolumeInfoStrings(v.volumeInfo) if v.volumeInfo.Version == 0 { v.volumeInfo.Version = uint32(needle.GetCurrentVersion()) @@ -56,6 +57,20 @@ func (v *Volume) maybeLoadVolumeInfo() (found bool) { } +// internVolumeInfoStrings shares the values every volume's .vif repeats. A +// tiered volume names its replication and its backend on every load, and the +// decode allocates a fresh copy of each, so a server holding millions of them +// otherwise holds millions of copies of the same handful of names. The remote +// key is left alone: it names one volume. +func internVolumeInfoStrings(volumeInfo *volume_server_pb.VolumeInfo) { + volumeInfo.Replication = internVolumeString(volumeInfo.Replication) + for _, remoteFile := range volumeInfo.GetFiles() { + remoteFile.BackendType = internVolumeString(remoteFile.BackendType) + remoteFile.BackendId = internVolumeString(remoteFile.BackendId) + remoteFile.Extension = internVolumeString(remoteFile.Extension) + } +} + func (v *Volume) HasRemoteFile() bool { return v.hasRemoteFile.Load() } diff --git a/weed/storage/volume_tier_intern_test.go b/weed/storage/volume_tier_intern_test.go new file mode 100644 index 000000000..253309669 --- /dev/null +++ b/weed/storage/volume_tier_intern_test.go @@ -0,0 +1,59 @@ +package storage + +import ( + "runtime" + "testing" + "unsafe" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/volume_info" +) + +// Every tiered volume names the same backend and replication, so a server +// holding millions of them must end up with one copy of each. The sharing has +// to hold across a collection: a table that lets its entries go hands the next +// volume a second copy. The remote key names a single volume and is left alone. +func TestVolumeInfoSharesTheStringsEveryVolumeRepeats(t *testing.T) { + dir := t.TempDir() + load := func(name, key string) *volume_server_pb.VolumeInfo { + t.Helper() + path := dir + "/" + name + if err := volume_info.SaveVolumeInfo(path, &volume_server_pb.VolumeInfo{ + Version: 3, + Replication: "001", + Files: []*volume_server_pb.RemoteFile{{ + BackendType: "s3", BackendId: "cold", Extension: ".dat", Key: key, + }}, + }); err != nil { + t.Fatal(err) + } + loaded, _, found, err := volume_info.MaybeLoadVolumeInfo(path) + if err != nil || !found { + t.Fatalf("load %s: found=%v err=%v", name, found, err) + } + internVolumeInfoStrings(loaded) + return loaded + } + + first := load("1.vif", "one") + runtime.GC() + second := load("2.vif", "two") + + shared := func(what, a, b string) { + t.Helper() + if a != b { + t.Fatalf("%s read back as %q then %q", what, a, b) + } + if unsafe.StringData(a) != unsafe.StringData(b) { + t.Errorf("%s was kept twice instead of shared", what) + } + } + shared("replication", first.Replication, second.Replication) + shared("backend type", first.Files[0].BackendType, second.Files[0].BackendType) + shared("backend id", first.Files[0].BackendId, second.Files[0].BackendId) + shared("extension", first.Files[0].Extension, second.Files[0].Extension) + + if first.Files[0].Key != "one" || second.Files[0].Key != "two" { + t.Errorf("remote keys came back as %q and %q", first.Files[0].Key, second.Files[0].Key) + } +}