<commit_before>package pex

import (
	"errors"
	"reflect"
	"sync"
	"time"

	log "github.com/sirupsen/logrus"
	cmn "github.com/tendermint/tmlibs/common"

	"github.com/bytom/p2p"
	"github.com/bytom/p2p/connection"
	"github.com/bytom/p2p/discover"
)

const (
	// PexChannel is a channel for PEX messages
	PexChannel = byte(0x00)

	minNumOutboundPeers      = 5
	maxPexMessageSize        = 1048576 // 1MB
	defaultMaxMsgCountByPeer = uint16(1000)
)

// PEXReactor handles peer exchange and ensures that an adequate number of peers are connected to the switch.
type PEXReactor struct {
	p2p.BaseReactor
	discv          *discover.Network
	msgCountByPeer *cmn.CMap
}

// NewPEXReactor creates new PEX reactor.
func NewPEXReactor(discv *discover.Network) *PEXReactor {
	r := &PEXReactor{
		discv:          discv,
		msgCountByPeer: cmn.NewCMap(),
	}
	r.BaseReactor = *p2p.NewBaseReactor("PEXReactor", r)
	return r
}

// OnStart implements BaseService
func (r *PEXReactor) OnStart() error {
	r.BaseReactor.OnStart()
	go r.ensurePeersRoutine()
	go r.flushMsgCountByPeer()
	return nil
}

// OnStop implements BaseService
func (r *PEXReactor) OnStop() {
	r.BaseReactor.OnStop()
}

// GetChannels implements Reactor
func (r *PEXReactor) GetChannels() []*connection.ChannelDescriptor {
	return []*connection.ChannelDescriptor{&connection.ChannelDescriptor{
		ID:                PexChannel,
		Priority:          1,
		SendQueueCapacity: 10,
	}}
}

// AddPeer adding peer to the address book
func (r *PEXReactor) AddPeer(p *p2p.Peer) error {
	if r.Switch.Peers().Size() <= r.Switch.Config.P2P.MaxNumPeers {
		return nil
	}

	nodes := make([]*discover.Node, 20)
	if n := r.discv.ReadRandomNodes(nodes); n == 0 {
		return nil
	}

	if r.SendAddrs(p, nodes) {
		<-time.After(1 * time.Second)
		r.Switch.StopPeerGracefully(p)
	}
	return errors.New("addPeer: reach the max peer, exchange then close")
}

// Receive implements Reactor by handling incoming PEX messages.
func (r *PEXReactor) Receive(chID byte, p *p2p.Peer, rawMsg []byte) {
	r.incrementMsgCount(p.RemoteAddr)
	if r.reachedMaxMsgLimit(p.RemoteAddr) {
		log.WithField("peer", p.RemoteAddr).Error("reached the max pex messages limit")
		r.Switch.StopPeerGracefully(p)
		return
	}

	_, msg, err := DecodeMessage(rawMsg)
	if err != nil {
		log.WithField("error", err).Error("failed to decoding pex message")
		r.Switch.StopPeerGracefully(p)
		return
	}

	switch msg := msg.(type) {
	case *pexRequestMessage:
		nodes := make([]*discover.Node, 20)
		if n := r.discv.ReadRandomNodes(nodes); n == 0 {
			return
		}

		if !r.SendAddrs(p, nodes) {
			log.Error("failed to send pex address message")
		}

	case *pexAddrsMessage:
	default:
		log.WithField("type", reflect.TypeOf(msg)).Error("Unknown message type")
	}
}

// RemovePeer implements Reactor.
func (r *PEXReactor) RemovePeer(p *p2p.Peer, reason interface{}) {}

// SendAddrs sends addrs to the peer.
func (r *PEXReactor) SendAddrs(p *p2p.Peer, nodes []*discover.Node) bool {
	addrs := []*p2p.NetAddress{}
	for _, node := range nodes {
		addrs = append(addrs, p2p.NewNetAddressIPPort(node.IP, node.TCP))
	}

	ok := p.TrySend(PexChannel, struct{ PexMessage }{&pexAddrsMessage{Addrs: addrs}})
	if !ok {
		r.Switch.StopPeerGracefully(p)
	}
	return ok
}

func (r *PEXReactor) dialPeerWorker(a *p2p.NetAddress, wg *sync.WaitGroup) {
	if err := r.Switch.DialPeerWithAddress(a); err != nil {
		log.WithFields(log.Fields{"addr": a, "err": err}).Error("dialPeerWorker fail on dial peer")
	}
	wg.Done()
}
<commit_msg>handle node is nil bug<commit_after>package pex

import (
	"errors"
	"reflect"
	"sync"
	"time"

	log "github.com/sirupsen/logrus"
	cmn "github.com/tendermint/tmlibs/common"

	"github.com/bytom/p2p"
	"github.com/bytom/p2p/connection"
	"github.com/bytom/p2p/discover"
)

const (
	// PexChannel is a channel for PEX messages
	PexChannel = byte(0x00)

	minNumOutboundPeers      = 5
	maxPexMessageSize        = 1048576 // 1MB
	defaultMaxMsgCountByPeer = uint16(1000)
)

// PEXReactor handles peer exchange and ensures that an adequate number of peers are connected to the switch.
type PEXReactor struct {
	p2p.BaseReactor
	discv          *discover.Network
	msgCountByPeer *cmn.CMap
}

// NewPEXReactor creates new PEX reactor.
func NewPEXReactor(discv *discover.Network) *PEXReactor {
	r := &PEXReactor{
		discv:          discv,
		msgCountByPeer: cmn.NewCMap(),
	}
	r.BaseReactor = *p2p.NewBaseReactor("PEXReactor", r)
	return r
}

// OnStart implements BaseService
func (r *PEXReactor) OnStart() error {
	r.BaseReactor.OnStart()
	go r.ensurePeersRoutine()
	go r.flushMsgCountByPeer()
	return nil
}

// OnStop implements BaseService
func (r *PEXReactor) OnStop() {
	r.BaseReactor.OnStop()
}

// GetChannels implements Reactor
func (r *PEXReactor) GetChannels() []*connection.ChannelDescriptor {
	return []*connection.ChannelDescriptor{&connection.ChannelDescriptor{
		ID:                PexChannel,
		Priority:          1,
		SendQueueCapacity: 10,
	}}
}

// AddPeer adding peer to the address book
func (r *PEXReactor) AddPeer(p *p2p.Peer) error {
	if r.Switch.Peers().Size() <= r.Switch.Config.P2P.MaxNumPeers {
		return nil
	}

	nodes := make([]*discover.Node, 20)
	if n := r.discv.ReadRandomNodes(nodes); n == 0 {
		return nil
	}

	if r.SendAddrs(p, nodes) {
		<-time.After(1 * time.Second)
		r.Switch.StopPeerGracefully(p)
	}
	return errors.New("addPeer: reach the max peer, exchange then close")
}

// Receive implements Reactor by handling incoming PEX messages.
func (r *PEXReactor) Receive(chID byte, p *p2p.Peer, rawMsg []byte) {
	r.incrementMsgCount(p.RemoteAddr)
	if r.reachedMaxMsgLimit(p.RemoteAddr) {
		log.WithField("peer", p.RemoteAddr).Error("reached the max pex messages limit")
		r.Switch.StopPeerGracefully(p)
		return
	}

	_, msg, err := DecodeMessage(rawMsg)
	if err != nil {
		log.WithField("error", err).Error("failed to decoding pex message")
		r.Switch.StopPeerGracefully(p)
		return
	}

	switch msg := msg.(type) {
	case *pexRequestMessage:
		nodes := make([]*discover.Node, 20)
		if n := r.discv.ReadRandomNodes(nodes); n == 0 {
			return
		}

		if !r.SendAddrs(p, nodes) {
			log.Error("failed to send pex address message")
		}

	case *pexAddrsMessage:
	default:
		log.WithField("type", reflect.TypeOf(msg)).Error("Unknown message type")
	}
}

// RemovePeer implements Reactor.
func (r *PEXReactor) RemovePeer(p *p2p.Peer, reason interface{}) {}

// SendAddrs sends addrs to the peer.
func (r *PEXReactor) SendAddrs(p *p2p.Peer, nodes []*discover.Node) bool {
	addrs := []*p2p.NetAddress{}
	for _, node := range nodes {
		if node == nil {
			break
		}
		addrs = append(addrs, p2p.NewNetAddressIPPort(node.IP, node.TCP))
	}

	ok := p.TrySend(PexChannel, struct{ PexMessage }{&pexAddrsMessage{Addrs: addrs}})
	if !ok {
		r.Switch.StopPeerGracefully(p)
	}
	return ok
}

func (r *PEXReactor) dialPeerWorker(a *p2p.NetAddress, wg *sync.WaitGroup) {
	if err := r.Switch.DialPeerWithAddress(a); err != nil {
		log.WithFields(log.Fields{"addr": a, "err": err}).Error("dialPeerWorker fail on dial peer")
	}
	wg.Done()
}

{'fraction_non_alphanumeric': 0.13486982501067007, 'fraction_numerical': 0.011665955327927158, 'mean_word_length': 4.046661880832735, 'pattern_counts': {'":': 4, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7283012', 'n_tokens_mistral': 2772, 'n_tokens_neox': 2633, 'n_words': 602}