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.

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