123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546 |
- // Copyright 2014 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 <http://www.gnu.org/licenses/>.
- package p2p
- import (
- "errors"
- "fmt"
- "io"
- "net"
- "sort"
- "sync"
- "time"
- "github.com/ethereum/go-ethereum/common/mclock"
- "github.com/ethereum/go-ethereum/event"
- "github.com/ethereum/go-ethereum/log"
- "github.com/ethereum/go-ethereum/metrics"
- "github.com/ethereum/go-ethereum/p2p/enode"
- "github.com/ethereum/go-ethereum/p2p/enr"
- "github.com/ethereum/go-ethereum/rlp"
- )
- var (
- ErrShuttingDown = errors.New("shutting down")
- )
- const (
- baseProtocolVersion = 5
- baseProtocolLength = uint64(16)
- baseProtocolMaxMsgSize = 2 * 1024
- snappyProtocolVersion = 5
- pingInterval = 15 * time.Second
- )
- const (
- // devp2p message codes
- handshakeMsg = 0x00
- discMsg = 0x01
- pingMsg = 0x02
- pongMsg = 0x03
- )
- // protoHandshake is the RLP structure of the protocol handshake.
- type protoHandshake struct {
- Version uint64
- Name string
- Caps []Cap
- ListenPort uint64
- ID []byte // secp256k1 public key
- // Ignore additional fields (for forward compatibility).
- Rest []rlp.RawValue `rlp:"tail"`
- }
- // PeerEventType is the type of peer events emitted by a p2p.Server
- type PeerEventType string
- const (
- // PeerEventTypeAdd is the type of event emitted when a peer is added
- // to a p2p.Server
- PeerEventTypeAdd PeerEventType = "add"
- // PeerEventTypeDrop is the type of event emitted when a peer is
- // dropped from a p2p.Server
- PeerEventTypeDrop PeerEventType = "drop"
- // PeerEventTypeMsgSend is the type of event emitted when a
- // message is successfully sent to a peer
- PeerEventTypeMsgSend PeerEventType = "msgsend"
- // PeerEventTypeMsgRecv is the type of event emitted when a
- // message is received from a peer
- PeerEventTypeMsgRecv PeerEventType = "msgrecv"
- )
- // PeerEvent is an event emitted when peers are either added or dropped from
- // a p2p.Server or when a message is sent or received on a peer connection
- type PeerEvent struct {
- Type PeerEventType `json:"type"`
- Peer enode.ID `json:"peer"`
- Error string `json:"error,omitempty"`
- Protocol string `json:"protocol,omitempty"`
- MsgCode *uint64 `json:"msg_code,omitempty"`
- MsgSize *uint32 `json:"msg_size,omitempty"`
- LocalAddress string `json:"local,omitempty"`
- RemoteAddress string `json:"remote,omitempty"`
- }
- // Peer represents a connected remote node.
- type Peer struct {
- rw *conn
- running map[string]*protoRW
- log log.Logger
- created mclock.AbsTime
- wg sync.WaitGroup
- protoErr chan error
- closed chan struct{}
- disc chan DiscReason
- // events receives message send / receive events if set
- events *event.Feed
- // Quorum
- testPipe *MsgPipeRW // for testing
- EthPeerRegistered chan struct{}
- EthPeerDisconnected chan struct{}
- }
- // NewPeer returns a peer for testing purposes.
- func NewPeer(id enode.ID, name string, caps []Cap) *Peer {
- pipe, _ := net.Pipe()
- node := enode.SignNull(new(enr.Record), id)
- conn := &conn{fd: pipe, transport: nil, node: node, caps: caps, name: name}
- peer := newPeer(log.Root(), conn, nil)
- close(peer.closed) // ensures Disconnect doesn't block
- return peer
- }
- // ID returns the node's public key.
- func (p *Peer) ID() enode.ID {
- return p.rw.node.ID()
- }
- // Node returns the peer's node descriptor.
- func (p *Peer) Node() *enode.Node {
- return p.rw.node
- }
- // Name returns an abbreviated form of the name
- func (p *Peer) Name() string {
- s := p.rw.name
- if len(s) > 20 {
- return s[:20] + "..."
- }
- return s
- }
- // Fullname returns the node name that the remote node advertised.
- func (p *Peer) Fullname() string {
- return p.rw.name
- }
- // Caps returns the capabilities (supported subprotocols) of the remote peer.
- func (p *Peer) Caps() []Cap {
- // TODO: maybe return copy
- return p.rw.caps
- }
- // RunningCap returns true if the peer is actively connected using any of the
- // enumerated versions of a specific protocol, meaning that at least one of the
- // versions is supported by both this node and the peer p.
- func (p *Peer) RunningCap(protocol string, versions []uint) bool {
- if proto, ok := p.running[protocol]; ok {
- for _, ver := range versions {
- if proto.Version == ver {
- return true
- }
- }
- }
- return false
- }
- // RemoteAddr returns the remote address of the network connection.
- func (p *Peer) RemoteAddr() net.Addr {
- return p.rw.fd.RemoteAddr()
- }
- // LocalAddr returns the local address of the network connection.
- func (p *Peer) LocalAddr() net.Addr {
- return p.rw.fd.LocalAddr()
- }
- // Disconnect terminates the peer connection with the given reason.
- // It returns immediately and does not wait until the connection is closed.
- func (p *Peer) Disconnect(reason DiscReason) {
- if p.testPipe != nil {
- p.testPipe.Close()
- }
- select {
- case p.disc <- reason:
- case <-p.closed:
- }
- // Quorum
- // if a quorum eth service subprotocol is waiting on EthPeerRegistered, notify the peer that it was not registered.
- select {
- case p.EthPeerDisconnected <- struct{}{}:
- default:
- }
- // Quorum
- }
- // String implements fmt.Stringer.
- func (p *Peer) String() string {
- id := p.ID()
- return fmt.Sprintf("Peer %x %v", id[:8], p.RemoteAddr())
- }
- // Inbound returns true if the peer is an inbound connection
- func (p *Peer) Inbound() bool {
- return p.rw.is(inboundConn)
- }
- func newPeer(log log.Logger, conn *conn, protocols []Protocol) *Peer {
- protomap := matchProtocols(protocols, conn.caps, conn)
- p := &Peer{
- rw: conn,
- running: protomap,
- created: mclock.Now(),
- disc: make(chan DiscReason),
- protoErr: make(chan error, len(protomap)+1), // protocols + pingLoop
- closed: make(chan struct{}),
- log: log.New("id", conn.node.ID(), "conn", conn.flags),
- // Quorum
- EthPeerRegistered: make(chan struct{}, 1),
- EthPeerDisconnected: make(chan struct{}, 1),
- }
- return p
- }
- func (p *Peer) Log() log.Logger {
- return p.log
- }
- func (p *Peer) run() (remoteRequested bool, err error) {
- var (
- writeStart = make(chan struct{}, 1)
- writeErr = make(chan error, 1)
- readErr = make(chan error, 1)
- reason DiscReason // sent to the peer
- )
- p.wg.Add(2)
- go p.readLoop(readErr)
- go p.pingLoop()
- // Start all protocol handlers.
- writeStart <- struct{}{}
- p.startProtocols(writeStart, writeErr)
- // Wait for an error or disconnect.
- loop:
- for {
- select {
- case err = <-writeErr:
- // A write finished. Allow the next write to start if
- // there was no error.
- if err != nil {
- reason = DiscNetworkError
- break loop
- }
- writeStart <- struct{}{}
- case err = <-readErr:
- if r, ok := err.(DiscReason); ok {
- remoteRequested = true
- reason = r
- } else {
- reason = DiscNetworkError
- }
- break loop
- case err = <-p.protoErr:
- reason = discReasonForError(err)
- break loop
- case err = <-p.disc:
- reason = discReasonForError(err)
- break loop
- }
- }
- close(p.closed)
- p.rw.close(reason)
- p.wg.Wait()
- return remoteRequested, err
- }
- func (p *Peer) pingLoop() {
- ping := time.NewTimer(pingInterval)
- defer p.wg.Done()
- defer ping.Stop()
- for {
- select {
- case <-ping.C:
- if err := SendItems(p.rw, pingMsg); err != nil {
- p.protoErr <- err
- return
- }
- ping.Reset(pingInterval)
- case <-p.closed:
- return
- }
- }
- }
- func (p *Peer) readLoop(errc chan<- error) {
- defer p.wg.Done()
- for {
- msg, err := p.rw.ReadMsg()
- if err != nil {
- errc <- err
- return
- }
- msg.ReceivedAt = time.Now()
- if err = p.handle(msg); err != nil {
- errc <- err
- return
- }
- }
- }
- func (p *Peer) handle(msg Msg) error {
- switch {
- case msg.Code == pingMsg:
- msg.Discard()
- go SendItems(p.rw, pongMsg)
- case msg.Code == discMsg:
- var reason [1]DiscReason
- // This is the last message. We don't need to discard or
- // check errors because, the connection will be closed after it.
- rlp.Decode(msg.Payload, &reason)
- return reason[0]
- case msg.Code < baseProtocolLength:
- // ignore other base protocol messages
- return msg.Discard()
- default:
- // it's a subprotocol message
- proto, err := p.getProto(msg.Code)
- if err != nil {
- return fmt.Errorf("msg code out of range: %v", msg.Code)
- }
- if metrics.Enabled {
- m := fmt.Sprintf("%s/%s/%d/%#02x", ingressMeterName, proto.Name, proto.Version, msg.Code-proto.offset)
- metrics.GetOrRegisterMeter(m, nil).Mark(int64(msg.meterSize))
- metrics.GetOrRegisterMeter(m+"/packets", nil).Mark(1)
- }
- select {
- case proto.in <- msg:
- return nil
- case <-p.closed:
- return io.EOF
- }
- }
- return nil
- }
- func countMatchingProtocols(protocols []Protocol, caps []Cap) int {
- n := 0
- for _, cap := range caps {
- for _, proto := range protocols {
- if proto.Name == cap.Name && proto.Version == cap.Version {
- n++
- }
- }
- }
- return n
- }
- // matchProtocols creates structures for matching named subprotocols.
- func matchProtocols(protocols []Protocol, caps []Cap, rw MsgReadWriter) map[string]*protoRW {
- sort.Sort(capsByNameAndVersion(caps))
- offset := baseProtocolLength
- result := make(map[string]*protoRW)
- outer:
- for _, cap := range caps {
- for _, proto := range protocols {
- if proto.Name == cap.Name && proto.Version == cap.Version {
- // If an old protocol version matched, revert it
- if old := result[cap.Name]; old != nil {
- offset -= old.Length
- }
- // Assign the new match
- result[cap.Name] = &protoRW{Protocol: proto, offset: offset, in: make(chan Msg), w: rw}
- offset += proto.Length
- continue outer
- }
- }
- }
- return result
- }
- func (p *Peer) startProtocols(writeStart <-chan struct{}, writeErr chan<- error) {
- p.wg.Add(len(p.running))
- for _, proto := range p.running {
- proto := proto
- proto.closed = p.closed
- proto.wstart = writeStart
- proto.werr = writeErr
- var rw MsgReadWriter = proto
- if p.events != nil {
- rw = newMsgEventer(rw, p.events, p.ID(), proto.Name, p.Info().Network.RemoteAddress, p.Info().Network.LocalAddress)
- }
- p.log.Trace(fmt.Sprintf("Starting protocol %s/%d", proto.Name, proto.Version))
- go func() {
- defer p.wg.Done()
- err := proto.Run(p, rw)
- if err == nil {
- p.log.Trace(fmt.Sprintf("Protocol %s/%d returned", proto.Name, proto.Version))
- err = errProtocolReturned
- } else if err != io.EOF {
- p.log.Trace(fmt.Sprintf("Protocol %s/%d failed", proto.Name, proto.Version), "err", err)
- }
- p.protoErr <- err
- }()
- }
- }
- // getProto finds the protocol responsible for handling
- // the given message code.
- func (p *Peer) getProto(code uint64) (*protoRW, error) {
- for _, proto := range p.running {
- if code >= proto.offset && code < proto.offset+proto.Length {
- return proto, nil
- }
- }
- return nil, newPeerError(errInvalidMsgCode, "%d", code)
- }
- type protoRW struct {
- Protocol
- in chan Msg // receives read messages
- closed <-chan struct{} // receives when peer is shutting down
- wstart <-chan struct{} // receives when write may start
- werr chan<- error // for write results
- offset uint64
- w MsgWriter
- }
- func (rw *protoRW) WriteMsg(msg Msg) (err error) {
- if msg.Code >= rw.Length {
- return newPeerError(errInvalidMsgCode, "not handled")
- }
- msg.meterCap = rw.cap()
- msg.meterCode = msg.Code
- msg.Code += rw.offset
- select {
- case <-rw.wstart:
- err = rw.w.WriteMsg(msg)
- // Report write status back to Peer.run. It will initiate
- // shutdown if the error is non-nil and unblock the next write
- // otherwise. The calling protocol code should exit for errors
- // as well but we don't want to rely on that.
- rw.werr <- err
- case <-rw.closed:
- err = ErrShuttingDown
- }
- return err
- }
- func (rw *protoRW) ReadMsg() (Msg, error) {
- select {
- case msg := <-rw.in:
- msg.Code -= rw.offset
- return msg, nil
- case <-rw.closed:
- return Msg{}, io.EOF
- }
- }
- // PeerInfo represents a short summary of the information known about a connected
- // peer. Sub-protocol independent fields are contained and initialized here, with
- // protocol specifics delegated to all connected sub-protocols.
- type PeerInfo struct {
- ENR string `json:"enr,omitempty"` // Ethereum Node Record
- Enode string `json:"enode"` // Node URL
- ID string `json:"id"` // Unique node identifier
- Name string `json:"name"` // Name of the node, including client type, version, OS, custom data
- Caps []string `json:"caps"` // Protocols advertised by this peer
- Network struct {
- LocalAddress string `json:"localAddress"` // Local endpoint of the TCP data connection
- RemoteAddress string `json:"remoteAddress"` // Remote endpoint of the TCP data connection
- Inbound bool `json:"inbound"`
- Trusted bool `json:"trusted"`
- Static bool `json:"static"`
- } `json:"network"`
- Protocols map[string]interface{} `json:"protocols"` // Sub-protocol specific metadata fields
- }
- // Info gathers and returns a collection of metadata known about a peer.
- func (p *Peer) Info() *PeerInfo {
- // Gather the protocol capabilities
- var caps []string
- for _, cap := range p.Caps() {
- caps = append(caps, cap.String())
- }
- // Assemble the generic peer metadata
- info := &PeerInfo{
- Enode: p.Node().URLv4(),
- ID: p.ID().String(),
- Name: p.Fullname(),
- Caps: caps,
- Protocols: make(map[string]interface{}),
- }
- if p.Node().Seq() > 0 {
- info.ENR = p.Node().String()
- }
- info.Network.LocalAddress = p.LocalAddr().String()
- info.Network.RemoteAddress = p.RemoteAddr().String()
- info.Network.Inbound = p.rw.is(inboundConn)
- info.Network.Trusted = p.rw.is(trustedConn)
- info.Network.Static = p.rw.is(staticDialedConn)
- // Gather all the running protocol infos
- for _, proto := range p.running {
- protoInfo := interface{}("unknown")
- if query := proto.Protocol.PeerInfo; query != nil {
- if metadata := query(p.ID()); metadata != nil {
- protoInfo = metadata
- } else {
- protoInfo = "handshake"
- }
- }
- info.Protocols[proto.Name] = protoInfo
- }
- return info
- }
- // Quorum
- // NewPeerPipe creates a peer for testing purposes.
- // The message pipe given as the last parameter is closed when
- // Disconnect is called on the peer.
- func NewPeerPipe(id enode.ID, name string, caps []Cap, pipe *MsgPipeRW) *Peer {
- p := NewPeer(id, name, caps)
- p.testPipe = pipe
- return p
- }
|