swarm/storage/localstore: optimize collectGarbage

This commit is contained in:
Janos Guljas 2018-12-14 13:06:53 +01:00
parent 5732c38b1a
commit ac9d153a1f
3 changed files with 54 additions and 17 deletions

View file

@ -21,17 +21,23 @@ import (
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/swarm/shed" "github.com/ethereum/go-ethereum/swarm/shed"
"github.com/syndtr/goleveldb/leveldb"
) )
// gcTargetRatio defines the target number of items var (
// in garbage collection index that will not be removed // gcTargetRatio defines the target number of items
// on garbage collection. The target number of items // in garbage collection index that will not be removed
// is calculated by gcTarget function. This value must be // on garbage collection. The target number of items
// in range (0,1]. For example, with 0.9 value, // is calculated by gcTarget function. This value must be
// garbage collection will leave 90% of defined capacity // in range (0,1]. For example, with 0.9 value,
// in database after its run. This prevents frequent // garbage collection will leave 90% of defined capacity
// garbage collection runt. // in database after its run. This prevents frequent
var gcTargetRatio = 0.9 // garbage collection runt.
gcTargetRatio = 0.9
// gcBatchSize limits the number of chunks in a single
// leveldb batch on garbage collection.
gcBatchSize int64 = 1000
)
// collectGarbage is a long running function that waits for // collectGarbage is a long running function that waits for
// collectGarbageTrigger channel to signal a garbage collection // collectGarbageTrigger channel to signal a garbage collection
@ -42,22 +48,50 @@ func (db *DB) collectGarbage() {
for { for {
select { select {
case <-db.collectGarbageTrigger: case <-db.collectGarbageTrigger:
batch := new(leveldb.Batch)
// sets a gc trigger if batch limit is reached
var triggerNextIteration bool
var collectedCount int64 var collectedCount int64
err := db.gcIndex.IterateAll(func(item shed.IndexItem) (stop bool, err error) { err := db.gcIndex.IterateAll(func(item shed.IndexItem) (stop bool, err error) {
gcSize := atomic.LoadInt64(&db.gcSize) gcSize := atomic.LoadInt64(&db.gcSize)
if gcSize <= target { if gcSize-collectedCount <= target {
return true, nil return true, nil
} }
err = db.set(ModeSetRemove, item.Address) // delete from retrieve, pull, gc
if err != nil { if db.useRetrievalCompositeIndex {
return false, err db.retrievalCompositeIndex.DeleteInBatch(batch, item)
} else {
db.retrievalDataIndex.DeleteInBatch(batch, item)
db.retrievalAccessIndex.DeleteInBatch(batch, item)
} }
db.pullIndex.DeleteInBatch(batch, item)
db.gcIndex.DeleteInBatch(batch, item)
collectedCount++ collectedCount++
if collectedCount >= gcBatchSize {
triggerNextIteration = true
return true, nil
}
return false, nil return false, nil
}) })
if err != nil { if err != nil {
log.Error("localstore collect garbage", "err", err) log.Error("localstore collect garbage", "err", err)
} }
err = db.shed.WriteBatch(batch)
if err != nil {
log.Error("localstore collect garbage write batch", "err", err)
} else {
// batch is written, decrement gcSize and check if another gc run is needed
db.incGCSize(-collectedCount)
if triggerNextIteration {
select {
case db.collectGarbageTrigger <- struct{}{}:
default:
}
}
}
if testHookCollectGarbage != nil { if testHookCollectGarbage != nil {
testHookCollectGarbage(collectedCount) testHookCollectGarbage(collectedCount)
} }

View file

@ -82,6 +82,8 @@ func TestDB_collectGarbage(t *testing.T) {
t.Errorf("total collected chunks %v, want %v", totalCollectedCount, wantTotalCollectedCount) t.Errorf("total collected chunks %v, want %v", totalCollectedCount, wantTotalCollectedCount)
} }
t.Run("pull index count", newIndexItemsCountTest(db.pullIndex, int(gcTarget)))
t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, int(gcTarget))) t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, int(gcTarget)))
t.Run("gc size", newIndexGCSizeTest(db)) t.Run("gc size", newIndexGCSizeTest(db))
@ -182,6 +184,8 @@ func TestDB_collectGarbage_withRequests(t *testing.T) {
t.Errorf("total collected chunks %v, want %v", totalCollectedCount, wantTotalCollectedCount) t.Errorf("total collected chunks %v, want %v", totalCollectedCount, wantTotalCollectedCount)
} }
t.Run("pull index count", newIndexItemsCountTest(db.pullIndex, int(gcTarget)))
t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, int(gcTarget))) t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, int(gcTarget)))
t.Run("gc size", newIndexGCSizeTest(db)) t.Run("gc size", newIndexGCSizeTest(db))

View file

@ -226,10 +226,9 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
} }
db.pullIndex.DeleteInBatch(batch, item) db.pullIndex.DeleteInBatch(batch, item)
db.gcIndex.DeleteInBatch(batch, item) db.gcIndex.DeleteInBatch(batch, item)
// TODO: optimize in garbage collection // a check is needed for decrementing gcSize
// get is too expensive operation // as delete is not reporting if the key/value pair
// Suggestion: remove ModeSetRemove and use this code // is deleted or not
// only in collectGarbage function
if _, err := db.gcIndex.Get(item); err == nil { if _, err := db.gcIndex.Get(item); err == nil {
db.incGCSize(-1) db.incGCSize(-1)
} }