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

Persistant state for kademlia (#282)

Browse source 
State persistence store:
- Refactored `Intervals`' `DBStore` and `MemStore` into `swarm/state#Store` (The stores now implement the `Store` interface which allows the consumers to be persistence-agnostic)
- Changed the store database filename
- Refactored the `StateStore` interface into `Store`
This commit is contained in:
Elad Nachmias 2018-03-04 12:15:57 +01:00 committed by GitHub
parent daffa9f471
commit f2e94a2c77
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
18 changed files with 403 additions and 171 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
swarm/network/discovery_test.go
View file

@ -29,7 +29,7 @@ import (
*/
func TestDiscovery(t *testing.T) {
params := NewHiveParams()
s, pp := newHiveTester(t, params)
s, pp := newHiveTester(t, params, 1, nil)
id := s.IDs[0]
raddr := NewAddrFromNodeID(id)

22
swarm/network/hive.go
View file

@ -17,7 +17,6 @@
package network
import (
"encoding/json"
"fmt"
"math/rand"
"sync"
@ -26,6 +25,7 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/swarm/state"
)
/*
@ -79,7 +79,7 @@ func NewHiveParams() *HiveParams {
type Hive struct {
*HiveParams // settings
Overlay // the overlay connectiviy driver
Store StateStore // storage interface to save peers across sessions
Store state.Store // storage interface to save peers across sessions
addPeer func(*discover.Node) // server callback to connect to a peer
// bookkeeping
lock sync.Mutex
@ -90,7 +90,7 @@ type Hive struct {
// HiveParams: config parameters
// Overlay: connectivity driver using a network topology
// StateStore: to save peers across sessions
func NewHive(params *HiveParams, overlay Overlay, store StateStore) *Hive {
func NewHive(params *HiveParams, overlay Overlay, store state.Store) *Hive {
return &Hive{
HiveParams: params,
Overlay: overlay,
@ -202,15 +202,13 @@ func ToAddr(pa OverlayPeer) *BzzAddr {
// loadPeers, savePeer implement persistence callback/
func (h *Hive) loadPeers() error {
data, err := h.Store.Load("peers")
var as []*BzzAddr
err := h.Store.Get("peers", &as)
if err != nil {
return err
}
if data == nil {
if err == state.ErrNotFound {
return nil
}
var as []*BzzAddr
if err := json.Unmarshal(data, &as); err != nil {
return err
}
return h.Register(toOverlayAddrs(as...))
@ -235,11 +233,7 @@ func (h *Hive) savePeers() error {
peers = append(peers, ToAddr(pa))
return true
})
data, err := json.Marshal(peers)
if err != nil {
return fmt.Errorf("could not encode peers: %v", err)
}
if err := h.Store.Save("peers", data); err != nil {
if err := h.Store.Put("peers", peers); err != nil {
return fmt.Errorf("could not save peers: %v", err)
}
return nil

66
swarm/network/hive_test.go
View file

@ -17,33 +17,40 @@
package network
import (
"io/ioutil"
"log"
"os"
"testing"
p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
"github.com/ethereum/go-ethereum/swarm/state"
)
func newHiveTester(t *testing.T, params *HiveParams) (*bzzTester, *Hive) {
func newHiveTester(t *testing.T, params *HiveParams, n int, store state.Store) (*bzzTester, *Hive) {
// setup
addr := RandomAddr() // tested peers peer address
to := NewKademlia(addr.OAddr, NewKadParams())
pp := NewHive(params, to, nil) // hive
pp := NewHive(params, to, store) // hive
return newBzzBaseTester(t, 1, addr, DiscoverySpec, pp.Run), pp
return newBzzBaseTester(t, n, addr, DiscoverySpec, pp.Run), pp
}
func TestRegisterAndConnect(t *testing.T) {
params := NewHiveParams()
s, pp := newHiveTester(t, params)
s, pp := newHiveTester(t, params, 1, nil)
id := s.IDs[0]
raddr := NewAddrFromNodeID(id)
pp.Register([]OverlayAddr{OverlayAddr(raddr)})
// start the hive and wait for the connection
pp.Start(s.Server)
err := pp.Start(s.Server)
if err != nil {
t.Fatal(err)
}
defer pp.Stop()
// retrieve and broadcast
err := s.TestDisconnected(&p2ptest.Disconnect{
err = s.TestDisconnected(&p2ptest.Disconnect{
Peer: s.IDs[0],
Error: nil,
})
@ -52,3 +59,50 @@ func TestRegisterAndConnect(t *testing.T) {
t.Fatalf("expected peer to connect")
}
}
func TestHiveStatePersistance(t *testing.T) {
log.SetOutput(os.Stdout)
dir, err := ioutil.TempDir("", "hive_test_store")
if err != nil {
panic(err)
}
defer os.RemoveAll(dir)
store, err := state.NewDBStore(dir) //start the hive with an empty dbstore
params := NewHiveParams()
s, pp := newHiveTester(t, params, 5, store)
peers := make(map[string]bool)
for _, id := range s.IDs {
raddr := NewAddrFromNodeID(id)
pp.Register([]OverlayAddr{OverlayAddr(raddr)})
peers[raddr.String()] = true
}
// start the hive and wait for the connection
err = pp.Start(s.Server)
if err != nil {
t.Fatal(err)
}
pp.Stop()
store.Close()
persistedStore, err := state.NewDBStore(dir) //start the hive with an empty dbstore
s1, pp := newHiveTester(t, params, 1, persistedStore)
//start the hive and wait for the connection
pp.Start(s1.Server)
i := 0
pp.Overlay.EachAddr(nil, 256, func(addr OverlayAddr, po int, nn bool) bool {
delete(peers, addr.(*BzzAddr).String())
i++
return true
})
if len(peers) != 0 || i != 5 {
t.Fatalf("invalid peers loaded")
}
}

9
swarm/network/protocol.go
View file

@ -31,6 +31,7 @@ import (
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/state"
)
//metrics variables
@ -102,12 +103,6 @@ type Conn interface {
Off() OverlayAddr
}
// StateStore is a container interface to save/load peers across sessions
type StateStore interface {
Load(string) ([]byte, error) // load peer
Save(string, []byte) error // save peer
}
// BzzConfig captures the config params used by the hive
type BzzConfig struct {
OverlayAddr []byte // base address of the overlay network
@ -128,7 +123,7 @@ type Bzz struct {
// * bzz config
// * overlay driver
// * peer store
func NewBzz(config *BzzConfig, kad Overlay, store StateStore) *Bzz {
func NewBzz(config *BzzConfig, kad Overlay, store state.Store) *Bzz {
return &Bzz{
Hive: NewHive(config.HiveParams, kad, store),
localAddr: &BzzAddr{config.OverlayAddr, config.UnderlayAddr},

6
swarm/network/stream/common_test.go
View file

@ -37,7 +37,7 @@ import (
p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
"github.com/ethereum/go-ethereum/swarm/state"
"github.com/ethereum/go-ethereum/swarm/storage"
)
@ -77,7 +77,7 @@ func NewStreamerService(ctx *adapters.ServiceContext) (node.Service, error) {
delivery := NewDelivery(kad, db)
deliveries[id] = delivery
netStore := storage.NewNetStore(store, nil)
r := NewRegistry(addr, delivery, netStore, intervals.NewMemStore(), defaultSkipCheck)
r := NewRegistry(addr, delivery, netStore, state.NewMemStore(), defaultSkipCheck)
RegisterSwarmSyncerServer(r, db)
RegisterSwarmSyncerClient(r, db)
go func() {
@ -107,7 +107,7 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora
db := storage.NewDBAPI(localStore)
delivery := NewDelivery(to, db)
streamer := NewRegistry(addr, delivery, localStore, intervals.NewMemStore(), defaultSkipCheck)
streamer := NewRegistry(addr, delivery, localStore, state.NewMemStore(), defaultSkipCheck)
teardown := func() {
streamer.Close()
removeDataDir()

78
swarm/network/stream/intervals/dbstore.go
View file

@ -1,78 +0,0 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package intervals
import (
"github.com/syndtr/goleveldb/leveldb"
)
// DBStore uses LevelDB to store intervals.
type DBStore struct {
db *leveldb.DB
}
// NewDBStore creates a new instance of DBStore.
func NewDBStore(path string) (s *DBStore, err error) {
db, err := leveldb.OpenFile(path, nil)
if err != nil {
return nil, err
}
return &DBStore{
db: db,
}, nil
}
// Get retrieves Intervals for a specific key. If there is no Intervals
// ErrNotFound is returned.
func (s *DBStore) Get(key string) (i *Intervals, err error) {
k := []byte(key)
has, err := s.db.Has(k, nil)
if err != nil {
return nil, ErrNotFound
}
if !has {
return nil, ErrNotFound
}
data, err := s.db.Get(k, nil)
if err == leveldb.ErrNotFound {
err = ErrNotFound
}
i = &Intervals{}
if err = i.UnmarshalBinary(data); err != nil {
return nil, err
}
return i, err
}
// Put stores Intervals for a specific key.
func (s *DBStore) Put(key string, i *Intervals) (err error) {
data, err := i.MarshalBinary()
if err != nil {
return err
}
return s.db.Put([]byte(key), data, nil)
}
// Delete removes Intervals stored under a specific key.
func (s *DBStore) Delete(key string) (err error) {
return s.db.Delete([]byte(key), nil)
}
// Close releases the resources used by the underlying LevelDB.
func (s *DBStore) Close() error {
return s.db.Close()
}

4
swarm/network/stream/intervals/dbstore_test.go
View file

@ -20,6 +20,8 @@ import (
"io/ioutil"
"os"
"testing"
"github.com/ethereum/go-ethereum/swarm/state"
)
// TestDBStore tests basic functionality of DBStore.
@ -30,7 +32,7 @@ func TestDBStore(t *testing.T) {
}
defer os.RemoveAll(dir)
store, err := NewDBStore(dir)
store, err := state.NewDBStore(dir)
if err != nil {
t.Fatal(err)
}

36
swarm/network/stream/intervals/store_test.go
View file

@ -16,27 +16,35 @@
package intervals
import "testing"
import (
"errors"
"testing"
"github.com/ethereum/go-ethereum/swarm/state"
)
var ErrNotFound = errors.New("not found")
// TestMemStore tests basic functionality of MemStore.
func TestMemStore(t *testing.T) {
testStore(t, NewMemStore())
testStore(t, state.NewMemStore())
}
// testStore is a helper function to test various Store implementations.
func testStore(t *testing.T, s Store) {
func testStore(t *testing.T, s state.Store) {
key1 := "key1"
i1 := NewIntervals(0)
i1.Add(10, 20)
if err := s.Put(key1, i1); err != nil {
t.Fatal(err)
}
g, err := s.Get(key1)
i := &Intervals{}
err := s.Get(key1, i)
if err != nil {
t.Fatal(err)
}
if g.String() != i1.String() {
t.Errorf("expected interval %s, got %s", i1, g)
if i.String() != i1.String() {
t.Errorf("expected interval %s, got %s", i1, i)
}
key2 := "key2"
@ -45,28 +53,28 @@ func testStore(t *testing.T, s Store) {
if err := s.Put(key2, i2); err != nil {
t.Fatal(err)
}
g, err = s.Get(key2)
err = s.Get(key2, i)
if err != nil {
t.Fatal(err)
}
if g.String() != i2.String() {
t.Errorf("expected interval %s, got %s", i2, g)
if i.String() != i2.String() {
t.Errorf("expected interval %s, got %s", i2, i)
}
if err := s.Delete(key1); err != nil {
t.Fatal(err)
}
if _, err := s.Get(key1); err != ErrNotFound {
t.Errorf("expected error %v, got %s", ErrNotFound, err)
if err := s.Get(key1, i); err != state.ErrNotFound {
t.Errorf("expected error %v, got %s", state.ErrNotFound, err)
}
if _, err := s.Get(key2); err != nil {
if err := s.Get(key2, i); err != nil {
t.Errorf("expected error %v, got %s", nil, err)
}
if err := s.Delete(key2); err != nil {
t.Fatal(err)
}
if _, err := s.Get(key2); err != ErrNotFound {
t.Errorf("expected error %v, got %s", ErrNotFound, err)
if err := s.Get(key2, i); err != state.ErrNotFound {
t.Errorf("expected error %v, got %s", state.ErrNotFound, err)
}
}

4
swarm/network/stream/intervals_test.go
View file

@ -31,8 +31,8 @@ import (
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
streamTesting "github.com/ethereum/go-ethereum/swarm/network/stream/testing"
"github.com/ethereum/go-ethereum/swarm/state"
"github.com/ethereum/go-ethereum/swarm/storage"
)
@ -51,7 +51,7 @@ func newIntervalsStreamerService(ctx *adapters.ServiceContext) (node.Service, er
delivery := NewDelivery(kad, db)
deliveries[id] = delivery
netStore := storage.NewNetStore(store, nil)
r := NewRegistry(addr, delivery, netStore, intervals.NewMemStore(), defaultSkipCheck)
r := NewRegistry(addr, delivery, netStore, state.NewMemStore(), defaultSkipCheck)
r.RegisterClientFunc(externalStreamName, func(p *Peer, t []byte, live bool) (Client, error) {
return newTestExternalClient(t, db), nil

11
swarm/network/stream/peer.go
View file

@ -26,6 +26,7 @@ import (
"github.com/ethereum/go-ethereum/p2p/protocols"
pq "github.com/ethereum/go-ethereum/swarm/network/priorityqueue"
"github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
"github.com/ethereum/go-ethereum/swarm/state"
"github.com/ethereum/go-ethereum/swarm/storage"
)
@ -236,21 +237,23 @@ func (p *Peer) getOrSetClient(s Stream, from, to uint64) (c *client, created boo
if s.Live {
// try to find previous history and live intervals and merge live into history
historyKey := peerStreamIntervalsKey(p, NewStream(s.Name, s.Key, false))
historyIntervals, err := p.streamer.intervalsStore.Get(historyKey)
historyIntervals := &intervals.Intervals{}
err := p.streamer.intervalsStore.Get(historyKey, historyIntervals)
switch err {
case nil:
liveIntervals, err := p.streamer.intervalsStore.Get(intervalsKey)
liveIntervals := &intervals.Intervals{}
err := p.streamer.intervalsStore.Get(intervalsKey, liveIntervals)
switch err {
case nil:
historyIntervals.Merge(liveIntervals)
if err := p.streamer.intervalsStore.Put(historyKey, historyIntervals); err != nil {
log.Error("stream set client: put history intervals", "stream", s, "peer", p, "err", err)
}
case intervals.ErrNotFound:
case state.ErrNotFound:
default:
log.Error("stream set client: get live intervals", "stream", s, "peer", p, "err", err)
}
case intervals.ErrNotFound:
case state.ErrNotFound:
default:
log.Error("stream set client: get history intervals", "stream", s, "peer", p, "err", err)
}

13
swarm/network/stream/stream.go
View file

@ -30,6 +30,7 @@ import (
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
"github.com/ethereum/go-ethereum/swarm/state"
"github.com/ethereum/go-ethereum/swarm/storage"
)
@ -56,11 +57,11 @@ type Registry struct {
peers map[discover.NodeID]*Peer
delivery *Delivery
store storage.ChunkStore
intervalsStore intervals.Store
intervalsStore state.Store
}
// NewRegistry is Streamer constructor
func NewRegistry(addr *network.BzzAddr, delivery *Delivery, store storage.ChunkStore, intervalsStore intervals.Store, skipCheck bool) *Registry {
func NewRegistry(addr *network.BzzAddr, delivery *Delivery, store storage.ChunkStore, intervalsStore state.Store, skipCheck bool) *Registry {
streamer := &Registry{
addr: addr,
skipCheck: skipCheck,
@ -297,7 +298,7 @@ type client struct {
next chan error
intervalsKey string
intervalsStore intervals.Store
intervalsStore state.Store
}
func peerStreamIntervalsKey(p *Peer, s Stream) string {
@ -305,7 +306,8 @@ func peerStreamIntervalsKey(p *Peer, s Stream) string {
}
func (c client) AddInterval(start, end uint64) (err error) {
i, err := c.intervalsStore.Get(c.intervalsKey)
i := &intervals.Intervals{}
err = c.intervalsStore.Get(c.intervalsKey, i)
if err != nil {
return err
}
@ -314,7 +316,8 @@ func (c client) AddInterval(start, end uint64) (err error) {
}
func (c client) NextInterval() (start, end uint64, err error) {
i, err := c.intervalsStore.Get(c.intervalsKey)
i := &intervals.Intervals{}
err = c.intervalsStore.Get(c.intervalsKey, i)
if err != nil {
return 0, 0, err
}

3
swarm/pss/client/client_test.go
View file

@ -21,6 +21,7 @@ import (
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/pss"
"github.com/ethereum/go-ethereum/swarm/state"
"github.com/ethereum/go-ethereum/swarm/storage"
whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
)
@ -211,7 +212,7 @@ func setupNetwork(numnodes int) (clients []*rpc.Client, err error) {
}
func newServices() adapters.Services {
stateStore := newTestStore()
stateStore := state.NewMemStore()
kademlias := make(map[discover.NodeID]*network.Kademlia)
kademlia := func(id discover.NodeID) *network.Kademlia {
if k, ok := kademlias[id]; ok {

3
swarm/pss/pss_test.go
View file

@ -30,6 +30,7 @@ import (
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/state"
"github.com/ethereum/go-ethereum/swarm/storage"
whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
)
@ -1112,7 +1113,7 @@ func setupNetwork(numnodes int) (clients []*rpc.Client, err error) {
}
func newServices() adapters.Services {
stateStore := newStateStore()
stateStore := state.NewMemStore()
kademlias := make(map[discover.NodeID]*network.Kademlia)
kademlia := func(id discover.NodeID) *network.Kademlia {
if k, ok := kademlias[id]; ok {

96
swarm/state/dbstore.go Normal file
View file

@ -0,0 +1,96 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package state
import (
"encoding"
"encoding/json"
"errors"
"github.com/syndtr/goleveldb/leveldb"
)
// ErrNotFound is returned when no results are returned from the database
var ErrNotFound = errors.New("ErrorNotFound")
// ErrInvalidArgument is returned when the argument type does not match the expected type
var ErrInvalidArgument = errors.New("ErrorInvalidArgument")
// DBStore uses LevelDB to store values.
type DBStore struct {
db *leveldb.DB
}
// NewDBStore creates a new instance of DBStore.
func NewDBStore(path string) (s *DBStore, err error) {
db, err := leveldb.OpenFile(path, nil)
if err != nil {
return nil, err
}
return &DBStore{
db: db,
}, nil
}
// Get retrieves a persisted value for a specific key. If there is no results
// ErrNotFound is returned. The provided parameter should be either a byte slice or
// a struct that implements the encoding.BinaryUnmarshaler interface
func (s *DBStore) Get(key string, i interface{}) (err error) {
has, err := s.db.Has([]byte(key), nil)
if err != nil || !has {
return ErrNotFound
}
data, err := s.db.Get([]byte(key), nil)
if err == leveldb.ErrNotFound {
return ErrNotFound
}
unmarshaler, ok := i.(encoding.BinaryUnmarshaler)
if !ok {
return json.Unmarshal(data, i)
}
return unmarshaler.UnmarshalBinary(data)
}
// Put stores an object that implements Binary for a specific key.
func (s *DBStore) Put(key string, i interface{}) (err error) {
bytes := []byte{}
marshaler, ok := i.(encoding.BinaryMarshaler)
if !ok {
if bytes, err = json.Marshal(i); err != nil {
return err
}
} else {
if bytes, err = marshaler.MarshalBinary(); err != nil {
return err
}
}
return s.db.Put([]byte(key), bytes, nil)
}
// Delete removes entries stored under a specific key.
func (s *DBStore) Delete(key string) (err error) {
return s.db.Delete([]byte(key), nil)
}
// Close releases the resources used by the underlying LevelDB.
func (s *DBStore) Close() error {
return s.db.Close()
}

122
swarm/state/dbstore_test.go Normal file
View file

@ -0,0 +1,122 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package state
import (
"bytes"
"errors"
"io/ioutil"
"os"
"strings"
"testing"
)
var ErrInvalidArraySize = errors.New("invalid byte array size")
var ErrInvalidValuePersisted = errors.New("invalid value was persisted to the db")
type SerializingType struct {
key string
value string
}
func (st *SerializingType) MarshalBinary() (data []byte, err error) {
d := []byte(strings.Join([]string{st.key, st.value}, ";"))
return d, nil
}
func (st *SerializingType) UnmarshalBinary(data []byte) (err error) {
d := bytes.Split(data, []byte(";"))
l := len(d)
if l == 0 {
return ErrInvalidArraySize
}
if l == 2 {
keyLen := len(d[0])
st.key = string(d[0][:keyLen])
valLen := len(d[1])
st.value = string(d[1][:valLen])
}
return nil
}
// TestDBStore tests basic functionality of DBStore.
func TestDBStore(t *testing.T) {
dir, err := ioutil.TempDir("", "db_store_test")
if err != nil {
panic(err)
}
defer os.RemoveAll(dir)
store, err := NewDBStore(dir)
if err != nil {
t.Fatal(err)
}
testStore(t, store)
store.Close()
persistedStore, err := NewDBStore(dir)
if err != nil {
t.Fatal(err)
}
defer persistedStore.Close()
testPersistedStore(t, persistedStore)
}
func testStore(t *testing.T, store Store) {
ser := &SerializingType{key: "key1", value: "value1"}
jsonify := []string{"a", "b", "c"}
err := store.Put(ser.key, ser)
if err != nil {
t.Fatal(err)
}
err = store.Put("key2", jsonify)
if err != nil {
t.Fatal(err)
}
}
func testPersistedStore(t *testing.T, store Store) {
ser := &SerializingType{}
err := store.Get("key1", ser)
if err != nil {
t.Fatal(err)
}
if ser.key != "key1" || ser.value != "value1" {
t.Fatal(ErrInvalidValuePersisted)
}
as := []string{}
err = store.Get("key2", &as)
if len(as) != 3 {
t.Fatalf("serialized array did not match expectation")
}
if as[0] != "a" || as[1] != "b" || as[2] != "c" {
t.Fatalf("elements serialized did not match expected values")
}
}

61
swarm/network/stream/intervals/store.go → swarm/state/memstore.go
View file

@ -14,64 +14,69 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package intervals
package state
import (
"errors"
"encoding"
"encoding/json"
"sync"
)
// ErrNotFound is returned by the Store implementation when the Interval
// for a specific key does not exist.
var ErrNotFound = errors.New("not found")
// Store defines methods required to get and retrieve Intervals for different keys.
// It is meant to be used for intervals persistence for different streams in the
// stream package.
type Store interface {
Get(key string) (i *Intervals, err error)
Put(key string, i *Intervals) (err error)
Delete(key string) (err error)
Close() error
}
// MemStore is the reference implementation of Store interface that is supposed
// to be used in tests.
type MemStore struct {
db map[string]*Intervals
db map[string][]byte
mu sync.RWMutex
}
// NewMemStore returns a new instance of MemStore.
func NewMemStore() *MemStore {
return &MemStore{
db: make(map[string]*Intervals),
db: make(map[string][]byte),
}
}
// Get retrieves Intervals for a specific key. If there is no Intervals
// Get retrieves a value stored for a specific key. If there is no value found,
// ErrNotFound is returned.
func (s *MemStore) Get(key string) (i *Intervals, err error) {
func (s *MemStore) Get(key string, i interface{}) (err error) {
s.mu.RLock()
defer s.mu.RUnlock()
i, ok := s.db[key]
bytes, ok := s.db[key]
if !ok {
return nil, ErrNotFound
return ErrNotFound
}
return i, nil
unmarshaler, ok := i.(encoding.BinaryUnmarshaler)
if !ok {
return json.Unmarshal(bytes, i)
}
return unmarshaler.UnmarshalBinary(bytes)
}
// Put stores Intervals for a specific key.
func (s *MemStore) Put(key string, i *Intervals) (err error) {
// Put stores a value for a specific key.
func (s *MemStore) Put(key string, i interface{}) (err error) {
s.mu.Lock()
defer s.mu.Unlock()
bytes := []byte{}
s.db[key] = i
marshaler, ok := i.(encoding.BinaryMarshaler)
if !ok {
if bytes, err = json.Marshal(i); err != nil {
return err
}
} else {
if bytes, err = marshaler.MarshalBinary(); err != nil {
return err
}
}
s.db[key] = bytes
return nil
}
// Delete removes Intervals stored under a specific key.
// Delete removes value stored under a specific key.
func (s *MemStore) Delete(key string) (err error) {
s.mu.Lock()
defer s.mu.Unlock()
@ -83,7 +88,7 @@ func (s *MemStore) Delete(key string) (err error) {
return nil
}
// Close doesnot do anything.
// Close does not do anything.
func (s *MemStore) Close() error {
return nil
}

26
swarm/state/store.go Normal file
View file

@ -0,0 +1,26 @@
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package state
// Store defines methods required to get, set, delete values for different keys
// and close the underlying resources.
type Store interface {
Get(key string, i interface{}) (err error)
Put(key string, i interface{}) (err error)
Delete(key string) (err error)
Close() error
}

8
swarm/swarm.go
View file

@ -47,8 +47,8 @@ import (
"github.com/ethereum/go-ethereum/swarm/fuse"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/stream"
"github.com/ethereum/go-ethereum/swarm/network/stream/intervals"
"github.com/ethereum/go-ethereum/swarm/pss"
"github.com/ethereum/go-ethereum/swarm/state"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/storage/mock"
)
@ -149,15 +149,15 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api.
db := storage.NewDBAPI(self.lstore)
delivery := stream.NewDelivery(to, db)
// TODO: decide on intervals store file location
intervalsStore, err := intervals.NewDBStore(filepath.Join(config.Path, "stream-intervals.db"))
stateStore, err := state.NewDBStore(filepath.Join(config.Path, "state-store.db"))
if err != nil {
return
}
self.streamer = stream.NewRegistry(addr, delivery, self.lstore, intervalsStore, false)
self.streamer = stream.NewRegistry(addr, delivery, self.lstore, stateStore, false)
stream.RegisterSwarmSyncerServer(self.streamer, db)
stream.RegisterSwarmSyncerClient(self.streamer, db)
self.bzz = network.NewBzz(bzzconfig, to, nil)
self.bzz = network.NewBzz(bzzconfig, to, stateStore)
// set up DPA, the cloud storage local access layer
dpaChunkStore := storage.NewNetStore(self.lstore, self.streamer.Retrieve)