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         // Create & add listener
  67         if !config.VaultMode {
  68                 p, address := protocolAndAddress(manager.config.P2P.ListenAddress)
  69                 l := p2p.NewDefaultListener(p, address, manager.config.P2P.SkipUPNP, nil)
  70                 manager.sw.AddListener(l)
  71         }
  72         manager.sw.SetNodeInfo(manager.makeNodeInfo())
  73         manager.sw.SetNodePrivKey(manager.privKey)
  74
  75         // Optionally, start the pex reactor
  76         //var addrBook *p2p.AddrBook
  77         if config.P2P.PexReactor {
  78                 manager.addrBook = p2p.NewAddrBook(config.P2P.AddrBookFile(), config.P2P.AddrBookStrict)
  79                 pexReactor := p2p.NewPEXReactor(manager.addrBook, manager.sw)
  80                 manager.sw.AddReactor("PEX", pexReactor)
  81         }
  82
  83         return manager, nil
  84 }
  85
  86 // Defaults to tcp
  87 func protocolAndAddress(listenAddr string) (string, string) {
  88         p, address := "tcp", listenAddr
  89         parts := strings.SplitN(address, "://", 2)
  90         if len(parts) == 2 {
  91                 p, address = parts[0], parts[1]
  92         }
  93         return p, address
  94 }
  95
  96 func (sm *SyncManager) makeNodeInfo() *p2p.NodeInfo {
  97         nodeInfo := &p2p.NodeInfo{
  98                 PubKey:  sm.privKey.PubKey().Unwrap().(crypto.PubKeyEd25519),
  99                 Moniker: sm.config.Moniker,
 100                 Network: sm.config.ChainID,
 101                 Version: version.Version,
 102                 Other: []string{
 103                         cmn.Fmt("wire_version=%v", wire.Version),
 104                         cmn.Fmt("p2p_version=%v", p2p.Version),
 105                 },
 106         }
 107
 108         if !sm.sw.IsListening() {
 109                 return nodeInfo
 110         }
 111
 112         p2pListener := sm.sw.Listeners()[0]
 113         p2pHost := p2pListener.ExternalAddress().IP.String()
 114         p2pPort := p2pListener.ExternalAddress().Port
 115
 116         // We assume that the rpcListener has the same ExternalAddress.
 117         // This is probably true because both P2P and RPC listeners use UPnP,
 118         // except of course if the rpc is only bound to localhost
 119         nodeInfo.ListenAddr = cmn.Fmt("%v:%v", p2pHost, p2pPort)
 120         return nodeInfo
 121 }
 122
 123 func (sm *SyncManager) netStart() error {
 124         // Start the switch
 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 _, smPeer := range peers {
 174                                 if smPeer == nil {
 175                                         continue
 176                                 }
 177                                 swPeer := smPeer.getPeer()
 178                                 log.Info("Tx broadcast error. Stop Peer.")
 179                                 sm.sw.StopPeerGracefully(swPeer)
 180                         }
 181                 case <-sm.quitSync:
 182                         return
 183                 }
 184         }
 185 }
 186
 187 func (sm *SyncManager) minedBroadcastLoop() {
 188         for {
 189                 select {
 190                 case blockHash := <-sm.newBlockCh:
 191                         block, err := sm.chain.GetBlockByHash(blockHash)
 192                         if err != nil {
 193                                 log.Errorf("Failed on mined broadcast loop get block %v", err)
 194                                 return
 195                         }
 196                         peers, err := sm.peers.BroadcastMinedBlock(block)
 197                         if err != nil {
 198                                 log.Errorf("Broadcast mine block error. %v", err)
 199                                 return
 200                         }
 201                         for _, smPeer := range peers {
 202                                 if smPeer == nil {
 203                                         continue
 204                                 }
 205                                 swPeer := smPeer.getPeer()
 206                                 log.Info("New mined block broadcast error. Stop Peer.")
 207                                 sm.sw.StopPeerGracefully(swPeer)
 208                         }
 209                 case <-sm.quitSync:
 210                         return
 211                 }
 212         }
 213 }
 214
 215 //NodeInfo get P2P peer node info
 216 func (sm *SyncManager) NodeInfo() *p2p.NodeInfo {
 217         return sm.sw.NodeInfo()
 218 }
 219
 220 //BlockKeeper get block keeper
 221 func (sm *SyncManager) BlockKeeper() *blockKeeper {
 222         return sm.blockKeeper
 223 }
 224
 225 //Peers get sync manager peer set
 226 func (sm *SyncManager) Peers() *peerSet {
 227         return sm.peers
 228 }
 229
 230 //DialSeeds dial seed peers
 231 func (sm *SyncManager) DialSeeds(seeds []string) error {
 232         return sm.sw.DialSeeds(sm.addrBook, seeds)
 233 }
 234
 235 //Switch get sync manager switch
 236 func (sm *SyncManager) Switch() *p2p.Switch {
 237         return sm.sw
 238 }