From 7f22abaead2e0a6da2c0c611ec9c4a84dfd66de4 Mon Sep 17 00:00:00 2001 From: zelig Date: Mon, 13 Mar 2017 10:33:41 +0700 Subject: [PATCH] swarm/network, p2p: fixes --- p2p/adapters/inproc.go | 5 +- p2p/protocols/protocol.go | 40 +++++++++---- p2p/protocols/protocol_test.go | 82 +++++++++++---------------- p2p/testing/exchange.go | 75 +++++++------------------ p2p/testing/peerpool.go | 48 ---------------- p2p/testing/sessions.go | 12 ++-- swarm/network/discovery.go | 97 ++++++++++++++++---------------- swarm/network/discovery_test.go | 19 +++---- swarm/network/hive.go | 41 +++++--------- swarm/network/hive_test.go | 54 ++++++++---------- swarm/network/kademlia.go | 23 ++++++-- swarm/network/protocol.go | 34 +---------- swarm/network/protocol_test.go | 99 +++++++++++---------------------- 13 files changed, 231 insertions(+), 398 deletions(-) delete mode 100644 p2p/testing/peerpool.go diff --git a/p2p/adapters/inproc.go b/p2p/adapters/inproc.go index 0821d07559..6c36ac5690 100644 --- a/p2p/adapters/inproc.go +++ b/p2p/adapters/inproc.go @@ -34,7 +34,6 @@ func newPeer(m Messenger) *Peer { } type Peer struct { - //RW p2p.MsgReadWriter Messenger Errc chan error Flushc chan bool @@ -174,9 +173,9 @@ func (self *SimNode) RunProtocol(id *NodeId, rw, rrw p2p.MsgReadWriter, runc cha p := p2p.NewPeer(id.NodeID, Name(id.Bytes()), []p2p.Cap{}) go func() { err := self.Run(p, rw) - glog.V(6).Infof("protocol quit on peer %v (connection with %v broken)", self.Id, id) + glog.V(6).Infof("protocol quit on peer %v (connection with %v broken: %v)", self.Id, id, err) <-runc - self.Disconnect(id.Bytes()) + // self.Disconnect(id.Bytes()) peer.Errc <- err }() return nil diff --git a/p2p/protocols/protocol.go b/p2p/protocols/protocol.go index 51d1ffeea0..17c93e5ea7 100644 --- a/p2p/protocols/protocol.go +++ b/p2p/protocols/protocol.go @@ -37,6 +37,7 @@ import ( "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/adapters" + "github.com/ethereum/go-ethereum/p2p/discover" ) // error codes used by this protocol scheme @@ -158,8 +159,11 @@ func (self *CodeMap) Register(msgs ...interface{}) { } } -func NewProtocol(protocolname string, protocolversion uint, run func(*Peer) error, na adapters.NodeAdapter, ct *CodeMap) *p2p.Protocol { +func NewProtocol(protocolname string, protocolversion uint, run func(*Peer) error, na adapters.NodeAdapter, ct *CodeMap, peerInfo func(id discover.NodeID) interface{}, nodeInfo func() interface{}) *p2p.Protocol { + // PeerInfo is an optional helper method to retrieve protocol specific metadata + // about a certain peer in the network. If an info retrieval function is set, + // but returns nil, it is assumed that the protocol handshake is still running. r := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { m := na.Messenger(rw) @@ -176,13 +180,19 @@ func NewProtocol(protocolname string, protocolversion uint, run func(*Peer) erro } return &p2p.Protocol{ - Name: protocolname, - Version: protocolversion, - Length: ct.Length(), - Run: r, + Name: protocolname, + Version: protocolversion, + Length: ct.Length(), + Run: r, + PeerInfo: peerInfo, + NodeInfo: nodeInfo, } } +type Disconnect struct { + err error +} + // A Peer represents a remote peer or protocol instance that is running on a peer connection with // a remote peer type Peer struct { @@ -192,6 +202,7 @@ type Peer struct { rw p2p.MsgReadWriter // p2p.MsgReadWriter to send messages to and read messages from handlers map[reflect.Type][]func(interface{}) error // message type -> message handler callback(s) map disconnect func() // Disconnect function set differently for testing + Err error } // NewPeer returns a new peer @@ -222,7 +233,7 @@ func (self *Peer) Register(msg interface{}, handler func(interface{}) error) uin if !found { panic(fmt.Sprintf("message type '%v' unknown ", typ)) } - glog.V(logger.Detail).Infof("registered handle for %v %v", msg, typ) + glog.V(logger.Detail).Infof("register handle for %v", typ) self.handlers[typ] = append(self.handlers[typ], handler) return code } @@ -233,9 +244,13 @@ func (self *Peer) Run() error { var err error for { _, err = self.handleIncoming() - if err != nil { - return err + if self.Err == nil { + err = self.Err } + for _, f := range self.handlers[reflect.TypeOf(&Disconnect{})] { + f(err) + } + return err } } @@ -245,8 +260,8 @@ func (self *Peer) Run() error { // TODO: may need to implement protocol drop only? don't want to kick off the peer // if they are useful for other protocols // overwrite Disconnect for testing, so that protocol readloop quits -func (self *Peer) Drop() { - self.disconnect() +func (self *Peer) Drop(err error) { + self.Err = err } // Send takes a message, encodes it in RLP, finds the right message code and sends the @@ -261,8 +276,9 @@ func (self *Peer) Send(msg interface{}) error { glog.V(logger.Detail).Infof("=> %v (%d)", msg, code) err := self.m.SendMsg(uint64(code), msg) if err != nil { - self.Drop() - return errorf(ErrWrite, "(msg code: %v): %v", code, err) + err = errorf(ErrWrite, "(msg code: %v): %v", code, err) + self.Drop(err) + return err } return nil } diff --git a/p2p/protocols/protocol_test.go b/p2p/protocols/protocol_test.go index 71f0291cb2..2c69c803a6 100644 --- a/p2p/protocols/protocol_test.go +++ b/p2p/protocols/protocol_test.go @@ -4,7 +4,9 @@ import ( "fmt" "sync" "testing" + "time" + "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/adapters" @@ -12,7 +14,7 @@ import ( ) func init() { - glog.SetV(6) + glog.SetV(logger.Detail) glog.SetToStderr(true) } @@ -55,9 +57,8 @@ const networkId = "420" // newProtocol sets up a protocol // the run function here demonstrates a typical protocol using peerPool, handshake // and messages registered to handlers -func newProtocol(pp *p2ptest.TestPeerPool, wg *sync.WaitGroup) func(adapters.NodeAdapter) adapters.ProtoCall { +func newProtocol(pp *TestPeerPool, wg *sync.WaitGroup) func(adapters.NodeAdapter) adapters.ProtoCall { ct := NewCodeMap("test", 42, 1024, &protoHandshake{}, &hs0{}, &kill{}, &drop{}) - return func(na adapters.NodeAdapter) adapters.ProtoCall { return func(p *p2p.Peer, rw p2p.MsgReadWriter) error { if wg != nil { @@ -69,13 +70,16 @@ func newProtocol(pp *p2ptest.TestPeerPool, wg *sync.WaitGroup) func(adapters.Nod // demonstrates use of peerPool, killing another peer connection as a response to a message peer.Register(&kill{}, func(msg interface{}) error { id := msg.(*kill).C - pp.Get(id).Drop() + // pp.Get(id).Drop(fmt.Errorf("killed")) + glog.V(logger.Detail).Infof("id %v killed", id) return nil }) // for testing we can trigger self induced disconnect upon receiving drop message peer.Register(&drop{}, func(msg interface{}) error { - return fmt.Errorf("received disconnect request") + glog.V(logger.Detail).Infof("dropped") + // return fmt.Errorf("dropped") + return }) // initiate one-off protohandshake and check validity @@ -110,20 +114,21 @@ func newProtocol(pp *p2ptest.TestPeerPool, wg *sync.WaitGroup) func(adapters.Nod return peer.Send(lhs) }) - // add/remove peer from pool + glog.V(logger.Detail).Infof("adding peer %v", peer) pp.Add(peer) defer pp.Remove(peer) - // this launches a forever read loop err = peer.Run() if wg != nil { wg.Done() } + glog.V(logger.Detail).Infof("peer %v protocol quitting: %v", peer, err) + return err } } } -func protocolTester(t *testing.T, pp *p2ptest.TestPeerPool, wg *sync.WaitGroup) *p2ptest.ExchangeSession { +func protocolTester(t *testing.T, pp *TestPeerPool, wg *sync.WaitGroup) *p2ptest.ExchangeSession { id := p2ptest.RandomNodeId() return p2ptest.NewProtocolTester(t, id, 2, newProtocol(pp, wg)) } @@ -153,7 +158,7 @@ func protoHandshakeExchange(id *adapters.NodeId, proto *protoHandshake) []p2ptes } func runProtoHandshake(t *testing.T, proto *protoHandshake, errs ...error) { - pp := p2ptest.NewTestPeerPool() + pp := NewTestPeerPool() s := protocolTester(t, pp, nil) // TODO: make this more than one handshake id := s.Ids[0] @@ -202,7 +207,7 @@ func moduleHandshakeExchange(id *adapters.NodeId, resp uint) []p2ptest.Exchange } func runModuleHandshake(t *testing.T, resp uint, errs ...error) { - pp := p2ptest.NewTestPeerPool() + pp := NewTestPeerPool() s := protocolTester(t, pp, nil) id := s.Ids[0] s.TestExchanges(protoHandshakeExchange(id, &protoHandshake{42, "420"})...) @@ -227,6 +232,7 @@ func testMultiPeerSetup(a, b *adapters.NodeId) []p2ptest.Exchange { return []p2ptest.Exchange{ p2ptest.Exchange{ + Label: "primary handshake", Expects: []p2ptest.Expect{ p2ptest.Expect{ Code: 0, @@ -241,6 +247,7 @@ func testMultiPeerSetup(a, b *adapters.NodeId) []p2ptest.Exchange { }, }, p2ptest.Exchange{ + Label: "module handshake", Triggers: []p2ptest.Trigger{ p2ptest.Trigger{ Code: 0, @@ -266,53 +273,30 @@ func testMultiPeerSetup(a, b *adapters.NodeId) []p2ptest.Exchange { }, }, }, - p2ptest.Exchange{ - Triggers: []p2ptest.Trigger{ - p2ptest.Trigger{ - Code: 1, - Msg: &hs0{41}, - Peer: a, - }, - p2ptest.Trigger{ - Code: 1, - Msg: &hs0{41}, - Peer: b, - }, - }, - }, - p2ptest.Exchange{ - Triggers: []p2ptest.Trigger{ - p2ptest.Trigger{ - Code: 1, - Msg: &hs0{1}, - Peer: a, - }, - }, - }, - p2ptest.Exchange{ - Expects: []p2ptest.Expect{ - p2ptest.Expect{ - Code: 1, - Msg: &hs0{43}, - Peer: a, - }, - }, - }, - } + + p2ptest.Exchange{Label: "alternative module handshake", Triggers: []p2ptest.Trigger{p2ptest.Trigger{Code: 1, Msg: &hs0{41}, Peer: a}, + p2ptest.Trigger{Code: 1, Msg: &hs0{41}, Peer: b}}}, + p2ptest.Exchange{Label: "repeated module handshake", Triggers: []p2ptest.Trigger{p2ptest.Trigger{Code: 1, Msg: &hs0{1}, Peer: a}}}, + p2ptest.Exchange{Label: "receiving repeated module handshake", Expects: []p2ptest.Expect{p2ptest.Expect{Code: 1, Msg: &hs0{43}, Peer: a}}}} } func runMultiplePeers(t *testing.T, peer int, errs ...error) { wg := &sync.WaitGroup{} - pp := p2ptest.NewTestPeerPool() + pp := NewTestPeerPool() s := protocolTester(t, pp, wg) s.TestExchanges(testMultiPeerSetup(s.Ids[0], s.Ids[1])...) // after some exchanges of messages, we can test state changes // here this is simply demonstrated by the peerPool - // after the handshake negotiations peers must be addded to the pool - if !pp.Has(s.Ids[0]) { - t.Fatalf("missing peer test-0: %v (%v)", pp, s.Ids) + // after the handshake negotiations peers must be added to the pool + // time.Sleep(1) + for !pp.Has(s.Ids[0]) { + time.Sleep(1) + glog.V(logger.Detail).Infof("missing peer test-0: %v (%v)", pp, s.Ids) } + // if !pp.Has(s.Ids[0]) { + // t.Fatalf("missing peer test-0: %v (%v)", pp, s.Ids) + // } if !pp.Has(s.Ids[1]) { t.Fatalf("missing peer test-1: %v (%v)", pp, s.Ids) } @@ -355,13 +339,13 @@ func runMultiplePeers(t *testing.T, peer int, errs ...error) { func TestMultiplePeersDropSelf(t *testing.T) { runMultiplePeers(t, 0, fmt.Errorf("p2p: read or write on closed message pipe"), - fmt.Errorf("Message handler error: (msg code 3): received disconnect request"), + fmt.Errorf("Message handler error: (msg code 3): killed"), ) } func TestMultiplePeersDropOther(t *testing.T) { runMultiplePeers(t, 1, - fmt.Errorf("Message handler error: (msg code 3): received disconnect request"), + fmt.Errorf("Message handler error: (msg code 3): dropped"), fmt.Errorf("p2p: read or write on closed message pipe"), ) } diff --git a/p2p/testing/exchange.go b/p2p/testing/exchange.go index 9f32f71531..83b731facd 100644 --- a/p2p/testing/exchange.go +++ b/p2p/testing/exchange.go @@ -1,7 +1,3 @@ -// Package protocols helpers_test make it easier to -// write protocol tests by providing convenience functions and structures -// protocols uses these helpers for its own tests -// but ideally should sit in p2p/protocols/testing/ subpackage package testing import ( @@ -34,7 +30,6 @@ type TestNetAdapter interface { } type TestMessenger interface { - // MsgPipe([]byte, []byte) p2p,MsgPipe ExpectMsg(uint64, interface{}) error TriggerMsg(uint64, interface{}) error } @@ -42,6 +37,7 @@ type TestMessenger interface { // exchanges are the basic units of protocol tests // an exchange is defined on a session type Exchange struct { + Label string Triggers []Trigger Expects []Expect } @@ -62,8 +58,8 @@ type Expect struct { } type Disconnect struct { - Peer *adapters.NodeId // the peer that expects the message - Error error + Peer *adapters.NodeId // discconnected peer + Error error // disconnect reason } // NewExchangeTestSession takes a network session and Messenger @@ -81,12 +77,11 @@ func NewExchangeTestSession(t *testing.T, n TestNetAdapter, ids []*adapters.Node } } -type TestPeerInfo struct { - //RW p2p.MsgReadWriter - Messenger TestMessenger - Flushc chan bool - Errc chan error -} +// type PeerTester struct { +// Messenger TestMessenger +// Flushc chan bool +// Errc chan error +// } // trigger sends messages from peers func (self *ExchangeTestSession) trigger(trig Trigger) error { @@ -102,10 +97,10 @@ func (self *ExchangeTestSession) trigger(trig Trigger) error { errc := make(chan error) go func() { - glog.V(6).Infof("trigger....") + glog.V(6).Infof("trigger %v (%v)....", trig.Msg, trig.Code) //errc <- self.TriggerMsg(rw, trig.Code, trig.Msg) errc <- m.(TestMessenger).TriggerMsg(trig.Code, trig.Msg) - glog.V(6).Infof("triggered") + glog.V(6).Infof("triggered %v (%v)", trig.Msg, trig.Code) }() t := trig.Timeout @@ -168,18 +163,13 @@ func (self *ExchangeTestSession) TestExchanges(exchanges ...Exchange) { // launch all triggers of this exchanges for i, e := range exchanges { - errc := make(chan error) + errc := make(chan error, 1) wg := &sync.WaitGroup{} for _, trig := range e.Triggers { - wg.Add(1) - // separate go routing to allow parallel requests - go func(t Trigger) { - defer wg.Done() - err := self.trigger(t) - if err != nil { - errc <- err - } - }(trig) + err := self.trigger(trig) + if err != nil { + errc <- err + } } // each expectation is spawned in separate go-routine @@ -217,40 +207,16 @@ func (self *ExchangeTestSession) TestExchanges(exchanges ...Exchange) { if err != nil { self.t.Fatalf("exchange failed with: %v", err) } else { - glog.V(6).Infof("exchange %v run successfully", i) + glog.V(6).Infof("exchange %v: '%v' run successfully", i, e.Label) } case <-alarm.C: - self.t.Fatalf("exchange timed out") + self.t.Fatalf("exchange %v: '%v' timed out", i, e.Label) } } } -type flushMsg struct{} - -func flushExchange(c int, ids ...*adapters.NodeId) Exchange { - var triggers []Trigger - for _, id := range ids { - triggers = append(triggers, - Trigger{ - Code: uint64(c), - Msg: &flushMsg{}, - Peer: id, - }) - } - return Exchange{ - Triggers: triggers, - } -} - -var FlushMsg = &flushMsg{} - -func (self *ExchangeTestSession) TestConnected(flush bool, peers ...*adapters.NodeId) { +func (self *ExchangeTestSession) TestConnected(peers ...*adapters.NodeId) { timeout := time.NewTimer(1000 * time.Millisecond) - var flushc chan bool - if !flush { - flushc = make(chan bool) - close(flushc) - } wg := &sync.WaitGroup{} wg.Add(len(peers)) for _, id := range peers { @@ -260,15 +226,12 @@ func (self *ExchangeTestSession) TestConnected(flush bool, peers ...*adapters.No for { peer := self.GetPeer(p) if peer != nil { - if flush { - flushc = peer.Flushc - } select { case <-timeout.C: self.t.Fatalf("exchange timed out waiting for peer %v to flush", p) case err := <-peer.Errc: self.t.Fatalf("peer %v disconnected with error %v", p, err) - case <-flushc: + case <-peer.Flushc: glog.V(6).Infof("peer %v is connected", p) return } diff --git a/p2p/testing/peerpool.go b/p2p/testing/peerpool.go deleted file mode 100644 index 269336877f..0000000000 --- a/p2p/testing/peerpool.go +++ /dev/null @@ -1,48 +0,0 @@ -package testing - -import ( - "sync" - - "github.com/ethereum/go-ethereum/p2p/adapters" - "github.com/ethereum/go-ethereum/p2p/discover" -) - -type TestPeer interface { - ID() discover.NodeID - Drop() -} - -// TestPeerPool is an example peerPool to demonstrate registration of peer connections -type TestPeerPool struct { - lock sync.Mutex - peers map[discover.NodeID]TestPeer -} - -func NewTestPeerPool() *TestPeerPool { - return &TestPeerPool{peers: make(map[discover.NodeID]TestPeer)} -} - -func (self *TestPeerPool) Add(p TestPeer) { - self.lock.Lock() - defer self.lock.Unlock() - self.peers[p.ID()] = p -} - -func (self *TestPeerPool) Remove(p TestPeer) { - self.lock.Lock() - defer self.lock.Unlock() - delete(self.peers, p.ID()) -} - -func (self *TestPeerPool) Has(n *adapters.NodeId) bool { - self.lock.Lock() - defer self.lock.Unlock() - _, ok := self.peers[n.NodeID] - return ok -} - -func (self *TestPeerPool) Get(n *adapters.NodeId) TestPeer { - self.lock.Lock() - defer self.lock.Unlock() - return self.peers[n.NodeID] -} diff --git a/p2p/testing/sessions.go b/p2p/testing/sessions.go index 4054c6e8b5..a68d5afedf 100644 --- a/p2p/testing/sessions.go +++ b/p2p/testing/sessions.go @@ -62,13 +62,7 @@ func NewProtocolTester(t *testing.T, id *adapters.NodeId, n int, run func(id ada return self } -func (self *ExchangeTestSession) Flush(code int, ids ...*adapters.NodeId) { - self.TestConnected(false, ids...) - glog.V(6).Infof("flushing peers %v (code %v)", ids, code) - self.TestExchanges(flushExchange(code, ids...)) - self.TestConnected(true, ids...) -} - +// func (self *ExchangeSession) Start(id *adapters.NodeId) error { err := self.network.NewNode(&simulations.NodeConfig{Id: id}) if err != nil { @@ -103,6 +97,10 @@ func (self *ExchangeSession) Connect(ids ...*adapters.NodeId) { } +// func (self *ExchangeSession) Id(i int) *adapters.NodeId { +// return self.network.Nodes[i].Id +// } + func RandomNodeId() *adapters.NodeId { key, err := crypto.GenerateKey() if err != nil { diff --git a/swarm/network/discovery.go b/swarm/network/discovery.go index e4633db8e0..6700630be8 100644 --- a/swarm/network/discovery.go +++ b/swarm/network/discovery.go @@ -1,56 +1,57 @@ package network import ( + "bytes" "fmt" + + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/p2p/discover" ) -// discovery bzz hive extension for efficient peer relaying -// this will be triggered by p2p/protocol already +// discovery bzz hive extension doing peer relaying +// can be switched off type discPeer struct { Peer - hive Hive - sub overlaySubscription - peers map[discover.NodeID]bool + hive *Hive + proxLimit uint8 + peers map[discover.NodeID]bool } -type overlaySubscription interface { - Subscribe(proxLimit uint) chan interface{} - SubscribeProxChange(proxLimit uint) chan interface{} +// NotifyPeer notifies the receiver remote end of a peer p or PO po. +// callback for overlay driver +func (self *discPeer) NotifyPeer(p Peer, po uint8) error { + if po < self.proxLimit || self.peers[p.ID()] { + return nil + } + resp := &peersMsg{ + Peers: []*peerAddr{p.(*bzzPeer).peerAddr}, + } + return p.Send(resp) +} + +// NotifyProx sends a subPeers Msg to the receiver notifying them about +// a change in the prox limit (radius of the set including the nearest X peers +// or first empty row) +// callback for overlay driver +func (self *discPeer) NotifyProx(po uint8) error { + return self.Send(&SubPeersMsg{ProxLimit: po, MinProxBinSize: 8}) } // new discovery contructor -func NewDiscovery(p Peer, h Hive) error { - overlay, ok := h.Overlay.(overlaySubscription) - if !ok { - return fmt.Errorf("overlay does not support subscription") - } +func NewDiscovery(p Peer, h *Hive) *discPeer { self := &discPeer{ - sub: overlay, hive: h, Peer: p, - peers: make(map[discover.NodeID]Peer), + peers: make(map[discover.NodeID]bool), } - c := sub.SubscribeProxChange() - go func() { - for { - select { - case <-h.quit: - return - case p := <-c: - resp := &peersMsg{ - Peers: []*peerAddr{p.PeerAddr.(*peerAddr)}, - } - p.Send(resp) - } - } - }() - p.Register(&subPeersMsg{}, self.handleSubPeersMsg) p.Register(&peersMsg{}, self.handlePeersMsg) p.Register(&getPeersMsg{}, self.handleGetPeersMsg) + p.Register(&SubPeersMsg{}, self.handleSubPeersMsg) - return nil + return self } /* @@ -61,6 +62,7 @@ The encoding of a peer is identical to that in the devp2p base protocol peers messages: [IP, Port, NodeID] note that a node's DPA address is not the NodeID but the hash of the NodeID. +TODO: To mitigate against spurious peers messages, requests should be remembered and correctness of responses should be checked @@ -70,7 +72,7 @@ disconnected // peersMsg encapsulates an array of peer addresses // used for communicating about known peers -// relevvant for bootstrapping connectivity and updating peersets +// relevant for bootstrapping connectivity and updating peersets type peersMsg struct { Peers []*peerAddr } @@ -81,8 +83,8 @@ func (self peersMsg) String() string { // getPeersMsg is sent to (random) peers to request (Max) peers of a specific order type getPeersMsg struct { - Order uint - Max uint + Order uint8 + Max uint8 } func (self getPeersMsg) String() string { @@ -90,20 +92,19 @@ func (self getPeersMsg) String() string { } // subPeers msg is communicating the depth/sharpness/focus of the overlay table of a peer -type subPeersMsg struct { - MinProxBinSize uint - ProxLimit uint - // Offset uint - // Batch uint +type SubPeersMsg struct { + MinProxBinSize uint8 + ProxLimit uint8 } -func (self subPeersMsg) String() string { - return fmt.Sprintf("request peers > PO%02d. ProxLimit: %02d", self.Request, self.ProxLimit) +func (self SubPeersMsg) String() string { + return fmt.Sprintf("%T: request peers > PO%02d. ", self, self.ProxLimit) } func (self *discPeer) handleSubPeersMsg(msg interface{}) error { - spm := msg.(*subPeersMsg) - self.sub.Subscribe(spm.ProxLimit) + spm := msg.(*SubPeersMsg) + self.proxLimit = spm.ProxLimit + return nil } // handlePeersMsg called by the protocol when receiving peerset (for target address) @@ -115,7 +116,7 @@ func (p *discPeer) handlePeersMsg(msg interface{}) error { for _, na := range msg.(*peersMsg).Peers { addr := PeerAddr(na) nas = append(nas, addr) - p.peers[NodeId(addr)] = true + p.peers[NodeId(addr).NodeID] = true } return p.hive.Register(nas...) } @@ -130,7 +131,7 @@ func (p *discPeer) handleGetPeersMsg(msg interface{}) error { var peers []*peerAddr alreadySent := p.peers i := 0 - p.Overlay.EachLivePeer(p.OverlayAddr(), int(req.Order), func(n Peer, po int) bool { + p.hive.EachLivePeer(p.OverlayAddr(), int(req.Order), func(n Peer, po int) bool { i++ if bytes.Compare(n.OverlayAddr(), p.OverlayAddr()) != 0 && // only send peers we have not sent before in this session @@ -149,9 +150,5 @@ func (p *discPeer) handleGetPeersMsg(msg interface{}) error { resp := &peersMsg{ Peers: peers, } - err := p.Send(resp) - if err != nil { - return err - } - return nil + return p.Send(resp) } diff --git a/swarm/network/discovery_test.go b/swarm/network/discovery_test.go index 3229188824..cf098f9caf 100644 --- a/swarm/network/discovery_test.go +++ b/swarm/network/discovery_test.go @@ -1,30 +1,25 @@ package network import ( - "sync" "testing" - "time" - "github.com/ethereum/go-ethereum/p2p/adapters" - // "github.com/ethereum/go-ethereum/p2p/discover" - "github.com/ethereum/go-ethereum/logger" - "github.com/ethereum/go-ethereum/logger/glog" - "github.com/ethereum/go-ethereum/p2p/protocols" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" ) func TestDiscovery(t *testing.T) { addr := RandomAddr() - to := NewTestOverlay(addr.OverlayAddr()) + to := NewKademlia(addr.OAddr, NewKadParams()) pp := NewHive(NewHiveParams(), to) ct := BzzCodeMap(HiveMsgs...) - s := newDiscoveryTester(t, 0, addr, pp, ct, nil) + s := newBzzTester(t, 1, addr, pp, ct, nil) + + s.runHandshakes() s.TestExchanges(p2ptest.Exchange{ Expects: []p2ptest.Expect{ p2ptest.Expect{ - Code: 1, - Msg: &subPeersMsg{uint(o)}, - Peer: id, + Code: 3, + Msg: &SubPeersMsg{ProxLimit: 0, MinProxBinSize: 8}, + Peer: s.ExchangeSession.Id(1), }, }, }) diff --git a/swarm/network/hive.go b/swarm/network/hive.go index 3434776814..03d5c11c6e 100644 --- a/swarm/network/hive.go +++ b/swarm/network/hive.go @@ -17,13 +17,12 @@ package network import ( - "bytes" - "fmt" "sync" "time" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/p2p/adapters" "github.com/ethereum/go-ethereum/p2p/discover" ) @@ -57,7 +56,7 @@ type Overlay interface { type Hive struct { *HiveParams // settings Overlay // the overlay topology driver - disc Discovery + // disc Discovery lock sync.Mutex quit chan bool @@ -72,8 +71,8 @@ const ( ) type HiveParams struct { - PeersBroadcastSetSize uint - MaxPeersPerRequest uint + PeersBroadcastSetSize uint8 + MaxPeersPerRequest uint8 CallInterval uint } @@ -95,18 +94,17 @@ func NewHive(params *HiveParams, overlay Overlay) *Hive { } } -// messages that hive regusters handles for +// messages that hive handles var HiveMsgs = []interface{}{ &getPeersMsg{}, &peersMsg{}, - &subPeersMsg{}, + &SubPeersMsg{}, } // Start receives network info only at startup -// listedAddr is a function to retrieve listening address to advertise to peers // connectPeer is a function to connect to a peer based on its NodeID or enode URL +// these are called on the p2p.Server which runs on the node // af() returns an arbitrary ticker channel -// there are called on the p2p.Server which runs on the node func (self *Hive) Start(connectPeer func(string) error, af func() <-chan time.Time) error { self.toggle = make(chan bool) @@ -137,10 +135,10 @@ func (self *Hive) Start(connectPeer func(string) error, af func() <-chan time.Ti } if want { req := &getPeersMsg{ - Order: uint(order), + Order: uint8(order), Max: self.MaxPeersPerRequest, } - var i uint + var i uint8 var err error self.EachLivePeer(nil, order, func(n Peer, po int) bool { glog.V(logger.Detail).Infof("%T sent to %v", req, n.ID()) @@ -218,15 +216,12 @@ func (self *Hive) wake() { // to register a connected (live) peer func (self *Hive) Add(p Peer) error { defer self.wake() + dp := NewDiscovery(p, self) glog.V(logger.Debug).Infof("to add new bee %v", p) - self.On(p) + self.On(dp) glog.V(logger.Warn).Infof("%v", self) - - // self.lock.Lock() - // self.peers[p.ID()] = p - // self.lock.Unlock() - - return NewDiscovery(p, self) + dp.NotifyProx(0) + return nil } // Remove called after peer is disconnected @@ -234,9 +229,6 @@ func (self *Hive) Remove(p Peer) { defer self.wake() glog.V(logger.Debug).Infof("remove bee %v", p) self.Off(p) - self.lock.Lock() - delete(self.peers, p.ID()) - self.lock.Unlock() } func (self *Hive) NodeInfo() interface{} { @@ -246,11 +238,8 @@ func (self *Hive) NodeInfo() interface{} { func (self *Hive) PeerInfo(id discover.NodeID) interface{} { self.lock.Lock() defer self.lock.Unlock() - p, ok := self.peers[id] - if !ok { - return nil - } - return interface{}(&peerAddr{p.OverlayAddr(), p.UnderlayAddr()}) + addr := NewPeerAddrFromNodeId(adapters.NewNodeId(id[:])) + return interface{}(addr) } func HexToBytes(s string) []byte { diff --git a/swarm/network/hive_test.go b/swarm/network/hive_test.go index 9fe0ac5e05..7df68742fc 100644 --- a/swarm/network/hive_test.go +++ b/swarm/network/hive_test.go @@ -9,7 +9,7 @@ import ( // "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" - "github.com/ethereum/go-ethereum/p2p/protocols" + // "github.com/ethereum/go-ethereum/p2p/protocols" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" ) @@ -37,22 +37,14 @@ func (self *testConnect) connect(na string) error { return nil } -func newBzzHiveTester(t *testing.T, n int, addr *peerAddr, pp PeerPool, ct *protocols.CodeMap, services func(Peer) error) *bzzTester { - s := p2ptest.NewProtocolTester(t, NodeId(addr), n, newTestBzzProtocol(addr, pp, ct, services)) - return &bzzTester{ - addr: addr, - flushCode: 3, - ExchangeSession: s, - } -} - func TestOverlayRegistration(t *testing.T) { // setup addr := RandomAddr() // tested peers peer address to := NewTestOverlay(addr.OverlayAddr()) // overlay topology driver pp := NewHive(NewHiveParams(), to) // hive ct := BzzCodeMap(HiveMsgs...) // bzz protocol code map - s := newBzzHiveTester(t, 1, addr, pp, ct, nil) + + s := newBzzTester(t, 1, addr, ct, nil) // connect to the other peer id := s.Ids[0] @@ -70,14 +62,18 @@ func TestRegisterAndConnect(t *testing.T) { to := NewTestOverlay(addr.OverlayAddr()) pp := NewHive(NewHiveParams(), to) ct := BzzCodeMap(HiveMsgs...) - s := newBzzHiveTester(t, 0, addr, pp, ct, nil) + s := newBzzTester(t, 0, addr, pp, ct, nil) // register the node with the peerPool id := p2ptest.RandomNodeId() + // pretend to start the node s.Start(id) + + // register another address raddr := NewPeerAddrFromNodeId(id) pp.Register(raddr) glog.V(5).Infof("%v", pp) + // start the hive and wait for the connection tc := &testConnect{ connectf: func(c string) error { @@ -88,42 +84,36 @@ func TestRegisterAndConnect(t *testing.T) { } pp.Start(tc.connect, tc.ping) tc.ticker <- time.Now() + + // run bzz handshake s.runHandshakes() if to.posMap[string(raddr.OverlayAddr())] == nil { t.Fatalf("Overlay#On not called on new peer") } - glog.V(6).Infof("check peer requests for %v", id) - // tc.ticker <- time.Now() - // shakeHands(s, addr, id) - // s.Flush(int(ct.Length())-1, 0) - // time.Sleep(3) + // retrieve and broadcast + glog.V(6).Infof("check peer requests for %v", id) ord := order(raddr.OverlayAddr()) o := 0 if ord == 0 { o = 1 } + s.TestExchanges(p2ptest.Exchange{ + Expects: []p2ptest.Expect{ + p2ptest.Expect{ + Code: 3, + Msg: &SubPeersMsg{ProxLimit: 0, MinProxBinSize: 8}, + Peer: s.ExchangeSession.Id(1), + }, + }, + }) s.TestExchanges(p2ptest.Exchange{ Expects: []p2ptest.Expect{ p2ptest.Expect{ Code: 1, - Msg: &getPeersMsg{uint(o), 5}, + Msg: &getPeersMsg{uint8(o), 5}, Peer: id, }, }, - // Triggers: []p2ptest.Trigger{ - // p2ptest.Trigger{ - // Code: 1, - // Msg: &getPeersMsg{0, 1}, - // Peer: 0, - // }, - // }, - // Expects: []p2ptest.Expect{ - // p2ptest.Expect{ - // Code: 1, - // Msg: &peersMsg{[]*peerAddr{RandomAddr()}}, - // Peer: 0, - // }, - // }, }) } diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go index c44adc066c..d50e34192f 100644 --- a/swarm/network/kademlia.go +++ b/swarm/network/kademlia.go @@ -50,6 +50,10 @@ adjust Prox (proxLimit and proxSize after an insertion/removal of nodes) caller holds the lock */ +type KadDiscovery interface { + NotifyPeer(Peer, uint8) error + NotifyProx(uint8) error +} type KadParams struct { // adjustable parameters @@ -199,12 +203,21 @@ func (self *Kademlia) On(p Peer) { }) kp.seenAt = time.Now() kp.retries = 0 - f := func(val pot.PotVal, po int) { - vp := val.(*KadPeer).Peer - dp.NotifyPeer(kp.PeerAddr, po) - } + prox := self.proxLimit() - self.conns.EachNeighbourAsync(pp, 256, 256, f, nil) + vp, ok := kp.Peer.(KadDiscovery) + if !ok { + glog.V(logger.Detail).Infof("not discovery peer") + return + } + // vp.NotifyProx(uint8(prox)) + f := func(val pot.PotVal, po int) { + glog.V(logger.Detail).Infof("peer %v nofified", vp) + dp := val.(KadDiscovery) + dp.NotifyPeer(kp.Peer, uint8(po)) + dp.NotifyProx(uint8(prox)) + } + self.conns.EachNeighbourAsync(pp, 255, 255, f, false) } // Off removes a peer from among live peers diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index 5bdb325027..72b40948c4 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -39,7 +39,6 @@ const ( // bzz is the bzz protocol view of a protocols.Peer (itself an extension of p2p.Peer) type bzzPeer struct { *protocols.Peer - hive PeerPool network adapters.NodeAdapter localAddr *peerAddr *peerAddr // remote address @@ -68,22 +67,10 @@ type Peer interface { ID() discover.NodeID // the key that uniquely identifies the Node for the peerPool Peers() map[discover.NodeID]bool Send(interface{}) error // can send messages - Drop() // disconnect this peer + Drop(error) // disconnect this peer Register(interface{}, func(interface{}) error) uint64 // register message-handler callbacks } -// PeerPool is the interface for the connectivity manager -// directly interacts with the p2p server to suggest connections -type PeerPool interface { - Add(Peer) error - Remove(Peer) -} - -type PeerInfo interface { - NodeInfo() interface{} - PeerInfo(discover.NodeID) interface{} -} - func BzzCodeMap(msgs ...interface{}) *protocols.CodeMap { ct := protocols.NewCodeMap(ProtocolName, Version, ProtocolMaxMsgSize) ct.Register(&bzzHandshake{}) @@ -93,13 +80,12 @@ func BzzCodeMap(msgs ...interface{}) *protocols.CodeMap { // Bzz is the protocol constructor // returns p2p.Protocol that is to be offered by the node.Service -func Bzz(localAddr []byte, hive PeerPool, na adapters.NodeAdapter, ct *protocols.CodeMap, services func(Peer) error) *p2p.Protocol { +func Bzz(localAddr []byte, na adapters.NodeAdapter, ct *protocols.CodeMap, services func(Peer) error) *p2p.Protocol { run := func(p *protocols.Peer) error { addr := &peerAddr{localAddr, na.LocalAddr()} bee := &bzzPeer{ Peer: p, - hive: hive, network: na, localAddr: addr, peers: make(map[discover.NodeID]bool), @@ -121,24 +107,10 @@ func Bzz(localAddr []byte, hive PeerPool, na adapters.NodeAdapter, ct *protocols } } - err = hive.Add(bee) - if err != nil { - glog.V(6).Infof("failed to add peer '%v' to hive: %v", bee.ID(), err) - return err - } - - defer hive.Remove(bee) return bee.Run() } - proto := protocols.NewProtocol(ProtocolName, Version, run, na, ct) - - if o, ok := hive.(PeerInfo); ok { - proto.NodeInfo = o.NodeInfo - proto.PeerInfo = o.PeerInfo - } - - return proto + return protocols.NewProtocol(ProtocolName, Version, run, na, ct, peerInfo, nodeInfo) } /* diff --git a/swarm/network/protocol_test.go b/swarm/network/protocol_test.go index 4fbf7fd59c..5ac1c122e3 100644 --- a/swarm/network/protocol_test.go +++ b/swarm/network/protocol_test.go @@ -4,7 +4,6 @@ import ( "fmt" "sync" "testing" - "time" "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/p2p/adapters" @@ -37,11 +36,11 @@ func bzzHandshakeExchange(lhs, rhs *bzzHandshake, id *adapters.NodeId) []p2ptest } } -func newTestBzzProtocol(addr *peerAddr, pp PeerPool, ct *protocols.CodeMap, services func(Peer) error) func(adapters.NodeAdapter) adapters.ProtoCall { +func newTestBzzProtocol(addr *peerAddr, ct *protocols.CodeMap, services func(Peer) error) func(adapters.NodeAdapter) adapters.ProtoCall { if ct == nil { ct = BzzCodeMap() } - ct.Register(p2ptest.FlushMsg) + // ct.Register(p2ptest.FlushMsg) return func(na adapters.NodeAdapter) adapters.ProtoCall { srv := func(p Peer) error { if services != nil { @@ -50,12 +49,12 @@ func newTestBzzProtocol(addr *peerAddr, pp PeerPool, ct *protocols.CodeMap, serv return err } } - id := p.ID() - p.Register(p2ptest.FlushMsg, func(interface{}) error { - flushc := na.(p2ptest.TestNetAdapter).GetPeer(&adapters.NodeId{id}).Flushc - flushc <- true - return nil - }) + // id := p.ID() + // p.Register(p2ptest.FlushMsg, func(interface{}) error { + // flushc := na.(p2ptest.TestNetAdapter).GetPeer(&adapters.NodeId{id}).Flushc + // flushc <- true + // return nil + // }) return nil } @@ -66,8 +65,8 @@ func newTestBzzProtocol(addr *peerAddr, pp PeerPool, ct *protocols.CodeMap, serv type bzzTester struct { *p2ptest.ExchangeSession - flushCode int - addr *peerAddr + // flushCode int + addr *peerAddr } // should test handshakes in one exchange? parallelisation @@ -81,27 +80,23 @@ func (s *bzzTester) testHandshake(lhs, rhs *bzzHandshake, disconnects ...*p2ptes } else { peers = []*adapters.NodeId{id} } - s.TestConnected(false, peers...) + s.TestConnected(peers...) s.TestExchanges(bzzHandshakeExchange(lhs, rhs, id)...) s.TestDisconnected(disconnects...) } -func (s *bzzTester) flush(ids ...*adapters.NodeId) { - s.Flush(s.flushCode, ids...) -} +// func (s *bzzTester) flush(ids ...*adapters.NodeId) { +// s.Flush(s.flushCode, ids...) +// } func (s *bzzTester) runHandshakes(ids ...*adapters.NodeId) { if len(ids) == 0 { ids = s.Ids } for _, id := range ids { - glog.V(6).Infof("\n\n\nrun handshake with %v", id) - time.Sleep(1) s.testHandshake(correctBzzHandshake(s.addr), correctBzzHandshake(NewPeerAddrFromNodeId(id))) - time.Sleep(1) } - glog.V(6).Infof("flush %v", ids) - s.flush(ids...) + } func correctBzzHandshake(addr *peerAddr) *bzzHandshake { @@ -109,16 +104,22 @@ func correctBzzHandshake(addr *peerAddr) *bzzHandshake { } func newBzzTester(t *testing.T, addr *peerAddr, pp PeerPool, ct *protocols.CodeMap, services func(Peer) error) *bzzTester { - s := p2ptest.NewProtocolTester(t, NodeId(addr), 1, newTestBzzProtocol(addr, pp, ct, services)) + + extraservices := func(p Peer) error { + pp.Add(p) + p.Register(&protocols.Disconnect{}, func(e interface{}) error { pp.Remove(p) }) + return services(p) + } + s := p2ptest.NewProtocolTester(t, NodeId(addr), 1, newTestBzzProtocol(addr, pp, ct, extarservices)) return &bzzTester{ - addr: addr, - flushCode: 1, + addr: addr, + // flushCode: 4, ExchangeSession: s, } } func TestBzzHandshakeNetworkIdMismatch(t *testing.T) { - pp := NewTestPeerPool() + pp := p2ptest.NewTestPeerPool() addr := RandomAddr() s := newBzzTester(t, addr, pp, nil, nil) id := s.Ids[0] @@ -130,7 +131,7 @@ func TestBzzHandshakeNetworkIdMismatch(t *testing.T) { } func TestBzzHandshakeVersionMismatch(t *testing.T) { - pp := NewTestPeerPool() + pp := p2ptest.NewTestPeerPool() addr := RandomAddr() s := newBzzTester(t, addr, pp, nil, nil) id := s.Ids[0] @@ -142,7 +143,7 @@ func TestBzzHandshakeVersionMismatch(t *testing.T) { } func TestBzzHandshakeSuccess(t *testing.T) { - pp := NewTestPeerPool() + pp := p2ptest.NewTestPeerPool() addr := RandomAddr() s := newBzzTester(t, addr, pp, nil, nil) id := s.Ids[0] @@ -153,13 +154,14 @@ func TestBzzHandshakeSuccess(t *testing.T) { } func TestBzzPeerPoolAdd(t *testing.T) { - pp := NewTestPeerPool() + pp := p2ptest.NewTestPeerPool() addr := RandomAddr() s := newBzzTester(t, addr, pp, nil, nil) id := s.Ids[0] glog.V(6).Infof("handshake with %v", id) s.runHandshakes() + // s.TestConnected() if !pp.Has(id) { t.Fatalf("peer '%v' not added: %v", id, pp) } @@ -167,7 +169,7 @@ func TestBzzPeerPoolAdd(t *testing.T) { func TestBzzPeerPoolRemove(t *testing.T) { addr := RandomAddr() - pp := NewTestPeerPool() + pp := p2ptest.NewTestPeerPool() s := newBzzTester(t, addr, pp, nil, nil) s.runHandshakes() @@ -181,7 +183,7 @@ func TestBzzPeerPoolRemove(t *testing.T) { func TestBzzPeerPoolBothAddRemove(t *testing.T) { addr := RandomAddr() - pp := NewTestPeerPool() + pp := p2ptest.NewTestPeerPool() s := newBzzTester(t, addr, pp, nil, nil) s.runHandshakes() @@ -199,7 +201,7 @@ func TestBzzPeerPoolBothAddRemove(t *testing.T) { func TestBzzPeerPoolNotAdd(t *testing.T) { addr := RandomAddr() - pp := NewTestPeerPool() + pp := p2ptest.NewTestPeerPool() s := newBzzTester(t, addr, pp, nil, nil) id := s.Ids[0] @@ -208,40 +210,3 @@ func TestBzzPeerPoolNotAdd(t *testing.T) { t.Fatalf("peer %v incorrectly added: %v", id, pp) } } - -// TestPeerPool is an example peerPool to demonstrate registration of peer connections -type TestPeerPool struct { - lock sync.Mutex - peers map[discover.NodeID]Peer -} - -func NewTestPeerPool() *TestPeerPool { - return &TestPeerPool{peers: make(map[discover.NodeID]Peer)} -} - -func (self *TestPeerPool) Add(p Peer) error { - self.lock.Lock() - defer self.lock.Unlock() - self.peers[p.ID()] = p - return nil -} - -func (self *TestPeerPool) Remove(p Peer) { - self.lock.Lock() - defer self.lock.Unlock() - // glog.V(6).Infof("removing peer %v", p.ID()) - delete(self.peers, p.ID()) -} - -func (self *TestPeerPool) Has(n *adapters.NodeId) bool { - self.lock.Lock() - defer self.lock.Unlock() - _, ok := self.peers[n.NodeID] - return ok -} - -func (self *TestPeerPool) Get(n *adapters.NodeId) Peer { - self.lock.Lock() - defer self.lock.Unlock() - return self.peers[n.NodeID] -}