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.

265 lines
6.3 KiB

9 years ago
9 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
9 years ago
9 years ago
9 years ago
9 years ago
9 years ago
  1. package p2p
  2. import (
  3. "bytes"
  4. "errors"
  5. "fmt"
  6. "math/rand"
  7. "reflect"
  8. "time"
  9. . "github.com/tendermint/go-common"
  10. "github.com/tendermint/go-wire"
  11. )
  12. var pexErrInvalidMessage = errors.New("Invalid PEX message")
  13. const (
  14. PexChannel = byte(0x00)
  15. ensurePeersPeriodSeconds = 30
  16. minNumOutboundPeers = 10
  17. maxPexMessageSize = 1048576 // 1MB
  18. )
  19. /*
  20. PEXReactor handles PEX (peer exchange) and ensures that an
  21. adequate number of peers are connected to the switch.
  22. */
  23. type PEXReactor struct {
  24. BaseReactor
  25. sw *Switch
  26. book *AddrBook
  27. }
  28. func NewPEXReactor(book *AddrBook) *PEXReactor {
  29. pexR := &PEXReactor{
  30. book: book,
  31. }
  32. pexR.BaseReactor = *NewBaseReactor(log, "PEXReactor", pexR)
  33. return pexR
  34. }
  35. func (pexR *PEXReactor) OnStart() error {
  36. pexR.BaseReactor.OnStart()
  37. pexR.book.Start()
  38. go pexR.ensurePeersRoutine()
  39. return nil
  40. }
  41. func (pexR *PEXReactor) OnStop() {
  42. pexR.BaseReactor.OnStop()
  43. pexR.book.Stop()
  44. }
  45. // Implements Reactor
  46. func (pexR *PEXReactor) GetChannels() []*ChannelDescriptor {
  47. return []*ChannelDescriptor{
  48. &ChannelDescriptor{
  49. ID: PexChannel,
  50. Priority: 1,
  51. SendQueueCapacity: 10,
  52. },
  53. }
  54. }
  55. // Implements Reactor
  56. func (pexR *PEXReactor) AddPeer(peer *Peer) {
  57. // Add the peer to the address book
  58. netAddr, err := NewNetAddressString(peer.ListenAddr)
  59. if err != nil {
  60. // this should never happen
  61. log.Error("Error in AddPeer: invalid peer address", "addr", peer.ListenAddr, "error", err)
  62. return
  63. }
  64. if peer.IsOutbound() {
  65. if pexR.book.NeedMoreAddrs() {
  66. pexR.RequestPEX(peer)
  67. }
  68. } else {
  69. // For inbound connections, the peer is its own source
  70. // (For outbound peers, the address is already in the books)
  71. pexR.book.AddAddress(netAddr, netAddr)
  72. }
  73. }
  74. // Implements Reactor
  75. func (pexR *PEXReactor) RemovePeer(peer *Peer, reason interface{}) {
  76. // TODO
  77. }
  78. // Implements Reactor
  79. // Handles incoming PEX messages.
  80. func (pexR *PEXReactor) Receive(chID byte, src *Peer, msgBytes []byte) {
  81. // decode message
  82. _, msg, err := DecodeMessage(msgBytes)
  83. if err != nil {
  84. log.Warn("Error decoding message", "error", err)
  85. return
  86. }
  87. log.Notice("Received message", "msg", msg)
  88. switch msg := msg.(type) {
  89. case *pexRequestMessage:
  90. // src requested some peers.
  91. // TODO: prevent abuse.
  92. pexR.SendAddrs(src, pexR.book.GetSelection())
  93. case *pexAddrsMessage:
  94. // We received some peer addresses from src.
  95. // TODO: prevent abuse.
  96. // (We don't want to get spammed with bad peers)
  97. srcAddr := src.Connection().RemoteAddress
  98. for _, addr := range msg.Addrs {
  99. if addr != nil {
  100. pexR.book.AddAddress(addr, srcAddr)
  101. }
  102. }
  103. default:
  104. log.Warn(Fmt("Unknown message type %v", reflect.TypeOf(msg)))
  105. }
  106. }
  107. // Asks peer for more addresses.
  108. func (pexR *PEXReactor) RequestPEX(peer *Peer) {
  109. peer.Send(PexChannel, struct{ PexMessage }{&pexRequestMessage{}})
  110. }
  111. func (pexR *PEXReactor) SendAddrs(peer *Peer, addrs []*NetAddress) {
  112. peer.Send(PexChannel, struct{ PexMessage }{&pexAddrsMessage{Addrs: addrs}})
  113. }
  114. // Ensures that sufficient peers are connected. (continuous)
  115. func (pexR *PEXReactor) ensurePeersRoutine() {
  116. // Randomize when routine starts
  117. time.Sleep(time.Duration(rand.Int63n(500*ensurePeersPeriodSeconds)) * time.Millisecond)
  118. // fire once immediately.
  119. pexR.ensurePeers()
  120. // fire periodically
  121. timer := NewRepeatTimer("pex", ensurePeersPeriodSeconds*time.Second)
  122. FOR_LOOP:
  123. for {
  124. select {
  125. case <-timer.Ch:
  126. pexR.ensurePeers()
  127. case <-pexR.Quit:
  128. break FOR_LOOP
  129. }
  130. }
  131. // Cleanup
  132. timer.Stop()
  133. }
  134. // Ensures that sufficient peers are connected. (once)
  135. func (pexR *PEXReactor) ensurePeers() {
  136. numOutPeers, _, numDialing := pexR.Switch.NumPeers()
  137. numToDial := minNumOutboundPeers - (numOutPeers + numDialing)
  138. log.Info("Ensure peers", "numOutPeers", numOutPeers, "numDialing", numDialing, "numToDial", numToDial)
  139. if numToDial <= 0 {
  140. return
  141. }
  142. toDial := NewCMap()
  143. // Try to pick numToDial addresses to dial.
  144. // TODO: improve logic.
  145. for i := 0; i < numToDial; i++ {
  146. newBias := MinInt(numOutPeers, 8)*10 + 10
  147. var picked *NetAddress
  148. // Try to fetch a new peer 3 times.
  149. // This caps the maximum number of tries to 3 * numToDial.
  150. for j := 0; j < 3; j++ {
  151. try := pexR.book.PickAddress(newBias)
  152. if try == nil {
  153. break
  154. }
  155. alreadySelected := toDial.Has(try.IP.String())
  156. alreadyDialing := pexR.Switch.IsDialing(try)
  157. alreadyConnected := pexR.Switch.Peers().Has(try.IP.String())
  158. if alreadySelected || alreadyDialing || alreadyConnected {
  159. /*
  160. log.Info("Cannot dial address", "addr", try,
  161. "alreadySelected", alreadySelected,
  162. "alreadyDialing", alreadyDialing,
  163. "alreadyConnected", alreadyConnected)
  164. */
  165. continue
  166. } else {
  167. log.Info("Will dial address", "addr", try)
  168. picked = try
  169. break
  170. }
  171. }
  172. if picked == nil {
  173. continue
  174. }
  175. toDial.Set(picked.IP.String(), picked)
  176. }
  177. // Dial picked addresses
  178. for _, item := range toDial.Values() {
  179. go func(picked *NetAddress) {
  180. _, err := pexR.Switch.DialPeerWithAddress(picked)
  181. if err != nil {
  182. pexR.book.MarkAttempt(picked)
  183. }
  184. }(item.(*NetAddress))
  185. }
  186. // If we need more addresses, pick a random peer and ask for more.
  187. if pexR.book.NeedMoreAddrs() {
  188. if peers := pexR.Switch.Peers().List(); len(peers) > 0 {
  189. i := rand.Int() % len(peers)
  190. peer := peers[i]
  191. log.Info("No addresses to dial. Sending pexRequest to random peer", "peer", peer)
  192. pexR.RequestPEX(peer)
  193. }
  194. }
  195. }
  196. //-----------------------------------------------------------------------------
  197. // Messages
  198. const (
  199. msgTypeRequest = byte(0x01)
  200. msgTypeAddrs = byte(0x02)
  201. )
  202. type PexMessage interface{}
  203. var _ = wire.RegisterInterface(
  204. struct{ PexMessage }{},
  205. wire.ConcreteType{&pexRequestMessage{}, msgTypeRequest},
  206. wire.ConcreteType{&pexAddrsMessage{}, msgTypeAddrs},
  207. )
  208. func DecodeMessage(bz []byte) (msgType byte, msg PexMessage, err error) {
  209. msgType = bz[0]
  210. n := new(int)
  211. r := bytes.NewReader(bz)
  212. msg = wire.ReadBinary(struct{ PexMessage }{}, r, maxPexMessageSize, n, &err).(struct{ PexMessage }).PexMessage
  213. return
  214. }
  215. /*
  216. A pexRequestMessage requests additional peer addresses.
  217. */
  218. type pexRequestMessage struct {
  219. }
  220. func (m *pexRequestMessage) String() string {
  221. return "[pexRequest]"
  222. }
  223. /*
  224. A message with announced peer addresses.
  225. */
  226. type pexAddrsMessage struct {
  227. Addrs []*NetAddress
  228. }
  229. func (m *pexAddrsMessage) String() string {
  230. return fmt.Sprintf("[pexAddrs %v]", m.Addrs)
  231. }