From 4319639f45a5cd8e32278487093d6a7f774677ab Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Wed, 30 Jan 2019 13:25:12 +0100 Subject: [PATCH] swarm/storage/localstore: global batch write lock --- swarm/storage/localstore/gc.go | 146 ++++++-------------- swarm/storage/localstore/gc_test.go | 34 +++-- swarm/storage/localstore/localstore.go | 119 ++-------------- swarm/storage/localstore/localstore_test.go | 90 +----------- swarm/storage/localstore/mode_get.go | 12 +- swarm/storage/localstore/mode_put.go | 21 +-- swarm/storage/localstore/mode_put_test.go | 84 ++++------- swarm/storage/localstore/mode_set.go | 23 +-- 8 files changed, 118 insertions(+), 411 deletions(-) diff --git a/swarm/storage/localstore/gc.go b/swarm/storage/localstore/gc.go index 05a4b5ec17..7571456e76 100644 --- a/swarm/storage/localstore/gc.go +++ b/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 { - db.globalMu.Lock() - defer db.globalMu.Unlock() + // protect database from changing idexes and gcSize + db.batchMu.Lock() + 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) { - return int64(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 +func (db *DB) gcTarget() (target uint64) { + return uint64(float64(db.capacity) * gcTargetRatio) } // triggerGarbageCollection signals collectGarbageWorker @@ -171,66 +133,40 @@ func (db *DB) triggerGarbageCollection() { } } -// writeGCSizeWorker writes gcSize on trigger event -// and waits writeGCSizeDelay after each write. -// It implements a linear backoff with delay of -// writeGCSizeDelay duration to avoid very frequent -// database operations. -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 - // iteration. This prevents frequent I/O operations. - select { - case <-time.After(10 * time.Second): - case <-db.close: - return - } - case <-db.close: - return - } +// incGCSizeInBatch changes gcSize field value +// by change which can be negative. +func (db *DB) incGCSizeInBatch(batch *leveldb.Batch, change int64) (err error) { + if change == 0 { + return nil } -} - -// 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) + gcSize, err := db.gcSize.Get() 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) diff --git a/swarm/storage/localstore/gc_test.go b/swarm/storage/localstore/gc_test.go index eb039a554a..90755582e7 100644 --- a/swarm/storage/localstore/gc_test.go +++ b/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) - defer setTestHookCollectGarbage(func(collectedCount int64) { + testHookCollectGarbageChan := make(chan uint64) + 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) - defer setTestHookCollectGarbage(func(collectedCount int64) { + testHookCollectGarbageChan := make(chan uint64) + 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 }) diff --git a/swarm/storage/localstore/localstore.go b/swarm/storage/localstore/localstore.go index a1908f134a..2b3e174cbe 100644 --- a/swarm/storage/localstore/localstore.go +++ b/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 - // it must be always read by getGCSize and - // set with incGCSize which are locking gcSizeMu - gcSize int64 - gcSizeMu sync.RWMutex + // field that stores number of intems in gc index + gcSize shed.Uint64Field + // 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 - - // 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 + batchMu 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 @@ -141,15 +121,13 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) { o = new(Options) } db = &DB{ - capacity: o.Capacity, - useGlobalLock: o.useGlobalLock, - baseKey: baseKey, - // channels collectGarbageTrigger and writeGCSizeTrigger - // need to be buffered with the size of 1 + capacity: o.Capacity, + baseKey: baseKey, + // channel collectGarbageTrigger + // 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{ diff --git a/swarm/storage/localstore/localstore_test.go b/swarm/storage/localstore/localstore_test.go index c7309d3cd8..3948cbd830 100644 --- a/swarm/storage/localstore/localstore_test.go +++ b/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) } diff --git a/swarm/storage/localstore/mode_get.go b/swarm/storage/localstore/mode_get.go index c5905834ce..de30a4dcde 100644 --- a/swarm/storage/localstore/mode_get.go +++ b/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.globalMu.Lock() - defer db.globalMu.Unlock() - } else { - unlock, err := db.lockAddr(item.Address) - if err != nil { - return err - } - defer unlock() - } + db.batchMu.Lock() + defer db.batchMu.Unlock() batch := new(leveldb.Batch) diff --git a/swarm/storage/localstore/mode_put.go b/swarm/storage/localstore/mode_put.go index 15e2e12412..2f948ae091 100644 --- a/swarm/storage/localstore/mode_put.go +++ b/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.globalMu.Lock() - defer db.globalMu.Unlock() - } else { - unlock, err := db.lockAddr(item.Address) - if err != nil { - return err - } - defer unlock() - } + db.batchMu.Lock() + defer db.batchMu.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)) diff --git a/swarm/storage/localstore/mode_put_test.go b/swarm/storage/localstore/mode_put_test.go index 164af12324..d8138e0395 100644 --- a/swarm/storage/localstore/mode_put_test.go +++ b/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-glob_lock-8 300 5032374 ns/op 2061207 B/op 1772 allocs/op -// BenchmarkPutUpload/count_100_parallel_2-addr_lock-8 300 5079732 ns/op 2081731 B/op 2370 allocs/op -// BenchmarkPutUpload/count_100_parallel_2-glob_lock-8 300 5179478 ns/op 2061380 B/op 1773 allocs/op -// BenchmarkPutUpload/count_100_parallel_4-addr_lock-8 500 3748581 ns/op 2081535 B/op 2323 allocs/op -// BenchmarkPutUpload/count_100_parallel_4-glob_lock-8 300 5367513 ns/op 2061337 B/op 1774 allocs/op -// BenchmarkPutUpload/count_100_parallel_8-addr_lock-8 500 3311724 ns/op 2082696 B/op 2297 allocs/op -// BenchmarkPutUpload/count_100_parallel_8-glob_lock-8 300 5677622 ns/op 2061636 B/op 1776 allocs/op -// BenchmarkPutUpload/count_100_parallel_16-addr_lock-8 500 3606605 ns/op 2085559 B/op 2282 allocs/op -// BenchmarkPutUpload/count_100_parallel_16-glob_lock-8 300 6057814 ns/op 2062032 B/op 1780 allocs/op -// BenchmarkPutUpload/count_100_parallel_32-addr_lock-8 500 3720995 ns/op 2089247 B/op 2280 allocs/op -// BenchmarkPutUpload/count_100_parallel_32-glob_lock-8 200 6186910 ns/op 2062744 B/op 1789 allocs/op -// BenchmarkPutUpload/count_1000_parallel_1-addr_lock-8 20 84397760 ns/op 25210142 B/op 23222 allocs/op -// BenchmarkPutUpload/count_1000_parallel_1-glob_lock-8 20 83432699 ns/op 25011813 B/op 17222 allocs/op -// BenchmarkPutUpload/count_1000_parallel_2-addr_lock-8 20 80471064 ns/op 25208653 B/op 23182 allocs/op -// BenchmarkPutUpload/count_1000_parallel_2-glob_lock-8 20 87841819 ns/op 25008899 B/op 17223 allocs/op -// BenchmarkPutUpload/count_1000_parallel_4-addr_lock-8 20 71364750 ns/op 25206981 B/op 22704 allocs/op -// BenchmarkPutUpload/count_1000_parallel_4-glob_lock-8 20 91491913 ns/op 25013307 B/op 17225 allocs/op -// BenchmarkPutUpload/count_1000_parallel_8-addr_lock-8 20 67776485 ns/op 25210323 B/op 22315 allocs/op -// BenchmarkPutUpload/count_1000_parallel_8-glob_lock-8 20 88658733 ns/op 25008864 B/op 17228 allocs/op -// BenchmarkPutUpload/count_1000_parallel_16-addr_lock-8 20 61599020 ns/op 25213746 B/op 22000 allocs/op -// BenchmarkPutUpload/count_1000_parallel_16-glob_lock-8 20 92734980 ns/op 25012744 B/op 17228 allocs/op -// BenchmarkPutUpload/count_1000_parallel_32-addr_lock-8 20 57465216 ns/op 25224471 B/op 21844 allocs/op -// BenchmarkPutUpload/count_1000_parallel_32-glob_lock-8 20 92420562 ns/op 25013237 B/op 17244 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 +// BenchmarkPutUpload/count_100_parallel_1-8 300 4955055 ns/op 2061388 B/op 1754 allocs/op +// BenchmarkPutUpload/count_100_parallel_2-8 300 5162484 ns/op 2061452 B/op 1755 allocs/op +// BenchmarkPutUpload/count_100_parallel_4-8 300 5260477 ns/op 2061655 B/op 1756 allocs/op +// BenchmarkPutUpload/count_100_parallel_8-8 300 5381812 ns/op 2061843 B/op 1758 allocs/op +// BenchmarkPutUpload/count_100_parallel_16-8 300 5477313 ns/op 2062115 B/op 1762 allocs/op +// BenchmarkPutUpload/count_100_parallel_32-8 300 5565273 ns/op 2062965 B/op 1775 allocs/op +// BenchmarkPutUpload/count_1000_parallel_1-8 20 75632247 ns/op 25009474 B/op 17204 allocs/op +// BenchmarkPutUpload/count_1000_parallel_2-8 20 78194544 ns/op 25009064 B/op 17205 allocs/op +// BenchmarkPutUpload/count_1000_parallel_4-8 20 77413001 ns/op 25010023 B/op 17206 allocs/op +// BenchmarkPutUpload/count_1000_parallel_8-8 20 77406586 ns/op 25010968 B/op 17206 allocs/op +// BenchmarkPutUpload/count_1000_parallel_16-8 20 81943323 ns/op 25006622 B/op 17209 allocs/op +// BenchmarkPutUpload/count_1000_parallel_32-8 20 84393475 ns/op 25009450 B/op 17222 allocs/op +// BenchmarkPutUpload/count_10000_parallel_1-8 2 612973544 ns/op 214429212 B/op 186539 allocs/op +// BenchmarkPutUpload/count_10000_parallel_2-8 2 613744836 ns/op 214525364 B/op 188857 allocs/op +// BenchmarkPutUpload/count_10000_parallel_4-8 2 619848337 ns/op 214437448 B/op 188043 allocs/op +// BenchmarkPutUpload/count_10000_parallel_8-8 2 612132728 ns/op 214492440 B/op 188061 allocs/op +// BenchmarkPutUpload/count_10000_parallel_16-8 2 625959679 ns/op 214493172 B/op 188840 allocs/op +// BenchmarkPutUpload/count_10000_parallel_32-8 2 652223974 ns/op 214648080 B/op 188916 allocs/op +// BenchmarkPutUpload/count_100000_parallel_1-8 1 22682989072 ns/op 2317757256 B/op 3486655 allocs/op +// BenchmarkPutUpload/count_100000_parallel_2-8 1 23928779747 ns/op 2339295256 B/op 3621696 allocs/op +// BenchmarkPutUpload/count_100000_parallel_4-8 1 22704591819 ns/op 2317971752 B/op 3495423 allocs/op +// BenchmarkPutUpload/count_100000_parallel_8-8 1 22654015490 ns/op 2320451336 B/op 3506505 allocs/op +// BenchmarkPutUpload/count_100000_parallel_16-8 1 23192344424 ns/op 2326781648 B/op 3540538 allocs/op +// BenchmarkPutUpload/count_100000_parallel_32-8 1 24188298331 ns/op 2344201416 B/op 3651945 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) - } - }) } } } diff --git a/swarm/storage/localstore/mode_set.go b/swarm/storage/localstore/mode_set.go index 5bf7e37dc0..7aeb2754b7 100644 --- a/swarm/storage/localstore/mode_set.go +++ b/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.globalMu.Lock() - defer db.globalMu.Unlock() - } else { - unlock, err := db.lockAddr(addr) - if err != nil { - return err - } - defer unlock() - } + db.batchMu.Lock() + defer db.batchMu.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))