forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 10:22:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

swarm/storage: Edit delete in test to avoid accesscnt increment

Browse source
This commit is contained in:
lash 2018-10-10 20:28:54 +02:00
parent c098932dc4
commit bfaf481765
2 changed files with 82 additions and 68 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

72
swarm/storage/ldbstore.go
View file

@ -17,7 +17,8 @@
// disk storage layer for the package bzz
// DbStore implements the ChunkStore interface and is used by the FileStore as
// persistent storage of chunks
// it implements purging based on access count allowing for external control of // max capacity
// it implements purging based on access count allowing for external control of
// max capacity
package storage
@ -45,6 +46,10 @@ const (
defaultGCRatio = 10
defaultMaxGCRound = 10000
defaultMaxGCBatch = 5000
wEntryCnt = 1 << 0
wIndexCnt = 1 << 1
wAccessCnt = 1 << 2
)
var (
@ -116,9 +121,8 @@ type LDBStore struct {
closed bool
batch *dbBatch
lock sync.RWMutex
//axxLock sync.RWMutex
quit chan struct{}
gc *garbage
quit chan struct{}
gc *garbage
// Functions encodeDataFunc is used to bypass
// the default functionality of DbStore with
@ -360,15 +364,7 @@ func (s *LDBStore) collectGarbage() error {
}
}
// commit batch changes
s.gc.batch.Put(keyEntryCnt, U64ToBytes(s.entryCnt))
l := s.gc.batch.Len()
if err := s.db.Write(s.gc.batch.Batch); err != nil {
s.lock.Unlock()
it.Release()
return fmt.Errorf("unable to write batch: %v", err)
}
log.Trace(fmt.Sprintf("batch write (%d entries)", l))
s.writeBatch(s.gc.batch, wEntryCnt)
s.lock.Unlock()
it.Release()
log.Trace("garbage collect batch done", "batch", singleIterationCount, "total", s.gc.count)
@ -607,36 +603,36 @@ func (s *LDBStore) ReIndex() {
log.Warn(fmt.Sprintf("Found %v errors out of %v entries", errorsFound, total))
}
// Delete is thread safe and removes a chunk and updates indices. Increments accesscnt
func (s *LDBStore) Delete(addr Address) error {
s.lock.Lock()
defer s.lock.Unlock()
ikey := getIndexKey(addr)
var indx dpaDBIndex
var idx dpaDBIndex
proximity := s.po(addr)
if !s.tryAccessIdx(ikey, proximity, &indx) {
if !s.tryAccessIdx(ikey, proximity, &idx) {
return fmt.Errorf("noent")
}
s.deleteNow(&indx, ikey, proximity)
return nil
return s.deleteNow(&idx, ikey, proximity)
}
func (s *LDBStore) deleteNow(idx *dpaDBIndex, idxKey []byte, po uint8) {
func (s *LDBStore) deleteNow(idx *dpaDBIndex, idxKey []byte, po uint8) error {
batch := new(leveldb.Batch)
s.delete(batch, idx, idxKey, po)
s.db.Write(batch)
return s.db.Write(batch)
}
// NOTE: decrements entrycount regardless if the chunk exists upon deletion. Risk of wrap to max uint64
func (s *LDBStore) delete(batch *leveldb.Batch, idx *dpaDBIndex, idxKey []byte, po uint8) {
metrics.GetOrRegisterCounter("ldbstore.delete", nil).Inc(1)
batch.Delete(idxKey)
gcIdxKey := getGCIdxKey(idx)
batch.Delete(gcIdxKey)
batch.Delete(getDataKey(idx.Idx, po))
dataKey := getDataKey(idx.Idx, po)
batch.Delete(dataKey)
batch.Delete(idxKey)
s.entryCnt--
dbEntryCount.Dec(1)
cntKey := make([]byte, 2)
@ -668,12 +664,13 @@ func (s *LDBStore) Put(ctx context.Context, chunk Chunk) error {
metrics.GetOrRegisterCounter("ldbstore.put", nil).Inc(1)
log.Trace("ldbstore.put", "key", chunk.Address())
s.lock.Lock()
ikey := getIndexKey(chunk.Address())
var index dpaDBIndex
po := s.po(chunk.Address())
s.lock.Lock()
if s.closed {
s.lock.Unlock()
return ErrDBClosed
@ -685,7 +682,7 @@ func (s *LDBStore) Put(ctx context.Context, chunk Chunk) error {
if err != nil {
s.doPut(chunk, &index, po)
} else {
log.Trace("ldbstore.put: chunk already exists, only update access", "key", chunk.Address)
log.Debug("ldbstore.put: chunk already exists, only update access", "key", chunk.Address(), "po", po)
decodeIndex(idata, &index)
}
index.Access = s.accessCnt
@ -749,31 +746,32 @@ func (s *LDBStore) writeBatches() {
func (s *LDBStore) writeCurrentBatch() error {
s.lock.Lock()
var b *dbBatch
b = s.batch
defer s.lock.Unlock()
b := s.batch
l := b.Len()
if l == 0 {
s.lock.Unlock()
return nil
}
e := s.entryCnt
d := s.dataIdx
a := s.accessCnt
s.batch = newBatch()
b.err = s.writeBatch(b, e, d, a)
b.err = s.writeBatch(b, wEntryCnt|wAccessCnt|wIndexCnt)
close(b.c)
s.lock.Unlock()
if e > s.capacity {
if s.entryCnt >= s.capacity {
go s.collectGarbage()
}
return nil
}
// must be called non concurrently
func (s *LDBStore) writeBatch(b *dbBatch, entryCnt, dataIdx, accessCnt uint64) error {
b.Put(keyEntryCnt, U64ToBytes(entryCnt))
b.Put(keyDataIdx, U64ToBytes(dataIdx))
b.Put(keyAccessCnt, U64ToBytes(accessCnt))
func (s *LDBStore) writeBatch(b *dbBatch, wFlag uint8) error {
if wFlag&wEntryCnt > 0 {
b.Put(keyEntryCnt, U64ToBytes(s.entryCnt))
}
if wFlag&wIndexCnt > 0 {
b.Put(keyDataIdx, U64ToBytes(s.dataIdx))
}
if wFlag&wAccessCnt > 0 {
b.Put(keyAccessCnt, U64ToBytes(s.accessCnt))
}
l := b.Len()
if err := s.db.Write(b.Batch); err != nil {
return fmt.Errorf("unable to write batch: %v", err)

78
swarm/storage/ldbstore_test.go
View file

@ -280,7 +280,7 @@ func TestLDBStoreWithoutCollectGarbage(t *testing.T) {
log.Info("ldbstore", "entrycnt", ldb.entryCnt, "accesscnt", ldb.accessCnt)
for _, ch := range chunks {
ret, err := ldb.get(ch.Address())
ret, err := ldb.Get(context.TODO(), ch.Address())
if err != nil {
t.Fatal(err)
}
@ -301,17 +301,17 @@ func TestLDBStoreWithoutCollectGarbage(t *testing.T) {
func TestLDBStoreCollectGarbage(t *testing.T) {
var cap int
cap := defaultMaxGCRound / 2
t.Run(fmt.Sprintf("A/%d/%d", cap, cap*4), testLDBStoreCollectGarbage)
t.Run(fmt.Sprintf("B/%d/%d", cap, cap*4), testLDBStoreRemoveThenCollectGarbage)
cap = defaultMaxGCRound
t.Run(fmt.Sprintf("A/%d/%d/%d", cap, cap*2, 10000), testLDBStoreCollectGarbage)
t.Run(fmt.Sprintf("A/%d/%d", cap, cap*4), testLDBStoreCollectGarbage)
t.Run(fmt.Sprintf("B/%d/%d", cap, cap*4), testLDBStoreRemoveThenCollectGarbage)
cap = defaultMaxGCRound / 2
t.Run(fmt.Sprintf("A/%d/%d/%d", cap, cap*4, 15000), testLDBStoreCollectGarbage)
t.Run(fmt.Sprintf("B/%d/%d/%d", cap, cap*4, 15000), testLDBStoreRemoveThenCollectGarbage)
cap = defaultMaxGCRound * 2
t.Run(fmt.Sprintf("A/%d/%d/%d", cap, cap*4, 60000), testLDBStoreCollectGarbage)
t.Run(fmt.Sprintf("B/%d/%d/%d", cap, cap*4, 60000), testLDBStoreRemoveThenCollectGarbage)
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
@ -326,18 +326,18 @@ func testLDBStoreCollectGarbage(t *testing.T) {
if err != nil {
t.Fatal(err)
}
expectMissing, err := strconv.Atoi(params[4])
if err != nil {
t.Fatal(err)
}
surplus := n - capacity
ldb, cleanup := newLDBStore(t)
ldb.setCapacity(uint64(capacity))
defer cleanup()
// retrieve the gc round target count for the db capacity
ldb.startGC(capacity)
roundTarget := ldb.gc.target
// split put counts to gc target count threshold, and wait for gc to finish inbetween
var allChunks []Chunk
remaining := n
for remaining > 0 {
@ -355,12 +355,12 @@ func testLDBStoreCollectGarbage(t *testing.T) {
allChunks = append(allChunks, chunks...)
log.Debug("ldbstore", "entrycnt", ldb.entryCnt, "accesscnt", ldb.accessCnt, "cap", capacity, "n", n)
// wait for garbage collection to kick in on the responsible actor
ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
defer cancel()
waitGc(ctx, ldb)
}
// attempt gets on all put chunks
var missing int
for _, ch := range allChunks {
ret, err := ldb.Get(context.TODO(), ch.Address())
@ -379,8 +379,9 @@ func testLDBStoreCollectGarbage(t *testing.T) {
log.Trace("got back chunk", "chunk", ret)
}
if missing < expectMissing {
t.Fatalf("gc failure: expected to miss %v chunks, but only %v are actually missing", expectMissing, missing)
// all surplus chunks should be missing
if missing != surplus+roundTarget {
t.Fatalf("gc failure: expected to miss %v chunks, but only %v are actually missing", surplus-roundTarget, missing)
}
log.Info("ldbstore", "total", n, "missing", missing, "entrycnt", ldb.entryCnt, "accesscnt", ldb.accessCnt)
@ -442,25 +443,45 @@ func testLDBStoreRemoveThenCollectGarbage(t *testing.T) {
}
ldb, cleanup := newLDBStore(t)
defer cleanup()
ldb.setCapacity(uint64(capacity))
surplus := n - capacity
chunks := []Chunk{}
for i := 0; i < n+surplus; i++ {
// put capacity count number of chunks
chunks := make([]Chunk, n)
for i := 0; i < n; i++ {
c := GenerateRandomChunk(ch.DefaultSize)
chunks = append(chunks, c)
chunks[i] = c
log.Trace("generate random chunk", "idx", i, "chunk", c)
}
for i := 0; i < n; i++ {
ldb.Put(context.TODO(), chunks[i])
err := ldb.Put(context.TODO(), chunks[i])
if err != nil {
t.Fatal(err)
}
}
ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
defer cancel()
waitGc(ctx, ldb)
// delete all chunks
// (only count the ones actually deleted, the rest will have been gc'd)
deletes := 0
for i := 0; i < n; i++ {
if ldb.Delete(chunks[i].Address()) == nil {
ikey := getIndexKey(chunks[i].Address())
idata, err := ldb.db.Get(ikey)
if err == nil {
deletes++
po := ldb.po(chunks[i].Address())
var idx dpaDBIndex
decodeIndex(idata, &idx)
err := ldb.deleteNow(&idx, ikey, po)
if err != nil {
t.Fatal(err)
}
}
}
@ -470,17 +491,13 @@ func testLDBStoreRemoveThenCollectGarbage(t *testing.T) {
t.Fatalf("ldb.entrCnt expected 0 got %v", ldb.entryCnt)
}
expAccessCnt := uint64(n + deletes)
// the manual deletes will have increased accesscnt, so we need to add this when we verify the current count
expAccessCnt := uint64(n)
if ldb.accessCnt != expAccessCnt {
t.Fatalf("ldb.accessCnt expected %v got %v", expAccessCnt, ldb.accessCnt)
}
cleanup()
ldb, cleanup = newLDBStore(t)
ldb.setCapacity(uint64(capacity))
defer cleanup()
// retrieve the gc round target count for the db capacity
ldb.startGC(capacity)
roundTarget := ldb.gc.target
@ -496,19 +513,18 @@ func testLDBStoreRemoveThenCollectGarbage(t *testing.T) {
remaining -= putCount
for putCount > 0 {
ldb.Put(context.TODO(), chunks[puts])
log.Debug("ldbstore", "entrycnt", ldb.entryCnt, "accesscnt", ldb.accessCnt, "cap", capacity, "n", n, "puts", puts, "remaining", remaining)
log.Debug("ldbstore", "entrycnt", ldb.entryCnt, "accesscnt", ldb.accessCnt, "cap", capacity, "n", n, "puts", puts, "remaining", remaining, "roundtarget", roundTarget)
puts++
putCount--
}
// wait for garbage collection to kick in on the responsible actor
ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
defer cancel()
waitGc(ctx, ldb)
}
// expect first surplus chunks to be missing, because they have the smallest access value
for i := 0; i < surplus; i++ {
for i := 0; i < surplus+roundTarget; i++ {
_, err := ldb.Get(context.TODO(), chunks[i].Address())
if err == nil {
t.Fatalf("expected surplus chunk %d to be missing, but got no error", i)
@ -516,7 +532,7 @@ func testLDBStoreRemoveThenCollectGarbage(t *testing.T) {
}
// expect last chunks to be present, as they have the largest access value
for i := surplus + 1; i < n; i++ {
for i := surplus + roundTarget; i < n; i++ {
ret, err := ldb.Get(context.TODO(), chunks[i].Address())
if err != nil {
t.Fatalf("chunk %v: expected no error, but got %s", i, err)