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/shed, swarm/storage/localstore: rename IndexItem to Item

Browse source
This commit is contained in:
Janos Guljas 2018-12-17 10:29:02 +01:00
parent e6e29f55ee
commit 750268d8d0
14 changed files with 187 additions and 187 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
swarm/shed/db.go
View file

@ -18,7 +18,7 @@
// more complex operations on storage data organized in fields and indexes. // more complex operations on storage data organized in fields and indexes.
// //
// Only type which holds logical information about swarm storage chunks data // Only type which holds logical information about swarm storage chunks data
// and metadata is IndexItem. This part is not generalized mostly for // and metadata is Item. This part is not generalized mostly for
// performance reasons. // performance reasons.
package shed package shed

38
swarm/shed/example_store_test.go
View file

@ -71,20 +71,20 @@ func New(path string) (s *Store, err error) {
} }
// Index storing actual chunk address, data and store timestamp. // Index storing actual chunk address, data and store timestamp.
s.retrievalIndex, err = db.NewIndex("Address->StoreTimestamp|Data", shed.IndexFuncs{ s.retrievalIndex, err = db.NewIndex("Address->StoreTimestamp|Data", shed.IndexFuncs{
EncodeKey: func(fields shed.IndexItem) (key []byte, err error) { EncodeKey: func(fields shed.Item) (key []byte, err error) {
return fields.Address, nil return fields.Address, nil
}, },
DecodeKey: func(key []byte) (e shed.IndexItem, err error) { DecodeKey: func(key []byte) (e shed.Item, err error) {
e.Address = key e.Address = key
return e, nil return e, nil
}, },
EncodeValue: func(fields shed.IndexItem) (value []byte, err error) { EncodeValue: func(fields shed.Item) (value []byte, err error) {
b := make([]byte, 8) b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(fields.StoreTimestamp)) binary.BigEndian.PutUint64(b, uint64(fields.StoreTimestamp))
value = append(b, fields.Data...) value = append(b, fields.Data...)
return value, nil return value, nil
}, },
DecodeValue: func(keyItem shed.IndexItem, value []byte) (e shed.IndexItem, err error) { DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8])) e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
e.Data = value[8:] e.Data = value[8:]
return e, nil return e, nil
@ -96,19 +96,19 @@ func New(path string) (s *Store, err error) {
// Index storing access timestamp for a particular address. // Index storing access timestamp for a particular address.
// It is needed in order to update gc index keys for iteration order. // It is needed in order to update gc index keys for iteration order.
s.accessIndex, err = db.NewIndex("Address->AccessTimestamp", shed.IndexFuncs{ s.accessIndex, err = db.NewIndex("Address->AccessTimestamp", shed.IndexFuncs{
EncodeKey: func(fields shed.IndexItem) (key []byte, err error) { EncodeKey: func(fields shed.Item) (key []byte, err error) {
return fields.Address, nil return fields.Address, nil
}, },
DecodeKey: func(key []byte) (e shed.IndexItem, err error) { DecodeKey: func(key []byte) (e shed.Item, err error) {
e.Address = key e.Address = key
return e, nil return e, nil
}, },
EncodeValue: func(fields shed.IndexItem) (value []byte, err error) { EncodeValue: func(fields shed.Item) (value []byte, err error) {
b := make([]byte, 8) b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(fields.AccessTimestamp)) binary.BigEndian.PutUint64(b, uint64(fields.AccessTimestamp))
return b, nil return b, nil
}, },
DecodeValue: func(keyItem shed.IndexItem, value []byte) (e shed.IndexItem, err error) { DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
e.AccessTimestamp = int64(binary.BigEndian.Uint64(value)) e.AccessTimestamp = int64(binary.BigEndian.Uint64(value))
return e, nil return e, nil
}, },
@ -118,23 +118,23 @@ func New(path string) (s *Store, err error) {
} }
// Index with keys ordered by access timestamp for garbage collection prioritization. // Index with keys ordered by access timestamp for garbage collection prioritization.
s.gcIndex, err = db.NewIndex("AccessTimestamp|StoredTimestamp|Address->nil", shed.IndexFuncs{ s.gcIndex, err = db.NewIndex("AccessTimestamp|StoredTimestamp|Address->nil", shed.IndexFuncs{
EncodeKey: func(fields shed.IndexItem) (key []byte, err error) { EncodeKey: func(fields shed.Item) (key []byte, err error) {
b := make([]byte, 16, 16+len(fields.Address)) b := make([]byte, 16, 16+len(fields.Address))
binary.BigEndian.PutUint64(b[:8], uint64(fields.AccessTimestamp)) binary.BigEndian.PutUint64(b[:8], uint64(fields.AccessTimestamp))
binary.BigEndian.PutUint64(b[8:16], uint64(fields.StoreTimestamp)) binary.BigEndian.PutUint64(b[8:16], uint64(fields.StoreTimestamp))
key = append(b, fields.Address...) key = append(b, fields.Address...)
return key, nil return key, nil
}, },
DecodeKey: func(key []byte) (e shed.IndexItem, err error) { DecodeKey: func(key []byte) (e shed.Item, err error) {
e.AccessTimestamp = int64(binary.BigEndian.Uint64(key[:8])) e.AccessTimestamp = int64(binary.BigEndian.Uint64(key[:8]))
e.StoreTimestamp = int64(binary.BigEndian.Uint64(key[8:16])) e.StoreTimestamp = int64(binary.BigEndian.Uint64(key[8:16]))
e.Address = key[16:] e.Address = key[16:]
return e, nil return e, nil
}, },
EncodeValue: func(fields shed.IndexItem) (value []byte, err error) { EncodeValue: func(fields shed.Item) (value []byte, err error) {
return nil, nil return nil, nil
}, },
DecodeValue: func(keyItem shed.IndexItem, value []byte) (e shed.IndexItem, err error) { DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
return e, nil return e, nil
}, },
}) })
@ -146,7 +146,7 @@ func New(path string) (s *Store, err error) {
// Put stores the chunk and sets it store timestamp. // Put stores the chunk and sets it store timestamp.
func (s *Store) Put(_ context.Context, ch storage.Chunk) (err error) { func (s *Store) Put(_ context.Context, ch storage.Chunk) (err error) {
return s.retrievalIndex.Put(shed.IndexItem{ return s.retrievalIndex.Put(shed.Item{
Address: ch.Address(), Address: ch.Address(),
Data: ch.Data(), Data: ch.Data(),
StoreTimestamp: time.Now().UTC().UnixNano(), StoreTimestamp: time.Now().UTC().UnixNano(),
@ -161,7 +161,7 @@ func (s *Store) Get(_ context.Context, addr storage.Address) (c storage.Chunk, e
batch := new(leveldb.Batch) batch := new(leveldb.Batch)
// Get the chunk data and storage timestamp. // Get the chunk data and storage timestamp.
item, err := s.retrievalIndex.Get(shed.IndexItem{ item, err := s.retrievalIndex.Get(shed.Item{
Address: addr, Address: addr,
}) })
if err != nil { if err != nil {
@ -172,13 +172,13 @@ func (s *Store) Get(_ context.Context, addr storage.Address) (c storage.Chunk, e
} }
// Get the chunk access timestamp. // Get the chunk access timestamp.
accessItem, err := s.accessIndex.Get(shed.IndexItem{ accessItem, err := s.accessIndex.Get(shed.Item{
Address: addr, Address: addr,
}) })
switch err { switch err {
case nil: case nil:
// Remove gc index entry if access timestamp is found. // Remove gc index entry if access timestamp is found.
err = s.gcIndex.DeleteInBatch(batch, shed.IndexItem{ err = s.gcIndex.DeleteInBatch(batch, shed.Item{
Address: item.Address, Address: item.Address,
StoreTimestamp: accessItem.AccessTimestamp, StoreTimestamp: accessItem.AccessTimestamp,
AccessTimestamp: item.StoreTimestamp, AccessTimestamp: item.StoreTimestamp,
@ -197,7 +197,7 @@ func (s *Store) Get(_ context.Context, addr storage.Address) (c storage.Chunk, e
accessTimestamp := time.Now().UTC().UnixNano() accessTimestamp := time.Now().UTC().UnixNano()
// Put new access timestamp in access index. // Put new access timestamp in access index.
err = s.accessIndex.PutInBatch(batch, shed.IndexItem{ err = s.accessIndex.PutInBatch(batch, shed.Item{
Address: addr, Address: addr,
AccessTimestamp: accessTimestamp, AccessTimestamp: accessTimestamp,
}) })
@ -206,7 +206,7 @@ func (s *Store) Get(_ context.Context, addr storage.Address) (c storage.Chunk, e
} }
// Put new access timestamp in gc index. // Put new access timestamp in gc index.
err = s.gcIndex.PutInBatch(batch, shed.IndexItem{ err = s.gcIndex.PutInBatch(batch, shed.Item{
Address: item.Address, Address: item.Address,
AccessTimestamp: accessTimestamp, AccessTimestamp: accessTimestamp,
StoreTimestamp: item.StoreTimestamp, StoreTimestamp: item.StoreTimestamp,
@ -244,7 +244,7 @@ func (s *Store) CollectGarbage() (err error) {
// New batch for a new cg round. // New batch for a new cg round.
trash := new(leveldb.Batch) trash := new(leveldb.Batch)
// Iterate through all index items and break when needed. // Iterate through all index items and break when needed.
err = s.gcIndex.IterateAll(func(item shed.IndexItem) (stop bool, err error) { err = s.gcIndex.IterateAll(func(item shed.Item) (stop bool, err error) {
// Remove the chunk. // Remove the chunk.
err = s.retrievalIndex.DeleteInBatch(trash, item) err = s.retrievalIndex.DeleteInBatch(trash, item)
if err != nil { if err != nil {

62
swarm/shed/index.go
View file

@ -20,19 +20,19 @@ import (
"github.com/syndtr/goleveldb/leveldb" "github.com/syndtr/goleveldb/leveldb"
) )
// IndexItem holds fields relevant to Swarm Chunk data and metadata. // Item holds fields relevant to Swarm Chunk data and metadata.
// All information required for swarm storage and operations // All information required for swarm storage and operations
// on that storage must be defined here. // on that storage must be defined here.
// This structure is logically connected to swarm storage, // This structure is logically connected to swarm storage,
// the only part of this package that is not generalized, // the only part of this package that is not generalized,
// mostly for performance reasons. // mostly for performance reasons.
// //
// IndexItem is a type that is used for retrieving, storing and encoding // Item is a type that is used for retrieving, storing and encoding
// chunk data and metadata. It is passed as an argument to Index encoding // chunk data and metadata. It is passed as an argument to Index encoding
// functions, get function and put function. // functions, get function and put function.
// But it is also returned with additional data from get function call // But it is also returned with additional data from get function call
// and as the argument in iterator function definition. // and as the argument in iterator function definition.
type IndexItem struct { type Item struct {
Address []byte Address []byte
Data []byte Data []byte
AccessTimestamp int64 AccessTimestamp int64
@ -43,9 +43,9 @@ type IndexItem struct {
} }
// Merge is a helper method to construct a new // Merge is a helper method to construct a new
// IndexItem by filling up fields with default values // Item by filling up fields with default values
// of a particular IndexItem with values from another one. // of a particular Item with values from another one.
func (i IndexItem) Merge(i2 IndexItem) (new IndexItem) { func (i Item) Merge(i2 Item) (new Item) {
if i.Address == nil { if i.Address == nil {
i.Address = i2.Address i.Address = i2.Address
} }
@ -67,26 +67,26 @@ func (i IndexItem) Merge(i2 IndexItem) (new IndexItem) {
// Index represents a set of LevelDB key value pairs that have common // Index represents a set of LevelDB key value pairs that have common
// prefix. It holds functions for encoding and decoding keys and values // prefix. It holds functions for encoding and decoding keys and values
// to provide transparent actions on saved data which inclide: // to provide transparent actions on saved data which inclide:
// - getting a particular IndexItem // - getting a particular Item
// - saving a particular IndexItem // - saving a particular Item
// - iterating over a sorted LevelDB keys // - iterating over a sorted LevelDB keys
// It implements IndexIteratorInterface interface. // It implements IndexIteratorInterface interface.
type Index struct { type Index struct {
db *DB db *DB
prefix []byte prefix []byte
encodeKeyFunc func(fields IndexItem) (key []byte, err error) encodeKeyFunc func(fields Item) (key []byte, err error)
decodeKeyFunc func(key []byte) (e IndexItem, err error) decodeKeyFunc func(key []byte) (e Item, err error)
encodeValueFunc func(fields IndexItem) (value []byte, err error) encodeValueFunc func(fields Item) (value []byte, err error)
decodeValueFunc func(keyFields IndexItem, value []byte) (e IndexItem, err error) decodeValueFunc func(keyFields Item, value []byte) (e Item, err error)
} }
// IndexFuncs structure defines functions for encoding and decoding // IndexFuncs structure defines functions for encoding and decoding
// LevelDB keys and values for a specific index. // LevelDB keys and values for a specific index.
type IndexFuncs struct { type IndexFuncs struct {
EncodeKey func(fields IndexItem) (key []byte, err error) EncodeKey func(fields Item) (key []byte, err error)
DecodeKey func(key []byte) (e IndexItem, err error) DecodeKey func(key []byte) (e Item, err error)
EncodeValue func(fields IndexItem) (value []byte, err error) EncodeValue func(fields Item) (value []byte, err error)
DecodeValue func(keyFields IndexItem, value []byte) (e IndexItem, err error) DecodeValue func(keyFields Item, value []byte) (e Item, err error)
} }
// NewIndex returns a new Index instance with defined name and // NewIndex returns a new Index instance with defined name and
@ -105,7 +105,7 @@ func (db *DB) NewIndex(name string, funcs IndexFuncs) (f Index, err error) {
// by appending the provided index id byte. // by appending the provided index id byte.
// This is needed to avoid collisions between keys of different // This is needed to avoid collisions between keys of different
// indexes as all index ids are unique. // indexes as all index ids are unique.
encodeKeyFunc: func(e IndexItem) (key []byte, err error) { encodeKeyFunc: func(e Item) (key []byte, err error) {
key, err = funcs.EncodeKey(e) key, err = funcs.EncodeKey(e)
if err != nil { if err != nil {
return nil, err return nil, err
@ -115,7 +115,7 @@ func (db *DB) NewIndex(name string, funcs IndexFuncs) (f Index, err error) {
// This function reverses the encodeKeyFunc constructed key // This function reverses the encodeKeyFunc constructed key
// to transparently work with index keys without their index ids. // to transparently work with index keys without their index ids.
// It assumes that index keys are prefixed with only one byte. // It assumes that index keys are prefixed with only one byte.
decodeKeyFunc: func(key []byte) (e IndexItem, err error) { decodeKeyFunc: func(key []byte) (e Item, err error) {
return funcs.DecodeKey(key[1:]) return funcs.DecodeKey(key[1:])
}, },
encodeValueFunc: funcs.EncodeValue, encodeValueFunc: funcs.EncodeValue,
@ -123,10 +123,10 @@ func (db *DB) NewIndex(name string, funcs IndexFuncs) (f Index, err error) {
}, nil }, nil
} }
// Get accepts key fields represented as IndexItem to retrieve a // Get accepts key fields represented as Item to retrieve a
// value from the index and return maximum available information // value from the index and return maximum available information
// from the index represented as another IndexItem. // from the index represented as another Item.
func (f Index) Get(keyFields IndexItem) (out IndexItem, err error) { func (f Index) Get(keyFields Item) (out Item, err error) {
key, err := f.encodeKeyFunc(keyFields) key, err := f.encodeKeyFunc(keyFields)
if err != nil { if err != nil {
return out, err return out, err
@ -142,9 +142,9 @@ func (f Index) Get(keyFields IndexItem) (out IndexItem, err error) {
return out.Merge(keyFields), nil return out.Merge(keyFields), nil
} }
// Put accepts IndexItem to encode information from it // Put accepts Item to encode information from it
// and save it to the database. // and save it to the database.
func (f Index) Put(i IndexItem) (err error) { func (f Index) Put(i Item) (err error) {
key, err := f.encodeKeyFunc(i) key, err := f.encodeKeyFunc(i)
if err != nil { if err != nil {
return err return err
@ -159,7 +159,7 @@ func (f Index) Put(i IndexItem) (err error) {
// PutInBatch is the same as Put method, but it just // PutInBatch is the same as Put method, but it just
// saves the key/value pair to the batch instead // saves the key/value pair to the batch instead
// directly to the database. // directly to the database.
func (f Index) PutInBatch(batch *leveldb.Batch, i IndexItem) (err error) { func (f Index) PutInBatch(batch *leveldb.Batch, i Item) (err error) {
key, err := f.encodeKeyFunc(i) key, err := f.encodeKeyFunc(i)
if err != nil { if err != nil {
return err return err
@ -172,9 +172,9 @@ func (f Index) PutInBatch(batch *leveldb.Batch, i IndexItem) (err error) {
return nil return nil
} }
// Delete accepts IndexItem to remove a key/value pair // Delete accepts Item to remove a key/value pair
// from the database based on its fields. // from the database based on its fields.
func (f Index) Delete(keyFields IndexItem) (err error) { func (f Index) Delete(keyFields Item) (err error) {
key, err := f.encodeKeyFunc(keyFields) key, err := f.encodeKeyFunc(keyFields)
if err != nil { if err != nil {
return err return err
@ -184,7 +184,7 @@ func (f Index) Delete(keyFields IndexItem) (err error) {
// DeleteInBatch is the same as Delete just the operation // DeleteInBatch is the same as Delete just the operation
// is performed on the batch instead on the database. // is performed on the batch instead on the database.
func (f Index) DeleteInBatch(batch *leveldb.Batch, keyFields IndexItem) (err error) { func (f Index) DeleteInBatch(batch *leveldb.Batch, keyFields Item) (err error) {
key, err := f.encodeKeyFunc(keyFields) key, err := f.encodeKeyFunc(keyFields)
if err != nil { if err != nil {
return err return err
@ -193,12 +193,12 @@ func (f Index) DeleteInBatch(batch *leveldb.Batch, keyFields IndexItem) (err err
return nil return nil
} }
// IndexIterFunc is a callback on every IndexItem that is decoded // IndexIterFunc is a callback on every Item that is decoded
// by iterating on an Index keys. // by iterating on an Index keys.
// By returning a true for stop variable, iteration will // By returning a true for stop variable, iteration will
// stop, and by returning the error, that error will be // stop, and by returning the error, that error will be
// propagated to the called iterator method on Index. // propagated to the called iterator method on Index.
type IndexIterFunc func(item IndexItem) (stop bool, err error) type IndexIterFunc func(item Item) (stop bool, err error)
// IterateAll iterates over all keys of the Index. // IterateAll iterates over all keys of the Index.
func (f Index) IterateAll(fn IndexIterFunc) (err error) { func (f Index) IterateAll(fn IndexIterFunc) (err error) {
@ -230,8 +230,8 @@ func (f Index) IterateAll(fn IndexIterFunc) (err error) {
} }
// IterateFrom iterates over Index keys starting from the key // IterateFrom iterates over Index keys starting from the key
// encoded from the provided IndexItem. // encoded from the provided Item.
func (f Index) IterateFrom(start IndexItem, fn IndexIterFunc) (err error) { func (f Index) IterateFrom(start Item, fn IndexIterFunc) (err error) {
startKey, err := f.encodeKeyFunc(start) startKey, err := f.encodeKeyFunc(start)
if err != nil { if err != nil {
return err return err

96
swarm/shed/index_test.go
View file

@ -29,20 +29,20 @@ import (
// Index functions for the index that is used in tests in this file. // Index functions for the index that is used in tests in this file.
var retrievalIndexFuncs = IndexFuncs{ var retrievalIndexFuncs = IndexFuncs{
EncodeKey: func(fields IndexItem) (key []byte, err error) { EncodeKey: func(fields Item) (key []byte, err error) {
return fields.Address, nil return fields.Address, nil
}, },
DecodeKey: func(key []byte) (e IndexItem, err error) { DecodeKey: func(key []byte) (e Item, err error) {
e.Address = key e.Address = key
return e, nil return e, nil
}, },
EncodeValue: func(fields IndexItem) (value []byte, err error) { EncodeValue: func(fields Item) (value []byte, err error) {
b := make([]byte, 8) b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(fields.StoreTimestamp)) binary.BigEndian.PutUint64(b, uint64(fields.StoreTimestamp))
value = append(b, fields.Data...) value = append(b, fields.Data...)
return value, nil return value, nil
}, },
DecodeValue: func(keyItem IndexItem, value []byte) (e IndexItem, err error) { DecodeValue: func(keyItem Item, value []byte) (e Item, err error) {
e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8])) e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
e.Data = value[8:] e.Data = value[8:]
return e, nil return e, nil
@ -60,7 +60,7 @@ func TestIndex(t *testing.T) {
} }
t.Run("put", func(t *testing.T) { t.Run("put", func(t *testing.T) {
want := IndexItem{ want := Item{
Address: []byte("put-hash"), Address: []byte("put-hash"),
Data: []byte("DATA"), Data: []byte("DATA"),
StoreTimestamp: time.Now().UTC().UnixNano(), StoreTimestamp: time.Now().UTC().UnixNano(),
@ -70,16 +70,16 @@ func TestIndex(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
got, err := index.Get(IndexItem{ got, err := index.Get(Item{
Address: want.Address, Address: want.Address,
}) })
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
checkIndexItem(t, got, want) checkItem(t, got, want)
t.Run("overwrite", func(t *testing.T) { t.Run("overwrite", func(t *testing.T) {
want := IndexItem{ want := Item{
Address: []byte("put-hash"), Address: []byte("put-hash"),
Data: []byte("New DATA"), Data: []byte("New DATA"),
StoreTimestamp: time.Now().UTC().UnixNano(), StoreTimestamp: time.Now().UTC().UnixNano(),
@ -89,18 +89,18 @@ func TestIndex(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
got, err := index.Get(IndexItem{ got, err := index.Get(Item{
Address: want.Address, Address: want.Address,
}) })
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
checkIndexItem(t, got, want) checkItem(t, got, want)
}) })
}) })
t.Run("put in batch", func(t *testing.T) { t.Run("put in batch", func(t *testing.T) {
want := IndexItem{ want := Item{
Address: []byte("put-in-batch-hash"), Address: []byte("put-in-batch-hash"),
Data: []byte("DATA"), Data: []byte("DATA"),
StoreTimestamp: time.Now().UTC().UnixNano(), StoreTimestamp: time.Now().UTC().UnixNano(),
@ -112,16 +112,16 @@ func TestIndex(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
got, err := index.Get(IndexItem{ got, err := index.Get(Item{
Address: want.Address, Address: want.Address,
}) })
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
checkIndexItem(t, got, want) checkItem(t, got, want)
t.Run("overwrite", func(t *testing.T) { t.Run("overwrite", func(t *testing.T) {
want := IndexItem{ want := Item{
Address: []byte("put-in-batch-hash"), Address: []byte("put-in-batch-hash"),
Data: []byte("New DATA"), Data: []byte("New DATA"),
StoreTimestamp: time.Now().UTC().UnixNano(), StoreTimestamp: time.Now().UTC().UnixNano(),
@ -133,13 +133,13 @@ func TestIndex(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
got, err := index.Get(IndexItem{ got, err := index.Get(Item{
Address: want.Address, Address: want.Address,
}) })
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
checkIndexItem(t, got, want) checkItem(t, got, want)
}) })
}) })
@ -150,13 +150,13 @@ func TestIndex(t *testing.T) {
address := []byte("put-in-batch-twice-hash") address := []byte("put-in-batch-twice-hash")
// put the first item // put the first item
index.PutInBatch(batch, IndexItem{ index.PutInBatch(batch, Item{
Address: address, Address: address,
Data: []byte("DATA"), Data: []byte("DATA"),
StoreTimestamp: time.Now().UTC().UnixNano(), StoreTimestamp: time.Now().UTC().UnixNano(),
}) })
want := IndexItem{ want := Item{
Address: address, Address: address,
Data: []byte("New DATA"), Data: []byte("New DATA"),
StoreTimestamp: time.Now().UTC().UnixNano(), StoreTimestamp: time.Now().UTC().UnixNano(),
@ -168,17 +168,17 @@ func TestIndex(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
got, err := index.Get(IndexItem{ got, err := index.Get(Item{
Address: address, Address: address,
}) })
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
checkIndexItem(t, got, want) checkItem(t, got, want)
}) })
t.Run("delete", func(t *testing.T) { t.Run("delete", func(t *testing.T) {
want := IndexItem{ want := Item{
Address: []byte("delete-hash"), Address: []byte("delete-hash"),
Data: []byte("DATA"), Data: []byte("DATA"),
StoreTimestamp: time.Now().UTC().UnixNano(), StoreTimestamp: time.Now().UTC().UnixNano(),
@ -188,15 +188,15 @@ func TestIndex(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
got, err := index.Get(IndexItem{ got, err := index.Get(Item{
Address: want.Address, Address: want.Address,
}) })
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
checkIndexItem(t, got, want) checkItem(t, got, want)
err = index.Delete(IndexItem{ err = index.Delete(Item{
Address: want.Address, Address: want.Address,
}) })
if err != nil { if err != nil {
@ -204,7 +204,7 @@ func TestIndex(t *testing.T) {
} }
wantErr := leveldb.ErrNotFound wantErr := leveldb.ErrNotFound
got, err = index.Get(IndexItem{ got, err = index.Get(Item{
Address: want.Address, Address: want.Address,
}) })
if err != wantErr { if err != wantErr {
@ -213,7 +213,7 @@ func TestIndex(t *testing.T) {
}) })
t.Run("delete in batch", func(t *testing.T) { t.Run("delete in batch", func(t *testing.T) {
want := IndexItem{ want := Item{
Address: []byte("delete-in-batch-hash"), Address: []byte("delete-in-batch-hash"),
Data: []byte("DATA"), Data: []byte("DATA"),
StoreTimestamp: time.Now().UTC().UnixNano(), StoreTimestamp: time.Now().UTC().UnixNano(),
@ -223,16 +223,16 @@ func TestIndex(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
got, err := index.Get(IndexItem{ got, err := index.Get(Item{
Address: want.Address, Address: want.Address,
}) })
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
checkIndexItem(t, got, want) checkItem(t, got, want)
batch := new(leveldb.Batch) batch := new(leveldb.Batch)
index.DeleteInBatch(batch, IndexItem{ index.DeleteInBatch(batch, Item{
Address: want.Address, Address: want.Address,
}) })
err = db.WriteBatch(batch) err = db.WriteBatch(batch)
@ -241,7 +241,7 @@ func TestIndex(t *testing.T) {
} }
wantErr := leveldb.ErrNotFound wantErr := leveldb.ErrNotFound
got, err = index.Get(IndexItem{ got, err = index.Get(Item{
Address: want.Address, Address: want.Address,
}) })
if err != wantErr { if err != wantErr {
@ -260,7 +260,7 @@ func TestIndex_iterate(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
items := []IndexItem{ items := []Item{
{ {
Address: []byte("iterate-hash-01"), Address: []byte("iterate-hash-01"),
Data: []byte("data80"), Data: []byte("data80"),
@ -290,7 +290,7 @@ func TestIndex_iterate(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
item04 := IndexItem{ item04 := Item{
Address: []byte("iterate-hash-04"), Address: []byte("iterate-hash-04"),
Data: []byte("data0"), Data: []byte("data0"),
} }
@ -306,12 +306,12 @@ func TestIndex_iterate(t *testing.T) {
t.Run("all", func(t *testing.T) { t.Run("all", func(t *testing.T) {
var i int var i int
err := index.IterateAll(func(item IndexItem) (stop bool, err error) { err := index.IterateAll(func(item Item) (stop bool, err error) {
if i > len(items)-1 { if i > len(items)-1 {
return true, fmt.Errorf("got unexpected index item: %#v", item) return true, fmt.Errorf("got unexpected index item: %#v", item)
} }
want := items[i] want := items[i]
checkIndexItem(t, item, want) checkItem(t, item, want)
i++ i++
return false, nil return false, nil
}) })
@ -323,12 +323,12 @@ func TestIndex_iterate(t *testing.T) {
t.Run("from", func(t *testing.T) { t.Run("from", func(t *testing.T) {
startIndex := 2 startIndex := 2
i := startIndex i := startIndex
err := index.IterateFrom(items[startIndex], func(item IndexItem) (stop bool, err error) { err := index.IterateFrom(items[startIndex], func(item Item) (stop bool, err error) {
if i > len(items)-1 { if i > len(items)-1 {
return true, fmt.Errorf("got unexpected index item: %#v", item) return true, fmt.Errorf("got unexpected index item: %#v", item)
} }
want := items[i] want := items[i]
checkIndexItem(t, item, want) checkItem(t, item, want)
i++ i++
return false, nil return false, nil
}) })
@ -341,12 +341,12 @@ func TestIndex_iterate(t *testing.T) {
var i int var i int
stopIndex := 3 stopIndex := 3
var count int var count int
err := index.IterateAll(func(item IndexItem) (stop bool, err error) { err := index.IterateAll(func(item Item) (stop bool, err error) {
if i > len(items)-1 { if i > len(items)-1 {
return true, fmt.Errorf("got unexpected index item: %#v", item) return true, fmt.Errorf("got unexpected index item: %#v", item)
} }
want := items[i] want := items[i]
checkIndexItem(t, item, want) checkItem(t, item, want)
count++ count++
if i == stopIndex { if i == stopIndex {
return true, nil return true, nil
@ -369,22 +369,22 @@ func TestIndex_iterate(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
secondIndexItem := IndexItem{ secondItem := Item{
Address: []byte("iterate-hash-10"), Address: []byte("iterate-hash-10"),
Data: []byte("data-second"), Data: []byte("data-second"),
} }
err = secondIndex.Put(secondIndexItem) err = secondIndex.Put(secondItem)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
var i int var i int
err = index.IterateAll(func(item IndexItem) (stop bool, err error) { err = index.IterateAll(func(item Item) (stop bool, err error) {
if i > len(items)-1 { if i > len(items)-1 {
return true, fmt.Errorf("got unexpected index item: %#v", item) return true, fmt.Errorf("got unexpected index item: %#v", item)
} }
want := items[i] want := items[i]
checkIndexItem(t, item, want) checkItem(t, item, want)
i++ i++
return false, nil return false, nil
}) })
@ -393,11 +393,11 @@ func TestIndex_iterate(t *testing.T) {
} }
i = 0 i = 0
err = secondIndex.IterateAll(func(item IndexItem) (stop bool, err error) { err = secondIndex.IterateAll(func(item Item) (stop bool, err error) {
if i > 1 { if i > 1 {
return true, fmt.Errorf("got unexpected index item: %#v", item) return true, fmt.Errorf("got unexpected index item: %#v", item)
} }
checkIndexItem(t, item, secondIndexItem) checkItem(t, item, secondItem)
i++ i++
return false, nil return false, nil
}) })
@ -418,7 +418,7 @@ func TestIndex_Count(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
items := []IndexItem{ items := []Item{
{ {
Address: []byte("iterate-hash-01"), Address: []byte("iterate-hash-01"),
Data: []byte("data80"), Data: []byte("data80"),
@ -461,7 +461,7 @@ func TestIndex_Count(t *testing.T) {
// update the index with another item // update the index with another item
item04 := IndexItem{ item04 := Item{
Address: []byte("iterate-hash-04"), Address: []byte("iterate-hash-04"),
Data: []byte("data0"), Data: []byte("data0"),
} }
@ -502,8 +502,8 @@ func TestIndex_Count(t *testing.T) {
} }
} }
// checkIndexItem is a test helper function that compares if two Index items are the same. // checkItem is a test helper function that compares if two Index items are the same.
func checkIndexItem(t *testing.T, got, want IndexItem) { func checkItem(t *testing.T, got, want Item) {
t.Helper() t.Helper()
if !bytes.Equal(got.Address, want.Address) { if !bytes.Equal(got.Address, want.Address) {

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

@ -53,7 +53,7 @@ func (db *DB) collectGarbage() {
// sets a gc trigger if batch limit is reached // sets a gc trigger if batch limit is reached
var triggerNextIteration bool var triggerNextIteration bool
var collectedCount int64 var collectedCount int64
err := db.gcIndex.IterateAll(func(item shed.IndexItem) (stop bool, err error) { err := db.gcIndex.IterateAll(func(item shed.Item) (stop bool, err error) {
gcSize := atomic.LoadInt64(&db.gcSize) gcSize := atomic.LoadInt64(&db.gcSize)
if gcSize-collectedCount <= target { if gcSize-collectedCount <= target {
return true, nil return true, nil

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

@ -137,9 +137,9 @@ func testDB_collectGarbage(t *testing.T, db *DB) {
t.Errorf("total collected chunks %v, want %v", totalCollectedCount, wantTotalCollectedCount) t.Errorf("total collected chunks %v, want %v", totalCollectedCount, wantTotalCollectedCount)
} }
t.Run("pull index count", newIndexItemsCountTest(db.pullIndex, int(gcTarget))) t.Run("pull index count", newItemsCountTest(db.pullIndex, int(gcTarget)))
t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, int(gcTarget))) t.Run("gc index count", newItemsCountTest(db.gcIndex, int(gcTarget)))
t.Run("gc size", newIndexGCSizeTest(db)) t.Run("gc size", newIndexGCSizeTest(db))
@ -259,9 +259,9 @@ func testDB_collectGarbage_withRequests(t *testing.T, db *DB) {
t.Errorf("total collected chunks %v, want %v", totalCollectedCount, wantTotalCollectedCount) t.Errorf("total collected chunks %v, want %v", totalCollectedCount, wantTotalCollectedCount)
} }
t.Run("pull index count", newIndexItemsCountTest(db.pullIndex, int(gcTarget))) t.Run("pull index count", newItemsCountTest(db.pullIndex, int(gcTarget)))
t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, int(gcTarget))) t.Run("gc index count", newItemsCountTest(db.gcIndex, int(gcTarget)))
t.Run("gc size", newIndexGCSizeTest(db)) t.Run("gc size", newIndexGCSizeTest(db))

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

@ -26,7 +26,7 @@ import (
// TestDB_pullIndex validates the ordering of keys in pull index. // TestDB_pullIndex validates the ordering of keys in pull index.
// Pull index key contains PO prefix which is calculated from // Pull index key contains PO prefix which is calculated from
// DB base key and chunk address. This is not an IndexItem field // DB base key and chunk address. This is not an Item field
// which are checked in Mode tests. // which are checked in Mode tests.
// This test uploads chunks, sorts them in expected order and // This test uploads chunks, sorts them in expected order and
// validates that pull index iterator will iterate it the same // validates that pull index iterator will iterate it the same
@ -61,7 +61,7 @@ func TestDB_pullIndex(t *testing.T) {
} }
} }
testIndexItemsOrder(t, db.pullIndex, chunks, func(i, j int) (less bool) { testItemsOrder(t, db.pullIndex, chunks, func(i, j int) (less bool) {
poi := storage.Proximity(db.baseKey, chunks[i].Address()) poi := storage.Proximity(db.baseKey, chunks[i].Address())
poj := storage.Proximity(db.baseKey, chunks[j].Address()) poj := storage.Proximity(db.baseKey, chunks[j].Address())
if poi < poj { if poi < poj {
@ -119,10 +119,10 @@ func testDB_gcIndex(t *testing.T, db *DB) {
} }
// check if all chunks are stored // check if all chunks are stored
newIndexItemsCountTest(db.pullIndex, chunkCount)(t) newItemsCountTest(db.pullIndex, chunkCount)(t)
// check that chunks are not collectable for garbage // check that chunks are not collectable for garbage
newIndexItemsCountTest(db.gcIndex, 0)(t) newItemsCountTest(db.gcIndex, 0)(t)
// set update gc test hook to signal when // set update gc test hook to signal when
// update gc goroutine is done by sending to // update gc goroutine is done by sending to
@ -145,7 +145,7 @@ func testDB_gcIndex(t *testing.T, db *DB) {
// the chunk is not synced // the chunk is not synced
// should not be in the garbace collection index // should not be in the garbace collection index
newIndexItemsCountTest(db.gcIndex, 0)(t) newItemsCountTest(db.gcIndex, 0)(t)
newIndexGCSizeTest(db)(t) newIndexGCSizeTest(db)(t)
}) })
@ -159,7 +159,7 @@ func testDB_gcIndex(t *testing.T, db *DB) {
} }
// the chunk is synced and should be in gc index // the chunk is synced and should be in gc index
newIndexItemsCountTest(db.gcIndex, 1)(t) newItemsCountTest(db.gcIndex, 1)(t)
newIndexGCSizeTest(db)(t) newIndexGCSizeTest(db)(t)
}) })
@ -174,7 +174,7 @@ func testDB_gcIndex(t *testing.T, db *DB) {
} }
} }
testIndexItemsOrder(t, db.gcIndex, chunks, nil) testItemsOrder(t, db.gcIndex, chunks, nil)
newIndexGCSizeTest(db)(t) newIndexGCSizeTest(db)(t)
}) })
@ -194,7 +194,7 @@ func testDB_gcIndex(t *testing.T, db *DB) {
chunks = append(chunks[:i], chunks[i+1:]...) chunks = append(chunks[:i], chunks[i+1:]...)
chunks = append(chunks, c) chunks = append(chunks, c)
testIndexItemsOrder(t, db.gcIndex, chunks, nil) testItemsOrder(t, db.gcIndex, chunks, nil)
newIndexGCSizeTest(db)(t) newIndexGCSizeTest(db)(t)
}) })
@ -215,7 +215,7 @@ func testDB_gcIndex(t *testing.T, db *DB) {
<-testHookUpdateGCChan <-testHookUpdateGCChan
} }
testIndexItemsOrder(t, db.gcIndex, chunks, nil) testItemsOrder(t, db.gcIndex, chunks, nil)
newIndexGCSizeTest(db)(t) newIndexGCSizeTest(db)(t)
}) })
@ -231,7 +231,7 @@ func testDB_gcIndex(t *testing.T, db *DB) {
// remove the chunk from the expected chunks in gc index // remove the chunk from the expected chunks in gc index
chunks = append(chunks[:i], chunks[i+1:]...) chunks = append(chunks[:i], chunks[i+1:]...)
testIndexItemsOrder(t, db.gcIndex, chunks, nil) testItemsOrder(t, db.gcIndex, chunks, nil)
newIndexGCSizeTest(db)(t) newIndexGCSizeTest(db)(t)
}) })

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

@ -151,11 +151,11 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
} }
if db.useRetrievalCompositeIndex { if db.useRetrievalCompositeIndex {
var ( var (
encodeValueFunc func(fields shed.IndexItem) (value []byte, err error) encodeValueFunc func(fields shed.Item) (value []byte, err error)
decodeValueFunc func(keyIndexItem shed.IndexItem, value []byte) (e shed.IndexItem, err error) decodeValueFunc func(keyItem shed.Item, value []byte) (e shed.Item, err error)
) )
if o.MockStore != nil { if o.MockStore != nil {
encodeValueFunc = func(fields shed.IndexItem) (value []byte, err error) { encodeValueFunc = func(fields shed.Item) (value []byte, err error) {
b := make([]byte, 16) b := make([]byte, 16)
binary.BigEndian.PutUint64(b[:8], uint64(fields.StoreTimestamp)) binary.BigEndian.PutUint64(b[:8], uint64(fields.StoreTimestamp))
binary.BigEndian.PutUint64(b[8:16], uint64(fields.AccessTimestamp)) binary.BigEndian.PutUint64(b[8:16], uint64(fields.AccessTimestamp))
@ -165,21 +165,21 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
} }
return b, nil return b, nil
} }
decodeValueFunc = func(keyIndexItem shed.IndexItem, value []byte) (e shed.IndexItem, err error) { decodeValueFunc = func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8])) e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
e.AccessTimestamp = int64(binary.BigEndian.Uint64(value[8:16])) e.AccessTimestamp = int64(binary.BigEndian.Uint64(value[8:16]))
e.Data, err = o.MockStore.Get(keyIndexItem.Address) e.Data, err = o.MockStore.Get(keyItem.Address)
return e, err return e, err
} }
} else { } else {
encodeValueFunc = func(fields shed.IndexItem) (value []byte, err error) { encodeValueFunc = func(fields shed.Item) (value []byte, err error) {
b := make([]byte, 16) b := make([]byte, 16)
binary.BigEndian.PutUint64(b[:8], uint64(fields.StoreTimestamp)) binary.BigEndian.PutUint64(b[:8], uint64(fields.StoreTimestamp))
binary.BigEndian.PutUint64(b[8:16], uint64(fields.AccessTimestamp)) binary.BigEndian.PutUint64(b[8:16], uint64(fields.AccessTimestamp))
value = append(b, fields.Data...) value = append(b, fields.Data...)
return value, nil return value, nil
} }
decodeValueFunc = func(keyIndexItem shed.IndexItem, value []byte) (e shed.IndexItem, err error) { decodeValueFunc = func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8])) e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
e.AccessTimestamp = int64(binary.BigEndian.Uint64(value[8:16])) e.AccessTimestamp = int64(binary.BigEndian.Uint64(value[8:16]))
e.Data = value[16:] e.Data = value[16:]
@ -188,10 +188,10 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
} }
// Index storing chunk data with stored and access timestamps. // Index storing chunk data with stored and access timestamps.
db.retrievalCompositeIndex, err = db.shed.NewIndex("Hash->StoredTimestamp|AccessTimestamp|Data", shed.IndexFuncs{ db.retrievalCompositeIndex, err = db.shed.NewIndex("Hash->StoredTimestamp|AccessTimestamp|Data", shed.IndexFuncs{
EncodeKey: func(fields shed.IndexItem) (key []byte, err error) { EncodeKey: func(fields shed.Item) (key []byte, err error) {
return fields.Address, nil return fields.Address, nil
}, },
DecodeKey: func(key []byte) (e shed.IndexItem, err error) { DecodeKey: func(key []byte) (e shed.Item, err error) {
e.Address = key e.Address = key
return e, nil return e, nil
}, },
@ -203,11 +203,11 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
} }
} else { } else {
var ( var (
encodeValueFunc func(fields shed.IndexItem) (value []byte, err error) encodeValueFunc func(fields shed.Item) (value []byte, err error)
decodeValueFunc func(keyIndexItem shed.IndexItem, value []byte) (e shed.IndexItem, err error) decodeValueFunc func(keyItem shed.Item, value []byte) (e shed.Item, err error)
) )
if o.MockStore != nil { if o.MockStore != nil {
encodeValueFunc = func(fields shed.IndexItem) (value []byte, err error) { encodeValueFunc = func(fields shed.Item) (value []byte, err error) {
b := make([]byte, 8) b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(fields.StoreTimestamp)) binary.BigEndian.PutUint64(b, uint64(fields.StoreTimestamp))
err = o.MockStore.Put(fields.Address, fields.Data) err = o.MockStore.Put(fields.Address, fields.Data)
@ -216,19 +216,19 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
} }
return b, nil return b, nil
} }
decodeValueFunc = func(keyIndexItem shed.IndexItem, value []byte) (e shed.IndexItem, err error) { decodeValueFunc = func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8])) e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
e.Data, err = o.MockStore.Get(keyIndexItem.Address) e.Data, err = o.MockStore.Get(keyItem.Address)
return e, err return e, err
} }
} else { } else {
encodeValueFunc = func(fields shed.IndexItem) (value []byte, err error) { encodeValueFunc = func(fields shed.Item) (value []byte, err error) {
b := make([]byte, 8) b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(fields.StoreTimestamp)) binary.BigEndian.PutUint64(b, uint64(fields.StoreTimestamp))
value = append(b, fields.Data...) value = append(b, fields.Data...)
return value, nil return value, nil
} }
decodeValueFunc = func(keyIndexItem shed.IndexItem, value []byte) (e shed.IndexItem, err error) { decodeValueFunc = func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8])) e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
e.Data = value[8:] e.Data = value[8:]
return e, nil return e, nil
@ -236,10 +236,10 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
} }
// Index storing actual chunk address, data and store timestamp. // Index storing actual chunk address, data and store timestamp.
db.retrievalDataIndex, err = db.shed.NewIndex("Address->StoreTimestamp|Data", shed.IndexFuncs{ db.retrievalDataIndex, err = db.shed.NewIndex("Address->StoreTimestamp|Data", shed.IndexFuncs{
EncodeKey: func(fields shed.IndexItem) (key []byte, err error) { EncodeKey: func(fields shed.Item) (key []byte, err error) {
return fields.Address, nil return fields.Address, nil
}, },
DecodeKey: func(key []byte) (e shed.IndexItem, err error) { DecodeKey: func(key []byte) (e shed.Item, err error) {
e.Address = key e.Address = key
return e, nil return e, nil
}, },
@ -252,19 +252,19 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
// Index storing access timestamp for a particular address. // Index storing access timestamp for a particular address.
// It is needed in order to update gc index keys for iteration order. // It is needed in order to update gc index keys for iteration order.
db.retrievalAccessIndex, err = db.shed.NewIndex("Address->AccessTimestamp", shed.IndexFuncs{ db.retrievalAccessIndex, err = db.shed.NewIndex("Address->AccessTimestamp", shed.IndexFuncs{
EncodeKey: func(fields shed.IndexItem) (key []byte, err error) { EncodeKey: func(fields shed.Item) (key []byte, err error) {
return fields.Address, nil return fields.Address, nil
}, },
DecodeKey: func(key []byte) (e shed.IndexItem, err error) { DecodeKey: func(key []byte) (e shed.Item, err error) {
e.Address = key e.Address = key
return e, nil return e, nil
}, },
EncodeValue: func(fields shed.IndexItem) (value []byte, err error) { EncodeValue: func(fields shed.Item) (value []byte, err error) {
b := make([]byte, 8) b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(fields.AccessTimestamp)) binary.BigEndian.PutUint64(b, uint64(fields.AccessTimestamp))
return b, nil return b, nil
}, },
DecodeValue: func(keyIndexItem shed.IndexItem, value []byte) (e shed.IndexItem, err error) { DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
e.AccessTimestamp = int64(binary.BigEndian.Uint64(value)) e.AccessTimestamp = int64(binary.BigEndian.Uint64(value))
return e, nil return e, nil
}, },
@ -275,22 +275,22 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
} }
// pull index allows history and live syncing per po bin // pull index allows history and live syncing per po bin
db.pullIndex, err = db.shed.NewIndex("PO|StoredTimestamp|Hash->nil", shed.IndexFuncs{ db.pullIndex, err = db.shed.NewIndex("PO|StoredTimestamp|Hash->nil", shed.IndexFuncs{
EncodeKey: func(fields shed.IndexItem) (key []byte, err error) { EncodeKey: func(fields shed.Item) (key []byte, err error) {
key = make([]byte, 41) key = make([]byte, 41)
key[0] = db.po(fields.Address) key[0] = db.po(fields.Address)
binary.BigEndian.PutUint64(key[1:9], uint64(fields.StoreTimestamp)) binary.BigEndian.PutUint64(key[1:9], uint64(fields.StoreTimestamp))
copy(key[9:], fields.Address[:]) copy(key[9:], fields.Address[:])
return key, nil return key, nil
}, },
DecodeKey: func(key []byte) (e shed.IndexItem, err error) { DecodeKey: func(key []byte) (e shed.Item, err error) {
e.Address = key[9:] e.Address = key[9:]
e.StoreTimestamp = int64(binary.BigEndian.Uint64(key[1:9])) e.StoreTimestamp = int64(binary.BigEndian.Uint64(key[1:9]))
return e, nil return e, nil
}, },
EncodeValue: func(fields shed.IndexItem) (value []byte, err error) { EncodeValue: func(fields shed.Item) (value []byte, err error) {
return nil, nil return nil, nil
}, },
DecodeValue: func(keyIndexItem shed.IndexItem, value []byte) (e shed.IndexItem, err error) { DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
return e, nil return e, nil
}, },
}) })
@ -299,21 +299,21 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
} }
// push index contains as yet unsynced chunks // push index contains as yet unsynced chunks
db.pushIndex, err = db.shed.NewIndex("StoredTimestamp|Hash->nil", shed.IndexFuncs{ db.pushIndex, err = db.shed.NewIndex("StoredTimestamp|Hash->nil", shed.IndexFuncs{
EncodeKey: func(fields shed.IndexItem) (key []byte, err error) { EncodeKey: func(fields shed.Item) (key []byte, err error) {
key = make([]byte, 40) key = make([]byte, 40)
binary.BigEndian.PutUint64(key[:8], uint64(fields.StoreTimestamp)) binary.BigEndian.PutUint64(key[:8], uint64(fields.StoreTimestamp))
copy(key[8:], fields.Address[:]) copy(key[8:], fields.Address[:])
return key, nil return key, nil
}, },
DecodeKey: func(key []byte) (e shed.IndexItem, err error) { DecodeKey: func(key []byte) (e shed.Item, err error) {
e.Address = key[8:] e.Address = key[8:]
e.StoreTimestamp = int64(binary.BigEndian.Uint64(key[:8])) e.StoreTimestamp = int64(binary.BigEndian.Uint64(key[:8]))
return e, nil return e, nil
}, },
EncodeValue: func(fields shed.IndexItem) (value []byte, err error) { EncodeValue: func(fields shed.Item) (value []byte, err error) {
return nil, nil return nil, nil
}, },
DecodeValue: func(keyIndexItem shed.IndexItem, value []byte) (e shed.IndexItem, err error) { DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
return e, nil return e, nil
}, },
}) })
@ -322,23 +322,23 @@ func New(path string, baseKey []byte, o *Options) (db *DB, err error) {
} }
// gc index for removable chunk ordered by ascending last access time // gc index for removable chunk ordered by ascending last access time
db.gcIndex, err = db.shed.NewIndex("AccessTimestamp|StoredTimestamp|Hash->nil", shed.IndexFuncs{ db.gcIndex, err = db.shed.NewIndex("AccessTimestamp|StoredTimestamp|Hash->nil", shed.IndexFuncs{
EncodeKey: func(fields shed.IndexItem) (key []byte, err error) { EncodeKey: func(fields shed.Item) (key []byte, err error) {
b := make([]byte, 16, 16+len(fields.Address)) b := make([]byte, 16, 16+len(fields.Address))
binary.BigEndian.PutUint64(b[:8], uint64(fields.AccessTimestamp)) binary.BigEndian.PutUint64(b[:8], uint64(fields.AccessTimestamp))
binary.BigEndian.PutUint64(b[8:16], uint64(fields.StoreTimestamp)) binary.BigEndian.PutUint64(b[8:16], uint64(fields.StoreTimestamp))
key = append(b, fields.Address...) key = append(b, fields.Address...)
return key, nil return key, nil
}, },
DecodeKey: func(key []byte) (e shed.IndexItem, err error) { DecodeKey: func(key []byte) (e shed.Item, err error) {
e.AccessTimestamp = int64(binary.BigEndian.Uint64(key[:8])) e.AccessTimestamp = int64(binary.BigEndian.Uint64(key[:8]))
e.StoreTimestamp = int64(binary.BigEndian.Uint64(key[8:16])) e.StoreTimestamp = int64(binary.BigEndian.Uint64(key[8:16]))
e.Address = key[16:] e.Address = key[16:]
return e, nil return e, nil
}, },
EncodeValue: func(fields shed.IndexItem) (value []byte, err error) { EncodeValue: func(fields shed.Item) (value []byte, err error) {
return nil, nil return nil, nil
}, },
DecodeValue: func(keyIndexItem shed.IndexItem, value []byte) (e shed.IndexItem, err error) { DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
return e, nil return e, nil
}, },
}) })
@ -397,17 +397,17 @@ func (db *DB) lockAddr(addr storage.Address) (unlock func(), err error) {
return func() { db.addressLocks.Delete(lockKey) }, nil return func() { db.addressLocks.Delete(lockKey) }, nil
} }
// chunkToItem creates new IndexItem with data provided by the Chunk. // chunkToItem creates new Item with data provided by the Chunk.
func chunkToItem(ch storage.Chunk) shed.IndexItem { func chunkToItem(ch storage.Chunk) shed.Item {
return shed.IndexItem{ return shed.Item{
Address: ch.Address(), Address: ch.Address(),
Data: ch.Data(), Data: ch.Data(),
} }
} }
// addressToItem creates new IndexItem with a provided address. // addressToItem creates new Item with a provided address.
func addressToItem(addr storage.Address) shed.IndexItem { func addressToItem(addr storage.Address) shed.Item {
return shed.IndexItem{ return shed.Item{
Address: addr, Address: addr,
} }
} }

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

@ -376,7 +376,7 @@ func newRetrieveIndexesTestWithAccess(db *DB, chunk storage.Chunk, storeTimestam
// chunk values are in the pull index. // chunk values are in the pull index.
func newPullIndexTest(db *DB, chunk storage.Chunk, storeTimestamp int64, wantError error) func(t *testing.T) { func newPullIndexTest(db *DB, chunk storage.Chunk, storeTimestamp int64, wantError error) func(t *testing.T) {
return func(t *testing.T) { return func(t *testing.T) {
item, err := db.pullIndex.Get(shed.IndexItem{ item, err := db.pullIndex.Get(shed.Item{
Address: chunk.Address(), Address: chunk.Address(),
StoreTimestamp: storeTimestamp, StoreTimestamp: storeTimestamp,
}) })
@ -393,7 +393,7 @@ func newPullIndexTest(db *DB, chunk storage.Chunk, storeTimestamp int64, wantErr
// chunk values are in the push index. // chunk values are in the push index.
func newPushIndexTest(db *DB, chunk storage.Chunk, storeTimestamp int64, wantError error) func(t *testing.T) { func newPushIndexTest(db *DB, chunk storage.Chunk, storeTimestamp int64, wantError error) func(t *testing.T) {
return func(t *testing.T) { return func(t *testing.T) {
item, err := db.pushIndex.Get(shed.IndexItem{ item, err := db.pushIndex.Get(shed.Item{
Address: chunk.Address(), Address: chunk.Address(),
StoreTimestamp: storeTimestamp, StoreTimestamp: storeTimestamp,
}) })
@ -410,7 +410,7 @@ func newPushIndexTest(db *DB, chunk storage.Chunk, storeTimestamp int64, wantErr
// chunk values are in the push index. // chunk values are in the push index.
func newGCIndexTest(db *DB, chunk storage.Chunk, storeTimestamp, accessTimestamp int64) func(t *testing.T) { func newGCIndexTest(db *DB, chunk storage.Chunk, storeTimestamp, accessTimestamp int64) func(t *testing.T) {
return func(t *testing.T) { return func(t *testing.T) {
item, err := db.gcIndex.Get(shed.IndexItem{ item, err := db.gcIndex.Get(shed.Item{
Address: chunk.Address(), Address: chunk.Address(),
StoreTimestamp: storeTimestamp, StoreTimestamp: storeTimestamp,
AccessTimestamp: accessTimestamp, AccessTimestamp: accessTimestamp,
@ -422,12 +422,12 @@ func newGCIndexTest(db *DB, chunk storage.Chunk, storeTimestamp, accessTimestamp
} }
} }
// newIndexItemsCountTest returns a test function that validates if // newItemsCountTest returns a test function that validates if
// an index contains expected number of key/value pairs. // an index contains expected number of key/value pairs.
func newIndexItemsCountTest(i shed.Index, want int) func(t *testing.T) { func newItemsCountTest(i shed.Index, want int) func(t *testing.T) {
return func(t *testing.T) { return func(t *testing.T) {
var c int var c int
i.IterateAll(func(item shed.IndexItem) (stop bool, err error) { i.IterateAll(func(item shed.Item) (stop bool, err error) {
c++ c++
return return
}) })
@ -442,7 +442,7 @@ func newIndexItemsCountTest(i shed.Index, want int) func(t *testing.T) {
func newIndexGCSizeTest(db *DB) func(t *testing.T) { func newIndexGCSizeTest(db *DB) func(t *testing.T) {
return func(t *testing.T) { return func(t *testing.T) {
var want int64 var want int64
db.gcIndex.IterateAll(func(item shed.IndexItem) (stop bool, err error) { db.gcIndex.IterateAll(func(item shed.Item) (stop bool, err error) {
want++ want++
return return
}) })
@ -460,17 +460,17 @@ type testIndexChunk struct {
storeTimestamp int64 storeTimestamp int64
} }
// testIndexItemsOrder tests the order of chunks in the index. If sortFunc is not nil, // testItemsOrder tests the order of chunks in the index. If sortFunc is not nil,
// chunks will be sorted with it before validation. // 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)) { func testItemsOrder(t *testing.T, i shed.Index, chunks []testIndexChunk, sortFunc func(i, j int) (less bool)) {
newIndexItemsCountTest(i, len(chunks))(t) newItemsCountTest(i, len(chunks))(t)
if sortFunc != nil { if sortFunc != nil {
sort.Slice(chunks, sortFunc) sort.Slice(chunks, sortFunc)
} }
var cursor int var cursor int
err := i.IterateAll(func(item shed.IndexItem) (stop bool, err error) { err := i.IterateAll(func(item shed.Item) (stop bool, err error) {
want := chunks[cursor].Address() want := chunks[cursor].Address()
got := item.Address got := item.Address
if !bytes.Equal(got, want) { if !bytes.Equal(got, want) {
@ -484,8 +484,8 @@ func testIndexItemsOrder(t *testing.T, i shed.Index, chunks []testIndexChunk, so
} }
} }
// validateItem is a helper function that checks IndexItem values. // validateItem is a helper function that checks Item values.
func validateItem(t *testing.T, item shed.IndexItem, address, data []byte, storeTimestamp, accessTimestamp int64) { func validateItem(t *testing.T, item shed.Item, address, data []byte, storeTimestamp, accessTimestamp int64) {
t.Helper() t.Helper()
if !bytes.Equal(item.Address, address) { if !bytes.Equal(item.Address, address) {

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

@ -65,9 +65,9 @@ func (g *Getter) Get(addr storage.Address) (chunk storage.Chunk, err error) {
return storage.NewChunk(out.Address, out.Data), nil return storage.NewChunk(out.Address, out.Data), nil
} }
// get returns IndexItem with from the retrieval index // get returns Item with from the retrieval index
// and updates other indexes. // and updates other indexes.
func (db *DB) get(mode ModeGet, addr storage.Address) (out shed.IndexItem, err error) { func (db *DB) get(mode ModeGet, addr storage.Address) (out shed.Item, err error) {
item := addressToItem(addr) item := addressToItem(addr)
if db.useRetrievalCompositeIndex { if db.useRetrievalCompositeIndex {
@ -120,7 +120,7 @@ func (db *DB) get(mode ModeGet, addr storage.Address) (out shed.IndexItem, err e
// a single item. Provided item is expected to have // a single item. Provided item is expected to have
// only Address and Data fields with non zero values, // only Address and Data fields with non zero values,
// which is ensured by the get function. // which is ensured by the get function.
func (db *DB) updateGC(item shed.IndexItem) (err error) { func (db *DB) updateGC(item shed.Item) (err error) {
unlock, err := db.lockAddr(item.Address) unlock, err := db.lockAddr(item.Address)
if err != nil { if err != nil {
return err return err

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

@ -85,7 +85,7 @@ func testModeGetRequestValues(t *testing.T, db *DB) {
t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, uploadTimestamp, 0)) t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, uploadTimestamp, 0))
t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 0)) t.Run("gc index count", newItemsCountTest(db.gcIndex, 0))
t.Run("gc size", newIndexGCSizeTest(db)) t.Run("gc size", newIndexGCSizeTest(db))
}) })
@ -116,7 +116,7 @@ func testModeGetRequestValues(t *testing.T, db *DB) {
t.Run("gc index", newGCIndexTest(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 index count", newItemsCountTest(db.gcIndex, 1))
t.Run("gc size", newIndexGCSizeTest(db)) t.Run("gc size", newIndexGCSizeTest(db))
}) })
@ -146,7 +146,7 @@ func testModeGetRequestValues(t *testing.T, db *DB) {
t.Run("gc index", newGCIndexTest(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 index count", newItemsCountTest(db.gcIndex, 1))
t.Run("gc size", newIndexGCSizeTest(db)) t.Run("gc size", newIndexGCSizeTest(db))
}) })
@ -200,7 +200,7 @@ func testModeGetSyncValues(t *testing.T, db *DB) {
t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, uploadTimestamp, 0)) t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, uploadTimestamp, 0))
t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 0)) t.Run("gc index count", newItemsCountTest(db.gcIndex, 0))
t.Run("gc size", newIndexGCSizeTest(db)) t.Run("gc size", newIndexGCSizeTest(db))
} }

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

@ -57,12 +57,12 @@ func (p *Putter) Put(ch storage.Chunk) (err error) {
return p.db.put(p.mode, chunkToItem(ch)) return p.db.put(p.mode, chunkToItem(ch))
} }
// put stores IndexItem to database and updates other // put stores Item to database and updates other
// indexes. It acquires lockAddr to protect two calls // indexes. It acquires lockAddr to protect two calls
// of this function for the same address in parallel. // of this function for the same address in parallel.
// IndexItem fields Address and Data must not be // Item fields Address and Data must not be
// with their nil values. // with their nil values.
func (db *DB) put(mode ModePut, item shed.IndexItem) (err error) { func (db *DB) put(mode ModePut, item shed.Item) (err error) {
// protect parallel updates // protect parallel updates
unlock, err := db.lockAddr(item.Address) unlock, err := db.lockAddr(item.Address)
if err != nil { if err != nil {

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

@ -64,7 +64,7 @@ func testModePutRequestValues(t *testing.T, db *DB) {
t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, wantTimestamp, wantTimestamp)) t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, wantTimestamp, wantTimestamp))
t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 1)) t.Run("gc index count", newItemsCountTest(db.gcIndex, 1))
t.Run("gc size", newIndexGCSizeTest(db)) t.Run("gc size", newIndexGCSizeTest(db))
}) })
@ -82,7 +82,7 @@ func testModePutRequestValues(t *testing.T, db *DB) {
t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, storeTimestamp, wantTimestamp)) t.Run("retrieve indexes", newRetrieveIndexesTestWithAccess(db, chunk, storeTimestamp, wantTimestamp))
t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 1)) t.Run("gc index count", newItemsCountTest(db.gcIndex, 1))
t.Run("gc size", newIndexGCSizeTest(db)) t.Run("gc size", newIndexGCSizeTest(db))
}) })

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

@ -58,11 +58,11 @@ func testModeSetAccessValues(t *testing.T, db *DB) {
t.Run("pull index", newPullIndexTest(db, chunk, wantTimestamp, nil)) t.Run("pull index", newPullIndexTest(db, chunk, wantTimestamp, nil))
t.Run("pull index count", newIndexItemsCountTest(db.pullIndex, 1)) t.Run("pull index count", newItemsCountTest(db.pullIndex, 1))
t.Run("gc index", newGCIndexTest(db, chunk, wantTimestamp, wantTimestamp)) t.Run("gc index", newGCIndexTest(db, chunk, wantTimestamp, wantTimestamp))
t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 1)) t.Run("gc index count", newItemsCountTest(db.gcIndex, 1))
t.Run("gc size", newIndexGCSizeTest(db)) t.Run("gc size", newIndexGCSizeTest(db))
} }
@ -111,7 +111,7 @@ func testModeSetSyncValues(t *testing.T, db *DB) {
t.Run("gc index", newGCIndexTest(db, chunk, wantTimestamp, wantTimestamp)) t.Run("gc index", newGCIndexTest(db, chunk, wantTimestamp, wantTimestamp))
t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 1)) t.Run("gc index count", newItemsCountTest(db.gcIndex, 1))
t.Run("gc size", newIndexGCSizeTest(db)) t.Run("gc size", newIndexGCSizeTest(db))
} }
@ -156,28 +156,28 @@ func testModeSetRemovalValues(t *testing.T, db *DB) {
if err != wantErr { if err != wantErr {
t.Errorf("got error %v, want %v", err, wantErr) t.Errorf("got error %v, want %v", err, wantErr)
} }
t.Run("retrieve index count", newIndexItemsCountTest(db.retrievalCompositeIndex, 0)) t.Run("retrieve index count", newItemsCountTest(db.retrievalCompositeIndex, 0))
} else { } else {
_, err := db.retrievalDataIndex.Get(addressToItem(chunk.Address())) _, err := db.retrievalDataIndex.Get(addressToItem(chunk.Address()))
if err != wantErr { if err != wantErr {
t.Errorf("got error %v, want %v", err, wantErr) t.Errorf("got error %v, want %v", err, wantErr)
} }
t.Run("retrieve data index count", newIndexItemsCountTest(db.retrievalDataIndex, 0)) t.Run("retrieve data index count", newItemsCountTest(db.retrievalDataIndex, 0))
// access index should not be set // access index should not be set
_, err = db.retrievalAccessIndex.Get(addressToItem(chunk.Address())) _, err = db.retrievalAccessIndex.Get(addressToItem(chunk.Address()))
if err != wantErr { if err != wantErr {
t.Errorf("got error %v, want %v", err, wantErr) t.Errorf("got error %v, want %v", err, wantErr)
} }
t.Run("retrieve access index count", newIndexItemsCountTest(db.retrievalAccessIndex, 0)) t.Run("retrieve access index count", newItemsCountTest(db.retrievalAccessIndex, 0))
} }
}) })
t.Run("pull index", newPullIndexTest(db, chunk, 0, leveldb.ErrNotFound)) t.Run("pull index", newPullIndexTest(db, chunk, 0, leveldb.ErrNotFound))
t.Run("pull index count", newIndexItemsCountTest(db.pullIndex, 0)) t.Run("pull index count", newItemsCountTest(db.pullIndex, 0))
t.Run("gc index count", newIndexItemsCountTest(db.gcIndex, 0)) t.Run("gc index count", newItemsCountTest(db.gcIndex, 0))
t.Run("gc size", newIndexGCSizeTest(db)) t.Run("gc size", newIndexGCSizeTest(db))