diff --git a/swarm/storage/localstore/accessor.go b/swarm/storage/localstore/accessor.go index 79574b68d4..7ab5a43963 100644 --- a/swarm/storage/localstore/accessor.go +++ b/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 { - return u.db.update(ctx, u.mode, chunkToItem(ch)) +func (u *Accessor) Put(_ context.Context, ch storage.Chunk) error { + return u.db.update(u.mode, chunkToItem(ch)) } // Get uses the underlying DB for the specific mode of access to get the chunk. diff --git a/swarm/storage/localstore/localstore.go b/swarm/storage/localstore/localstore.go index d6fc0c7608..79e321db25 100644 --- a/swarm/storage/localstore/localstore.go +++ b/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 - sizeCounter shed.Uint64Field + schemaName shed.StringField // 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 - 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 + updateLocks sync.Map } // 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) { diff --git a/swarm/storage/localstore/mode.go b/swarm/storage/localstore/mode.go index ec67a387f9..fc597308fb 100644 --- a/swarm/storage/localstore/mode.go +++ b/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, -// and also in access for updating access timestamp and gc index. -// This function calls updateBatch to perform operations -// 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() +var ( + updateLockTimeout = 3 * time.Second + updateLockCheckDelay = 30 * time.Microsecond +) - // check if the database is not closed - 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 +// update 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.sizeCounter.IncInBatch(b.Batch) + db.pullIndex.PutInBatch(batch, item) 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.pushIndex.PutInBatch(b.Batch, item) - db.sizeCounter.IncInBatch(b.Batch) + db.pullIndex.PutInBatch(batch, item) + db.pushIndex.PutInBatch(batch, item) 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.gcIndex.PutInBatch(b.Batch, item) + db.pushIndex.DeleteInBatch(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.retrievalAccessIndex.DeleteInBatch(b.Batch, item) + db.retrievalDataIndex.DeleteInBatch(batch, item) + db.retrievalAccessIndex.DeleteInBatch(batch, item) } - db.pullIndex.DeleteInBatch(b.Batch, item) - db.gcIndex.DeleteInBatch(b.Batch, item) - db.sizeCounter.DecInBatch(b.Batch) + db.pullIndex.DeleteInBatch(batch, item) + db.gcIndex.DeleteInBatch(batch, item) default: return ErrInvalidMode } - return nil + + return db.shed.WriteBatch(batch) } diff --git a/swarm/storage/localstore/mode_test.go b/swarm/storage/localstore/mode_test.go index f47e8ded40..8b4ea48a97 100644 --- a/swarm/storage/localstore/mode_test.go +++ b/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()