forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

swarm/storage/mock: implement listings methods for mem and rpc stores

Browse source
This commit is contained in:
Janos Guljas 2019-02-07 15:26:14 +01:00
parent d212535ddd
commit b1d414ea41
7 changed files with 457 additions and 54 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

64
swarm/storage/mock/db/db.go
View file

@ -21,6 +21,7 @@ import (
"archive/tar"
"bytes"
"encoding/json"
"errors"
"io"
"io/ioutil"
@ -64,7 +65,7 @@ func (s *GlobalStore) NewNodeStore(addr common.Address) *mock.NodeStore {
// Get returns chunk data if the chunk with key exists for node
// on address addr.
func (s *GlobalStore) Get(addr common.Address, key []byte) (data []byte, err error) {
has, err := s.db.Has(nodeDBKey(addr, key), nil)
has, err := s.db.Has(dbKeyNodes(addr, key), nil)
if err != nil {
return nil, mock.ErrNotFound
}
@ -81,7 +82,8 @@ func (s *GlobalStore) Get(addr common.Address, key []byte) (data []byte, err err
// Put saves the chunk data for node with address addr.
func (s *GlobalStore) Put(addr common.Address, key []byte, data []byte) error {
batch := new(leveldb.Batch)
batch.Put(nodeDBKey(addr, key), nil)
batch.Put(dbKeyNodes(addr, key), nil)
batch.Put(dbKeyKeys(addr, key), nil)
batch.Put(dataDBKey(key), data)
return s.db.Write(batch, nil)
}
@ -89,19 +91,40 @@ func (s *GlobalStore) Put(addr common.Address, key []byte, data []byte) error {
// Delete removes the chunk reference to node with address addr.
func (s *GlobalStore) Delete(addr common.Address, key []byte) error {
batch := new(leveldb.Batch)
batch.Delete(nodeDBKey(addr, key))
batch.Delete(dbKeyNodes(addr, key))
batch.Delete(dbKeyKeys(addr, key))
return s.db.Write(batch, nil)
}
// HasKey returns whether a node with addr contains the key.
func (s *GlobalStore) HasKey(addr common.Address, key []byte) bool {
has, err := s.db.Has(nodeDBKey(addr, key), nil)
has, err := s.db.Has(dbKeyNodes(addr, key), nil)
if err != nil {
has = false
}
return has
}
func (s *GlobalStore) Keys(startKey []byte, limit int) (keys mock.Keys, err error) {
// TODO: implement
return keys, errors.New("not implemented")
}
func (s *GlobalStore) Nodes(startAddr *common.Address, limit int) (nodes mock.Nodes, err error) {
// TODO: implement
return nodes, errors.New("not implemented")
}
func (s *GlobalStore) NodeKeys(addr common.Address, startKey []byte, limit int) (keys mock.Keys, err error) {
// TODO: implement
return keys, errors.New("not implemented")
}
func (s *GlobalStore) KeyNodes(key []byte, startAddr *common.Address, limit int) (nodes mock.Nodes, err error) {
// TODO: implement
return nodes, errors.New("not implemented")
}
// Import reads tar archive from a reader that contains exported chunk data.
// It returns the number of chunks imported and an error.
func (s *GlobalStore) Import(r io.Reader) (n int, err error) {
@ -128,7 +151,7 @@ func (s *GlobalStore) Import(r io.Reader) (n int, err error) {
batch := new(leveldb.Batch)
for _, addr := range c.Addrs {
batch.Put(nodeDBKeyHex(addr, hdr.Name), nil)
batch.Put(dbKeyNodesHex(addr, hdr.Name), nil)
}
batch.Put(dataDBKey(common.Hex2Bytes(hdr.Name)), c.Data)
@ -150,7 +173,7 @@ func (s *GlobalStore) Export(w io.Writer) (n int, err error) {
buf := bytes.NewBuffer(make([]byte, 0, 1024))
encoder := json.NewEncoder(buf)
iter := s.db.NewIterator(util.BytesPrefix(nodeKeyPrefix), nil)
iter := s.db.NewIterator(util.BytesPrefix(nodesPrefix), nil)
defer iter.Release()
var currentKey string
@ -189,7 +212,7 @@ func (s *GlobalStore) Export(w io.Writer) (n int, err error) {
}
for iter.Next() {
k := bytes.TrimPrefix(iter.Key(), nodeKeyPrefix)
k := bytes.TrimPrefix(iter.Key(), nodesPrefix)
i := bytes.Index(k, []byte("-"))
if i < 0 {
continue
@ -222,22 +245,31 @@ func (s *GlobalStore) Export(w io.Writer) (n int, err error) {
}
var (
nodeKeyPrefix = []byte("node-")
dataKeyPrefix = []byte("data-")
nodesPrefix = []byte("nodes-")
keysPrefix = []byte("keys-")
dataPrefix = []byte("data-")
)
// nodeDBKey constructs a database key for key/node mappings.
func nodeDBKey(addr common.Address, key []byte) []byte {
return nodeDBKeyHex(addr, common.Bytes2Hex(key))
// dbKeyNodes constructs a database key for key/node mappings.
func dbKeyNodes(addr common.Address, key []byte) []byte {
return dbKeyNodesHex(addr, common.Bytes2Hex(key))
}
// nodeDBKeyHex constructs a database key for key/node mappings
// dbKeyNodesHex constructs a database key for key/node mappings
// using the hexadecimal string representation of the key.
func nodeDBKeyHex(addr common.Address, hexKey string) []byte {
return append(append(nodeKeyPrefix, []byte(hexKey+"-")...), addr[:]...)
func dbKeyNodesHex(addr common.Address, hexKey string) []byte {
return append(append(nodesPrefix, []byte(hexKey+"-")...), addr[:]...)
}
func dbKeyKeys(addr common.Address, key []byte) []byte {
return dbKeyKeysHex(addr, common.Bytes2Hex(key))
}
func dbKeyKeysHex(addr common.Address, hexKey string) []byte {
return append(append(keysPrefix, []byte(addr.Hex()+"-")...), hexKey[:]...)
}
// dataDBkey constructs a database key for key/data storage.
func dataDBKey(key []byte) []byte {
return append(dataKeyPrefix, key...)
return append(dataPrefix, key...)
}

278
swarm/storage/mock/mem/mem.go
View file

@ -23,8 +23,10 @@ import (
"archive/tar"
"bytes"
"encoding/json"
"errors"
"io"
"io/ioutil"
"sort"
"sync"
"github.com/ethereum/go-ethereum/common"
@ -34,16 +36,27 @@ import (
// GlobalStore stores all chunk data and also keys and node addresses relations.
// It implements mock.GlobalStore interface.
type GlobalStore struct {
nodes map[string]map[common.Address]struct{}
data map[string][]byte
mu sync.Mutex
// holds a slice of keys per node
nodeKeys map[common.Address][][]byte
// holds which key is stored on which nodes
keyNodes map[string][]common.Address
// all node addresses
nodes []common.Address
// all keys
keys [][]byte
// all keys data
data map[string][]byte
mu sync.RWMutex
}
// NewGlobalStore creates a new instance of GlobalStore.
func NewGlobalStore() *GlobalStore {
return &GlobalStore{
nodes: make(map[string]map[common.Address]struct{}),
data: make(map[string][]byte),
nodeKeys: make(map[common.Address][][]byte),
keyNodes: make(map[string][]common.Address),
nodes: make([]common.Address, 0),
keys: make([][]byte, 0),
data: make(map[string][]byte),
}
}
@ -56,10 +69,10 @@ func (s *GlobalStore) NewNodeStore(addr common.Address) *mock.NodeStore {
// Get returns chunk data if the chunk with key exists for node
// on address addr.
func (s *GlobalStore) Get(addr common.Address, key []byte) (data []byte, err error) {
s.mu.Lock()
defer s.mu.Unlock()
s.mu.RLock()
defer s.mu.RUnlock()
if _, ok := s.nodes[string(key)][addr]; !ok {
if _, has := s.nodeKeyIndex(addr, key); !has {
return nil, mock.ErrNotFound
}
@ -75,11 +88,33 @@ func (s *GlobalStore) Put(addr common.Address, key []byte, data []byte) error {
s.mu.Lock()
defer s.mu.Unlock()
if _, ok := s.nodes[string(key)]; !ok {
s.nodes[string(key)] = make(map[common.Address]struct{})
if i, found := s.nodeKeyIndex(addr, key); !found {
s.nodeKeys[addr] = append(s.nodeKeys[addr], nil)
copy(s.nodeKeys[addr][i+1:], s.nodeKeys[addr][i:])
s.nodeKeys[addr][i] = key
}
s.nodes[string(key)][addr] = struct{}{}
if i, found := s.keyNodeIndex(key, addr); !found {
k := string(key)
s.keyNodes[k] = append(s.keyNodes[k], addr)
copy(s.keyNodes[k][i+1:], s.keyNodes[k][i:])
s.keyNodes[k][i] = addr
}
if i, found := s.nodeIndex(addr); !found {
s.nodes = append(s.nodes, addr)
copy(s.nodes[i+1:], s.nodes[i:])
s.nodes[i] = addr
}
if i, found := s.keyIndex(key); !found {
s.keys = append(s.keys, nil)
copy(s.keys[i+1:], s.keys[i:])
s.keys[i] = key
}
s.data[string(key)] = data
return nil
}
@ -88,24 +123,184 @@ func (s *GlobalStore) Delete(addr common.Address, key []byte) error {
s.mu.Lock()
defer s.mu.Unlock()
var count int
if _, ok := s.nodes[string(key)]; ok {
delete(s.nodes[string(key)], addr)
count = len(s.nodes[string(key)])
if i, has := s.nodeKeyIndex(addr, key); has {
s.nodeKeys[addr] = append(s.nodeKeys[addr][:i], s.nodeKeys[addr][i+1:]...)
}
if count == 0 {
delete(s.data, string(key))
k := string(key)
if i, on := s.keyNodeIndex(key, addr); on {
s.keyNodes[k] = append(s.keyNodes[k][:i], s.keyNodes[k][i+1:]...)
}
if len(s.nodeKeys[addr]) == 0 {
if i, found := s.nodeIndex(addr); found {
s.nodes = append(s.nodes[:i], s.nodes[i+1:]...)
}
}
if len(s.keyNodes[k]) == 0 {
if i, found := s.keyIndex(key); found {
s.keys = append(s.keys[:i], s.keys[i+1:]...)
}
}
return nil
}
// HasKey returns whether a node with addr contains the key.
func (s *GlobalStore) HasKey(addr common.Address, key []byte) bool {
s.mu.Lock()
defer s.mu.Unlock()
func (s *GlobalStore) HasKey(addr common.Address, key []byte) (yes bool) {
s.mu.RLock()
defer s.mu.RUnlock()
_, ok := s.nodes[string(key)][addr]
return ok
_, yes = s.nodeKeyIndex(addr, key)
return yes
}
func (s *GlobalStore) keyIndex(key []byte) (index int, found bool) {
l := len(s.keys)
index = sort.Search(l, func(i int) bool {
return bytes.Compare(s.keys[i], key) >= 0
})
found = index < l && bytes.Equal(s.keys[index], key)
return index, found
}
func (s *GlobalStore) nodeIndex(addr common.Address) (index int, found bool) {
l := len(s.nodes)
index = sort.Search(l, func(i int) bool {
return bytes.Compare(s.nodes[i][:], addr[:]) >= 0
})
found = index < l && bytes.Equal(s.nodes[index][:], addr[:])
return index, found
}
func (s *GlobalStore) nodeKeyIndex(addr common.Address, key []byte) (index int, found bool) {
l := len(s.nodeKeys[addr])
index = sort.Search(l, func(i int) bool {
return bytes.Compare(s.nodeKeys[addr][i], key) >= 0
})
found = index < l && bytes.Equal(s.nodeKeys[addr][index], key)
return index, found
}
func (s *GlobalStore) keyNodeIndex(key []byte, addr common.Address) (index int, found bool) {
k := string(key)
l := len(s.keyNodes[k])
index = sort.Search(l, func(i int) bool {
return bytes.Compare(s.keyNodes[k][i][:], addr[:]) >= 0
})
found = index < l && s.keyNodes[k][index] == addr
return index, found
}
func (s *GlobalStore) Keys(startKey []byte, limit int) (keys mock.Keys, err error) {
s.mu.RLock()
defer s.mu.RUnlock()
var i int
if startKey != nil {
var found bool
i, found = s.keyIndex(startKey)
if !found {
return keys, errors.New("start key not found")
}
}
total := len(s.keys)
max := maxIndex(i, limit, total)
keys.Keys = make([][]byte, 0, max-i)
for ; i < max; i++ {
keys.Keys = append(keys.Keys, append([]byte(nil), s.keys[i]...))
}
if total > max {
keys.Next = s.keys[max]
}
return keys, nil
}
func (s *GlobalStore) Nodes(startAddr *common.Address, limit int) (nodes mock.Nodes, err error) {
s.mu.RLock()
defer s.mu.RUnlock()
var i int
if startAddr != nil {
var found bool
i, found = s.nodeIndex(*startAddr)
if !found {
return nodes, errors.New("start address not found")
}
}
total := len(s.nodes)
max := maxIndex(i, limit, total)
nodes.Addrs = make([]common.Address, 0, max-i)
for ; i < max; i++ {
nodes.Addrs = append(nodes.Addrs, s.nodes[i])
}
if total > max {
nodes.Next = &s.nodes[max]
}
return nodes, nil
}
func (s *GlobalStore) NodeKeys(addr common.Address, startKey []byte, limit int) (keys mock.Keys, err error) {
s.mu.RLock()
defer s.mu.RUnlock()
var i int
if startKey != nil {
var found bool
i, found = s.nodeKeyIndex(addr, startKey)
if !found {
return keys, errors.New("start key not found")
}
}
total := len(s.nodeKeys[addr])
max := maxIndex(i, limit, total)
keys.Keys = make([][]byte, 0, max-i)
for ; i < max; i++ {
keys.Keys = append(keys.Keys, append([]byte(nil), s.nodeKeys[addr][i]...))
}
if total > max {
keys.Next = s.nodeKeys[addr][max]
}
return keys, nil
}
func (s *GlobalStore) KeyNodes(key []byte, startAddr *common.Address, limit int) (nodes mock.Nodes, err error) {
s.mu.RLock()
defer s.mu.RUnlock()
var i int
if startAddr != nil {
var found bool
i, found = s.keyNodeIndex(key, *startAddr)
if !found {
return nodes, errors.New("start address not found")
}
}
total := len(s.keyNodes[string(key)])
max := maxIndex(i, limit, total)
nodes.Addrs = make([]common.Address, 0, max-i)
for ; i < max; i++ {
nodes.Addrs = append(nodes.Addrs, s.keyNodes[string(key)][i])
}
if total > max {
nodes.Next = &s.keyNodes[string(key)][max]
}
return nodes, nil
}
func maxIndex(start, limit, total int) (max int) {
if limit <= 0 {
limit = mock.DefaultLimit
}
if limit > mock.MaxLimit {
limit = mock.MaxLimit
}
max = total
if start+limit < max {
max = start + limit
}
return max
}
// Import reads tar archive from a reader that contains exported chunk data.
@ -135,14 +330,26 @@ func (s *GlobalStore) Import(r io.Reader) (n int, err error) {
return n, err
}
addrs := make(map[common.Address]struct{})
for _, a := range c.Addrs {
addrs[a] = struct{}{}
key := common.Hex2Bytes(hdr.Name)
s.keyNodes[string(key)] = c.Addrs
for _, addr := range c.Addrs {
if i, has := s.nodeKeyIndex(addr, key); !has {
s.nodeKeys[addr] = append(s.nodeKeys[addr], nil)
copy(s.nodeKeys[addr][i+1:], s.nodeKeys[addr][i:])
s.nodeKeys[addr][i] = key
}
if i, found := s.nodeIndex(addr); !found {
s.nodes = append(s.nodes, addr)
copy(s.nodes[i+1:], s.nodes[i:])
s.nodes[i] = addr
}
}
key := string(common.Hex2Bytes(hdr.Name))
s.nodes[key] = addrs
s.data[key] = c.Data
if i, found := s.keyIndex(key); !found {
s.keys = append(s.keys, nil)
copy(s.keys[i+1:], s.keys[i:])
s.keys[i] = key
}
s.data[string(key)] = c.Data
n++
}
return n, err
@ -151,23 +358,18 @@ func (s *GlobalStore) Import(r io.Reader) (n int, err error) {
// Export writes to a writer a tar archive with all chunk data from
// the store. It returns the number of chunks exported and an error.
func (s *GlobalStore) Export(w io.Writer) (n int, err error) {
s.mu.Lock()
defer s.mu.Unlock()
s.mu.RLock()
defer s.mu.RUnlock()
tw := tar.NewWriter(w)
defer tw.Close()
buf := bytes.NewBuffer(make([]byte, 0, 1024))
encoder := json.NewEncoder(buf)
for key, addrs := range s.nodes {
al := make([]common.Address, 0, len(addrs))
for a := range addrs {
al = append(al, a)
}
for key, addrs := range s.keyNodes {
buf.Reset()
if err = encoder.Encode(mock.ExportedChunk{
Addrs: al,
Addrs: addrs,
Data: s.data[key],
}); err != nil {
return n, err

4
swarm/storage/mock/mem/mem_test.go
View file

@ -28,6 +28,10 @@ func TestGlobalStore(t *testing.T) {
test.MockStore(t, NewGlobalStore(), 100)
}
func TestGlobalStoreListings(t *testing.T) {
test.MockStoreListings(t, NewGlobalStore(), 1000)
}
// TestImportExport is running tests for importing and
// exporting data between two GlobalStores
// using test.ImportExport function.

23
swarm/storage/mock/mock.go
View file

@ -39,6 +39,11 @@ import (
"github.com/ethereum/go-ethereum/common"
)
const (
DefaultLimit = 100
MaxLimit = 1000
)
// ErrNotFound indicates that the chunk is not found.
var ErrNotFound = errors.New("not found")
@ -76,6 +81,10 @@ func (n *NodeStore) Delete(key []byte) error {
return n.store.Delete(n.addr, key)
}
func (n *NodeStore) Keys(startKey []byte, limit int) (keys Keys, err error) {
return n.store.NodeKeys(n.addr, startKey, limit)
}
// GlobalStorer defines methods for mock db store
// that stores chunk data for all swarm nodes.
// It is used in tests to construct mock NodeStores
@ -85,12 +94,26 @@ type GlobalStorer interface {
Put(addr common.Address, key []byte, data []byte) error
Delete(addr common.Address, key []byte) error
HasKey(addr common.Address, key []byte) bool
Keys(startKey []byte, limit int) (keys Keys, err error)
Nodes(startAddr *common.Address, limit int) (nodes Nodes, err error)
NodeKeys(addr common.Address, startKey []byte, limit int) (keys Keys, err error)
KeyNodes(key []byte, startAddr *common.Address, limit int) (nodes Nodes, err error)
// NewNodeStore creates an instance of NodeStore
// to be used by a single swarm node with
// address addr.
NewNodeStore(addr common.Address) *NodeStore
}
type Keys struct {
Keys [][]byte
Next []byte
}
type Nodes struct {
Addrs []common.Address
Next *common.Address
}
// Importer defines method for importing mock store data
// from an exported tar archive.
type Importer interface {

20
swarm/storage/mock/rpc/rpc.go
View file

@ -88,3 +88,23 @@ func (s *GlobalStore) HasKey(addr common.Address, key []byte) bool {
}
return has
}
func (s *GlobalStore) Keys(startKey []byte, limit int) (keys mock.Keys, err error) {
err = s.client.Call(&keys, "mockStore_keys", startKey, limit)
return keys, err
}
func (s *GlobalStore) Nodes(startAddr *common.Address, limit int) (nodes mock.Nodes, err error) {
err = s.client.Call(&nodes, "mockStore_nodes", startAddr, limit)
return nodes, err
}
func (s *GlobalStore) NodeKeys(addr common.Address, startKey []byte, limit int) (keys mock.Keys, err error) {
err = s.client.Call(&keys, "mockStore_nodeKeys", addr, startKey, limit)
return keys, err
}
func (s *GlobalStore) KeyNodes(key []byte, startAddr *common.Address, limit int) (nodes mock.Nodes, err error) {
err = s.client.Call(&nodes, "mockStore_keyNodes", key, startAddr, limit)
return nodes, err
}

14
swarm/storage/mock/rpc/rpc_test.go
View file

@ -39,3 +39,17 @@ func TestRPCStore(t *testing.T) {
test.MockStore(t, store, 30)
}
func TestGlobalStoreListings(t *testing.T) {
serverStore := mem.NewGlobalStore()
server := rpc.NewServer()
if err := server.RegisterName("mockStore", serverStore); err != nil {
t.Fatal(err)
}
store := NewGlobalStore(rpc.DialInProc(server))
defer store.Close()
test.MockStoreListings(t, store, 1000)
}

108
swarm/storage/mock/test/test.go
View file

@ -20,6 +20,7 @@ package test
import (
"bytes"
"encoding/binary"
"fmt"
"io"
"strconv"
@ -170,6 +171,113 @@ func MockStore(t *testing.T, globalStore mock.GlobalStorer, n int) {
})
}
func MockStoreListings(t *testing.T, globalStore mock.GlobalStorer, n int) {
addrs := make([]common.Address, n)
for i := 0; i < n; i++ {
addrs[i] = common.HexToAddress(strconv.FormatInt(int64(i)+1, 16))
}
type chunk struct {
key []byte
data []byte
}
const chunksPerNode = 5
keys := make([][]byte, n*chunksPerNode)
for i := 0; i < n*chunksPerNode; i++ {
b := make([]byte, 8)
binary.BigEndian.PutUint64(b, uint64(i))
keys[i] = b
}
nodeKeys := make(map[common.Address][][]byte)
keyNodes := make(map[string][]common.Address)
for i := 0; i < chunksPerNode; i++ {
for j := 0; j < n; j++ {
addr := addrs[j]
key := keys[(i*n)+j]
err := globalStore.Put(addr, key, []byte("data"))
if err != nil {
t.Fatal(err)
}
nodeKeys[addr] = append(nodeKeys[addr], key)
keyNodes[string(key)] = append(keyNodes[string(key)], addr)
}
var startKey []byte
var gotKeys [][]byte
for {
keys, err := globalStore.Keys(startKey, 0)
if err != nil {
t.Fatal(err)
}
gotKeys = append(gotKeys, keys.Keys...)
if keys.Next == nil {
break
}
startKey = keys.Next
}
wantKeys := keys[:(i+1)*n]
if fmt.Sprint(gotKeys) != fmt.Sprint(wantKeys) {
t.Fatalf("got #%v keys %v, want %v", i+1, gotKeys, wantKeys)
}
var startNode *common.Address
var gotNodes []common.Address
for {
nodes, err := globalStore.Nodes(startNode, 0)
if err != nil {
t.Fatal(err)
}
gotNodes = append(gotNodes, nodes.Addrs...)
if nodes.Next == nil {
break
}
startNode = nodes.Next
}
wantNodes := addrs
if fmt.Sprint(gotNodes) != fmt.Sprint(wantNodes) {
t.Fatalf("got #%v nodes %v, want %v", i+1, gotNodes, wantNodes)
}
for addr, wantKeys := range nodeKeys {
var startKey []byte
var gotKeys [][]byte
for {
keys, err := globalStore.NodeKeys(addr, startKey, 0)
if err != nil {
t.Fatal(err)
}
gotKeys = append(gotKeys, keys.Keys...)
if keys.Next == nil {
break
}
startKey = keys.Next
}
if fmt.Sprint(gotKeys) != fmt.Sprint(wantKeys) {
t.Fatalf("got #%v %s node keys %v, want %v", i+1, addr.Hex(), gotKeys, wantKeys)
}
}
for key, wantNodes := range keyNodes {
var startNode *common.Address
var gotNodes []common.Address
for {
nodes, err := globalStore.KeyNodes([]byte(key), startNode, 0)
if err != nil {
t.Fatal(err)
}
gotNodes = append(gotNodes, nodes.Addrs...)
if nodes.Next == nil {
break
}
startNode = nodes.Next
}
if fmt.Sprint(gotNodes) != fmt.Sprint(wantNodes) {
t.Fatalf("got #%v %x key nodes %v, want %v", i+1, []byte(key), gotNodes, wantNodes)
}
}
}
}
// ImportExport saves chunks to the outStore, exports them to the tar archive,
// imports tar archive to the inStore and checks if all chunks are imported correctly.
func ImportExport(t *testing.T, outStore, inStore mock.GlobalStorer, n int) {