import (
"encoding/json"
- "errors"
"fmt"
"math/rand"
"net"
+ "strings"
"sync"
"time"
dbm "github.com/tendermint/tmlibs/db"
cfg "github.com/bytom/config"
+ "github.com/bytom/errors"
"github.com/bytom/p2p/trust"
)
const (
- reconnectAttempts = 30
- reconnectInterval = 3 * time.Second
+ reconnectAttempts = 10
+ reconnectInterval = 10 * time.Second
bannedPeerKey = "BannedPeer"
defaultBanDuration = time.Hour * 24
- peerBannedTM = 20
)
var ErrConnectBannedPeer = errors.New("Connect banned peer")
type Switch struct {
cmn.BaseService
- config *cfg.P2PConfig
- peerConfig *PeerConfig
- listeners []Listener
- reactors map[string]Reactor
- chDescs []*ChannelDescriptor
- reactorsByCh map[byte]Reactor
- peers *PeerSet
- dialing *cmn.CMap
- nodeInfo *NodeInfo // our node info
- nodePrivKey crypto.PrivKeyEd25519 // our node privkey
- bannedPeer map[string]time.Time
- db dbm.DB
- TrustMetricStore *trust.TrustMetricStore
- ScamPeerCh chan *Peer
- mtx sync.Mutex
+ config *cfg.P2PConfig
+ peerConfig *PeerConfig
+ listeners []Listener
+ reactors map[string]Reactor
+ chDescs []*ChannelDescriptor
+ reactorsByCh map[byte]Reactor
+ peers *PeerSet
+ dialing *cmn.CMap
+ nodeInfo *NodeInfo // our node info
+ nodePrivKey crypto.PrivKeyEd25519 // our node privkey
+ bannedPeer map[string]time.Time
+ db dbm.DB
+ mtx sync.Mutex
filterConnByAddr func(net.Addr) error
filterConnByPubKey func(crypto.PubKeyEd25519) error
var (
ErrSwitchDuplicatePeer = errors.New("Duplicate peer")
+ ErrConnectSelf = errors.New("Connect self")
+ ErrPeerConnected = errors.New("Peer is connected")
)
func NewSwitch(config *cfg.P2PConfig, trustHistoryDB dbm.DB) *Switch {
dialing: cmn.NewCMap(),
nodeInfo: nil,
db: trustHistoryDB,
- ScamPeerCh: make(chan *Peer),
}
sw.BaseService = *cmn.NewBaseService(nil, "P2P Switch", sw)
- sw.TrustMetricStore = trust.NewTrustMetricStore(trustHistoryDB, trust.DefaultConfig())
- sw.TrustMetricStore.Start()
sw.bannedPeer = make(map[string]time.Time)
if datajson := sw.db.Get([]byte(bannedPeerKey)); datajson != nil {
return nil
}
}
- go sw.scamPeerHandler()
+ trust.Init()
return sw
}
return err
}
- tm := trust.NewMetric()
-
- tm.Start()
- sw.TrustMetricStore.AddPeerTrustMetric(peer.mconn.RemoteAddress.IP.String(), tm)
-
log.WithField("peer", peer).Info("Added peer")
return nil
}
}
addrBook.AddAddress(netAddr, ourAddr)
}
+
addrBook.Save()
}
-
- // permute the list, dial them in random order.
+ //permute the list, dial them in random order.
perm := rand.Perm(len(netAddrs))
- for i := 0; i < len(perm); i++ {
- go func(i int) {
- time.Sleep(time.Duration(rand.Int63n(3000)) * time.Millisecond)
- j := perm[i]
- sw.dialSeed(netAddrs[j])
- }(i)
+ for i := 0; i < len(perm)/2; i++ {
+ j := perm[i]
+ sw.dialSeed(netAddrs[j])
}
+
return nil
}
if err := sw.checkBannedPeer(addr.IP.String()); err != nil {
return nil, err
}
-
+ if strings.Compare(addr.IP.String(), sw.nodeInfo.ListenHost()) == 0 {
+ return nil, ErrConnectSelf
+ }
+ for _, v := range sw.Peers().list {
+ if strings.Compare(v.mconn.RemoteAddress.IP.String(), addr.IP.String()) == 0 {
+ return nil, ErrPeerConnected
+ }
+ }
sw.dialing.Set(addr.IP.String(), addr)
defer sw.dialing.Delete(addr.IP.String())
}
log.WithFields(log.Fields{
"address": addr,
- "error": err,
}).Info("Dialed and added peer")
return peer, nil
}
sw.stopAndRemovePeer(peer, reason)
if peer.IsPersistent() {
- go func() {
- log.WithField("peer", peer).Info("Reconnecting to peer")
- for i := 1; i < reconnectAttempts; i++ {
- if !sw.IsRunning() {
- return
- }
+ log.WithField("peer", peer).Info("Reconnecting to peer")
+ for i := 1; i < reconnectAttempts; i++ {
+ if !sw.IsRunning() {
+ return
+ }
- peer, err := sw.DialPeerWithAddress(addr, true)
- if err != nil {
- if i == reconnectAttempts {
- log.WithFields(log.Fields{
- "retries": i,
- "error": err,
- }).Info("Error reconnecting to peer. Giving up")
- return
- }
+ peer, err := sw.DialPeerWithAddress(addr, true)
+ if err != nil {
+ if i == reconnectAttempts {
log.WithFields(log.Fields{
"retries": i,
"error": err,
- }).Info("Error reconnecting to peer. Trying again")
- time.Sleep(reconnectInterval)
- continue
+ }).Info("Error reconnecting to peer. Giving up")
+ return
}
- log.WithField("peer", peer).Info("Reconnected to peer")
- return
+ if errors.Root(err) == ErrConnectBannedPeer || errors.Root(err) == ErrPeerConnected || errors.Root(err) == ErrSwitchDuplicatePeer || errors.Root(err) == ErrConnectSelf {
+ log.WithField("error", err).Info("Error reconnecting to peer. ")
+ return
+ }
+
+ log.WithFields(log.Fields{
+ "retries": i,
+ "error": err,
+ }).Info("Error reconnecting to peer. Trying again")
+ time.Sleep(reconnectInterval)
+ continue
}
- }()
+
+ log.WithField("peer", peer).Info("Reconnected to peer")
+ return
+ }
}
}
}
func (sw *Switch) stopAndRemovePeer(peer *Peer, reason interface{}) {
- sw.peers.Remove(peer)
- peer.Stop()
for _, reactor := range sw.reactors {
reactor.RemovePeer(peer, reason)
}
+ sw.peers.Remove(peer)
+ peer.Stop()
}
func (sw *Switch) listenerRoutine(l Listener) {
// ignore connection if we already have enough
maxPeers := sw.config.MaxNumPeers
if maxPeers <= sw.peers.Size() {
+ // close inConn
+ inConn.Close()
log.WithFields(log.Fields{
"address": inConn.RemoteAddr().String(),
"numPeers": sw.peers.Size(),
// New inbound connection!
err := sw.addPeerWithConnectionAndConfig(inConn, sw.peerConfig)
if err != nil {
+ // conn close for returing err
+ inConn.Close()
log.WithFields(log.Fields{
"address": inConn.RemoteAddr().String(),
"error": err,
peer, err := newInboundPeerWithConfig(conn, sw.reactorsByCh, sw.chDescs, sw.StopPeerForError, sw.nodePrivKey, config)
if err != nil {
- conn.Close()
return err
}
peer.SetLogger(sw.Logger.With("peer", conn.RemoteAddr()))
if err = sw.AddPeer(peer); err != nil {
- conn.Close()
return err
}
return nil
}
-func (sw *Switch) scamPeerHandler() {
- for src := range sw.ScamPeerCh {
- var tm *trust.TrustMetric
- key := src.Connection().RemoteAddress.IP.String()
- if tm = sw.TrustMetricStore.GetPeerTrustMetric(key); tm == nil {
- log.Errorf("Can't get peer trust metric")
- continue
- }
- sw.delTrustMetric(tm, src)
- }
-}
-
-func (sw *Switch) AddScamPeer(src *Peer) {
- sw.ScamPeerCh <- src
-}
-
-func (sw *Switch) delTrustMetric(tm *trust.TrustMetric, src *Peer) {
- key := src.Connection().RemoteAddress.IP.String()
- tm.BadEvents(1)
- if tm.TrustScore() < peerBannedTM {
- sw.AddBannedPeer(src)
- sw.TrustMetricStore.PeerDisconnected(key)
- sw.StopPeerGracefully(src)
- }
-}
-
func (sw *Switch) checkBannedPeer(peer string) error {
if banEnd, ok := sw.bannedPeer[peer]; ok {
if time.Now().Before(banEnd) {
OSDN Git Service
