swarm/storage: Revert silly garbageC, correct test expects

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()
+	// calculate the amount of chunks to collect and reset counter
-	defer it.Release()
-
 	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()
-		s.garbageC <- struct{}{}
+			it.Release()
+			return fmt.Errorf("unable to write batch: %v", err)
+		}
+		log.Trace(fmt.Sprintf("batch write (%d entries)", l))
+		s.lock.Unlock()
+		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)
+	if !s.tryAccessIdx(ikey, proximity, &indx) {
-
+		return fmt.Errorf("noent")
-	s.lock.Lock()
+	}
-	defer s.lock.Unlock()
 
 	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
-	}
 	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()
-	}
 	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)
+	b.Put(keyEntryCnt, 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(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()
 }

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("A/%d/%d/%d", cap, cap*4, 15000), testLDBStoreCollectGarbage)
-	t.Run(fmt.Sprintf("B/%d/%d", cap, cap*4), testLDBStoreRemoveThenCollectGarbage)
+	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("A/%d/%d/%d", cap, cap*4, 60000), testLDBStoreCollectGarbage)
-	t.Run(fmt.Sprintf("B/%d/%d", cap, cap*4), testLDBStoreRemoveThenCollectGarbage)
+	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))
+	ldb.startGC(capacity)
+	roundTarget := ldb.gc.target
+
+	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())
 		}
-	log.Info("ldbstore", "entrycnt", ldb.entryCnt, "accesscnt", ldb.accessCnt)
+		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 {
+	if missing < expectMissing {
-		t.Fatalf("gc failure: expected to miss %v chunks, but only %v are actually missing", n-capacity, missing)
+		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++ {
+	remaining := n
-		ldb.Put(context.TODO(), chunks[i])
+	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
+		// 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++ {
 		_, 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)