swarm/storage: working garbage collection

swarm/storage/ldbstore.go

@@ -30,6 +30,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"sort"
 	"sync"
 
 	"github.com/ethereum/go-ethereum/log"
@@ -50,7 +51,6 @@ const (
 	defaultDbCapacity = 5000000
 	defaultRadius     = 0 // not yet used
 
-	gcArraySize      = 10000
 	gcArrayFreeRatio = 0.1
 )
 
@@ -69,6 +69,7 @@ type gcItem struct {
 	idx    uint64
 	value  uint64
 	idxKey []byte
+	po     uint8
 }
 
 type LDBStore struct {
@@ -83,7 +84,6 @@ type LDBStore struct {
 
 	gcPos      []byte
 	gcStartPos []byte
-	gcArray    []*gcItem
 
 	hashfunc SwarmHasher
 	po       func(Key) uint8
@@ -138,6 +138,7 @@ func NewLDBStore(path string, hash SwarmHasher, capacity uint64, po func(Key) ui
 	data, _ := s.db.Get(keyEntryCnt)
 	s.entryCnt = BytesToU64(data)
 	s.entryCnt++
+	log.Trace("NewLDBStore s.entryCnt++", "entryCnt", s.entryCnt)
 	data, _ = s.db.Get(keyAccessCnt)
 	s.accessCnt = BytesToU64(data)
 	s.accessCnt++
@@ -147,7 +148,6 @@ func NewLDBStore(path string, hash SwarmHasher, capacity uint64, po func(Key) ui
 
 	s.gcStartPos = make([]byte, 1)
 	s.gcStartPos[0] = keyIndex
-	s.gcArray = make([]*gcItem, gcArraySize)
 	s.gcPos, _ = s.db.Get(keyGCPos)
 	if s.gcPos == nil {
 		s.gcPos = s.gcStartPos
@@ -195,10 +195,6 @@ func U64ToBytes(val uint64) []byte {
 	return data
 }
 
-func getIndexGCValue(index *dpaDBIndex) uint64 {
-	return index.Access
-}
-
 func (s *LDBStore) updateIndexAccess(index *dpaDBIndex) {
 	index.Access = s.accessCnt
 }
@@ -292,61 +288,55 @@ func gcListSelect(list []*gcItem, left int, right int, n int) int {
 
 func (s *LDBStore) collectGarbage(ratio float32) {
 	it := s.db.NewIterator()
-	if it.Seek(s.gcPos) {
+	defer it.Release()
-		s.gcPos = it.Key()
+
-	} else {
+	garbage := []*gcItem{}
-		s.gcPos = nil
-	}
 	gcnt := 0
 
-	for (gcnt < gcArraySize) && (uint64(gcnt) < s.entryCnt) {
+	for ok := it.Seek([]byte{keyIndex}); ok && (gcnt < 5000) && (uint64(gcnt) < s.entryCnt); ok = it.Next() {
-
+		key := it.Key()
-		if (s.gcPos == nil) || (s.gcPos[0] != keyIndex) {
+		val := it.Value()
-			it.Seek(s.gcStartPos)
+		if (key == nil) || (key[0] != keyIndex) {
-			if it.Valid() {
-				s.gcPos = it.Key()
-			} else {
-				s.gcPos = s.gcStartPos
-			}
-		}
-
-		if (s.gcPos == nil) || (s.gcPos[0] != keyIndex) {
 			break
 		}
 
-		gci := new(gcItem)
+		log.Trace("iterator", "key", fmt.Sprintf("%x", key), "value", fmt.Sprintf("%x", val))
-		gci.idxKey = s.gcPos
+
 		var index dpaDBIndex
-		decodeIndex(it.Value(), &index)
-		gci.idx = index.Idx
-		// the smaller, the more likely to be gc'd
-		gci.value = getIndexGCValue(&index)
-		s.gcArray[gcnt] = gci
-		gcnt++
-		it.Next()
-		if it.Valid() {
-			s.gcPos = it.Key()
-		} else {
-			s.gcPos = nil
-		}
-	}
-	it.Release()
 
-	if gcnt == 0 {
+		hash := key[1:]
+		decodeIndex(val, &index)
+		po := s.po(hash)
+
+		kkey := make([]byte, len(key))
+		copy(kkey, key)
+
+		gci := &gcItem{
+			idxKey: kkey,
+			idx:    index.Idx,
+			value:  index.Access,
+			po:     po,
+		}
+
+		log.Trace("gci.idxKey", "gcnt", gcnt, "idxKey", fmt.Sprintf("%x", gci.idxKey), "idx", gci.idx, "gci.value", gci.value)
+
+		garbage = append(garbage, gci)
 		gcnt++
 	}
-	cutidx := gcListSelect(s.gcArray, 0, gcnt-1, int(float32(gcnt)*ratio))
-	cutval := s.gcArray[cutidx].value
 
-	// actual gc
+	sort.Slice(garbage[:gcnt], func(i, j int) bool { return garbage[i].value < garbage[j].value })
-	for i := 0; i < gcnt; i++ {
+
-		if s.gcArray[i].value <= cutval {
+	for k := 0; k < gcnt; k++ {
-			gcCounter.Inc(1)
+		log.Trace("gcArray[]", "k", k, "idx", garbage[k].idx, "idxKey", fmt.Sprintf("%x", garbage[k].idxKey), "value", garbage[k].value)
-			s.delete(s.gcArray[i].idx, s.gcArray[i].idxKey, s.po(Key(s.gcPos[1:])))
-		}
 	}
 
-	s.db.Put(keyGCPos, s.gcPos)
+	cutoff := int(float32(gcnt) * ratio)
+	log.Trace("cutoff", "cut", cutoff, "gcnt", gcnt)
+	for i := 0; i < cutoff; i++ {
+		s.delete(garbage[i].idx, garbage[i].idxKey, garbage[i].po)
+	}
+
+	//s.db.Put(keyGCPos, s.gcPos)
 }
 
 // Export writes all chunks from the store to a tar archive, returning the
@@ -520,11 +510,14 @@ func (s *LDBStore) ReIndex() {
 }
 
 func (s *LDBStore) delete(idx uint64, idxKey []byte, po uint8) {
+	log.Trace("LDBStore delete()", "idx", idx, "idxKey", fmt.Sprintf("%x", idxKey), "po", po)
+
 	batch := new(leveldb.Batch)
 	batch.Delete(idxKey)
 	batch.Delete(getDataKey(idx, po))
 	dbStoreDeleteCounter.Inc(1)
 	s.entryCnt--
+	log.Trace("delete s.entryCnt--", "entryCnt", s.entryCnt)
 	s.bucketCnt[po]--
 	cntKey := make([]byte, 2)
 	cntKey[0] = keyDistanceCnt
@@ -594,6 +587,7 @@ func (s *LDBStore) doPut(chunk *Chunk, index *dpaDBIndex, po uint8) {
 	index.Idx = s.dataIdx
 	s.bucketCnt[po] = s.dataIdx
 	s.entryCnt++
+	log.Trace("doPut entryCnt++", "entryCnt", s.entryCnt)
 	s.dataIdx++
 
 	cntKey := make([]byte, 2)
@@ -612,17 +606,17 @@ func (s *LDBStore) writeBatches() {
 		c := s.batchC
 		s.batchC = make(chan bool)
 		s.batch = new(leveldb.Batch)
-		s.lock.Unlock()
 		err := s.writeBatch(b, e, d, a)
 		// TODO: set this error on the batch, then tell the chunk
 		if err != nil {
-			log.Error(fmt.Sprintf("DbStore: spawn batch write (%d chunks): %v", b.Len(), err))
+			log.Error(fmt.Sprintf("spawn batch write (%d entries): %v", b.Len(), err))
 		}
 		close(c)
-		if e >= s.capacity {
+		if e >= s.capacity && int(float32(e-1)*gcArrayFreeRatio) > 0 {
-			log.Trace(fmt.Sprintf("DbStore: collecting garbage...(%d chunks)", e))
+			log.Trace(fmt.Sprintf("collecting garbage (%d chunks)", e))
 			s.collectGarbage(gcArrayFreeRatio)
 		}
+		s.lock.Unlock()
 	}
 	log.Trace(fmt.Sprintf("DbStore: quit batch write loop"))
 }
@@ -636,7 +630,7 @@ func (s *LDBStore) writeBatch(b *leveldb.Batch, entryCnt, dataIdx, accessCnt uin
 	if err := s.db.Write(b); err != nil {
 		return fmt.Errorf("unable to write batch: %v", err)
 	}
-	log.Trace(fmt.Sprintf("DbStore: batch write (%d chunks) complete", l))
+	log.Trace(fmt.Sprintf("batch write (%d entries)", l))
 	return nil
 }
 

swarm/storage/ldbstore_test.go

@@ -23,7 +23,6 @@ import (
 	"os"
 	"sync"
 	"testing"
-	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
@@ -353,11 +352,6 @@ func TestLDBStoreCollectGarbage(t *testing.T) {
 
 	for i := 0; i < n; i++ {
 		ldb.Put(chunks[i])
-
-		if i%100 == 0 {
-			log.Info("sleeping 1 sec...")
-			time.Sleep(1 * time.Second)
-		}
 	}
 
 	// wait for all chunks to be stored before ending the test are cleaning up
@@ -367,8 +361,6 @@ func TestLDBStoreCollectGarbage(t *testing.T) {
 
 	log.Info("ldbstore", "entrycnt", ldb.entryCnt, "accesscnt", ldb.accessCnt)
 
-	time.Sleep(5 * time.Second)
-
 	var missing int
 	for i := 0; i < n; i++ {
 		ret, err := ldb.Get(chunks[i].Key)
@@ -419,7 +411,7 @@ func TestLDBStoreAddRemove(t *testing.T) {
 		go ldb.Put(chunks[i])
 	}
 
-	// wait for all chunks to be stored before ending the test are cleaning up
+	// wait for all chunks to be stored before continuing
 	for i := 0; i < n; i++ {
 		<-chunks[i].dbStoredC
 	}
@@ -460,3 +452,79 @@ func TestLDBStoreAddRemove(t *testing.T) {
 		}
 	}
 }
+
+// TestLDBStoreRemoveThenCollectGarbage tests that we can delete chunks and that we can trigger garbage collection
+func TestLDBStoreRemoveThenCollectGarbage(t *testing.T) {
+	log.PrintOrigins(true)
+	log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
+
+	capacity := 10
+
+	ldb, cleanup := newLDBStore(t)
+	ldb.setCapacity(uint64(capacity))
+
+	n := 7
+
+	chunks := []*Chunk{}
+	for i := 0; i < capacity; i++ {
+		c := NewRandomChunk(chunkSize)
+		chunks = append(chunks, c)
+		log.Info("generate random chunk", "idx", i, "chunk", c)
+	}
+
+	for i := 0; i < n; i++ {
+		ldb.Put(chunks[i])
+	}
+
+	// wait for all chunks to be stored before continuing
+	for i := 0; i < n; i++ {
+		<-chunks[i].dbStoredC
+	}
+
+	// delete all chunks
+	for i := 0; i < n; i++ {
+		key := chunks[i].Key
+		ikey := getIndexKey(key)
+
+		var indx dpaDBIndex
+		ldb.tryAccessIdx(ikey, &indx)
+
+		ldb.delete(indx.Idx, ikey, ldb.po(key))
+	}
+
+	log.Info("ldbstore", "entrycnt", ldb.entryCnt, "accesscnt", ldb.accessCnt)
+
+	cleanup()
+
+	ldb, cleanup = newLDBStore(t)
+	ldb.setCapacity(uint64(capacity))
+
+	n = 10
+
+	for i := 0; i < n; i++ {
+		ldb.Put(chunks[i])
+	}
+
+	// wait for all chunks to be stored before continuing
+	for i := 0; i < n; i++ {
+		<-chunks[i].dbStoredC
+	}
+
+	// expect for first chunk to be missing
+	idx := 0
+	ret, err := ldb.Get(chunks[idx].Key)
+	if err == nil || ret != nil {
+		t.Fatal("expected first chunk to be missing, but got no error")
+	}
+
+	// expect for last chunk to be present
+	idx = 9
+	ret, err = ldb.Get(chunks[idx].Key)
+	if err != nil {
+		t.Fatalf("expected no error, but got %s", err)
+	}
+
+	if !bytes.Equal(ret.SData, chunks[idx].SData) {
+		t.Fatal("expected to get the same data back, but got smth else")
+	}
+}