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.

247 lines
5.2 KiB

10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
10 years ago
  1. package p2p
  2. import (
  3. "bytes"
  4. "errors"
  5. "fmt"
  6. "sync/atomic"
  7. "time"
  8. . "github.com/tendermint/tendermint/binary"
  9. . "github.com/tendermint/tendermint/common"
  10. )
  11. var pexErrInvalidMessage = errors.New("Invalid PEX message")
  12. const (
  13. PexCh = byte(0x00)
  14. ensurePeersPeriodSeconds = 30
  15. minNumOutboundPeers = 10
  16. maxNumPeers = 50
  17. )
  18. /*
  19. PEXReactor handles PEX (peer exchange) and ensures that an
  20. adequate number of peers are connected to the switch.
  21. */
  22. type PEXReactor struct {
  23. sw *Switch
  24. quit chan struct{}
  25. started uint32
  26. stopped uint32
  27. book *AddrBook
  28. }
  29. func NewPEXReactor(book *AddrBook) *PEXReactor {
  30. pexR := &PEXReactor{
  31. quit: make(chan struct{}),
  32. book: book,
  33. }
  34. return pexR
  35. }
  36. // Implements Reactor
  37. func (pexR *PEXReactor) Start(sw *Switch) {
  38. if atomic.CompareAndSwapUint32(&pexR.started, 0, 1) {
  39. log.Info("Starting PEXReactor")
  40. pexR.sw = sw
  41. go pexR.ensurePeersRoutine()
  42. }
  43. }
  44. // Implements Reactor
  45. func (pexR *PEXReactor) Stop() {
  46. if atomic.CompareAndSwapUint32(&pexR.stopped, 0, 1) {
  47. log.Info("Stopping PEXReactor")
  48. close(pexR.quit)
  49. }
  50. }
  51. // Implements Reactor
  52. func (pexR *PEXReactor) GetChannels() []*ChannelDescriptor {
  53. return []*ChannelDescriptor{
  54. &ChannelDescriptor{
  55. Id: PexCh,
  56. Priority: 1,
  57. },
  58. }
  59. }
  60. // Implements Reactor
  61. func (pexR *PEXReactor) AddPeer(peer *Peer) {
  62. if peer.IsOutbound() {
  63. pexR.SendAddrs(peer, pexR.book.OurAddresses())
  64. if pexR.book.NeedMoreAddrs() {
  65. pexR.RequestPEX(peer)
  66. }
  67. }
  68. }
  69. // Implements Reactor
  70. func (pexR *PEXReactor) RemovePeer(peer *Peer, reason interface{}) {
  71. // TODO
  72. }
  73. // Implements Reactor
  74. // Handles incoming PEX messages.
  75. func (pexR *PEXReactor) Receive(chId byte, src *Peer, msgBytes []byte) {
  76. // decode message
  77. msg, err := DecodeMessage(msgBytes)
  78. if err != nil {
  79. log.Warn("Error decoding message", "error", err)
  80. return
  81. }
  82. log.Info("Received message", "msg", msg)
  83. switch msg.(type) {
  84. case *pexRequestMessage:
  85. // src requested some peers.
  86. // TODO: prevent abuse.
  87. addrs := pexR.book.GetSelection()
  88. msg := &pexAddrsMessage{Addrs: addrs}
  89. queued := src.TrySend(PexCh, msg)
  90. if !queued {
  91. // ignore
  92. }
  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.(*pexAddrsMessage).Addrs {
  99. pexR.book.AddAddress(addr, srcAddr)
  100. }
  101. default:
  102. // Ignore unknown message.
  103. }
  104. }
  105. // Asks peer for more addresses.
  106. func (pexR *PEXReactor) RequestPEX(peer *Peer) {
  107. peer.TrySend(PexCh, &pexRequestMessage{})
  108. }
  109. func (pexR *PEXReactor) SendAddrs(peer *Peer, addrs []*NetAddress) {
  110. peer.Send(PexCh, &pexAddrsMessage{Addrs: addrs})
  111. }
  112. // Ensures that sufficient peers are connected. (continuous)
  113. func (pexR *PEXReactor) ensurePeersRoutine() {
  114. // fire once immediately.
  115. pexR.ensurePeers()
  116. // fire periodically
  117. timer := NewRepeatTimer(ensurePeersPeriodSeconds * time.Second)
  118. FOR_LOOP:
  119. for {
  120. select {
  121. case <-timer.Ch:
  122. pexR.ensurePeers()
  123. case <-pexR.quit:
  124. break FOR_LOOP
  125. }
  126. }
  127. // Cleanup
  128. timer.Stop()
  129. }
  130. // Ensures that sufficient peers are connected. (once)
  131. func (pexR *PEXReactor) ensurePeers() {
  132. numOutPeers, _, numDialing := pexR.sw.NumPeers()
  133. numToDial := minNumOutboundPeers - (numOutPeers + numDialing)
  134. if numToDial <= 0 {
  135. return
  136. }
  137. toDial := NewCMap()
  138. // Try to pick numToDial addresses to dial.
  139. // TODO: improve logic.
  140. for i := 0; i < numToDial; i++ {
  141. newBias := MinInt(numOutPeers, 8)*10 + 10
  142. var picked *NetAddress
  143. // Try to fetch a new peer 3 times.
  144. // This caps the maximum number of tries to 3 * numToDial.
  145. for j := 0; i < 3; j++ {
  146. picked = pexR.book.PickAddress(newBias)
  147. if picked == nil {
  148. return
  149. }
  150. if toDial.Has(picked.String()) ||
  151. pexR.sw.IsDialing(picked) ||
  152. pexR.sw.Peers().Has(picked.String()) {
  153. continue
  154. } else {
  155. break
  156. }
  157. }
  158. if picked == nil {
  159. continue
  160. }
  161. toDial.Set(picked.String(), picked)
  162. }
  163. // Dial picked addresses
  164. for _, item := range toDial.Values() {
  165. picked := item.(*NetAddress)
  166. go func() {
  167. _, err := pexR.sw.DialPeerWithAddress(picked)
  168. if err != nil {
  169. pexR.book.MarkAttempt(picked)
  170. }
  171. }()
  172. }
  173. }
  174. //-----------------------------------------------------------------------------
  175. // Messages
  176. const (
  177. msgTypeUnknown = byte(0x00)
  178. msgTypeRequest = byte(0x01)
  179. msgTypeAddrs = byte(0x02)
  180. )
  181. // TODO: check for unnecessary extra bytes at the end.
  182. func DecodeMessage(bz []byte) (msg interface{}, err error) {
  183. n := new(int64)
  184. msgType := bz[0]
  185. r := bytes.NewReader(bz)
  186. // log.Debug(Fmt("decoding msg bytes: %X", bz))
  187. switch msgType {
  188. case msgTypeRequest:
  189. msg = &pexRequestMessage{}
  190. case msgTypeAddrs:
  191. msg = ReadBinary(&pexAddrsMessage{}, r, n, &err)
  192. default:
  193. msg = nil
  194. }
  195. return
  196. }
  197. /*
  198. A pexRequestMessage requests additional peer addresses.
  199. */
  200. type pexRequestMessage struct {
  201. }
  202. func (m *pexRequestMessage) TypeByte() byte { return msgTypeRequest }
  203. func (m *pexRequestMessage) String() string {
  204. return "[pexRequest]"
  205. }
  206. /*
  207. A message with announced peer addresses.
  208. */
  209. type pexAddrsMessage struct {
  210. Addrs []*NetAddress
  211. }
  212. func (m *pexAddrsMessage) TypeByte() byte { return msgTypeAddrs }
  213. func (m *pexAddrsMessage) String() string {
  214. return fmt.Sprintf("[pexAddrs %v]", m.Addrs)
  215. }