diff --git a/p2p/discover/common.go b/p2p/discover/common.go index c9f0477def..c36e8dcc3a 100644 --- a/p2p/discover/common.go +++ b/p2p/discover/common.go @@ -19,7 +19,6 @@ package discover import ( "crypto/ecdsa" "net" - "time" "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/log" @@ -36,39 +35,29 @@ type UDPConn interface { LocalAddr() net.Addr } +type V5Config struct { + ProtocolID *[6]byte +} + // Config holds settings for the discovery listener. type Config struct { // These settings are required and configure the UDP listener: PrivateKey *ecdsa.PrivateKey - // All remaining settings are optional. - - // Packet handling configuration: + // These settings are optional: NetRestrict *netutil.Netlist // list of allowed IP networks + Bootnodes []*enode.Node // list of bootstrap nodes Unhandled chan<- ReadPacket // unhandled packets are sent on this channel + Log log.Logger // if set, log messages go here - // Node table configuration: - Bootnodes []*enode.Node // list of bootstrap nodes - PingInterval time.Duration // speed of node liveness check - RefreshInterval time.Duration // used in bucket refresh - - // The options below are useful in very specific cases, like in unit tests. + // V5ProtocolID configures the discv5 protocol identifier. V5ProtocolID *[6]byte - Log log.Logger // if set, log messages go here + ValidSchemes enr.IdentityScheme // allowed identity schemes Clock mclock.Clock } func (cfg Config) withDefaults() Config { - // Node table configuration: - if cfg.PingInterval == 0 { - cfg.PingInterval = 10 * time.Second - } - if cfg.RefreshInterval == 0 { - cfg.RefreshInterval = 30 * time.Minute - } - - // Debug/test settings: if cfg.Log == nil { cfg.Log = log.Root() } diff --git a/p2p/discover/table.go b/p2p/discover/table.go index 1397348aec..41d5ac6e34 100644 --- a/p2p/discover/table.go +++ b/p2p/discover/table.go @@ -53,10 +53,12 @@ const ( bucketIPLimit, bucketSubnet = 2, 24 // at most 2 addresses from the same /24 tableIPLimit, tableSubnet = 10, 24 - copyNodesInterval = 30 * time.Second - seedMinTableTime = 5 * time.Minute - seedCount = 30 - seedMaxAge = 5 * 24 * time.Hour + refreshInterval = 30 * time.Minute + revalidateInterval = 10 * time.Second + copyNodesInterval = 30 * time.Second + seedMinTableTime = 5 * time.Minute + seedCount = 30 + seedMaxAge = 5 * 24 * time.Hour ) // Table is the 'node table', a Kademlia-like index of neighbor nodes. The table keeps @@ -69,12 +71,9 @@ type Table struct { rand *mrand.Rand // source of randomness, periodically reseeded ips netutil.DistinctNetSet - db *enode.DB // database of known nodes - net transport - cfg Config - log log.Logger - - // loop channels + log log.Logger + db *enode.DB // database of known nodes + net transport refreshReq chan chan struct{} initDone chan struct{} closeReq chan struct{} @@ -100,21 +99,19 @@ type bucket struct { ips netutil.DistinctNetSet } -func newTable(t transport, db *enode.DB, cfg Config) (*Table, error) { - cfg = cfg.withDefaults() +func newTable(t transport, db *enode.DB, bootnodes []*enode.Node, log log.Logger) (*Table, error) { tab := &Table{ net: t, db: db, - cfg: cfg, - log: cfg.Log, refreshReq: make(chan chan struct{}), initDone: make(chan struct{}), closeReq: make(chan struct{}), closed: make(chan struct{}), rand: mrand.New(mrand.NewSource(0)), ips: netutil.DistinctNetSet{Subnet: tableSubnet, Limit: tableIPLimit}, + log: log, } - if err := tab.setFallbackNodes(cfg.Bootnodes); err != nil { + if err := tab.setFallbackNodes(bootnodes); err != nil { return nil, err } for i := range tab.buckets { @@ -128,24 +125,6 @@ func newTable(t transport, db *enode.DB, cfg Config) (*Table, error) { return tab, nil } -// Nodes returns all nodes contained in the table. -func (tab *Table) Nodes() []*enode.Node { - if !tab.isInitDone() { - return nil - } - - tab.mutex.Lock() - defer tab.mutex.Unlock() - - var nodes []*enode.Node - for _, b := range &tab.buckets { - for _, n := range b.entries { - nodes = append(nodes, unwrapNode(n)) - } - } - return nodes -} - func (tab *Table) self() *enode.Node { return tab.net.Self() } @@ -159,6 +138,29 @@ func (tab *Table) seedRand() { tab.mutex.Unlock() } +// 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) { + if !tab.isInitDone() { + return 0 + } + tab.mutex.Lock() + defer tab.mutex.Unlock() + + var nodes []*enode.Node + for _, b := range &tab.buckets { + for _, n := range b.entries { + nodes = append(nodes, unwrapNode(n)) + } + } + // Shuffle. + for i := 0; i < len(nodes); i++ { + j := tab.rand.Intn(len(nodes)) + nodes[i], nodes[j] = nodes[j], nodes[i] + } + return copy(buf, nodes) +} + // getNode returns the node with the given ID or nil if it isn't in the table. func (tab *Table) getNode(id enode.ID) *enode.Node { tab.mutex.Lock() @@ -216,7 +218,7 @@ func (tab *Table) refresh() <-chan struct{} { func (tab *Table) loop() { var ( revalidate = time.NewTimer(tab.nextRevalidateTime()) - refresh = time.NewTimer(tab.nextRefreshTime()) + refresh = time.NewTicker(refreshInterval) copyNodes = time.NewTicker(copyNodesInterval) refreshDone = make(chan struct{}) // where doRefresh reports completion revalidateDone chan struct{} // where doRevalidate reports completion @@ -249,7 +251,6 @@ loop: close(ch) } waiting, refreshDone = nil, nil - refresh.Reset(tab.nextRefreshTime()) case <-revalidate.C: revalidateDone = make(chan struct{}) go tab.doRevalidate(revalidateDone) @@ -372,15 +373,7 @@ func (tab *Table) nextRevalidateTime() time.Duration { tab.mutex.Lock() defer tab.mutex.Unlock() - return time.Duration(tab.rand.Int63n(int64(tab.cfg.PingInterval))) -} - -func (tab *Table) nextRefreshTime() time.Duration { - tab.mutex.Lock() - defer tab.mutex.Unlock() - - half := tab.cfg.RefreshInterval / 2 - return half + time.Duration(tab.rand.Int63n(int64(half))) + return time.Duration(tab.rand.Int63n(int64(revalidateInterval))) } // copyLiveNodes adds nodes from the table to the database if they have been in the table @@ -488,12 +481,10 @@ func (tab *Table) addSeenNode(n *node) { // Can't add: IP limit reached. return } - // Add to end of bucket: b.entries = append(b.entries, n) b.replacements = deleteNode(b.replacements, n) n.addedAt = time.Now() - if tab.nodeAddedHook != nil { tab.nodeAddedHook(n) } @@ -532,12 +523,10 @@ func (tab *Table) addVerifiedNode(n *node) { // Can't add: IP limit reached. return } - // Add to front of bucket. b.entries, _ = pushNode(b.entries, n, bucketSize) b.replacements = deleteNode(b.replacements, n) n.addedAt = time.Now() - if tab.nodeAddedHook != nil { tab.nodeAddedHook(n) } diff --git a/p2p/discover/table_test.go b/p2p/discover/table_test.go index 2781dd4225..1ef63fe010 100644 --- a/p2p/discover/table_test.go +++ b/p2p/discover/table_test.go @@ -247,6 +247,41 @@ func TestTable_findnodeByID(t *testing.T) { } } +func TestTable_ReadRandomNodesGetAll(t *testing.T) { + cfg := &quick.Config{ + MaxCount: 200, + Rand: rand.New(rand.NewSource(time.Now().Unix())), + Values: func(args []reflect.Value, rand *rand.Rand) { + args[0] = reflect.ValueOf(make([]*enode.Node, rand.Intn(1000))) + }, + } + test := func(buf []*enode.Node) bool { + transport := newPingRecorder() + tab, db := newTestTable(transport) + defer db.Close() + defer tab.close() + <-tab.initDone + + for i := 0; i < len(buf); i++ { + ld := cfg.Rand.Intn(len(tab.buckets)) + fillTable(tab, []*node{nodeAtDistance(tab.self().ID(), ld, intIP(ld))}) + } + gotN := tab.ReadRandomNodes(buf) + if gotN != tab.len() { + t.Errorf("wrong number of nodes, got %d, want %d", gotN, tab.len()) + return false + } + if hasDuplicates(wrapNodes(buf[:gotN])) { + t.Errorf("result contains duplicates") + return false + } + return true + } + if err := quick.Check(test, cfg); err != nil { + t.Error(err) + } +} + type closeTest struct { Self enode.ID Target enode.ID diff --git a/p2p/discover/table_util_test.go b/p2p/discover/table_util_test.go index 5b5e9a0431..527afbb77e 100644 --- a/p2p/discover/table_util_test.go +++ b/p2p/discover/table_util_test.go @@ -28,6 +28,7 @@ import ( "sync" "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/enr" ) @@ -41,9 +42,8 @@ func init() { } func newTestTable(t transport) (*Table, *enode.DB) { - cfg := Config{} db, _ := enode.OpenDB("") - tab, _ := newTable(t, db, cfg) + tab, _ := newTable(t, db, nil, log.Root()) go tab.loop() return tab, db } diff --git a/p2p/discover/v4_udp.go b/p2p/discover/v4_udp.go index d61a52c395..21fee9c793 100644 --- a/p2p/discover/v4_udp.go +++ b/p2p/discover/v4_udp.go @@ -142,7 +142,7 @@ func ListenV4(c UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv4, error) { log: cfg.Log, } - tab, err := newTable(t, ln.Database(), cfg) + tab, err := newTable(t, ln.Database(), cfg.Bootnodes, t.log) if err != nil { return nil, err } diff --git a/p2p/discover/v5_udp.go b/p2p/discover/v5_udp.go index 7bed9dbcfd..6a86e72a50 100644 --- a/p2p/discover/v5_udp.go +++ b/p2p/discover/v5_udp.go @@ -174,7 +174,7 @@ func newUDPv5(conn UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv5, error) { cancelCloseCtx: cancelCloseCtx, } t.talk = newTalkSystem(t) - tab, err := newTable(t, t.db, cfg) + tab, err := newTable(t, t.db, cfg.Bootnodes, cfg.Log) if err != nil { return nil, err }