From e0002f8dd7c94820a162bd1f40bb5db64bb40901 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 24 Oct 2020 01:34:31 -0700 Subject: [PATCH] check existing volumes for writable status --- weed/topology/volume_layout.go | 27 +++++++++++++++++++-------- 1 file changed, 19 insertions(+), 8 deletions(-) diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index dddcfc9c9..ffe36e95b 100644 --- a/weed/topology/volume_layout.go +++ b/weed/topology/volume_layout.go @@ -139,7 +139,7 @@ func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) { vl.accessLock.Lock() defer vl.accessLock.Unlock() - defer vl.ensureCorrectWritables(v) + defer vl.ensureCorrectWritables(v.Id) defer vl.rememberOversizedVolume(v, dn) if _, ok := vl.vid2location[v.Id]; !ok { @@ -189,7 +189,7 @@ func (vl *VolumeLayout) UnRegisterVolume(v *storage.VolumeInfo, dn *DataNode) { vl.readonlyVolumes.Remove(v.Id, dn) vl.oversizedVolumes.Remove(v.Id, dn) - vl.ensureCorrectWritables(v) + vl.ensureCorrectWritables(v.Id) if location.Length() == 0 { delete(vl.vid2location, v.Id) @@ -202,19 +202,30 @@ func (vl *VolumeLayout) EnsureCorrectWritables(v *storage.VolumeInfo) { vl.accessLock.Lock() defer vl.accessLock.Unlock() - vl.ensureCorrectWritables(v) + vl.ensureCorrectWritables(v.Id) } -func (vl *VolumeLayout) ensureCorrectWritables(v *storage.VolumeInfo) { - if vl.enoughCopies(v.Id) && vl.isWritable(v) { - if !vl.oversizedVolumes.IsTrue(v.Id) { - vl.setVolumeWritable(v.Id) +func (vl *VolumeLayout) ensureCorrectWritables(vid needle.VolumeId) { + if vl.enoughCopies(vid) && vl.isAllWritable(vid) { + if !vl.oversizedVolumes.IsTrue(vid) { + vl.setVolumeWritable(vid) } } else { - vl.removeFromWritable(v.Id) + vl.removeFromWritable(vid) } } +func (vl *VolumeLayout) isAllWritable(vid needle.VolumeId) bool { + for _, dn := range vl.vid2location[vid].list { + if v, found := dn.volumes[vid]; found { + if v.ReadOnly { + return false + } + } + } + return true +} + func (vl *VolumeLayout) isOversized(v *storage.VolumeInfo) bool { return uint64(v.Size) >= vl.volumeSizeLimit }