Change WC size estimations (#3106)

Author: roman-khimov

CHANGELOG.md

@@ -21,6 +21,7 @@ Changelog for NeoFS Node
 
 ### Changed
 - Number of cuncurrenly handled notifications from the chain was increased from 10 to 300 for IR (#3068)
+- Write-cache size estimations (#3106)
 
 ### Removed
 - Drop creating new eacl tables with public keys (#3096)

pkg/local_object_storage/blobstor/fstree/fstree.go

@@ -139,7 +139,7 @@ func addressFromString(s string) (*oid.Address, error) {
 
 // Iterate iterates over all stored objects.
 func (t *FSTree) Iterate(objHandler func(addr oid.Address, data []byte, id []byte) error, errorHandler func(addr oid.Address, err error) error) error {
-	return t.iterate(0, []string{t.RootPath}, objHandler, errorHandler, nil)
+	return t.iterate(0, []string{t.RootPath}, objHandler, errorHandler, nil, nil)
 }
 
 // IterateAddresses iterates over all objects stored in the underlying storage
@@ -152,13 +152,27 @@ func (t *FSTree) IterateAddresses(f func(addr oid.Address) error, ignoreErrors b
 		errorHandler = func(oid.Address, error) error { return nil }
 	}
 
-	return t.iterate(0, []string{t.RootPath}, nil, errorHandler, f)
+	return t.iterate(0, []string{t.RootPath}, nil, errorHandler, f, nil)
+}
+
+// IterateSizes iterates over all objects stored in the underlying storage
+// and passes their addresses and sizes into f. If f returns an error, IterateSizes
+// returns it and breaks. ignoreErrors allows to continue if internal errors
+// happen.
+func (t *FSTree) IterateSizes(f func(addr oid.Address, size uint64) error, ignoreErrors bool) error {
+	var errorHandler func(oid.Address, error) error
+	if ignoreErrors {
+		errorHandler = func(oid.Address, error) error { return nil }
+	}
+
+	return t.iterate(0, []string{t.RootPath}, nil, errorHandler, nil, f)
 }
 
 func (t *FSTree) iterate(depth uint64, curPath []string,
 	objHandler func(oid.Address, []byte, []byte) error,
 	errorHandler func(oid.Address, error) error,
-	addrHandler func(oid.Address) error) error {
+	addrHandler func(oid.Address) error,
+	sizeHandler func(oid.Address, uint64) error) error {
 	curName := strings.Join(curPath[1:], "")
 	dir := filepath.Join(curPath...)
 	des, err := os.ReadDir(dir)
@@ -177,7 +191,7 @@ func (t *FSTree) iterate(depth uint64, curPath []string,
 		curPath[l] = des[i].Name()
 
 		if !isLast && des[i].IsDir() {
-			err := t.iterate(depth+1, curPath, objHandler, errorHandler, addrHandler)
+			err := t.iterate(depth+1, curPath, objHandler, errorHandler, addrHandler, sizeHandler)
 			if err != nil {
 				// Must be error from handler in case errors are ignored.
 				// Need to report.
@@ -199,24 +213,36 @@ func (t *FSTree) iterate(depth uint64, curPath []string,
 		} else {
 			var data []byte
 			p := filepath.Join(curPath...)
-			data, err = getRawObjectBytes(addr.Object(), p)
-			if err != nil && errors.Is(err, apistatus.ObjectNotFound{}) {
-				continue
-			}
-			if err == nil {
-				data, err = t.Decompress(data)
-			}
-			if err != nil {
-				if errorHandler != nil {
-					err = errorHandler(*addr, err)
-					if err == nil {
-						continue
+			if sizeHandler != nil {
+				err = filepath.Walk(p, func(path string, info os.FileInfo, _ error) error {
+					if !info.IsDir() {
+						err = sizeHandler(*addr, uint64(info.Size()))
+						if err != nil {
+							return err
+						}
 					}
+					return nil
+				})
+			} else {
+				data, err = getRawObjectBytes(addr.Object(), p)
+				if err != nil && errors.Is(err, apistatus.ObjectNotFound{}) {
+					continue
+				}
+				if err == nil {
+					data, err = t.Decompress(data)
+				}
+				if err != nil {
+					if errorHandler != nil {
+						err = errorHandler(*addr, err)
+						if err == nil {
+							continue
+						}
+					}
+					return fmt.Errorf("read file %q: %w", p, err)
 				}
-				return fmt.Errorf("read file %q: %w", p, err)
-			}
 
-			err = objHandler(*addr, data, []byte{})
+				err = objHandler(*addr, data, []byte{})
+			}
 		}
 
 		if err != nil {
@@ -456,29 +482,6 @@ func (t *FSTree) GetRange(addr oid.Address, from uint64, length uint64) ([]byte,
 	return payload[from:to], nil
 }
 
-// NumberOfObjects walks the file tree rooted at FSTree's root
-// and returns number of stored objects.
-func (t *FSTree) NumberOfObjects() (uint64, error) {
-	var counter uint64
-
-	// it is simpler to just consider every file
-	// that is not directory as an object
-	err := filepath.WalkDir(t.RootPath,
-		func(_ string, d fs.DirEntry, _ error) error {
-			if !d.IsDir() {
-				counter++
-			}
-
-			return nil
-		},
-	)
-	if err != nil {
-		return 0, fmt.Errorf("could not walk through %q directory: %w", t.RootPath, err)
-	}
-
-	return counter, nil
-}
-
 // Type is fstree storage type used in logs and configuration.
 const Type = "fstree"
 

pkg/local_object_storage/writecache/delete.go

@@ -24,7 +24,7 @@ func (c *cache) Delete(addr oid.Address) error {
 			storagelog.StorageTypeField(wcStorageType),
 			storagelog.OpField("DELETE"),
 		)
-		c.objCounters.DecFS()
+		c.objCounters.Delete(addr)
 	}
 
 	return err

pkg/local_object_storage/writecache/migrate_test.go

@@ -0,0 +1,87 @@
+package writecache
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
+	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
+	objecttest "github.com/nspcc-dev/neofs-sdk-go/object/test"
+	"github.com/stretchr/testify/require"
+	"go.etcd.io/bbolt"
+)
+
+func TestMigrateFromBolt(t *testing.T) {
+	c, b, _ := newCache(t)
+
+	wc := c.(*cache)
+	path := filepath.Join(wc.path, dbName)
+
+	require.NoError(t, wc.Close())
+
+	t.Run("ok, no database", func(t *testing.T) {
+		require.NoError(t, wc.migrate())
+	})
+
+	db, err := bbolt.Open(path, os.ModePerm, &bbolt.Options{
+		NoFreelistSync: true,
+		ReadOnly:       false,
+		Timeout:        time.Second,
+	})
+	require.NoError(t, err)
+
+	t.Run("couldn't open database", func(t *testing.T) {
+		err := wc.migrate()
+		require.Error(t, err)
+		fmt.Println(err)
+	})
+
+	require.NoError(t, db.Close())
+	t.Run("no default bucket", func(t *testing.T) {
+		err := wc.migrate()
+		require.Error(t, err)
+		fmt.Println(err)
+	})
+
+	db, err = bbolt.Open(path, os.ModePerm, &bbolt.Options{
+		NoFreelistSync: true,
+		ReadOnly:       false,
+		Timeout:        time.Second,
+	})
+	require.NoError(t, err)
+
+	obj := objecttest.Object()
+
+	var addr oid.Address
+	addr.SetObject(obj.GetID())
+	addr.SetContainer(obj.GetContainerID())
+
+	require.NoError(t, db.Batch(func(tx *bbolt.Tx) error {
+		b, err := tx.CreateBucketIfNotExists(defaultBucket)
+		require.NoError(t, err)
+
+		require.NoError(t, b.Put([]byte(addr.String()), obj.Marshal()))
+
+		return nil
+	}))
+
+	t.Run("migrate object", func(t *testing.T) {
+		require.NoError(t, db.Close())
+		require.NoError(t, wc.migrate())
+
+		_, err := wc.Get(addr)
+		require.Error(t, err, apistatus.ObjectNotFound{})
+
+		bObject, err := b.Get(addr, []byte{})
+		require.NoError(t, err)
+		require.Equal(t, obj.GetID(), bObject.GetID())
+		require.Equal(t, obj.GetContainerID(), bObject.GetContainerID())
+		require.Equal(t, obj.Marshal(), bObject.Marshal())
+
+		_, err = os.Stat(path)
+		require.Error(t, err, os.ErrNotExist)
+	})
+}

pkg/local_object_storage/writecache/options.go

@@ -38,7 +38,7 @@ type options struct {
 	// maxCacheSize is the maximum total size of all objects saved in cache.
 	// 1 GiB by default.
 	maxCacheSize uint64
-	// objCounters contains atomic counters for the number of objects stored in cache.
+	// objCounters contains object list along with sizes and overall size of cache.
 	objCounters counters
 	// noSync is true iff FSTree allows unsynchronized writes.
 	noSync bool

pkg/local_object_storage/writecache/put.go

@@ -39,8 +39,9 @@ func (c *cache) Put(addr oid.Address, obj *objectSDK.Object, data []byte) error
 
 // put writes object to FSTree and pushes it to the flush workers queue.
 func (c *cache) put(addr oid.Address, obj objectInfo) error {
-	cacheSz := c.estimateCacheSize()
-	if c.maxCacheSize < c.incSizeFS(cacheSz) {
+	cacheSz := c.objCounters.Size()
+	objSz := uint64(len(obj.data))
+	if c.maxCacheSize < cacheSz+objSz {
 		return ErrOutOfSpace
 	}
 
@@ -54,7 +55,7 @@ func (c *cache) put(addr oid.Address, obj objectInfo) error {
 		c.compressFlags[obj.addr] = struct{}{}
 		c.mtx.Unlock()
 	}
-	c.objCounters.IncFS()
+	c.objCounters.Add(addr, objSz)
 	storagelog.Write(c.log,
 		storagelog.AddressField(obj.addr),
 		storagelog.StorageTypeField(wcStorageType),

pkg/local_object_storage/writecache/state.go

@@ -2,41 +2,49 @@ package writecache
 
 import (
 	"fmt"
-	"math"
-	"sync/atomic"
+	"sync"
+
+	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
 )
 
-func (c *cache) estimateCacheSize() uint64 {
-	return c.objCounters.FS() * c.maxObjectSize
+type counters struct {
+	mu     sync.Mutex
+	objMap map[oid.Address]uint64
+	size   uint64
 }
 
-func (c *cache) incSizeFS(sz uint64) uint64 {
-	return sz + c.maxObjectSize
-}
+func (x *counters) Add(addr oid.Address, size uint64) {
+	x.mu.Lock()
+	defer x.mu.Unlock()
 
-type counters struct {
-	cFS atomic.Uint64
+	x.size += size
+	x.objMap[addr] = size
 }
 
-func (x *counters) IncFS() {
-	x.cFS.Add(1)
-}
+func (x *counters) Delete(addr oid.Address) {
+	x.mu.Lock()
+	defer x.mu.Unlock()
 
-func (x *counters) DecFS() {
-	x.cFS.Add(math.MaxUint32)
+	x.size -= x.objMap[addr]
+	delete(x.objMap, addr)
 }
 
-func (x *counters) FS() uint64 {
-	return x.cFS.Load()
+func (x *counters) Size() uint64 {
+	x.mu.Lock()
+	defer x.mu.Unlock()
+	return x.size
 }
 
 func (c *cache) initCounters() error {
-	inFS, err := c.fsTree.NumberOfObjects()
+	var sizeHandler = func(addr oid.Address, size uint64) error {
+		c.objCounters.Add(addr, size)
+		return nil
+	}
+
+	err := c.fsTree.IterateSizes(sizeHandler, false)
 	if err != nil {
 		return fmt.Errorf("could not read write-cache FS counter: %w", err)
 	}
 
-	c.objCounters.cFS.Store(inFS)
-
 	return nil
 }

pkg/local_object_storage/writecache/storage.go

@@ -97,7 +97,7 @@ func (c *cache) deleteFromDisk(keys []string) []string {
 				storagelog.StorageTypeField(wcStorageType),
 				storagelog.OpField("DELETE"),
 			)
-			c.objCounters.DecFS()
+			c.objCounters.Delete(addr)
 		}
 	}
 

pkg/local_object_storage/writecache/writecache.go

@@ -101,6 +101,9 @@ func New(opts ...Option) Cache {
 			log:           zap.NewNop(),
 			maxObjectSize: defaultMaxObjectSize,
 			maxCacheSize:  defaultMaxCacheSize,
+			objCounters: counters{
+				objMap: make(map[oid.Address]uint64),
+			},
 		},
 	}