mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-20 10:52:25 +00:00
p2p: dial using iterator
This commit is contained in:
Felix Lange
parent
52a1841ee1
commit
83e2b8547b
5 changed files with 150 additions and 247 deletions
145
p2p/dial.go
145
p2p/dial.go
|
|
@ -33,12 +33,7 @@ const (
|
|||
// private networks.
|
||||
dialHistoryExpiration = inboundThrottleTime + 5*time.Second
|
||||
|
||||
// Discovery lookups are throttled and can only run
|
||||
// once every few seconds.
|
||||
lookupInterval = 4 * time.Second
|
||||
|
||||
// If no peers are found for this amount of time, the initial bootnodes are
|
||||
// attempted to be connected.
|
||||
// If no peers are found for this amount of time, the initial bootnodes are dialed.
|
||||
fallbackInterval = 20 * time.Second
|
||||
|
||||
// Endpoint resolution is throttled with bounded backoff.
|
||||
|
|
@ -52,6 +47,10 @@ type NodeDialer interface {
|
|||
Dial(*enode.Node) (net.Conn, error)
|
||||
}
|
||||
|
||||
type nodeResolver interface {
|
||||
Resolve(*enode.Node) *enode.Node
|
||||
}
|
||||
|
||||
// TCPDialer implements the NodeDialer interface by using a net.Dialer to
|
||||
// create TCP connections to nodes in the network
|
||||
type TCPDialer struct {
|
||||
|
|
@ -69,7 +68,6 @@ func (t TCPDialer) Dial(dest *enode.Node) (net.Conn, error) {
|
|||
// of the main loop in Server.run.
|
||||
type dialstate struct {
|
||||
maxDynDials int
|
||||
ntab discoverTable
|
||||
netrestrict *netutil.Netlist
|
||||
self enode.ID
|
||||
bootnodes []*enode.Node // default dials when there are no peers
|
||||
|
|
@ -79,55 +77,23 @@ type dialstate struct {
|
|||
lookupRunning bool
|
||||
dialing map[enode.ID]connFlag
|
||||
lookupBuf []*enode.Node // current discovery lookup results
|
||||
randomNodes []*enode.Node // filled from Table
|
||||
static map[enode.ID]*dialTask
|
||||
hist expHeap
|
||||
}
|
||||
|
||||
type discoverTable interface {
|
||||
Close()
|
||||
Resolve(*enode.Node) *enode.Node
|
||||
LookupRandom() []*enode.Node
|
||||
ReadRandomNodes([]*enode.Node) int
|
||||
}
|
||||
|
||||
type task interface {
|
||||
Do(*Server)
|
||||
}
|
||||
|
||||
// A dialTask is generated for each node that is dialed. Its
|
||||
// fields cannot be accessed while the task is running.
|
||||
type dialTask struct {
|
||||
flags connFlag
|
||||
dest *enode.Node
|
||||
lastResolved time.Time
|
||||
resolveDelay time.Duration
|
||||
}
|
||||
|
||||
// discoverTask runs discovery table operations.
|
||||
// Only one discoverTask is active at any time.
|
||||
// discoverTask.Do performs a random lookup.
|
||||
type discoverTask struct {
|
||||
results []*enode.Node
|
||||
}
|
||||
|
||||
// A waitExpireTask is generated if there are no other tasks
|
||||
// to keep the loop in Server.run ticking.
|
||||
type waitExpireTask struct {
|
||||
time.Duration
|
||||
}
|
||||
|
||||
func newDialState(self enode.ID, ntab discoverTable, maxdyn int, cfg *Config) *dialstate {
|
||||
func newDialState(self enode.ID, maxdyn int, cfg *Config) *dialstate {
|
||||
s := &dialstate{
|
||||
maxDynDials: maxdyn,
|
||||
ntab: ntab,
|
||||
self: self,
|
||||
netrestrict: cfg.NetRestrict,
|
||||
log: cfg.Logger,
|
||||
static: make(map[enode.ID]*dialTask),
|
||||
dialing: make(map[enode.ID]connFlag),
|
||||
bootnodes: make([]*enode.Node, len(cfg.BootstrapNodes)),
|
||||
randomNodes: make([]*enode.Node, maxdyn/2),
|
||||
}
|
||||
copy(s.bootnodes, cfg.BootstrapNodes)
|
||||
if s.log == nil {
|
||||
|
|
@ -151,10 +117,6 @@ func (s *dialstate) removeStatic(n *enode.Node) {
|
|||
}
|
||||
|
||||
func (s *dialstate) newTasks(nRunning int, peers map[enode.ID]*Peer, now time.Time) []task {
|
||||
if s.start.IsZero() {
|
||||
s.start = now
|
||||
}
|
||||
|
||||
var newtasks []task
|
||||
addDial := func(flag connFlag, n *enode.Node) bool {
|
||||
if err := s.checkDial(n, peers); err != nil {
|
||||
|
|
@ -166,20 +128,9 @@ func (s *dialstate) newTasks(nRunning int, peers map[enode.ID]*Peer, now time.Ti
|
|||
return true
|
||||
}
|
||||
|
||||
// Compute number of dynamic dials necessary at this point.
|
||||
needDynDials := s.maxDynDials
|
||||
for _, p := range peers {
|
||||
if p.rw.is(dynDialedConn) {
|
||||
needDynDials--
|
||||
}
|
||||
if s.start.IsZero() {
|
||||
s.start = now
|
||||
}
|
||||
for _, flag := range s.dialing {
|
||||
if flag&dynDialedConn != 0 {
|
||||
needDynDials--
|
||||
}
|
||||
}
|
||||
|
||||
// Expire the dial history on every invocation.
|
||||
s.hist.expire(now)
|
||||
|
||||
// Create dials for static nodes if they are not connected.
|
||||
|
|
@ -194,6 +145,20 @@ func (s *dialstate) newTasks(nRunning int, peers map[enode.ID]*Peer, now time.Ti
|
|||
newtasks = append(newtasks, t)
|
||||
}
|
||||
}
|
||||
|
||||
// Compute number of dynamic dials needed.
|
||||
needDynDials := s.maxDynDials
|
||||
for _, p := range peers {
|
||||
if p.rw.is(dynDialedConn) {
|
||||
needDynDials--
|
||||
}
|
||||
}
|
||||
for _, flag := range s.dialing {
|
||||
if flag&dynDialedConn != 0 {
|
||||
needDynDials--
|
||||
}
|
||||
}
|
||||
|
||||
// If we don't have any peers whatsoever, try to dial a random bootnode. This
|
||||
// scenario is useful for the testnet (and private networks) where the discovery
|
||||
// table might be full of mostly bad peers, making it hard to find good ones.
|
||||
|
|
@ -201,24 +166,12 @@ func (s *dialstate) newTasks(nRunning int, peers map[enode.ID]*Peer, now time.Ti
|
|||
bootnode := s.bootnodes[0]
|
||||
s.bootnodes = append(s.bootnodes[:0], s.bootnodes[1:]...)
|
||||
s.bootnodes = append(s.bootnodes, bootnode)
|
||||
|
||||
if addDial(dynDialedConn, bootnode) {
|
||||
needDynDials--
|
||||
}
|
||||
}
|
||||
// Use random nodes from the table for half of the necessary
|
||||
// dynamic dials.
|
||||
randomCandidates := needDynDials / 2
|
||||
if randomCandidates > 0 {
|
||||
n := s.ntab.ReadRandomNodes(s.randomNodes)
|
||||
for i := 0; i < randomCandidates && i < n; i++ {
|
||||
if addDial(dynDialedConn, s.randomNodes[i]) {
|
||||
needDynDials--
|
||||
}
|
||||
}
|
||||
}
|
||||
// Create dynamic dials from random lookup results, removing tried
|
||||
// items from the result buffer.
|
||||
|
||||
// Create dynamic dials from discovery results.
|
||||
i := 0
|
||||
for ; i < len(s.lookupBuf) && needDynDials > 0; i++ {
|
||||
if addDial(dynDialedConn, s.lookupBuf[i]) {
|
||||
|
|
@ -226,10 +179,11 @@ func (s *dialstate) newTasks(nRunning int, peers map[enode.ID]*Peer, now time.Ti
|
|||
}
|
||||
}
|
||||
s.lookupBuf = s.lookupBuf[:copy(s.lookupBuf, s.lookupBuf[i:])]
|
||||
|
||||
// Launch a discovery lookup if more candidates are needed.
|
||||
if len(s.lookupBuf) < needDynDials && !s.lookupRunning {
|
||||
s.lookupRunning = true
|
||||
newtasks = append(newtasks, &discoverTask{})
|
||||
newtasks = append(newtasks, &discoverTask{want: needDynDials - len(s.lookupBuf)})
|
||||
}
|
||||
|
||||
// Launch a timer to wait for the next node to expire if all
|
||||
|
|
@ -279,6 +233,15 @@ func (s *dialstate) taskDone(t task, now time.Time) {
|
|||
}
|
||||
}
|
||||
|
||||
// A dialTask is generated for each node that is dialed. Its
|
||||
// fields cannot be accessed while the task is running.
|
||||
type dialTask struct {
|
||||
flags connFlag
|
||||
dest *enode.Node
|
||||
lastResolved time.Time
|
||||
resolveDelay time.Duration
|
||||
}
|
||||
|
||||
func (t *dialTask) Do(srv *Server) {
|
||||
if t.dest.Incomplete() {
|
||||
if !t.resolve(srv) {
|
||||
|
|
@ -304,8 +267,8 @@ func (t *dialTask) Do(srv *Server) {
|
|||
// discovery network with useless queries for nodes that don't exist.
|
||||
// The backoff delay resets when the node is found.
|
||||
func (t *dialTask) resolve(srv *Server) bool {
|
||||
if srv.ntab == nil {
|
||||
srv.log.Debug("Can't resolve node", "id", t.dest.ID, "err", "discovery is disabled")
|
||||
if srv.staticNodeResolver == nil {
|
||||
srv.log.Debug("Can't resolve node", "id", t.dest.ID(), "err", "discovery is disabled")
|
||||
return false
|
||||
}
|
||||
if t.resolveDelay == 0 {
|
||||
|
|
@ -314,20 +277,20 @@ func (t *dialTask) resolve(srv *Server) bool {
|
|||
if time.Since(t.lastResolved) < t.resolveDelay {
|
||||
return false
|
||||
}
|
||||
resolved := srv.ntab.Resolve(t.dest)
|
||||
resolved := srv.staticNodeResolver.Resolve(t.dest)
|
||||
t.lastResolved = time.Now()
|
||||
if resolved == nil {
|
||||
t.resolveDelay *= 2
|
||||
if t.resolveDelay > maxResolveDelay {
|
||||
t.resolveDelay = maxResolveDelay
|
||||
}
|
||||
srv.log.Debug("Resolving node failed", "id", t.dest.ID, "newdelay", t.resolveDelay)
|
||||
srv.log.Debug("Resolving node failed", "id", t.dest.ID(), "newdelay", t.resolveDelay)
|
||||
return false
|
||||
}
|
||||
// The node was found.
|
||||
t.resolveDelay = initialResolveDelay
|
||||
t.dest = resolved
|
||||
srv.log.Debug("Resolved node", "id", t.dest.ID, "addr", &net.TCPAddr{IP: t.dest.IP(), Port: t.dest.TCP()})
|
||||
srv.log.Debug("Resolved node", "id", t.dest.ID(), "addr", &net.TCPAddr{IP: t.dest.IP(), Port: t.dest.TCP()})
|
||||
return true
|
||||
}
|
||||
|
||||
|
|
@ -350,26 +313,34 @@ func (t *dialTask) String() string {
|
|||
return fmt.Sprintf("%v %x %v:%d", t.flags, id[:8], t.dest.IP(), t.dest.TCP())
|
||||
}
|
||||
|
||||
// discoverTask runs discovery table operations.
|
||||
// Only one discoverTask is active at any time.
|
||||
// discoverTask.Do performs a random lookup.
|
||||
type discoverTask struct {
|
||||
want int
|
||||
results []*enode.Node
|
||||
}
|
||||
|
||||
func (t *discoverTask) Do(srv *Server) {
|
||||
// newTasks generates a lookup task whenever dynamic dials are
|
||||
// necessary. Lookups need to take some time, otherwise the
|
||||
// event loop spins too fast.
|
||||
next := srv.lastLookup.Add(lookupInterval)
|
||||
if now := time.Now(); now.Before(next) {
|
||||
time.Sleep(next.Sub(now))
|
||||
}
|
||||
srv.lastLookup = time.Now()
|
||||
t.results = srv.ntab.LookupRandom()
|
||||
t.results = enode.ReadNodes(srv.discmix, t.want)
|
||||
}
|
||||
|
||||
func (t *discoverTask) String() string {
|
||||
s := "discovery lookup"
|
||||
s := "discovery query"
|
||||
if len(t.results) > 0 {
|
||||
s += fmt.Sprintf(" (%d results)", len(t.results))
|
||||
} else {
|
||||
s += fmt.Sprintf(" (want %d)", t.want)
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// A waitExpireTask is generated if there are no other tasks
|
||||
// to keep the loop in Server.run ticking.
|
||||
type waitExpireTask struct {
|
||||
time.Duration
|
||||
}
|
||||
|
||||
func (t waitExpireTask) Do(*Server) {
|
||||
time.Sleep(t.Duration)
|
||||
}
|
||||
|
|
|
|||
213
p2p/dial_test.go
213
p2p/dial_test.go
|
|
@ -73,7 +73,7 @@ func runDialTest(t *testing.T, test dialtest) {
|
|||
t.Errorf("ERROR round %d: got %v\nwant %v\nstate: %v\nrunning: %v",
|
||||
i, spew.Sdump(new), spew.Sdump(round.new), spew.Sdump(test.init), spew.Sdump(running))
|
||||
}
|
||||
t.Logf("round %d new tasks: %s", i, strings.TrimSpace(spew.Sdump(new)))
|
||||
t.Logf("round %d (running %d) new tasks: %s", i, running, strings.TrimSpace(spew.Sdump(new)))
|
||||
|
||||
// Time advances by 16 seconds on every round.
|
||||
vtime = vtime.Add(16 * time.Second)
|
||||
|
|
@ -81,19 +81,11 @@ func runDialTest(t *testing.T, test dialtest) {
|
|||
}
|
||||
}
|
||||
|
||||
type fakeTable []*enode.Node
|
||||
|
||||
func (t fakeTable) Self() *enode.Node { return new(enode.Node) }
|
||||
func (t fakeTable) Close() {}
|
||||
func (t fakeTable) LookupRandom() []*enode.Node { return nil }
|
||||
func (t fakeTable) Resolve(*enode.Node) *enode.Node { return nil }
|
||||
func (t fakeTable) ReadRandomNodes(buf []*enode.Node) int { return copy(buf, t) }
|
||||
|
||||
// This test checks that dynamic dials are launched from discovery results.
|
||||
func TestDialStateDynDial(t *testing.T) {
|
||||
config := &Config{Logger: testlog.Logger(t, log.LvlTrace)}
|
||||
runDialTest(t, dialtest{
|
||||
init: newDialState(enode.ID{}, fakeTable{}, 5, config),
|
||||
init: newDialState(enode.ID{}, 5, config),
|
||||
rounds: []round{
|
||||
// A discovery query is launched.
|
||||
{
|
||||
|
|
@ -102,7 +94,9 @@ func TestDialStateDynDial(t *testing.T) {
|
|||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(1), nil)}},
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(2), nil)}},
|
||||
},
|
||||
new: []task{&discoverTask{}},
|
||||
new: []task{
|
||||
&discoverTask{want: 3},
|
||||
},
|
||||
},
|
||||
// Dynamic dials are launched when it completes.
|
||||
{
|
||||
|
|
@ -188,7 +182,7 @@ func TestDialStateDynDial(t *testing.T) {
|
|||
},
|
||||
new: []task{
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(7), nil)},
|
||||
&discoverTask{},
|
||||
&discoverTask{want: 2},
|
||||
},
|
||||
},
|
||||
// Peer 7 is connected, but there still aren't enough dynamic peers
|
||||
|
|
@ -218,7 +212,7 @@ func TestDialStateDynDial(t *testing.T) {
|
|||
&discoverTask{},
|
||||
},
|
||||
new: []task{
|
||||
&discoverTask{},
|
||||
&discoverTask{want: 2},
|
||||
},
|
||||
},
|
||||
},
|
||||
|
|
@ -235,35 +229,37 @@ func TestDialStateDynDialBootnode(t *testing.T) {
|
|||
},
|
||||
Logger: testlog.Logger(t, log.LvlTrace),
|
||||
}
|
||||
table := fakeTable{
|
||||
newNode(uintID(4), nil),
|
||||
newNode(uintID(5), nil),
|
||||
newNode(uintID(6), nil),
|
||||
newNode(uintID(7), nil),
|
||||
newNode(uintID(8), nil),
|
||||
}
|
||||
runDialTest(t, dialtest{
|
||||
init: newDialState(enode.ID{}, table, 5, config),
|
||||
init: newDialState(enode.ID{}, 5, config),
|
||||
rounds: []round{
|
||||
// 2 dynamic dials attempted, bootnodes pending fallback interval
|
||||
{
|
||||
new: []task{
|
||||
&discoverTask{want: 5},
|
||||
},
|
||||
},
|
||||
{
|
||||
done: []task{
|
||||
&discoverTask{
|
||||
results: []*enode.Node{
|
||||
newNode(uintID(4), nil),
|
||||
newNode(uintID(5), nil),
|
||||
},
|
||||
},
|
||||
},
|
||||
new: []task{
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(4), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(5), nil)},
|
||||
&discoverTask{},
|
||||
&discoverTask{want: 3},
|
||||
},
|
||||
},
|
||||
// No dials succeed, bootnodes still pending fallback interval
|
||||
{},
|
||||
// 1 bootnode attempted as fallback interval was reached
|
||||
{
|
||||
done: []task{
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(4), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(5), nil)},
|
||||
},
|
||||
},
|
||||
// No dials succeed, bootnodes still pending fallback interval
|
||||
{},
|
||||
// No dials succeed, 2 dynamic dials attempted and 1 bootnode too as fallback interval was reached
|
||||
{
|
||||
new: []task{
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(1), nil)},
|
||||
},
|
||||
|
|
@ -275,15 +271,12 @@ func TestDialStateDynDialBootnode(t *testing.T) {
|
|||
},
|
||||
new: []task{
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(2), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(4), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(5), nil)},
|
||||
},
|
||||
},
|
||||
// No dials succeed, 3rd bootnode is attempted
|
||||
{
|
||||
done: []task{
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(2), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(5), nil)},
|
||||
},
|
||||
new: []task{
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(3), nil)},
|
||||
|
|
@ -293,115 +286,19 @@ func TestDialStateDynDialBootnode(t *testing.T) {
|
|||
{
|
||||
done: []task{
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(3), nil)},
|
||||
},
|
||||
new: []task{},
|
||||
},
|
||||
// Random dial succeeds, no more bootnodes are attempted
|
||||
{
|
||||
new: []task{
|
||||
&waitExpireTask{3 * time.Second},
|
||||
},
|
||||
peers: []*Peer{
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(4), nil)}},
|
||||
},
|
||||
done: []task{
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(1), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(4), nil)},
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
func TestDialStateDynDialFromTable(t *testing.T) {
|
||||
// This table always returns the same random nodes
|
||||
// in the order given below.
|
||||
table := fakeTable{
|
||||
newNode(uintID(1), nil),
|
||||
newNode(uintID(2), nil),
|
||||
newNode(uintID(3), nil),
|
||||
newNode(uintID(4), nil),
|
||||
newNode(uintID(5), nil),
|
||||
newNode(uintID(6), nil),
|
||||
newNode(uintID(7), nil),
|
||||
newNode(uintID(8), nil),
|
||||
}
|
||||
|
||||
runDialTest(t, dialtest{
|
||||
init: newDialState(enode.ID{}, table, 10, &Config{Logger: testlog.Logger(t, log.LvlTrace)}),
|
||||
rounds: []round{
|
||||
// 5 out of 8 of the nodes returned by ReadRandomNodes are dialed.
|
||||
{
|
||||
new: []task{
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(1), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(2), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(3), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(4), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(5), nil)},
|
||||
&discoverTask{},
|
||||
},
|
||||
},
|
||||
// Dialing nodes 1,2 succeeds. Dials from the lookup are launched.
|
||||
{
|
||||
peers: []*Peer{
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(1), nil)}},
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(2), nil)}},
|
||||
},
|
||||
done: []task{
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(1), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(2), nil)},
|
||||
&discoverTask{results: []*enode.Node{
|
||||
newNode(uintID(10), nil),
|
||||
newNode(uintID(11), nil),
|
||||
newNode(uintID(12), nil),
|
||||
newNode(uintID(6), nil),
|
||||
}},
|
||||
},
|
||||
new: []task{
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(10), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(11), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(12), nil)},
|
||||
&discoverTask{},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(6), nil)},
|
||||
&discoverTask{want: 4},
|
||||
},
|
||||
},
|
||||
// Dialing nodes 3,4,5 fails. The dials from the lookup succeed.
|
||||
// Random dial succeeds, no more bootnodes are attempted
|
||||
{
|
||||
peers: []*Peer{
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(1), nil)}},
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(2), nil)}},
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(10), nil)}},
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(11), nil)}},
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(12), nil)}},
|
||||
},
|
||||
done: []task{
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(3), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(4), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(5), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(10), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(11), nil)},
|
||||
&dialTask{flags: dynDialedConn, dest: newNode(uintID(12), nil)},
|
||||
},
|
||||
},
|
||||
// Waiting for expiry. No waitExpireTask is launched because the
|
||||
// discovery query is still running.
|
||||
{
|
||||
peers: []*Peer{
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(1), nil)}},
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(2), nil)}},
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(10), nil)}},
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(11), nil)}},
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(12), nil)}},
|
||||
},
|
||||
},
|
||||
// Nodes 3,4 are not tried again because only the first two
|
||||
// returned random nodes (nodes 1,2) are tried and they're
|
||||
// already connected.
|
||||
{
|
||||
peers: []*Peer{
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(1), nil)}},
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(2), nil)}},
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(10), nil)}},
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(11), nil)}},
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(12), nil)}},
|
||||
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(6), nil)}},
|
||||
},
|
||||
},
|
||||
},
|
||||
|
|
@ -416,11 +313,11 @@ func newNode(id enode.ID, ip net.IP) *enode.Node {
|
|||
return enode.SignNull(&r, id)
|
||||
}
|
||||
|
||||
// This test checks that candidates that do not match the netrestrict list are not dialed.
|
||||
// // This test checks that candidates that do not match the netrestrict list are not dialed.
|
||||
func TestDialStateNetRestrict(t *testing.T) {
|
||||
// This table always returns the same random nodes
|
||||
// in the order given below.
|
||||
table := fakeTable{
|
||||
nodes := []*enode.Node{
|
||||
newNode(uintID(1), net.ParseIP("127.0.0.1")),
|
||||
newNode(uintID(2), net.ParseIP("127.0.0.2")),
|
||||
newNode(uintID(3), net.ParseIP("127.0.0.3")),
|
||||
|
|
@ -434,12 +331,23 @@ func TestDialStateNetRestrict(t *testing.T) {
|
|||
restrict.Add("127.0.2.0/24")
|
||||
|
||||
runDialTest(t, dialtest{
|
||||
init: newDialState(enode.ID{}, table, 10, &Config{NetRestrict: restrict}),
|
||||
init: newDialState(enode.ID{}, 10, &Config{NetRestrict: restrict}),
|
||||
rounds: []round{
|
||||
{
|
||||
new: []task{
|
||||
&dialTask{flags: dynDialedConn, dest: table[4]},
|
||||
&discoverTask{},
|
||||
&discoverTask{want: 10},
|
||||
},
|
||||
},
|
||||
{
|
||||
done: []task{
|
||||
&discoverTask{results: nodes},
|
||||
},
|
||||
new: []task{
|
||||
&dialTask{flags: dynDialedConn, dest: nodes[4]},
|
||||
&dialTask{flags: dynDialedConn, dest: nodes[5]},
|
||||
&dialTask{flags: dynDialedConn, dest: nodes[6]},
|
||||
&dialTask{flags: dynDialedConn, dest: nodes[7]},
|
||||
&discoverTask{want: 6},
|
||||
},
|
||||
},
|
||||
},
|
||||
|
|
@ -459,7 +367,7 @@ func TestDialStateStaticDial(t *testing.T) {
|
|||
Logger: testlog.Logger(t, log.LvlTrace),
|
||||
}
|
||||
runDialTest(t, dialtest{
|
||||
init: newDialState(enode.ID{}, fakeTable{}, 0, config),
|
||||
init: newDialState(enode.ID{}, 0, config),
|
||||
rounds: []round{
|
||||
// Static dials are launched for the nodes that
|
||||
// aren't yet connected.
|
||||
|
|
@ -544,7 +452,7 @@ func TestDialStateCache(t *testing.T) {
|
|||
Logger: testlog.Logger(t, log.LvlTrace),
|
||||
}
|
||||
runDialTest(t, dialtest{
|
||||
init: newDialState(enode.ID{}, fakeTable{}, 0, config),
|
||||
init: newDialState(enode.ID{}, 0, config),
|
||||
rounds: []round{
|
||||
// Static dials are launched for the nodes that
|
||||
// aren't yet connected.
|
||||
|
|
@ -618,8 +526,8 @@ func TestDialResolve(t *testing.T) {
|
|||
Dialer: TCPDialer{&net.Dialer{Deadline: time.Now().Add(-5 * time.Minute)}},
|
||||
}
|
||||
resolved := newNode(uintID(1), net.IP{127, 0, 55, 234})
|
||||
table := &resolveMock{answer: resolved}
|
||||
state := newDialState(enode.ID{}, table, 0, config)
|
||||
resolver := &resolveMock{answer: resolved}
|
||||
state := newDialState(enode.ID{}, 0, config)
|
||||
|
||||
// Check that the task is generated with an incomplete ID.
|
||||
dest := newNode(uintID(1), nil)
|
||||
|
|
@ -630,10 +538,14 @@ func TestDialResolve(t *testing.T) {
|
|||
}
|
||||
|
||||
// Now run the task, it should resolve the ID once.
|
||||
srv := &Server{ntab: table, log: config.Logger, Config: *config}
|
||||
srv := &Server{
|
||||
Config: *config,
|
||||
log: config.Logger,
|
||||
staticNodeResolver: resolver,
|
||||
}
|
||||
tasks[0].Do(srv)
|
||||
if !reflect.DeepEqual(table.resolveCalls, []*enode.Node{dest}) {
|
||||
t.Fatalf("wrong resolve calls, got %v", table.resolveCalls)
|
||||
if !reflect.DeepEqual(resolver.calls, []*enode.Node{dest}) {
|
||||
t.Fatalf("wrong resolve calls, got %v", resolver.calls)
|
||||
}
|
||||
|
||||
// Report it as done to the dialer, which should update the static node record.
|
||||
|
|
@ -666,18 +578,13 @@ func uintID(i uint32) enode.ID {
|
|||
return id
|
||||
}
|
||||
|
||||
// implements discoverTable for TestDialResolve
|
||||
// for TestDialResolve
|
||||
type resolveMock struct {
|
||||
resolveCalls []*enode.Node
|
||||
answer *enode.Node
|
||||
calls []*enode.Node
|
||||
answer *enode.Node
|
||||
}
|
||||
|
||||
func (t *resolveMock) Resolve(n *enode.Node) *enode.Node {
|
||||
t.resolveCalls = append(t.resolveCalls, n)
|
||||
t.calls = append(t.calls, n)
|
||||
return t.answer
|
||||
}
|
||||
|
||||
func (t *resolveMock) Self() *enode.Node { return new(enode.Node) }
|
||||
func (t *resolveMock) Close() {}
|
||||
func (t *resolveMock) LookupRandom() []*enode.Node { return nil }
|
||||
func (t *resolveMock) ReadRandomNodes(buf []*enode.Node) int { return 0 }
|
||||
|
|
|
|||
|
|
@ -54,6 +54,11 @@ type Protocol struct {
|
|||
// but returns nil, it is assumed that the protocol handshake is still running.
|
||||
PeerInfo func(id enode.ID) interface{}
|
||||
|
||||
// DialCandidates, if non-nil, is a way to tell Server about protocol-specific nodes
|
||||
// that should be dialed. The server continuously reads nodes from the iterator and
|
||||
// attempts to create connections to them.
|
||||
DialCandidates enode.Iterator
|
||||
|
||||
// Attributes contains protocol specific information for the node record.
|
||||
Attributes []enr.Entry
|
||||
}
|
||||
|
|
|
|||
|
|
@ -167,16 +167,20 @@ type Server struct {
|
|||
lock sync.Mutex // protects running
|
||||
running bool
|
||||
|
||||
nodedb *enode.DB
|
||||
localnode *enode.LocalNode
|
||||
ntab discoverTable
|
||||
listener net.Listener
|
||||
ourHandshake *protoHandshake
|
||||
DiscV5 *discv5.Network
|
||||
loopWG sync.WaitGroup // loop, listenLoop
|
||||
peerFeed event.Feed
|
||||
log log.Logger
|
||||
|
||||
nodedb *enode.DB
|
||||
localnode *enode.LocalNode
|
||||
ntab *discover.UDPv4
|
||||
DiscV5 *discv5.Network
|
||||
discmix *enode.FairMix
|
||||
|
||||
staticNodeResolver nodeResolver
|
||||
|
||||
// Channels into the run loop.
|
||||
quit chan struct{}
|
||||
addstatic chan *enode.Node
|
||||
|
|
@ -470,7 +474,7 @@ func (srv *Server) Start() (err error) {
|
|||
}
|
||||
|
||||
dynPeers := srv.maxDialedConns()
|
||||
dialer := newDialState(srv.localnode.ID(), srv.ntab, dynPeers, &srv.Config)
|
||||
dialer := newDialState(srv.localnode.ID(), dynPeers, &srv.Config)
|
||||
srv.loopWG.Add(1)
|
||||
go srv.run(dialer)
|
||||
return nil
|
||||
|
|
@ -521,6 +525,18 @@ func (srv *Server) setupLocalNode() error {
|
|||
}
|
||||
|
||||
func (srv *Server) setupDiscovery() error {
|
||||
srv.discmix = enode.NewFairMix(fallbackInterval)
|
||||
|
||||
// Add protocol-specific discovery sources.
|
||||
added := make(map[string]bool)
|
||||
for _, proto := range srv.Protocols {
|
||||
if proto.DialCandidates != nil && !added[proto.Name] {
|
||||
srv.discmix.AddSource(proto.DialCandidates)
|
||||
added[proto.Name] = true
|
||||
}
|
||||
}
|
||||
|
||||
// Don't listen on UDP endpoint if DHT is disabled.
|
||||
if srv.NoDiscovery && !srv.DiscoveryV5 {
|
||||
return nil
|
||||
}
|
||||
|
|
@ -562,7 +578,10 @@ func (srv *Server) setupDiscovery() error {
|
|||
return err
|
||||
}
|
||||
srv.ntab = ntab
|
||||
srv.discmix.AddSource(ntab.RandomNodes())
|
||||
srv.staticNodeResolver = ntab
|
||||
}
|
||||
|
||||
// Discovery V5
|
||||
if srv.DiscoveryV5 {
|
||||
var ntab *discv5.Network
|
||||
|
|
@ -620,6 +639,7 @@ func (srv *Server) run(dialstate dialer) {
|
|||
srv.log.Info("Started P2P networking", "self", srv.localnode.Node().URLv4())
|
||||
defer srv.loopWG.Done()
|
||||
defer srv.nodedb.Close()
|
||||
defer srv.discmix.Close()
|
||||
|
||||
var (
|
||||
peers = make(map[enode.ID]*Peer)
|
||||
|
|
|
|||
|
|
@ -233,8 +233,8 @@ func TestServerTaskScheduling(t *testing.T) {
|
|||
Config: Config{MaxPeers: 10},
|
||||
localnode: enode.NewLocalNode(db, newkey()),
|
||||
nodedb: db,
|
||||
discmix: enode.NewFairMix(0),
|
||||
quit: make(chan struct{}),
|
||||
ntab: fakeTable{},
|
||||
running: true,
|
||||
log: log.New(),
|
||||
}
|
||||
|
|
@ -282,9 +282,9 @@ func TestServerManyTasks(t *testing.T) {
|
|||
quit: make(chan struct{}),
|
||||
localnode: enode.NewLocalNode(db, newkey()),
|
||||
nodedb: db,
|
||||
ntab: fakeTable{},
|
||||
running: true,
|
||||
log: log.New(),
|
||||
discmix: enode.NewFairMix(0),
|
||||
}
|
||||
done = make(chan *testTask)
|
||||
start, end = 0, 0
|
||||
|
|
|
|||
Loading…
Reference in a new issue