swarm/storage/localstore: implement basic garbage collection

swarm/shed/index.go

@@ -262,3 +262,18 @@ func (f Index) IterateFrom(start IndexItem, fn IndexIterFunc) (err error) {
 	}
 	return it.Error()
 }
+
+// Count returns the number of items in index.
+func (f Index) Count() (count int, err error) {
+	it := f.db.NewIterator()
+	defer it.Release()
+
+	for ok := it.Seek(f.prefix); ok; ok = it.Next() {
+		key := it.Key()
+		if key[0] != f.prefix[0] {
+			break
+		}
+		count++
+	}
+	return count, it.Error()
+}

swarm/shed/index_test.go

@@ -407,6 +407,101 @@ func TestIndex_iterate(t *testing.T) {
 	})
 }
 
+// TestIndex_Count tests if Index.Count returns the correct
+// number of items.
+func TestIndex_Count(t *testing.T) {
+	db, cleanupFunc := newTestDB(t)
+	defer cleanupFunc()
+
+	index, err := db.NewIndex("retrieval", retrievalIndexFuncs)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	items := []IndexItem{
+		{
+			Address: []byte("iterate-hash-01"),
+			Data:    []byte("data80"),
+		},
+		{
+			Address: []byte("iterate-hash-03"),
+			Data:    []byte("data22"),
+		},
+		{
+			Address: []byte("iterate-hash-05"),
+			Data:    []byte("data41"),
+		},
+		{
+			Address: []byte("iterate-hash-02"),
+			Data:    []byte("data84"),
+		},
+		{
+			Address: []byte("iterate-hash-06"),
+			Data:    []byte("data1"),
+		},
+	}
+	batch := new(leveldb.Batch)
+	for _, i := range items {
+		index.PutInBatch(batch, i)
+	}
+	err = db.WriteBatch(batch)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	got, err := index.Count()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	want := len(items)
+	if got != want {
+		t.Errorf("got %v items count, want %v", got, want)
+	}
+
+	// update the index with another item
+
+	item04 := IndexItem{
+		Address: []byte("iterate-hash-04"),
+		Data:    []byte("data0"),
+	}
+	err = index.Put(item04)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	got, err = index.Count()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	want = len(items) + 1
+	if got != want {
+		t.Errorf("got %v items count, want %v", got, want)
+	}
+
+	// delete some items
+
+	deleteCount := 3
+
+	for _, item := range items[:deleteCount] {
+		err := index.Delete(item)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	got, err = index.Count()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	want = len(items) + 1 - deleteCount
+	if got != want {
+		t.Errorf("got %v items count, want %v", got, want)
+	}
+}
+
 // checkIndexItem is a test helper function that compares if two Index items are the same.
 func checkIndexItem(t *testing.T, got, want IndexItem) {
 	t.Helper()

swarm/storage/localstore/gc.go

@@ -0,0 +1,91 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package localstore
+
+import (
+	"sync/atomic"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/swarm/shed"
+)
+
+// gcTargetRatio defines the target number of items
+// in garbage collection index that will not be removed
+// on garbage collection. The target number of items
+// is calculated by gcTarget function. This value must be
+// in range (0,1]. For example, with 0.9 value,
+// garbage collection will leave 90% of defined capacity
+// in database after its run. This prevents frequent
+// garbage collection runt.
+var gcTargetRatio = 0.9
+
+// collectGarbage 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() {
+	target := db.gcTarget()
+	for {
+		select {
+		case <-db.collectGarbageTrigger:
+			var collectedCount int64
+			err := db.gcIndex.IterateAll(func(item shed.IndexItem) (stop bool, err error) {
+				gcSize := atomic.LoadInt64(&db.gcSize)
+				if gcSize <= target {
+					return true, nil
+				}
+				err = db.set(ModeSetRemove, item.Address)
+				if err != nil {
+					return false, err
+				}
+				collectedCount++
+				return false, nil
+			})
+			if err != nil {
+				log.Error("localstore collect garbage", "err", err)
+			}
+			if testHookCollectGarbage != nil {
+				testHookCollectGarbage(collectedCount)
+			}
+		case <-db.close:
+			return
+		}
+	}
+}
+
+// gcTrigger retruns the absolute value for garbage collection
+// target value, calculated from db.capacity and gcTargetRatio.
+func (db *DB) gcTarget() (target int64) {
+	return int64(float64(db.capacity) * gcTargetRatio)
+}
+
+// incGCSize increments gcSize by the provided number.
+// If count is negative, it will decrement gcSize.
+func (db *DB) incGCSize(count int64) {
+	new := atomic.AddInt64(&db.gcSize, count)
+	if new >= db.capacity {
+		select {
+		case db.collectGarbageTrigger <- struct{}{}:
+		default:
+		}
+	}
+}
+
+// testHookCollectGarbage is a hook that can provide
+// information when a garbage collection run is done
+// and how many items it removed.
+var testHookCollectGarbage func(collectedCount int64)

swarm/storage/localstore/gc_test.go

@@ -0,0 +1,275 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package localstore
+
+import (
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+// TestDB_collectGarbage tests garbage collection runs
+// by uploading and syncing a number of chunks.
+func TestDB_collectGarbage(t *testing.T) {
+	db, cleanupFunc := newTestDB(t, &Options{
+		Capacity: 100,
+	})
+	defer cleanupFunc()
+
+	uploader := db.NewPutter(ModePutUpload)
+	syncer := db.NewSetter(ModeSetSync)
+
+	chunkCount := 150
+
+	testHookCollectGarbageChan := make(chan int64)
+	defer setTestHookCollectGarbage(func(collectedCount int64) {
+		testHookCollectGarbageChan <- collectedCount
+	})()
+
+	addrs := make([]storage.Address, 0)
+
+	// upload random chunks
+	for i := 0; i < chunkCount; i++ {
+		chunk := generateRandomChunk()
+
+		err := uploader.Put(chunk)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		err = syncer.Set(chunk.Address())
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		addrs = append(addrs, chunk.Address())
+	}
+
+	gcTarget := db.gcTarget()
+
+	var totalCollectedCount int64
+	for {
+		select {
+		case c := <-testHookCollectGarbageChan:
+			totalCollectedCount += c
+		case <-time.After(10 * time.Second):
+			t.Error("collect garbage timeout")
+		}
+		gcSize := atomic.LoadInt64(&db.gcSize)
+		if gcSize == gcTarget {
+			break
+		}
+	}
+
+	wantTotalCollectedCount := int64(chunkCount) - gcTarget
+	if totalCollectedCount != wantTotalCollectedCount {
+		t.Errorf("total collected chunks %v, want %v", totalCollectedCount, wantTotalCollectedCount)
+	}
+
+	t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, int(gcTarget)))
+
+	t.Run("gc size", newIndexGCSizeTest(db))
+
+	// the first synced chunk should be removed
+	t.Run("get the first synced chunk", func(t *testing.T) {
+		_, err := db.NewGetter(ModeGetRequest).Get(addrs[0])
+		if err != storage.ErrChunkNotFound {
+			t.Errorf("got error %v, want %v", err, storage.ErrChunkNotFound)
+		}
+	})
+
+	// last synced chunk should not be removed
+	t.Run("get most recent synced chunk", func(t *testing.T) {
+		_, err := db.NewGetter(ModeGetRequest).Get(addrs[len(addrs)-1])
+		if err != nil {
+			t.Fatal(err)
+		}
+	})
+}
+
+// TestDB_collectGarbage_withRequests tests garbage collection
+// runs by uploading, syncing and requesting a number of chunks.
+func TestDB_collectGarbage_withRequests(t *testing.T) {
+	db, cleanupFunc := newTestDB(t, &Options{
+		Capacity: 100,
+	})
+	defer cleanupFunc()
+
+	uploader := db.NewPutter(ModePutUpload)
+	syncer := db.NewSetter(ModeSetSync)
+
+	testHookCollectGarbageChan := make(chan int64)
+	defer setTestHookCollectGarbage(func(collectedCount int64) {
+		testHookCollectGarbageChan <- collectedCount
+	})()
+
+	addrs := make([]storage.Address, 0)
+
+	// upload random chunks just up to the capacity
+	for i := 0; i < int(db.capacity)-1; i++ {
+		chunk := generateRandomChunk()
+
+		err := uploader.Put(chunk)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		err = syncer.Set(chunk.Address())
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		addrs = append(addrs, chunk.Address())
+	}
+
+	// request the latest synced chunk
+	// to prioritize it in the gc index
+	// not to be collected
+	_, err := db.NewGetter(ModeGetRequest).Get(addrs[0])
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// upload and sync another chunk to trigger
+	// garbage collection
+	chunk := generateRandomChunk()
+	err = uploader.Put(chunk)
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = syncer.Set(chunk.Address())
+	if err != nil {
+		t.Fatal(err)
+	}
+	addrs = append(addrs, chunk.Address())
+
+	// wait for garbage collection
+
+	gcTarget := db.gcTarget()
+
+	var totalCollectedCount int64
+	for {
+		select {
+		case c := <-testHookCollectGarbageChan:
+			totalCollectedCount += c
+		case <-time.After(10 * time.Second):
+			t.Error("collect garbage timeout")
+		}
+		gcSize := atomic.LoadInt64(&db.gcSize)
+		if gcSize == gcTarget {
+			break
+		}
+	}
+
+	wantTotalCollectedCount := int64(len(addrs)) - gcTarget
+	if totalCollectedCount != wantTotalCollectedCount {
+		t.Errorf("total collected chunks %v, want %v", totalCollectedCount, wantTotalCollectedCount)
+	}
+
+	t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, int(gcTarget)))
+
+	t.Run("gc size", newIndexGCSizeTest(db))
+
+	// requested chunk should not be removed
+	t.Run("get requested chunk", func(t *testing.T) {
+		_, err := db.NewGetter(ModeGetRequest).Get(addrs[0])
+		if err != nil {
+			t.Fatal(err)
+		}
+	})
+
+	// the second synced chunk should be removed
+	t.Run("get gc-ed chunk", func(t *testing.T) {
+		_, err := db.NewGetter(ModeGetRequest).Get(addrs[1])
+		if err != storage.ErrChunkNotFound {
+			t.Errorf("got error %v, want %v", err, storage.ErrChunkNotFound)
+		}
+	})
+
+	// last synced chunk should not be removed
+	t.Run("get most recent synced chunk", func(t *testing.T) {
+		_, err := db.NewGetter(ModeGetRequest).Get(addrs[len(addrs)-1])
+		if err != nil {
+			t.Fatal(err)
+		}
+	})
+}
+
+// setTestHookCollectGarbage sets testHookCollectGarbage and
+// returns a function that will reset it to the
+// value before the change.
+func setTestHookCollectGarbage(h func(collectedCount int64)) (reset func()) {
+	current := testHookCollectGarbage
+	reset = func() { testHookCollectGarbage = current }
+	testHookCollectGarbage = h
+	return reset
+}
+
+// TestSetTestHookCollectGarbage tests if setTestHookCollectGarbage changes
+// testHookCollectGarbage function correctly and if its reset function
+// resets the original function.
+func TestSetTestHookCollectGarbage(t *testing.T) {
+	// Set the current function after the test finishes.
+	defer func(h func(collectedCount int64)) { testHookCollectGarbage = h }(testHookCollectGarbage)
+
+	// expected value for the unchanged function
+	original := 1
+	// expected value for the changed function
+	changed := 2
+
+	// this variable will be set with two different functions
+	var got int
+
+	// define the original (unchanged) functions
+	testHookCollectGarbage = func(_ int64) {
+		got = original
+	}
+
+	// set got variable
+	testHookCollectGarbage(0)
+
+	// test if got variable is set correctly
+	if got != original {
+		t.Errorf("got hook value %v, want %v", got, original)
+	}
+
+	// set the new function
+	reset := setTestHookCollectGarbage(func(_ int64) {
+		got = changed
+	})
+
+	// set got variable
+	testHookCollectGarbage(0)
+
+	// test if got variable is set correctly to changed value
+	if got != changed {
+		t.Errorf("got hook value %v, want %v", got, changed)
+	}
+
+	// set the function to the original one
+	reset()
+
+	// set got variable
+	testHookCollectGarbage(0)
+
+	// test if got variable is set correctly to original value
+	if got != original {
+		t.Errorf("got hook value %v, want %v", got, original)
+	}
+}

swarm/storage/localstore/localstore.go

@@ -21,7 +21,6 @@ import (
 	"encoding/hex"
 	"errors"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/ethereum/go-ethereum/swarm/shed"
@@ -41,9 +40,13 @@ 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
+var (
+	// Default value for Capacity DB option.
+	defaultCapacity int64 = 5000000
+	// Limit the number of goroutines created by Getters
+	// that call updateGC function. Value 0 sets no limit.
+	maxParallelUpdateGC = 1000
+)
 
 // DB is the local store implementation and holds
 // database related objects.
@@ -71,6 +74,11 @@ type DB struct {
 
 	// number of elements in garbage collection index
 	gcSize int64
+	// garbage collection is triggered when gcSize exceeds
+	// the capacity value
+	capacity int64
+
+	collectGarbageTrigger chan struct{}
 
 	// a buffered channel acting as a semaphore
 	// to limit the maximal number of goroutines
@@ -80,6 +88,10 @@ type DB struct {
 	baseKey []byte
 
 	addressLocks sync.Map
+
+	// this channel is closed when close function is called
+	// to terminate other goroutines
+	close chan struct{}
 }
 
 // Options struct holds optional parameters for configuring DB.
@@ -99,6 +111,9 @@ type Options struct {
 	// of swarm nodes with chunk data deduplication provided by
 	// the mock global store.
 	MockStore *mock.NodeStore
+	// Capacity is a limit that triggers garbage collection when
+	// number of items in gcIndex equals or exceeds it.
+	Capacity int64
 }
 
 // New returns a new DB.  All fields and indexes are initialized
@@ -109,8 +124,17 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
 		o = new(Options)
 	}
 	db = &DB{
+		capacity:                   o.Capacity,
 		baseKey:                    baseKey,
 		useRetrievalCompositeIndex: o.UseRetrievalCompositeIndex,
+		// this channel needs to be buffered with the size of 1
+		// to signal another garbage collection run if it
+		// is triggered during already running one
+		collectGarbageTrigger: make(chan struct{}, 1),
+		close:                 make(chan struct{}),
+	}
+	if db.capacity <= 0 {
+		db.capacity = defaultCapacity
 	}
 	if maxParallelUpdateGC > 0 {
 		db.updateGCSem = make(chan struct{}, maxParallelUpdateGC)
@@ -321,18 +345,20 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
 	if err != nil {
 		return nil, err
 	}
+	// start garbage collection worker
+	go db.collectGarbage()
 	// count number of elements in garbage collection index
-	var gcSize int64
-	db.gcIndex.IterateAll(func(_ shed.IndexItem) (stop bool, err error) {
-		gcSize++
-		return false, nil
-	})
-	atomic.AddInt64(&db.gcSize, gcSize)
+	gcSize, err := db.gcIndex.Count()
+	if err != nil {
+		return nil, err
+	}
+	db.incGCSize(int64(gcSize))
 	return db, nil
 }
 
 // Close closes the underlying database.
 func (db *DB) Close() (err error) {
+	close(db.close)
 	return db.shed.Close()
 }
 

swarm/storage/localstore/localstore_test.go

@@ -166,10 +166,10 @@ func TestDB_updateGCSem(t *testing.T) {
 // goos: darwin
 // goarch: amd64
 // pkg: github.com/ethereum/go-ethereum/swarm/storage/localstore
-// BenchmarkNew/1000-8         	     200	  11684285 ns/op	 9556056 B/op	   10005 allocs/op
-// BenchmarkNew/10000-8        	     100	  15161036 ns/op	10539571 B/op	    7799 allocs/op
-// BenchmarkNew/100000-8       	      20	  74270386 ns/op	18234588 B/op	   24382 allocs/op
-// BenchmarkNew/1000000-8      	       2	 942098251 ns/op	48747500 B/op	  274976 allocs/op
+// BenchmarkNew/1000-8         	     200	  11672414 ns/op	 9570960 B/op	   10008 allocs/op
+// BenchmarkNew/10000-8        	     100	  14890609 ns/op	10490118 B/op	    7759 allocs/op
+// BenchmarkNew/100000-8       	      20	  58334080 ns/op	17763157 B/op	   22978 allocs/op
+// BenchmarkNew/1000000-8      	       2	 748595153 ns/op	45297404 B/op	  253242 allocs/op
 // PASS
 func BenchmarkNew(b *testing.B) {
 	if testing.Short() {
@@ -178,8 +178,8 @@ func BenchmarkNew(b *testing.B) {
 	for _, count := range []int{
 		1000,
 		10000,
-		// 100000,
-		// 1000000,
+		100000,
+		1000000,
 	} {
 		b.Run(strconv.Itoa(count), func(b *testing.B) {
 			dir, err := ioutil.TempDir("", "localstore-new-benchmark")
@@ -502,69 +502,6 @@ func validateItem(t *testing.T, item shed.IndexItem, address, data []byte, store
 	}
 }
 
-// setTestHookUpdateGC sets testHookUpdateGC and
-// returns a function that will reset it to the
-// value before the change.
-func setTestHookUpdateGC(h func()) (reset func()) {
-	current := testHookUpdateGC
-	reset = func() { testHookUpdateGC = current }
-	testHookUpdateGC = h
-	return reset
-}
-
-// TestSetTestHookUpdateGC tests if setTestHookUpdateGC changes
-// testHookUpdateGC function correctly and if its reset function
-// resets the original function.
-func TestSetTestHookUpdateGC(t *testing.T) {
-	// Set the current function after the test finishes.
-	defer func(h func()) { testHookUpdateGC = h }(testHookUpdateGC)
-
-	// expected value for the unchanged function
-	original := 1
-	// expected value for the changed function
-	changed := 2
-
-	// this variable will be set with two different functions
-	var got int
-
-	// define the original (unchanged) functions
-	testHookUpdateGC = func() {
-		got = original
-	}
-
-	// set got variable
-	testHookUpdateGC()
-
-	// test if got variable is set correctly
-	if got != original {
-		t.Errorf("got hook value %v, want %v", got, original)
-	}
-
-	// set the new function
-	reset := setTestHookUpdateGC(func() {
-		got = changed
-	})
-
-	// set got variable
-	testHookUpdateGC()
-
-	// test if got variable is set correctly to changed value
-	if got != changed {
-		t.Errorf("got hook value %v, want %v", got, changed)
-	}
-
-	// set the function to the original one
-	reset()
-
-	// set got variable
-	testHookUpdateGC()
-
-	// test if got variable is set correctly to original value
-	if got != original {
-		t.Errorf("got hook value %v, want %v", got, original)
-	}
-}
-
 // setNow replaces now function and
 // returns a function that will reset it to the
 // value before the change.

swarm/storage/localstore/mode_get_test.go

@@ -204,3 +204,66 @@ func testModeGetSyncValues(t *testing.T, db *DB) {
 
 	t.Run("gc size", newIndexGCSizeTest(db))
 }
+
+// setTestHookUpdateGC sets testHookUpdateGC and
+// returns a function that will reset it to the
+// value before the change.
+func setTestHookUpdateGC(h func()) (reset func()) {
+	current := testHookUpdateGC
+	reset = func() { testHookUpdateGC = current }
+	testHookUpdateGC = h
+	return reset
+}
+
+// TestSetTestHookUpdateGC tests if setTestHookUpdateGC changes
+// testHookUpdateGC function correctly and if its reset function
+// resets the original function.
+func TestSetTestHookUpdateGC(t *testing.T) {
+	// Set the current function after the test finishes.
+	defer func(h func()) { testHookUpdateGC = h }(testHookUpdateGC)
+
+	// expected value for the unchanged function
+	original := 1
+	// expected value for the changed function
+	changed := 2
+
+	// this variable will be set with two different functions
+	var got int
+
+	// define the original (unchanged) functions
+	testHookUpdateGC = func() {
+		got = original
+	}
+
+	// set got variable
+	testHookUpdateGC()
+
+	// test if got variable is set correctly
+	if got != original {
+		t.Errorf("got hook value %v, want %v", got, original)
+	}
+
+	// set the new function
+	reset := setTestHookUpdateGC(func() {
+		got = changed
+	})
+
+	// set got variable
+	testHookUpdateGC()
+
+	// test if got variable is set correctly to changed value
+	if got != changed {
+		t.Errorf("got hook value %v, want %v", got, changed)
+	}
+
+	// set the function to the original one
+	reset()
+
+	// set got variable
+	testHookUpdateGC()
+
+	// test if got variable is set correctly to original value
+	if got != original {
+		t.Errorf("got hook value %v, want %v", got, original)
+	}
+}

swarm/storage/localstore/mode_put.go

@@ -17,8 +17,6 @@
 package localstore
 
 import (
-	"sync/atomic"
-
 	"github.com/ethereum/go-ethereum/swarm/shed"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 	"github.com/syndtr/goleveldb/leveldb"
@@ -114,7 +112,7 @@ func (db *DB) put(mode ModePut, item shed.IndexItem) (err error) {
 		if item.AccessTimestamp != 0 {
 			// delete current entry from the gc index
 			db.gcIndex.DeleteInBatch(batch, item)
-			atomic.AddInt64(&db.gcSize, -1)
+			db.incGCSize(-1)
 		}
 		if item.StoreTimestamp == 0 {
 			item.StoreTimestamp = now()
@@ -129,7 +127,7 @@ func (db *DB) put(mode ModePut, item shed.IndexItem) (err error) {
 		}
 		// add new entry to gc index
 		db.gcIndex.PutInBatch(batch, item)
-		atomic.AddInt64(&db.gcSize, 1)
+		db.incGCSize(1)
 
 		if db.useRetrievalCompositeIndex {
 			db.retrievalCompositeIndex.PutInBatch(batch, item)

swarm/storage/localstore/mode_set.go

@@ -17,8 +17,6 @@
 package localstore
 
 import (
-	"sync/atomic"
-
 	"github.com/ethereum/go-ethereum/swarm/storage"
 	"github.com/syndtr/goleveldb/leveldb"
 )
@@ -88,7 +86,7 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
 				item.AccessTimestamp = i.AccessTimestamp
 				item.StoreTimestamp = i.StoreTimestamp
 				db.gcIndex.DeleteInBatch(batch, item)
-				atomic.AddInt64(&db.gcSize, -1)
+				db.incGCSize(-1)
 			case leveldb.ErrNotFound:
 				db.pullIndex.DeleteInBatch(batch, item)
 				item.AccessTimestamp = now()
@@ -113,7 +111,7 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
 			case nil:
 				item.AccessTimestamp = i.AccessTimestamp
 				db.gcIndex.DeleteInBatch(batch, item)
-				atomic.AddInt64(&db.gcSize, -1)
+				db.incGCSize(-1)
 			case leveldb.ErrNotFound:
 				// the chunk is not accessed before
 			default:
@@ -124,7 +122,7 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
 		}
 		db.pullIndex.PutInBatch(batch, item)
 		db.gcIndex.PutInBatch(batch, item)
-		atomic.AddInt64(&db.gcSize, 1)
+		db.incGCSize(1)
 
 	case ModeSetSync:
 		// delete from push, insert to gc
@@ -157,7 +155,7 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
 				// the chunk is accessed before
 				// remove the current gc index item
 				db.gcIndex.DeleteInBatch(batch, item)
-				atomic.AddInt64(&db.gcSize, -1)
+				db.incGCSize(-1)
 			}
 		} else {
 			i, err := db.retrievalDataIndex.Get(item)
@@ -179,7 +177,7 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
 			case nil:
 				item.AccessTimestamp = i.AccessTimestamp
 				db.gcIndex.DeleteInBatch(batch, item)
-				atomic.AddInt64(&db.gcSize, -1)
+				db.incGCSize(-1)
 			case leveldb.ErrNotFound:
 				// the chunk is not accessed before
 			default:
@@ -190,7 +188,7 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
 		}
 		db.pushIndex.DeleteInBatch(batch, item)
 		db.gcIndex.PutInBatch(batch, item)
-		atomic.AddInt64(&db.gcSize, 1)
+		db.incGCSize(1)
 
 	case ModeSetRemove:
 		// delete from retrieve, pull, gc
@@ -230,8 +228,10 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
 		db.gcIndex.DeleteInBatch(batch, item)
 		// TODO: optimize in garbage collection
 		// get is too expensive operation
+		// Suggestion: remove ModeSetRemove and use this code
+		// only in collectGarbage function
 		if _, err := db.gcIndex.Get(item); err == nil {
-			atomic.AddInt64(&db.gcSize, -1)
+			db.incGCSize(-1)
 		}
 
 	default: