swarm/storage/localstore: BenchmarkPutUpload and global lock option

swarm/storage/localstore/gc.go

@@ -81,11 +81,16 @@ func (db *DB) collectGarbage() (collectedCount int64, done bool, err error) {
 	done = true
 	err = db.gcIndex.Iterate(func(item shed.Item) (stop bool, err error) {
 		// protect parallel updates
+		if db.useGlobalLock {
+			db.globalMu.Lock()
+			defer db.globalMu.Unlock()
+		} else {
 			unlock, err := db.lockAddr(item.Address)
 			if err != nil {
 				return false, err
 			}
 			defer unlock()
+		}
 
 		gcSize := db.getGCSize()
 		if gcSize-collectedCount <= target {

swarm/storage/localstore/localstore.go

@@ -104,6 +104,12 @@ type DB struct {
 
 	addressLocks sync.Map
 
+	// 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
 	close chan struct{}
@@ -122,6 +128,9 @@ type Options struct {
 	Capacity int64
 	// 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
@@ -133,6 +142,7 @@ 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
 		// need to be buffered with the size of 1

swarm/storage/localstore/mode_get.go

@@ -113,11 +113,16 @@ 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.globalMu.Lock()
+		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,11 +64,16 @@ 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.globalMu.Lock()
+		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_test.go

@@ -18,6 +18,7 @@ package localstore
 
 import (
 	"bytes"
+	"fmt"
 	"sync"
 	"testing"
 	"time"
@@ -201,3 +202,129 @@ func TestModePutUpload_parallel(t *testing.T) {
 		}
 	}
 }
+
+// BenchmarkPutUpload runs a series of benchmarks that upload
+// a specific number of chunks in parallel.
+//
+// Measurements on MacBook Pro (Retina, 15-inch, Mid 2014)
+//
+// # go test -benchmem -run=none github.com/ethereum/go-ethereum/swarm/storage/localstore -bench BenchmarkPutUpload -v
+//
+// goos: darwin
+// goarch: amd64
+// pkg: github.com/ethereum/go-ethereum/swarm/storage/localstore
+// BenchmarkPutUpload/count_100_parallel_1-addr_lock-8         	     300         5955129 ns/op     2500357 B/op	    2672 allocs/op
+// BenchmarkPutUpload/count_100_parallel_1-glob_lock-8         	     300         5693210 ns/op     2480057 B/op	    2070 allocs/op
+// BenchmarkPutUpload/count_100_parallel_2-addr_lock-8         	     300         5147344 ns/op     2500580 B/op	    2673 allocs/op
+// BenchmarkPutUpload/count_100_parallel_2-glob_lock-8         	     300         5801207 ns/op     2480237 B/op	    2072 allocs/op
+// BenchmarkPutUpload/count_100_parallel_4-addr_lock-8         	     500         3900634 ns/op     2500283 B/op	    2630 allocs/op
+// BenchmarkPutUpload/count_100_parallel_4-glob_lock-8         	     300         5956225 ns/op     2480160 B/op	    2071 allocs/op
+// BenchmarkPutUpload/count_100_parallel_8-addr_lock-8         	     500         3204571 ns/op     2500840 B/op	    2604 allocs/op
+// BenchmarkPutUpload/count_100_parallel_8-glob_lock-8         	     200         5804689 ns/op     2480354 B/op	    2073 allocs/op
+// BenchmarkPutUpload/count_100_parallel_16-addr_lock-8        	     500         3209578 ns/op     2502570 B/op	    2609 allocs/op
+// BenchmarkPutUpload/count_100_parallel_16-glob_lock-8        	     300         5868150 ns/op     2480533 B/op	    2076 allocs/op
+// BenchmarkPutUpload/count_100_parallel_32-addr_lock-8        	     500         3091060 ns/op     2503923 B/op	    2634 allocs/op
+// BenchmarkPutUpload/count_100_parallel_32-glob_lock-8        	     300         5620684 ns/op     2481332 B/op	    2087 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_1-addr_lock-8        	      20        83724617 ns/op    29397827 B/op	   26226 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_1-glob_lock-8        	      20        79737650 ns/op    29202973 B/op	   20228 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_2-addr_lock-8        	      20        73382431 ns/op    29405901 B/op	   26234 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_2-glob_lock-8        	      20        87743895 ns/op    29200106 B/op	   20230 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_4-addr_lock-8        	      20        59550383 ns/op    29397483 B/op	   25761 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_4-glob_lock-8        	      20        80713765 ns/op    29195823 B/op	   20232 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_8-addr_lock-8        	      30        54826082 ns/op    29405468 B/op	   25448 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_8-glob_lock-8        	      20        82545759 ns/op    29205908 B/op	   20233 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_16-addr_lock-8       	      30        53334438 ns/op    29406540 B/op	   25332 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_16-glob_lock-8       	      20        81493550 ns/op    29205267 B/op	   20233 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_32-addr_lock-8       	      30        51840371 ns/op    29411834 B/op	   25336 allocs/op
+// BenchmarkPutUpload/count_1000_parallel_32-glob_lock-8       	      20        80898167 ns/op    29209452 B/op	   20234 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_1-addr_lock-8       	       2       668323148 ns/op   259038900 B/op	  280705 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_1-glob_lock-8       	       2       679351952 ns/op   257010124 B/op	  219969 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_2-addr_lock-8       	       2       666368239 ns/op   258808396 B/op	  278026 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_2-glob_lock-8       	       2       670005612 ns/op   256970316 B/op	  219983 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_4-addr_lock-8       	       2       551150500 ns/op   258527680 B/op	  272697 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_4-glob_lock-8       	       2       685501375 ns/op   256762796 B/op	  219901 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_8-addr_lock-8       	       2       518875154 ns/op   258491000 B/op	  268423 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_8-glob_lock-8       	       2       692095806 ns/op   256747644 B/op	  219858 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_16-addr_lock-8      	       2       528648421 ns/op   257939932 B/op	  264513 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_16-glob_lock-8      	       2       716251691 ns/op   256762568 B/op	  219120 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_32-addr_lock-8      	       3       473578608 ns/op   257253077 B/op	  259673 allocs/op
+// BenchmarkPutUpload/count_10000_parallel_32-glob_lock-8      	       2       676274817 ns/op   256824384 B/op	  219168 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_1-addr_lock-8      	       1     24740576226 ns/op  2778786256 B/op	 4525586 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_1-glob_lock-8      	       1     24704378905 ns/op  2760701208 B/op	 3930715 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_2-addr_lock-8      	       1     24391650224 ns/op  2778239744 B/op	 4501266 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_2-glob_lock-8      	       1     25900543952 ns/op  2750693384 B/op	 3870144 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_4-addr_lock-8      	       1     23036622183 ns/op  2756547704 B/op	 4316307 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_4-glob_lock-8      	       1     25068711098 ns/op  2761207392 B/op	 3935577 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_8-addr_lock-8      	       1     21948692932 ns/op  2742785760 B/op	 4196817 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_8-glob_lock-8      	       1     24591707861 ns/op  2760381320 B/op	 3929831 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_16-addr_lock-8     	       1     22399527760 ns/op  2750030272 B/op	 4218608 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_16-glob_lock-8     	       1     24758066757 ns/op  2749799200 B/op	 3864641 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_32-addr_lock-8     	       1     23118686208 ns/op  2762324560 B/op	 4283463 allocs/op
+// BenchmarkPutUpload/count_100000_parallel_32-glob_lock-8     	       1     25448525628 ns/op  2771420720 B/op	 3998428 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.5x at best.
+func BenchmarkPutUpload(b *testing.B) {
+	for _, count := range []int{
+		100,
+		1000,
+		10000,
+		100000,
+	} {
+		for _, maxParallelUploads := range []int{
+			1,
+			2,
+			4,
+			8,
+			16,
+			32,
+		} {
+			name := fmt.Sprintf("count %v parallel %v", count, maxParallelUploads)
+			b.Run(name+"-addr_lock", 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)
+				}
+			})
+		}
+	}
+}
+
+// benchmarkPutUpload runs a benchmark by uploading a specific number
+// of chunks with specified max parallel uploads.
+func benchmarkPutUpload(b *testing.B, o *Options, count, maxParallelUploads int) {
+	b.StopTimer()
+	db, cleanupFunc := newTestDB(b, o)
+	defer cleanupFunc()
+
+	uploader := db.NewPutter(ModePutUpload)
+	errs := make(chan error)
+	b.StartTimer()
+
+	go func() {
+		sem := make(chan struct{}, maxParallelUploads)
+		for i := 0; i < count; i++ {
+			sem <- struct{}{}
+
+			go func() {
+				defer func() { <-sem }()
+
+				chunk := generateFakeRandomChunk()
+				errs <- uploader.Put(chunk)
+			}()
+		}
+	}()
+
+	for i := 0; i < count; i++ {
+		err := <-errs
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}

swarm/storage/localstore/mode_set.go

@@ -63,11 +63,16 @@ 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.globalMu.Lock()
+		defer db.globalMu.Unlock()
+	} else {
 		unlock, err := db.lockAddr(addr)
 		if err != nil {
 			return err
 		}
 		defer unlock()
+	}
 
 	batch := new(leveldb.Batch)