add ut

plugin/consensus/para/paramultidownload.go

@@ -17,19 +17,18 @@ import (
 )
 
 const (
-	maxRollbackHeight      int64 = 10000
-	defaultInvNumPerJob          = 20       // 20inv task 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倍，需要剔除
+	maxRollbackHeight      = 10000    // 据当前主链高度回滚的余量
+	defaultInvNumPerJob    = 20       // 20inv task per job
+	defaultJobBufferNum    = 20       // channel buffer num for done job process
+	maxBlockSize           = 20000000 // 单次1000block size累积超过20M 需保存到localdb
+	downTimesFastThreshold = 450      //  单个server 下载超过450次，平均20次用20s来计算，下载7分钟左右检查有没有差别比较大的
+	downTimesSlowThreshold = 30       //  慢的server小于30次，则小于快server的15倍，需要剔除
 )
 
 type connectCli struct {
 	ip        string
 	conn      types.Chain33Client
 	downTimes int64
-	isFail    bool
 }
 
 //invertory 是每次请求的最小单位，每次请求最多MaxBlockCountPerTime
@@ -355,22 +354,22 @@ func (d *downloadJob) checkDownLoadRate() {
 		return
 	}
 
-	var fastConns, slowConns []*connectCli
-
+	var found bool
 	for _, conn := range d.mDldCli.conns {
 		if conn.downTimes >= downTimesFastThreshold {
-			fastConns = append(fastConns, conn)
-		}
-		if conn.downTimes <= downTimesSlowThreshold {
-			slowConns = append(slowConns, conn)
+			found = true
+			break
 		}
 	}
 
-	if len(fastConns) > 0 {
-		for _, conn := range slowConns {
-			conn.isFail = true
-			plog.Info("paramultiDownload.checkDownLoadRate removed server", "ip", conn.ip, "times", conn.downTimes)
+	if found {
+		var fastConns []*connectCli
+		for _, conn := range d.mDldCli.conns {
+			if conn.downTimes > downTimesSlowThreshold {
+				fastConns = append(fastConns, conn)
+			}
 		}
+		d.mDldCli.conns = fastConns
 		d.mDldCli.connsCheckDone = true
 	}
 
@@ -445,9 +444,7 @@ func (d *downloadJob) getInvBlocks(inv *inventory, connPool chan *connectCli) {
 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
-		}
+		connPool <- conn
 	}
 
 	for _, inv := range invs {

plugin/consensus/para/paramultidownload_test.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 (
+	"testing"
+
+	"github.com/33cn/chain33/types"
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	count = 3
+)
+
+func getTestInv(height int64, isDone bool) *inventory {
+
+	headDetail := &types.ParaTxDetail{
+		Type: types.AddBlock,
+		Header: &types.Header{
+			ParentHash: []byte(string(height - 1)),
+			Hash:       []byte(string(height))}}
+	endDetail := &types.ParaTxDetail{
+		Type: types.AddBlock,
+		Header: &types.Header{
+			ParentHash: []byte(string(height + count - 2)),
+			Hash:       []byte(string(height + count - 1))}}
+	txs1 := &types.ParaTxDetails{Items: []*types.ParaTxDetail{headDetail, endDetail, endDetail}}
+	return &inventory{
+		start:  height,
+		end:    height + count - 1,
+		txs:    txs1,
+		isDone: isDone,
+	}
+}
+
+func TestVerifyInvs(t *testing.T) {
+	var invs []*inventory
+	start := int64(1)
+	inv1 := getTestInv(start, false)
+	invs = append(invs, inv1)
+
+	end := start + count
+	inv2 := getTestInv(end, true)
+	invs = append(invs, inv2)
+
+	end += count
+	inv3 := getTestInv(end, false)
+	invs = append(invs, inv3)
+
+	end += count
+	inv4 := getTestInv(end, true)
+	invs = append(invs, inv4)
+
+	end += count
+	inv5 := getTestInv(end, true)
+	invs = append(invs, inv5)
+
+	end += count
+	inv6 := getTestInv(end, false)
+	invs = append(invs, inv6)
+
+	end += count
+	inv7 := getTestInv(end, true)
+	invs = append(invs, inv7)
+
+	end += count
+	inv8 := getTestInv(end, true)
+	invs = append(invs, inv8)
+
+	end += count
+	inv9 := getTestInv(end, false)
+	invs = append(invs, inv9)
+
+	wrongItems := []*inventory{inv1, inv3, inv6, inv9}
+
+	preBlock := &types.ParaTxDetail{
+		Type:   types.AddBlock,
+		Header: &types.Header{Hash: []byte(string(start - 1))},
+	}
+	d := &downloadJob{
+		parentBlock: preBlock,
+		invs:        invs,
+	}
+
+	retry := d.verifyInvs()
+	assert.Equal(t, wrongItems, retry)
+}
+
+func TestCheckDownLoadRate(t *testing.T) {
+	conn1 := &connectCli{downTimes: downTimesFastThreshold + 1}
+	conn2 := &connectCli{downTimes: downTimesFastThreshold - 20}
+	conn3 := &connectCli{downTimes: downTimesSlowThreshold - 10}
+	conn4 := &connectCli{downTimes: downTimesSlowThreshold - 20}
+	conn5 := &connectCli{downTimes: downTimesSlowThreshold + 5}
+	mCli := &multiDldClient{conns: []*connectCli{conn1, conn2, conn3, conn4, conn5}}
+	d := &downloadJob{
+		mDldCli: mCli,
+	}
+
+	expectConn := []*connectCli{conn1, conn2, conn5}
+
+	d.checkDownLoadRate()
+	assert.Equal(t, true, d.mDldCli.connsCheckDone)
+	assert.Equal(t, expectConn, d.mDldCli.conns)
+}
+
+func TestGetInvs(t *testing.T) {
+	subCfg := &subConfig{BatchFetchBlockCount: 1000}
+	cli := &client{subCfg: subCfg}
+	mCli := &multiDldClient{paraClient: cli}
+
+	var start, end int64
+	start = 345800
+	end = start + maxRollbackHeight + 3
+	invs := mCli.getInvs(start, end)
+	assert.Equal(t, 11, len(invs))
+	assert.Equal(t, invs[0].end+1, invs[1].start)
+	assert.Equal(t, invs[3].end+1, invs[4].start)
+	assert.Equal(t, invs[9].end+1, invs[10].start)
+
+}