func (sm *SyncManager) QueueInv(inv *msg.Inv, peer *peer.Peer) {
	if atomic.LoadInt32(&sm.shutdown) != 0 {
		return
	}
	sm.msgChan <- &invMsg{inv: inv, peer: peer}
}
QueueInv adds the passed inv message and peer to the block handling queue.
package netsync
import (
	"fmt"
	"sync"
	"sync/atomic"
	"github.com/elastos/Elastos.ELA.SideChain/blockchain"
	"github.com/elastos/Elastos.ELA.SideChain/events"
	"github.com/elastos/Elastos.ELA.SideChain/mempool"
	"github.com/elastos/Elastos.ELA.SideChain/pact"
	"github.com/elastos/Elastos.ELA.SideChain/peer"
	"github.com/elastos/Elastos.ELA.SideChain/types"
	"github.com/elastos/Elastos.ELA/errors"
	"github.com/elastos/Elastos.ELA/common"
	"github.com/elastos/Elastos.ELA/p2p"
	"github.com/elastos/Elastos.ELA/p2p/msg"
)
const (
	maxRejectedTxns = 1000
	maxRequestedBlocks = msg.MaxInvPerMsg
	maxRequestedTxns = msg.MaxInvPerMsg
)
var zeroHash common.Uint256
type newPeerMsg struct {
	peer *peer.Peer
}
type blockMsg struct {
	block *types.Block
	peer  *peer.Peer
	reply chan struct{}
}
type invMsg struct {
	inv  *msg.Inv
	peer *peer.Peer
}
type donePeerMsg struct {
	peer *peer.Peer
}
type txMsg struct {
	tx    *types.Transaction
	peer  *peer.Peer
	reply chan struct{}
}
type getSyncPeerMsg struct {
	reply chan uint64
}
type processBlockResponse struct {
	isOrphan bool
	err      error
}
type processBlockMsg struct {
	block *types.Block
	reply chan processBlockResponse
}
type isCurrentMsg struct {
	reply chan bool
}
type pauseMsg struct {
	unpause <-chan struct{}
}
type peerSyncState struct {
	syncCandidate   bool
	requestQueue    []*msg.InvVect
	requestedTxns   map[common.Uint256]struct{}
	requestedBlocks map[common.Uint256]struct{}
}
type SyncManager struct {
	peerNotifier PeerNotifier
	started      int32
	shutdown     int32
	chain        *blockchain.BlockChain
	txMemPool    *mempool.TxPool
	msgChan      chan interface{}
	wg           sync.WaitGroup
	quit         chan struct{}
	rejectedTxns    map[common.Uint256]struct{}
	requestedTxns   map[common.Uint256]struct{}
	requestedBlocks map[common.Uint256]struct{}
	syncPeer        *peer.Peer
	peerStates      map[*peer.Peer]*peerSyncState
}
func (sm *SyncManager) startSync() {
	if sm.syncPeer != nil {
		return
	}
	bestHeight := sm.chain.GetBestHeight()
	var bestPeer *peer.Peer
	for peer, state := range sm.peerStates {
		if !state.syncCandidate {
			continue
		}
		if peer.Height() < bestHeight {
			state.syncCandidate = false
			continue
		}
		bestPeer = peer
		break
	}
	if bestPeer != nil {
		if bestPeer.Height() == bestHeight {
			return
		}
		sm.requestedBlocks = make(map[common.Uint256]struct{})
		locator, err := sm.chain.LatestBlockLocator()
		if err != nil {
			log.Errorf("Failed to get block locator for the "+
				"latest block: %v", err)
			return
		}
		log.Infof("Syncing to block height %d from peer %v",
			bestPeer.Height(), bestPeer.Addr())
		bestPeer.PushGetBlocksMsg(locator, &zeroHash)
		sm.syncPeer = bestPeer
	} else {
		log.Warnf("No sync peer candidates available")
	}
}
func (sm *SyncManager) isSyncCandidate(peer *peer.Peer) bool {
	if peer.Services()&pact.SFNodeNetwork != pact.SFNodeNetwork {
		return false
	}
	return true
}
func (sm *SyncManager) handleNewPeerMsg(peer *peer.Peer) {
	if atomic.LoadInt32(&sm.shutdown) != 0 {
		return
	}
	log.Infof("New valid peer %s", peer)
	isSyncCandidate := sm.isSyncCandidate(peer)
	sm.peerStates[peer] = &peerSyncState{
		syncCandidate:   isSyncCandidate,
		requestedTxns:   make(map[common.Uint256]struct{}),
		requestedBlocks: make(map[common.Uint256]struct{}),
	}
	if isSyncCandidate && sm.syncPeer == nil {
		sm.startSync()
	}
}
func (sm *SyncManager) handleDonePeerMsg(peer *peer.Peer) {
	state, exists := sm.peerStates[peer]
	if !exists {
		log.Warnf("Received done peer message for unknown peer %s", peer)
		return
	}
	delete(sm.peerStates, peer)
	log.Infof("Lost peer %s", peer)
	for txHash := range state.requestedTxns {
		delete(sm.requestedTxns, txHash)
	}
	for blockHash := range state.requestedBlocks {
		delete(sm.requestedBlocks, blockHash)
	}
	if sm.syncPeer == peer {
		sm.syncPeer = nil
		sm.startSync()
	}
}
func (sm *SyncManager) handleTxMsg(tmsg *txMsg) {
	peer := tmsg.peer
	state, exists := sm.peerStates[peer]
	if !exists {
		log.Warnf("Received tx message from unknown peer %s", peer)
		return
	}
	txHash := tmsg.tx.Hash()
	if _, exists = sm.rejectedTxns[txHash]; exists {
		log.Debugf("Ignoring unsolicited previously rejected "+
			"transaction %v from %s", txHash, peer)
		return
	}
	delete(state.requestedTxns, txHash)
	delete(sm.requestedTxns, txHash)
	err := sm.txMemPool.AppendToTxPool(tmsg.tx)
	if err != nil {
		sm.rejectedTxns[txHash] = struct{}{}
		sm.limitMap(sm.rejectedTxns, maxRejectedTxns)
		elaErr := errors.SimpleWithMessage(errors.ErrTxPoolFailure, err,
			fmt.Sprintf("AppendToTxPool fail tx hash %s", tmsg.tx.Hash()))
		peer.PushRejectMsg(p2p.CmdTx, elaErr, &txHash, false)
		return
	}
	iv := msg.NewInvVect(msg.InvTypeTx, &txHash)
	sm.peerNotifier.RelayInventory(iv, tmsg.tx)
}
func (sm *SyncManager) current() bool {
	if sm.syncPeer == nil {
		return true
	}
	if sm.chain.BestChain.Height < sm.syncPeer.Height() {
		return false
	}
	return true
}
func (sm *SyncManager) handleBlockMsg(bmsg *blockMsg) {
	peer := bmsg.peer
	state, exists := sm.peerStates[peer]
	if !exists {
		log.Warnf("Received block message from unknown peer %s", peer)
		return
	}
	blockHash := bmsg.block.Hash()
	if _, exists = state.requestedBlocks[blockHash]; !exists {
		log.Warnf("Got unrequested block %v from %s -- "+
			"disconnecting", blockHash, peer.Addr())
		peer.Disconnect()
		return
	}
	delete(state.requestedBlocks, blockHash)
	delete(sm.requestedBlocks, blockHash)
	_, isOrphan, err := sm.chain.ProcessBlock(bmsg.block)
	if err != nil {
		elaErr := errors.SimpleWithMessage(errors.ErrP2pReject, err,
			fmt.Sprintf("Rejected block %v from %s", blockHash, peer))
		log.Info(elaErr.Error())
		peer.PushRejectMsg(p2p.CmdBlock, elaErr, &blockHash, false)
		return
	}
	if isOrphan {
		orphanRoot := sm.chain.GetOrphanRoot(&blockHash)
		locator, err := sm.chain.LatestBlockLocator()
		if err != nil {
			log.Warnf("Failed to get block locator for the "+
				"latest block: %v", err)
		} else {
			peer.PushGetBlocksMsg(locator, orphanRoot)
		}
	} else {
		sm.rejectedTxns = make(map[common.Uint256]struct{})
	}
}
func (sm *SyncManager) haveInventory(invVect *msg.InvVect) (bool, error) {
	switch invVect.Type {
	case msg.InvTypeBlock:
		return sm.chain.HaveBlock(&invVect.Hash)
	case msg.InvTypeTx:
		if sm.txMemPool.HaveTransaction(invVect.Hash) {
			return true, nil
		}
		return false, nil
	}
	return true, nil
}
func (sm *SyncManager) handleInvMsg(imsg *invMsg) {
	peer := imsg.peer
	state, exists := sm.peerStates[peer]
	if !exists {
		log.Warnf("Received inv message from unknown peer %s", peer)
		return
	}
	invVects := imsg.inv.InvList
	if peer != sm.syncPeer && !sm.current() {
		return
	}
	for _, iv := range invVects {
		switch iv.Type {
		case msg.InvTypeBlock:
		case msg.InvTypeTx:
		default:
			continue
		}
		peer.AddKnownInventory(iv)
		haveInv, err := sm.haveInventory(iv)
		if err != nil {
			log.Warnf("Unexpected failure when checking for "+
				"existing inventory during inv message "+
				"processing: %v", err)
			continue
		}
		if !haveInv {
			if iv.Type == msg.InvTypeTx {
				if _, exists := sm.rejectedTxns[iv.Hash]; exists {
					continue
				}
			}
			state.requestQueue = append(state.requestQueue, iv)
			continue
		}
		if iv.Type == msg.InvTypeBlock {
			if sm.chain.IsKnownOrphan(&iv.Hash) {
				orphanRoot := sm.chain.GetOrphanRoot(&iv.Hash)
				locator, err := sm.chain.LatestBlockLocator()
				if err != nil {
					log.Errorf("PEER: Failed to get block "+
						"locator for the latest block: "+
						"%v", err)
					continue
				}
				peer.PushGetBlocksMsg(locator, orphanRoot)
				continue
			}
		}
	}
	numRequested := 0
	gdmsg := msg.NewGetData()
	requestQueue := state.requestQueue
	for len(requestQueue) != 0 {
		iv := requestQueue[0]
		requestQueue[0] = nil
		requestQueue = requestQueue[1:]
		switch iv.Type {
		case msg.InvTypeBlock:
			if _, exists := sm.requestedBlocks[iv.Hash]; !exists {
				sm.requestedBlocks[iv.Hash] = struct{}{}
				sm.limitMap(sm.requestedBlocks, maxRequestedBlocks)
				state.requestedBlocks[iv.Hash] = struct{}{}
				gdmsg.AddInvVect(iv)
				numRequested++
			}
		case msg.InvTypeTx:
			if _, exists := sm.requestedTxns[iv.Hash]; !exists {
				sm.requestedTxns[iv.Hash] = struct{}{}
				sm.limitMap(sm.requestedTxns, maxRequestedTxns)
				state.requestedTxns[iv.Hash] = struct{}{}
				gdmsg.AddInvVect(iv)
				numRequested++
			}
		}
		if numRequested >= msg.MaxInvPerMsg {
			break
		}
	}
	state.requestQueue = requestQueue
	if len(gdmsg.InvList) > 0 {
		peer.QueueMessage(gdmsg, nil)
	}
}
func (sm *SyncManager) limitMap(m map[common.Uint256]struct{}, limit int) {
	if len(m)+1 > limit {
		for txHash := range m {
			delete(m, txHash)
			return
		}
	}
}
func (sm *SyncManager) blockHandler() {
out:
	for {
		select {
		case m := <-sm.msgChan:
			switch msg := m.(type) {
			case *newPeerMsg:
				sm.handleNewPeerMsg(msg.peer)
			case *txMsg:
				sm.handleTxMsg(msg)
				msg.reply <- struct{}{}
			case *blockMsg:
				sm.handleBlockMsg(msg)
				msg.reply <- struct{}{}
			case *invMsg:
				sm.handleInvMsg(msg)
			case *donePeerMsg:
				sm.handleDonePeerMsg(msg.peer)
			case getSyncPeerMsg:
				var peerID uint64
				if sm.syncPeer != nil {
					peerID = sm.syncPeer.ID()
				}
				msg.reply <- peerID
			case processBlockMsg:
				_, isOrphan, err := sm.chain.ProcessBlock(msg.block)
				if err != nil {
					msg.reply <- processBlockResponse{
						isOrphan: false,
						err:      err,
					}
				}
				msg.reply <- processBlockResponse{
					isOrphan: isOrphan,
					err:      nil,
				}
			case isCurrentMsg:
				msg.reply <- sm.current()
			case pauseMsg:
				<-msg.unpause
			default:
				log.Warnf("Invalid message type in block "+
					"handler: %T", msg)
			}
		case <-sm.quit:
			break out
		}
	}
	sm.wg.Done()
}
func (sm *SyncManager) handleBlockchainEvents(event *events.Event) {
	switch event.Type {
	case events.ETBlockAccepted:
		if !sm.current() {
			return
		}
		block, ok := event.Data.(*types.Block)
		if !ok {
			log.Warnf("Chain accepted event is not a block.")
			break
		}
		blockHash := block.Hash()
		iv := msg.NewInvVect(msg.InvTypeBlock, &blockHash)
		sm.peerNotifier.RelayInventory(iv, block.Header)
	case events.ETBlockConnected:
		block, ok := event.Data.(*types.Block)
		if !ok {
			log.Warnf("Chain connected event is not a block.")
			break
		}
		sm.txMemPool.CleanSubmittedTransactions(block)
	case events.ETBlockDisconnected:
		block, ok := event.Data.(*types.Block)
		if !ok {
			log.Warnf("Chain disconnected event is not a block.")
			break
		}
		for _, tx := range block.Transactions[1:] {
			err := sm.txMemPool.MaybeAcceptTransaction(tx)
			if err != nil {
				sm.txMemPool.RemoveTransaction(tx)
			}
		}
	}
}
func (sm *SyncManager) NewPeer(peer *peer.Peer) {
	if atomic.LoadInt32(&sm.shutdown) != 0 {
		return
	}
	sm.msgChan <- &newPeerMsg{peer: peer}
}
func (sm *SyncManager) QueueTx(tx *types.Transaction, peer *peer.Peer, done chan struct{}) {
	if atomic.LoadInt32(&sm.shutdown) != 0 {
		done <- struct{}{}
		return
	}
	sm.msgChan <- &txMsg{tx: tx, peer: peer, reply: done}
}
func (sm *SyncManager) QueueBlock(block *types.Block, peer *peer.Peer, done chan struct{}) {
	if atomic.LoadInt32(&sm.shutdown) != 0 {
		done <- struct{}{}
		return
	}
	sm.msgChan <- &blockMsg{block: block, peer: peer, reply: done}
}