From 37bf2006d58e76ce8c1a93716cba8c65b4dd1a3e Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Mon, 17 Sep 2018 13:04:46 +0200 Subject: [PATCH 1/8] swarm: schemas and migrations --- swarm/swarm.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/swarm/swarm.go b/swarm/swarm.go index aea0989a1e..3c6bda445e 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -159,6 +159,11 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e return nil, err } + err = lstore.Migrate() + if err != nil { + return nil, err + } + self.netStore, err = storage.NewNetStore(lstore, nil) if err != nil { return nil, err From e958529a4380acfba098997757c5eada9d2c6f0e Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Mon, 17 Sep 2018 13:34:57 +0200 Subject: [PATCH 2/8] swarm/storage: extract keySchema --- swarm/storage/ldbstore.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go index 2a7f51cb38..415120552f 100644 --- a/swarm/storage/ldbstore.go +++ b/swarm/storage/ldbstore.go @@ -60,7 +60,7 @@ var ( keyDataIdx = []byte{4} keyData = byte(6) keyDistanceCnt = byte(7) - keySchema = []byte{8} + keySchema = byte(8) ) var ( From dcc062205b997d20af978566aed0dd6561393a85 Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Tue, 18 Sep 2018 13:35:39 +0200 Subject: [PATCH 3/8] swarm/storage: use validators to cleanup LevelDB --- swarm/storage/ldbstore.go | 2 +- swarm/swarm.go | 10 +++++----- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go index 415120552f..2a7f51cb38 100644 --- a/swarm/storage/ldbstore.go +++ b/swarm/storage/ldbstore.go @@ -60,7 +60,7 @@ var ( keyDataIdx = []byte{4} keyData = byte(6) keyDistanceCnt = byte(7) - keySchema = byte(8) + keySchema = []byte{8} ) var ( diff --git a/swarm/swarm.go b/swarm/swarm.go index 3c6bda445e..c31fbfb15b 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -159,11 +159,6 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e return nil, err } - err = lstore.Migrate() - if err != nil { - return nil, err - } - self.netStore, err = storage.NewNetStore(lstore, nil) if err != nil { return nil, err @@ -207,6 +202,11 @@ func NewSwarm(config *api.Config, mockStore *mock.NodeStore) (self *Swarm, err e return nil, err } + err = lstore.Migrate() + if err != nil { + return nil, err + } + log.Debug("Setup local storage") self.bzz = network.NewBzz(bzzconfig, to, stateStore, stream.Spec, self.streamer.Run) From 481f0ab1c44d4b9af308bbd42f58d8bb5020bd66 Mon Sep 17 00:00:00 2001 From: lash Date: Wed, 3 Oct 2018 08:36:18 +0200 Subject: [PATCH 4/8] swarm/storage: Add accesscount-indexcount index on put --- swarm/storage/ldbstore.go | 22 ++++++++++++---------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go index 2a7f51cb38..80aedfa47d 100644 --- a/swarm/storage/ldbstore.go +++ b/swarm/storage/ldbstore.go @@ -61,6 +61,7 @@ var ( keyData = byte(6) keyDistanceCnt = byte(7) keySchema = []byte{8} + keyAccessIdx = []byte{9} // access to chunk data index, used by garbage collection in ascending order from first entry ) var ( @@ -225,6 +226,14 @@ func getDataKey(idx uint64, po uint8) []byte { return key } +func getAccessIdxKey(idx *dpaDBIndex) []byte { + key := make([]byte, 17) + key[0] = keyAccessIdx[0] + binary.BigEndian.PutUint64(key[1:], index.Access) + binary.BigEndian.PutUint64(key[9:], index.Idx) + return key +} + func encodeIndex(index *dpaDBIndex) []byte { data, _ := rlp.EncodeToBytes(index) return data @@ -602,6 +611,9 @@ func (s *LDBStore) Put(ctx context.Context, chunk Chunk) error { idata = encodeIndex(&index) s.batch.Put(ikey, idata) + // add the access-chunkindex index for garbage collection + accessIdx := getAccessIdxKey(&index) + s.batch.Put(accessIdx, nil) s.lock.Unlock() select { @@ -813,16 +825,6 @@ func newMockGetDataFunc(mockStore *mock.NodeStore) func(addr Address) (data []by } } -func (s *LDBStore) updateAccessCnt(addr Address) { - - s.lock.Lock() - defer s.lock.Unlock() - - var index dpaDBIndex - s.tryAccessIdx(getIndexKey(addr), &index) // result_chn == nil, only update access cnt - -} - func (s *LDBStore) setCapacity(c uint64) { s.lock.Lock() defer s.lock.Unlock() From d9da5641e772c6061519079470094f2cd27ffa5e Mon Sep 17 00:00:00 2001 From: lash Date: Wed, 3 Oct 2018 11:10:06 +0200 Subject: [PATCH 5/8] swarm/storage: WIP Expand GC access idx to direct ptr params --- swarm/storage/ldbstore.go | 49 +++++++++++++++++++++++++++------------ 1 file changed, 34 insertions(+), 15 deletions(-) diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go index 80aedfa47d..11ee4d221f 100644 --- a/swarm/storage/ldbstore.go +++ b/swarm/storage/ldbstore.go @@ -61,7 +61,7 @@ var ( keyData = byte(6) keyDistanceCnt = byte(7) keySchema = []byte{8} - keyAccessIdx = []byte{9} // access to chunk data index, used by garbage collection in ascending order from first entry + keyGCIdx = []byte{9} // access to chunk data index, used by garbage collection in ascending order from first entry ) var ( @@ -226,14 +226,21 @@ func getDataKey(idx uint64, po uint8) []byte { return key } -func getAccessIdxKey(idx *dpaDBIndex) []byte { - key := make([]byte, 17) - key[0] = keyAccessIdx[0] +func getGCIdxKey(index *dpaDBIndex) []byte { + key := make([]byte, 9) + key[0] = keyGCIdx[0] binary.BigEndian.PutUint64(key[1:], index.Access) - binary.BigEndian.PutUint64(key[9:], index.Idx) return key } +func getGCIdxValue(index *dpaDBIndex, po uint8, addr Address) []byte { + val := make([]byte, 41) // po = 1, index.Index = 8, Address = 32 + val[0] = po + binary.BigEndian.PutUint64(val[1:], index.Idx) + copy(val[9:], addr) + return val +} + func encodeIndex(index *dpaDBIndex) []byte { data, _ := rlp.EncodeToBytes(index) return data @@ -266,7 +273,9 @@ func (s *LDBStore) collectGarbage(ratio float32) { garbage := []*gcItem{} gcnt := 0 + //maxGcnt := + //for ok := it.Seek([]byte{keyGCIdx}); ok && (gcnt < maxGcnt) for ok := it.Seek([]byte{keyIndex}); ok && (gcnt < maxGCitems) && (uint64(gcnt) < s.entryCnt); ok = it.Next() { itkey := it.Key() @@ -542,17 +551,20 @@ func (s *LDBStore) Delete(addr Address) { ikey := getIndexKey(addr) var indx dpaDBIndex - s.tryAccessIdx(ikey, &indx) + proximity := s.po(addr) + s.tryAccessIdx(ikey, proximity, &indx) - s.delete(indx.Idx, ikey, s.po(addr)) + s.delete(&indx, ikey, proximity) } -func (s *LDBStore) delete(idx uint64, idxKey []byte, po uint8) { +func (s *LDBStore) delete(idx *dpaDBIndex, idxKey []byte, po uint8) { metrics.GetOrRegisterCounter("ldbstore.delete", nil).Inc(1) batch := new(leveldb.Batch) batch.Delete(idxKey) - batch.Delete(getDataKey(idx, po)) + accessIdxKey := getAccessIdxKey(idx.Index, idx.Access) + batch.Delete(accessIdxKey) + batch.Delete(getDataKey(idx.Inde.Indexx, po)) s.entryCnt-- dbEntryCount.Dec(1) cntKey := make([]byte, 2) @@ -612,8 +624,9 @@ func (s *LDBStore) Put(ctx context.Context, chunk Chunk) error { s.batch.Put(ikey, idata) // add the access-chunkindex index for garbage collection - accessIdx := getAccessIdxKey(&index) - s.batch.Put(accessIdx, nil) + gcIdxKey := getGCIdxKey(&index) + gcIdxData := getGCIdxData(&index, po, chunk.Address) + s.batch.Put(gcIdxKey, gcIdxData) s.lock.Unlock() select { @@ -630,6 +643,7 @@ func (s *LDBStore) Put(ctx context.Context, chunk Chunk) error { } // force putting into db, does not check access index +// NOTE chunks put directly through this method will currently NOT be handled by garbage collection func (s *LDBStore) doPut(chunk Chunk, index *dpaDBIndex, po uint8) { data := s.encodeDataFunc(chunk) dkey := getDataKey(s.dataIdx, po) @@ -725,17 +739,22 @@ func newMockEncodeDataFunc(mockStore *mock.NodeStore) func(chunk Chunk) []byte { } // try to find index; if found, update access cnt and return true -func (s *LDBStore) tryAccessIdx(ikey []byte, index *dpaDBIndex) bool { +func (s *LDBStore) tryAccessIdx(ikey []byte, po uint8, index *dpaDBIndex) bool { idata, err := s.db.Get(ikey) if err != nil { return false } decodeIndex(idata, index) + oldGCIdxKey := getAccessIdxKey(&index) s.batch.Put(keyAccessCnt, U64ToBytes(s.accessCnt)) s.accessCnt++ index.Access = s.accessCnt idata = encodeIndex(index) s.batch.Put(ikey, idata) + newGCIdxKey := getGCIdxKey(&index) + newGCIdxData := getGCIdxData(&index, po, ikey) + s.batch.Delete(oldGCIdxKey) + s.batch.Put(newGCIdxKey, newGCIdxData) select { case s.batchesC <- struct{}{}: default: @@ -781,7 +800,8 @@ func (s *LDBStore) get(addr Address) (chunk *chunk, err error) { if s.closed { return nil, ErrDBClosed } - if s.tryAccessIdx(getIndexKey(addr), &indx) { + proximity := s.po(addr) + if s.tryAccessIdx(getIndexKey(addr), proximity, &indx) { var data []byte if s.getDataFunc != nil { // if getDataFunc is defined, use it to retrieve the chunk data @@ -792,13 +812,12 @@ func (s *LDBStore) get(addr Address) (chunk *chunk, err error) { } } else { // default DbStore functionality to retrieve chunk data - proximity := s.po(addr) datakey := getDataKey(indx.Idx, proximity) data, err = s.db.Get(datakey) log.Trace("ldbstore.get retrieve", "key", addr, "indexkey", indx.Idx, "datakey", fmt.Sprintf("%x", datakey), "proximity", proximity) if err != nil { log.Trace("ldbstore.get chunk found but could not be accessed", "key", addr, "err", err) - s.delete(indx.Idx, getIndexKey(addr), s.po(addr)) + s.delete(&indx, getIndexKey(addr), s.po(addr)) return } } From 9e68d6d46fc3c0a40cc3317509cc8d9698990bd9 Mon Sep 17 00:00:00 2001 From: lash Date: Thu, 4 Oct 2018 08:26:58 +0200 Subject: [PATCH 6/8] swarm/storage: GC pass test with new access index --- swarm/storage/ldbstore.go | 73 ++++++++++++++++++++-------------- swarm/storage/ldbstore_test.go | 9 +++-- 2 files changed, 49 insertions(+), 33 deletions(-) diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go index 11ee4d221f..3f67cd409d 100644 --- a/swarm/storage/ldbstore.go +++ b/swarm/storage/ldbstore.go @@ -45,7 +45,7 @@ import ( const ( gcArrayFreeRatio = 0.1 - maxGCitems = 5000 // max number of items to be gc'd per call to collectGarbage() + maxGCItems = 5000 // max number of items to be gc'd per call to collectGarbage() ) var ( @@ -61,7 +61,7 @@ var ( keyData = byte(6) keyDistanceCnt = byte(7) keySchema = []byte{8} - keyGCIdx = []byte{9} // access to chunk data index, used by garbage collection in ascending order from first entry + keyGCIdx = byte(9) // access to chunk data index, used by garbage collection in ascending order from first entry ) var ( @@ -69,7 +69,7 @@ var ( ) type gcItem struct { - idx uint64 + idx *dpaDBIndex value uint64 idxKey []byte po uint8 @@ -228,7 +228,7 @@ func getDataKey(idx uint64, po uint8) []byte { func getGCIdxKey(index *dpaDBIndex) []byte { key := make([]byte, 9) - key[0] = keyGCIdx[0] + key[0] = keyGCIdx binary.BigEndian.PutUint64(key[1:], index.Access) return key } @@ -241,6 +241,16 @@ func getGCIdxValue(index *dpaDBIndex, po uint8, addr Address) []byte { return val } +func parseGCIdxEntry(accessCnt []byte, val []byte) (index *dpaDBIndex, po uint8, addr Address) { + index = &dpaDBIndex{ + Idx: binary.BigEndian.Uint64(val[1:]), + Access: binary.BigEndian.Uint64(accessCnt), + } + po = val[0] + addr = val[9:] + return +} + func encodeIndex(index *dpaDBIndex) []byte { data, _ := rlp.EncodeToBytes(index) return data @@ -272,32 +282,32 @@ func (s *LDBStore) collectGarbage(ratio float32) { defer it.Release() garbage := []*gcItem{} - gcnt := 0 - //maxGcnt := + var gcnt uint64 + var maxGcnt uint64 + if s.entryCnt >= maxGCItems { + maxGcnt = maxGCItems * gcArrayFreeRatio + } else { + maxGcnt = uint64(float64(s.entryCnt) * gcArrayFreeRatio) + } - //for ok := it.Seek([]byte{keyGCIdx}); ok && (gcnt < maxGcnt) - for ok := it.Seek([]byte{keyIndex}); ok && (gcnt < maxGCitems) && (uint64(gcnt) < s.entryCnt); ok = it.Next() { + for ok := it.Seek([]byte{keyGCIdx}); ok && (gcnt < maxGcnt); ok = it.Next() { itkey := it.Key() - if (itkey == nil) || (itkey[0] != keyIndex) { + if (itkey == nil) || (itkey[0] != keyGCIdx) { break } - // it.Key() contents change on next call to it.Next(), so we must copy it - key := make([]byte, len(it.Key())) - copy(key, it.Key()) - val := it.Value() + index, po, hash := parseGCIdxEntry(itkey[1:], val) + keyIdx := make([]byte, 33) + keyIdx[0] = keyIndex + copy(keyIdx[1:], hash) - var index dpaDBIndex - - hash := key[1:] - decodeIndex(val, &index) - po := s.po(hash) + log.Trace("parse gc", "index", index, "po", po, "hash", hash) gci := &gcItem{ - idxKey: key, - idx: index.Idx, + idxKey: keyIdx, + idx: index, value: index.Access, // the smaller, the more likely to be gc'd. see sort comparator below. po: po, } @@ -308,8 +318,8 @@ func (s *LDBStore) collectGarbage(ratio float32) { sort.Slice(garbage[:gcnt], func(i, j int) bool { return garbage[i].value < garbage[j].value }) - cutoff := int(float32(gcnt) * ratio) - metrics.GetOrRegisterCounter("ldbstore.collectgarbage.delete", nil).Inc(int64(cutoff)) + cutoff := int(float32(gcnt)) //* ratio) + metrics.GetOrRegisterCounter("ldbstore.collectgarbage.delete", nil).Inc(int64(maxGcnt)) for i := 0; i < cutoff; i++ { s.delete(garbage[i].idx, garbage[i].idxKey, garbage[i].po) @@ -492,7 +502,7 @@ func (s *LDBStore) Cleanup(f func(*chunk) bool) { // if chunk is to be removed if f(c) { log.Warn("chunk for cleanup", "key", fmt.Sprintf("%x", key), "ck", fmt.Sprintf("%x", ck), "dkey", fmt.Sprintf("%x", datakey), "dataidx", index.Idx, "po", po, "len data", len(data), "len sdata", len(c.sdata), "size", cs) - s.delete(index.Idx, getIndexKey(key[1:]), po) + s.delete(&index, getIndexKey(key[1:]), po) removed++ errorsFound++ } @@ -562,9 +572,12 @@ func (s *LDBStore) delete(idx *dpaDBIndex, idxKey []byte, po uint8) { batch := new(leveldb.Batch) batch.Delete(idxKey) - accessIdxKey := getAccessIdxKey(idx.Index, idx.Access) - batch.Delete(accessIdxKey) - batch.Delete(getDataKey(idx.Inde.Indexx, po)) + gcIdxKey := getGCIdxKey(idx) + batch.Delete(gcIdxKey) + batch.Delete(getDataKey(idx.Idx, po)) + if s.entryCnt == 0 { + panic("") + } s.entryCnt-- dbEntryCount.Dec(1) cntKey := make([]byte, 2) @@ -625,7 +638,7 @@ func (s *LDBStore) Put(ctx context.Context, chunk Chunk) error { // add the access-chunkindex index for garbage collection gcIdxKey := getGCIdxKey(&index) - gcIdxData := getGCIdxData(&index, po, chunk.Address) + gcIdxData := getGCIdxValue(&index, po, chunk.Address()) s.batch.Put(gcIdxKey, gcIdxData) s.lock.Unlock() @@ -745,14 +758,14 @@ func (s *LDBStore) tryAccessIdx(ikey []byte, po uint8, index *dpaDBIndex) bool { return false } decodeIndex(idata, index) - oldGCIdxKey := getAccessIdxKey(&index) + oldGCIdxKey := getGCIdxKey(index) s.batch.Put(keyAccessCnt, U64ToBytes(s.accessCnt)) s.accessCnt++ index.Access = s.accessCnt idata = encodeIndex(index) s.batch.Put(ikey, idata) - newGCIdxKey := getGCIdxKey(&index) - newGCIdxData := getGCIdxData(&index, po, ikey) + newGCIdxKey := getGCIdxKey(index) + newGCIdxData := getGCIdxValue(index, po, ikey) s.batch.Delete(oldGCIdxKey) s.batch.Put(newGCIdxKey, newGCIdxData) select { diff --git a/swarm/storage/ldbstore_test.go b/swarm/storage/ldbstore_test.go index 75b5d6aa95..9bd17ba5e1 100644 --- a/swarm/storage/ldbstore_test.go +++ b/swarm/storage/ldbstore_test.go @@ -385,8 +385,10 @@ func TestLDBStoreAddRemove(t *testing.T) { // TestLDBStoreRemoveThenCollectGarbage tests that we can delete chunks and that we can trigger garbage collection func TestLDBStoreRemoveThenCollectGarbage(t *testing.T) { - capacity := 11 - surplus := 4 + //capacity := 11 + //surplus := 4 + capacity := 10000 + surplus := 10000 ldb, cleanup := newLDBStore(t) ldb.setCapacity(uint64(capacity)) @@ -423,7 +425,8 @@ func TestLDBStoreRemoveThenCollectGarbage(t *testing.T) { cleanup() ldb, cleanup = newLDBStore(t) - capacity = 10 + //capacity = 10 + capacity = 10000 ldb.setCapacity(uint64(capacity)) defer cleanup() From f72602631221679f99e89b28007613a335a00566 Mon Sep 17 00:00:00 2001 From: lash Date: Thu, 4 Oct 2018 08:45:51 +0200 Subject: [PATCH 7/8] swarm/storage: Add GC test over and under maxGCItems --- swarm/storage/ldbstore.go | 3 --- swarm/storage/ldbstore_test.go | 45 ++++++++++++++++++++++++++-------- 2 files changed, 35 insertions(+), 13 deletions(-) diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go index 3f67cd409d..c4fe1ce24d 100644 --- a/swarm/storage/ldbstore.go +++ b/swarm/storage/ldbstore.go @@ -575,9 +575,6 @@ func (s *LDBStore) delete(idx *dpaDBIndex, idxKey []byte, po uint8) { gcIdxKey := getGCIdxKey(idx) batch.Delete(gcIdxKey) batch.Delete(getDataKey(idx.Idx, po)) - if s.entryCnt == 0 { - panic("") - } s.entryCnt-- dbEntryCount.Dec(1) cntKey := make([]byte, 2) diff --git a/swarm/storage/ldbstore_test.go b/swarm/storage/ldbstore_test.go index 9bd17ba5e1..9acac83ebd 100644 --- a/swarm/storage/ldbstore_test.go +++ b/swarm/storage/ldbstore_test.go @@ -22,6 +22,8 @@ import ( "fmt" "io/ioutil" "os" + "strconv" + "strings" "testing" "time" @@ -296,11 +298,30 @@ func TestLDBStoreWithoutCollectGarbage(t *testing.T) { } } +func TestLDBStoreCollectGarbage(t *testing.T) { + + cap := maxGCItems / 2 + t.Run(fmt.Sprintf("A/%d/%d", cap, cap*4), testLDBStoreCollectGarbage) + t.Run(fmt.Sprintf("B/%d/%d", cap, cap*4), testLDBStoreRemoveThenCollectGarbage) + + cap = maxGCItems * 2 + t.Run(fmt.Sprintf("A/%d/%d", cap, cap*4), testLDBStoreCollectGarbage) + t.Run(fmt.Sprintf("B/%d/%d", cap, cap*4), testLDBStoreRemoveThenCollectGarbage) +} + // TestLDBStoreCollectGarbage tests that we can put more chunks than LevelDB's capacity, and // retrieve only some of them, because garbage collection must have cleared some of them -func TestLDBStoreCollectGarbage(t *testing.T) { - capacity := 500 - n := 2000 +func testLDBStoreCollectGarbage(t *testing.T) { + + params := strings.Split(t.Name(), "/") + capacity, err := strconv.Atoi(params[2]) + if err != nil { + t.Fatal(err) + } + n, err := strconv.Atoi(params[3]) + if err != nil { + t.Fatal(err) + } ldb, cleanup := newLDBStore(t) ldb.setCapacity(uint64(capacity)) @@ -384,11 +405,17 @@ func TestLDBStoreAddRemove(t *testing.T) { } // TestLDBStoreRemoveThenCollectGarbage tests that we can delete chunks and that we can trigger garbage collection -func TestLDBStoreRemoveThenCollectGarbage(t *testing.T) { - //capacity := 11 - //surplus := 4 - capacity := 10000 - surplus := 10000 +func testLDBStoreRemoveThenCollectGarbage(t *testing.T) { + + params := strings.Split(t.Name(), "/") + capacity, err := strconv.Atoi(params[2]) + if err != nil { + t.Fatal(err) + } + surplus, err := strconv.Atoi(params[3]) + if err != nil { + t.Fatal(err) + } ldb, cleanup := newLDBStore(t) ldb.setCapacity(uint64(capacity)) @@ -425,8 +452,6 @@ func TestLDBStoreRemoveThenCollectGarbage(t *testing.T) { cleanup() ldb, cleanup = newLDBStore(t) - //capacity = 10 - capacity = 10000 ldb.setCapacity(uint64(capacity)) defer cleanup() From 9886921a7e866143e41e757e8c06359e40abeff4 Mon Sep 17 00:00:00 2001 From: lash Date: Thu, 4 Oct 2018 10:36:36 +0200 Subject: [PATCH 8/8] swarm/storage: Add sane gc test params + test accesscnt!=indexcnt --- swarm/storage/ldbstore.go | 14 ++++++----- swarm/storage/ldbstore_test.go | 45 +++++++++++++++++++++++++++++++++- 2 files changed, 52 insertions(+), 7 deletions(-) diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go index c4fe1ce24d..c783570674 100644 --- a/swarm/storage/ldbstore.go +++ b/swarm/storage/ldbstore.go @@ -170,6 +170,13 @@ func NewLDBStore(params *LDBStoreParams) (s *LDBStore, err error) { return s, nil } +func (s *LDBStore) getGCCount() uint64 { + if s.entryCnt >= maxGCItems { + return maxGCItems * gcArrayFreeRatio + } + return uint64(float64(s.entryCnt) * gcArrayFreeRatio) +} + // NewMockDbStore creates a new instance of DbStore with // mockStore set to a provided value. If mockStore argument is nil, // this function behaves exactly as NewDbStore. @@ -283,12 +290,7 @@ func (s *LDBStore) collectGarbage(ratio float32) { garbage := []*gcItem{} var gcnt uint64 - var maxGcnt uint64 - if s.entryCnt >= maxGCItems { - maxGcnt = maxGCItems * gcArrayFreeRatio - } else { - maxGcnt = uint64(float64(s.entryCnt) * gcArrayFreeRatio) - } + maxGcnt := s.getGCCount() for ok := it.Seek([]byte{keyGCIdx}); ok && (gcnt < maxGcnt); ok = it.Next() { itkey := it.Key() diff --git a/swarm/storage/ldbstore_test.go b/swarm/storage/ldbstore_test.go index 9acac83ebd..6e8df2e42a 100644 --- a/swarm/storage/ldbstore_test.go +++ b/swarm/storage/ldbstore_test.go @@ -312,7 +312,6 @@ func TestLDBStoreCollectGarbage(t *testing.T) { // TestLDBStoreCollectGarbage tests that we can put more chunks than LevelDB's capacity, and // retrieve only some of them, because garbage collection must have cleared some of them func testLDBStoreCollectGarbage(t *testing.T) { - params := strings.Split(t.Name(), "/") capacity, err := strconv.Atoi(params[2]) if err != nil { @@ -483,3 +482,47 @@ func testLDBStoreRemoveThenCollectGarbage(t *testing.T) { } } } + +// TestLDBStoreCollectGarbageAccessUnlikeIndex tests garbage collection where accesscount differs from indexcount +func TestLDBStoreCollectGarbageAccessUnlikeIndex(t *testing.T) { + + capacity := maxGCItems + n := capacity - 1 + + ldb, cleanup := newLDBStore(t) + ldb.setCapacity(uint64(capacity)) + defer cleanup() + + chunks, err := mputRandomChunks(ldb, n, int64(ch.DefaultSize)) + if err != nil { + t.Fatal(err.Error()) + } + log.Info("ldbstore", "entrycnt", ldb.entryCnt, "accesscnt", ldb.accessCnt) + + // set first added capacity/2 chunks to highest accesscount + for i := 0; i < capacity/2; i++ { + ldb.Get(context.TODO(), chunks[i].Address()) + } + _, err = mputRandomChunks(ldb, 2, int64(ch.DefaultSize)) + if err != nil { + t.Fatal(err.Error()) + } + + // wait for garbage collection to kick in on the responsible actor + time.Sleep(1 * time.Second) + + var missing int + for _, ch := range chunks[:capacity/2] { + ret, err := ldb.Get(context.Background(), ch.Address()) + if err == ErrChunkNotFound || err == ldberrors.ErrNotFound { + t.Fatalf("fail find chunk %s: %v", ch.Address(), err) + } + + if !bytes.Equal(ret.Data(), ch.Data()) { + t.Fatal("expected to get the same data back, but got smth else") + } + log.Trace("got back chunk", "chunk", ret) + } + + log.Info("ldbstore", "total", n, "missing", missing, "entrycnt", ldb.entryCnt, "accesscnt", ldb.accessCnt) +}