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         p, address := protocolAndAddress(manager.config.P2P.ListenAddress)
  68         l := p2p.NewDefaultListener(p, address, manager.config.P2P.SkipUPNP, nil)
  69         manager.sw.AddListener(l)
  70         manager.sw.SetNodeInfo(manager.makeNodeInfo())
  71         manager.sw.SetNodePrivKey(manager.privKey)
  72
  73         // Optionally, start the pex reactor
  74         //var addrBook *p2p.AddrBook
  75         if config.P2P.PexReactor {
  76                 manager.addrBook = p2p.NewAddrBook(config.P2P.AddrBookFile(), config.P2P.AddrBookStrict)
  77                 pexReactor := p2p.NewPEXReactor(manager.addrBook, manager.sw)
  78                 manager.sw.AddReactor("PEX", pexReactor)
  79         }
  80
  81         return manager, nil
  82 }
  83
  84 // Defaults to tcp
  85 func protocolAndAddress(listenAddr string) (string, string) {
  86         p, address := "tcp", listenAddr
  87         parts := strings.SplitN(address, "://", 2)
  88         if len(parts) == 2 {
  89                 p, address = parts[0], parts[1]
  90         }
  91         return p, address
  92 }
  93
  94 func (sm *SyncManager) makeNodeInfo() *p2p.NodeInfo {
  95         nodeInfo := &p2p.NodeInfo{
  96                 PubKey:  sm.privKey.PubKey().Unwrap().(crypto.PubKeyEd25519),
  97                 Moniker: sm.config.Moniker,
  98                 Network: sm.config.ChainID,
  99                 Version: version.Version,
 100                 Other: []string{
 101                         cmn.Fmt("wire_version=%v", wire.Version),
 102                         cmn.Fmt("p2p_version=%v", p2p.Version),
 103                 },
 104         }
 105
 106         if !sm.sw.IsListening() {
 107                 return nodeInfo
 108         }
 109
 110         p2pListener := sm.sw.Listeners()[0]
 111         p2pHost := p2pListener.ExternalAddress().IP.String()
 112         p2pPort := p2pListener.ExternalAddress().Port
 113
 114         // We assume that the rpcListener has the same ExternalAddress.
 115         // This is probably true because both P2P and RPC listeners use UPnP,
 116         // except of course if the rpc is only bound to localhost
 117         nodeInfo.ListenAddr = cmn.Fmt("%v:%v", p2pHost, p2pPort)
 118         return nodeInfo
 119 }
 120
 121 func (sm *SyncManager) netStart() error {
 122         // Start the switch
 123         _, err := sm.sw.Start()
 124         if err != nil {
 125                 return err
 126         }
 127
 128         // If seeds exist, add them to the address book and dial out
 129         if sm.config.P2P.Seeds != "" {
 130                 // dial out
 131                 seeds := strings.Split(sm.config.P2P.Seeds, ",")
 132                 if err := sm.DialSeeds(seeds); err != nil {
 133                         return err
 134                 }
 135         }
 136
 137         return nil
 138 }
 139
 140 //Start start sync manager service
 141 func (sm *SyncManager) Start() {
 142         go sm.netStart()
 143         // broadcast transactions
 144         go sm.txBroadcastLoop()
 145
 146         // broadcast mined blocks
 147         go sm.minedBroadcastLoop()
 148
 149         // start sync handlers
 150         go sm.syncer()
 151
 152         go sm.txsyncLoop()
 153 }
 154
 155 //Stop stop sync manager
 156 func (sm *SyncManager) Stop() {
 157         close(sm.quitSync)
 158         sm.sw.Stop()
 159 }
 160
 161 func (sm *SyncManager) txBroadcastLoop() {
 162         newTxCh := sm.txPool.GetNewTxCh()
 163         for {
 164                 select {
 165                 case newTx := <-newTxCh:
 166                         peers, err := sm.peers.BroadcastTx(newTx)
 167                         if err != nil {
 168                                 log.Errorf("Broadcast new tx error. %v", err)
 169                                 return
 170                         }
 171                         for _, smPeer := range peers {
 172                                 if smPeer == nil {
 173                                         continue
 174                                 }
 175                                 swPeer := smPeer.getPeer()
 176                                 log.Info("Tx broadcast error. Stop Peer.")
 177                                 sm.sw.StopPeerGracefully(swPeer)
 178                         }
 179                 case <-sm.quitSync:
 180                         return
 181                 }
 182         }
 183 }
 184
 185 func (sm *SyncManager) minedBroadcastLoop() {
 186         for {
 187                 select {
 188                 case blockHash := <-sm.newBlockCh:
 189                         block, err := sm.chain.GetBlockByHash(blockHash)
 190                         if err != nil {
 191                                 log.Errorf("Failed on mined broadcast loop get block %v", err)
 192                                 return
 193                         }
 194                         peers, err := sm.peers.BroadcastMinedBlock(block)
 195                         if err != nil {
 196                                 log.Errorf("Broadcast mine block error. %v", err)
 197                                 return
 198                         }
 199                         for _, smPeer := range peers {
 200                                 if smPeer == nil {
 201                                         continue
 202                                 }
 203                                 swPeer := smPeer.getPeer()
 204                                 log.Info("New mined block broadcast error. Stop Peer.")
 205                                 sm.sw.StopPeerGracefully(swPeer)
 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 }