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: remove composite retrieval index

Browse source
This commit is contained in:
Janos Guljas 2018-12-18 15:24:41 +01:00
parent e17fec20f2
commit 560532303b
12 changed files with 219 additions and 663 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
swarm/storage/localstore/gc.go
View file

@ -87,12 +87,8 @@ func (db *DB) collectGarbage() (collectedCount int64, done bool, err error) {
return true, nil
}
// delete from retrieve, pull, gc
if db.useRetrievalCompositeIndex {
db.retrievalCompositeIndex.DeleteInBatch(batch, item)
} else {
db.retrievalDataIndex.DeleteInBatch(batch, item)
db.retrievalAccessIndex.DeleteInBatch(batch, item)
}
db.retrievalDataIndex.DeleteInBatch(batch, item)
db.retrievalAccessIndex.DeleteInBatch(batch, item)
db.pullIndex.DeleteInBatch(batch, item)
db.gcIndex.DeleteInBatch(batch, item)
collectedCount++

57
swarm/storage/localstore/gc_test.go
View file

@ -38,19 +38,6 @@ func TestDB_collectGarbageWorker(t *testing.T) {
testDB_collectGarbageWorker(t, db)
}
// TestDB_collectGarbageWorker_useRetrievalCompositeIndex tests
// garbage collection runs by uploading and syncing a number
// of chunks using composite retrieval index.
func TestDB_collectGarbageWorker_useRetrievalCompositeIndex(t *testing.T) {
db, cleanupFunc := newTestDB(t, &Options{
Capacity: 100,
UseRetrievalCompositeIndex: true,
})
defer cleanupFunc()
testDB_collectGarbageWorker(t, db)
}
// TestDB_collectGarbageWorker_multipleBatches tests garbage
// collection runs by uploading and syncing a number of
// chunks by having multiple smaller batches.
@ -68,25 +55,6 @@ func TestDB_collectGarbageWorker_multipleBatches(t *testing.T) {
testDB_collectGarbageWorker(t, db)
}
// TestDB_collectGarbageWorker_multipleBatches_useRetrievalCompositeIndex
// tests garbage collection runs by uploading and syncing a number
// of chunks using composite retrieval index and having multiple
// smaller batches.
func TestDB_collectGarbageWorker_multipleBatches_useRetrievalCompositeIndex(t *testing.T) {
// lower the maximal number of chunks in a single
// gc batch to ensure multiple batches.
defer func(s int64) { gcBatchSize = s }(gcBatchSize)
gcBatchSize = 2
db, cleanupFunc := newTestDB(t, &Options{
Capacity: 100,
UseRetrievalCompositeIndex: true,
})
defer cleanupFunc()
testDB_collectGarbageWorker(t, db)
}
// testDB_collectGarbageWorker is a helper test function to test
// garbage collection runs by uploading and syncing a number of chunks.
func testDB_collectGarbageWorker(t *testing.T, db *DB) {
@ -163,34 +131,15 @@ func testDB_collectGarbageWorker(t *testing.T, db *DB) {
})
}
// TestDB_collectGarbageWorker_withRequests tests garbage collection
// runs by uploading, syncing and requesting a number of chunks.
// TestDB_collectGarbageWorker_withRequests is a helper test function
// to test garbage collection runs by uploading, syncing and
// requesting a number of chunks.
func TestDB_collectGarbageWorker_withRequests(t *testing.T) {
db, cleanupFunc := newTestDB(t, &Options{
Capacity: 100,
})
defer cleanupFunc()
testDB_collectGarbageWorker_withRequests(t, db)
}
// TestDB_collectGarbageWorker_withRequests_useRetrievalCompositeIndex
// tests garbage collection runs by uploading, syncing and
// requesting a number of chunks using composite retrieval index.
func TestDB_collectGarbageWorker_withRequests_useRetrievalCompositeIndex(t *testing.T) {
db, cleanupFunc := newTestDB(t, &Options{
Capacity: 100,
UseRetrievalCompositeIndex: true,
})
defer cleanupFunc()
testDB_collectGarbageWorker_withRequests(t, db)
}
// testDB_collectGarbageWorker_withRequests is a helper test function
// to test garbage collection runs by uploading, syncing and
// requesting a number of chunks.
func testDB_collectGarbageWorker_withRequests(t *testing.T, db *DB) {
uploader := db.NewPutter(ModePutUpload)
syncer := db.NewSetter(ModeSetSync)

17
swarm/storage/localstore/index_test.go
View file

@ -80,24 +80,13 @@ func TestDB_pullIndex(t *testing.T) {
})
}
// 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, 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

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

@ -58,16 +58,9 @@ type DB struct {
// filed that stores number of intems in gc index
storedGCSize shed.Uint64Field
// this flag is for benchmarking two types of retrieval indexes
// - single retrieval composite index retrievalCompositeIndex
// - two separated indexes for data and access time
// - retrievalDataIndex
// - retrievalAccessIndex
useRetrievalCompositeIndex bool
// retrieval indexes
retrievalCompositeIndex shed.Index
retrievalDataIndex shed.Index
retrievalAccessIndex shed.Index
retrievalDataIndex shed.Index
retrievalAccessIndex shed.Index
// sync indexes
pushIndex shed.Index
pullIndex shed.Index
@ -107,15 +100,6 @@ type DB struct {
// Options struct holds optional parameters for configuring DB.
type Options struct {
// UseRetrievalCompositeIndex option is for benchmarking
// two types of retrieval indexes:
// - single retrieval composite index retrievalCompositeIndex
// - two separated indexes for data and access time
// - retrievalDataIndex
// - retrievalAccessIndex
// It should be temporary until the decision is reached
// about the retrieval index structure.
UseRetrievalCompositeIndex bool
// MockStore is a mock node store that is used to store
// chunk data in a central store. It can be used to reduce
// total storage space requirements in testing large number
@ -135,9 +119,8 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
o = new(Options)
}
db = &DB{
capacity: o.Capacity,
baseKey: baseKey,
useRetrievalCompositeIndex: o.UseRetrievalCompositeIndex,
capacity: o.Capacity,
baseKey: baseKey,
// channels collectGarbageTrigger and writeGCSizeTrigger
// need to be buffered with the size of 1
// to signal another event if it
@ -167,130 +150,77 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
if err != nil {
return nil, err
}
if db.useRetrievalCompositeIndex {
var (
encodeValueFunc func(fields shed.Item) (value []byte, err error)
decodeValueFunc func(keyItem shed.Item, value []byte) (e shed.Item, err error)
)
if o.MockStore != nil {
encodeValueFunc = func(fields shed.Item) (value []byte, err error) {
b := make([]byte, 16)
binary.BigEndian.PutUint64(b[:8], uint64(fields.StoreTimestamp))
binary.BigEndian.PutUint64(b[8:16], uint64(fields.AccessTimestamp))
err = o.MockStore.Put(fields.Address, fields.Data)
if err != nil {
return nil, err
}
return b, nil
}
decodeValueFunc = func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
e.AccessTimestamp = int64(binary.BigEndian.Uint64(value[8:16]))
e.Data, err = o.MockStore.Get(keyItem.Address)
return e, err
}
} else {
encodeValueFunc = func(fields shed.Item) (value []byte, err error) {
b := make([]byte, 16)
binary.BigEndian.PutUint64(b[:8], uint64(fields.StoreTimestamp))
binary.BigEndian.PutUint64(b[8:16], uint64(fields.AccessTimestamp))
value = append(b, fields.Data...)
return value, nil
}
decodeValueFunc = func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
e.AccessTimestamp = int64(binary.BigEndian.Uint64(value[8:16]))
e.Data = value[16:]
return e, nil
// Functions for retrieval data index.
var (
encodeValueFunc func(fields shed.Item) (value []byte, err error)
decodeValueFunc func(keyItem shed.Item, value []byte) (e shed.Item, err error)
)
if o.MockStore != nil {
encodeValueFunc = func(fields shed.Item) (value []byte, err error) {
b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(fields.StoreTimestamp))
err = o.MockStore.Put(fields.Address, fields.Data)
if err != nil {
return nil, err
}
return b, nil
}
// Index storing chunk data with stored and access timestamps.
db.retrievalCompositeIndex, err = db.shed.NewIndex("Hash->StoredTimestamp|AccessTimestamp|Data", shed.IndexFuncs{
EncodeKey: func(fields shed.Item) (key []byte, err error) {
return fields.Address, nil
},
DecodeKey: func(key []byte) (e shed.Item, err error) {
e.Address = key
return e, nil
},
EncodeValue: encodeValueFunc,
DecodeValue: decodeValueFunc,
})
if err != nil {
return nil, err
decodeValueFunc = func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
e.Data, err = o.MockStore.Get(keyItem.Address)
return e, err
}
} else {
var (
encodeValueFunc func(fields shed.Item) (value []byte, err error)
decodeValueFunc func(keyItem shed.Item, value []byte) (e shed.Item, err error)
)
if o.MockStore != nil {
encodeValueFunc = func(fields shed.Item) (value []byte, err error) {
b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(fields.StoreTimestamp))
err = o.MockStore.Put(fields.Address, fields.Data)
if err != nil {
return nil, err
}
return b, nil
}
decodeValueFunc = func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
e.Data, err = o.MockStore.Get(keyItem.Address)
return e, err
}
} else {
encodeValueFunc = func(fields shed.Item) (value []byte, err error) {
b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(fields.StoreTimestamp))
value = append(b, fields.Data...)
return value, nil
}
decodeValueFunc = func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
e.Data = value[8:]
return e, nil
}
encodeValueFunc = func(fields shed.Item) (value []byte, err error) {
b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(fields.StoreTimestamp))
value = append(b, fields.Data...)
return value, nil
}
// Index storing actual chunk address, data and store timestamp.
db.retrievalDataIndex, err = db.shed.NewIndex("Address->StoreTimestamp|Data", shed.IndexFuncs{
EncodeKey: func(fields shed.Item) (key []byte, err error) {
return fields.Address, nil
},
DecodeKey: func(key []byte) (e shed.Item, err error) {
e.Address = key
return e, nil
},
EncodeValue: encodeValueFunc,
DecodeValue: decodeValueFunc,
})
if err != nil {
return nil, err
}
// Index storing access timestamp for a particular address.
// It is needed in order to update gc index keys for iteration order.
db.retrievalAccessIndex, err = db.shed.NewIndex("Address->AccessTimestamp", shed.IndexFuncs{
EncodeKey: func(fields shed.Item) (key []byte, err error) {
return fields.Address, nil
},
DecodeKey: func(key []byte) (e shed.Item, err error) {
e.Address = key
return e, nil
},
EncodeValue: func(fields shed.Item) (value []byte, err error) {
b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(fields.AccessTimestamp))
return b, nil
},
DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
e.AccessTimestamp = int64(binary.BigEndian.Uint64(value))
return e, nil
},
})
if err != nil {
return nil, err
decodeValueFunc = func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
e.Data = value[8:]
return e, nil
}
}
// Index storing actual chunk address, data and store timestamp.
db.retrievalDataIndex, err = db.shed.NewIndex("Address->StoreTimestamp|Data", shed.IndexFuncs{
EncodeKey: func(fields shed.Item) (key []byte, err error) {
return fields.Address, nil
},
DecodeKey: func(key []byte) (e shed.Item, err error) {
e.Address = key
return e, nil
},
EncodeValue: encodeValueFunc,
DecodeValue: decodeValueFunc,
})
if err != nil {
return nil, err
}
// Index storing access timestamp for a particular address.
// It is needed in order to update gc index keys for iteration order.
db.retrievalAccessIndex, err = db.shed.NewIndex("Address->AccessTimestamp", shed.IndexFuncs{
EncodeKey: func(fields shed.Item) (key []byte, err error) {
return fields.Address, nil
},
DecodeKey: func(key []byte) (e shed.Item, err error) {
e.Address = key
return e, nil
},
EncodeValue: func(fields shed.Item) (value []byte, err error) {
b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(fields.AccessTimestamp))
return b, nil
},
DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
e.AccessTimestamp = int64(binary.BigEndian.Uint64(value))
return e, nil
},
})
if err != nil {
return nil, err
}
// pull index allows history and live syncing per po bin
db.pullIndex, err = db.shed.NewIndex("PO|StoredTimestamp|Hash->nil", shed.IndexFuncs{
EncodeKey: func(fields shed.Item) (key []byte, err error) {

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

@ -61,36 +61,6 @@ func TestDB(t *testing.T) {
}
}
// TestDB_useRetrievalCompositeIndex checks if optional argument
// WithRetrievalCompositeIndex to New constructor is setting the
// correct state.
func TestDB_useRetrievalCompositeIndex(t *testing.T) {
t.Run("set true", func(t *testing.T) {
db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: true})
defer cleanupFunc()
if !db.useRetrievalCompositeIndex {
t.Error("useRetrievalCompositeIndex is not set to true")
}
})
t.Run("set false", func(t *testing.T) {
db, cleanupFunc := newTestDB(t, &Options{UseRetrievalCompositeIndex: false})
defer cleanupFunc()
if db.useRetrievalCompositeIndex {
t.Error("useRetrievalCompositeIndex is not set to false")
}
})
t.Run("unset", func(t *testing.T) {
db, cleanupFunc := newTestDB(t, nil)
defer cleanupFunc()
if db.useRetrievalCompositeIndex {
t.Error("useRetrievalCompositeIndex is not set to false")
}
})
}
// TestDB_updateGCSem tests maxParallelUpdateGC limit.
// This test temporary sets the limit to a low number,
// makes updateGC function execution time longer by
@ -321,25 +291,17 @@ func TestGenerateFakeRandomChunk(t *testing.T) {
// 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)
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)
}
// 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)
}
}
}
@ -348,26 +310,18 @@ func newRetrieveIndexesTest(db *DB, chunk storage.Chunk, storeTimestamp, accessT
// 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)
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)
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)
}
}
}

43
swarm/storage/localstore/mode_get.go
View file

@ -70,19 +70,9 @@ func (g *Getter) Get(addr storage.Address) (chunk storage.Chunk, err error) {
func (db *DB) get(mode ModeGet, addr storage.Address) (out shed.Item, 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
}
out, err = db.retrievalDataIndex.Get(item)
if err != nil {
return out, err
}
switch mode {
// update the access timestamp and gc index
@ -133,19 +123,14 @@ func (db *DB) updateGC(item shed.Item) (err error) {
// 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
}
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
@ -157,11 +142,7 @@ func (db *DB) updateGC(item shed.Item) (err error) {
// update access timestamp
item.AccessTimestamp = now()
// update retrieve access index
if db.useRetrievalCompositeIndex {
db.retrievalCompositeIndex.PutInBatch(batch, item)
} else {
db.retrievalAccessIndex.PutInBatch(batch, item)
}
db.retrievalAccessIndex.PutInBatch(batch, item)
// add new entry to gc index
db.gcIndex.PutInBatch(batch, item)

36
swarm/storage/localstore/mode_get_test.go
View file

@ -22,27 +22,11 @@ import (
"time"
)
// TestModeGetRequest validates internal data operations and state
// for ModeGetRequest on DB with default configuration.
// TestModeGetRequest validates ModeGetRequest index values on the provided DB.
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
@ -152,27 +136,11 @@ func testModeGetRequestValues(t *testing.T, db *DB) {
})
}
// TestModeGetSync validates internal data operations and state
// for ModeGetSync on DB with default configuration.
// TestModeGetSync validates ModeGetSync index values on the provided DB.
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

73
swarm/storage/localstore/mode_put.go
View file

@ -78,36 +78,23 @@ func (db *DB) put(mode ModePut, item shed.Item) (err error) {
// 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
}
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
@ -120,32 +107,20 @@ func (db *DB) put(mode ModePut, item shed.Item) (err error) {
// update access timestamp
item.AccessTimestamp = now()
// update retrieve access index
if db.useRetrievalCompositeIndex {
db.retrievalCompositeIndex.PutInBatch(batch, item)
} else {
db.retrievalAccessIndex.PutInBatch(batch, item)
}
db.retrievalAccessIndex.PutInBatch(batch, item)
// add new entry to gc index
db.gcIndex.PutInBatch(batch, item)
db.gcUncountedHashesIndex.PutInBatch(batch, item)
db.incGCSize(1)
if db.useRetrievalCompositeIndex {
db.retrievalCompositeIndex.PutInBatch(batch, item)
} else {
db.retrievalDataIndex.PutInBatch(batch, item)
db.retrievalAccessIndex.PutInBatch(batch, item)
}
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.retrievalDataIndex.PutInBatch(batch, item)
db.pullIndex.PutInBatch(batch, item)
db.pushIndex.PutInBatch(batch, item)
@ -153,11 +128,7 @@ func (db *DB) put(mode ModePut, item shed.Item) (err error) {
// put to indexes: retrieve, pull
item.StoreTimestamp = now()
if db.useRetrievalCompositeIndex {
db.retrievalCompositeIndex.PutInBatch(batch, item)
} else {
db.retrievalDataIndex.PutInBatch(batch, item)
}
db.retrievalDataIndex.PutInBatch(batch, item)
db.pullIndex.PutInBatch(batch, item)
default:

54
swarm/storage/localstore/mode_put_test.go
View file

@ -21,27 +21,11 @@ import (
"time"
)
// TestModePutRequest validates internal data operations and state
// for ModePutRequest on DB with default configuration.
// TestModePutRequest validates ModePutRequest index values on the provided DB.
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()
@ -88,27 +72,11 @@ func testModePutRequestValues(t *testing.T, db *DB) {
})
}
// TestModePutSync validates internal data operations and state
// for ModePutSync on DB with default configuration.
// TestModePutSync validates ModePutSync index values on the provided DB.
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
@ -126,27 +94,11 @@ func testModePutSyncValues(t *testing.T, db *DB) {
t.Run("pull index", newPullIndexTest(db, chunk, wantTimestamp, nil))
}
// TestModePutUpload validates internal data operations and state
// for ModePutUpload on DB with default configuration.
// TestModePutUpload validates ModePutUpload index values on the provided DB.
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

180
swarm/storage/localstore/mode_set.go
View file

@ -79,47 +79,31 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
// 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)
db.incGCSize(-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)
db.incGCSize(-1)
case leveldb.ErrNotFound:
// the chunk is not accessed before
default:
return err
}
item.AccessTimestamp = now()
db.retrievalAccessIndex.PutInBatch(batch, item)
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)
db.incGCSize(-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)
db.gcUncountedHashesIndex.PutInBatch(batch, item)
@ -131,62 +115,33 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
// 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
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
}
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)
db.incGCSize(-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)
db.incGCSize(-1)
case leveldb.ErrNotFound:
// the chunk is not accessed before
default:
return err
}
item.AccessTimestamp = now()
db.retrievalAccessIndex.PutInBatch(batch, item)
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)
db.incGCSize(-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)
db.gcUncountedHashesIndex.PutInBatch(batch, item)
@ -198,34 +153,23 @@ func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
// 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
i, err := db.retrievalAccessIndex.Get(item)
switch err {
case nil:
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
case leveldb.ErrNotFound:
default:
return err
}
if db.useRetrievalCompositeIndex {
db.retrievalCompositeIndex.DeleteInBatch(batch, item)
} else {
db.retrievalDataIndex.DeleteInBatch(batch, item)
db.retrievalAccessIndex.DeleteInBatch(batch, item)
i, err = db.retrievalDataIndex.Get(item)
if err != nil {
return err
}
item.StoreTimestamp = i.StoreTimestamp
db.retrievalDataIndex.DeleteInBatch(batch, item)
db.retrievalAccessIndex.DeleteInBatch(batch, item)
db.pullIndex.DeleteInBatch(batch, item)
db.gcIndex.DeleteInBatch(batch, item)
db.gcUncountedHashesIndex.DeleteInBatch(batch, item)

84
swarm/storage/localstore/mode_set_test.go
View file

@ -23,27 +23,11 @@ import (
"github.com/syndtr/goleveldb/leveldb"
)
// TestModeSetAccess validates internal data operations and state
// for ModeSetAccess on DB with default configuration.
// TestModeSetAccess validates ModeSetAccess index values on the provided DB.
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()
@ -67,27 +51,11 @@ func testModeSetAccessValues(t *testing.T, db *DB) {
t.Run("gc size", newIndexGCSizeTest(db))
}
// TestModeSetSync validates internal data operations and state
// for ModeSetSync on DB with default configuration.
// TestModeSetSync validates ModeSetSync index values on the provided DB.
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()
@ -116,27 +84,11 @@ func testModeSetSyncValues(t *testing.T, db *DB) {
t.Run("gc size", newIndexGCSizeTest(db))
}
// TestModeSetRemoval validates internal data operations and state
// for ModeSetRemoval on DB with default configuration.
// TestModeSetRemoval validates ModeSetRemoval index values on the provided DB.
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)
@ -151,26 +103,18 @@ func testModeSetRemovalValues(t *testing.T, db *DB) {
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", newItemsCountTest(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", newItemsCountTest(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", newItemsCountTest(db.retrievalAccessIndex, 0))
_, 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", newItemsCountTest(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", newItemsCountTest(db.retrievalAccessIndex, 0))
})
t.Run("pull index", newPullIndexTest(db, chunk, 0, leveldb.ErrNotFound))

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

@ -23,13 +23,7 @@ import (
"github.com/ethereum/go-ethereum/swarm/storage"
)
// BenchmarkRetrievalIndexes compares two different retrieval
// index schemas:
// - single retrieval composite index retrievalCompositeIndex
// - two separated indexes for data and access time
// - retrievalDataIndex
// - retrievalAccessIndex
// This benchmark uploads a number of chunks in order to measure
// BenchmarkRetrievalIndexes uploads a number of chunks in order to measure
// total time of updating their retrieval indexes by setting them
// to synced state and requesting them.
//
@ -42,12 +36,9 @@ import (
// goos: darwin
// goarch: amd64
// pkg: github.com/ethereum/go-ethereum/swarm/storage/localstore
// 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
// BenchmarkRetrievalIndexes/1000-8 20 75556686 ns/op 19033493 B/op 84500 allocs/op
// BenchmarkRetrievalIndexes/10000-8 1 1079084922 ns/op 382792064 B/op 1429644 allocs/op
// BenchmarkRetrievalIndexes/100000-8 1 16891305737 ns/op 2629165304 B/op 12465019 allocs/op
// PASS
func BenchmarkRetrievalIndexes(b *testing.B) {
for _, count := range []int{
@ -60,11 +51,6 @@ func BenchmarkRetrievalIndexes(b *testing.B) {
benchmarkRetrievalIndexes(b, nil, count)
}
})
b.Run(strconv.Itoa(count)+"-composite", func(b *testing.B) {
for n := 0; n < b.N; n++ {
benchmarkRetrievalIndexes(b, &Options{UseRetrievalCompositeIndex: true}, count)
}
})
}
}
@ -122,12 +108,9 @@ 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 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
// BenchmarkUpload/1000-8 20 59437463 ns/op 25205193 B/op 23208 allocs/op
// BenchmarkUpload/10000-8 2 580646362 ns/op 216532932 B/op 248090 allocs/op
// BenchmarkUpload/100000-8 1 22373390892 ns/op 2323055312 B/op 3995903 allocs/op
// PASS
func BenchmarkUpload(b *testing.B) {
for _, count := range []int{
@ -135,16 +118,11 @@ func BenchmarkUpload(b *testing.B) {
10000,
100000,
} {
b.Run(strconv.Itoa(count)+"-split", func(b *testing.B) {
b.Run(strconv.Itoa(count), func(b *testing.B) {
for n := 0; n < b.N; n++ {
benchmarkUpload(b, nil, count)
}
})
b.Run(strconv.Itoa(count)+"-composite", func(b *testing.B) {
for n := 0; n < b.N; n++ {
benchmarkUpload(b, &Options{UseRetrievalCompositeIndex: true}, count)
}
})
}
}