diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go index 9ff7d5e5b..ca640b96c 100644 --- a/weed/storage/disk_location.go +++ b/weed/storage/disk_location.go @@ -383,7 +383,7 @@ func (l *DiskLocation) UnloadVolume(vid needle.VolumeId) error { func (l *DiskLocation) unmountVolumeByCollection(collectionName string) map[needle.VolumeId]*Volume { deltaVols := make(map[needle.VolumeId]*Volume, 0) for k, v := range l.volumes { - if v.Collection == collectionName && !v.isCompacting && !v.isCommitCompacting { + if v.Collection == collectionName && !v.isCompacting.Load() && !v.isCommitCompacting.Load() { deltaVols[k] = v } } diff --git a/weed/storage/volume.go b/weed/storage/volume.go index dd8ecbdce..9c12f8e25 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -5,6 +5,7 @@ import ( "path" "strconv" "sync" + "sync/atomic" "time" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" @@ -45,8 +46,8 @@ type Volume struct { lastCompactRevision uint16 ldbTimeout int64 - isCompacting bool - isCommitCompacting bool + isCompacting atomic.Bool + isCommitCompacting atomic.Bool volumeInfoRWLock sync.RWMutex volumeInfo *volume_server_pb.VolumeInfo @@ -231,7 +232,7 @@ func (v *Volume) Close() { } func (v *Volume) doClose() { - for v.isCommitCompacting { + for v.isCommitCompacting.Load() { time.Sleep(521 * time.Millisecond) glog.Warningf("Volume Close wait for compaction %d", v.Id) } diff --git a/weed/storage/volume_vacuum.go b/weed/storage/volume_vacuum.go index 5dc5d8233..aed10d9f5 100644 --- a/weed/storage/volume_vacuum.go +++ b/weed/storage/volume_vacuum.go @@ -64,14 +64,11 @@ func (v *Volume) CompactByVolumeData(opts *CompactOptions) error { //v.accessLock.Lock() //defer v.accessLock.Unlock() //glog.V(3).Infof("Got Compaction lock...") - if v.isCompacting || v.isCommitCompacting { + if v.isCommitCompacting.Load() || !v.isCompacting.CompareAndSwap(false, true) { glog.V(0).Infof("Volume %d is already compacting...", v.Id) return nil } - v.isCompacting = true - defer func() { - v.isCompacting = false - }() + defer v.isCompacting.Store(false) v.lastCompactIndexOffset = v.IndexFileSize() v.lastCompactRevision = v.SuperBlock.CompactionRevision @@ -102,14 +99,11 @@ func (v *Volume) CompactByIndex(opts *CompactOptions) error { } glog.V(3).Infof("Compact2 volume %d ...", v.Id) - if v.isCompacting || v.isCommitCompacting { + if v.isCommitCompacting.Load() || !v.isCompacting.CompareAndSwap(false, true) { glog.V(0).Infof("Volume %d is already compacting2 ...", v.Id) return nil } - v.isCompacting = true - defer func() { - v.isCompacting = false - }() + defer v.isCompacting.Store(false) v.lastCompactIndexOffset = v.IndexFileSize() v.lastCompactRevision = v.SuperBlock.CompactionRevision @@ -142,14 +136,11 @@ func (v *Volume) CommitCompact() error { } glog.V(0).Infof("Committing volume %d vacuuming...", v.Id) - if v.isCommitCompacting { + if !v.isCommitCompacting.CompareAndSwap(false, true) { glog.V(0).Infof("Volume %d is already commit compacting ...", v.Id) return nil } - v.isCommitCompacting = true - defer func() { - v.isCommitCompacting = false - }() + defer v.isCommitCompacting.Store(false) v.dataFileAccessLock.Lock() defer v.dataFileAccessLock.Unlock() diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go index 8cb00bc15..4885f45a2 100644 --- a/weed/storage/volume_write.go +++ b/weed/storage/volume_write.go @@ -70,7 +70,7 @@ func (v *Volume) Destroy(onlyEmpty bool) (err error) { return } } - if v.isCompacting || v.isCommitCompacting { + if v.isCompacting.Load() || v.isCommitCompacting.Load() { err = fmt.Errorf("volume %d is compacting", v.Id) return }