atomic

weed/pb/filer_pb/filer_client_bfs.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -23,7 +24,7 @@ func TraverseBfs(ctx context.Context, filerClient FilerClient, parentPath util.F
 	pending.Add(1)
 	queue.Enqueue(parentPath)
 
-	var once sync.Once
+	var hasError int32
 	var firstErr error
 
 	enqueue := func(p util.FullPath) bool {
@@ -64,10 +65,10 @@ func TraverseBfs(ctx context.Context, filerClient FilerClient, parentPath util.F
 				if ctx.Err() == nil {
 					processErr := processOneDirectory(ctx, filerClient, dir, enqueue, fn)
 					if processErr != nil {
-						once.Do(func() {
+						if atomic.CompareAndSwapInt32(&hasError, 0, 1) {
 							firstErr = processErr
 							cancel()
-						})
+						}
 					}
 				}
 				pending.Done()

weed/shell/command_fs_meta_save.go

@@ -155,7 +155,6 @@ func doTraverseBfsAndSaving(filerClient filer_pb.FilerClient, writer io.Writer,
 	}()
 
 	var dirCount, fileCount uint64
-	var once sync.Once
 	var firstErr error
 	var hasErr atomic.Bool
 
@@ -176,10 +175,8 @@ func doTraverseBfsAndSaving(filerClient filer_pb.FilerClient, writer io.Writer,
 			return firstErr
 		}
 		if genErr := genFn(protoMessage, outputChan); genErr != nil {
-			once.Do(func() {
-				firstErr = genErr
-				hasErr.Store(true)
-			})
+			firstErr = genErr
+			hasErr.Store(true)
 			return genErr
 		} else {
 			if e.IsDirectory {
@@ -208,11 +205,10 @@ func doTraverseBfsAndSaving(filerClient filer_pb.FilerClient, writer io.Writer,
 			return firstErr
 		}
 		if genErr := genFn(protoMessage, outputChan); genErr != nil {
-			once.Do(func() {
+			if hasErr.CompareAndSwap(false, true) {
 				firstErr = genErr
-				hasErr.Store(true)
 				cancel()
-			})
+			}
 			return genErr
 		}