diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go index d9f6b62340..ca9d2403e7 100644 --- a/swarm/storage/ldbstore.go +++ b/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) { - s.gcPos = it.Key() - } else { - s.gcPos = nil - } + defer it.Release() + + garbage := []*gcItem{} gcnt := 0 - for (gcnt < gcArraySize) && (uint64(gcnt) < s.entryCnt) { - - if (s.gcPos == nil) || (s.gcPos[0] != keyIndex) { - it.Seek(s.gcStartPos) - if it.Valid() { - s.gcPos = it.Key() - } else { - s.gcPos = s.gcStartPos - } - } - - if (s.gcPos == nil) || (s.gcPos[0] != keyIndex) { + for ok := it.Seek([]byte{keyIndex}); ok && (gcnt < 5000) && (uint64(gcnt) < s.entryCnt); ok = it.Next() { + key := it.Key() + val := it.Value() + if (key == nil) || (key[0] != keyIndex) { break } - gci := new(gcItem) - gci.idxKey = s.gcPos + log.Trace("iterator", "key", fmt.Sprintf("%x", key), "value", fmt.Sprintf("%x", val)) + 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 - for i := 0; i < gcnt; i++ { - if s.gcArray[i].value <= cutval { - gcCounter.Inc(1) - s.delete(s.gcArray[i].idx, s.gcArray[i].idxKey, s.po(Key(s.gcPos[1:]))) - } + sort.Slice(garbage[:gcnt], func(i, j int) bool { return garbage[i].value < garbage[j].value }) + + for k := 0; k < gcnt; k++ { + log.Trace("gcArray[]", "k", k, "idx", garbage[k].idx, "idxKey", fmt.Sprintf("%x", garbage[k].idxKey), "value", garbage[k].value) } - 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 { - log.Trace(fmt.Sprintf("DbStore: collecting garbage...(%d chunks)", e)) + if e >= s.capacity && int(float32(e-1)*gcArrayFreeRatio) > 0 { + 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 } diff --git a/swarm/storage/ldbstore_test.go b/swarm/storage/ldbstore_test.go index 82f5b053be..928d8dc94a 100644 --- a/swarm/storage/ldbstore_test.go +++ b/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") + } +}