Remove ants pool from engine put

pkg/local_object_storage/engine/control.go

@@ -81,15 +81,15 @@ func (e *StorageEngine) close(releasePools bool) error {
 	defer e.mtx.RUnlock()
 
 	for id, sh := range e.shards {
-		if releasePools {
-			sh.pool.Release()
-		}
 		if err := sh.Close(); err != nil {
 			e.log.Debug("could not close shard",
 				zap.String("id", id),
 				zap.Error(err),
 			)
 		}
+		if releasePools {
+			close(sh.putCh)
+		}
 	}
 
 	return nil
@@ -188,7 +188,17 @@ func (rCfg *ReConfiguration) AddShard(id string, opts []shard.Option) {
 // Reload reloads StorageEngine's configuration in runtime.
 func (e *StorageEngine) Reload(rcfg ReConfiguration) error {
 	e.mtx.Lock()
-	e.shardPoolSize = rcfg.shardPoolSize
+	if rcfg.shardPoolSize != e.shardPoolSize {
+		e.shardPoolSize = rcfg.shardPoolSize
+		for id, sh := range e.shards {
+			close(sh.putCh)
+			sh.putCh = make(chan putTask)
+			for range e.shardPoolSize {
+				go sh.shardPutThread()
+			}
+			e.shards[id] = sh
+		}
+	}
 	e.mtx.Unlock()
 
 	type reloadInfo struct {
@@ -197,9 +207,6 @@ func (e *StorageEngine) Reload(rcfg ReConfiguration) error {
 	}
 
 	e.mtx.RLock()
-	for _, sh := range e.shards {
-		sh.pool.Tune(int(e.shardPoolSize))
-	}
 
 	var shardsToRemove []string // shards IDs
 	var shardsToAdd []string    // shard config identifiers (blobstor paths concatenation)

pkg/local_object_storage/engine/ec.go

@@ -47,7 +47,7 @@ func (e *StorageEngine) GetECPart(cnr cid.ID, parent oid.ID, pi iec.PartInfo) (o
 		return object.Object{}, nil, e.blockErr
 	}
 
-	s := e.sortShardsFn(e, oid.NewAddress(cnr, parent))
+	s := e.sortShardsFn(e, parent)
 
 	var partID oid.ID
 loop:
@@ -147,7 +147,7 @@ func (e *StorageEngine) GetECPartRange(cnr cid.ID, parent oid.ID, pi iec.PartInf
 		return 0, nil, e.blockErr
 	}
 
-	s := e.sortShardsFn(e, oid.NewAddress(cnr, parent))
+	s := e.sortShardsFn(e, parent)
 
 	var partID oid.ID
 loop:
@@ -216,7 +216,7 @@ func (e *StorageEngine) HeadECPart(cnr cid.ID, parent oid.ID, pi iec.PartInfo) (
 		return object.Object{}, e.blockErr
 	}
 
-	s := e.sortShardsFn(e, oid.NewAddress(cnr, parent))
+	s := e.sortShardsFn(e, parent)
 
 	var partID oid.ID
 loop:

pkg/local_object_storage/engine/engine.go

@@ -12,7 +12,6 @@ import (
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard/mode"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/util/logicerr"
-	"github.com/nspcc-dev/neofs-node/pkg/util"
 	cid "github.com/nspcc-dev/neofs-sdk-go/container/id"
 	"github.com/nspcc-dev/neofs-sdk-go/object"
 	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
@@ -37,7 +36,7 @@ type StorageEngine struct {
 	blockMtx sync.RWMutex
 	blockErr error
 
-	sortShardsFn func(*StorageEngine, oid.Address) []shardWrapper
+	sortShardsFn func(*StorageEngine, oid.ID) []shardWrapper
 }
 
 // interface of [shard.Shard] used by [StorageEngine] for overriding in tests.
@@ -51,11 +50,19 @@ type shardInterface interface {
 	HeadECPart(cid.ID, oid.ID, iec.PartInfo) (object.Object, error)
 }
 
+type putTask struct {
+	addr   oid.Address
+	obj    *object.Object
+	objBin []byte
+	retCh  chan error
+}
+
 type shardWrapper struct {
 	errorCount *atomic.Uint32
-	pool       util.WorkerPool
 	*shard.Shard
 	shardIface shardInterface // TODO: make Shard a shardInterface
+	putCh      chan putTask
+	engine     *StorageEngine
 }
 
 type setModeRequest struct {

pkg/local_object_storage/engine/engine_test.go

@@ -26,7 +26,6 @@ import (
 	oidtest "github.com/nspcc-dev/neofs-sdk-go/object/id/test"
 	usertest "github.com/nspcc-dev/neofs-sdk-go/user/test"
 	"github.com/nspcc-dev/tzhash/tz"
-	"github.com/panjf2000/ants/v2"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 )
@@ -84,18 +83,11 @@ func testNewEngineWithShards(shards ...*shard.Shard) *StorageEngine {
 	engine := New()
 
 	for _, s := range shards {
-		pool, err := ants.NewPool(10, ants.WithNonblocking(true))
+		err := engine.addShard(s)
 		if err != nil {
 			panic(err)
 		}
-
-		engine.shards[s.ID().String()] = shardWrapper{
-			errorCount: new(atomic.Uint32),
-			pool:       pool,
-			Shard:      s,
-		}
 	}
-
 	return engine
 }
 
@@ -194,7 +186,7 @@ func newEngineWithFixedShardOrder(ss []shardInterface) *StorageEngine {
 		}
 	}
 
-	e.sortShardsFn = func(*StorageEngine, oid.Address) []shardWrapper {
+	e.sortShardsFn = func(*StorageEngine, oid.ID) []shardWrapper {
 		return ws
 	}
 

pkg/local_object_storage/engine/exists.go

@@ -14,7 +14,7 @@ func (e *StorageEngine) exists(addr oid.Address) (bool, error) {
 		defer elapsed(e.metrics.AddExistsDuration)()
 	}
 
-	for _, sh := range e.sortedShards(addr) {
+	for _, sh := range e.sortedShards(addr.Object()) {
 		exists, err := sh.Exists(addr, false)
 		if err != nil {
 			if shard.IsErrRemoved(err) {

pkg/local_object_storage/engine/gc_test.go

@@ -386,7 +386,7 @@ func TestSplitObjectExpirationWithoutLink(t *testing.T) {
 
 	require.Eventually(t, func() bool {
 		// Check this way because Get returns ErrObjectNotFound for expired error
-		for _, sh := range e.sortedShards(parentAddr) {
+		for _, sh := range e.sortedShards(parentID) {
 			_, err = sh.Get(parentAddr, false)
 			if errors.Is(err, statusSDK.ErrObjectNotFound) {
 				return true
@@ -496,7 +496,7 @@ func TestSplitObjectExpirationWithLinkNotFound(t *testing.T) {
 
 	require.Eventually(t, func() bool {
 		// Check this way because Get returns ErrObjectNotFound for expired error
-		for _, sh := range e.sortedShards(parentAddr) {
+		for _, sh := range e.sortedShards(parentID) {
 			_, err = sh.Get(parentAddr, false)
 			if errors.Is(err, statusSDK.ErrObjectNotFound) {
 				return true

pkg/local_object_storage/engine/get.go

@@ -62,7 +62,7 @@ func (e *StorageEngine) get(addr oid.Address, shardFunc func(s *shard.Shard, ign
 		metaError     error
 	)
 
-	for _, sh := range e.sortedShards(addr) {
+	for _, sh := range e.sortedShards(addr.Object()) {
 		noMeta := sh.GetMode().NoMetabase()
 		hasDegraded = hasDegraded || noMeta
 
@@ -118,7 +118,7 @@ func (e *StorageEngine) get(addr oid.Address, shardFunc func(s *shard.Shard, ign
 	// If the object is not found but is present in metabase,
 	// try to fetch it from blobstor directly. If it is found in any
 	// blobstor, increase the error counter for the shard which contains the meta.
-	for _, sh := range e.sortedShards(addr) {
+	for _, sh := range e.sortedShards(addr.Object()) {
 		if sh.GetMode().NoMetabase() {
 			// Already visited.
 			continue

pkg/local_object_storage/engine/head.go

@@ -36,7 +36,7 @@ func (e *StorageEngine) Head(addr oid.Address, raw bool) (*object.Object, error)
 
 	var splitInfo *object.SplitInfo
 
-	for _, sh := range e.sortedShards(addr) {
+	for _, sh := range e.sortedShards(addr.Object()) {
 		res, err := sh.Head(addr, raw)
 		if err != nil {
 			var siErr *object.SplitInfoError

pkg/local_object_storage/engine/inhume.go

@@ -49,7 +49,7 @@ func (e *StorageEngine) processAddrDelete(addr oid.Address, deleteFunc func(*sha
 		err      error
 		root     bool
 		siNoLink *object.SplitInfo
-		shards   = e.sortedShards(addr)
+		shards   = e.sortedShards(addr.Object())
 	)
 
 	// see if the object is root

pkg/local_object_storage/engine/inhume_test.go

@@ -133,7 +133,7 @@ func TestStorageEngine_Inhume(t *testing.T) {
 
 		var wrongShardID string
 
-		for i, sh := range e.sortedShards(addr) {
+		for i, sh := range e.sortedShards(addr.Object()) {
 			if i != 0 {
 				wrongShardID = sh.ID().String()
 			}