From cbb510bb3c2f1fd0ab2e8f73e3e702c91112f4f0 Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Mon, 3 Dec 2018 12:31:24 +0100 Subject: [PATCH] swarm/storage/localstore: add WithRetrievalCompositeIndex --- swarm/storage/localstore/accessor_test.go | 21 ++- swarm/storage/localstore/localstore.go | 148 +++++++++++++++----- swarm/storage/localstore/localstore_test.go | 34 ++++- swarm/storage/localstore/mode.go | 53 +++++-- 4 files changed, 210 insertions(+), 46 deletions(-) diff --git a/swarm/storage/localstore/accessor_test.go b/swarm/storage/localstore/accessor_test.go index 74f4e5323f..9e35e1cdf8 100644 --- a/swarm/storage/localstore/accessor_test.go +++ b/swarm/storage/localstore/accessor_test.go @@ -25,13 +25,28 @@ import ( ) // 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. +// for different accessors. func TestAccessors(t *testing.T) { db, cleanupFunc := newTestDB(t) defer cleanupFunc() + testAccessors(t, db) +} + +// TestAccessors tests most basic Put and Get functionalities +// for different accessors by using retrieval composite index. +func TestAccessors_WithRetrievalCompositeIndex(t *testing.T) { + db, cleanupFunc := newTestDB(t, WithRetrievalCompositeIndex(true)) + defer cleanupFunc() + + testAccessors(t, db) +} + +// 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, diff --git a/swarm/storage/localstore/localstore.go b/swarm/storage/localstore/localstore.go index 843fc04825..bb05ab950c 100644 --- a/swarm/storage/localstore/localstore.go +++ b/swarm/storage/localstore/localstore.go @@ -44,13 +44,25 @@ var ( type DB struct { shed *shed.DB - // fields and indexes - schemaName shed.StringField - sizeCounter shed.Uint64Field - retrievalIndex shed.Index - pushIndex shed.Index - pullIndex shed.Index - gcIndex shed.Index + // fields + schemaName shed.StringField + sizeCounter shed.Uint64Field + + // this flag is for banchmarking 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 + // sync indexes + pushIndex shed.Index + pullIndex shed.Index + // garbage collection index + gcIndex shed.Index baseKey []byte @@ -61,10 +73,26 @@ type DB struct { close chan struct{} // closed on Close, signals other goroutines to terminate } +// Option is a function that sets optional field values on DB. +// It is used as a variadic parameter to New constructor. +type Option func(*DB) + +// WithRetrievalCompositeIndex is the optional variadic parameter to New constructor +// to use the single retrieval composite index instead two separate for data +// and access timestamp. This option is used for benchmarking this two types of +// retrieval schemas for performance. Composite retrieval index performes less seeks +// on retrieval as it has two times less key/value pairs then alternative approach, +// but it needs to write chunk data on every access timestamp change. +func WithRetrievalCompositeIndex(use bool) Option { + return func(db *DB) { + db.useRetrievalCompositeIndex = use + } +} + // New returns a new DB. All fields and indexes are initialized // and possible conflicts with schema from existing database is checked. // One goroutine for writing batches is created. -func New(path string, baseKey []byte) (db *DB, err error) { +func New(path string, baseKey []byte, opts ...Option) (db *DB, err error) { db = &DB{ baseKey: baseKey, batch: newBatch(), @@ -72,6 +100,11 @@ func New(path string, baseKey []byte) (db *DB, err error) { close: make(chan struct{}), writeDone: make(chan struct{}), } + + for _, o := range opts { + o(db) + } + db.shed, err = shed.NewDB(path) if err != nil { return nil, err @@ -85,30 +118,81 @@ func New(path string, baseKey []byte) (db *DB, err error) { if err != nil { return nil, err } - db.retrievalIndex, err = db.shed.NewIndex("Hash->StoredTimestamp|AccessTimestamp|Data", shed.IndexFuncs{ - EncodeKey: func(fields shed.IndexItem) (key []byte, err error) { - return fields.Address, nil - }, - DecodeKey: func(key []byte) (e shed.IndexItem, err error) { - e.Address = key - return e, nil - }, - EncodeValue: func(fields shed.IndexItem) (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 - }, - DecodeValue: func(value []byte) (e shed.IndexItem, 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 - }, - }) - if err != nil { - return nil, err + if db.useRetrievalCompositeIndex { + // 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.IndexItem) (key []byte, err error) { + return fields.Address, nil + }, + DecodeKey: func(key []byte) (e shed.IndexItem, err error) { + e.Address = key + return e, nil + }, + EncodeValue: func(fields shed.IndexItem) (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 + }, + DecodeValue: func(value []byte) (e shed.IndexItem, 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 + }, + }) + if err != nil { + return nil, err + } + } else { + // Index storing actual chunk address, data and store timestamp. + db.retrievalDataIndex, err = db.shed.NewIndex("Address->StoreTimestamp|Data", shed.IndexFuncs{ + EncodeKey: func(fields shed.IndexItem) (key []byte, err error) { + return fields.Address, nil + }, + DecodeKey: func(key []byte) (e shed.IndexItem, err error) { + e.Address = key + return e, nil + }, + EncodeValue: func(fields shed.IndexItem) (value []byte, err error) { + b := make([]byte, 8) + binary.BigEndian.PutUint64(b, uint64(fields.StoreTimestamp)) + value = append(b, fields.Data...) + return value, nil + }, + DecodeValue: func(value []byte) (e shed.IndexItem, err error) { + e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8])) + e.Data = value[8:] + return e, nil + }, + }) + 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.IndexItem) (key []byte, err error) { + return fields.Address, nil + }, + DecodeKey: func(key []byte) (e shed.IndexItem, err error) { + e.Address = key + return e, nil + }, + EncodeValue: func(fields shed.IndexItem) (value []byte, err error) { + b := make([]byte, 8) + binary.BigEndian.PutUint64(b, uint64(fields.AccessTimestamp)) + return b, nil + }, + DecodeValue: func(value []byte) (e shed.IndexItem, 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{ diff --git a/swarm/storage/localstore/localstore_test.go b/swarm/storage/localstore/localstore_test.go index 1a12c27703..e019e0f058 100644 --- a/swarm/storage/localstore/localstore_test.go +++ b/swarm/storage/localstore/localstore_test.go @@ -26,10 +26,40 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage" ) +// TestWithRetrievalCompositeIndex checks if optional argument +// WithRetrievalCompositeIndex to New constructor is setting the +// correct state. +func TestWithRetrievalCompositeIndex(t *testing.T) { + t.Run("set true", func(t *testing.T) { + db, cleanupFunc := newTestDB(t, WithRetrievalCompositeIndex(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, WithRetrievalCompositeIndex(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) + defer cleanupFunc() + + if db.useRetrievalCompositeIndex { + t.Error("useRetrievalCompositeIndex is not set to false") + } + }) +} + // newTestDB is a helper function that constructs a // temporary database and returns a cleanup function that must // be called to remove the data. -func newTestDB(t *testing.T) (db *DB, cleanupFunc func()) { +func newTestDB(t *testing.T, opts ...Option) (db *DB, cleanupFunc func()) { t.Helper() dir, err := ioutil.TempDir("", "shed-test") @@ -41,7 +71,7 @@ func newTestDB(t *testing.T) (db *DB, cleanupFunc func()) { if _, err := rand.Read(baseKey); err != nil { t.Fatal(err) } - db, err = New(dir, baseKey) + db, err = New(dir, baseKey, opts...) if err != nil { cleanupFunc() t.Fatal(err) diff --git a/swarm/storage/localstore/mode.go b/swarm/storage/localstore/mode.go index cba7d0f89c..c42bcee4fe 100644 --- a/swarm/storage/localstore/mode.go +++ b/swarm/storage/localstore/mode.go @@ -64,9 +64,16 @@ func ModeName(m Mode) (name string) { // This function utilizes different indexes depending on // the Mode. func (db *DB) access(mode Mode, item shed.IndexItem) (out shed.IndexItem, err error) { - out, err = db.retrievalIndex.Get(item) - if err != nil { - return out, err + if db.useRetrievalCompositeIndex { + out, err = db.retrievalCompositeIndex.Get(item) + if err != nil { + return out, err + } + } else { + out, err = db.retrievalDataIndex.Get(item) + if err != nil { + return out, err + } } switch mode { case ModeRequest: @@ -137,7 +144,11 @@ func (db *DB) updateBatch(b *batch, mode Mode, item shed.IndexItem) (err error) // put to indexes: retrieve, pull item.StoreTimestamp = now() item.AccessTimestamp = now() - db.retrievalIndex.PutInBatch(b.Batch, item) + if db.useRetrievalCompositeIndex { + db.retrievalCompositeIndex.PutInBatch(b.Batch, item) + } else { + db.retrievalDataIndex.PutInBatch(b.Batch, item) + } db.pullIndex.PutInBatch(b.Batch, item) db.sizeCounter.IncInBatch(b.Batch) @@ -145,7 +156,11 @@ func (db *DB) updateBatch(b *batch, mode Mode, item shed.IndexItem) (err error) // put to indexes: retrieve, push, pull item.StoreTimestamp = now() item.AccessTimestamp = now() - db.retrievalIndex.PutInBatch(b.Batch, item) + if db.useRetrievalCompositeIndex { + db.retrievalCompositeIndex.PutInBatch(b.Batch, item) + } else { + db.retrievalDataIndex.PutInBatch(b.Batch, item) + } db.pullIndex.PutInBatch(b.Batch, item) db.pushIndex.PutInBatch(b.Batch, item) @@ -153,13 +168,23 @@ func (db *DB) updateBatch(b *batch, mode Mode, item shed.IndexItem) (err error) // put to indexes: retrieve, gc item.StoreTimestamp = now() item.AccessTimestamp = now() - db.retrievalIndex.PutInBatch(b.Batch, item) + if db.useRetrievalCompositeIndex { + db.retrievalCompositeIndex.PutInBatch(b.Batch, item) + } else { + db.retrievalDataIndex.PutInBatch(b.Batch, item) + db.retrievalAccessIndex.PutInBatch(b.Batch, item) + } db.gcIndex.PutInBatch(b.Batch, item) case ModeSynced: // delete from push, insert to gc item.StoreTimestamp = now() - db.retrievalIndex.PutInBatch(b.Batch, item) + if db.useRetrievalCompositeIndex { + db.retrievalCompositeIndex.PutInBatch(b.Batch, item) + } else { + db.retrievalDataIndex.PutInBatch(b.Batch, item) + db.retrievalAccessIndex.PutInBatch(b.Batch, item) + } db.pushIndex.DeleteInBatch(b.Batch, item) db.gcIndex.PutInBatch(b.Batch, item) @@ -167,12 +192,22 @@ func (db *DB) updateBatch(b *batch, mode Mode, item shed.IndexItem) (err error) // update accessTimeStamp in retrieve, gc db.gcIndex.DeleteInBatch(b.Batch, item) item.AccessTimestamp = now() - db.retrievalIndex.PutInBatch(b.Batch, item) + if db.useRetrievalCompositeIndex { + db.retrievalCompositeIndex.PutInBatch(b.Batch, item) + } else { + db.retrievalDataIndex.PutInBatch(b.Batch, item) + db.retrievalAccessIndex.PutInBatch(b.Batch, item) + } db.gcIndex.PutInBatch(b.Batch, item) case modeRemoval: // delete from retrieve, pull, gc - db.retrievalIndex.DeleteInBatch(b.Batch, item) + if db.useRetrievalCompositeIndex { + db.retrievalCompositeIndex.DeleteInBatch(b.Batch, item) + } else { + db.retrievalDataIndex.DeleteInBatch(b.Batch, item) + db.retrievalAccessIndex.DeleteInBatch(b.Batch, item) + } db.pullIndex.DeleteInBatch(b.Batch, item) db.gcIndex.DeleteInBatch(b.Batch, item) db.sizeCounter.DecInBatch(b.Batch)