peer.go 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462
  1. // Copyright 2014 The go-ethereum Authors
  2. // This file is part of the go-ethereum library.
  3. //
  4. // The go-ethereum library is free software: you can redistribute it and/or modify
  5. // it under the terms of the GNU Lesser General Public License as published by
  6. // the Free Software Foundation, either version 3 of the License, or
  7. // (at your option) any later version.
  8. //
  9. // The go-ethereum library is distributed in the hope that it will be useful,
  10. // but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. // GNU Lesser General Public License for more details.
  13. //
  14. // You should have received a copy of the GNU Lesser General Public License
  15. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
  16. package p2p
  17. import (
  18. "fmt"
  19. "io"
  20. "net"
  21. "sort"
  22. "sync"
  23. "time"
  24. "github.com/ethereum/go-ethereum/common/mclock"
  25. "github.com/ethereum/go-ethereum/event"
  26. "github.com/ethereum/go-ethereum/log"
  27. "github.com/ethereum/go-ethereum/p2p/discover"
  28. "github.com/ethereum/go-ethereum/rlp"
  29. )
  30. const (
  31. baseProtocolVersion = 5
  32. baseProtocolLength = uint64(16)
  33. baseProtocolMaxMsgSize = 2 * 1024
  34. snappyProtocolVersion = 5
  35. pingInterval = 15 * time.Second
  36. )
  37. const (
  38. // devp2p message codes
  39. handshakeMsg = 0x00
  40. discMsg = 0x01
  41. pingMsg = 0x02
  42. pongMsg = 0x03
  43. )
  44. // protoHandshake is the RLP structure of the protocol handshake.
  45. type protoHandshake struct {
  46. Version uint64
  47. Name string
  48. Caps []Cap
  49. ListenPort uint64
  50. ID discover.NodeID
  51. // Ignore additional fields (for forward compatibility).
  52. Rest []rlp.RawValue `rlp:"tail"`
  53. }
  54. // PeerEventType is the type of peer events emitted by a p2p.Server
  55. type PeerEventType string
  56. const (
  57. // PeerEventTypeAdd is the type of event emitted when a peer is added
  58. // to a p2p.Server
  59. PeerEventTypeAdd PeerEventType = "add"
  60. // PeerEventTypeDrop is the type of event emitted when a peer is
  61. // dropped from a p2p.Server
  62. PeerEventTypeDrop PeerEventType = "drop"
  63. // PeerEventTypeMsgSend is the type of event emitted when a
  64. // message is successfully sent to a peer
  65. PeerEventTypeMsgSend PeerEventType = "msgsend"
  66. // PeerEventTypeMsgRecv is the type of event emitted when a
  67. // message is received from a peer
  68. PeerEventTypeMsgRecv PeerEventType = "msgrecv"
  69. )
  70. // PeerEvent is an event emitted when peers are either added or dropped from
  71. // a p2p.Server or when a message is sent or received on a peer connection
  72. type PeerEvent struct {
  73. Type PeerEventType `json:"type"`
  74. Peer discover.NodeID `json:"peer"`
  75. Error string `json:"error,omitempty"`
  76. Protocol string `json:"protocol,omitempty"`
  77. MsgCode *uint64 `json:"msg_code,omitempty"`
  78. MsgSize *uint32 `json:"msg_size,omitempty"`
  79. }
  80. // Peer represents a connected remote node.
  81. type Peer struct {
  82. rw *conn
  83. running map[string]*protoRW
  84. log log.Logger
  85. created mclock.AbsTime
  86. wg sync.WaitGroup
  87. protoErr chan error
  88. closed chan struct{}
  89. disc chan DiscReason
  90. // events receives message send / receive events if set
  91. events *event.Feed
  92. }
  93. // NewPeer returns a peer for testing purposes.
  94. func NewPeer(id discover.NodeID, name string, caps []Cap) *Peer {
  95. pipe, _ := net.Pipe()
  96. conn := &conn{fd: pipe, transport: nil, id: id, caps: caps, name: name}
  97. peer := newPeer(conn, nil)
  98. close(peer.closed) // ensures Disconnect doesn't block
  99. return peer
  100. }
  101. // ID returns the node's public key.
  102. func (p *Peer) ID() discover.NodeID {
  103. return p.rw.id
  104. }
  105. // Name returns the node name that the remote node advertised.
  106. func (p *Peer) Name() string {
  107. return p.rw.name
  108. }
  109. // Caps returns the capabilities (supported subprotocols) of the remote peer.
  110. func (p *Peer) Caps() []Cap {
  111. // TODO: maybe return copy
  112. return p.rw.caps
  113. }
  114. // RemoteAddr returns the remote address of the network connection.
  115. func (p *Peer) RemoteAddr() net.Addr {
  116. return p.rw.fd.RemoteAddr()
  117. }
  118. // LocalAddr returns the local address of the network connection.
  119. func (p *Peer) LocalAddr() net.Addr {
  120. return p.rw.fd.LocalAddr()
  121. }
  122. // Disconnect terminates the peer connection with the given reason.
  123. // It returns immediately and does not wait until the connection is closed.
  124. func (p *Peer) Disconnect(reason DiscReason) {
  125. select {
  126. case p.disc <- reason:
  127. case <-p.closed:
  128. }
  129. }
  130. // String implements fmt.Stringer.
  131. func (p *Peer) String() string {
  132. return fmt.Sprintf("Peer %x %v", p.rw.id[:8], p.RemoteAddr())
  133. }
  134. // Inbound returns true if the peer is an inbound connection
  135. func (p *Peer) Inbound() bool {
  136. return p.rw.flags&inboundConn != 0
  137. }
  138. func newPeer(conn *conn, protocols []Protocol) *Peer {
  139. protomap := matchProtocols(protocols, conn.caps, conn)
  140. p := &Peer{
  141. rw: conn,
  142. running: protomap,
  143. created: mclock.Now(),
  144. disc: make(chan DiscReason),
  145. protoErr: make(chan error, len(protomap)+1), // protocols + pingLoop
  146. closed: make(chan struct{}),
  147. log: log.New("id", conn.id, "conn", conn.flags),
  148. }
  149. return p
  150. }
  151. func (p *Peer) Log() log.Logger {
  152. return p.log
  153. }
  154. func (p *Peer) run() (remoteRequested bool, err error) {
  155. var (
  156. writeStart = make(chan struct{}, 1)
  157. writeErr = make(chan error, 1)
  158. readErr = make(chan error, 1)
  159. reason DiscReason // sent to the peer
  160. )
  161. p.wg.Add(2)
  162. go p.readLoop(readErr)
  163. go p.pingLoop()
  164. // Start all protocol handlers.
  165. writeStart <- struct{}{}
  166. p.startProtocols(writeStart, writeErr)
  167. // Wait for an error or disconnect.
  168. loop:
  169. for {
  170. select {
  171. case err = <-writeErr:
  172. // A write finished. Allow the next write to start if
  173. // there was no error.
  174. if err != nil {
  175. reason = DiscNetworkError
  176. break loop
  177. }
  178. writeStart <- struct{}{}
  179. case err = <-readErr:
  180. if r, ok := err.(DiscReason); ok {
  181. remoteRequested = true
  182. reason = r
  183. } else {
  184. reason = DiscNetworkError
  185. }
  186. break loop
  187. case err = <-p.protoErr:
  188. reason = discReasonForError(err)
  189. break loop
  190. case err = <-p.disc:
  191. reason = discReasonForError(err)
  192. break loop
  193. }
  194. }
  195. close(p.closed)
  196. p.rw.close(reason)
  197. p.wg.Wait()
  198. return remoteRequested, err
  199. }
  200. func (p *Peer) pingLoop() {
  201. ping := time.NewTimer(pingInterval)
  202. defer p.wg.Done()
  203. defer ping.Stop()
  204. for {
  205. select {
  206. case <-ping.C:
  207. if err := SendItems(p.rw, pingMsg); err != nil {
  208. p.protoErr <- err
  209. return
  210. }
  211. ping.Reset(pingInterval)
  212. case <-p.closed:
  213. return
  214. }
  215. }
  216. }
  217. func (p *Peer) readLoop(errc chan<- error) {
  218. defer p.wg.Done()
  219. for {
  220. msg, err := p.rw.ReadMsg()
  221. if err != nil {
  222. errc <- err
  223. return
  224. }
  225. msg.ReceivedAt = time.Now()
  226. if err = p.handle(msg); err != nil {
  227. errc <- err
  228. return
  229. }
  230. }
  231. }
  232. func (p *Peer) handle(msg Msg) error {
  233. switch {
  234. case msg.Code == pingMsg:
  235. msg.Discard()
  236. go SendItems(p.rw, pongMsg)
  237. case msg.Code == discMsg:
  238. var reason [1]DiscReason
  239. // This is the last message. We don't need to discard or
  240. // check errors because, the connection will be closed after it.
  241. rlp.Decode(msg.Payload, &reason)
  242. return reason[0]
  243. case msg.Code < baseProtocolLength:
  244. // ignore other base protocol messages
  245. return msg.Discard()
  246. default:
  247. // it's a subprotocol message
  248. proto, err := p.getProto(msg.Code)
  249. if err != nil {
  250. return fmt.Errorf("msg code out of range: %v", msg.Code)
  251. }
  252. select {
  253. case proto.in <- msg:
  254. return nil
  255. case <-p.closed:
  256. return io.EOF
  257. }
  258. }
  259. return nil
  260. }
  261. func countMatchingProtocols(protocols []Protocol, caps []Cap) int {
  262. n := 0
  263. for _, cap := range caps {
  264. for _, proto := range protocols {
  265. if proto.Name == cap.Name && proto.Version == cap.Version {
  266. n++
  267. }
  268. }
  269. }
  270. return n
  271. }
  272. // matchProtocols creates structures for matching named subprotocols.
  273. func matchProtocols(protocols []Protocol, caps []Cap, rw MsgReadWriter) map[string]*protoRW {
  274. sort.Sort(capsByNameAndVersion(caps))
  275. offset := baseProtocolLength
  276. result := make(map[string]*protoRW)
  277. outer:
  278. for _, cap := range caps {
  279. for _, proto := range protocols {
  280. if proto.Name == cap.Name && proto.Version == cap.Version {
  281. // If an old protocol version matched, revert it
  282. if old := result[cap.Name]; old != nil {
  283. offset -= old.Length
  284. }
  285. // Assign the new match
  286. result[cap.Name] = &protoRW{Protocol: proto, offset: offset, in: make(chan Msg), w: rw}
  287. offset += proto.Length
  288. continue outer
  289. }
  290. }
  291. }
  292. return result
  293. }
  294. func (p *Peer) startProtocols(writeStart <-chan struct{}, writeErr chan<- error) {
  295. p.wg.Add(len(p.running))
  296. for _, proto := range p.running {
  297. proto := proto
  298. proto.closed = p.closed
  299. proto.wstart = writeStart
  300. proto.werr = writeErr
  301. var rw MsgReadWriter = proto
  302. if p.events != nil {
  303. rw = newMsgEventer(rw, p.events, p.ID(), proto.Name)
  304. }
  305. p.log.Trace(fmt.Sprintf("Starting protocol %s/%d", proto.Name, proto.Version))
  306. go func() {
  307. err := proto.Run(p, rw)
  308. if err == nil {
  309. p.log.Trace(fmt.Sprintf("Protocol %s/%d returned", proto.Name, proto.Version))
  310. err = errProtocolReturned
  311. } else if err != io.EOF {
  312. p.log.Trace(fmt.Sprintf("Protocol %s/%d failed", proto.Name, proto.Version), "err", err)
  313. }
  314. p.protoErr <- err
  315. p.wg.Done()
  316. }()
  317. }
  318. }
  319. // getProto finds the protocol responsible for handling
  320. // the given message code.
  321. func (p *Peer) getProto(code uint64) (*protoRW, error) {
  322. for _, proto := range p.running {
  323. if code >= proto.offset && code < proto.offset+proto.Length {
  324. return proto, nil
  325. }
  326. }
  327. return nil, newPeerError(errInvalidMsgCode, "%d", code)
  328. }
  329. type protoRW struct {
  330. Protocol
  331. in chan Msg // receices read messages
  332. closed <-chan struct{} // receives when peer is shutting down
  333. wstart <-chan struct{} // receives when write may start
  334. werr chan<- error // for write results
  335. offset uint64
  336. w MsgWriter
  337. }
  338. func (rw *protoRW) WriteMsg(msg Msg) (err error) {
  339. if msg.Code >= rw.Length {
  340. return newPeerError(errInvalidMsgCode, "not handled")
  341. }
  342. msg.Code += rw.offset
  343. select {
  344. case <-rw.wstart:
  345. err = rw.w.WriteMsg(msg)
  346. // Report write status back to Peer.run. It will initiate
  347. // shutdown if the error is non-nil and unblock the next write
  348. // otherwise. The calling protocol code should exit for errors
  349. // as well but we don't want to rely on that.
  350. rw.werr <- err
  351. case <-rw.closed:
  352. err = fmt.Errorf("shutting down")
  353. }
  354. return err
  355. }
  356. func (rw *protoRW) ReadMsg() (Msg, error) {
  357. select {
  358. case msg := <-rw.in:
  359. msg.Code -= rw.offset
  360. return msg, nil
  361. case <-rw.closed:
  362. return Msg{}, io.EOF
  363. }
  364. }
  365. // PeerInfo represents a short summary of the information known about a connected
  366. // peer. Sub-protocol independent fields are contained and initialized here, with
  367. // protocol specifics delegated to all connected sub-protocols.
  368. type PeerInfo struct {
  369. ID string `json:"id"` // Unique node identifier (also the encryption key)
  370. Name string `json:"name"` // Name of the node, including client type, version, OS, custom data
  371. Caps []string `json:"caps"` // Sum-protocols advertised by this particular peer
  372. Network struct {
  373. LocalAddress string `json:"localAddress"` // Local endpoint of the TCP data connection
  374. RemoteAddress string `json:"remoteAddress"` // Remote endpoint of the TCP data connection
  375. Inbound bool `json:"inbound"`
  376. Trusted bool `json:"trusted"`
  377. Static bool `json:"static"`
  378. } `json:"network"`
  379. Protocols map[string]interface{} `json:"protocols"` // Sub-protocol specific metadata fields
  380. }
  381. // Info gathers and returns a collection of metadata known about a peer.
  382. func (p *Peer) Info() *PeerInfo {
  383. // Gather the protocol capabilities
  384. var caps []string
  385. for _, cap := range p.Caps() {
  386. caps = append(caps, cap.String())
  387. }
  388. // Assemble the generic peer metadata
  389. info := &PeerInfo{
  390. ID: p.ID().String(),
  391. Name: p.Name(),
  392. Caps: caps,
  393. Protocols: make(map[string]interface{}),
  394. }
  395. info.Network.LocalAddress = p.LocalAddr().String()
  396. info.Network.RemoteAddress = p.RemoteAddr().String()
  397. info.Network.Inbound = p.rw.is(inboundConn)
  398. info.Network.Trusted = p.rw.is(trustedConn)
  399. info.Network.Static = p.rw.is(staticDialedConn)
  400. // Gather all the running protocol infos
  401. for _, proto := range p.running {
  402. protoInfo := interface{}("unknown")
  403. if query := proto.Protocol.PeerInfo; query != nil {
  404. if metadata := query(p.ID()); metadata != nil {
  405. protoInfo = metadata
  406. } else {
  407. protoInfo = "handshake"
  408. }
  409. }
  410. info.Protocols[proto.Name] = protoInfo
  411. }
  412. return info
  413. }