From 836c60b2c92fe9993205f34ac18f1824b30f8ad0 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 3 Nov 2026 12:48:16 -0800 Subject: [PATCH] topology: refresh EC file and delete counts from heartbeats (#12378) * topology: refresh EC file and delete counts from heartbeats UpdateEcShards only re-stored a heartbeated EC record when shard membership changed, so FileCount and DeleteCount stayed at whatever the first heartbeat after registration carried, while the volume server reports fresh tombstone counts on every full EC heartbeat. Per-collection object counts and EC vacuum candidate detection read the stored values. Compare the counts per (volume, disk) against the record already in hand and copy them onto the stored record in place, so the refresh raises no new/deleted shard events and rebuilds no maps. A message reporting 0/0 did not count (older volume servers, and the incremental mount message omits the fields), so it keeps the stored values, also across a membership-triggered rewrite. The incremental path's existing branch picks counts up the same way. * topology: swap EC shard records instead of mutating them in place GetEcShards hands out the stored *EcVolumeInfo pointers and releases ecShardsLock; CollectionStatistics, CollectionEcVolumeStats, ToDiskInfo and the master's shard broadcast read fields through them afterwards. The heartbeat count refresh wrote FileCount/DeleteCount into those records under the lock, which the readers never take, so it was a data race and a reader could pair a FileCount from one heartbeat with a DeleteCount from the next. Disk.AddOrUpdateEcShard had the same write on the incremental path, and its shard-bits/ratio merge was already in place. Stored records are now never written after they are published: a count refresh, an incremental merge and a shard delete each build a copy beside the stored record and swap the map entry under ecShardsLock, so a reader holding the old pointer keeps one heartbeat's snapshot. The refresh compares against whatever record is current under the lock, since the incremental path may have replaced it since UpdateEcShards looked. The 0/0-means-not-reported rule is unchanged. The count tests now assert that the stored pointer changes while a previously obtained one still reads its old values, and a -race test runs full and incremental count refreshes against readers; it failed on the previous code and passes now. --- weed/topology/data_node_ec.go | 18 ++ weed/topology/data_node_ec_counts_test.go | 241 ++++++++++++++++++++++ weed/topology/disk_ec.go | 46 ++++- 3 files changed, 298 insertions(+), 7 deletions(-) create mode 100644 weed/topology/data_node_ec_counts_test.go diff --git a/weed/topology/data_node_ec.go b/weed/topology/data_node_ec.go index de38fa6cb2..095cc30212 100644 --- a/weed/topology/data_node_ec.go +++ b/weed/topology/data_node_ec.go @@ -78,6 +78,12 @@ func (dn *DataNode) UpdateEcShards(actualShards []*erasure_coding.EcVolumeInfo) deletedShards = append(deletedShards, d) deletedShardCount += d.ShardsInfo.Count() } + // Counts move without any shard changing hands, and the + // rewrite below only runs on a membership change, so publish + // the new counts on their own. + if ecCountsChanged(ecShards, actualEcShards) { + disk.refreshEcShardCounts(key.vid, key.diskId, actualEcShards.FileCount, actualEcShards.DeleteCount) + } } if (newShardCount - deletedShardCount) != 0 { @@ -113,6 +119,18 @@ func (dn *DataNode) UpdateEcShards(actualShards []*erasure_coding.EcVolumeInfo) return } +// ecCountsChanged reports whether a heartbeat moves the stored FileCount or +// DeleteCount. A message reporting 0/0 did not count (older volume servers and +// the incremental mount message omit the fields), so it keeps the stored values +// and carries them onto the incoming record in case the map is rewritten. +func ecCountsChanged(stored, actual *erasure_coding.EcVolumeInfo) bool { + if actual.FileCount == 0 && actual.DeleteCount == 0 { + actual.FileCount, actual.DeleteCount = stored.FileCount, stored.DeleteCount + return false + } + return actual.FileCount != stored.FileCount || actual.DeleteCount != stored.DeleteCount +} + // getEcShardsLocked returns the flat list of per-physical-disk EC shard // entries across all children. Caller MUST hold dn.Lock (or RLock); each // disk's ecShardsLock is taken internally via Disk.GetEcShards. diff --git a/weed/topology/data_node_ec_counts_test.go b/weed/topology/data_node_ec_counts_test.go new file mode 100644 index 0000000000..e91ee7ef56 --- /dev/null +++ b/weed/topology/data_node_ec_counts_test.go @@ -0,0 +1,241 @@ +package topology + +import ( + "sync" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/sequence" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +func newEcCountsTestNode(t *testing.T) (*Topology, *DataNode) { + t.Helper() + topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false) + dc := topo.GetOrCreateDataCenter("dc1") + rack := dc.GetOrCreateRack("rack1") + dn := rack.GetOrCreateDataNode("127.0.0.1", 34534, 0, "127.0.0.1", "", map[string]uint32{"": 100}) + return topo, dn +} + +func ecMessageWithCounts(vid uint32, diskId uint32, shardIds []erasure_coding.ShardId, fileCount, deleteCount uint64) *master_pb.VolumeEcShardInformationMessage { + m := buildEcShardMessage(vid, "col", "", diskId, shardIds) + m.FileCount = fileCount + m.DeleteCount = deleteCount + return m +} + +func storedEcCounts(t *testing.T, dn *DataNode, vid uint32, diskId uint32) (*erasure_coding.EcVolumeInfo, uint64, uint64) { + t.Helper() + for _, ev := range dn.GetEcShards() { + if ev.VolumeId == needle.VolumeId(vid) && ev.DiskId == diskId { + return ev, ev.FileCount, ev.DeleteCount + } + } + t.Fatalf("volume %d disk %d: no stored ec shard record", vid, diskId) + return nil, 0, 0 +} + +func TestSyncRefreshesEcCountsWithoutShardChange(t *testing.T) { + topo, dn := newEcCountsTestNode(t) + shards := []erasure_coding.ShardId{0, 1, 2} + + topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + ecMessageWithCounts(100, 0, shards, 1000, 10), + }, dn) + before, fileCount, deleteCount := storedEcCounts(t, dn, 100, 0) + if fileCount != 1000 || deleteCount != 10 { + t.Fatalf("initial counts = %d/%d, want 1000/10", fileCount, deleteCount) + } + + // Same shards, more tombstones: the next full heartbeat must land. + newShards, deletedShards := topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + ecMessageWithCounts(100, 0, shards, 1000, 250), + }, dn) + if len(newShards) != 0 || len(deletedShards) != 0 { + t.Errorf("count-only heartbeat reported %d new / %d deleted shards, want none", len(newShards), len(deletedShards)) + } + after, fileCount, deleteCount := storedEcCounts(t, dn, 100, 0) + if fileCount != 1000 || deleteCount != 250 { + t.Errorf("counts after heartbeat = %d/%d, want 1000/250", fileCount, deleteCount) + } + // Readers keep the pointers GetEcShards handed out after the lock is + // released, so a refresh must publish a new record rather than write + // into the one they may still be reading. + if after == before { + t.Errorf("count-only heartbeat updated the stored record in place instead of swapping in a copy") + } + if before.FileCount != 1000 || before.DeleteCount != 10 { + t.Errorf("previously obtained record now reads %d/%d, want the 1000/10 snapshot it was taken with", before.FileCount, before.DeleteCount) + } + if after.ShardsInfo.Count() != len(shards) { + t.Errorf("stored shard count = %d, want %d", after.ShardsInfo.Count(), len(shards)) + } + locs, ok := topo.LookupEcShards(100) + if !ok { + t.Fatal("volume 100 lost its shard locations") + } + for _, sid := range shards { + if len(locs.Locations[sid]) != 1 { + t.Errorf("shard %d has %d locations, want 1", sid, len(locs.Locations[sid])) + } + } +} + +func TestSyncKeepsEcCountsWhenNotReported(t *testing.T) { + topo, dn := newEcCountsTestNode(t) + shards := []erasure_coding.ShardId{0, 1, 2} + + topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + ecMessageWithCounts(100, 0, shards, 1000, 10), + }, dn) + + // An older volume server reports no counts: keep what we have. + newShards, deletedShards := topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + ecMessageWithCounts(100, 0, shards, 0, 0), + }, dn) + if len(newShards) != 0 || len(deletedShards) != 0 { + t.Errorf("unchanged heartbeat reported %d new / %d deleted shards, want none", len(newShards), len(deletedShards)) + } + if _, fileCount, deleteCount := storedEcCounts(t, dn, 100, 0); fileCount != 1000 || deleteCount != 10 { + t.Errorf("counts after 0/0 heartbeat = %d/%d, want 1000/10 kept", fileCount, deleteCount) + } + + // A membership change rewrites the record; an unreported count must not zero it. + newShards, deletedShards = topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + ecMessageWithCounts(100, 0, append(shards, 3), 0, 0), + }, dn) + if len(newShards) != 1 || len(deletedShards) != 0 { + t.Fatalf("adding a shard reported %d new / %d deleted shards, want 1 / 0", len(newShards), len(deletedShards)) + } + if newShards[0].ShardsInfo.Count() != 1 || !newShards[0].ShardsInfo.Has(3) { + t.Errorf("new shard delta = %v, want shard 3 only", newShards[0].ShardsInfo.Ids()) + } + if _, fileCount, deleteCount := storedEcCounts(t, dn, 100, 0); fileCount != 1000 || deleteCount != 10 { + t.Errorf("counts after rewrite with 0/0 = %d/%d, want 1000/10 kept", fileCount, deleteCount) + } +} + +func TestSyncMembershipChangeCarriesNewEcCounts(t *testing.T) { + topo, dn := newEcCountsTestNode(t) + + topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + ecMessageWithCounts(100, 0, []erasure_coding.ShardId{0, 1, 2}, 1000, 10), + }, dn) + + newShards, deletedShards := topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + ecMessageWithCounts(100, 0, []erasure_coding.ShardId{0, 1, 3}, 1000, 300), + }, dn) + if len(newShards) != 1 || len(deletedShards) != 1 { + t.Fatalf("shard swap reported %d new / %d deleted shards, want 1 / 1", len(newShards), len(deletedShards)) + } + if !newShards[0].ShardsInfo.Has(3) || newShards[0].ShardsInfo.Count() != 1 { + t.Errorf("new shard delta = %v, want shard 3 only", newShards[0].ShardsInfo.Ids()) + } + if !deletedShards[0].ShardsInfo.Has(2) || deletedShards[0].ShardsInfo.Count() != 1 { + t.Errorf("deleted shard delta = %v, want shard 2 only", deletedShards[0].ShardsInfo.Ids()) + } + if _, fileCount, deleteCount := storedEcCounts(t, dn, 100, 0); fileCount != 1000 || deleteCount != 300 { + t.Errorf("counts after shard swap = %d/%d, want 1000/300", fileCount, deleteCount) + } +} + +func TestIncrementalSyncRefreshesEcCounts(t *testing.T) { + topo, dn := newEcCountsTestNode(t) + + topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + ecMessageWithCounts(100, 0, []erasure_coding.ShardId{0, 1}, 1000, 10), + }, dn) + + // The mount message carries no counts: keep the stored ones. + topo.IncrementalSyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + ecMessageWithCounts(100, 0, []erasure_coding.ShardId{2}, 0, 0), + }, nil, dn) + record, fileCount, deleteCount := storedEcCounts(t, dn, 100, 0) + if fileCount != 1000 || deleteCount != 10 { + t.Errorf("counts after 0/0 mount = %d/%d, want 1000/10 kept", fileCount, deleteCount) + } + if record.ShardsInfo.Count() != 3 { + t.Errorf("stored shard count after mount = %d, want 3", record.ShardsInfo.Count()) + } + + // A message that does count refreshes the existing record, again by + // swapping in a copy so an escaped pointer keeps its snapshot. + topo.IncrementalSyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + ecMessageWithCounts(100, 0, []erasure_coding.ShardId{3}, 1000, 400), + }, nil, dn) + after, fileCount, deleteCount := storedEcCounts(t, dn, 100, 0) + if fileCount != 1000 || deleteCount != 400 { + t.Errorf("counts after counted mount = %d/%d, want 1000/400", fileCount, deleteCount) + } + if after == record { + t.Errorf("counted mount updated the stored record in place instead of swapping in a copy") + } + if record.FileCount != 1000 || record.DeleteCount != 10 || record.ShardsInfo.Count() != 3 { + t.Errorf("previously obtained record now reads %d/%d with %d shards, want the 1000/10, 3-shard snapshot it was taken with", + record.FileCount, record.DeleteCount, record.ShardsInfo.Count()) + } +} + +// TestEcCountRefreshDoesNotRaceReaders runs full and incremental heartbeats +// that only move the counts against readers that keep reading the records +// GetEcShards handed out, the way CollectionStatistics and ToDiskInfo do. +// Run with -race: an in-place refresh is a write to memory a reader holds +// without the lock. Without -race it still checks that every reader sees +// one heartbeat's pair, never a FileCount from one and a DeleteCount from +// another. +func TestEcCountRefreshDoesNotRaceReaders(t *testing.T) { + topo, dn := newEcCountsTestNode(t) + shards := []erasure_coding.ShardId{0, 1, 2} + pairs := [][2]uint64{{1000, 10}, {2000, 20}} + + topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + ecMessageWithCounts(100, 0, shards, pairs[0][0], pairs[0][1]), + }, dn) + + done := make(chan struct{}) + var readers sync.WaitGroup + for r := 0; r < 4; r++ { + readers.Add(1) + go func() { + defer readers.Done() + for { + select { + case <-done: + return + default: + } + for _, ev := range dn.GetEcShards() { + fileCount, deleteCount := ev.FileCount, ev.DeleteCount + if (fileCount != pairs[0][0] || deleteCount != pairs[0][1]) && (fileCount != pairs[1][0] || deleteCount != pairs[1][1]) { + t.Errorf("reader saw counts %d/%d, which no heartbeat sent", fileCount, deleteCount) + return + } + } + topo.CollectionEcVolumeStats("col") + } + }() + } + + for i := 1; i <= 300; i++ { + pair := pairs[i%2] + topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + ecMessageWithCounts(100, 0, shards, pair[0], pair[1]), + }, dn) + pair = pairs[(i+1)%2] + // A re-mount of a shard the node already holds takes the + // existing-record branch of AddOrUpdateEcShard. + topo.IncrementalSyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + ecMessageWithCounts(100, 0, shards[:1], pair[0], pair[1]), + }, nil, dn) + } + close(done) + readers.Wait() + + if record, fileCount, deleteCount := storedEcCounts(t, dn, 100, 0); fileCount != pairs[1][0] || deleteCount != pairs[1][1] { + t.Errorf("final counts = %d/%d, want %d/%d", fileCount, deleteCount, pairs[1][0], pairs[1][1]) + } else if record.ShardsInfo.Count() != len(shards) { + t.Errorf("final shard count = %d, want %d", record.ShardsInfo.Count(), len(shards)) + } +} diff --git a/weed/topology/disk_ec.go b/weed/topology/disk_ec.go index d92bedf0da..4aca6b2824 100644 --- a/weed/topology/disk_ec.go +++ b/weed/topology/disk_ec.go @@ -6,6 +6,10 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage/types" ) +// GetEcShards returns the stored records themselves, not copies. Callers +// read them after the lock is released, which is safe only because no stored +// record is ever written in place: every change swaps a replacement into the +// map, so a caller holding an old pointer sees one heartbeat's snapshot. func (d *Disk) GetEcShards() (ret []*erasure_coding.EcVolumeInfo) { d.ecShardsLock.RLock() defer d.ecShardsLock.RUnlock() @@ -40,9 +44,16 @@ func (d *Disk) AddOrUpdateEcShard(s *erasure_coding.EcVolumeInfo) { byDisk[diskId] = s delta = s.ShardsInfo.Count() } else { - oldCount := existing.ShardsInfo.Count() - existing.ShardsInfo.Add(s.ShardsInfo) - delta = existing.ShardsInfo.Count() - oldCount + // Build the merged record beside the stored one and swap it in; + // see GetEcShards for why the stored one must stay as it is. + updated := *existing + updated.ShardsInfo = existing.ShardsInfo.Plus(s.ShardsInfo) + delta = updated.ShardsInfo.Count() - existing.ShardsInfo.Count() + // 0/0 means the message did not count (the mount message omits the fields) + if s.FileCount != 0 || s.DeleteCount != 0 { + updated.FileCount, updated.DeleteCount = s.FileCount, s.DeleteCount + } + byDisk[diskId] = &updated } if delta != 0 { @@ -52,6 +63,25 @@ func (d *Disk) AddOrUpdateEcShard(s *erasure_coding.EcVolumeInfo) { } } +// refreshEcShardCounts publishes the counts a full heartbeat reported for one +// (volume, disk) record. The stored record is left untouched: a copy carrying +// the new counts takes its place, so a reader that already holds the old +// pointer keeps reading one heartbeat's FileCount/DeleteCount pair. The +// comparison runs here, under the lock, against whatever record is current, +// since the incremental path may have replaced it since the caller looked. +func (d *Disk) refreshEcShardCounts(vid needle.VolumeId, diskId types.DiskId, fileCount, deleteCount uint64) { + d.ecShardsLock.Lock() + defer d.ecShardsLock.Unlock() + + existing, ok := d.ecShards[vid][diskId] + if !ok || (existing.FileCount == fileCount && existing.DeleteCount == deleteCount) { + return + } + updated := *existing + updated.FileCount, updated.DeleteCount = fileCount, deleteCount + d.ecShards[vid][diskId] = &updated +} + func (d *Disk) DeleteEcShard(s *erasure_coding.EcVolumeInfo) { d.ecShardsLock.Lock() defer d.ecShardsLock.Unlock() @@ -65,20 +95,22 @@ func (d *Disk) DeleteEcShard(s *erasure_coding.EcVolumeInfo) { if !ok { return } - oldCount := existing.ShardsInfo.Count() - existing.ShardsInfo.Subtract(s.ShardsInfo) - delta := existing.ShardsInfo.Count() - oldCount + updated := *existing + updated.ShardsInfo = existing.ShardsInfo.Minus(s.ShardsInfo) + delta := updated.ShardsInfo.Count() - existing.ShardsInfo.Count() if delta != 0 { d.UpAdjustDiskUsageDelta(types.ToDiskType(string(d.Id())), &DiskUsageCounts{ ecShardCount: int64(delta), }) } - if existing.ShardsInfo.Count() == 0 { + if updated.ShardsInfo.Count() == 0 { delete(byDisk, diskId) if len(byDisk) == 0 { delete(d.ecShards, s.VolumeId) } + } else if delta != 0 { + byDisk[diskId] = &updated } }