From 0da915652b773bd9aff1feeba531f1d1a68ccbb9 Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Mon, 8 Jul 2019 09:56:28 +0200 Subject: [PATCH] p2p: fix dial tests --- p2p/dial.go | 92 ++++++++++++------------ p2p/dial_test.go | 178 ++++++++++++++++++++++++++--------------------- p2p/server.go | 3 + 3 files changed, 147 insertions(+), 126 deletions(-) diff --git a/p2p/dial.go b/p2p/dial.go index d958b6df6b..ad02aa1ed7 100644 --- a/p2p/dial.go +++ b/p2p/dial.go @@ -35,6 +35,7 @@ const ( // private networks. dialHistoryExpiration = inboundThrottleTime + 5*time.Second + // Timeout for NextNode on the discovery iterator. discoveryTimeout = 4 * time.Second // 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) } +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 { @@ -85,29 +90,6 @@ 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 { - 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 { s := &dialstate{ 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 { - 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 { @@ -155,20 +133,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. @@ -183,6 +150,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. @@ -190,14 +171,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-- } } - // 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]) { @@ -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) { if t.dest.Incomplete() { 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. // The backoff delay resets when the node is found. 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") return false } @@ -294,7 +282,7 @@ 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 @@ -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()) } +// 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) { ctx, cancel := context.WithTimeout(context.Background(), discoveryTimeout) defer cancel() @@ -346,6 +342,12 @@ func (t *discoverTask) String() string { 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) } diff --git a/p2p/dial_test.go b/p2p/dial_test.go index 2637d98189..6189ec4d0b 100644 --- a/p2p/dial_test.go +++ b/p2p/dial_test.go @@ -29,6 +29,7 @@ import ( "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/enr" + "github.com/ethereum/go-ethereum/p2p/netutil" ) 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", 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) @@ -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, 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{ &dialTask{flags: dynDialedConn, dest: newNode(uintID(1), nil)}, }, @@ -261,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)}, @@ -279,20 +286,19 @@ func TestDialStateDynDialBootnode(t *testing.T) { { done: []task{ &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 { - 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)}, + {rw: &conn{flags: dynDialedConn, node: newNode(uintID(6), 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. -// func TestDialStateNetRestrict(t *testing.T) { -// // This table always returns the same random nodes -// // in the order given below. -// table := fakeTable{ -// 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")), -// newNode(uintID(4), net.ParseIP("127.0.0.4")), -// newNode(uintID(5), net.ParseIP("127.0.2.5")), -// newNode(uintID(6), net.ParseIP("127.0.2.6")), -// newNode(uintID(7), net.ParseIP("127.0.2.7")), -// newNode(uintID(8), net.ParseIP("127.0.2.8")), -// } -// restrict := new(netutil.Netlist) -// restrict.Add("127.0.2.0/24") -// -// runDialTest(t, dialtest{ -// init: newDialState(enode.ID{}, table, 10, &Config{NetRestrict: restrict}), -// rounds: []round{ -// { -// new: []task{ -// &dialTask{flags: dynDialedConn, dest: table[4]}, -// &discoverTask{}, -// }, -// }, -// }, -// }) -// } +func TestDialStateNetRestrict(t *testing.T) { + // This table always returns the same random nodes + // in the order given below. + 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")), + newNode(uintID(4), net.ParseIP("127.0.0.4")), + newNode(uintID(5), net.ParseIP("127.0.2.5")), + newNode(uintID(6), net.ParseIP("127.0.2.6")), + newNode(uintID(7), net.ParseIP("127.0.2.7")), + newNode(uintID(8), net.ParseIP("127.0.2.8")), + } + restrict := new(netutil.Netlist) + restrict.Add("127.0.2.0/24") + + runDialTest(t, dialtest{ + init: newDialState(enode.ID{}, 10, &Config{NetRestrict: restrict}), + rounds: []round{ + { + new: []task{ + &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}, + }, + }, + }, + }) +} // This test checks that static dials are launched. func TestDialStateStaticDial(t *testing.T) { @@ -503,36 +520,40 @@ func TestDialStateCache(t *testing.T) { }) } -// func TestDialResolve(t *testing.T) { -// config := &Config{ -// Logger: testlog.Logger(t, log.LvlTrace), -// 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) -// -// // Check that the task is generated with an incomplete ID. -// dest := newNode(uintID(1), nil) -// state.addStatic(dest) -// tasks := state.newTasks(0, nil, time.Time{}) -// if !reflect.DeepEqual(tasks, []task{&dialTask{flags: staticDialedConn, dest: dest}}) { -// t.Fatalf("expected dial task, got %#v", tasks) -// } -// -// // Now run the task, it should resolve the ID once. -// srv := &Server{ntab: table, log: config.Logger, Config: *config} -// tasks[0].Do(srv) -// if !reflect.DeepEqual(table.resolveCalls, []*enode.Node{dest}) { -// t.Fatalf("wrong resolve calls, got %v", table.resolveCalls) -// } -// -// // 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") -// } -// } +func TestDialResolve(t *testing.T) { + config := &Config{ + Logger: testlog.Logger(t, log.LvlTrace), + Dialer: TCPDialer{&net.Dialer{Deadline: time.Now().Add(-5 * time.Minute)}}, + } + resolved := newNode(uintID(1), net.IP{127, 0, 55, 234}) + 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) + state.addStatic(dest) + tasks := state.newTasks(0, nil, time.Time{}) + if !reflect.DeepEqual(tasks, []task{&dialTask{flags: staticDialedConn, dest: dest}}) { + t.Fatalf("expected dial task, got %#v", tasks) + } + + // Now run the task, it should resolve the ID once. + srv := &Server{ + Config: *config, + log: config.Logger, + staticNodeResolver: resolver, + } + tasks[0].Do(srv) + 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. + 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. func sametasks(a, b []task) bool { @@ -557,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 } diff --git a/p2p/server.go b/p2p/server.go index 2e410c18b6..329afed126 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -180,6 +180,8 @@ type Server struct { DiscV5 *discv5.Network discmix *discutil.FairMix + staticNodeResolver nodeResolver + // Channels into the run loop. quit chan struct{} addstatic chan *enode.Node @@ -564,6 +566,7 @@ func (srv *Server) setupDiscovery() error { } srv.ntab = ntab srv.discmix.AddSource(ntab.RandomNodes()) + srv.staticNodeResolver = ntab } // Discovery V5