diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go index 2e92a595c1..e6eb857941 100644 --- a/swarm/storage/ldbstore.go +++ b/swarm/storage/ldbstore.go @@ -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) diff --git a/swarm/storage/ldbstore_test.go b/swarm/storage/ldbstore_test.go index 244cd1870b..55d06b9697 100644 --- a/swarm/storage/ldbstore_test.go +++ b/swarm/storage/ldbstore_test.go @@ -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)