storage/localstore: cleanup and squash of GC changes

Author: tzdybal

storage/localstore/gc.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
+	"github.com/ethersphere/swarm/chunk"
 	"github.com/ethersphere/swarm/shed"
 	"github.com/syndtr/goleveldb/leveldb"
 )
@@ -40,6 +41,36 @@ var (
 	gcBatchSize uint64 = 200
 )
 
+type GCPolicy interface {
+	GetEvictionMetric(addr chunk.Address) Fraction
+	GetAdmissionMetric(addr chunk.Address) Fraction
+}
+
+// based on https://hackmd.io/t-OQFK3mTsGfrpLCqDrdlw#Synced-chunks
+type DefaultGCPolicy struct {
+	db *DB
+}
+
+// based on https://hackmd.io/t-OQFK3mTsGfrpLCqDrdlw#Synced-chunks
+func (p *DefaultGCPolicy) GetEvictionMetric(addr chunk.Address) Fraction {
+	item := addressToItem(addr)
+	po := int(p.db.po(item.Address))
+	responsibilityRadius := p.db.getResponsibilityRadius()
+
+	if po < responsibilityRadius {
+		// More Distant Chunks
+		n := uint64(responsibilityRadius - po)
+		return Fraction{Numerator: n, Denominator: n + 1}
+	}
+	// Most Proximate Chunks
+	return Fraction{Numerator: 1, Denominator: 3}
+}
+
+// GetAdmissionMetric is currently not in use
+func (p *DefaultGCPolicy) GetAdmissionMetric(addr chunk.Address) Fraction {
+	return Fraction{1, 1}
+}
+
 // 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
@@ -96,7 +127,7 @@ func (db *DB) collectGarbage() (collectedCount uint64, done bool, err error) {
 
 	// run through the recently pinned chunks and
 	// remove them from the gcIndex before iterating through gcIndex
-	err = db.removeChunksInExcludeIndexFromGC()
+	err = db.syncExcludeAndGCIndex()
 	if err != nil {
 		log.Error("localstore exclude pinned chunks", "err", err)
 		return 0, true, err
@@ -148,8 +179,8 @@ func (db *DB) collectGarbage() (collectedCount uint64, done bool, err error) {
 	return collectedCount, done, err
 }
 
-// removeChunksInExcludeIndexFromGC removed any recently chunks in the exclude Index, from the gcIndex.
-func (db *DB) removeChunksInExcludeIndexFromGC() (err error) {
+// syncExcludeAndGCIndex removes any chunks in the exclude Index, from the gcIndex.
+func (db *DB) syncExcludeAndGCIndex() (err error) {
 	metricName := "localstore/gc/exclude"
 	metrics.GetOrRegisterCounter(metricName, nil).Inc(1)
 	defer totalTimeMetric(metricName, time.Now())
@@ -275,20 +306,8 @@ func (db *DB) updateGCQuantiles() (err error) {
 	}
 	var newQuantiles quantiles
 	for _, q := range gcQuantiles {
-		item, position, found := gcQuantiles.Get(q.Fraction)
-		if !found {
-			continue
-		}
 		newPosition := quantilePosition(gcSize, q.Numerator, q.Denominator)
-		diff := uint64Diff(position, newPosition)
-		if diff == 0 {
-			continue
-		}
-		newItem, err := db.gcIndex.Offset(&item, diff)
-		if err != nil {
-			return err
-		}
-		newQuantiles.Set(q.Fraction, newItem, newPosition)
+		newQuantiles = newQuantiles.Set(q.Fraction, q.Item, newPosition)
 	}
 	return db.gcQuantiles.Put(newQuantiles)
 }

storage/localstore/gc_test.go

@@ -578,5 +578,4 @@ func TestUpdateGCQuantiles(t *testing.T) {
 	defer db.Close()
 
 	db.collectGarbage()
-	db.collectGarbage()
 }

storage/localstore/localstore.go

@@ -89,6 +89,8 @@ type DB struct {
 	// pin files Index
 	pinIndex shed.Index
 
+	gcPolicy GCPolicy
+
 	// field that stores number of intems in gc index
 	gcSize shed.Uint64Field
 
@@ -183,6 +185,7 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
 		collectGarbageWorkerDone: make(chan struct{}),
 		putToGCCheck:             o.PutToGCCheck,
 	}
+	db.gcPolicy = &DefaultGCPolicy{db}
 	if db.capacity <= 0 {
 		db.capacity = defaultCapacity
 	}

storage/localstore/mode_set.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/ethersphere/swarm/shed"
 	"time"
 
 	"github.com/ethereum/go-ethereum/metrics"
@@ -201,7 +202,7 @@ func (db *DB) setSync(batch *leveldb.Batch, addr chunk.Address, mode chunk.ModeS
 	// provided by the access function, and it is not
 	// a property of a chunk provided to Accessor.Put.
 
-	i, err := db.retrievalDataIndex.Get(item)
+	accessIndexItem, err := db.retrievalDataIndex.Get(item)
 	if err != nil {
 		if err == leveldb.ErrNotFound {
 			// chunk is not found,
@@ -213,8 +214,8 @@ func (db *DB) setSync(batch *leveldb.Batch, addr chunk.Address, mode chunk.ModeS
 		}
 		return 0, err
 	}
-	item.StoreTimestamp = i.StoreTimestamp
-	item.BinID = i.BinID
+	item.StoreTimestamp = accessIndexItem.StoreTimestamp
+	item.BinID = accessIndexItem.BinID
 
 	switch mode {
 	case chunk.ModeSetSyncPull:
@@ -286,10 +287,10 @@ func (db *DB) setSync(batch *leveldb.Batch, addr chunk.Address, mode chunk.ModeS
 		db.pushIndex.DeleteInBatch(batch, item)
 	}
 
-	i, err = db.retrievalAccessIndex.Get(item)
+	accessIndexItem, err = db.retrievalAccessIndex.Get(item)
 	switch err {
 	case nil:
-		item.AccessTimestamp = i.AccessTimestamp
+		item.AccessTimestamp = accessIndexItem.AccessTimestamp
 		db.gcIndex.DeleteInBatch(batch, item)
 		gcSizeChange--
 	case leveldb.ErrNotFound:
@@ -306,50 +307,57 @@ func (db *DB) setSync(batch *leveldb.Batch, addr chunk.Address, mode chunk.ModeS
 		return 0, err
 	}
 	if !pinned {
-		// set access timestamp based on quantile calculated from
-		// chunk proximity and db.responsibilityRadius
-		f := chunkQuantileFraction(int(db.po(item.Address)), db.getResponsibilityRadius())
-		gcSize, err := db.gcSize.Get()
+		err := db.addToGC(batch, item, accessIndexItem)
 		if err != nil {
 			return 0, err
 		}
-		position := quantilePosition(gcSize, f.Numerator, f.Denominator)
-		var gcQuantiles quantiles
-		err = db.gcQuantiles.Get(&gcQuantiles)
-		if err != nil && err != leveldb.ErrNotFound {
-			return 0, err
+		gcSizeChange++
+	}
+
+	return gcSizeChange, nil
+}
+
+func (db *DB) addToGC(batch *leveldb.Batch, item shed.Item, i shed.Item) error {
+	// set access timestamp based on quantile calculated from
+	// chunk proximity and db.responsibilityRadius
+	f := db.gcPolicy.GetEvictionMetric(item.Address)
+	gcSize, err := db.gcSize.Get()
+	if err != nil {
+		return err
+	}
+	position := quantilePosition(gcSize, f.Numerator, f.Denominator)
+	var gcQuantiles quantiles
+	err = db.gcQuantiles.Get(&gcQuantiles)
+	if err != nil && err != leveldb.ErrNotFound {
+		return err
+	}
+	var found bool
+	for _, q := range gcQuantiles {
+		if q.Fraction == f {
+			item.AccessTimestamp = q.Item.AccessTimestamp + 1
+			found = true
+			break
 		}
-		var found bool
-		for _, q := range gcQuantiles {
-			if q.Fraction == f {
-				item.AccessTimestamp = q.Item.AccessTimestamp + 1
-				found = true
-				break
-			}
+	}
+	if len(gcQuantiles) > 0 && !found {
+		var shift int64
+		if closest := gcQuantiles.Closest(f); closest == nil {
+			shift = int64(position)
+		} else {
+			shift = int64(closest.Position) - int64(position)
 		}
-		if len(gcQuantiles) > 0 && !found {
-			var shift int64
-			if closest := gcQuantiles.Closest(f); closest == nil {
-				shift = int64(position)
-			} else {
-				shift = int64(closest.Position) - int64(position)
-			}
-			i, err = db.gcIndex.Offset(nil, shift)
-			if err != nil {
-				return 0, err
-			}
+		i, err = db.gcIndex.Offset(nil, shift)
+		if err == nil {
 			item.AccessTimestamp = i.AccessTimestamp + 1
 		}
-		gcQuantiles = gcQuantiles.Set(f, item, position)
-		db.gcQuantiles.PutInBatch(batch, gcQuantiles)
-
-		db.retrievalAccessIndex.PutInBatch(batch, item)
-		db.pushIndex.DeleteInBatch(batch, item)
-		db.gcIndex.PutInBatch(batch, item)
-		gcSizeChange++
 	}
+	gcQuantiles = gcQuantiles.Set(f, item, position)
+	db.gcQuantiles.PutInBatch(batch, gcQuantiles)
 
-	return gcSizeChange, nil
+	db.retrievalAccessIndex.PutInBatch(batch, item)
+	db.pushIndex.DeleteInBatch(batch, item)
+	db.gcIndex.PutInBatch(batch, item)
+	return nil
 }
 
 // setRemove removes the chunk by updating indexes:

storage/localstore/quantile.go

@@ -66,26 +66,15 @@ func (q quantiles) Set(f Fraction, item shed.Item, position uint64) quantiles {
 			return q
 		}
 	}
-	q = append(q, quantile{
+	newQ := append(q, quantile{
 		Fraction: f,
 		Item:     item,
 	})
-	sort.Sort(q)
-	return q
+	sort.Sort(newQ)
+	return newQ
 }
 
 func quantilePosition(total, numerator, denominator uint64) uint64 {
 	return total / denominator * numerator
 }
 
-// based on https://hackmd.io/t-OQFK3mTsGfrpLCqDrdlw#Synced-chunks
-// TODO: review and document exact quantiles for chunks
-func chunkQuantileFraction(po, responsibilityRadius int) Fraction {
-	if po < responsibilityRadius {
-		// More Distant Chunks
-		n := uint64(responsibilityRadius - po)
-		return Fraction{Numerator: n, Denominator: n + 1}
-	}
-	// Most Proximate Chunks
-	return Fraction{Numerator: 1, Denominator: 3}
-}