diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go index c5157ae02d..2e92a595c1 100644 --- a/swarm/storage/ldbstore.go +++ b/swarm/storage/ldbstore.go @@ -113,7 +113,6 @@ type LDBStore struct { po func(Address) uint8 batchesC chan struct{} - garbageC chan struct{} closed bool batch *dbBatch lock sync.RWMutex @@ -150,7 +149,6 @@ func NewLDBStore(params *LDBStoreParams) (s *LDBStore, err error) { s.quit = make(chan struct{}) s.batchesC = make(chan struct{}, 1) - s.garbageC = make(chan struct{}, 1) go s.writeBatches() s.batch = newBatch() // associate encodeData with default functionality @@ -306,13 +304,13 @@ func decodeData(addr Address, data []byte) (*chunk, error) { return NewChunk(addr, data[32:]), nil } -func (s *LDBStore) collectGarbage() { +func (s *LDBStore) collectGarbage() error { // the running param prevents duplicate gc from starting when one is already running s.lock.Lock() if s.gc.running { s.lock.Unlock() - return + return nil } s.gc.running = true defer func() { @@ -324,45 +322,61 @@ func (s *LDBStore) collectGarbage() { metrics.GetOrRegisterCounter("ldbstore.collectgarbage", nil).Inc(1) - it := s.db.NewIterator() - defer it.Release() - + // calculate the amount of chunks to collect and reset counter s.startGC(int(s.entryCnt)) - log.Debug("collectGarbage", "count", s.gc.target, "entryCnt", s.entryCnt) + log.Debug("collectGarbage", "target", s.gc.target, "entryCnt", s.entryCnt) var totalDeleted int - ok := it.Seek([]byte{keyGCIdx}) for s.gc.count < s.gc.target { + it := s.db.NewIterator() + ok := it.Seek([]byte{keyGCIdx}) var singleIterationCount int + + // every batch needs a lock so we avoid entries changing accessidx in the meantime s.lock.Lock() for ; ok && (singleIterationCount < s.gc.maxBatch); ok = it.Next() { - itkey := it.Key() + // quit if no more access index keys + itkey := it.Key() if (itkey == nil) || (itkey[0] != keyGCIdx) { break } + // get chunk data entry from access index val := it.Value() index, po, hash := parseGCIdxEntry(itkey[1:], val) keyIdx := make([]byte, 33) keyIdx[0] = keyIndex copy(keyIdx[1:], hash) + // add delete operation to batch s.delete(s.gc.batch.Batch, index, keyIdx, po) singleIterationCount++ s.gc.count++ - if s.gc.count > s.gc.maxRound { + + // break if target is not on max garbage batch boundary + if s.gc.count >= s.gc.target { break } } + + // 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.lock.Unlock() - s.garbageC <- struct{}{} + it.Release() log.Trace("garbage collect batch done", "batch", singleIterationCount, "total", s.gc.count) } log.Debug("garbage collect done", "c", s.gc.count) metrics.GetOrRegisterCounter("ldbstore.collectgarbage.delete", nil).Inc(int64(totalDeleted)) - + return nil } // Export writes all chunks from the store to a tar archive, returning the @@ -593,17 +607,20 @@ func (s *LDBStore) ReIndex() { log.Warn(fmt.Sprintf("Found %v errors out of %v entries", errorsFound, total)) } -func (s *LDBStore) Delete(addr Address) { +func (s *LDBStore) Delete(addr Address) error { + s.lock.Lock() + defer s.lock.Unlock() + ikey := getIndexKey(addr) var indx dpaDBIndex proximity := s.po(addr) - s.tryAccessIdx(ikey, proximity, &indx) - - s.lock.Lock() - defer s.lock.Unlock() + if !s.tryAccessIdx(ikey, proximity, &indx) { + return fmt.Errorf("noent") + } s.deleteNow(&indx, ikey, proximity) + return nil } func (s *LDBStore) deleteNow(idx *dpaDBIndex, idxKey []byte, po uint8) { @@ -612,6 +629,7 @@ func (s *LDBStore) deleteNow(idx *dpaDBIndex, idxKey []byte, po uint8) { 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) @@ -719,31 +737,20 @@ func (s *LDBStore) writeBatches() { log.Debug("DbStore: quit batch write loop") return case <-s.batchesC: - err := s.writeCurrentBatch(false) + err := s.writeCurrentBatch() if err != nil { log.Debug("DbStore: quit batch write loop", "err", err.Error()) return } - case <-s.garbageC: - err := s.writeCurrentBatch(true) - if err != nil { - log.Debug("DbStore: quit batch garbage write loop", "err", err.Error()) - return - } - } } } -func (s *LDBStore) writeCurrentBatch(garbage bool) error { +func (s *LDBStore) writeCurrentBatch() error { s.lock.Lock() var b *dbBatch - if garbage { - b = s.gc.batch - } else { - b = s.batch - } + b = s.batch l := b.Len() if l == 0 { s.lock.Unlock() @@ -752,15 +759,11 @@ func (s *LDBStore) writeCurrentBatch(garbage bool) error { e := s.entryCnt d := s.dataIdx a := s.accessCnt - if garbage { - s.gc.batch = newBatch() - } else { - s.batch = newBatch() - } + s.batch = newBatch() b.err = s.writeBatch(b, e, d, a) close(b.c) s.lock.Unlock() - if e > s.capacity && !garbage { + if e > s.capacity { go s.collectGarbage() } return nil @@ -768,13 +771,7 @@ func (s *LDBStore) writeCurrentBatch(garbage bool) error { // must be called non concurrently func (s *LDBStore) writeBatch(b *dbBatch, entryCnt, dataIdx, accessCnt uint64) error { - ub := U64ToBytes(entryCnt) - if len(ub) != 8 || len(keyEntryCnt) != 1 { - e := fmt.Errorf("ub fail: %d -> %v . %d", entryCnt, ub, len(keyEntryCnt)) - log.Error("key", "e", e) - return e - } - b.Put(keyEntryCnt, ub) + b.Put(keyEntryCnt, U64ToBytes(entryCnt)) b.Put(keyDataIdx, U64ToBytes(dataIdx)) b.Put(keyAccessCnt, U64ToBytes(accessCnt)) l := b.Len() @@ -800,8 +797,6 @@ 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, po uint8, index *dpaDBIndex) bool { - //s.axxLock.Lock() - //defer s.axxLock.Unlock() idata, err := s.db.Get(ikey) if err != nil { return false @@ -923,7 +918,7 @@ func (s *LDBStore) Close() { s.closed = true s.lock.Unlock() // force writing out current batch - s.writeCurrentBatch(false) + s.writeCurrentBatch() close(s.batchesC) s.db.Close() } diff --git a/swarm/storage/ldbstore_test.go b/swarm/storage/ldbstore_test.go index 8f5285e026..244cd1870b 100644 --- a/swarm/storage/ldbstore_test.go +++ b/swarm/storage/ldbstore_test.go @@ -304,15 +304,14 @@ func TestLDBStoreCollectGarbage(t *testing.T) { var cap int cap = defaultMaxGCRound - t.Run(fmt.Sprintf("A/%d/%d", cap, cap*2+1), testLDBStoreCollectGarbage) + t.Run(fmt.Sprintf("A/%d/%d/%d", cap, cap*2, 10000), testLDBStoreCollectGarbage) 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) - + 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", cap, cap*4), testLDBStoreCollectGarbage) - t.Run(fmt.Sprintf("B/%d/%d", cap, cap*4), testLDBStoreRemoveThenCollectGarbage) + 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) } // TestLDBStoreCollectGarbage tests that we can put more chunks than LevelDB's capacity, and @@ -327,24 +326,43 @@ func testLDBStoreCollectGarbage(t *testing.T) { if err != nil { t.Fatal(err) } + expectMissing, err := strconv.Atoi(params[4]) + if err != nil { + t.Fatal(err) + } 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) + ldb.startGC(capacity) + roundTarget := ldb.gc.target - // 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) + var allChunks []Chunk + remaining := n + for remaining > 0 { + var putCount int + if remaining < roundTarget { + putCount = remaining + } else { + putCount = roundTarget + } + remaining -= putCount + chunks, err := mputRandomChunks(ldb, putCount, int64(ch.DefaultSize)) + if err != nil { + t.Fatal(err.Error()) + } + 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) + } var missing int - for _, ch := range chunks { + for _, ch := range allChunks { ret, err := ldb.Get(context.TODO(), ch.Address()) if err == ErrChunkNotFound || err == ldberrors.ErrNotFound { missing++ @@ -361,8 +379,8 @@ func testLDBStoreCollectGarbage(t *testing.T) { log.Trace("got back chunk", "chunk", ret) } - if missing < n-capacity { - t.Fatalf("gc failure: expected to miss %v chunks, but only %v are actually missing", n-capacity, missing) + if missing < expectMissing { + t.Fatalf("gc failure: expected to miss %v chunks, but only %v are actually missing", expectMissing, missing) } log.Info("ldbstore", "total", n, "missing", missing, "entrycnt", ldb.entryCnt, "accesscnt", ldb.accessCnt) @@ -418,7 +436,7 @@ func testLDBStoreRemoveThenCollectGarbage(t *testing.T) { if err != nil { t.Fatal(err) } - surplus, err := strconv.Atoi(params[3]) + n, err := strconv.Atoi(params[3]) if err != nil { t.Fatal(err) } @@ -426,8 +444,7 @@ func testLDBStoreRemoveThenCollectGarbage(t *testing.T) { ldb, cleanup := newLDBStore(t) ldb.setCapacity(uint64(capacity)) - n := capacity - + surplus := n - capacity chunks := []Chunk{} for i := 0; i < n+surplus; i++ { c := GenerateRandomChunk(ch.DefaultSize) @@ -440,8 +457,11 @@ func testLDBStoreRemoveThenCollectGarbage(t *testing.T) { } // delete all chunks + deletes := 0 for i := 0; i < n; i++ { - ldb.Delete(chunks[i].Address()) //&indx, ikey, proximity) + if ldb.Delete(chunks[i].Address()) == nil { + deletes++ + } } log.Info("ldbstore", "entrycnt", ldb.entryCnt, "accesscnt", ldb.accessCnt) @@ -450,7 +470,7 @@ func testLDBStoreRemoveThenCollectGarbage(t *testing.T) { t.Fatalf("ldb.entrCnt expected 0 got %v", ldb.entryCnt) } - expAccessCnt := uint64(n * 2) + expAccessCnt := uint64(n + deletes) if ldb.accessCnt != expAccessCnt { t.Fatalf("ldb.accessCnt expected %v got %v", expAccessCnt, ldb.accessCnt) } @@ -461,27 +481,42 @@ func testLDBStoreRemoveThenCollectGarbage(t *testing.T) { ldb.setCapacity(uint64(capacity)) defer cleanup() - n = capacity + surplus + ldb.startGC(capacity) + roundTarget := ldb.gc.target - for i := 0; i < n; i++ { - ldb.Put(context.TODO(), chunks[i]) + remaining := n + var puts int + for remaining > 0 { + var putCount int + if remaining < roundTarget { + putCount = remaining + } else { + putCount = roundTarget + } + 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) + 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) } - // wait for garbage collection - 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++ { _, err := ldb.Get(context.TODO(), chunks[i].Address()) if err == nil { - t.Fatal("expected surplus chunk to be missing, but got no error") + t.Fatalf("expected surplus chunk %d to be missing, but got no error", i) } } // expect last chunks to be present, as they have the largest access value - for i := surplus; i < surplus+capacity; i++ { + for i := surplus + 1; 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)