diff --git a/swarm/network/discovery.go b/swarm/network/discovery.go
index fb7152cba1..71ed974d2f 100644
--- a/swarm/network/discovery.go
+++ b/swarm/network/discovery.go
@@ -25,9 +25,9 @@ import (
// 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 {
- *bzzPeer
+ *BzzPeer
overlay Overlay
sentPeers bool // whether we already sent peer closer to this address
mtx sync.Mutex
@@ -36,10 +36,10 @@ type discPeer struct {
}
// NewDiscovery constructs a discovery peer
-func newDiscovery(p *bzzPeer, o Overlay) *discPeer {
+func newDiscovery(p *BzzPeer, o Overlay) *discPeer {
d := &discPeer{
overlay: o,
- bzzPeer: p,
+ BzzPeer: p,
peers: make(map[string]bool),
}
// record remote as seen so we never send a peer its own record
diff --git a/swarm/network/discovery_test.go b/swarm/network/discovery_test.go
index ee90683a73..50e1f468b6 100644
--- a/swarm/network/discovery_test.go
+++ b/swarm/network/discovery_test.go
@@ -33,7 +33,7 @@ func TestDiscovery(t *testing.T) {
addr := RandomAddr()
to := NewKademlia(addr.OAddr, NewKadParams())
- run := func(p *bzzPeer) error {
+ run := func(p *BzzPeer) error {
dp := newDiscovery(p, to)
to.On(p)
defer to.Off(p)
diff --git a/swarm/network/hive.go b/swarm/network/hive.go
index d72d7c5e7c..a309d93983 100644
--- a/swarm/network/hive.go
+++ b/swarm/network/hive.go
@@ -159,7 +159,7 @@ func (h *Hive) connect() {
}
// Run protocol run function
-func (h *Hive) Run(p *bzzPeer) error {
+func (h *Hive) Run(p *BzzPeer) error {
dp := newDiscovery(p, h)
depth, changed := h.On(dp)
// if we want discovery, advertise changed depth of depth
@@ -191,7 +191,7 @@ func ToAddr(pa OverlayPeer) *BzzAddr {
if p, ok := pa.(*discPeer); ok {
return p.BzzAddr
}
- return pa.(*bzzPeer).BzzAddr
+ return pa.(*BzzPeer).BzzAddr
}
// loadPeers, savePeer implement persistence callback/
diff --git a/swarm/network/kademlia_test.go b/swarm/network/kademlia_test.go
index 20bfd7daaf..01ed72c582 100644
--- a/swarm/network/kademlia_test.go
+++ b/swarm/network/kademlia_test.go
@@ -70,7 +70,7 @@ func newTestKademlia(b string) *testKademlia {
}
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 {
diff --git a/swarm/network/lightnode.go b/swarm/network/light/lightnode.go
similarity index 67%
rename from swarm/network/lightnode.go
rename to swarm/network/light/lightnode.go
index 8ee7f5b22e..7bf769d468 100644
--- a/swarm/network/lightnode.go
+++ b/swarm/network/light/lightnode.go
@@ -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
//
// 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
// along with the go-ethereum library. If not, see .
-package network
+package light
import (
"errors"
+ "github.com/ethereum/go-ethereum/swarm/network/stream"
"github.com/ethereum/go-ethereum/swarm/storage"
)
// RemoteReader implements IncomingStreamer
type RemoteSectionReader struct {
- db *DbAccess
+ db *storage.DBAPI
start uint64
end uint64
hashes chan []byte
@@ -35,7 +36,7 @@ type RemoteSectionReader struct {
}
// NewRemoteReader is the constructor for RemoteReader
-func NewRemoteSectionReader(root []byte, db *DbAccess) *RemoteSectionReader {
+func NewRemoteSectionReader(root []byte, db *storage.DBAPI) *RemoteSectionReader {
return &RemoteSectionReader{
db: db,
root: root,
@@ -45,7 +46,7 @@ func NewRemoteSectionReader(root []byte, db *DbAccess) *RemoteSectionReader {
}
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
if chunk.ReqC == nil || !created {
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
return nil
}
@@ -75,9 +76,9 @@ func (r *RemoteSectionReader) Read(b []byte) (n int64, err error) {
return l, nil
}
var end bool
- for i := 0; !end && i < len(r.currentHashes); i += HashSize {
- hash := r.currentHashes[i : i+HashSize]
- chunk, err := r.db.get(hash)
+ for i := 0; !end && i < len(r.currentHashes); i += stream.HashSize {
+ hash := r.currentHashes[i : i+stream.HashSize]
+ chunk, err := r.db.Get(hash)
if err != nil {
return n, err
}
@@ -96,9 +97,9 @@ func (r *RemoteSectionReader) Read(b []byte) (n int64, err error) {
return n, errors.New("aborted")
case hashes := <-r.hashes:
var i int
- for ; !end && i < len(hashes); i += HashSize {
- hash := hashes[i : i+HashSize]
- chunk, err := r.db.get(hash)
+ for ; !end && i < len(hashes); i += stream.HashSize {
+ hash := hashes[i : i+stream.HashSize]
+ chunk, err := r.db.Get(hash)
if err != nil {
return n, err
}
@@ -120,12 +121,12 @@ func (r *RemoteSectionReader) Read(b []byte) (n int64, err error) {
type RemoteSectionServer struct {
// quit chan struct{}
root []byte
- db *DbAccess
+ db *storage.DBAPI
r *storage.LazyChunkReader
}
// 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{
db: db,
r: r,
@@ -134,7 +135,7 @@ func NewRemoteSectionServer(db *DbAccess, r *storage.LazyChunkReader) *RemoteSec
// GetData retrieves the actual chunk from localstore
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 {
return nil
}
@@ -142,26 +143,26 @@ func (s *RemoteSectionServer) GetData(key []byte) []byte {
}
// GetBatch retrieves the next batch of hashes from the dbstore
-func (s *RemoteSectionServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) {
- if to > from+batchSize {
- to = from + batchSize
+func (s *RemoteSectionServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint64, *stream.HandoverProof, error) {
+ if to > from+stream.BatchSize {
+ to = from + stream.BatchSize
}
- batch := make([]byte, (to-from)*HashSize)
+ batch := make([]byte, (to-from)*stream.HashSize)
s.r.ReadAt(batch, int64(from))
return batch, from, to, nil, nil
}
// RegisterRemoteSectionReader registers RemoteSectionReader on light downstream node
-func RegisterRemoteSectionReader(s *Streamer, db *DbAccess) {
- s.RegisterIncomingStreamer("REMOTE_SECTION", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
+func RegisterRemoteSectionReader(s *stream.Registry, db *storage.DBAPI) {
+ s.RegisterClientFunc("REMOTE_SECTION", func(p *stream.Peer, t []byte) (stream.Client, error) {
return NewRemoteSectionReader(t, db), nil
})
}
// RegisterRemoteSectionServer registers RemoteSectionServer outgoing streamer on
// upstream light server node
-func RegisterRemoteSectionServer(s *Streamer, db *DbAccess, rf func([]byte) *storage.LazyChunkReader) {
- s.RegisterOutgoingStreamer("REMOTE_SECTION", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
+func RegisterRemoteSectionServer(s *stream.Registry, db *storage.DBAPI, rf func([]byte) *storage.LazyChunkReader) {
+ s.RegisterServerFunc("REMOTE_SECTION", func(p *stream.Peer, t []byte) (stream.Server, error) {
r := rf(t)
return NewRemoteSectionServer(db, r), nil
})
@@ -169,16 +170,16 @@ func RegisterRemoteSectionServer(s *Streamer, db *DbAccess, rf func([]byte) *sto
// RegisterRemoteDownloader registers RemoteDownloader incoming streamer
// on downstream light node
-// func RegisterRemoteDownloader(s *Streamer, db *DbAccess) {
-// s.RegisterIncomingStreamer("REMOTE_DOWNLOADER", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
+// func RegisterRemoteDownloader(s *Streamer, db *storage.DBAPI) {
+// s.RegisterIncomingStreamer("REMOTE_DOWNLOADER", func(p *stream.Peer, t []byte) (IncomingStreamer, error) {
// return NewRemoteDownloader(t, db), nil
// })
// }
//
// // RegisterRemoteDownloadServer registers RemoteDownloadServer outgoing streamer on
// // upstream light server node
-// func RegisterRemoteDownloadServer(s *Streamer, db *DbAccess, rf func([]byte) *storage.LazyChunkReader) {
-// s.RegisterOutgoingStreamer("REMOTE_DOWNLOADER", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
+// func RegisterRemoteDownloadServer(s *Streamer, db *storage.DBAPI, rf func([]byte) *storage.LazyChunkReader) {
+// s.RegisterOutgoingStreamer("REMOTE_DOWNLOADER", func(p *stream.Peer, t []byte) (OutgoingStreamer, error) {
// r := rf(t)
// return NewRemoteDownloadServer(db, r), nil
// })
diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go
index 4f5906b4ae..53776e5ba6 100644
--- a/swarm/network/protocol.go
+++ b/swarm/network/protocol.go
@@ -103,7 +103,6 @@ type BzzConfig struct {
// Bzz is the swarm protocol bundle
type Bzz struct {
- Streamer *Streamer
*Hive
localAddr *BzzAddr
mtx sync.Mutex
@@ -115,9 +114,8 @@ type Bzz struct {
// * bzz config
// * overlay driver
// * peer store
-func NewBzz(config *BzzConfig, kad Overlay, store StateStore, streamer *Streamer) *Bzz {
+func NewBzz(config *BzzConfig, kad Overlay, store StateStore) *Bzz {
return &Bzz{
- Streamer: streamer,
Hive: NewHive(config.HiveParams, kad, store),
localAddr: &BzzAddr{config.OverlayAddr, config.UnderlayAddr},
handshakes: make(map[discover.NodeID]*HandshakeMsg),
@@ -143,7 +141,7 @@ func (b *Bzz) NodeInfo() interface{} {
// * handshake/hive
// * discovery
func (b *Bzz) Protocols() []p2p.Protocol {
- protocols := []p2p.Protocol{
+ return []p2p.Protocol{
{
Name: BzzSpec.Name,
Version: BzzSpec.Version,
@@ -160,17 +158,6 @@ func (b *Bzz) Protocols() []p2p.Protocol {
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
@@ -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
// arguments:
// * 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
// on return the session is terminated and the peer is disconnected
// the protocol waits for the bzz handshake is negotiated
-// 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 {
+// 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 {
return func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
// wait for the bzz protocol to perform the handshake
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 {
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
- peer := &bzzPeer{
+ // the handshake has succeeded so construct the BzzPeer and run the protocol
+ peer := &BzzPeer{
Peer: protocols.NewPeer(p, rw, spec),
localAddr: b.localAddr,
BzzAddr: handshake.peerAddr,
@@ -257,9 +244,9 @@ func (b *Bzz) runBzz(p *p2p.Peer, rw p2p.MsgReadWriter) error {
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
-type bzzPeer struct {
+type BzzPeer struct {
*protocols.Peer // represents the connection for online peers
localAddr *BzzAddr // local Peers address
*BzzAddr // remote address -> implements Addr interface = protocols.Peer
@@ -267,12 +254,12 @@ type bzzPeer struct {
}
// Off returns the overlay peer record for offline persistance
-func (p *bzzPeer) Off() OverlayAddr {
+func (p *BzzPeer) Off() OverlayAddr {
return p.BzzAddr
}
// LastActive returns the time the peer was last active
-func (p *bzzPeer) LastActive() time.Time {
+func (p *BzzPeer) LastActive() time.Time {
return p.lastActive
}
diff --git a/swarm/network/protocol_test.go b/swarm/network/protocol_test.go
index 1d7e165f02..fdabddb1c3 100644
--- a/swarm/network/protocol_test.go
+++ b/swarm/network/protocol_test.go
@@ -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)
- srv := func(p *bzzPeer) error {
+ srv := func(p *BzzPeer) error {
defer close(cs[p.ID().String()])
return run(p)
}
protocall := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
- return srv(&bzzPeer{
+ return srv(&BzzPeer{
Peer: protocols.NewPeer(p, rw, spec),
localAddr: addr,
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 {
- extraservices := func(p *bzzPeer) error {
+ extraservices := func(p *BzzPeer) error {
pp.Add(p)
defer pp.Remove(p)
if services == nil {
diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go
new file mode 100644
index 0000000000..c46cc47144
--- /dev/null
+++ b/swarm/network/stream/common_test.go
@@ -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 .
+
+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")
+ }
+ }
+}
diff --git a/swarm/network/requests.go b/swarm/network/stream/delivery.go
similarity index 61%
rename from swarm/network/requests.go
rename to swarm/network/stream/delivery.go
index 3495cfceb9..d242dd01f3 100644
--- a/swarm/network/requests.go
+++ b/swarm/network/stream/delivery.go
@@ -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.
//
// 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
// along with the go-ethereum library. If not, see .
-package network
+package stream
import (
"errors"
@@ -23,51 +23,52 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/discover"
+ "github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/storage"
)
-const retrieveRequestStream = "RETRIEVE_REQUEST"
+const swarmChunkServerStreamName = "RETRIEVE_REQUEST"
type Delivery struct {
- dbAccess *DbAccess
- overlay Overlay
+ db *storage.DBAPI
+ overlay network.Overlay
receiveC chan *ChunkDeliveryMsg
- getPeer func(discover.NodeID) *StreamerPeer
+ getPeer func(discover.NodeID) *Peer
quit chan struct{}
}
-func NewDelivery(overlay Overlay, dbAccess *DbAccess) *Delivery {
- self := &Delivery{
- dbAccess: dbAccess,
+func NewDelivery(overlay network.Overlay, db *storage.DBAPI) *Delivery {
+ d := &Delivery{
+ db: db,
overlay: overlay,
receiveC: make(chan *ChunkDeliveryMsg, 10),
}
- go self.processReceivedChunks()
- return self
+ go d.processReceivedChunks()
+ return d
}
-// RetrieveRequestStreamer implements OutgoingStreamer
-type RetrieveRequestStreamer struct {
+// SwarmChunkServer implements OutgoingStreamer
+type SwarmChunkServer struct {
deliveryC chan []byte
batchC chan []byte
- dbAccess *DbAccess
+ db *storage.DBAPI
currentLen uint64
}
-// NewRetrieveRequestStreamer is RetrieveRequestStreamer constructor
-func NewRetrieveRequestStreamer(dbAccess *DbAccess) *RetrieveRequestStreamer {
- s := &RetrieveRequestStreamer{
+// NewSwarmChunkServer is SwarmChunkServer constructor
+func NewSwarmChunkServer(db *storage.DBAPI) *SwarmChunkServer {
+ s := &SwarmChunkServer{
deliveryC: make(chan []byte),
batchC: make(chan []byte),
- dbAccess: dbAccess,
+ db: db,
}
go s.processDeliveries()
return s
}
// processDeliveries handles delivered chunk hashes
-func (s *RetrieveRequestStreamer) processDeliveries() {
+func (s *SwarmChunkServer) processDeliveries() {
var hashes []byte
var batchC chan []byte
for {
@@ -83,7 +84,7 @@ func (s *RetrieveRequestStreamer) processDeliveries() {
}
// 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
from = s.currentLen
s.currentLen += uint64(len(hashes))
@@ -92,8 +93,8 @@ func (s *RetrieveRequestStreamer) SetNextBatch(_, _ uint64) (hashes []byte, from
}
// GetData retrives chunk data from db store
-func (s *RetrieveRequestStreamer) GetData(key []byte) []byte {
- chunk, _ := s.dbAccess.get(storage.Key(key))
+func (s *SwarmChunkServer) GetData(key []byte) []byte {
+ chunk, _ := s.db.Get(storage.Key(key))
return chunk.SData
}
@@ -103,16 +104,16 @@ type RetrieveRequestMsg struct {
SkipCheck bool
}
-func (self *Delivery) handleRetrieveRequestMsg(sp *StreamerPeer, req *RetrieveRequestMsg) error {
- s, err := sp.getOutgoingStreamer(retrieveRequestStream)
+func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) error {
+ s, err := sp.getServer(swarmChunkServerStreamName)
if err != nil {
return err
}
- streamer := s.OutgoingStreamer.(*RetrieveRequestStreamer)
- chunk, created := self.dbAccess.getOrCreateRequest(req.Key)
+ streamer := s.Server.(*SwarmChunkServer)
+ chunk, created := d.db.GetOrCreateRequest(req.Key)
if chunk.ReqC != nil {
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
}
}
@@ -122,7 +123,7 @@ func (self *Delivery) handleRetrieveRequestMsg(sp *StreamerPeer, req *RetrieveRe
select {
case <-chunk.ReqC:
- case <-self.quit:
+ case <-d.quit:
return
case <-t.C:
return
@@ -149,21 +150,21 @@ type ChunkDeliveryMsg struct {
SData []byte // the stored chunk Data (incl size)
}
-func (self *Delivery) handleChunkDeliveryMsg(req *ChunkDeliveryMsg) error {
- chunk, err := self.dbAccess.get(req.Key)
+func (d *Delivery) handleChunkDeliveryMsg(req *ChunkDeliveryMsg) error {
+ chunk, err := d.db.Get(req.Key)
if err != nil {
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
}
-func (self *Delivery) processReceivedChunks() {
- for req := range self.receiveC {
- chunk, err := self.dbAccess.get(req.Key)
+func (d *Delivery) processReceivedChunks() {
+ for req := range d.receiveC {
+ chunk, err := d.db.Get(req.Key)
if err != nil {
continue
}
@@ -171,23 +172,23 @@ func (self *Delivery) processReceivedChunks() {
select {
case <-chunk.ReqC:
default:
- self.dbAccess.put(chunk)
+ d.db.Put(chunk)
close(chunk.ReqC)
}
}
}
// 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
- self.overlay.EachConn(hash, 255, func(p OverlayConn, po int, nn bool) bool {
- spId := p.(Peer).ID()
+ d.overlay.EachConn(hash, 255, func(p network.OverlayConn, po int, nn bool) bool {
+ spId := p.(*network.BzzPeer).ID()
for _, p := range peersToSkip {
if p == spId {
return true
}
}
- sp := self.getPeer(spId)
+ sp := d.getPeer(spId)
// TODO: skip light nodes that do not accept retrieve requests
err := sp.SendPriority(&RetrieveRequestMsg{
Key: hash,
diff --git a/swarm/network/request_test.go b/swarm/network/stream/delivery_test.go
similarity index 97%
rename from swarm/network/request_test.go
rename to swarm/network/stream/delivery_test.go
index 8ae1d3c344..317132be61 100644
--- a/swarm/network/request_test.go
+++ b/swarm/network/stream/delivery_test.go
@@ -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.
//
// 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
// along with the go-ethereum library. If not, see .
-package network
+package stream
import (
"bytes"
@@ -405,8 +405,8 @@ func newDeliveryService(ctx *adapters.ServiceContext) (node.Service, error) {
addr := NewAddrFromNodeID(id)
kad := NewKademlia(addr.Over(), NewKadParams())
localStore := localStores[nodeCount]
- dbAccess := NewDbAccess(localStore.(*storage.LocalStore))
- streamer := NewStreamer(NewDelivery(kad, dbAccess))
+ db := NewDBAPI(localStore.(*storage.LocalStore))
+ streamer := NewStreamerRegistry(NewDelivery(kad, db))
if nodeCount == 0 {
// the delivery service for the pivot node is assigned globally
// 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 {
- bzzPeer := &bzzPeer{
+ BzzPeer := &BzzPeer{
Peer: protocols.NewPeer(p, rw, StreamerSpec),
localAddr: b.addr,
BzzAddr: NewAddrFromNodeID(p.ID()),
}
- b.streamer.delivery.overlay.On(bzzPeer)
- defer b.streamer.delivery.overlay.Off(bzzPeer)
+ b.streamer.delivery.overlay.On(BzzPeer)
+ defer b.streamer.delivery.overlay.Off(BzzPeer)
go func() {
// each node Subscribes to each other's retrieveRequestStream
// 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)
}
}()
- return b.streamer.Run(bzzPeer)
+ return b.streamer.Run(BzzPeer)
}
diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go
new file mode 100644
index 0000000000..1e5e281b91
--- /dev/null
+++ b/swarm/network/stream/messages.go
@@ -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 .
+
+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{}
diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go
new file mode 100644
index 0000000000..5d2461a3c6
--- /dev/null
+++ b/swarm/network/stream/peer.go
@@ -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 .
+
+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
+}
diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go
new file mode 100644
index 0000000000..469ed9f126
--- /dev/null
+++ b/swarm/network/stream/stream.go
@@ -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 .
+
+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,
+ },
+ }
+}
diff --git a/swarm/network/streamer_test.go b/swarm/network/stream/streamer_test.go
similarity index 94%
rename from swarm/network/streamer_test.go
rename to swarm/network/stream/streamer_test.go
index 447713a33d..250573ade4 100644
--- a/swarm/network/streamer_test.go
+++ b/swarm/network/stream/streamer_test.go
@@ -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.
//
// 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
// along with the go-ethereum library. If not, see .
-package network
+package stream
import (
"bytes"
@@ -92,7 +92,7 @@ func TestStreamerDownstreamSubscribeMsgExchange(t *testing.T) {
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{
t: t,
}, nil
@@ -134,7 +134,7 @@ func TestStreamerUpstreamSubscribeMsgExchange(t *testing.T) {
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{
t: t,
}, nil
@@ -188,7 +188,7 @@ func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) {
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{
t: t,
}, nil
diff --git a/swarm/network/syncer.go b/swarm/network/stream/syncer.go
similarity index 50%
rename from swarm/network/syncer.go
rename to swarm/network/stream/syncer.go
index ec8a42808e..9523e4e440 100644
--- a/swarm/network/syncer.go
+++ b/swarm/network/stream/syncer.go
@@ -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.
//
// 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
// along with the go-ethereum library. If not, see .
-package network
+package stream
import (
"bytes"
@@ -29,88 +29,52 @@ import (
)
const (
- batchSize = 2
- // batchSize = 128
+ BatchSize = 2
+ // BatchSize = 128
)
-// wrapper of db-s to provide mockable custom local chunk store access to syncer
-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
+// SwarmSyncerServer implements an OutgoingStreamer for history syncing on bins
// offered streams:
// * live request delivery with or without checkback
// * (live/non-live historical) chunk syncing per proximity bin
-type OutgoingSwarmSyncer struct {
+type SwarmSyncerServer struct {
po uint8
- db *DbAccess
+ db *storage.DBAPI
sessionAt uint64
start uint64
}
-// NewOutgoingSwarmSyncer is contructor for OutgoingSwarmSyncer
-func NewOutgoingSwarmSyncer(live bool, po uint8, db *DbAccess) (*OutgoingSwarmSyncer, error) {
- sessionAt := db.currentBucketStorageIndex(po)
+// NewSwarmSyncerServer is contructor for SwarmSyncerServer
+func NewSwarmSyncerServer(live bool, po uint8, db *storage.DBAPI) (*SwarmSyncerServer, error) {
+ sessionAt := db.CurrentBucketStorageIndex(po)
var start uint64
if live {
start = sessionAt
}
- self := &OutgoingSwarmSyncer{
+ return &SwarmSyncerServer{
po: po,
db: db,
sessionAt: sessionAt,
start: start,
- }
- return self, nil
+ }, nil
}
const maxPO = 32
-func RegisterOutgoingSyncer(streamer *Streamer, db *DbAccess) {
- streamer.RegisterOutgoingStreamer("SYNC", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) {
+func RegisterSwarmSyncerServer(streamer *Registry, db *storage.DBAPI) {
+ streamer.RegisterServerFunc("SYNC", func(p *Peer, t []byte) (Server, error) {
po := uint8(t[0])
// 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)
// })
}
// GetSection retrieves the actual chunk from localstore
-func (self *OutgoingSwarmSyncer) GetData(key []byte) []byte {
- chunk, err := self.db.get(storage.Key(key))
+func (s *SwarmSyncerServer) GetData(key []byte) []byte {
+ chunk, err := s.db.Get(storage.Key(key))
if err != nil {
return nil
}
@@ -118,23 +82,23 @@ func (self *OutgoingSwarmSyncer) GetData(key []byte) []byte {
}
// 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
i := 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
}
ticker := time.NewTicker(10 * time.Millisecond)
defer ticker.Stop()
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[:]...)
i++
to = idx
- return i < batchSize
+ return i < BatchSize
})
if err != nil {
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
}
-// IncomingSwarmSyncer
-type IncomingSwarmSyncer struct {
+// SwarmSyncerClient
+type SwarmSyncerClient struct {
sessionAt uint64
nextC chan struct{}
sessionRoot storage.Key
sessionReader storage.LazySectionReader
retrieveC chan *storage.Chunk
storeC chan *storage.Chunk
- dbAccess *DbAccess
+ db *storage.DBAPI
chunker storage.Chunker
currentRoot storage.Key
requestFunc func(chunk *storage.Chunk)
end, start uint64
}
-// NewIncomingSwarmSyncer is a contructor for provable data exchange syncer
-func NewIncomingSwarmSyncer(p Peer, dbAccess *DbAccess, chunker storage.Chunker) (*IncomingSwarmSyncer, error) {
- self := &IncomingSwarmSyncer{
- dbAccess: dbAccess,
- chunker: chunker,
- }
- return self, nil
+// NewSwarmSyncerClient is a contructor for provable data exchange syncer
+func NewSwarmSyncerClient(_ *Peer, db *storage.DBAPI, chunker storage.Chunker) (*SwarmSyncerClient, error) {
+ return &SwarmSyncerClient{
+ db: db,
+ chunker: chunker,
+ }, nil
}
// // 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)
// RunChunkRequestor(p, retrieveC)
// storeC := make(storage.Chunk, chunksCap)
// RunChunkStorer(store, storeC)
-// self := &IncomingSwarmSyncer{
+// s := &SwarmSyncerClient{
// po: po,
// priority: priority,
// sessionAt: sessionAt,
@@ -191,10 +154,10 @@ func NewIncomingSwarmSyncer(p Peer, dbAccess *DbAccess, chunker storage.Chunker)
// retrieveC: retrieveC,
// 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
// func StartSyncing(s *Streamer, peerId discover.NodeID, po uint8, nn bool) {
// lastPO := po
@@ -208,15 +171,15 @@ func NewIncomingSwarmSyncer(p Peer, dbAccess *DbAccess, chunker storage.Chunker)
// }
// }
-func RegisterIncomingSyncer(streamer *Streamer, db *DbAccess) {
- streamer.RegisterIncomingStreamer("SYNC", func(p *StreamerPeer, t []byte) (IncomingStreamer, error) {
- return NewIncomingSwarmSyncer(p, db, nil)
+func RegisterSwarmSyncerClient(streamer *Registry, db *storage.DBAPI) {
+ streamer.RegisterClientFunc("SYNC", func(p *Peer, t []byte) (Client, error) {
+ return NewSwarmSyncerClient(p, db, nil)
})
}
// NeedData
-func (self *IncomingSwarmSyncer) NeedData(key []byte) (wait func()) {
- chunk, _ := self.dbAccess.getOrCreateRequest(key)
+func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) {
+ chunk, _ := s.db.GetOrCreateRequest(key)
// TODO: we may want to request from this peer anyway even if the request exists
if chunk.ReqC == nil {
return nil
@@ -226,29 +189,29 @@ func (self *IncomingSwarmSyncer) NeedData(key []byte) (wait func()) {
}
// BatchDone
-func (self *IncomingSwarmSyncer) BatchDone(s string, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) {
- if self.chunker != nil {
- return func() (*TakeoverProof, error) { return self.TakeoverProof(s, from, hashes, root) }
+func (s *SwarmSyncerClient) BatchDone(streamName string, from uint64, hashes []byte, root []byte) func() (*TakeoverProof, error) {
+ if s.chunker != nil {
+ return func() (*TakeoverProof, error) { return s.TakeoverProof(streamName, from, hashes, root) }
}
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
- if self.chunker != nil {
- if from > self.sessionAt { // for live syncing currentRoot is always updated
- //expRoot, err := self.chunker.Append(self.currentRoot, bytes.NewReader(hashes), self.retrieveC, self.storeC)
- expRoot, _, err := self.chunker.Append(self.currentRoot, bytes.NewReader(hashes), self.retrieveC)
+ if s.chunker != nil {
+ if from > s.sessionAt { // for live syncing currentRoot is always updated
+ //expRoot, err := s.chunker.Append(s.currentRoot, bytes.NewReader(hashes), s.retrieveC, s.storeC)
+ expRoot, _, err := s.chunker.Append(s.currentRoot, bytes.NewReader(hashes), s.retrieveC)
if err != nil {
return nil, err
}
if !bytes.Equal(root, expRoot) {
return nil, fmt.Errorf("HandoverProof mismatch")
}
- self.currentRoot = root
+ s.currentRoot = root
} else {
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 {
return nil, err
}
@@ -258,12 +221,12 @@ func (self *IncomingSwarmSyncer) TakeoverProof(s string, from uint64, hashes []b
}
return nil, nil
}
- self.end += uint64(len(hashes)) / HashSize
+ s.end += uint64(len(hashes)) / HashSize
takeover := &Takeover{
- Stream: s,
- // Key: self.Key,
- Start: self.start,
- End: self.end,
+ Stream: streamName,
+ // Key: s.Key,
+ Start: s.start,
+ End: s.end,
Root: root,
}
// serialise and sign
diff --git a/swarm/network/syncer_test.go b/swarm/network/stream/syncer_test.go
similarity index 85%
rename from swarm/network/syncer_test.go
rename to swarm/network/stream/syncer_test.go
index 893be367f5..c1a9bd6fac 100644
--- a/swarm/network/syncer_test.go
+++ b/swarm/network/stream/syncer_test.go
@@ -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.
//
// 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
// along with the go-ethereum library. If not, see .
-package network
+package stream
import (
"context"
@@ -32,6 +32,7 @@ import (
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
+ "github.com/ethereum/go-ethereum/swarm/network"
"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) {
- dbAccesses := make([]*DbAccess, nodes)
+ dbs := make([]*storage.DBAPI, nodes)
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) {
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
for i := 1; i < nodes; i++ {
- dbAccesses[i].iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool {
- _, err := dbAccesses[0].get(key)
+ dbs[i].iterator(0, math.MaxUint64, po, func(key storage.Key, index uint64) bool {
+ _, err := dbs[0].get(key)
if err == nil {
found++
}
@@ -105,7 +106,7 @@ func testSyncBetweenNodes(nodes, conns, size int, skipCheck bool, po uint8) func
}
}
toAddr := func(id discover.NodeID) *BzzAddr {
- addr := NewAddrFromNodeID(id)
+ addr := network.NewAddrFromNodeID(id)
addr.OAddr[0] = byte(0)
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) {
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
addr.OAddr[0] = byte(0)
kad := NewKademlia(addr.Over(), NewKadParams())
localStore := localStores[nodeCount]
- dbAccess := NewDbAccess(localStore.(*storage.LocalStore))
- streamer := NewStreamer(NewDelivery(kad, dbAccess))
- RegisterIncomingSyncer(streamer, dbAccess)
- RegisterOutgoingSyncer(streamer, dbAccess)
+ db := NewDbAccess(localStore.(*storage.LocalStore))
+ streamer := NewRegistry(NewDelivery(kad, db))
+ RegisterIncomingSyncer(streamer, db)
+ RegisterOutgoingSyncer(streamer, db)
self := &testStreamerService{
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 {
- addr := NewAddrFromNodeID(p.ID())
+ addr := network.NewAddrFromNodeID(p.ID())
addr.OAddr[0] = byte(0)
- bzzPeer := &bzzPeer{
- Peer: protocols.NewPeer(p, rw, StreamerSpec),
+ BzzPeer := &BzzPeer{
+ Peer: protocols.NewPeer(p, rw, Spec),
localAddr: b.addr,
BzzAddr: addr,
}
- b.streamer.delivery.overlay.On(bzzPeer)
- defer b.streamer.delivery.overlay.Off(bzzPeer)
+ b.streamer.delivery.overlay.On(BzzPeer)
+ defer b.streamer.delivery.overlay.Off(BzzPeer)
// if len(addr) > b.index+1 && bytes.Equal(addrs[b.index+1], addr) {
go func() {
// 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)
}
diff --git a/swarm/network/streamer_common_test.go b/swarm/network/stream/testing/testing.go
similarity index 57%
rename from swarm/network/streamer_common_test.go
rename to swarm/network/stream/testing/testing.go
index 7690ed75c8..e0da4d0336 100644
--- a/swarm/network/streamer_common_test.go
+++ b/swarm/network/stream/testing/testing.go
@@ -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.
//
// 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
// along with the go-ethereum library. If not, see .
-package network
+package testing
import (
"context"
"errors"
- "flag"
"fmt"
- "io"
"io/ioutil"
"math/rand"
"os"
@@ -33,44 +31,23 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
- "github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/p2p/simulations"
"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/swarm/network"
+ "github.com/ethereum/go-ethereum/swarm/network/stream"
"github.com/ethereum/go-ethereum/swarm/storage"
)
-var services = adapters.Services{
- "delivery": newDeliveryService,
- "syncer": newSyncerService,
-}
-
var (
- adapter = flag.String("adapter", "sim", "type of simulation: sim|socket|exec|docker")
- loglevel = flag.Int("loglevel", 2, "verbosity of logs")
+ LocalStores []storage.ChunkStore
+ Addrs []network.Addr
+ NodeCount int
)
-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
- localStores []storage.ChunkStore
- addrs []Addr
- fileHash storage.Key
- nodeCount int
-)
-
-func setLocalStores(addrs ...Addr) (func(), error) {
+func setLocalStores(addrs ...network.Addr) (func(), error) {
var datadirs []string
- localStores = make([]storage.ChunkStore, len(addrs))
+ LocalStores = make([]storage.ChunkStore, len(addrs))
var err error
for i, addr := range addrs {
// TODO: remove temp datadir after test
@@ -85,7 +62,7 @@ func setLocalStores(addrs ...Addr) (func(), error) {
break
}
datadirs = append(datadirs, datadir)
- localStores[i] = localStore
+ LocalStores[i] = localStore
}
teardown := func() {
for _, datadir := range datadirs {
@@ -95,27 +72,12 @@ func setLocalStores(addrs ...Addr) (func(), error) {
return teardown, err
}
-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 testSimulation(t *testing.T, simf func(adapters.NodeAdapter) (*simulations.StepResult, error)) {
+func testSimulation(t *testing.T, services adapters.Services, adapter string, simf func(adapters.NodeAdapter) (*simulations.StepResult, error)) {
var err error
var result *simulations.StepResult
startedAt := time.Now()
- switch *adapter {
+ switch adapter {
case "sim":
t.Logf("simadapter")
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))
}
-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
net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{
ID: "0",
@@ -166,8 +128,8 @@ func runSimulation(nodes, conns int, serviceName string, toAddr func(discover.No
})
defer net.Shutdown()
ids := make([]discover.NodeID, nodes)
- nodeCount = 0
- addrs = make([]Addr, nodes)
+ NodeCount = 0
+ Addrs = make([]network.Addr, nodes)
// start nodes
for i := 0; i < nodes; i++ {
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)
}
ids[i] = node.ID()
- addrs[i] = toAddr(ids[i])
+ Addrs[i] = toAddr(ids[i])
}
// set nodes number of localstores globally available
- teardown, err := setLocalStores(addrs...)
+ teardown, err := setLocalStores(Addrs...)
defer teardown()
if err != nil {
return nil, err
@@ -212,11 +174,11 @@ func runSimulation(nodes, conns int, serviceName string, toAddr func(discover.No
}
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
// if retriee requests have arrived
- dpa := storage.NewDPA(localStores[0], storage.NewChunkerParams())
+ dpa := storage.NewDPA(LocalStores[0], storage.NewChunkerParams())
dpa.Start()
defer dpa.Stop()
timeout := 300 * time.Second
@@ -233,47 +195,6 @@ func runSimulation(nodes, conns int, serviceName string, toAddr func(discover.No
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 {
index uint32
stores []storage.ChunkStore
@@ -301,34 +222,24 @@ func (rrs *roundRobinStore) Close() {
}
}
-func waitForPeers(streamer *Streamer, 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")
- }
- }
-}
-
-type testStreamerService struct {
+type TestStreamerService struct {
index int
- addr *BzzAddr
- streamer *Streamer
- run func(p *p2p.Peer, rw p2p.MsgReadWriter) error
+ addr *network.BzzAddr
+ streamer *stream.Registry
+ 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{
{
- Name: StreamerSpec.Name,
- Version: StreamerSpec.Version,
- Length: StreamerSpec.Length(),
+ Name: stream.Spec.Name,
+ Version: stream.Spec.Version,
+ Length: stream.Spec.Length(),
Run: tds.run,
// NodeInfo: ,
// 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{}
}
-func (b *testStreamerService) Start(server *p2p.Server) error {
+func (b *TestStreamerService) Start(server *p2p.Server) error {
return nil
}
-func (b *testStreamerService) Stop() error {
+func (b *TestStreamerService) Stop() error {
return nil
}
diff --git a/swarm/network/streamer.go b/swarm/network/streamer.go
deleted file mode 100644
index b0e8b9eb3b..0000000000
--- a/swarm/network/streamer.go
+++ /dev/null
@@ -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 .
-
-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)
- }
-}
diff --git a/swarm/storage/dbaccess.go b/swarm/storage/dbaccess.go
new file mode 100644
index 0000000000..69a659564a
--- /dev/null
+++ b/swarm/storage/dbaccess.go
@@ -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 .
+
+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)
+}
diff --git a/swarm/swarm.go b/swarm/swarm.go
index bc6533875b..6280d0fce7 100644
--- a/swarm/swarm.go
+++ b/swarm/swarm.go
@@ -39,6 +39,7 @@ import (
httpapi "github.com/ethereum/go-ethereum/swarm/api/http"
"github.com/ethereum/go-ethereum/swarm/fuse"
"github.com/ethereum/go-ethereum/swarm/network"
+ "github.com/ethereum/go-ethereum/swarm/network/stream"
"github.com/ethereum/go-ethereum/swarm/pss"
"github.com/ethereum/go-ethereum/swarm/storage"
"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
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
- streamer *network.Streamer
+ streamer *stream.Registry
//cloud storage.CloudStore // procurement, cloud storage backend (can multi-cloud)
bzz *network.Bzz // the logistic manager
backend chequebook.Backend // simple blockchain Backend
@@ -129,13 +130,13 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e
HiveParams: config.HiveParams,
}
- dbAccess := network.NewDbAccess(self.lstore)
- delivery := network.NewDelivery(to, dbAccess)
- self.streamer = network.NewStreamer(delivery)
- network.RegisterOutgoingSyncer(self.streamer, dbAccess)
- network.RegisterIncomingSyncer(self.streamer, dbAccess)
+ db := storage.NewDBAPI(self.lstore)
+ delivery := stream.NewDelivery(to, db)
+ self.streamer = stream.NewRegistry(delivery)
+ stream.RegisterSwarmSyncerServer(self.streamer, db)
+ 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
dpaChunkStore := storage.NewNetStore(self.lstore, self.streamer.Retrieve)
@@ -271,6 +272,11 @@ func (self *Swarm) Protocols() (protos []p2p.Protocol) {
protos = append(protos, p)
}
}
+ if self.streamer != nil {
+ for _, p := range self.streamer.Protocols() {
+ protos = append(protos, p)
+ }
+ }
return
}