package p2p import ( "bytes" "errors" "fmt" "io" "sync/atomic" "time" . "github.com/tendermint/tendermint/binary" . "github.com/tendermint/tendermint/common" ) var pexErrInvalidMessage = errors.New("Invalid PEX message") const ( PexCh = byte(0x00) ensurePeersPeriodSeconds = 30 minNumOutboundPeers = 10 maxNumPeers = 50 ) /* PEXReactor handles PEX (peer exchange) and ensures that an adequate number of peers are connected to the switch. */ type PEXReactor struct { sw *Switch quit chan struct{} started uint32 stopped uint32 book *AddrBook } func NewPEXReactor(sw *Switch, book *AddrBook) *PEXReactor { pexR := &PEXReactor{ sw: sw, quit: make(chan struct{}), book: book, } return pexR } func (pexR *PEXReactor) Start() { if atomic.CompareAndSwapUint32(&pexR.started, 0, 1) { log.Info("Starting PEXReactor") go pexR.ensurePeersRoutine() } } func (pexR *PEXReactor) Stop() { if atomic.CompareAndSwapUint32(&pexR.stopped, 0, 1) { log.Info("Stopping PEXReactor") close(pexR.quit) } } // Asks peer for more addresses. func (pexR *PEXReactor) RequestPEX(peer *Peer) { peer.TrySend(PexCh, &pexRequestMessage{}) } func (pexR *PEXReactor) SendAddrs(peer *Peer, addrs []*NetAddress) { peer.Send(PexCh, &pexRddrsMessage{Addrs: addrs}) } // Implements Reactor func (pexR *PEXReactor) GetChannels() []*ChannelDescriptor { return []*ChannelDescriptor{ &ChannelDescriptor{ Id: PexCh, Priority: 1, }, } } // Implements Reactor func (pexR *PEXReactor) AddPeer(peer *Peer) { if peer.IsOutbound() { pexR.SendAddrs(peer, pexR.book.OurAddresses()) if pexR.book.NeedMoreAddrs() { pexR.RequestPEX(peer) } } } // Implements Reactor func (pexR *PEXReactor) RemovePeer(peer *Peer, err error) { // TODO } // Implements Reactor // Handles incoming PEX messages. func (pexR *PEXReactor) Receive(chId byte, src *Peer, msgBytes []byte) { // decode message msg := decodeMessage(msgBytes) log.Info("requestRoutine received %v", msg) switch msg.(type) { case *pexRequestMessage: // src requested some peers. // TODO: prevent abuse. addrs := pexR.book.GetSelection() msg := &pexRddrsMessage{Addrs: addrs} queued := src.TrySend(PexCh, msg) if !queued { // ignore } case *pexRddrsMessage: // We received some peer addresses from src. // TODO: prevent abuse. // (We don't want to get spammed with bad peers) srcAddr := src.RemoteAddress() for _, addr := range msg.(*pexRddrsMessage).Addrs { pexR.book.AddAddress(addr, srcAddr) } default: // Ignore unknown message. } } // Ensures that sufficient peers are connected. (continuous) func (pexR *PEXReactor) ensurePeersRoutine() { // fire once immediately. pexR.ensurePeers() // fire periodically timer := NewRepeatTimer(ensurePeersPeriodSeconds * time.Second) FOR_LOOP: for { select { case <-timer.Ch: pexR.ensurePeers() case <-pexR.quit: break FOR_LOOP } } // Cleanup timer.Stop() } // Ensures that sufficient peers are connected. (once) func (pexR *PEXReactor) ensurePeers() { numOutPeers, _, numDialing := pexR.sw.NumPeers() numToDial := minNumOutboundPeers - (numOutPeers + numDialing) if numToDial <= 0 { return } toDial := NewCMap() // Try to pick numToDial addresses to dial. // TODO: improve logic. for i := 0; i < numToDial; i++ { newBias := MinInt(numOutPeers, 8)*10 + 10 var picked *NetAddress // Try to fetch a new peer 3 times. // This caps the maximum number of tries to 3 * numToDial. for j := 0; i < 3; j++ { picked = pexR.book.PickAddress(newBias) if picked == nil { return } if toDial.Has(picked.String()) || pexR.sw.IsDialing(picked) || pexR.sw.Peers().Has(picked.String()) { continue } else { break } } if picked == nil { continue } toDial.Set(picked.String(), picked) } // Dial picked addresses for _, item := range toDial.Values() { picked := item.(*NetAddress) go func() { _, err := pexR.sw.DialPeerWithAddress(picked) if err != nil { pexR.book.MarkAttempt(picked) } }() } } //----------------------------------------------------------------------------- /* Messages */ const ( msgTypeUnknown = byte(0x00) msgTypeRequest = byte(0x01) msgTypeAddrs = byte(0x02) ) // TODO: check for unnecessary extra bytes at the end. func decodeMessage(bz []byte) (msg interface{}) { var n int64 var err error // log.Debug("decoding msg bytes: %X", bz) switch bz[0] { case msgTypeRequest: return &pexRequestMessage{} case msgTypeAddrs: return readPexAddrsMessage(bytes.NewReader(bz[1:]), &n, &err) default: return nil } } /* A pexRequestMessage requests additional peer addresses. */ type pexRequestMessage struct { } func (m *pexRequestMessage) WriteTo(w io.Writer) (n int64, err error) { WriteByte(w, msgTypeRequest, &n, &err) return } func (m *pexRequestMessage) String() string { return "[pexRequest]" } /* A message with announced peer addresses. */ type pexRddrsMessage struct { Addrs []*NetAddress } func readPexAddrsMessage(r io.Reader, n *int64, err *error) *pexRddrsMessage { numAddrs := int(ReadUInt32(r, n, err)) addrs := []*NetAddress{} for i := 0; i < numAddrs; i++ { addr := ReadNetAddress(r, n, err) addrs = append(addrs, addr) } return &pexRddrsMessage{ Addrs: addrs, } } func (m *pexRddrsMessage) WriteTo(w io.Writer) (n int64, err error) { WriteByte(w, msgTypeAddrs, &n, &err) WriteUInt32(w, uint32(len(m.Addrs)), &n, &err) for _, addr := range m.Addrs { WriteBinary(w, addr, &n, &err) } return } func (m *pexRddrsMessage) String() string { return fmt.Sprintf("[pexRddrs %v]", m.Addrs) }