forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 18:32:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

swarm/storage/localstore: accessors redesign

Browse source
This commit is contained in:
Janos Guljas 2018-12-13 10:25:48 +01:00
parent f2299f4703
commit e6bdda7078
14 changed files with 1796 additions and 1618 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

74
swarm/storage/localstore/accessor.go
View file

@ -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 <http://www.gnu.org/licenses/>.
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,
}
}

217
swarm/storage/localstore/accessor_test.go
View file

@ -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 <http://www.gnu.org/licenses/>.
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)
}
}

238
swarm/storage/localstore/index_test.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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)
})
}

51
swarm/storage/localstore/localstore.go
View file

@ -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

360
swarm/storage/localstore/localstore_test.go
View file

@ -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)
}
}

369
swarm/storage/localstore/mode.go
View file

@ -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 <http://www.gnu.org/licenses/>.
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)
}

161
swarm/storage/localstore/mode_get.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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()

206
swarm/storage/localstore/mode_get_test.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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))
}

169
swarm/storage/localstore/mode_put.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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)
}

167
swarm/storage/localstore/mode_put_test.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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))
}

242
swarm/storage/localstore/mode_set.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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)
}

184
swarm/storage/localstore/mode_set_test.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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))
}

918
swarm/storage/localstore/mode_test.go
View file

@ -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 <http://www.gnu.org/licenses/>.
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)
}
}

58
swarm/storage/localstore/retrieval_index_test.go
View file

@ -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)
}