mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/chrislusf/seaweedfs synced 2024-07-03 07:36:45 +02:00
Code Issues Projects Releases Wiki Activity

volume: avoid reprocessing the same volume

Browse source 
fix https://github.com/chrislusf/seaweedfs/issues/1682
This commit is contained in:
Chris Lu 2020-12-17 13:03:39 -08:00
parent 986cbdf7d9
commit e2076201d7
1 changed files with 23 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
weed/storage/disk_location.go
View file

@ -53,7 +53,7 @@ func NewDiskLocation(dir string, maxVolumeCount int, minFreeSpacePercent float32
}
func volumeIdFromFileName(filename string) (needle.VolumeId, string, error) {
if strings.HasSuffix(filename, ".idx") || strings.HasSuffix(filename, ".vif") {
if isValidVolume(filename) {
base := filename[:len(filename)-4]
collection, volumeId, err := parseCollectionVolumeId(base)
return volumeId, collection, err
@ -71,15 +71,26 @@ func parseCollectionVolumeId(base string) (collection string, vid needle.VolumeI
return collection, vol, err
}
func isValidVolume(basename string) bool {
return strings.HasSuffix(basename, ".idx") || strings.HasSuffix(basename, ".vif")
}
func getValidVolumeName(basename string) string {
if isValidVolume(basename) {
return basename[:len(basename)-4]
}
return ""
}
func (l *DiskLocation) loadExistingVolume(fileInfo os.FileInfo, needleMapKind NeedleMapType) bool {
basename := fileInfo.Name()
if fileInfo.IsDir() {
return false
}
if !strings.HasSuffix(basename, ".idx") && !strings.HasSuffix(basename, ".vif") {
volumeName := getValidVolumeName(basename)
if volumeName == "" {
return false
}
volumeName := basename[:len(basename)-4]
// check for incomplete volume
noteFile := l.Directory + "/" + volumeName + ".note"
@ -126,9 +137,17 @@ func (l *DiskLocation) concurrentLoadingVolumes(needleMapKind NeedleMapType, con
task_queue := make(chan os.FileInfo, 10*concurrency)
go func() {
foundVolumeNames := make(map[string]bool)
if fileInfos, err := ioutil.ReadDir(l.Directory); err == nil {
for _, fi := range fileInfos {
task_queue <- fi
volumeName := getValidVolumeName(fi.Name())
if volumeName == "" {
continue
}
if _, found := foundVolumeNames[volumeName]; !found {
foundVolumeNames[volumeName] = true
task_queue <- fi
}
}
}
close(task_queue)