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

p2p, p2p/discover: integrate enode.LocalNode

Browse source 
Startup is much faster because it doesn't need to wait for the external
IP query anymore.
This commit is contained in:
Felix Lange 2018-09-21 11:13:42 +02:00
parent cbd0a2cef1
commit 16b135060b
11 changed files with 240 additions and 207 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
p2p/dial.go
View file

@ -71,6 +71,7 @@ type dialstate struct {
maxDynDials int
ntab discoverTable
netrestrict *netutil.Netlist
self enode.ID
lookupRunning bool
dialing map[enode.ID]connFlag
@ -84,7 +85,6 @@ type dialstate struct {
}
type discoverTable interface {
Self() *enode.Node
Close()
Resolve(*enode.Node) *enode.Node
LookupRandom() []*enode.Node
@ -126,10 +126,11 @@ type waitExpireTask struct {
time.Duration
}
func newDialState(static []*enode.Node, bootnodes []*enode.Node, ntab discoverTable, maxdyn int, netrestrict *netutil.Netlist) *dialstate {
func newDialState(self enode.ID, static []*enode.Node, bootnodes []*enode.Node, ntab discoverTable, maxdyn int, netrestrict *netutil.Netlist) *dialstate {
s := &dialstate{
maxDynDials: maxdyn,
ntab: ntab,
self: self,
netrestrict: netrestrict,
static: make(map[enode.ID]*dialTask),
dialing: make(map[enode.ID]connFlag),
@ -266,7 +267,7 @@ func (s *dialstate) checkDial(n *enode.Node, peers map[enode.ID]*Peer) error {
return errAlreadyDialing
case peers[n.ID()] != nil:
return errAlreadyConnected
case s.ntab != nil && n.ID() == s.ntab.Self().ID():
case n.ID() == s.self:
return errSelf
case s.netrestrict != nil && !s.netrestrict.Contains(n.IP()):
return errNotWhitelisted

16
p2p/dial_test.go
View file

@ -89,7 +89,7 @@ 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) {
runDialTest(t, dialtest{
init: newDialState(nil, nil, fakeTable{}, 5, nil),
init: newDialState(enode.ID{}, nil, nil, fakeTable{}, 5, nil),
rounds: []round{
// A discovery query is launched.
{
@ -236,7 +236,7 @@ func TestDialStateDynDialBootnode(t *testing.T) {
newNode(uintID(8), nil),
}
runDialTest(t, dialtest{
init: newDialState(nil, bootnodes, table, 5, nil),
init: newDialState(enode.ID{}, nil, bootnodes, table, 5, nil),
rounds: []round{
// 2 dynamic dials attempted, bootnodes pending fallback interval
{
@ -324,7 +324,7 @@ func TestDialStateDynDialFromTable(t *testing.T) {
}
runDialTest(t, dialtest{
init: newDialState(nil, nil, table, 10, nil),
init: newDialState(enode.ID{}, nil, nil, table, 10, nil),
rounds: []round{
// 5 out of 8 of the nodes returned by ReadRandomNodes are dialed.
{
@ -430,7 +430,7 @@ func TestDialStateNetRestrict(t *testing.T) {
restrict.Add("127.0.2.0/24")
runDialTest(t, dialtest{
init: newDialState(nil, nil, table, 10, restrict),
init: newDialState(enode.ID{}, nil, nil, table, 10, restrict),
rounds: []round{
{
new: []task{
@ -453,7 +453,7 @@ func TestDialStateStaticDial(t *testing.T) {
}
runDialTest(t, dialtest{
init: newDialState(wantStatic, nil, fakeTable{}, 0, nil),
init: newDialState(enode.ID{}, wantStatic, nil, fakeTable{}, 0, nil),
rounds: []round{
// Static dials are launched for the nodes that
// aren't yet connected.
@ -557,7 +557,7 @@ func TestDialStaticAfterReset(t *testing.T) {
},
}
dTest := dialtest{
init: newDialState(wantStatic, nil, fakeTable{}, 0, nil),
init: newDialState(enode.ID{}, wantStatic, nil, fakeTable{}, 0, nil),
rounds: rounds,
}
runDialTest(t, dTest)
@ -578,7 +578,7 @@ func TestDialStateCache(t *testing.T) {
}
runDialTest(t, dialtest{
init: newDialState(wantStatic, nil, fakeTable{}, 0, nil),
init: newDialState(enode.ID{}, wantStatic, nil, fakeTable{}, 0, nil),
rounds: []round{
// Static dials are launched for the nodes that
// aren't yet connected.
@ -640,7 +640,7 @@ func TestDialStateCache(t *testing.T) {
func TestDialResolve(t *testing.T) {
resolved := newNode(uintID(1), net.IP{127, 0, 55, 234})
table := &resolveMock{answer: resolved}
state := newDialState(nil, nil, table, 0, nil)
state := newDialState(enode.ID{}, nil, nil, table, 0, nil)
// Check that the task is generated with an incomplete ID.
dest := newNode(uintID(1), nil)

27
p2p/discover/table.go
View file

@ -72,21 +72,20 @@ type Table struct {
ips netutil.DistinctNetSet
db *enode.DB // database of known nodes
net transport
refreshReq chan chan struct{}
initDone chan struct{}
closeReq chan struct{}
closed chan struct{}
nodeAddedHook func(*node) // for testing
net transport
self *node // metadata of the local node
}
// transport is implemented by the UDP transport.
// it is an interface so we can test without opening lots of UDP
// sockets and without generating a private key.
type transport interface {
self() *enode.Node
ping(enode.ID, *net.UDPAddr) error
findnode(toid enode.ID, addr *net.UDPAddr, target encPubkey) ([]*node, error)
close()
@ -100,11 +99,10 @@ type bucket struct {
ips netutil.DistinctNetSet
}
func newTable(t transport, self *enode.Node, db *enode.DB, bootnodes []*enode.Node) (*Table, error) {
func newTable(t transport, db *enode.DB, bootnodes []*enode.Node) (*Table, error) {
tab := &Table{
net: t,
db: db,
self: wrapNode(self),
refreshReq: make(chan chan struct{}),
initDone: make(chan struct{}),
closeReq: make(chan struct{}),
@ -127,6 +125,10 @@ func newTable(t transport, self *enode.Node, db *enode.DB, bootnodes []*enode.No
return tab, nil
}
func (tab *Table) self() *enode.Node {
return tab.net.self()
}
func (tab *Table) seedRand() {
var b [8]byte
crand.Read(b[:])
@ -136,11 +138,6 @@ func (tab *Table) seedRand() {
tab.mutex.Unlock()
}
// Self returns the local node.
func (tab *Table) Self() *enode.Node {
return unwrapNode(tab.self)
}
// ReadRandomNodes fills the given slice with random nodes from the table. The results
// are guaranteed to be unique for a single invocation, no node will appear twice.
func (tab *Table) ReadRandomNodes(buf []*enode.Node) (n int) {
@ -257,7 +254,7 @@ func (tab *Table) lookup(targetKey encPubkey, refreshIfEmpty bool) []*node {
)
// don't query further if we hit ourself.
// unlikely to happen often in practice.
asked[tab.self.ID()] = true
asked[tab.self().ID()] = true
for {
tab.mutex.Lock()
@ -408,7 +405,7 @@ func (tab *Table) doRefresh(done chan struct{}) {
// Run self lookup to discover new neighbor nodes.
// We can only do this if we have a secp256k1 identity.
var key ecdsa.PublicKey
if err := tab.self.Load((*enode.Secp256k1)(&key)); err == nil {
if err := tab.self().Load((*enode.Secp256k1)(&key)); err == nil {
tab.lookup(encodePubkey(&key), false)
}
@ -530,7 +527,7 @@ func (tab *Table) len() (n int) {
// bucket returns the bucket for the given node ID hash.
func (tab *Table) bucket(id enode.ID) *bucket {
d := enode.LogDist(tab.self.ID(), id)
d := enode.LogDist(tab.self().ID(), id)
if d <= bucketMinDistance {
return tab.buckets[0]
}
@ -543,7 +540,7 @@ func (tab *Table) bucket(id enode.ID) *bucket {
//
// The caller must not hold tab.mutex.
func (tab *Table) add(n *node) {
if n.ID() == tab.self.ID() {
if n.ID() == tab.self().ID() {
return
}
@ -576,7 +573,7 @@ func (tab *Table) stuff(nodes []*node) {
defer tab.mutex.Unlock()
for _, n := range nodes {
if n.ID() == tab.self.ID() {
if n.ID() == tab.self().ID() {
continue // don't add self
}
b := tab.bucket(n.ID())

10
p2p/discover/table_test.go
View file

@ -141,7 +141,7 @@ func TestTable_IPLimit(t *testing.T) {
defer db.Close()
for i := 0; i < tableIPLimit+1; i++ {
n := nodeAtDistance(tab.self.ID(), i, net.IP{172, 0, 1, byte(i)})
n := nodeAtDistance(tab.self().ID(), i, net.IP{172, 0, 1, byte(i)})
tab.add(n)
}
if tab.len() > tableIPLimit {
@ -158,7 +158,7 @@ func TestTable_BucketIPLimit(t *testing.T) {
d := 3
for i := 0; i < bucketIPLimit+1; i++ {
n := nodeAtDistance(tab.self.ID(), d, net.IP{172, 0, 1, byte(i)})
n := nodeAtDistance(tab.self().ID(), d, net.IP{172, 0, 1, byte(i)})
tab.add(n)
}
if tab.len() > bucketIPLimit {
@ -240,7 +240,7 @@ func TestTable_ReadRandomNodesGetAll(t *testing.T) {
for i := 0; i < len(buf); i++ {
ld := cfg.Rand.Intn(len(tab.buckets))
tab.stuff([]*node{nodeAtDistance(tab.self.ID(), ld, intIP(ld))})
tab.stuff([]*node{nodeAtDistance(tab.self().ID(), ld, intIP(ld))})
}
gotN := tab.ReadRandomNodes(buf)
if gotN != tab.len() {
@ -510,6 +510,10 @@ type preminedTestnet struct {
dists [hashBits + 1][]encPubkey
}
func (tn *preminedTestnet) self() *enode.Node {
return nullNode
}
func (tn *preminedTestnet) findnode(toid enode.ID, toaddr *net.UDPAddr, target encPubkey) ([]*node, error) {
// current log distance is encoded in port number
// fmt.Println("findnode query at dist", toaddr.Port)

25
p2p/discover/table_util_test.go
View file

@ -28,12 +28,17 @@ import (
"github.com/ethereum/go-ethereum/p2p/enr"
)
func newTestTable(t transport) (*Table, *enode.DB) {
var nullNode *enode.Node
func init() {
var r enr.Record
r.Set(enr.IP{0, 0, 0, 0})
n := enode.SignNull(&r, enode.ID{})
nullNode = enode.SignNull(&r, enode.ID{})
}
func newTestTable(t transport) (*Table, *enode.DB) {
db, _ := enode.OpenDB("")
tab, _ := newTable(t, n, db, nil)
tab, _ := newTable(t, db, nil)
return tab, db
}
@ -70,10 +75,10 @@ func intIP(i int) net.IP {
// fillBucket inserts nodes into the given bucket until it is full.
func fillBucket(tab *Table, n *node) (last *node) {
ld := enode.LogDist(tab.self.ID(), n.ID())
ld := enode.LogDist(tab.self().ID(), n.ID())
b := tab.bucket(n.ID())
for len(b.entries) < bucketSize {
b.entries = append(b.entries, nodeAtDistance(tab.self.ID(), ld, intIP(ld)))
b.entries = append(b.entries, nodeAtDistance(tab.self().ID(), ld, intIP(ld)))
}
return b.entries[bucketSize-1]
}
@ -81,15 +86,25 @@ func fillBucket(tab *Table, n *node) (last *node) {
type pingRecorder struct {
mu sync.Mutex
dead, pinged map[enode.ID]bool
n *enode.Node
}
func newPingRecorder() *pingRecorder {
var r enr.Record
r.Set(enr.IP{0, 0, 0, 0})
n := enode.SignNull(&r, enode.ID{})
return &pingRecorder{
dead: make(map[enode.ID]bool),
pinged: make(map[enode.ID]bool),
n: n,
}
}
func (t *pingRecorder) self() *enode.Node {
return nullNode
}
func (t *pingRecorder) findnode(toid enode.ID, toaddr *net.UDPAddr, target encPubkey) ([]*node, error) {
return nil, nil
}

82
p2p/discover/udp.go
View file

@ -118,9 +118,11 @@ type (
)
func makeEndpoint(addr *net.UDPAddr, tcpPort uint16) rpcEndpoint {
ip := addr.IP.To4()
if ip == nil {
ip = addr.IP.To16()
ip := net.IP{}
if ip4 := addr.IP.To4(); ip4 != nil {
ip = ip4
} else if ip6 := addr.IP.To16(); ip6 != nil {
ip = ip6
}
return rpcEndpoint{IP: ip, UDP: uint16(addr.Port), TCP: tcpPort}
}
@ -165,20 +167,19 @@ type conn interface {
LocalAddr() net.Addr
}
// udp implements the RPC protocol.
// udp implements the discovery v4 UDP wire protocol.
type udp struct {
conn conn
netrestrict *netutil.Netlist
priv *ecdsa.PrivateKey
ourEndpoint rpcEndpoint
localNode *enode.LocalNode
db *enode.DB
tab *Table
nat nat.Interface
addpending chan *pending
gotreply chan reply
closing chan struct{}
nat nat.Interface
*Table
closing chan struct{}
}
// pending represents a pending reply.
@ -230,62 +231,58 @@ type Config struct {
PrivateKey *ecdsa.PrivateKey
// These settings are optional:
AnnounceAddr *net.UDPAddr // local address announced in the DHT
NodeDBPath string // if set, the node database is stored at this filesystem location
NetRestrict *netutil.Netlist // network whitelist
Bootnodes []*enode.Node // list of bootstrap nodes
Unhandled chan<- ReadPacket // unhandled packets are sent on this channel
NetRestrict *netutil.Netlist // network whitelist
Bootnodes []*enode.Node // list of bootstrap nodes
Unhandled chan<- ReadPacket // unhandled packets are sent on this channel
}
// ListenUDP returns a new table that listens for UDP packets on laddr.
func ListenUDP(c conn, cfg Config) (*Table, error) {
tab, _, err := newUDP(c, cfg)
func ListenUDP(c conn, ln *enode.LocalNode, cfg Config) (*Table, error) {
tab, _, err := newUDP(c, ln, cfg)
if err != nil {
return nil, err
}
log.Info("UDP listener up", "self", tab.self)
return tab, nil
}
func newUDP(c conn, cfg Config) (*Table, *udp, error) {
realaddr := c.LocalAddr().(*net.UDPAddr)
if cfg.AnnounceAddr != nil {
realaddr = cfg.AnnounceAddr
}
self := enode.NewV4(&cfg.PrivateKey.PublicKey, realaddr.IP, realaddr.Port, realaddr.Port)
db, err := enode.OpenDB(cfg.NodeDBPath)
if err != nil {
return nil, nil, err
}
func newUDP(c conn, ln *enode.LocalNode, cfg Config) (*Table, *udp, error) {
udp := &udp{
conn: c,
priv: cfg.PrivateKey,
netrestrict: cfg.NetRestrict,
localNode: ln,
db: ln.Database(),
closing: make(chan struct{}),
gotreply: make(chan reply),
addpending: make(chan *pending),
}
// TODO: separate TCP port
udp.ourEndpoint = makeEndpoint(realaddr, uint16(realaddr.Port))
tab, err := newTable(udp, self, db, cfg.Bootnodes)
tab, err := newTable(udp, ln.Database(), cfg.Bootnodes)
if err != nil {
return nil, nil, err
}
udp.Table = tab
udp.tab = tab
go udp.loop()
go udp.readLoop(cfg.Unhandled)
return udp.Table, udp, nil
return udp.tab, udp, nil
}
func (t *udp) self() *enode.Node {
return t.localNode.Node()
}
func (t *udp) close() {
close(t.closing)
t.conn.Close()
t.db.Close()
// TODO: wait for the loops to end.
}
func (t *udp) ourEndpoint() rpcEndpoint {
n := t.self()
a := &net.UDPAddr{IP: n.IP(), Port: n.UDP()}
return makeEndpoint(a, uint16(n.TCP()))
}
// ping sends a ping message to the given node and waits for a reply.
func (t *udp) ping(toid enode.ID, toaddr *net.UDPAddr) error {
return <-t.sendPing(toid, toaddr, nil)
@ -296,7 +293,7 @@ func (t *udp) ping(toid enode.ID, toaddr *net.UDPAddr) error {
func (t *udp) sendPing(toid enode.ID, toaddr *net.UDPAddr, callback func()) <-chan error {
req := &ping{
Version: 4,
From: t.ourEndpoint,
From: t.ourEndpoint(),
To: makeEndpoint(toaddr, 0), // TODO: maybe use known TCP port from DB
Expiration: uint64(time.Now().Add(expiration).Unix()),
}
@ -313,6 +310,7 @@ func (t *udp) sendPing(toid enode.ID, toaddr *net.UDPAddr, callback func()) <-ch
}
return ok
})
t.localNode.UDPContact(toaddr)
t.write(toaddr, req.name(), packet)
return errc
}
@ -629,10 +627,11 @@ func (req *ping) handle(t *udp, from *net.UDPAddr, fromKey encPubkey, mac []byte
n := wrapNode(enode.NewV4(key, from.IP, int(req.From.TCP), from.Port))
t.handleReply(n.ID(), pingPacket, req)
if time.Since(t.db.LastPongReceived(n.ID())) > bondExpiration {
t.sendPing(n.ID(), from, func() { t.addThroughPing(n) })
t.sendPing(n.ID(), from, func() { t.tab.addThroughPing(n) })
} else {
t.addThroughPing(n)
t.tab.addThroughPing(n)
}
t.localNode.UDPEndpointStatement(from, &net.UDPAddr{IP: req.To.IP, Port: int(req.To.UDP)})
t.db.UpdateLastPingReceived(n.ID(), time.Now())
return nil
}
@ -647,6 +646,7 @@ func (req *pong) handle(t *udp, from *net.UDPAddr, fromKey encPubkey, mac []byte
if !t.handleReply(fromID, pongPacket, req) {
return errUnsolicitedReply
}
t.localNode.UDPEndpointStatement(from, &net.UDPAddr{IP: req.To.IP, Port: int(req.To.UDP)})
t.db.UpdateLastPongReceived(fromID, time.Now())
return nil
}
@ -668,9 +668,9 @@ func (req *findnode) handle(t *udp, from *net.UDPAddr, fromKey encPubkey, mac []
return errUnknownNode
}
target := enode.ID(crypto.Keccak256Hash(req.Target[:]))
t.mutex.Lock()
closest := t.closest(target, bucketSize).entries
t.mutex.Unlock()
t.tab.mutex.Lock()
closest := t.tab.closest(target, bucketSize).entries
t.tab.mutex.Unlock()
p := neighbors{Expiration: uint64(time.Now().Add(expiration).Unix())}
var sent bool

11
p2p/discover/udp_test.go
View file

@ -71,7 +71,9 @@ func newUDPTest(t *testing.T) *udpTest {
remotekey: newkey(),
remoteaddr: &net.UDPAddr{IP: net.IP{10, 0, 1, 99}, Port: 30303},
}
test.table, test.udp, _ = newUDP(test.pipe, Config{PrivateKey: test.localkey})
db, _ := enode.OpenDB("")
ln := enode.NewLocalNode(db, test.localkey)
test.table, test.udp, _ = newUDP(test.pipe, ln, Config{PrivateKey: test.localkey})
// Wait for initial refresh so the table doesn't send unexpected findnode.
<-test.table.initDone
return test
@ -355,12 +357,13 @@ func TestUDP_successfulPing(t *testing.T) {
// remote is unknown, the table pings back.
hash, _ := test.waitPacketOut(func(p *ping) error {
if !reflect.DeepEqual(p.From, test.udp.ourEndpoint) {
t.Errorf("got ping.From %v, want %v", p.From, test.udp.ourEndpoint)
if !reflect.DeepEqual(p.From, test.udp.ourEndpoint()) {
t.Errorf("got ping.From %#v, want %#v", p.From, test.udp.ourEndpoint())
}
wantTo := rpcEndpoint{
// The mirrored UDP address is the UDP packet sender.
IP: test.remoteaddr.IP, UDP: uint16(test.remoteaddr.Port),
IP: test.remoteaddr.IP,
UDP: uint16(test.remoteaddr.Port),
TCP: 0,
}
if !reflect.DeepEqual(p.To, wantTo) {

3
p2p/discv5/udp.go
View file

@ -230,7 +230,8 @@ type udp struct {
}
// ListenUDP returns a new table that listens for UDP packets on laddr.
func ListenUDP(priv *ecdsa.PrivateKey, conn conn, realaddr *net.UDPAddr, nodeDBPath string, netrestrict *netutil.Netlist) (*Network, error) {
func ListenUDP(priv *ecdsa.PrivateKey, conn conn, nodeDBPath string, netrestrict *netutil.Netlist) (*Network, error) {
realaddr := conn.LocalAddr().(*net.UDPAddr)
transport, err := listenUDP(priv, conn, realaddr)
if err != nil {
return nil, err

10
p2p/protocol.go
View file

@ -20,6 +20,7 @@ import (
"fmt"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/enr"
)
// Protocol represents a P2P subprotocol implementation.
@ -52,6 +53,9 @@ type Protocol struct {
// 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.
PeerInfo func(id enode.ID) interface{}
// Attributes contains protocol specific information for the node record.
Attributes []enr.Entry
}
func (p Protocol) cap() Cap {
@ -64,10 +68,6 @@ type Cap struct {
Version uint
}
func (cap Cap) RlpData() interface{} {
return []interface{}{cap.Name, cap.Version}
}
func (cap Cap) String() string {
return fmt.Sprintf("%s/%d", cap.Name, cap.Version)
}
@ -79,3 +79,5 @@ func (cs capsByNameAndVersion) Swap(i, j int) { cs[i], cs[j] = cs[j], cs[i] }
func (cs capsByNameAndVersion) Less(i, j int) bool {
return cs[i].Name < cs[j].Name || (cs[i].Name == cs[j].Name && cs[i].Version < cs[j].Version)
}
func (capsByNameAndVersion) ENRKey() string { return "cap" }

230
p2p/server.go
View file

@ -20,9 +20,11 @@ package p2p
import (
"bytes"
"crypto/ecdsa"
"encoding/hex"
"errors"
"fmt"
"net"
"sort"
"sync"
"sync/atomic"
"time"
@ -37,6 +39,7 @@ import (
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/nat"
"github.com/ethereum/go-ethereum/p2p/netutil"
"github.com/ethereum/go-ethereum/rlp"
)
const (
@ -160,6 +163,8 @@ type Server struct {
lock sync.Mutex // protects running
running bool
nodedb *enode.DB
localnode *enode.LocalNode
ntab discoverTable
listener net.Listener
ourHandshake *protoHandshake
@ -347,43 +352,13 @@ func (srv *Server) SubscribeEvents(ch chan *PeerEvent) event.Subscription {
// Self returns the local node's endpoint information.
func (srv *Server) Self() *enode.Node {
srv.lock.Lock()
running, listener, ntab := srv.running, srv.listener, srv.ntab
ln := srv.localnode
srv.lock.Unlock()
if !running {
if ln == nil {
return enode.NewV4(&srv.PrivateKey.PublicKey, net.ParseIP("0.0.0.0"), 0, 0)
}
return srv.makeSelf(listener, ntab)
}
func (srv *Server) makeSelf(listener net.Listener, ntab discoverTable) *enode.Node {
// If the node is running but discovery is off, manually assemble the node infos.
if ntab == nil {
addr := srv.tcpAddr(listener)
return enode.NewV4(&srv.PrivateKey.PublicKey, addr.IP, addr.Port, 0)
}
// Otherwise return the discovery node.
return ntab.Self()
}
func (srv *Server) tcpAddr(listener net.Listener) net.TCPAddr {
addr := net.TCPAddr{IP: net.IP{0, 0, 0, 0}}
if listener == nil {
return addr // Inbound connections disabled, use zero address.
}
// Otherwise inject the listener address too.
if a, ok := listener.Addr().(*net.TCPAddr); ok {
addr = *a
}
if srv.NAT != nil {
if ip, err := srv.NAT.ExternalIP(); err == nil {
addr.IP = ip
}
}
if addr.IP.IsUnspecified() {
addr.IP = net.IP{127, 0, 0, 1}
}
return addr
return ln.Node()
}
// Stop terminates the server and all active peer connections.
@ -395,6 +370,7 @@ func (srv *Server) Stop() {
return
}
srv.running = false
srv.nodedb.Close()
if srv.listener != nil {
// this unblocks listener Accept
srv.listener.Close()
@ -443,7 +419,9 @@ func (srv *Server) Start() (err error) {
if srv.log == nil {
srv.log = log.New()
}
srv.log.Info("Starting P2P networking")
if srv.NoDial && srv.ListenAddr == "" {
srv.log.Warn("P2P server will be useless, neither dialing nor listening")
}
// static fields
if srv.PrivateKey == nil {
@ -466,65 +444,113 @@ func (srv *Server) Start() (err error) {
srv.peerOp = make(chan peerOpFunc)
srv.peerOpDone = make(chan struct{})
var (
conn *net.UDPConn
sconn *sharedUDPConn
realaddr *net.UDPAddr
unhandled chan discover.ReadPacket
)
if !srv.NoDiscovery || srv.DiscoveryV5 {
addr, err := net.ResolveUDPAddr("udp", srv.ListenAddr)
if err != nil {
if err := srv.setupLocalNode(); err != nil {
return err
}
if srv.ListenAddr != "" {
if err := srv.setupListening(); err != nil {
return err
}
conn, err = net.ListenUDP("udp", addr)
if err != nil {
return err
}
realaddr = conn.LocalAddr().(*net.UDPAddr)
if srv.NAT != nil {
if !realaddr.IP.IsLoopback() {
go nat.Map(srv.NAT, srv.quit, "udp", realaddr.Port, realaddr.Port, "ethereum discovery")
}
// TODO: react to external IP changes over time.
if ext, err := srv.NAT.ExternalIP(); err == nil {
realaddr = &net.UDPAddr{IP: ext, Port: realaddr.Port}
}
}
}
if err := srv.setupDiscovery(); err != nil {
return err
}
if !srv.NoDiscovery && srv.DiscoveryV5 {
unhandled = make(chan discover.ReadPacket, 100)
sconn = &sharedUDPConn{conn, unhandled}
dynPeers := srv.maxDialedConns()
dialer := newDialState(srv.localnode.ID(), srv.StaticNodes, srv.BootstrapNodes, srv.ntab, dynPeers, srv.NetRestrict)
srv.loopWG.Add(1)
go srv.run(dialer)
return nil
}
func (srv *Server) setupLocalNode() error {
// Create the devp2p handshake.
pubkey := crypto.FromECDSAPub(&srv.PrivateKey.PublicKey)
srv.ourHandshake = &protoHandshake{Version: baseProtocolVersion, Name: srv.Name, ID: pubkey[1:]}
for _, p := range srv.Protocols {
srv.ourHandshake.Caps = append(srv.ourHandshake.Caps, p.cap())
}
sort.Sort(capsByNameAndVersion(srv.ourHandshake.Caps))
// Create the local node.
db, err := enode.OpenDB(srv.Config.NodeDatabase)
if err != nil {
return err
}
srv.nodedb = db
srv.localnode = enode.NewLocalNode(db, srv.PrivateKey)
srv.localnode.SetFallbackIP(net.IP{127, 0, 0, 1})
srv.localnode.Set(capsByNameAndVersion(srv.ourHandshake.Caps))
// TODO: check conflicts
for _, p := range srv.Protocols {
for _, e := range p.Attributes {
srv.localnode.Set(e)
}
}
if srv.NAT != nil {
// Ask the router about the IP. This takes a while and blocks startup,
// do it in the background.
srv.loopWG.Add(1)
go func() {
defer srv.loopWG.Done()
if ip, err := srv.NAT.ExternalIP(); err == nil {
srv.localnode.SetStaticIP(ip)
}
}()
}
return nil
}
func (srv *Server) setupDiscovery() error {
if srv.NoDiscovery && !srv.DiscoveryV5 {
return nil
}
// node table
addr, err := net.ResolveUDPAddr("udp", srv.ListenAddr)
if err != nil {
return err
}
conn, err := net.ListenUDP("udp", addr)
if err != nil {
return err
}
realaddr := conn.LocalAddr().(*net.UDPAddr)
srv.log.Debug("UDP listener up", "addr", realaddr)
if srv.NAT != nil {
if !realaddr.IP.IsLoopback() {
go nat.Map(srv.NAT, srv.quit, "udp", realaddr.Port, realaddr.Port, "ethereum discovery")
}
}
srv.localnode.SetFallbackUDP(realaddr.Port)
// Discovery V4
var unhandled chan discover.ReadPacket
var sconn *sharedUDPConn
if !srv.NoDiscovery {
cfg := discover.Config{
PrivateKey: srv.PrivateKey,
AnnounceAddr: realaddr,
NodeDBPath: srv.NodeDatabase,
NetRestrict: srv.NetRestrict,
Bootnodes: srv.BootstrapNodes,
Unhandled: unhandled,
if srv.DiscoveryV5 {
unhandled = make(chan discover.ReadPacket, 100)
sconn = &sharedUDPConn{conn, unhandled}
}
ntab, err := discover.ListenUDP(conn, cfg)
cfg := discover.Config{
PrivateKey: srv.PrivateKey,
NetRestrict: srv.NetRestrict,
Bootnodes: srv.BootstrapNodes,
Unhandled: unhandled,
}
ntab, err := discover.ListenUDP(conn, srv.localnode, cfg)
if err != nil {
return err
}
srv.ntab = ntab
}
// Discovery V5
if srv.DiscoveryV5 {
var (
ntab *discv5.Network
err error
)
var ntab *discv5.Network
var err error
if sconn != nil {
ntab, err = discv5.ListenUDP(srv.PrivateKey, sconn, realaddr, "", srv.NetRestrict) //srv.NodeDatabase)
ntab, err = discv5.ListenUDP(srv.PrivateKey, sconn, "", srv.NetRestrict)
} else {
ntab, err = discv5.ListenUDP(srv.PrivateKey, conn, realaddr, "", srv.NetRestrict) //srv.NodeDatabase)
ntab, err = discv5.ListenUDP(srv.PrivateKey, conn, "", srv.NetRestrict)
}
if err != nil {
return err
@ -534,32 +560,10 @@ func (srv *Server) Start() (err error) {
}
srv.DiscV5 = ntab
}
dynPeers := srv.maxDialedConns()
dialer := newDialState(srv.StaticNodes, srv.BootstrapNodes, srv.ntab, dynPeers, srv.NetRestrict)
// handshake
pubkey := crypto.FromECDSAPub(&srv.PrivateKey.PublicKey)
srv.ourHandshake = &protoHandshake{Version: baseProtocolVersion, Name: srv.Name, ID: pubkey[1:]}
for _, p := range srv.Protocols {
srv.ourHandshake.Caps = append(srv.ourHandshake.Caps, p.cap())
}
// listen/dial
if srv.ListenAddr != "" {
if err := srv.startListening(); err != nil {
return err
}
}
if srv.NoDial && srv.ListenAddr == "" {
srv.log.Warn("P2P server will be useless, neither dialing nor listening")
}
srv.loopWG.Add(1)
go srv.run(dialer)
return nil
}
func (srv *Server) startListening() error {
func (srv *Server) setupListening() error {
// Launch the TCP listener.
listener, err := net.Listen("tcp", srv.ListenAddr)
if err != nil {
@ -568,8 +572,11 @@ func (srv *Server) startListening() error {
laddr := listener.Addr().(*net.TCPAddr)
srv.ListenAddr = laddr.String()
srv.listener = listener
srv.localnode.SetStaticTCP(laddr.Port)
srv.loopWG.Add(1)
go srv.listenLoop()
// Map the TCP listening port if NAT is configured.
if !laddr.IP.IsLoopback() && srv.NAT != nil {
srv.loopWG.Add(1)
@ -589,6 +596,8 @@ type dialer interface {
}
func (srv *Server) run(dialstate dialer) {
srv.log.Info("Started P2P networking", "self", srv.localnode.Node())
defer srv.loopWG.Done()
var (
peers = make(map[enode.ID]*Peer)
@ -781,7 +790,7 @@ func (srv *Server) encHandshakeChecks(peers map[enode.ID]*Peer, inboundCount int
return DiscTooManyPeers
case peers[c.node.ID()] != nil:
return DiscAlreadyConnected
case c.node.ID() == srv.Self().ID():
case c.node.ID() == srv.localnode.ID():
return DiscSelf
default:
return nil
@ -802,15 +811,11 @@ func (srv *Server) maxDialedConns() int {
return srv.MaxPeers / r
}
type tempError interface {
Temporary() bool
}
// listenLoop runs in its own goroutine and accepts
// inbound connections.
func (srv *Server) listenLoop() {
defer srv.loopWG.Done()
srv.log.Info("RLPx listener up", "self", srv.Self())
srv.log.Debug("TCP listener up", "addr", srv.listener.Addr())
tokens := defaultMaxPendingPeers
if srv.MaxPendingPeers > 0 {
@ -831,7 +836,7 @@ func (srv *Server) listenLoop() {
)
for {
fd, err = srv.listener.Accept()
if tempErr, ok := err.(tempError); ok && tempErr.Temporary() {
if netutil.IsTemporaryError(err) {
srv.log.Debug("Temporary read error", "err", err)
continue
} else if err != nil {
@ -864,10 +869,6 @@ func (srv *Server) listenLoop() {
// as a peer. It returns when the connection has been added as a peer
// or the handshakes have failed.
func (srv *Server) SetupConn(fd net.Conn, flags connFlag, dialDest *enode.Node) error {
self := srv.Self()
if self == nil {
return errors.New("shutdown")
}
c := &conn{fd: fd, transport: srv.newTransport(fd), flags: flags, cont: make(chan error)}
err := srv.setupConn(c, flags, dialDest)
if err != nil {
@ -1003,6 +1004,7 @@ type NodeInfo struct {
ID string `json:"id"` // Unique node identifier (also the encryption key)
Name string `json:"name"` // Name of the node, including client type, version, OS, custom data
Enode string `json:"enode"` // Enode URL for adding this peer from remote peers
ENR string `json:"enr"` // Ethereum Node Record
IP string `json:"ip"` // IP address of the node
Ports struct {
Discovery int `json:"discovery"` // UDP listening port for discovery protocol
@ -1014,9 +1016,8 @@ type NodeInfo struct {
// NodeInfo gathers and returns a collection of metadata known about the host.
func (srv *Server) NodeInfo() *NodeInfo {
node := srv.Self()
// Gather and assemble the generic node infos
node := srv.Self()
info := &NodeInfo{
Name: srv.Name,
Enode: node.String(),
@ -1027,6 +1028,9 @@ func (srv *Server) NodeInfo() *NodeInfo {
}
info.Ports.Discovery = node.UDP()
info.Ports.Listener = node.TCP()
if enc, err := rlp.EncodeToBytes(node.Record()); err == nil {
info.ENR = "0x" + hex.EncodeToString(enc)
}
// Gather all the running protocol infos (only once per protocol type)
for _, proto := range srv.Protocols {

26
p2p/server_test.go
View file

@ -225,12 +225,15 @@ func TestServerTaskScheduling(t *testing.T) {
// The Server in this test isn't actually running
// because we're only interested in what run does.
db, _ := enode.OpenDB("")
srv := &Server{
Config: Config{MaxPeers: 10},
quit: make(chan struct{}),
ntab: fakeTable{},
running: true,
log: log.New(),
Config: Config{MaxPeers: 10},
localnode: enode.NewLocalNode(db, newkey()),
nodedb: db,
quit: make(chan struct{}),
ntab: fakeTable{},
running: true,
log: log.New(),
}
srv.loopWG.Add(1)
go func() {
@ -271,11 +274,14 @@ func TestServerManyTasks(t *testing.T) {
}
var (
srv = &Server{
quit: make(chan struct{}),
ntab: fakeTable{},
running: true,
log: log.New(),
db, _ = enode.OpenDB("")
srv = &Server{
quit: make(chan struct{}),
localnode: enode.NewLocalNode(db, newkey()),
nodedb: db,
ntab: fakeTable{},
running: true,
log: log.New(),
}
done = make(chan *testTask)
start, end = 0, 0