From 5732c38b1ad01d54c9b5d006c230e4fb004c93d0 Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Thu, 13 Dec 2018 15:52:44 +0100 Subject: [PATCH] swarm/storage/localstore: implement basic garbage collection --- swarm/shed/index.go | 15 ++ swarm/shed/index_test.go | 95 +++++++ swarm/storage/localstore/gc.go | 91 +++++++ swarm/storage/localstore/gc_test.go | 275 ++++++++++++++++++++ swarm/storage/localstore/localstore.go | 46 +++- swarm/storage/localstore/localstore_test.go | 75 +----- swarm/storage/localstore/mode_get_test.go | 63 +++++ swarm/storage/localstore/mode_put.go | 6 +- swarm/storage/localstore/mode_set.go | 18 +- 9 files changed, 592 insertions(+), 92 deletions(-) create mode 100644 swarm/storage/localstore/gc.go create mode 100644 swarm/storage/localstore/gc_test.go 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: