swarm/storage/localstore: change how batches are written

swarm/storage/localstore/accessor.go

@@ -41,8 +41,8 @@ func (db *DB) Accessor(mode Mode) *Accessor {
 }
 
 // Put uses the underlying DB for the specific mode of update to store the chunk.
-func (u *Accessor) Put(ctx context.Context, ch storage.Chunk) error {
+func (u *Accessor) Put(_ context.Context, ch storage.Chunk) error {
-	return u.db.update(ctx, u.mode, chunkToItem(ch))
+	return u.db.update(u.mode, chunkToItem(ch))
 }
 
 // Get uses the underlying DB for the specific mode of access to get the chunk.

swarm/storage/localstore/localstore.go

@@ -27,17 +27,16 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/storage/mock"
 )
 
-const (
-	// maximal time for DB.Close must return
-	closeTimeout = 10 * time.Second
-)
-
 var (
 	// ErrInvalidMode is retuned when an unknown Mode
 	// is provided to the function.
 	ErrInvalidMode = errors.New("invalid mode")
 	// ErrDBClosed is returned when database is closed.
 	ErrDBClosed = errors.New("db closed")
+	// ErrUpdateLockTimeout is returned when the same chunk
+	// is updated in parallel and one of the updates
+	// takse longer then the configured timeout duration.
+	ErrUpdateLockTimeout = errors.New("update lock timeout")
 )
 
 // DB is the local store implementation and holds
@@ -46,8 +45,7 @@ type DB struct {
 	shed *shed.DB
 
 	// fields
-	schemaName  shed.StringField
+	schemaName shed.StringField
-	sizeCounter shed.Uint64Field
 
 	// this flag is for banchmarking two types of retrieval indexes
 	// - single retrieval composite index retrievalCompositeIndex
@@ -67,11 +65,7 @@ type DB struct {
 
 	baseKey []byte
 
-	batch        *batch        // current batch
+	updateLocks sync.Map
-	mu           sync.RWMutex  // mutex for accessing current batch
-	writeTrigger chan struct{} // channel to signal current write batch
-	writeDone    chan struct{} // closed when writeBatches function returns
-	close        chan struct{} // closed on Close, signals other goroutines to terminate
 }
 
 // Options struct holds optional parameters for configuring DB.
@@ -90,10 +84,6 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
 	db = &DB{
 		baseKey:                    baseKey,
 		useRetrievalCompositeIndex: o.UseRetrievalCompositeIndex,
-		batch:                      newBatch(),
-		writeTrigger:               make(chan struct{}, 1),
-		close:                      make(chan struct{}),
-		writeDone:                  make(chan struct{}),
 	}
 
 	db.shed, err = shed.NewDB(path)
@@ -105,10 +95,6 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
 	if err != nil {
 		return nil, err
 	}
-	db.sizeCounter, err = db.shed.NewUint64Field("size")
-	if err != nil {
-		return nil, err
-	}
 	if db.useRetrievalCompositeIndex {
 		var (
 			encodeValueFunc func(fields shed.IndexItem) (value []byte, err error)
@@ -305,54 +291,14 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
 	if err != nil {
 		return nil, err
 	}
-	// start goroutine that writes batches
-	go db.writeBatches()
 	return db, nil
 }
 
 // Close closes the underlying database.
 func (db *DB) Close() (err error) {
-	// signal other goroutines that
-	// the database is closing
-	close(db.close)
-	select {
-	// wait for writeBatches to write
-	// the last batch
-	case <-db.writeDone:
-	// closing timeout
-	case <-time.After(closeTimeout):
-	}
 	return db.shed.Close()
 }
 
-// writeBatches is a forever loop handing out the current batch apply
-// the batch when the db is free.
-func (db *DB) writeBatches() {
-	// close the writeDone channel
-	// so the DB.Close can return
-	defer close(db.writeDone)
-
-	write := func() {
-		db.mu.Lock()
-		b := db.batch
-		db.batch = newBatch()
-		db.mu.Unlock()
-		b.Err = db.shed.WriteBatch(b.Batch)
-		close(b.Done)
-	}
-	for {
-		select {
-		case <-db.writeTrigger:
-			write()
-		case <-db.close:
-			// check it there is a batch
-			// left to be written
-			write()
-			return
-		}
-	}
-}
-
 // po computes the proximity order between the address
 // and database base key.
 func (db *DB) po(addr storage.Address) (bin uint8) {

swarm/storage/localstore/mode.go

@@ -17,7 +17,8 @@
 package localstore
 
 import (
-	"context"
+	"encoding/hex"
+	"time"
 
 	"github.com/ethereum/go-ethereum/swarm/shed"
 	"github.com/syndtr/goleveldb/leveldb"
@@ -81,96 +82,67 @@ func (db *DB) access(mode Mode, item shed.IndexItem) (out shed.IndexItem, err er
 	switch mode {
 	case ModeRequest, modeAccess:
 		// update the access timestamp and fc index
-		return out, db.update(context.TODO(), mode, out)
+		return out, db.update(mode, out)
 	default:
 		// all other modes are not updating the index
 	}
 	return out, nil
 }
 
-// update is called by an Accessor with a specific Mode,
+var (
-// and also in access for updating access timestamp and gc index.
+	updateLockTimeout    = 3 * time.Second
-// This function calls updateBatch to perform operations
+	updateLockCheckDelay = 30 * time.Microsecond
-// on indexes and fields within a single batch.
+)
-func (db *DB) update(ctx context.Context, mode Mode, item shed.IndexItem) error {
-	db.mu.RLock()
-	b := db.batch
-	db.mu.RUnlock()
 
-	// check if the database is not closed
+// update performs different operations on fields and indexes
-	select {
-	case <-db.close:
-		return ErrDBClosed
-	default:
-	}
-
-	// call the update with the provided mode
-	err := db.updateBatch(b, mode, item)
-	if err != nil {
-		return err
-	}
-	// trigger the writeBatches loop
-	select {
-	case db.writeTrigger <- struct{}{}:
-	default:
-	}
-	// wait for batch to be written and return batch error
-	// this is in order for Put calls to be synchronous
-	select {
-	case <-b.Done:
-	case <-ctx.Done():
-		return ctx.Err()
-	}
-	return b.Err
-}
-
-// batch wraps leveldb.Batch extending it with a done channel.
-type batch struct {
-	*leveldb.Batch
-	Done chan struct{} // to signal when batch is written
-	Err  error         // error resulting from write
-}
-
-// newBatch constructs a new batch.
-func newBatch() *batch {
-	return &batch{
-		Batch: new(leveldb.Batch),
-		Done:  make(chan struct{}),
-	}
-}
-
-// updateBatch performs different operations on fields and indexes
 // depending on the provided Mode.
-func (db *DB) updateBatch(b *batch, mode Mode, item shed.IndexItem) (err error) {
+// It protects parallel updates of items with the same address
+// with updateLocks map and waiting using a simple for loop.
+func (db *DB) update(mode Mode, item shed.IndexItem) (err error) {
+	// protect parallel updates
+	start := time.Now()
+	lockKey := hex.EncodeToString(item.Address)
+	for {
+		_, loaded := db.updateLocks.LoadOrStore(lockKey, struct{}{})
+		if !loaded {
+			break
+		}
+		time.Sleep(updateLockCheckDelay)
+		if time.Since(start) > updateLockTimeout {
+			return ErrUpdateLockTimeout
+		}
+	}
+	defer db.updateLocks.Delete(lockKey)
+
+	batch := new(leveldb.Batch)
+
 	switch mode {
 	case ModeSyncing:
 		// put to indexes: retrieve, pull
 		item.StoreTimestamp = now()
 		if db.useRetrievalCompositeIndex {
-			db.retrievalCompositeIndex.PutInBatch(b.Batch, item)
+			db.retrievalCompositeIndex.PutInBatch(batch, item)
 		} else {
-			db.retrievalDataIndex.PutInBatch(b.Batch, item)
+			db.retrievalDataIndex.PutInBatch(batch, item)
 		}
-		db.pullIndex.PutInBatch(b.Batch, item)
+		db.pullIndex.PutInBatch(batch, item)
-		db.sizeCounter.IncInBatch(b.Batch)
 
 	case ModeUpload:
 		// put to indexes: retrieve, push, pull
 		item.StoreTimestamp = now()
 		if db.useRetrievalCompositeIndex {
-			db.retrievalCompositeIndex.PutInBatch(b.Batch, item)
+			db.retrievalCompositeIndex.PutInBatch(batch, item)
 		} else {
-			db.retrievalDataIndex.PutInBatch(b.Batch, item)
+			db.retrievalDataIndex.PutInBatch(batch, item)
 		}
-		db.pullIndex.PutInBatch(b.Batch, item)
+		db.pullIndex.PutInBatch(batch, item)
-		db.pushIndex.PutInBatch(b.Batch, item)
+		db.pushIndex.PutInBatch(batch, item)
-		db.sizeCounter.IncInBatch(b.Batch)
 
 	case ModeRequest:
 		// update accessTimeStamp in retrieve, gc
 
 		if db.useRetrievalCompositeIndex {
-			// access timestap is already populated
+			// access timestamp is already populated
 			// in the provided item, passed from access function.
 		} else {
 			i, err := db.retrievalAccessIndex.Get(item)
@@ -189,17 +161,17 @@ func (db *DB) updateBatch(b *batch, mode Mode, item shed.IndexItem) (err error)
 			return nil
 		}
 		// delete current entry from the gc index
-		db.gcIndex.DeleteInBatch(b.Batch, item)
+		db.gcIndex.DeleteInBatch(batch, item)
 		// update access timestamp
 		item.AccessTimestamp = now()
 		// update retrieve access index
 		if db.useRetrievalCompositeIndex {
-			db.retrievalCompositeIndex.PutInBatch(b.Batch, item)
+			db.retrievalCompositeIndex.PutInBatch(batch, item)
 		} else {
-			db.retrievalAccessIndex.PutInBatch(b.Batch, item)
+			db.retrievalAccessIndex.PutInBatch(batch, item)
 		}
 		// add new entry to gc index
-		db.gcIndex.PutInBatch(b.Batch, item)
+		db.gcIndex.PutInBatch(batch, item)
 
 	case ModeSynced:
 		// delete from push, insert to gc
@@ -215,7 +187,7 @@ func (db *DB) updateBatch(b *batch, mode Mode, item shed.IndexItem) (err error)
 					// no need to update gc index
 					// just delete from the push index
 					// if it is there
-					db.pushIndex.DeleteInBatch(b.Batch, item)
+					db.pushIndex.DeleteInBatch(batch, item)
 					return nil
 				}
 				return err
@@ -226,7 +198,7 @@ func (db *DB) updateBatch(b *batch, mode Mode, item shed.IndexItem) (err error)
 				// the chunk is not accessed before
 				// set access time for gc index
 				item.AccessTimestamp = now()
-				db.retrievalCompositeIndex.PutInBatch(b.Batch, item)
+				db.retrievalCompositeIndex.PutInBatch(batch, item)
 			}
 		} else {
 			i, err := db.retrievalDataIndex.Get(item)
@@ -236,7 +208,7 @@ func (db *DB) updateBatch(b *batch, mode Mode, item shed.IndexItem) (err error)
 					// no need to update gc index
 					// just delete from the push index
 					// if it is there
-					db.pushIndex.DeleteInBatch(b.Batch, item)
+					db.pushIndex.DeleteInBatch(batch, item)
 					return nil
 				}
 				return err
@@ -247,24 +219,24 @@ func (db *DB) updateBatch(b *batch, mode Mode, item shed.IndexItem) (err error)
 			switch err {
 			case nil:
 				item.AccessTimestamp = i.AccessTimestamp
-				db.gcIndex.DeleteInBatch(b.Batch, item)
+				db.gcIndex.DeleteInBatch(batch, item)
 			case leveldb.ErrNotFound:
 				// the chunk is not accessed before
 			default:
 				return err
 			}
 			item.AccessTimestamp = now()
-			db.retrievalAccessIndex.PutInBatch(b.Batch, item)
+			db.retrievalAccessIndex.PutInBatch(batch, item)
 		}
-		db.pushIndex.DeleteInBatch(b.Batch, item)
+		db.pushIndex.DeleteInBatch(batch, item)
-		db.gcIndex.PutInBatch(b.Batch, item)
+		db.gcIndex.PutInBatch(batch, item)
 
 	// Q: modeAccess and ModeRequest are very similar,  why do we need both?
 	case modeAccess:
 		// update accessTimeStamp in retrieve, pull, gc
 
 		if db.useRetrievalCompositeIndex {
-			// access timestap is already populated
+			// access timestamp is already populated
 			// in the provided item, passed from access function.
 		} else {
 			i, err := db.retrievalAccessIndex.Get(item)
@@ -278,24 +250,24 @@ func (db *DB) updateBatch(b *batch, mode Mode, item shed.IndexItem) (err error)
 			}
 		}
 		// Q: why do we need to update this index?
-		db.pullIndex.PutInBatch(b.Batch, item)
+		db.pullIndex.PutInBatch(batch, item)
 		if item.AccessTimestamp == 0 {
 			// chunk is not yes synced
 			// do not add it to the gc index
 			return nil
 		}
 		// delete current entry from the gc index
-		db.gcIndex.DeleteInBatch(b.Batch, item)
+		db.gcIndex.DeleteInBatch(batch, item)
 		// update access timestamp
 		item.AccessTimestamp = now()
 		// update retrieve access index
 		if db.useRetrievalCompositeIndex {
-			db.retrievalCompositeIndex.PutInBatch(b.Batch, item)
+			db.retrievalCompositeIndex.PutInBatch(batch, item)
 		} else {
-			db.retrievalAccessIndex.PutInBatch(b.Batch, item)
+			db.retrievalAccessIndex.PutInBatch(batch, item)
 		}
 		// add new entry to gc index
-		db.gcIndex.PutInBatch(b.Batch, item)
+		db.gcIndex.PutInBatch(batch, item)
 
 	case modeRemoval:
 		// delete from retrieve, pull, gc
@@ -326,17 +298,17 @@ func (db *DB) updateBatch(b *batch, mode Mode, item shed.IndexItem) (err error)
 			item.StoreTimestamp = i.StoreTimestamp
 		}
 		if db.useRetrievalCompositeIndex {
-			db.retrievalCompositeIndex.DeleteInBatch(b.Batch, item)
+			db.retrievalCompositeIndex.DeleteInBatch(batch, item)
 		} else {
-			db.retrievalDataIndex.DeleteInBatch(b.Batch, item)
+			db.retrievalDataIndex.DeleteInBatch(batch, item)
-			db.retrievalAccessIndex.DeleteInBatch(b.Batch, item)
+			db.retrievalAccessIndex.DeleteInBatch(batch, item)
 		}
-		db.pullIndex.DeleteInBatch(b.Batch, item)
+		db.pullIndex.DeleteInBatch(batch, item)
-		db.gcIndex.DeleteInBatch(b.Batch, item)
+		db.gcIndex.DeleteInBatch(batch, item)
-		db.sizeCounter.DecInBatch(b.Batch)
 
 	default:
 		return ErrInvalidMode
 	}
-	return nil
+
+	return db.shed.WriteBatch(batch)
 }

swarm/storage/localstore/mode_test.go

@@ -62,23 +62,14 @@ func testModeSyncingValues(t *testing.T, db *DB) {
 		return wantTimestamp
 	}
 
-	wantSize, err := db.sizeCounter.Get()
+	err := a.Put(context.Background(), chunk)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	err = a.Put(context.Background(), chunk)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	wantSize++
-
 	t.Run("retrieve indexes", newRetrieveIndexesTest(db, chunk, wantTimestamp, 0))
 
 	t.Run("pull index", newPullIndexTest(db, chunk, wantTimestamp, nil))
-
-	t.Run("size counter", testSizeCounter(db, wantSize))
 }
 
 // TestModeUpload validates internal data operations and state
@@ -112,25 +103,16 @@ func testModeUploadValues(t *testing.T, db *DB) {
 		return wantTimestamp
 	}
 
-	wantSize, err := db.sizeCounter.Get()
+	err := a.Put(context.Background(), chunk)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	err = a.Put(context.Background(), chunk)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	wantSize++
-
 	t.Run("retrieve indexes", newRetrieveIndexesTest(db, chunk, wantTimestamp, 0))
 
 	t.Run("pull index", newPullIndexTest(db, chunk, wantTimestamp, nil))
 
 	t.Run("push index", newPushIndexTest(db, chunk, wantTimestamp, nil))
-
-	t.Run("size counter", testSizeCounter(db, wantSize))
 }
 
 // TestModeRequest validates internal data operations and state
@@ -437,13 +419,6 @@ func testModeRemovalValues(t *testing.T, db *DB) {
 
 	a = db.Accessor(modeRemoval)
 
-	wantSize, err := db.sizeCounter.Get()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	wantSize--
-
 	err = a.Put(context.Background(), chunk)
 	if err != nil {
 		t.Fatal(err)
@@ -478,8 +453,6 @@ func testModeRemovalValues(t *testing.T, db *DB) {
 	t.Run("pull index count", newIndexItemsCountTest(db.pullIndex, 0))
 
 	t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 0))
-
-	t.Run("size counter", testSizeCounter(db, wantSize))
 }
 
 // TestDB_pullIndex validates the ordering of keys in pull index.
@@ -874,20 +847,6 @@ func testIndexItemsOrder(t *testing.T, i shed.Index, chunks []testIndexChunk, so
 	}
 }
 
-// testSizeCounter returns a test function that validates the expected
-// value from sizeCounter field.
-func testSizeCounter(db *DB, wantSize uint64) func(t *testing.T) {
-	return func(t *testing.T) {
-		got, err := db.sizeCounter.Get()
-		if err != nil {
-			t.Fatal(err)
-		}
-		if got != wantSize {
-			t.Errorf("got size counter value %v, want %v", got, wantSize)
-		}
-	}
-}
-
 // validateItem is a helper function that checks IndexItem values.
 func validateItem(t *testing.T, item shed.IndexItem, address, data []byte, storeTimestamp, accessTimestamp int64) {
 	t.Helper()