swarm/storage/localstore: global batch write lock

swarm/storage/localstore/gc.go

@@ -17,8 +17,6 @@
 package localstore
 
 import (
-	"time"
-
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/swarm/shed"
 	"github.com/syndtr/goleveldb/leveldb"
@@ -36,7 +34,7 @@ var (
 	gcTargetRatio = 0.9
 	// gcBatchSize limits the number of chunks in a single
 	// leveldb batch on garbage collection.
-	gcBatchSize int64 = 1000
+	gcBatchSize uint64 = 1000
 )
 
 // collectGarbageWorker is a long running function that waits for
@@ -74,27 +72,21 @@ func (db *DB) collectGarbageWorker() {
 // is false, another call to this function is needed to collect
 // the rest of the garbage as the batch size limit is reached.
 // This function is called in collectGarbageWorker.
-func (db *DB) collectGarbage() (collectedCount int64, done bool, err error) {
+func (db *DB) collectGarbage() (collectedCount uint64, done bool, err error) {
 	batch := new(leveldb.Batch)
 	target := db.gcTarget()
 
-	if db.useGlobalLock {
+	// protect database from changing idexes and gcSize
-		db.globalMu.Lock()
+	db.batchMu.Lock()
-		defer db.globalMu.Unlock()
+	defer db.batchMu.Unlock()
+
+	gcSize, err := db.gcSize.Get()
+	if err != nil {
+		return 0, true, err
 	}
 
 	done = true
 	err = db.gcIndex.Iterate(func(item shed.Item) (stop bool, err error) {
-		// protect parallel updates
-		if !db.useGlobalLock {
-			unlock, err := db.lockAddr(item.Address)
-			if err != nil {
-				return false, err
-			}
-			defer unlock()
-		}
-
-		gcSize := db.getGCSize()
 		if gcSize-collectedCount <= target {
 			return true, nil
 		}
@@ -116,49 +108,19 @@ func (db *DB) collectGarbage() (collectedCount int64, done bool, err error) {
 		return 0, false, err
 	}
 
+	db.gcSize.PutInBatch(batch, gcSize-collectedCount)
+
 	err = db.shed.WriteBatch(batch)
 	if err != nil {
 		return 0, false, err
 	}
-	// batch is written, decrement gcSize
-	db.incGCSize(-collectedCount)
 	return collectedCount, done, nil
 }
 
 // gcTrigger retruns the absolute value for garbage collection
 // target value, calculated from db.capacity and gcTargetRatio.
-func (db *DB) gcTarget() (target int64) {
+func (db *DB) gcTarget() (target uint64) {
-	return int64(float64(db.capacity) * gcTargetRatio)
+	return uint64(float64(db.capacity) * gcTargetRatio)
-}
-
-// incGCSize increments gcSize by the provided number.
-// If count is negative, it will decrement gcSize.
-func (db *DB) incGCSize(count int64) {
-	if count == 0 {
-		return
-	}
-
-	db.gcSizeMu.Lock()
-	new := db.gcSize + count
-	db.gcSize = new
-	db.gcSizeMu.Unlock()
-
-	select {
-	case db.writeGCSizeTrigger <- struct{}{}:
-	default:
-	}
-	if new >= db.capacity {
-		db.triggerGarbageCollection()
-	}
-}
-
-// 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
@@ -171,66 +133,40 @@ func (db *DB) triggerGarbageCollection() {
 	}
 }
 
-// writeGCSizeWorker writes gcSize on trigger event
+// incGCSizeInBatch changes gcSize field value
-// and waits writeGCSizeDelay after each write.
+// by change which can be negative.
-// It implements a linear backoff with delay of
+func (db *DB) incGCSizeInBatch(batch *leveldb.Batch, change int64) (err error) {
-// writeGCSizeDelay duration to avoid very frequent
+	if change == 0 {
-// database operations.
+		return nil
-func (db *DB) writeGCSizeWorker() {
-	for {
-		select {
-		case <-db.writeGCSizeTrigger:
-			err := db.writeGCSize(db.getGCSize())
-			if err != nil {
-				log.Error("localstore write gc size", "err", err)
 	}
-			// Wait some time before writing gc size in the next
+	gcSize, err := db.gcSize.Get()
-			// iteration. This prevents frequent I/O operations.
-			select {
-			case <-time.After(10 * time.Second):
-			case <-db.close:
-				return
-			}
-		case <-db.close:
-			return
-		}
-	}
-}
-
-// writeGCSize stores the number of items in gcIndex.
-// It removes all hashes from gcUncountedHashesIndex
-// not to include them on the next DB initialization
-// (New function) when gcSize is counted.
-func (db *DB) writeGCSize(gcSize int64) (err error) {
-	const maxBatchSize = 1000
-
-	batch := new(leveldb.Batch)
-	db.storedGCSize.PutInBatch(batch, uint64(gcSize))
-	batchSize := 1
-
-	// use only one iterator as it acquires its snapshot
-	// not to remove hashes from index that are added
-	// after stored gc size is written
-	err = db.gcUncountedHashesIndex.Iterate(func(item shed.Item) (stop bool, err error) {
-		db.gcUncountedHashesIndex.DeleteInBatch(batch, item)
-		batchSize++
-		if batchSize >= maxBatchSize {
-			err = db.shed.WriteBatch(batch)
-			if err != nil {
-				return false, err
-			}
-			batch.Reset()
-			batchSize = 0
-		}
-		return false, nil
-	}, nil)
 	if err != nil {
 		return err
 	}
-	return db.shed.WriteBatch(batch)
+
+	var new uint64
+	if change > 0 {
+		new = gcSize + uint64(change)
+	} else {
+		// 'change' is an int64 and is negative
+		// a conversion is needed with correct sign
+		c := uint64(-change)
+		if c > gcSize {
+			// protect uint64 undeflow
+			return nil
+		}
+		new = gcSize - c
+	}
+	db.gcSize.PutInBatch(batch, new)
+
+	// trigger garbage collection if we reached the capacity
+	if new >= db.capacity {
+		db.triggerGarbageCollection()
+	}
+	return nil
 }
 
 // testHookCollectGarbage is a hook that can provide
 // information when a garbage collection run is done
 // and how many items it removed.
-var testHookCollectGarbage func(collectedCount int64)
+var testHookCollectGarbage func(collectedCount uint64)

swarm/storage/localstore/gc_test.go

@@ -38,7 +38,7 @@ func TestDB_collectGarbageWorker(t *testing.T) {
 func TestDB_collectGarbageWorker_multipleBatches(t *testing.T) {
 	// lower the maximal number of chunks in a single
 	// gc batch to ensure multiple batches.
-	defer func(s int64) { gcBatchSize = s }(gcBatchSize)
+	defer func(s uint64) { gcBatchSize = s }(gcBatchSize)
 	gcBatchSize = 2
 
 	testDB_collectGarbageWorker(t)
@@ -49,8 +49,8 @@ func TestDB_collectGarbageWorker_multipleBatches(t *testing.T) {
 func testDB_collectGarbageWorker(t *testing.T) {
 	chunkCount := 150
 
-	testHookCollectGarbageChan := make(chan int64)
+	testHookCollectGarbageChan := make(chan uint64)
-	defer setTestHookCollectGarbage(func(collectedCount int64) {
+	defer setTestHookCollectGarbage(func(collectedCount uint64) {
 		testHookCollectGarbageChan <- collectedCount
 	})()
 
@@ -89,7 +89,10 @@ func testDB_collectGarbageWorker(t *testing.T) {
 		case <-time.After(10 * time.Second):
 			t.Error("collect garbage timeout")
 		}
-		gcSize := db.getGCSize()
+		gcSize, err := db.gcSize.Get()
+		if err != nil {
+			t.Fatal(err)
+		}
 		if gcSize == gcTarget {
 			break
 		}
@@ -139,8 +142,8 @@ func TestDB_collectGarbageWorker_withRequests(t *testing.T) {
 	uploader := db.NewPutter(ModePutUpload)
 	syncer := db.NewSetter(ModeSetSync)
 
-	testHookCollectGarbageChan := make(chan int64)
+	testHookCollectGarbageChan := make(chan uint64)
-	defer setTestHookCollectGarbage(func(collectedCount int64) {
+	defer setTestHookCollectGarbage(func(collectedCount uint64) {
 		testHookCollectGarbageChan <- collectedCount
 	})()
 
@@ -188,7 +191,7 @@ func TestDB_collectGarbageWorker_withRequests(t *testing.T) {
 
 	gcTarget := db.gcTarget()
 
-	var totalCollectedCount int64
+	var totalCollectedCount uint64
 	for {
 		select {
 		case c := <-testHookCollectGarbageChan:
@@ -196,13 +199,16 @@ func TestDB_collectGarbageWorker_withRequests(t *testing.T) {
 		case <-time.After(10 * time.Second):
 			t.Error("collect garbage timeout")
 		}
-		gcSize := db.getGCSize()
+		gcSize, err := db.gcSize.Get()
+		if err != nil {
+			t.Fatal(err)
+		}
 		if gcSize == gcTarget {
 			break
 		}
 	}
 
-	wantTotalCollectedCount := int64(len(addrs)) - gcTarget
+	wantTotalCollectedCount := uint64(len(addrs)) - gcTarget
 	if totalCollectedCount != wantTotalCollectedCount {
 		t.Errorf("total collected chunks %v, want %v", totalCollectedCount, wantTotalCollectedCount)
 	}
@@ -290,14 +296,12 @@ func TestDB_gcSize(t *testing.T) {
 	}
 
 	t.Run("gc index size", newIndexGCSizeTest(db))
-
-	t.Run("gc uncounted hashes index count", newItemsCountTest(db.gcUncountedHashesIndex, 0))
 }
 
 // setTestHookCollectGarbage sets testHookCollectGarbage and
 // returns a function that will reset it to the
 // value before the change.
-func setTestHookCollectGarbage(h func(collectedCount int64)) (reset func()) {
+func setTestHookCollectGarbage(h func(collectedCount uint64)) (reset func()) {
 	current := testHookCollectGarbage
 	reset = func() { testHookCollectGarbage = current }
 	testHookCollectGarbage = h
@@ -309,7 +313,7 @@ func setTestHookCollectGarbage(h func(collectedCount int64)) (reset func()) {
 // resets the original function.
 func TestSetTestHookCollectGarbage(t *testing.T) {
 	// Set the current function after the test finishes.
-	defer func(h func(collectedCount int64)) { testHookCollectGarbage = h }(testHookCollectGarbage)
+	defer func(h func(collectedCount uint64)) { testHookCollectGarbage = h }(testHookCollectGarbage)
 
 	// expected value for the unchanged function
 	original := 1
@@ -320,7 +324,7 @@ func TestSetTestHookCollectGarbage(t *testing.T) {
 	var got int
 
 	// define the original (unchanged) functions
-	testHookCollectGarbage = func(_ int64) {
+	testHookCollectGarbage = func(_ uint64) {
 		got = original
 	}
 
@@ -333,7 +337,7 @@ func TestSetTestHookCollectGarbage(t *testing.T) {
 	}
 
 	// set the new function
-	reset := setTestHookCollectGarbage(func(_ int64) {
+	reset := setTestHookCollectGarbage(func(_ uint64) {
 		got = changed
 	})
 

swarm/storage/localstore/localstore.go

@@ -18,12 +18,10 @@ package localstore
 
 import (
 	"encoding/binary"
-	"encoding/hex"
 	"errors"
 	"sync"
 	"time"
 
-	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/swarm/shed"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 	"github.com/ethereum/go-ethereum/swarm/storage/mock"
@@ -41,7 +39,7 @@ var (
 
 var (
 	// Default value for Capacity DB option.
-	defaultCapacity int64 = 5000000
+	defaultCapacity uint64 = 5000000
 	// Limit the number of goroutines created by Getters
 	// that call updateGC function. Value 0 sets no limit.
 	maxParallelUpdateGC = 1000
@@ -54,8 +52,6 @@ type DB struct {
 
 	// schema name of loaded data
 	schemaName shed.StringField
-	// field that stores number of intems in gc index
-	storedGCSize shed.Uint64Field
 
 	// retrieval indexes
 	retrievalDataIndex   shed.Index
@@ -74,23 +70,16 @@ type DB struct {
 
 	// garbage collection index
 	gcIndex shed.Index
-	// index that stores hashes that are not
-	// counted in and saved to storedGCSize
-	gcUncountedHashesIndex shed.Index
 
-	// number of elements in garbage collection index
+	// field that stores number of intems in gc index
-	// it must be always read by getGCSize and
+	gcSize shed.Uint64Field
-	// set with incGCSize which are locking gcSizeMu
+
-	gcSize   int64
-	gcSizeMu sync.RWMutex
 	// garbage collection is triggered when gcSize exceeds
 	// the capacity value
-	capacity int64
+	capacity uint64
 
 	// triggers garbage collection event loop
 	collectGarbageTrigger chan struct{}
-	// triggers write gc size event loop
-	writeGCSizeTrigger chan struct{}
 
 	// a buffered channel acting as a semaphore
 	// to limit the maximal number of goroutines
@@ -102,13 +91,7 @@ type DB struct {
 
 	baseKey []byte
 
-	addressLocks sync.Map
+	batchMu sync.Mutex
-
-	// useGlobalLock specifies that DB should not perform
-	// any batch writes in parallel. This is for benchmarks only.
-	useGlobalLock bool
-	// This is for benchmarks only.
-	globalMu sync.Mutex
 
 	// this channel is closed when close function is called
 	// to terminate other goroutines
@@ -125,12 +108,9 @@ type Options struct {
 	MockStore *mock.NodeStore
 	// Capacity is a limit that triggers garbage collection when
 	// number of items in gcIndex equals or exceeds it.
-	Capacity int64
+	Capacity uint64
 	// MetricsPrefix defines a prefix for metrics names.
 	MetricsPrefix string
-	// useGlobalLock specifies that DB should not perform
-	// any batch writes in parallel. This is for benchmarks only.
-	useGlobalLock bool
 }
 
 // New returns a new DB.  All fields and indexes are initialized
@@ -142,14 +122,12 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
 	}
 	db = &DB{
 		capacity: o.Capacity,
-		useGlobalLock: o.useGlobalLock,
 		baseKey:  baseKey,
-		// channels collectGarbageTrigger and writeGCSizeTrigger
+		// channel collectGarbageTrigger
-		// need to be buffered with the size of 1
+		// needs to be buffered with the size of 1
 		// to signal another event if it
 		// is triggered during already running function
 		collectGarbageTrigger: make(chan struct{}, 1),
-		writeGCSizeTrigger:    make(chan struct{}, 1),
 		close:                 make(chan struct{}),
 	}
 	if db.capacity <= 0 {
@@ -169,7 +147,7 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
 		return nil, err
 	}
 	// Persist gc size.
-	db.storedGCSize, err = db.shed.NewUint64Field("gc-size")
+	db.gcSize, err = db.shed.NewUint64Field("gc-size")
 	if err != nil {
 		return nil, err
 	}
@@ -320,48 +298,7 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
 	if err != nil {
 		return nil, err
 	}
-	// gc uncounted hashes index keeps hashes that are in gc index
-	// but not counted in and saved to storedGCSize
-	db.gcUncountedHashesIndex, err = db.shed.NewIndex("Hash->nil", shed.IndexFuncs{
-		EncodeKey: func(fields shed.Item) (key []byte, err error) {
-			return fields.Address, nil
-		},
-		DecodeKey: func(key []byte) (e shed.Item, err error) {
-			e.Address = key
-			return e, nil
-		},
-		EncodeValue: func(fields shed.Item) (value []byte, err error) {
-			return nil, nil
-		},
-		DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
-			return e, nil
-		},
-	})
-	if err != nil {
-		return nil, err
-	}
 
-	// count number of elements in garbage collection index
-	gcSize, err := db.storedGCSize.Get()
-	if err != nil {
-		return nil, err
-	}
-	// get number of uncounted hashes
-	gcUncountedSize, err := db.gcUncountedHashesIndex.Count()
-	if err != nil {
-		return nil, err
-	}
-	gcSize += uint64(gcUncountedSize)
-	// remove uncounted hashes from the index and
-	// save the total gcSize after uncounted hashes are removed
-	err = db.writeGCSize(int64(gcSize))
-	if err != nil {
-		return nil, err
-	}
-	db.incGCSize(int64(gcSize))
-
-	// start worker to write gc size
-	go db.writeGCSizeWorker()
 	// start garbage collection worker
 	go db.collectGarbageWorker()
 	return db, nil
@@ -371,9 +308,6 @@ 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(db.getGCSize()); err != nil {
-		log.Error("localstore: write gc size", "err", err)
-	}
 	return db.shed.Close()
 }
 
@@ -383,35 +317,6 @@ func (db *DB) po(addr storage.Address) (bin uint8) {
 	return uint8(storage.Proximity(db.baseKey, addr))
 }
 
-var (
-	// Maximal time for lockAddr to wait until it
-	// returns error.
-	addressLockTimeout = 3 * time.Second
-	// duration between two lock checks in lockAddr.
-	addressLockCheckDelay = 30 * time.Microsecond
-)
-
-// lockAddr sets the lock on a particular address
-// using addressLocks sync.Map and returns unlock function.
-// If the address is locked this function will check it
-// in a for loop for addressLockTimeout time, after which
-// it will return ErrAddressLockTimeout error.
-func (db *DB) lockAddr(addr storage.Address) (unlock func(), err error) {
-	start := time.Now()
-	lockKey := hex.EncodeToString(addr)
-	for {
-		_, loaded := db.addressLocks.LoadOrStore(lockKey, struct{}{})
-		if !loaded {
-			break
-		}
-		time.Sleep(addressLockCheckDelay)
-		if time.Since(start) > addressLockTimeout {
-			return nil, ErrAddressLockTimeout
-		}
-	}
-	return func() { db.addressLocks.Delete(lockKey) }, nil
-}
-
 // chunkToItem creates new Item with data provided by the Chunk.
 func chunkToItem(ch storage.Chunk) shed.Item {
 	return shed.Item{

swarm/storage/localstore/localstore_test.go

@@ -23,7 +23,6 @@ import (
 	"math/rand"
 	"os"
 	"sort"
-	"strconv"
 	"sync"
 	"testing"
 	"time"
@@ -117,88 +116,6 @@ func TestDB_updateGCSem(t *testing.T) {
 	}
 }
 
-// BenchmarkNew measures the time that New function
-// needs to initialize and count the number of key/value
-// pairs in GC index.
-// This benchmark generates a number of chunks, uploads them,
-// sets them to synced state for them to enter the GC index,
-// and measures the execution time of New function by creating
-// new databases with the same data directory.
-//
-// This benchmark takes significant amount of time.
-//
-// Measurements on MacBook Pro (Retina, 15-inch, Mid 2014) show
-// that New function executes around 1s for database with 1M chunks.
-//
-// # go test -benchmem -run=none github.com/ethereum/go-ethereum/swarm/storage/localstore -bench BenchmarkNew -v -timeout 20m
-// goos: darwin
-// goarch: amd64
-// pkg: github.com/ethereum/go-ethereum/swarm/storage/localstore
-// BenchmarkNew/1000-8         	     200	  11672414 ns/op	 9570960 B/op	   10008 allocs/op
-// BenchmarkNew/10000-8        	     100	  14890609 ns/op	10490118 B/op	    7759 allocs/op
-// BenchmarkNew/100000-8       	      20	  58334080 ns/op	17763157 B/op	   22978 allocs/op
-// BenchmarkNew/1000000-8      	       2	 748595153 ns/op	45297404 B/op	  253242 allocs/op
-// PASS
-func BenchmarkNew(b *testing.B) {
-	if testing.Short() {
-		b.Skip("skipping benchmark in short mode")
-	}
-	for _, count := range []int{
-		1000,
-		10000,
-		100000,
-		1000000,
-	} {
-		b.Run(strconv.Itoa(count), func(b *testing.B) {
-			dir, err := ioutil.TempDir("", "localstore-new-benchmark")
-			if err != nil {
-				b.Fatal(err)
-			}
-			defer os.RemoveAll(dir)
-			baseKey := make([]byte, 32)
-			if _, err := rand.Read(baseKey); err != nil {
-				b.Fatal(err)
-			}
-			db, err := New(dir, baseKey, nil)
-			if err != nil {
-				b.Fatal(err)
-			}
-			uploader := db.NewPutter(ModePutUpload)
-			syncer := db.NewSetter(ModeSetSync)
-			for i := 0; i < count; i++ {
-				chunk := generateFakeRandomChunk()
-				err := uploader.Put(chunk)
-				if err != nil {
-					b.Fatal(err)
-				}
-				err = syncer.Set(chunk.Address())
-				if err != nil {
-					b.Fatal(err)
-				}
-			}
-			err = db.Close()
-			if err != nil {
-				b.Fatal(err)
-			}
-			b.ResetTimer()
-
-			for n := 0; n < b.N; n++ {
-				b.StartTimer()
-				db, err := New(dir, baseKey, nil)
-				b.StopTimer()
-
-				if err != nil {
-					b.Fatal(err)
-				}
-				err = db.Close()
-				if err != nil {
-					b.Fatal(err)
-				}
-			}
-		})
-	}
-}
-
 // newTestDB is a helper function that constructs a
 // temporary database and returns a cleanup function that must
 // be called to remove the data.
@@ -396,7 +313,7 @@ func newItemsCountTest(i shed.Index, want int) func(t *testing.T) {
 // value is the same as the number of items in DB.gcIndex.
 func newIndexGCSizeTest(db *DB) func(t *testing.T) {
 	return func(t *testing.T) {
-		var want int64
+		var want uint64
 		err := db.gcIndex.Iterate(func(item shed.Item) (stop bool, err error) {
 			want++
 			return
@@ -404,7 +321,10 @@ func newIndexGCSizeTest(db *DB) func(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		got := db.getGCSize()
+		got, err := db.gcSize.Get()
+		if err != nil {
+			t.Fatal(err)
+		}
 		if got != want {
 			t.Errorf("got gc size %v, want %v", got, want)
 		}

swarm/storage/localstore/mode_get.go

@@ -113,16 +113,8 @@ func (db *DB) get(mode ModeGet, addr storage.Address) (out shed.Item, err error)
 // only Address and Data fields with non zero values,
 // which is ensured by the get function.
 func (db *DB) updateGC(item shed.Item) (err error) {
-	if db.useGlobalLock {
+	db.batchMu.Lock()
-		db.globalMu.Lock()
+	defer db.batchMu.Unlock()
-		defer db.globalMu.Unlock()
-	} else {
-		unlock, err := db.lockAddr(item.Address)
-		if err != nil {
-			return err
-		}
-		defer unlock()
-	}
 
 	batch := new(leveldb.Batch)
 

swarm/storage/localstore/mode_put.go

@@ -64,16 +64,8 @@ func (p *Putter) Put(ch storage.Chunk) (err error) {
 // with their nil values.
 func (db *DB) put(mode ModePut, item shed.Item) (err error) {
 	// protect parallel updates
-	if db.useGlobalLock {
+	db.batchMu.Lock()
-		db.globalMu.Lock()
+	defer db.batchMu.Unlock()
-		defer db.globalMu.Unlock()
-	} else {
-		unlock, err := db.lockAddr(item.Address)
-		if err != nil {
-			return err
-		}
-		defer unlock()
-	}
 
 	batch := new(leveldb.Batch)
 
@@ -121,7 +113,6 @@ func (db *DB) put(mode ModePut, item shed.Item) (err error) {
 		db.retrievalAccessIndex.PutInBatch(batch, item)
 		// add new entry to gc index
 		db.gcIndex.PutInBatch(batch, item)
-		db.gcUncountedHashesIndex.PutInBatch(batch, item)
 		gcSizeChange++
 
 		db.retrievalDataIndex.PutInBatch(batch, item)
@@ -148,12 +139,14 @@ func (db *DB) put(mode ModePut, item shed.Item) (err error) {
 		return ErrInvalidMode
 	}
 
-	err = db.shed.WriteBatch(batch)
+	err = db.incGCSizeInBatch(batch, gcSizeChange)
 	if err != nil {
 		return err
 	}
-	if gcSizeChange != 0 {
+
-		db.incGCSize(gcSizeChange)
+	err = db.shed.WriteBatch(batch)
+	if err != nil {
+		return err
 	}
 	if triggerPullFeed {
 		db.triggerPullSubscriptions(db.po(item.Address))

swarm/storage/localstore/mode_put_test.go

@@ -213,60 +213,31 @@ func TestModePutUpload_parallel(t *testing.T) {
 // goos: darwin
 // goarch: amd64
 // pkg: github.com/ethereum/go-ethereum/swarm/storage/localstore
-// BenchmarkPutUpload/count_100_parallel_1-addr_lock-8         	     300	   5075184 ns/op	 2081455 B/op	    2374 allocs/op
+// BenchmarkPutUpload/count_100_parallel_1-8         	     300	   4955055 ns/op	 2061388 B/op	    1754 allocs/op
-// BenchmarkPutUpload/count_100_parallel_1-glob_lock-8         	     300	   5032374 ns/op	 2061207 B/op	    1772 allocs/op
+// BenchmarkPutUpload/count_100_parallel_2-8         	     300	   5162484 ns/op	 2061452 B/op	    1755 allocs/op
-// BenchmarkPutUpload/count_100_parallel_2-addr_lock-8         	     300	   5079732 ns/op	 2081731 B/op	    2370 allocs/op
+// BenchmarkPutUpload/count_100_parallel_4-8         	     300	   5260477 ns/op	 2061655 B/op	    1756 allocs/op
-// BenchmarkPutUpload/count_100_parallel_2-glob_lock-8         	     300	   5179478 ns/op	 2061380 B/op	    1773 allocs/op
+// BenchmarkPutUpload/count_100_parallel_8-8         	     300	   5381812 ns/op	 2061843 B/op	    1758 allocs/op
-// BenchmarkPutUpload/count_100_parallel_4-addr_lock-8         	     500	   3748581 ns/op	 2081535 B/op	    2323 allocs/op
+// BenchmarkPutUpload/count_100_parallel_16-8        	     300	   5477313 ns/op	 2062115 B/op	    1762 allocs/op
-// BenchmarkPutUpload/count_100_parallel_4-glob_lock-8         	     300	   5367513 ns/op	 2061337 B/op	    1774 allocs/op
+// BenchmarkPutUpload/count_100_parallel_32-8        	     300	   5565273 ns/op	 2062965 B/op	    1775 allocs/op
-// BenchmarkPutUpload/count_100_parallel_8-addr_lock-8         	     500	   3311724 ns/op	 2082696 B/op	    2297 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_1-8        	      20	  75632247 ns/op	25009474 B/op	   17204 allocs/op
-// BenchmarkPutUpload/count_100_parallel_8-glob_lock-8         	     300	   5677622 ns/op	 2061636 B/op	    1776 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_2-8        	      20	  78194544 ns/op	25009064 B/op	   17205 allocs/op
-// BenchmarkPutUpload/count_100_parallel_16-addr_lock-8        	     500	   3606605 ns/op	 2085559 B/op	    2282 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_4-8        	      20	  77413001 ns/op	25010023 B/op	   17206 allocs/op
-// BenchmarkPutUpload/count_100_parallel_16-glob_lock-8        	     300	   6057814 ns/op	 2062032 B/op	    1780 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_8-8        	      20	  77406586 ns/op	25010968 B/op	   17206 allocs/op
-// BenchmarkPutUpload/count_100_parallel_32-addr_lock-8        	     500	   3720995 ns/op	 2089247 B/op	    2280 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_16-8       	      20	  81943323 ns/op	25006622 B/op	   17209 allocs/op
-// BenchmarkPutUpload/count_100_parallel_32-glob_lock-8        	     200	   6186910 ns/op	 2062744 B/op	    1789 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_32-8       	      20	  84393475 ns/op	25009450 B/op	   17222 allocs/op
-// BenchmarkPutUpload/count_1000_parallel_1-addr_lock-8        	      20	  84397760 ns/op	25210142 B/op	   23222 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_1-8       	       2	 612973544 ns/op	214429212 B/op	  186539 allocs/op
-// BenchmarkPutUpload/count_1000_parallel_1-glob_lock-8        	      20	  83432699 ns/op	25011813 B/op	   17222 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_2-8       	       2	 613744836 ns/op	214525364 B/op	  188857 allocs/op
-// BenchmarkPutUpload/count_1000_parallel_2-addr_lock-8        	      20	  80471064 ns/op	25208653 B/op	   23182 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_4-8       	       2	 619848337 ns/op	214437448 B/op	  188043 allocs/op
-// BenchmarkPutUpload/count_1000_parallel_2-glob_lock-8        	      20	  87841819 ns/op	25008899 B/op	   17223 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_8-8       	       2	 612132728 ns/op	214492440 B/op	  188061 allocs/op
-// BenchmarkPutUpload/count_1000_parallel_4-addr_lock-8        	      20	  71364750 ns/op	25206981 B/op	   22704 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_16-8      	       2	 625959679 ns/op	214493172 B/op	  188840 allocs/op
-// BenchmarkPutUpload/count_1000_parallel_4-glob_lock-8        	      20	  91491913 ns/op	25013307 B/op	   17225 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_32-8      	       2	 652223974 ns/op	214648080 B/op	  188916 allocs/op
-// BenchmarkPutUpload/count_1000_parallel_8-addr_lock-8        	      20	  67776485 ns/op	25210323 B/op	   22315 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_1-8      	       1	22682989072 ns/op	2317757256 B/op	 3486655 allocs/op
-// BenchmarkPutUpload/count_1000_parallel_8-glob_lock-8        	      20	  88658733 ns/op	25008864 B/op	   17228 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_2-8      	       1	23928779747 ns/op	2339295256 B/op	 3621696 allocs/op
-// BenchmarkPutUpload/count_1000_parallel_16-addr_lock-8       	      20	  61599020 ns/op	25213746 B/op	   22000 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_4-8      	       1	22704591819 ns/op	2317971752 B/op	 3495423 allocs/op
-// BenchmarkPutUpload/count_1000_parallel_16-glob_lock-8       	      20	  92734980 ns/op	25012744 B/op	   17228 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_8-8      	       1	22654015490 ns/op	2320451336 B/op	 3506505 allocs/op
-// BenchmarkPutUpload/count_1000_parallel_32-addr_lock-8       	      20	  57465216 ns/op	25224471 B/op	   21844 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_16-8     	       1	23192344424 ns/op	2326781648 B/op	 3540538 allocs/op
-// BenchmarkPutUpload/count_1000_parallel_32-glob_lock-8       	      20	  92420562 ns/op	25013237 B/op	   17244 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_32-8     	       1	24188298331 ns/op	2344201416 B/op	 3651945 allocs/op
-// BenchmarkPutUpload/count_10000_parallel_1-addr_lock-8       	       2	 611387455 ns/op	216747724 B/op	  248218 allocs/op
-// BenchmarkPutUpload/count_10000_parallel_1-glob_lock-8       	       2	 616212255 ns/op	214871528 B/op	  188983 allocs/op
-// BenchmarkPutUpload/count_10000_parallel_2-addr_lock-8       	       2	 576871975 ns/op	216552736 B/op	  246849 allocs/op
-// BenchmarkPutUpload/count_10000_parallel_2-glob_lock-8       	       2	 601008305 ns/op	214713748 B/op	  188931 allocs/op
-// BenchmarkPutUpload/count_10000_parallel_4-addr_lock-8       	       2	 551001371 ns/op	216701032 B/op	  241935 allocs/op
-// BenchmarkPutUpload/count_10000_parallel_4-glob_lock-8       	       2	 605576690 ns/op	214719292 B/op	  188949 allocs/op
-// BenchmarkPutUpload/count_10000_parallel_8-addr_lock-8       	       2	 504949238 ns/op	216431280 B/op	  236326 allocs/op
-// BenchmarkPutUpload/count_10000_parallel_8-glob_lock-8       	       2	 611631748 ns/op	214809276 B/op	  188957 allocs/op
-// BenchmarkPutUpload/count_10000_parallel_16-addr_lock-8      	       3	 510030296 ns/op	216088080 B/op	  231171 allocs/op
-// BenchmarkPutUpload/count_10000_parallel_16-glob_lock-8      	       2	 611416284 ns/op	214855916 B/op	  189724 allocs/op
-// BenchmarkPutUpload/count_10000_parallel_32-addr_lock-8      	       3	 481631118 ns/op	215341840 B/op	  224716 allocs/op
-// BenchmarkPutUpload/count_10000_parallel_32-glob_lock-8      	       2	 633612977 ns/op	214904164 B/op	  189775 allocs/op
-// BenchmarkPutUpload/count_100000_parallel_1-addr_lock-8      	       1	23289076334 ns/op	2354337552 B/op	 4190917 allocs/op
-// BenchmarkPutUpload/count_100000_parallel_1-glob_lock-8      	       1	22155535580 ns/op	2312803760 B/op	 3455566 allocs/op
-// BenchmarkPutUpload/count_100000_parallel_2-addr_lock-8      	       1	21908455154 ns/op	2328191128 B/op	 4014009 allocs/op
-// BenchmarkPutUpload/count_100000_parallel_2-glob_lock-8      	       1	22956308053 ns/op	2325078528 B/op	 3530270 allocs/op
-// BenchmarkPutUpload/count_100000_parallel_4-addr_lock-8      	       1	22334786914 ns/op	2338677488 B/op	 4028700 allocs/op
-// BenchmarkPutUpload/count_100000_parallel_4-glob_lock-8      	       1	23222406988 ns/op	2334153480 B/op	 3580197 allocs/op
-// BenchmarkPutUpload/count_100000_parallel_8-addr_lock-8      	       1	21569685948 ns/op	2322310120 B/op	 3880022 allocs/op
-// BenchmarkPutUpload/count_100000_parallel_8-glob_lock-8      	       1	22730998001 ns/op	2318311616 B/op	 3494378 allocs/op
-// BenchmarkPutUpload/count_100000_parallel_16-addr_lock-8     	       1	22005406658 ns/op	2324345744 B/op	 3862100 allocs/op
-// BenchmarkPutUpload/count_100000_parallel_16-glob_lock-8     	       1	24246335163 ns/op	2341373784 B/op	 3626749 allocs/op
-// BenchmarkPutUpload/count_100000_parallel_32-addr_lock-8     	       1	22764682771 ns/op	2332867552 B/op	 3896808 allocs/op
-// BenchmarkPutUpload/count_100000_parallel_32-glob_lock-8     	       1	24617688531 ns/op	2343609240 B/op	 3647404 allocs/op
 // PASS
-//
-// As expected, global lock introduces performance penalty, but in much less degree then expected.
-// Higher levels of parallelization do not give high level of performance boost. For 8 parallel
-// uploads on 8 core benchmark, the speedup is only ~1.72x at best. There is no significant difference
-// when a larger number of chunks is uploaded.
 func BenchmarkPutUpload(b *testing.B) {
 	for _, count := range []int{
 		100,
@@ -283,16 +254,11 @@ func BenchmarkPutUpload(b *testing.B) {
 			32,
 		} {
 			name := fmt.Sprintf("count %v parallel %v", count, maxParallelUploads)
-			b.Run(name+"-addr_lock", func(b *testing.B) {
+			b.Run(name, func(b *testing.B) {
 				for n := 0; n < b.N; n++ {
 					benchmarkPutUpload(b, nil, count, maxParallelUploads)
 				}
 			})
-			b.Run(name+"-glob_lock", func(b *testing.B) {
-				for n := 0; n < b.N; n++ {
-					benchmarkPutUpload(b, &Options{useGlobalLock: true}, count, maxParallelUploads)
-				}
-			})
 		}
 	}
 }

swarm/storage/localstore/mode_set.go

@@ -63,16 +63,8 @@ func (s *Setter) Set(addr storage.Address) (err error) {
 // of this function for the same address in parallel.
 func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
 	// protect parallel updates
-	if db.useGlobalLock {
+	db.batchMu.Lock()
-		db.globalMu.Lock()
+	defer db.batchMu.Unlock()
-		defer db.globalMu.Unlock()
-	} else {
-		unlock, err := db.lockAddr(addr)
-		if err != nil {
-			return err
-		}
-		defer unlock()
-	}
 
 	batch := new(leveldb.Batch)
 
@@ -118,7 +110,6 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
 		db.pullIndex.PutInBatch(batch, item)
 		triggerPullFeed = true
 		db.gcIndex.PutInBatch(batch, item)
-		db.gcUncountedHashesIndex.PutInBatch(batch, item)
 		gcSizeChange++
 
 	case ModeSetSync:
@@ -156,7 +147,6 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
 		db.retrievalAccessIndex.PutInBatch(batch, item)
 		db.pushIndex.DeleteInBatch(batch, item)
 		db.gcIndex.PutInBatch(batch, item)
-		db.gcUncountedHashesIndex.PutInBatch(batch, item)
 		gcSizeChange++
 
 	case modeSetRemove:
@@ -184,7 +174,6 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
 		db.retrievalAccessIndex.DeleteInBatch(batch, item)
 		db.pullIndex.DeleteInBatch(batch, item)
 		db.gcIndex.DeleteInBatch(batch, item)
-		db.gcUncountedHashesIndex.DeleteInBatch(batch, item)
 		// a check is needed for decrementing gcSize
 		// as delete is not reporting if the key/value pair
 		// is deleted or not
@@ -196,12 +185,14 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
 		return ErrInvalidMode
 	}
 
-	err = db.shed.WriteBatch(batch)
+	err = db.incGCSizeInBatch(batch, gcSizeChange)
 	if err != nil {
 		return err
 	}
-	if gcSizeChange != 0 {
+
-		db.incGCSize(gcSizeChange)
+	err = db.shed.WriteBatch(batch)
+	if err != nil {
+		return err
 	}
 	if triggerPullFeed {
 		db.triggerPullSubscriptions(db.po(item.Address))