gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

add locking to volume layout vacuum

Browse source
This commit is contained in:
Chris Lu 2018-10-18 20:34:43 -07:00
parent 46eb77f9bb
commit 201c24c110
2 changed files with 33 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
weed/topology/topology_event_handling.go
View file

@ -40,6 +40,10 @@ func (t *Topology) SetVolumeCapacityFull(volumeInfo storage.VolumeInfo) bool {
if !vl.SetVolumeCapacityFull(volumeInfo.Id) { if !vl.SetVolumeCapacityFull(volumeInfo.Id) {
return false return false
} }
vl.accessLock.RLock()
defer vl.accessLock.RUnlock()
for _, dn := range vl.vid2location[volumeInfo.Id].list { for _, dn := range vl.vid2location[volumeInfo.Id].list {
if !volumeInfo.ReadOnly { if !volumeInfo.ReadOnly {
dn.UpAdjustActiveVolumeCountDelta(-1) dn.UpAdjustActiveVolumeCountDelta(-1)

46
weed/topology/topology_vacuum.go
View file

@ -122,29 +122,41 @@ func (t *Topology) Vacuum(garbageThreshold float64, preallocate int64) int {
for _, vl := range c.storageType2VolumeLayout.Items() { for _, vl := range c.storageType2VolumeLayout.Items() {
if vl != nil { if vl != nil {
volumeLayout := vl.(*VolumeLayout) volumeLayout := vl.(*VolumeLayout)
for vid, locationlist := range volumeLayout.vid2location { vacuumOneVolumeLayout(volumeLayout, c, garbageThreshold, preallocate)
volumeLayout.accessLock.RLock()
isReadOnly, hasValue := volumeLayout.readonlyVolumes[vid]
volumeLayout.accessLock.RUnlock()
if hasValue && isReadOnly {
continue
}
glog.V(0).Infof("check vacuum on collection:%s volume:%d", c.Name, vid)
if batchVacuumVolumeCheck(volumeLayout, vid, locationlist, garbageThreshold) {
if batchVacuumVolumeCompact(volumeLayout, vid, locationlist, preallocate) {
batchVacuumVolumeCommit(volumeLayout, vid, locationlist)
}
}
}
} }
} }
} }
return 0 return 0
} }
func vacuumOneVolumeLayout(volumeLayout *VolumeLayout, c *Collection, garbageThreshold float64, preallocate int64) {
volumeLayout.accessLock.RLock()
tmpMap := make(map[storage.VolumeId]*VolumeLocationList)
for vid, locationlist := range volumeLayout.vid2location {
tmpMap[vid] = locationlist
}
volumeLayout.accessLock.RUnlock()
for vid, locationlist := range tmpMap {
volumeLayout.accessLock.RLock()
isReadOnly, hasValue := volumeLayout.readonlyVolumes[vid]
volumeLayout.accessLock.RUnlock()
if hasValue && isReadOnly {
continue
}
glog.V(0).Infof("check vacuum on collection:%s volume:%d", c.Name, vid)
if batchVacuumVolumeCheck(volumeLayout, vid, locationlist, garbageThreshold) {
if batchVacuumVolumeCompact(volumeLayout, vid, locationlist, preallocate) {
batchVacuumVolumeCommit(volumeLayout, vid, locationlist)
}
}
}
}
type VacuumVolumeResult struct { type VacuumVolumeResult struct {
Result bool Result bool
Error string Error string