path: root/p2p/peer.go

// 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 (
    "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/p2p/discover"
    "github.com/ethereum/go-ethereum/rlp"
)

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         discover.NodeID

    // 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     discover.NodeID `json:"peer"`
    Error    string          `json:"error,omitempty"`
    Protocol string          `json:"protocol,omitempty"`
    MsgCode  *uint64         `json:"msg_code,omitempty"`
    MsgSize  *uint32         `json:"msg_size,omitempty"`
}

// Peer represents a connected remote node.
type Peer struct {
    rw        *conn
    isInbound bool // Cached from rw.flags to avoid a race condition
    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
}

// NewPeer returns a peer for testing purposes.
func NewPeer(id discover.NodeID, name string, caps []Cap) *Peer {
    pipe, _ := net.Pipe()
    conn := &conn{fd: pipe, transport: nil, id: id, caps: caps, name: name}
    peer := newPeer(conn, nil)
    close(peer.closed) // ensures Disconnect doesn't block
    return peer
}

// ID returns the node's public key.
func (p *Peer) ID() discover.NodeID {
    return p.rw.id
}

// Name returns the node name that the remote node advertised.
func (p *Peer) Name() 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
}

// 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) {
    select {
    case p.disc <- reason:
    case <-p.closed:
    }
}

// String implements fmt.Stringer.
func (p *Peer) String() string {
    return fmt.Sprintf("Peer %x %v", p.rw.id[:8], p.RemoteAddr())
}

// Inbound returns true if the peer is an inbound connection
func (p *Peer) Inbound() bool {
    return p.isInbound
}

func newPeer(conn *conn, protocols []Protocol) *Peer {
    protomap := matchProtocols(protocols, conn.caps, conn)
    p := &Peer{
        rw:        conn,
        isInbound: conn.is(inboundConn),
        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.id, "conn", conn.flags),
    }
    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)
        }
        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.log.Trace(fmt.Sprintf("Starting protocol %s/%d", proto.Name, proto.Version))
        go func() {
            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
            p.wg.Done()
        }()
    }
}

// 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        // receices 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.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 = fmt.Errorf("shutting down")
    }
    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 {
    ID      string   `json:"id"`   // Unique node identifier (also the encryption key)
    Name    string   `json:"name"` // Name of the node, including client type, version, OS, custom data
    Caps    []string `json:"caps"` // Sum-protocols advertised by this particular 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{
        ID:        p.ID().String(),
        Name:      p.Name(),
        Caps:      caps,
        Protocols: make(map[string]interface{}),
    }
    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
}