swarm/storage/localstore: add semaphore for updateGC goroutine

swarm/storage/localstore/localstore.go

@@ -41,6 +41,10 @@ var (
 	ErraddressLockTimeout = errors.New("update lock timeout")
 )
 
+// Limit the number of goroutines created by Getters
+// that call updateGC function. Value 0 sets no limit.
+var maxParallelUpdateGC = 1000
+
 // DB is the local store implementation and holds
 // database related objects.
 type DB struct {
@@ -68,6 +72,11 @@ type DB struct {
 	// number of elements in garbage collection index
 	gcSize int64
 
+	// a buffered channel acting as a semaphore
+	// to limit the maximal number of goroutines
+	// created by Getters to call updateGC function
+	updateGCSem chan struct{}
+
 	baseKey []byte
 
 	addressLocks sync.Map
@@ -103,6 +112,9 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
 		baseKey:                    baseKey,
 		useRetrievalCompositeIndex: o.UseRetrievalCompositeIndex,
 	}
+	if maxParallelUpdateGC > 0 {
+		db.updateGCSem = make(chan struct{}, maxParallelUpdateGC)
+	}
 
 	db.shed, err = shed.NewDB(path)
 	if err != nil {

swarm/storage/localstore/localstore_test.go

@@ -24,6 +24,7 @@ import (
 	"os"
 	"sort"
 	"strconv"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -90,6 +91,64 @@ func TestDB_useRetrievalCompositeIndex(t *testing.T) {
 	})
 }
 
+// TestDB_updateGCSem tests maxParallelUpdateGC limit.
+// This test temporary sets the limit to a low number,
+// makes updateGC function execution time longer by
+// setting a custom testHookUpdateGC function with a sleep
+// and a count current and maximal number of goroutines.
+func TestDB_updateGCSem(t *testing.T) {
+	defer func(m int) { maxParallelUpdateGC = m }(maxParallelUpdateGC)
+	maxParallelUpdateGC = 3
+
+	db, cleanupFunc := newTestDB(t, nil)
+	defer cleanupFunc()
+
+	chunk := generateRandomChunk()
+
+	err := db.NewPutter(ModePutUpload).Put(chunk)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	updateGCSleep := time.Second
+	var count int
+	var max int
+	var mu sync.Mutex
+	defer setTestHookUpdateGC(func() {
+		mu.Lock()
+		// add to the count of current goroutines
+		count++
+		if count > max {
+			// set maximal detected numbers of goroutines
+			max = count
+		}
+		mu.Unlock()
+
+		// wait for some time to ensure multiple parallel goroutines
+		time.Sleep(updateGCSleep)
+
+		mu.Lock()
+		count--
+		mu.Unlock()
+	})()
+
+	getter := db.NewGetter(ModeGetRequest)
+
+	// get more chunks then maxParallelUpdateGC
+	// in time shorter then updateGCSleep
+	for i := 0; i < 5; i++ {
+		_, err = getter.Get(chunk.Address())
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	if max != maxParallelUpdateGC {
+		t.Errorf("got max %v, want %v", max, maxParallelUpdateGC)
+	}
+
+}
+
 // BenchmarkNew measures the time that New function
 // needs to initialize and count the number of key/value
 // pairs in GC index.

swarm/storage/localstore/mode_get.go

@@ -87,7 +87,17 @@ func (db *DB) get(mode ModeGet, addr storage.Address) (out shed.IndexItem, err e
 	switch mode {
 	// update the access timestamp and gc index
 	case ModeGetRequest:
+		if db.updateGCSem != nil {
+			// wait before creating new goroutines
+			// if updateGCSem buffer id full
+			db.updateGCSem <- struct{}{}
+		}
 		go func() {
+			if db.updateGCSem != nil {
+				// free a spot in updateGCSem buffer
+				// for a new goroutine
+				defer func() { <-db.updateGCSem }()
+			}
 			err := db.updateGC(out)
 			if err != nil {
 				log.Error("localstore update gc", "err", err)