From 5c8bd94513a06fe6e74cb5636807cc599bf7d81d Mon Sep 17 00:00:00 2001 From: lash Date: Fri, 28 Sep 2018 22:18:55 +0200 Subject: [PATCH 1/5] swarm/storage: Add ordered garbage collection test for ldb --- swarm/storage/ldbstore_test.go | 106 +++++++++++++++++++++++++++++++++ 1 file changed, 106 insertions(+) diff --git a/swarm/storage/ldbstore_test.go b/swarm/storage/ldbstore_test.go index 75b5d6aa95..46a7887de1 100644 --- a/swarm/storage/ldbstore_test.go +++ b/swarm/storage/ldbstore_test.go @@ -19,6 +19,7 @@ package storage import ( "bytes" "context" + "encoding/binary" "fmt" "io/ioutil" "os" @@ -340,6 +341,111 @@ func TestLDBStoreCollectGarbage(t *testing.T) { log.Info("ldbstore", "total", n, "missing", missing, "entrycnt", ldb.entryCnt, "accesscnt", ldb.accessCnt) } +// TestLDBStoreCollectGarbageOrdered checks if the most recently added chunks according to capacity are left after garbage collection +func TestLDBStoreCollectGarbageOrdered(t *testing.T) { + capacity := 10000 + chunkCount := 20000 + hasher := MakeHashFunc(DefaultHash)() + + // four byte value incremented sequentially as chunk data (one chunk has 1024 values) + var byteValue uint32 = 0 + + // lru buffer + cursor := 0 + buf := make([][ch.DefaultSize]byte, capacity) + + // record keeping + madeChunks := make([]Chunk, capacity) + madeAddrs := make([]Address, capacity) + matchAddrs := make([]Address, capacity) + + // needed for hashing (all chunks are full chunks here) + meta := make([]byte, 8) + binary.LittleEndian.PutUint64(meta, uint64(ch.DefaultSize)) + + // the store + store, cleanup := newLDBStore(t) + store.setCapacity(uint64(capacity)) + defer cleanup() + + for i := 0; i < chunkCount; i++ { + hasher.ResetWithLength(meta) + + // write (same) sequential data to buffer and hasher + for j := 0; j < ch.DefaultSize; j += 4 { // uint32 intervals + byteValueByte := [4]byte{} + binary.LittleEndian.PutUint32(byteValueByte[:], byteValue) + copy(buf[cursor][j:], byteValueByte[:]) + hasher.Write(byteValueByte[:]) + byteValue++ + } + + // create chunk, add to record keeping and put chunk in store + madeChunks[cursor] = NewChunk(hasher.Sum(nil), buf[cursor][:]) + madeAddrs[cursor] = madeChunks[cursor].Address() + matchAddrs[cursor] = madeChunks[cursor].Address() + savedChunk, err := mput(store, 1, func(n int64) Chunk { return madeChunks[cursor] }) + if err != nil { + t.Fatalf("store put fail: %v", err) + } else if !bytes.Equal(savedChunk[0].Address(), madeAddrs[cursor]) { // probably redundant but let's be careful for now + t.Fatalf("saved addr mismatch %x/%x: %v", savedChunk[0].Address(), madeAddrs[cursor], err) + } + + log.Debug("putting", "address", matchAddrs[cursor]) + + // wrap cursor on capacity. + cursor++ + cursor %= capacity + } + + log.Info("chunks put, sir", "cursor", cursor, "lastvalue", byteValue, "count", len(madeAddrs)) + + // madeAddrs should now contain only the last added chunks. + var matches uint64 + var seq uint64 + err := mget(store, madeAddrs, func(h Address, retrievedChunk Chunk) error { + oldMatch := matches + seq++ + + // matchedAddr originally equal to madeAddr + // when an element is found, remove it + for i, matchedAddr := range matchAddrs { + if bytes.Equal(matchedAddr, h) { + matchAddrs[i] = matchAddrs[len(matchAddrs)-1] + + // last one needs special treatment + if len(matchAddrs) == 1 { + matchAddrs = []Address{} + } else { + matchAddrs = matchAddrs[:len(matchAddrs)-1] + } + + matches++ + log.Debug("found match", "addr", h, "match", matches, "seq", seq, "left", len(matchAddrs)) + break + } + } + + // we don't seem to reach this, which suggests retrieve fails are handled further up...? + if oldMatch == matches { + log.Warn("no match", "addr", h, "left", len(matchAddrs)) + return fmt.Errorf("not found (%d): %x", matches, h) + } + + return nil + }) + + // check the retrieve errors + if err != nil { + t.Fatalf("matches %d/%d, retrieve fail: %v", matches, capacity, err) + } + + // if all elements are found the array should be empty + if len(matchAddrs) > 0 { + t.Fatalf("expected 0 chunks in match array, have %d", len(matchAddrs)) + } +} + // TestLDBStoreAddRemove tests that we can put and then delete a given chunk func TestLDBStoreAddRemove(t *testing.T) { ldb, cleanup := newLDBStore(t) From 50df43c67054fef118060ef84b1cecb5caefee15 Mon Sep 17 00:00:00 2001 From: lash Date: Mon, 1 Oct 2018 09:45:54 +0200 Subject: [PATCH 2/5] swarm/storage: Factor in accesscount in ordered gc test Amended access count increment in tryAccessIdx to only affect requested chunk and not the offset of the next added chunk --- swarm/storage/ldbstore.go | 6 +++-- swarm/storage/ldbstore_test.go | 48 ++++++++++++++++++++++------------ 2 files changed, 36 insertions(+), 18 deletions(-) diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go index bde627394e..94c7428512 100644 --- a/swarm/storage/ldbstore.go +++ b/swarm/storage/ldbstore.go @@ -719,14 +719,16 @@ func (s *LDBStore) tryAccessIdx(ikey []byte, index *dpaDBIndex) bool { } decodeIndex(idata, index) s.batch.Put(keyAccessCnt, U64ToBytes(s.accessCnt)) - s.accessCnt++ - index.Access = s.accessCnt + // presumably, we only want to increase the access count of the chunk in question, and not the offset of any future ones? + //s.accessCnt++ + index.Access = s.accessCnt + 1 idata = encodeIndex(index) s.batch.Put(ikey, idata) select { case s.batchesC <- struct{}{}: default: } + log.Trace("tryaccessidx", "addr", fmt.Sprintf("%x", ikey[1:]), "indexdata", index, "data", idata) return true } diff --git a/swarm/storage/ldbstore_test.go b/swarm/storage/ldbstore_test.go index 46a7887de1..1e0ca887d1 100644 --- a/swarm/storage/ldbstore_test.go +++ b/swarm/storage/ldbstore_test.go @@ -345,7 +345,9 @@ func TestLDBStoreCollectGarbage(t *testing.T) { func TestLDBStoreCollectGarbageOrdered(t *testing.T) { capacity := 10000 chunkCount := 20000 + gcThreshold := int(maxGCitems * gcArrayFreeRatio) hasher := MakeHashFunc(DefaultHash)() + writeBatchTolerance := 128 // according to log ldb seems to write in batches of 6 // four byte value incremented sequentially as chunk data (one chunk has 1024 values) var byteValue uint32 = 0 @@ -355,9 +357,10 @@ func TestLDBStoreCollectGarbageOrdered(t *testing.T) { buf := make([][ch.DefaultSize]byte, capacity) // record keeping - madeChunks := make([]Chunk, capacity) - madeAddrs := make([]Address, capacity) - matchAddrs := make([]Address, capacity) + chunkSaveCount := gcThreshold - writeBatchTolerance + madeChunks := make([]Chunk, chunkSaveCount) + madeAddrs := make([]Address, chunkSaveCount) + matchAddrs := make([]Address, chunkSaveCount) // needed for hashing (all chunks are full chunks here) meta := make([]byte, 8) @@ -368,6 +371,7 @@ func TestLDBStoreCollectGarbageOrdered(t *testing.T) { store.setCapacity(uint64(capacity)) defer cleanup() + log.Info("gc ordered test", "gcthreshold", gcThreshold, "savecount", chunkSaveCount, "cap", capacity, "count", chunkCount) for i := 0; i < chunkCount; i++ { hasher.ResetWithLength(meta) @@ -380,25 +384,37 @@ func TestLDBStoreCollectGarbageOrdered(t *testing.T) { byteValue++ } - // create chunk, add to record keeping and put chunk in store - madeChunks[cursor] = NewChunk(hasher.Sum(nil), buf[cursor][:]) - madeAddrs[cursor] = madeChunks[cursor].Address() - matchAddrs[cursor] = madeChunks[cursor].Address() - savedChunk, err := mput(store, 1, func(n int64) Chunk { return madeChunks[cursor] }) + // create and put chunk + newChunk := NewChunk(hasher.Sum(nil), buf[cursor][:]) + _, err := mput(store, 1, func(n int64) Chunk { return newChunk }) if err != nil { t.Fatalf("store put fail: %v", err) - } else if !bytes.Equal(savedChunk[0].Address(), madeAddrs[cursor]) { // probably redundant but let's be careful for now - t.Fatalf("saved addr mismatch %x/%x: %v", savedChunk[0].Address(), madeAddrs[cursor], err) } - log.Debug("putting", "address", matchAddrs[cursor]) + log.Trace("putting", "address", newChunk.Address(), "i", i) + + // add to record keeping if it's among the last chunkSaveCount chunks + if i > chunkCount-chunkSaveCount { + madeChunks[cursor] = newChunk + madeAddrs[cursor] = madeChunks[cursor].Address() + matchAddrs[cursor] = madeChunks[cursor].Address() + + // get the chunk at least gcThreshold times. That should put the chunk access count comfortable above the limit of any previously added chunks (and give time to flush the db batch writes, too) + for i := 0; i < gcThreshold; i++ { + log.Trace("accessing", "address", madeChunks[cursor].Address()) + store.Get(context.TODO(), madeChunks[cursor].Address()) + } + + cursor++ + + // wrap cursor on capacity. + //cursor %= capacity + + } - // wrap cursor on capacity. - cursor++ - cursor %= capacity } - log.Info("chunks put, sir", "cursor", cursor, "lastvalue", byteValue, "count", len(madeAddrs)) + log.Info("chunks put, sir", "cursor", cursor, "capacity", capacity, "lastvalue", byteValue, "count", len(madeAddrs)) // madeAddrs should now contain only the last added chunks. var matches uint64 @@ -437,7 +453,7 @@ func TestLDBStoreCollectGarbageOrdered(t *testing.T) { // check the retrieve errors if err != nil { - t.Fatalf("matches %d/%d, retrieve fail: %v", matches, capacity, err) + t.Fatalf("matches %d/%d, retrieve fail: %v", matches, chunkSaveCount, err) } // if all elements are found the array should be empty From 9f7b0f5813bae2f243ef56fb758581f49bd0c936 Mon Sep 17 00:00:00 2001 From: lash Date: Mon, 1 Oct 2018 09:50:26 +0200 Subject: [PATCH 3/5] swarm/storage: Add const dependent test params to gc ordered test --- swarm/storage/ldbstore_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/swarm/storage/ldbstore_test.go b/swarm/storage/ldbstore_test.go index 1e0ca887d1..59fcc6f509 100644 --- a/swarm/storage/ldbstore_test.go +++ b/swarm/storage/ldbstore_test.go @@ -343,9 +343,9 @@ func TestLDBStoreCollectGarbage(t *testing.T) { // TestLDBStoreCollectGarbageOrdered checks if the most recently added chunks according to capacity are left after garbage collection func TestLDBStoreCollectGarbageOrdered(t *testing.T) { - capacity := 10000 - chunkCount := 20000 gcThreshold := int(maxGCitems * gcArrayFreeRatio) + capacity := maxGCitems * 2 + chunkCount := capacity * 2 hasher := MakeHashFunc(DefaultHash)() writeBatchTolerance := 128 // according to log ldb seems to write in batches of 6 From 759fdeb65beadb4355f5c9fc11a692bf90b3dce6 Mon Sep 17 00:00:00 2001 From: lash Date: Mon, 1 Oct 2018 10:44:23 +0200 Subject: [PATCH 4/5] swarm/storage: Revert access count change --- swarm/storage/ldbstore.go | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go index 94c7428512..2bf121dc64 100644 --- a/swarm/storage/ldbstore.go +++ b/swarm/storage/ldbstore.go @@ -719,9 +719,8 @@ func (s *LDBStore) tryAccessIdx(ikey []byte, index *dpaDBIndex) bool { } decodeIndex(idata, index) s.batch.Put(keyAccessCnt, U64ToBytes(s.accessCnt)) - // presumably, we only want to increase the access count of the chunk in question, and not the offset of any future ones? - //s.accessCnt++ - index.Access = s.accessCnt + 1 + s.accessCnt++ + index.Access = s.accessCnt idata = encodeIndex(index) s.batch.Put(ikey, idata) select { From f1f7ba92e201f135c81100091832337bd046b8b4 Mon Sep 17 00:00:00 2001 From: lash Date: Tue, 2 Oct 2018 12:08:00 +0200 Subject: [PATCH 5/5] swarm/storage: Remove premature iterator termination condition in gc --- swarm/storage/ldbstore.go | 8 +- swarm/storage/ldbstore_test.go | 129 ++------------------------------- 2 files changed, 11 insertions(+), 126 deletions(-) diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go index 2bf121dc64..831747b688 100644 --- a/swarm/storage/ldbstore.go +++ b/swarm/storage/ldbstore.go @@ -248,7 +248,7 @@ func decodeData(addr Address, data []byte) (*chunk, error) { } func (s *LDBStore) collectGarbage(ratio float32) { - log.Trace("collectGarbage", "ratio", ratio) + log.Trace("collectGarbage", "ratio", ratio, "entrycnt", s.entryCnt) metrics.GetOrRegisterCounter("ldbstore.collectgarbage", nil).Inc(1) @@ -258,7 +258,7 @@ func (s *LDBStore) collectGarbage(ratio float32) { garbage := []*gcItem{} gcnt := 0 - for ok := it.Seek([]byte{keyIndex}); ok && (gcnt < maxGCitems) && (uint64(gcnt) < s.entryCnt); ok = it.Next() { + for ok := it.Seek([]byte{keyIndex}); ok && (uint64(gcnt) < s.entryCnt); ok = it.Next() { itkey := it.Key() if (itkey == nil) || (itkey[0] != keyIndex) { @@ -290,7 +290,11 @@ func (s *LDBStore) collectGarbage(ratio float32) { sort.Slice(garbage[:gcnt], func(i, j int) bool { return garbage[i].value < garbage[j].value }) + if gcnt > maxGCitems { + gcnt = maxGCitems + } cutoff := int(float32(gcnt) * ratio) + metrics.GetOrRegisterCounter("ldbstore.collectgarbage.delete", nil).Inc(int64(cutoff)) for i := 0; i < cutoff; i++ { diff --git a/swarm/storage/ldbstore_test.go b/swarm/storage/ldbstore_test.go index 59fcc6f509..dc675c3824 100644 --- a/swarm/storage/ldbstore_test.go +++ b/swarm/storage/ldbstore_test.go @@ -341,127 +341,6 @@ func TestLDBStoreCollectGarbage(t *testing.T) { log.Info("ldbstore", "total", n, "missing", missing, "entrycnt", ldb.entryCnt, "accesscnt", ldb.accessCnt) } -// TestLDBStoreCollectGarbageOrdered checks if the most recently added chunks according to capacity are left after garbage collection -func TestLDBStoreCollectGarbageOrdered(t *testing.T) { - gcThreshold := int(maxGCitems * gcArrayFreeRatio) - capacity := maxGCitems * 2 - chunkCount := capacity * 2 - hasher := MakeHashFunc(DefaultHash)() - writeBatchTolerance := 128 // according to log ldb seems to write in batches of 6 - - // four byte value incremented sequentially as chunk data (one chunk has 1024 values) - var byteValue uint32 = 0 - - // lru buffer - cursor := 0 - buf := make([][ch.DefaultSize]byte, capacity) - - // record keeping - chunkSaveCount := gcThreshold - writeBatchTolerance - madeChunks := make([]Chunk, chunkSaveCount) - madeAddrs := make([]Address, chunkSaveCount) - matchAddrs := make([]Address, chunkSaveCount) - - // needed for hashing (all chunks are full chunks here) - meta := make([]byte, 8) - binary.LittleEndian.PutUint64(meta, uint64(ch.DefaultSize)) - - // the store - store, cleanup := newLDBStore(t) - store.setCapacity(uint64(capacity)) - defer cleanup() - - log.Info("gc ordered test", "gcthreshold", gcThreshold, "savecount", chunkSaveCount, "cap", capacity, "count", chunkCount) - for i := 0; i < chunkCount; i++ { - hasher.ResetWithLength(meta) - - // write (same) sequential data to buffer and hasher - for j := 0; j < ch.DefaultSize; j += 4 { // uint32 intervals - byteValueByte := [4]byte{} - binary.LittleEndian.PutUint32(byteValueByte[:], byteValue) - copy(buf[cursor][j:], byteValueByte[:]) - hasher.Write(byteValueByte[:]) - byteValue++ - } - - // create and put chunk - newChunk := NewChunk(hasher.Sum(nil), buf[cursor][:]) - _, err := mput(store, 1, func(n int64) Chunk { return newChunk }) - if err != nil { - t.Fatalf("store put fail: %v", err) - } - - log.Trace("putting", "address", newChunk.Address(), "i", i) - - // add to record keeping if it's among the last chunkSaveCount chunks - if i > chunkCount-chunkSaveCount { - madeChunks[cursor] = newChunk - madeAddrs[cursor] = madeChunks[cursor].Address() - matchAddrs[cursor] = madeChunks[cursor].Address() - - // get the chunk at least gcThreshold times. That should put the chunk access count comfortable above the limit of any previously added chunks (and give time to flush the db batch writes, too) - for i := 0; i < gcThreshold; i++ { - log.Trace("accessing", "address", madeChunks[cursor].Address()) - store.Get(context.TODO(), madeChunks[cursor].Address()) - } - - cursor++ - - // wrap cursor on capacity. - //cursor %= capacity - - } - - } - - log.Info("chunks put, sir", "cursor", cursor, "capacity", capacity, "lastvalue", byteValue, "count", len(madeAddrs)) - - // madeAddrs should now contain only the last added chunks. - var matches uint64 - var seq uint64 - err := mget(store, madeAddrs, func(h Address, retrievedChunk Chunk) error { - oldMatch := matches - seq++ - - // matchedAddr originally equal to madeAddr - // when an element is found, remove it - for i, matchedAddr := range matchAddrs { - if bytes.Equal(matchedAddr, h) { - matchAddrs[i] = matchAddrs[len(matchAddrs)-1] - - // last one needs special treatment - if len(matchAddrs) == 1 { - matchAddrs = []Address{} - } else { - matchAddrs = matchAddrs[:len(matchAddrs)-1] - } - - matches++ - log.Debug("found match", "addr", h, "match", matches, "seq", seq, "left", len(matchAddrs)) - break - } - } - - // we don't seem to reach this, which suggests retrieve fails are handled further up...? - if oldMatch == matches { - log.Warn("no match", "addr", h, "left", len(matchAddrs)) - return fmt.Errorf("not found (%d): %x", matches, h) - } - - return nil - }) - - // check the retrieve errors - if err != nil { - t.Fatalf("matches %d/%d, retrieve fail: %v", matches, chunkSaveCount, err) - } - - // if all elements are found the array should be empty - if len(matchAddrs) > 0 { - t.Fatalf("expected 0 chunks in match array, have %d", len(matchAddrs)) - } -} - // TestLDBStoreAddRemove tests that we can put and then delete a given chunk func TestLDBStoreAddRemove(t *testing.T) { ldb, cleanup := newLDBStore(t) @@ -507,8 +386,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)) @@ -545,7 +426,7 @@ func TestLDBStoreRemoveThenCollectGarbage(t *testing.T) { cleanup() ldb, cleanup = newLDBStore(t) - capacity = 10 + //capacity = 10000 ldb.setCapacity(uint64(capacity)) defer cleanup()