[#1367] writecache: Flush from FSTree concurrently

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

internal/logs/logs.go

@@ -544,4 +544,5 @@ const (
 	FailedToSealWritecacheAsync                                          = "failed to seal writecache async"
 	WritecacheShrinkSkippedNotEmpty                                      = "writecache shrink skipped: not empty"
 	BlobovniczatreeFailedToRemoveRebuildTempFile                         = "failed to remove rebuild temp file"
+	WritecacheCantGetObject                                              = "can't get an object from fstree"
 )

pkg/local_object_storage/writecache/cache.go

@@ -10,7 +10,7 @@ import (
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/metaerr"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard/mode"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
-	objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
+	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
 	"go.etcd.io/bbolt"
 	"go.uber.org/zap"
 )
@@ -37,9 +37,8 @@ type cache struct {
 const wcStorageType = "write-cache"
 
 type objectInfo struct {
-	addr string
+	addr oid.Address
-	data []byte
+	size uint64
-	obj  *objectSDK.Object
 }
 
 const (

pkg/local_object_storage/writecache/flush.go

@@ -13,10 +13,12 @@ import (
 	objectCore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
+	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/fstree"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/metaerr"
 	meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard/mode"
 	"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
+	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"
 	"go.etcd.io/bbolt"
@@ -29,7 +31,7 @@ const (
 	// defaultFlushWorkersCount is number of workers for putting objects in main storage.
 	defaultFlushWorkersCount = 20
 	// defaultFlushInterval is default time interval between successive flushes.
-	defaultFlushInterval = time.Second
+	defaultFlushInterval = 10 * time.Second
 )
 
 var errIterationCompleted = errors.New("iteration completed")
@@ -41,23 +43,41 @@ func (c *cache) runFlushLoop(ctx context.Context) {
 	}
 	c.wg.Add(1)
 	go func() {
-		c.workerFlushBig(ctx)
+		defer c.wg.Done()
-		c.wg.Done()
+		c.pushToFlushQueue(ctx)
 	}()
+
+	for range c.workersCount {
+		c.wg.Add(1)
+		go c.workerFlush(ctx)
+	}
 }
 
-func (c *cache) workerFlushBig(ctx context.Context) {
+func (c *cache) pushToFlushQueue(ctx context.Context) {
-	tick := time.NewTicker(defaultFlushInterval * 10)
+	tick := time.NewTicker(defaultFlushInterval)
 	for {
 		select {
 		case <-tick.C:
 			c.modeMtx.RLock()
 			if c.readOnly() || c.noMetabase() {
 				c.modeMtx.RUnlock()
-				break
+				continue
 			}
 
-			_ = c.flushFSTree(ctx, true)
+			err := c.fsTree.IterateInfo(ctx, func(oi fstree.ObjectInfo) error {
+				select {
+				case c.flushCh <- objectInfo{
+					addr: oi.Address,
+					size: oi.DataSize,
+				}:
+					return nil
+				case <-ctx.Done():
+					return ctx.Err()
+				}
+			})
+			if err != nil {
+				c.log.Warn(logs.BlobstorErrorOccurredDuringTheIteration, zap.Error(err))
+			}
 
 			c.modeMtx.RUnlock()
 		case <-ctx.Done():
@@ -66,6 +86,37 @@ func (c *cache) workerFlushBig(ctx context.Context) {
 	}
 }
 
+func (c *cache) workerFlush(ctx context.Context) {
+	defer c.wg.Done()
+
+	var objInfo objectInfo
+	for {
+		select {
+		case objInfo = <-c.flushCh:
+		case <-ctx.Done():
+			return
+		}
+
+		res, err := c.fsTree.Get(ctx, common.GetPrm{
+			Address: objInfo.addr,
+		})
+		if err != nil {
+			if !errors.As(err, new(*apistatus.ObjectNotFound)) {
+				c.reportFlushError(logs.WritecacheCantGetObject, objInfo.addr.EncodeToString(), metaerr.Wrap(err))
+			}
+			continue
+		}
+
+		err = c.flushObject(ctx, res.Object, res.RawData, StorageTypeFSTree)
+		if err != nil {
+			// Error is handled in flushObject.
+			continue
+		}
+
+		c.deleteFromDisk(ctx, objInfo.addr)
+	}
+}
+
 func (c *cache) reportFlushError(msg string, addr string, err error) {
 	if c.reportError != nil {
 		c.reportError(msg, err)