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.

354 lines
9.9 KiB

9 years ago
8 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
7 years ago
9 years ago
7 years ago
8 years ago
7 years ago
8 years ago
9 years ago
8 years ago
7 years ago
8 years ago
9 years ago
7 years ago
8 years ago
9 years ago
9 years ago
8 years ago
9 years ago
9 years ago
9 years ago
prevent nil addr Error: ``` Error: runtime error: invalid memoryaddress or nil pointer dereference\nStack: goroutine 549 [running]:\nruntime/debug.Stack(0x0, 0x0, 0x0)\n\t/usr/local/go/src/runtime/debug/stack.go:24 +0x80\ngithub.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.(*MConnection)._recover(0xc821723b00)\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/connection.go:173 +0x53\npanic(0xbe1500, 0xc820012080)\n\t/usr/local/go/src/runtime/panic.go:443 +0x4e9\ngithub.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.(*NetAddress).Valid(0x0, 0x0)\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/netaddress.go:125 +0x1c\ngithub.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.(*NetAddress).Routable(0x0, 0xc8217bb740)\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/netaddress.go:117 +0x25\ngithub.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.(*AddrBook).addAddress(0xc820108380, 0x0, 0xc821739590)\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/addrbook.go:524 +0x45\ngithub.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.(*AddrBook).AddAddress(0xc820108380, 0x0, 0xc821739590)\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/addrbook.go:160 +0x286\ngithub.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.(*PEXReactor).Receive(0xc82000be60, 0xc820149f00, 0xc8218163f0, 0xc82184e000, 0x5b, 0x1000)\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/pex_reactor.go:109 +0x457\ngithub.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.newPeer.func1(0xc82011d500, 0xc82184e000, 0x5b, 0x1000)\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/peer.go:58 +0x202\ngithub.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.(*MConnection).recvRoutine(0xc821723b00)\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/connection.go:439 +0x1177\ncreated by github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.(*MConnection).OnStart\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/connection.go:138 +0x1a1\n ```
8 years ago
prevent nil addr Error: ``` Error: runtime error: invalid memoryaddress or nil pointer dereference\nStack: goroutine 549 [running]:\nruntime/debug.Stack(0x0, 0x0, 0x0)\n\t/usr/local/go/src/runtime/debug/stack.go:24 +0x80\ngithub.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.(*MConnection)._recover(0xc821723b00)\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/connection.go:173 +0x53\npanic(0xbe1500, 0xc820012080)\n\t/usr/local/go/src/runtime/panic.go:443 +0x4e9\ngithub.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.(*NetAddress).Valid(0x0, 0x0)\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/netaddress.go:125 +0x1c\ngithub.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.(*NetAddress).Routable(0x0, 0xc8217bb740)\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/netaddress.go:117 +0x25\ngithub.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.(*AddrBook).addAddress(0xc820108380, 0x0, 0xc821739590)\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/addrbook.go:524 +0x45\ngithub.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.(*AddrBook).AddAddress(0xc820108380, 0x0, 0xc821739590)\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/addrbook.go:160 +0x286\ngithub.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.(*PEXReactor).Receive(0xc82000be60, 0xc820149f00, 0xc8218163f0, 0xc82184e000, 0x5b, 0x1000)\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/pex_reactor.go:109 +0x457\ngithub.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.newPeer.func1(0xc82011d500, 0xc82184e000, 0x5b, 0x1000)\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/peer.go:58 +0x202\ngithub.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.(*MConnection).recvRoutine(0xc821723b00)\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/connection.go:439 +0x1177\ncreated by github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p.(*MConnection).OnStart\n\t/go/src/github.com/tendermint/tendermint/vendor/github.com/tendermint/go-p2p/connection.go:138 +0x1a1\n ```
8 years ago
9 years ago
8 years ago
9 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
9 years ago
9 years ago
8 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
9 years ago
9 years ago
7 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
  1. package p2p
  2. import (
  3. "bytes"
  4. "fmt"
  5. "math/rand"
  6. "reflect"
  7. "time"
  8. wire "github.com/tendermint/go-wire"
  9. cmn "github.com/tendermint/tmlibs/common"
  10. )
  11. const (
  12. // PexChannel is a channel for PEX messages
  13. PexChannel = byte(0x00)
  14. // period to ensure peers connected
  15. defaultEnsurePeersPeriod = 30 * time.Second
  16. minNumOutboundPeers = 10
  17. maxPexMessageSize = 1048576 // 1MB
  18. // maximum messages one peer can send to us during `msgCountByPeerFlushInterval`
  19. defaultMaxMsgCountByPeer = 1000
  20. msgCountByPeerFlushInterval = 1 * time.Hour
  21. )
  22. // PEXReactor handles PEX (peer exchange) and ensures that an
  23. // adequate number of peers are connected to the switch.
  24. //
  25. // It uses `AddrBook` (address book) to store `NetAddress`es of the peers.
  26. //
  27. // ## Preventing abuse
  28. //
  29. // For now, it just limits the number of messages from one peer to
  30. // `defaultMaxMsgCountByPeer` messages per `msgCountByPeerFlushInterval` (1000
  31. // msg/hour).
  32. //
  33. // NOTE [2017-01-17]:
  34. // Limiting is fine for now. Maybe down the road we want to keep track of the
  35. // quality of peer messages so if peerA keeps telling us about peers we can't
  36. // connect to then maybe we should care less about peerA. But I don't think
  37. // that kind of complexity is priority right now.
  38. type PEXReactor struct {
  39. BaseReactor
  40. book *AddrBook
  41. ensurePeersPeriod time.Duration
  42. // tracks message count by peer, so we can prevent abuse
  43. msgCountByPeer *cmn.CMap
  44. maxMsgCountByPeer uint16
  45. }
  46. // NewPEXReactor creates new PEX reactor.
  47. func NewPEXReactor(b *AddrBook) *PEXReactor {
  48. r := &PEXReactor{
  49. book: b,
  50. ensurePeersPeriod: defaultEnsurePeersPeriod,
  51. msgCountByPeer: cmn.NewCMap(),
  52. maxMsgCountByPeer: defaultMaxMsgCountByPeer,
  53. }
  54. r.BaseReactor = *NewBaseReactor("PEXReactor", r)
  55. return r
  56. }
  57. // OnStart implements BaseService
  58. func (r *PEXReactor) OnStart() error {
  59. if err := r.BaseReactor.OnStart(); err != nil {
  60. return err
  61. }
  62. err := r.book.Start()
  63. if err != nil && err != cmn.ErrAlreadyStarted {
  64. return err
  65. }
  66. go r.ensurePeersRoutine()
  67. go r.flushMsgCountByPeer()
  68. return nil
  69. }
  70. // OnStop implements BaseService
  71. func (r *PEXReactor) OnStop() {
  72. r.BaseReactor.OnStop()
  73. r.book.Stop()
  74. }
  75. // GetChannels implements Reactor
  76. func (r *PEXReactor) GetChannels() []*ChannelDescriptor {
  77. return []*ChannelDescriptor{
  78. {
  79. ID: PexChannel,
  80. Priority: 1,
  81. SendQueueCapacity: 10,
  82. },
  83. }
  84. }
  85. // AddPeer implements Reactor by adding peer to the address book (if inbound)
  86. // or by requesting more addresses (if outbound).
  87. func (r *PEXReactor) AddPeer(p Peer) {
  88. if p.IsOutbound() {
  89. // For outbound peers, the address is already in the books.
  90. // Either it was added in DialSeeds or when we
  91. // received the peer's address in r.Receive
  92. if r.book.NeedMoreAddrs() {
  93. r.RequestPEX(p)
  94. }
  95. } else { // For inbound connections, the peer is its own source
  96. addr, err := NewNetAddressString(p.NodeInfo().ListenAddr)
  97. if err != nil {
  98. // peer gave us a bad ListenAddr. TODO: punish
  99. r.Logger.Error("Error in AddPeer: invalid peer address", "addr", p.NodeInfo().ListenAddr, "err", err)
  100. return
  101. }
  102. r.book.AddAddress(addr, addr)
  103. }
  104. }
  105. // RemovePeer implements Reactor.
  106. func (r *PEXReactor) RemovePeer(p Peer, reason interface{}) {
  107. // If we aren't keeping track of local temp data for each peer here, then we
  108. // don't have to do anything.
  109. }
  110. // Receive implements Reactor by handling incoming PEX messages.
  111. func (r *PEXReactor) Receive(chID byte, src Peer, msgBytes []byte) {
  112. srcAddrStr := src.NodeInfo().RemoteAddr
  113. srcAddr, err := NewNetAddressString(srcAddrStr)
  114. if err != nil {
  115. // this should never happen. TODO: cancel conn
  116. r.Logger.Error("Error in Receive: invalid peer address", "addr", srcAddrStr, "err", err)
  117. return
  118. }
  119. r.IncrementMsgCountForPeer(srcAddrStr)
  120. if r.ReachedMaxMsgCountForPeer(srcAddrStr) {
  121. r.Logger.Error("Maximum number of messages reached for peer", "peer", srcAddrStr)
  122. // TODO remove src from peers?
  123. return
  124. }
  125. _, msg, err := DecodeMessage(msgBytes)
  126. if err != nil {
  127. r.Logger.Error("Error decoding message", "err", err)
  128. return
  129. }
  130. r.Logger.Info("Received message", "msg", msg)
  131. switch msg := msg.(type) {
  132. case *pexRequestMessage:
  133. // src requested some peers.
  134. // NOTE: we might send an empty selection
  135. r.SendAddrs(src, r.book.GetSelection())
  136. case *pexAddrsMessage:
  137. // We received some peer addresses from src.
  138. // TODO: (We don't want to get spammed with bad peers)
  139. for _, addr := range msg.Addrs {
  140. if addr != nil {
  141. r.book.AddAddress(addr, srcAddr)
  142. }
  143. }
  144. default:
  145. r.Logger.Error(fmt.Sprintf("Unknown message type %v", reflect.TypeOf(msg)))
  146. }
  147. }
  148. // RequestPEX asks peer for more addresses.
  149. func (r *PEXReactor) RequestPEX(p Peer) {
  150. p.Send(PexChannel, struct{ PexMessage }{&pexRequestMessage{}})
  151. }
  152. // SendAddrs sends addrs to the peer.
  153. func (r *PEXReactor) SendAddrs(p Peer, addrs []*NetAddress) {
  154. p.Send(PexChannel, struct{ PexMessage }{&pexAddrsMessage{Addrs: addrs}})
  155. }
  156. // SetEnsurePeersPeriod sets period to ensure peers connected.
  157. func (r *PEXReactor) SetEnsurePeersPeriod(d time.Duration) {
  158. r.ensurePeersPeriod = d
  159. }
  160. // SetMaxMsgCountByPeer sets maximum messages one peer can send to us during 'msgCountByPeerFlushInterval'.
  161. func (r *PEXReactor) SetMaxMsgCountByPeer(v uint16) {
  162. r.maxMsgCountByPeer = v
  163. }
  164. // ReachedMaxMsgCountForPeer returns true if we received too many
  165. // messages from peer with address `addr`.
  166. // NOTE: assumes the value in the CMap is non-nil
  167. func (r *PEXReactor) ReachedMaxMsgCountForPeer(addr string) bool {
  168. return r.msgCountByPeer.Get(addr).(uint16) >= r.maxMsgCountByPeer
  169. }
  170. // Increment or initialize the msg count for the peer in the CMap
  171. func (r *PEXReactor) IncrementMsgCountForPeer(addr string) {
  172. var count uint16
  173. countI := r.msgCountByPeer.Get(addr)
  174. if countI != nil {
  175. count = countI.(uint16)
  176. }
  177. count++
  178. r.msgCountByPeer.Set(addr, count)
  179. }
  180. // Ensures that sufficient peers are connected. (continuous)
  181. func (r *PEXReactor) ensurePeersRoutine() {
  182. // Randomize when routine starts
  183. ensurePeersPeriodMs := r.ensurePeersPeriod.Nanoseconds() / 1e6
  184. time.Sleep(time.Duration(rand.Int63n(ensurePeersPeriodMs)) * time.Millisecond)
  185. // fire once immediately.
  186. r.ensurePeers()
  187. // fire periodically
  188. ticker := time.NewTicker(r.ensurePeersPeriod)
  189. for {
  190. select {
  191. case <-ticker.C:
  192. r.ensurePeers()
  193. case <-r.Quit:
  194. ticker.Stop()
  195. return
  196. }
  197. }
  198. }
  199. // ensurePeers ensures that sufficient peers are connected. (once)
  200. //
  201. // Old bucket / New bucket are arbitrary categories to denote whether an
  202. // address is vetted or not, and this needs to be determined over time via a
  203. // heuristic that we haven't perfected yet, or, perhaps is manually edited by
  204. // the node operator. It should not be used to compute what addresses are
  205. // already connected or not.
  206. //
  207. // TODO Basically, we need to work harder on our good-peer/bad-peer marking.
  208. // What we're currently doing in terms of marking good/bad peers is just a
  209. // placeholder. It should not be the case that an address becomes old/vetted
  210. // upon a single successful connection.
  211. func (r *PEXReactor) ensurePeers() {
  212. numOutPeers, _, numDialing := r.Switch.NumPeers()
  213. numToDial := minNumOutboundPeers - (numOutPeers + numDialing)
  214. r.Logger.Info("Ensure peers", "numOutPeers", numOutPeers, "numDialing", numDialing, "numToDial", numToDial)
  215. if numToDial <= 0 {
  216. return
  217. }
  218. // bias to prefer more vetted peers when we have fewer connections.
  219. // not perfect, but somewhate ensures that we prioritize connecting to more-vetted
  220. newBias := cmn.MinInt(numOutPeers, 8)*10 + 10
  221. toDial := make(map[string]*NetAddress)
  222. // Try maxAttempts times to pick numToDial addresses to dial
  223. maxAttempts := numToDial * 3
  224. for i := 0; i < maxAttempts && len(toDial) < numToDial; i++ {
  225. try := r.book.PickAddress(newBias)
  226. if try == nil {
  227. continue
  228. }
  229. if _, selected := toDial[try.IP.String()]; selected {
  230. continue
  231. }
  232. if dialling := r.Switch.IsDialing(try); dialling {
  233. continue
  234. }
  235. if connected := r.Switch.Peers().Has(try.IP.String()); connected {
  236. continue
  237. }
  238. r.Logger.Info("Will dial address", "addr", try)
  239. toDial[try.IP.String()] = try
  240. }
  241. // Dial picked addresses
  242. for _, item := range toDial {
  243. go func(picked *NetAddress) {
  244. _, err := r.Switch.DialPeerWithAddress(picked, false)
  245. if err != nil {
  246. r.book.MarkAttempt(picked)
  247. }
  248. }(item)
  249. }
  250. // If we need more addresses, pick a random peer and ask for more.
  251. if r.book.NeedMoreAddrs() {
  252. if peers := r.Switch.Peers().List(); len(peers) > 0 {
  253. i := rand.Int() % len(peers) // nolint: gas
  254. peer := peers[i]
  255. r.Logger.Info("No addresses to dial. Sending pexRequest to random peer", "peer", peer)
  256. r.RequestPEX(peer)
  257. }
  258. }
  259. }
  260. func (r *PEXReactor) flushMsgCountByPeer() {
  261. ticker := time.NewTicker(msgCountByPeerFlushInterval)
  262. for {
  263. select {
  264. case <-ticker.C:
  265. r.msgCountByPeer.Clear()
  266. case <-r.Quit:
  267. ticker.Stop()
  268. return
  269. }
  270. }
  271. }
  272. //-----------------------------------------------------------------------------
  273. // Messages
  274. const (
  275. msgTypeRequest = byte(0x01)
  276. msgTypeAddrs = byte(0x02)
  277. )
  278. // PexMessage is a primary type for PEX messages. Underneath, it could contain
  279. // either pexRequestMessage, or pexAddrsMessage messages.
  280. type PexMessage interface{}
  281. var _ = wire.RegisterInterface(
  282. struct{ PexMessage }{},
  283. wire.ConcreteType{&pexRequestMessage{}, msgTypeRequest},
  284. wire.ConcreteType{&pexAddrsMessage{}, msgTypeAddrs},
  285. )
  286. // DecodeMessage implements interface registered above.
  287. func DecodeMessage(bz []byte) (msgType byte, msg PexMessage, err error) {
  288. msgType = bz[0]
  289. n := new(int)
  290. r := bytes.NewReader(bz)
  291. msg = wire.ReadBinary(struct{ PexMessage }{}, r, maxPexMessageSize, n, &err).(struct{ PexMessage }).PexMessage
  292. return
  293. }
  294. /*
  295. A pexRequestMessage requests additional peer addresses.
  296. */
  297. type pexRequestMessage struct {
  298. }
  299. func (m *pexRequestMessage) String() string {
  300. return "[pexRequest]"
  301. }
  302. /*
  303. A message with announced peer addresses.
  304. */
  305. type pexAddrsMessage struct {
  306. Addrs []*NetAddress
  307. }
  308. func (m *pexAddrsMessage) String() string {
  309. return fmt.Sprintf("[pexAddrs %v]", m.Addrs)
  310. }