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

swarm/network, swarm/storage: initial netowork/stream refactor

Browse source
This commit is contained in:
Janos Guljas 2018-01-18 17:53:29 +01:00
parent 0265566534
commit 35609bec2e
20 changed files with 1119 additions and 989 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
swarm/network/discovery.go
View file

@ -25,9 +25,9 @@ import (
// discovery bzz extension for requesting and relaying node address records // discovery bzz extension for requesting and relaying node address records
// discPeer wraps bzzPeer and embeds an Overlay connectivity driver // discPeer wraps BzzPeer and embeds an Overlay connectivity driver
type discPeer struct { type discPeer struct {
*bzzPeer *BzzPeer
overlay Overlay overlay Overlay
sentPeers bool // whether we already sent peer closer to this address sentPeers bool // whether we already sent peer closer to this address
mtx sync.Mutex mtx sync.Mutex
@ -36,10 +36,10 @@ type discPeer struct {
} }
// NewDiscovery constructs a discovery peer // NewDiscovery constructs a discovery peer
func newDiscovery(p *bzzPeer, o Overlay) *discPeer { func newDiscovery(p *BzzPeer, o Overlay) *discPeer {
d := &discPeer{ d := &discPeer{
overlay: o, overlay: o,
bzzPeer: p, BzzPeer: p,
peers: make(map[string]bool), peers: make(map[string]bool),
} }
// record remote as seen so we never send a peer its own record // record remote as seen so we never send a peer its own record

2
swarm/network/discovery_test.go
View file

@ -33,7 +33,7 @@ func TestDiscovery(t *testing.T) {
addr := RandomAddr() addr := RandomAddr()
to := NewKademlia(addr.OAddr, NewKadParams()) to := NewKademlia(addr.OAddr, NewKadParams())
run := func(p *bzzPeer) error { run := func(p *BzzPeer) error {
dp := newDiscovery(p, to) dp := newDiscovery(p, to)
to.On(p) to.On(p)
defer to.Off(p) defer to.Off(p)

4
swarm/network/hive.go
View file

@ -159,7 +159,7 @@ func (h *Hive) connect() {
} }
// Run protocol run function // Run protocol run function
func (h *Hive) Run(p *bzzPeer) error { func (h *Hive) Run(p *BzzPeer) error {
dp := newDiscovery(p, h) dp := newDiscovery(p, h)
depth, changed := h.On(dp) depth, changed := h.On(dp)
// if we want discovery, advertise changed depth of depth // if we want discovery, advertise changed depth of depth
@ -191,7 +191,7 @@ func ToAddr(pa OverlayPeer) *BzzAddr {
if p, ok := pa.(*discPeer); ok { if p, ok := pa.(*discPeer); ok {
return p.BzzAddr return p.BzzAddr
} }
return pa.(*bzzPeer).BzzAddr return pa.(*BzzPeer).BzzAddr
} }
// loadPeers, savePeer implement persistence callback/ // loadPeers, savePeer implement persistence callback/

2
swarm/network/kademlia_test.go
View file

@ -70,7 +70,7 @@ func newTestKademlia(b string) *testKademlia {
} }
func (k *testKademlia) newTestKadPeer(s string) Peer { func (k *testKademlia) newTestKadPeer(s string) Peer {
return &testDropPeer{&bzzPeer{BzzAddr: testKadPeerAddr(s)}, k.dropc} return &testDropPeer{&BzzPeer{BzzAddr: testKadPeerAddr(s)}, k.dropc}
} }
func (k *testKademlia) On(ons ...string) *testKademlia { func (k *testKademlia) On(ons ...string) *testKademlia {

55
swarm/network/lightnode.go → swarm/network/light/lightnode.go
View file

@ -1,4 +1,4 @@
// Copyright 2017 The go-ethereum Authors // Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.d // This file is part of the go-ethereum library.d
// //
// The go-ethereum library is free software: you can redistribute it and/or modify // The go-ethereum library is free software: you can redistribute it and/or modify
@ -14,17 +14,18 @@
// You should have received a copy of the GNU Lesser General Public License // 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/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package network package light
import ( import (
"errors" "errors"
"github.com/ethereum/go-ethereum/swarm/network/stream"
"github.com/ethereum/go-ethereum/swarm/storage" "github.com/ethereum/go-ethereum/swarm/storage"
) )
// RemoteReader implements IncomingStreamer // RemoteReader implements IncomingStreamer
type RemoteSectionReader struct { type RemoteSectionReader struct {
db *DbAccess db *storage.DBAPI
start uint64 start uint64
end uint64 end uint64
hashes chan []byte hashes chan []byte
@ -35,7 +36,7 @@ type RemoteSectionReader struct {
} }
// NewRemoteReader is the constructor for RemoteReader // NewRemoteReader is the constructor for RemoteReader
func NewRemoteSectionReader(root []byte, db *DbAccess) *RemoteSectionReader { func NewRemoteSectionReader(root []byte, db *storage.DBAPI) *RemoteSectionReader {
return &RemoteSectionReader{ return &RemoteSectionReader{
db: db, db: db,
root: root, root: root,
@ -45,7 +46,7 @@ func NewRemoteSectionReader(root []byte, db *DbAccess) *RemoteSectionReader {
} }
func (r *RemoteSectionReader) NeedData(key []byte) func() { func (r *RemoteSectionReader) NeedData(key []byte) func() {
chunk, created := r.db.getOrCreateRequest(storage.Key(key)) chunk, created := r.db.GetOrCreateRequest(storage.Key(key))
// TODO: we may want to request from this peer anyway even if the request exists // TODO: we may want to request from this peer anyway even if the request exists
if chunk.ReqC == nil || !created { if chunk.ReqC == nil || !created {
return nil return nil
@ -58,7 +59,7 @@ func (r *RemoteSectionReader) NeedData(key []byte) func() {
} }
} }
func (r *RemoteSectionReader) BatchDone(s string, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) { func (r *RemoteSectionReader) BatchDone(s string, from uint64, hashes []byte, root []byte) func() (*stream.TakeoverProof, error) {
r.hashes <- hashes r.hashes <- hashes
return nil return nil
} }
@ -75,9 +76,9 @@ func (r *RemoteSectionReader) Read(b []byte) (n int64, err error) {
return l, nil return l, nil
} }
var end bool var end bool
for i := 0; !end && i < len(r.currentHashes); i += HashSize { for i := 0; !end && i < len(r.currentHashes); i += stream.HashSize {
hash := r.currentHashes[i : i+HashSize] hash := r.currentHashes[i : i+stream.HashSize]
chunk, err := r.db.get(hash) chunk, err := r.db.Get(hash)
if err != nil { if err != nil {
return n, err return n, err
} }
@ -96,9 +97,9 @@ func (r *RemoteSectionReader) Read(b []byte) (n int64, err error) {
return n, errors.New("aborted") return n, errors.New("aborted")
case hashes := <-r.hashes: case hashes := <-r.hashes:
var i int var i int
for ; !end && i < len(hashes); i += HashSize { for ; !end && i < len(hashes); i += stream.HashSize {
hash := hashes[i : i+HashSize] hash := hashes[i : i+stream.HashSize]
chunk, err := r.db.get(hash) chunk, err := r.db.Get(hash)
if err != nil { if err != nil {
return n, err return n, err
} }
@ -120,12 +121,12 @@ func (r *RemoteSectionReader) Read(b []byte) (n int64, err error) {
type RemoteSectionServer struct { type RemoteSectionServer struct {
// quit chan struct{} // quit chan struct{}
root []byte root []byte
db *DbAccess db *storage.DBAPI
r *storage.LazyChunkReader r *storage.LazyChunkReader
} }
// NewRemoteReader is the constructor for RemoteReader // NewRemoteReader is the constructor for RemoteReader
func NewRemoteSectionServer(db *DbAccess, r *storage.LazyChunkReader) *RemoteSectionServer { func NewRemoteSectionServer(db *storage.DBAPI, r *storage.LazyChunkReader) *RemoteSectionServer {
return &RemoteSectionServer{ return &RemoteSectionServer{
db: db, db: db,
r: r, r: r,
@ -134,7 +135,7 @@ func NewRemoteSectionServer(db *DbAccess, r *storage.LazyChunkReader) *RemoteSec
// GetData retrieves the actual chunk from localstore // GetData retrieves the actual chunk from localstore
func (s *RemoteSectionServer) GetData(key []byte) []byte { func (s *RemoteSectionServer) GetData(key []byte) []byte {
chunk, err := s.db.get(storage.Key(key)) chunk, err := s.db.Get(storage.Key(key))
if err != nil { if err != nil {
return nil return nil
} }
@ -142,26 +143,26 @@ func (s *RemoteSectionServer) GetData(key []byte) []byte {
} }
// GetBatch retrieves the next batch of hashes from the dbstore // GetBatch retrieves the next batch of hashes from the dbstore
func (s *RemoteSectionServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) { func (s *RemoteSectionServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint64, *stream.HandoverProof, error) {
if to > from+batchSize { if to > from+stream.BatchSize {
to = from + batchSize to = from + stream.BatchSize
} }
batch := make([]byte, (to-from)*HashSize) batch := make([]byte, (to-from)*stream.HashSize)
s.r.ReadAt(batch, int64(from)) s.r.ReadAt(batch, int64(from))
return batch, from, to, nil, nil return batch, from, to, nil, nil
} }
// RegisterRemoteSectionReader registers RemoteSectionReader on light downstream node // RegisterRemoteSectionReader registers RemoteSectionReader on light downstream node
func RegisterRemoteSectionReader(s *Streamer, db *DbAccess) { func RegisterRemoteSectionReader(s *stream.Registry, db *storage.DBAPI) {
s.RegisterIncomingStreamer("REMOTE_SECTION", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) { s.RegisterClientFunc("REMOTE_SECTION", func(p *stream.Peer, t []byte) (stream.Client, error) {
return NewRemoteSectionReader(t, db), nil return NewRemoteSectionReader(t, db), nil
}) })
} }
// RegisterRemoteSectionServer registers RemoteSectionServer outgoing streamer on // RegisterRemoteSectionServer registers RemoteSectionServer outgoing streamer on
// upstream light server node // upstream light server node
func RegisterRemoteSectionServer(s *Streamer, db *DbAccess, rf func([]byte) *storage.LazyChunkReader) { func RegisterRemoteSectionServer(s *stream.Registry, db *storage.DBAPI, rf func([]byte) *storage.LazyChunkReader) {
s.RegisterOutgoingStreamer("REMOTE_SECTION", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) { s.RegisterServerFunc("REMOTE_SECTION", func(p *stream.Peer, t []byte) (stream.Server, error) {
r := rf(t) r := rf(t)
return NewRemoteSectionServer(db, r), nil return NewRemoteSectionServer(db, r), nil
}) })
@ -169,16 +170,16 @@ func RegisterRemoteSectionServer(s *Streamer, db *DbAccess, rf func([]byte) *sto
// RegisterRemoteDownloader registers RemoteDownloader incoming streamer // RegisterRemoteDownloader registers RemoteDownloader incoming streamer
// on downstream light node // on downstream light node
// func RegisterRemoteDownloader(s *Streamer, db *DbAccess) { // func RegisterRemoteDownloader(s *Streamer, db *storage.DBAPI) {
// s.RegisterIncomingStreamer("REMOTE_DOWNLOADER", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) { // s.RegisterIncomingStreamer("REMOTE_DOWNLOADER", func(p *stream.Peer, t []byte) (IncomingStreamer, error) {
// return NewRemoteDownloader(t, db), nil // return NewRemoteDownloader(t, db), nil
// }) // })
// } // }
// //
// // RegisterRemoteDownloadServer registers RemoteDownloadServer outgoing streamer on // // RegisterRemoteDownloadServer registers RemoteDownloadServer outgoing streamer on
// // upstream light server node // // upstream light server node
// func RegisterRemoteDownloadServer(s *Streamer, db *DbAccess, rf func([]byte) *storage.LazyChunkReader) { // func RegisterRemoteDownloadServer(s *Streamer, db *storage.DBAPI, rf func([]byte) *storage.LazyChunkReader) {
// s.RegisterOutgoingStreamer("REMOTE_DOWNLOADER", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) { // s.RegisterOutgoingStreamer("REMOTE_DOWNLOADER", func(p *stream.Peer, t []byte) (OutgoingStreamer, error) {
// r := rf(t) // r := rf(t)
// return NewRemoteDownloadServer(db, r), nil // return NewRemoteDownloadServer(db, r), nil
// }) // })

35
swarm/network/protocol.go
View file

@ -103,7 +103,6 @@ type BzzConfig struct {
// Bzz is the swarm protocol bundle // Bzz is the swarm protocol bundle
type Bzz struct { type Bzz struct {
Streamer *Streamer
*Hive *Hive
localAddr *BzzAddr localAddr *BzzAddr
mtx sync.Mutex mtx sync.Mutex
@ -115,9 +114,8 @@ type Bzz struct {
// * bzz config // * bzz config
// * overlay driver // * overlay driver
// * peer store // * peer store
func NewBzz(config *BzzConfig, kad Overlay, store StateStore, streamer *Streamer) *Bzz { func NewBzz(config *BzzConfig, kad Overlay, store StateStore) *Bzz {
return &Bzz{ return &Bzz{
Streamer: streamer,
Hive: NewHive(config.HiveParams, kad, store), Hive: NewHive(config.HiveParams, kad, store),
localAddr: &BzzAddr{config.OverlayAddr, config.UnderlayAddr}, localAddr: &BzzAddr{config.OverlayAddr, config.UnderlayAddr},
handshakes: make(map[discover.NodeID]*HandshakeMsg), handshakes: make(map[discover.NodeID]*HandshakeMsg),
@ -143,7 +141,7 @@ func (b *Bzz) NodeInfo() interface{} {
// * handshake/hive // * handshake/hive
// * discovery // * discovery
func (b *Bzz) Protocols() []p2p.Protocol { func (b *Bzz) Protocols() []p2p.Protocol {
protocols := []p2p.Protocol{ return []p2p.Protocol{
{ {
Name: BzzSpec.Name, Name: BzzSpec.Name,
Version: BzzSpec.Version, Version: BzzSpec.Version,
@ -160,17 +158,6 @@ func (b *Bzz) Protocols() []p2p.Protocol {
PeerInfo: b.Hive.PeerInfo, PeerInfo: b.Hive.PeerInfo,
}, },
} }
if b.Streamer != nil {
protocols = append(protocols, p2p.Protocol{
Name: StreamerSpec.Name,
Version: StreamerSpec.Version,
Length: StreamerSpec.Length(),
Run: b.RunProtocol(StreamerSpec, b.Streamer.Run),
NodeInfo: b.Streamer.NodeInfo,
PeerInfo: b.Streamer.PeerInfo,
})
}
return protocols
} }
// APIs returns the APIs offered by bzz // APIs returns the APIs offered by bzz
@ -188,12 +175,12 @@ func (b *Bzz) APIs() []rpc.API {
// returns a p2p protocol run function that can be assigned to p2p.Protocol#Run field // returns a p2p protocol run function that can be assigned to p2p.Protocol#Run field
// arguments: // arguments:
// * p2p protocol spec // * p2p protocol spec
// * run function taking bzzPeer as argument // * run function taking BzzPeer as argument
// this run function is meant to block for the duration of the protocol session // this run function is meant to block for the duration of the protocol session
// on return the session is terminated and the peer is disconnected // on return the session is terminated and the peer is disconnected
// the protocol waits for the bzz handshake is negotiated // the protocol waits for the bzz handshake is negotiated
// the overlay address on the bzzPeer is set from the remote handshake // the overlay address on the BzzPeer is set from the remote handshake
func (b *Bzz) RunProtocol(spec *protocols.Spec, run func(*bzzPeer) error) func(*p2p.Peer, p2p.MsgReadWriter) error { func (b *Bzz) RunProtocol(spec *protocols.Spec, run func(*BzzPeer) error) func(*p2p.Peer, p2p.MsgReadWriter) error {
return func(p *p2p.Peer, rw p2p.MsgReadWriter) error { return func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
// wait for the bzz protocol to perform the handshake // wait for the bzz protocol to perform the handshake
handshake, _ := b.GetHandshake(p.ID()) handshake, _ := b.GetHandshake(p.ID())
@ -206,8 +193,8 @@ func (b *Bzz) RunProtocol(spec *protocols.Spec, run func(*bzzPeer) error) func(*
if handshake.err != nil { if handshake.err != nil {
return fmt.Errorf("%08x: %s protocol closed: %v", b.BaseAddr()[:4], spec.Name, handshake.err) return fmt.Errorf("%08x: %s protocol closed: %v", b.BaseAddr()[:4], spec.Name, handshake.err)
} }
// the handshake has succeeded so construct the bzzPeer and run the protocol // the handshake has succeeded so construct the BzzPeer and run the protocol
peer := &bzzPeer{ peer := &BzzPeer{
Peer: protocols.NewPeer(p, rw, spec), Peer: protocols.NewPeer(p, rw, spec),
localAddr: b.localAddr, localAddr: b.localAddr,
BzzAddr: handshake.peerAddr, BzzAddr: handshake.peerAddr,
@ -257,9 +244,9 @@ func (b *Bzz) runBzz(p *p2p.Peer, rw p2p.MsgReadWriter) error {
return errors.New("received multiple handshakes") return errors.New("received multiple handshakes")
} }
// bzzPeer is the bzz protocol view of a protocols.Peer (itself an extension of p2p.Peer) // BzzPeer is the bzz protocol view of a protocols.Peer (itself an extension of p2p.Peer)
// implements the Peer interface and all interfaces Peer implements: Addr, OverlayPeer // implements the Peer interface and all interfaces Peer implements: Addr, OverlayPeer
type bzzPeer struct { type BzzPeer struct {
*protocols.Peer // represents the connection for online peers *protocols.Peer // represents the connection for online peers
localAddr *BzzAddr // local Peers address localAddr *BzzAddr // local Peers address
*BzzAddr // remote address -> implements Addr interface = protocols.Peer *BzzAddr // remote address -> implements Addr interface = protocols.Peer
@ -267,12 +254,12 @@ type bzzPeer struct {
} }
// Off returns the overlay peer record for offline persistance // Off returns the overlay peer record for offline persistance
func (p *bzzPeer) Off() OverlayAddr { func (p *BzzPeer) Off() OverlayAddr {
return p.BzzAddr return p.BzzAddr
} }
// LastActive returns the time the peer was last active // LastActive returns the time the peer was last active
func (p *bzzPeer) LastActive() time.Time { func (p *BzzPeer) LastActive() time.Time {
return p.lastActive return p.lastActive
} }

8
swarm/network/protocol_test.go
View file

@ -78,16 +78,16 @@ func HandshakeMsgExchange(lhs, rhs *HandshakeMsg, id discover.NodeID) []p2ptest.
} }
} }
func newBzzBaseTester(t *testing.T, n int, addr *BzzAddr, spec *protocols.Spec, run func(*bzzPeer) error) *bzzTester { func newBzzBaseTester(t *testing.T, n int, addr *BzzAddr, spec *protocols.Spec, run func(*BzzPeer) error) *bzzTester {
cs := make(map[string]chan bool) cs := make(map[string]chan bool)
srv := func(p *bzzPeer) error { srv := func(p *BzzPeer) error {
defer close(cs[p.ID().String()]) defer close(cs[p.ID().String()])
return run(p) return run(p)
} }
protocall := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { protocall := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
return srv(&bzzPeer{ return srv(&BzzPeer{
Peer: protocols.NewPeer(p, rw, spec), Peer: protocols.NewPeer(p, rw, spec),
localAddr: addr, localAddr: addr,
BzzAddr: NewAddrFromNodeID(p.ID()), BzzAddr: NewAddrFromNodeID(p.ID()),
@ -115,7 +115,7 @@ type bzzTester struct {
func newBzzTester(t *testing.T, n int, addr *BzzAddr, pp *p2ptest.TestPeerPool, spec *protocols.Spec, services func(Peer) error) *bzzTester { func newBzzTester(t *testing.T, n int, addr *BzzAddr, pp *p2ptest.TestPeerPool, spec *protocols.Spec, services func(Peer) error) *bzzTester {
extraservices := func(p *bzzPeer) error { extraservices := func(p *BzzPeer) error {
pp.Add(p) pp.Add(p)
defer pp.Remove(p) defer pp.Remove(p)
if services == nil { if services == nil {

130
swarm/network/stream/common_test.go Normal file
View file

@ -0,0 +1,130 @@
// 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 stream
import (
"errors"
"flag"
"io"
"io/ioutil"
"os"
"testing"
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/storage"
)
var (
adapter = flag.String("adapter", "sim", "type of simulation: sim|socket|exec|docker")
loglevel = flag.Int("loglevel", 2, "verbosity of logs")
)
var services = adapters.Services{
"delivery": newDeliveryService,
"syncer": newSyncerService,
}
func init() {
flag.Parse()
// register the Delivery service which will run as a devp2p
// protocol when using the exec adapter
adapters.RegisterServices(services)
log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
}
var (
delivery *Delivery
fileHash storage.Key
)
func mustReadAll(dpa *storage.DPA, hash storage.Key) (int, error) {
r := dpa.Retrieve(fileHash)
buf := make([]byte, 1024)
var n, total int
var err error
for (total == 0 || n > 0) && err == nil {
n, err = r.ReadAt(buf, int64(total))
total += n
}
if err != nil && err != io.EOF {
return total, err
}
return total, nil
}
func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *storage.LocalStore, func(), error) {
// setup
addr := network.RandomAddr() // tested peers peer address
to := network.NewKademlia(addr.OAddr, network.NewKadParams())
// temp datadir
datadir, err := ioutil.TempDir("", "streamer")
if err != nil {
return nil, nil, nil, func() {}, err
}
teardown := func() {
os.RemoveAll(datadir)
}
localStore, err := storage.NewTestLocalStoreForAddr(datadir, addr.Over())
if err != nil {
return nil, nil, nil, teardown, err
}
db := storage.NewDBAPI(localStore)
delivery := NewDelivery(to, db)
streamer := NewRegistry(delivery)
run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
BzzPeer := &BzzPeer{
Peer: protocols.NewPeer(p, rw, Spec),
localAddr: addr,
BzzAddr: network.NewAddrFromNodeID(p.ID()),
}
to.On(BzzPeer)
return streamer.Run(BzzPeer)
}
protocolTester := p2ptest.NewProtocolTester(t, network.NewNodeIDFromAddr(addr), 1, run)
err = waitForPeers(streamer, 1*time.Second)
if err != nil {
return nil, nil, nil, nil, errors.New("timeout: peer is not created")
}
return protocolTester, streamer, localStore, teardown, nil
}
func waitForPeers(streamer *Registry, timeout time.Duration) error {
ticker := time.NewTicker(10 * time.Millisecond)
timeoutTimer := time.NewTimer(timeout)
for {
select {
case <-ticker.C:
if len(streamer.peers) > 0 {
return nil
}
case <-timeoutTimer.C:
return errors.New("timeout")
}
}
}

81
swarm/network/requests.go → swarm/network/stream/delivery.go
View file

@ -1,4 +1,4 @@
// Copyright 2016 The go-ethereum Authors // Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library. // This file is part of the go-ethereum library.
// //
// The go-ethereum library is free software: you can redistribute it and/or modify // The go-ethereum library is free software: you can redistribute it and/or modify
@ -14,7 +14,7 @@
// You should have received a copy of the GNU Lesser General Public License // 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/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package network package stream
import ( import (
"errors" "errors"
@ -23,51 +23,52 @@ import (
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/storage" "github.com/ethereum/go-ethereum/swarm/storage"
) )
const retrieveRequestStream = "RETRIEVE_REQUEST" const swarmChunkServerStreamName = "RETRIEVE_REQUEST"
type Delivery struct { type Delivery struct {
dbAccess *DbAccess db *storage.DBAPI
overlay Overlay overlay network.Overlay
receiveC chan *ChunkDeliveryMsg receiveC chan *ChunkDeliveryMsg
getPeer func(discover.NodeID) *StreamerPeer getPeer func(discover.NodeID) *Peer
quit chan struct{} quit chan struct{}
} }
func NewDelivery(overlay Overlay, dbAccess *DbAccess) *Delivery { func NewDelivery(overlay network.Overlay, db *storage.DBAPI) *Delivery {
self := &Delivery{ d := &Delivery{
dbAccess: dbAccess, db: db,
overlay: overlay, overlay: overlay,
receiveC: make(chan *ChunkDeliveryMsg, 10), receiveC: make(chan *ChunkDeliveryMsg, 10),
} }
go self.processReceivedChunks() go d.processReceivedChunks()
return self return d
} }
// RetrieveRequestStreamer implements OutgoingStreamer // SwarmChunkServer implements OutgoingStreamer
type RetrieveRequestStreamer struct { type SwarmChunkServer struct {
deliveryC chan []byte deliveryC chan []byte
batchC chan []byte batchC chan []byte
dbAccess *DbAccess db *storage.DBAPI
currentLen uint64 currentLen uint64
} }
// NewRetrieveRequestStreamer is RetrieveRequestStreamer constructor // NewSwarmChunkServer is SwarmChunkServer constructor
func NewRetrieveRequestStreamer(dbAccess *DbAccess) *RetrieveRequestStreamer { func NewSwarmChunkServer(db *storage.DBAPI) *SwarmChunkServer {
s := &RetrieveRequestStreamer{ s := &SwarmChunkServer{
deliveryC: make(chan []byte), deliveryC: make(chan []byte),
batchC: make(chan []byte), batchC: make(chan []byte),
dbAccess: dbAccess, db: db,
} }
go s.processDeliveries() go s.processDeliveries()
return s return s
} }
// processDeliveries handles delivered chunk hashes // processDeliveries handles delivered chunk hashes
func (s *RetrieveRequestStreamer) processDeliveries() { func (s *SwarmChunkServer) processDeliveries() {
var hashes []byte var hashes []byte
var batchC chan []byte var batchC chan []byte
for { for {
@ -83,7 +84,7 @@ func (s *RetrieveRequestStreamer) processDeliveries() {
} }
// SetNextBatch // SetNextBatch
func (s *RetrieveRequestStreamer) SetNextBatch(_, _ uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error) { func (s *SwarmChunkServer) SetNextBatch(_, _ uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error) {
hashes = <-s.batchC hashes = <-s.batchC
from = s.currentLen from = s.currentLen
s.currentLen += uint64(len(hashes)) s.currentLen += uint64(len(hashes))
@ -92,8 +93,8 @@ func (s *RetrieveRequestStreamer) SetNextBatch(_, _ uint64) (hashes []byte, from
} }
// GetData retrives chunk data from db store // GetData retrives chunk data from db store
func (s *RetrieveRequestStreamer) GetData(key []byte) []byte { func (s *SwarmChunkServer) GetData(key []byte) []byte {
chunk, _ := s.dbAccess.get(storage.Key(key)) chunk, _ := s.db.Get(storage.Key(key))
return chunk.SData return chunk.SData
} }
@ -103,16 +104,16 @@ type RetrieveRequestMsg struct {
SkipCheck bool SkipCheck bool
} }
func (self *Delivery) handleRetrieveRequestMsg(sp *StreamerPeer, req *RetrieveRequestMsg) error { func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) error {
s, err := sp.getOutgoingStreamer(retrieveRequestStream) s, err := sp.getServer(swarmChunkServerStreamName)
if err != nil { if err != nil {
return err return err
} }
streamer := s.OutgoingStreamer.(*RetrieveRequestStreamer) streamer := s.Server.(*SwarmChunkServer)
chunk, created := self.dbAccess.getOrCreateRequest(req.Key) chunk, created := d.db.GetOrCreateRequest(req.Key)
if chunk.ReqC != nil { if chunk.ReqC != nil {
if created { if created {
if err := self.RequestFromPeers(chunk.Key[:], false, sp.ID()); err != nil { if err := d.RequestFromPeers(chunk.Key[:], false, sp.ID()); err != nil {
return nil return nil
} }
} }
@ -122,7 +123,7 @@ func (self *Delivery) handleRetrieveRequestMsg(sp *StreamerPeer, req *RetrieveRe
select { select {
case <-chunk.ReqC: case <-chunk.ReqC:
case <-self.quit: case <-d.quit:
return return
case <-t.C: case <-t.C:
return return
@ -149,21 +150,21 @@ type ChunkDeliveryMsg struct {
SData []byte // the stored chunk Data (incl size) SData []byte // the stored chunk Data (incl size)
} }
func (self *Delivery) handleChunkDeliveryMsg(req *ChunkDeliveryMsg) error { func (d *Delivery) handleChunkDeliveryMsg(req *ChunkDeliveryMsg) error {
chunk, err := self.dbAccess.get(req.Key) chunk, err := d.db.Get(req.Key)
if err != nil { if err != nil {
return err return err
} }
self.receiveC <- req d.receiveC <- req
log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, self)) log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, d))
return nil return nil
} }
func (self *Delivery) processReceivedChunks() { func (d *Delivery) processReceivedChunks() {
for req := range self.receiveC { for req := range d.receiveC {
chunk, err := self.dbAccess.get(req.Key) chunk, err := d.db.Get(req.Key)
if err != nil { if err != nil {
continue continue
} }
@ -171,23 +172,23 @@ func (self *Delivery) processReceivedChunks() {
select { select {
case <-chunk.ReqC: case <-chunk.ReqC:
default: default:
self.dbAccess.put(chunk) d.db.Put(chunk)
close(chunk.ReqC) close(chunk.ReqC)
} }
} }
} }
// RequestFromPeers sends a chunk retrieve request to // RequestFromPeers sends a chunk retrieve request to
func (self *Delivery) RequestFromPeers(hash []byte, skipCheck bool, peersToSkip ...discover.NodeID) error { func (d *Delivery) RequestFromPeers(hash []byte, skipCheck bool, peersToSkip ...discover.NodeID) error {
var success bool var success bool
self.overlay.EachConn(hash, 255, func(p OverlayConn, po int, nn bool) bool { d.overlay.EachConn(hash, 255, func(p network.OverlayConn, po int, nn bool) bool {
spId := p.(Peer).ID() spId := p.(*network.BzzPeer).ID()
for _, p := range peersToSkip { for _, p := range peersToSkip {
if p == spId { if p == spId {
return true return true
} }
} }
sp := self.getPeer(spId) sp := d.getPeer(spId)
// TODO: skip light nodes that do not accept retrieve requests // TODO: skip light nodes that do not accept retrieve requests
err := sp.SendPriority(&RetrieveRequestMsg{ err := sp.SendPriority(&RetrieveRequestMsg{
Key: hash, Key: hash,

16
swarm/network/request_test.go → swarm/network/stream/delivery_test.go
View file

@ -1,4 +1,4 @@
// Copyright 2016 The go-ethereum Authors // Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library. // This file is part of the go-ethereum library.
// //
// The go-ethereum library is free software: you can redistribute it and/or modify // The go-ethereum library is free software: you can redistribute it and/or modify
@ -14,7 +14,7 @@
// You should have received a copy of the GNU Lesser General Public License // 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/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package network package stream
import ( import (
"bytes" "bytes"
@ -405,8 +405,8 @@ func newDeliveryService(ctx *adapters.ServiceContext) (node.Service, error) {
addr := NewAddrFromNodeID(id) addr := NewAddrFromNodeID(id)
kad := NewKademlia(addr.Over(), NewKadParams()) kad := NewKademlia(addr.Over(), NewKadParams())
localStore := localStores[nodeCount] localStore := localStores[nodeCount]
dbAccess := NewDbAccess(localStore.(*storage.LocalStore)) db := NewDBAPI(localStore.(*storage.LocalStore))
streamer := NewStreamer(NewDelivery(kad, dbAccess)) streamer := NewStreamerRegistry(NewDelivery(kad, db))
if nodeCount == 0 { if nodeCount == 0 {
// the delivery service for the pivot node is assigned globally // the delivery service for the pivot node is assigned globally
// so that the simulation action call can use it for the // so that the simulation action call can use it for the
@ -423,13 +423,13 @@ func newDeliveryService(ctx *adapters.ServiceContext) (node.Service, error) {
} }
func (b *testStreamerService) runDelivery(p *p2p.Peer, rw p2p.MsgReadWriter) error { func (b *testStreamerService) runDelivery(p *p2p.Peer, rw p2p.MsgReadWriter) error {
bzzPeer := &bzzPeer{ BzzPeer := &BzzPeer{
Peer: protocols.NewPeer(p, rw, StreamerSpec), Peer: protocols.NewPeer(p, rw, StreamerSpec),
localAddr: b.addr, localAddr: b.addr,
BzzAddr: NewAddrFromNodeID(p.ID()), BzzAddr: NewAddrFromNodeID(p.ID()),
} }
b.streamer.delivery.overlay.On(bzzPeer) b.streamer.delivery.overlay.On(BzzPeer)
defer b.streamer.delivery.overlay.Off(bzzPeer) defer b.streamer.delivery.overlay.Off(BzzPeer)
go func() { go func() {
// each node Subscribes to each other's retrieveRequestStream // each node Subscribes to each other's retrieveRequestStream
// need to wait till an aynchronous process registers the peers in streamer.peers // need to wait till an aynchronous process registers the peers in streamer.peers
@ -440,5 +440,5 @@ func (b *testStreamerService) runDelivery(p *p2p.Peer, rw p2p.MsgReadWriter) err
log.Warn("error in subscribe", "err", err) log.Warn("error in subscribe", "err", err)
} }
}() }()
return b.streamer.Run(bzzPeer) return b.streamer.Run(BzzPeer)
} }

228
swarm/network/stream/messages.go Normal file
View file

@ -0,0 +1,228 @@
// 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 stream
import (
"errors"
"fmt"
"sync"
"github.com/ethereum/go-ethereum/log"
bv "github.com/ethereum/go-ethereum/swarm/network/bitvector"
"github.com/ethereum/go-ethereum/swarm/storage"
)
// Handover represents a statement that the upstream peer hands over the stream section
type Handover struct {
Stream string // name of stream
Start, End uint64 // index of hashes
Root []byte // Root hash for indexed segment inclusion proofs
}
// HandoverProof represents a signed statement that the upstream peer handed over the stream section
type HandoverProof struct {
Sig []byte // Sign(Hash(Serialisation(Handover)))
*Handover
}
// Takeover represents a statement that downstream peer took over (stored all data)
// handed over
type Takeover Handover
// TakeoverProof represents a signed statement that the downstream peer took over
// the stream section
type TakeoverProof struct {
Sig []byte // Sign(Hash(Serialisation(Takeover)))
*Takeover
}
// TakeoverProofMsg is the protocol msg sent by downstream peer
type TakeoverProofMsg TakeoverProof
// String pretty prints TakeoverProofMsg
func (m TakeoverProofMsg) String() string {
return fmt.Sprintf("Stream: '%v' [%v-%v], Root: %x, Sig: %x", m.Stream, m.Start, m.End, m.Root, m.Sig)
}
// SubcribeMsg is the protocol msg for requesting a stream(section)
type SubscribeMsg struct {
Stream string
Key []byte
From, To uint64
Priority uint8 // delivered on priority channel
}
func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) error {
f, err := p.streamer.GetServerFunc(req.Stream)
if err != nil {
return err
}
s, err := f(p, req.Key)
if err != nil {
return err
}
os, err := p.setServer(req.Stream, req.Key, s, req.Priority)
if err != nil {
return nil
}
log.Debug("received subscription", "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
go p.SendOfferedHashes(os, req.From, req.To)
return nil
}
// OfferedHashesMsg is the protocol msg for offering to hand over a
// stream section
type OfferedHashesMsg struct {
Stream string // name of Stream
Key []byte // subtype or key
From, To uint64 // peer and db-specific entry count
Hashes []byte // stream of hashes (128)
*HandoverProof // HandoverProof
}
// String pretty prints OfferedHashesMsg
func (m OfferedHashesMsg) String() string {
return fmt.Sprintf("Stream '%v' [%v-%v] (%v)", m.Stream, m.From, m.To, len(m.Hashes)/HashSize)
}
// handleOfferedHashesMsg protocol msg handler calls the incoming streamer interface
// Filter method
func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
sk := req.Stream
sk += keyToString(req.Key)
s, err := p.getClient(sk)
if err != nil {
return err
}
hashes := req.Hashes
want, err := bv.New(len(hashes) / HashSize)
if err != nil {
return fmt.Errorf("error initiaising bitvector of length %v: %v", len(hashes)/HashSize, err)
}
wg := sync.WaitGroup{}
for i := 0; i < len(hashes); i += HashSize {
hash := hashes[i : i+HashSize]
if wait := s.NeedData(hash); wait != nil {
want.Set(i/HashSize, true)
wg.Add(1)
// create request and wait until the chunk data arrives and is stored
go func(w func()) {
w()
wg.Done()
}(wait)
}
}
go func() {
wg.Wait()
if tf := s.BatchDone(req.Stream, req.From, hashes, req.Root); tf != nil {
tp, err := tf()
if err != nil {
return
}
p.SendPriority(tp, s.priority)
}
s.next <- struct{}{}
}()
// only send wantedKeysMsg if all missing chunks of the previous batch arrived
// except
if s.live {
s.sessionAt = req.From
}
from, to := s.nextBatch(req.To)
log.Debug("received batch", "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
if from == to {
return nil
}
msg := &WantedHashesMsg{
Stream: req.Stream,
Key: req.Key,
Want: want.Bytes(),
From: from,
To: to,
}
go func() {
select {
case <-s.next:
case <-s.quit:
return
}
log.Debug("want batch", "stream", msg.Stream, "Key", msg.Key, "from", msg.From, "to", msg.To)
p.SendPriority(msg, s.priority)
}()
return nil
}
// WantedHashesMsg is the protocol msg data for signaling which hashes
// offered in OfferedHashesMsg downstream peer actually wants sent over
type WantedHashesMsg struct {
Stream string // name of stream
Key []byte // subtype or key
Want []byte // bitvector indicating which keys of the batch needed
From, To uint64 // next interval offset - empty if not to be continued
}
// String pretty prints WantedHashesMsg
func (m WantedHashesMsg) String() string {
return fmt.Sprintf("Stream '%v', Want: %x, Next: [%v-%v]", m.Stream, m.Want, m.From, m.To)
}
// handleWantedHashesMsg protocol msg handler
// * sends the next batch of unsynced keys
// * sends the actual data chunks as per WantedHashesMsg
func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error {
log.Debug("received wanted batch", "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
s, err := p.getServer(req.Stream + keyToString(req.Key))
if err != nil {
log.Debug(err.Error())
return err
}
hashes := s.currentBatch
// launch in go routine since GetBatch blocks until new hashes arrive
go p.SendOfferedHashes(s, req.From, req.To)
l := len(hashes) / HashSize
want, err := bv.NewFromBytes(req.Want, l)
if err != nil {
return fmt.Errorf("error initiaising bitvector of length %v: %v", l, err)
}
for i := 0; i < l; i++ {
if want.Get(i) {
hash := hashes[i*HashSize : (i+1)*HashSize]
data := s.GetData(hash)
if data == nil {
return errors.New("not found")
}
chunk := storage.NewChunk(hash, nil)
chunk.SData = data
if err := p.Deliver(chunk, s.priority); err != nil {
return err
}
}
}
return nil
}
func (p *Peer) handleTakeoverProofMsg(req *TakeoverProofMsg) error {
_, err := p.getServer(req.Stream)
if err != nil {
return err
}
// store the strongest takeoverproof for the stream in streamer
return nil
}
type UnsubscribeMsg struct{}

164
swarm/network/stream/peer.go Normal file
View file

@ -0,0 +1,164 @@
// 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 stream
import (
"context"
"fmt"
"sync"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/protocols"
pq "github.com/ethereum/go-ethereum/swarm/network/priorityqueue"
"github.com/ethereum/go-ethereum/swarm/storage"
)
// Peer is the Peer extention for the streaming protocol
type Peer struct {
*protocols.Peer
streamer *Registry
pq *pq.PriorityQueue
outgoingMu sync.RWMutex
incomingMu sync.RWMutex
servers map[string]*server
clients map[string]*client
quit chan struct{}
}
// NewPeer is the constructor for Peer
func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer {
p := &Peer{
Peer: peer,
pq: pq.New(int(PriorityQueue), PriorityQueueCap),
streamer: streamer,
servers: make(map[string]*server),
clients: make(map[string]*client),
quit: make(chan struct{}),
}
ctx, cancel := context.WithCancel(context.Background())
go p.pq.Run(ctx, func(i interface{}) { p.Send(i) })
go func() {
<-p.quit
cancel()
}()
return p
}
// Deliver sends a storeRequestMsg protocol message to the peer
func (p *Peer) Deliver(chunk *storage.Chunk, priority uint8) error {
msg := &ChunkDeliveryMsg{
Key: chunk.Key,
SData: chunk.SData,
}
return p.pq.Push(nil, msg, int(priority))
}
// Deliver sends a storeRequestMsg protocol message to the peer
func (p *Peer) SendPriority(msg interface{}, priority uint8) error {
return p.pq.Push(nil, msg, int(priority))
}
// SendOfferedHashes sends OfferedHashesMsg protocol msg
func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error {
hashes, from, to, proof, err := s.SetNextBatch(f, t)
if err != nil {
return err
}
if proof == nil {
proof = &HandoverProof{
Handover: &Handover{},
}
}
s.currentBatch = hashes
msg := &OfferedHashesMsg{
HandoverProof: proof,
Hashes: hashes,
From: from,
To: to,
Stream: s.stream,
Key: s.key,
}
log.Debug("Swarm syncer offer batch", "stream", s.stream, "key", s.key, "len", len(hashes), "from", from, "to", to)
return p.SendPriority(msg, s.priority)
}
func (p *Peer) getServer(s string) (*server, error) {
p.outgoingMu.RLock()
defer p.outgoingMu.RUnlock()
server := p.servers[s]
if server == nil {
return nil, fmt.Errorf("server '%v' not provided to peer %v", s, p.ID())
}
return server, nil
}
func (p *Peer) getClient(s string) (*client, error) {
p.incomingMu.RLock()
defer p.incomingMu.RUnlock()
client := p.clients[s]
if client == nil {
return nil, fmt.Errorf("client '%v' not provided to peer %v", s, p.ID())
}
return client, nil
}
func (p *Peer) setServer(s string, key []byte, o Server, priority uint8) (*server, error) {
p.outgoingMu.Lock()
defer p.outgoingMu.Unlock()
sk := s + keyToString(key)
if p.servers[sk] != nil {
return nil, fmt.Errorf("server %v already registered", sk)
}
os := &server{
Server: o,
priority: priority,
stream: s,
key: key,
}
p.servers[sk] = os
return os, nil
}
func (p *Peer) setClient(s string, key []byte, i Client, priority uint8, live bool) error {
p.incomingMu.Lock()
defer p.incomingMu.Unlock()
sk := s + keyToString(key)
if p.clients[sk] != nil {
return fmt.Errorf("client %v already registered", sk)
}
next := make(chan struct{}, 1)
// var intervals *Intervals
// if !live {
// key := s + p.ID().String()
// intervals = NewIntervals(key, p.streamer)
// }
p.clients[sk] = &client{
Client: i,
// intervals: intervals,
live: live,
priority: priority,
next: next,
stream: s,
key: key,
}
next <- struct{}{} // this is to allow wantedKeysMsg before first batch arrives
return nil
}

316
swarm/network/stream/stream.go Normal file
View file

@ -0,0 +1,316 @@
// 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 stream
import (
"fmt"
"math"
"sync"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/swarm/storage"
)
const (
Low uint8 = iota
Mid
High
Top
PriorityQueue // number of queues
PriorityQueueCap = 3 // queue capacity
HashSize = 32
)
// Registry registry for outgoing and incoming streamer constructors
type Registry struct {
clientMu sync.RWMutex
serverMu sync.RWMutex
peersMu sync.RWMutex
serverFuncs map[string]func(*Peer, []byte) (Server, error)
clientFuncs map[string]func(*Peer, []byte) (Client, error)
peers map[discover.NodeID]*Peer
delivery *Delivery
}
// NewRegistry is Streamer constructor
func NewRegistry(delivery *Delivery) *Registry {
streamer := &Registry{
serverFuncs: make(map[string]func(*Peer, []byte) (Server, error)),
clientFuncs: make(map[string]func(*Peer, []byte) (Client, error)),
peers: make(map[discover.NodeID]*Peer),
delivery: delivery,
}
delivery.getPeer = streamer.getPeer
streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, t []byte) (Server, error) {
return NewSwarmChunkServer(delivery.db), nil
})
streamer.RegisterClientFunc(swarmChunkServerStreamName, func(p *Peer, t []byte) (Client, error) {
return NewSwarmSyncerClient(p, delivery.db, nil)
})
return streamer
}
// RegisterClient registers an incoming streamer constructor
func (r *Registry) RegisterClientFunc(stream string, f func(*Peer, []byte) (Client, error)) {
r.clientMu.Lock()
defer r.clientMu.Unlock()
r.clientFuncs[stream] = f
}
// RegisterServer registers an outgoing streamer constructor
func (r *Registry) RegisterServerFunc(stream string, f func(*Peer, []byte) (Server, error)) {
r.serverMu.Lock()
defer r.serverMu.Unlock()
r.serverFuncs[stream] = f
}
// GetClient accessor for incoming streamer constructors
func (r *Registry) GetClientFunc(stream string) (func(*Peer, []byte) (Client, error), error) {
r.clientMu.RLock()
defer r.clientMu.RUnlock()
f := r.clientFuncs[stream]
if f == nil {
return nil, fmt.Errorf("stream %v not registered", stream)
}
return f, nil
}
// GetServer accessor for incoming streamer constructors
func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte) (Server, error), error) {
r.serverMu.RLock()
defer r.serverMu.RUnlock()
f := r.serverFuncs[stream]
if f == nil {
return nil, fmt.Errorf("stream %v not registered", stream)
}
return f, nil
}
// Subscribe initiates the streamer
func (r *Registry) Subscribe(peerId discover.NodeID, s string, t []byte, from, to uint64, priority uint8, live bool) error {
f, err := r.GetClientFunc(s)
if err != nil {
return err
}
peer := r.getPeer(peerId)
if peer == nil {
return fmt.Errorf("peer not found %v", peerId)
}
is, err := f(peer, t)
if err != nil {
return err
}
err = peer.setClient(s, t, is, priority, live)
if err != nil {
return err
}
msg := &SubscribeMsg{
Stream: s,
Key: t,
// Live: live,
From: from,
To: to,
Priority: priority,
}
log.Debug("Subscribe ", "peer", peerId, "stream", s, "key", t, "from", from, "to", to)
peer.SendPriority(msg, priority)
return nil
}
func (r *Registry) Retrieve(chunk *storage.Chunk) error {
return r.delivery.RequestFromPeers(chunk.Key[:], false)
}
func (r *Registry) NodeInfo() interface{} {
return nil
}
func (r *Registry) PeerInfo(id discover.NodeID) interface{} {
return nil
}
func (r *Registry) getPeer(peerId discover.NodeID) *Peer {
r.peersMu.RLock()
defer r.peersMu.RUnlock()
return r.peers[peerId]
}
func (r *Registry) setPeer(peer *Peer) {
r.peersMu.Lock()
r.peers[peer.ID()] = peer
r.peersMu.Unlock()
}
func (r *Registry) deletePeer(peer *Peer) {
r.peersMu.Lock()
delete(r.peers, peer.ID())
r.peersMu.Unlock()
}
// Run protocol run function
func (r *Registry) Run(p *protocols.Peer) error {
sp := NewPeer(p, r)
// load saved intervals
r.setPeer(sp)
defer r.deletePeer(sp)
defer close(sp.quit)
return sp.Run(sp.HandleMsg)
}
// HandleMsg is the message handler that delegates incoming messages
func (p *Peer) HandleMsg(msg interface{}) error {
switch msg := msg.(type) {
case *SubscribeMsg:
return p.handleSubscribeMsg(msg)
case *OfferedHashesMsg:
return p.handleOfferedHashesMsg(msg)
case *TakeoverProofMsg:
return p.handleTakeoverProofMsg(msg)
case *WantedHashesMsg:
return p.handleWantedHashesMsg(msg)
case *ChunkDeliveryMsg:
return p.streamer.delivery.handleChunkDeliveryMsg(msg)
case *RetrieveRequestMsg:
return p.streamer.delivery.handleRetrieveRequestMsg(p, msg)
default:
return fmt.Errorf("unknown message type: %T", msg)
}
}
func keyToString(key []byte) string {
l := len(key)
if l == 0 {
return ""
}
return fmt.Sprintf("%s-%d", string(key[:l-1]), uint8(key[l-1]))
}
type server struct {
Server
priority uint8
currentBatch []byte
stream string
key []byte
}
// Server interface for outgoing peer Streamer
type Server interface {
SetNextBatch(uint64, uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error)
GetData([]byte) []byte
}
type client struct {
Client
priority uint8
sessionAt uint64
live bool
stream string
key []byte
quit chan struct{}
next chan struct{}
}
// Client interface for incoming peer Streamer
type Client interface {
NeedData([]byte) func()
BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error)
}
// NextBatch adjusts the indexes by inspecting the intervals
func (c *client) nextBatch(from uint64) (nextFrom uint64, nextTo uint64) {
var intervals []uint64
if c.live {
if len(intervals) == 0 {
intervals = []uint64{c.sessionAt, from}
} else {
intervals[1] = from
}
nextFrom = from
} else if from >= c.sessionAt { // history sync complete
intervals = nil
nextFrom = from
nextTo = math.MaxUint64
} else if len(intervals) > 2 && from >= intervals[2] { // filled a gap in the intervals
intervals = append(intervals[:1], intervals[3:]...)
nextFrom = intervals[1]
if len(intervals) > 2 {
nextTo = intervals[2]
} else {
nextTo = c.sessionAt
}
} else {
nextFrom = from
intervals[1] = from
nextTo = c.sessionAt
}
// b.intervals.set(intervals)
return nextFrom, nextTo
}
// Spec is the spec of the streamer protocol.
var Spec = &protocols.Spec{
Name: "stream",
Version: 1,
MaxMsgSize: 10 * 1024 * 1024,
Messages: []interface{}{
UnsubscribeMsg{},
OfferedHashesMsg{},
WantedHashesMsg{},
TakeoverProofMsg{},
SubscribeMsg{},
RetrieveRequestMsg{},
ChunkDeliveryMsg{},
},
}
func (r *Registry) Protocols() []p2p.Protocol {
return []p2p.Protocol{
{
Name: Spec.Name,
Version: Spec.Version,
Length: Spec.Length(),
Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
peer := protocols.NewPeer(p, rw, Spec)
return r.Run(peer)
},
NodeInfo: r.NodeInfo,
PeerInfo: r.PeerInfo,
},
}
}

10
swarm/network/streamer_test.go → swarm/network/stream/streamer_test.go
View file

@ -1,4 +1,4 @@
// Copyright 2016 The go-ethereum Authors // Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library. // This file is part of the go-ethereum library.
// //
// The go-ethereum library is free software: you can redistribute it and/or modify // The go-ethereum library is free software: you can redistribute it and/or modify
@ -14,7 +14,7 @@
// You should have received a copy of the GNU Lesser General Public License // 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/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package network package stream
import ( import (
"bytes" "bytes"
@ -92,7 +92,7 @@ func TestStreamerDownstreamSubscribeMsgExchange(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
streamer.RegisterIncomingStreamer("foo", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) { streamer.RegisterClientFunc("foo", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
return &testIncomingStreamer{ return &testIncomingStreamer{
t: t, t: t,
}, nil }, nil
@ -134,7 +134,7 @@ func TestStreamerUpstreamSubscribeMsgExchange(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
streamer.RegisterOutgoingStreamer("foo", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) { streamer.RegisterServerFunc("foo", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
return &testOutgoingStreamer{ return &testOutgoingStreamer{
t: t, t: t,
}, nil }, nil
@ -188,7 +188,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
streamer.RegisterIncomingStreamer("foo", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) { streamer.RegisterClientFunc("foo", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
return &testIncomingStreamer{ return &testIncomingStreamer{
t: t, t: t,
}, nil }, nil

151
swarm/network/syncer.go → swarm/network/stream/syncer.go
View file

@ -1,4 +1,4 @@
// Copyright 2016 The go-ethereum Authors // Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library. // This file is part of the go-ethereum library.
// //
// The go-ethereum library is free software: you can redistribute it and/or modify // The go-ethereum library is free software: you can redistribute it and/or modify
@ -14,7 +14,7 @@
// You should have received a copy of the GNU Lesser General Public License // 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/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package network package stream
import ( import (
"bytes" "bytes"
@ -29,88 +29,52 @@ import (
) )
const ( const (
batchSize = 2 BatchSize = 2
// batchSize = 128 // BatchSize = 128
) )
// wrapper of db-s to provide mockable custom local chunk store access to syncer // SwarmSyncerServer implements an OutgoingStreamer for history syncing on bins
type DbAccess struct {
db *storage.DbStore
loc *storage.LocalStore
}
func NewDbAccess(loc *storage.LocalStore) *DbAccess {
return &DbAccess{loc.DbStore.(*storage.DbStore), loc}
}
// to obtain the chunks from key or request db entry only
func (self *DbAccess) get(key storage.Key) (*storage.Chunk, error) {
return self.loc.Get(key)
}
// current storage counter of chunk db
func (self *DbAccess) currentBucketStorageIndex(po uint8) uint64 {
return self.db.CurrentBucketStorageIndex(po)
}
// iteration storage counter and proximity order
func (self *DbAccess) iterator(from uint64, to uint64, po uint8, f func(storage.Key, uint64) bool) error {
return self.db.SyncIterator(from, to, po, f)
}
// to obtain the chunks from key or request db entry only
func (self *DbAccess) getOrCreateRequest(key storage.Key) (*storage.Chunk, bool) {
return self.loc.GetOrCreateRequest(key)
}
// to obtain the chunks from key or request db entry only
func (self *DbAccess) put(chunk *storage.Chunk) {
self.loc.Put(chunk)
}
// OutgoingSwarmSyncer implements an OutgoingStreamer for history syncing on bins
// offered streams: // offered streams:
// * live request delivery with or without checkback // * live request delivery with or without checkback
// * (live/non-live historical) chunk syncing per proximity bin // * (live/non-live historical) chunk syncing per proximity bin
type OutgoingSwarmSyncer struct { type SwarmSyncerServer struct {
po uint8 po uint8
db *DbAccess db *storage.DBAPI
sessionAt uint64 sessionAt uint64
start uint64 start uint64
} }
// NewOutgoingSwarmSyncer is contructor for OutgoingSwarmSyncer // NewSwarmSyncerServer is contructor for SwarmSyncerServer
func NewOutgoingSwarmSyncer(live bool, po uint8, db *DbAccess) (*OutgoingSwarmSyncer, error) { func NewSwarmSyncerServer(live bool, po uint8, db *storage.DBAPI) (*SwarmSyncerServer, error) {
sessionAt := db.currentBucketStorageIndex(po) sessionAt := db.CurrentBucketStorageIndex(po)
var start uint64 var start uint64
if live { if live {
start = sessionAt start = sessionAt
} }
self := &OutgoingSwarmSyncer{ return &SwarmSyncerServer{
po: po, po: po,
db: db, db: db,
sessionAt: sessionAt, sessionAt: sessionAt,
start: start, start: start,
} }, nil
return self, nil
} }
const maxPO = 32 const maxPO = 32
func RegisterOutgoingSyncer(streamer *Streamer, db *DbAccess) { func RegisterSwarmSyncerServer(streamer *Registry, db *storage.DBAPI) {
streamer.RegisterOutgoingStreamer("SYNC", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) { streamer.RegisterServerFunc("SYNC", func(p *Peer, t []byte) (Server, error) {
po := uint8(t[0]) po := uint8(t[0])
// TODO: make this work for HISTORY too // TODO: make this work for HISTORY too
return NewOutgoingSwarmSyncer(false, po, db) return NewSwarmSyncerServer(false, po, db)
}) })
// streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) { // streamer.RegisterOutgoingStreamer(stream, func(p *Peer) (OutgoingStreamer, error) {
// return NewOutgoingProvableSwarmSyncer(po, db) // return NewOutgoingProvableSwarmSyncer(po, db)
// }) // })
} }
// GetSection retrieves the actual chunk from localstore // GetSection retrieves the actual chunk from localstore
func (self *OutgoingSwarmSyncer) GetData(key []byte) []byte { func (s *SwarmSyncerServer) GetData(key []byte) []byte {
chunk, err := self.db.get(storage.Key(key)) chunk, err := s.db.Get(storage.Key(key))
if err != nil { if err != nil {
return nil return nil
} }
@ -118,23 +82,23 @@ func (self *OutgoingSwarmSyncer) GetData(key []byte) []byte {
} }
// GetBatch retrieves the next batch of hashes from the dbstore // GetBatch retrieves the next batch of hashes from the dbstore
func (self *OutgoingSwarmSyncer) SetNextBatch(from, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) { func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) {
var batch []byte var batch []byte
i := 0 i := 0
if from == 0 { if from == 0 {
from = self.start from = s.start
} }
if to <= from || from >= self.sessionAt { if to <= from || from >= s.sessionAt {
to = math.MaxUint64 to = math.MaxUint64
} }
ticker := time.NewTicker(10 * time.Millisecond) ticker := time.NewTicker(10 * time.Millisecond)
defer ticker.Stop() defer ticker.Stop()
for range ticker.C { for range ticker.C {
err := self.db.iterator(from, to, self.po, func(key storage.Key, idx uint64) bool { err := s.db.Iterator(from, to, s.po, func(key storage.Key, idx uint64) bool {
batch = append(batch, key[:]...) batch = append(batch, key[:]...)
i++ i++
to = idx to = idx
return i < batchSize return i < BatchSize
}) })
if err != nil { if err != nil {
return nil, 0, 0, nil, err return nil, 0, 0, nil, err
@ -144,41 +108,40 @@ func (self *OutgoingSwarmSyncer) SetNextBatch(from, to uint64) ([]byte, uint64,
} }
} }
log.Debug("Swarm syncer offer batch", "po", self.po, "len", i, "from", from, "to", to, "current store count", self.db.currentBucketStorageIndex(self.po)) log.Debug("Swarm syncer offer batch", "po", s.po, "len", i, "from", from, "to", to, "current store count", s.db.CurrentBucketStorageIndex(s.po))
return batch, from, to + 1, nil, nil return batch, from, to + 1, nil, nil
} }
// IncomingSwarmSyncer // SwarmSyncerClient
type IncomingSwarmSyncer struct { type SwarmSyncerClient struct {
sessionAt uint64 sessionAt uint64
nextC chan struct{} nextC chan struct{}
sessionRoot storage.Key sessionRoot storage.Key
sessionReader storage.LazySectionReader sessionReader storage.LazySectionReader
retrieveC chan *storage.Chunk retrieveC chan *storage.Chunk
storeC chan *storage.Chunk storeC chan *storage.Chunk
dbAccess *DbAccess db *storage.DBAPI
chunker storage.Chunker chunker storage.Chunker
currentRoot storage.Key currentRoot storage.Key
requestFunc func(chunk *storage.Chunk) requestFunc func(chunk *storage.Chunk)
end, start uint64 end, start uint64
} }
// NewIncomingSwarmSyncer is a contructor for provable data exchange syncer // NewSwarmSyncerClient is a contructor for provable data exchange syncer
func NewIncomingSwarmSyncer(p Peer, dbAccess *DbAccess, chunker storage.Chunker) (*IncomingSwarmSyncer, error) { func NewSwarmSyncerClient(_ *Peer, db *storage.DBAPI, chunker storage.Chunker) (*SwarmSyncerClient, error) {
self := &IncomingSwarmSyncer{ return &SwarmSyncerClient{
dbAccess: dbAccess, db: db,
chunker: chunker, chunker: chunker,
} }, nil
return self, nil
} }
// // NewIncomingProvableSwarmSyncer is a contructor for provable data exchange syncer // // NewIncomingProvableSwarmSyncer is a contructor for provable data exchange syncer
// func NewIncomingProvableSwarmSyncer(po int, priority int, index uint64, sessionAt uint64, intervals []uint64, sessionRoot storage.Key, chunker *storage.PyramidChunker, store storage.ChunkStore, p Peer) *IncomingSwarmSyncer { // func NewIncomingProvableSwarmSyncer(po int, priority int, index uint64, sessionAt uint64, intervals []uint64, sessionRoot storage.Key, chunker *storage.PyramidChunker, store storage.ChunkStore, p Peer) *SwarmSyncerClient {
// retrieveC := make(storage.Chunk, chunksCap) // retrieveC := make(storage.Chunk, chunksCap)
// RunChunkRequestor(p, retrieveC) // RunChunkRequestor(p, retrieveC)
// storeC := make(storage.Chunk, chunksCap) // storeC := make(storage.Chunk, chunksCap)
// RunChunkStorer(store, storeC) // RunChunkStorer(store, storeC)
// self := &IncomingSwarmSyncer{ // s := &SwarmSyncerClient{
// po: po, // po: po,
// priority: priority, // priority: priority,
// sessionAt: sessionAt, // sessionAt: sessionAt,
@ -191,10 +154,10 @@ func NewIncomingSwarmSyncer(p Peer, dbAccess *DbAccess, chunker storage.Chunker)
// retrieveC: retrieveC, // retrieveC: retrieveC,
// storeC: storeC, // storeC: storeC,
// } // }
// return self // return s
// } // }
// // StartSyncing is called on the StreamerPeer to start the syncing process // // StartSyncing is called on the Peer to start the syncing process
// // the idea is that it is called only after kademlia is close to healthy // // the idea is that it is called only after kademlia is close to healthy
// func StartSyncing(s *Streamer, peerId discover.NodeID, po uint8, nn bool) { // func StartSyncing(s *Streamer, peerId discover.NodeID, po uint8, nn bool) {
// lastPO := po // lastPO := po
@ -208,15 +171,15 @@ func NewIncomingSwarmSyncer(p Peer, dbAccess *DbAccess, chunker storage.Chunker)
// } // }
// } // }
func RegisterIncomingSyncer(streamer *Streamer, db *DbAccess) { func RegisterSwarmSyncerClient(streamer *Registry, db *storage.DBAPI) {
streamer.RegisterIncomingStreamer("SYNC", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) { streamer.RegisterClientFunc("SYNC", func(p *Peer, t []byte) (Client, error) {
return NewIncomingSwarmSyncer(p, db, nil) return NewSwarmSyncerClient(p, db, nil)
}) })
} }
// NeedData // NeedData
func (self *IncomingSwarmSyncer) NeedData(key []byte) (wait func()) { func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) {
chunk, _ := self.dbAccess.getOrCreateRequest(key) chunk, _ := s.db.GetOrCreateRequest(key)
// TODO: we may want to request from this peer anyway even if the request exists // TODO: we may want to request from this peer anyway even if the request exists
if chunk.ReqC == nil { if chunk.ReqC == nil {
return nil return nil
@ -226,29 +189,29 @@ func (self *IncomingSwarmSyncer) NeedData(key []byte) (wait func()) {
} }
// BatchDone // BatchDone
func (self *IncomingSwarmSyncer) BatchDone(s string, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) { func (s *SwarmSyncerClient) BatchDone(streamName string, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) {
if self.chunker != nil { if s.chunker != nil {
return func() (*TakeoverProof, error) { return self.TakeoverProof(s, from, hashes, root) } return func() (*TakeoverProof, error) { return s.TakeoverProof(streamName, from, hashes, root) }
} }
return nil return nil
} }
func (self *IncomingSwarmSyncer) TakeoverProof(s string, from uint64, hashes []byte, root storage.Key) (*TakeoverProof, error) { func (s *SwarmSyncerClient) TakeoverProof(streamName string, from uint64, hashes []byte, root storage.Key) (*TakeoverProof, error) {
// for provable syncer currentRoot is non-zero length // for provable syncer currentRoot is non-zero length
if self.chunker != nil { if s.chunker != nil {
if from > self.sessionAt { // for live syncing currentRoot is always updated if from > s.sessionAt { // for live syncing currentRoot is always updated
//expRoot, err := self.chunker.Append(self.currentRoot, bytes.NewReader(hashes), self.retrieveC, self.storeC) //expRoot, err := s.chunker.Append(s.currentRoot, bytes.NewReader(hashes), s.retrieveC, s.storeC)
expRoot, _, err := self.chunker.Append(self.currentRoot, bytes.NewReader(hashes), self.retrieveC) expRoot, _, err := s.chunker.Append(s.currentRoot, bytes.NewReader(hashes), s.retrieveC)
if err != nil { if err != nil {
return nil, err return nil, err
} }
if !bytes.Equal(root, expRoot) { if !bytes.Equal(root, expRoot) {
return nil, fmt.Errorf("HandoverProof mismatch") return nil, fmt.Errorf("HandoverProof mismatch")
} }
self.currentRoot = root s.currentRoot = root
} else { } else {
expHashes := make([]byte, len(hashes)) expHashes := make([]byte, len(hashes))
_, err := self.sessionReader.ReadAt(expHashes, int64(self.end*HashSize)) _, err := s.sessionReader.ReadAt(expHashes, int64(s.end*HashSize))
if err != nil && err != io.EOF { if err != nil && err != io.EOF {
return nil, err return nil, err
} }
@ -258,12 +221,12 @@ func (self *IncomingSwarmSyncer) TakeoverProof(s string, from uint64, hashes []b
} }
return nil, nil return nil, nil
} }
self.end += uint64(len(hashes)) / HashSize s.end += uint64(len(hashes)) / HashSize
takeover := &Takeover{ takeover := &Takeover{
Stream: s, Stream: streamName,
// Key: self.Key, // Key: s.Key,
Start: self.start, Start: s.start,
End: self.end, End: s.end,
Root: root, Root: root,
} }
// serialise and sign // serialise and sign

37
swarm/network/syncer_test.go → swarm/network/stream/syncer_test.go
View file

@ -1,4 +1,4 @@
// Copyright 2016 The go-ethereum Authors // Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library. // This file is part of the go-ethereum library.
// //
// The go-ethereum library is free software: you can redistribute it and/or modify // The go-ethereum library is free software: you can redistribute it and/or modify
@ -14,7 +14,7 @@
// You should have received a copy of the GNU Lesser General Public License // 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/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package network package stream
import ( import (
"context" "context"
@ -32,6 +32,7 @@ import (
"github.com/ethereum/go-ethereum/p2p/protocols" "github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/p2p/simulations" "github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/storage" "github.com/ethereum/go-ethereum/swarm/storage"
) )
@ -74,10 +75,10 @@ func testSyncBetweenNodes(nodes, conns, size int, skipCheck bool, po uint8) func
} }
check := func(net *simulations.Network, dpa *storage.DPA) func(ctx context.Context, id discover.NodeID) (bool, error) { check := func(net *simulations.Network, dpa *storage.DPA) func(ctx context.Context, id discover.NodeID) (bool, error) {
dbAccesses := make([]*DbAccess, nodes) dbs := make([]*storage.DBAPI, nodes)
for i := 0; i < nodes; i++ { for i := 0; i < nodes; i++ {
dbAccesses[i] = NewDbAccess(localStores[i].(*storage.LocalStore)) dbs[i] = NewDbAccess(localStores[i].(*storage.LocalStore))
} }
return func(ctx context.Context, id discover.NodeID) (bool, error) { return func(ctx context.Context, id discover.NodeID) (bool, error) {
if id != net.Nodes[0].ID() { if id != net.Nodes[0].ID() {
@ -91,8 +92,8 @@ func testSyncBetweenNodes(nodes, conns, size int, skipCheck bool, po uint8) func
var found, total int var found, total int
for i := 1; i < nodes; i++ { for i := 1; i < nodes; i++ {
dbAccesses[i].iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool { dbs[i].iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool {
_, err := dbAccesses[0].get(key) _, err := dbs[0].get(key)
if err == nil { if err == nil {
found++ found++
} }
@ -105,7 +106,7 @@ func testSyncBetweenNodes(nodes, conns, size int, skipCheck bool, po uint8) func
} }
} }
toAddr := func(id discover.NodeID) *BzzAddr { toAddr := func(id discover.NodeID) *BzzAddr {
addr := NewAddrFromNodeID(id) addr := network.NewAddrFromNodeID(id)
addr.OAddr[0] = byte(0) addr.OAddr[0] = byte(0)
return addr return addr
} }
@ -123,15 +124,15 @@ func testSyncBetweenNodes(nodes, conns, size int, skipCheck bool, po uint8) func
func newSyncerService(ctx *adapters.ServiceContext) (node.Service, error) { func newSyncerService(ctx *adapters.ServiceContext) (node.Service, error) {
id := ctx.Config.ID id := ctx.Config.ID
addr := NewAddrFromNodeID(id) addr := network.NewAddrFromNodeID(id)
// for the test we make all peers share 8 bits so that syncing full bins make sense // for the test we make all peers share 8 bits so that syncing full bins make sense
addr.OAddr[0] = byte(0) addr.OAddr[0] = byte(0)
kad := NewKademlia(addr.Over(), NewKadParams()) kad := NewKademlia(addr.Over(), NewKadParams())
localStore := localStores[nodeCount] localStore := localStores[nodeCount]
dbAccess := NewDbAccess(localStore.(*storage.LocalStore)) db := NewDbAccess(localStore.(*storage.LocalStore))
streamer := NewStreamer(NewDelivery(kad, dbAccess)) streamer := NewRegistry(NewDelivery(kad, db))
RegisterIncomingSyncer(streamer, dbAccess) RegisterIncomingSyncer(streamer, db)
RegisterOutgoingSyncer(streamer, dbAccess) RegisterOutgoingSyncer(streamer, db)
self := &testStreamerService{ self := &testStreamerService{
index: nodeCount, index: nodeCount,
@ -144,15 +145,15 @@ func newSyncerService(ctx *adapters.ServiceContext) (node.Service, error) {
} }
func (b *testStreamerService) runSyncer(p *p2p.Peer, rw p2p.MsgReadWriter) error { func (b *testStreamerService) runSyncer(p *p2p.Peer, rw p2p.MsgReadWriter) error {
addr := NewAddrFromNodeID(p.ID()) addr := network.NewAddrFromNodeID(p.ID())
addr.OAddr[0] = byte(0) addr.OAddr[0] = byte(0)
bzzPeer := &bzzPeer{ BzzPeer := &BzzPeer{
Peer: protocols.NewPeer(p, rw, StreamerSpec), Peer: protocols.NewPeer(p, rw, Spec),
localAddr: b.addr, localAddr: b.addr,
BzzAddr: addr, BzzAddr: addr,
} }
b.streamer.delivery.overlay.On(bzzPeer) b.streamer.delivery.overlay.On(BzzPeer)
defer b.streamer.delivery.overlay.Off(bzzPeer) defer b.streamer.delivery.overlay.Off(BzzPeer)
// if len(addr) > b.index+1 && bytes.Equal(addrs[b.index+1], addr) { // if len(addr) > b.index+1 && bytes.Equal(addrs[b.index+1], addr) {
go func() { go func() {
// each node Subscribes to each other's retrieveRequestStream // each node Subscribes to each other's retrieveRequestStream
@ -164,5 +165,5 @@ func (b *testStreamerService) runSyncer(p *p2p.Peer, rw p2p.MsgReadWriter) error
} }
}() }()
// } // }
return b.streamer.Run(bzzPeer) return b.streamer.Run(BzzPeer)
} }

159
swarm/network/streamer_common_test.go → swarm/network/stream/testing/testing.go
View file

@ -1,4 +1,4 @@
// Copyright 2016 The go-ethereum Authors // Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library. // This file is part of the go-ethereum library.
// //
// The go-ethereum library is free software: you can redistribute it and/or modify // The go-ethereum library is free software: you can redistribute it and/or modify
@ -14,14 +14,12 @@
// You should have received a copy of the GNU Lesser General Public License // 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/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package network package testing
import ( import (
"context" "context"
"errors" "errors"
"flag"
"fmt" "fmt"
"io"
"io/ioutil" "io/ioutil"
"math/rand" "math/rand"
"os" "os"
@ -33,44 +31,23 @@ import (
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/p2p/simulations" "github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/p2p/simulations/adapters"
p2ptest "github.com/ethereum/go-ethereum/p2p/testing"
"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/stream"
"github.com/ethereum/go-ethereum/swarm/storage" "github.com/ethereum/go-ethereum/swarm/storage"
) )
var services = adapters.Services{
"delivery": newDeliveryService,
"syncer": newSyncerService,
}
var ( var (
adapter = flag.String("adapter", "sim", "type of simulation: sim|socket|exec|docker") LocalStores []storage.ChunkStore
loglevel = flag.Int("loglevel", 2, "verbosity of logs") Addrs []network.Addr
NodeCount int
) )
func init() { func setLocalStores(addrs ...network.Addr) (func(), error) {
flag.Parse()
// register the Delivery service which will run as a devp2p
// protocol when using the exec adapter
adapters.RegisterServices(services)
log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
}
var (
delivery *Delivery
localStores []storage.ChunkStore
addrs []Addr
fileHash storage.Key
nodeCount int
)
func setLocalStores(addrs ...Addr) (func(), error) {
var datadirs []string var datadirs []string
localStores = make([]storage.ChunkStore, len(addrs)) LocalStores = make([]storage.ChunkStore, len(addrs))
var err error var err error
for i, addr := range addrs { for i, addr := range addrs {
// TODO: remove temp datadir after test // TODO: remove temp datadir after test
@ -85,7 +62,7 @@ func setLocalStores(addrs ...Addr) (func(), error) {
break break
} }
datadirs = append(datadirs, datadir) datadirs = append(datadirs, datadir)
localStores[i] = localStore LocalStores[i] = localStore
} }
teardown := func() { teardown := func() {
for _, datadir := range datadirs { for _, datadir := range datadirs {
@ -95,27 +72,12 @@ func setLocalStores(addrs ...Addr) (func(), error) {
return teardown, err return teardown, err
} }
func mustReadAll(dpa *storage.DPA, hash storage.Key) (int, error) { func testSimulation(t *testing.T, services adapters.Services, adapter string, simf func(adapters.NodeAdapter) (*simulations.StepResult, error)) {
r := dpa.Retrieve(fileHash)
buf := make([]byte, 1024)
var n, total int
var err error
for (total == 0 || n > 0) && err == nil {
n, err = r.ReadAt(buf, int64(total))
total += n
}
if err != nil && err != io.EOF {
return total, err
}
return total, nil
}
func testSimulation(t *testing.T, simf func(adapters.NodeAdapter) (*simulations.StepResult, error)) {
var err error var err error
var result *simulations.StepResult var result *simulations.StepResult
startedAt := time.Now() startedAt := time.Now()
switch *adapter { switch adapter {
case "sim": case "sim":
t.Logf("simadapter") t.Logf("simadapter")
result, err = simf(adapters.NewSimAdapter(services)) result, err = simf(adapters.NewSimAdapter(services))
@ -158,7 +120,7 @@ func testSimulation(t *testing.T, simf func(adapters.NodeAdapter) (*simulations.
t.Logf("Setup: %s, shutdown: %s", result.StartedAt.Sub(startedAt), finishedAt.Sub(result.FinishedAt)) t.Logf("Setup: %s, shutdown: %s", result.StartedAt.Sub(startedAt), finishedAt.Sub(result.FinishedAt))
} }
func runSimulation(nodes, conns int, serviceName string, toAddr func(discover.NodeID) *BzzAddr, action func(*simulations.Network) func(context.Context) error, trigger func(*simulations.Network) chan discover.NodeID, check func(*simulations.Network, *storage.DPA) func(context.Context, discover.NodeID) (bool, error), adapter adapters.NodeAdapter) (*simulations.StepResult, error) { func runSimulation(nodes, conns int, serviceName string, toAddr func(discover.NodeID) *network.BzzAddr, action func(*simulations.Network) func(context.Context) error, trigger func(*simulations.Network) chan discover.NodeID, check func(*simulations.Network, *storage.DPA) func(context.Context, discover.NodeID) (bool, error), adapter adapters.NodeAdapter) (*simulations.StepResult, error) {
// create network // create network
net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{ net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{
ID: "0", ID: "0",
@ -166,8 +128,8 @@ func runSimulation(nodes, conns int, serviceName string, toAddr func(discover.No
}) })
defer net.Shutdown() defer net.Shutdown()
ids := make([]discover.NodeID, nodes) ids := make([]discover.NodeID, nodes)
nodeCount = 0 NodeCount = 0
addrs = make([]Addr, nodes) Addrs = make([]network.Addr, nodes)
// start nodes // start nodes
for i := 0; i < nodes; i++ { for i := 0; i < nodes; i++ {
node, err := net.NewNode() node, err := net.NewNode()
@ -175,10 +137,10 @@ func runSimulation(nodes, conns int, serviceName string, toAddr func(discover.No
return nil, fmt.Errorf("error creating node: %s", err) return nil, fmt.Errorf("error creating node: %s", err)
} }
ids[i] = node.ID() ids[i] = node.ID()
addrs[i] = toAddr(ids[i]) Addrs[i] = toAddr(ids[i])
} }
// set nodes number of localstores globally available // set nodes number of localstores globally available
teardown, err := setLocalStores(addrs...) teardown, err := setLocalStores(Addrs...)
defer teardown() defer teardown()
if err != nil { if err != nil {
return nil, err return nil, err
@ -212,11 +174,11 @@ func runSimulation(nodes, conns int, serviceName string, toAddr func(discover.No
} }
wg.Wait() wg.Wait()
log.Debug(fmt.Sprintf("nodes: %v", len(addrs))) log.Debug(fmt.Sprintf("nodes: %v", len(Addrs)))
// create an only locally retrieving dpa for the pivot node to test // create an only locally retrieving dpa for the pivot node to test
// if retriee requests have arrived // if retriee requests have arrived
dpa := storage.NewDPA(localStores[0], storage.NewChunkerParams()) dpa := storage.NewDPA(LocalStores[0], storage.NewChunkerParams())
dpa.Start() dpa.Start()
defer dpa.Stop() defer dpa.Stop()
timeout := 300 * time.Second timeout := 300 * time.Second
@ -233,47 +195,6 @@ func runSimulation(nodes, conns int, serviceName string, toAddr func(discover.No
return result, nil return result, nil
} }
func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Streamer, *storage.LocalStore, func(), error) {
// setup
addr := RandomAddr() // tested peers peer address
to := NewKademlia(addr.OAddr, NewKadParams())
// temp datadir
datadir, err := ioutil.TempDir("", "streamer")
if err != nil {
return nil, nil, nil, func() {}, err
}
teardown := func() {
os.RemoveAll(datadir)
}
localStore, err := storage.NewTestLocalStoreForAddr(datadir, addr.Over())
if err != nil {
return nil, nil, nil, teardown, err
}
dbAccess := NewDbAccess(localStore)
delivery := NewDelivery(to, dbAccess)
streamer := NewStreamer(delivery)
run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
bzzPeer := &bzzPeer{
Peer: protocols.NewPeer(p, rw, StreamerSpec),
localAddr: addr,
BzzAddr: NewAddrFromNodeID(p.ID()),
}
to.On(bzzPeer)
return streamer.Run(bzzPeer)
}
protocolTester := p2ptest.NewProtocolTester(t, NewNodeIDFromAddr(addr), 1, run)
err = waitForPeers(streamer, 1*time.Second)
if err != nil {
return nil, nil, nil, nil, errors.New("timeout: peer is not created")
}
return protocolTester, streamer, localStore, teardown, nil
}
type roundRobinStore struct { type roundRobinStore struct {
index uint32 index uint32
stores []storage.ChunkStore stores []storage.ChunkStore
@ -301,34 +222,24 @@ func (rrs *roundRobinStore) Close() {
} }
} }
func waitForPeers(streamer *Streamer, timeout time.Duration) error { type TestStreamerService struct {
ticker := time.NewTicker(10 * time.Millisecond)
timeoutTimer := time.NewTimer(timeout)
for {
select {
case <-ticker.C:
if len(streamer.peers) > 0 {
return nil
}
case <-timeoutTimer.C:
return errors.New("timeout")
}
}
}
type testStreamerService struct {
index int index int
addr *BzzAddr addr *network.BzzAddr
streamer *Streamer streamer *stream.Registry
run func(p *p2p.Peer, rw p2p.MsgReadWriter) error run func(s *TestStreamerService, p *p2p.Peer, rw p2p.MsgReadWriter) error
} }
func (tds *testStreamerService) Protocols() []p2p.Protocol { func NewTestStreamerService(run func(s *TestStreamerService, p *p2p.Peer, rw p2p.MsgReadWriter) error) TestStreamerService {
t := &TestStreamerService{}
t.run = run
}
func (tds *TestStreamerService) Protocols() []p2p.Protocol {
return []p2p.Protocol{ return []p2p.Protocol{
{ {
Name: StreamerSpec.Name, Name: stream.Spec.Name,
Version: StreamerSpec.Version, Version: stream.Spec.Version,
Length: StreamerSpec.Length(), Length: stream.Spec.Length(),
Run: tds.run, Run: tds.run,
// NodeInfo: , // NodeInfo: ,
// PeerInfo: , // PeerInfo: ,
@ -336,14 +247,14 @@ func (tds *testStreamerService) Protocols() []p2p.Protocol {
} }
} }
func (b *testStreamerService) APIs() []rpc.API { func (b *TestStreamerService) APIs() []rpc.API {
return []rpc.API{} return []rpc.API{}
} }
func (b *testStreamerService) Start(server *p2p.Server) error { func (b *TestStreamerService) Start(server *p2p.Server) error {
return nil return nil
} }
func (b *testStreamerService) Stop() error { func (b *TestStreamerService) Stop() error {
return nil return nil
} }

630
swarm/network/streamer.go
View file

@ -1,630 +0,0 @@
// Copyright 2016 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 network
import (
"context"
"errors"
"fmt"
"math"
"sync"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/protocols"
bv "github.com/ethereum/go-ethereum/swarm/network/bitvector"
pq "github.com/ethereum/go-ethereum/swarm/network/priorityqueue"
"github.com/ethereum/go-ethereum/swarm/storage"
)
const (
Low uint8 = iota
Mid
High
Top
PriorityQueue // number of queues
PriorityQueueCap = 3 // queue capacity
HashSize = 32
)
// Handover represents a statement that the upstream peer hands over the stream section
type Handover struct {
Stream string // name of stream
Start, End uint64 // index of hashes
Root []byte // Root hash for indexed segment inclusion proofs
}
// HandoverProof represents a signed statement that the upstream peer handed over the stream section
type HandoverProof struct {
Sig []byte // Sign(Hash(Serialisation(Handover)))
*Handover
}
// Takeover represents a statement that downstream peer took over (stored all data)
// handed over
type Takeover Handover
// TakeoverProof represents a signed statement that the downstream peer took over
// the stream section
type TakeoverProof struct {
Sig []byte // Sign(Hash(Serialisation(Takeover)))
*Takeover
}
// TakeoverProofMsg is the protocol msg sent by downstream peer
type TakeoverProofMsg TakeoverProof
// String pretty prints TakeoverProofMsg
func (self TakeoverProofMsg) String() string {
return fmt.Sprintf("Stream: '%v' [%v-%v], Root: %x, Sig: %x", self.Stream, self.Start, self.End, self.Root, self.Sig)
}
// SubcribeMsg is the protocol msg for requesting a stream(section)
type SubscribeMsg struct {
Stream string
Key []byte
From, To uint64
Priority uint8 // delivered on priority channel
}
// OfferedHashesMsg is the protocol msg for offering to hand over a
// stream section
type OfferedHashesMsg struct {
Stream string // name of Stream
Key []byte // subtype or key
From, To uint64 // peer and db-specific entry count
Hashes []byte // stream of hashes (128)
*HandoverProof // HandoverProof
}
// String pretty prints OfferedHashesMsg
func (self OfferedHashesMsg) String() string {
return fmt.Sprintf("Stream '%v' [%v-%v] (%v)", self.Stream, self.From, self.To, len(self.Hashes)/HashSize)
}
// WantedHashesMsg is the protocol msg data for signaling which hashes
// offered in OfferedHashesMsg downstream peer actually wants sent over
type WantedHashesMsg struct {
Stream string // name of stream
Key []byte // subtype or key
Want []byte // bitvector indicating which keys of the batch needed
From, To uint64 // next interval offset - empty if not to be continued
}
// String pretty prints WantedHashesMsg
func (self WantedHashesMsg) String() string {
return fmt.Sprintf("Stream '%v', Want: %x, Next: [%v-%v]", self.Stream, self.Want, self.From, self.To)
}
func keyToString(key []byte) string {
l := len(key)
if l == 0 {
return ""
}
return fmt.Sprintf("%s-%d", string(key[:l-1]), uint8(key[l-1]))
}
// Streamer registry for outgoing and incoming streamer constructors
type Streamer struct {
incomingLock sync.RWMutex
outgoingLock sync.RWMutex
peersLock sync.RWMutex
outgoing map[string]func(*StreamerPeer, []byte) (OutgoingStreamer, error)
incoming map[string]func(*StreamerPeer, []byte) (IncomingStreamer, error)
peers map[discover.NodeID]*StreamerPeer
delivery *Delivery
}
// NewStreamer is Streamer constructor
func NewStreamer(delivery *Delivery) *Streamer {
streamer := &Streamer{
outgoing: make(map[string]func(*StreamerPeer, []byte) (OutgoingStreamer, error)),
incoming: make(map[string]func(*StreamerPeer, []byte) (IncomingStreamer, error)),
peers: make(map[discover.NodeID]*StreamerPeer),
delivery: delivery,
}
delivery.getPeer = streamer.getPeer
streamer.RegisterOutgoingStreamer(retrieveRequestStream, func(_ *StreamerPeer, t []byte) (OutgoingStreamer, error) {
return NewRetrieveRequestStreamer(delivery.dbAccess), nil
})
streamer.RegisterIncomingStreamer(retrieveRequestStream, func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
return NewIncomingSwarmSyncer(p, delivery.dbAccess, nil)
})
return streamer
}
func (self *Streamer) Retrieve(chunk *storage.Chunk) error {
return self.delivery.RequestFromPeers(chunk.Key[:], false)
}
// RegisterIncomingStreamer registers an incoming streamer constructor
func (self *Streamer) RegisterIncomingStreamer(stream string, f func(*StreamerPeer, []byte) (IncomingStreamer, error)) {
self.incomingLock.Lock()
defer self.incomingLock.Unlock()
self.incoming[stream] = f
}
// RegisterOutgoingStreamer registers an outgoing streamer constructor
func (self *Streamer) RegisterOutgoingStreamer(stream string, f func(*StreamerPeer, []byte) (OutgoingStreamer, error)) {
self.outgoingLock.Lock()
defer self.outgoingLock.Unlock()
self.outgoing[stream] = f
}
// GetIncomingStreamer accessor for incoming streamer constructors
func (self *Streamer) GetIncomingStreamer(stream string) (func(*StreamerPeer, []byte) (IncomingStreamer, error), error) {
self.incomingLock.RLock()
defer self.incomingLock.RUnlock()
f := self.incoming[stream]
if f == nil {
return nil, fmt.Errorf("stream %v not registered", stream)
}
return f, nil
}
// GetOutgoingStreamer accessor for incoming streamer constructors
func (self *Streamer) GetOutgoingStreamer(stream string) (func(*StreamerPeer, []byte) (OutgoingStreamer, error), error) {
self.outgoingLock.RLock()
defer self.outgoingLock.RUnlock()
f := self.outgoing[stream]
if f == nil {
return nil, fmt.Errorf("stream %v not registered", stream)
}
return f, nil
}
func (self *Streamer) NodeInfo() interface{} {
return nil
}
func (self *Streamer) PeerInfo(id discover.NodeID) interface{} {
return nil
}
type outgoingStreamer struct {
OutgoingStreamer
priority uint8
currentBatch []byte
stream string
key []byte
}
// OutgoingStreamer interface for outgoing peer Streamer
type OutgoingStreamer interface {
SetNextBatch(uint64, uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error)
GetData([]byte) []byte
}
type incomingStreamer struct {
IncomingStreamer
priority uint8
sessionAt uint64
live bool
stream string
key []byte
quit chan struct{}
next chan struct{}
}
// IncomingStreamer interface for incoming peer Streamer
type IncomingStreamer interface {
NeedData([]byte) func()
BatchDone(string, uint64, []byte, []byte) func() (*TakeoverProof, error)
}
// StreamerPeer is the Peer extention for the streaming protocol
type StreamerPeer struct {
Peer
streamer *Streamer
pq *pq.PriorityQueue
//netStore storage.ChunkStore
outgoingLock sync.RWMutex
incomingLock sync.RWMutex
outgoing map[string]*outgoingStreamer
incoming map[string]*incomingStreamer
quit chan struct{}
}
// NewStreamerPeer is the constructor for StreamerPeer
func NewStreamerPeer(p Peer, streamer *Streamer) *StreamerPeer {
self := &StreamerPeer{
Peer: p,
pq: pq.New(int(PriorityQueue), PriorityQueueCap),
streamer: streamer,
outgoing: make(map[string]*outgoingStreamer),
incoming: make(map[string]*incomingStreamer),
quit: make(chan struct{}),
}
ctx, cancel := context.WithCancel(context.Background())
go self.pq.Run(ctx, func(i interface{}) { p.Send(i) })
go func() {
<-self.quit
cancel()
}()
return self
}
func (self *Streamer) getPeer(peerId discover.NodeID) *StreamerPeer {
self.peersLock.RLock()
defer self.peersLock.RUnlock()
return self.peers[peerId]
}
func (self *Streamer) setPeer(peer *StreamerPeer) {
self.peersLock.Lock()
self.peers[peer.ID()] = peer
self.peersLock.Unlock()
}
func (self *Streamer) deletePeer(peer *StreamerPeer) {
self.peersLock.Lock()
delete(self.peers, peer.ID())
self.peersLock.Unlock()
}
func (self *StreamerPeer) getOutgoingStreamer(s string) (*outgoingStreamer, error) {
self.outgoingLock.RLock()
defer self.outgoingLock.RUnlock()
streamer := self.outgoing[s]
if streamer == nil {
return nil, fmt.Errorf("outgoing stream '%v' not provided to peer %v", s, self.ID())
}
return streamer, nil
}
func (self *StreamerPeer) getIncomingStreamer(s string) (*incomingStreamer, error) {
self.incomingLock.RLock()
defer self.incomingLock.RUnlock()
streamer := self.incoming[s]
if streamer == nil {
return nil, fmt.Errorf("incoming stream '%v' not provided to peer %v", s, self.ID())
}
return streamer, nil
}
func (self *StreamerPeer) setOutgoingStreamer(s string, key []byte, o OutgoingStreamer, priority uint8) (*outgoingStreamer, error) {
self.outgoingLock.Lock()
defer self.outgoingLock.Unlock()
sk := s + keyToString(key)
if self.outgoing[sk] != nil {
return nil, fmt.Errorf("stream %v already registered", sk)
}
os := &outgoingStreamer{
OutgoingStreamer: o,
priority: priority,
stream: s,
key: key,
}
self.outgoing[sk] = os
return os, nil
}
func (self *StreamerPeer) setIncomingStreamer(s string, key []byte, i IncomingStreamer, priority uint8, live bool) error {
self.incomingLock.Lock()
defer self.incomingLock.Unlock()
sk := s + keyToString(key)
if self.incoming[sk] != nil {
return fmt.Errorf("stream %v already registered", sk)
}
next := make(chan struct{}, 1)
// var intervals *Intervals
// if !live {
// key := s + self.ID().String()
// intervals = NewIntervals(key, self.streamer)
// }
self.incoming[sk] = &incomingStreamer{
IncomingStreamer: i,
// intervals: intervals,
live: live,
priority: priority,
next: next,
stream: s,
key: key,
}
next <- struct{}{} // this is to allow wantedKeysMsg before first batch arrives
return nil
}
// NextBatch adjusts the indexes by inspecting the intervals
func (self *incomingStreamer) nextBatch(from uint64) (nextFrom uint64, nextTo uint64) {
var intervals []uint64
if self.live {
if len(intervals) == 0 {
intervals = []uint64{self.sessionAt, from}
} else {
intervals[1] = from
}
nextFrom = from
} else if from >= self.sessionAt { // history sync complete
intervals = nil
nextFrom = from
nextTo = math.MaxUint64
} else if len(intervals) > 2 && from >= intervals[2] { // filled a gap in the intervals
intervals = append(intervals[:1], intervals[3:]...)
nextFrom = intervals[1]
if len(intervals) > 2 {
nextTo = intervals[2]
} else {
nextTo = self.sessionAt
}
} else {
nextFrom = from
intervals[1] = from
nextTo = self.sessionAt
}
// self.intervals.set(intervals)
return nextFrom, nextTo
}
// Subscribe initiates the streamer
func (self *Streamer) Subscribe(peerId discover.NodeID, s string, t []byte, from, to uint64, priority uint8, live bool) error {
f, err := self.GetIncomingStreamer(s)
if err != nil {
return err
}
peer := self.getPeer(peerId)
if peer == nil {
return fmt.Errorf("peer not found %v", peerId)
}
is, err := f(peer, t)
if err != nil {
return err
}
err = peer.setIncomingStreamer(s, t, is, priority, live)
if err != nil {
return err
}
msg := &SubscribeMsg{
Stream: s,
Key: t,
// Live: live,
From: from,
To: to,
Priority: priority,
}
log.Debug("Subscribe ", "peer", peerId, "stream", s, "key", t, "from", from, "to", to)
peer.SendPriority(msg, priority)
return nil
}
func (self *StreamerPeer) handleSubscribeMsg(req *SubscribeMsg) error {
f, err := self.streamer.GetOutgoingStreamer(req.Stream)
if err != nil {
return err
}
s, err := f(self, req.Key)
if err != nil {
return err
}
os, err := self.setOutgoingStreamer(req.Stream, req.Key, s, req.Priority)
if err != nil {
return nil
}
log.Debug("received subscription", "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
go self.SendOfferedHashes(os, req.From, req.To)
return nil
}
// handleOfferedHashesMsg protocol msg handler calls the incoming streamer interface
// Filter method
func (self *StreamerPeer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
sk := req.Stream
sk += keyToString(req.Key)
s, err := self.getIncomingStreamer(sk)
if err != nil {
return err
}
hashes := req.Hashes
want, err := bv.New(len(hashes) / HashSize)
if err != nil {
return fmt.Errorf("error initiaising bitvector of length %v: %v", len(hashes)/HashSize, err)
}
wg := sync.WaitGroup{}
for i := 0; i < len(hashes); i += HashSize {
hash := hashes[i : i+HashSize]
if wait := s.NeedData(hash); wait != nil {
want.Set(i/HashSize, true)
wg.Add(1)
// create request and wait until the chunk data arrives and is stored
go func(w func()) {
w()
wg.Done()
}(wait)
}
}
go func() {
wg.Wait()
if tf := s.BatchDone(req.Stream, req.From, hashes, req.Root); tf != nil {
tp, err := tf()
if err != nil {
return
}
self.SendPriority(tp, s.priority)
}
s.next <- struct{}{}
}()
// only send wantedKeysMsg if all missing chunks of the previous batch arrived
// except
if s.live {
s.sessionAt = req.From
}
from, to := s.nextBatch(req.To)
log.Debug("received batch", "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
if from == to {
return nil
}
msg := &WantedHashesMsg{
Stream: req.Stream,
Key: req.Key,
Want: want.Bytes(),
From: from,
To: to,
}
go func() {
select {
case <-s.next:
case <-s.quit:
return
}
log.Debug("want batch", "stream", msg.Stream, "Key", msg.Key, "from", msg.From, "to", msg.To)
self.SendPriority(msg, s.priority)
}()
return nil
}
// handleWantedHashesMsg protocol msg handler
// * sends the next batch of unsynced keys
// * sends the actual data chunks as per WantedHashesMsg
func (self *StreamerPeer) handleWantedHashesMsg(req *WantedHashesMsg) error {
log.Debug("received wanted batch", "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
s, err := self.getOutgoingStreamer(req.Stream + keyToString(req.Key))
if err != nil {
log.Debug(err.Error())
return err
}
hashes := s.currentBatch
// launch in go routine since GetBatch blocks until new hashes arrive
go self.SendOfferedHashes(s, req.From, req.To)
l := len(hashes) / HashSize
want, err := bv.NewFromBytes(req.Want, l)
if err != nil {
return fmt.Errorf("error initiaising bitvector of length %v: %v", l, err)
}
for i := 0; i < l; i++ {
if want.Get(i) {
hash := hashes[i*HashSize : (i+1)*HashSize]
data := s.GetData(hash)
if data == nil {
return errors.New("not found")
}
chunk := storage.NewChunk(hash, nil)
chunk.SData = data
if err := self.Deliver(chunk, s.priority); err != nil {
return err
}
}
}
return nil
}
func (self *StreamerPeer) handleTakeoverProofMsg(req *TakeoverProofMsg) error {
_, err := self.getOutgoingStreamer(req.Stream)
if err != nil {
return err
}
// store the strongest takeoverproof for the stream in streamer
return nil
}
// Deliver sends a storeRequestMsg protocol message to the peer
func (self *StreamerPeer) Deliver(chunk *storage.Chunk, priority uint8) error {
msg := &ChunkDeliveryMsg{
Key: chunk.Key,
SData: chunk.SData,
}
return self.pq.Push(nil, msg, int(priority))
}
// Deliver sends a storeRequestMsg protocol message to the peer
func (self *StreamerPeer) SendPriority(msg interface{}, priority uint8) error {
return self.pq.Push(nil, msg, int(priority))
}
// SendOfferedHashes sends OfferedHashesMsg protocol msg
func (self *StreamerPeer) SendOfferedHashes(s *outgoingStreamer, f, t uint64) error {
hashes, from, to, proof, err := s.SetNextBatch(f, t)
if err != nil {
return err
}
if proof == nil {
proof = &HandoverProof{
Handover: &Handover{},
}
}
s.currentBatch = hashes
msg := &OfferedHashesMsg{
HandoverProof: proof,
Hashes: hashes,
From: from,
To: to,
Stream: s.stream,
Key: s.key,
}
log.Debug("Swarm syncer offer batch", "stream", s.stream, "key", s.key, "len", len(hashes), "from", from, "to", to)
return self.SendPriority(msg, s.priority)
}
// StreamerSpec is the spec of the streamer protocol.
var StreamerSpec = &protocols.Spec{
Name: "stream",
Version: 1,
MaxMsgSize: 10 * 1024 * 1024,
Messages: []interface{}{
HandshakeMsg{},
OfferedHashesMsg{},
WantedHashesMsg{},
TakeoverProofMsg{},
SubscribeMsg{},
RetrieveRequestMsg{},
ChunkDeliveryMsg{},
},
}
// Run protocol run function
func (s *Streamer) Run(p *bzzPeer) error {
sp := NewStreamerPeer(p, s)
// load saved intervals
s.setPeer(sp)
defer s.deletePeer(sp)
defer close(sp.quit)
return sp.Run(sp.HandleMsg)
}
// HandleMsg is the message handler that delegates incoming messages
func (self *StreamerPeer) HandleMsg(msg interface{}) error {
switch msg := msg.(type) {
case *SubscribeMsg:
return self.handleSubscribeMsg(msg)
case *OfferedHashesMsg:
return self.handleOfferedHashesMsg(msg)
case *TakeoverProofMsg:
return self.handleTakeoverProofMsg(msg)
case *WantedHashesMsg:
return self.handleWantedHashesMsg(msg)
case *ChunkDeliveryMsg:
return self.streamer.delivery.handleChunkDeliveryMsg(msg)
case *RetrieveRequestMsg:
return self.streamer.delivery.handleRetrieveRequestMsg(self, msg)
default:
return fmt.Errorf("unknown message type: %T", msg)
}
}

52
swarm/storage/dbaccess.go Normal file
View file

@ -0,0 +1,52 @@
// 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 storage
// wrapper of db-s to provide mockable custom local chunk store access to syncer
type DBAPI struct {
db *DbStore
loc *LocalStore
}
func NewDBAPI(loc *LocalStore) *DBAPI {
return &DBAPI{loc.DbStore.(*DbStore), loc}
}
// to obtain the chunks from key or request db entry only
func (self *DBAPI) Get(key Key) (*Chunk, error) {
return self.loc.Get(key)
}
// current storage counter of chunk db
func (self *DBAPI) CurrentBucketStorageIndex(po uint8) uint64 {
return self.db.CurrentBucketStorageIndex(po)
}
// iteration storage counter and proximity order
func (self *DBAPI) Iterator(from uint64, to uint64, po uint8, f func(Key, uint64) bool) error {
return self.db.SyncIterator(from, to, po, f)
}
// to obtain the chunks from key or request db entry only
func (self *DBAPI) GetOrCreateRequest(key Key) (*Chunk, bool) {
return self.loc.GetOrCreateRequest(key)
}
// to obtain the chunks from key or request db entry only
func (self *DBAPI) Put(chunk *Chunk) {
self.loc.Put(chunk)
}

20
swarm/swarm.go
View file

@ -39,6 +39,7 @@ import (
httpapi "github.com/ethereum/go-ethereum/swarm/api/http" httpapi "github.com/ethereum/go-ethereum/swarm/api/http"
"github.com/ethereum/go-ethereum/swarm/fuse" "github.com/ethereum/go-ethereum/swarm/fuse"
"github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/stream"
"github.com/ethereum/go-ethereum/swarm/pss" "github.com/ethereum/go-ethereum/swarm/pss"
"github.com/ethereum/go-ethereum/swarm/storage" "github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/storage/mock" "github.com/ethereum/go-ethereum/swarm/storage/mock"
@ -53,7 +54,7 @@ type Swarm struct {
//storage storage.ChunkStore // internal access to storage, common interface to cloud storage backends //storage storage.ChunkStore // internal access to storage, common interface to cloud storage backends
dpa *storage.DPA // distributed preimage archive, the local API to the storage with document level storage/retrieval support dpa *storage.DPA // distributed preimage archive, the local API to the storage with document level storage/retrieval support
//depo network.StorageHandler // remote request handler, interface between bzz protocol and the storage //depo network.StorageHandler // remote request handler, interface between bzz protocol and the storage
streamer *network.Streamer streamer *stream.Registry
//cloud storage.CloudStore // procurement, cloud storage backend (can multi-cloud) //cloud storage.CloudStore // procurement, cloud storage backend (can multi-cloud)
bzz *network.Bzz // the logistic manager bzz *network.Bzz // the logistic manager
backend chequebook.Backend // simple blockchain Backend backend chequebook.Backend // simple blockchain Backend
@ -129,13 +130,13 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e
HiveParams: config.HiveParams, HiveParams: config.HiveParams,
} }
dbAccess := network.NewDbAccess(self.lstore) db := storage.NewDBAPI(self.lstore)
delivery := network.NewDelivery(to, dbAccess) delivery := stream.NewDelivery(to, db)
self.streamer = network.NewStreamer(delivery) self.streamer = stream.NewRegistry(delivery)
network.RegisterOutgoingSyncer(self.streamer, dbAccess) stream.RegisterSwarmSyncerServer(self.streamer, db)
network.RegisterIncomingSyncer(self.streamer, dbAccess) stream.RegisterSwarmSyncerClient(self.streamer, db)
self.bzz = network.NewBzz(bzzconfig, to, nil, self.streamer) self.bzz = network.NewBzz(bzzconfig, to, nil)
// set up DPA, the cloud storage local access layer // set up DPA, the cloud storage local access layer
dpaChunkStore := storage.NewNetStore(self.lstore, self.streamer.Retrieve) dpaChunkStore := storage.NewNetStore(self.lstore, self.streamer.Retrieve)
@ -271,6 +272,11 @@ func (self *Swarm) Protocols() (protos []p2p.Protocol) {
protos = append(protos, p) protos = append(protos, p)
} }
} }
if self.streamer != nil {
for _, p := range self.streamer.Protocols() {
protos = append(protos, p)
}
}
return return
} }