In-depth understanding of Ethereum P2P network design from the code layer

foreword

foreword

This article will give a brief overview of the development of P2P networks. At the same time, it will briefly introduce the node discovery mechanism, distributed hash table, node lookup, node addition, and node removal in Ethereum from the perspective of source code. A brief analysis of P2P network security design.

basic knowledge

basic knowledge

P2P network

The P2P network is different from the traditional CS structure. In the P2P network, each node can be either a client or a server. The communication protocol between nodes is generally implemented directly through Socket.

The development of P2P technology has gone through the following four stages of development:

1. Centralized:

It is the simplest routing method in the P2P network mode, that is, there is a central node, which stores the index information of all other nodes. The index information generally includes node IP, port, node resources, etc. The advantage of centralized routing is that it is simple in structure and easy to implement. Easy, but the disadvantages are also obvious. Since the central node needs to store the routing information of all nodes, when the node scale continues to expand, performance bottlenecks are prone to occur, and there is also a single node failure problem

2. Distributed:

It means that the central node is removed, a random network is established between P2P nodes, and a connection channel is randomly established between a newly added node and a node in the P2P network, thereby forming a random topology structure. When a new node joins the network, it randomly Select an existing node and establish a neighbor relationship. After the node establishes a connection with the neighbor node, it needs to broadcast the entire network to let the entire network know the existence of the node."The specific broadcast steps are: the node first broadcasts to the neighbor nodes, and after the neighbor node receives the broadcast message, it continues to broadcast to its own neighbor nodes, and so on, this broadcast method is also called"flooding mechanism

, and the problem of the flooding mechanism is poor controllability, which mainly includes two major problems: one is that it is easy to form a flooding cycle, for example, node A sends a message to node B to node C, and node C broadcasts to node C A, this forms a cycle. Another problem is the subcontracting of message responses. For example, if the resource that node A wants to request is owned by many nodes, then in a short period of time, a large number of nodes will send response messages to node A at the same time. This is likely to cause node A to crash instantly

The method of eliminating flooding loop can learn from the control of flooding broadcast in IP network routing protocols. One method is to set a TTL value for each query message. Every time a flooding message is forwarded, the TTL value will be reduced by 1. When the node accepts When the TTL is 0, the message will not be forwarded, which can avoid the infinite loop of the query message in the network, and can also set a unique flag for the flood message, and will not forward the received duplicate message so as to avoid the infinite loop. The method of responding to storms can perform network segmentation at the data link layer to reduce cross-segment broadcasting of messages

3. Hybrid:

The hybrid type is actually a mixed centralized and distributed structure. There are many distributed networks composed of super nodes in the network, and each super node has multiple ordinary nodes to form a local centralized network with it. A new ordinary node can be added first. Select a super node to communicate, and the super node will push other super node lists to the new joining node. The joining node decides to choose the specific super node as the parent node according to the status of the super node in the list. The flood broadcast of this structure just happens Between super nodes, large-scale flooding can be avoided. In practical applications, the hybrid structure is a relatively flexible and effective networking architecture, and the implementation difficulty is relatively small. Therefore, many systems are currently based on the hybrid structure. Develop and implement

Four. Structuring:

A structured P2P network is a distributed network structure. Different from the distributed structure mentioned above, a distributed network is a random network, while a structured network organizes all nodes in an orderly manner according to a certain structure, such as forming a Ring network or tree network, structured network is generally based on distributed hash table (Distributed Hash Table, DHI) algorithm in specific implementation, specific implementation schemes include Chord, Pasty, CAN, Kademlia and other algorithms

The four network structures are compared as follows:

node discovery

Node discovery is the first step for any node to access the P2P network. Node discovery can be divided into two types:

- Initial node discovery:

Refers to the node is a brand new node that has never been running. The node does not have any data of other nodes in the network. At this time, the node finds that it can only rely on the hard-coded seed nodes in the node to obtain the information of the P2P network.

- Known node discovery:

The node has been running before, and the node database saves the information of other nodes in the network. At this time, the node finds that it can rely on the nodes summarized in the node database to obtain the information of the P2P network, so as to build its own network topology

seed node"In the P2P network, the initial node will quickly discover other nodes in the network through some long-term stable nodes when starting up. These nodes are called"seed node

DNS-Seed：

(In general, the seed node information will be hard-coded in the code). In general, the seed nodes can be divided into two types:"also known as"DNS seed node

IP-Seed：

, DNS is a domain name query service provided by the Internet. It maps domain names and IP addresses and stores them in a distributed database. When we access the DNS server, provide it with a domain name, and the DNS server will correspond to the domain name IP address returned

That is, the IP address of the seed node is hard-coded into the code, and the addresses of these hard-coded nodes are called seed nodes

KDA algorithm

HashTable

Kademlia is a distributed hash table (DHT) technology. Compared with other DHT technologies, the KDA algorithm uses the XOR algorithm to calculate the distance between nodes, and then establishes a new DHT topology. This algorithm can greatly improve The query speed of the route.

A hash table is a container for storing key-value pairs. Key-value pairs are called Key/Value pairs. The hash table data structure contains N buckets. For a specific hash table, N (the number of buckets) is usually fixed, so you can number each bucket, 0~N-1, and the bucket is used to store key-value pairs. It can be simply understood as a dynamic array, which stores Multiple key-value pairs.

HashCollision

The following figure shows the search principle of the hash table. We can easily and quickly obtain the value through the Key. When using a certain key to search, first use a certain hash function to calculate the hash value of the key, and the obtained hash The value is usually an integer, and then the hash value is used to perform a modulo operation on N (the number of buckets) (remainder of division), and the corresponding bucket number can be calculated

When it comes to hash tables, we have to mention hash table collisions. When two different Keys perform hash calculations to obtain the same hash value, it is the so-called hash function collision. Once this happens, the two keys The corresponding two key-value pairs will be stored in the same bucket (bucket). Another hash collision is that although the calculated hash values ​​are different, the same bucket number is obtained after the modulo operation. Sometimes, two key-value pairs will be stored in a bucket. The hash collision principle is shown in the figure below:

If a hash table has no collision at all when storing data, then there are only 0 or 1 key-value pairs in each bucket, which makes it very fast to look up. On the contrary, if a hash table appears when storing data Serious collisions will cause many key-value pairs to be stored in certain buckets. When searching for keys, it is necessary to compare whether the keys are the same in this bucket one by one, and the search efficiency will become very low~

distributed hash table

The distributed hash table is conceptually similar to the traditional hash table, the difference is that the traditional hash table is mainly used in a certain software on a single machine, and the distributed hash table is mainly used in a distributed system (at this time, distributed The nodes of the system can be commonly understood as buckets in the hash table). The distributed hash table is mainly used to store a large amount (even massive) of data. The principle of the distributed hash table is shown in the following figure:

Source code analysis

The underlying P2PServer of Ethereum can be roughly divided into the following three layers:

- top layer:

The specific implementation of each protocol in Ethereum

- middle layer:

The p2p communication link layer in Ethereum is responsible for starting monitoring, processing new connections or maintaining connections

- Bottom layer:

The IO layer of the data communication network in Ethereum is mainly responsible for the management of the routing table and the read and write operations of the database

table structure

// filedir:go-ethereum-1.10.2\p2p\discover\table.go L40
const (
   alpha           = 3  // Kademlia concurrency factor
   bucketSize      = 16 // Kademlia bucket size
   maxReplacements = 10 // Size of per-bucket replacement list
   // We keep buckets for the upper 1/15 of distances because
   // it's very unlikely we'll ever encounter a node that's closer.
   hashBits          = len(common.Hash{}) * 8
   nBuckets          = hashBits / 15       // Number of buckets
   bucketMinDistance = hashBits - nBuckets // Log distance of closest bucket
   // IP address limits.
   bucketIPLimit, bucketSubnet = 2, 24 // at most 2 addresses from the same /24
   tableIPLimit, tableSubnet   = 10, 24
   refreshInterval    = 30 * time.Minute
   revalidateInterval = 10 * time.Second
   copyNodesInterval  = 30 * time.Second
   seedMinTableTime   = 5 * time.Minute
   seedCount          = 30
   seedMaxAge         = 5 * 24 * time.Hour
)
type Table struct {
   mutex   sync.Mutex        // protects buckets, bucket content, nursery, rand
   buckets [nBuckets]*bucket // index of known nodes by distance
   nursery []*node           // bootstrap nodes
   rand    *mrand.Rand       // source of randomness, periodically reseeded
   ips     netutil.DistinctNetSet
   log        log.Logger
   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
}
type bucket struct {
   entries      []*node // live entries, sorted by time of last contact
   replacements []*node // recently seen nodes to be used if revalidation fails
   ips          netutil.DistinctNetSet
}

The table data structure looks like this:

• A few key variables:

• buckets: K buckets, each K bucket contains nodes (sorted in descending order according to recent activity), used to list known node indexes by distance

• Nursery: seed node, when a node is started, it can link up to 35 seed nodes, five of which are officially designated by Ethereum, and the other 30 are extracted from the database

• db: the database used to store P2P nodes (there are two in Ethereum, and the other is used to store data on the chain)

refreshReq: pipeline to refresh K-bucket events

table creation

// filedir:go-ethereum-1.10.2\p2p\discover\table.go L102
func newTable(t transport, db *enode.DB, bootnodes []*enode.Node, log log.Logger) (*Table, error) {
   tab := &Table{
       net:        t,
       db:         db,
       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(bootnodes); err != nil {
       return nil, err
   }
   for i := range tab.buckets {
       tab.buckets[i] = &bucket{
           ips: netutil.DistinctNetSet{Subnet: bucketSubnet, Limit: bucketIPLimit},
       }
   }
   tab.seedRand()
   tab.loadSeedNodes()
   return tab, nil
}

The newTable function is used to create a new table:

In the above code, a Table object tab is first initialized with the incoming parameters, and the setFallbackNodes function is called to set the initial link nodes (that is, to obtain 5 nurse nodes, which will be used later if the table is empty or there is no node information in the database. to link the network), and then use a for loop combined with the function ValidateComplete to verify whether the node is valid

   for i := range tab.buckets {
       tab.buckets[i] = &bucket{
           ips: netutil.DistinctNetSet{Subnet: bucketSubnet, Limit: bucketIPLimit},
       }
   }

Then initialize the K bucket:

   tab.seedRand()
   tab.loadSeedNodes()
   return tab, nil

Then randomly select 30 nodes from table.buckets and load the seed node to the corresponding bucket:

// filedir: go-ethereum-1.10.2\p2p\discover\table.go    L302
func (tab *Table) loadSeedNodes() {
   seeds := wrapNodes(tab.db.QuerySeeds(seedCount, seedMaxAge))
   seeds = append(seeds, tab.nursery...)
   for i := range seeds {
       seed := seeds[i]
       age := log.Lazy{Fn: func() interface{} { return time.Since(tab.db.LastPongReceived(seed.ID(), seed.IP())) }}
       tab.log.Trace("Found seed node in database", "id", seed.ID(), "addr", seed.addr(), "age", age)
       tab.addSeenNode(seed)
   }
}

The specific implementation of the loadSeedNodes function is as follows (the seedCount here is the global variable defined at the top of table.go, with a value of 30):

// filedir: go-ethereum-1.10.2\p2p\discover\table.go L458
func (tab *Table) addSeenNode(n *node) {
   if n.ID() == tab.self().ID() {
       return
   }
   tab.mutex.Lock()
   defer tab.mutex.Unlock()
   b := tab.bucket(n.ID())
   if contains(b.entries, n.ID()) {
       // Already in bucket, don't add.
       return
   }
   if len(b.entries) >= bucketSize {
       // Bucket full, maybe add as replacement.
       tab.addReplacement(b, n)
       return
   }
   if !tab.addIP(b, n.IP()) {
       // 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)
   }
}

The addSeenNode here is used to add nodes to the bucket. Here, it will check whether the node to be added already exists and whether the bucket is full. If it is full, call tab.addReplacement(b, n) to add the node to the replacement list. Then add the IP, and then update the bucket:

event listener

// filedir: go-ethereum-1.10.2\p2p\discover\table.go L55
   refreshInterval    = 30 * time.Minute
   revalidateInterval = 10 * time.Second
   copyNodesInterval  = 30 * time.Second
// filedir: go-ethereum-1.10.2\p2p\discover\table.go  L218
// loop schedules runs of doRefresh, doRevalidate and copyLiveNodes.
func (tab *Table) loop() {
   var (
       revalidate     = time.NewTimer(tab.nextRevalidateTime())
       refresh        = time.NewTicker(refreshInterval)
       copyNodes      = time.NewTicker(copyNodesInterval)
       refreshDone    = make(chan struct{})           // where doRefresh reports completion
       revalidateDone chan struct{}                   // where doRevalidate reports completion
       waiting        = []chan struct{}{tab.initDone} // holds waiting callers while doRefresh runs
   )
   defer refresh.Stop()
   defer revalidate.Stop()
   defer copyNodes.Stop()
   // Start initial refresh.
   go tab.doRefresh(refreshDone)
loop:
   for {
       select {
The loop function is the main loop in table.go. Local variables that will be used later are defined at the beginning of the function, and then the bucket is refreshed through deRefresh. The loop here will automatically refresh the K bucket every 30 minutes. Verify whether the K-bucket nodes can be pinged every 10 seconds, and store the nodes in the K-bucket for more than 5 minutes in the local database every 30 seconds as stable nodes:
           tab.seedRand()
           if refreshDone == nil {
               refreshDone = make(chan struct{})
               go tab.doRefresh(refreshDone)
           }
case <-refresh.C: //Refresh k bucket events regularly, refreshInterval=30 min
           waiting = append(waiting, req)
           if refreshDone == nil {
               refreshDone = make(chan struct{})
               go tab.doRefresh(refreshDone)
           }
case req := <-tab.refreshReq: //Refresh the request event of k bucket
           for _, ch := range waiting {
               close(ch)
           }
           waiting, refreshDone = nil, nil
case <-refreshDone: // doRefresh completed
           revalidateDone = make(chan struct{})
           go tab.doRevalidate(revalidateDone)
case <-revalidate.C: // Verify the validity of k-bucket nodes, 10 seconds
           revalidate.Reset(tab.nextRevalidateTime())
           revalidateDone = nil
case <-revalidateDone: // Verify the validity of the K bucket node is completed
           go tab.copyLiveNodes()
       case <-tab.closeReq:
           break loop
       }
   }
   if refreshDone != nil {
       <-refreshDone
   }
   for _, ch := range waiting {
       close(ch)
   }
   if revalidateDone != nil {
       <-revalidateDone
   }
   close(tab.closed)
case <-copyNodes.C: // Store the nodes in the database regularly (30 seconds). If a node exists in the k-bucket for more than 5 minutes, it is considered a stable node

} node lookup

// 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()
   defer tab.mutex.Unlock()
   b := tab.bucket(id)
   for _, e := range b.entries {
       if e.ID() == id {
           return unwrapNode(e)
       }
   }
   return nil
}

The getNode function is used to find a node by ID, and returns nil if it does not exist:

node discovery

// filedir:go-ethereum-1.10.2\p2p\discover\table.go L278
func (tab *Table) doRefresh(done chan struct{}) {
   defer close(done)
   tab.loadSeedNodes()
   // Run self lookup to discover new neighbor nodes.
   tab.net.lookupSelf()
   for i := 0; i < 3; i++ {
       tab.net.lookupRandom()
   }
}

The Ethereum distributed network adopts a structured network model, and its implementation scheme uses the Kademlia protocol. Below we briefly introduce node discovery. In Ethereum, the value of k is 16, which means that each k bucket contains 16 nodes. 256 k-buckets, K-buckets record the node’s NodeId, Distance, Endpoint, IP and other information, and sort them according to the distance from the Target node. The node search is implemented by doRefresh():

// filedir: go-ethereum-1.10.2\p2p\discover\table.go L302
func (tab *Table) loadSeedNodes() {
   seeds := wrapNodes(tab.db.QuerySeeds(seedCount, seedMaxAge))
   seeds = append(seeds, tab.nursery...)
   for i := range seeds {
       seed := seeds[i]
       age := log.Lazy{Fn: func() interface{} { return time.Since(tab.db.LastPongReceived(seed.ID(), seed.IP())) }}
       tab.log.Trace("Found seed node in database", "id", seed.ID(), "addr", seed.addr(), "age", age)
       tab.addSeenNode(seed)
   }
}
// filedir:go-ethereum-1.10.2\p2p\enode\nodedb.go L440
// QuerySeeds retrieves random nodes to be used as potential seed nodes
// for bootstrapping.
func (db *DB) QuerySeeds(n int, maxAge time.Duration) []*Node {
   var (
       now   = time.Now()
       nodes = make([]*Node, 0, n)
       it    = db.lvl.NewIterator(nil, nil)
       id    ID
   )
   defer it.Release()
seek:
   for seeks := 0; len(nodes) < n && seeks < n*5; seeks++ {
       ctr := id[0]
       rand.Read(id[:])
       id[0] = ctr + id[0]%16
       it.Seek(nodeKey(id))
       n := nextNode(it)
       if n == nil {
           id[0] = 0
           continue seek // iterator exhausted
       }
       if now.Sub(db.LastPongReceived(n.ID(), n.IP())) > maxAge {
           continue seek
       }
       for i := range nodes {
           if nodes[i].ID() == n.ID() {
               continue seek // duplicate
           }
       }
       nodes = append(nodes, n)
   }
   return nodes
}

As you can see from the above code, first call tab.loadSeedNodes() to load nodes from the database and insert them into the table:

// filedir: go-ethereum-1.10.2\p2p\discover\v5_udp.go L281
// lookupSelf looks up our own node ID.
// This is needed to satisfy the transport interface.
func (t *UDPv5) lookupSelf() []*enode.Node {
   return t.newLookup(t.closeCtx, t.Self().ID()).run()
}
// filedir: go-ethereum-1.10.2\p2p\discover\v5_udp.go L292
func (t *UDPv5) newLookup(ctx context.Context, target enode.ID) *lookup {
   return newLookup(ctx, t.tab, target, func(n *node) ([]*node, error) {
       return t.lookupWorker(n, target)
   })
}
func newLookup(ctx context.Context, tab *Table, target enode.ID, q queryFunc) *lookup {
   it := &lookup{
       tab:       tab,
       queryfunc: q,
       asked:     make(map[enode.ID]bool),
       seen:      make(map[enode.ID]bool),
       result:    nodesByDistance{target: target},
       replyCh:   make(chan []*node, alpha),
       cancelCh:  ctx.Done(),
       queries:   -1,
   }
   // Don't query further if we hit ourself.
   // Unlikely to happen often in practice.
   it.asked[tab.self().ID()] = true
   return it
}

Then use lookupSelf to discover new nodes. Here, the ID of the current node will be used first to run newLookup to discover neighbor nodes:

// filedir: go-ethereum-1.10.2\p2p\discover\v5_udp.go L275
func (t *UDPv5) lookupRandom() []*enode.Node {
   return t.newRandomLookup(t.closeCtx).run()
}
// filedir: go-ethereum-1.10.2\p2p\discover\v5_udp.go L287
func (t *UDPv5) newRandomLookup(ctx context.Context) *lookup {
   var target enode.ID
   crand.Read(target[:])
   return t.newLookup(ctx, target)
}
func (t *UDPv5) newLookup(ctx context.Context, target enode.ID) *lookup {
   return newLookup(ctx, t.tab, target, func(n *node) ([]*node, error) {
       return t.lookupWorker(n, target)
   })
}
// lookupWorker performs FINDNODE calls against a single node during lookup.
func (t *UDPv5) lookupWorker(destNode *node, target enode.ID) ([]*node, error) {
   var (
       dists = lookupDistances(target, destNode.ID())
       nodes = nodesByDistance{target: target}
       err   error
   )
   var r []*enode.Node
   r, err = t.findnode(unwrapNode(destNode), dists)
   if err == errClosed {
       return nil, err
   }
   for _, n := range r {
       if n.ID() != t.Self().ID() {
           nodes.push(wrapNode(n), findnodeResultLimit)
       }
   }
   return nodes.entries, err
}
// filedir:go-ethereum-1.10.2\p2p\discover\v5_udp.go  L362
// findnode calls FINDNODE on a node and waits for responses.
func (t *UDPv5) findnode(n *enode.Node, distances []uint) ([]*enode.Node, error) {
   resp := t.call(n, v5wire.NodesMsg, &v5wire.Findnode{Distances: distances})
   return t.waitForNodes(resp, distances)
}
func (t *UDPv5) call(node *enode.Node, responseType byte, packet v5wire.Packet) *callV5 {
   c := &callV5{
       node:         node,
       packet:       packet,
       responseType: responseType,
       reqid:        make([]byte, 8),
       ch:           make(chan v5wire.Packet, 1),
       err:          make(chan error, 1),
   }
   // Assign request ID.
   crand.Read(c.reqid)
   packet.SetRequestID(c.reqid)
   // Send call to dispatch.
   select {
   case t.callCh <- c:
   case <-t.closeCtx.Done():
       c.err <- errClosed
   }
   return c
}

Finally, a random target is performed for lookup:

service structure

// filedir:go-ethereum-1.10.2\p2p\server.go  L160
// Server manages all peer connections.
type Server struct {
   // Config fields may not be modified while the server is running.
   Config
   // Hooks for testing. These are useful because we can inhibit
   // the whole protocol stack.
   newTransport func(net.Conn, *ecdsa.PublicKey) transport
   newPeerHook  func(*Peer)
   listenFunc   func(network, addr string) (net.Listener, error)
   lock    sync.Mutex // protects running
   running bool
   listener     net.Listener
   ourHandshake *protoHandshake
   loopWG       sync.WaitGroup // loop, listenLoop
   peerFeed     event.Feed
   log          log.Logger
   nodedb    *enode.DB
   localnode *enode.LocalNode
   ntab      *discover.UDPv4
   DiscV5    *discover.UDPv5
   discmix   *enode.FairMix
   dialsched *dialScheduler
   // Channels into the run loop.
   quit                    chan struct{}
   addtrusted              chan *enode.Node
   removetrusted           chan *enode.Node
   peerOp                  chan peerOpFunc
   peerOpDone              chan struct{}
   delpeer                 chan peerDrop
   checkpointPostHandshake chan *conn
   checkpointAddPeer       chan *conn
   // State of run loop and listenLoop.
   inboundHistory expHeap
}

The data structure on the server side is as follows:

// Config holds Server options.
type Config struct {
   PrivateKey *ecdsa.PrivateKey `toml:"-"`
   MaxPeers int
   MaxPendingPeers int `toml:",omitempty"`
   DialRatio int `toml:",omitempty"`
   NoDiscovery bool
   DiscoveryV5 bool `toml:",omitempty"`
   Name string `toml:"-"`
   BootstrapNodes []*enode.Node
   BootstrapNodesV5 []*enode.Node `toml:",omitempty"`
   StaticNodes []*enode.Node
   TrustedNodes []*enode.Node
   NetRestrict *netutil.Netlist `toml:",omitempty"`
   NodeDatabase string `toml:",omitempty"`
   Protocols []Protocol `toml:"-"`
   ListenAddr string
   NAT nat.Interface `toml:",omitempty"`
   Dialer NodeDialer `toml:"-"`
   NoDial bool `toml:",omitempty"`
   EnableMsgEvents bool
   Logger log.Logger `toml:",omitempty"`
   clock mclock.Clock
}

#####

Server configuration (local node secret key, dial-up ratio, maximum number of node connections, dial-up ratio, event records, etc.):

new node

// filedir:go-ethereum-1.10.2\p2p\server.go  L318
func (srv *Server) AddPeer(node *enode.Node) {
   srv.dialsched.addStatic(node)
}
// filedir:go-ethereum-1.10.2\p2p\dial.go  L190
// addStatic adds a static dial candidate.
func (d *dialScheduler) addStatic(n *enode.Node) {
   select {
   case d.addStaticCh <- n:
   case <-d.ctx.Done():
   }
}

The AddPeer function is used to add a given node, and its implementation code is as follows:

// AddTrustedPeer adds the given node to a reserved whitelist which allows the
// node to always connect, even if the slot are full.
func (srv *Server) AddTrustedPeer(node *enode.Node) {
   select {
   case srv.addtrusted <- node:
   case <-srv.quit:
   }
}

The AddTrustedPeer function is used to add a trusted node:

remove node

// filedir:go-ethereum-1.10.2\p2p\server.go  L325
func (srv *Server) RemovePeer(node *enode.Node) {
   var (
       ch  chan *PeerEvent
       sub event.Subscription
   )
   // Disconnect the peer on the main loop.
   srv.doPeerOp(func(peers map[enode.ID]*Peer) {
       srv.dialsched.removeStatic(node)
       if peer := peers[node.ID()]; peer != nil {
           ch = make(chan *PeerEvent, 1)
           sub = srv.peerFeed.Subscribe(ch)
           peer.Disconnect(DiscRequested)
       }
   })
   // Wait for the peer connection to end.
   if ch != nil {
       defer sub.Unsubscribe()
       for ev := range ch {
           if ev.Peer == node.ID() && ev.Type == PeerEventTypeDrop {
               return
           }
       }
   }
}
// filedir:go-ethereum-1.10.2\p2p\dial.go  L198
// removeStatic removes a static dial candidate.
func (d *dialScheduler) removeStatic(n *enode.Node) {
   select {
   case d.remStaticCh <- n:
   case <-d.ctx.Done():
   }
}

The RemovePeer function is used to remove a node and disconnect from the node:

// RemoveTrustedPeer removes the given node from the trusted peer set.
func (srv *Server) RemoveTrustedPeer(node *enode.Node) {
   select {
   case srv.removetrusted <- node:
   case <-srv.quit:
   }
}

The RemoveTrustedPeer function is used to remove a trusted peer:

terminate service

func (srv *Server) Stop() {
   srv.lock.Lock()
   if !srv.running {
       srv.lock.Unlock()
       return
   }
   srv.running = false
   if srv.listener != nil {
       // this unblocks listener Accept
       srv.listener.Close()
   }
   close(srv.quit)
   srv.lock.Unlock()
   srv.loopWG.Wait()
}

The Stop function is used to terminate the node operation, the specific code is as follows:

service start

// filedir:go-ethereum-1.10.2\p2p\server.go  L433
func (srv *Server) Start() (err error) {
   srv.lock.Lock()
   defer srv.lock.Unlock()
   if srv.running {
       return errors.New("server already running")
   }
   srv.running = true
   srv.log = srv.Config.Logger
   if srv.log == nil {
       srv.log = log.Root()
   }
   if srv.clock == nil {
       srv.clock = mclock.System{}
   }
   if srv.NoDial && srv.ListenAddr == "" {
       srv.log.Warn("P2P server will be useless, neither dialing nor listening")
   }
   // static fields
   if srv.PrivateKey == nil {
       return errors.New("Server.PrivateKey must be set to a non-nil key")
   }
   if srv.newTransport == nil {
       srv.newTransport = newRLPX
   }
   if srv.listenFunc == nil {
       srv.listenFunc = net.Listen
   }
   srv.quit = make(chan struct{})
   srv.delpeer = make(chan peerDrop)
   srv.checkpointPostHandshake = make(chan *conn)
   srv.checkpointAddPeer = make(chan *conn)
   srv.addtrusted = make(chan *enode.Node)
   srv.removetrusted = make(chan *enode.Node)
   srv.peerOp = make(chan peerOpFunc)
   srv.peerOpDone = make(chan struct{})
   if err := srv.setupLocalNode(); err != nil {
       return err
   }
   if srv.ListenAddr != "" {
       if err := srv.setupListening(); err != nil {
           return err
       }
   }
   if err := srv.setupDiscovery(); err != nil {
       return err
   }
   srv.setupDialScheduler()
   srv.loopWG.Add(1)
   go srv.run()
   return nil
}

The start function located in go-ethereum-1.10.2\p2p\server.go is used to start a P2P node:

// Start starts running the server.
// Servers can not be re-used after stopping.
func (srv *Server) Start() (err error) {
   srv.lock.Lock()
   defer srv.lock.Unlock()
   if srv.running {
       return errors.New("server already running")
   }
   srv.running = true
   srv.log = srv.Config.Logger
   if srv.log == nil {
       srv.log = log.Root()
   }
   if srv.clock == nil {
       srv.clock = mclock.System{}
   }
   if srv.NoDial && srv.ListenAddr == "" {
       srv.log.Warn("P2P server will be useless, neither dialing nor listening")
   }
   // static fields
   if srv.PrivateKey == nil {
       return errors.New("Server.PrivateKey must be set to a non-nil key")
   }
   if srv.newTransport == nil {
       srv.newTransport = newRLPX
   }
   if srv.listenFunc == nil {
       srv.listenFunc = net.Listen
   }
   srv.quit = make(chan struct{})
   srv.delpeer = make(chan peerDrop)
   srv.checkpointPostHandshake = make(chan *conn)
   srv.checkpointAddPeer = make(chan *conn)
   srv.addtrusted = make(chan *enode.Node)
   srv.removetrusted = make(chan *enode.Node)
   srv.peerOp = make(chan peerOpFunc)
   srv.peerOpDone = make(chan struct{})

Here, first check whether the current node is in the running state, if it is, return directly and give an error message, if not, set srv.running to true, then enter the service startup process, then check whether the log is enabled, etc., and then initialize the configuration P2P service information:

Then call setupLocalNode to start a local node, and establish a local monitor, and then configure a DiscoveryV5 network protocol to generate a node routing table.

   srv.setupDialScheduler()
   srv.loopWG.Add(1)
   go srv.run()
   return nil
}

Then call setupDialScheduler to start the active dial-up connection process, and then open a coroutine in which to maintain the peer:

func (srv *Server) setupDialScheduler() {
   config := dialConfig{
       self:           srv.localnode.ID(),
       maxDialPeers:   srv.maxDialedConns(),
       maxActiveDials: srv.MaxPendingPeers,
       log:            srv.Logger,
       netRestrict:    srv.NetRestrict,
       dialer:         srv.Dialer,
       clock:          srv.clock,
   }
   if srv.ntab != nil {
       config.resolver = srv.ntab
   }
   if config.dialer == nil {
       config.dialer = tcpDialer{&net.Dialer{Timeout: defaultDialTimeout}}
   }
   srv.dialsched = newDialScheduler(config, srv.discmix, srv.SetupConn)
   for _, n := range srv.StaticNodes {
       srv.dialsched.addStatic(n)
   }
}

The setupDialScheduler code is as follows, here the connection is established through newDialScheduler, the parameter discmix determines the node set when actively establishing a connection, it is an iterator, and the setupConn connection establishment function is passed in at the same time:

// filedir:go-ethereum-1.10.2\p2p\dial.go   L162
func newDialScheduler(config dialConfig, it enode.Iterator, setupFunc dialSetupFunc) *dialScheduler {
   d := &dialScheduler{
       dialConfig:  config.withDefaults(),
       setupFunc:   setupFunc,
       dialing:     make(map[enode.ID]*dialTask),
       static:      make(map[enode.ID]*dialTask),
       peers:       make(map[enode.ID]connFlag),
       doneCh:      make(chan *dialTask),
       nodesIn:     make(chan *enode.Node),
       addStaticCh: make(chan *enode.Node),
       remStaticCh: make(chan *enode.Node),
       addPeerCh:   make(chan *conn),
       remPeerCh:   make(chan *conn),
   }
   d.lastStatsLog = d.clock.Now()
   d.ctx, d.cancel = context.WithCancel(context.Background())
   d.wg.Add(2)
   go d.readNodes(it)
   go d.loop(it)
   return d
}

The newDialScheduler function is as follows, where the node is obtained from the iterator through d.readNodes(it), and then passed to d.loop(it) through the channel for connection:

service monitoring

func (srv *Server) setupListening() error {
   // Launch the listener.
   listener, err := srv.listenFunc("tcp", srv.ListenAddr)
   if err != nil {
       return err
   }
   srv.listener = listener
   srv.ListenAddr = listener.Addr().String()
   // Update the local node record and map the TCP listening port if NAT is configured.
   if tcp, ok := listener.Addr().(*net.TCPAddr); ok {
       srv.localnode.Set(enr.TCP(tcp.Port))
       if !tcp.IP.IsLoopback() && srv.NAT != nil {
           srv.loopWG.Add(1)
           go func() {
               nat.Map(srv.NAT, srv.quit, "tcp", tcp.Port, tcp.Port, "ethereum p2p")
               srv.loopWG.Done()
           }()
       }
   }
   srv.loopWG.Add(1)
   go srv.listenLoop()
   return nil
}

In the above service startup process, there is a setupListening function, which is used to listen to events. The specific code is as follows:

// listenLoop runs in its own goroutine and accepts
// inbound connections.
func (srv *Server) listenLoop() {
   srv.log.Debug("TCP listener up", "addr", srv.listener.Addr())
   // The slots channel limits accepts of new connections.
   tokens := defaultMaxPendingPeers
   if srv.MaxPendingPeers > 0 {
       tokens = srv.MaxPendingPeers
   }
   slots := make(chan struct{}, tokens)
   for i := 0; i < tokens; i++ {
       slots <- struct{}{}
   }
   // Wait for slots to be returned on exit. This ensures all connection goroutines
   // are down before listenLoop returns.
   defer srv.loopWG.Done()
   defer func() {
       for i := 0; i < cap(slots); i++ {
           <-slots
       }
   }()
   for {
       // Wait for a free slot before accepting.
       <-slots
       var (
           fd      net.Conn
           err     error
           lastLog time.Time
       )
       for {
           fd, err = srv.listener.Accept()
           if netutil.IsTemporaryError(err) {
               if time.Since(lastLog) > 1*time.Second {
                   srv.log.Debug("Temporary read error", "err", err)
                   lastLog = time.Now()
               }
               time.Sleep(time.Millisecond * 200)
               continue
           } else if err != nil {
               srv.log.Debug("Read error", "err", err)
               slots <- struct{}{}
               return
           }
           break
       }
       remoteIP := netutil.AddrIP(fd.RemoteAddr())
       if err := srv.checkInboundConn(remoteIP); err != nil {
           srv.log.Debug("Rejected inbound connection", "addr", fd.RemoteAddr(), "err", err)
           fd.Close()
           slots <- struct{}{}
           continue
       }
       if remoteIP != nil {
           var addr *net.TCPAddr
           if tcp, ok := fd.RemoteAddr().(*net.TCPAddr); ok {
               addr = tcp
           }
           fd = newMeteredConn(fd, true, addr)
           srv.log.Trace("Accepted connection", "addr", fd.RemoteAddr())
       }
       go func() {
           srv.SetupConn(fd, inboundConn, nil)
           slots <- struct{}{}
       }()
   }
}

In the above code, another srv.listenLoop() is called, which is an infinite loop goroutine, which will listen to the port and receive external requests:

// SetupConn runs the handshakes and attempts to add the connection
// 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 {
   c := &conn{fd: fd, flags: flags, cont: make(chan error)}
   if dialDest == nil {
       c.transport = srv.newTransport(fd, nil)
   } else {
       c.transport = srv.newTransport(fd, dialDest.Pubkey())
   }
   err := srv.setupConn(c, flags, dialDest)
   if err != nil {
       c.close(err)
   }
   return err
}

The SetupConn here mainly executes the handshake protocol, and tries to create the link as a peer object:

func (srv *Server) setupConn(c *conn, flags connFlag, dialDest *enode.Node) error {
   // Prevent leftover pending conns from entering the handshake.
   srv.lock.Lock()
   running := srv.running
   srv.lock.Unlock()
   if !running {
       return errServerStopped
   }
   // If dialing, figure out the remote public key.
   var dialPubkey *ecdsa.PublicKey
In the above code, the srv.setupConn(c, flags, dialDest) function is called again, which is used to execute the handshake protocol:
       dialPubkey = new(ecdsa.PublicKey)
       if err := dialDest.Load((*enode.Secp256k1)(dialPubkey)); err != nil {
           err = errors.New("dial destination doesn't have a secp256k1 public key")
           srv.log.Trace("Setting up connection failed", "addr", c.fd.RemoteAddr(), "conn", c.flags, "err", err)
           return err
       }
   }
   // Run the RLPx handshake.
   remotePubkey,if dialDest != nil { // dest=nil passive connection, dest != nil active connection
   if err != nil {
       srv.log.Trace("Failed RLPx handshake", "addr", c.fd.RemoteAddr(), "conn", c.flags, "err", err)
       return err
   }
   if dialDest != nil {
       c.node = dialDest
   } else {
       c.node = nodeFromConn(remotePubkey, c.fd)
   }
   clog := srv.log.New("id", c.node.ID(), "addr", c.fd.RemoteAddr(), "conn", c.flags)
   err = srv.checkpoint(c, srv.checkpointPostHandshake)
   if err != nil {
       clog.Trace("Rejected peer", "err", err)
       return err
   }
   // Run the capability negotiation handshake.
   phs,err := c.doEncHandshake(srv.PrivateKey) // public key exchange, determine the handshake at the RLPx level of the shared secret key
   if err != nil {
       clog.Trace("Failed p2p handshake", "err", err)
       return err
   }
   if id := c.node.ID(); !bytes.Equal(crypto.Keccak256(phs.ID), id[:]) {
       clog.Trace("Wrong devp2p handshake identity", "phsid", hex.EncodeToString(phs.ID))
       return DiscUnexpectedIdentity
   }
   c.caps, c.name = phs.Caps, phs.Name
   err = srv.checkpoint(c,err := c.doProtoHandshake(srv.ourHandshake) // Handshake at the protocol level, that is, p2p handshake, come and go
   if err != nil {
       clog.Trace("Rejected peer", "err", err)
       return err
   }
   return nil
}

srv.checkpointAddPeer) // status verification

// filedir:go-ethereum-1.10.2\p2p\transport.go  L123
func (t *rlpxTransport) doEncHandshake(prv *ecdsa.PrivateKey) (*ecdsa.PublicKey, error) {
   t.conn.SetDeadline(time.Now().Add(handshakeTimeout))
   return t.conn.Handshake(prv)
}

The key handshake is implemented through the deEncHandshake function, and the Handshake() function is called in the function:

// filedir:go-ethereum-1.10.2\p2p\rlpx\rlpx.go   L253
func (c *Conn) Handshake(prv *ecdsa.PrivateKey) (*ecdsa.PublicKey, error) {
   var (
       sec Secrets
       err error
   )
The Handshake code is as follows, where the corresponding handshake logic will be executed according to whether it is an active handshake or a passive handshake:
       sec, err = initiatorEncHandshake(c.conn, prv,if c.dialDest != nil { //Active handshake
c.dialDest) //Proactively initiate key verification handshake ends, confirm the shared key
       sec, err = receiverEncHandshake(c.conn, prv)
   }
   if err != nil {
       return nil, err
   }
   c.InitWithSecrets(sec)
   return sec.remote, err
}

} else { // passive handshake

// filedir:go-ethereum-1.10.2\p2p\rlpx\rlpx.go  L477
func initiatorEncHandshake(conn io.ReadWriter, prv *ecdsa.PrivateKey, remote *ecdsa.PublicKey) (s Secrets, err error) {
   h := &encHandshake{initiator: true, remote: ecies.ImportECDSAPublic(remote)}
   authMsg, err := h.makeAuthMsg(prv)
   if err != nil {
       return s, err
   }
   authPacket, err := sealEIP8(authMsg, h)
   if err != nil {
       return s, err
   }
   if _, err = conn.Write(authPacket); err != nil {
       return s, err
   }
   authRespMsg := new(authRespV4)
   authRespPacket, err := readHandshakeMsg(authRespMsg, encAuthRespLen, prv, conn)
   if err != nil {
       return s, err
   }
   if err := h.handleAuthResp(authRespMsg); err != nil {
       return s, err
   }
   return h.secrets(authPacket, authRespPacket)
}

The process of actively initiating the handshake is as follows. Here, makeAuthMsg will be called to generate the Auth identity information, including the signature, the public key and version number corresponding to the signature generated by the random nonce, and then the sealEIP8 method is called to perform rlpx encoding, and then the encrypted handshake is initiated, and then received Return the authResp message, verify and decrypt it, obtain the public key of the other party, and then generate AES, MAC:

func receiverEncHandshake(conn io.ReadWriter, prv *ecdsa.PrivateKey) (s Secrets, err error) {
   authMsg := new(authMsgV4)
   authPacket, err := readHandshakeMsg(authMsg, encAuthMsgLen, prv, conn)
   if err != nil {
       return s, err
   }
   h := new(encHandshake)
   if err := h.handleAuthMsg(authMsg, prv); err != nil {
       return s, err
   }
   authRespMsg, err := h.makeAuthResp()
   if err != nil {
       return s, err
   }
   var authRespPacket []byte
   if authMsg.gotPlain {
       authRespPacket, err = authRespMsg.sealPlain(h)
   } else {
       authRespPacket, err = sealEIP8(authRespMsg, h)
   }
   if err != nil {
       return s, err
   }
   if _, err = conn.Write(authRespPacket); err != nil {
       return s, err
   }
   return h.secrets(authPacket, authRespPacket)
}

receiverEncHandshake is as follows, which is almost the same as initiatorEncHandshake:

func (t *rlpxTransport) doProtoHandshake(our *protoHandshake) (their *protoHandshake, err error) {
   werr := make(chan error, 1)
   go func() { werr <- Send(t, handshakeMsg, our) }()  
   if their, err = readProtocolHandshake(t); err != nil {
       <-werr // make sure the write terminates too
       return nil, err
   }
   if err := <-werr; err != nil {
       return nil, fmt.Errorf("write error: %v", err)
   }
   // If the protocol version supports Snappy encoding, upgrade immediately
   t.conn.SetSnappy(their.Version >= snappyProtocolVersion)
   return their, nil
}

Then complete the protocol handshake operation through doProtoHandshake, call send here to send a handshake operation, and then read the returned information through readProtocolHandshake, and then check:

service cycle

func (srv *Server) run() {
   srv.log.Info("Started P2P networking", "self", srv.localnode.Node().URLv4())
   defer srv.loopWG.Done()
   defer srv.nodedb.Close()
   defer srv.discmix.Close()
   defer srv.dialsched.stop()
   var (
       peers        = make(map[enode.ID]*Peer)
       inboundCount = 0
       trusted      = make(map[enode.ID]bool, len(srv.TrustedNodes))
   )
   for _, n := range srv.TrustedNodes {
       trusted[n.ID()] = true
   }
running:
   for {
       select {
       case <-srv.quit:
           break running
       case n := <-srv.addtrusted:
           srv.log.Trace("Adding trusted node", "node", n)
           trusted[n.ID()] = true
           if p, ok := peers[n.ID()]; ok {
               p.rw.set(trustedConn, true)
           }
       case n := <-srv.removetrusted:
           srv.log.Trace("Removing trusted node", "node", n)
           delete(trusted, n.ID())
           if p, ok := peers[n.ID()]; ok {
               p.rw.set(trustedConn, false)
           }
       case op := <-srv.peerOp:
           op(peers)
           srv.peerOpDone <- struct{}{}
       case c := <-srv.checkpointPostHandshake:
           if trusted[c.node.ID()] {
               c.flags |= trustedConn
           }
           c.cont <- srv.postHandshakeChecks(peers, inboundCount, c)
       case c := <-srv.checkpointAddPeer:
           err := srv.addPeerChecks(peers, inboundCount, c)
           if err == nil {
               p := srv.launchPeer(c)
               peers[c.node.ID()] = p
               srv.log.Debug("Adding p2p peer", "peercount", len(peers), "id", p.ID(), "conn", c.flags, "addr", p.RemoteAddr(), "name", p.Name())
               srv.dialsched.peerAdded(c)
               if p.Inbound() {
                   inboundCount++
               }
           }
           c.cont <- err
       case pd := <-srv.delpeer:
           // A peer disconnected.
           d := common.PrettyDuration(mclock.Now() - pd.created)
           delete(peers, pd.ID())
           srv.log.Debug("Removing p2p peer", "peercount", len(peers), "id", pd.ID(), "duration", d, "req", pd.requested, "err", pd.err)
           srv.dialsched.peerRemoved(pd.rw)
           if pd.Inbound() {
               inboundCount--
           }
       }
   }
   srv.log.Trace("P2P networking is spinning down")
   if srv.ntab != nil {
       srv.ntab.Close()
   }
   if srv.DiscV5 != nil {
       srv.DiscV5.Close()
   }
   for _, p := range peers {
       p.Disconnect(DiscQuitting)
   }
   for len(peers) > 0 {
       p := <-srv.delpeer
       p.log.Trace("<-delpeer (spindown)")
       delete(peers, p.ID())
   }
}

The run function is the main loop of the service, monitoring server termination, adding trust nodes, removing trust nodes, adding check nodes, etc.:

node information

// NodeInfo gathers and returns a collection of metadata known about the host.
func (srv *Server) NodeInfo() *NodeInfo {
   node := srv.Self()
   info := &NodeInfo{
       Name:       srv.Name,
       Enode:      node.URLv4(),
       ID:         node.ID().String(),
       IP:       &nb