forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

p2p: fix dial tests

Browse source
This commit is contained in:
Felix Lange 2019-07-08 09:56:28 +02:00
parent b317a484d2
commit 0da915652b
3 changed files with 147 additions and 126 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

92
p2p/dial.go
View file

@ -35,6 +35,7 @@ const (
// private networks. // private networks.
dialHistoryExpiration = inboundThrottleTime + 5*time.Second dialHistoryExpiration = inboundThrottleTime + 5*time.Second
// Timeout for NextNode on the discovery iterator.
discoveryTimeout = 4 * time.Second discoveryTimeout = 4 * time.Second
// If no peers are found for this amount of time, the initial bootnodes are dialed. // If no peers are found for this amount of time, the initial bootnodes are dialed.
@ -51,6 +52,10 @@ type NodeDialer interface {
Dial(*enode.Node) (net.Conn, error) 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 // TCPDialer implements the NodeDialer interface by using a net.Dialer to
// create TCP connections to nodes in the network // create TCP connections to nodes in the network
type TCPDialer struct { type TCPDialer struct {
@ -85,29 +90,6 @@ type task interface {
Do(*Server) 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 {
want int
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, maxdyn int, cfg *Config) *dialstate { func newDialState(self enode.ID, maxdyn int, cfg *Config) *dialstate {
s := &dialstate{ s := &dialstate{
maxDynDials: maxdyn, maxDynDials: maxdyn,
@ -140,10 +122,6 @@ func (s *dialstate) removeStatic(n *enode.Node) {
} }
func (s *dialstate) newTasks(nRunning int, peers map[enode.ID]*Peer, now time.Time) []task { 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 var newtasks []task
addDial := func(flag connFlag, n *enode.Node) bool { addDial := func(flag connFlag, n *enode.Node) bool {
if err := s.checkDial(n, peers); err != nil { if err := s.checkDial(n, peers); err != nil {
@ -155,20 +133,9 @@ func (s *dialstate) newTasks(nRunning int, peers map[enode.ID]*Peer, now time.Ti
return true return true
} }
// Compute number of dynamic dials necessary at this point. if s.start.IsZero() {
needDynDials := s.maxDynDials s.start = now
for _, p := range peers {
if p.rw.is(dynDialedConn) {
needDynDials--
}
} }
for _, flag := range s.dialing {
if flag&dynDialedConn != 0 {
needDynDials--
}
}
// Expire the dial history on every invocation.
s.hist.expire(now) s.hist.expire(now)
// Create dials for static nodes if they are not connected. // Create dials for static nodes if they are not connected.
@ -183,6 +150,20 @@ func (s *dialstate) newTasks(nRunning int, peers map[enode.ID]*Peer, now time.Ti
newtasks = append(newtasks, t) 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 // 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 // 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. // table might be full of mostly bad peers, making it hard to find good ones.
@ -190,14 +171,12 @@ func (s *dialstate) newTasks(nRunning int, peers map[enode.ID]*Peer, now time.Ti
bootnode := s.bootnodes[0] bootnode := s.bootnodes[0]
s.bootnodes = append(s.bootnodes[:0], s.bootnodes[1:]...) s.bootnodes = append(s.bootnodes[:0], s.bootnodes[1:]...)
s.bootnodes = append(s.bootnodes, bootnode) s.bootnodes = append(s.bootnodes, bootnode)
if addDial(dynDialedConn, bootnode) { if addDial(dynDialedConn, bootnode) {
needDynDials-- needDynDials--
} }
} }
// Create dynamic dials from random lookup results, removing tried // Create dynamic dials from discovery results.
// items from the result buffer.
i := 0 i := 0
for ; i < len(s.lookupBuf) && needDynDials > 0; i++ { for ; i < len(s.lookupBuf) && needDynDials > 0; i++ {
if addDial(dynDialedConn, s.lookupBuf[i]) { if addDial(dynDialedConn, s.lookupBuf[i]) {
@ -259,6 +238,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) { func (t *dialTask) Do(srv *Server) {
if t.dest.Incomplete() { if t.dest.Incomplete() {
if !t.resolve(srv) { if !t.resolve(srv) {
@ -284,7 +272,7 @@ func (t *dialTask) Do(srv *Server) {
// discovery network with useless queries for nodes that don't exist. // discovery network with useless queries for nodes that don't exist.
// The backoff delay resets when the node is found. // The backoff delay resets when the node is found.
func (t *dialTask) resolve(srv *Server) bool { func (t *dialTask) resolve(srv *Server) bool {
if srv.ntab == nil { if srv.staticNodeResolver == nil {
srv.log.Debug("Can't resolve node", "id", t.dest.ID, "err", "discovery is disabled") srv.log.Debug("Can't resolve node", "id", t.dest.ID, "err", "discovery is disabled")
return false return false
} }
@ -294,7 +282,7 @@ func (t *dialTask) resolve(srv *Server) bool {
if time.Since(t.lastResolved) < t.resolveDelay { if time.Since(t.lastResolved) < t.resolveDelay {
return false return false
} }
resolved := srv.ntab.Resolve(t.dest) resolved := srv.staticNodeResolver.Resolve(t.dest)
t.lastResolved = time.Now() t.lastResolved = time.Now()
if resolved == nil { if resolved == nil {
t.resolveDelay *= 2 t.resolveDelay *= 2
@ -330,6 +318,14 @@ func (t *dialTask) String() string {
return fmt.Sprintf("%v %x %v:%d", t.flags, id[:8], t.dest.IP(), t.dest.TCP()) 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) { func (t *discoverTask) Do(srv *Server) {
ctx, cancel := context.WithTimeout(context.Background(), discoveryTimeout) ctx, cancel := context.WithTimeout(context.Background(), discoveryTimeout)
defer cancel() defer cancel()
@ -346,6 +342,12 @@ func (t *discoverTask) String() string {
return s 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) { func (t waitExpireTask) Do(*Server) {
time.Sleep(t.Duration) time.Sleep(t.Duration)
} }

178
p2p/dial_test.go
View file

@ -29,6 +29,7 @@ import (
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/enr" "github.com/ethereum/go-ethereum/p2p/enr"
"github.com/ethereum/go-ethereum/p2p/netutil"
) )
func init() { func init() {
@ -72,7 +73,7 @@ func runDialTest(t *testing.T, test dialtest) {
t.Errorf("ERROR round %d: got %v\nwant %v\nstate: %v\nrunning: %v", 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)) 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. // Time advances by 16 seconds on every round.
vtime = vtime.Add(16 * time.Second) vtime = vtime.Add(16 * time.Second)
@ -245,11 +246,20 @@ func TestDialStateDynDialBootnode(t *testing.T) {
}, },
}, },
}, },
new: []task{
&dialTask{flags: dynDialedConn, dest: newNode(uintID(4), nil)},
&dialTask{flags: dynDialedConn, dest: newNode(uintID(5), nil)},
&discoverTask{want: 3},
},
}, },
// No dials succeed, bootnodes still pending fallback interval // No dials succeed, bootnodes still pending fallback interval
{}, {},
// No dials succeed, 2 dynamic dials attempted and 1 bootnode too as fallback interval was reached // 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)},
},
new: []task{ new: []task{
&dialTask{flags: dynDialedConn, dest: newNode(uintID(1), nil)}, &dialTask{flags: dynDialedConn, dest: newNode(uintID(1), nil)},
}, },
@ -261,15 +271,12 @@ func TestDialStateDynDialBootnode(t *testing.T) {
}, },
new: []task{ new: []task{
&dialTask{flags: dynDialedConn, dest: newNode(uintID(2), nil)}, &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 // No dials succeed, 3rd bootnode is attempted
{ {
done: []task{ done: []task{
&dialTask{flags: dynDialedConn, dest: newNode(uintID(2), nil)}, &dialTask{flags: dynDialedConn, dest: newNode(uintID(2), nil)},
&dialTask{flags: dynDialedConn, dest: newNode(uintID(5), nil)},
}, },
new: []task{ new: []task{
&dialTask{flags: dynDialedConn, dest: newNode(uintID(3), nil)}, &dialTask{flags: dynDialedConn, dest: newNode(uintID(3), nil)},
@ -279,20 +286,19 @@ func TestDialStateDynDialBootnode(t *testing.T) {
{ {
done: []task{ done: []task{
&dialTask{flags: dynDialedConn, dest: newNode(uintID(3), nil)}, &dialTask{flags: dynDialedConn, dest: newNode(uintID(3), nil)},
&discoverTask{results: []*enode.Node{
newNode(uintID(6), nil),
}},
},
new: []task{
&dialTask{flags: dynDialedConn, dest: newNode(uintID(6), nil)},
&discoverTask{want: 4},
}, },
new: []task{},
}, },
// Random dial succeeds, no more bootnodes are attempted // Random dial succeeds, no more bootnodes are attempted
{ {
new: []task{
&waitExpireTask{3 * time.Second},
},
peers: []*Peer{ peers: []*Peer{
{rw: &conn{flags: dynDialedConn, node: newNode(uintID(4), nil)}}, {rw: &conn{flags: dynDialedConn, node: newNode(uintID(6), nil)}},
},
done: []task{
&dialTask{flags: dynDialedConn, dest: newNode(uintID(1), nil)},
&dialTask{flags: dynDialedConn, dest: newNode(uintID(4), nil)},
}, },
}, },
}, },
@ -308,34 +314,45 @@ func newNode(id enode.ID, ip net.IP) *enode.Node {
} }
// // 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) { func TestDialStateNetRestrict(t *testing.T) {
// // This table always returns the same random nodes // This table always returns the same random nodes
// // in the order given below. // in the order given below.
// table := fakeTable{ nodes := []*enode.Node{
// newNode(uintID(1), net.ParseIP("127.0.0.1")), newNode(uintID(1), net.ParseIP("127.0.0.1")),
// newNode(uintID(2), net.ParseIP("127.0.0.2")), newNode(uintID(2), net.ParseIP("127.0.0.2")),
// newNode(uintID(3), net.ParseIP("127.0.0.3")), newNode(uintID(3), net.ParseIP("127.0.0.3")),
// newNode(uintID(4), net.ParseIP("127.0.0.4")), newNode(uintID(4), net.ParseIP("127.0.0.4")),
// newNode(uintID(5), net.ParseIP("127.0.2.5")), newNode(uintID(5), net.ParseIP("127.0.2.5")),
// newNode(uintID(6), net.ParseIP("127.0.2.6")), newNode(uintID(6), net.ParseIP("127.0.2.6")),
// newNode(uintID(7), net.ParseIP("127.0.2.7")), newNode(uintID(7), net.ParseIP("127.0.2.7")),
// newNode(uintID(8), net.ParseIP("127.0.2.8")), newNode(uintID(8), net.ParseIP("127.0.2.8")),
// } }
// restrict := new(netutil.Netlist) restrict := new(netutil.Netlist)
// restrict.Add("127.0.2.0/24") restrict.Add("127.0.2.0/24")
//
// runDialTest(t, dialtest{ runDialTest(t, dialtest{
// init: newDialState(enode.ID{}, table, 10, &Config{NetRestrict: restrict}), init: newDialState(enode.ID{}, 10, &Config{NetRestrict: restrict}),
// rounds: []round{ rounds: []round{
// { {
// new: []task{ new: []task{
// &dialTask{flags: dynDialedConn, dest: table[4]}, &discoverTask{want: 10},
// &discoverTask{}, },
// }, },
// }, {
// }, 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},
},
},
},
})
}
// This test checks that static dials are launched. // This test checks that static dials are launched.
func TestDialStateStaticDial(t *testing.T) { func TestDialStateStaticDial(t *testing.T) {
@ -503,36 +520,40 @@ func TestDialStateCache(t *testing.T) {
}) })
} }
// func TestDialResolve(t *testing.T) { func TestDialResolve(t *testing.T) {
// config := &Config{ config := &Config{
// Logger: testlog.Logger(t, log.LvlTrace), Logger: testlog.Logger(t, log.LvlTrace),
// Dialer: TCPDialer{&net.Dialer{Deadline: time.Now().Add(-5 * time.Minute)}}, Dialer: TCPDialer{&net.Dialer{Deadline: time.Now().Add(-5 * time.Minute)}},
// } }
// resolved := newNode(uintID(1), net.IP{127, 0, 55, 234}) resolved := newNode(uintID(1), net.IP{127, 0, 55, 234})
// table := &resolveMock{answer: resolved} resolver := &resolveMock{answer: resolved}
// state := newDialState(enode.ID{}, table, 0, config) state := newDialState(enode.ID{}, 0, config)
//
// // Check that the task is generated with an incomplete ID. // Check that the task is generated with an incomplete ID.
// dest := newNode(uintID(1), nil) dest := newNode(uintID(1), nil)
// state.addStatic(dest) state.addStatic(dest)
// tasks := state.newTasks(0, nil, time.Time{}) tasks := state.newTasks(0, nil, time.Time{})
// if !reflect.DeepEqual(tasks, []task{&dialTask{flags: staticDialedConn, dest: dest}}) { if !reflect.DeepEqual(tasks, []task{&dialTask{flags: staticDialedConn, dest: dest}}) {
// t.Fatalf("expected dial task, got %#v", tasks) t.Fatalf("expected dial task, got %#v", tasks)
// } }
//
// // Now run the task, it should resolve the ID once. // Now run the task, it should resolve the ID once.
// srv := &Server{ntab: table, log: config.Logger, Config: *config} srv := &Server{
// tasks[0].Do(srv) Config: *config,
// if !reflect.DeepEqual(table.resolveCalls, []*enode.Node{dest}) { log: config.Logger,
// t.Fatalf("wrong resolve calls, got %v", table.resolveCalls) staticNodeResolver: resolver,
// } }
// tasks[0].Do(srv)
// // Report it as done to the dialer, which should update the static node record. if !reflect.DeepEqual(resolver.calls, []*enode.Node{dest}) {
// state.taskDone(tasks[0], time.Now()) t.Fatalf("wrong resolve calls, got %v", resolver.calls)
// if state.static[uintID(1)].dest != resolved { }
// t.Fatalf("state.dest not updated")
// } // Report it as done to the dialer, which should update the static node record.
// } state.taskDone(tasks[0], time.Now())
if state.static[uintID(1)].dest != resolved {
t.Fatalf("state.dest not updated")
}
}
// compares task lists but doesn't care about the order. // compares task lists but doesn't care about the order.
func sametasks(a, b []task) bool { func sametasks(a, b []task) bool {
@ -557,18 +578,13 @@ func uintID(i uint32) enode.ID {
return id return id
} }
// implements discoverTable for TestDialResolve // for TestDialResolve
type resolveMock struct { type resolveMock struct {
resolveCalls []*enode.Node calls []*enode.Node
answer *enode.Node answer *enode.Node
} }
func (t *resolveMock) Resolve(n *enode.Node) *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 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 }

3
p2p/server.go
View file

@ -180,6 +180,8 @@ type Server struct {
DiscV5 *discv5.Network DiscV5 *discv5.Network
discmix *discutil.FairMix discmix *discutil.FairMix
staticNodeResolver nodeResolver
// Channels into the run loop. // Channels into the run loop.
quit chan struct{} quit chan struct{}
addstatic chan *enode.Node addstatic chan *enode.Node
@ -564,6 +566,7 @@ func (srv *Server) setupDiscovery() error {
} }
srv.ntab = ntab srv.ntab = ntab
srv.discmix.AddSource(ntab.RandomNodes()) srv.discmix.AddSource(ntab.RandomNodes())
srv.staticNodeResolver = ntab
} }
// Discovery V5 // Discovery V5