diff --git a/swarm/storage/localstore/gc.go b/swarm/storage/localstore/gc.go index de0bb13136..00e02d8bca 100644 --- a/swarm/storage/localstore/gc.go +++ b/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) } diff --git a/swarm/storage/localstore/gc_test.go b/swarm/storage/localstore/gc_test.go index 9f5f497669..da32b9dcbb 100644 --- a/swarm/storage/localstore/gc_test.go +++ b/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: - totalCollectedCount += c + case <-testHookCollectGarbageChan: 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 } diff --git a/swarm/storage/localstore/localstore.go b/swarm/storage/localstore/localstore.go index 6266a3959f..0c75489c72 100644 --- a/swarm/storage/localstore/localstore.go +++ b/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 - gcSize int64 + // 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() diff --git a/swarm/storage/localstore/localstore_test.go b/swarm/storage/localstore/localstore_test.go index 4be6df537e..549f049c84 100644 --- a/swarm/storage/localstore/localstore_test.go +++ b/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) } diff --git a/swarm/storage/localstore/subscription_pull_test.go b/swarm/storage/localstore/subscription_pull_test.go index fb804b04c4..ecd6185f20 100644 --- a/swarm/storage/localstore/subscription_pull_test.go +++ b/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) {