server.go 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806
  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 implements the Ethereum p2p network protocols.
  17. package p2p
  18. import (
  19. "crypto/ecdsa"
  20. "errors"
  21. "fmt"
  22. "net"
  23. "sync"
  24. "time"
  25. "github.com/ethereum/go-ethereum/logger"
  26. "github.com/ethereum/go-ethereum/logger/glog"
  27. "github.com/ethereum/go-ethereum/p2p/discover"
  28. "github.com/ethereum/go-ethereum/p2p/discv5"
  29. "github.com/ethereum/go-ethereum/p2p/nat"
  30. )
  31. const (
  32. defaultDialTimeout = 15 * time.Second
  33. refreshPeersInterval = 30 * time.Second
  34. staticPeerCheckInterval = 15 * time.Second
  35. // Maximum number of concurrently handshaking inbound connections.
  36. maxAcceptConns = 50
  37. // Maximum number of concurrently dialing outbound connections.
  38. maxActiveDialTasks = 16
  39. // Maximum time allowed for reading a complete message.
  40. // This is effectively the amount of time a connection can be idle.
  41. frameReadTimeout = 30 * time.Second
  42. // Maximum amount of time allowed for writing a complete message.
  43. frameWriteTimeout = 20 * time.Second
  44. )
  45. var errServerStopped = errors.New("server stopped")
  46. var srvjslog = logger.NewJsonLogger()
  47. // Config holds Server options.
  48. type Config struct {
  49. // This field must be set to a valid secp256k1 private key.
  50. PrivateKey *ecdsa.PrivateKey
  51. // MaxPeers is the maximum number of peers that can be
  52. // connected. It must be greater than zero.
  53. MaxPeers int
  54. // MaxPendingPeers is the maximum number of peers that can be pending in the
  55. // handshake phase, counted separately for inbound and outbound connections.
  56. // Zero defaults to preset values.
  57. MaxPendingPeers int
  58. // Discovery specifies whether the peer discovery mechanism should be started
  59. // or not. Disabling is usually useful for protocol debugging (manual topology).
  60. Discovery bool
  61. // DiscoveryV5 specifies whether the the new topic-discovery based V5 discovery
  62. // protocol should be started or not.
  63. DiscoveryV5 bool
  64. // Listener address for the V5 discovery protocol UDP traffic.
  65. DiscoveryV5Addr string
  66. // Name sets the node name of this server.
  67. // Use common.MakeName to create a name that follows existing conventions.
  68. Name string
  69. // BootstrapNodes are used to establish connectivity
  70. // with the rest of the network.
  71. BootstrapNodes []*discover.Node
  72. // BootstrapNodesV5 are used to establish connectivity
  73. // with the rest of the network using the V5 discovery
  74. // protocol.
  75. BootstrapNodesV5 []*discv5.Node
  76. // Static nodes are used as pre-configured connections which are always
  77. // maintained and re-connected on disconnects.
  78. StaticNodes []*discover.Node
  79. // Trusted nodes are used as pre-configured connections which are always
  80. // allowed to connect, even above the peer limit.
  81. TrustedNodes []*discover.Node
  82. // NodeDatabase is the path to the database containing the previously seen
  83. // live nodes in the network.
  84. NodeDatabase string
  85. // Protocols should contain the protocols supported
  86. // by the server. Matching protocols are launched for
  87. // each peer.
  88. Protocols []Protocol
  89. // If ListenAddr is set to a non-nil address, the server
  90. // will listen for incoming connections.
  91. //
  92. // If the port is zero, the operating system will pick a port. The
  93. // ListenAddr field will be updated with the actual address when
  94. // the server is started.
  95. ListenAddr string
  96. // If set to a non-nil value, the given NAT port mapper
  97. // is used to make the listening port available to the
  98. // Internet.
  99. NAT nat.Interface
  100. // If Dialer is set to a non-nil value, the given Dialer
  101. // is used to dial outbound peer connections.
  102. Dialer *net.Dialer
  103. // If NoDial is true, the server will not dial any peers.
  104. NoDial bool
  105. }
  106. // Server manages all peer connections.
  107. type Server struct {
  108. // Config fields may not be modified while the server is running.
  109. Config
  110. // Hooks for testing. These are useful because we can inhibit
  111. // the whole protocol stack.
  112. newTransport func(net.Conn) transport
  113. newPeerHook func(*Peer)
  114. lock sync.Mutex // protects running
  115. running bool
  116. ntab discoverTable
  117. listener net.Listener
  118. ourHandshake *protoHandshake
  119. lastLookup time.Time
  120. DiscV5 *discv5.Network
  121. // These are for Peers, PeerCount (and nothing else).
  122. peerOp chan peerOpFunc
  123. peerOpDone chan struct{}
  124. quit chan struct{}
  125. addstatic chan *discover.Node
  126. removestatic chan *discover.Node
  127. posthandshake chan *conn
  128. addpeer chan *conn
  129. delpeer chan *Peer
  130. loopWG sync.WaitGroup // loop, listenLoop
  131. }
  132. type peerOpFunc func(map[discover.NodeID]*Peer)
  133. type connFlag int
  134. const (
  135. dynDialedConn connFlag = 1 << iota
  136. staticDialedConn
  137. inboundConn
  138. trustedConn
  139. )
  140. // conn wraps a network connection with information gathered
  141. // during the two handshakes.
  142. type conn struct {
  143. fd net.Conn
  144. transport
  145. flags connFlag
  146. cont chan error // The run loop uses cont to signal errors to setupConn.
  147. id discover.NodeID // valid after the encryption handshake
  148. caps []Cap // valid after the protocol handshake
  149. name string // valid after the protocol handshake
  150. }
  151. type transport interface {
  152. // The two handshakes.
  153. doEncHandshake(prv *ecdsa.PrivateKey, dialDest *discover.Node) (discover.NodeID, error)
  154. doProtoHandshake(our *protoHandshake) (*protoHandshake, error)
  155. // The MsgReadWriter can only be used after the encryption
  156. // handshake has completed. The code uses conn.id to track this
  157. // by setting it to a non-nil value after the encryption handshake.
  158. MsgReadWriter
  159. // transports must provide Close because we use MsgPipe in some of
  160. // the tests. Closing the actual network connection doesn't do
  161. // anything in those tests because NsgPipe doesn't use it.
  162. close(err error)
  163. }
  164. func (c *conn) String() string {
  165. s := c.flags.String() + " conn"
  166. if (c.id != discover.NodeID{}) {
  167. s += fmt.Sprintf(" %x", c.id[:8])
  168. }
  169. s += " " + c.fd.RemoteAddr().String()
  170. return s
  171. }
  172. func (f connFlag) String() string {
  173. s := ""
  174. if f&trustedConn != 0 {
  175. s += " trusted"
  176. }
  177. if f&dynDialedConn != 0 {
  178. s += " dyn dial"
  179. }
  180. if f&staticDialedConn != 0 {
  181. s += " static dial"
  182. }
  183. if f&inboundConn != 0 {
  184. s += " inbound"
  185. }
  186. if s != "" {
  187. s = s[1:]
  188. }
  189. return s
  190. }
  191. func (c *conn) is(f connFlag) bool {
  192. return c.flags&f != 0
  193. }
  194. // Peers returns all connected peers.
  195. func (srv *Server) Peers() []*Peer {
  196. var ps []*Peer
  197. select {
  198. // Note: We'd love to put this function into a variable but
  199. // that seems to cause a weird compiler error in some
  200. // environments.
  201. case srv.peerOp <- func(peers map[discover.NodeID]*Peer) {
  202. for _, p := range peers {
  203. ps = append(ps, p)
  204. }
  205. }:
  206. <-srv.peerOpDone
  207. case <-srv.quit:
  208. }
  209. return ps
  210. }
  211. // PeerCount returns the number of connected peers.
  212. func (srv *Server) PeerCount() int {
  213. var count int
  214. select {
  215. case srv.peerOp <- func(ps map[discover.NodeID]*Peer) { count = len(ps) }:
  216. <-srv.peerOpDone
  217. case <-srv.quit:
  218. }
  219. return count
  220. }
  221. // AddPeer connects to the given node and maintains the connection until the
  222. // server is shut down. If the connection fails for any reason, the server will
  223. // attempt to reconnect the peer.
  224. func (srv *Server) AddPeer(node *discover.Node) {
  225. select {
  226. case srv.addstatic <- node:
  227. case <-srv.quit:
  228. }
  229. }
  230. // RemovePeer disconnects from the given node
  231. func (srv *Server) RemovePeer(node *discover.Node) {
  232. select {
  233. case srv.removestatic <- node:
  234. case <-srv.quit:
  235. }
  236. }
  237. // Self returns the local node's endpoint information.
  238. func (srv *Server) Self() *discover.Node {
  239. srv.lock.Lock()
  240. defer srv.lock.Unlock()
  241. // If the server's not running, return an empty node
  242. if !srv.running {
  243. return &discover.Node{IP: net.ParseIP("0.0.0.0")}
  244. }
  245. // If the node is running but discovery is off, manually assemble the node infos
  246. if srv.ntab == nil {
  247. // Inbound connections disabled, use zero address
  248. if srv.listener == nil {
  249. return &discover.Node{IP: net.ParseIP("0.0.0.0"), ID: discover.PubkeyID(&srv.PrivateKey.PublicKey)}
  250. }
  251. // Otherwise inject the listener address too
  252. addr := srv.listener.Addr().(*net.TCPAddr)
  253. return &discover.Node{
  254. ID: discover.PubkeyID(&srv.PrivateKey.PublicKey),
  255. IP: addr.IP,
  256. TCP: uint16(addr.Port),
  257. }
  258. }
  259. // Otherwise return the live node infos
  260. return srv.ntab.Self()
  261. }
  262. // Stop terminates the server and all active peer connections.
  263. // It blocks until all active connections have been closed.
  264. func (srv *Server) Stop() {
  265. srv.lock.Lock()
  266. defer srv.lock.Unlock()
  267. if !srv.running {
  268. return
  269. }
  270. srv.running = false
  271. if srv.listener != nil {
  272. // this unblocks listener Accept
  273. srv.listener.Close()
  274. }
  275. close(srv.quit)
  276. srv.loopWG.Wait()
  277. }
  278. // Start starts running the server.
  279. // Servers can not be re-used after stopping.
  280. func (srv *Server) Start() (err error) {
  281. srv.lock.Lock()
  282. defer srv.lock.Unlock()
  283. if srv.running {
  284. return errors.New("server already running")
  285. }
  286. srv.running = true
  287. glog.V(logger.Info).Infoln("Starting Server")
  288. // static fields
  289. if srv.PrivateKey == nil {
  290. return fmt.Errorf("Server.PrivateKey must be set to a non-nil key")
  291. }
  292. if srv.newTransport == nil {
  293. srv.newTransport = newRLPX
  294. }
  295. if srv.Dialer == nil {
  296. srv.Dialer = &net.Dialer{Timeout: defaultDialTimeout}
  297. }
  298. srv.quit = make(chan struct{})
  299. srv.addpeer = make(chan *conn)
  300. srv.delpeer = make(chan *Peer)
  301. srv.posthandshake = make(chan *conn)
  302. srv.addstatic = make(chan *discover.Node)
  303. srv.removestatic = make(chan *discover.Node)
  304. srv.peerOp = make(chan peerOpFunc)
  305. srv.peerOpDone = make(chan struct{})
  306. // node table
  307. if srv.Discovery {
  308. ntab, err := discover.ListenUDP(srv.PrivateKey, srv.ListenAddr, srv.NAT, srv.NodeDatabase)
  309. if err != nil {
  310. return err
  311. }
  312. if err := ntab.SetFallbackNodes(srv.BootstrapNodes); err != nil {
  313. return err
  314. }
  315. srv.ntab = ntab
  316. }
  317. if srv.DiscoveryV5 {
  318. ntab, err := discv5.ListenUDP(srv.PrivateKey, srv.DiscoveryV5Addr, srv.NAT, "") //srv.NodeDatabase)
  319. if err != nil {
  320. return err
  321. }
  322. if err := ntab.SetFallbackNodes(srv.BootstrapNodesV5); err != nil {
  323. return err
  324. }
  325. srv.DiscV5 = ntab
  326. }
  327. dynPeers := (srv.MaxPeers + 1) / 2
  328. if !srv.Discovery {
  329. dynPeers = 0
  330. }
  331. dialer := newDialState(srv.StaticNodes, srv.ntab, dynPeers)
  332. // handshake
  333. srv.ourHandshake = &protoHandshake{Version: baseProtocolVersion, Name: srv.Name, ID: discover.PubkeyID(&srv.PrivateKey.PublicKey)}
  334. for _, p := range srv.Protocols {
  335. srv.ourHandshake.Caps = append(srv.ourHandshake.Caps, p.cap())
  336. }
  337. // listen/dial
  338. if srv.ListenAddr != "" {
  339. if err := srv.startListening(); err != nil {
  340. return err
  341. }
  342. }
  343. if srv.NoDial && srv.ListenAddr == "" {
  344. glog.V(logger.Warn).Infoln("I will be kind-of useless, neither dialing nor listening.")
  345. }
  346. srv.loopWG.Add(1)
  347. go srv.run(dialer)
  348. srv.running = true
  349. return nil
  350. }
  351. func (srv *Server) startListening() error {
  352. // Launch the TCP listener.
  353. listener, err := net.Listen("tcp", srv.ListenAddr)
  354. if err != nil {
  355. return err
  356. }
  357. laddr := listener.Addr().(*net.TCPAddr)
  358. srv.ListenAddr = laddr.String()
  359. srv.listener = listener
  360. srv.loopWG.Add(1)
  361. go srv.listenLoop()
  362. // Map the TCP listening port if NAT is configured.
  363. if !laddr.IP.IsLoopback() && srv.NAT != nil {
  364. srv.loopWG.Add(1)
  365. go func() {
  366. nat.Map(srv.NAT, srv.quit, "tcp", laddr.Port, laddr.Port, "ethereum p2p")
  367. srv.loopWG.Done()
  368. }()
  369. }
  370. return nil
  371. }
  372. type dialer interface {
  373. newTasks(running int, peers map[discover.NodeID]*Peer, now time.Time) []task
  374. taskDone(task, time.Time)
  375. addStatic(*discover.Node)
  376. removeStatic(*discover.Node)
  377. }
  378. func (srv *Server) run(dialstate dialer) {
  379. defer srv.loopWG.Done()
  380. var (
  381. peers = make(map[discover.NodeID]*Peer)
  382. trusted = make(map[discover.NodeID]bool, len(srv.TrustedNodes))
  383. taskdone = make(chan task, maxActiveDialTasks)
  384. runningTasks []task
  385. queuedTasks []task // tasks that can't run yet
  386. )
  387. // Put trusted nodes into a map to speed up checks.
  388. // Trusted peers are loaded on startup and cannot be
  389. // modified while the server is running.
  390. for _, n := range srv.TrustedNodes {
  391. trusted[n.ID] = true
  392. }
  393. // removes t from runningTasks
  394. delTask := func(t task) {
  395. for i := range runningTasks {
  396. if runningTasks[i] == t {
  397. runningTasks = append(runningTasks[:i], runningTasks[i+1:]...)
  398. break
  399. }
  400. }
  401. }
  402. // starts until max number of active tasks is satisfied
  403. startTasks := func(ts []task) (rest []task) {
  404. i := 0
  405. for ; len(runningTasks) < maxActiveDialTasks && i < len(ts); i++ {
  406. t := ts[i]
  407. glog.V(logger.Detail).Infoln("new task:", t)
  408. go func() { t.Do(srv); taskdone <- t }()
  409. runningTasks = append(runningTasks, t)
  410. }
  411. return ts[i:]
  412. }
  413. scheduleTasks := func() {
  414. // Start from queue first.
  415. queuedTasks = append(queuedTasks[:0], startTasks(queuedTasks)...)
  416. // Query dialer for new tasks and start as many as possible now.
  417. if len(runningTasks) < maxActiveDialTasks {
  418. nt := dialstate.newTasks(len(runningTasks)+len(queuedTasks), peers, time.Now())
  419. queuedTasks = append(queuedTasks, startTasks(nt)...)
  420. }
  421. }
  422. running:
  423. for {
  424. scheduleTasks()
  425. select {
  426. case <-srv.quit:
  427. // The server was stopped. Run the cleanup logic.
  428. glog.V(logger.Detail).Infoln("<-quit: spinning down")
  429. break running
  430. case n := <-srv.addstatic:
  431. // This channel is used by AddPeer to add to the
  432. // ephemeral static peer list. Add it to the dialer,
  433. // it will keep the node connected.
  434. glog.V(logger.Detail).Infoln("<-addstatic:", n)
  435. dialstate.addStatic(n)
  436. case n := <-srv.removestatic:
  437. // This channel is used by RemovePeer to send a
  438. // disconnect request to a peer and begin the
  439. // stop keeping the node connected
  440. glog.V(logger.Detail).Infoln("<-removestatic:", n)
  441. dialstate.removeStatic(n)
  442. if p, ok := peers[n.ID]; ok {
  443. p.Disconnect(DiscRequested)
  444. }
  445. case op := <-srv.peerOp:
  446. // This channel is used by Peers and PeerCount.
  447. op(peers)
  448. srv.peerOpDone <- struct{}{}
  449. case t := <-taskdone:
  450. // A task got done. Tell dialstate about it so it
  451. // can update its state and remove it from the active
  452. // tasks list.
  453. glog.V(logger.Detail).Infoln("<-taskdone:", t)
  454. dialstate.taskDone(t, time.Now())
  455. delTask(t)
  456. case c := <-srv.posthandshake:
  457. // A connection has passed the encryption handshake so
  458. // the remote identity is known (but hasn't been verified yet).
  459. if trusted[c.id] {
  460. // Ensure that the trusted flag is set before checking against MaxPeers.
  461. c.flags |= trustedConn
  462. }
  463. glog.V(logger.Detail).Infoln("<-posthandshake:", c)
  464. // TODO: track in-progress inbound node IDs (pre-Peer) to avoid dialing them.
  465. c.cont <- srv.encHandshakeChecks(peers, c)
  466. case c := <-srv.addpeer:
  467. // At this point the connection is past the protocol handshake.
  468. // Its capabilities are known and the remote identity is verified.
  469. glog.V(logger.Detail).Infoln("<-addpeer:", c)
  470. err := srv.protoHandshakeChecks(peers, c)
  471. if err != nil {
  472. glog.V(logger.Detail).Infof("Not adding %v as peer: %v", c, err)
  473. } else {
  474. // The handshakes are done and it passed all checks.
  475. p := newPeer(c, srv.Protocols)
  476. peers[c.id] = p
  477. go srv.runPeer(p)
  478. }
  479. // The dialer logic relies on the assumption that
  480. // dial tasks complete after the peer has been added or
  481. // discarded. Unblock the task last.
  482. c.cont <- err
  483. case p := <-srv.delpeer:
  484. // A peer disconnected.
  485. glog.V(logger.Detail).Infoln("<-delpeer:", p)
  486. delete(peers, p.ID())
  487. }
  488. }
  489. // Terminate discovery. If there is a running lookup it will terminate soon.
  490. if srv.ntab != nil {
  491. srv.ntab.Close()
  492. }
  493. if srv.DiscV5 != nil {
  494. srv.DiscV5.Close()
  495. }
  496. // Disconnect all peers.
  497. for _, p := range peers {
  498. p.Disconnect(DiscQuitting)
  499. }
  500. // Wait for peers to shut down. Pending connections and tasks are
  501. // not handled here and will terminate soon-ish because srv.quit
  502. // is closed.
  503. glog.V(logger.Detail).Infof("ignoring %d pending tasks at spindown", len(runningTasks))
  504. for len(peers) > 0 {
  505. p := <-srv.delpeer
  506. glog.V(logger.Detail).Infoln("<-delpeer (spindown):", p)
  507. delete(peers, p.ID())
  508. }
  509. }
  510. func (srv *Server) protoHandshakeChecks(peers map[discover.NodeID]*Peer, c *conn) error {
  511. // Drop connections with no matching protocols.
  512. if len(srv.Protocols) > 0 && countMatchingProtocols(srv.Protocols, c.caps) == 0 {
  513. return DiscUselessPeer
  514. }
  515. // Repeat the encryption handshake checks because the
  516. // peer set might have changed between the handshakes.
  517. return srv.encHandshakeChecks(peers, c)
  518. }
  519. func (srv *Server) encHandshakeChecks(peers map[discover.NodeID]*Peer, c *conn) error {
  520. switch {
  521. case !c.is(trustedConn|staticDialedConn) && len(peers) >= srv.MaxPeers:
  522. return DiscTooManyPeers
  523. case peers[c.id] != nil:
  524. return DiscAlreadyConnected
  525. case c.id == srv.Self().ID:
  526. return DiscSelf
  527. default:
  528. return nil
  529. }
  530. }
  531. type tempError interface {
  532. Temporary() bool
  533. }
  534. // listenLoop runs in its own goroutine and accepts
  535. // inbound connections.
  536. func (srv *Server) listenLoop() {
  537. defer srv.loopWG.Done()
  538. glog.V(logger.Info).Infoln("Listening on", srv.listener.Addr())
  539. // This channel acts as a semaphore limiting
  540. // active inbound connections that are lingering pre-handshake.
  541. // If all slots are taken, no further connections are accepted.
  542. tokens := maxAcceptConns
  543. if srv.MaxPendingPeers > 0 {
  544. tokens = srv.MaxPendingPeers
  545. }
  546. slots := make(chan struct{}, tokens)
  547. for i := 0; i < tokens; i++ {
  548. slots <- struct{}{}
  549. }
  550. for {
  551. // Wait for a handshake slot before accepting.
  552. <-slots
  553. var (
  554. fd net.Conn
  555. err error
  556. )
  557. for {
  558. fd, err = srv.listener.Accept()
  559. if tempErr, ok := err.(tempError); ok && tempErr.Temporary() {
  560. glog.V(logger.Debug).Infof("Temporary read error: %v", err)
  561. continue
  562. } else if err != nil {
  563. glog.V(logger.Debug).Infof("Read error: %v", err)
  564. return
  565. }
  566. break
  567. }
  568. fd = newMeteredConn(fd, true)
  569. glog.V(logger.Debug).Infof("Accepted conn %v\n", fd.RemoteAddr())
  570. // Spawn the handler. It will give the slot back when the connection
  571. // has been established.
  572. go func() {
  573. srv.setupConn(fd, inboundConn, nil)
  574. slots <- struct{}{}
  575. }()
  576. }
  577. }
  578. // setupConn runs the handshakes and attempts to add the connection
  579. // as a peer. It returns when the connection has been added as a peer
  580. // or the handshakes have failed.
  581. func (srv *Server) setupConn(fd net.Conn, flags connFlag, dialDest *discover.Node) {
  582. // Prevent leftover pending conns from entering the handshake.
  583. srv.lock.Lock()
  584. running := srv.running
  585. srv.lock.Unlock()
  586. c := &conn{fd: fd, transport: srv.newTransport(fd), flags: flags, cont: make(chan error)}
  587. if !running {
  588. c.close(errServerStopped)
  589. return
  590. }
  591. // Run the encryption handshake.
  592. var err error
  593. if c.id, err = c.doEncHandshake(srv.PrivateKey, dialDest); err != nil {
  594. glog.V(logger.Debug).Infof("%v faild enc handshake: %v", c, err)
  595. c.close(err)
  596. return
  597. }
  598. // For dialed connections, check that the remote public key matches.
  599. if dialDest != nil && c.id != dialDest.ID {
  600. c.close(DiscUnexpectedIdentity)
  601. glog.V(logger.Debug).Infof("%v dialed identity mismatch, want %x", c, dialDest.ID[:8])
  602. return
  603. }
  604. if err := srv.checkpoint(c, srv.posthandshake); err != nil {
  605. glog.V(logger.Debug).Infof("%v failed checkpoint posthandshake: %v", c, err)
  606. c.close(err)
  607. return
  608. }
  609. // Run the protocol handshake
  610. phs, err := c.doProtoHandshake(srv.ourHandshake)
  611. if err != nil {
  612. glog.V(logger.Debug).Infof("%v failed proto handshake: %v", c, err)
  613. c.close(err)
  614. return
  615. }
  616. if phs.ID != c.id {
  617. glog.V(logger.Debug).Infof("%v wrong proto handshake identity: %x", c, phs.ID[:8])
  618. c.close(DiscUnexpectedIdentity)
  619. return
  620. }
  621. c.caps, c.name = phs.Caps, phs.Name
  622. if err := srv.checkpoint(c, srv.addpeer); err != nil {
  623. glog.V(logger.Debug).Infof("%v failed checkpoint addpeer: %v", c, err)
  624. c.close(err)
  625. return
  626. }
  627. // If the checks completed successfully, runPeer has now been
  628. // launched by run.
  629. }
  630. // checkpoint sends the conn to run, which performs the
  631. // post-handshake checks for the stage (posthandshake, addpeer).
  632. func (srv *Server) checkpoint(c *conn, stage chan<- *conn) error {
  633. select {
  634. case stage <- c:
  635. case <-srv.quit:
  636. return errServerStopped
  637. }
  638. select {
  639. case err := <-c.cont:
  640. return err
  641. case <-srv.quit:
  642. return errServerStopped
  643. }
  644. }
  645. // runPeer runs in its own goroutine for each peer.
  646. // it waits until the Peer logic returns and removes
  647. // the peer.
  648. func (srv *Server) runPeer(p *Peer) {
  649. glog.V(logger.Debug).Infof("Added %v\n", p)
  650. srvjslog.LogJson(&logger.P2PConnected{
  651. RemoteId: p.ID().String(),
  652. RemoteAddress: p.RemoteAddr().String(),
  653. RemoteVersionString: p.Name(),
  654. NumConnections: srv.PeerCount(),
  655. })
  656. if srv.newPeerHook != nil {
  657. srv.newPeerHook(p)
  658. }
  659. discreason := p.run()
  660. // Note: run waits for existing peers to be sent on srv.delpeer
  661. // before returning, so this send should not select on srv.quit.
  662. srv.delpeer <- p
  663. glog.V(logger.Debug).Infof("Removed %v (%v)\n", p, discreason)
  664. srvjslog.LogJson(&logger.P2PDisconnected{
  665. RemoteId: p.ID().String(),
  666. NumConnections: srv.PeerCount(),
  667. })
  668. }
  669. // NodeInfo represents a short summary of the information known about the host.
  670. type NodeInfo struct {
  671. ID string `json:"id"` // Unique node identifier (also the encryption key)
  672. Name string `json:"name"` // Name of the node, including client type, version, OS, custom data
  673. Enode string `json:"enode"` // Enode URL for adding this peer from remote peers
  674. IP string `json:"ip"` // IP address of the node
  675. Ports struct {
  676. Discovery int `json:"discovery"` // UDP listening port for discovery protocol
  677. Listener int `json:"listener"` // TCP listening port for RLPx
  678. } `json:"ports"`
  679. ListenAddr string `json:"listenAddr"`
  680. Protocols map[string]interface{} `json:"protocols"`
  681. }
  682. // NodeInfo gathers and returns a collection of metadata known about the host.
  683. func (srv *Server) NodeInfo() *NodeInfo {
  684. node := srv.Self()
  685. // Gather and assemble the generic node infos
  686. info := &NodeInfo{
  687. Name: srv.Name,
  688. Enode: node.String(),
  689. ID: node.ID.String(),
  690. IP: node.IP.String(),
  691. ListenAddr: srv.ListenAddr,
  692. Protocols: make(map[string]interface{}),
  693. }
  694. info.Ports.Discovery = int(node.UDP)
  695. info.Ports.Listener = int(node.TCP)
  696. // Gather all the running protocol infos (only once per protocol type)
  697. for _, proto := range srv.Protocols {
  698. if _, ok := info.Protocols[proto.Name]; !ok {
  699. nodeInfo := interface{}("unknown")
  700. if query := proto.NodeInfo; query != nil {
  701. nodeInfo = proto.NodeInfo()
  702. }
  703. info.Protocols[proto.Name] = nodeInfo
  704. }
  705. }
  706. return info
  707. }
  708. // PeersInfo returns an array of metadata objects describing connected peers.
  709. func (srv *Server) PeersInfo() []*PeerInfo {
  710. // Gather all the generic and sub-protocol specific infos
  711. infos := make([]*PeerInfo, 0, srv.PeerCount())
  712. for _, peer := range srv.Peers() {
  713. if peer != nil {
  714. infos = append(infos, peer.Info())
  715. }
  716. }
  717. // Sort the result array alphabetically by node identifier
  718. for i := 0; i < len(infos); i++ {
  719. for j := i + 1; j < len(infos); j++ {
  720. if infos[i].ID > infos[j].ID {
  721. infos[i], infos[j] = infos[j], infos[i]
  722. }
  723. }
  724. }
  725. return infos
  726. }