diff --git a/p2p/doc.go b/p2p/doc.go
new file mode 100644
index 0000000000..a89a2d3122
--- /dev/null
+++ b/p2p/doc.go
@@ -0,0 +1,61 @@
+// Copyright 2015 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+/*
+Package p2p implements the devp2p protocol suite.
+
+The devp2p suite is a framework for the definition of RLP-based
+peer-to-peer protocols.
+
+From Nodes to Peers and Protocols
+
+Devp2p distinguishes nodes and peers. A node is any devp2p-capable
+host participating in the network. Peers are nodes to which a
+connection has been established. Nodes are identified by their Node
+ID, a secp256k1 public key.
+
+On any particular connection, one or more protocols are spoken. The
+protocols understood by the local node are declared when creating the
+local Server. When the connection is established, the sets of
+available protocols are matched against each other and the Run
+function of each protocol present on both sides is launched.
+
+RLPx
+
+Connections between peers use the RLPx wire protocol, which provides
+an encrypted and authenticated communication channel over TCP. RLPx
+supports concurrent transfer of protocol messages, ensuring that all
+protocols get an equal share of the available bandwidth.
+
+Connection Handling
+
+Package p2p establishes peer connections automatically by selecting
+randomly from the pool of all existing nodes in the network. The
+connectivity graph approaches an unstructured network with low
+diameter. If a protocol requires stronger connectivity properties
+(e.g. when building a structured overlay), the protocol implementation
+should provide information about preferred nodes on its Prefer channel.
+
+In order to accomodate new nodes joining the network, non-preferred
+peer connections may be terminated under certain conditions. Protocol
+implementations are free to terminate connections at any time by
+simply returning from the Run function.
+
+Users can configure static connectivity targets through the AddPeer
+method of Server. It will attempt to keep such nodes connected at all
+times.
+*/
+package p2p
diff --git a/p2p/dial.go b/p2p/internal/dialsched.go
similarity index 53%
rename from p2p/dial.go
rename to p2p/internal/dialsched.go
index 0fd3a4cf52..42be4b5572 100644
--- a/p2p/dial.go
+++ b/p2p/internal/dialsched.go
@@ -14,20 +14,49 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see .
-package p2p
+package p2pint
import (
"container/heap"
- "crypto/rand"
"fmt"
- "net"
"time"
- "github.com/ethereum/go-ethereum/logger"
- "github.com/ethereum/go-ethereum/logger/glog"
"github.com/ethereum/go-ethereum/p2p/discover"
)
+/*
+
+Dial Candidate Selection Algorithm
+
+This algorithm is responsible for picking dynamic dial candidates from
+various sources.
+
+- Random nodes from the the discovery table. Nodes in the discovery
+ table are usually long-lived, and querying the local table is fast.
+- Results from random discovery lookups.
+- Future: Results from discovery topic queries. Protocols may declare
+ topic hashes that they wish to find peers for.
+- Suggestions drawn from the (optional) PeerSuggestions channel for
+ each protocol. While protocol suggestions take precedence over other
+ sources, random connections are still required for bootstrapping and
+ attack resistance.
+
+The algorithm randomises output from available sources. There
+are more potential candidates than available connection slots. Not all
+sources can provide equally many candidates. Dial candidates should be
+a uniform selection out of all potential candidates.
+
+Avoid 'close nodes' bias. The XOR distance metric used by p2p/discover
+provides a useful network structure for the DHT, but can lead to less
+than ideal TCP connectivity if used naively. Since the discovery table
+prefers close nodes, the candidate selection algorithm must balance
+distances to overcome this bias.
+
+Honour per-protocol connection limits. If enough nodes are connected
+to satisfy the maximum number of peers for a given protocol, no new
+nodes need to be found.
+*/
+
const (
// This is the amount of time spent waiting in between
// redialing a certain node.
@@ -38,29 +67,23 @@ const (
lookupInterval = 4 * time.Second
)
-// dialstate schedules dials and discovery lookups.
-// it get's a chance to compute new tasks on every iteration
+// DialState schedules dials, discovery lookups and collects
+// peer sugggestions from protocols.
+// It gets a chance to compute new tasks on every iteration
// of the main loop in Server.run.
-type dialstate struct {
- maxDynDials int
- ntab discoverTable
+type DialState struct {
+ peers *PeerSet
+ ntab DiscoverTable
+ maxDynDials int // per protocol
lookupRunning bool
bootstrapped bool
+ lookupBuf []*discover.Node // current discovery lookup results
+ randomNodes []*discover.Node // filled from Table
- dialing map[discover.NodeID]connFlag
- lookupBuf []*discover.Node // current discovery lookup results
- randomNodes []*discover.Node // filled from Table
- static map[discover.NodeID]*discover.Node
- hist *dialHistory
-}
-
-type discoverTable interface {
- Self() *discover.Node
- Close()
- Bootstrap([]*discover.Node)
- Lookup(target discover.NodeID) []*discover.Node
- ReadRandomNodes([]*discover.Node) int
+ dialing map[discover.NodeID]Flag
+ static map[discover.NodeID]*discover.Node
+ hist *dialHistory
}
// the dial history remembers recent dials.
@@ -72,83 +95,77 @@ type pastDial struct {
exp time.Time
}
-type task interface {
- Do(*Server)
+type DiscoverTable interface {
+ Self() *discover.Node
+ Close()
+ Bootstrap([]*discover.Node)
+ Lookup(target discover.NodeID) []*discover.Node
+ ReadRandomNodes([]*discover.Node) int
}
-// A dialTask is generated for each node that is dialed.
-type dialTask struct {
- flags connFlag
- dest *discover.Node
+// A DialTask is generated for each node that is dialed.
+type DialTask struct {
+ Flags Flag
+ Dest *discover.Node
}
-// discoverTask runs discovery table operations.
-// Only one discoverTask is active at any time.
+// DiscoverTask runs discovery table operations.
+// Only one DiscoverTask is active at any time.
//
// If bootstrap is true, the task runs Table.Bootstrap,
// otherwise it performs a random lookup and leaves the
// results in the task.
-type discoverTask struct {
- bootstrap bool
- results []*discover.Node
+type DiscoverTask struct {
+ Bootstrap bool
+ Results []*discover.Node
}
-// A waitExpireTask is generated if there are no other tasks
-// to keep the loop in Server.run ticking.
-type waitExpireTask struct {
+// A WaitExpireTask is generated if there are no other tasks
+// running and the dial history is non-empty. This ensures that
+// the loop executing the tasks keeps ticking.
+type WaitExpireTask struct {
time.Duration
}
-func newDialState(static []*discover.Node, ntab discoverTable, maxdyn int) *dialstate {
- s := &dialstate{
+func NewDialState(ps *PeerSet, ntab DiscoverTable, maxdyn int) *DialState {
+ s := &DialState{
+ peers: ps,
maxDynDials: maxdyn,
ntab: ntab,
static: make(map[discover.NodeID]*discover.Node),
- dialing: make(map[discover.NodeID]connFlag),
+ dialing: make(map[discover.NodeID]Flag),
randomNodes: make([]*discover.Node, maxdyn/2),
hist: new(dialHistory),
}
- for _, n := range static {
- s.static[n.ID] = n
- }
return s
}
-func (s *dialstate) addStatic(n *discover.Node) {
+func (s *DialState) AddStatic(n *discover.Node) {
s.static[n.ID] = n
}
-func (s *dialstate) newTasks(nRunning int, peers map[discover.NodeID]*Peer, now time.Time) []task {
- var newtasks []task
- addDial := func(flag connFlag, n *discover.Node) bool {
+// newTasks is called from the main loop to new tasks.
+func (s *DialState) NewTasks(nRunning int, now time.Time) []interface{} {
+ var newtasks []interface{}
+ addDial := func(flag Flag, n *discover.Node) bool {
_, dialing := s.dialing[n.ID]
- if dialing || peers[n.ID] != nil || s.hist.contains(n.ID) {
+ if dialing || s.peers.IsConnected(n.ID) || s.hist.contains(n.ID) {
return false
}
s.dialing[n.ID] = flag
- newtasks = append(newtasks, &dialTask{flags: flag, dest: n})
+ newtasks = append(newtasks, &DialTask{Flags: flag, Dest: n})
return true
}
// Compute number of dynamic dials necessary at this point.
- needDynDials := s.maxDynDials
- for _, p := range peers {
- if p.rw.is(dynDialedConn) {
- needDynDials--
- }
- }
- for _, flag := range s.dialing {
- if flag&dynDialedConn != 0 {
- needDynDials--
- }
- }
+ needDynDials := s.needDynDials()
// Expire the dial history on every invocation.
s.hist.expire(now)
// Create dials for static nodes if they are not connected.
for _, n := range s.static {
- addDial(staticDialedConn, n)
+ addDial(StaticDialedConn, n)
}
// Use random nodes from the table for half of the necessary
@@ -157,7 +174,7 @@ func (s *dialstate) newTasks(nRunning int, peers map[discover.NodeID]*Peer, now
if randomCandidates > 0 && s.bootstrapped {
n := s.ntab.ReadRandomNodes(s.randomNodes)
for i := 0; i < randomCandidates && i < n; i++ {
- if addDial(dynDialedConn, s.randomNodes[i]) {
+ if addDial(DynDialedConn, s.randomNodes[i]) {
needDynDials--
}
}
@@ -166,7 +183,7 @@ func (s *dialstate) newTasks(nRunning int, peers map[discover.NodeID]*Peer, now
// items from the result buffer.
i := 0
for ; i < len(s.lookupBuf) && needDynDials > 0; i++ {
- if addDial(dynDialedConn, s.lookupBuf[i]) {
+ if addDial(DynDialedConn, s.lookupBuf[i]) {
needDynDials--
}
}
@@ -176,7 +193,7 @@ func (s *dialstate) newTasks(nRunning int, peers map[discover.NodeID]*Peer, now
// results.
if len(s.lookupBuf) < needDynDials && !s.lookupRunning {
s.lookupRunning = true
- newtasks = append(newtasks, &discoverTask{bootstrap: !s.bootstrapped})
+ newtasks = append(newtasks, &DiscoverTask{Bootstrap: !s.bootstrapped})
}
// Launch a timer to wait for the next node to expire if all
@@ -184,61 +201,50 @@ func (s *dialstate) newTasks(nRunning int, peers map[discover.NodeID]*Peer, now
// This should prevent cases where the dialer logic is not ticked
// because there are no pending events.
if nRunning == 0 && len(newtasks) == 0 && s.hist.Len() > 0 {
- t := &waitExpireTask{s.hist.min().exp.Sub(now)}
+ t := &WaitExpireTask{s.hist.min().exp.Sub(now)}
newtasks = append(newtasks, t)
}
return newtasks
}
-func (s *dialstate) taskDone(t task, now time.Time) {
+// TaskDone should be called when a task has finished.
+func (s *DialState) taskDone(t interface{}, now time.Time) {
switch t := t.(type) {
- case *dialTask:
- s.hist.add(t.dest.ID, now.Add(dialHistoryExpiration))
- delete(s.dialing, t.dest.ID)
- case *discoverTask:
- if t.bootstrap {
+ case *DialTask:
+ s.hist.add(t.Dest.ID, now.Add(dialHistoryExpiration))
+ delete(s.dialing, t.Dest.ID)
+ case *DiscoverTask:
+ if t.Bootstrap {
s.bootstrapped = true
}
s.lookupRunning = false
- s.lookupBuf = append(s.lookupBuf, t.results...)
+ s.lookupBuf = append(s.lookupBuf, t.Results...)
+ case *WaitExpireTask:
+ // nothing to do here
+ default:
+ panic("unknown task type")
}
}
-func (t *dialTask) Do(srv *Server) {
- addr := &net.TCPAddr{IP: t.dest.IP, Port: int(t.dest.TCP)}
- glog.V(logger.Debug).Infof("dialing %v\n", t.dest)
- fd, err := srv.Dialer.Dial("tcp", addr.String())
- if err != nil {
- glog.V(logger.Detail).Infof("dial error: %v", err)
- return
+// computes number of required dynamic dials
+func (s *DialState) needDynDials() int {
+ need := s.maxDynDials
+ for _, p := range s.peers.NumDynPeers() {
+ need--
}
- mfd := newMeteredConn(fd, false)
-
- srv.setupConn(mfd, t.flags, t.dest)
-}
-func (t *dialTask) String() string {
- return fmt.Sprintf("%v %x %v:%d", t.flags, t.dest.ID[:8], t.dest.IP, t.dest.TCP)
+ for _, flag := range s.dialing {
+ if flag&DynDialedConn != 0 {
+v need--
+ }
+ }
+ return need
}
-func (t *discoverTask) Do(srv *Server) {
- if t.bootstrap {
- srv.ntab.Bootstrap(srv.BootstrapNodes)
- return
- }
- // 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()
- var target discover.NodeID
- rand.Read(target[:])
- t.results = srv.ntab.Lookup(target)
+func (t *DialTask) String() string {
+ return fmt.Sprintf("%v %x %v:%d", t.Flags, t.Dest.ID[:8], t.Dest.IP, t.Dest.TCP)
}
-func (t *discoverTask) String() (s string) {
+func (t *DiscoverTask) String() (s string) {
if t.bootstrap {
s = "discovery bootstrap"
} else {
@@ -250,10 +256,7 @@ func (t *discoverTask) String() (s string) {
return s
}
-func (t waitExpireTask) Do(*Server) {
- time.Sleep(t.Duration)
-}
-func (t waitExpireTask) String() string {
+func (t *WaitExpireTask) String() string {
return fmt.Sprintf("wait for dial hist expire (%v)", t.Duration)
}
diff --git a/p2p/internal/peerset.go b/p2p/internal/peerset.go
new file mode 100644
index 0000000000..090017e6a7
--- /dev/null
+++ b/p2p/internal/peerset.go
@@ -0,0 +1,117 @@
+// Copyright 2015 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package p2pint
+
+import "github.com/ethereum/go-ethereum/p2p/discover"
+
+type Flag int
+
+const (
+ DynDialedConn Flag = 1 << iota
+ StaticDialedConn
+ InboundConn
+ TrustedConn
+)
+
+// PeerSet tracks connected peers and various indices
+// associcated with them.
+type PeerSet struct {
+ limit int
+ counts map[string]int
+ all map[discover.NodeID]Peer
+ preferred map[discover.NodeID][]string
+ static map[discover.NodeID]struct{}
+}
+
+type Peer interface {
+ ID() discover.NodeID
+ ActiveProtocols() []string
+ Flag() Flag
+}
+
+func NewPeerSet(limit int) *PeerSet {
+ return &PeerSet{
+ limit: limit,
+ counts: make(map[string]int),
+ all: make(map[discover.NodeID]psPeer),
+ preferred: make(map[discover.NodeID][]string),
+ static: make(map[discover.NodeID]struct{}),
+ }
+}
+
+func (ps *PeerSet) Add(p psPeer) {
+ for _, protoName := range p.activeProtocols() {
+ ps.counts[protoName]++
+ }
+ ps.all[p.ID()] = p
+}
+
+func (ps *PeerSet) Remove(p psPeer) {
+ for _, protoName := range p.activeProtocols() {
+ ps.setIdle(p, protoName)
+ ps.counts[protoName]--
+ }
+ delete(ps.all, p.ID())
+}
+
+func (ps *PeerSet) NumPeers() int {
+ return len(ps.all)
+}
+
+func (ps *PeerSet) IsAtCapacity() bool {
+ for _, numProtoPeers := range ps.counts {
+ if numProtoPeers < ps.limit {
+ return false
+ }
+ }
+ return true
+}
+
+func (ps *PeerSet) IsConnected(id discover.NodeID) bool {
+ _, ok := ps.all[id]
+ return ok
+}
+
+// disables auto-disconnect for a particular peer
+func (ps *PeerSet) SetPreferred(p psPeer, protoName string) {
+ pl := ps.preferred[p.ID()]
+ for _, name := range pl {
+ if name == protoName {
+ return
+ }
+ }
+ ps.preferred[p.ID()] = append(pl, protoName)
+}
+
+// opposite of setPreferred, removes from ps.preferred
+func (ps *PeerSet) SetIdle(p psPeer, protoName string) {
+ pl := ps.preferred[p.ID()]
+ pos := -1
+ for i, name := range pl {
+ if name == protoName {
+ pos = i
+ break
+ }
+ }
+ if pos >= 0 {
+ if len(pl) == 1 {
+ delete(ps.preferred, p.ID())
+ } else {
+ ps.preferred[p.ID()] = append(pl[:pos], pl[pos:]...)
+ }
+ }
+}
diff --git a/p2p/protocol.go b/p2p/protocol.go
index ee747ba23d..9db34c3cc6 100644
--- a/p2p/protocol.go
+++ b/p2p/protocol.go
@@ -35,22 +35,30 @@ type Protocol struct {
// by the protocol.
Length uint64
- // Run is called in a new groutine when the protocol has been
+ // Run is called in a new goroutine when the protocol has been
// negotiated with a peer. It should read and write messages from
// rw. The Payload for each message must be fully consumed.
//
- // The peer connection is closed when Start returns. It should return
+ // The peer connection is closed when Run returns. It should return
// any protocol-level error (such as an I/O error) that is
// encountered.
Run func(peer *Peer, rw MsgReadWriter) error
- // NodeInfo is an optional helper method to retrieve protocol specific metadata
- // about the host node.
+ // Prefer, if non-nil, should receive nodes that the protocol
+ // implementation wants to be connected to. If a preferred node is
+ // not connected yet, package p2p will attempt to establish a
+ // connection. A successful send on this channel does not
+ // guarantee that a connection will actually be established.
+ Prefer <-chan *discover.Node
+
+ // NodeInfo is an optional helper method to retrieve protocol
+ // specific metadata about the host node.
NodeInfo func() interface{}
- // PeerInfo is an optional helper method to retrieve protocol specific metadata
- // 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 is an optional helper method to retrieve protocol
+ // specific metadata 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 discover.NodeID) interface{}
}
@@ -64,10 +72,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)
}
diff --git a/p2p/server.go b/p2p/server.go
index 7991585f17..501cc00f6c 100644
--- a/p2p/server.go
+++ b/p2p/server.go
@@ -14,11 +14,11 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see .
-// Package p2p implements the Ethereum p2p network protocols.
package p2p
import (
"crypto/ecdsa"
+ "crypto/rand"
"errors"
"fmt"
"net"
@@ -28,6 +28,7 @@ import (
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
"github.com/ethereum/go-ethereum/p2p/discover"
+ "github.com/ethereum/go-ethereum/p2p/internal"
"github.com/ethereum/go-ethereum/p2p/nat"
)
@@ -148,21 +149,12 @@ type Server struct {
type peerOpFunc func(map[discover.NodeID]*Peer)
-type connFlag int
-
-const (
- dynDialedConn connFlag = 1 << iota
- staticDialedConn
- inboundConn
- trustedConn
-)
-
// conn wraps a network connection with information gathered
// during the two handshakes.
type conn struct {
fd net.Conn
transport
- flags connFlag
+ flags p2pint.Flag
cont chan error // The run loop uses cont to signal errors to setupConn.
id discover.NodeID // valid after the encryption handshake
caps []Cap // valid after the protocol handshake
@@ -429,92 +421,36 @@ func (srv *Server) run(dialstate dialer) {
for _, t := range pt[:start] {
t := t
glog.V(logger.Detail).Infoln("new task:", t)
- go func() { t.Do(srv); taskdone <- t }()
- }
- copy(pt, pt[start:])
- pendingTasks = pt[:len(pt)-start]
+ go func() { srv.runDialTask(t); taskdone <- t }()
+ case *p2pint.DialTask:
+ addr := &net.TCPAddr{IP: t.dest.IP, Port: int(t.dest.TCP)}
+ glog.V(logger.Debug).Infof("dialing %v\n", t.dest)
+ fd, err := srv.Dialer.Dial("tcp", addr.String())
+ if err != nil {
+ glog.V(logger.Detail).Infof("dial error: %v", err)
+ return
}
- }
+ mfd := newMeteredConn(fd, false)
+ srv.setupConn(mfd, t.flags, t.dest)
-running:
- for {
- // Query the dialer for new tasks and launch them.
- now := time.Now()
- nt := dialstate.newTasks(len(pendingTasks)+len(tasks), peers, now)
- scheduleTasks(nt)
-
- select {
- case <-srv.quit:
- // The server was stopped. Run the cleanup logic.
- glog.V(logger.Detail).Infoln("<-quit: spinning down")
- break running
- case n := <-srv.addstatic:
- // This channel is used by AddPeer to add to the
- // ephemeral static peer list. Add it to the dialer,
- // it will keep the node connected.
- glog.V(logger.Detail).Infoln("<-addstatic:", n)
- dialstate.addStatic(n)
- case op := <-srv.peerOp:
- // This channel is used by Peers and PeerCount.
- op(peers)
- srv.peerOpDone <- struct{}{}
- case t := <-taskdone:
- // A task got done. Tell dialstate about it so it
- // can update its state and remove it from the active
- // tasks list.
- glog.V(logger.Detail).Infoln("<-taskdone:", t)
- dialstate.taskDone(t, now)
- delTask(t)
- case c := <-srv.posthandshake:
- // A connection has passed the encryption handshake so
- // the remote identity is known (but hasn't been verified yet).
- if trusted[c.id] {
- // Ensure that the trusted flag is set before checking against MaxPeers.
- c.flags |= trustedConn
- }
- glog.V(logger.Detail).Infoln("<-posthandshake:", c)
- // TODO: track in-progress inbound node IDs (pre-Peer) to avoid dialing them.
- c.cont <- srv.encHandshakeChecks(peers, c)
- case c := <-srv.addpeer:
- // At this point the connection is past the protocol handshake.
- // Its capabilities are known and the remote identity is verified.
- glog.V(logger.Detail).Infoln("<-addpeer:", c)
- err := srv.protoHandshakeChecks(peers, c)
- if err != nil {
- glog.V(logger.Detail).Infof("Not adding %v as peer: %v", c, err)
- } else {
- // The handshakes are done and it passed all checks.
- p := newPeer(c, srv.Protocols)
- peers[c.id] = p
- go srv.runPeer(p)
- }
- // The dialer logic relies on the assumption that
- // dial tasks complete after the peer has been added or
- // discarded. Unblock the task last.
- c.cont <- err
- case p := <-srv.delpeer:
- // A peer disconnected.
- glog.V(logger.Detail).Infoln("<-delpeer:", p)
- delete(peers, p.ID())
+ case *p2pint.DiscoverTask:
+ // 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()
+ var target discover.NodeID
+ rand.Read(target[:])
+ t.results = srv.ntab.Lookup(target)
- // Terminate discovery. If there is a running lookup it will terminate soon.
- if srv.ntab != nil {
- srv.ntab.Close()
- }
- // Disconnect all peers.
- for _, p := range peers {
- p.Disconnect(DiscQuitting)
- }
- // Wait for peers to shut down. Pending connections and tasks are
- // not handled here and will terminate soon-ish because srv.quit
- // is closed.
- glog.V(logger.Detail).Infof("ignoring %d pending tasks at spindown", len(tasks))
- for len(peers) > 0 {
- p := <-srv.delpeer
- glog.V(logger.Detail).Infoln("<-delpeer (spindown):", p)
- delete(peers, p.ID())
+ case *p2pint.WaitExpireTask:
+ time.Sleep(t.Duration)
+
+ default:
+ panic("unknown task type")
}
}