[#874] engine: Revert Check object existance concurrently

This reverts commit f526f49995.

Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>

pkg/local_object_storage/blobovnicza/exists.go

@@ -21,22 +21,10 @@ func (b *Blobovnicza) Exists(ctx context.Context, addr oid.Address) (bool, error
 		))
 	defer span.End()
 
-	select {
-	case <-ctx.Done():
-		return false, ctx.Err()
-	default:
-	}
-
 	addrKey := addressKey(addr)
 
 	err := b.boltDB.View(func(tx *bbolt.Tx) error {
 		return tx.ForEach(func(bucketName []byte, buck *bbolt.Bucket) error {
-			select {
-			case <-ctx.Done():
-				return ctx.Err()
-			default:
-			}
-
 			if isNonDataBucket(bucketName) {
 				return nil
 			}

pkg/local_object_storage/blobovnicza/get.go

@@ -51,12 +51,6 @@ func (b *Blobovnicza) Get(ctx context.Context, prm GetPrm) (GetRes, error) {
 		))
 	defer span.End()
 
-	select {
-	case <-ctx.Done():
-		return GetRes{}, ctx.Err()
-	default:
-	}
-
 	var (
 		data    []byte
 		addrKey = addressKey(prm.addr)
@@ -64,12 +58,6 @@ func (b *Blobovnicza) Get(ctx context.Context, prm GetPrm) (GetRes, error) {
 
 	if err := b.boltDB.View(func(tx *bbolt.Tx) error {
 		return tx.ForEach(func(bucketName []byte, buck *bbolt.Bucket) error {
-			select {
-			case <-ctx.Done():
-				return ctx.Err()
-			default:
-			}
-
 			if isNonDataBucket(bucketName) {
 				return nil
 			}

pkg/local_object_storage/blobstor/blobovniczatree/get.go

@@ -94,11 +94,6 @@ func (b *Blobovniczas) Get(ctx context.Context, prm common.GetPrm) (res common.G
 //
 // returns error if object could not be read from any blobovnicza of the same level.
 func (b *Blobovniczas) getObjectFromLevel(ctx context.Context, prm blobovnicza.GetPrm, blzPath string) (common.GetRes, error) {
-	select {
-	case <-ctx.Done():
-		return common.GetRes{}, ctx.Err()
-	default:
-	}
 	// open blobovnicza (cached inside)
 	shBlz := b.getBlobovnicza(blzPath)
 	blz, err := shBlz.Open()

pkg/local_object_storage/blobstor/blobovniczatree/iterate.go

@@ -230,12 +230,6 @@ func (b *Blobovniczas) iterateSortedDBPaths(ctx context.Context, addr oid.Addres
 }
 
 func (b *Blobovniczas) iterateSordedDBPathsInternal(ctx context.Context, path string, addr oid.Address, f func(string) (bool, error)) (bool, error) {
-	select {
-	case <-ctx.Done():
-		return false, ctx.Err()
-	default:
-	}
-
 	sysPath := filepath.Join(b.rootPath, path)
 	entries, err := os.ReadDir(sysPath)
 	if os.IsNotExist(err) && b.readOnly && path == "" { // non initialized tree in read only mode
@@ -258,12 +252,6 @@ func (b *Blobovniczas) iterateSordedDBPathsInternal(ctx context.Context, path st
 
 	if len(dbIdxs) > 0 {
 		for _, dbIdx := range dbIdxs {
-			select {
-			case <-ctx.Done():
-				return false, ctx.Err()
-			default:
-			}
-
 			dbPath := filepath.Join(path, u64ToHexStringExt(dbIdx))
 			stop, err := f(dbPath)
 			if err != nil {
@@ -278,12 +266,6 @@ func (b *Blobovniczas) iterateSordedDBPathsInternal(ctx context.Context, path st
 	if len(dirIdxs) > 0 {
 		hrw.SortSliceByValue(dirIdxs, addressHash(&addr, path))
 		for _, dirIdx := range dirIdxs {
-			select {
-			case <-ctx.Done():
-				return false, ctx.Err()
-			default:
-			}
-
 			dirPath := filepath.Join(path, u64ToHexString(dirIdx))
 			stop, err := b.iterateSordedDBPathsInternal(ctx, dirPath, addr, f)
 			if err != nil {

pkg/local_object_storage/blobstor/exists.go

@@ -3,7 +3,6 @@ package blobstor
 import (
 	"context"
 	"encoding/hex"
-	"errors"
 	"time"
 
 	"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
@@ -58,30 +57,27 @@ func (b *BlobStor) Exists(ctx context.Context, prm common.ExistsPrm) (common.Exi
 	// error     | found       | log the error, return true, nil
 	// error     | not found   | return the error
 	// error     | error       | log the first error, return the second
-	var storageErrors []error
+	var errors []error
 	for i := range b.storage {
 		res, err := b.storage[i].Storage.Exists(ctx, prm)
 		if err == nil && res.Exists {
 			exists = true
 			return res, nil
 		} else if err != nil {
-			if errors.Is(err, context.Canceled) {
-				return common.ExistsRes{}, err
-			}
-			storageErrors = append(storageErrors, err)
+			errors = append(errors, err)
 		}
 	}
 
-	if len(storageErrors) == 0 {
+	if len(errors) == 0 {
 		return common.ExistsRes{}, nil
 	}
 
-	for _, err := range storageErrors[:len(storageErrors)-1] {
+	for _, err := range errors[:len(errors)-1] {
 		b.log.Warn(logs.BlobstorErrorOccurredDuringObjectExistenceChecking,
 			zap.Stringer("address", prm.Address),
 			zap.String("error", err.Error()),
 			zap.String("trace_id", tracingPkg.GetTraceID(ctx)))
 	}
 
-	return common.ExistsRes{}, storageErrors[len(storageErrors)-1]
+	return common.ExistsRes{}, errors[len(errors)-1]
 }

pkg/local_object_storage/blobstor/fstree/fstree.go

@@ -285,12 +285,6 @@ func (t *FSTree) Exists(ctx context.Context, prm common.ExistsPrm) (common.Exist
 		))
 	defer span.End()
 
-	select {
-	case <-ctx.Done():
-		return common.ExistsRes{}, ctx.Err()
-	default:
-	}
-
 	p := t.treePath(prm.Address)
 
 	_, err := os.Stat(p)

pkg/local_object_storage/engine/engine_test.go

@@ -39,12 +39,6 @@ func BenchmarkExists(b *testing.B) {
 	b.Run("8 shards", func(b *testing.B) {
 		benchmarkExists(b, 8)
 	})
-	b.Run("12 shards", func(b *testing.B) {
-		benchmarkExists(b, 12)
-	})
-	b.Run("16 shards", func(b *testing.B) {
-		benchmarkExists(b, 12)
-	})
 }
 
 func benchmarkExists(b *testing.B, shardNum int) {

pkg/local_object_storage/engine/exists.go

@@ -3,70 +3,54 @@ package engine
 import (
 	"context"
 	"errors"
-	"sync/atomic"
 
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard"
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
 	apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
 	objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
 	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
-	"golang.org/x/sync/errgroup"
 )
 
 func (e *StorageEngine) exists(ctx context.Context, addr oid.Address) (bool, error) {
 	var shPrm shard.ExistsPrm
 	shPrm.SetAddress(addr)
-
-	var exists atomic.Bool
-	eg, egCtx := errgroup.WithContext(ctx)
-	egCtx, cancel := context.WithCancel(egCtx)
-	defer cancel()
+	alreadyRemoved := false
+	exists := false
 
 	e.iterateOverSortedShards(addr, func(_ int, sh hashedShard) (stop bool) {
-		select {
-		case <-egCtx.Done():
-			return true
-		default:
+		res, err := sh.Exists(ctx, shPrm)
+		if err != nil {
+			if client.IsErrObjectAlreadyRemoved(err) {
+				alreadyRemoved = true
+
+				return true
+			}
+
+			var siErr *objectSDK.SplitInfoError
+			if errors.As(err, &siErr) {
+				return true
+			}
+
+			if shard.IsErrObjectExpired(err) {
+				return true
+			}
+
+			if !client.IsErrObjectNotFound(err) {
+				e.reportShardError(sh, "could not check existence of object in shard", err)
+			}
+			return false
 		}
 
-		eg.Go(func() error {
-			res, err := sh.Exists(egCtx, shPrm)
-			if err != nil {
-				if errors.Is(err, context.Canceled) {
-					return err
-				}
+		if !exists {
+			exists = res.Exists()
+		}
 
-				if client.IsErrObjectAlreadyRemoved(err) {
-					return err
-				}
-
-				var siErr *objectSDK.SplitInfoError
-				if errors.As(err, &siErr) {
-					return err
-				}
-
-				if shard.IsErrObjectExpired(err) {
-					return err
-				}
-
-				if !client.IsErrObjectNotFound(err) {
-					e.reportShardError(sh, "could not check existence of object in shard", err)
-				}
-				return nil
-			}
-			if res.Exists() {
-				exists.Store(true)
-				cancel()
-			}
-			return nil
-		})
 		return false
 	})
 
-	err := eg.Wait()
-	if client.IsErrObjectAlreadyRemoved(err) {
+	if alreadyRemoved {
 		return false, new(apistatus.ObjectAlreadyRemoved)
 	}
 
-	return exists.Load(), nil
+	return exists, nil
 }

pkg/local_object_storage/engine/put.go

@@ -78,6 +78,8 @@ func (e *StorageEngine) put(ctx context.Context, prm PutPrm) error {
 
 	addr := object.AddressOf(prm.obj)
 
+	// In #1146 this check was parallelized, however, it became
+	// much slower on fast machines for 4 shards.
 	_, err := e.exists(ctx, addr)
 	if err != nil {
 		return err

pkg/local_object_storage/metabase/exists.go

@@ -67,12 +67,6 @@ func (db *DB) Exists(ctx context.Context, prm ExistsPrm) (res ExistsRes, err err
 		return res, ErrDegradedMode
 	}
 
-	select {
-	case <-ctx.Done():
-		return res, ctx.Err()
-	default:
-	}
-
 	currEpoch := db.epochState.CurrentEpoch()
 
 	err = db.boltDB.View(func(tx *bbolt.Tx) error {

pkg/local_object_storage/shard/exists.go

@@ -47,12 +47,6 @@ func (s *Shard) Exists(ctx context.Context, prm ExistsPrm) (ExistsRes, error) {
 		))
 	defer span.End()
 
-	select {
-	case <-ctx.Done():
-		return ExistsRes{}, ctx.Err()
-	default:
-	}
-
 	var exists bool
 	var err error