diff --git a/swarm/shed/index.go b/swarm/shed/index.go
index 70bb15f74b..9a4e2c0ced 100644
--- a/swarm/shed/index.go
+++ b/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()
+}
diff --git a/swarm/shed/index_test.go b/swarm/shed/index_test.go
index 33003b1f23..77f091202b 100644
--- a/swarm/shed/index_test.go
+++ b/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()
diff --git a/swarm/storage/localstore/gc.go b/swarm/storage/localstore/gc.go
new file mode 100644
index 0000000000..871c244e8d
--- /dev/null
+++ b/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 .
+
+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)
diff --git a/swarm/storage/localstore/gc_test.go b/swarm/storage/localstore/gc_test.go
new file mode 100644
index 0000000000..634a3d98de
--- /dev/null
+++ b/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 .
+
+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)
+ }
+}
diff --git a/swarm/storage/localstore/localstore.go b/swarm/storage/localstore/localstore.go
index 483ce458eb..81d45529ce 100644
--- a/swarm/storage/localstore/localstore.go
+++ b/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()
}
diff --git a/swarm/storage/localstore/localstore_test.go b/swarm/storage/localstore/localstore_test.go
index dce3c4ad06..b7ed351772 100644
--- a/swarm/storage/localstore/localstore_test.go
+++ b/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.
diff --git a/swarm/storage/localstore/mode_get_test.go b/swarm/storage/localstore/mode_get_test.go
index e928ec1f4a..eadb55cc59 100644
--- a/swarm/storage/localstore/mode_get_test.go
+++ b/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)
+ }
+}
diff --git a/swarm/storage/localstore/mode_put.go b/swarm/storage/localstore/mode_put.go
index a2e1fcdb5d..811a5ba558 100644
--- a/swarm/storage/localstore/mode_put.go
+++ b/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)
diff --git a/swarm/storage/localstore/mode_set.go b/swarm/storage/localstore/mode_set.go
index f6d161a96a..e8ac1ff63e 100644
--- a/swarm/storage/localstore/mode_set.go
+++ b/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: