Parallelize WC flush process (#3179)

Author: cthulhu-rider

CHANGELOG.md

@@ -40,6 +40,7 @@ Changelog for NeoFS Node
 - `ObjectService.GetRange(Hash)` ops now handle zero ranges as full payload (#3071)
 - Add some GAS to system fee of notary transactions (#3176)
 - IR now applies sane limits to containers' storage policies recently added to the protocol (#3075)
+- Make flushing objects in write-cache in parallel (#3179)
 
 ### Removed
 - Drop creating new eacl tables with public keys (#3096)

pkg/local_object_storage/writecache/flush.go

@@ -18,31 +18,73 @@ import (
 const (
 	// defaultFlushInterval is default time interval between successive flushes.
 	defaultFlushInterval = 10 * time.Second
+	// defaultWorkerCount is a default number of workers that flush objects.
+	defaultWorkerCount = 20
 )
 
 // runFlushLoop starts background workers which periodically flush objects to the blobstor.
 func (c *cache) runFlushLoop() {
+	for i := range c.workersCount {
+		c.wg.Add(1)
+		go c.flushWorker(i)
+	}
+
 	c.wg.Add(1)
-	go func() {
-		defer c.wg.Done()
-		tick := time.NewTicker(defaultFlushInterval)
-		for {
-			select {
-			case <-tick.C:
-				c.modeMtx.RLock()
-				if c.readOnly() {
-					c.modeMtx.RUnlock()
-					break
-				}
+	go c.flushScheduler()
+}
 
-				_ = c.flush(true)
+func (c *cache) flushScheduler() {
+	defer c.wg.Done()
+	ticker := time.NewTicker(defaultFlushInterval)
+	defer ticker.Stop()
 
+	for {
+		select {
+		case <-ticker.C:
+			c.modeMtx.RLock()
+			if c.readOnly() {
 				c.modeMtx.RUnlock()
-			case <-c.closeCh:
+				continue
+			}
+			c.modeMtx.RUnlock()
+			err := c.fsTree.IterateAddresses(func(addr oid.Address) error {
+				select {
+				case c.flushCh <- addr:
+					return nil
+				case <-c.closeCh:
+					return errors.New("closed during iteration")
+				}
+			}, true)
+			if err != nil {
+				c.log.Warn("iteration failed", zap.Error(err))
+			}
+		case <-c.closeCh:
+			return
+		}
+	}
+}
+
+func (c *cache) flushWorker(id int) {
+	defer c.wg.Done()
+	for {
+		select {
+		case addr, ok := <-c.flushCh:
+			if !ok {
 				return
 			}
+			c.modeMtx.RLock()
+			if c.readOnly() {
+				c.modeMtx.RUnlock()
+				continue
+			}
+			if err := c.flushSingle(addr, true); err != nil {
+				c.log.Warn("flush failed", zap.Int("worker", id), zap.Error(err))
+			}
+			c.modeMtx.RUnlock()
+		case <-c.closeCh:
+			return
 		}
-	}()
+	}
 }
 
 func (c *cache) reportFlushError(msg string, addr string, err error) {
@@ -55,56 +97,52 @@ func (c *cache) reportFlushError(msg string, addr string, err error) {
 	}
 }
 
-func (c *cache) flush(ignoreErrors bool) error {
-	var addrHandler = func(addr oid.Address) error {
-		sAddr := addr.EncodeToString()
-
-		data, err := c.fsTree.GetBytes(addr)
-		if err != nil {
-			if errors.As(err, new(apistatus.ObjectNotFound)) {
-				// an object can be removed b/w iterating over it
-				// and reading its payload; not an error
-				return nil
-			}
+func (c *cache) flushSingle(addr oid.Address, ignoreErrors bool) error {
+	sAddr := addr.EncodeToString()
 
-			c.reportFlushError("can't read a file", sAddr, err)
-			if ignoreErrors {
-				return nil
-			}
-			return err
+	data, err := c.fsTree.GetBytes(addr)
+	if err != nil {
+		if errors.As(err, new(apistatus.ObjectNotFound)) {
+			// an object can be removed b/w iterating over it
+			// and reading its payload; not an error
+			return nil
 		}
 
-		var obj object.Object
-		err = obj.Unmarshal(data)
-		if err != nil {
-			c.reportFlushError("can't unmarshal an object", sAddr, err)
-			if ignoreErrors {
-				return nil
-			}
-			return err
+		c.reportFlushError("can't read a file", sAddr, err)
+		if ignoreErrors {
+			return nil
 		}
+		return err
+	}
 
-		err = c.flushObject(&obj, data)
-		if err != nil {
-			return err
+	var obj object.Object
+	err = obj.Unmarshal(data)
+	if err != nil {
+		c.reportFlushError("can't unmarshal an object", sAddr, err)
+		if ignoreErrors {
+			return nil
 		}
+		return err
+	}
 
-		err = c.fsTree.Delete(addr)
-		if err != nil && !errors.As(err, new(apistatus.ObjectNotFound)) {
-			c.log.Error("can't remove object from write-cache", zap.Error(err))
-		} else if err == nil {
-			storagelog.Write(c.log,
-				storagelog.AddressField(addr),
-				storagelog.StorageTypeField(wcStorageType),
-				storagelog.OpField("DELETE"),
-			)
-			c.objCounters.Delete(addr)
-		}
+	err = c.flushObject(&obj, data)
+	if err != nil {
+		return err
+	}
 
-		return nil
+	err = c.fsTree.Delete(addr)
+	if err != nil && !errors.As(err, new(apistatus.ObjectNotFound)) {
+		c.log.Error("can't remove object from write-cache", zap.Error(err))
+	} else if err == nil {
+		storagelog.Write(c.log,
+			storagelog.AddressField(addr),
+			storagelog.StorageTypeField(wcStorageType),
+			storagelog.OpField("DELETE"),
+		)
+		c.objCounters.Delete(addr)
 	}
 
-	return c.fsTree.IterateAddresses(addrHandler, ignoreErrors)
+	return nil
 }
 
 // flushObject is used to write object directly to the main storage.
@@ -150,3 +188,9 @@ func (c *cache) Flush(ignoreErrors bool) error {
 
 	return c.flush(ignoreErrors)
 }
+
+func (c *cache) flush(ignoreErrors bool) error {
+	return c.fsTree.IterateAddresses(func(addr oid.Address) error {
+		return c.flushSingle(addr, ignoreErrors)
+	}, ignoreErrors)
+}

pkg/local_object_storage/writecache/flush_test.go

@@ -1,10 +1,12 @@
 package writecache
 
 import (
+	"fmt"
 	"os"
 	"path/filepath"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	objectCore "github.com/nspcc-dev/neofs-node/pkg/core/object"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor"
@@ -204,6 +206,79 @@ func TestFlush(t *testing.T) {
 	})
 }
 
+func TestFlushPerformance(t *testing.T) {
+	objectCounts := []int{100, 1000}
+	workerCounts := []int{1, 4, 16}
+
+	for _, objCount := range objectCounts {
+		for _, workerCount := range workerCounts {
+			t.Run(fmt.Sprintf("objects=%d_workers=%d", objCount, workerCount), func(t *testing.T) {
+				t.Parallel()
+				wc, bs, mb := newCache(t, WithFlushWorkersCount(workerCount))
+				defer wc.Close()
+
+				objects := make([]objectPair, objCount)
+				for i := range objects {
+					objects[i] = putObject(t, wc, 1+(i%2)*1024)
+				}
+				for _, obj := range objects {
+					_, err := wc.Get(obj.addr)
+					require.NoError(t, err)
+				}
+				require.NoError(t, wc.Close())
+
+				require.NoError(t, bs.SetMode(mode.ReadWrite))
+				require.NoError(t, mb.SetMode(mode.ReadWrite))
+
+				require.NoError(t, wc.Open(false))
+				require.NoError(t, wc.Init())
+				start := time.Now()
+				waitForFlush(t, wc, objects)
+				duration := time.Since(start)
+
+				for i := range objects {
+					id, err := mb.StorageID(objects[i].addr)
+					require.NoError(t, err)
+
+					res, err := bs.Get(objects[i].addr, id)
+					require.NoError(t, err)
+					require.Equal(t, objects[i].obj, res)
+				}
+				require.Equal(t, uint64(0), wc.(*cache).objCounters.Size())
+				for _, obj := range objects {
+					_, err := wc.Get(obj.addr)
+					require.Error(t, err)
+				}
+
+				t.Logf("Flush took %v for %d objects with %d workers", duration-defaultFlushInterval, objCount, workerCount)
+			})
+		}
+	}
+}
+
+func waitForFlush(t *testing.T, wc Cache, objects []objectPair) {
+	timeout := time.After(60 * time.Second)
+	ticker := time.NewTicker(100 * time.Millisecond)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			cachedCount := 0
+			for _, obj := range objects {
+				if _, err := wc.Get(obj.addr); err == nil {
+					cachedCount++
+				}
+			}
+			if cachedCount == 0 {
+				return
+			}
+		case <-timeout:
+			t.Fatalf("Flush did not complete within 60 seconds, %d objects still cached", len(objects))
+		}
+	}
+}
+
 func putObject(t *testing.T, c Cache, size int) objectPair {
 	obj, data := newObject(t, size)
 

pkg/local_object_storage/writecache/options.go

@@ -38,6 +38,8 @@ type options struct {
 	// maxCacheSize is the maximum total size of all objects saved in cache.
 	// 1 GiB by default.
 	maxCacheSize uint64
+	// workersCount is the number of workers flushing objects in parallel.
+	workersCount int
 	// objCounters contains object list along with sizes and overall size of cache.
 	objCounters counters
 	// noSync is true iff FSTree allows unsynchronized writes.
@@ -83,6 +85,15 @@ func WithMaxObjectSize(sz uint64) Option {
 	}
 }
 
+// WithFlushWorkersCount sets number of workers to flushing objects in parallel.
+func WithFlushWorkersCount(c int) Option {
+	return func(o *options) {
+		if c > 0 {
+			o.workersCount = c
+		}
+	}
+}
+
 // WithMaxCacheSize sets maximum write-cache size in bytes.
 func WithMaxCacheSize(sz uint64) Option {
 	return func(o *options) {

pkg/local_object_storage/writecache/state.go

@@ -37,6 +37,9 @@ func (x *counters) Size() uint64 {
 
 func (c *cache) initCounters() error {
 	var sizeHandler = func(addr oid.Address, size uint64) error {
+		if _, ok := c.objCounters.objMap[addr]; ok {
+			return nil
+		}
 		c.objCounters.Add(addr, size)
 		return nil
 	}

pkg/local_object_storage/writecache/writecache.go

@@ -63,6 +63,8 @@ type cache struct {
 	// whether object should be compressed.
 	compressFlags map[string]struct{}
 
+	// flushCh is a channel with objects to flush.
+	flushCh chan oid.Address
 	// closeCh is close channel.
 	closeCh chan struct{}
 	// wg is a wait group for flush workers.
@@ -92,7 +94,8 @@ var (
 // New creates new writecache instance.
 func New(opts ...Option) Cache {
 	c := &cache{
-		mode: mode.ReadWrite,
+		flushCh: make(chan oid.Address),
+		mode:    mode.ReadWrite,
 
 		compressFlags: make(map[string]struct{}),
 		options: options{
@@ -102,6 +105,7 @@ func New(opts ...Option) Cache {
 			objCounters: counters{
 				objMap: make(map[oid.Address]uint64),
 			},
+			workersCount: defaultWorkerCount,
 		},
 	}