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.

357 lines
10 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
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
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
8 years ago
9 years ago
8 years ago
9 years ago
8 years ago
8 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
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. r.BaseReactor.OnStart()
  60. r.book.Start()
  61. go r.ensurePeersRoutine()
  62. go r.flushMsgCountByPeer()
  63. return nil
  64. }
  65. // OnStop implements BaseService
  66. func (r *PEXReactor) OnStop() {
  67. r.BaseReactor.OnStop()
  68. r.book.Stop()
  69. }
  70. // GetChannels implements Reactor
  71. func (r *PEXReactor) GetChannels() []*ChannelDescriptor {
  72. return []*ChannelDescriptor{
  73. &ChannelDescriptor{
  74. ID: PexChannel,
  75. Priority: 1,
  76. SendQueueCapacity: 10,
  77. },
  78. }
  79. }
  80. // AddPeer implements Reactor by adding peer to the address book (if inbound)
  81. // or by requesting more addresses (if outbound).
  82. func (r *PEXReactor) AddPeer(p Peer) {
  83. if p.IsOutbound() {
  84. // For outbound peers, the address is already in the books.
  85. // Either it was added in DialSeeds or when we
  86. // received the peer's address in r.Receive
  87. if r.book.NeedMoreAddrs() {
  88. r.RequestPEX(p)
  89. }
  90. } else { // For inbound connections, the peer is its own source
  91. addr, err := NewNetAddressString(p.NodeInfo().ListenAddr)
  92. if err != nil {
  93. // this should never happen
  94. r.Logger.Error("Error in AddPeer: invalid peer address", "addr", p.NodeInfo().ListenAddr, "err", err)
  95. return
  96. }
  97. r.book.AddAddress(addr, addr)
  98. }
  99. }
  100. // RemovePeer implements Reactor.
  101. func (r *PEXReactor) RemovePeer(p Peer, reason interface{}) {
  102. // If we aren't keeping track of local temp data for each peer here, then we
  103. // don't have to do anything.
  104. }
  105. // Receive implements Reactor by handling incoming PEX messages.
  106. func (r *PEXReactor) Receive(chID byte, src Peer, msgBytes []byte) {
  107. srcAddrStr := src.NodeInfo().RemoteAddr
  108. srcAddr, _ := NewNetAddressString(srcAddrStr)
  109. r.IncrementMsgCountForPeer(srcAddrStr)
  110. if r.ReachedMaxMsgCountForPeer(srcAddrStr) {
  111. r.Logger.Error("Maximum number of messages reached for peer", "peer", srcAddrStr)
  112. // TODO remove src from peers?
  113. return
  114. }
  115. _, msg, err := DecodeMessage(msgBytes)
  116. if err != nil {
  117. r.Logger.Error("Error decoding message", "err", err)
  118. return
  119. }
  120. r.Logger.Info("Received message", "msg", msg)
  121. switch msg := msg.(type) {
  122. case *pexRequestMessage:
  123. // src requested some peers.
  124. r.SendAddrs(src, r.book.GetSelection())
  125. case *pexAddrsMessage:
  126. // We received some peer addresses from src.
  127. // (We don't want to get spammed with bad peers)
  128. for _, addr := range msg.Addrs {
  129. if addr != nil {
  130. r.book.AddAddress(addr, srcAddr)
  131. }
  132. }
  133. default:
  134. r.Logger.Error(fmt.Sprintf("Unknown message type %v", reflect.TypeOf(msg)))
  135. }
  136. }
  137. // RequestPEX asks peer for more addresses.
  138. func (r *PEXReactor) RequestPEX(p Peer) {
  139. p.Send(PexChannel, struct{ PexMessage }{&pexRequestMessage{}})
  140. }
  141. // SendAddrs sends addrs to the peer.
  142. func (r *PEXReactor) SendAddrs(p Peer, addrs []*NetAddress) {
  143. p.Send(PexChannel, struct{ PexMessage }{&pexAddrsMessage{Addrs: addrs}})
  144. }
  145. // SetEnsurePeersPeriod sets period to ensure peers connected.
  146. func (r *PEXReactor) SetEnsurePeersPeriod(d time.Duration) {
  147. r.ensurePeersPeriod = d
  148. }
  149. // SetMaxMsgCountByPeer sets maximum messages one peer can send to us during 'msgCountByPeerFlushInterval'.
  150. func (r *PEXReactor) SetMaxMsgCountByPeer(v uint16) {
  151. r.maxMsgCountByPeer = v
  152. }
  153. // ReachedMaxMsgCountForPeer returns true if we received too many
  154. // messages from peer with address `addr`.
  155. // NOTE: assumes the value in the CMap is non-nil
  156. func (r *PEXReactor) ReachedMaxMsgCountForPeer(addr string) bool {
  157. return r.msgCountByPeer.Get(addr).(uint16) >= r.maxMsgCountByPeer
  158. }
  159. // Increment or initialize the msg count for the peer in the CMap
  160. func (r *PEXReactor) IncrementMsgCountForPeer(addr string) {
  161. var count uint16
  162. countI := r.msgCountByPeer.Get(addr)
  163. if countI != nil {
  164. count = countI.(uint16)
  165. }
  166. count++
  167. r.msgCountByPeer.Set(addr, count)
  168. }
  169. // Ensures that sufficient peers are connected. (continuous)
  170. func (r *PEXReactor) ensurePeersRoutine() {
  171. // Randomize when routine starts
  172. ensurePeersPeriodMs := r.ensurePeersPeriod.Nanoseconds() / 1e6
  173. time.Sleep(time.Duration(rand.Int63n(ensurePeersPeriodMs)) * time.Millisecond)
  174. // fire once immediately.
  175. r.ensurePeers()
  176. // fire periodically
  177. ticker := time.NewTicker(r.ensurePeersPeriod)
  178. for {
  179. select {
  180. case <-ticker.C:
  181. r.ensurePeers()
  182. case <-r.Quit:
  183. ticker.Stop()
  184. return
  185. }
  186. }
  187. }
  188. // ensurePeers ensures that sufficient peers are connected. (once)
  189. //
  190. // Old bucket / New bucket are arbitrary categories to denote whether an
  191. // address is vetted or not, and this needs to be determined over time via a
  192. // heuristic that we haven't perfected yet, or, perhaps is manually edited by
  193. // the node operator. It should not be used to compute what addresses are
  194. // already connected or not.
  195. //
  196. // TODO Basically, we need to work harder on our good-peer/bad-peer marking.
  197. // What we're currently doing in terms of marking good/bad peers is just a
  198. // placeholder. It should not be the case that an address becomes old/vetted
  199. // upon a single successful connection.
  200. func (r *PEXReactor) ensurePeers() {
  201. numOutPeers, _, numDialing := r.Switch.NumPeers()
  202. numToDial := minNumOutboundPeers - (numOutPeers + numDialing)
  203. r.Logger.Info("Ensure peers", "numOutPeers", numOutPeers, "numDialing", numDialing, "numToDial", numToDial)
  204. if numToDial <= 0 {
  205. return
  206. }
  207. toDial := make(map[string]*NetAddress)
  208. // Try to pick numToDial addresses to dial.
  209. for i := 0; i < numToDial; i++ {
  210. // The purpose of newBias is to first prioritize old (more vetted) peers
  211. // when we have few connections, but to allow for new (less vetted) peers
  212. // if we already have many connections. This algorithm isn't perfect, but
  213. // it somewhat ensures that we prioritize connecting to more-vetted
  214. // peers.
  215. newBias := cmn.MinInt(numOutPeers, 8)*10 + 10
  216. var picked *NetAddress
  217. // Try to fetch a new peer 3 times.
  218. // This caps the maximum number of tries to 3 * numToDial.
  219. for j := 0; j < 3; j++ {
  220. try := r.book.PickAddress(newBias)
  221. if try == nil {
  222. break
  223. }
  224. _, alreadySelected := toDial[try.IP.String()]
  225. alreadyDialing := r.Switch.IsDialing(try)
  226. alreadyConnected := r.Switch.Peers().Has(try.IP.String())
  227. if alreadySelected || alreadyDialing || alreadyConnected {
  228. // r.Logger.Info("Cannot dial address", "addr", try,
  229. // "alreadySelected", alreadySelected,
  230. // "alreadyDialing", alreadyDialing,
  231. // "alreadyConnected", alreadyConnected)
  232. continue
  233. } else {
  234. r.Logger.Info("Will dial address", "addr", try)
  235. picked = try
  236. break
  237. }
  238. }
  239. if picked == nil {
  240. continue
  241. }
  242. toDial[picked.IP.String()] = picked
  243. }
  244. // Dial picked addresses
  245. for _, item := range toDial {
  246. go func(picked *NetAddress) {
  247. _, err := r.Switch.DialPeerWithAddress(picked, false)
  248. if err != nil {
  249. r.book.MarkAttempt(picked)
  250. }
  251. }(item)
  252. }
  253. // If we need more addresses, pick a random peer and ask for more.
  254. if r.book.NeedMoreAddrs() {
  255. if peers := r.Switch.Peers().List(); len(peers) > 0 {
  256. i := rand.Int() % len(peers)
  257. peer := peers[i]
  258. r.Logger.Info("No addresses to dial. Sending pexRequest to random peer", "peer", peer)
  259. r.RequestPEX(peer)
  260. }
  261. }
  262. }
  263. func (r *PEXReactor) flushMsgCountByPeer() {
  264. ticker := time.NewTicker(msgCountByPeerFlushInterval)
  265. for {
  266. select {
  267. case <-ticker.C:
  268. r.msgCountByPeer.Clear()
  269. case <-r.Quit:
  270. ticker.Stop()
  271. return
  272. }
  273. }
  274. }
  275. //-----------------------------------------------------------------------------
  276. // Messages
  277. const (
  278. msgTypeRequest = byte(0x01)
  279. msgTypeAddrs = byte(0x02)
  280. )
  281. // PexMessage is a primary type for PEX messages. Underneath, it could contain
  282. // either pexRequestMessage, or pexAddrsMessage messages.
  283. type PexMessage interface{}
  284. var _ = wire.RegisterInterface(
  285. struct{ PexMessage }{},
  286. wire.ConcreteType{&pexRequestMessage{}, msgTypeRequest},
  287. wire.ConcreteType{&pexAddrsMessage{}, msgTypeAddrs},
  288. )
  289. // DecodeMessage implements interface registered above.
  290. func DecodeMessage(bz []byte) (msgType byte, msg PexMessage, err error) {
  291. msgType = bz[0]
  292. n := new(int)
  293. r := bytes.NewReader(bz)
  294. msg = wire.ReadBinary(struct{ PexMessage }{}, r, maxPexMessageSize, n, &err).(struct{ PexMessage }).PexMessage
  295. return
  296. }
  297. /*
  298. A pexRequestMessage requests additional peer addresses.
  299. */
  300. type pexRequestMessage struct {
  301. }
  302. func (m *pexRequestMessage) String() string {
  303. return "[pexRequest]"
  304. }
  305. /*
  306. A message with announced peer addresses.
  307. */
  308. type pexAddrsMessage struct {
  309. Addrs []*NetAddress
  310. }
  311. func (m *pexAddrsMessage) String() string {
  312. return fmt.Sprintf("[pexAddrs %v]", m.Addrs)
  313. }