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.

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