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

swarm/network, p2p: fixes

Browse source
This commit is contained in:
zelig 2017-03-13 10:33:41 +07:00 committed by Lewis Marshall
parent 8cd9adcf38
commit 7f22abaead
13 changed files with 231 additions and 398 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
p2p/adapters/inproc.go
View file

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

40
p2p/protocols/protocol.go
View file

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

82
p2p/protocols/protocol_test.go
View file

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

75
p2p/testing/exchange.go
View file

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

48
p2p/testing/peerpool.go
View file

@ -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]
}

12
p2p/testing/sessions.go
View file

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

97
swarm/network/discovery.go
View file

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

19
swarm/network/discovery_test.go
View file

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

41
swarm/network/hive.go
View file

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

54
swarm/network/hive_test.go
View file

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

23
swarm/network/kademlia.go
View file

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

34
swarm/network/protocol.go
View file

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

99
swarm/network/protocol_test.go
View file

@ -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]
}