swarm/storage/localstore: replace atomics with mutexes for gcSize and tests

swarm/storage/localstore/gc.go

@@ -17,7 +17,6 @@
 package localstore
 
 import (
-	"sync/atomic"
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
@@ -88,7 +87,7 @@ func (db *DB) collectGarbage() (collectedCount int64, done bool, err error) {
 		}
 		defer unlock()
 
-		gcSize := atomic.LoadInt64(&db.gcSize)
+		gcSize := db.getGCSize()
 		if gcSize-collectedCount <= target {
 			return true, nil
 		}
@@ -131,7 +130,12 @@ func (db *DB) incGCSize(count int64) {
 	if count == 0 {
 		return
 	}
-	new := atomic.AddInt64(&db.gcSize, count)
+
+	db.gcSizeMu.Lock()
+	new := db.gcSize + count
+	db.gcSize = new
+	db.gcSizeMu.Unlock()
+
 	select {
 	case db.writeGCSizeTrigger <- struct{}{}:
 	default:
@@ -141,6 +145,15 @@ func (db *DB) incGCSize(count int64) {
 	}
 }
 
+// getGCSize returns gcSize value by locking it
+// with gcSizeMu mutex.
+func (db *DB) getGCSize() (count int64) {
+	db.gcSizeMu.RLock()
+	count = db.gcSize
+	db.gcSizeMu.RUnlock()
+	return count
+}
+
 // triggerGarbageCollection signals collectGarbageWorker
 // to call collectGarbage.
 func (db *DB) triggerGarbageCollection() {
@@ -159,7 +172,7 @@ func (db *DB) writeGCSizeWorker() {
 	for {
 		select {
 		case <-db.writeGCSizeTrigger:
-			err := db.writeGCSize(atomic.LoadInt64(&db.gcSize))
+			err := db.writeGCSize(db.getGCSize())
 			if err != nil {
 				log.Error("localstore write gc size", "err", err)
 			}

swarm/storage/localstore/gc_test.go

@@ -20,7 +20,6 @@ import (
 	"io/ioutil"
 	"math/rand"
 	"os"
-	"sync/atomic"
 	"testing"
 	"time"
 
@@ -89,25 +88,18 @@ func testDB_collectGarbageWorker(t *testing.T, db *DB) {
 
 	gcTarget := db.gcTarget()
 
-	var totalCollectedCount int64
 	for {
 		select {
-		case c := <-testHookCollectGarbageChan:
+		case <-testHookCollectGarbageChan:
-			totalCollectedCount += c
 		case <-time.After(10 * time.Second):
 			t.Error("collect garbage timeout")
 		}
-		gcSize := atomic.LoadInt64(&db.gcSize)
+		gcSize := db.getGCSize()
 		if gcSize == gcTarget {
 			break
 		}
 	}
 
-	wantTotalCollectedCount := int64(chunkCount) - gcTarget
-	if totalCollectedCount != wantTotalCollectedCount {
-		t.Errorf("total collected chunks %v, want %v", totalCollectedCount, wantTotalCollectedCount)
-	}
-
 	t.Run("pull index count", newItemsCountTest(db.pullIndex, int(gcTarget)))
 
 	t.Run("gc index count", newItemsCountTest(db.gcIndex, int(gcTarget)))
@@ -200,7 +192,7 @@ func TestDB_collectGarbageWorker_withRequests(t *testing.T) {
 		case <-time.After(10 * time.Second):
 			t.Error("collect garbage timeout")
 		}
-		gcSize := atomic.LoadInt64(&db.gcSize)
+		gcSize := db.getGCSize()
 		if gcSize == gcTarget {
 			break
 		}

swarm/storage/localstore/localstore.go

@@ -21,7 +21,6 @@ import (
 	"encoding/hex"
 	"errors"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
@@ -82,7 +81,10 @@ type DB struct {
 	gcUncountedHashesIndex shed.Index
 
 	// number of elements in garbage collection index
+	// it must be always read by getGCSize and
+	// set with incGCSize which are locking gcSizeMu
 	gcSize   int64
+	gcSizeMu sync.RWMutex
 	// garbage collection is triggered when gcSize exceeds
 	// the capacity value
 	capacity int64
@@ -361,7 +363,7 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
 func (db *DB) Close() (err error) {
 	close(db.close)
 	db.updateGCWG.Wait()
-	if err := db.writeGCSize(atomic.LoadInt64(&db.gcSize)); err != nil {
+	if err := db.writeGCSize(db.getGCSize()); err != nil {
 		log.Error("localstore: write gc size", "err", err)
 	}
 	return db.shed.Close()

swarm/storage/localstore/localstore_test.go

@@ -25,7 +25,6 @@ import (
 	"sort"
 	"strconv"
 	"sync"
-	"sync/atomic"
 	"testing"
 	"time"
 
@@ -408,7 +407,7 @@ func newIndexGCSizeTest(db *DB) func(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		got := atomic.LoadInt64(&db.gcSize)
+		got := db.getGCSize()
 		if got != want {
 			t.Errorf("got gc size %v, want %v", got, want)
 		}

swarm/storage/localstore/subscription_pull_test.go

@@ -20,7 +20,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"sync/atomic"
+	"sync"
 	"testing"
 	"time"
 
@@ -143,8 +143,12 @@ func TestDB_SubscribePull_since(t *testing.T) {
 	var wantedChunksCount int
 
 	lastTimestamp := time.Now().UTC().UnixNano()
+	var lastTimestampMu sync.RWMutex
 	defer setNow(func() (t int64) {
-		return atomic.AddInt64(&lastTimestamp, 1)
+		lastTimestampMu.Lock()
+		defer lastTimestampMu.Unlock()
+		lastTimestamp++
+		return lastTimestamp
 	})()
 
 	uploadRandomChunks := func(count int, wanted bool) (last map[uint8]ChunkDescriptor) {
@@ -167,9 +171,13 @@ func TestDB_SubscribePull_since(t *testing.T) {
 				wantedChunksCount++
 			}
 
+			lastTimestampMu.RLock()
+			storeTimestamp := lastTimestamp
+			lastTimestampMu.RUnlock()
+
 			last[bin] = ChunkDescriptor{
 				Address:        chunk.Address(),
-				StoreTimestamp: atomic.LoadInt64(&lastTimestamp),
+				StoreTimestamp: storeTimestamp,
 			}
 		}
 		return last
@@ -222,8 +230,12 @@ func TestDB_SubscribePull_until(t *testing.T) {
 	var wantedChunksCount int
 
 	lastTimestamp := time.Now().UTC().UnixNano()
+	var lastTimestampMu sync.RWMutex
 	defer setNow(func() (t int64) {
-		return atomic.AddInt64(&lastTimestamp, 1)
+		lastTimestampMu.Lock()
+		defer lastTimestampMu.Unlock()
+		lastTimestamp++
+		return lastTimestamp
 	})()
 
 	uploadRandomChunks := func(count int, wanted bool) (last map[uint8]ChunkDescriptor) {
@@ -246,9 +258,13 @@ func TestDB_SubscribePull_until(t *testing.T) {
 				wantedChunksCount++
 			}
 
+			lastTimestampMu.RLock()
+			storeTimestamp := lastTimestamp
+			lastTimestampMu.RUnlock()
+
 			last[bin] = ChunkDescriptor{
 				Address:        chunk.Address(),
-				StoreTimestamp: atomic.LoadInt64(&lastTimestamp),
+				StoreTimestamp: storeTimestamp,
 			}
 		}
 		return last
@@ -302,8 +318,12 @@ func TestDB_SubscribePull_sinceAndUntil(t *testing.T) {
 	var wantedChunksCount int
 
 	lastTimestamp := time.Now().UTC().UnixNano()
+	var lastTimestampMu sync.RWMutex
 	defer setNow(func() (t int64) {
-		return atomic.AddInt64(&lastTimestamp, 1)
+		lastTimestampMu.Lock()
+		defer lastTimestampMu.Unlock()
+		lastTimestamp++
+		return lastTimestamp
 	})()
 
 	uploadRandomChunks := func(count int, wanted bool) (last map[uint8]ChunkDescriptor) {
@@ -326,9 +346,13 @@ func TestDB_SubscribePull_sinceAndUntil(t *testing.T) {
 				wantedChunksCount++
 			}
 
+			lastTimestampMu.RLock()
+			storeTimestamp := lastTimestamp
+			lastTimestampMu.RUnlock()
+
 			last[bin] = ChunkDescriptor{
 				Address:        chunk.Address(),
-				StoreTimestamp: atomic.LoadInt64(&lastTimestamp),
+				StoreTimestamp: storeTimestamp,
 			}
 		}
 		return last
@@ -410,6 +434,9 @@ func readPullSubscriptionBin(ctx context.Context, bin uint8, ch <-chan ChunkDesc
 			if !ok {
 				return
 			}
+			if i+1 > len(addrs[bin]) {
+				errChan <- fmt.Errorf("got more chunk addresses %v, then expected %v, for bin %v", i+1, len(addrs[bin]), bin)
+			}
 			want := addrs[bin][i]
 			var err error
 			if !bytes.Equal(got.Address, want) {