diff --git a/p2p/protocols/protocol.go b/p2p/protocols/protocol.go index 2b44d933d5..fe0c792b36 100644 --- a/p2p/protocols/protocol.go +++ b/p2p/protocols/protocol.go @@ -159,7 +159,7 @@ func (self *CodeMap) Register(msgs ...interface{}) { } } -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 { +func NewProtocol(protocolname string, protocolversion uint, run func(*Peer) error, na adapters.NodeAdapter, ct *CodeMap, peerInfo func(id discover.NodeID) interface{}, nodeInfo func() interface{}, connectHook func(*Peer)) *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, @@ -169,7 +169,9 @@ func NewProtocol(protocolname string, protocolversion uint, run func(*Peer) erro m := na.Messenger(rw) peer := NewPeer(p, ct, m) - + if connectHook != nil { + connectHook(peer) + } return run(peer) } @@ -191,12 +193,12 @@ type Disconnect struct { // A Peer represents a remote peer or protocol instance that is running on a peer connection with // a remote peer type Peer struct { - ct *CodeMap // CodeMap for the protocol - m adapters.Messenger // defines senf and receive - *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 - Err error + ct *CodeMap // CodeMap for the protocol + m adapters.Messenger // defines senf and receive + *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 + Err error } // NewPeer returns a new peer @@ -205,10 +207,10 @@ type Peer struct { // the third argument is the CodeMap describing the protocol messages and options 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), + ct: ct, + m: m, + Peer: p, + handlers: make(map[reflect.Type][]func(interface{}) error), } } @@ -243,7 +245,7 @@ func (self *Peer) Run() error { if self.Err != nil { break } - + } for _, f := range self.handlers[reflect.TypeOf(&Disconnect{})] { f(err) @@ -285,7 +287,6 @@ func (self *Peer) DisconnectHook(f func(e interface{}) error) { 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/testing/exchange.go b/p2p/testing/exchange.go index 83b731facd..d43fd9cb8f 100644 --- a/p2p/testing/exchange.go +++ b/p2p/testing/exchange.go @@ -77,19 +77,12 @@ func NewExchangeTestSession(t *testing.T, n TestNetAdapter, ids []*adapters.Node } } -// type PeerTester struct { -// Messenger TestMessenger -// Flushc chan bool -// Errc chan error -// } - // trigger sends messages from peers func (self *ExchangeTestSession) trigger(trig Trigger) error { peer := self.GetPeer(trig.Peer) if peer == nil { panic(fmt.Sprintf("trigger: peer %v does not exist (1- %v)", trig.Peer, len(self.Ids))) } - //rw := peer.RW m := peer.Messenger if m == nil { return fmt.Errorf("trigger: peer %v unreachable", trig.Peer) @@ -98,7 +91,6 @@ func (self *ExchangeTestSession) trigger(trig Trigger) error { go func() { 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 %v (%v)", trig.Msg, trig.Code) }() @@ -129,7 +121,6 @@ func (self *ExchangeTestSession) expect(exp Expect) error { if peer == nil { panic(fmt.Sprintf("expect: peer %v does not exist (1- %v)", exp.Peer, len(self.Ids))) } - //rw := peer.RW m := peer.Messenger if m == nil { return fmt.Errorf("trigger: peer %v unreachable", exp.Peer) @@ -138,7 +129,6 @@ func (self *ExchangeTestSession) expect(exp Expect) error { errc := make(chan error) go func() { glog.V(6).Infof("waiting for msg, %v", exp.Msg) - //errc <- self.ExpectMsg(rw, exp.Code, exp.Msg) errc <- m.(TestMessenger).ExpectMsg(exp.Code, exp.Msg) }() @@ -215,40 +205,40 @@ func (self *ExchangeTestSession) TestExchanges(exchanges ...Exchange) { } } -func (self *ExchangeTestSession) TestConnected(peers ...*adapters.NodeId) { - timeout := time.NewTimer(1000 * time.Millisecond) - wg := &sync.WaitGroup{} - wg.Add(len(peers)) - for _, id := range peers { - ticker := time.NewTicker(100 * time.Millisecond) - go func(p *adapters.NodeId) { - defer wg.Done() - for { - peer := self.GetPeer(p) - if peer != nil { - 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 <-peer.Flushc: - glog.V(6).Infof("peer %v is connected", p) - return - } - } - select { - case <-ticker.C: - glog.V(6).Infof("waiting for %v to connect", p) - case <-timeout.C: - self.t.Fatalf("exchange timed out waiting for peer %v to connect", p) - } - } - }(id) - } - wg.Wait() - glog.V(6).Infof("checking complete") +// func (self *ExchangeTestSession) TestConnected(peers ...*adapters.NodeId) { +// timeout := time.NewTimer(1000 * time.Millisecond) +// wg := &sync.WaitGroup{} +// wg.Add(len(peers)) +// for _, id := range peers { +// ticker := time.NewTicker(100 * time.Millisecond) +// go func(p *adapters.NodeId) { +// defer wg.Done() +// for { +// peer := self.GetPeer(p) +// if peer != nil { +// 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 <-peer.Flushc: +// glog.V(6).Infof("peer %v is connected", p) +// return +// } +// } +// select { +// case <-ticker.C: +// glog.V(6).Infof("waiting for %v to connect", p) +// case <-timeout.C: +// self.t.Fatalf("timed out waiting for peer %v to connect", p) +// } +// } +// }(id) +// } +// wg.Wait() +// glog.V(6).Infof("checking complete") -} +// } func (self *ExchangeTestSession) TestDisconnected(disconnects ...*Disconnect) { for _, disconnect := range disconnects { @@ -259,7 +249,7 @@ func (self *ExchangeTestSession) TestDisconnected(disconnects ...*Disconnect) { select { case derr := <-errc: if !((err == nil && derr == nil) || err != nil && derr != nil && err.Error() == derr.Error()) { - self.t.Fatalf("unexpected error on peer %v: '%v', wanted '%v'", id, derr, err) + self.t.Fatalf("unexpected error on peer %v. expected '%v', got '%v'", id, err, derr) } case <-alarm.C: self.t.Fatalf("exchange timed out waiting for peer %v to disconnect", id) diff --git a/swarm/network/discovery_test.go b/swarm/network/discovery_test.go index c57e5512ee..cd5f306b94 100644 --- a/swarm/network/discovery_test.go +++ b/swarm/network/discovery_test.go @@ -7,20 +7,17 @@ import ( ) /*** - * + * * - after connect, that outgoing subpeersmsg is sent - * + * */ func TestDiscovery(t *testing.T) { addr := RandomAddr() to := NewKademlia(addr.OAddr, NewKadParams()) - pp := p2ptest.NewTestPeerPool() - //pp := NewHive(NewHiveParams(), to) ct := BzzCodeMap(HiveMsgs...) - + services := func(p Peer) error { dp := NewDiscovery(p, to) - //pp.Add(p) to.On(dp) p.DisconnectHook(func(e interface{}) error { dp := e.(Peer) @@ -29,16 +26,8 @@ func TestDiscovery(t *testing.T) { }) 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 := newBzzBaseTester(t, 1, addr, ct, services) s.runHandshakes() s.TestExchanges(p2ptest.Exchange{ diff --git a/swarm/network/hive_test.go b/swarm/network/hive_test.go index 1d73ed190f..e62a6a4c31 100644 --- a/swarm/network/hive_test.go +++ b/swarm/network/hive_test.go @@ -5,16 +5,14 @@ import ( "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" + "github.com/ethereum/go-ethereum/p2p/adapters" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" ) func init() { - glog.SetV(logger.Detail) + glog.SetV(logger.Warn) glog.SetToStderr(true) } @@ -51,44 +49,63 @@ func TestOverlayRegistration(t *testing.T) { }) 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, - } + + // 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, + // } + + s := newBzzBaseTester(t, 1, addr, ct, services) 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() - to := NewTestOverlay(addr.OverlayAddr()) - pp := NewHive(NewHiveParams(), to) - ct := BzzCodeMap(HiveMsgs...) - s := newBzzTester(t, addr, pp, ct, nil) + // 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 + } - // register the node with the peerPool - id := p2ptest.RandomNodeId() - // pretend to start the node - s.Start(id) + s := newBzzBaseTester(t, 1, addr, ct, services) + // protocall := func(na adapters.NodeAdapter) adapters.ProtoCall { + // protocol := Bzz(addr.OverlayAddr(), na, ct, services, nil, nil) + // return protocol.Run + // } - // register another address + // es := p2ptest.NewProtocolTester(t, NodeId(addr), 1, protocall) + + // s := &bzzTester{ + // addr: addr, + // ExchangeSession: es, + // } + + id := s.Ids[0] raddr := NewPeerAddrFromNodeId(id) + pp.Register(raddr) glog.V(5).Infof("%v", pp) @@ -103,40 +120,8 @@ func TestRegisterAndConnect(t *testing.T) { pp.Start(tc.connect, tc.ping) 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") } @@ -149,15 +134,7 @@ func TestRegisterAndConnect(t *testing.T) { o = 1 } s.TestExchanges(p2ptest.Exchange{ - Expects: []p2ptest.Expect{ - p2ptest.Expect{ - Code: 3, - Msg: &SubPeersMsg{ProxLimit: 0, MinProxBinSize: 8}, - Peer: s.ExchangeSession.Ids[1], - }, - }, - }) - s.TestExchanges(p2ptest.Exchange{ + Label: "getPeers message", Expects: []p2ptest.Expect{ p2ptest.Expect{ Code: 1, @@ -166,4 +143,14 @@ func TestRegisterAndConnect(t *testing.T) { }, }, }) + s.TestExchanges(p2ptest.Exchange{ + Label: "subPeers message outgoing", + Expects: []p2ptest.Expect{ + p2ptest.Expect{ + Code: 3, + Msg: &SubPeersMsg{ProxLimit: 0, MinProxBinSize: 8}, + Peer: id, + }, + }, + }) } diff --git a/swarm/network/protocol.go b/swarm/network/protocol.go index d30daea541..6a484a6fb9 100644 --- a/swarm/network/protocol.go +++ b/swarm/network/protocol.go @@ -50,7 +50,6 @@ func (self *bzzPeer) LastActive() time.Time { return self.lastActive } - // implemented by peerAddr type PeerAddr interface { OverlayAddr() []byte @@ -60,8 +59,8 @@ type PeerAddr interface { // the Peer interface that peerPool needs type Peer interface { PeerAddr - String() string // pretty printable the Node - ID() discover.NodeID // the key that uniquely identifies the Node for the peerPool + String() string // pretty printable the Node + ID() discover.NodeID // the key that uniquely identifies the Node for the peerPool Send(interface{}) error // can send messages Drop(error) // disconnect this peer Register(interface{}, func(interface{}) error) uint64 // register message-handler callbacks @@ -77,7 +76,7 @@ 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, na adapters.NodeAdapter, ct *protocols.CodeMap, services func(Peer) error, peerInfo func(id discover.NodeID) interface{}, nodeInfo func() interface{}) *p2p.Protocol { +func Bzz(localAddr []byte, na adapters.NodeAdapter, ct *protocols.CodeMap, services func(Peer) error, peerInfo func(id discover.NodeID) interface{}, nodeInfo func() interface{}, connectHook func(*protocols.Peer)) *p2p.Protocol { run := func(p *protocols.Peer) error { addr := &peerAddr{localAddr, na.LocalAddr()} @@ -106,7 +105,7 @@ func Bzz(localAddr []byte, na adapters.NodeAdapter, ct *protocols.CodeMap, servi return bee.Run() } - return protocols.NewProtocol(ProtocolName, Version, run, na, ct, peerInfo, nodeInfo) + return protocols.NewProtocol(ProtocolName, Version, run, na, ct, peerInfo, nodeInfo, connectHook) } /* diff --git a/swarm/network/protocol_test.go b/swarm/network/protocol_test.go index 3f03f45d21..4d8ae502e5 100644 --- a/swarm/network/protocol_test.go +++ b/swarm/network/protocol_test.go @@ -34,18 +34,48 @@ func bzzHandshakeExchange(lhs, rhs *bzzHandshake, id *adapters.NodeId) []p2ptest } } - -func newBzzTester(t *testing.T, addr *peerAddr, pp *p2ptest.TestPeerPool, ct *protocols.CodeMap, services func(Peer) error) *bzzTester { +func newBzzBaseTester(t *testing.T, n int, addr *peerAddr, ct *protocols.CodeMap, services func(Peer) error) *bzzTester { if ct == nil { ct = BzzCodeMap() } + + flushc := make(map[string]chan bool) + + connectHook := func(p *protocols.Peer) { + close(flushc[p.ID().String()]) + } + protocall := func(na adapters.NodeAdapter) adapters.ProtoCall { + protocol := Bzz(addr.OverlayAddr(), na, ct, services, nil, nil, connectHook) + return protocol.Run + } + + s := p2ptest.NewProtocolTester(t, NodeId(addr), n, protocall) + + for _, id := range s.Ids { + flushc[id.String()] = make(chan bool) + } + return &bzzTester{ + addr: addr, + flushc: flushc, + ExchangeSession: s, + } +} + +type bzzTester struct { + *p2ptest.ExchangeSession + flushc map[string]chan bool + addr *peerAddr +} + +func newBzzTester(t *testing.T, n int, addr *peerAddr, pp *p2ptest.TestPeerPool, ct *protocols.CodeMap, services func(Peer) error) *bzzTester { + extraservices := func(p Peer) error { pp.Add(p) p.DisconnectHook(func(e interface{}) error { pp.Remove(p) return nil }) - + if services != nil { err := services(p) if err != nil { @@ -54,25 +84,7 @@ func newBzzTester(t *testing.T, addr *peerAddr, pp *p2ptest.TestPeerPool, ct *pr } return 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, - ExchangeSession: s, - } -} - -type bzzTester struct { - *p2ptest.ExchangeSession - // flushCode int - addr *peerAddr + return newBzzBaseTester(t, n, addr, ct, extraservices) } // should test handshakes in one exchange? parallelisation @@ -86,21 +98,19 @@ func (s *bzzTester) testHandshake(lhs, rhs *bzzHandshake, disconnects ...*p2ptes } else { peers = []*adapters.NodeId{id} } - 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) runHandshakes(ids ...*adapters.NodeId) { if len(ids) == 0 { ids = s.Ids } for _, id := range ids { - s.testHandshake(correctBzzHandshake(s.addr), correctBzzHandshake(NewPeerAddrFromNodeId(id))) + go func() { + <-s.flushc[id.String()] + s.testHandshake(correctBzzHandshake(s.addr), correctBzzHandshake(NewPeerAddrFromNodeId(id))) + }() } } @@ -112,7 +122,7 @@ func correctBzzHandshake(addr *peerAddr) *bzzHandshake { func TestBzzHandshakeNetworkIdMismatch(t *testing.T) { pp := p2ptest.NewTestPeerPool() addr := RandomAddr() - s := newBzzTester(t, addr, pp, nil, nil) + s := newBzzTester(t, 1, addr, pp, nil, nil) id := s.Ids[0] s.testHandshake( correctBzzHandshake(addr), @@ -124,7 +134,7 @@ func TestBzzHandshakeNetworkIdMismatch(t *testing.T) { func TestBzzHandshakeVersionMismatch(t *testing.T) { pp := p2ptest.NewTestPeerPool() addr := RandomAddr() - s := newBzzTester(t, addr, pp, nil, nil) + s := newBzzTester(t, 1, addr, pp, nil, nil) id := s.Ids[0] s.testHandshake( correctBzzHandshake(addr), @@ -136,7 +146,7 @@ func TestBzzHandshakeVersionMismatch(t *testing.T) { func TestBzzHandshakeSuccess(t *testing.T) { pp := p2ptest.NewTestPeerPool() addr := RandomAddr() - s := newBzzTester(t, addr, pp, nil, nil) + s := newBzzTester(t, 1, addr, pp, nil, nil) id := s.Ids[0] s.testHandshake( correctBzzHandshake(addr), @@ -147,12 +157,11 @@ func TestBzzHandshakeSuccess(t *testing.T) { func TestBzzPeerPoolAdd(t *testing.T) { pp := p2ptest.NewTestPeerPool() addr := RandomAddr() - s := newBzzTester(t, addr, pp, nil, nil) + s := newBzzTester(t, 1, 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) } @@ -161,7 +170,7 @@ func TestBzzPeerPoolAdd(t *testing.T) { func TestBzzPeerPoolRemove(t *testing.T) { addr := RandomAddr() pp := p2ptest.NewTestPeerPool() - s := newBzzTester(t, addr, pp, nil, nil) + s := newBzzTester(t, 1, addr, pp, nil, nil) s.runHandshakes() id := s.Ids[0] @@ -175,7 +184,7 @@ func TestBzzPeerPoolRemove(t *testing.T) { func TestBzzPeerPoolBothAddRemove(t *testing.T) { addr := RandomAddr() pp := p2ptest.NewTestPeerPool() - s := newBzzTester(t, addr, pp, nil, nil) + s := newBzzTester(t, 1, addr, pp, nil, nil) s.runHandshakes() id := s.Ids[0] @@ -193,7 +202,7 @@ func TestBzzPeerPoolBothAddRemove(t *testing.T) { func TestBzzPeerPoolNotAdd(t *testing.T) { addr := RandomAddr() pp := p2ptest.NewTestPeerPool() - s := newBzzTester(t, addr, pp, nil, nil) + s := newBzzTester(t, 1, addr, pp, nil, nil) id := s.Ids[0] s.testHandshake(correctBzzHandshake(addr), &bzzHandshake{0, 321, NewPeerAddrFromNodeId(id)}, &p2ptest.Disconnect{Peer: id, Error: fmt.Errorf("network id mismatch 321 (!= 322)")})