以太坊源码分析之P2P网络启动

目录

• p2p源码目录
• 启动p2p网络Start()设置本地节点
• 设置监听
• 设置节点发现
• 创建DialState

p2p源码目录

discover/          基于UDP的节点发现V4协议discv5/            节点发现V5协议enode/             节点信息enr/               以太坊节点记录（ethereum node records)nat/               网络地址转换，用于内网穿透netutil/protocol/simulations/       本地p2p网络的模拟器dial.go            建立连接请求，以任务的形式message.go         定义了读写的接口metrics.go         计时器和计量器工具peer.go            节点protocol.go        子协议rlpx.go            加密传输协议server.go          底层p2p网络的函数入口

启动p2p网络

启动p2p网络主要会做以下几件事：

1. 发现远程节点，建立相邻节点列表
2. 监听远程节点发过来的建立TCP请求
3. 向远程节点发送建立TCP连接请求

首先找到p2p网络启动的入口：

Start()

start

函数主要做了以下6件事：

1. 初始化server的字段
2. 设置本地节点setupLocalNode
3. 设置监听TCP连接请求setupListening
4. 设置节点发现（setupDiscovery）V4版本
5. 设置最大可以主动发起的连接为50/3
6. srv.run(dialer) 发起建立TCP连接请求

其中setupLocalNode、setupListening、setupDiscovery、newDialState、srv.run(dialer)是我们要重点分析的函数。

设置本地节点

进入到setupLocalNode中：

①：创建devp2p握手

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))

握手协议包括协议版本号，节点名称和节点的公钥，存入到Caps中要根据名称和协议排序。

②：创建本地节点

db, err := enode.OpenDB(srv.Config.NodeDatabase)if err != nil {return err}srv.nodedb = dbsrv.localnode = enode.NewLocalNode(db, srv.PrivateKey)srv.localnode.SetFallbackIP(net.IP{127, 0, 0, 1})// TODO: check conflictsfor _, p := range srv.Protocols {for _, e := range p.Attributes {srv.localnode.Set(e)}}

首先从节点数据库中去获取节点信息，如果不存在则新建本地节点并设置默认IP，同时将节点记录的协议特定信息存入到本地节点中。

设置监听

进入到

setupListening

:

①：启动监听器

②：如果配置了NAT，则更新本地节点记录并映射TCP监听端口

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()}()}}

③：开启P2P监听，接收

inbound

连接

srv.listenLoop()

这个函数需要进一步分析：

主要有以下逻辑：

• 首先

  defaultMaxPendingPeers

  这个字段指的是

  inbound

  和

  outbound

  连接，默认最大值为50

• 将监听的连接返回给

  listener

  fd, err = srv.listener.Accept()

• 获取监听的连接的地址并检查这个连接

  remoteIP :ad8= netutil.AddrIP(fd.RemoteAddr())if err := srv.checkInboundConn(fd, remoteIP); err != nil {.....}

  checkInboundConn

  主要是做了以下的判断：

  拒绝不符合NetRestrict的连接（NetRestrict是指已经限定了某些连接，除此之外会拒绝）

• 拒绝尝试过多的节点

go func() {srv.SetupConn(fd, inboundConn, nil)// 连接建立过程（将连接添加为peer）slots <- struct{}{}}()

var dialPubkey *ecdsa.PublicKeyif dialDest != nil {dialPubkey = new(ecdsa.PublicKey)if err := dialDest.Load((*enode.Secp256k1)(dialPubkey)); err != nil {return errors.New(\"dial destination doesn\'t have a secp256k1 public key\")}}

remotePubkey, err := c.doEncHandshake(srv.PrivateKey, dialPubkey)

if dialDest != nil {// For dialed connections, check that the remote public key matches.//对于拨号连接，请检查远程公钥是否匹配if dialPubkey.X.Cmp(remotePubkey.X) != 0 || dialPubkey.Y.Cmp(remotePubkey.Y) != 0 {return DiscUnexpectedIdentity}c.node = dialDest} else {c.node = nodeFromConn(remotePubkey, c.fd)}

phs, err := c.doProtoHandshake(srv.ourHandshake)

设置节点发现

进入到

srv.setupDiscovery()

①：添加特定于协议的发现源

added := make(map[string]bool)for _, proto := range srv.Protocols {if proto.DialCandidates != nil && !added[proto.Name] {srv.discmix.AddSource(proto.DialCandidates)added[proto.Name] = true}}

②：如果DHT禁用的话，就不要在UDP上监听

if srv.NoDiscovery && !srv.DiscoveryV5 {return nil}

③：监听给定的socket 上的发现的包

ntab, err := discover.ListenUDP(conn, srv.localnode, cfg)

创建DialState

dialstate负责拨号和查找发现。

①：初始化dialstate

s := &dialstate{maxDynDials: maxdyn,self:        self,netrestrict: cfg.NetRestrict,log:         cfg.Logger,static:      make(map[enode.ID]*dialTask),dialing:     make(map[enode.ID]connFlag),bootnodes:   make([]*enode.Node, len(cfg.BootstrapNodes)),}

②：加入初始引导节点

copy(s.bootnodes, cfg.BootstrapNodes)if s.log == nil {s.log = log.Root()}

③： 加入静态节点

for _, n := range cfg.StaticNodes {s.addStatic(n)}

bootnodes

是初始引导节点，在节点没有接收到任何节点的连接请求，也没有节点可以给我们邻居节点的时候，就去连接

bootnodes

，它硬编码在了以太坊的源码中。

static

是静态节点，如果我们想和某些节点保持长期的连接，就把它们加入到静态节点的列表中

接下来就是到了运行p2p网络的时候了，主要的函数是：

go srv.run(dialer)

运行p2p网络

srv.run(dialer)

在p2p网络启动时候，我们会监听远程节点发送过来的TCP请求，到了运行p2p网络的时候，我们则会向远程节点发起TCP的连接请求。首先我们要知道我们所说的发起TCP连接请求可以形容成拨号，每个拨号都是以任务的形式存在，进入到

srv.run(dialer)

分析

整个函数就是一个循环，介绍下它的主要功能：

发起TCP连接任务

scheduleTasks()

scheduleTasks

主要是从queued task 中去获取任务，通过查询dialer以查找新任务并立即启动尽可能多的任务，我们这里要注意个变量

maxActiveDialTasks

,它的默认值为16 ，而安排任务的核心方法是：

nt := dialstate.newTasks(len(runningTasks)+len(queuedTasks), peers, time.Now())

主要做了以下几件事：

①：为没有连接的静态节点创建拨号任务

for id, t := range s.static {err := s.checkDial(t.dest, peers)switch err {case errNotWhitelisted, errSelf:s.log.Warn(\"Removing static dial candidate\", \"id\", t.dest.ID, \"addr\", &net.TCPAddr{IP: t.dest.IP(), Port: t.dest.TCP()}, \"err\", err)delete(s.static, t.dest.ID())case nil:s.dialing[id] = t.flagsnewtasks = append(newtasks, t)}}

首先对拨号节点进行校验：正在连接，已经连接，是本身，不在白名单中，最近连接过的都会报错，并且不是在白名单中的和自身的节点会直接从静态节点列表中删除，校验通过的创建任务。

②：计算所需的动态拨号数

needDynDials := s.maxDynDialsfor _, p := range peers {if p.rw.is(dynDialedConn) {needDynDials--}}for _, flag := range s.dialing {if flag&dynDialedConn != 0 {needDynDials--}}

maxPeers/radio

if len(peers) == 0 && len(s.bootnodes) > 0 && needDynDials > 0 && now.Sub(s.start) > fallbackInterval {bootnode := s.bootnodes[0]s.bootnodes = append(s.bootnodes[:0], s.bootnodes[1:]...)s.bootnodes = append(s.bootnodes, bootnode)if addDial(dynDialedConn, bootnode) {needDynDials--}}

s.lookupBuf

lookupBuf

for ; i < len(s.lookupBuf) && needDynDials > 0; i++ {if addDial(dynDialedConn, s.lookupBuf[i]) {needDynDials--}}s.lookupBuf = s.lookupBuf[:copy(s.lookupBuf, s.lookupBuf[i:])]if len(s.lookupBuf) < needDynDials && !s.lookupRunning {s.lookupRunning = truenewtasks = append(newtasks, &discoverTask{want: needDynDials - len(s.lookupBuf)})}

if nRunning == 0 && len(newtasks) == 0 && s.hist.Len() > 0 {t := &waitExpireTask{s.hist.nextExpiry().Sub(now)}newtasks = append(newtasks, t)}

newTasks

startTasks := func(ts []task) (rest []task) {i := 0for ; len(runningTasks) < maxActiveDialTasks && i < len(ts); i++ {t := ts[i]srv.log.Trace(\"New dial task\", \"task\", t)go func() { t.Do(srv); taskdone <- t }()runningTasks = append(runningTasks, t)}return ts[i:]}

t.Do(srv);

dialTask

func (t *dialTask) Do(srv *Server) {if t.dest.Incomplete() {if !t.resolve(srv) {return}}err := t.dial(srv, t.dest)if err != nil {srv.log.Trace(\"Dial error\", \"task\", t, \"err\", err)// Try resolving the ID of static nodes if dialing failed.if _, ok := err.(*dialError); ok && t.flags&staticDialedConn != 0 {if t.resolve(srv) {t.dial(srv, t.dest)}}}}

t.dail

// 实际的网络连接操作func (t *dialTask) dial(srv *Server, dest *enode.Node) error {fd, err := srv.Dialer.Dial(dest)if err != nil {return &dialError{err}}mfd := newMeteredConn(fd, false, &net.TCPAddr{IP: dest.IP(), Port: dest.TCP()})return srv.SetupConn(mfd, t.flags, dest)}

case <-srv.quit:break running

case n := <-srv.addstatic:srv.log.Trace(\"Adding static node\", \"node\", n)dialstate.addStatic(n)

case n := <-srv.removestatic:dialstate.removeStatic(n)if p, ok := peers[n.ID()]; ok {p.Disconnect(DiscRequested)}

case n := <-srv.addtrusted:trusted[n.ID()] = true

case n := <-srv.removetrusted:delete(trusted, n.ID())

case t := <-taskdone:newTasksdialstate.taskDone(t, time.Now())delTask(t)

case c := <-srv.checkpointPostHandshake:c.cont <- srv.postHandshakeChecks(peers, inboundCount, c)

err := srv.addPeerChecks(peers, inboundCount, c)if err == nil {// 握手完成，所有检查完毕p := newPeer(srv.log, c, srv.Protocols)//启用了消息事件就把peerfeed传给peerif srv.EnableMsgEvents {p.events = &srv.peerFeed}name := truncateName(c.name)p.RemoteAddr(), \"peers\", len(peers)+1, \"name\", name)go srv.runPeer(p) // 重点peers[c.node.ID()] = pif p.Inbound() {inboundCount++}if conn, ok := c.fd.(*meteredConn); ok {conn.handshakeDone(p)//}}

addPeerChecks

handshakeDone

case pd := <-srv.delpeer:d := common.PrettyDuration(mclock.Now() - pd.created)pd.log.Debug(\"Removing p2p peer\", \"addr\", pd.RemoteAddr(), \"peers\", len(peers)-1, \"duration\", d, \"req\", pd.requested, \"err\", pd.err)delete(peers, pd.ID())if pd.Inbound() {inboundCount--}