add multi-download

plugin/consensus/para/kv.go

@@ -14,6 +14,11 @@ func calcTitleHeightKey(title string, height int64) []byte {
 	return []byte(fmt.Sprintf("%s-TH-%s-%d", types.ConsensusParaTxsPrefix, title, height))
 }
 
+//temp main height tx data for big size
+func calcTitleMainHeightKey(title string, height int64) []byte {
+	return []byte(fmt.Sprintf("%s-TMH-%s-%d", types.ConsensusParaTxsPrefix, title, height))
+}
+
 func calcTitleLastHeightKey(title string) []byte {
 	return []byte(fmt.Sprintf("%s-TLH-%s", types.ConsensusParaTxsPrefix, title))
 }

plugin/consensus/para/para.go

@@ -29,9 +29,6 @@ import (
 )
 
 const (
-	addAct int64 = 1 //add para block action
-	delAct int64 = 2 //reference blockstore.go, del para block action
-
 	minBlockNum = 100 //min block number startHeight before lastHeight in mainchain
 
 	genesisBlockTime int64 = 1514533390
@@ -64,10 +61,12 @@ type client struct {
 	caughtUp        int32
 	commitMsgClient *commitMsgClient
 	blockSyncClient *blockSyncClient
+	multiDldCli     *multiDldClient
 	authAccount     string
 	privateKey      crypto.PrivKey
 	wg              sync.WaitGroup
 	subCfg          *subConfig
+	isClosed        int32
 	quitCreate      chan struct{}
 }
 
@@ -89,6 +88,9 @@ type subConfig struct {
 	FetchFilterParaTxsEnable        uint32 `json:"fetchFilterParaTxsEnable,omitempty"`
 	BatchFetchBlockCount            int64  `json:"batchFetchBlockCount,omitempty"`
 	ParaConsensStartHeight          int64  `json:"paraConsensStartHeight,omitempty"`
+	MultiDownloadOpen               int32  `json:"multiDownloadOpen,omitempty"`
+	MultiDownInvNumPerJob           int64  `json:"multiDownInvNumPerJob,omitempty"`
+	MultiDownJobBuffNum             uint32 `json:"multiDownJobBuffNum,omitempty"`
 }
 
 // New function to init paracross env
@@ -197,6 +199,21 @@ func New(cfg *types.Consensus, sub []byte) queue.Module {
 		para.blockSyncClient.maxSyncErrCount = subcfg.MaxSyncErrCount
 	}
 
+	para.multiDldCli = &multiDldClient{
+		paraClient:   para,
+		invNumPerJob: defaultInvNumPerJob,
+		jobBufferNum: defaultJobBufferNum,
+	}
+	if subcfg.MultiDownInvNumPerJob > 0 {
+		para.multiDldCli.invNumPerJob = subcfg.MultiDownInvNumPerJob
+	}
+	if subcfg.MultiDownJobBuffNum > 0 {
+		para.multiDldCli.jobBufferNum = subcfg.MultiDownJobBuffNum
+	}
+	if subcfg.MultiDownloadOpen > 0 {
+		para.multiDldCli.multiDldOpen = true
+	}
+
 	c.SetChild(para)
 	return para
 }
@@ -208,6 +225,7 @@ func (client *client) CheckBlock(parent *types.Block, current *types.BlockDetail
 }
 
 func (client *client) Close() {
+	atomic.StoreInt32(&client.isClosed, 1)
 	close(client.commitMsgClient.quit)
 	close(client.quitCreate)
 	close(client.blockSyncClient.quitChan)
@@ -218,6 +236,10 @@ func (client *client) Close() {
 	plog.Info("consensus para closed")
 }
 
+func (client *client) isCancel() bool {
+	return atomic.LoadInt32(&client.isClosed) == 1
+}
+
 func (client *client) SetQueueClient(c queue.Client) {
 	plog.Info("Enter SetQueueClient method of Para consensus")
 	client.InitClient(c, func() {

plugin/consensus/para/paracreate.go

@@ -93,7 +93,7 @@ func (client *client) getLastLocalHeight() (int64, error) {
 	if err != nil {
 		return -1, err
 	}
-	if len(value) == 0 {
+	if len(value) == 0 || value[0] == nil {
 		return -1, types.ErrNotFound
 	}
 
@@ -114,7 +114,7 @@ func (client *client) getLocalBlockByHeight(height int64) (*pt.ParaLocalDbBlock,
 	if err != nil {
 		return nil, err
 	}
-	if len(value) == 0 {
+	if len(value) == 0 || value[0] == nil {
 		return nil, types.ErrNotFound
 	}
 
@@ -264,15 +264,15 @@ func (client *client) switchMatchedBlockOnChain(startHeight int64) (int64, []byt
 }
 
 func (client *client) switchHashMatchedBlock() (int64, []byte, error) {
-	mainSeq, mainHash, err := client.switchLocalHashMatchedBlock()
+	mainSeq, localBlock, err := client.switchLocalHashMatchedBlock()
 	if err != nil {
 		return client.switchMatchedBlockOnChain(0)
 	}
-	return mainSeq, mainHash, nil
+	return mainSeq, localBlock.MainHash, nil
 }
 
 //
-func (client *client) switchLocalHashMatchedBlock() (int64, []byte, error) {
+func (client *client) switchLocalHashMatchedBlock() (int64, *pt.ParaLocalDbBlock, error) {
 	lastBlock, err := client.getLastLocalBlock()
 	if err != nil {
 		plog.Error("Parachain RequestLastBlock fail", "err", err)
@@ -286,6 +286,11 @@ func (client *client) switchLocalHashMatchedBlock() (int64, []byte, error) {
 		}
 		//当前block结构已经有mainHash和MainHeight但是从blockchain获取的block还没有写入，以后如果获取到，可以替换从minerTx获取
 		plog.Info("switchLocalHashMatchedBlock", "height", height, "mainHeight", block.MainHeight, "mainHash", hex.EncodeToString(block.MainHash))
+		mainHash, err := client.GetHashByHeightOnMainChain(block.MainHeight)
+		if err != nil || !bytes.Equal(mainHash, block.MainHash) {
+			continue
+		}
+
 		mainSeq, err := client.GetSeqByHashOnMainChain(block.MainHash)
 		if err != nil {
 			continue
@@ -298,8 +303,8 @@ func (client *client) switchLocalHashMatchedBlock() (int64, []byte, error) {
 		}
 
 		plog.Info("switchLocalHashMatchedBlock succ", "currHeight", height, "initHeight", lastBlock.Height,
-			"currSeq", mainSeq, "currMainBlockHash", hex.EncodeToString(block.MainHash))
-		return mainSeq, block.MainHash, nil
+			"currSeq", mainSeq, "mainHeight", block.MainHeight, "currMainBlockHash", hex.EncodeToString(block.MainHash))
+		return mainSeq, block, nil
 	}
 	return -2, nil, pt.ErrParaCurHashNotMatch
 }
@@ -317,7 +322,7 @@ func (client *client) getBatchSeqCount(currSeq int64) (int64, error) {
 			atomic.StoreInt32(&client.caughtUp, 1)
 		}
 		if fetchFilterParaTxsEnable && lastSeq-currSeq > client.subCfg.BatchFetchBlockCount {
-			return client.subCfg.BatchFetchBlockCount, nil
+			return client.subCfg.BatchFetchBlockCount - 1, nil
 		}
 		return 0, nil
 	}
@@ -337,9 +342,23 @@ func (client *client) getBatchSeqCount(currSeq int64) (int64, error) {
 
 }
 
+func getParentHash(block *types.ParaTxDetail) []byte {
+	if block.Type == types.AddBlock {
+		return block.Header.ParentHash
+	}
+
+	return block.Header.Hash
+}
+
+func getVerifyHash(block *types.ParaTxDetail) []byte {
+	if block.Type == types.AddBlock {
+		return block.Header.Hash
+	}
+
+	return block.Header.ParentHash
+}
 func verifyMainBlockHash(preMainBlockHash []byte, mainBlock *types.ParaTxDetail) error {
-	if (bytes.Equal(preMainBlockHash, mainBlock.Header.ParentHash) && mainBlock.Type == addAct) ||
-		(bytes.Equal(preMainBlockHash, mainBlock.Header.Hash) && mainBlock.Type == delAct) {
+	if bytes.Equal(preMainBlockHash, getParentHash(mainBlock)) {
 		return nil
 	}
 	plog.Error("verifyMainBlockHash", "preMainBlockHash", hex.EncodeToString(preMainBlockHash),
@@ -355,16 +374,29 @@ func verifyMainBlocks(preMainBlockHash []byte, mainBlocks *types.ParaTxDetails)
 		if err != nil {
 			return err
 		}
-		if block.Type == addAct {
-			pre = block.Header.Hash
-		} else {
-			pre = block.Header.ParentHash
-		}
-
+		pre = getVerifyHash(block)
 	}
 	return nil
 }
 
+func verifyMainBlocksInternal(mainBlocks *types.ParaTxDetails) error {
+	return verifyMainBlocks(getParentHash(mainBlocks.Items[0]), mainBlocks)
+}
+
+func isValidSeqType(ty int64) bool {
+	return ty == types.AddBlock || ty == types.DelBlock
+}
+
+func validMainBlocks(txs *types.ParaTxDetails) *types.ParaTxDetails {
+	for i, item := range txs.Items {
+		if item == nil || !isValidSeqType(item.Type) {
+			txs.Items = txs.Items[:i]
+			return txs
+		}
+	}
+	return txs
+}
+
 func (client *client) requestTxsFromBlock(currSeq int64, preMainBlockHash []byte) (*types.ParaTxDetails, error) {
 	blockSeq, err := client.GetBlockOnMainBySeq(currSeq)
 	if err != nil {
@@ -374,6 +406,10 @@ func (client *client) requestTxsFromBlock(currSeq int64, preMainBlockHash []byte
 	txDetail := blockSeq.Detail.FilterParaTxsByTitle(types.GetTitle())
 	txDetail.Type = blockSeq.Seq.Type
 
+	if !isValidSeqType(txDetail.Type) {
+		return nil, types.ErrInvalidParam
+	}
+
 	err = verifyMainBlockHash(preMainBlockHash, txDetail)
 	if err != nil {
 		plog.Error("requestTxsFromBlock", "curr seq", currSeq, "preMainBlockHash", hex.EncodeToString(preMainBlockHash))
@@ -383,12 +419,13 @@ func (client *client) requestTxsFromBlock(currSeq int64, preMainBlockHash []byte
 }
 
 func (client *client) requestFilterParaTxs(currSeq int64, count int64, preMainBlockHash []byte) (*types.ParaTxDetails, error) {
-	req := &types.ReqParaTxByTitle{Start: currSeq, End: currSeq + count, Title: types.GetTitle()}
+	req := &types.ReqParaTxByTitle{IsSeq: true, Start: currSeq, End: currSeq + count, Title: types.GetTitle()}
 	details, err := client.GetParaTxByTitle(req)
 	if err != nil {
 		return nil, err
 	}
 
+	details = validMainBlocks(details)
 	err = verifyMainBlocks(preMainBlockHash, details)
 	if err != nil {
 		plog.Error("requestTxsOnlyPara", "curSeq", currSeq, "count", count, "preMainBlockHash", hex.EncodeToString(preMainBlockHash))
@@ -430,26 +467,24 @@ func (client *client) procLocalBlock(mainBlock *types.ParaTxDetail) (bool, error
 		"lastBlockMainHash", common.ToHex(lastBlock.MainHash), "currMainHeight", lastSeqMainHeight,
 		"curMainHash", common.ToHex(mainBlock.Header.Hash), "seqTy", mainBlock.Type)
 
-	if mainBlock.Type == delAct {
+	if mainBlock.Type == types.DelBlock {
 		if len(txs) == 0 {
 			if lastSeqMainHeight > lastBlock.MainHeight {
 				return false, nil
 			}
-			plog.Info("Delete empty block")
+			plog.Info("Delete empty block", "height", lastBlock.Height)
 		}
 		return true, client.delLocalBlock(lastBlock.Height)
 
-	} else if mainBlock.Type == addAct {
-		if len(txs) == 0 {
-			if lastSeqMainHeight-lastBlock.MainHeight < client.subCfg.EmptyBlockInterval {
-				return false, nil
-			}
-			plog.Info("Create empty block")
+	}
+	//AddAct
+	if len(txs) == 0 {
+		if lastSeqMainHeight-lastBlock.MainHeight < client.subCfg.EmptyBlockInterval {
+			return false, nil
 		}
-		return true, client.createLocalBlock(lastBlock, txs, mainBlock)
-
+		plog.Info("Create empty block", "newHeight", lastBlock.Height+1)
 	}
-	return false, types.ErrInvalidParam
+	return true, client.createLocalBlock(lastBlock, txs, mainBlock)
 
 }
 
@@ -472,6 +507,8 @@ func (client *client) procLocalBlocks(mainBlocks *types.ParaTxDetails) error {
 }
 
 func (client *client) CreateBlock() {
+	client.multiDldCli.tryMultiServerDownload()
+
 	lastSeq, lastSeqMainHash, err := client.getLastLocalBlockSeq()
 	if err != nil {
 		plog.Error("Parachain CreateBlock getLastLocalBlockSeq fail", "err", err.Error())
@@ -502,9 +539,9 @@ out:
 				continue
 			}
 
-			if count+1 != int64(len(paraTxs.Items)) {
-				plog.Error("para CreateBlock count not match", "count", count+1, "items", len(paraTxs.Items))
-				continue
+			if count != int64(len(paraTxs.Items)) {
+				plog.Debug("para CreateBlock count not match", "count", count, "items", len(paraTxs.Items))
+				count = int64(len(paraTxs.Items))
 			}
 
 			err = client.procLocalBlocks(paraTxs)
@@ -516,11 +553,11 @@ out:
 			}
 
 			//重新设定seq和lastSeqMainHash
-			lastSeqMainHash = paraTxs.Items[count].Header.Hash
-			if paraTxs.Items[count].Type == delAct {
-				lastSeqMainHash = paraTxs.Items[count].Header.ParentHash
+			lastSeqMainHash = paraTxs.Items[count-1].Header.Hash
+			if paraTxs.Items[count-1].Type == types.DelBlock {
+				lastSeqMainHash = paraTxs.Items[count-1].Header.ParentHash
 			}
-			currSeq = currSeq + count + 1
+			currSeq = currSeq + count
 
 		}
 	}

plugin/consensus/para/paramultidownload.go

+// Copyright Fuzamei Corp. 2018 All Rights Reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package para
+
+import (
+	"context"
+	"sync"
+
+	"strings"
+
+	"github.com/33cn/chain33/rpc/grpcclient"
+	"github.com/33cn/chain33/types"
+)
+
+const (
+	maxRollbackHeight      int64 = 10000
+	defaultInvNumPerJob          = 20       // 1000 block per inv,  20inv per job
+	defaultJobBufferNum          = 20       // channel buffer num for done job process
+	maxBlockSize                 = 20000000 // 单次1000block size累积超过20M 需保存到localdb
+	downTimesFastThreshold       = 600      //  单个server 下载超过600次，平均20次用20s，下载10分钟左右检查有没有差别比较大的
+	downTimesSlowThreshold       = 40       //  慢的server小于40次，则小于快server的15倍，需要剔除
+)
+
+type connectCli struct {
+	ip        string
+	conn      types.Chain33Client
+	downTimes int64
+	isFail    bool
+}
+
+//invertory 是每次请求的最小单位，每次请求最多MaxBlockCountPerTime
+type inventory struct {
+	start     int64
+	end       int64
+	curHeight int64
+	txs       *types.ParaTxDetails
+	isDone    bool
+	isSaveDb  bool
+	connCli   *connectCli
+}
+
+type downloadJob struct {
+	mDldCli     *multiDldClient
+	parentBlock *types.ParaTxDetail
+	wg          sync.WaitGroup
+	invs        []*inventory
+}
+
+func newDownLoadJob(cli *multiDldClient) *downloadJob {
+	return &downloadJob{
+		mDldCli: cli,
+	}
+}
+
+type multiDldClient struct {
+	paraClient     *client
+	jobBufferNum   uint32
+	invNumPerJob   int64
+	conns          []*connectCli
+	connsCheckDone bool
+	multiDldOpen   bool
+	wg             sync.WaitGroup
+}
+
+func (m *multiDldClient) getInvs(startHeight, endHeight int64) []*inventory {
+	var invs []*inventory
+	if endHeight > startHeight && endHeight-startHeight > maxRollbackHeight {
+		for i := startHeight; i < endHeight; i += types.MaxBlockCountPerTime {
+			inv := new(inventory)
+			inv.txs = &types.ParaTxDetails{}
+			inv.start = i
+			inv.end = i + types.MaxBlockCountPerTime - 1
+			if inv.end > endHeight {
+				inv.end = endHeight
+				invs = append(invs, inv)
+				return invs
+			}
+			invs = append(invs, inv)
+		}
+	}
+	return invs
+}
+
+func (m *multiDldClient) tryMultiServerDownload() {
+	if !m.multiDldOpen {
+		return
+	}
+
+	paraRemoteGrpcIps := types.Conf("config.consensus.sub.para").GStr("ParaRemoteGrpcClient")
+	ips := strings.Split(paraRemoteGrpcIps, ",")
+	var conns []*connectCli
+	for _, ip := range ips {
+		conn, err := grpcclient.NewMainChainClient(ip)
+		if err == nil {
+			conns = append(conns, &connectCli{conn: conn, ip: ip})
+		}
+	}
+	if len(conns) == 0 {
+		plog.Info("multiDownload not valid ips")
+		return
+	}
+	m.conns = conns
+
+	curMainHeight, err := m.paraClient.GetLastHeightOnMainChain()
+	if err != nil {
+		return
+	}
+
+	//如果切换不成功，则不进行多服务下载
+	_, localBlock, err := m.paraClient.switchLocalHashMatchedBlock()
+	if err != nil {
+		return
+	}
+
+	//获取批量下载区间和数量，给curMainHeight留10000的回滚buffer
+	totalInvs := m.getInvs(localBlock.MainHeight+1, curMainHeight-maxRollbackHeight)
+	totalInvsNum := int64(len(totalInvs))
+	if totalInvsNum == 0 {
+		plog.Info("multiDownload no invs need download")
+		return
+	}
+
+	plog.Info("tryMultiServerDownload", "start", localBlock.MainHeight+1, "end", curMainHeight-maxRollbackHeight, "totalInvs", totalInvsNum)
+
+	jobsCh := make(chan *downloadJob, m.jobBufferNum)
+	m.wg.Add(1)
+	go m.processDoneJobs(jobsCh)
+
+	preBlock := &types.ParaTxDetail{
+		Type:   types.AddBlock,
+		Header: &types.Header{Hash: localBlock.MainHash, Height: localBlock.MainHeight},
+	}
+	for i := int64(0); i < totalInvsNum; i += m.invNumPerJob {
+		end := i + m.invNumPerJob
+		if end > totalInvsNum {
+			end = totalInvsNum
+		}
+		job := newDownLoadJob(m)
+		job.invs = append(job.invs, totalInvs[i:end]...)
+		job.parentBlock = preBlock
+		job.GetBlocks()
+		if m.paraClient.isCancel() {
+			break
+		}
+		jobsCh <- job
+		plog.Info("tryMultiServerDownload", "start", i, "end", end, "total", totalInvsNum)
+		preBlock = job.invs[len(job.invs)-1].getLastBlock(job)
+
+	}
+	close(jobsCh)
+	m.wg.Wait()
+}
+
+func (i *inventory) getFirstBlock(d *downloadJob) *types.ParaTxDetail {
+	if i.isSaveDb {
+		block, err := d.getBlockFromDb(i.start)
+		if err != nil {
+			panic(err)
+		}
+		return block
+	}
+	return i.txs.Items[0]
+}
+
+func (i *inventory) getLastBlock(d *downloadJob) *types.ParaTxDetail {
+	if !i.isDone {
+		return nil
+	}
+	if i.isSaveDb {
+		block, err := d.getBlockFromDb(i.end)
+		if err != nil {
+			panic(err)
+		}
+		return block
+	}
+	return i.txs.Items[len(i.txs.Items)-1]
+}
+
+func (m *multiDldClient) processDoneJobs(ch chan *downloadJob) {
+	defer m.wg.Done()
+
+	for job := range ch {
+		job.process()
+	}
+}
+
+func (d *downloadJob) process() {
+	for _, inv := range d.invs {
+		if inv.isSaveDb {
+			for i := inv.start; i <= inv.end; i++ {
+				block, err := d.getBlockFromDb(i)
+				if err != nil {
+					panic(err)
+				}
+				_, err = d.mDldCli.paraClient.procLocalBlock(block)
+				if err != nil {
+					panic(err)
+				}
+
+			}
+			continue
+		}
+
+		//block需要严格顺序执行，数据库错误，panic 重新来过
+		err := d.mDldCli.paraClient.procLocalBlocks(inv.txs)
+		if err != nil {
+			panic(err)
+		}
+
+	}
+}
+
+func (d *downloadJob) getPreVerifyBlock(inv *inventory) (*types.ParaTxDetail, error) {
+	if inv.isSaveDb {
+
+		lastBlock, err := d.getBlockFromDb(inv.curHeight - 1)
+		if err != nil {
+			return nil, err
+		}
+		return lastBlock, nil
+	}
+	if len(inv.txs.Items) != 0 {
+		return inv.txs.Items[len(inv.txs.Items)-1], nil
+	}
+	return nil, nil
+}
+
+func (d *downloadJob) verifyDownloadBlock(inv *inventory, blocks *types.ParaTxDetails) error {
+	//返回区块内部校验
+	err := verifyMainBlocksInternal(blocks)
+	if err != nil {
+		plog.Error("verifyDownloadBlock internal", "ip", inv.connCli.ip)
+		return err
+	}
+
+	//跟已下载的区块校验
+	verifyBlock, err := d.getPreVerifyBlock(inv)
+	if err != nil {
+		plog.Error("verifyDownloadBlock.getPreVerifyBlock", "ip", inv.connCli.ip)
+		return err
+	}
+	if verifyBlock != nil {
+		err = verifyMainBlockHash(getVerifyHash(verifyBlock), blocks.Items[0])
+		if err != nil {
+			d.rmvBatchMainBlocks(inv)
+			inv.curHeight = inv.start
+			inv.txs.Items = nil
+			inv.isSaveDb = false
+			plog.Error("verifyDownloadBlock.verfiy", "ip", inv.connCli.ip)
+			return err
+		}
+	}
+	return nil
+}
+
+func (d *downloadJob) saveMainBlock(height int64, block *types.ParaTxDetail) error {
+	set := &types.LocalDBSet{}
+
+	key := calcTitleMainHeightKey(types.GetTitle(), height)
+	kv := &types.KeyValue{Key: key, Value: types.Encode(block)}
+	set.KV = append(set.KV, kv)
+
+	return d.mDldCli.paraClient.setLocalDb(set)
+}
+
+func (d *downloadJob) saveBatchMainBlocks(txs *types.ParaTxDetails) error {
+	set := &types.LocalDBSet{}
+
+	for _, block := range txs.Items {
+		key := calcTitleMainHeightKey(types.GetTitle(), block.Header.Height)
+		kv := &types.KeyValue{Key: key, Value: types.Encode(block)}
+		set.KV = append(set.KV, kv)
+	}
+
+	return d.mDldCli.paraClient.setLocalDb(set)
+}
+
+func (d *downloadJob) rmvBatchMainBlocks(inv *inventory) error {
+	set := &types.LocalDBSet{}
+
+	for i := inv.start; i <= inv.curHeight; i++ {
+		key := calcTitleMainHeightKey(types.GetTitle(), i)
+		kv := &types.KeyValue{Key: key, Value: nil}
+		set.KV = append(set.KV, kv)
+	}
+
+	return d.mDldCli.paraClient.setLocalDb(set)
+}
+
+func (d *downloadJob) getBlockFromDb(height int64) (*types.ParaTxDetail, error) {
+	key := calcTitleMainHeightKey(types.GetTitle(), height)
+	set := &types.LocalDBGet{Keys: [][]byte{key}}
+
+	value, err := d.mDldCli.paraClient.getLocalDb(set, len(set.Keys))
+	if err != nil {
+		return nil, err
+	}
+	if len(value) == 0 || value[0] == nil {
+		return nil, types.ErrNotFound
+	}
+
+	var tx types.ParaTxDetail
+	err = types.Decode(value[0], &tx)
+	if err != nil {
+		return nil, err
+	}
+	return &tx, nil
+}
+
+func (d *downloadJob) checkInv(lastRetry, pre *types.ParaTxDetail, inv *inventory) error {
+	if !inv.isDone {
+		return types.ErrNotFound
+	}
+
+	if lastRetry == pre {
+		return nil
+	}
+
+	return verifyMainBlockHash(getVerifyHash(pre), inv.getFirstBlock(d))
+
+}
+
+// 对一个job里面的invs之间头尾做校验， 如果后一个跟之前的校验不过，放入retry，retry后面的一个inv暂时跳过校验，继续和后面的做校验
+func (d *downloadJob) verifyInvs() []*inventory {
+	var retryItems []*inventory
+	pre := d.parentBlock
+	var lastRetry *types.ParaTxDetail
+	for _, inv := range d.invs {
+		err := d.checkInv(lastRetry, pre, inv)
+		if err != nil {
+			plog.Info("verifyInvs", "height", inv.start)
+			retryItems = append(retryItems, inv)
+			lastRetry = inv.getLastBlock(d)
+		}
+		pre = inv.getLastBlock(d)
+
+	}
+
+	return retryItems
+
+}
+
+func (d *downloadJob) checkDownLoadRate() {
+	if d.mDldCli.connsCheckDone {
+		return
+	}
+
+	var fastConns, slowConns []*connectCli
+
+	for _, conn := range d.mDldCli.conns {
+		if conn.downTimes >= downTimesFastThreshold {
+			fastConns = append(fastConns, conn)
+		}
+		if conn.downTimes <= downTimesSlowThreshold {
+			slowConns = append(slowConns, conn)
+		}
+	}
+
+	if len(fastConns) > 0 {
+		for _, conn := range slowConns {
+			conn.isFail = true
+			plog.Info("paramultiDownload.checkDownLoadRate removed server", "ip", conn.ip, "times", conn.downTimes)
+		}
+		d.mDldCli.connsCheckDone = true
+	}
+
+}
+
+func (d *downloadJob) requestMainBlocks(inv *inventory) (*types.ParaTxDetails, error) {
+	req := &types.ReqParaTxByTitle{IsSeq: false, Start: inv.curHeight, End: inv.end, Title: types.GetTitle()}
+	txs, err := inv.connCli.conn.GetParaTxByTitle(context.Background(), req)
+	if err != nil {
+		return nil, err
+	}
+
+	for i, item := range txs.Items {
+		if item != nil && item.Header.Height != int64(i)+req.Start {
+			plog.Error("requestMainBlocks block notmatch", "expect", int64(i)+req.Start, "height", item.Header.Height)
+			return nil, types.ErrBlockHeightNoMatch
+		}
+	}
+
+	//只获取最前面的有效交易
+	return validMainBlocks(txs), nil
+}
+
+func (d *downloadJob) getInvBlocks(inv *inventory, connPool chan *connectCli) {
+	defer func() {
+		d.wg.Done()
+	}()
+
+	inv.curHeight = inv.start
+	plog.Debug("getInvBlocks begin", "start", inv.start, "end", inv.end)
+	for {
+		txs, err := d.requestMainBlocks(inv)
+		if err != nil {
+			plog.Error("getInvBlocks connect error", "err", err, "ip", inv.connCli.ip)
+			return
+		}
+		if len(txs.Items) == 0 {
+			plog.Error("getInvBlocks not items down", "ip", inv.connCli.ip)
+			continue
+		}
+
+		err = d.verifyDownloadBlock(inv, txs)
+		if err != nil {
+			continue
+		}
+
+		//save  之前save到db，后面区块全部save到db
+		if inv.isSaveDb {
+			d.saveBatchMainBlocks(txs)
+		} else {
+			inv.txs.Items = append(inv.txs.Items, txs.Items...)
+		}
+
+		//check done
+		if txs.Items[len(txs.Items)-1].Header.Height == inv.end {
+			inv.connCli.downTimes++
+			plog.Info("getInvs done", "start", inv.start, "end", inv.end, "downtimes", inv.connCli.downTimes, "ip", inv.connCli.ip)
+			inv.isDone = true
+			connPool <- inv.connCli
+			return
+		}
+		if !inv.isSaveDb && types.Size(inv.txs) > maxBlockSize {
+			d.saveBatchMainBlocks(inv.txs)
+			inv.txs.Items = nil
+			inv.isSaveDb = true
+		}
+		inv.curHeight = txs.Items[len(txs.Items)-1].Header.Height + 1
+
+	}
+}
+
+// getInvs download the block
+func (d *downloadJob) getInvs(invs []*inventory) {
+	connPool := make(chan *connectCli, len(d.mDldCli.conns))
+	for _, conn := range d.mDldCli.conns {
+		if !conn.isFail {
+			connPool <- conn
+		}
+
+	}
+
+	for _, inv := range invs {
+		if d.mDldCli.paraClient.isCancel() {
+			break
+		}
+		inv.connCli = <-connPool
+		d.wg.Add(1)
+		go d.getInvBlocks(inv, connPool)
+
+	}
+
+	//等待下载任务
+	d.wg.Wait()
+}
+
+// GetBlocks get blocks information
+func (d *downloadJob) GetBlocks() {
+	invs := d.invs
+	for {
+		d.getInvs(invs)
+		invs = d.verifyInvs()
+		d.checkDownLoadRate()
+		if len(invs) == 0 {
+			return
+		}
+
+		if d.mDldCli.paraClient.isCancel() {
+			return
+		}
+
+	}
+}

plugin/consensus/para/paratxrequest_test.go

@@ -150,7 +150,7 @@ func TestVerifyMainBlocks(t *testing.T) {
 		Hash:       hash1,
 	}
 	block1 := &types.ParaTxDetail{
-		Type:   addAct,
+		Type:   types.AddBlock,
 		Header: header1,
 	}
 
@@ -159,7 +159,7 @@ func TestVerifyMainBlocks(t *testing.T) {
 		Hash:       hash2,
 	}
 	block2 := &types.ParaTxDetail{
-		Type:   addAct,
+		Type:   types.AddBlock,
 		Header: header2,
 	}
 
@@ -168,7 +168,7 @@ func TestVerifyMainBlocks(t *testing.T) {
 		Hash:       hash3,
 	}
 	block3 := &types.ParaTxDetail{
-		Type:   addAct,
+		Type:   types.AddBlock,
 		Header: header3,
 	}
 
@@ -178,7 +178,7 @@ func TestVerifyMainBlocks(t *testing.T) {
 		Hash:       hash3,
 	}
 	block4 := &types.ParaTxDetail{
-		Type:   delAct,
+		Type:   types.DelBlock,
 		Header: header4,
 	}
 	//del2
@@ -187,7 +187,7 @@ func TestVerifyMainBlocks(t *testing.T) {
 		Hash:       hash2,
 	}
 	block5 := &types.ParaTxDetail{
-		Type:   delAct,
+		Type:   types.DelBlock,
 		Header: header5,
 	}
 
@@ -196,7 +196,7 @@ func TestVerifyMainBlocks(t *testing.T) {
 		Hash:       hash6,
 	}
 	block6 := &types.ParaTxDetail{
-		Type:   addAct,
+		Type:   types.AddBlock,
 		Header: header6,
 	}