cfg "github.com/vapor/config"
"github.com/vapor/consensus"
+ dbm "github.com/vapor/database/leveldb"
"github.com/vapor/event"
msgs "github.com/vapor/netsync/messages"
"github.com/vapor/netsync/peers"
// Chain is the interface for Bytom core
type Chain interface {
BestBlockHeader() *types.BlockHeader
+ LastIrreversibleHeader() *types.BlockHeader
BestBlockHeight() uint64
GetBlockByHash(*bc.Hash) (*types.Block, error)
GetBlockByHeight(uint64) (*types.Block, error)
}
//NewChainManager create a chain sync manager.
-func NewManager(config *cfg.Config, sw Switch, chain Chain, mempool Mempool, dispatcher *event.Dispatcher, peers *peers.PeerSet) (*Manager, error) {
+func NewManager(config *cfg.Config, sw Switch, chain Chain, mempool Mempool, dispatcher *event.Dispatcher, peers *peers.PeerSet, fastSyncDB dbm.DB) (*Manager, error) {
manager := &Manager{
sw: sw,
mempool: mempool,
chain: chain,
- blockKeeper: newBlockKeeper(chain, peers),
+ blockKeeper: newBlockKeeper(chain, peers, fastSyncDB),
peers: peers,
txSyncCh: make(chan *txSyncMsg),
quit: make(chan struct{}),
rawData, err := block.MarshalText()
if err != nil {
log.WithFields(log.Fields{"module": logModule, "err": err}).Error("fail on handleGetBlocksMsg marshal block")
- continue
+ return
}
if totalSize+len(rawData) > msgs.MaxBlockchainResponseSize/2 {
}
func (m *Manager) handleGetHeadersMsg(peer *peers.Peer, msg *msgs.GetHeadersMessage) {
- headers, err := m.blockKeeper.locateHeaders(msg.GetBlockLocator(), msg.GetStopHash())
+ headers, err := m.blockKeeper.locateHeaders(msg.GetBlockLocator(), msg.GetStopHash(), msg.GetSkip(), maxNumOfHeadersPerMsg)
if err != nil || len(headers) == 0 {
log.WithFields(log.Fields{"module": logModule, "err": err}).Debug("fail on handleGetHeadersMsg locateHeaders")
return
func (m *Manager) handleStatusMsg(basePeer peers.BasePeer, msg *msgs.StatusMessage) {
if peer := m.peers.GetPeer(basePeer.ID()); peer != nil {
- peer.SetStatus(msg.Height, msg.GetHash())
- return
+ peer.SetBestStatus(msg.BestHeight, msg.GetBestHash())
+ peer.SetIrreversibleStatus(msg.IrreversibleHeight, msg.GetIrreversibleHash())
}
}
return
}
+ m.peers.MarkTx(peer.ID(), tx.ID)
if isOrphan, err := m.chain.ValidateTx(tx); err != nil && err != core.ErrDustTx && !isOrphan {
m.peers.ProcessIllegal(peer.ID(), security.LevelMsgIllegal, "fail on validate tx transaction")
}
- m.peers.MarkTx(peer.ID(), tx.ID)
}
func (m *Manager) handleTransactionsMsg(peer *peers.Peer, msg *msgs.TransactionsMessage) {
}
for _, tx := range txs {
+ m.peers.MarkTx(peer.ID(), tx.ID)
if isOrphan, err := m.chain.ValidateTx(tx); err != nil && !isOrphan {
m.peers.ProcessIllegal(peer.ID(), security.LevelMsgIllegal, "fail on validate tx transaction")
return
}
- m.peers.MarkTx(peer.ID(), tx.ID)
}
}
return errors.New("invalid peer")
}
- if err := p.SendStatus(m.chain.BestBlockHeader()); err != nil {
+ if err := p.SendStatus(m.chain.BestBlockHeader(), m.chain.LastIrreversibleHeader()); err != nil {
m.peers.RemovePeer(p.ID())
return err
}
if err != nil {
return err
}
-
+ m.blockKeeper.start()
go m.broadcastTxsLoop()
go m.syncMempoolLoop()
//Stop stop sync manager
func (m *Manager) Stop() {
+ m.blockKeeper.stop()
close(m.quit)
}