From e89768d4298c36f89e272b495a6464df9b3d27d1 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 10 Mar 2026 12:31:04 -0700 Subject: [PATCH] use atomic.Bool for compaction flags to prevent concurrent vacuum races The isCompacting and isCommitCompacting flags were plain bools read and written from multiple goroutines without synchronization. This allowed concurrent vacuums on the same volume to pass the guard checks and run simultaneously, leading to the nil pointer crash. Using atomic.Bool with CompareAndSwap ensures only one compaction or commit can run per volume at a time. Fixes #8591 --- weed/storage/disk_location.go | 2 +- weed/storage/volume.go | 7 ++++--- weed/storage/volume_vacuum.go | 21 ++++++--------------- weed/storage/volume_write.go | 2 +- 4 files changed, 12 insertions(+), 20 deletions(-) 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 }