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

@@ -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.

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))
+	}
+}

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
 	}
 }