diff --git a/swarm/storage/localstore/gc.go b/swarm/storage/localstore/gc.go index 80a42b8712..fc5205e77d 100644 --- a/swarm/storage/localstore/gc.go +++ b/swarm/storage/localstore/gc.go @@ -18,6 +18,7 @@ package localstore import ( "sync/atomic" + "time" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/swarm/shed" @@ -39,56 +40,24 @@ var ( gcBatchSize int64 = 1000 ) -// collectGarbage is a long running function that waits for +// collectGarbageWorker is a long running function that waits for // collectGarbageTrigger channel to signal a garbage collection // run. GC run iterates on gcIndex and removes older items // form retrieval and other indexes. -func (db *DB) collectGarbage() { - target := db.gcTarget() +func (db *DB) collectGarbageWorker() { for { select { case <-db.collectGarbageTrigger: - batch := new(leveldb.Batch) - - // sets a gc trigger if batch limit is reached - var triggerNextIteration bool - var collectedCount int64 - err := db.gcIndex.IterateAll(func(item shed.Item) (stop bool, err error) { - gcSize := atomic.LoadInt64(&db.gcSize) - if gcSize-collectedCount <= target { - return true, nil - } - // delete from retrieve, pull, gc - if db.useRetrievalCompositeIndex { - db.retrievalCompositeIndex.DeleteInBatch(batch, item) - } else { - db.retrievalDataIndex.DeleteInBatch(batch, item) - db.retrievalAccessIndex.DeleteInBatch(batch, item) - } - db.pullIndex.DeleteInBatch(batch, item) - db.gcIndex.DeleteInBatch(batch, item) - collectedCount++ - if collectedCount >= gcBatchSize { - triggerNextIteration = true - return true, nil - } - return false, nil - }) + // TODO: Add comment about done + collectedCount, done, err := db.collectGarbage() if err != nil { log.Error("localstore collect garbage", "err", err) } - - err = db.shed.WriteBatch(batch) - if err != nil { - log.Error("localstore collect garbage write batch", "err", err) - } else { - // batch is written, decrement gcSize and check if another gc run is needed - db.incGCSize(-collectedCount) - if triggerNextIteration { - select { - case db.collectGarbageTrigger <- struct{}{}: - default: - } + // check if another gc run is needed + if !done { + select { + case db.collectGarbageTrigger <- struct{}{}: + default: } } @@ -101,6 +70,53 @@ func (db *DB) collectGarbage() { } } +// collectGarbage removes chunks from retrieval and other +// indexes if maximal number of chunks in database is reached. +// This function returns the number of removed chunks. If done +// 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) { + batch := new(leveldb.Batch) + target := db.gcTarget() + + done = true + err = db.gcIndex.IterateAll(func(item shed.Item) (stop bool, err error) { + gcSize := atomic.LoadInt64(&db.gcSize) + if gcSize-collectedCount <= target { + return true, nil + } + // delete from retrieve, pull, gc + if db.useRetrievalCompositeIndex { + db.retrievalCompositeIndex.DeleteInBatch(batch, item) + } else { + db.retrievalDataIndex.DeleteInBatch(batch, item) + db.retrievalAccessIndex.DeleteInBatch(batch, item) + } + db.pullIndex.DeleteInBatch(batch, item) + db.gcIndex.DeleteInBatch(batch, item) + collectedCount++ + if collectedCount >= gcBatchSize { + // bach size limit reached, + // another gc run is needed + done = false + return true, nil + } + return false, nil + }) + if err != nil { + return 0, false, err + } + + 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) { @@ -110,7 +126,14 @@ func (db *DB) gcTarget() (target int64) { // 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 + } new := atomic.AddInt64(&db.gcSize, count) + select { + case db.writeGCSizeTrigger <- struct{}{}: + default: + } if new >= db.capacity { select { case db.collectGarbageTrigger <- struct{}{}: @@ -119,6 +142,46 @@ func (db *DB) incGCSize(count int64) { } } +var writeGCSizeDelay = 10 * time.Second + +// writeGCSizeWorker calls writeGCSize function +// on writeGCSizeTrigger receive. It implements a +// backoff with delay of writeGCSizeDelay duration +// to avoid very frequent database operations. +func (db *DB) writeGCSizeWorker() { + for { + select { + case <-db.writeGCSizeTrigger: + err := db.writeGCSize() + if err != nil { + log.Error("localstore write gc size", "err", err) + } + select { + case <-time.After(writeGCSizeDelay): + 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 database initialization +// when gcSize is counted. +func (db *DB) writeGCSize() (err error) { + gcSize := atomic.LoadInt64(&db.gcSize) + err = db.storedGCSize.Put(uint64(gcSize)) + if err != nil { + return err + } + return db.gcUncountedHashesIndex.IterateAll(func(item shed.Item) (stop bool, err error) { + return false, db.gcUncountedHashesIndex.Delete(item) + }) +} + // testHookCollectGarbage is a hook that can provide // information when a garbage collection run is done // and how many items it removed. diff --git a/swarm/storage/localstore/gc_test.go b/swarm/storage/localstore/gc_test.go index 07e7f52fe9..69d10e94fa 100644 --- a/swarm/storage/localstore/gc_test.go +++ b/swarm/storage/localstore/gc_test.go @@ -17,6 +17,9 @@ package localstore import ( + "io/ioutil" + "math/rand" + "os" "sync/atomic" "testing" "time" @@ -24,34 +27,34 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage" ) -// TestDB_collectGarbage tests garbage collection runs +// TestDB_collectGarbageWorker tests garbage collection runs // by uploading and syncing a number of chunks. -func TestDB_collectGarbage(t *testing.T) { +func TestDB_collectGarbageWorker(t *testing.T) { db, cleanupFunc := newTestDB(t, &Options{ Capacity: 100, }) defer cleanupFunc() - testDB_collectGarbage(t, db) + testDB_collectGarbageWorker(t, db) } -// TestDB_collectGarbage_useRetrievalCompositeIndex tests +// TestDB_collectGarbageWorker_useRetrievalCompositeIndex tests // garbage collection runs by uploading and syncing a number // of chunks using composite retrieval index. -func TestDB_collectGarbage_useRetrievalCompositeIndex(t *testing.T) { +func TestDB_collectGarbageWorker_useRetrievalCompositeIndex(t *testing.T) { db, cleanupFunc := newTestDB(t, &Options{ Capacity: 100, UseRetrievalCompositeIndex: true, }) defer cleanupFunc() - testDB_collectGarbage(t, db) + testDB_collectGarbageWorker(t, db) } -// TestDB_collectGarbage_multipleBatches tests garbage +// TestDB_collectGarbageWorker_multipleBatches tests garbage // collection runs by uploading and syncing a number of // chunks by having multiple smaller batches. -func TestDB_collectGarbage_multipleBatches(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) @@ -62,14 +65,14 @@ func TestDB_collectGarbage_multipleBatches(t *testing.T) { }) defer cleanupFunc() - testDB_collectGarbage(t, db) + testDB_collectGarbageWorker(t, db) } -// TestDB_collectGarbage_multipleBatches_useRetrievalCompositeIndex +// TestDB_collectGarbageWorker_multipleBatches_useRetrievalCompositeIndex // tests garbage collection runs by uploading and syncing a number // of chunks using composite retrieval index and having multiple // smaller batches. -func TestDB_collectGarbage_multipleBatches_useRetrievalCompositeIndex(t *testing.T) { +func TestDB_collectGarbageWorker_multipleBatches_useRetrievalCompositeIndex(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) @@ -81,12 +84,12 @@ func TestDB_collectGarbage_multipleBatches_useRetrievalCompositeIndex(t *testing }) defer cleanupFunc() - testDB_collectGarbage(t, db) + testDB_collectGarbageWorker(t, db) } -// testDB_collectGarbage is a helper test function to test +// testDB_collectGarbageWorker is a helper test function to test // garbage collection runs by uploading and syncing a number of chunks. -func testDB_collectGarbage(t *testing.T, db *DB) { +func testDB_collectGarbageWorker(t *testing.T, db *DB) { uploader := db.NewPutter(ModePutUpload) syncer := db.NewSetter(ModeSetSync) @@ -160,34 +163,34 @@ func testDB_collectGarbage(t *testing.T, db *DB) { }) } -// TestDB_collectGarbage_withRequests tests garbage collection +// TestDB_collectGarbageWorker_withRequests tests garbage collection // runs by uploading, syncing and requesting a number of chunks. -func TestDB_collectGarbage_withRequests(t *testing.T) { +func TestDB_collectGarbageWorker_withRequests(t *testing.T) { db, cleanupFunc := newTestDB(t, &Options{ Capacity: 100, }) defer cleanupFunc() - testDB_collectGarbage_withRequests(t, db) + testDB_collectGarbageWorker_withRequests(t, db) } -// TestDB_collectGarbage_withRequests_useRetrievalCompositeIndex +// TestDB_collectGarbageWorker_withRequests_useRetrievalCompositeIndex // tests garbage collection runs by uploading, syncing and // requesting a number of chunks using composite retrieval index. -func TestDB_collectGarbage_withRequests_useRetrievalCompositeIndex(t *testing.T) { +func TestDB_collectGarbageWorker_withRequests_useRetrievalCompositeIndex(t *testing.T) { db, cleanupFunc := newTestDB(t, &Options{ Capacity: 100, UseRetrievalCompositeIndex: true, }) defer cleanupFunc() - testDB_collectGarbage_withRequests(t, db) + testDB_collectGarbageWorker_withRequests(t, db) } -// testDB_collectGarbage_withRequests is a helper test function +// testDB_collectGarbageWorker_withRequests is a helper test function // to test garbage collection runs by uploading, syncing and // requesting a number of chunks. -func testDB_collectGarbage_withRequests(t *testing.T, db *DB) { +func testDB_collectGarbageWorker_withRequests(t *testing.T, db *DB) { uploader := db.NewPutter(ModePutUpload) syncer := db.NewSetter(ModeSetSync) @@ -290,6 +293,69 @@ func testDB_collectGarbage_withRequests(t *testing.T, db *DB) { }) } +// TestDB_gcSize checks if gcSize has a correct value after +// database is initialized with existing data. +func TestDB_gcSize(t *testing.T) { + dir, err := ioutil.TempDir("", "localstore-stored-gc-size") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(dir) + baseKey := make([]byte, 32) + if _, err := rand.Read(baseKey); err != nil { + t.Fatal(err) + } + db, err := New(dir, baseKey, nil) + if err != nil { + t.Fatal(err) + } + + uploader := db.NewPutter(ModePutUpload) + syncer := db.NewSetter(ModeSetSync) + + count := 100 + + for i := 0; i < count; i++ { + chunk := generateRandomChunk() + + err := uploader.Put(chunk) + if err != nil { + t.Fatal(err) + } + + err = syncer.Set(chunk.Address()) + if err != nil { + t.Fatal(err) + } + } + + err = db.Close() + if err != nil { + t.Fatal(err) + } + + db, err = New(dir, baseKey, nil) + if err != nil { + t.Fatal(err) + } + + t.Run("gc index size", newIndexGCSizeTest(db)) + + t.Run("gc uncounted hashes index count", newItemsCountTest(db.gcUncountedHashesIndex, 0)) +} + +func testStoredGCSize(t *testing.T, db *DB, want uint64) { + t.Helper() + + got, err := db.storedGCSize.Get() + if err != nil { + t.Fatal(err) + } + if got != want { + t.Errorf("got stored gc size %v, want %v", got, want) + } +} + // setTestHookCollectGarbage sets testHookCollectGarbage and // returns a function that will reset it to the // value before the change. diff --git a/swarm/storage/localstore/localstore.go b/swarm/storage/localstore/localstore.go index f3ef0387a2..e7269909a8 100644 --- a/swarm/storage/localstore/localstore.go +++ b/swarm/storage/localstore/localstore.go @@ -34,10 +34,10 @@ var ( ErrInvalidMode = errors.New("invalid mode") // ErrDBClosed is returned when database is closed. ErrDBClosed = errors.New("db closed") - // ErraddressLockTimeout is returned when the same chunk + // ErrAddressLockTimeout is returned when the same chunk // is updated in parallel and one of the updates // takes longer then the configured timeout duration. - ErraddressLockTimeout = errors.New("update lock timeout") + ErrAddressLockTimeout = errors.New("address lock timeout") ) var ( @@ -53,8 +53,10 @@ var ( type DB struct { shed *shed.DB - // fields + // schema name of loaded data schemaName shed.StringField + // filed that stores number of intems in gc index + storedGCSize shed.Uint64Field // this flag is for benchmarking two types of retrieval indexes // - single retrieval composite index retrievalCompositeIndex @@ -71,6 +73,9 @@ type DB struct { pullIndex shed.Index // 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 gcSize int64 @@ -78,12 +83,18 @@ type DB struct { // the capacity value capacity int64 + // 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 // created by Getters to call updateGC function updateGCSem chan struct{} + // a wait group to ensure all updateGC goroutines + // are done before closing the database + updateGCWG sync.WaitGroup baseKey []byte @@ -127,10 +138,12 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) { capacity: o.Capacity, baseKey: baseKey, useRetrievalCompositeIndex: o.UseRetrievalCompositeIndex, - // this channel needs to be buffered with the size of 1 - // to signal another garbage collection run if it - // is triggered during already running one + // channels collectGarbageTrigger and writeGCSizeTrigger + // need 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 { @@ -149,6 +162,11 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) { if err != nil { return nil, err } + // Persist gc size. + db.storedGCSize, err = db.shed.NewUint64Field("gc-size") + if err != nil { + return nil, err + } if db.useRetrievalCompositeIndex { var ( encodeValueFunc func(fields shed.Item) (value []byte, err error) @@ -345,20 +363,63 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) { if err != nil { return nil, err } - // start garbage collection worker - go db.collectGarbage() + // 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.gcIndex.Count() + 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 + err = db.gcUncountedHashesIndex.IterateAll(func(item shed.Item) (stop bool, err error) { + return false, db.gcUncountedHashesIndex.Delete(item) + }) + if err != nil { + return nil, err + } + // save the total gcSize after uncounted hashes are removed + err = db.storedGCSize.Put(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 } // Close closes the underlying database. func (db *DB) Close() (err error) { close(db.close) + db.updateGCWG.Wait() return db.shed.Close() } @@ -380,7 +441,7 @@ var ( // 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. +// it will return ErrAddressLockTimeout error. func (db *DB) lockAddr(addr storage.Address) (unlock func(), err error) { start := time.Now() lockKey := hex.EncodeToString(addr) @@ -391,7 +452,7 @@ func (db *DB) lockAddr(addr storage.Address) (unlock func(), err error) { } time.Sleep(addressLockCheckDelay) if time.Since(start) > addressLockTimeout { - return nil, ErraddressLockTimeout + return nil, ErrAddressLockTimeout } } return func() { db.addressLocks.Delete(lockKey) }, nil diff --git a/swarm/storage/localstore/mode_get.go b/swarm/storage/localstore/mode_get.go index d5d862114a..02a402590c 100644 --- a/swarm/storage/localstore/mode_get.go +++ b/swarm/storage/localstore/mode_get.go @@ -92,7 +92,9 @@ func (db *DB) get(mode ModeGet, addr storage.Address) (out shed.Item, err error) // if updateGCSem buffer id full db.updateGCSem <- struct{}{} } + db.updateGCWG.Add(1) go func() { + defer db.updateGCWG.Done() if db.updateGCSem != nil { // free a spot in updateGCSem buffer // for a new goroutine diff --git a/swarm/storage/localstore/mode_put.go b/swarm/storage/localstore/mode_put.go index 7b8d600f66..3eb22f3e97 100644 --- a/swarm/storage/localstore/mode_put.go +++ b/swarm/storage/localstore/mode_put.go @@ -127,6 +127,7 @@ func (db *DB) put(mode ModePut, item shed.Item) (err error) { } // add new entry to gc index db.gcIndex.PutInBatch(batch, item) + db.gcUncountedHashesIndex.PutInBatch(batch, item) db.incGCSize(1) if db.useRetrievalCompositeIndex { diff --git a/swarm/storage/localstore/mode_set.go b/swarm/storage/localstore/mode_set.go index 83c0e54656..ddaa789439 100644 --- a/swarm/storage/localstore/mode_set.go +++ b/swarm/storage/localstore/mode_set.go @@ -122,6 +122,7 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) { } db.pullIndex.PutInBatch(batch, item) db.gcIndex.PutInBatch(batch, item) + db.gcUncountedHashesIndex.PutInBatch(batch, item) db.incGCSize(1) case ModeSetSync: @@ -188,6 +189,7 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) { } db.pushIndex.DeleteInBatch(batch, item) db.gcIndex.PutInBatch(batch, item) + db.gcUncountedHashesIndex.PutInBatch(batch, item) db.incGCSize(1) case ModeSetRemove: @@ -226,6 +228,7 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) { } 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