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

volume: load volume can optionally be skipped, if ec volume exists

Browse source 
fix https://github.com/chrislusf/seaweedfs/issues/2489
This commit is contained in:
chrislu 2021-12-05 02:28:52 -08:00
parent 73e3e77904
commit 488afa5002
1 changed files with 11 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
weed/storage/disk_location.go
View file

@ -84,7 +84,7 @@ func getValidVolumeName(basename string) string {
return "" return ""
} }
func (l *DiskLocation) loadExistingVolume(dirEntry os.DirEntry, needleMapKind NeedleMapKind) bool { func (l *DiskLocation) loadExistingVolume(dirEntry os.DirEntry, needleMapKind NeedleMapKind, skipIfEcVolumesExists bool) bool {
basename := dirEntry.Name() basename := dirEntry.Name()
if dirEntry.IsDir() { if dirEntry.IsDir() {
return false return false
@ -94,9 +94,11 @@ func (l *DiskLocation) loadExistingVolume(dirEntry os.DirEntry, needleMapKind Ne
return false return false
} }
// skip ec volumes // skip if ec volumes exists
if util.FileExists(l.Directory + "/" + volumeName + ".ecx") { if skipIfEcVolumesExists {
return false if util.FileExists(l.Directory + "/" + volumeName + ".ecx") {
return false
}
} }
// check for incomplete volume // check for incomplete volume
@ -166,7 +168,7 @@ func (l *DiskLocation) concurrentLoadingVolumes(needleMapKind NeedleMapKind, con
go func() { go func() {
defer wg.Done() defer wg.Done()
for fi := range task_queue { for fi := range task_queue {
_ = l.loadExistingVolume(fi, needleMapKind) _ = l.loadExistingVolume(fi, needleMapKind, true)
} }
}() }()
} }
@ -246,7 +248,7 @@ func (l *DiskLocation) deleteVolumeById(vid needle.VolumeId) (found bool, e erro
func (l *DiskLocation) LoadVolume(vid needle.VolumeId, needleMapKind NeedleMapKind) bool { func (l *DiskLocation) LoadVolume(vid needle.VolumeId, needleMapKind NeedleMapKind) bool {
if fileInfo, found := l.LocateVolume(vid); found { if fileInfo, found := l.LocateVolume(vid); found {
return l.loadExistingVolume(fileInfo, needleMapKind) return l.loadExistingVolume(fileInfo, needleMapKind, false)
} }
return false return false
} }
@ -332,9 +334,12 @@ func (l *DiskLocation) Close() {
} }
func (l *DiskLocation) LocateVolume(vid needle.VolumeId) (os.DirEntry, bool) { func (l *DiskLocation) LocateVolume(vid needle.VolumeId) (os.DirEntry, bool) {
println("LocateVolume", vid, "on", l.Directory)
if dirEntries, err := os.ReadDir(l.Directory); err == nil { if dirEntries, err := os.ReadDir(l.Directory); err == nil {
for _, entry := range dirEntries { for _, entry := range dirEntries {
println("checking", entry.Name(), "...")
volId, _, err := volumeIdFromFileName(entry.Name()) volId, _, err := volumeIdFromFileName(entry.Name())
println("volId", volId, "err", err)
if vid == volId && err == nil { if vid == volId && err == nil {
return entry, true return entry, true
} }