123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462 |
- // 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
- 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.rw.flags&inboundConn != 0
- }
- func newPeer(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.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
- }
|