diff --git a/internal/logs/logs.go b/internal/logs/logs.go index 87e4e0b43..7aef6873e 100644 --- a/internal/logs/logs.go +++ b/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" ) diff --git a/pkg/local_object_storage/writecache/cache.go b/pkg/local_object_storage/writecache/cache.go index f2280f2f4..b298f812a 100644 --- a/pkg/local_object_storage/writecache/cache.go +++ b/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 - data []byte - obj *objectSDK.Object + addr oid.Address + size uint64 } const ( diff --git a/pkg/local_object_storage/writecache/flush.go b/pkg/local_object_storage/writecache/flush.go index 074756e32..d06896ed5 100644 --- a/pkg/local_object_storage/writecache/flush.go +++ b/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) - c.wg.Done() + defer 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) { - tick := time.NewTicker(defaultFlushInterval * 10) +func (c *cache) pushToFlushQueue(ctx context.Context) { + 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)