Eth Protocol

// New creates a new Ethereum object (including the
// initialisation of the common Ethereum object)
func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
    ...
    // Ensure configuration values are compatible and sane
    eth := &Ethereum{
    }
    ...
    // Register the backend on the node
    stack.RegisterAPIs(eth.APIs())
    stack.RegisterProtocols(eth.Protocols())
    stack.RegisterLifecycle(eth)
    ...
    return eth, nil
}

// Protocols returns all the currently configured
// network protocols to start.
func (s *Ethereum) Protocols() []p2p.Protocol {
    protos := eth.MakeProtocols((*ethHandler)(s.handler), s.networkID, s.ethDialCandidates)
    if s.config.SnapshotCache > 0 {
        protos = append(protos, snap.MakeProtocols((*snapHandler)(s.handler), s.snapDialCandidates)...)
    }
    return protos
}

// MakeProtocols constructs the P2P protocol definitions for `eth`.
func MakeProtocols(backend Backend, network uint64, dnsdisc enode.Iterator) []p2p.Protocol {
    protocols := make([]p2p.Protocol, len(ProtocolVersions))
    for i, version := range ProtocolVersions {
        version := version // Closure

        protocols[i] = p2p.Protocol{
            Name:    ProtocolName,
            Version: version,
            Length:  protocolLengths[version],
            // 注意：启动协议的时候执行这里的 Run 函数。
            Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
                peer := NewPeer(version, p, rw, backend.TxPool())
                defer peer.Close()

                return backend.RunPeer(peer, func(peer *Peer) error {
                    return Handle(backend, peer)
                })
            },
            NodeInfo: func() interface{} {
                return nodeInfo(backend.Chain(), network)
            },
            PeerInfo: func(id enode.ID) interface{} {
                return backend.PeerInfo(id)
            },
            Attributes:     []enr.Entry{currentENREntry(backend.Chain())},
            DialCandidates: dnsdisc,
        }
    }
    return protocols
}

// ProtocolVersions are the supported versions of the `eth` protocol (first
// is primary).
var ProtocolVersions = []uint{ETH67, ETH66}

    Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error {
        peer := NewPeer(version, p, rw, backend.TxPool())
        defer peer.Close()
        return backend.RunPeer(peer, func(peer *Peer) error {
            return Handle(backend, peer)
        })
    },

func NewPeer(version uint, p *p2p.Peer, rw p2p.MsgReadWriter, txpool TxPool) *Peer {
    peer := &Peer{
        ...
    }
    // Start up all the broadcasters
    go peer.broadcastBlocks()
    go peer.broadcastTransactions()
    go peer.announceTransactions()
    go peer.dispatcher()

    return peer
}

// runEthPeer registers an eth peer into the joint eth/snap peerset, adds it to
// various subsystems and starts handling messages.
func (h *handler) runEthPeer(peer *eth.Peer, handler eth.Handler) error {
    ...
    if err := peer.Handshake(h.networkID, td, hash, genesis.Hash(), forkID, h.forkFilter); err != nil {
        peer.Log().Debug("Ethereum handshake failed", "err", err)
        return err
    }
    ...
    // Handle incoming messages until the connection is torn down
    return handler(peer)

// Handshake executes the eth protocol handshake, negotiating version number,
// network IDs, difficulties, head and genesis blocks.
func (p *Peer) Handshake(network uint64, td *big.Int, head common.Hash, genesis common.Hash, forkID forkid.ID,
forkFilter forkid.Filter) error {
    ...
    // Send out own handshake in a new thread
    var status StatusPacket // safe to read after two values have been received from errc

    go func() {
        errc <- p2p.Send(p.rw, StatusMsg, &StatusPacket{
            ProtocolVersion: uint32(p.version),
            NetworkID:       network,
            TD:              td,
            Head:            head,
            Genesis:         genesis,
            ForkID:          forkID,
        })
    }()
    go func() {
        errc <- p.readStatus(network, &status, genesis, forkFilter)
    }()
    ...
    p.td, p.head = status.TD, status.Head
    ...
    return nil
}

// Handle is invoked whenever an `eth` connection is made that successfully passes
// the protocol handshake. This method will keep processing messages until the
// connection is torn down.
func Handle(backend Backend, peer *Peer) error {
    for {
        if err := handleMessage(backend, peer); err != nil {
            peer.Log().Debug("Message handling failed in `eth`", "err", err)
            return err
        }
    }
}

// handleMessage is invoked whenever an inbound message is received from a remote
// peer. The remote connection is torn down upon returning any error.
func handleMessage(backend Backend, peer *Peer) error {
    // Read the next message from the remote peer, and ensure it's fully consumed
    msg, err := peer.rw.ReadMsg()
    if err != nil {
        return err
    }
    if msg.Size > maxMessageSize {
        return fmt.Errorf("%w: %v > %v", errMsgTooLarge, msg.Size, maxMessageSize)
    }
    defer msg.Discard()

    var handlers = eth66
    if peer.Version() >= ETH67 {
        handlers = eth67
    }

    // Track the amount of time it takes to serve the request and run the handler
    if metrics.Enabled {
    }
    if handler := handlers[msg.Code]; handler != nil {
        return handler(backend, msg, peer)
    }
    return fmt.Errorf("%w: %v", errInvalidMsgCode, msg.Code)
}

// maxMessageSize is the maximum cap on the size of a protocol message.
const maxMessageSize = 10 * 1024 * 1024

var eth66 = map[uint64]msgHandler{
    NewBlockHashesMsg:             handleNewBlockhashes,
    NewBlockMsg:                   handleNewBlock,
    ...
}

var eth67 = map[uint64]msgHandler{
    NewBlockHashesMsg:             handleNewBlockhashes,
    NewBlockMsg:                   handleNewBlock,
    TransactionsMsg:               handleTransactions,
    ...
}