From f2e94a2c7737bfac987839682bb4fd5a60893250 Mon Sep 17 00:00:00 2001 From: Elad Nachmias Date: Sun, 4 Mar 2018 12:15:57 +0100 Subject: [PATCH] Persistant state for kademlia (#282) 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` --- swarm/network/discovery_test.go | 2 +- swarm/network/hive.go | 26 ++-- swarm/network/hive_test.go | 66 +++++++++- swarm/network/protocol.go | 9 +- swarm/network/stream/common_test.go | 6 +- swarm/network/stream/intervals/dbstore.go | 78 ----------- .../network/stream/intervals/dbstore_test.go | 4 +- swarm/network/stream/intervals/store_test.go | 36 ++++-- swarm/network/stream/intervals_test.go | 4 +- swarm/network/stream/peer.go | 11 +- swarm/network/stream/stream.go | 13 +- swarm/pss/client/client_test.go | 3 +- swarm/pss/pss_test.go | 3 +- swarm/state/dbstore.go | 96 ++++++++++++++ swarm/state/dbstore_test.go | 122 ++++++++++++++++++ .../intervals/store.go => state/memstore.go} | 61 +++++---- swarm/state/store.go | 26 ++++ swarm/swarm.go | 8 +- 18 files changed, 403 insertions(+), 171 deletions(-) delete mode 100644 swarm/network/stream/intervals/dbstore.go create mode 100644 swarm/state/dbstore.go create mode 100644 swarm/state/dbstore_test.go rename swarm/{network/stream/intervals/store.go => state/memstore.go} (59%) create mode 100644 swarm/state/store.go diff --git a/swarm/network/discovery_test.go b/swarm/network/discovery_test.go index 4cdb341ce6..0427d81cad 100644 --- a/swarm/network/discovery_test.go +++ b/swarm/network/discovery_test.go @@ -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) diff --git a/swarm/network/hive.go b/swarm/network/hive.go index 05004d1b8c..172403379d 100644 --- a/swarm/network/hive.go +++ b/swarm/network/hive.go @@ -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") - if err != nil { - return err - } - if data == nil { - return nil - } var as []*BzzAddr - if err := json.Unmarshal(data, &as); err != nil { + + err := h.Store.Get("peers", &as) + if err != nil { + if err == state.ErrNotFound { + return 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 diff --git a/swarm/network/hive_test.go b/swarm/network/hive_test.go index daa9389291..c2abfb2aad 100644 --- a/swarm/network/hive_test.go +++ b/swarm/network/hive_test.go @@ -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") + } +} diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index 9191980038..f59ab7c71d 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -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}, diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index 2a8245413b..f7c767b352 100644 --- a/swarm/network/stream/common_test.go +++ b/swarm/network/stream/common_test.go @@ -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() diff --git a/swarm/network/stream/intervals/dbstore.go b/swarm/network/stream/intervals/dbstore.go deleted file mode 100644 index 1849c09433..0000000000 --- a/swarm/network/stream/intervals/dbstore.go +++ /dev/null @@ -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 . - -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() -} diff --git a/swarm/network/stream/intervals/dbstore_test.go b/swarm/network/stream/intervals/dbstore_test.go index 75a7ddbfb4..6716e593ab 100644 --- a/swarm/network/stream/intervals/dbstore_test.go +++ b/swarm/network/stream/intervals/dbstore_test.go @@ -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) } diff --git a/swarm/network/stream/intervals/store_test.go b/swarm/network/stream/intervals/store_test.go index 0b7344345f..5efb6ae8a6 100644 --- a/swarm/network/stream/intervals/store_test.go +++ b/swarm/network/stream/intervals/store_test.go @@ -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) } } diff --git a/swarm/network/stream/intervals_test.go b/swarm/network/stream/intervals_test.go index bda123e7c0..fe90bb8c88 100644 --- a/swarm/network/stream/intervals_test.go +++ b/swarm/network/stream/intervals_test.go @@ -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 diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index c9b623e792..0bc079b1cc 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -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) } diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index d0e230eb37..360567c559 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -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 } diff --git a/swarm/pss/client/client_test.go b/swarm/pss/client/client_test.go index ae6b423382..9ce4e856a8 100644 --- a/swarm/pss/client/client_test.go +++ b/swarm/pss/client/client_test.go @@ -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 { diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 242f653463..190d704f59 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -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 { diff --git a/swarm/state/dbstore.go b/swarm/state/dbstore.go new file mode 100644 index 0000000000..5e5c172b25 --- /dev/null +++ b/swarm/state/dbstore.go @@ -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 . + +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() +} diff --git a/swarm/state/dbstore_test.go b/swarm/state/dbstore_test.go new file mode 100644 index 0000000000..6683e788f7 --- /dev/null +++ b/swarm/state/dbstore_test.go @@ -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 . + +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") + } +} diff --git a/swarm/network/stream/intervals/store.go b/swarm/state/memstore.go similarity index 59% rename from swarm/network/stream/intervals/store.go rename to swarm/state/memstore.go index d5a3bb8437..140697bdd0 100644 --- a/swarm/network/stream/intervals/store.go +++ b/swarm/state/memstore.go @@ -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 . -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 } diff --git a/swarm/state/store.go b/swarm/state/store.go new file mode 100644 index 0000000000..fb7fe258f1 --- /dev/null +++ b/swarm/state/store.go @@ -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 . + +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 +} diff --git a/swarm/swarm.go b/swarm/swarm.go index 91ec805511..9966da228d 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -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)