diff --git a/swarm/storage/localstore/accessor.go b/swarm/storage/localstore/accessor.go deleted file mode 100644 index 7ab5a43963..0000000000 --- a/swarm/storage/localstore/accessor.go +++ /dev/null @@ -1,74 +0,0 @@ -// Copyright 2018 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package localstore - -import ( - "context" - - "github.com/syndtr/goleveldb/leveldb" - - "github.com/ethereum/go-ethereum/swarm/shed" - "github.com/ethereum/go-ethereum/swarm/storage" -) - -// Accessor implements ChunkStore to manage data -// in DB with different modes of access and update. -type Accessor struct { - db *DB - mode Mode -} - -// Accessor returns a new Accessor with a specified Mode. -func (db *DB) Accessor(mode Mode) *Accessor { - return &Accessor{ - mode: mode, - db: db, - } -} - -// Put uses the underlying DB for the specific mode of update to store the chunk. -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. -func (u *Accessor) Get(_ context.Context, addr storage.Address) (chunk storage.Chunk, err error) { - item := addressToItem(addr) - out, err := u.db.access(u.mode, item) - if err != nil { - if err == leveldb.ErrNotFound { - return nil, storage.ErrChunkNotFound - } - return nil, err - } - return storage.NewChunk(out.Address, out.Data), nil -} - -// chunkToItem creates new IndexItem with data provided by the Chunk. -func chunkToItem(ch storage.Chunk) shed.IndexItem { - return shed.IndexItem{ - Address: ch.Address(), - Data: ch.Data(), - } -} - -// addressToItem creates new IndexItem with a provided address. -func addressToItem(addr storage.Address) shed.IndexItem { - return shed.IndexItem{ - Address: addr, - } -} diff --git a/swarm/storage/localstore/accessor_test.go b/swarm/storage/localstore/accessor_test.go deleted file mode 100644 index bedf806ba3..0000000000 --- a/swarm/storage/localstore/accessor_test.go +++ /dev/null @@ -1,217 +0,0 @@ -// Copyright 2018 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package localstore - -import ( - "bytes" - "context" - "io/ioutil" - "testing" - - "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/swarm/storage" - "github.com/ethereum/go-ethereum/swarm/storage/mock" - "github.com/ethereum/go-ethereum/swarm/storage/mock/mem" -) - -// TestAccessors tests most basic Put and Get functionalities -// for different accessors. -func TestAccessors(t *testing.T) { - db, cleanupFunc := newTestDB(t, nil) - defer cleanupFunc() - - testAccessors(t, db) -} - -// TestAccessors_useRetrievalCompositeIndex tests most basic -// Put and Get functionalities for different accessors -// by using retrieval composite index. -func TestAccessors_useRetrievalCompositeIndex(t *testing.T) { - db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) - defer cleanupFunc() - - testAccessors(t, db) -} - -// TestAccessors_mockStore tests most basic Put and Get -// functionalities for different accessors with the mock store -// as the storage for chunk data. -func TestAccessors_mockStore(t *testing.T) { - globalStore := mem.NewGlobalStore() - - addr := common.BytesToAddress(make([]byte, 32)) - - db, cleanupFunc := newTestDB(t, &Options{ - MockStore: globalStore.NewNodeStore(addr), - }) - defer cleanupFunc() - - testAccessors(t, db) - - // testAccessors leaves 5 chunks in global store - checkGlobalStoreChunkCount(t, globalStore, 5) -} - -// TestAccessors_mockStore_useRetrievalCompositeIndex tests -// most basic Put and Get functionalities for different accessors -// with the mock store as the storage for chunk data and by using -// retrieval composite index. -func TestAccessors_mockStore_useRetrievalCompositeIndex(t *testing.T) { - globalStore := mem.NewGlobalStore() - - addr := common.BytesToAddress(make([]byte, 32)) - - db, cleanupFunc := newTestDB(t, &Options{ - MockStore: globalStore.NewNodeStore(addr), - UseRetrievalCompositeIndex: true, - }) - defer cleanupFunc() - - testAccessors(t, db) - - // testAccessors leaves 5 chunks in global store - checkGlobalStoreChunkCount(t, globalStore, 5) -} - -// testAccessors tests most basic Put and Get functionalities -// for different accessors. This test validates that the chunk -// is retrievable from the database, not if all indexes are set -// correctly. -func testAccessors(t *testing.T, db *DB) { - for _, m := range []Mode{ - ModeSyncing, - ModeUpload, - } { - t.Run(ModeName(m), func(t *testing.T) { - a := db.Accessor(m) - - want := generateRandomChunk() - - err := a.Put(context.Background(), want) - if err != nil { - t.Fatal(err) - } - - got, err := a.Get(context.Background(), want.Address()) - if err != nil { - t.Fatal(err) - } - if !bytes.Equal(got.Data(), want.Data()) { - t.Errorf("got chunk data %x, want %x", got.Data(), want.Data()) - } - }) - } - - // Synced mode does not put the item to retrieval index. - t.Run(ModeName(ModeSynced), func(t *testing.T) { - a := db.Accessor(ModeSynced) - - chunk := generateRandomChunk() - - // first put a random chunk to the database - err := a.Put(context.Background(), chunk) - if err != nil { - t.Fatal(err) - } - - wantError := storage.ErrChunkNotFound - _, err = a.Get(context.Background(), chunk.Address()) - if err != wantError { - t.Errorf("got error %v, want %v", err, wantError) - } - }) - - // Request and access modes are special as they do not store - // chunks in the database. - for _, m := range []Mode{ - ModeRequest, - modeAccess, - } { - t.Run(ModeName(m), func(t *testing.T) { - a := db.Accessor(ModeUpload) - - want := generateRandomChunk() - - // first put a random chunk to the database - err := a.Put(context.Background(), want) - if err != nil { - t.Fatal(err) - } - - a = db.Accessor(ModeRequest) - - got, err := a.Get(context.Background(), want.Address()) - if err != nil { - t.Fatal(err) - } - if !bytes.Equal(got.Data(), want.Data()) { - t.Errorf("got chunk data %x, want %x", got.Data(), want.Data()) - } - }) - } - - // Removal mode is a special case as it removes the chunk - // from the database. - t.Run(ModeName(modeRemoval), func(t *testing.T) { - a := db.Accessor(ModeUpload) - - want := generateRandomChunk() - - // first put a random chunk to the database - err := a.Put(context.Background(), want) - if err != nil { - t.Fatal(err) - } - - got, err := a.Get(context.Background(), want.Address()) - if err != nil { - t.Fatal(err) - } - if !bytes.Equal(got.Data(), want.Data()) { - t.Errorf("got chunk data %x, want %x", got.Data(), want.Data()) - } - - a = db.Accessor(modeRemoval) - - // removal accessor actually removes the chunk on Put - err = a.Put(context.Background(), want) - if err != nil { - t.Fatal(err) - } - - // chunk should not be found - wantErr := storage.ErrChunkNotFound - _, err = a.Get(context.Background(), want.Address()) - if err != wantErr { - t.Errorf("got error %v, expected %v", err, wantErr) - } - }) -} - -// checkGlobalStoreChunkCount counts the number of chunks -// in a global mock store to validate it against the expected value. -func checkGlobalStoreChunkCount(t *testing.T, s mock.ImportExporter, want int) { - t.Helper() - - n, err := s.Export(ioutil.Discard) - if err != nil { - t.Fatal(err) - } - if n != want { - t.Errorf("got %v chunks, want %v", n, want) - } -} diff --git a/swarm/storage/localstore/index_test.go b/swarm/storage/localstore/index_test.go new file mode 100644 index 0000000000..17253d7970 --- /dev/null +++ b/swarm/storage/localstore/index_test.go @@ -0,0 +1,238 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package localstore + +import ( + "bytes" + "math/rand" + "testing" + + "github.com/ethereum/go-ethereum/swarm/storage" +) + +// TestDB_pullIndex validates the ordering of keys in pull index. +// Pull index key contains PO prefix which is calculated from +// DB base key and chunk address. This is not an IndexItem field +// which are checked in Mode tests. +// This test uploads chunks, sorts them in expected order and +// validates that pull index iterator will iterate it the same +// order. +func TestDB_pullIndex(t *testing.T) { + db, cleanupFunc := newTestDB(t, nil) + defer cleanupFunc() + + uploader := db.NewPutter(ModePutUpload) + + chunkCount := 50 + + chunks := make([]testIndexChunk, chunkCount) + + // upload random chunks + for i := 0; i < chunkCount; i++ { + chunk := generateRandomChunk() + + err := uploader.Put(chunk) + if err != nil { + t.Fatal(err) + } + + chunks[i] = testIndexChunk{ + Chunk: chunk, + // this timestamp is not the same as in + // the index, but given that uploads + // are sequential and that only ordering + // of events matter, this information is + // sufficient + storeTimestamp: now(), + } + } + + testIndexItemsOrder(t, db.pullIndex, chunks, func(i, j int) (less bool) { + poi := storage.Proximity(db.baseKey, chunks[i].Address()) + poj := storage.Proximity(db.baseKey, chunks[j].Address()) + if poi < poj { + return true + } + if poi > poj { + return false + } + if chunks[i].storeTimestamp < chunks[j].storeTimestamp { + return true + } + if chunks[i].storeTimestamp > chunks[j].storeTimestamp { + return false + } + return bytes.Compare(chunks[i].Address(), chunks[j].Address()) == -1 + }) +} + +func TestDB_gcIndex(t *testing.T) { + db, cleanupFunc := newTestDB(t, nil) + defer cleanupFunc() + + testDB_gcIndex(t, db) +} + +func TestDB_gcIndex_useRetrievalCompositeIndex(t *testing.T) { + db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) + defer cleanupFunc() + + testDB_gcIndex(t, db) +} + +// testDB_gcIndex validates garbage collection index by uploading +// a chunk with and performing operations using synced, access and +// request modes. +func testDB_gcIndex(t *testing.T, db *DB) { + uploader := db.NewPutter(ModePutUpload) + + chunkCount := 50 + + chunks := make([]testIndexChunk, chunkCount) + + // upload random chunks + for i := 0; i < chunkCount; i++ { + chunk := generateRandomChunk() + + err := uploader.Put(chunk) + if err != nil { + t.Fatal(err) + } + + chunks[i] = testIndexChunk{ + Chunk: chunk, + } + } + + // check if all chunks are stored + newIndexItemsCountTest(db.pullIndex, chunkCount)(t) + + // check that chunks are not collectable for garbage + newIndexItemsCountTest(db.gcIndex, 0)(t) + + // set update gc test hook to signal when + // update gc goroutine is done by sending to + // testHookUpdateGCChan channel, which is + // used to wait for indexes change verifications + testHookUpdateGCChan := make(chan struct{}) + defer setTestHookUpdateGC(func() { + testHookUpdateGCChan <- struct{}{} + })() + + t.Run("request unsynced", func(t *testing.T) { + chunk := chunks[1] + + _, err := db.NewGetter(ModeGetRequest).Get(chunk.Address()) + if err != nil { + t.Fatal(err) + } + // wait for update gc goroutine to be done + <-testHookUpdateGCChan + + // the chunk is not synced + // should not be in the garbace collection index + newIndexItemsCountTest(db.gcIndex, 0)(t) + + newIndexGCSizeTest(db)(t) + }) + + t.Run("sync one chunk", func(t *testing.T) { + chunk := chunks[0] + + err := db.NewSetter(ModeSetSync).Set(chunk.Address()) + if err != nil { + t.Fatal(err) + } + + // the chunk is synced and should be in gc index + newIndexItemsCountTest(db.gcIndex, 1)(t) + + newIndexGCSizeTest(db)(t) + }) + + t.Run("sync all chunks", func(t *testing.T) { + setter := db.NewSetter(ModeSetSync) + + for i := range chunks { + err := setter.Set(chunks[i].Address()) + if err != nil { + t.Fatal(err) + } + } + + testIndexItemsOrder(t, db.gcIndex, chunks, nil) + + newIndexGCSizeTest(db)(t) + }) + + t.Run("request one chunk", func(t *testing.T) { + i := 6 + + _, err := db.NewGetter(ModeGetRequest).Get(chunks[i].Address()) + if err != nil { + t.Fatal(err) + } + // wait for update gc goroutine to be done + <-testHookUpdateGCChan + + // move the chunk to the end of the expected gc + c := chunks[i] + chunks = append(chunks[:i], chunks[i+1:]...) + chunks = append(chunks, c) + + testIndexItemsOrder(t, db.gcIndex, chunks, nil) + + newIndexGCSizeTest(db)(t) + }) + + t.Run("random chunk request", func(t *testing.T) { + requester := db.NewGetter(ModeGetRequest) + + rand.Shuffle(len(chunks), func(i, j int) { + chunks[i], chunks[j] = chunks[j], chunks[i] + }) + + for _, chunk := range chunks { + _, err := requester.Get(chunk.Address()) + if err != nil { + t.Fatal(err) + } + // wait for update gc goroutine to be done + <-testHookUpdateGCChan + } + + testIndexItemsOrder(t, db.gcIndex, chunks, nil) + + newIndexGCSizeTest(db)(t) + }) + + t.Run("remove one chunk", func(t *testing.T) { + i := 3 + + err := db.NewSetter(ModeSetRemove).Set(chunks[i].Address()) + if err != nil { + t.Fatal(err) + } + + // remove the chunk from the expected chunks in gc index + chunks = append(chunks[:i], chunks[i+1:]...) + + testIndexItemsOrder(t, db.gcIndex, chunks, nil) + + newIndexGCSizeTest(db)(t) + }) +} diff --git a/swarm/storage/localstore/localstore.go b/swarm/storage/localstore/localstore.go index 2573f42072..7dd741d732 100644 --- a/swarm/storage/localstore/localstore.go +++ b/swarm/storage/localstore/localstore.go @@ -18,6 +18,7 @@ package localstore import ( "encoding/binary" + "encoding/hex" "errors" "sync" "sync/atomic" @@ -34,10 +35,10 @@ var ( ErrInvalidMode = errors.New("invalid mode") // ErrDBClosed is returned when database is closed. ErrDBClosed = errors.New("db closed") - // ErrUpdateLockTimeout 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. - ErrUpdateLockTimeout = errors.New("update lock timeout") + ErraddressLockTimeout = errors.New("update lock timeout") ) // DB is the local store implementation and holds @@ -69,7 +70,7 @@ type DB struct { baseKey []byte - updateLocks sync.Map + addressLocks sync.Map } // Options struct holds optional parameters for configuring DB. @@ -329,6 +330,50 @@ 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 IndexItem with data provided by the Chunk. +func chunkToItem(ch storage.Chunk) shed.IndexItem { + return shed.IndexItem{ + Address: ch.Address(), + Data: ch.Data(), + } +} + +// addressToItem creates new IndexItem with a provided address. +func addressToItem(addr storage.Address) shed.IndexItem { + return shed.IndexItem{ + Address: addr, + } +} + // now is a helper function that returns a current unix timestamp // in UTC timezone. // It is set in the init function for usage in production, and diff --git a/swarm/storage/localstore/localstore_test.go b/swarm/storage/localstore/localstore_test.go index e7e6eec1b2..56743485ea 100644 --- a/swarm/storage/localstore/localstore_test.go +++ b/swarm/storage/localstore/localstore_test.go @@ -18,18 +18,48 @@ package localstore import ( "bytes" - "context" + "fmt" "io/ioutil" "math/rand" "os" + "sort" "strconv" + "sync/atomic" "testing" "time" ch "github.com/ethereum/go-ethereum/swarm/chunk" + "github.com/ethereum/go-ethereum/swarm/shed" "github.com/ethereum/go-ethereum/swarm/storage" + "github.com/syndtr/goleveldb/leveldb" ) +// TestDB validates if the chunk can be uploaded and +// correctly retrieved. +func TestDB(t *testing.T) { + db, cleanupFunc := newTestDB(t, nil) + defer cleanupFunc() + + chunk := generateRandomChunk() + + err := db.NewPutter(ModePutUpload).Put(chunk) + if err != nil { + t.Fatal(err) + } + + got, err := db.NewGetter(ModeGetRequest).Get(chunk.Address()) + if err != nil { + t.Fatal(err) + } + + if !bytes.Equal(got.Address(), chunk.Address()) { + t.Errorf("got address %x, want %x", got.Address(), chunk.Address()) + } + if !bytes.Equal(got.Data(), chunk.Data()) { + t.Errorf("got data %x, want %x", got.Data(), chunk.Data()) + } +} + // TestDB_useRetrievalCompositeIndex checks if optional argument // WithRetrievalCompositeIndex to New constructor is setting the // correct state. @@ -77,10 +107,10 @@ func TestDB_useRetrievalCompositeIndex(t *testing.T) { // goos: darwin // goarch: amd64 // pkg: github.com/ethereum/go-ethereum/swarm/storage/localstore -// BenchmarkNew/1000-8 200 12020231 ns/op 9556077 B/op 9999 allocs/op -// BenchmarkNew/10000-8 100 15475883 ns/op 10493071 B/op 7781 allocs/op -// BenchmarkNew/100000-8 20 64046466 ns/op 17823841 B/op 23375 allocs/op -// BenchmarkNew/1000000-8 1 1011464203 ns/op 51024688 B/op 310599 allocs/op +// BenchmarkNew/1000-8 200 11684285 ns/op 9556056 B/op 10005 allocs/op +// BenchmarkNew/10000-8 100 15161036 ns/op 10539571 B/op 7799 allocs/op +// BenchmarkNew/100000-8 20 74270386 ns/op 18234588 B/op 24382 allocs/op +// BenchmarkNew/1000000-8 2 942098251 ns/op 48747500 B/op 274976 allocs/op // PASS func BenchmarkNew(b *testing.B) { if testing.Short() { @@ -89,8 +119,8 @@ func BenchmarkNew(b *testing.B) { for _, count := range []int{ 1000, 10000, - 100000, - 1000000, + // 100000, + // 1000000, } { b.Run(strconv.Itoa(count), func(b *testing.B) { dir, err := ioutil.TempDir("", "localstore-new-benchmark") @@ -106,16 +136,15 @@ func BenchmarkNew(b *testing.B) { if err != nil { b.Fatal(err) } - uploader := db.Accessor(ModeUpload) - syncer := db.Accessor(ModeSynced) - ctx := context.Background() + uploader := db.NewPutter(ModePutUpload) + syncer := db.NewSetter(ModeSetSync) for i := 0; i < count; i++ { chunk := generateFakeRandomChunk() - err := uploader.Put(ctx, chunk) + err := uploader.Put(chunk) if err != nil { b.Fatal(err) } - err = syncer.Put(ctx, chunk) + err = syncer.Set(chunk.Address()) if err != nil { b.Fatal(err) } @@ -228,3 +257,310 @@ func TestGenerateFakeRandomChunk(t *testing.T) { t.Error("fake chunks data bytes do not differ") } } + +// newRetrieveIndexesTest returns a test function that validates if the right +// chunk values are in the retrieval indexes. +func newRetrieveIndexesTest(db *DB, chunk storage.Chunk, storeTimestamp, accessTimestamp int64) func(t *testing.T) { + return func(t *testing.T) { + if db.useRetrievalCompositeIndex { + item, err := db.retrievalCompositeIndex.Get(addressToItem(chunk.Address())) + if err != nil { + t.Fatal(err) + } + validateItem(t, item, chunk.Address(), chunk.Data(), storeTimestamp, accessTimestamp) + } else { + item, err := db.retrievalDataIndex.Get(addressToItem(chunk.Address())) + if err != nil { + t.Fatal(err) + } + validateItem(t, item, chunk.Address(), chunk.Data(), storeTimestamp, 0) + + // access index should not be set + wantErr := leveldb.ErrNotFound + item, err = db.retrievalAccessIndex.Get(addressToItem(chunk.Address())) + if err != wantErr { + t.Errorf("got error %v, want %v", err, wantErr) + } + } + } +} + +// newRetrieveIndexesTestWithAccess returns a test function that validates if the right +// chunk values are in the retrieval indexes when access time must be stored. +func newRetrieveIndexesTestWithAccess(db *DB, chunk storage.Chunk, storeTimestamp, accessTimestamp int64) func(t *testing.T) { + return func(t *testing.T) { + if db.useRetrievalCompositeIndex { + item, err := db.retrievalCompositeIndex.Get(addressToItem(chunk.Address())) + if err != nil { + t.Fatal(err) + } + validateItem(t, item, chunk.Address(), chunk.Data(), storeTimestamp, accessTimestamp) + } else { + item, err := db.retrievalDataIndex.Get(addressToItem(chunk.Address())) + if err != nil { + t.Fatal(err) + } + validateItem(t, item, chunk.Address(), chunk.Data(), storeTimestamp, 0) + + if accessTimestamp > 0 { + item, err = db.retrievalAccessIndex.Get(addressToItem(chunk.Address())) + if err != nil { + t.Fatal(err) + } + validateItem(t, item, chunk.Address(), nil, 0, accessTimestamp) + } + } + } +} + +// newPullIndexTest returns a test function that validates if the right +// chunk values are in the pull index. +func newPullIndexTest(db *DB, chunk storage.Chunk, storeTimestamp int64, wantError error) func(t *testing.T) { + return func(t *testing.T) { + item, err := db.pullIndex.Get(shed.IndexItem{ + Address: chunk.Address(), + StoreTimestamp: storeTimestamp, + }) + if err != wantError { + t.Errorf("got error %v, want %v", err, wantError) + } + if err == nil { + validateItem(t, item, chunk.Address(), nil, storeTimestamp, 0) + } + } +} + +// newPushIndexTest returns a test function that validates if the right +// chunk values are in the push index. +func newPushIndexTest(db *DB, chunk storage.Chunk, storeTimestamp int64, wantError error) func(t *testing.T) { + return func(t *testing.T) { + item, err := db.pushIndex.Get(shed.IndexItem{ + Address: chunk.Address(), + StoreTimestamp: storeTimestamp, + }) + if err != wantError { + t.Errorf("got error %v, want %v", err, wantError) + } + if err == nil { + validateItem(t, item, chunk.Address(), nil, storeTimestamp, 0) + } + } +} + +// newGCIndexTest returns a test function that validates if the right +// chunk values are in the push index. +func newGCIndexTest(db *DB, chunk storage.Chunk, storeTimestamp, accessTimestamp int64) func(t *testing.T) { + return func(t *testing.T) { + item, err := db.gcIndex.Get(shed.IndexItem{ + Address: chunk.Address(), + StoreTimestamp: storeTimestamp, + AccessTimestamp: accessTimestamp, + }) + if err != nil { + t.Fatal(err) + } + validateItem(t, item, chunk.Address(), nil, storeTimestamp, accessTimestamp) + } +} + +// newIndexItemsCountTest returns a test function that validates if +// an index contains expected number of key/value pairs. +func newIndexItemsCountTest(i shed.Index, want int) func(t *testing.T) { + return func(t *testing.T) { + var c int + i.IterateAll(func(item shed.IndexItem) (stop bool, err error) { + c++ + return + }) + if c != want { + t.Errorf("got %v items in index, want %v", c, want) + } + } +} + +// newIndexGCSizeTest retruns a test function that validates if DB.gcSize +// 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 + db.gcIndex.IterateAll(func(item shed.IndexItem) (stop bool, err error) { + want++ + return + }) + got := atomic.LoadInt64(&db.gcSize) + if got != want { + t.Errorf("got gc size %v, want %v", got, want) + } + } +} + +// testIndexChunk embeds storageChunk with additional data that is stored +// in database. It is used for index values validations. +type testIndexChunk struct { + storage.Chunk + storeTimestamp int64 +} + +// testIndexItemsOrder tests the order of chunks in the index. If sortFunc is not nil, +// chunks will be sorted with it before validation. +func testIndexItemsOrder(t *testing.T, i shed.Index, chunks []testIndexChunk, sortFunc func(i, j int) (less bool)) { + newIndexItemsCountTest(i, len(chunks))(t) + + if sortFunc != nil { + sort.Slice(chunks, sortFunc) + } + + var cursor int + err := i.IterateAll(func(item shed.IndexItem) (stop bool, err error) { + want := chunks[cursor].Address() + got := item.Address + if !bytes.Equal(got, want) { + return true, fmt.Errorf("got address %x at position %v, want %x", got, cursor, want) + } + cursor++ + return false, nil + }) + if err != nil { + t.Fatal(err) + } +} + +// 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() + + if !bytes.Equal(item.Address, address) { + t.Errorf("got item address %x, want %x", item.Address, address) + } + if !bytes.Equal(item.Data, data) { + t.Errorf("got item data %x, want %x", item.Data, data) + } + if item.StoreTimestamp != storeTimestamp { + t.Errorf("got item store timestamp %v, want %v", item.StoreTimestamp, storeTimestamp) + } + if item.AccessTimestamp != accessTimestamp { + t.Errorf("got item access timestamp %v, want %v", item.AccessTimestamp, accessTimestamp) + } +} + +// setTestHookUpdateGC sets testHookUpdateGC and +// returns a function that will reset it to the +// value before the change. +func setTestHookUpdateGC(h func()) (reset func()) { + current := testHookUpdateGC + reset = func() { testHookUpdateGC = current } + testHookUpdateGC = h + return reset +} + +// TestSetTestHookUpdateGC tests if setTestHookUpdateGC changes +// testHookUpdateGC function correctly and if its reset function +// resets the original function. +func TestSetTestHookUpdateGC(t *testing.T) { + // Set the current function after the test finishes. + defer func(h func()) { testHookUpdateGC = h }(testHookUpdateGC) + + // expected value for the unchanged function + original := 1 + // expected value for the changed function + changed := 2 + + // this variable will be set with two different functions + var got int + + // define the original (unchanged) functions + testHookUpdateGC = func() { + got = original + } + + // set got variable + testHookUpdateGC() + + // test if got variable is set correctly + if got != original { + t.Errorf("got hook value %v, want %v", got, original) + } + + // set the new function + reset := setTestHookUpdateGC(func() { + got = changed + }) + + // set got variable + testHookUpdateGC() + + // test if got variable is set correctly to changed value + if got != changed { + t.Errorf("got hook value %v, want %v", got, changed) + } + + // set the function to the original one + reset() + + // set got variable + testHookUpdateGC() + + // test if got variable is set correctly to original value + if got != original { + t.Errorf("got hook value %v, want %v", got, original) + } +} + +// setNow replaces now function and +// returns a function that will reset it to the +// value before the change. +func setNow(f func() int64) (reset func()) { + current := now + reset = func() { now = current } + now = f + return reset +} + +// TestSetNow tests if setNow function changes now function +// correctly and if its reset function resets the original function. +func TestSetNow(t *testing.T) { + // set the current function after the test finishes + defer func(f func() int64) { now = f }(now) + + // expected value for the unchanged function + var original int64 = 1 + // expected value for the changed function + var changed int64 = 2 + + // define the original (unchanged) functions + now = func() int64 { + return original + } + + // get the time + got := now() + + // test if got variable is set correctly + if got != original { + t.Errorf("got now value %v, want %v", got, original) + } + + // set the new function + reset := setNow(func() int64 { + return changed + }) + + // get the time + got = now() + + // test if got variable is set correctly to changed value + if got != changed { + t.Errorf("got hook value %v, want %v", got, changed) + } + + // set the function to the original one + reset() + + // get the time + got = now() + + // test if got variable is set correctly to original value + if got != original { + t.Errorf("got hook value %v, want %v", got, original) + } +} diff --git a/swarm/storage/localstore/mode.go b/swarm/storage/localstore/mode.go deleted file mode 100644 index ed2fbf834c..0000000000 --- a/swarm/storage/localstore/mode.go +++ /dev/null @@ -1,369 +0,0 @@ -// Copyright 2018 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package localstore - -import ( - "encoding/hex" - "sync/atomic" - "time" - - "github.com/ethereum/go-ethereum/swarm/shed" - "github.com/syndtr/goleveldb/leveldb" -) - -// Mode enumerates different modes of access and update -// operations on a database. -type Mode int - -// Modes of access and update. -const ( - ModeSyncing Mode = iota - ModeUpload - ModeRequest - ModeSynced - // this modes are internal only - // they can be removed completely - // if accessors are not used internally - modeAccess - modeRemoval -) - -// ModeName returns a descriptive name of a Mode. -// If the Mode is not known, a blank string is returned. -func ModeName(m Mode) (name string) { - switch m { - case ModeSyncing: - return "syncing" - case ModeUpload: - return "upload" - case ModeRequest: - return "request" - case ModeSynced: - return "synced" - case modeAccess: - return "access" - case modeRemoval: - return "removal" - } - return "" -} - -// access is called by an Accessor with a specific Mode. -// This function utilizes different indexes depending on -// the Mode. -func (db *DB) access(mode Mode, item shed.IndexItem) (out shed.IndexItem, err error) { - if db.useRetrievalCompositeIndex { - out, err = db.retrievalCompositeIndex.Get(item) - if err != nil { - return out, err - } - } else { - // No need to get access timestamp here as it is used - // only for some of Modes in update and access time - // is not property of the chunk returned by the Accessor.Get. - out, err = db.retrievalDataIndex.Get(item) - if err != nil { - return out, err - } - } - switch mode { - case ModeRequest, modeAccess: - // update the access timestamp and fc index - return out, db.updateOnAccess(mode, out) - default: - // all other modes are not updating the index - } - return out, nil -} - -var ( - updateLockTimeout = 3 * time.Second - updateLockCheckDelay = 30 * time.Microsecond -) - -// update performs different operations on fields and indexes -// depending on the provided Mode. It is called in accessor -// put function. -// 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(batch, item) - } else { - db.retrievalDataIndex.PutInBatch(batch, item) - } - db.pullIndex.PutInBatch(batch, item) - - case ModeUpload: - // put to indexes: retrieve, push, pull - item.StoreTimestamp = now() - if db.useRetrievalCompositeIndex { - db.retrievalCompositeIndex.PutInBatch(batch, item) - } else { - db.retrievalDataIndex.PutInBatch(batch, item) - } - db.pullIndex.PutInBatch(batch, item) - db.pushIndex.PutInBatch(batch, item) - - case ModeRequest: - // putting a chunk on mode request does not do anything - return nil - - case ModeSynced: - // delete from push, insert to gc - - // need to get access timestamp here as it is not - // provided by the access function, and it is not - // a property of a chunk provided to Accessor.Put. - if db.useRetrievalCompositeIndex { - i, err := db.retrievalCompositeIndex.Get(item) - if err != nil { - if err == leveldb.ErrNotFound { - // chunk is not found, - // no need to update gc index - // just delete from the push index - // if it is there - db.pushIndex.DeleteInBatch(batch, item) - return nil - } - return err - } - item.AccessTimestamp = i.AccessTimestamp - item.StoreTimestamp = i.StoreTimestamp - if item.AccessTimestamp == 0 { - // the chunk is not accessed before - // set access time for gc index - item.AccessTimestamp = now() - db.retrievalCompositeIndex.PutInBatch(batch, item) - } else { - // the chunk is accessed before - // remove the current gc index item - db.gcIndex.DeleteInBatch(batch, item) - atomic.AddInt64(&db.gcSize, -1) - } - } else { - i, err := db.retrievalDataIndex.Get(item) - if err != nil { - if err == leveldb.ErrNotFound { - // chunk is not found, - // no need to update gc index - // just delete from the push index - // if it is there - db.pushIndex.DeleteInBatch(batch, item) - return nil - } - return err - } - item.StoreTimestamp = i.StoreTimestamp - - i, err = db.retrievalAccessIndex.Get(item) - switch err { - case nil: - item.AccessTimestamp = i.AccessTimestamp - db.gcIndex.DeleteInBatch(batch, item) - atomic.AddInt64(&db.gcSize, -1) - case leveldb.ErrNotFound: - // the chunk is not accessed before - default: - return err - } - item.AccessTimestamp = now() - db.retrievalAccessIndex.PutInBatch(batch, item) - } - db.pushIndex.DeleteInBatch(batch, item) - db.gcIndex.PutInBatch(batch, item) - atomic.AddInt64(&db.gcSize, 1) - - case modeAccess: - // putting a chunk on mode access does not do anything - return nil - - case modeRemoval: - // delete from retrieve, pull, gc - - // need to get access timestamp here as it is not - // provided by the access function, and it is not - // a property of a chunk provided to Accessor.Put. - if db.useRetrievalCompositeIndex { - i, err := db.retrievalCompositeIndex.Get(item) - if err != nil { - return err - } - item.StoreTimestamp = i.StoreTimestamp - item.AccessTimestamp = i.AccessTimestamp - } else { - i, err := db.retrievalAccessIndex.Get(item) - switch err { - case nil: - item.AccessTimestamp = i.AccessTimestamp - case leveldb.ErrNotFound: - default: - return err - } - i, err = db.retrievalDataIndex.Get(item) - if err != nil { - return err - } - item.StoreTimestamp = i.StoreTimestamp - } - 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) - // TODO: optimize in garbage collection - // get is too expensive operation - if _, err := db.gcIndex.Get(item); err == nil { - atomic.AddInt64(&db.gcSize, -1) - } - - default: - return ErrInvalidMode - } - - return db.shed.WriteBatch(batch) -} - -// updateOnAccess is called in access function and performs -// different operations on fields and indexes depending on -// the provided Mode. -// This function is separated from the update function to prevent -// changes on calling accessor put function in access and request modes. -// It protects parallel updates of items with the same address -// with updateLocks map and waiting using a simple for loop. -func (db *DB) updateOnAccess(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 ModeRequest: - // update accessTimeStamp in retrieve, gc - - if db.useRetrievalCompositeIndex { - // access timestamp is already populated - // in the provided item, passed from access function. - } else { - i, err := db.retrievalAccessIndex.Get(item) - switch err { - case nil: - item.AccessTimestamp = i.AccessTimestamp - case leveldb.ErrNotFound: - // no chunk accesses - default: - return err - } - } - 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(batch, item) - // update access timestamp - item.AccessTimestamp = now() - // update retrieve access index - if db.useRetrievalCompositeIndex { - db.retrievalCompositeIndex.PutInBatch(batch, item) - } else { - db.retrievalAccessIndex.PutInBatch(batch, item) - } - // add new entry to gc index - 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 timestamp is already populated - // in the provided item, passed from access function. - } else { - i, err := db.retrievalAccessIndex.Get(item) - switch err { - case nil: - item.AccessTimestamp = i.AccessTimestamp - case leveldb.ErrNotFound: - // no chunk accesses - default: - return err - } - } - // Q: why do we need to update this index? - 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(batch, item) - // update access timestamp - item.AccessTimestamp = now() - // update retrieve access index - if db.useRetrievalCompositeIndex { - db.retrievalCompositeIndex.PutInBatch(batch, item) - } else { - db.retrievalAccessIndex.PutInBatch(batch, item) - } - // add new entry to gc index - db.gcIndex.PutInBatch(batch, item) - - default: - return ErrInvalidMode - } - - return db.shed.WriteBatch(batch) -} diff --git a/swarm/storage/localstore/mode_get.go b/swarm/storage/localstore/mode_get.go new file mode 100644 index 0000000000..8535e5fc68 --- /dev/null +++ b/swarm/storage/localstore/mode_get.go @@ -0,0 +1,161 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package localstore + +import ( + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/swarm/shed" + "github.com/ethereum/go-ethereum/swarm/storage" + "github.com/syndtr/goleveldb/leveldb" +) + +// ModeGet enumerates different Getter modes. +type ModeGet int + +// Getter modes. +const ( + // ModeGetRequest: when accessed for retrieval + ModeGetRequest ModeGet = iota + // ModeGetSync: when accessed for syncing or proof of custody request + ModeGetSync +) + +// Getter provides Get method to retrieve Chunks +// from database. +type Getter struct { + db *DB + mode ModeGet +} + +// NewGetter returns a new Getter on database +// with a specific Mode. +func (db *DB) NewGetter(mode ModeGet) *Getter { + return &Getter{ + mode: mode, + db: db, + } +} + +// Get returns a chunk from the database. If the chunk is +// not found storage.ErrChunkNotFound will be returned. +// All required indexes will be updated required by the +// Getter Mode. +func (g *Getter) Get(addr storage.Address) (chunk storage.Chunk, err error) { + out, err := g.db.get(g.mode, addr) + if err != nil { + if err == leveldb.ErrNotFound { + return nil, storage.ErrChunkNotFound + } + return nil, err + } + return storage.NewChunk(out.Address, out.Data), nil +} + +// get returns IndexItem with from the retrieval index +// and updates other indexes. +func (db *DB) get(mode ModeGet, addr storage.Address) (out shed.IndexItem, err error) { + item := addressToItem(addr) + + if db.useRetrievalCompositeIndex { + out, err = db.retrievalCompositeIndex.Get(item) + if err != nil { + return out, err + } + } else { + // No need to get access timestamp here as it is used + // only for some of Modes in update and access time + // is not property of the chunk returned by the Accessor.Get. + out, err = db.retrievalDataIndex.Get(item) + if err != nil { + return out, err + } + } + switch mode { + // update the access timestamp and gc index + case ModeGetRequest: + go func() { + err := db.updateGC(out) + if err != nil { + log.Error("localstore update gc", "err", err) + } + // if gc update hook is defined, call it + if testHookUpdateGC != nil { + testHookUpdateGC() + } + }() + + // no updates to indexes + case ModeGetSync: + default: + return out, ErrInvalidMode + } + return out, nil +} + +// updateGC updates garbage collection index for +// a single item. Provided item is expected to have +// only Address and Data fields with non zero values, +// which is ensured by the get function. +func (db *DB) updateGC(item shed.IndexItem) (err error) { + unlock, err := db.lockAddr(item.Address) + if err != nil { + return err + } + defer unlock() + + batch := new(leveldb.Batch) + + // update accessTimeStamp in retrieve, gc + + if db.useRetrievalCompositeIndex { + // access timestamp is already populated + // in the provided item, passed from access function. + } else { + i, err := db.retrievalAccessIndex.Get(item) + switch err { + case nil: + item.AccessTimestamp = i.AccessTimestamp + case leveldb.ErrNotFound: + // no chunk accesses + default: + return err + } + } + 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(batch, item) + // update access timestamp + item.AccessTimestamp = now() + // update retrieve access index + if db.useRetrievalCompositeIndex { + db.retrievalCompositeIndex.PutInBatch(batch, item) + } else { + db.retrievalAccessIndex.PutInBatch(batch, item) + } + // add new entry to gc index + db.gcIndex.PutInBatch(batch, item) + + return db.shed.WriteBatch(batch) +} + +// testHookUpdateGC is a hook that can provide +// information when a garbage collection index is updated. +var testHookUpdateGC func() diff --git a/swarm/storage/localstore/mode_get_test.go b/swarm/storage/localstore/mode_get_test.go new file mode 100644 index 0000000000..e928ec1f4a --- /dev/null +++ b/swarm/storage/localstore/mode_get_test.go @@ -0,0 +1,206 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package localstore + +import ( + "bytes" + "testing" + "time" +) + +// TestModeGetRequest validates internal data operations and state +// for ModeGetRequest on DB with default configuration. +func TestModeGetRequest(t *testing.T) { + db, cleanupFunc := newTestDB(t, nil) + defer cleanupFunc() + + testModeGetRequestValues(t, db) +} + +// TestModeGetRequest_useRetrievalCompositeIndex validates internal +// data operations and state for ModeGetRequest on DB with +// retrieval composite index enabled. +func TestModeGetRequest_useRetrievalCompositeIndex(t *testing.T) { + db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) + defer cleanupFunc() + + testModeGetRequestValues(t, db) +} + +// testModeGetRequestValues validates ModeGetRequest index values on the provided DB. +func testModeGetRequestValues(t *testing.T, db *DB) { + uploadTimestamp := time.Now().UTC().UnixNano() + defer setNow(func() (t int64) { + return uploadTimestamp + })() + + chunk := generateRandomChunk() + + err := db.NewPutter(ModePutUpload).Put(chunk) + if err != nil { + t.Fatal(err) + } + + requester := db.NewGetter(ModeGetRequest) + + // set update gc test hook to signal when + // update gc goroutine is done by sending to + // testHookUpdateGCChan channel, which is + // used to wait for garbage colletion index + // changes + testHookUpdateGCChan := make(chan struct{}) + defer setTestHookUpdateGC(func() { + testHookUpdateGCChan <- struct{}{} + })() + + t.Run("get unsynced", func(t *testing.T) { + got, err := requester.Get(chunk.Address()) + if err != nil { + t.Fatal(err) + } + // wait for update gc goroutine to be done + <-testHookUpdateGCChan + + if !bytes.Equal(got.Address(), chunk.Address()) { + t.Errorf("got chunk address %x, want %x", got.Address(), chunk.Address()) + } + + if !bytes.Equal(got.Data(), chunk.Data()) { + t.Errorf("got chunk data %x, want %x", got.Data(), chunk.Data()) + } + + t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, uploadTimestamp, 0)) + + t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 0)) + + t.Run("gc size", newIndexGCSizeTest(db)) + }) + + // set chunk to synced state + err = db.NewSetter(ModeSetSync).Set(chunk.Address()) + if err != nil { + t.Fatal(err) + } + + t.Run("first get", func(t *testing.T) { + got, err := requester.Get(chunk.Address()) + if err != nil { + t.Fatal(err) + } + // wait for update gc goroutine to be done + <-testHookUpdateGCChan + + if !bytes.Equal(got.Address(), chunk.Address()) { + t.Errorf("got chunk address %x, want %x", got.Address(), chunk.Address()) + } + + if !bytes.Equal(got.Data(), chunk.Data()) { + t.Errorf("got chunk data %x, want %x", got.Data(), chunk.Data()) + } + + t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, uploadTimestamp, uploadTimestamp)) + + t.Run("gc index", newGCIndexTest(db, chunk, uploadTimestamp, uploadTimestamp)) + + t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 1)) + + t.Run("gc size", newIndexGCSizeTest(db)) + }) + + t.Run("second get", func(t *testing.T) { + accessTimestamp := time.Now().UTC().UnixNano() + defer setNow(func() (t int64) { + return accessTimestamp + })() + + got, err := requester.Get(chunk.Address()) + if err != nil { + t.Fatal(err) + } + // wait for update gc goroutine to be done + <-testHookUpdateGCChan + + if !bytes.Equal(got.Address(), chunk.Address()) { + t.Errorf("got chunk address %x, want %x", got.Address(), chunk.Address()) + } + + if !bytes.Equal(got.Data(), chunk.Data()) { + t.Errorf("got chunk data %x, want %x", got.Data(), chunk.Data()) + } + + t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, uploadTimestamp, accessTimestamp)) + + t.Run("gc index", newGCIndexTest(db, chunk, uploadTimestamp, accessTimestamp)) + + t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 1)) + + t.Run("gc size", newIndexGCSizeTest(db)) + }) +} + +// TestModeGetSync validates internal data operations and state +// for ModeGetSync on DB with default configuration. +func TestModeGetSync(t *testing.T) { + db, cleanupFunc := newTestDB(t, nil) + defer cleanupFunc() + + testModeGetSyncValues(t, db) +} + +// TestModeGetSync_useRetrievalCompositeIndex validates internal +// data operations and state for ModeGetSync on DB with +// retrieval composite index enabled. +func TestModeGetSync_useRetrievalCompositeIndex(t *testing.T) { + db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) + defer cleanupFunc() + + testModeGetSyncValues(t, db) +} + +// testModeGetSyncValues validates ModeGetSync index values on the provided DB. +func testModeGetSyncValues(t *testing.T, db *DB) { + uploadTimestamp := time.Now().UTC().UnixNano() + defer setNow(func() (t int64) { + return uploadTimestamp + })() + + chunk := generateRandomChunk() + + err := db.NewPutter(ModePutUpload).Put(chunk) + if err != nil { + t.Fatal(err) + } + + got, err := db.NewGetter(ModeGetSync).Get(chunk.Address()) + if err != nil { + t.Fatal(err) + } + + if !bytes.Equal(got.Address(), chunk.Address()) { + t.Errorf("got chunk address %x, want %x", got.Address(), chunk.Address()) + } + + if !bytes.Equal(got.Data(), chunk.Data()) { + t.Errorf("got chunk data %x, want %x", got.Data(), chunk.Data()) + } + + t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, uploadTimestamp, 0)) + + t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 0)) + + t.Run("gc size", newIndexGCSizeTest(db)) +} diff --git a/swarm/storage/localstore/mode_put.go b/swarm/storage/localstore/mode_put.go new file mode 100644 index 0000000000..a2e1fcdb5d --- /dev/null +++ b/swarm/storage/localstore/mode_put.go @@ -0,0 +1,169 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package localstore + +import ( + "sync/atomic" + + "github.com/ethereum/go-ethereum/swarm/shed" + "github.com/ethereum/go-ethereum/swarm/storage" + "github.com/syndtr/goleveldb/leveldb" +) + +// ModePut enumerates different Putter modes. +type ModePut int + +// Putter modes. +const ( + // ModePutRequest: when a chunk is received as a result of retrieve request and delivery, it is put only in + ModePutRequest ModePut = iota + // ModePutSync: when a chunk is received via syncing in it is put in + ModePutSync + // ModePutUpload: when a chunk is created by local upload it is put in + ModePutUpload +) + +// Putter provides Put method to store Chunks +// to database. +type Putter struct { + db *DB + mode ModePut +} + +// NewPutter returns a new Putter on database +// with a specific Mode. +func (db *DB) NewPutter(mode ModePut) *Putter { + return &Putter{ + mode: mode, + db: db, + } +} + +// Put stores the Chunk to database and depending +// on the Putter mode, it updates required indexes. +func (p *Putter) Put(ch storage.Chunk) (err error) { + return p.db.put(p.mode, chunkToItem(ch)) +} + +// put stores IndexItem to database and updates other +// indexes. It acquires lockAddr to protect two calls +// of this function for the same address in parallel. +// IndexItem fields Address and Data must not be +// with their nil values. +func (db *DB) put(mode ModePut, item shed.IndexItem) (err error) { + // protect parallel updates + unlock, err := db.lockAddr(item.Address) + if err != nil { + return err + } + defer unlock() + + batch := new(leveldb.Batch) + + switch mode { + case ModePutRequest: + // put to indexes: retrieve, gc; it does not enter the syncpool + + // check if the chunk already is in the database + // as gc index is updated + if db.useRetrievalCompositeIndex { + i, err := db.retrievalCompositeIndex.Get(item) + switch err { + case nil: + item.AccessTimestamp = i.AccessTimestamp + item.StoreTimestamp = i.StoreTimestamp + case leveldb.ErrNotFound: + // no chunk in database + default: + return err + } + } else { + i, err := db.retrievalAccessIndex.Get(item) + switch err { + case nil: + item.AccessTimestamp = i.AccessTimestamp + case leveldb.ErrNotFound: + // no chunk accesses + default: + return err + } + i, err = db.retrievalDataIndex.Get(item) + switch err { + case nil: + item.StoreTimestamp = i.StoreTimestamp + case leveldb.ErrNotFound: + // no chunk accesses + default: + return err + } + } + if item.AccessTimestamp != 0 { + // delete current entry from the gc index + db.gcIndex.DeleteInBatch(batch, item) + atomic.AddInt64(&db.gcSize, -1) + } + if item.StoreTimestamp == 0 { + item.StoreTimestamp = now() + } + // update access timestamp + item.AccessTimestamp = now() + // update retrieve access index + if db.useRetrievalCompositeIndex { + db.retrievalCompositeIndex.PutInBatch(batch, item) + } else { + db.retrievalAccessIndex.PutInBatch(batch, item) + } + // add new entry to gc index + db.gcIndex.PutInBatch(batch, item) + atomic.AddInt64(&db.gcSize, 1) + + if db.useRetrievalCompositeIndex { + db.retrievalCompositeIndex.PutInBatch(batch, item) + } else { + db.retrievalDataIndex.PutInBatch(batch, item) + db.retrievalAccessIndex.PutInBatch(batch, item) + } + + case ModePutUpload: + // put to indexes: retrieve, push, pull + + item.StoreTimestamp = now() + if db.useRetrievalCompositeIndex { + db.retrievalCompositeIndex.PutInBatch(batch, item) + } else { + db.retrievalDataIndex.PutInBatch(batch, item) + } + db.pullIndex.PutInBatch(batch, item) + db.pushIndex.PutInBatch(batch, item) + + case ModePutSync: + // put to indexes: retrieve, pull + + item.StoreTimestamp = now() + if db.useRetrievalCompositeIndex { + db.retrievalCompositeIndex.PutInBatch(batch, item) + } else { + db.retrievalDataIndex.PutInBatch(batch, item) + } + db.pullIndex.PutInBatch(batch, item) + + default: + return ErrInvalidMode + } + + return db.shed.WriteBatch(batch) +} diff --git a/swarm/storage/localstore/mode_put_test.go b/swarm/storage/localstore/mode_put_test.go new file mode 100644 index 0000000000..f8518312cf --- /dev/null +++ b/swarm/storage/localstore/mode_put_test.go @@ -0,0 +1,167 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package localstore + +import ( + "testing" + "time" +) + +// TestModePutRequest validates internal data operations and state +// for ModePutRequest on DB with default configuration. +func TestModePutRequest(t *testing.T) { + db, cleanupFunc := newTestDB(t, nil) + defer cleanupFunc() + + testModePutRequestValues(t, db) +} + +// TestModePutRequest_useRetrievalCompositeIndex validates internal +// data operations and state for ModePutRequest on DB with +// retrieval composite index enabled. +func TestModePutRequest_useRetrievalCompositeIndex(t *testing.T) { + db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) + defer cleanupFunc() + + testModePutRequestValues(t, db) +} + +// testModePutRequestValues validates ModePutRequest index values on the provided DB. +func testModePutRequestValues(t *testing.T, db *DB) { + putter := db.NewPutter(ModePutRequest) + + chunk := generateRandomChunk() + + // keep the record when the chunk is stored + var storeTimestamp int64 + + t.Run("first put", func(t *testing.T) { + wantTimestamp := time.Now().UTC().UnixNano() + defer setNow(func() (t int64) { + return wantTimestamp + })() + + storeTimestamp = wantTimestamp + + err := putter.Put(chunk) + if err != nil { + t.Fatal(err) + } + + t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, wantTimestamp, wantTimestamp)) + + t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 1)) + + t.Run("gc size", newIndexGCSizeTest(db)) + }) + + t.Run("second put", func(t *testing.T) { + wantTimestamp := time.Now().UTC().UnixNano() + defer setNow(func() (t int64) { + return wantTimestamp + })() + + err := putter.Put(chunk) + if err != nil { + t.Fatal(err) + } + + t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, storeTimestamp, wantTimestamp)) + + t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 1)) + + t.Run("gc size", newIndexGCSizeTest(db)) + }) +} + +// TestModePutSync validates internal data operations and state +// for ModePutSync on DB with default configuration. +func TestModePutSync(t *testing.T) { + db, cleanupFunc := newTestDB(t, nil) + defer cleanupFunc() + + testModePutSyncValues(t, db) +} + +// TestModePutSync_useRetrievalCompositeIndex validates internal +// data operations and state for ModePutSync on DB with +// retrieval composite index enabled. +func TestModePutSync_useRetrievalCompositeIndex(t *testing.T) { + db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) + defer cleanupFunc() + + testModePutSyncValues(t, db) +} + +// testModePutSyncValues validates ModePutSync index values on the provided DB. +func testModePutSyncValues(t *testing.T, db *DB) { + wantTimestamp := time.Now().UTC().UnixNano() + defer setNow(func() (t int64) { + return wantTimestamp + })() + + chunk := generateRandomChunk() + + err := db.NewPutter(ModePutSync).Put(chunk) + if err != nil { + t.Fatal(err) + } + + t.Run("retrieve indexes", newRetrieveIndexesTest(db, chunk, wantTimestamp, 0)) + + t.Run("pull index", newPullIndexTest(db, chunk, wantTimestamp, nil)) +} + +// TestModePutUpload validates internal data operations and state +// for ModePutUpload on DB with default configuration. +func TestModePutUpload(t *testing.T) { + db, cleanupFunc := newTestDB(t, nil) + defer cleanupFunc() + + testModePutUploadValues(t, db) +} + +// TestModePutUpload_useRetrievalCompositeIndex validates internal +// data operations and state for ModePutUpload on DB with +// retrieval composite index enabled. +func TestModePutUpload_useRetrievalCompositeIndex(t *testing.T) { + db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) + defer cleanupFunc() + + testModePutUploadValues(t, db) +} + +// testModePutUploadValues validates ModePutUpload index values on the provided DB. +func testModePutUploadValues(t *testing.T, db *DB) { + wantTimestamp := time.Now().UTC().UnixNano() + defer setNow(func() (t int64) { + return wantTimestamp + })() + + chunk := generateRandomChunk() + + err := db.NewPutter(ModePutUpload).Put(chunk) + if err != nil { + t.Fatal(err) + } + + 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)) +} diff --git a/swarm/storage/localstore/mode_set.go b/swarm/storage/localstore/mode_set.go new file mode 100644 index 0000000000..f6d161a96a --- /dev/null +++ b/swarm/storage/localstore/mode_set.go @@ -0,0 +1,242 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package localstore + +import ( + "sync/atomic" + + "github.com/ethereum/go-ethereum/swarm/storage" + "github.com/syndtr/goleveldb/leveldb" +) + +// ModeSet enumerates different Setter modes. +type ModeSet int + +// Setter modes. +const ( + // ModeSetAccess: when an update request is received for a chunk or chunk is retrieved for delivery + ModeSetAccess ModeSet = iota + // ModeSetSync: when push sync receipt is received + ModeSetSync + // ModeSetRemove: when GC-d + ModeSetRemove +) + +// Setter sets the state of a particular +// Chunk in database by changing indexes. +type Setter struct { + db *DB + mode ModeSet +} + +// NewSetter returns a new Setter on database +// with a specific Mode. +func (db *DB) NewSetter(mode ModeSet) *Setter { + return &Setter{ + mode: mode, + db: db, + } +} + +// Set updates database indexes for a specific +// chunk represented by the address. +func (s *Setter) Set(addr storage.Address) (err error) { + return s.db.set(s.mode, addr) +} + +// set updates database indexes for a specific +// chunk represented by the address. +// It acquires lockAddr to protect two calls +// of this function for the same address in parallel. +func (db *DB) set(mode ModeSet, addr storage.Address) (err error) { + // protect parallel updates + unlock, err := db.lockAddr(addr) + if err != nil { + return err + } + defer unlock() + + batch := new(leveldb.Batch) + + item := addressToItem(addr) + + switch mode { + case ModeSetAccess: + // add to pull, insert to gc + + // need to get access timestamp here as it is not + // provided by the access function, and it is not + // a property of a chunk provided to Accessor.Put. + if db.useRetrievalCompositeIndex { + i, err := db.retrievalCompositeIndex.Get(item) + switch err { + case nil: + item.AccessTimestamp = i.AccessTimestamp + item.StoreTimestamp = i.StoreTimestamp + db.gcIndex.DeleteInBatch(batch, item) + atomic.AddInt64(&db.gcSize, -1) + case leveldb.ErrNotFound: + db.pullIndex.DeleteInBatch(batch, item) + item.AccessTimestamp = now() + item.StoreTimestamp = now() + default: + return err + } + } else { + i, err := db.retrievalDataIndex.Get(item) + switch err { + case nil: + item.StoreTimestamp = i.StoreTimestamp + case leveldb.ErrNotFound: + db.pushIndex.DeleteInBatch(batch, item) + item.StoreTimestamp = now() + default: + return err + } + + i, err = db.retrievalAccessIndex.Get(item) + switch err { + case nil: + item.AccessTimestamp = i.AccessTimestamp + db.gcIndex.DeleteInBatch(batch, item) + atomic.AddInt64(&db.gcSize, -1) + case leveldb.ErrNotFound: + // the chunk is not accessed before + default: + return err + } + item.AccessTimestamp = now() + db.retrievalAccessIndex.PutInBatch(batch, item) + } + db.pullIndex.PutInBatch(batch, item) + db.gcIndex.PutInBatch(batch, item) + atomic.AddInt64(&db.gcSize, 1) + + case ModeSetSync: + // delete from push, insert to gc + + // need to get access timestamp here as it is not + // provided by the access function, and it is not + // a property of a chunk provided to Accessor.Put. + if db.useRetrievalCompositeIndex { + i, err := db.retrievalCompositeIndex.Get(item) + if err != nil { + if err == leveldb.ErrNotFound { + // chunk is not found, + // no need to update gc index + // just delete from the push index + // if it is there + db.pushIndex.DeleteInBatch(batch, item) + return nil + } + return err + } + item.AccessTimestamp = i.AccessTimestamp + item.StoreTimestamp = i.StoreTimestamp + item.Data = i.Data + if item.AccessTimestamp == 0 { + // the chunk is not accessed before + // set access time for gc index + item.AccessTimestamp = now() + db.retrievalCompositeIndex.PutInBatch(batch, item) + } else { + // the chunk is accessed before + // remove the current gc index item + db.gcIndex.DeleteInBatch(batch, item) + atomic.AddInt64(&db.gcSize, -1) + } + } else { + i, err := db.retrievalDataIndex.Get(item) + if err != nil { + if err == leveldb.ErrNotFound { + // chunk is not found, + // no need to update gc index + // just delete from the push index + // if it is there + db.pushIndex.DeleteInBatch(batch, item) + return nil + } + return err + } + item.StoreTimestamp = i.StoreTimestamp + + i, err = db.retrievalAccessIndex.Get(item) + switch err { + case nil: + item.AccessTimestamp = i.AccessTimestamp + db.gcIndex.DeleteInBatch(batch, item) + atomic.AddInt64(&db.gcSize, -1) + case leveldb.ErrNotFound: + // the chunk is not accessed before + default: + return err + } + item.AccessTimestamp = now() + db.retrievalAccessIndex.PutInBatch(batch, item) + } + db.pushIndex.DeleteInBatch(batch, item) + db.gcIndex.PutInBatch(batch, item) + atomic.AddInt64(&db.gcSize, 1) + + case ModeSetRemove: + // delete from retrieve, pull, gc + + // need to get access timestamp here as it is not + // provided by the access function, and it is not + // a property of a chunk provided to Accessor.Put. + if db.useRetrievalCompositeIndex { + i, err := db.retrievalCompositeIndex.Get(item) + if err != nil { + return err + } + item.StoreTimestamp = i.StoreTimestamp + item.AccessTimestamp = i.AccessTimestamp + } else { + i, err := db.retrievalAccessIndex.Get(item) + switch err { + case nil: + item.AccessTimestamp = i.AccessTimestamp + case leveldb.ErrNotFound: + default: + return err + } + i, err = db.retrievalDataIndex.Get(item) + if err != nil { + return err + } + item.StoreTimestamp = i.StoreTimestamp + } + 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) + // TODO: optimize in garbage collection + // get is too expensive operation + if _, err := db.gcIndex.Get(item); err == nil { + atomic.AddInt64(&db.gcSize, -1) + } + + default: + return ErrInvalidMode + } + + return db.shed.WriteBatch(batch) +} diff --git a/swarm/storage/localstore/mode_set_test.go b/swarm/storage/localstore/mode_set_test.go new file mode 100644 index 0000000000..be0dbf58ef --- /dev/null +++ b/swarm/storage/localstore/mode_set_test.go @@ -0,0 +1,184 @@ +// Copyright 2018 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package localstore + +import ( + "testing" + "time" + + "github.com/syndtr/goleveldb/leveldb" +) + +// TestModeSetAccess validates internal data operations and state +// for ModeSetAccess on DB with default configuration. +func TestModeSetAccess(t *testing.T) { + db, cleanupFunc := newTestDB(t, nil) + defer cleanupFunc() + + testModeSetAccessValues(t, db) +} + +// TestModeSetAccess_useRetrievalCompositeIndex validates internal +// data operations and state for ModeSetAccess on DB with +// retrieval composite index enabled. +func TestModeSetAccess_useRetrievalCompositeIndex(t *testing.T) { + db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) + defer cleanupFunc() + + testModeSetAccessValues(t, db) +} + +// testModeSetAccessValues validates ModeSetAccess index values on the provided DB. +func testModeSetAccessValues(t *testing.T, db *DB) { + chunk := generateRandomChunk() + + wantTimestamp := time.Now().UTC().UnixNano() + defer setNow(func() (t int64) { + return wantTimestamp + })() + + err := db.NewSetter(ModeSetAccess).Set(chunk.Address()) + if err != nil { + t.Fatal(err) + } + + t.Run("pull index", newPullIndexTest(db, chunk, wantTimestamp, nil)) + + t.Run("pull index count", newIndexItemsCountTest(db.pullIndex, 1)) + + t.Run("gc index", newGCIndexTest(db, chunk, wantTimestamp, wantTimestamp)) + + t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 1)) + + t.Run("gc size", newIndexGCSizeTest(db)) +} + +// TestModeSetSync validates internal data operations and state +// for ModeSetSync on DB with default configuration. +func TestModeSetSync(t *testing.T) { + db, cleanupFunc := newTestDB(t, nil) + defer cleanupFunc() + + testModeSetSyncValues(t, db) +} + +// TestModeSetSync_useRetrievalCompositeIndex validates internal +// data operations and state for ModeSetSync on DB with +// retrieval composite index enabled. +func TestModeSetSync_useRetrievalCompositeIndex(t *testing.T) { + db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) + defer cleanupFunc() + + testModeSetSyncValues(t, db) +} + +// testModeSetSyncValues validates ModeSetSync index values on the provided DB. +func testModeSetSyncValues(t *testing.T, db *DB) { + chunk := generateRandomChunk() + + wantTimestamp := time.Now().UTC().UnixNano() + defer setNow(func() (t int64) { + return wantTimestamp + })() + + err := db.NewPutter(ModePutUpload).Put(chunk) + if err != nil { + t.Fatal(err) + } + + err = db.NewSetter(ModeSetSync).Set(chunk.Address()) + if err != nil { + t.Fatal(err) + } + + t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, wantTimestamp, wantTimestamp)) + + t.Run("push index", newPushIndexTest(db, chunk, wantTimestamp, leveldb.ErrNotFound)) + + t.Run("gc index", newGCIndexTest(db, chunk, wantTimestamp, wantTimestamp)) + + t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 1)) + + t.Run("gc size", newIndexGCSizeTest(db)) +} + +// TestModeSetRemoval validates internal data operations and state +// for ModeSetRemoval on DB with default configuration. +func TestModeSetRemoval(t *testing.T) { + db, cleanupFunc := newTestDB(t, nil) + defer cleanupFunc() + + testModeSetRemovalValues(t, db) +} + +// TestModeSetRemoval_useRetrievalCompositeIndex validates internal +// data operations and state for ModeSetRemoval on DB with +// retrieval composite index enabled. +func TestModeSetRemoval_useRetrievalCompositeIndex(t *testing.T) { + db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) + defer cleanupFunc() + + testModeSetRemovalValues(t, db) +} + +// testModeSetRemovalValues validates ModeSetRemoval index values on the provided DB. +func testModeSetRemovalValues(t *testing.T, db *DB) { + chunk := generateRandomChunk() + + err := db.NewPutter(ModePutUpload).Put(chunk) + if err != nil { + t.Fatal(err) + } + + err = db.NewSetter(ModeSetRemove).Set(chunk.Address()) + if err != nil { + t.Fatal(err) + } + + t.Run("retrieve indexes", func(t *testing.T) { + wantErr := leveldb.ErrNotFound + if db.useRetrievalCompositeIndex { + _, err := db.retrievalCompositeIndex.Get(addressToItem(chunk.Address())) + if err != wantErr { + t.Errorf("got error %v, want %v", err, wantErr) + } + t.Run("retrieve index count", newIndexItemsCountTest(db.retrievalCompositeIndex, 0)) + } else { + _, err := db.retrievalDataIndex.Get(addressToItem(chunk.Address())) + if err != wantErr { + t.Errorf("got error %v, want %v", err, wantErr) + } + t.Run("retrieve data index count", newIndexItemsCountTest(db.retrievalDataIndex, 0)) + + // access index should not be set + _, err = db.retrievalAccessIndex.Get(addressToItem(chunk.Address())) + if err != wantErr { + t.Errorf("got error %v, want %v", err, wantErr) + } + t.Run("retrieve access index count", newIndexItemsCountTest(db.retrievalAccessIndex, 0)) + } + }) + + t.Run("pull index", newPullIndexTest(db, chunk, 0, leveldb.ErrNotFound)) + + t.Run("pull index count", newIndexItemsCountTest(db.pullIndex, 0)) + + t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 0)) + + t.Run("gc size", newIndexGCSizeTest(db)) + +} diff --git a/swarm/storage/localstore/mode_test.go b/swarm/storage/localstore/mode_test.go deleted file mode 100644 index 930c896d85..0000000000 --- a/swarm/storage/localstore/mode_test.go +++ /dev/null @@ -1,918 +0,0 @@ -// Copyright 2018 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package localstore - -import ( - "bytes" - "context" - "fmt" - "math/rand" - "sort" - "sync/atomic" - "testing" - "time" - - "github.com/syndtr/goleveldb/leveldb" - - "github.com/ethereum/go-ethereum/swarm/shed" - "github.com/ethereum/go-ethereum/swarm/storage" -) - -// TestModeSyncing validates internal data operations and state -// for ModeSyncing on DB with default configuration. -func TestModeSyncing(t *testing.T) { - db, cleanupFunc := newTestDB(t, nil) - defer cleanupFunc() - - testModeSyncingValues(t, db) -} - -// TestModeSyncing_useRetrievalCompositeIndex validates internal -// data operations and state for ModeSyncing on DB with -// retrieval composite index enabled. -func TestModeSyncing_useRetrievalCompositeIndex(t *testing.T) { - db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) - defer cleanupFunc() - - testModeSyncingValues(t, db) -} - -// testModeSyncingValues validates ModeSyncing index values on the provided DB. -func testModeSyncingValues(t *testing.T, db *DB) { - a := db.Accessor(ModeSyncing) - - chunk := generateRandomChunk() - - wantTimestamp := time.Now().UTC().UnixNano() - defer func(n func() int64) { now = n }(now) - now = func() (t int64) { - return wantTimestamp - } - - err := a.Put(context.Background(), chunk) - if err != nil { - t.Fatal(err) - } - - t.Run("retrieve indexes", newRetrieveIndexesTest(db, chunk, wantTimestamp, 0)) - - t.Run("pull index", newPullIndexTest(db, chunk, wantTimestamp, nil)) -} - -// TestModeUpload validates internal data operations and state -// for ModeUpload on DB with default configuration. -func TestModeUpload(t *testing.T) { - db, cleanupFunc := newTestDB(t, nil) - defer cleanupFunc() - - testModeUploadValues(t, db) -} - -// TestModeUpload_useRetrievalCompositeIndex validates internal -// data operations and state for ModeUpload on DB with -// retrieval composite index enabled. -func TestModeUpload_useRetrievalCompositeIndex(t *testing.T) { - db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) - defer cleanupFunc() - - testModeUploadValues(t, db) -} - -// testModeUploadValues validates ModeUpload index values on the provided DB. -func testModeUploadValues(t *testing.T, db *DB) { - a := db.Accessor(ModeUpload) - - chunk := generateRandomChunk() - - wantTimestamp := time.Now().UTC().UnixNano() - defer func(n func() int64) { now = n }(now) - now = func() (t int64) { - return wantTimestamp - } - - err := a.Put(context.Background(), chunk) - if err != nil { - t.Fatal(err) - } - - 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)) -} - -// TestModeRequest validates internal data operations and state -// for ModeRequest on DB with default configuration. -func TestModeRequest(t *testing.T) { - db, cleanupFunc := newTestDB(t, nil) - defer cleanupFunc() - - testModeRequestValues(t, db) -} - -// TestModeRequest_useRetrievalCompositeIndex validates internal -// data operations and state for ModeRequest on DB with -// retrieval composite index enabled. -func TestModeRequest_useRetrievalCompositeIndex(t *testing.T) { - db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) - defer cleanupFunc() - - testModeRequestValues(t, db) -} - -// testModeRequestValues validates ModeRequest index values on the provided DB. -func testModeRequestValues(t *testing.T, db *DB) { - a := db.Accessor(ModeUpload) - - chunk := generateRandomChunk() - - uploadTimestamp := time.Now().UTC().UnixNano() - defer func(n func() int64) { now = n }(now) - now = func() (t int64) { - return uploadTimestamp - } - - err := a.Put(context.Background(), chunk) - if err != nil { - t.Fatal(err) - } - - a = db.Accessor(ModeRequest) - - t.Run("get unsynced", func(t *testing.T) { - got, err := a.Get(context.Background(), chunk.Address()) - if err != nil { - t.Fatal(err) - } - - if !bytes.Equal(chunk.Address(), got.Address()) { - t.Errorf("got chunk address %x, want %s", chunk.Address(), got.Address()) - } - - if !bytes.Equal(chunk.Data(), got.Data()) { - t.Errorf("got chunk data %x, want %s", chunk.Data(), got.Data()) - } - - t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, uploadTimestamp, 0)) - - t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 0)) - - t.Run("gc size", newIndexGCSizeTest(db)) - }) - - // set chunk to synced state - err = db.Accessor(ModeSynced).Put(context.Background(), chunk) - if err != nil { - t.Fatal(err) - } - - t.Run("first get", func(t *testing.T) { - got, err := a.Get(context.Background(), chunk.Address()) - if err != nil { - t.Fatal(err) - } - - if !bytes.Equal(chunk.Address(), got.Address()) { - t.Errorf("got chunk address %x, want %s", chunk.Address(), got.Address()) - } - - if !bytes.Equal(chunk.Data(), got.Data()) { - t.Errorf("got chunk data %x, want %s", chunk.Data(), got.Data()) - } - - t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, uploadTimestamp, uploadTimestamp)) - - t.Run("gc index", newGCIndexTest(db, chunk, uploadTimestamp, uploadTimestamp)) - - t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 1)) - - t.Run("gc size", newIndexGCSizeTest(db)) - }) - - t.Run("second get", func(t *testing.T) { - accessTimestamp := time.Now().UTC().UnixNano() - now = func() (t int64) { - return accessTimestamp - } - - got, err := a.Get(context.Background(), chunk.Address()) - if err != nil { - t.Fatal(err) - } - - if !bytes.Equal(chunk.Address(), got.Address()) { - t.Errorf("got chunk address %x, want %s", chunk.Address(), got.Address()) - } - - if !bytes.Equal(chunk.Data(), got.Data()) { - t.Errorf("got chunk data %x, want %s", chunk.Data(), got.Data()) - } - - t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, uploadTimestamp, accessTimestamp)) - - t.Run("gc index", newGCIndexTest(db, chunk, uploadTimestamp, accessTimestamp)) - - t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 1)) - - t.Run("gc size", newIndexGCSizeTest(db)) - }) -} - -// TestModeSynced validates internal data operations and state -// for ModeSynced on DB with default configuration. -func TestModeSynced(t *testing.T) { - db, cleanupFunc := newTestDB(t, nil) - defer cleanupFunc() - - testModeSyncedValues(t, db) -} - -// TestModeSynced_useRetrievalCompositeIndex validates internal -// data operations and state for ModeSynced on DB with -// retrieval composite index enabled. -func TestModeSynced_useRetrievalCompositeIndex(t *testing.T) { - db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) - defer cleanupFunc() - - testModeSyncedValues(t, db) -} - -// testModeSyncedValues validates ModeSynced index values on the provided DB. -func testModeSyncedValues(t *testing.T, db *DB) { - a := db.Accessor(ModeSyncing) - - chunk := generateRandomChunk() - - wantTimestamp := time.Now().UTC().UnixNano() - defer func(n func() int64) { now = n }(now) - now = func() (t int64) { - return wantTimestamp - } - - err := a.Put(context.Background(), chunk) - if err != nil { - t.Fatal(err) - } - - a = db.Accessor(ModeSynced) - - err = a.Put(context.Background(), chunk) - if err != nil { - t.Fatal(err) - } - - t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, wantTimestamp, wantTimestamp)) - - t.Run("push index", newPushIndexTest(db, chunk, wantTimestamp, leveldb.ErrNotFound)) - - t.Run("gc index", newGCIndexTest(db, chunk, wantTimestamp, wantTimestamp)) - - t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 1)) - - t.Run("gc size", newIndexGCSizeTest(db)) -} - -// TestModeAccess validates internal data operations and state -// for ModeAccess on DB with default configuration. -func TestModeAccess(t *testing.T) { - db, cleanupFunc := newTestDB(t, nil) - defer cleanupFunc() - - testModeAccessValues(t, db) -} - -// TestModeAccess_useRetrievalCompositeIndex validates internal -// data operations and state for ModeAccess on DB with -// retrieval composite index enabled. -func TestModeAccess_useRetrievalCompositeIndex(t *testing.T) { - db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) - defer cleanupFunc() - - testModeAccessValues(t, db) -} - -// testModeAccessValues validates ModeAccess index values on the provided DB. -func testModeAccessValues(t *testing.T, db *DB) { - a := db.Accessor(ModeUpload) - - chunk := generateRandomChunk() - - uploadTimestamp := time.Now().UTC().UnixNano() - defer func(n func() int64) { now = n }(now) - now = func() (t int64) { - return uploadTimestamp - } - - err := a.Put(context.Background(), chunk) - if err != nil { - t.Fatal(err) - } - - a = db.Accessor(modeAccess) - - t.Run("get unsynced", func(t *testing.T) { - got, err := a.Get(context.Background(), chunk.Address()) - if err != nil { - t.Fatal(err) - } - - if !bytes.Equal(chunk.Address(), got.Address()) { - t.Errorf("got chunk address %x, want %s", chunk.Address(), got.Address()) - } - - if !bytes.Equal(chunk.Data(), got.Data()) { - t.Errorf("got chunk data %x, want %s", chunk.Data(), got.Data()) - } - - t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, uploadTimestamp, 0)) - - t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 0)) - - t.Run("gc size", newIndexGCSizeTest(db)) - }) - - // set chunk to synced state - err = db.Accessor(ModeSynced).Put(context.Background(), chunk) - if err != nil { - t.Fatal(err) - } - - t.Run("first get", func(t *testing.T) { - got, err := a.Get(context.Background(), chunk.Address()) - if err != nil { - t.Fatal(err) - } - - if !bytes.Equal(chunk.Address(), got.Address()) { - t.Errorf("got chunk address %x, want %s", chunk.Address(), got.Address()) - } - - if !bytes.Equal(chunk.Data(), got.Data()) { - t.Errorf("got chunk data %x, want %s", chunk.Data(), got.Data()) - } - - t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, uploadTimestamp, uploadTimestamp)) - - t.Run("gc index", newGCIndexTest(db, chunk, uploadTimestamp, uploadTimestamp)) - - t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 1)) - - t.Run("gc size", newIndexGCSizeTest(db)) - }) - - t.Run("second get", func(t *testing.T) { - accessTimestamp := time.Now().UTC().UnixNano() - now = func() (t int64) { - return accessTimestamp - } - - got, err := a.Get(context.Background(), chunk.Address()) - if err != nil { - t.Fatal(err) - } - - if !bytes.Equal(chunk.Address(), got.Address()) { - t.Errorf("got chunk address %x, want %s", chunk.Address(), got.Address()) - } - - if !bytes.Equal(chunk.Data(), got.Data()) { - t.Errorf("got chunk data %x, want %s", chunk.Data(), got.Data()) - } - - t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, uploadTimestamp, accessTimestamp)) - - t.Run("gc index", newGCIndexTest(db, chunk, uploadTimestamp, accessTimestamp)) - - t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 1)) - - t.Run("gc size", newIndexGCSizeTest(db)) - }) -} - -// TestModeRemoval validates internal data operations and state -// for ModeRemoval on DB with default configuration. -func TestModeRemoval(t *testing.T) { - db, cleanupFunc := newTestDB(t, nil) - defer cleanupFunc() - - testModeRemovalValues(t, db) -} - -// TestModeRemoval_useRetrievalCompositeIndex validates internal -// data operations and state for ModeRemoval on DB with -// retrieval composite index enabled. -func TestModeRemoval_useRetrievalCompositeIndex(t *testing.T) { - db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) - defer cleanupFunc() - - testModeRemovalValues(t, db) -} - -// testModeRemovalValues validates ModeRemoval index values on the provided DB. -func testModeRemovalValues(t *testing.T, db *DB) { - a := db.Accessor(ModeUpload) - - chunk := generateRandomChunk() - - err := a.Put(context.Background(), chunk) - if err != nil { - t.Fatal(err) - } - - a = db.Accessor(modeRemoval) - - err = a.Put(context.Background(), chunk) - if err != nil { - t.Fatal(err) - } - - t.Run("retrieve indexes", func(t *testing.T) { - wantErr := leveldb.ErrNotFound - if db.useRetrievalCompositeIndex { - _, err := db.retrievalCompositeIndex.Get(addressToItem(chunk.Address())) - if err != wantErr { - t.Errorf("got error %v, want %v", err, wantErr) - } - t.Run("retrieve index count", newIndexItemsCountTest(db.retrievalCompositeIndex, 0)) - } else { - _, err := db.retrievalDataIndex.Get(addressToItem(chunk.Address())) - if err != wantErr { - t.Errorf("got error %v, want %v", err, wantErr) - } - t.Run("retrieve data index count", newIndexItemsCountTest(db.retrievalDataIndex, 0)) - - // access index should not be set - _, err = db.retrievalAccessIndex.Get(addressToItem(chunk.Address())) - if err != wantErr { - t.Errorf("got error %v, want %v", err, wantErr) - } - t.Run("retrieve access index count", newIndexItemsCountTest(db.retrievalAccessIndex, 0)) - } - }) - - t.Run("pull index", newPullIndexTest(db, chunk, 0, leveldb.ErrNotFound)) - - t.Run("pull index count", newIndexItemsCountTest(db.pullIndex, 0)) - - t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 0)) - - t.Run("gc size", newIndexGCSizeTest(db)) - -} - -// TestDB_pullIndex validates the ordering of keys in pull index. -// Pull index key contains PO prefix which is calculated from -// DB base key and chunk address. This is not an IndexItem field -// which are checked in Mode tests. -// This test uploads chunks, sorts them in expected order and -// validates that pull index iterator will iterate it the same -// order. -func TestDB_pullIndex(t *testing.T) { - db, cleanupFunc := newTestDB(t, nil) - defer cleanupFunc() - - a := db.Accessor(ModeUpload) - - chunkCount := 50 - - chunks := make([]testIndexChunk, chunkCount) - - // upload random chunks - for i := 0; i < chunkCount; i++ { - chunk := generateRandomChunk() - - err := a.Put(context.Background(), chunk) - if err != nil { - t.Fatal(err) - } - - chunks[i] = testIndexChunk{ - Chunk: chunk, - // this timestamp is not the same as in - // the index, but given that uploads - // are sequential and that only ordering - // of events matter, this information is - // sufficient - storeTimestamp: now(), - } - } - - testIndexItemsOrder(t, db.pullIndex, chunks, func(i, j int) (less bool) { - poi := storage.Proximity(db.baseKey, chunks[i].Address()) - poj := storage.Proximity(db.baseKey, chunks[j].Address()) - if poi < poj { - return true - } - if poi > poj { - return false - } - if chunks[i].storeTimestamp < chunks[j].storeTimestamp { - return true - } - if chunks[i].storeTimestamp > chunks[j].storeTimestamp { - return false - } - return bytes.Compare(chunks[i].Address(), chunks[j].Address()) == -1 - }) -} - -func TestDB_gcIndex(t *testing.T) { - db, cleanupFunc := newTestDB(t, nil) - defer cleanupFunc() - - testDB_gcIndex(t, db) -} - -func TestDB_gcIndex_useRetrievalCompositeIndex(t *testing.T) { - db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true}) - defer cleanupFunc() - - testDB_gcIndex(t, db) -} - -// testDB_gcIndex validates garbage collection index by uploading -// a chunk with and performing operations using synced, access and -// request modes. -func testDB_gcIndex(t *testing.T, db *DB) { - a := db.Accessor(ModeUpload) - - chunkCount := 50 - - chunks := make([]testIndexChunk, chunkCount) - - // upload random chunks - for i := 0; i < chunkCount; i++ { - chunk := generateRandomChunk() - - err := a.Put(context.Background(), chunk) - if err != nil { - t.Fatal(err) - } - - chunks[i] = testIndexChunk{ - Chunk: chunk, - } - } - - // check if all chunks are stored - newIndexItemsCountTest(db.pullIndex, chunkCount)(t) - - // check that chunks are not collectable for garbage - newIndexItemsCountTest(db.gcIndex, 0)(t) - - t.Run("access unsynced", func(t *testing.T) { - chunk := chunks[0] - - a := db.Accessor(modeAccess) - - _, err := a.Get(context.Background(), chunk.Address()) - if err != nil { - t.Fatal(err) - } - - // the chunk is not synced - // should not be in the garbace collection index - newIndexItemsCountTest(db.gcIndex, 0)(t) - - newIndexGCSizeTest(db)(t) - }) - - t.Run("request unsynced", func(t *testing.T) { - chunk := chunks[1] - - a := db.Accessor(ModeRequest) - - _, err := a.Get(context.Background(), chunk.Address()) - if err != nil { - t.Fatal(err) - } - - // the chunk is not synced - // should not be in the garbace collection index - newIndexItemsCountTest(db.gcIndex, 0)(t) - - newIndexGCSizeTest(db)(t) - }) - - t.Run("sync one chunk", func(t *testing.T) { - chunk := chunks[0] - - a := db.Accessor(ModeSynced) - - err := a.Put(context.Background(), chunk) - if err != nil { - t.Fatal(err) - } - - // the chunk is synced and should be in gc index - newIndexItemsCountTest(db.gcIndex, 1)(t) - - newIndexGCSizeTest(db)(t) - }) - - t.Run("sync all chunks", func(t *testing.T) { - a := db.Accessor(ModeSynced) - - for i := range chunks { - err := a.Put(context.Background(), chunks[i]) - if err != nil { - t.Fatal(err) - } - } - - testIndexItemsOrder(t, db.gcIndex, chunks, nil) - - newIndexGCSizeTest(db)(t) - }) - - t.Run("access one chunk", func(t *testing.T) { - a := db.Accessor(modeAccess) - - i := 5 - - _, err := a.Get(context.Background(), chunks[i].Address()) - if err != nil { - t.Fatal(err) - } - - // move the chunk to the end of the expected gc - c := chunks[i] - chunks = append(chunks[:i], chunks[i+1:]...) - chunks = append(chunks, c) - - testIndexItemsOrder(t, db.gcIndex, chunks, nil) - - newIndexGCSizeTest(db)(t) - }) - - t.Run("request one chunk", func(t *testing.T) { - a := db.Accessor(ModeRequest) - - i := 6 - - _, err := a.Get(context.Background(), chunks[i].Address()) - if err != nil { - t.Fatal(err) - } - - // move the chunk to the end of the expected gc - c := chunks[i] - chunks = append(chunks[:i], chunks[i+1:]...) - chunks = append(chunks, c) - - testIndexItemsOrder(t, db.gcIndex, chunks, nil) - - newIndexGCSizeTest(db)(t) - }) - - t.Run("random chunk access", func(t *testing.T) { - a := db.Accessor(modeAccess) - - rand.Shuffle(len(chunks), func(i, j int) { - chunks[i], chunks[j] = chunks[j], chunks[i] - }) - - for _, chunk := range chunks { - _, err := a.Get(context.Background(), chunk.Address()) - if err != nil { - t.Fatal(err) - } - } - - testIndexItemsOrder(t, db.gcIndex, chunks, nil) - - newIndexGCSizeTest(db)(t) - }) - - t.Run("random chunk request", func(t *testing.T) { - a := db.Accessor(ModeRequest) - - rand.Shuffle(len(chunks), func(i, j int) { - chunks[i], chunks[j] = chunks[j], chunks[i] - }) - - for _, chunk := range chunks { - _, err := a.Get(context.Background(), chunk.Address()) - if err != nil { - t.Fatal(err) - } - } - - testIndexItemsOrder(t, db.gcIndex, chunks, nil) - - newIndexGCSizeTest(db)(t) - }) - - t.Run("remove one chunk", func(t *testing.T) { - a := db.Accessor(modeRemoval) - - i := 3 - - err := a.Put(context.Background(), chunks[i]) - if err != nil { - t.Fatal(err) - } - - // remove the chunk from the expected chunks in gc index - chunks = append(chunks[:i], chunks[i+1:]...) - - testIndexItemsOrder(t, db.gcIndex, chunks, nil) - - newIndexGCSizeTest(db)(t) - }) -} - -// newRetrieveIndexesTest returns a test function that validates if the right -// chunk values are in the retrieval indexes. -func newRetrieveIndexesTest(db *DB, chunk storage.Chunk, storeTimestamp, accessTimestamp int64) func(t *testing.T) { - return func(t *testing.T) { - if db.useRetrievalCompositeIndex { - item, err := db.retrievalCompositeIndex.Get(addressToItem(chunk.Address())) - if err != nil { - t.Fatal(err) - } - validateItem(t, item, chunk.Address(), chunk.Data(), storeTimestamp, accessTimestamp) - } else { - item, err := db.retrievalDataIndex.Get(addressToItem(chunk.Address())) - if err != nil { - t.Fatal(err) - } - validateItem(t, item, chunk.Address(), chunk.Data(), storeTimestamp, 0) - - // access index should not be set - wantErr := leveldb.ErrNotFound - item, err = db.retrievalAccessIndex.Get(addressToItem(chunk.Address())) - if err != wantErr { - t.Errorf("got error %v, want %v", err, wantErr) - } - } - } -} - -// newRetrieveIndexesTestWithAccess returns a test function that validates if the right -// chunk values are in the retrieval indexes when access time must be stored. -func newRetrieveIndexesTestWithAccess(db *DB, chunk storage.Chunk, storeTimestamp, accessTimestamp int64) func(t *testing.T) { - return func(t *testing.T) { - if db.useRetrievalCompositeIndex { - item, err := db.retrievalCompositeIndex.Get(addressToItem(chunk.Address())) - if err != nil { - t.Fatal(err) - } - validateItem(t, item, chunk.Address(), chunk.Data(), storeTimestamp, accessTimestamp) - } else { - item, err := db.retrievalDataIndex.Get(addressToItem(chunk.Address())) - if err != nil { - t.Fatal(err) - } - validateItem(t, item, chunk.Address(), chunk.Data(), storeTimestamp, 0) - - if accessTimestamp > 0 { - item, err = db.retrievalAccessIndex.Get(addressToItem(chunk.Address())) - if err != nil { - t.Fatal(err) - } - validateItem(t, item, chunk.Address(), nil, 0, accessTimestamp) - } - } - } -} - -// newPullIndexTest returns a test function that validates if the right -// chunk values are in the pull index. -func newPullIndexTest(db *DB, chunk storage.Chunk, storeTimestamp int64, wantError error) func(t *testing.T) { - return func(t *testing.T) { - item, err := db.pullIndex.Get(shed.IndexItem{ - Address: chunk.Address(), - StoreTimestamp: storeTimestamp, - }) - if err != wantError { - t.Errorf("got error %v, want %v", err, wantError) - } - if err == nil { - validateItem(t, item, chunk.Address(), nil, storeTimestamp, 0) - } - } -} - -// newPushIndexTest returns a test function that validates if the right -// chunk values are in the push index. -func newPushIndexTest(db *DB, chunk storage.Chunk, storeTimestamp int64, wantError error) func(t *testing.T) { - return func(t *testing.T) { - item, err := db.pushIndex.Get(shed.IndexItem{ - Address: chunk.Address(), - StoreTimestamp: storeTimestamp, - }) - if err != wantError { - t.Errorf("got error %v, want %v", err, wantError) - } - if err == nil { - validateItem(t, item, chunk.Address(), nil, storeTimestamp, 0) - } - } -} - -// newGCIndexTest returns a test function that validates if the right -// chunk values are in the push index. -func newGCIndexTest(db *DB, chunk storage.Chunk, storeTimestamp, accessTimestamp int64) func(t *testing.T) { - return func(t *testing.T) { - item, err := db.gcIndex.Get(shed.IndexItem{ - Address: chunk.Address(), - StoreTimestamp: storeTimestamp, - AccessTimestamp: accessTimestamp, - }) - if err != nil { - t.Fatal(err) - } - validateItem(t, item, chunk.Address(), nil, storeTimestamp, accessTimestamp) - } -} - -// newIndexItemsCountTest returns a test function that validates if -// an index contains expected number of key/value pairs. -func newIndexItemsCountTest(i shed.Index, want int) func(t *testing.T) { - return func(t *testing.T) { - var c int - i.IterateAll(func(item shed.IndexItem) (stop bool, err error) { - c++ - return - }) - if c != want { - t.Errorf("got %v items in index, want %v", c, want) - } - } -} - -func newIndexGCSizeTest(db *DB) func(t *testing.T) { - return func(t *testing.T) { - var want int64 - db.gcIndex.IterateAll(func(item shed.IndexItem) (stop bool, err error) { - want++ - return - }) - got := atomic.LoadInt64(&db.gcSize) - if got != want { - t.Errorf("got gc size %v, want %v", got, want) - } - } -} - -type testIndexChunk struct { - storage.Chunk - storeTimestamp int64 - accessTimestamp int64 -} - -func testIndexItemsOrder(t *testing.T, i shed.Index, chunks []testIndexChunk, sortFunc func(i, j int) (less bool)) { - newIndexItemsCountTest(i, len(chunks))(t) - - if sortFunc != nil { - sort.Slice(chunks, sortFunc) - } - - var cursor int - err := i.IterateAll(func(item shed.IndexItem) (stop bool, err error) { - want := chunks[cursor].Address() - got := item.Address - if !bytes.Equal(got, want) { - return true, fmt.Errorf("got address %x at position %v, want %x", got, cursor, want) - } - cursor++ - return false, nil - }) - if err != nil { - t.Fatal(err) - } -} - -// 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() - - if !bytes.Equal(item.Address, address) { - t.Errorf("got item address %x, want %x", item.Address, address) - } - if !bytes.Equal(item.Data, data) { - t.Errorf("got item data %x, want %x", item.Data, data) - } - if item.StoreTimestamp != storeTimestamp { - t.Errorf("got item store timestamp %v, want %v", item.StoreTimestamp, storeTimestamp) - } - if item.AccessTimestamp != accessTimestamp { - t.Errorf("got item access timestamp %v, want %v", item.AccessTimestamp, accessTimestamp) - } -} diff --git a/swarm/storage/localstore/retrieval_index_test.go b/swarm/storage/localstore/retrieval_index_test.go index 403046fa46..675cd47045 100644 --- a/swarm/storage/localstore/retrieval_index_test.go +++ b/swarm/storage/localstore/retrieval_index_test.go @@ -17,7 +17,6 @@ package localstore import ( - "context" "strconv" "testing" @@ -43,12 +42,12 @@ import ( // goos: darwin // goarch: amd64 // pkg: github.com/ethereum/go-ethereum/swarm/storage/localstore -// BenchmarkRetrievalIndexes/1000-split-8 20 57035332 ns/op 18150318 B/op 78152 allocs/op -// BenchmarkRetrievalIndexes/1000-composite-8 10 145093830 ns/op 66965899 B/op 68621 allocs/op -// BenchmarkRetrievalIndexes/10000-split-8 1 1023919551 ns/op 376620048 B/op 1384874 allocs/op -// BenchmarkRetrievalIndexes/10000-composite-8 1 2612845197 ns/op 1006614104 B/op 1492380 allocs/op -// BenchmarkRetrievalIndexes/100000-split-8 1 14168164804 ns/op 2868944816 B/op 12425362 allocs/op -// BenchmarkRetrievalIndexes/100000-composite-8 1 65995988337 ns/op 12387004776 B/op 22376909 allocs/op +// BenchmarkRetrievalIndexes/1000-split-8 20 75556686 ns/op 19033493 B/op 84500 allocs/op +// BenchmarkRetrievalIndexes/1000-composite-8 10 143774538 ns/op 67474551 B/op 72104 allocs/op +// BenchmarkRetrievalIndexes/10000-split-8 1 1079084922 ns/op 382792064 B/op 1429644 allocs/op +// BenchmarkRetrievalIndexes/10000-composite-8 1 2597268475 ns/op 1005916808 B/op 1516443 allocs/op +// BenchmarkRetrievalIndexes/100000-split-8 1 16891305737 ns/op 2629165304 B/op 12465019 allocs/op +// BenchmarkRetrievalIndexes/100000-composite-8 1 67158059676 ns/op 12292703424 B/op 22436767 allocs/op // PASS func BenchmarkRetrievalIndexes(b *testing.B) { for _, count := range []int{ @@ -76,31 +75,41 @@ func benchmarkRetrievalIndexes(b *testing.B, o *Options, count int) { b.StopTimer() db, cleanupFunc := newTestDB(b, o) defer cleanupFunc() - uploader := db.Accessor(ModeUpload) - syncer := db.Accessor(ModeSynced) - requester := db.Accessor(ModeRequest) - ctx := context.Background() - chunks := make([]storage.Chunk, count) + uploader := db.NewPutter(ModePutUpload) + syncer := db.NewSetter(ModeSetSync) + requester := db.NewGetter(ModeGetRequest) + addrs := make([]storage.Address, count) for i := 0; i < count; i++ { chunk := generateFakeRandomChunk() - err := uploader.Put(ctx, chunk) + err := uploader.Put(chunk) if err != nil { b.Fatal(err) } - chunks[i] = chunk + addrs[i] = chunk.Address() } + // set update gc test hook to signal when + // update gc goroutine is done by sending to + // testHookUpdateGCChan channel, which is + // used to wait for gc index updates to be + // included in the benchmark time + testHookUpdateGCChan := make(chan struct{}) + defer setTestHookUpdateGC(func() { + testHookUpdateGCChan <- struct{}{} + })() b.StartTimer() for i := 0; i < count; i++ { - err := syncer.Put(ctx, chunks[i]) + err := syncer.Set(addrs[i]) if err != nil { b.Fatal(err) } - _, err = requester.Get(ctx, chunks[i].Address()) + _, err = requester.Get(addrs[i]) if err != nil { b.Fatal(err) } + // wait for update gc goroutine to be done + <-testHookUpdateGCChan } } @@ -113,12 +122,12 @@ func benchmarkRetrievalIndexes(b *testing.B, o *Options, count int) { // goos: darwin // goarch: amd64 // pkg: github.com/ethereum/go-ethereum/swarm/storage/localstore -// BenchmarkUpload/1000-split-8 20 99501623 ns/op 25164178 B/op 22202 allocs/op -// BenchmarkUpload/1000-composite-8 20 103449118 ns/op 25177986 B/op 22204 allocs/op -// BenchmarkUpload/10000-split-8 2 670290376 ns/op 216382840 B/op 239645 allocs/op -// BenchmarkUpload/10000-composite-8 2 667137525 ns/op 216377176 B/op 238854 allocs/op -// BenchmarkUpload/100000-split-8 1 26074429894 ns/op 2326850952 B/op 3932893 allocs/op -// BenchmarkUpload/100000-composite-8 1 26242346728 ns/op 2331055096 B/op 3957569 allocs/op +// BenchmarkUpload/1000-split-8 20 59437463 ns/op 25205193 B/op 23208 allocs/op +// BenchmarkUpload/1000-composite-8 20 59823642 ns/op 25204900 B/op 23202 allocs/op +// BenchmarkUpload/10000-split-8 2 580646362 ns/op 216532932 B/op 248090 allocs/op +// BenchmarkUpload/10000-composite-8 2 589351080 ns/op 216540740 B/op 248007 allocs/op +// BenchmarkUpload/100000-split-8 1 22373390892 ns/op 2323055312 B/op 3995903 allocs/op +// BenchmarkUpload/100000-composite-8 1 22090725078 ns/op 2320312976 B/op 3969219 allocs/op // PASS func BenchmarkUpload(b *testing.B) { for _, count := range []int{ @@ -146,8 +155,7 @@ func benchmarkUpload(b *testing.B, o *Options, count int) { b.StopTimer() db, cleanupFunc := newTestDB(b, o) defer cleanupFunc() - uploader := db.Accessor(ModeUpload) - ctx := context.Background() + uploader := db.NewPutter(ModePutUpload) chunks := make([]storage.Chunk, count) for i := 0; i < count; i++ { chunk := generateFakeRandomChunk() @@ -156,7 +164,7 @@ func benchmarkUpload(b *testing.B, o *Options, count int) { b.StartTimer() for i := 0; i < count; i++ { - err := uploader.Put(ctx, chunks[i]) + err := uploader.Put(chunks[i]) if err != nil { b.Fatal(err) }