swarm/storage/localstore: persist gcSize

swarm/storage/localstore/gc.go

@@ -18,6 +18,7 @@ package localstore
 
 import (
 	"sync/atomic"
+	"time"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/swarm/shed"
@@ -39,21 +40,48 @@ var (
 	gcBatchSize int64 = 1000
 )
 
-// collectGarbage is a long running function that waits for
+// collectGarbageWorker is a long running function that waits for
 // collectGarbageTrigger channel to signal a garbage collection
 // run. GC run iterates on gcIndex and removes older items
 // form retrieval and other indexes.
-func (db *DB) collectGarbage() {
+func (db *DB) collectGarbageWorker() {
-	target := db.gcTarget()
 	for {
 		select {
 		case <-db.collectGarbageTrigger:
-			batch := new(leveldb.Batch)
+			// TODO: Add comment about done
+			collectedCount, done, err := db.collectGarbage()
+			if err != nil {
+				log.Error("localstore collect garbage", "err", err)
+			}
+			// check if another gc run is needed
+			if !done {
+				select {
+				case db.collectGarbageTrigger <- struct{}{}:
+				default:
+				}
+			}
 
-			// sets a gc trigger if batch limit is reached
+			if testHookCollectGarbage != nil {
-			var triggerNextIteration bool
+				testHookCollectGarbage(collectedCount)
-			var collectedCount int64
+			}
-			err := db.gcIndex.IterateAll(func(item shed.Item) (stop bool, err error) {
+		case <-db.close:
+			return
+		}
+	}
+}
+
+// collectGarbage removes chunks from retrieval and other
+// indexes if maximal number of chunks in database is reached.
+// This function returns the number of removed chunks. If done
+// is false, another call to this function is needed to collect
+// the rest of the garbage as the batch size limit is reached.
+// This function is called in collectGarbageWorker.
+func (db *DB) collectGarbage() (collectedCount int64, done bool, err error) {
+	batch := new(leveldb.Batch)
+	target := db.gcTarget()
+
+	done = true
+	err = db.gcIndex.IterateAll(func(item shed.Item) (stop bool, err error) {
 		gcSize := atomic.LoadInt64(&db.gcSize)
 		if gcSize-collectedCount <= target {
 			return true, nil
@@ -69,36 +97,24 @@ func (db *DB) collectGarbage() {
 		db.gcIndex.DeleteInBatch(batch, item)
 		collectedCount++
 		if collectedCount >= gcBatchSize {
-					triggerNextIteration = true
+			// bach size limit reached,
+			// another gc run is needed
+			done = false
 			return true, nil
 		}
 		return false, nil
 	})
 	if err != nil {
-				log.Error("localstore collect garbage", "err", err)
+		return 0, false, err
 	}
 
 	err = db.shed.WriteBatch(batch)
 	if err != nil {
-				log.Error("localstore collect garbage write batch", "err", err)
+		return 0, false, err
-			} else {
+	}
-				// batch is written, decrement gcSize and check if another gc run is needed
+	// batch is written, decrement gcSize
 	db.incGCSize(-collectedCount)
-				if triggerNextIteration {
+	return collectedCount, done, nil
-					select {
-					case db.collectGarbageTrigger <- struct{}{}:
-					default:
-					}
-				}
-			}
-
-			if testHookCollectGarbage != nil {
-				testHookCollectGarbage(collectedCount)
-			}
-		case <-db.close:
-			return
-		}
-	}
 }
 
 // gcTrigger retruns the absolute value for garbage collection
@@ -110,7 +126,14 @@ func (db *DB) gcTarget() (target int64) {
 // incGCSize increments gcSize by the provided number.
 // If count is negative, it will decrement gcSize.
 func (db *DB) incGCSize(count int64) {
+	if count == 0 {
+		return
+	}
 	new := atomic.AddInt64(&db.gcSize, count)
+	select {
+	case db.writeGCSizeTrigger <- struct{}{}:
+	default:
+	}
 	if new >= db.capacity {
 		select {
 		case db.collectGarbageTrigger <- struct{}{}:
@@ -119,6 +142,46 @@ func (db *DB) incGCSize(count int64) {
 	}
 }
 
+var writeGCSizeDelay = 10 * time.Second
+
+// writeGCSizeWorker calls writeGCSize function
+// on writeGCSizeTrigger receive. It implements a
+// backoff with delay of writeGCSizeDelay duration
+// to avoid very frequent database operations.
+func (db *DB) writeGCSizeWorker() {
+	for {
+		select {
+		case <-db.writeGCSizeTrigger:
+			err := db.writeGCSize()
+			if err != nil {
+				log.Error("localstore write gc size", "err", err)
+			}
+			select {
+			case <-time.After(writeGCSizeDelay):
+			case <-db.close:
+				return
+			}
+		case <-db.close:
+			return
+		}
+	}
+}
+
+// writeGCSize stores the number of items in gcIndex.
+// It removes all hashes from gcUncountedHashesIndex
+// not to include them on the next database initialization
+// when gcSize is counted.
+func (db *DB) writeGCSize() (err error) {
+	gcSize := atomic.LoadInt64(&db.gcSize)
+	err = db.storedGCSize.Put(uint64(gcSize))
+	if err != nil {
+		return err
+	}
+	return db.gcUncountedHashesIndex.IterateAll(func(item shed.Item) (stop bool, err error) {
+		return false, db.gcUncountedHashesIndex.Delete(item)
+	})
+}
+
 // testHookCollectGarbage is a hook that can provide
 // information when a garbage collection run is done
 // and how many items it removed.

swarm/storage/localstore/gc_test.go

@@ -17,6 +17,9 @@
 package localstore
 
 import (
+	"io/ioutil"
+	"math/rand"
+	"os"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -24,34 +27,34 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
-// TestDB_collectGarbage tests garbage collection runs
+// TestDB_collectGarbageWorker tests garbage collection runs
 // by uploading and syncing a number of chunks.
-func TestDB_collectGarbage(t *testing.T) {
+func TestDB_collectGarbageWorker(t *testing.T) {
 	db, cleanupFunc := newTestDB(t, &Options{
 		Capacity: 100,
 	})
 	defer cleanupFunc()
 
-	testDB_collectGarbage(t, db)
+	testDB_collectGarbageWorker(t, db)
 }
 
-// TestDB_collectGarbage_useRetrievalCompositeIndex tests
+// TestDB_collectGarbageWorker_useRetrievalCompositeIndex tests
 // garbage collection runs by uploading and syncing a number
 // of chunks using composite retrieval index.
-func TestDB_collectGarbage_useRetrievalCompositeIndex(t *testing.T) {
+func TestDB_collectGarbageWorker_useRetrievalCompositeIndex(t *testing.T) {
 	db, cleanupFunc := newTestDB(t, &Options{
 		Capacity:                   100,
 		UseRetrievalCompositeIndex: true,
 	})
 	defer cleanupFunc()
 
-	testDB_collectGarbage(t, db)
+	testDB_collectGarbageWorker(t, db)
 }
 
-// TestDB_collectGarbage_multipleBatches tests garbage
+// TestDB_collectGarbageWorker_multipleBatches tests garbage
 // collection runs by uploading and syncing a number of
 // chunks by having multiple smaller batches.
-func TestDB_collectGarbage_multipleBatches(t *testing.T) {
+func TestDB_collectGarbageWorker_multipleBatches(t *testing.T) {
 	// lower the maximal number of chunks in a single
 	// gc batch to ensure multiple batches.
 	defer func(s int64) { gcBatchSize = s }(gcBatchSize)
@@ -62,14 +65,14 @@ func TestDB_collectGarbage_multipleBatches(t *testing.T) {
 	})
 	defer cleanupFunc()
 
-	testDB_collectGarbage(t, db)
+	testDB_collectGarbageWorker(t, db)
 }
 
-// TestDB_collectGarbage_multipleBatches_useRetrievalCompositeIndex
+// TestDB_collectGarbageWorker_multipleBatches_useRetrievalCompositeIndex
 // tests garbage collection runs by uploading and syncing a number
 // of chunks using composite retrieval index and having multiple
 // smaller batches.
-func TestDB_collectGarbage_multipleBatches_useRetrievalCompositeIndex(t *testing.T) {
+func TestDB_collectGarbageWorker_multipleBatches_useRetrievalCompositeIndex(t *testing.T) {
 	// lower the maximal number of chunks in a single
 	// gc batch to ensure multiple batches.
 	defer func(s int64) { gcBatchSize = s }(gcBatchSize)
@@ -81,12 +84,12 @@ func TestDB_collectGarbage_multipleBatches_useRetrievalCompositeIndex(t *testing
 	})
 	defer cleanupFunc()
 
-	testDB_collectGarbage(t, db)
+	testDB_collectGarbageWorker(t, db)
 }
 
-// testDB_collectGarbage is a helper test function to test
+// testDB_collectGarbageWorker is a helper test function to test
 // garbage collection runs by uploading and syncing a number of chunks.
-func testDB_collectGarbage(t *testing.T, db *DB) {
+func testDB_collectGarbageWorker(t *testing.T, db *DB) {
 	uploader := db.NewPutter(ModePutUpload)
 	syncer := db.NewSetter(ModeSetSync)
 
@@ -160,34 +163,34 @@ func testDB_collectGarbage(t *testing.T, db *DB) {
 	})
 }
 
-// TestDB_collectGarbage_withRequests tests garbage collection
+// TestDB_collectGarbageWorker_withRequests tests garbage collection
 // runs by uploading, syncing and requesting a number of chunks.
-func TestDB_collectGarbage_withRequests(t *testing.T) {
+func TestDB_collectGarbageWorker_withRequests(t *testing.T) {
 	db, cleanupFunc := newTestDB(t, &Options{
 		Capacity: 100,
 	})
 	defer cleanupFunc()
 
-	testDB_collectGarbage_withRequests(t, db)
+	testDB_collectGarbageWorker_withRequests(t, db)
 }
 
-// TestDB_collectGarbage_withRequests_useRetrievalCompositeIndex
+// TestDB_collectGarbageWorker_withRequests_useRetrievalCompositeIndex
 // tests garbage collection runs by uploading, syncing and
 // requesting a number of chunks using composite retrieval index.
-func TestDB_collectGarbage_withRequests_useRetrievalCompositeIndex(t *testing.T) {
+func TestDB_collectGarbageWorker_withRequests_useRetrievalCompositeIndex(t *testing.T) {
 	db, cleanupFunc := newTestDB(t, &Options{
 		Capacity:                   100,
 		UseRetrievalCompositeIndex: true,
 	})
 	defer cleanupFunc()
 
-	testDB_collectGarbage_withRequests(t, db)
+	testDB_collectGarbageWorker_withRequests(t, db)
 }
 
-// testDB_collectGarbage_withRequests is a helper test function
+// testDB_collectGarbageWorker_withRequests is a helper test function
 // to test garbage collection runs by uploading, syncing and
 // requesting a number of chunks.
-func testDB_collectGarbage_withRequests(t *testing.T, db *DB) {
+func testDB_collectGarbageWorker_withRequests(t *testing.T, db *DB) {
 	uploader := db.NewPutter(ModePutUpload)
 	syncer := db.NewSetter(ModeSetSync)
 
@@ -290,6 +293,69 @@ func testDB_collectGarbage_withRequests(t *testing.T, db *DB) {
 	})
 }
 
+// TestDB_gcSize checks if gcSize has a correct value after
+// database is initialized with existing data.
+func TestDB_gcSize(t *testing.T) {
+	dir, err := ioutil.TempDir("", "localstore-stored-gc-size")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(dir)
+	baseKey := make([]byte, 32)
+	if _, err := rand.Read(baseKey); err != nil {
+		t.Fatal(err)
+	}
+	db, err := New(dir, baseKey, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	uploader := db.NewPutter(ModePutUpload)
+	syncer := db.NewSetter(ModeSetSync)
+
+	count := 100
+
+	for i := 0; i < count; i++ {
+		chunk := generateRandomChunk()
+
+		err := uploader.Put(chunk)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		err = syncer.Set(chunk.Address())
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	err = db.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	db, err = New(dir, baseKey, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	t.Run("gc index size", newIndexGCSizeTest(db))
+
+	t.Run("gc uncounted hashes index count", newItemsCountTest(db.gcUncountedHashesIndex, 0))
+}
+
+func testStoredGCSize(t *testing.T, db *DB, want uint64) {
+	t.Helper()
+
+	got, err := db.storedGCSize.Get()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got != want {
+		t.Errorf("got stored gc size %v, want %v", got, want)
+	}
+}
+
 // setTestHookCollectGarbage sets testHookCollectGarbage and
 // returns a function that will reset it to the
 // value before the change.

swarm/storage/localstore/localstore.go

@@ -34,10 +34,10 @@ var (
 	ErrInvalidMode = errors.New("invalid mode")
 	// ErrDBClosed is returned when database is closed.
 	ErrDBClosed = errors.New("db closed")
-	// ErraddressLockTimeout is returned when the same chunk
+	// ErrAddressLockTimeout is returned when the same chunk
 	// is updated in parallel and one of the updates
 	// takes longer then the configured timeout duration.
-	ErraddressLockTimeout = errors.New("update lock timeout")
+	ErrAddressLockTimeout = errors.New("address lock timeout")
 )
 
 var (
@@ -53,8 +53,10 @@ var (
 type DB struct {
 	shed *shed.DB
 
-	// fields
+	// schema name of loaded data
 	schemaName shed.StringField
+	// filed that stores number of intems in gc index
+	storedGCSize shed.Uint64Field
 
 	// this flag is for benchmarking two types of retrieval indexes
 	// - single retrieval composite index retrievalCompositeIndex
@@ -71,6 +73,9 @@ type DB struct {
 	pullIndex shed.Index
 	// garbage collection index
 	gcIndex shed.Index
+	// index that stores hashes that are not
+	// counted in and saved to storedGCSize
+	gcUncountedHashesIndex shed.Index
 
 	// number of elements in garbage collection index
 	gcSize int64
@@ -78,12 +83,18 @@ type DB struct {
 	// the capacity value
 	capacity int64
 
+	// triggers garbage collection event loop
 	collectGarbageTrigger chan struct{}
+	// triggers write gc size event loop
+	writeGCSizeTrigger chan struct{}
 
 	// a buffered channel acting as a semaphore
 	// to limit the maximal number of goroutines
 	// created by Getters to call updateGC function
 	updateGCSem chan struct{}
+	// a wait group to ensure all updateGC goroutines
+	// are done before closing the database
+	updateGCWG sync.WaitGroup
 
 	baseKey []byte
 
@@ -127,10 +138,12 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
 		capacity:                   o.Capacity,
 		baseKey:                    baseKey,
 		useRetrievalCompositeIndex: o.UseRetrievalCompositeIndex,
-		// this channel needs to be buffered with the size of 1
+		// channels collectGarbageTrigger and writeGCSizeTrigger
-		// to signal another garbage collection run if it
+		// need to be buffered with the size of 1
-		// is triggered during already running one
+		// to signal another event if it
+		// is triggered during already running function
 		collectGarbageTrigger: make(chan struct{}, 1),
+		writeGCSizeTrigger:    make(chan struct{}, 1),
 		close:                 make(chan struct{}),
 	}
 	if db.capacity <= 0 {
@@ -149,6 +162,11 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
 	if err != nil {
 		return nil, err
 	}
+	// Persist gc size.
+	db.storedGCSize, err = db.shed.NewUint64Field("gc-size")
+	if err != nil {
+		return nil, err
+	}
 	if db.useRetrievalCompositeIndex {
 		var (
 			encodeValueFunc func(fields shed.Item) (value []byte, err error)
@@ -345,20 +363,63 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
 	if err != nil {
 		return nil, err
 	}
-	// start garbage collection worker
+	// gc uncounted hashes index keeps hashes that are in gc index
-	go db.collectGarbage()
+	// but not counted in and saved to storedGCSize
+	db.gcUncountedHashesIndex, err = db.shed.NewIndex("Hash->nil", shed.IndexFuncs{
+		EncodeKey: func(fields shed.Item) (key []byte, err error) {
+			return fields.Address, nil
+		},
+		DecodeKey: func(key []byte) (e shed.Item, err error) {
+			e.Address = key
+			return e, nil
+		},
+		EncodeValue: func(fields shed.Item) (value []byte, err error) {
+			return nil, nil
+		},
+		DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
+			return e, nil
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+
 	// count number of elements in garbage collection index
-	gcSize, err := db.gcIndex.Count()
+	gcSize, err := db.storedGCSize.Get()
+	if err != nil {
+		return nil, err
+	}
+	// get number of uncounted hashes
+	gcUncountedSize, err := db.gcUncountedHashesIndex.Count()
+	if err != nil {
+		return nil, err
+	}
+	gcSize += uint64(gcUncountedSize)
+	// remove uncounted hashes from the index
+	err = db.gcUncountedHashesIndex.IterateAll(func(item shed.Item) (stop bool, err error) {
+		return false, db.gcUncountedHashesIndex.Delete(item)
+	})
+	if err != nil {
+		return nil, err
+	}
+	// save the total gcSize after uncounted hashes are removed
+	err = db.storedGCSize.Put(gcSize)
 	if err != nil {
 		return nil, err
 	}
 	db.incGCSize(int64(gcSize))
+
+	// start worker to write gc size
+	go db.writeGCSizeWorker()
+	// start garbage collection worker
+	go db.collectGarbageWorker()
 	return db, nil
 }
 
 // Close closes the underlying database.
 func (db *DB) Close() (err error) {
 	close(db.close)
+	db.updateGCWG.Wait()
 	return db.shed.Close()
 }
 
@@ -380,7 +441,7 @@ var (
 // using addressLocks sync.Map and returns unlock function.
 // If the address is locked this function will check it
 // in a for loop for addressLockTimeout time, after which
-// it will return ErraddressLockTimeout error.
+// it will return ErrAddressLockTimeout error.
 func (db *DB) lockAddr(addr storage.Address) (unlock func(), err error) {
 	start := time.Now()
 	lockKey := hex.EncodeToString(addr)
@@ -391,7 +452,7 @@ func (db *DB) lockAddr(addr storage.Address) (unlock func(), err error) {
 		}
 		time.Sleep(addressLockCheckDelay)
 		if time.Since(start) > addressLockTimeout {
-			return nil, ErraddressLockTimeout
+			return nil, ErrAddressLockTimeout
 		}
 	}
 	return func() { db.addressLocks.Delete(lockKey) }, nil

swarm/storage/localstore/mode_get.go

@@ -92,7 +92,9 @@ func (db *DB) get(mode ModeGet, addr storage.Address) (out shed.Item, err error)
 			// if updateGCSem buffer id full
 			db.updateGCSem <- struct{}{}
 		}
+		db.updateGCWG.Add(1)
 		go func() {
+			defer db.updateGCWG.Done()
 			if db.updateGCSem != nil {
 				// free a spot in updateGCSem buffer
 				// for a new goroutine

swarm/storage/localstore/mode_put.go

@@ -127,6 +127,7 @@ func (db *DB) put(mode ModePut, item shed.Item) (err error) {
 		}
 		// add new entry to gc index
 		db.gcIndex.PutInBatch(batch, item)
+		db.gcUncountedHashesIndex.PutInBatch(batch, item)
 		db.incGCSize(1)
 
 		if db.useRetrievalCompositeIndex {

swarm/storage/localstore/mode_set.go

@@ -122,6 +122,7 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
 		}
 		db.pullIndex.PutInBatch(batch, item)
 		db.gcIndex.PutInBatch(batch, item)
+		db.gcUncountedHashesIndex.PutInBatch(batch, item)
 		db.incGCSize(1)
 
 	case ModeSetSync:
@@ -188,6 +189,7 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
 		}
 		db.pushIndex.DeleteInBatch(batch, item)
 		db.gcIndex.PutInBatch(batch, item)
+		db.gcUncountedHashesIndex.PutInBatch(batch, item)
 		db.incGCSize(1)
 
 	case ModeSetRemove:
@@ -226,6 +228,7 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
 		}
 		db.pullIndex.DeleteInBatch(batch, item)
 		db.gcIndex.DeleteInBatch(batch, item)
+		db.gcUncountedHashesIndex.DeleteInBatch(batch, item)
 		// a check is needed for decrementing gcSize
 		// as delete is not reporting if the key/value pair
 		// is deleted or not