update chain33

vendor/github.com/33cn/chain33/client/mock_system_test.go

@@ -71,6 +71,10 @@ func (mock *mockClient) Wait(msg *queue.Message) (*queue.Message, error) {
 	return mock.c.Wait(msg)
 }
 
+func (mock *mockClient) Reply(msg *queue.Message) {
+	mock.c.Reply(msg)
+}
+
 func (mock *mockClient) WaitTimeout(msg *queue.Message, timeout time.Duration) (*queue.Message, error) {
 	return mock.c.WaitTimeout(msg, timeout)
 }

vendor/github.com/33cn/chain33/queue/client.go

@@ -34,6 +34,7 @@ type Client interface {
 	Wait(msg *Message) (*Message, error)                               //等待消息处理完成
 	WaitTimeout(msg *Message, timeout time.Duration) (*Message, error) //等待消息处理完成
 	Recv() chan *Message
+	Reply(msg *Message)
 	Sub(topic string) //订阅消息
 	Close()
 	CloseQueue() (*types.Reply, error)
@@ -101,6 +102,16 @@ func (client *client) NewMessage(topic string, ty int64, data interface{}) (msg
 	return NewMessage(id, topic, ty, data)
 }
 
+func (client *client) Reply(msg *Message) {
+	if msg.chReply != nil {
+		msg.Reply(msg)
+		return
+	}
+	if msg.callback != nil {
+		client.q.callback <- msg
+	}
+}
+
 // WaitTimeout 等待时间 msg 消息 timeout 超时时间
 func (client *client) WaitTimeout(msg *Message, timeout time.Duration) (*Message, error) {
 	if msg.chReply == nil {
@@ -183,10 +194,10 @@ func (client *client) isEnd(data *Message, ok bool) bool {
 	if !ok {
 		return true
 	}
-	if atomic.LoadInt32(&client.isClosed) == 1 {
+	if data.Data == nil && data.ID == 0 && data.Ty == 0 {
 		return true
 	}
-	if data.Data == nil && data.ID == 0 && data.Ty == 0 {
+	if atomic.LoadInt32(&client.isClosed) == 1 {
 		return true
 	}
 	return false

vendor/github.com/33cn/chain33/queue/mocks/client.go

@@ -72,6 +72,11 @@ func (_m *Client) Recv() chan *queue.Message {
 	return r0
 }
 
+// Reply provides a mock function with given fields: msg
+func (_m *Client) Reply(msg *queue.Message) {
+	_m.Called(msg)
+}
+
 // Send provides a mock function with given fields: msg, waitReply
 func (_m *Client) Send(msg *queue.Message, waitReply bool) error {
 	ret := _m.Called(msg, waitReply)

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

@@ -66,13 +66,33 @@ type queue struct {
 	mu        sync.Mutex
 	done      chan struct{}
 	interrupt chan struct{}
+	callback  chan *Message
 	isClose   int32
 	name      string
 }
 
 // New new queue struct
 func New(name string) Queue {
-	q := &queue{chanSubs: make(map[string]*chanSub), name: name, done: make(chan struct{}, 1), interrupt: make(chan struct{}, 1)}
+	q := &queue{
+		chanSubs:  make(map[string]*chanSub),
+		name:      name,
+		done:      make(chan struct{}, 1),
+		interrupt: make(chan struct{}, 1),
+		callback:  make(chan *Message, 1024),
+	}
+	go func() {
+		for {
+			select {
+			case <-q.done:
+				fmt.Println("closing chain33 callback")
+				return
+			case msg := <-q.callback:
+				if msg.callback != nil {
+					msg.callback(msg)
+				}
+			}
+		}
+	}()
 	return q
 }
 
@@ -88,7 +108,8 @@ func (q *queue) Start() {
 	// Block until a signal is received.
 	select {
 	case <-q.done:
-		atomic.StoreInt32(&q.isClose, 1)
+		fmt.Println("closing chain33 done")
+		//atomic.StoreInt32(&q.isClose, 1)
 		break
 	case <-q.interrupt:
 		fmt.Println("closing chain33")
@@ -130,7 +151,11 @@ func (q *queue) chanSub(topic string) *chanSub {
 	defer q.mu.Unlock()
 	_, ok := q.chanSubs[topic]
 	if !ok {
-		q.chanSubs[topic] = &chanSub{make(chan *Message, defaultChanBuffer), make(chan *Message, defaultLowChanBuffer), 0}
+		q.chanSubs[topic] = &chanSub{
+			high:    make(chan *Message, defaultChanBuffer),
+			low:     make(chan *Message, defaultLowChanBuffer),
+			isClose: 0,
+		}
 	}
 	return q.chanSubs[topic]
 }
@@ -157,6 +182,10 @@ func (q *queue) send(msg *Message, timeout time.Duration) (err error) {
 	if sub.isClose == 1 {
 		return types.ErrChannelClosed
 	}
+	if timeout == -1 {
+		sub.high <- msg
+		return nil
+	}
 	defer func() {
 		res := recover()
 		if res != nil {
@@ -172,10 +201,6 @@ func (q *queue) send(msg *Message, timeout time.Duration) (err error) {
 			return ErrQueueChannelFull
 		}
 	}
-	if timeout == -1 {
-		sub.high <- msg
-		return nil
-	}
 	t := time.NewTimer(timeout)
 	defer t.Stop()
 	select {
@@ -212,13 +237,13 @@ func (q *queue) sendLowTimeout(msg *Message, timeout time.Duration) error {
 	if sub.isClose == 1 {
 		return types.ErrChannelClosed
 	}
-	if timeout == 0 {
-		return q.sendAsyn(msg)
-	}
 	if timeout == -1 {
 		sub.low <- msg
 		return nil
 	}
+	if timeout == 0 {
+		return q.sendAsyn(msg)
+	}
 	t := time.NewTimer(timeout)
 	defer t.Stop()
 	select {
@@ -237,11 +262,12 @@ func (q *queue) Client() Client {
 
 // Message message struct
 type Message struct {
-	Topic   string
-	Ty      int64
-	ID      int64
-	Data    interface{}
-	chReply chan *Message
+	Topic    string
+	Ty       int64
+	ID       int64
+	Data     interface{}
+	chReply  chan *Message
+	callback func(msg *Message)
 }
 
 // NewMessage new message
@@ -255,6 +281,17 @@ func NewMessage(id int64, topic string, ty int64, data interface{}) (msg *Messag
 	return msg
 }
 
+// NewMessageCallback reply block
+func NewMessageCallback(id int64, topic string, ty int64, data interface{}, callback func(msg *Message)) (msg *Message) {
+	msg = &Message{}
+	msg.ID = id
+	msg.Ty = ty
+	msg.Data = data
+	msg.Topic = topic
+	msg.callback = callback
+	return msg
+}
+
 // GetData get message data
 func (msg *Message) GetData() interface{} {
 	if _, ok := msg.Data.(error); ok {

vendor/github.com/33cn/chain33/queue/queue_test.go

@@ -5,6 +5,7 @@
 package queue
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
@@ -132,6 +133,7 @@ func TestHighLow(t *testing.T) {
 			msg := client.NewMessage("mempool", types.EventTx, "hello")
 			err := client.SendTimeout(msg, false, 0)
 			if err != nil {
+				fmt.Println(err)
 				break
 			}
 		}
@@ -211,24 +213,50 @@ func TestPrintMessage(t *testing.T) {
 	t.Log(msg)
 }
 
+func TestChanSubCallback(t *testing.T) {
+	q := New("channel")
+	client := q.Client()
+	client.Sub("hello")
+	done := make(chan struct{}, 1025)
+	go func() {
+		for i := 0; i < 1025; i++ {
+			sub := q.(*queue).chanSub("hello")
+			msg := NewMessageCallback(1, "", 0, nil, func(msg *Message) {
+				done <- struct{}{}
+			})
+			sub.high <- msg
+		}
+	}()
+	for i := 0; i < 1025; i++ {
+		msg := <-client.Recv()
+		client.Reply(msg)
+	}
+	for i := 0; i < 1025; i++ {
+		<-done
+	}
+}
+
 func BenchmarkSendMessage(b *testing.B) {
 	q := New("channel")
 	//mempool
+	b.ReportAllocs()
 	go func() {
 		client := q.Client()
 		client.Sub("mempool")
 		defer client.Close()
 		for msg := range client.Recv() {
-			if msg.Ty == types.EventTx {
-				msg.Reply(client.NewMessage("mempool", types.EventReply, types.Reply{IsOk: true, Msg: []byte("word")}))
-			}
+			go func(msg *Message) {
+				if msg.Ty == types.EventTx {
+					msg.Reply(client.NewMessage("mempool", types.EventReply, types.Reply{IsOk: true, Msg: []byte("word")}))
+				}
+			}(msg)
 		}
 	}()
 	go q.Start()
 	client := q.Client()
 	//high 优先级
+	msg := client.NewMessage("mempool", types.EventTx, "hello")
 	for i := 0; i < b.N; i++ {
-		msg := client.NewMessage("mempool", types.EventTx, "hello")
 		err := client.Send(msg, true)
 		if err != nil {
 			b.Error(err)
@@ -277,3 +305,121 @@ func BenchmarkIntChan(b *testing.B) {
 		ch <- 1
 	}
 }
+
+func BenchmarkChanSub(b *testing.B) {
+	q := New("channel")
+	done := make(chan struct{})
+	go func() {
+		for i := 0; i < b.N; i++ {
+			q.(*queue).chanSub("hello")
+		}
+		done <- struct{}{}
+	}()
+	for i := 0; i < b.N; i++ {
+		q.(*queue).chanSub("hello")
+	}
+	<-done
+}
+
+func BenchmarkChanSub2(b *testing.B) {
+	q := New("channel")
+	client := q.Client()
+	client.Sub("hello")
+
+	go func() {
+		for i := 0; i < b.N; i++ {
+			sub := q.(*queue).chanSub("hello")
+			msg := NewMessage(1, "", 0, nil)
+			sub.high <- msg
+			_, err := client.Wait(msg)
+			if err != nil {
+				b.Fatal(err)
+			}
+		}
+	}()
+	for i := 0; i < b.N; i++ {
+		msg := <-client.Recv()
+		msg.Reply(msg)
+	}
+}
+
+func BenchmarkChanSub3(b *testing.B) {
+	q := New("channel")
+	client := q.Client()
+	client.Sub("hello")
+
+	go func() {
+		for i := 0; i < b.N; i++ {
+			sub := q.(*queue).chanSub("hello")
+			msg := NewMessage(1, "", 0, nil)
+			sub.high <- msg
+		}
+	}()
+	for i := 0; i < b.N; i++ {
+		msg := <-client.Recv()
+		msg.Reply(msg)
+	}
+}
+
+func BenchmarkChanSub4(b *testing.B) {
+	q := New("channel")
+	client := q.Client()
+	client.Sub("hello")
+
+	go func() {
+		for i := 0; i < b.N; i++ {
+			sub := q.(*queue).chanSub("hello")
+			msg := &Message{ID: 1}
+			sub.high <- msg
+		}
+	}()
+	for i := 0; i < b.N; i++ {
+		<-client.Recv()
+	}
+}
+
+func BenchmarkChanSubCallback(b *testing.B) {
+	q := New("channel")
+	client := q.Client()
+	client.Sub("hello")
+	done := make(chan struct{}, 1024)
+	go func() {
+		for i := 0; i < b.N; i++ {
+			sub := q.(*queue).chanSub("hello")
+			msg := NewMessageCallback(1, "", 0, nil, func(msg *Message) {
+				done <- struct{}{}
+			})
+			sub.high <- msg
+		}
+	}()
+	go func() {
+		for i := 0; i < b.N; i++ {
+			msg := <-client.Recv()
+			client.Reply(msg)
+		}
+	}()
+	for i := 0; i < b.N; i++ {
+		<-done
+	}
+}
+
+func BenchmarkChanSubCallback2(b *testing.B) {
+	q := New("channel")
+	client := q.Client()
+	client.Sub("hello")
+	go func() {
+		for i := 0; i < b.N; i++ {
+			sub := q.(*queue).chanSub("hello")
+			done := make(chan struct{}, 1)
+			msg := NewMessageCallback(1, "", 0, nil, func(msg *Message) {
+				done <- struct{}{}
+			})
+			sub.high <- msg
+			<-done
+		}
+	}()
+	for i := 0; i < b.N; i++ {
+		msg := <-client.Recv()
+		client.Reply(msg)
+	}
+}