Simplify engine internals

pkg/local_object_storage/engine/container.go

@@ -16,40 +16,32 @@
 //
 // Returns an error if executions are blocked (see BlockExecution).
 func (e *StorageEngine) ContainerSize(cnr cid.ID) (uint64, error) {
-	var (
-		err  error
-		size uint64
-	)
 	if e.metrics != nil {
 		defer elapsed(e.metrics.AddEstimateContainerSizeDuration)()
 	}
 
-	err = e.execIfNotBlocked(func() error {
-		size, err = e.containerSize(cnr)
-		return err
-	})
+	e.blockMtx.RLock()
+	defer e.blockMtx.RUnlock()
 
-	return size, err
-}
+	if e.blockErr != nil {
+		return 0, e.blockErr
+	}
 
-func (e *StorageEngine) containerSize(cnr cid.ID) (uint64, error) {
 	var size uint64
 
-	e.iterateOverUnsortedShards(func(sh hashedShard) (stop bool) {
+	for _, sh := range e.unsortedShards() {
 		var csPrm shard.ContainerSizePrm
 		csPrm.SetContainerID(cnr)
 
 		csRes, err := sh.Shard.ContainerSize(csPrm)
 		if err != nil {
 			e.reportShardError(sh, "can't get container size", err,
 				zap.Stringer("container_id", cnr))
-			return false
+			continue
 		}
 
 		size += csRes.Size()
-
-		return false
-	})
+	}
 
 	return size, nil
 }
@@ -58,40 +50,32 @@
 //
 // Returns an error if executions are blocked (see BlockExecution).
 func (e *StorageEngine) ListContainers() ([]cid.ID, error) {
-	var (
-		res []cid.ID
-		err error
-	)
 	if e.metrics != nil {
 		defer elapsed(e.metrics.AddListContainersDuration)()
 	}
 
-	err = e.execIfNotBlocked(func() error {
-		res, err = e.listContainers()
-		return err
-	})
+	e.blockMtx.RLock()
+	defer e.blockMtx.RUnlock()
 
-	return res, err
-}
+	if e.blockErr != nil {
+		return nil, e.blockErr
+	}
 
-func (e *StorageEngine) listContainers() ([]cid.ID, error) {
 	uniqueIDs := make(map[cid.ID]struct{})
 
-	e.iterateOverUnsortedShards(func(sh hashedShard) (stop bool) {
+	for _, sh := range e.unsortedShards() {
 		res, err := sh.Shard.ListContainers(shard.ListContainersPrm{})
 		if err != nil {
 			e.reportShardError(sh, "can't get list of containers", err)
-			return false
+			continue
 		}
 
 		for _, cnr := range res.Containers() {
 			if _, ok := uniqueIDs[cnr]; !ok {
 				uniqueIDs[cnr] = struct{}{}
 			}
 		}
-
-		return false
-	})
+	}
 
 	result := make([]cid.ID, 0, len(uniqueIDs))
 	for cnr := range uniqueIDs {
@@ -103,27 +87,30 @@
 
 // DeleteContainer deletes container's objects that engine stores.
 func (e *StorageEngine) DeleteContainer(ctx context.Context, cID cid.ID) error {
-	return e.execIfNotBlocked(func() error {
-		var wg errgroup.Group
+	e.blockMtx.RLock()
+	defer e.blockMtx.RUnlock()
+
+	if e.blockErr != nil {
+		return e.blockErr
+	}
 
-		e.iterateOverUnsortedShards(func(hs hashedShard) bool {
-			wg.Go(func() error {
-				err := hs.Shard.DeleteContainer(ctx, cID)
-				if err != nil {
-					err = fmt.Errorf("container cleanup in %s shard: %w", hs.ID(), err)
-					e.log.Warn("container cleanup", zap.Error(err))
+	var wg errgroup.Group
 
-					return err
-				}
+	for _, sh := range e.unsortedShards() {
+		wg.Go(func() error {
+			err := sh.Shard.DeleteContainer(ctx, cID)
+			if err != nil {
+				err = fmt.Errorf("container cleanup in %s shard: %w", sh.ID(), err)
+				e.log.Warn("container cleanup", zap.Error(err))
 
-				return nil
-			})
+				return err
+			}
 
-			return false
+			return nil
 		})
+	}
 
-		return wg.Wait()
-	})
+	return wg.Wait()
 }
 
 func (e *StorageEngine) deleteNotFoundContainers() error {

pkg/local_object_storage/engine/control.go

@@ -99,38 +99,23 @@ func (e *StorageEngine) close(releasePools bool) error {
 	return nil
 }
 
-// executes op if execution is not blocked, otherwise returns blocking error.
-//
-// Can be called concurrently with setBlockExecErr.
-func (e *StorageEngine) execIfNotBlocked(op func() error) error {
-	e.blockExec.mtx.RLock()
-	defer e.blockExec.mtx.RUnlock()
-
-	if e.blockExec.err != nil {
-		return e.blockExec.err
-	}
-
-	return op()
-}
-
 // sets the flag of blocking execution of all data operations according to err:
 //   - err != nil, then blocks the execution. If exec wasn't blocked, calls close method
 //     (if err == errClosed => additionally releases pools and does not allow to resume executions).
 //   - otherwise, resumes execution. If exec was blocked, calls open method.
 //
 // Can be called concurrently with exec. In this case it waits for all executions to complete.
 func (e *StorageEngine) setBlockExecErr(err error) error {
-	e.blockExec.mtx.Lock()
-	defer e.blockExec.mtx.Unlock()
+	e.blockMtx.Lock()
+	defer e.blockMtx.Unlock()
 
-	prevErr := e.blockExec.err
+	prevErr := e.blockErr
 
-	wasClosed := errors.Is(prevErr, errClosed)
-	if wasClosed {
+	if errors.Is(prevErr, errClosed) {
 		return errClosed
 	}
 
-	e.blockExec.err = err
+	e.blockErr = err
 
 	if err == nil {
 		if prevErr != nil { // block -> ok

pkg/local_object_storage/engine/delete.go

@@ -15,9 +15,13 @@
 		defer elapsed(e.metrics.AddDeleteDuration)()
 	}
 
-	return e.execIfNotBlocked(func() error {
-		return e.deleteObj(addr, true)
-	})
+	e.blockMtx.RLock()
+	defer e.blockMtx.RUnlock()
+
+	if e.blockErr != nil {
+		return e.blockErr
+	}
+	return e.deleteObj(addr, true)
 }
 
 func (e *StorageEngine) deleteObj(addr oid.Address, force bool) error {

pkg/local_object_storage/engine/engine.go

@@ -30,11 +30,8 @@ type StorageEngine struct {
 	setModeCh chan setModeRequest
 	wg        sync.WaitGroup
 
-	blockExec struct {
-		mtx sync.RWMutex
-
-		err error
-	}
+	blockMtx sync.RWMutex
+	blockErr error
 }
 
 type shardWrapper struct {
@@ -137,7 +134,7 @@ func (e *StorageEngine) reportShardErrorBackground(id string, msg string, err er
 // reportShardError checks that the amount of errors doesn't exceed the configured threshold.
 // If it does, shard is set to read-only mode.
 func (e *StorageEngine) reportShardError(
-	sh hashedShard,
+	sh shardWrapper,
 	msg string,
 	err error,
 	fields ...zap.Field) {

pkg/local_object_storage/engine/evacuate.go

@@ -16,7 +16,7 @@ import (
 const defaultEvacuateBatchSize = 100
 
 type pooledShard struct {
-	hashedShard
+	shardWrapper
 	pool util.WorkerPool
 }
 
@@ -63,8 +63,8 @@ func (e *StorageEngine) Evacuate(shardIDs []*shard.ID, ignoreErrors bool, faultH
 	shards := make([]pooledShard, 0, len(e.shards))
 	for id := range e.shards {
 		shards = append(shards, pooledShard{
-			hashedShard: hashedShard(e.shards[id]),
-			pool:        e.shardPools[id],
+			shardWrapper: e.shards[id],
+			pool:         e.shardPools[id],
 		})
 	}
 	e.mtx.RUnlock()
@@ -125,7 +125,7 @@ mainLoop:
 					if _, ok := shardMap[shards[j].ID().String()]; ok {
 						continue
 					}
-					putDone, exists := e.putToShard(shards[j].hashedShard, j, shards[j].pool, addr, getRes.Object(), nil, 0)
+					putDone, exists := e.putToShard(shards[j].shardWrapper, j, shards[j].pool, addr, getRes.Object(), nil, 0)
 					if putDone || exists {
 						if putDone {
 							e.log.Debug("object is moved to another shard",

pkg/local_object_storage/engine/exists.go

@@ -12,45 +12,33 @@
 func (e *StorageEngine) exists(addr oid.Address) (bool, error) {
 	var shPrm shard.ExistsPrm
 	shPrm.SetAddress(addr)
-	alreadyRemoved := false
-	exists := false
 
-	e.iterateOverSortedShards(addr, func(_ int, sh hashedShard) (stop bool) {
+	for _, sh := range e.sortedShards(addr) {
 		res, err := sh.Exists(shPrm)
 		if err != nil {
 			if shard.IsErrRemoved(err) {
-				alreadyRemoved = true
-
-				return true
+				return false, apistatus.ObjectAlreadyRemoved{}
 			}
 
 			var siErr *objectSDK.SplitInfoError
 			if errors.As(err, &siErr) {
-				return true
+				return false, nil
 			}
 
 			if shard.IsErrObjectExpired(err) {
-				return true
+				return false, nil
 			}
 
 			if !shard.IsErrNotFound(err) {
 				e.reportShardError(sh, "could not check existence of object in shard", err)
 			}
-			return false
+			continue
 		}
 
-		if !exists {
-			exists = res.Exists()
+		if res.Exists() {
+			return true, nil
 		}
-
-		return false
-	})
-
-	if alreadyRemoved {
-		var errRemoved apistatus.ObjectAlreadyRemoved
-
-		return false, errRemoved
 	}
 
-	return exists, nil
+	return false, nil
 }