swarm/network: cherry pick dynamic implementation

This commit is contained in:
Fabio Barone 2019-03-21 18:53:25 -05:00
parent 7b6422b4a7
commit f3d7d684eb
2 changed files with 62 additions and 41 deletions

View file

@ -19,7 +19,10 @@ package network
import ( import (
"bytes" "bytes"
"context" "context"
"math/rand" "crypto/ecdsa"
"crypto/rand"
"fmt"
"net"
"testing" "testing"
"time" "time"
@ -72,9 +75,6 @@ func TestDiscovery(t *testing.T) {
// to another peer with a given depth and kademlia // to another peer with a given depth and kademlia
func TestSubpeersMsg(t *testing.T) { func TestSubpeersMsg(t *testing.T) {
// This is the defined depth
testDepth := rand.Intn(4) + 1
// construct ProtocolTester and hive // construct ProtocolTester and hive
params := NewHiveParams() params := NewHiveParams()
// setup // setup
@ -86,73 +86,78 @@ func TestSubpeersMsg(t *testing.T) {
to := NewKademlia(addr, NewKadParams()) to := NewKademlia(addr, NewKadParams())
hive := NewHive(params, to, nil) // hive hive := NewHive(params, to, nil) // hive
s, err := newBzzBaseTester(t, 1, prvkey, DiscoverySpec, hive.Run) numOfTestNodes := 12
s, err := newBzzBaseTester(t, numOfTestNodes, prvkey, DiscoverySpec, hive.Run)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
// register some addresses in specific bins (must coincide with testDepth) // the control node is the only one from the ProtocolTester
registerBzzAddr(0, hive, true) // bin 0 control := s.Nodes[numOfTestNodes-1]
registerBzzAddr(0, hive, true) // bin 0
registerBzzAddr(1, hive, true) // bin 1 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
registerBzzAddr(1, hive, true) // bin 1 defer cancel()
registerBzzAddr(3, hive, true) // bin 3
registerBzzAddr(4, hive, true) // bin 4
registerBzzAddr(3, hive, false) // add a known but not connected peer
registerBzzAddr(1, hive, false) // add a known but not connected peer
// start the hive // start the hive
hive.Start(s.Server) hive.Start(s.Server)
defer hive.Stop() defer hive.Stop()
// the remote node is the only one from the ProtocolTester // we need to wait until the control node is actually connected to our hive
remote := s.Nodes[0]
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
// we need to wait until the remote node is actually connected to our hive
WAIT_PIVOT: WAIT_PIVOT:
for { for {
select { select {
case <-ctx.Done(): case <-ctx.Done():
t.Fatal("Timed out waiting for the remote node to connect") t.Fatal("Timed out waiting for the control node to connect")
case <-time.After(100 * time.Millisecond): case <-time.After(100 * time.Millisecond):
if _, ok := hive.peers[remote.ID()]; ok { if len(hive.peers) == len(s.BzzAddrs) {
break WAIT_PIVOT break WAIT_PIVOT
} }
} }
} }
// get BzzAddr of the remote // get BzzAddr of the control
remoteAddress := hive.peers[remote.ID()] controlBzz := s.BzzAddrs[control.ID()].Over()
remoteBzz := remoteAddress.BzzAddr.Over()
controlKad := NewKademlia(controlBzz, NewKadParams())
for _, p := range s.BzzAddrs {
if !bytes.Equal(p.Over(), controlBzz) {
controlKad.On(NewPeer(&BzzPeer{nil, p, time.Now(), false}, controlKad))
}
}
controlDepth := controlKad.NeighbourhoodDepth()
// now we need to identify which peers are expected // now we need to identify which peers are expected
// iterate the hive's connection and only add peers below testDepth // iterate the hive's connection and only add peers below testDepth
var expectedPeers []*BzzAddr var expectedPeers []*BzzAddr
hive.EachConn(remoteBzz, 255, func(p *Peer, po int) bool { hive.EachConn(controlBzz, 255, func(p *Peer, po int) bool {
if po < testDepth { if po < controlDepth {
return false return false
} }
// don't add the remote node itself to expectedPeers; // don't add the control node itself to expectedPeers;
// the remote node was not added as // the control node was not added as
if !bytes.Equal(p.BzzAddr.Over(), remoteBzz) { if !bytes.Equal(p.BzzAddr.Over(), controlBzz) {
expectedPeers = append(expectedPeers, p.BzzAddr) expectedPeers = append(expectedPeers, p.BzzAddr)
} }
return true return true
}) })
hiveDepth := hive.NeighbourhoodDepth()
// the test exchange is as follows: // the test exchange is as follows:
// 1. Trigger a subPeersMsg from remote to our hive // 1. Trigger a subPeersMsg from control to our hive
// 2. Hive will respond with peersMsg with the set of expected peers // 2. Hive will respond with peersMsg with the set of expected peers
if controlDepth == 0 {
controlDepth = 1
}
err = s.TestExchanges(p2ptest.Exchange{ err = s.TestExchanges(p2ptest.Exchange{
Label: "incoming subPeersMsg", Label: "incoming subPeersMsg",
Expects: []p2ptest.Expect{ Expects: []p2ptest.Expect{
{ {
Code: 1, Code: 1,
Msg: &subPeersMsg{Depth: uint8(2)}, Msg: &subPeersMsg{Depth: uint8(hiveDepth)},
Peer: remote.ID(), Peer: control.ID(),
}, },
}, },
}, },
@ -161,8 +166,8 @@ WAIT_PIVOT:
Triggers: []p2ptest.Trigger{ Triggers: []p2ptest.Trigger{
{ {
Code: 1, Code: 1,
Msg: &subPeersMsg{Depth: uint8(testDepth)}, Msg: &subPeersMsg{Depth: uint8(controlDepth)},
Peer: remote.ID(), Peer: control.ID(),
Timeout: 3 * time.Second, Timeout: 3 * time.Second,
}, },
}, },
@ -170,12 +175,23 @@ WAIT_PIVOT:
{ {
Code: 0, Code: 0,
Msg: &peersMsg{Peers: expectedPeers}, Msg: &peersMsg{Peers: expectedPeers},
Peer: remote.ID(), Peer: control.ID(),
Timeout: 3 * time.Second, Timeout: 3 * time.Second,
}, },
}, },
}) })
// for some configurations, there will be no advertised peers due to the
// distance of the control peer to the hive and the set of connected peers
// in this case, no peersMsg will be sent out, and we would run into a time out
// catch this edge case
if len(expectedPeers) == 0 {
if err == nil || !strings.Contains(err.Error(), "timed out") {
t.Fatal("expected timeout but didn't")
}
return
}
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }

View file

@ -72,6 +72,7 @@ func HandshakeMsgExchange(lhs, rhs *HandshakeMsg, id enode.ID) []p2ptest.Exchang
func newBzzBaseTester(t *testing.T, n int, prvkey *ecdsa.PrivateKey, spec *protocols.Spec, run func(*BzzPeer) error) (*bzzTester, error) { func newBzzBaseTester(t *testing.T, n int, prvkey *ecdsa.PrivateKey, spec *protocols.Spec, run func(*BzzPeer) error) (*bzzTester, error) {
cs := make(map[string]chan bool) cs := make(map[string]chan bool)
bzzAddrs := make(map[enode.ID]*BzzAddr)
srv := func(p *BzzPeer) error { srv := func(p *BzzPeer) error {
defer func() { defer func() {
@ -83,7 +84,9 @@ func newBzzBaseTester(t *testing.T, n int, prvkey *ecdsa.PrivateKey, spec *proto
} }
protocol := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { protocol := func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
return srv(&BzzPeer{Peer: protocols.NewPeer(p, rw, spec), BzzAddr: NewAddr(p.Node())}) bzzAddr := NewAddr(p.Node())
bzzAddrs[p.Node().ID()] = bzzAddr
return srv(&BzzPeer{Peer: protocols.NewPeer(p, rw, spec), BzzAddr: bzzAddr})
} }
s := p2ptest.NewProtocolTester(prvkey, n, protocol) s := p2ptest.NewProtocolTester(prvkey, n, protocol)
@ -109,11 +112,13 @@ func newBzzBaseTester(t *testing.T, n int, prvkey *ecdsa.PrivateKey, spec *proto
addr: addr, addr: addr,
ProtocolTester: s, ProtocolTester: s,
cs: cs, cs: cs,
BzzAddrs: bzzAddrs,
}, nil }, nil
} }
type bzzTester struct { type bzzTester struct {
*p2ptest.ProtocolTester *p2ptest.ProtocolTester
BzzAddrs map[enode.ID]*BzzAddr
addr *BzzAddr addr *BzzAddr
cs map[string]chan bool cs map[string]chan bool
bzz *Bzz bzz *Bzz