p2p/pex_reactor.go

   1 package p2p
   2
   3 import (
   4         "bytes"
   5         "fmt"
   6         "math/rand"
   7         "reflect"
   8         "time"
   9
  10         log "github.com/sirupsen/logrus"
  11         wire "github.com/tendermint/go-wire"
  12         cmn "github.com/tendermint/tmlibs/common"
  13 )
  14
  15 const (
  16         // PexChannel is a channel for PEX messages
  17         PexChannel = byte(0x00)
  18
  19         // period to ensure peers connected
  20         defaultEnsurePeersPeriod = 30 * time.Second
  21         minNumOutboundPeers      = 10
  22         maxPexMessageSize        = 1048576 // 1MB
  23
  24         // maximum messages one peer can send to us during `msgCountByPeerFlushInterval`
  25         defaultMaxMsgCountByPeer    = 1000
  26         msgCountByPeerFlushInterval = 1 * time.Hour
  27 )
  28
  29 // PEXReactor handles PEX (peer exchange) and ensures that an
  30 // adequate number of peers are connected to the switch.
  31 //
  32 // It uses `AddrBook` (address book) to store `NetAddress`es of the peers.
  33 //
  34 // ## Preventing abuse
  35 //
  36 // For now, it just limits the number of messages from one peer to
  37 // `defaultMaxMsgCountByPeer` messages per `msgCountByPeerFlushInterval` (1000
  38 // msg/hour).
  39 //
  40 // NOTE [2017-01-17]:
  41 //   Limiting is fine for now. Maybe down the road we want to keep track of the
  42 //   quality of peer messages so if peerA keeps telling us about peers we can't
  43 //   connect to then maybe we should care less about peerA. But I don't think
  44 //   that kind of complexity is priority right now.
  45 type PEXReactor struct {
  46         BaseReactor
  47
  48         sw                *Switch
  49         book              *AddrBook
  50         ensurePeersPeriod time.Duration
  51
  52         // tracks message count by peer, so we can prevent abuse
  53         msgCountByPeer    *cmn.CMap
  54         maxMsgCountByPeer uint16
  55 }
  56
  57 // NewPEXReactor creates new PEX reactor.
  58 func NewPEXReactor(b *AddrBook) *PEXReactor {
  59         r := &PEXReactor{
  60                 book:              b,
  61                 ensurePeersPeriod: defaultEnsurePeersPeriod,
  62                 msgCountByPeer:    cmn.NewCMap(),
  63                 maxMsgCountByPeer: defaultMaxMsgCountByPeer,
  64         }
  65         r.BaseReactor = *NewBaseReactor("PEXReactor", r)
  66         return r
  67 }
  68
  69 // OnStart implements BaseService
  70 func (r *PEXReactor) OnStart() error {
  71         r.BaseReactor.OnStart()
  72         r.book.Start()
  73         go r.ensurePeersRoutine()
  74         go r.flushMsgCountByPeer()
  75         return nil
  76 }
  77
  78 // OnStop implements BaseService
  79 func (r *PEXReactor) OnStop() {
  80         r.BaseReactor.OnStop()
  81         r.book.Stop()
  82 }
  83
  84 // GetChannels implements Reactor
  85 func (r *PEXReactor) GetChannels() []*ChannelDescriptor {
  86         return []*ChannelDescriptor{
  87                 &ChannelDescriptor{
  88                         ID:                PexChannel,
  89                         Priority:          1,
  90                         SendQueueCapacity: 10,
  91                 },
  92         }
  93 }
  94
  95 // AddPeer implements Reactor by adding peer to the address book (if inbound)
  96 // or by requesting more addresses (if outbound).
  97 func (r *PEXReactor) AddPeer(p *Peer) {
  98         if p.IsOutbound() {
  99                 // For outbound peers, the address is already in the books.
 100                 // Either it was added in DialSeeds or when we
 101                 // received the peer's address in r.Receive
 102                 if r.book.NeedMoreAddrs() {
 103                         r.RequestPEX(p)
 104                 }
 105         } else { // For inbound connections, the peer is its own source
 106                 addr, err := NewNetAddressString(p.ListenAddr)
 107                 if err != nil {
 108                         // this should never happen
 109                         log.WithFields(log.Fields{
 110                                 "addr":  p.ListenAddr,
 111                                 "error": err,
 112                         }).Error("Error in AddPeer: Invalid peer address")
 113                         return
 114                 }
 115                 r.book.AddAddress(addr, addr)
 116         }
 117 }
 118
 119 // RemovePeer implements Reactor.
 120 func (r *PEXReactor) RemovePeer(p *Peer, reason interface{}) {
 121         // If we aren't keeping track of local temp data for each peer here, then we
 122         // don't have to do anything.
 123 }
 124
 125 // Receive implements Reactor by handling incoming PEX messages.
 126 func (r *PEXReactor) Receive(chID byte, src *Peer, msgBytes []byte) {
 127         srcAddr := src.Connection().RemoteAddress
 128         srcAddrStr := srcAddr.String()
 129
 130         r.IncrementMsgCountForPeer(srcAddrStr)
 131         if r.ReachedMaxMsgCountForPeer(srcAddrStr) {
 132                 log.WithField("peer", srcAddrStr).Error("Maximum number of messages reached for peer")
 133                 // TODO remove src from peers?
 134                 return
 135         }
 136
 137         _, msg, err := DecodeMessage(msgBytes)
 138         if err != nil {
 139                 log.WithField("error", err).Error("Error decoding message")
 140                 return
 141         }
 142         log.WithField("msg", msg).Info("Reveived message")
 143
 144         switch msg := msg.(type) {
 145         case *pexRequestMessage:
 146                 // src requested some peers.
 147                 r.SendAddrs(src, r.book.GetSelection())
 148         case *pexAddrsMessage:
 149                 // We received some peer addresses from src.
 150                 // (We don't want to get spammed with bad peers)
 151                 for _, addr := range msg.Addrs {
 152                         if addr != nil {
 153                                 r.book.AddAddress(addr, srcAddr)
 154                         }
 155                 }
 156         default:
 157                 log.WithField("type", reflect.TypeOf(msg)).Error("Unknown message type")
 158         }
 159 }
 160
 161 // RequestPEX asks peer for more addresses.
 162 func (r *PEXReactor) RequestPEX(p *Peer) {
 163         p.Send(PexChannel, struct{ PexMessage }{&pexRequestMessage{}})
 164 }
 165
 166 // SendAddrs sends addrs to the peer.
 167 func (r *PEXReactor) SendAddrs(p *Peer, addrs []*NetAddress) {
 168         p.Send(PexChannel, struct{ PexMessage }{&pexAddrsMessage{Addrs: addrs}})
 169 }
 170
 171 // SetEnsurePeersPeriod sets period to ensure peers connected.
 172 func (r *PEXReactor) SetEnsurePeersPeriod(d time.Duration) {
 173         r.ensurePeersPeriod = d
 174 }
 175
 176 // SetMaxMsgCountByPeer sets maximum messages one peer can send to us during 'msgCountByPeerFlushInterval'.
 177 func (r *PEXReactor) SetMaxMsgCountByPeer(v uint16) {
 178         r.maxMsgCountByPeer = v
 179 }
 180
 181 // ReachedMaxMsgCountForPeer returns true if we received too many
 182 // messages from peer with address `addr`.
 183 // NOTE: assumes the value in the CMap is non-nil
 184 func (r *PEXReactor) ReachedMaxMsgCountForPeer(addr string) bool {
 185         return r.msgCountByPeer.Get(addr).(uint16) >= r.maxMsgCountByPeer
 186 }
 187
 188 // Increment or initialize the msg count for the peer in the CMap
 189 func (r *PEXReactor) IncrementMsgCountForPeer(addr string) {
 190         var count uint16
 191         countI := r.msgCountByPeer.Get(addr)
 192         if countI != nil {
 193                 count = countI.(uint16)
 194         }
 195         count++
 196         r.msgCountByPeer.Set(addr, count)
 197 }
 198
 199 // Ensures that sufficient peers are connected. (continuous)
 200 func (r *PEXReactor) ensurePeersRoutine() {
 201         // Randomize when routine starts
 202         ensurePeersPeriodMs := r.ensurePeersPeriod.Nanoseconds() / 1e6
 203         time.Sleep(time.Duration(rand.Int63n(ensurePeersPeriodMs)) * time.Millisecond)
 204
 205         // fire once immediately.
 206         r.ensurePeers()
 207
 208         // fire periodically
 209         ticker := time.NewTicker(r.ensurePeersPeriod)
 210
 211         for {
 212                 select {
 213                 case <-ticker.C:
 214                         r.ensurePeers()
 215                 case <-r.Quit:
 216                         ticker.Stop()
 217                         return
 218                 }
 219         }
 220 }
 221
 222 // ensurePeers ensures that sufficient peers are connected. (once)
 223 //
 224 // Old bucket / New bucket are arbitrary categories to denote whether an
 225 // address is vetted or not, and this needs to be determined over time via a
 226 // heuristic that we haven't perfected yet, or, perhaps is manually edited by
 227 // the node operator. It should not be used to compute what addresses are
 228 // already connected or not.
 229 //
 230 // TODO Basically, we need to work harder on our good-peer/bad-peer marking.
 231 // What we're currently doing in terms of marking good/bad peers is just a
 232 // placeholder. It should not be the case that an address becomes old/vetted
 233 // upon a single successful connection.
 234 func (r *PEXReactor) ensurePeers() {
 235         numOutPeers, _, numDialing := r.Switch.NumPeers()
 236         numToDial := minNumOutboundPeers - (numOutPeers + numDialing)
 237         log.WithFields(log.Fields{
 238                 "numOutPeers": numOutPeers,
 239                 "numDialing":  numDialing,
 240                 "numToDial":   numToDial,
 241         }).Info("Ensure peers")
 242         if numToDial <= 0 {
 243                 return
 244         }
 245
 246         toDial := make(map[string]*NetAddress)
 247
 248         // Try to pick numToDial addresses to dial.
 249         for i := 0; i < numToDial; i++ {
 250                 // The purpose of newBias is to first prioritize old (more vetted) peers
 251                 // when we have few connections, but to allow for new (less vetted) peers
 252                 // if we already have many connections. This algorithm isn't perfect, but
 253                 // it somewhat ensures that we prioritize connecting to more-vetted
 254                 // peers.
 255                 newBias := cmn.MinInt(numOutPeers, 8)*10 + 10
 256                 var picked *NetAddress
 257                 // Try to fetch a new peer 3 times.
 258                 // This caps the maximum number of tries to 3 * numToDial.
 259                 for j := 0; j < 3; j++ {
 260                         try := r.book.PickAddress(newBias)
 261                         if try == nil {
 262                                 break
 263                         }
 264                         _, alreadySelected := toDial[try.IP.String()]
 265                         alreadyDialing := r.Switch.IsDialing(try)
 266                         alreadyConnected := r.Switch.Peers().Has(try.IP.String())
 267                         if alreadySelected || alreadyDialing || alreadyConnected {
 268                                 continue
 269                         } else {
 270                                 log.WithField("addr", try).Info("Will dial address")
 271                                 picked = try
 272                                 break
 273                         }
 274                 }
 275                 if picked == nil {
 276                         continue
 277                 }
 278                 toDial[picked.IP.String()] = picked
 279         }
 280
 281         // Dial picked addresses
 282         for _, item := range toDial {
 283                 go func(picked *NetAddress) {
 284                         _, err := r.Switch.DialPeerWithAddress(picked, false)
 285                         if err != nil {
 286                                 r.book.MarkAttempt(picked)
 287                         }
 288                 }(item)
 289         }
 290
 291         // If we need more addresses, pick a random peer and ask for more.
 292         if r.book.NeedMoreAddrs() {
 293                 if peers := r.Switch.Peers().List(); len(peers) > 0 {
 294                         i := rand.Int() % len(peers)
 295                         peer := peers[i]
 296                         log.WithField("peer", peer).Info("No addresses to dial. Sending pexRequest to random peer")
 297                         r.RequestPEX(peer)
 298                 }
 299         }
 300 }
 301
 302 func (r *PEXReactor) flushMsgCountByPeer() {
 303         ticker := time.NewTicker(msgCountByPeerFlushInterval)
 304
 305         for {
 306                 select {
 307                 case <-ticker.C:
 308                         r.msgCountByPeer.Clear()
 309                 case <-r.Quit:
 310                         ticker.Stop()
 311                         return
 312                 }
 313         }
 314 }
 315
 316 //-----------------------------------------------------------------------------
 317 // Messages
 318
 319 const (
 320         msgTypeRequest = byte(0x01)
 321         msgTypeAddrs   = byte(0x02)
 322 )
 323
 324 // PexMessage is a primary type for PEX messages. Underneath, it could contain
 325 // either pexRequestMessage, or pexAddrsMessage messages.
 326 type PexMessage interface{}
 327
 328 var _ = wire.RegisterInterface(
 329         struct{ PexMessage }{},
 330         wire.ConcreteType{&pexRequestMessage{}, msgTypeRequest},
 331         wire.ConcreteType{&pexAddrsMessage{}, msgTypeAddrs},
 332 )
 333
 334 // DecodeMessage implements interface registered above.
 335 func DecodeMessage(bz []byte) (msgType byte, msg PexMessage, err error) {
 336         msgType = bz[0]
 337         n := new(int)
 338         r := bytes.NewReader(bz)
 339         msg = wire.ReadBinary(struct{ PexMessage }{}, r, maxPexMessageSize, n, &err).(struct{ PexMessage }).PexMessage
 340         return
 341 }
 342
 343 /*
 344 A pexRequestMessage requests additional peer addresses.
 345 */
 346 type pexRequestMessage struct {
 347 }
 348
 349 func (m *pexRequestMessage) String() string {
 350         return "[pexRequest]"
 351 }
 352
 353 /*
 354 A message with announced peer addresses.
 355 */
 356 type pexAddrsMessage struct {
 357         Addrs []*NetAddress
 358 }
 359
 360 func (m *pexAddrsMessage) String() string {
 361         return fmt.Sprintf("[pexAddrs %v]", m.Addrs)
 362 }