swarm/storage/localstore: add WithRetrievalCompositeIndex

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
+// for different accessors.
-// is retrievable from the database, not if all indexes are set
-// correctly.
 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,

swarm/storage/localstore/localstore.go

@@ -44,13 +44,25 @@ var (
 type DB struct {
 	shed *shed.DB
 
-	// fields and indexes
+	// fields
-	schemaName     shed.StringField
+	schemaName  shed.StringField
-	sizeCounter    shed.Uint64Field
+	sizeCounter shed.Uint64Field
-	retrievalIndex shed.Index
+
-	pushIndex      shed.Index
+	// this flag is for banchmarking two types of retrieval indexes
-	pullIndex      shed.Index
+	// - single retrieval composite index retrievalCompositeIndex
-	gcIndex        shed.Index
+	// - 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{
+	if db.useRetrievalCompositeIndex {
-		EncodeKey: func(fields shed.IndexItem) (key []byte, err error) {
+		// Index storing chunk data with stored and access timestamps.
-			return fields.Address, nil
+		db.retrievalCompositeIndex, err = db.shed.NewIndex("Hash->StoredTimestamp|AccessTimestamp|Data", shed.IndexFuncs{
-		},
+			EncodeKey: func(fields shed.IndexItem) (key []byte, err error) {
-		DecodeKey: func(key []byte) (e shed.IndexItem, err error) {
+				return fields.Address, nil
-			e.Address = key
+			},
-			return e, nil
+			DecodeKey: func(key []byte) (e shed.IndexItem, err error) {
-		},
+				e.Address = key
-		EncodeValue: func(fields shed.IndexItem) (value []byte, err error) {
+				return e, nil
-			b := make([]byte, 16)
+			},
-			binary.BigEndian.PutUint64(b[:8], uint64(fields.StoreTimestamp))
+			EncodeValue: func(fields shed.IndexItem) (value []byte, err error) {
-			binary.BigEndian.PutUint64(b[8:16], uint64(fields.AccessTimestamp))
+				b := make([]byte, 16)
-			value = append(b, fields.Data...)
+				binary.BigEndian.PutUint64(b[:8], uint64(fields.StoreTimestamp))
-			return value, nil
+				binary.BigEndian.PutUint64(b[8:16], uint64(fields.AccessTimestamp))
-		},
+				value = append(b, fields.Data...)
-		DecodeValue: func(value []byte) (e shed.IndexItem, err error) {
+				return value, nil
-			e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
+			},
-			e.AccessTimestamp = int64(binary.BigEndian.Uint64(value[8:16]))
+			DecodeValue: func(value []byte) (e shed.IndexItem, err error) {
-			e.Data = value[16:]
+				e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
-			return e, nil
+				e.AccessTimestamp = int64(binary.BigEndian.Uint64(value[8:16]))
-		},
+				e.Data = value[16:]
-	})
+				return e, nil
-	if err != nil {
+			},
-		return nil, err
+		})
+		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{

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)

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 db.useRetrievalCompositeIndex {
-	if err != nil {
+		out, err = db.retrievalCompositeIndex.Get(item)
-		return out, err
+		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)