netsync/handle.go

   1 package netsync
   2
   3 import (
   4         "strings"
   5
   6         log "github.com/sirupsen/logrus"
   7         "github.com/tendermint/go-crypto"
   8         "github.com/tendermint/go-wire"
   9         cmn "github.com/tendermint/tmlibs/common"
  10         dbm "github.com/tendermint/tmlibs/db"
  11
  12         cfg "github.com/bytom/config"
  13         "github.com/bytom/p2p"
  14         core "github.com/bytom/protocol"
  15         "github.com/bytom/protocol/bc"
  16         "github.com/bytom/version"
  17 )
  18
  19 //SyncManager Sync Manager is responsible for the business layer information synchronization
  20 type SyncManager struct {
  21         networkID uint64
  22         sw        *p2p.Switch
  23         addrBook  *p2p.AddrBook // known peers
  24
  25         privKey     crypto.PrivKeyEd25519 // local node's p2p key
  26         chain       *core.Chain
  27         txPool      *core.TxPool
  28         fetcher     *Fetcher
  29         blockKeeper *blockKeeper
  30         peers       *peerSet
  31
  32         newBlockCh    chan *bc.Hash
  33         newPeerCh     chan struct{}
  34         txSyncCh      chan *txsync
  35         dropPeerCh    chan *string
  36         quitSync      chan struct{}
  37         config        *cfg.Config
  38         synchronising int32
  39 }
  40
  41 //NewSyncManager create a sync manager
  42 func NewSyncManager(config *cfg.Config, chain *core.Chain, txPool *core.TxPool, newBlockCh chan *bc.Hash) (*SyncManager, error) {
  43         // Create the protocol manager with the base fields
  44         manager := &SyncManager{
  45                 txPool:     txPool,
  46                 chain:      chain,
  47                 privKey:    crypto.GenPrivKeyEd25519(),
  48                 config:     config,
  49                 quitSync:   make(chan struct{}),
  50                 newBlockCh: newBlockCh,
  51                 newPeerCh:  make(chan struct{}),
  52                 txSyncCh:   make(chan *txsync),
  53                 dropPeerCh: make(chan *string, maxQuitReq),
  54                 peers:      newPeerSet(),
  55         }
  56
  57         trustHistoryDB := dbm.NewDB("trusthistory", config.DBBackend, config.DBDir())
  58         manager.sw = p2p.NewSwitch(config.P2P, trustHistoryDB)
  59
  60         manager.blockKeeper = newBlockKeeper(manager.chain, manager.sw, manager.peers, manager.dropPeerCh)
  61         manager.fetcher = NewFetcher(chain, manager.sw, manager.peers)
  62
  63         protocolReactor := NewProtocolReactor(chain, txPool, manager.sw, manager.blockKeeper, manager.fetcher, manager.peers, manager.newPeerCh, manager.txSyncCh, manager.dropPeerCh)
  64         manager.sw.AddReactor("PROTOCOL", protocolReactor)
  65
  66         // Optionally, start the pex reactor
  67         //var addrBook *p2p.AddrBook
  68         if config.P2P.PexReactor {
  69                 manager.addrBook = p2p.NewAddrBook(config.P2P.AddrBookFile(), config.P2P.AddrBookStrict)
  70                 pexReactor := p2p.NewPEXReactor(manager.addrBook)
  71                 manager.sw.AddReactor("PEX", pexReactor)
  72         }
  73
  74         return manager, nil
  75 }
  76
  77 // Defaults to tcp
  78 func protocolAndAddress(listenAddr string) (string, string) {
  79         p, address := "tcp", listenAddr
  80         parts := strings.SplitN(address, "://", 2)
  81         if len(parts) == 2 {
  82                 p, address = parts[0], parts[1]
  83         }
  84         return p, address
  85 }
  86
  87 func (sm *SyncManager) makeNodeInfo() *p2p.NodeInfo {
  88         nodeInfo := &p2p.NodeInfo{
  89                 PubKey:  sm.privKey.PubKey().Unwrap().(crypto.PubKeyEd25519),
  90                 Moniker: sm.config.Moniker,
  91                 Network: sm.config.ChainID,
  92                 Version: version.Version,
  93                 Other: []string{
  94                         cmn.Fmt("wire_version=%v", wire.Version),
  95                         cmn.Fmt("p2p_version=%v", p2p.Version),
  96                 },
  97         }
  98
  99         if !sm.sw.IsListening() {
 100                 return nodeInfo
 101         }
 102
 103         p2pListener := sm.sw.Listeners()[0]
 104         p2pHost := p2pListener.ExternalAddress().IP.String()
 105         p2pPort := p2pListener.ExternalAddress().Port
 106
 107         // We assume that the rpcListener has the same ExternalAddress.
 108         // This is probably true because both P2P and RPC listeners use UPnP,
 109         // except of course if the rpc is only bound to localhost
 110         nodeInfo.ListenAddr = cmn.Fmt("%v:%v", p2pHost, p2pPort)
 111         return nodeInfo
 112 }
 113
 114 func (sm *SyncManager) netStart() error {
 115         // Create & add listener
 116         p, address := protocolAndAddress(sm.config.P2P.ListenAddress)
 117
 118         l := p2p.NewDefaultListener(p, address, sm.config.P2P.SkipUPNP, nil)
 119
 120         sm.sw.AddListener(l)
 121
 122         // Start the switch
 123         sm.sw.SetNodeInfo(sm.makeNodeInfo())
 124         sm.sw.SetNodePrivKey(sm.privKey)
 125         _, err := sm.sw.Start()
 126         if err != nil {
 127                 return err
 128         }
 129
 130         // If seeds exist, add them to the address book and dial out
 131         if sm.config.P2P.Seeds != "" {
 132                 // dial out
 133                 seeds := strings.Split(sm.config.P2P.Seeds, ",")
 134                 if err := sm.DialSeeds(seeds); err != nil {
 135                         return err
 136                 }
 137         }
 138
 139         return nil
 140 }
 141
 142 //Start start sync manager service
 143 func (sm *SyncManager) Start() {
 144         go sm.netStart()
 145         // broadcast transactions
 146         go sm.txBroadcastLoop()
 147
 148         // broadcast mined blocks
 149         go sm.minedBroadcastLoop()
 150
 151         // start sync handlers
 152         go sm.syncer()
 153
 154         go sm.txsyncLoop()
 155 }
 156
 157 //Stop stop sync manager
 158 func (sm *SyncManager) Stop() {
 159         close(sm.quitSync)
 160         sm.sw.Stop()
 161 }
 162
 163 func (sm *SyncManager) txBroadcastLoop() {
 164         newTxCh := sm.txPool.GetNewTxCh()
 165         for {
 166                 select {
 167                 case newTx := <-newTxCh:
 168                         peers,err:=sm.peers.BroadcastTx(newTx)
 169                         if err != nil {
 170                                 log.Errorf("Broadcast new tx error. %v", err)
 171                                 return
 172                         }
 173                         for _, peer := range peers {
 174                                 if ban := peer.addBanScore(0, 50, "Broadcast new tx error"); ban {
 175                                         peer := sm.peers.Peer(peer.id).getPeer()
 176                                         sm.sw.AddBannedPeer(peer)
 177                                         sm.sw.StopPeerGracefully(peer)
 178                                 }
 179                         }
 180                 case <-sm.quitSync:
 181                         return
 182                 }
 183         }
 184 }
 185
 186 func (sm *SyncManager) minedBroadcastLoop() {
 187         for {
 188                 select {
 189                 case blockHash := <-sm.newBlockCh:
 190                         block, err := sm.chain.GetBlockByHash(blockHash)
 191                         if err != nil {
 192                                 log.Errorf("Failed on mined broadcast loop get block %v", err)
 193                                 return
 194                         }
 195                         peers, err := sm.peers.BroadcastMinedBlock(block)
 196                         if err != nil {
 197                                 log.Errorf("Broadcast mine block error. %v", err)
 198                                 return
 199                         }
 200                         for _, peer := range peers {
 201                                 if ban := peer.addBanScore(0, 50, "Broadcast block error"); ban {
 202                                         peer := sm.peers.Peer(peer.id).getPeer()
 203                                         sm.sw.AddBannedPeer(peer)
 204                                         sm.sw.StopPeerGracefully(peer)
 205                                 }
 206                         }
 207                 case <-sm.quitSync:
 208                         return
 209                 }
 210         }
 211 }
 212
 213 //NodeInfo get P2P peer node info
 214 func (sm *SyncManager) NodeInfo() *p2p.NodeInfo {
 215         return sm.sw.NodeInfo()
 216 }
 217
 218 //BlockKeeper get block keeper
 219 func (sm *SyncManager) BlockKeeper() *blockKeeper {
 220         return sm.blockKeeper
 221 }
 222
 223 //Peers get sync manager peer set
 224 func (sm *SyncManager) Peers() *peerSet {
 225         return sm.peers
 226 }
 227
 228 //DialSeeds dial seed peers
 229 func (sm *SyncManager) DialSeeds(seeds []string) error {
 230         return sm.sw.DialSeeds(sm.addrBook, seeds)
 231 }
 232
 233 //Switch get sync manager switch
 234 func (sm *SyncManager) Switch() *p2p.Switch {
 235         return sm.sw
 236 }