forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-27 07:06:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

swarm/network: WIP testing for protocol

Browse source
This commit is contained in:
zelig 2017-03-14 14:19:26 +07:00 committed by Lewis Marshall
parent 140c7f7f23
commit 1eedb4a2c8
6 changed files with 159 additions and 184 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
p2p/protocols/protocol.go
View file

@ -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),
}
}
@ -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

78
p2p/testing/exchange.go
View file

@ -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)

13
swarm/network/discovery_test.go
View file

@ -14,13 +14,10 @@ import (
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{

111
swarm/network/hive_test.go
View file

@ -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)
}
@ -52,17 +50,19 @@ 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
}
// 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)
// es := p2ptest.NewProtocolTester(t, NodeId(addr), 1, protocall)
s := &bzzTester{
addr: addr,
ExchangeSession: es,
}
// s := &bzzTester{
// addr: addr,
// ExchangeSession: es,
// }
s := newBzzBaseTester(t, 1, addr, ct, services)
id := s.Ids[0]
raddr := NewPeerAddrFromNodeId(id)
@ -76,36 +76,6 @@ func TestOverlayRegistration(t *testing.T) {
}
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)
// 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 {
s.Connect(adapters.NewNodeIdFromHex(c))
return nil
},
ticker: make(chan time.Time),
}
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
@ -120,22 +90,37 @@ func TestRegisterAndConnect(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 := 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
// }
es := p2ptest.NewProtocolTester(t, NodeId(addr), 1, protocall)
// es := p2ptest.NewProtocolTester(t, NodeId(addr), 1, protocall)
// s := &bzzTester{
// addr: addr,
// ExchangeSession: es,
// }
s := &bzzTester{
addr: addr,
ExchangeSession: es,
}
id := s.Ids[0]
raddr := NewPeerAddrFromNodeId(id)
s.runHandshakes()
pp.Register(raddr)
glog.V(5).Infof("%v", pp)
// start the hive and wait for the connection
tc := &testConnect{
connectf: func(c string) error {
s.Connect(adapters.NewNodeIdFromHex(c))
return nil
},
ticker: make(chan time.Time),
}
pp.Start(tc.connect, tc.ping)
tc.ticker <- time.Now()
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,
},
},
})
}

9
swarm/network/protocol.go
View file

@ -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)
}
/*

79
swarm/network/protocol_test.go
View file

@ -34,11 +34,41 @@ 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 {
@ -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)")})