mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-17 12:46:59 +00:00
Chris LuandGitHub
f1f720f5da
* refactor(types): add DiskId type for physical-disk identifiers Names the uint32 physical-disk index that volume servers carry in VolumeEcShardInformationMessage / VolumeInformationMessage, so EC shard tracking that needs to distinguish disks within a DataNode can use a dedicated type instead of an untyped uint32. No behaviour change. * fix(master): register EC shards per physical disk on full heartbeat sync (#9212) When a volume's EC shards are spread across multiple physical disks on the same volume server (common after ec.balance / ec.rebuild on multi-disk nodes), the volume server emits one VolumeEcShardInformationMessage per (disk, volume) in its heartbeat. The master's DataNode.UpdateEcShards was building a `map[VolumeId]*EcVolumeInfo` with last-write-wins, and doUpdateEcShards then overwrote `disk.ecShards[vid]` once per message, so all but the final disk's shards were silently dropped. Only the topology-global ecShardMap (built via RegisterEcShards in a per-message loop) stayed correct, which hid the problem from `topo.LookupEcShards` but broke everything that reads the DataNode/Disk view — volume.list, admin UI, ec.rebuild dry-run ("only 6 shards, skipping"), and `DiskInfo.EcShardInfos` which the shell's ec.balance / ec.rebuild planners group by `eci.DiskId`. Change the shape of `Disk.ecShards` from map[VolumeId]*EcVolumeInfo to map[VolumeId]map[types.DiskId]*EcVolumeInfo so every physical disk keeps its own entry. UpdateEcShards aggregates incoming messages by (vid, diskId) rather than vid alone; Add/Delete/ HasVolumesById and HasEcShards consult the nested map; doUpdateEcShards rewrites the nested structure from the aggregated map. Per-physical-disk attribution survives through DataNode.ToDataNodeInfo -> DiskInfo.EcShardInfos, matching the wire format the volume server produces and what downstream admin tooling expects. Delta sync (AddOrUpdateEcShard / DeleteEcShard) already merged via ShardsInfo.Add, so this only affects the full-sync path that runs on heartbeat reconnect. Adds data_node_ec_multi_disk_test.go with two regression tests that fail on pre-fix master: - TestEcShardsAcrossMultipleDisksOnSameNode: volume 15 spread over 3 disks (matches the bug report's volume-2 row); asserts every shard visible via LookupEcShards, DataNode.GetEcShards, and ToDataNodeInfo's per-disk EcShardInfos entries. - TestEcShardsAfterRestartHeartbeat: minimal 2-disk full sync case. * fix(topology): tighten locking around EC shard map access Addresses review comments on #9219: * DataNode.UpdateEcShards now holds dn.Lock for the full read-diff-write cycle, matching UpdateVolumes' model, so concurrent heartbeats can no longer interleave their getOrCreateDisk / UpAdjustDiskUsageDelta updates with each other. Introduces a private getEcShardsLocked helper for reads under the held lock; renames doUpdateEcShards to doUpdateEcShardsLocked for the same reason. * DataNode.HasEcShards now takes each disk's ecShardsLock while reading disk.ecShards, closing a pre-existing map race with concurrent Add/Delete/Update writers. * doUpdateEcShardsLocked takes each disk's ecShardsLock around the reset-and-rewrite so readers (GetEcShards, HasEcShards) see a consistent map state rather than a partially-rebuilt one. * Disk.GetEcShards' slice-capacity hint now accounts for the nested per-physical-disk entries (sum of inner lengths) instead of underestimating by the unique-volume count.
107 lines
2.5 KiB
Go
107 lines
2.5 KiB
Go
package topology
|
|
|
|
import (
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/types"
|
|
)
|
|
|
|
func (d *Disk) GetEcShards() (ret []*erasure_coding.EcVolumeInfo) {
|
|
d.ecShardsLock.RLock()
|
|
defer d.ecShardsLock.RUnlock()
|
|
// Total entries = sum of per-volume per-disk entries, which typically
|
|
// exceeds the number of unique volumes once shards span multiple disks.
|
|
total := 0
|
|
for _, byDisk := range d.ecShards {
|
|
total += len(byDisk)
|
|
}
|
|
ret = make([]*erasure_coding.EcVolumeInfo, 0, total)
|
|
for _, byDisk := range d.ecShards {
|
|
for _, ecVolumeInfo := range byDisk {
|
|
ret = append(ret, ecVolumeInfo)
|
|
}
|
|
}
|
|
return ret
|
|
}
|
|
|
|
func (d *Disk) AddOrUpdateEcShard(s *erasure_coding.EcVolumeInfo) {
|
|
d.ecShardsLock.Lock()
|
|
defer d.ecShardsLock.Unlock()
|
|
|
|
byDisk, ok := d.ecShards[s.VolumeId]
|
|
if !ok {
|
|
byDisk = make(map[types.DiskId]*erasure_coding.EcVolumeInfo, 1)
|
|
d.ecShards[s.VolumeId] = byDisk
|
|
}
|
|
|
|
diskId := types.DiskId(s.DiskId)
|
|
delta := 0
|
|
if existing, ok := byDisk[diskId]; !ok {
|
|
byDisk[diskId] = s
|
|
delta = s.ShardsInfo.Count()
|
|
} else {
|
|
oldCount := existing.ShardsInfo.Count()
|
|
existing.ShardsInfo.Add(s.ShardsInfo)
|
|
delta = existing.ShardsInfo.Count() - oldCount
|
|
}
|
|
|
|
if delta != 0 {
|
|
d.UpAdjustDiskUsageDelta(types.ToDiskType(string(d.Id())), &DiskUsageCounts{
|
|
ecShardCount: int64(delta),
|
|
})
|
|
}
|
|
}
|
|
|
|
func (d *Disk) DeleteEcShard(s *erasure_coding.EcVolumeInfo) {
|
|
d.ecShardsLock.Lock()
|
|
defer d.ecShardsLock.Unlock()
|
|
|
|
byDisk, ok := d.ecShards[s.VolumeId]
|
|
if !ok {
|
|
return
|
|
}
|
|
diskId := types.DiskId(s.DiskId)
|
|
existing, ok := byDisk[diskId]
|
|
if !ok {
|
|
return
|
|
}
|
|
oldCount := existing.ShardsInfo.Count()
|
|
existing.ShardsInfo.Subtract(s.ShardsInfo)
|
|
delta := existing.ShardsInfo.Count() - oldCount
|
|
|
|
if delta != 0 {
|
|
d.UpAdjustDiskUsageDelta(types.ToDiskType(string(d.Id())), &DiskUsageCounts{
|
|
ecShardCount: int64(delta),
|
|
})
|
|
}
|
|
if existing.ShardsInfo.Count() == 0 {
|
|
delete(byDisk, diskId)
|
|
if len(byDisk) == 0 {
|
|
delete(d.ecShards, s.VolumeId)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (d *Disk) HasVolumesById(id needle.VolumeId) (hasVolumeId bool) {
|
|
// check whether normal volumes has this volume id
|
|
d.RLock()
|
|
_, ok := d.volumes[id]
|
|
if ok {
|
|
hasVolumeId = true
|
|
}
|
|
d.RUnlock()
|
|
|
|
if hasVolumeId {
|
|
return
|
|
}
|
|
|
|
// check whether ec shards has this volume id
|
|
d.ecShardsLock.RLock()
|
|
if byDisk, ok := d.ecShards[id]; ok && len(byDisk) > 0 {
|
|
hasVolumeId = true
|
|
}
|
|
d.ecShardsLock.RUnlock()
|
|
|
|
return
|
|
}
|