From c0e1cbe9ed087cfc9d5da4840e4dd73ec73b3e26 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 8 May 2026 23:33:33 -0700 Subject: [PATCH] review: fix race, accounting, recovery, non-EIO streak break MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addressing PR #9382 review: - Data race on lastIoError: guard lastIoError + lastIoErrorCount with a RWMutex and expose them through note/clear/get helpers so the heartbeat reader sees a consistent snapshot. Verified with -race. - Collection-size accounting: when a volume is quarantined for sustained EIO, skip the entire per-volume bookkeeping (`continue`) instead of flipping shouldDeleteVolume — the old branch subtracted a size that was never added, dragging the collection gauge to zero / negative. - Recoverability: MarkVolumeWritable now also calls clearIoError so an operator can rejoin a quarantined replica. The next failed op re-arms the streak if the disk is still bad. - Non-EIO streak break: a non-EIO error (e.g. ENOSPC) now resets the consecutive-EIO counter, so a sequence EIO,EIO,ENOSPC,EIO is treated as a streak of one — the counter only tracks consecutive EIOs. Reads already call checkReadWriteError (volume_read.go), so successful reads also clear the streak — no change needed there. --- weed/storage/store.go | 43 ++++++----- weed/storage/volume.go | 40 ++++++++-- weed/storage/volume_io_error_test.go | 110 +++++++++++++++++++++------ weed/storage/volume_write.go | 12 +-- 4 files changed, 153 insertions(+), 52 deletions(-) diff --git a/weed/storage/store.go b/weed/storage/store.go index e967e8246..4fe1accac 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -415,34 +415,35 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { if maxFileKey < curMaxFileKey { maxFileKey = curMaxFileKey } - shouldDeleteVolume := false - if v.lastIoError != nil && v.lastIoErrorCount.Load() >= IoErrorTolerance { + if ioErr, ioCount := v.getIoErrorState(); ioErr != nil && ioCount >= IoErrorTolerance { // Sustained EIO: stop announcing this replica so the master // re-replicates from healthy peers, and mark it read-only so // further writes fail fast instead of producing more EIOs. // Never physically delete the data — the disk may be // transiently bad and this could be the last good copy. - shouldDeleteVolume = true glog.Warningf("volume %d has %d consecutive IO errors, marking read-only and unreporting from master: %v", - v.Id, v.lastIoErrorCount.Load(), v.lastIoError) - if !v.noWriteOrDelete { - v.noWriteLock.Lock() - v.noWriteOrDelete = true - v.noWriteLock.Unlock() - } + v.Id, ioCount, ioErr) + v.noWriteLock.Lock() + v.noWriteOrDelete = true + v.noWriteLock.Unlock() + // Skip per-volume size and read-only bookkeeping: a + // quarantined replica should not be summed into the + // collection's reported total nor counted in the + // read-only stats. Recovery via MarkVolumeWritable + // resets the error state so it can rejoin. + continue + } + + shouldDeleteVolume := false + if !v.expired(volumeMessage.Size, s.GetVolumeSizeLimit()) { + volumeMessages = append(volumeMessages, volumeMessage) } else { - if !v.expired(volumeMessage.Size, s.GetVolumeSizeLimit()) { - volumeMessages = append(volumeMessages, volumeMessage) + if v.expiredLongEnough(MAX_TTL_VOLUME_REMOVAL_DELAY) { + deleteVids = append(deleteVids, v.Id) + shouldDeleteVolume = true } else { - if v.expiredLongEnough(MAX_TTL_VOLUME_REMOVAL_DELAY) { - if !shouldDeleteVolume { - deleteVids = append(deleteVids, v.Id) - shouldDeleteVolume = true - } - } else { - glog.V(0).Infof("volume %d is expired", v.Id) - } + glog.V(0).Infof("volume %d is expired", v.Id) } } @@ -681,6 +682,10 @@ func (s *Store) MarkVolumeWritable(i needle.VolumeId) error { v.noWriteOrDelete = false v.PersistReadOnly(false) v.noWriteLock.Unlock() + // Clear any sustained-EIO state so the next CollectHeartbeat can + // announce the volume again. If the disk is still bad, the next + // failed op will re-arm the streak. + v.clearIoError() return nil } diff --git a/weed/storage/volume.go b/weed/storage/volume.go index b89e7d8b4..c33a806a2 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -54,12 +54,42 @@ type Volume struct { diskId uint32 // ID of this volume's disk in Store.Locations array // lastIoError is the most recent EIO from a read/write/delete; cleared - // on the next successful op. lastIoErrorCount tracks consecutive EIOs - // so CollectHeartbeat can require a sustained failure before unmounting - // the replica — protects against a transient hardware/network blip - // hitting multiple replicas at once and stranding the only good copy. + // on the next successful or non-EIO op. lastIoErrorCount tracks + // consecutive EIOs so CollectHeartbeat can require a sustained failure + // before unmounting the replica — protects against a transient + // hardware/network blip hitting multiple replicas at once and + // stranding the only good copy. Both fields are guarded together so + // the heartbeat reader sees a consistent (err, count) snapshot. lastIoError error - lastIoErrorCount atomic.Int32 + lastIoErrorCount int32 + lastIoErrorLock sync.RWMutex +} + +// noteIoError records an EIO and increments the consecutive-error +// counter. Caller has already verified errors.Is(err, syscall.EIO). +func (v *Volume) noteIoError(err error) { + v.lastIoErrorLock.Lock() + defer v.lastIoErrorLock.Unlock() + v.lastIoError = err + v.lastIoErrorCount++ +} + +// clearIoError resets the EIO streak. Called on any successful op or on +// a non-EIO error (which still breaks the EIO streak — only sustained +// EIOs are diagnostic of a failing volume). +func (v *Volume) clearIoError() { + v.lastIoErrorLock.Lock() + defer v.lastIoErrorLock.Unlock() + v.lastIoError = nil + v.lastIoErrorCount = 0 +} + +// getIoErrorState returns the latest EIO and its consecutive-error count +// as a single atomic snapshot. +func (v *Volume) getIoErrorState() (error, int32) { + v.lastIoErrorLock.RLock() + defer v.lastIoErrorLock.RUnlock() + return v.lastIoError, v.lastIoErrorCount } func NewVolume(dirname string, dirIdx string, collection string, id needle.VolumeId, needleMapKind NeedleMapKind, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, ver needle.Version, memoryMapMaxSizeMb uint32, ldbTimeout int64) (v *Volume, e error) { diff --git a/weed/storage/volume_io_error_test.go b/weed/storage/volume_io_error_test.go index f7725e3c0..647ecdf27 100644 --- a/weed/storage/volume_io_error_test.go +++ b/weed/storage/volume_io_error_test.go @@ -3,6 +3,7 @@ package storage import ( "errors" "fmt" + "sync" "syscall" "testing" ) @@ -10,33 +11,54 @@ import ( func TestCheckReadWriteErrorTracksConsecutiveEIO(t *testing.T) { v := &Volume{} - // non-EIO errors must not advance the counter. - v.checkReadWriteError(errors.New("some other error")) - if got := v.lastIoErrorCount.Load(); got != 0 { - t.Fatalf("non-EIO error advanced counter: got %d", got) - } - if v.lastIoError != nil { - t.Fatalf("non-EIO error set lastIoError: %v", v.lastIoError) - } - // each EIO bumps the counter. for i := int32(1); i <= 5; i++ { v.checkReadWriteError(fmt.Errorf("disk failed: %w", syscall.EIO)) - if got := v.lastIoErrorCount.Load(); got != i { - t.Fatalf("after %d EIO(s): counter = %d, want %d", i, got, i) - } - if v.lastIoError == nil { - t.Fatalf("after %d EIO(s): lastIoError is nil", i) + _, count := v.getIoErrorState() + if count != i { + t.Fatalf("after %d EIO(s): counter = %d, want %d", i, count, i) } } // a single success resets both fields. v.checkReadWriteError(nil) - if got := v.lastIoErrorCount.Load(); got != 0 { - t.Fatalf("success did not reset counter: got %d", got) + if err, count := v.getIoErrorState(); err != nil || count != 0 { + t.Fatalf("success did not reset state: err=%v count=%d", err, count) } - if v.lastIoError != nil { - t.Fatalf("success did not clear lastIoError: %v", v.lastIoError) +} + +func TestCheckReadWriteErrorNonEIOResetsStreak(t *testing.T) { + v := &Volume{} + + // build up a 2-EIO streak. + v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) + v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) + if _, count := v.getIoErrorState(); count != 2 { + t.Fatalf("expected count=2 after two EIOs, got %d", count) + } + + // a non-EIO error breaks the streak — only sustained EIOs are + // diagnostic of a failing disk. + v.checkReadWriteError(fmt.Errorf("other: %w", syscall.ENOSPC)) + if err, count := v.getIoErrorState(); err != nil || count != 0 { + t.Fatalf("non-EIO did not reset streak: err=%v count=%d", err, count) + } + + // a fresh EIO starts the streak from 1, not 3. + v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) + if _, count := v.getIoErrorState(); count != 1 { + t.Fatalf("EIO after non-EIO did not restart streak: count=%d, want 1", count) + } +} + +func TestCheckReadWriteErrorIgnoresPlainError(t *testing.T) { + v := &Volume{} + + // non-EIO error with no prior streak should be a no-op (count + // stays 0, no spurious lastIoError). + v.checkReadWriteError(errors.New("some other error")) + if err, count := v.getIoErrorState(); err != nil || count != 0 { + t.Fatalf("non-EIO with no prior streak set state: err=%v count=%d", err, count) } } @@ -47,15 +69,59 @@ func TestIoErrorToleranceGate(t *testing.T) { for i := 0; i < IoErrorTolerance-1; i++ { v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) } - if v.lastIoErrorCount.Load() >= IoErrorTolerance { + if _, count := v.getIoErrorState(); count >= IoErrorTolerance { t.Fatalf("counter %d already crossed tolerance %d after %d errors", - v.lastIoErrorCount.Load(), IoErrorTolerance, IoErrorTolerance-1) + count, IoErrorTolerance, IoErrorTolerance-1) } // one more crosses the threshold. v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) - if got := v.lastIoErrorCount.Load(); got < IoErrorTolerance { + if _, count := v.getIoErrorState(); count < IoErrorTolerance { t.Fatalf("counter %d below tolerance %d after %d errors", - got, IoErrorTolerance, IoErrorTolerance) + count, IoErrorTolerance, IoErrorTolerance) } } + +func TestIoErrorStateIsRaceFree(t *testing.T) { + // Drives both writers (checkReadWriteError) and a reader + // (getIoErrorState) concurrently; relies on `go test -race` to + // detect any unprotected access on lastIoError / lastIoErrorCount. + v := &Volume{} + + var wg sync.WaitGroup + stop := make(chan struct{}) + + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) + } + } + }() + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + v.checkReadWriteError(nil) + } + } + }() + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 1000; i++ { + v.getIoErrorState() + } + close(stop) + }() + wg.Wait() +} diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go index 8823889ba..8d5916786 100644 --- a/weed/storage/volume_write.go +++ b/weed/storage/volume_write.go @@ -26,16 +26,16 @@ const IoErrorTolerance = 3 func (v *Volume) checkReadWriteError(err error) { if err == nil { - if v.lastIoError != nil { - v.lastIoError = nil - } - v.lastIoErrorCount.Store(0) + v.clearIoError() return } if errors.Is(err, syscall.EIO) { - v.lastIoError = err - v.lastIoErrorCount.Add(1) + v.noteIoError(err) + return } + // non-EIO error breaks the EIO streak — only sustained EIOs should + // be treated as a failing volume. + v.clearIoError() } // isFileUnchanged checks whether this needle to write is same as last one.