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: "bytom",
  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         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                         sm.peers.BroadcastTx(newTx)
 169
 170                 case <-sm.quitSync:
 171                         return
 172                 }
 173         }
 174 }
 175
 176 func (sm *SyncManager) minedBroadcastLoop() {
 177         for {
 178                 select {
 179                 case blockHash := <-sm.newBlockCh:
 180                         block, err := sm.chain.GetBlockByHash(blockHash)
 181                         if err != nil {
 182                                 log.Errorf("Failed on mined broadcast loop get block %v", err)
 183                                 return
 184                         }
 185                         sm.peers.BroadcastMinedBlock(block)
 186                 case <-sm.quitSync:
 187                         return
 188                 }
 189         }
 190 }
 191
 192 //NodeInfo get P2P peer node info
 193 func (sm *SyncManager) NodeInfo() *p2p.NodeInfo {
 194         return sm.sw.NodeInfo()
 195 }
 196
 197 //BlockKeeper get block keeper
 198 func (sm *SyncManager) BlockKeeper() *blockKeeper {
 199         return sm.blockKeeper
 200 }
 201
 202 //Peers get sync manager peer set
 203 func (sm *SyncManager) Peers() *peerSet {
 204         return sm.peers
 205 }
 206
 207 //DialSeeds dial seed peers
 208 func (sm *SyncManager) DialSeeds(seeds []string) error {
 209         return sm.sw.DialSeeds(sm.addrBook, seeds)
 210 }
 211
 212 //Switch get sync manager switch
 213 func (sm *SyncManager) Switch() *p2p.Switch {
 214         return sm.sw
 215 }
 216
 217 func (sm *SyncManager) removePeer(peerID string) {
 218         sm.peers.DropPeer(peerID)
 219         log.Debug("Removing peer", "peerID:", peerID)
 220 }