Merge pull request #543 from vipwzw/add_test_mempool_price

plugin/mempool/price/cache.go

@@ -3,122 +3,66 @@ package price
 import (
 	"github.com/33cn/chain33/common/skiplist"
 	"github.com/33cn/chain33/system/mempool"
-	"github.com/33cn/chain33/types"
 	"github.com/golang/protobuf/proto"
 )
 
-var mempoolDupResendInterval int64 = 600 // mempool内交易过期时间，10分钟
-
 // Queue 价格队列模式(价格=手续费/交易字节数,价格高者优先,同价则时间早优先)
 type Queue struct {
-	txMap     map[string]*skiplist.SkipValue
-	txList    *skiplist.SkipList
+	*skiplist.Queue
 	subConfig subConfig
 }
 
-// NewQueue 创建队列
-func NewQueue(subcfg subConfig) *Queue {
-	return &Queue{
-		make(map[string]*skiplist.SkipValue, subcfg.PoolCacheSize),
-		skiplist.NewSkipList(&skiplist.SkipValue{Score: -1, Value: nil}),
-		subcfg,
-	}
+type priceScore struct {
+	*mempool.Item
 }
 
-func (cache *Queue) newSkipValue(item *mempool.Item) (*skiplist.SkipValue, error) {
+func (item *priceScore) GetScore() int64 {
 	txSize := proto.Size(item.Value)
-	return &skiplist.SkipValue{Score: item.Value.Fee / int64(txSize), Value: item}, nil
+	return item.Value.Fee / int64(txSize)
 }
 
-//Exist 是否存在
-func (cache *Queue) Exist(hash string) bool {
-	_, exists := cache.txMap[hash]
-	return exists
+func (item *priceScore) Hash() []byte {
+	return item.Value.Hash()
 }
 
-//GetItem 获取数据通过 key
-func (cache *Queue) GetItem(hash string) (*mempool.Item, error) {
-	if k, exist := cache.txMap[hash]; exist {
-		return k.Value.(*mempool.Item), nil
+func (item *priceScore) Compare(cmp skiplist.Scorer) int {
+	it := cmp.(*priceScore)
+	//时间越小，权重越高
+	if item.EnterTime < it.EnterTime {
+		return skiplist.Big
+	}
+	if item.EnterTime == it.EnterTime {
+		return skiplist.Equal
 	}
-	return nil, types.ErrNotFound
+	return skiplist.Small
 }
 
-// Push 把给定tx添加到Queue；如果tx已经存在Queue中或Mempool已满则返回对应error
-func (cache *Queue) Push(item *mempool.Item) error {
-	hash := item.Value.Hash()
-	if cache.Exist(string(hash)) {
-		s := cache.txMap[string(hash)]
-		addedItem := s.Value.(*mempool.Item)
-		addedTime := addedItem.EnterTime
-		if types.Now().Unix()-addedTime < mempoolDupResendInterval {
-			return types.ErrTxExist
-		}
-		// 超过2分钟之后的重发交易返回nil，再次发送给P2P，但是不再次加入mempool
-		// 并修改其enterTime，以避免该交易一直在节点间被重发
-		newEnterTime := types.Now().Unix()
-		resendItem := &mempool.Item{Value: item.Value, Priority: item.Value.Fee, EnterTime: newEnterTime}
-		var err error
-		sv, err := cache.newSkipValue(resendItem)
-		if err != nil {
-			return err
-		}
-		cache.Remove(string(hash))
-		cache.txList.Insert(sv)
-		cache.txMap[string(hash)] = sv
-		// ------------------
-		return nil
+// NewQueue 创建队列
+func NewQueue(subcfg subConfig) *Queue {
+	return &Queue{
+		Queue:     skiplist.NewQueue(subcfg.PoolCacheSize),
+		subConfig: subcfg,
 	}
+}
 
-	it := &mempool.Item{Value: item.Value, Priority: item.Value.Fee, EnterTime: item.EnterTime}
-	sv, err := cache.newSkipValue(it)
+//GetItem 获取数据通过 key
+func (cache *Queue) GetItem(hash string) (*mempool.Item, error) {
+	item, err := cache.Queue.GetItem(hash)
 	if err != nil {
-		return err
-	}
-	if int64(cache.txList.Len()) >= cache.subConfig.PoolCacheSize {
-		tail := cache.txList.GetIterator().Last()
-		//价格高存留
-		switch sv.Compare(tail) {
-		case -1:
-			cache.Remove(string(tail.Value.(*mempool.Item).Value.Hash()))
-		case 0:
-			if sv.Value.(*mempool.Item).EnterTime < tail.Value.(*mempool.Item).EnterTime {
-				cache.Remove(string(tail.Value.(*mempool.Item).Value.Hash()))
-				break
-			}
-			return types.ErrMemFull
-		case 1:
-			return types.ErrMemFull
-		default:
-			return types.ErrMemFull
-		}
+		return nil, err
 	}
-	cache.txList.Insert(sv)
-	cache.txMap[string(hash)] = sv
-	return nil
+	return item.(*priceScore).Item, nil
 }
 
-// Remove 删除数据
-func (cache *Queue) Remove(hash string) error {
-	cache.txList.Delete(cache.txMap[hash])
-	delete(cache.txMap, hash)
-	return nil
-}
-
-// Size 数据总数
-func (cache *Queue) Size() int {
-	return cache.txList.Len()
+//Push 加入数据到队列
+func (cache *Queue) Push(item *mempool.Item) error {
+	return cache.Queue.Push(&priceScore{Item: item})
 }
 
-// Walk 遍历整个队列
-func (cache *Queue) Walk(count int, cb func(value *mempool.Item) bool) {
-	i := 0
-	cache.txList.Walk(func(item interface{}) bool {
-		if !cb(item.(*mempool.Item)) {
-			return false
-		}
-		i++
-		return i != count
+//Walk 获取数据通过 key
+func (cache *Queue) Walk(count int, cb func(tx *mempool.Item) bool) {
+	cache.Queue.Walk(count, func(item skiplist.Scorer) bool {
+		return cb(item.(*priceScore).Item)
 	})
 }
 
@@ -132,12 +76,9 @@ func (cache *Queue) GetProperFee() int64 {
 	i := 0
 	var txSize int
 	var feeRate int64
-	cache.txList.Walk(func(tx interface{}) bool {
-		if i == 100 {
-			return false
-		}
-		txSize = proto.Size(tx.(*mempool.Item).Value)
-		feeRate = tx.(*mempool.Item).Value.Fee / int64(txSize/1000+1)
+	cache.Walk(100, func(item *mempool.Item) bool {
+		txSize = proto.Size(item.Value)
+		feeRate = item.Value.Fee / int64(txSize/1000+1)
 		sumFeeRate += feeRate
 		i++
 		return true

plugin/mempool/price/cache_test.go

 package price
 
 import (
+	"log"
 	"testing"
 
 	"github.com/33cn/chain33/common"
@@ -9,8 +10,12 @@ import (
 	cty "github.com/33cn/chain33/system/dapp/coins/types"
 	drivers "github.com/33cn/chain33/system/mempool"
 	"github.com/33cn/chain33/types"
+	"github.com/33cn/chain33/util"
+	"github.com/33cn/chain33/util/testnode"
 	"github.com/golang/protobuf/proto"
 	"github.com/stretchr/testify/assert"
+
+	_ "github.com/33cn/chain33/system"
 )
 
 var (
@@ -134,12 +139,11 @@ func TestQueueDirection(t *testing.T) {
 	cache.Push(item3)
 	cache.Push(item4)
 	cache.Push(item5)
-	cache.txList.Print()
 	i := 0
-	lastScore := cache.txList.GetIterator().First().Score
+	lastScore := cache.First().GetScore()
 	var tmpScore int64
 	cache.Walk(5, func(value *drivers.Item) bool {
-		tmpScore = cache.txMap[string(value.Value.Hash())].Score
+		tmpScore = cache.CreateSkipValue(&priceScore{Item: value}).Score
 		if lastScore < tmpScore {
 			return false
 		}
@@ -148,7 +152,7 @@ func TestQueueDirection(t *testing.T) {
 		return true
 	})
 	assert.Equal(t, 5, i)
-	assert.Equal(t, true, lastScore == cache.txList.GetIterator().Last().Score)
+	assert.Equal(t, true, lastScore == cache.Last().GetScore())
 }
 
 func TestGetProperFee(t *testing.T) {
@@ -162,3 +166,73 @@ func TestGetProperFee(t *testing.T) {
 	txSize2 := proto.Size(item4.Value)
 	assert.Equal(t, (item1.Value.Fee/int64(txSize1/1000+1)+item4.Value.Fee/int64(txSize2/1000+1))/2, cache.GetProperFee())
 }
+
+func TestRealNodeMempool(t *testing.T) {
+	mock33 := testnode.New("chain33.test.toml", nil)
+	defer mock33.Close()
+	mock33.Listen()
+	mock33.WaitHeight(0)
+	mock33.SendHot()
+	mock33.WaitHeight(1)
+	n := 20
+	done := make(chan struct{}, n)
+	keys := make([]crypto.PrivKey, n)
+	for i := 0; i < n; i++ {
+		addr, priv := util.Genaddress()
+		tx := util.CreateCoinsTx(mock33.GetHotKey(), addr, 10*types.Coin)
+		mock33.SendTx(tx)
+		keys[i] = priv
+	}
+	mock33.Wait()
+	for i := 0; i < n; i++ {
+		go func(priv crypto.PrivKey) {
+			for i := 0; i < 100; i++ {
+				tx := util.CreateCoinsTx(priv, mock33.GetGenesisAddress(), types.Coin/1000)
+				reply, err := mock33.GetAPI().SendTx(tx)
+				if err != nil {
+					log.Println(err)
+					continue
+				}
+				//发送交易组
+				tx1 := util.CreateCoinsTx(priv, mock33.GetGenesisAddress(), types.Coin/1000)
+				tx2 := util.CreateCoinsTx(priv, mock33.GetGenesisAddress(), types.Coin/1000)
+				txgroup, err := types.CreateTxGroup([]*types.Transaction{tx1, tx2})
+				if err != nil {
+					log.Println(err)
+					continue
+				}
+				for i := 0; i < len(txgroup.GetTxs()); i++ {
+					err = txgroup.SignN(i, types.SECP256K1, priv)
+					if err != nil {
+						t.Error(err)
+						return
+					}
+				}
+				reply, err = mock33.GetAPI().SendTx(txgroup.Tx())
+				if err != nil {
+					log.Println(err)
+					continue
+				}
+				mock33.SetLastSend(reply.GetMsg())
+			}
+			done <- struct{}{}
+		}(keys[i])
+	}
+	for i := 0; i < n; i++ {
+		<-done
+	}
+	for {
+		txs, err := mock33.GetAPI().GetMempool()
+		assert.Nil(t, err)
+		println("len", len(txs.GetTxs()))
+		if len(txs.GetTxs()) > 0 {
+			mock33.Wait()
+			continue
+		}
+		break
+	}
+	peer, err := mock33.GetAPI().PeerInfo()
+	assert.Nil(t, err)
+	assert.Equal(t, len(peer.Peers), 0)
+	//assert.Equal(t, peer.Peers[0].MempoolSize, int32(0))
+}

plugin/mempool/price/chain33.test.toml

-Title="chain33"
+Title="local"
 TestNet=true
 
 [log]
@@ -41,7 +41,7 @@ enableTxQuickIndex=false
 [p2p]
 port=13802
 seeds=["47.104.125.151:13802","47.104.125.97:13802","47.104.125.177:13802"]
-enable=true
+enable=false
 isSeed=true
 serverStart=true
 msgCacheSize=10240
@@ -65,7 +65,7 @@ keyFile="key.pem"
 
 [mempool]
 name="price"
-poolCacheSize=10240
+poolCacheSize=200
 minTxFee=100000
 maxTxNumPerAccount=100
 
@@ -188,4 +188,5 @@ superManager=[
     "1Bsg9j6gW83sShoee1fZAt9TkUjcrCgA9S",
     "12qyocayNF7Lv6C9qW4avxs2E7U41fKSfv",
     "1Q8hGLfoGe63efeWa8fJ4Pnukhkngt6poK"
-]
\ No newline at end of file
+]
+

vendor/github.com/33cn/chain33/blockchain/restore.go

@@ -9,9 +9,9 @@ import (
 
 	"fmt"
 
+	dbm "github.com/33cn/chain33/common/db"
 	"github.com/33cn/chain33/common/version"
 	"github.com/33cn/chain33/types"
-	dbm "github.com/33cn/chain33/common/db"
 )
 
 // Upgrade 升级localDB和storeDB

vendor/github.com/33cn/chain33/common/db/db_test.go

@@ -47,6 +47,51 @@ MAIN_LOOP:
 	return string(chars)
 }
 
+func testDBIteratorAllKey(t *testing.T, db DB) {
+	var datas = [][]byte{
+		[]byte("aa0"), []byte("aa1"), []byte("bb0"), []byte("bb1"), []byte("cc0"), []byte("cc1"),
+	}
+	for _, v := range datas {
+		db.Set(v, v)
+	}
+	//一次遍历
+	it := db.Iterator(nil, types.EmptyValue, false)
+	i := 0
+	for it.Rewind(); it.Valid(); it.Next() {
+		assert.Equal(t, it.Key(), datas[i])
+		db.Delete(it.Key())
+		i++
+		if i == 2 {
+			break
+		}
+	}
+	it.Close()
+	//从第3个开始遍历
+	it = db.Iterator([]byte("aa1"), types.EmptyValue, false)
+	i = 2
+	for it.Rewind(); it.Valid(); it.Next() {
+		assert.Equal(t, it.Key(), datas[i])
+		db.Delete(it.Key())
+		i++
+		if i == 4 {
+			break
+		}
+	}
+	it.Close()
+	//从第5个开始遍历
+	it = db.Iterator([]byte("bb1"), types.EmptyValue, false)
+	i = 4
+	for it.Rewind(); it.Valid(); it.Next() {
+		assert.Equal(t, it.Key(), datas[i])
+		db.Delete(it.Key())
+		i++
+		if i == 6 {
+			break
+		}
+	}
+	it.Close()
+}
+
 // 迭代测试
 func testDBIterator(t *testing.T, db DB) {
 	t.Log("test Set")

vendor/github.com/33cn/chain33/common/db/go_level_db_test.go

@@ -29,6 +29,16 @@ func TestGoLevelDBIterator(t *testing.T) {
 	testDBIterator(t, leveldb)
 }
 
+func TestGoLevelDBIteratorAll(t *testing.T) {
+	dir, err := ioutil.TempDir("", "goleveldb")
+	require.NoError(t, err)
+	t.Log(dir)
+	leveldb, err := NewGoLevelDB("goleveldb", dir, 128)
+	require.NoError(t, err)
+	defer leveldb.Close()
+	testDBIteratorAllKey(t, leveldb)
+}
+
 func TestGoLevelDBIteratorDel(t *testing.T) {
 	dir, err := ioutil.TempDir("", "goleveldb")
 	require.NoError(t, err)

vendor/github.com/33cn/chain33/common/db/list_helper.go

@@ -105,9 +105,8 @@ func (db *ListHelper) IteratorScan(prefix []byte, key []byte, count int32, direc
 	return
 }
 
-//IteratorScanFromFirst 从头迭代
-func (db *ListHelper) IteratorScanFromFirst(prefix []byte, count int32) (values [][]byte) {
-	it := db.db.Iterator(prefix, nil, false)
+func (db *ListHelper) iteratorScan(prefix []byte, count int32, reverse bool) (values [][]byte) {
+	it := db.db.Iterator(prefix, nil, reverse)
 	defer it.Close()
 	var i int32
 	for it.Rewind(); it.Valid(); it.Next() {
@@ -127,27 +126,14 @@ func (db *ListHelper) IteratorScanFromFirst(prefix []byte, count int32) (values
 	return
 }
 
+//IteratorScanFromFirst 从头迭代
+func (db *ListHelper) IteratorScanFromFirst(prefix []byte, count int32) (values [][]byte) {
+	return db.iteratorScan(prefix, count, false)
+}
+
 //IteratorScanFromLast 从尾迭代
 func (db *ListHelper) IteratorScanFromLast(prefix []byte, count int32) (values [][]byte) {
-	it := db.db.Iterator(prefix, nil, true)
-	defer it.Close()
-
-	var i int32
-	for it.Rewind(); it.Valid(); it.Next() {
-		value := it.ValueCopy()
-		if it.Error() != nil {
-			listlog.Error("PrefixScan it.Value()", "error", it.Error())
-			values = nil
-			return
-		}
-		// blog.Debug("PrefixScan", "key", string(item.Key()), "value", value)
-		values = append(values, value)
-		i++
-		if i == count {
-			break
-		}
-	}
-	return
+	return db.iteratorScan(prefix, count, true)
 }
 
 //PrefixCount 前缀数量

vendor/github.com/33cn/chain33/common/skiplist/queue.go

+package skiplist
+
+import (
+	"container/list"
+
+	"github.com/33cn/chain33/types"
+)
+
+//Scorer 接口实现 Value的 Score 功能
+type Scorer interface {
+	GetScore() int64
+	Hash() []byte
+	//在score相同情况下的比较
+	Compare(Scorer) int
+}
+
+// Queue skiplist 实现的一个 按照score 排序的队列，score相同的按照元素到的先后排序
+type Queue struct {
+	txMap   map[string]*list.Element
+	txList  *SkipList
+	maxsize int64
+}
+
+// NewQueue 创建队列
+func NewQueue(maxsize int64) *Queue {
+	return &Queue{
+		txMap:   make(map[string]*list.Element),
+		txList:  NewSkipList(&SkipValue{Score: -1, Value: nil}),
+		maxsize: maxsize,
+	}
+}
+
+/*
+为了处理相同 Score 的问题，需要一个队列保存相同 Score 下面的交易
+*/
+func (cache *Queue) insertSkipValue(item Scorer) *list.Element {
+	skvalue := cache.CreateSkipValue(item)
+	value := cache.txList.Find(skvalue)
+	var txlist *list.List
+	if value == nil {
+		txlist = list.New()
+		skvalue.Value = txlist
+		cache.txList.Insert(skvalue)
+	} else {
+		txlist = value.Value.(*list.List)
+	}
+	return txlist.PushBack(item)
+}
+
+func (cache *Queue) deleteSkipValue(item *list.Element) error {
+	if item == nil {
+		return nil
+	}
+	skvalue := cache.CreateSkipValue(item.Value.(Scorer))
+	value := cache.txList.Find(skvalue)
+	var txlist *list.List
+	if value == nil {
+		return types.ErrNotFound
+	}
+	txlist = value.Value.(*list.List)
+	txlist.Remove(item)
+	if txlist.Len() == 0 {
+		cache.txList.Delete(value)
+	}
+	return nil
+}
+
+//CreateSkipValue 创建一个 仅仅有 score 的Value
+func (cache *Queue) CreateSkipValue(item Scorer) *SkipValue {
+	skvalue := &SkipValue{Score: item.GetScore()}
+	return skvalue
+}
+
+//MaxSize 最大的cache数量
+func (cache *Queue) MaxSize() int64 {
+	return cache.maxsize
+}
+
+//Exist 是否存在
+func (cache *Queue) Exist(hash string) bool {
+	_, exists := cache.txMap[hash]
+	return exists
+}
+
+//GetItem 获取数据通过 key
+func (cache *Queue) GetItem(hash string) (Scorer, error) {
+	if k, exist := cache.txMap[hash]; exist {
+		return k.Value.(Scorer), nil
+	}
+	return nil, types.ErrNotFound
+}
+
+//Insert Scorer item to queue
+func (cache *Queue) Insert(hash string, item Scorer) {
+	cache.txMap[hash] = cache.insertSkipValue(item)
+}
+
+// Push item 到队列中，如果插入的数据优先级比队列中更大，那么弹出优先级最小的，然后插入这个数据，否则报错
+func (cache *Queue) Push(item Scorer) error {
+	hash := item.Hash()
+	if cache.Exist(string(hash)) {
+		return types.ErrTxExist
+	}
+	sv := cache.CreateSkipValue(item)
+	if int64(cache.Size()) >= cache.maxsize {
+		tail := cache.Last()
+		lasthash := string(tail.Hash())
+		cmp := sv.Compare(cache.CreateSkipValue(tail))
+		if cmp == Big || (cmp == Equal && item.Compare(tail) == Big) {
+			err := cache.Remove(lasthash)
+			if err != nil {
+				return err
+			}
+		} else {
+			return types.ErrMemFull
+		}
+	}
+	cache.Insert(string(hash), item)
+	return nil
+}
+
+// Remove 删除数据
+func (cache *Queue) Remove(hash string) error {
+	elm, ok := cache.txMap[hash]
+	if !ok {
+		return types.ErrNotFound
+	}
+	//保证txMap中先删除，这个用于计数
+	delete(cache.txMap, hash)
+	err := cache.deleteSkipValue(elm)
+	if err != nil {
+		println("queue_data_crash")
+		return err
+	}
+	return nil
+}
+
+// Size 数据总数
+func (cache *Queue) Size() int {
+	return len(cache.txMap)
+}
+
+//Last 取出最后一个交易
+func (cache *Queue) Last() Scorer {
+	if cache.Size() == 0 {
+		return nil
+	}
+	tailqueue := cache.txList.GetIterator().Last()
+	tail := tailqueue.Value.(*list.List).Back().Value.(Scorer)
+	return tail
+}
+
+//First 取出第一个交易
+func (cache *Queue) First() Scorer {
+	if cache.Size() == 0 {
+		return nil
+	}
+	tailqueue := cache.txList.GetIterator().First()
+	tail := tailqueue.Value.(*list.List).Front().Value.(Scorer)
+	return tail
+}
+
+// Walk 遍历整个队列
+func (cache *Queue) Walk(count int, cb func(value Scorer) bool) {
+	i := 0
+	cache.txList.Walk(func(item interface{}) bool {
+		l := item.(*list.List)
+		for e := l.Front(); e != nil; e = e.Next() {
+			if !cb(e.Value.(Scorer)) {
+				return false
+			}
+			i++
+			if i == count {
+				return false
+			}
+		}
+		return true
+	})
+}

vendor/github.com/33cn/chain33/common/skiplist/queue_test.go

+package skiplist
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/33cn/chain33/types"
+	"github.com/stretchr/testify/assert"
+)
+
+type scorer struct {
+	score int64
+	hash  string
+}
+
+func (item *scorer) GetScore() int64 {
+	return item.score
+}
+
+func (item *scorer) Hash() []byte {
+	return []byte(item.hash)
+}
+
+func (item *scorer) Compare(cmp Scorer) int {
+	switch item.GetScore() - cmp.GetScore() {
+	case Equal:
+		return Equal
+	case Big:
+		return Big
+	case Small:
+		return Small
+	default:
+		return 0
+	}
+}
+
+var (
+	sc1 = &scorer{1, "111"}
+	sc2 = &scorer{2, "222"}
+	sc3 = &scorer{3, "333"}
+	sc4 = &scorer{4, "444"}
+)
+
+func TestQueuePush(t *testing.T) {
+	q := NewQueue(10)
+	q.Push(sc1)
+	assert.Equal(t, 1, q.Size())
+	q.Push(sc2)
+	assert.Equal(t, 2, q.Size())
+	assert.Equal(t, sc2, q.First())
+	assert.Equal(t, sc1, q.Last())
+}
+
+func TestQueueFind(t *testing.T) {
+	q := NewQueue(10)
+	q.Push(sc1)
+	f1, _ := q.GetItem(string(sc1.Hash()))
+	assert.Equal(t, sc1, f1)
+	q.Push(sc2)
+	f2, _ := q.GetItem(string(sc2.Hash()))
+	assert.Equal(t, sc2, f2)
+	q.Push(sc3)
+	f3, _ := q.GetItem(string(sc3.Hash()))
+	assert.Equal(t, sc3, f3)
+	f4, err := q.GetItem(string(sc4.Hash()))
+	assert.Equal(t, nil, f4)
+	assert.Equal(t, types.ErrNotFound, err)
+}
+
+func TestQueueDelete(t *testing.T) {
+	q := NewQueue(10)
+	q.Push(sc1)
+	q.Push(sc2)
+	q.Push(sc3)
+	q.Remove(string(sc3.Hash()))
+	assert.Equal(t, 2, q.Size())
+	f3, err := q.GetItem(string(sc3.Hash()))
+	assert.Equal(t, nil, f3)
+	assert.Equal(t, types.ErrNotFound, err)
+}
+
+func TestQueueWalk(t *testing.T) {
+	q := NewQueue(10)
+	q.Push(sc1)
+	q.Push(sc2)
+	var data [2]string
+	i := 0
+	q.Walk(0, func(value Scorer) bool {
+		data[i] = string(value.Hash())
+		i++
+		return true
+	})
+	assert.Equal(t, data[0], "222")
+	assert.Equal(t, data[1], "111")
+
+	var data2 [2]string
+	i = 0
+	q.Walk(0, func(value Scorer) bool {
+		data2[i] = string(value.Hash())
+		i++
+		return false
+	})
+	assert.Equal(t, data2[0], "222")
+	assert.Equal(t, data2[1], "")
+
+	i = 0
+	q.Walk(0, func(value Scorer) bool {
+		data2[i] = string(value.Hash())
+		i++
+		return !(i == 2)
+	})
+	assert.Equal(t, data2[0], "222")
+	assert.Equal(t, data2[1], "111")
+}
+
+type scoreint struct {
+	data int64
+}
+
+func (s *scoreint) GetScore() int64 {
+	return s.data
+}
+
+func (s *scoreint) Hash() []byte {
+	return []byte(fmt.Sprint(s.data))
+}
+
+func (s *scoreint) Compare(b Scorer) int {
+	return Big
+}
+
+func TestQueue(t *testing.T) {
+	queue := NewQueue(10)
+	for i := 0; i < 11; i++ {
+		err := queue.Push(&scoreint{data: int64(i)})
+		assert.Nil(t, err)
+	}
+	assert.Equal(t, int64(10), queue.MaxSize())
+	err := queue.Push(&scoreint{data: int64(0)})
+	assert.Equal(t, err, types.ErrMemFull)
+	err = queue.Push(&scoreint{data: int64(1)})
+	assert.Equal(t, err, types.ErrTxExist)
+	assert.Equal(t, int64(10), queue.MaxSize())
+
+	item := &scoreint{data: int64(1)}
+	item2, err := queue.GetItem(string(item.Hash()))
+	assert.Nil(t, err)
+	assert.Equal(t, item.Hash(), item2.Hash())
+}

vendor/github.com/33cn/chain33/common/skiplist/skiplist.go

@@ -14,14 +14,21 @@ type SkipValue struct {
 	Value interface{}
 }
 
+//Compare Const
+const (
+	Big   = -1
+	Small = 1
+	Equal = 0
+)
+
 // Compare 比较函数,这样的比较排序是从大到小
 func (v *SkipValue) Compare(value *SkipValue) int {
 	if v.Score > value.Score {
-		return -1
+		return Big
 	} else if v.Score == value.Score {
-		return 0
+		return Equal
 	}
-	return 1
+	return Small
 }
 
 // skipListNode 跳跃表节点
@@ -227,6 +234,9 @@ func (sl *SkipList) Insert(value *SkipValue) int {
 
 // Delete 删除节点
 func (sl *SkipList) Delete(value *SkipValue) int {
+	if value == nil {
+		return 0
+	}
 	var update [maxLevel]*skipListNode
 	x := sl.header
 	for i := sl.level - 1; i >= 0; i-- {

vendor/github.com/33cn/chain33/common/skiplist/skiplist_test.go

@@ -39,9 +39,10 @@ func TestDelete(t *testing.T) {
 	l := NewSkipList(nil)
 	l.Insert(s1)
 	l.Insert(s2)
-	l.Delete(s1)
-	assert.Equal(t, 1, l.Len())
-	assert.Equal(t, (*SkipValue)(nil), l.Find(s1))
+	l.Insert(s3)
+	l.Delete(s3)
+	assert.Equal(t, 2, l.Len())
+	assert.Equal(t, (*SkipValue)(nil), l.Find(s3))
 	assert.Equal(t, s2, l.Find(s2))
 }
 

vendor/github.com/33cn/chain33/util/testnode/testnode.go

@@ -12,6 +12,7 @@ import (
 	"math/rand"
 	"os"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/33cn/chain33/account"
@@ -67,6 +68,7 @@ type Chain33Mock struct {
 	sub      *types.ConfigSubModule
 	datadir  string
 	lastsend []byte
+	mu       sync.Mutex
 }
 
 //GetDefaultConfig :
@@ -369,10 +371,17 @@ func (mock *Chain33Mock) SendTx(tx *types.Transaction) []byte {
 	if err != nil {
 		panic(err)
 	}
-	mock.lastsend = reply.GetMsg()
+	mock.SetLastSend(reply.GetMsg())
 	return reply.GetMsg()
 }
 
+//SetLastSend :
+func (mock *Chain33Mock) SetLastSend(hash []byte) {
+	mock.mu.Lock()
+	mock.lastsend = hash
+	mock.mu.Unlock()
+}
+
 //SendTxRPC :
 func (mock *Chain33Mock) SendTxRPC(tx *types.Transaction) []byte {
 	var txhash string