From 140c7f7f23e0c1e498b3bc16f6294840a25b6c58 Mon Sep 17 00:00:00 2001 From: nolash Date: Mon, 13 Mar 2017 14:45:27 +0100 Subject: [PATCH] swarm/network: WIP protocol testfixes --- p2p/adapters/inproc.go | 11 +++--- p2p/protocols/protocol.go | 31 +++++++++-------- p2p/protocols/protocol_test.go | 6 ++-- p2p/testing/peerpool.go | 53 +++++++++++++++++++++++++++++ swarm/network/discovery.go | 12 +++---- swarm/network/discovery_test.go | 33 ++++++++++++++++-- swarm/network/hive.go | 4 +-- swarm/network/hive_test.go | 60 ++++++++++++++++++++++++++++++--- swarm/network/protocol.go | 1 + swarm/network/protocol_test.go | 24 +++++++------ 10 files changed, 188 insertions(+), 47 deletions(-) create mode 100644 p2p/testing/peerpool.go diff --git a/p2p/adapters/inproc.go b/p2p/adapters/inproc.go index 6c36ac5690..c808ebd6da 100644 --- a/p2p/adapters/inproc.go +++ b/p2p/adapters/inproc.go @@ -130,7 +130,8 @@ func (self *SimNode) Disconnect(rid []byte) error { // peer = na.(*SimNode).GetPeer(self.Id) // peer.RW = nil glog.V(6).Infof("dropped peer %v", id) - return self.network.DidDisconnect(self.Id, id) + + return nil } func (self *SimNode) Connect(rid []byte) error { @@ -155,7 +156,7 @@ func (self *SimNode) Connect(rid []byte) error { if err != nil { return fmt.Errorf("cannot run protocol (%v -> %v): %v", id, self.Id, err) } - self.network.DidConnect(self.Id, id) + return nil } @@ -172,11 +173,13 @@ func (self *SimNode) RunProtocol(id *NodeId, rw, rrw p2p.MsgReadWriter, runc cha peer = self.setPeer(id, self.Messenger(rrw)) p := p2p.NewPeer(id.NodeID, Name(id.Bytes()), []p2p.Cap{}) go func() { + self.network.DidConnect(self.Id, id) err := self.Run(p, rw) - 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 + glog.V(6).Infof("protocol quit on peer %v (connection with %v broken: %v)", self.Id, id, err) + self.network.DidDisconnect(self.Id, id) }() return nil } diff --git a/p2p/protocols/protocol.go b/p2p/protocols/protocol.go index 17c93e5ea7..2b44d933d5 100644 --- a/p2p/protocols/protocol.go +++ b/p2p/protocols/protocol.go @@ -168,12 +168,7 @@ func NewProtocol(protocolname string, protocolversion uint, run func(*Peer) erro m := na.Messenger(rw) - disc := func() { - id := p.ID() - na.Disconnect(id[:]) - } - - peer := NewPeer(p, ct, m, disc) + peer := NewPeer(p, ct, m) return run(peer) @@ -201,7 +196,6 @@ type Peer struct { *p2p.Peer // the p2p.Peer object representing the remote 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 } @@ -209,13 +203,12 @@ type Peer struct { // this constructor is called by the p2p.Protocol#Run function // the first two arguments are comming the arguments passed to p2p.Protocol.Run function // the third argument is the CodeMap describing the protocol messages and options -func NewPeer(p *p2p.Peer, ct *CodeMap, m adapters.Messenger, disconn func()) *Peer { +func NewPeer(p *p2p.Peer, ct *CodeMap, m adapters.Messenger) *Peer { return &Peer{ ct: ct, m: m, Peer: p, handlers: make(map[reflect.Type][]func(interface{}) error), - disconnect: disconn, } } @@ -244,14 +237,18 @@ func (self *Peer) Run() error { var err error for { _, err = self.handleIncoming() - if self.Err == nil { - err = self.Err + if self.Err != nil { + self.Err = err } - for _, f := range self.handlers[reflect.TypeOf(&Disconnect{})] { - f(err) + if self.Err != nil { + break } - return err + } + for _, f := range self.handlers[reflect.TypeOf(&Disconnect{})] { + f(err) + } + return err } // Drop disconnects a peer. @@ -283,6 +280,12 @@ func (self *Peer) Send(msg interface{}) error { return nil } +func (self *Peer) DisconnectHook(f func(e interface{}) error) { + typ := reflect.TypeOf(&Disconnect{}) + self.handlers[typ] = append(self.handlers[typ], f) +} + + // handleIncoming(code) // is called each cycle of the main forever loop that handles and dispatches incoming messages // if this returns an error the loop returns and the peer is disconnected with the error diff --git a/p2p/protocols/protocol_test.go b/p2p/protocols/protocol_test.go index 2c69c803a6..3f14a79350 100644 --- a/p2p/protocols/protocol_test.go +++ b/p2p/protocols/protocol_test.go @@ -158,7 +158,7 @@ func protoHandshakeExchange(id *adapters.NodeId, proto *protoHandshake) []p2ptes } func runProtoHandshake(t *testing.T, proto *protoHandshake, errs ...error) { - pp := NewTestPeerPool() + pp := p2ptest.NewTestPeerPool() s := protocolTester(t, pp, nil) // TODO: make this more than one handshake id := s.Ids[0] @@ -207,7 +207,7 @@ func moduleHandshakeExchange(id *adapters.NodeId, resp uint) []p2ptest.Exchange } func runModuleHandshake(t *testing.T, resp uint, errs ...error) { - pp := NewTestPeerPool() + pp := p2ptest.NewTestPeerPool() s := protocolTester(t, pp, nil) id := s.Ids[0] s.TestExchanges(protoHandshakeExchange(id, &protoHandshake{42, "420"})...) @@ -282,7 +282,7 @@ func testMultiPeerSetup(a, b *adapters.NodeId) []p2ptest.Exchange { func runMultiplePeers(t *testing.T, peer int, errs ...error) { wg := &sync.WaitGroup{} - pp := NewTestPeerPool() + pp := p2ptest.NewTestPeerPool() s := protocolTester(t, pp, wg) s.TestExchanges(testMultiPeerSetup(s.Ids[0], s.Ids[1])...) diff --git a/p2p/testing/peerpool.go b/p2p/testing/peerpool.go new file mode 100644 index 0000000000..41e971ed7a --- /dev/null +++ b/p2p/testing/peerpool.go @@ -0,0 +1,53 @@ +package testing + +import ( + "sync" + + "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" +) + +type TestPeer interface { + ID() discover.NodeID + Drop(error) +} + +// 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() + glog.V(logger.Detail).Infof("pp add peer %v", p.ID()) + 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/swarm/network/discovery.go b/swarm/network/discovery.go index 6700630be8..c9cddb0417 100644 --- a/swarm/network/discovery.go +++ b/swarm/network/discovery.go @@ -9,12 +9,12 @@ import ( "github.com/ethereum/go-ethereum/p2p/discover" ) -// discovery bzz hive extension doing peer relaying +// discovery bzz overlay extension doing peer relaying // can be switched off type discPeer struct { Peer - hive *Hive + overlay Overlay proxLimit uint8 peers map[discover.NodeID]bool } @@ -40,9 +40,9 @@ func (self *discPeer) NotifyProx(po uint8) error { } // new discovery contructor -func NewDiscovery(p Peer, h *Hive) *discPeer { +func NewDiscovery(p Peer, o Overlay) *discPeer { self := &discPeer{ - hive: h, + overlay: o, Peer: p, peers: make(map[discover.NodeID]bool), } @@ -118,7 +118,7 @@ func (p *discPeer) handlePeersMsg(msg interface{}) error { nas = append(nas, addr) p.peers[NodeId(addr).NodeID] = true } - return p.hive.Register(nas...) + return p.overlay.Register(nas...) } // handleGetPeersMsg is called by the protocol when receiving a @@ -131,7 +131,7 @@ func (p *discPeer) handleGetPeersMsg(msg interface{}) error { var peers []*peerAddr alreadySent := p.peers i := 0 - p.hive.EachLivePeer(p.OverlayAddr(), int(req.Order), func(n Peer, po int) bool { + p.overlay.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 diff --git a/swarm/network/discovery_test.go b/swarm/network/discovery_test.go index a86c3058c0..c57e5512ee 100644 --- a/swarm/network/discovery_test.go +++ b/swarm/network/discovery_test.go @@ -6,12 +6,39 @@ import ( p2ptest "github.com/ethereum/go-ethereum/p2p/testing" ) +/*** + * + * - after connect, that outgoing subpeersmsg is sent + * + */ func TestDiscovery(t *testing.T) { addr := RandomAddr() to := NewKademlia(addr.OAddr, NewKadParams()) - pp := NewHive(NewHiveParams(), to) + pp := p2ptest.NewTestPeerPool() + //pp := NewHive(NewHiveParams(), to) ct := BzzCodeMap(HiveMsgs...) - s := newBzzTester(t, addr, pp, ct, nil) + + services := func(p Peer) error { + dp := NewDiscovery(p, to) + //pp.Add(p) + to.On(dp) + p.DisconnectHook(func(e interface{}) error { + dp := e.(Peer) + to.Off(dp) + return nil + }) + return nil + } + /* + protocall := func (na adapters.NodeAdapter) adapters.ProtoCall { + protocol := Bzz(addr.OverlayAddr(), na, ct, services, nil, nil) + return protocol.Run + } + + s := p2ptest.NewProtocolTester(t, NodeId(addr), 1, protocall) +*/ + + s := newBzzTester(t, addr, pp, ct, services) s.runHandshakes() s.TestExchanges(p2ptest.Exchange{ @@ -19,7 +46,7 @@ func TestDiscovery(t *testing.T) { p2ptest.Expect{ Code: 3, Msg: &SubPeersMsg{ProxLimit: 0, MinProxBinSize: 8}, - Peer: s.ExchangeSession.Ids[1], + Peer: s.ExchangeSession.Ids[0], }, }, }) diff --git a/swarm/network/hive.go b/swarm/network/hive.go index 03d5c11c6e..8a43e65be1 100644 --- a/swarm/network/hive.go +++ b/swarm/network/hive.go @@ -216,11 +216,11 @@ func (self *Hive) wake() { // to register a connected (live) peer func (self *Hive) Add(p Peer) error { defer self.wake() - dp := NewDiscovery(p, self) + dp := NewDiscovery(p, self.Overlay) glog.V(logger.Debug).Infof("to add new bee %v", p) self.On(dp) glog.V(logger.Warn).Infof("%v", self) - dp.NotifyProx(0) + //dp.NotifyProx(0) return nil } diff --git a/swarm/network/hive_test.go b/swarm/network/hive_test.go index aaba6a6784..1d73ed190f 100644 --- a/swarm/network/hive_test.go +++ b/swarm/network/hive_test.go @@ -43,22 +43,40 @@ func TestOverlayRegistration(t *testing.T) { to := NewTestOverlay(addr.OverlayAddr()) // overlay topology driver pp := NewHive(NewHiveParams(), to) // hive ct := BzzCodeMap(HiveMsgs...) // bzz protocol code map - - s := newBzzTester(t, addr, pp, ct, nil) - - // connect to the other peer + services := func(p Peer) error { + pp.Add(p) + p.DisconnectHook(func(e interface{}) error { + pp.Remove(p) + return nil + }) + return nil + } + + protocall := func (na adapters.NodeAdapter) adapters.ProtoCall { + protocol := Bzz(addr.OverlayAddr(), na, ct, services, nil, nil) + return protocol.Run + } + + es := p2ptest.NewProtocolTester(t, NodeId(addr), 1, protocall) + + s := &bzzTester{ + addr: addr, + ExchangeSession: es, + } id := s.Ids[0] raddr := NewPeerAddrFromNodeId(id) + s.runHandshakes() // hive should have called the overlay if to.posMap[string(raddr.OverlayAddr())] == nil { t.Fatalf("Overlay#On not called on new peer") } + } func TestRegisterAndConnect(t *testing.T) { - addr := RandomAddr() + /*addr := RandomAddr() to := NewTestOverlay(addr.OverlayAddr()) pp := NewHive(NewHiveParams(), to) ct := BzzCodeMap(HiveMsgs...) @@ -86,7 +104,39 @@ func TestRegisterAndConnect(t *testing.T) { tc.ticker <- time.Now() // run bzz handshake + s.runHandshakes()*/ + + // 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 + services := func(p Peer) error { + pp.Add(p) + p.DisconnectHook(func(e interface{}) error { + pp.Remove(p) + return nil + }) + return nil + } + + protocall := func (na adapters.NodeAdapter) adapters.ProtoCall { + protocol := Bzz(addr.OverlayAddr(), na, ct, services, nil, nil) + return protocol.Run + } + + es := p2ptest.NewProtocolTester(t, NodeId(addr), 1, protocall) + + s := &bzzTester{ + addr: addr, + ExchangeSession: es, + } + id := s.Ids[0] + raddr := NewPeerAddrFromNodeId(id) + s.runHandshakes() + + if to.posMap[string(raddr.OverlayAddr())] == nil { t.Fatalf("Overlay#On not called on new peer") } diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index d076cee260..d30daea541 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -65,6 +65,7 @@ type Peer interface { Send(interface{}) error // can send messages Drop(error) // disconnect this peer Register(interface{}, func(interface{}) error) uint64 // register message-handler callbacks + DisconnectHook(func(interface{}) error) } func BzzCodeMap(msgs ...interface{}) *protocols.CodeMap { diff --git a/swarm/network/protocol_test.go b/swarm/network/protocol_test.go index d2f520be55..3f03f45d21 100644 --- a/swarm/network/protocol_test.go +++ b/swarm/network/protocol_test.go @@ -34,14 +34,18 @@ func bzzHandshakeExchange(lhs, rhs *bzzHandshake, id *adapters.NodeId) []p2ptest } } -func newBzzTester(t *testing.T, addr *peerAddr, pp *Hive, ct *protocols.CodeMap, services func(Peer) error) *bzzTester { + +func newBzzTester(t *testing.T, addr *peerAddr, pp *p2ptest.TestPeerPool, ct *protocols.CodeMap, services func(Peer) error) *bzzTester { if ct == nil { ct = BzzCodeMap() } extraservices := func(p Peer) error { pp.Add(p) - p.Register(&protocols.Disconnect{}, func(e interface{}) error { pp.Remove(p) }) - + p.DisconnectHook(func(e interface{}) error { + pp.Remove(p) + return nil + }) + if services != nil { err := services(p) if err != nil { @@ -50,14 +54,14 @@ func newBzzTester(t *testing.T, addr *peerAddr, pp *Hive, ct *protocols.CodeMap, } return nil } - - protocall := func(na adapters.NodeAdapter) adapters.ProtoCall { - protocol := Bzz(addr.OverlayAddr(), pp, na, ct, extraservices, nil, nil) + + protocall := func (na adapters.NodeAdapter) adapters.ProtoCall { + protocol := Bzz(addr.OverlayAddr(), na, ct, extraservices, nil, nil) return protocol.Run } - + s := p2ptest.NewProtocolTester(t, NodeId(addr), 1, protocall) - + return &bzzTester{ addr: addr, // flushCode: 4, @@ -161,7 +165,7 @@ func TestBzzPeerPoolRemove(t *testing.T) { s.runHandshakes() id := s.Ids[0] - pp.Get(id).Drop() + pp.Get(id).Drop(nil) s.TestDisconnected(&p2ptest.Disconnect{id, fmt.Errorf("p2p: read or write on closed message pipe")}) if pp.Has(id) { t.Fatalf("peer '%v' not removed: %v", id, pp) @@ -179,7 +183,7 @@ func TestBzzPeerPoolBothAddRemove(t *testing.T) { t.Fatalf("peer '%v' not added: %v", id, pp) } - pp.Get(id).Drop() + pp.Get(id).Drop(nil) s.TestDisconnected(&p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("p2p: read or write on closed message pipe")}) if pp.Has(id) { t.Fatalf("peer '%v' not removed: %v", id, pp)