From 79abcac95e865a9516dc8c9f39ec2777f558615b Mon Sep 17 00:00:00 2001 From: lash Date: Mon, 5 Mar 2018 17:32:25 +0100 Subject: [PATCH 1/2] swarm/network: Bundle syncer in swarm.go --- swarm/network/kademlia.go | 6 ++ swarm/network/protocol.go | 24 +++++--- swarm/network/stream/messages.go | 17 ++++++ swarm/network/stream/stream.go | 96 ++++++++++++++++++++++++++++++-- swarm/swarm.go | 4 +- 5 files changed, 133 insertions(+), 14 deletions(-) diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go index aa13923379..c120b0861e 100644 --- a/swarm/network/kademlia.go +++ b/swarm/network/kademlia.go @@ -329,6 +329,12 @@ func (k *Kademlia) Off(p OverlayConn) { } } +func (k *Kademlia) EachBin(base []byte, pof pot.Pof, o int, eachBinFunc func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool) { + k.lock.RLock() + defer k.lock.RUnlock() + k.conns.EachBin(base, pof, o, eachBinFunc) +} + // EachConn is an iterator with args (base, po, f) applies f to each live peer // that has proximity order po or less as measured from the base // if base is nil, kademlia base address is used diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index f59ab7c71d..325b49bfd3 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -113,9 +113,11 @@ type BzzConfig struct { // Bzz is the swarm protocol bundle type Bzz struct { *Hive - localAddr *BzzAddr - mtx sync.Mutex - handshakes map[discover.NodeID]*HandshakeMsg + localAddr *BzzAddr + mtx sync.Mutex + handshakes map[discover.NodeID]*HandshakeMsg + streamerSpec *protocols.Spec + streamerRun func(*BzzPeer) error } // NewBzz is the swarm protocol constructor @@ -123,11 +125,13 @@ type Bzz struct { // * bzz config // * overlay driver // * peer store -func NewBzz(config *BzzConfig, kad Overlay, store state.Store) *Bzz { +func NewBzz(config *BzzConfig, kad Overlay, store state.Store, streamerSpec *protocols.Spec, streamerRun func(*BzzPeer) error) *Bzz { return &Bzz{ - Hive: NewHive(config.HiveParams, kad, store), - localAddr: &BzzAddr{config.OverlayAddr, config.UnderlayAddr}, - handshakes: make(map[discover.NodeID]*HandshakeMsg), + Hive: NewHive(config.HiveParams, kad, store), + localAddr: &BzzAddr{config.OverlayAddr, config.UnderlayAddr}, + handshakes: make(map[discover.NodeID]*HandshakeMsg), + streamerRun: streamerRun, + streamerSpec: streamerSpec, } } @@ -166,6 +170,12 @@ func (b *Bzz) Protocols() []p2p.Protocol { NodeInfo: b.Hive.NodeInfo, PeerInfo: b.Hive.PeerInfo, }, + { + Name: b.streamerSpec.Name, + Version: b.streamerSpec.Version, + Length: b.streamerSpec.Length(), + Run: b.RunProtocol(b.streamerSpec, b.streamerRun), + }, } } diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index 0c2ffae6ef..675a4a64d8 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -61,6 +61,23 @@ type SubscribeMsg struct { Priority uint8 // delivered on priority channel } +// RequestSubscriptionMsg is the protocol msg for a node to request subscription to a +// specific stream +type RequestSubscriptionMsg struct { + Stream Stream + History *Range `rlp:"nil"` + Priority uint8 // delivered on priority channel +} + +func (p *Peer) handleRequestSubscription(req *RequestSubscriptionMsg) (err error) { + log.Debug(fmt.Sprintf("handleRequestSubscription: streamer %s to subscribe to %s with stream %s", p.streamer.addr.ID(), p.ID(), req.Stream)) + err = p.streamer.Subscribe(p.ID(), req.Stream, req.History, req.Priority) + if err != nil { + return err + } + return nil +} + func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) (err error) { defer func() { if err != nil { diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index 360567c559..0b8c23b362 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -22,12 +22,12 @@ import ( "math" "sync" - "github.com/ethereum/go-ethereum/p2p" - "github.com/ethereum/go-ethereum/rpc" - "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/pot" + "github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/network/stream/intervals" "github.com/ethereum/go-ethereum/swarm/state" @@ -123,6 +123,26 @@ func (r *Registry) GetServerFunc(stream string) (func(*Peer, []byte, bool) (Serv return f, nil } +func (r *Registry) RequestSubscription(peerId discover.NodeID, s Stream, h *Range, prio uint8) error { + // check if the stream is registered + if _, err := r.GetClientFunc(s.Name); err != nil { + return err + } + + peer := r.getPeer(peerId) + if peer == nil { + return fmt.Errorf("peer not found %v", peerId) + } + + msg := &RequestSubscriptionMsg{ + Stream: s, + History: h, + Priority: prio, + } + log.Debug("RequestSubscription ", "peer", peerId, "stream", s, "history", h) + return peer.Send(msg) +} + // Subscribe initiates the streamer func (r *Registry) Subscribe(peerId discover.NodeID, s Stream, h *Range, priority uint8) error { // check if the stream is registered @@ -225,12 +245,71 @@ func (r *Registry) peersCount() (c int) { } // Run protocol run function -func (r *Registry) run(p *protocols.Peer) error { - sp := NewPeer(p, r) +func (r *Registry) Run(p *network.BzzPeer) error { + sp := NewPeer(p.Peer, r) r.setPeer(sp) defer r.deletePeer(sp) defer close(sp.quit) defer sp.close() + + var kadDepth int + + r.delivery.overlay.EachConn(nil, 256, func(addr network.OverlayConn, po int, nn bool) bool { + // TODO: stop or expose by kademlia + if nn { + kadDepth = po + } + return true + }) + + kad, ok := r.delivery.overlay.(*network.Kademlia) + if !ok { + return fmt.Errorf("Not a Kademlia!") + } + + var startPo int + var endPo int + var i int + var err error + + //iterate over each bin and solicit needed subscription to bins + kad.EachBin(r.addr.Over(), pot.DefaultPof(256), 0, func(po, size int, f func(func(val pot.Val, i int) bool) bool) bool { + + //identify begin and start index of the bin(s) we want to subscribe to + if po < kadDepth { + //not nn + endPo = po + if i > 0 { + startPo = endPo + 1 + } + } else if endPo < kadDepth || endPo == 0 { + if po == 0 && kadDepth == 0 { + startPo = endPo + } else { + startPo = endPo + 1 + } + endPo = maxPO + } + + // now iterate and subscribe + for bin := po - startPo; bin <= endPo; bin++ { + + f(func(val pot.Val, i int) bool { + // a := val.(network.OverlayPeer) + log.Debug(fmt.Sprintf("Requesting subscription by: registry %s from peer %s for bin: %d", r.addr.ID(), p.ID(), bin)) + + err = r.RequestSubscription(p.ID(), NewStream("SYNC", []byte{uint8(bin)}, true), &Range{}, Top) + if err != nil { + log.Error(fmt.Sprintf("Error in RequestSubsciption! %v", err)) + return false + } + return true + }) + } + i++ + return true + }) + return sp.Run(sp.HandleMsg) } @@ -239,7 +318,7 @@ func (r *Registry) runProtocol(p *p2p.Peer, rw p2p.MsgReadWriter) error { bzzPeer := network.NewBzzTestPeer(peer, r.addr) r.delivery.overlay.On(bzzPeer) defer r.delivery.overlay.Off(bzzPeer) - return r.run(peer) + return r.Run(bzzPeer) } // HandleMsg is the message handler that delegates incoming messages @@ -270,6 +349,9 @@ func (p *Peer) HandleMsg(msg interface{}) error { case *RetrieveRequestMsg: return p.streamer.delivery.handleRetrieveRequestMsg(p, msg) + case *RequestSubscriptionMsg: + return p.handleRequestSubscription(msg) + default: return fmt.Errorf("unknown message type: %T", msg) } @@ -428,6 +510,7 @@ var Spec = &protocols.Spec{ RetrieveRequestMsg{}, ChunkDeliveryMsg{}, SubscribeErrorMsg{}, + RequestSubscriptionMsg{}, }, } @@ -457,6 +540,7 @@ func (r *Registry) APIs() []rpc.API { func (r *Registry) Start(server *p2p.Server) error { r.api.dpa.Start() + log.Info("Streamer started") return nil } diff --git a/swarm/swarm.go b/swarm/swarm.go index 9966da228d..1fc554ecc0 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -157,7 +157,7 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, config *api. stream.RegisterSwarmSyncerServer(self.streamer, db) stream.RegisterSwarmSyncerClient(self.streamer, db) - self.bzz = network.NewBzz(bzzconfig, to, stateStore) + self.bzz = network.NewBzz(bzzconfig, to, stateStore, stream.Spec, self.streamer.Run) // set up DPA, the cloud storage local access layer dpaChunkStore := storage.NewNetStore(self.lstore, self.streamer.Retrieve) @@ -378,6 +378,7 @@ func (self *Swarm) Start(srv *p2p.Server) error { self.periodicallyUpdateGauges() startCounter.Inc(1) + self.streamer.Start(srv) return nil } @@ -413,6 +414,7 @@ func (self *Swarm) Stop() error { } self.sfs.Stop() stopCounter.Inc(1) + self.streamer.Stop() return self.bzz.Stop() } From 86df48a2a1f87914dfcfa5dd7fe03de31ee53e7d Mon Sep 17 00:00:00 2001 From: Balint Gabor Date: Mon, 5 Mar 2018 18:16:46 +0100 Subject: [PATCH 2/2] swarm: Fix bzz initialization from tests --- swarm/network/protocol.go | 9 ++++++--- swarm/network/protocol_test.go | 2 +- swarm/network/simulations/discovery/discovery_test.go | 2 +- swarm/network/simulations/overlay.go | 2 +- swarm/pss/client/client_test.go | 2 +- swarm/pss/pss_test.go | 2 +- 6 files changed, 11 insertions(+), 8 deletions(-) diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index 325b49bfd3..d90ba22f30 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -154,7 +154,7 @@ func (b *Bzz) NodeInfo() interface{} { // * handshake/hive // * discovery func (b *Bzz) Protocols() []p2p.Protocol { - return []p2p.Protocol{ + protocol := []p2p.Protocol{ { Name: BzzSpec.Name, Version: BzzSpec.Version, @@ -170,13 +170,16 @@ func (b *Bzz) Protocols() []p2p.Protocol { NodeInfo: b.Hive.NodeInfo, PeerInfo: b.Hive.PeerInfo, }, - { + } + if b.streamerSpec != nil && b.streamerRun != nil { + protocol = append(protocol, p2p.Protocol{ Name: b.streamerSpec.Name, Version: b.streamerSpec.Version, Length: b.streamerSpec.Length(), Run: b.RunProtocol(b.streamerSpec, b.streamerRun), - }, + }) } + return protocol } // APIs returns the APIs offered by bzz diff --git a/swarm/network/protocol_test.go b/swarm/network/protocol_test.go index 19bc9bea7c..c9a8da1dd1 100644 --- a/swarm/network/protocol_test.go +++ b/swarm/network/protocol_test.go @@ -137,7 +137,7 @@ func newBzzHandshakeTester(t *testing.T, n int, addr *BzzAddr) *bzzTester { HiveParams: NewHiveParams(), } kad := NewKademlia(addr.OAddr, NewKadParams()) - bzz := NewBzz(config, kad, nil) + bzz := NewBzz(config, kad, nil, nil, nil) s := p2ptest.NewProtocolTester(t, NewNodeIDFromAddr(addr), 1, bzz.runBzz) diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go index 66f69794bf..5a29028f47 100644 --- a/swarm/network/simulations/discovery/discovery_test.go +++ b/swarm/network/simulations/discovery/discovery_test.go @@ -331,5 +331,5 @@ func newService(ctx *adapters.ServiceContext) (node.Service, error) { HiveParams: hp, } - return network.NewBzz(config, kad, nil), nil + return network.NewBzz(config, kad, nil, nil, nil), nil } diff --git a/swarm/network/simulations/overlay.go b/swarm/network/simulations/overlay.go index f69813e755..a82b1cd100 100644 --- a/swarm/network/simulations/overlay.go +++ b/swarm/network/simulations/overlay.go @@ -69,7 +69,7 @@ func (s *Simulation) NewService(ctx *adapters.ServiceContext) (node.Service, err HiveParams: hp, } - return network.NewBzz(config, kad, store), nil + return network.NewBzz(config, kad, store, nil, nil), nil } func createMockers() map[string]*simulations.MockerConfig { diff --git a/swarm/pss/client/client_test.go b/swarm/pss/client/client_test.go index 9ce4e856a8..10f0171197 100644 --- a/swarm/pss/client/client_test.go +++ b/swarm/pss/client/client_test.go @@ -263,7 +263,7 @@ func newServices() adapters.Services { UnderlayAddr: addr.Under(), HiveParams: hp, } - return network.NewBzz(config, kademlia(ctx.Config.ID), stateStore), nil + return network.NewBzz(config, kademlia(ctx.Config.ID), stateStore, nil, nil), nil }, } } diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 190d704f59..e26a7e600a 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -1192,7 +1192,7 @@ func newServices() adapters.Services { UnderlayAddr: addr.Under(), HiveParams: hp, } - return network.NewBzz(config, kademlia(ctx.Config.ID), stateStore), nil + return network.NewBzz(config, kademlia(ctx.Config.ID), stateStore, nil, nil), nil }, } }