You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

702 lines
20 KiB

9 years ago
9 years ago
8 years ago
7 years ago
9 years ago
9 years ago
7 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
7 years ago
9 years ago
9 years ago
9 years ago
7 years ago
9 years ago
9 years ago
8 years ago
7 years ago
9 years ago
7 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
8 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
8 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
7 years ago
p2p: introduce peerConn to simplify peer creation (#1226) * expose AuthEnc in the P2P config if AuthEnc is true, dialed peers must have a node ID in the address and it must match the persistent pubkey from the secret handshake. Refs #1157 * fixes after my own review * fix docs * fix build failure ``` p2p/pex/pex_reactor_test.go:288:88: cannot use seed.NodeInfo().NetAddress() (type *p2p.NetAddress) as type string in array or slice literal ``` * p2p: introduce peerConn to simplify peer creation * Introduce `peerConn` containing the known fields of `peer` * `peer` only created in `sw.addPeer` once handshake is complete and NodeInfo is checked * Eliminates some mutable variables and makes the code flow better * Simplifies the `newXxxPeer` funcs * Use ID instead of PubKey where possible. * SetPubKeyFilter -> SetIDFilter * nodeInfo.Validate takes ID * remove peer.PubKey() * persistent node ids * fixes from review * test: use ip_plus_id.sh more * fix invalid memory panic during fast_sync test ``` 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: panic: runtime error: invalid memory address or nil pointer dereference 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: [signal SIGSEGV: segmentation violation code=0x1 addr=0x20 pc=0x98dd3e] 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: goroutine 3432 [running]: 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: github.com/tendermint/tendermint/p2p.newOutboundPeerConn(0xc423fd1380, 0xc420933e00, 0x1, 0x1239a60, 0 xc420128c40, 0x2, 0x42caf6, 0xc42001f300, 0xc422831d98, 0xc4227951c0, ...) 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/peer.go:123 +0x31e 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: github.com/tendermint/tendermint/p2p.(*Switch).addOutboundPeerWithConfig(0xc4200ad040, 0xc423fd1380, 0 xc420933e00, 0xc423f48801, 0x28, 0x2) 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/switch.go:455 +0x12b 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: github.com/tendermint/tendermint/p2p.(*Switch).DialPeerWithAddress(0xc4200ad040, 0xc423fd1380, 0x1, 0x 0, 0x0) 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/switch.go:371 +0xdc 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: github.com/tendermint/tendermint/p2p.(*Switch).reconnectToPeer(0xc4200ad040, 0x123e000, 0xc42007bb00) 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/switch.go:290 +0x25f 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: created by github.com/tendermint/tendermint/p2p.(*Switch).StopPeerForError 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/switch.go:256 +0x1b7 ```
7 years ago
p2p: introduce peerConn to simplify peer creation (#1226) * expose AuthEnc in the P2P config if AuthEnc is true, dialed peers must have a node ID in the address and it must match the persistent pubkey from the secret handshake. Refs #1157 * fixes after my own review * fix docs * fix build failure ``` p2p/pex/pex_reactor_test.go:288:88: cannot use seed.NodeInfo().NetAddress() (type *p2p.NetAddress) as type string in array or slice literal ``` * p2p: introduce peerConn to simplify peer creation * Introduce `peerConn` containing the known fields of `peer` * `peer` only created in `sw.addPeer` once handshake is complete and NodeInfo is checked * Eliminates some mutable variables and makes the code flow better * Simplifies the `newXxxPeer` funcs * Use ID instead of PubKey where possible. * SetPubKeyFilter -> SetIDFilter * nodeInfo.Validate takes ID * remove peer.PubKey() * persistent node ids * fixes from review * test: use ip_plus_id.sh more * fix invalid memory panic during fast_sync test ``` 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: panic: runtime error: invalid memory address or nil pointer dereference 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: [signal SIGSEGV: segmentation violation code=0x1 addr=0x20 pc=0x98dd3e] 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: goroutine 3432 [running]: 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: github.com/tendermint/tendermint/p2p.newOutboundPeerConn(0xc423fd1380, 0xc420933e00, 0x1, 0x1239a60, 0 xc420128c40, 0x2, 0x42caf6, 0xc42001f300, 0xc422831d98, 0xc4227951c0, ...) 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/peer.go:123 +0x31e 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: github.com/tendermint/tendermint/p2p.(*Switch).addOutboundPeerWithConfig(0xc4200ad040, 0xc423fd1380, 0 xc420933e00, 0xc423f48801, 0x28, 0x2) 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/switch.go:455 +0x12b 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: github.com/tendermint/tendermint/p2p.(*Switch).DialPeerWithAddress(0xc4200ad040, 0xc423fd1380, 0x1, 0x 0, 0x0) 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/switch.go:371 +0xdc 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: github.com/tendermint/tendermint/p2p.(*Switch).reconnectToPeer(0xc4200ad040, 0x123e000, 0xc42007bb00) 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/switch.go:290 +0x25f 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: created by github.com/tendermint/tendermint/p2p.(*Switch).StopPeerForError 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/switch.go:256 +0x1b7 ```
7 years ago
p2p: introduce peerConn to simplify peer creation (#1226) * expose AuthEnc in the P2P config if AuthEnc is true, dialed peers must have a node ID in the address and it must match the persistent pubkey from the secret handshake. Refs #1157 * fixes after my own review * fix docs * fix build failure ``` p2p/pex/pex_reactor_test.go:288:88: cannot use seed.NodeInfo().NetAddress() (type *p2p.NetAddress) as type string in array or slice literal ``` * p2p: introduce peerConn to simplify peer creation * Introduce `peerConn` containing the known fields of `peer` * `peer` only created in `sw.addPeer` once handshake is complete and NodeInfo is checked * Eliminates some mutable variables and makes the code flow better * Simplifies the `newXxxPeer` funcs * Use ID instead of PubKey where possible. * SetPubKeyFilter -> SetIDFilter * nodeInfo.Validate takes ID * remove peer.PubKey() * persistent node ids * fixes from review * test: use ip_plus_id.sh more * fix invalid memory panic during fast_sync test ``` 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: panic: runtime error: invalid memory address or nil pointer dereference 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: [signal SIGSEGV: segmentation violation code=0x1 addr=0x20 pc=0x98dd3e] 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: goroutine 3432 [running]: 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: github.com/tendermint/tendermint/p2p.newOutboundPeerConn(0xc423fd1380, 0xc420933e00, 0x1, 0x1239a60, 0 xc420128c40, 0x2, 0x42caf6, 0xc42001f300, 0xc422831d98, 0xc4227951c0, ...) 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/peer.go:123 +0x31e 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: github.com/tendermint/tendermint/p2p.(*Switch).addOutboundPeerWithConfig(0xc4200ad040, 0xc423fd1380, 0 xc420933e00, 0xc423f48801, 0x28, 0x2) 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/switch.go:455 +0x12b 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: github.com/tendermint/tendermint/p2p.(*Switch).DialPeerWithAddress(0xc4200ad040, 0xc423fd1380, 0x1, 0x 0, 0x0) 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/switch.go:371 +0xdc 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: github.com/tendermint/tendermint/p2p.(*Switch).reconnectToPeer(0xc4200ad040, 0x123e000, 0xc42007bb00) 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/switch.go:290 +0x25f 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: created by github.com/tendermint/tendermint/p2p.(*Switch).StopPeerForError 2018-02-21T06:30:05Z box887.localdomain docker/local_testnet_4[14907]: #011/go/src/github.com/tendermint/tendermint/p2p/switch.go:256 +0x1b7 ```
7 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
  1. package pex
  2. import (
  3. "fmt"
  4. "reflect"
  5. "sort"
  6. "sync"
  7. "time"
  8. amino "github.com/tendermint/go-amino"
  9. cmn "github.com/tendermint/tmlibs/common"
  10. "github.com/tendermint/tendermint/p2p"
  11. "github.com/tendermint/tendermint/p2p/conn"
  12. )
  13. type Peer = p2p.Peer
  14. const (
  15. // PexChannel is a channel for PEX messages
  16. PexChannel = byte(0x00)
  17. // over-estimate of max NetAddress size
  18. // hexID (40) + IP (16) + Port (2) + Name (100) ...
  19. // NOTE: dont use massive DNS name ..
  20. maxAddressSize = 256
  21. // NOTE: amplificaiton factor!
  22. // small request results in up to maxMsgSize response
  23. maxMsgSize = maxAddressSize * maxGetSelection
  24. // ensure we have enough peers
  25. defaultEnsurePeersPeriod = 30 * time.Second
  26. defaultMinNumOutboundPeers = p2p.DefaultMinNumOutboundPeers
  27. // Seed/Crawler constants
  28. // We want seeds to only advertise good peers. Therefore they should wait at
  29. // least as long as we expect it to take for a peer to become good before
  30. // disconnecting.
  31. // see consensus/reactor.go: blocksToContributeToBecomeGoodPeer
  32. // 10000 blocks assuming 1s blocks ~ 2.7 hours.
  33. defaultSeedDisconnectWaitPeriod = 3 * time.Hour
  34. defaultCrawlPeerInterval = 2 * time.Minute // don't redial for this. TODO: back-off. what for?
  35. defaultCrawlPeersPeriod = 30 * time.Second // check some peers every this
  36. maxAttemptsToDial = 16 // ~ 35h in total (last attempt - 18h)
  37. // if node connects to seed, it does not have any trusted peers.
  38. // Especially in the beginning, node should have more trusted peers than
  39. // untrusted.
  40. biasToSelectNewPeers = 30 // 70 to select good peers
  41. )
  42. // PEXReactor handles PEX (peer exchange) and ensures that an
  43. // adequate number of peers are connected to the switch.
  44. //
  45. // It uses `AddrBook` (address book) to store `NetAddress`es of the peers.
  46. //
  47. // ## Preventing abuse
  48. //
  49. // Only accept pexAddrsMsg from peers we sent a corresponding pexRequestMsg too.
  50. // Only accept one pexRequestMsg every ~defaultEnsurePeersPeriod.
  51. type PEXReactor struct {
  52. p2p.BaseReactor
  53. book AddrBook
  54. config *PEXReactorConfig
  55. ensurePeersPeriod time.Duration // TODO: should go in the config
  56. // maps to prevent abuse
  57. requestsSent *cmn.CMap // ID->struct{}: unanswered send requests
  58. lastReceivedRequests *cmn.CMap // ID->time.Time: last time peer requested from us
  59. attemptsToDial sync.Map // address (string) -> {number of attempts (int), last time dialed (time.Time)}
  60. }
  61. func (r *PEXReactor) minReceiveRequestInterval() time.Duration {
  62. // NOTE: must be less than ensurePeersPeriod, otherwise we'll request
  63. // peers too quickly from others and they'll think we're bad!
  64. return r.ensurePeersPeriod / 3
  65. }
  66. // PEXReactorConfig holds reactor specific configuration data.
  67. type PEXReactorConfig struct {
  68. // Seed/Crawler mode
  69. SeedMode bool
  70. // Seeds is a list of addresses reactor may use
  71. // if it can't connect to peers in the addrbook.
  72. Seeds []string
  73. // PrivatePeerIDs is a list of peer IDs, which must not be gossiped to other
  74. // peers.
  75. PrivatePeerIDs []string
  76. }
  77. type _attemptsToDial struct {
  78. number int
  79. lastDialed time.Time
  80. }
  81. // NewPEXReactor creates new PEX reactor.
  82. func NewPEXReactor(b AddrBook, config *PEXReactorConfig) *PEXReactor {
  83. r := &PEXReactor{
  84. book: b,
  85. config: config,
  86. ensurePeersPeriod: defaultEnsurePeersPeriod,
  87. requestsSent: cmn.NewCMap(),
  88. lastReceivedRequests: cmn.NewCMap(),
  89. }
  90. r.BaseReactor = *p2p.NewBaseReactor("PEXReactor", r)
  91. return r
  92. }
  93. // OnStart implements BaseService
  94. func (r *PEXReactor) OnStart() error {
  95. if err := r.BaseReactor.OnStart(); err != nil {
  96. return err
  97. }
  98. err := r.book.Start()
  99. if err != nil && err != cmn.ErrAlreadyStarted {
  100. return err
  101. }
  102. // return err if user provided a bad seed address
  103. // or a host name that we cant resolve
  104. if err := r.checkSeeds(); err != nil {
  105. return err
  106. }
  107. // Check if this node should run
  108. // in seed/crawler mode
  109. if r.config.SeedMode {
  110. go r.crawlPeersRoutine()
  111. } else {
  112. go r.ensurePeersRoutine()
  113. }
  114. return nil
  115. }
  116. // OnStop implements BaseService
  117. func (r *PEXReactor) OnStop() {
  118. r.BaseReactor.OnStop()
  119. r.book.Stop()
  120. }
  121. // GetChannels implements Reactor
  122. func (r *PEXReactor) GetChannels() []*conn.ChannelDescriptor {
  123. return []*conn.ChannelDescriptor{
  124. {
  125. ID: PexChannel,
  126. Priority: 1,
  127. SendQueueCapacity: 10,
  128. },
  129. }
  130. }
  131. // AddPeer implements Reactor by adding peer to the address book (if inbound)
  132. // or by requesting more addresses (if outbound).
  133. func (r *PEXReactor) AddPeer(p Peer) {
  134. if p.IsOutbound() {
  135. // For outbound peers, the address is already in the books -
  136. // either via DialPeersAsync or r.Receive.
  137. // Ask it for more peers if we need.
  138. if r.book.NeedMoreAddrs() {
  139. r.RequestAddrs(p)
  140. }
  141. } else {
  142. // inbound peer is its own source
  143. addr := p.NodeInfo().NetAddress()
  144. src := addr
  145. // ignore private addrs
  146. if isAddrPrivate(addr, r.config.PrivatePeerIDs) {
  147. return
  148. }
  149. // add to book. dont RequestAddrs right away because
  150. // we don't trust inbound as much - let ensurePeersRoutine handle it.
  151. err := r.book.AddAddress(addr, src)
  152. r.logErrAddrBook(err)
  153. }
  154. }
  155. func (r *PEXReactor) logErrAddrBook(err error) {
  156. if err != nil {
  157. switch err.(type) {
  158. case ErrAddrBookNilAddr:
  159. r.Logger.Error("Failed to add new address", "err", err)
  160. default:
  161. // non-routable, self, full book, etc.
  162. r.Logger.Debug("Failed to add new address", "err", err)
  163. }
  164. }
  165. }
  166. // RemovePeer implements Reactor.
  167. func (r *PEXReactor) RemovePeer(p Peer, reason interface{}) {
  168. id := string(p.ID())
  169. r.requestsSent.Delete(id)
  170. r.lastReceivedRequests.Delete(id)
  171. }
  172. // Receive implements Reactor by handling incoming PEX messages.
  173. func (r *PEXReactor) Receive(chID byte, src Peer, msgBytes []byte) {
  174. msg, err := DecodeMessage(msgBytes)
  175. if err != nil {
  176. r.Logger.Error("Error decoding message", "src", src, "chId", chID, "msg", msg, "err", err, "bytes", msgBytes)
  177. r.Switch.StopPeerForError(src, err)
  178. return
  179. }
  180. r.Logger.Debug("Received message", "src", src, "chId", chID, "msg", msg)
  181. switch msg := msg.(type) {
  182. case *pexRequestMessage:
  183. // Check we're not receiving too many requests
  184. if err := r.receiveRequest(src); err != nil {
  185. r.Switch.StopPeerForError(src, err)
  186. return
  187. }
  188. // Seeds disconnect after sending a batch of addrs
  189. // NOTE: this is a prime candidate for amplification attacks
  190. // so it's important we
  191. // 1) restrict how frequently peers can request
  192. // 2) limit the output size
  193. if r.config.SeedMode {
  194. r.SendAddrs(src, r.book.GetSelectionWithBias(biasToSelectNewPeers))
  195. r.Switch.StopPeerGracefully(src)
  196. } else {
  197. r.SendAddrs(src, r.book.GetSelection())
  198. }
  199. case *pexAddrsMessage:
  200. // If we asked for addresses, add them to the book
  201. if err := r.ReceiveAddrs(msg.Addrs, src); err != nil {
  202. r.Switch.StopPeerForError(src, err)
  203. return
  204. }
  205. default:
  206. r.Logger.Error(fmt.Sprintf("Unknown message type %v", reflect.TypeOf(msg)))
  207. }
  208. }
  209. // enforces a minimum amount of time between requests
  210. func (r *PEXReactor) receiveRequest(src Peer) error {
  211. id := string(src.ID())
  212. v := r.lastReceivedRequests.Get(id)
  213. if v == nil {
  214. // initialize with empty time
  215. lastReceived := time.Time{}
  216. r.lastReceivedRequests.Set(id, lastReceived)
  217. return nil
  218. }
  219. lastReceived := v.(time.Time)
  220. if lastReceived.Equal(time.Time{}) {
  221. // first time gets a free pass. then we start tracking the time
  222. lastReceived = time.Now()
  223. r.lastReceivedRequests.Set(id, lastReceived)
  224. return nil
  225. }
  226. now := time.Now()
  227. minInterval := r.minReceiveRequestInterval()
  228. if now.Sub(lastReceived) < minInterval {
  229. return fmt.Errorf("Peer (%v) sent next PEX request too soon. lastReceived: %v, now: %v, minInterval: %v. Disconnecting",
  230. src.ID(),
  231. lastReceived,
  232. now,
  233. minInterval,
  234. )
  235. }
  236. r.lastReceivedRequests.Set(id, now)
  237. return nil
  238. }
  239. // RequestAddrs asks peer for more addresses if we do not already
  240. // have a request out for this peer.
  241. func (r *PEXReactor) RequestAddrs(p Peer) {
  242. r.Logger.Debug("Request addrs", "from", p)
  243. id := string(p.ID())
  244. if r.requestsSent.Has(id) {
  245. return
  246. }
  247. r.requestsSent.Set(id, struct{}{})
  248. p.Send(PexChannel, cdc.MustMarshalBinary(&pexRequestMessage{}))
  249. }
  250. // ReceiveAddrs adds the given addrs to the addrbook if theres an open
  251. // request for this peer and deletes the open request.
  252. // If there's no open request for the src peer, it returns an error.
  253. func (r *PEXReactor) ReceiveAddrs(addrs []*p2p.NetAddress, src Peer) error {
  254. id := string(src.ID())
  255. if !r.requestsSent.Has(id) {
  256. return cmn.NewError("Received unsolicited pexAddrsMessage")
  257. }
  258. r.requestsSent.Delete(id)
  259. srcAddr := src.NodeInfo().NetAddress()
  260. for _, netAddr := range addrs {
  261. // NOTE: GetSelection methods should never return nil addrs
  262. if netAddr == nil {
  263. return cmn.NewError("received nil addr")
  264. }
  265. // ignore private peers
  266. // TODO: give private peers to AddrBook so it can enforce this on AddAddress.
  267. // We'd then have to check for ErrPrivatePeer on AddAddress here, which is
  268. // an error we just ignore (maybe peer is probing us for our private peers :P)
  269. if isAddrPrivate(netAddr, r.config.PrivatePeerIDs) {
  270. continue
  271. }
  272. err := r.book.AddAddress(netAddr, srcAddr)
  273. r.logErrAddrBook(err)
  274. }
  275. return nil
  276. }
  277. // SendAddrs sends addrs to the peer.
  278. func (r *PEXReactor) SendAddrs(p Peer, netAddrs []*p2p.NetAddress) {
  279. p.Send(PexChannel, cdc.MustMarshalBinary(&pexAddrsMessage{Addrs: netAddrs}))
  280. }
  281. // SetEnsurePeersPeriod sets period to ensure peers connected.
  282. func (r *PEXReactor) SetEnsurePeersPeriod(d time.Duration) {
  283. r.ensurePeersPeriod = d
  284. }
  285. // Ensures that sufficient peers are connected. (continuous)
  286. func (r *PEXReactor) ensurePeersRoutine() {
  287. var (
  288. seed = cmn.NewRand()
  289. jitter = seed.Int63n(r.ensurePeersPeriod.Nanoseconds())
  290. )
  291. // Randomize first round of communication to avoid thundering herd.
  292. // If no potential peers are present directly start connecting so we guarantee
  293. // swift setup with the help of configured seeds.
  294. if r.hasPotentialPeers() {
  295. time.Sleep(time.Duration(jitter))
  296. }
  297. // fire once immediately.
  298. // ensures we dial the seeds right away if the book is empty
  299. r.ensurePeers()
  300. // fire periodically
  301. ticker := time.NewTicker(r.ensurePeersPeriod)
  302. for {
  303. select {
  304. case <-ticker.C:
  305. r.ensurePeers()
  306. case <-r.Quit():
  307. ticker.Stop()
  308. return
  309. }
  310. }
  311. }
  312. // ensurePeers ensures that sufficient peers are connected. (once)
  313. //
  314. // heuristic that we haven't perfected yet, or, perhaps is manually edited by
  315. // the node operator. It should not be used to compute what addresses are
  316. // already connected or not.
  317. func (r *PEXReactor) ensurePeers() {
  318. var (
  319. out, in, dial = r.Switch.NumPeers()
  320. numToDial = defaultMinNumOutboundPeers - (out + dial)
  321. )
  322. r.Logger.Info(
  323. "Ensure peers",
  324. "numOutPeers", out,
  325. "numInPeers", in,
  326. "numDialing", dial,
  327. "numToDial", numToDial,
  328. )
  329. if numToDial <= 0 {
  330. return
  331. }
  332. // bias to prefer more vetted peers when we have fewer connections.
  333. // not perfect, but somewhate ensures that we prioritize connecting to more-vetted
  334. // NOTE: range here is [10, 90]. Too high ?
  335. newBias := cmn.MinInt(out, 8)*10 + 10
  336. toDial := make(map[p2p.ID]*p2p.NetAddress)
  337. // Try maxAttempts times to pick numToDial addresses to dial
  338. maxAttempts := numToDial * 3
  339. for i := 0; i < maxAttempts && len(toDial) < numToDial; i++ {
  340. try := r.book.PickAddress(newBias)
  341. if try == nil {
  342. continue
  343. }
  344. if _, selected := toDial[try.ID]; selected {
  345. continue
  346. }
  347. if dialling := r.Switch.IsDialing(try.ID); dialling {
  348. continue
  349. }
  350. if connected := r.Switch.Peers().Has(try.ID); connected {
  351. continue
  352. }
  353. // TODO: consider moving some checks from toDial into here
  354. // so we don't even consider dialing peers that we want to wait
  355. // before dialling again, or have dialed too many times already
  356. r.Logger.Info("Will dial address", "addr", try)
  357. toDial[try.ID] = try
  358. }
  359. // Dial picked addresses
  360. for _, addr := range toDial {
  361. go r.dialPeer(addr)
  362. }
  363. // If we need more addresses, pick a random peer and ask for more.
  364. if r.book.NeedMoreAddrs() {
  365. peers := r.Switch.Peers().List()
  366. peersCount := len(peers)
  367. if peersCount > 0 {
  368. peer := peers[cmn.RandInt()%peersCount] // nolint: gas
  369. r.Logger.Info("We need more addresses. Sending pexRequest to random peer", "peer", peer)
  370. r.RequestAddrs(peer)
  371. }
  372. }
  373. // If we are not connected to nor dialing anybody, fallback to dialing a seed.
  374. if out+in+dial+len(toDial) == 0 {
  375. r.Logger.Info("No addresses to dial nor connected peers. Falling back to seeds")
  376. r.dialSeeds()
  377. }
  378. }
  379. func (r *PEXReactor) dialAttemptsInfo(addr *p2p.NetAddress) (attempts int, lastDialed time.Time) {
  380. _attempts, ok := r.attemptsToDial.Load(addr.DialString())
  381. if !ok {
  382. return
  383. }
  384. atd := _attempts.(_attemptsToDial)
  385. return atd.number, atd.lastDialed
  386. }
  387. func (r *PEXReactor) dialPeer(addr *p2p.NetAddress) {
  388. attempts, lastDialed := r.dialAttemptsInfo(addr)
  389. if attempts > maxAttemptsToDial {
  390. r.Logger.Error("Reached max attempts to dial", "addr", addr, "attempts", attempts)
  391. r.book.MarkBad(addr)
  392. return
  393. }
  394. // exponential backoff if it's not our first attempt to dial given address
  395. if attempts > 0 {
  396. jitterSeconds := time.Duration(cmn.RandFloat64() * float64(time.Second)) // 1s == (1e9 ns)
  397. backoffDuration := jitterSeconds + ((1 << uint(attempts)) * time.Second)
  398. sinceLastDialed := time.Since(lastDialed)
  399. if sinceLastDialed < backoffDuration {
  400. r.Logger.Debug("Too early to dial", "addr", addr, "backoff_duration", backoffDuration, "last_dialed", lastDialed, "time_since", sinceLastDialed)
  401. return
  402. }
  403. }
  404. err := r.Switch.DialPeerWithAddress(addr, false)
  405. if err != nil {
  406. r.Logger.Error("Dialing failed", "addr", addr, "err", err, "attempts", attempts)
  407. // TODO: detect more "bad peer" scenarios
  408. if _, ok := err.(p2p.ErrSwitchAuthenticationFailure); ok {
  409. r.book.MarkBad(addr)
  410. r.attemptsToDial.Delete(addr.DialString())
  411. } else {
  412. r.book.MarkAttempt(addr)
  413. // FIXME: if the addr is going to be removed from the addrbook (hard to
  414. // tell at this point), we need to Delete it from attemptsToDial, not
  415. // record another attempt.
  416. // record attempt
  417. r.attemptsToDial.Store(addr.DialString(), _attemptsToDial{attempts + 1, time.Now()})
  418. }
  419. } else {
  420. // cleanup any history
  421. r.attemptsToDial.Delete(addr.DialString())
  422. }
  423. }
  424. // check seed addresses are well formed
  425. func (r *PEXReactor) checkSeeds() error {
  426. lSeeds := len(r.config.Seeds)
  427. if lSeeds == 0 {
  428. return nil
  429. }
  430. _, errs := p2p.NewNetAddressStrings(r.config.Seeds)
  431. for _, err := range errs {
  432. if err != nil {
  433. return err
  434. }
  435. }
  436. return nil
  437. }
  438. // randomly dial seeds until we connect to one or exhaust them
  439. func (r *PEXReactor) dialSeeds() {
  440. lSeeds := len(r.config.Seeds)
  441. if lSeeds == 0 {
  442. return
  443. }
  444. seedAddrs, _ := p2p.NewNetAddressStrings(r.config.Seeds)
  445. perm := cmn.RandPerm(lSeeds)
  446. // perm := r.Switch.rng.Perm(lSeeds)
  447. for _, i := range perm {
  448. // dial a random seed
  449. seedAddr := seedAddrs[i]
  450. err := r.Switch.DialPeerWithAddress(seedAddr, false)
  451. if err == nil {
  452. return
  453. }
  454. r.Switch.Logger.Error("Error dialing seed", "err", err, "seed", seedAddr)
  455. }
  456. r.Switch.Logger.Error("Couldn't connect to any seeds")
  457. }
  458. // AttemptsToDial returns the number of attempts to dial specific address. It
  459. // returns 0 if never attempted or successfully connected.
  460. func (r *PEXReactor) AttemptsToDial(addr *p2p.NetAddress) int {
  461. lAttempts, attempted := r.attemptsToDial.Load(addr.DialString())
  462. if attempted {
  463. return lAttempts.(_attemptsToDial).number
  464. }
  465. return 0
  466. }
  467. //----------------------------------------------------------
  468. // Explores the network searching for more peers. (continuous)
  469. // Seed/Crawler Mode causes this node to quickly disconnect
  470. // from peers, except other seed nodes.
  471. func (r *PEXReactor) crawlPeersRoutine() {
  472. // Do an initial crawl
  473. r.crawlPeers()
  474. // Fire periodically
  475. ticker := time.NewTicker(defaultCrawlPeersPeriod)
  476. for {
  477. select {
  478. case <-ticker.C:
  479. r.attemptDisconnects()
  480. r.crawlPeers()
  481. case <-r.Quit():
  482. return
  483. }
  484. }
  485. }
  486. // hasPotentialPeers indicates if there is a potential peer to connect to, by
  487. // consulting the Switch as well as the AddrBook.
  488. func (r *PEXReactor) hasPotentialPeers() bool {
  489. out, in, dial := r.Switch.NumPeers()
  490. return out+in+dial > 0 && len(r.book.ListOfKnownAddresses()) > 0
  491. }
  492. // crawlPeerInfo handles temporary data needed for the
  493. // network crawling performed during seed/crawler mode.
  494. type crawlPeerInfo struct {
  495. // The listening address of a potential peer we learned about
  496. Addr *p2p.NetAddress
  497. // The last time we attempt to reach this address
  498. LastAttempt time.Time
  499. // The last time we successfully reached this address
  500. LastSuccess time.Time
  501. }
  502. // oldestFirst implements sort.Interface for []crawlPeerInfo
  503. // based on the LastAttempt field.
  504. type oldestFirst []crawlPeerInfo
  505. func (of oldestFirst) Len() int { return len(of) }
  506. func (of oldestFirst) Swap(i, j int) { of[i], of[j] = of[j], of[i] }
  507. func (of oldestFirst) Less(i, j int) bool { return of[i].LastAttempt.Before(of[j].LastAttempt) }
  508. // getPeersToCrawl returns addresses of potential peers that we wish to validate.
  509. // NOTE: The status information is ordered as described above.
  510. func (r *PEXReactor) getPeersToCrawl() []crawlPeerInfo {
  511. var of oldestFirst
  512. // TODO: be more selective
  513. addrs := r.book.ListOfKnownAddresses()
  514. for _, addr := range addrs {
  515. if len(addr.ID()) == 0 {
  516. continue // dont use peers without id
  517. }
  518. of = append(of, crawlPeerInfo{
  519. Addr: addr.Addr,
  520. LastAttempt: addr.LastAttempt,
  521. LastSuccess: addr.LastSuccess,
  522. })
  523. }
  524. sort.Sort(of)
  525. return of
  526. }
  527. // crawlPeers will crawl the network looking for new peer addresses. (once)
  528. func (r *PEXReactor) crawlPeers() {
  529. peerInfos := r.getPeersToCrawl()
  530. now := time.Now()
  531. // Use addresses we know of to reach additional peers
  532. for _, pi := range peerInfos {
  533. // Do not attempt to connect with peers we recently dialed
  534. if now.Sub(pi.LastAttempt) < defaultCrawlPeerInterval {
  535. continue
  536. }
  537. // Otherwise, attempt to connect with the known address
  538. err := r.Switch.DialPeerWithAddress(pi.Addr, false)
  539. if err != nil {
  540. r.book.MarkAttempt(pi.Addr)
  541. continue
  542. }
  543. // Ask for more addresses
  544. peer := r.Switch.Peers().Get(pi.Addr.ID)
  545. if peer != nil {
  546. r.RequestAddrs(peer)
  547. }
  548. }
  549. }
  550. // attemptDisconnects checks if we've been with each peer long enough to disconnect
  551. func (r *PEXReactor) attemptDisconnects() {
  552. for _, peer := range r.Switch.Peers().List() {
  553. if peer.Status().Duration < defaultSeedDisconnectWaitPeriod {
  554. continue
  555. }
  556. if peer.IsPersistent() {
  557. continue
  558. }
  559. r.Switch.StopPeerGracefully(peer)
  560. }
  561. }
  562. // isAddrPrivate returns true if addr.ID is a private ID.
  563. func isAddrPrivate(addr *p2p.NetAddress, privatePeerIDs []string) bool {
  564. for _, id := range privatePeerIDs {
  565. if string(addr.ID) == id {
  566. return true
  567. }
  568. }
  569. return false
  570. }
  571. //-----------------------------------------------------------------------------
  572. // Messages
  573. // PexMessage is a primary type for PEX messages. Underneath, it could contain
  574. // either pexRequestMessage, or pexAddrsMessage messages.
  575. type PexMessage interface{}
  576. func RegisterPexMessage(cdc *amino.Codec) {
  577. cdc.RegisterInterface((*PexMessage)(nil), nil)
  578. cdc.RegisterConcrete(&pexRequestMessage{}, "tendermint/p2p/PexRequestMessage", nil)
  579. cdc.RegisterConcrete(&pexAddrsMessage{}, "tendermint/p2p/PexAddrsMessage", nil)
  580. }
  581. // DecodeMessage implements interface registered above.
  582. func DecodeMessage(bz []byte) (msg PexMessage, err error) {
  583. if len(bz) > maxMsgSize {
  584. return msg, fmt.Errorf("Msg exceeds max size (%d > %d)",
  585. len(bz), maxMsgSize)
  586. }
  587. err = cdc.UnmarshalBinary(bz, &msg)
  588. return
  589. }
  590. /*
  591. A pexRequestMessage requests additional peer addresses.
  592. */
  593. type pexRequestMessage struct {
  594. }
  595. func (m *pexRequestMessage) String() string {
  596. return "[pexRequest]"
  597. }
  598. /*
  599. A message with announced peer addresses.
  600. */
  601. type pexAddrsMessage struct {
  602. Addrs []*p2p.NetAddress
  603. }
  604. func (m *pexAddrsMessage) String() string {
  605. return fmt.Sprintf("[pexAddrs %v]", m.Addrs)
  606. }