[FAB-1298] Remove queueing from broadcast

https://jira.hyperledger.org/browse/FAB-1298

Per some testing from Bishop Brock, the underlying HTTP2 stream for gRPC
seems to handle windowing on its own gracefully and likely more
efficiently (and certainly more simply) than we are able to do it at a
higher layer.

This changeset removes the queueing concept from Broadcast, and pushes
the flow control back into the HTTP2/gRPC layer.  This simplifies the
code significantly, and appears to give us a performance improvement, so
this seems like a win across the board.

Change-Id: I4f519a188f0138d5b6fa8abf2432bf5e4a59342d
Signed-off-by: Jason Yellick <[email protected]>

Author: Jason Yellick

orderer/common/broadcast/broadcast.go

@@ -58,65 +58,19 @@ type Support interface {
 }
 
 type handlerImpl struct {
-	queueSize int
-	sm        SupportManager
+	sm SupportManager
 }
 
 // NewHandlerImpl constructs a new implementation of the Handler interface
-func NewHandlerImpl(sm SupportManager, queueSize int) Handler {
+func NewHandlerImpl(sm SupportManager) Handler {
 	return &handlerImpl{
-		queueSize: queueSize,
-		sm:        sm,
+		sm: sm,
 	}
 }
 
 // Handle starts a service thread for a given gRPC connection and services the broadcast connection
 func (bh *handlerImpl) Handle(srv ab.AtomicBroadcast_BroadcastServer) error {
-	b := newBroadcaster(bh)
-	defer close(b.queue)
-	go b.drainQueue()
-	return b.queueEnvelopes(srv)
-}
-
-type msgAndSupport struct {
-	msg     *cb.Envelope
-	support Support
-}
-
-type broadcaster struct {
-	bs       *handlerImpl
-	queue    chan *msgAndSupport
-	exitChan chan struct{}
-}
-
-func newBroadcaster(bs *handlerImpl) *broadcaster {
-	b := &broadcaster{
-		bs:       bs,
-		queue:    make(chan *msgAndSupport, bs.queueSize),
-		exitChan: make(chan struct{}),
-	}
-	return b
-}
-
-func (b *broadcaster) drainQueue() {
-	defer close(b.exitChan)
-	for msgAndSupport := range b.queue {
-		if !msgAndSupport.support.Enqueue(msgAndSupport.msg) {
-			logger.Debugf("Consenter instructed us to shut down")
-			return
-		}
-	}
-	logger.Debugf("Exiting because the queue channel closed")
-}
-
-func (b *broadcaster) queueEnvelopes(srv ab.AtomicBroadcast_BroadcastServer) error {
-
 	for {
-		select {
-		case <-b.exitChan:
-			return nil
-		default:
-		}
 		msg, err := srv.Recv()
 		if err != nil {
 			return err
@@ -129,32 +83,36 @@ func (b *broadcaster) queueEnvelopes(srv ab.AtomicBroadcast_BroadcastServer) err
 			return srv.Send(&ab.BroadcastResponse{Status: cb.Status_BAD_REQUEST})
 		}
 
-		support, ok := b.bs.sm.GetChain(payload.Header.ChainHeader.ChainID)
+		support, ok := bh.sm.GetChain(payload.Header.ChainHeader.ChainID)
 		if !ok {
 			// Chain not found, maybe create one?
 			if payload.Header.ChainHeader.Type != int32(cb.HeaderType_CONFIGURATION_TRANSACTION) {
-				err = srv.Send(&ab.BroadcastResponse{Status: cb.Status_NOT_FOUND})
-			} else {
-				logger.Debugf("Proposing new chain")
-				err = srv.Send(&ab.BroadcastResponse{Status: b.bs.sm.ProposeChain(msg)})
+				return srv.Send(&ab.BroadcastResponse{Status: cb.Status_NOT_FOUND})
 			}
-		} else {
-			// Normal transaction for existing chain
-			_, filterErr := support.Filters().Apply(msg)
-
-			if filterErr != nil {
-				logger.Debugf("Rejecting broadcast message")
-				err = srv.Send(&ab.BroadcastResponse{Status: cb.Status_BAD_REQUEST})
-			} else {
-				select {
-				case b.queue <- &msgAndSupport{msg: msg, support: support}:
-					err = srv.Send(&ab.BroadcastResponse{Status: cb.Status_SUCCESS})
-				default:
-					err = srv.Send(&ab.BroadcastResponse{Status: cb.Status_SERVICE_UNAVAILABLE})
-				}
+
+			logger.Debugf("Proposing new chain")
+			err = srv.Send(&ab.BroadcastResponse{Status: bh.sm.ProposeChain(msg)})
+			if err != nil {
+				return err
 			}
+			continue
+		}
+
+		// Normal transaction for existing chain
+		_, filterErr := support.Filters().Apply(msg)
+
+		if filterErr != nil {
+			logger.Debugf("Rejecting broadcast message")
+			return srv.Send(&ab.BroadcastResponse{Status: cb.Status_BAD_REQUEST})
+		}
+
+		if !support.Enqueue(msg) {
+			logger.Debugf("Consenter instructed us to shut down")
+			return srv.Send(&ab.BroadcastResponse{Status: cb.Status_SERVICE_UNAVAILABLE})
 		}
 
+		err = srv.Send(&ab.BroadcastResponse{Status: cb.Status_SUCCESS})
+
 		if err != nil {
 			return err
 		}

orderer/common/broadcast/broadcast_test.go

@@ -19,6 +19,7 @@ package broadcast
 import (
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/hyperledger/fabric/orderer/common/filter"
 	cb "github.com/hyperledger/fabric/protos/common"
@@ -73,21 +74,13 @@ func (mm *mockSupportManager) ProposeChain(configTx *cb.Envelope) cb.Status {
 			filter.EmptyRejectRule,
 			filter.AcceptRule,
 		}),
-		queue: make(chan *cb.Envelope),
 	}
 	return cb.Status_SUCCESS
 }
 
-func (mm *mockSupportManager) halt() {
-	for _, chain := range mm.chains {
-		chain.halt()
-	}
-}
-
 type mockSupport struct {
-	filters *filter.RuleSet
-	queue   chan *cb.Envelope
-	done    bool
+	filters       *filter.RuleSet
+	rejectEnqueue bool
 }
 
 func (ms *mockSupport) Filters() *filter.RuleSet {
@@ -96,16 +89,7 @@ func (ms *mockSupport) Filters() *filter.RuleSet {
 
 // Enqueue sends a message for ordering
 func (ms *mockSupport) Enqueue(env *cb.Envelope) bool {
-	ms.queue <- env
-	return !ms.done
-}
-
-func (ms *mockSupport) halt() {
-	ms.done = true
-	select {
-	case <-ms.queue:
-	default:
-	}
+	return !ms.rejectEnqueue
 }
 
 func makeConfigMessage(chainID string) *cb.Envelope {
@@ -137,29 +121,31 @@ func makeMessage(chainID string, data []byte) *cb.Envelope {
 	}
 }
 
-func getMultichainManager() *mockSupportManager {
+func getMockSupportManager() (*mockSupportManager, *mockSupport) {
 	filters := filter.NewRuleSet([]filter.Rule{
 		filter.EmptyRejectRule,
 		filter.AcceptRule,
 	})
 	mm := &mockSupportManager{
 		chains: make(map[string]*mockSupport),
 	}
-	mm.chains[string(systemChain)] = &mockSupport{
+	mSysChain := &mockSupport{
 		filters: filters,
-		queue:   make(chan *cb.Envelope),
 	}
-	return mm
+	mm.chains[string(systemChain)] = mSysChain
+	return mm, mSysChain
 }
 
-func TestQueueOverflow(t *testing.T) {
-	mm := getMultichainManager()
-	defer mm.halt()
-	bh := NewHandlerImpl(mm, 2)
+func TestEnqueueFailure(t *testing.T) {
+	mm, mSysChain := getMockSupportManager()
+	bh := NewHandlerImpl(mm)
 	m := newMockB()
 	defer close(m.recvChan)
-	b := newBroadcaster(bh.(*handlerImpl))
-	go b.queueEnvelopes(m)
+	done := make(chan struct{})
+	go func() {
+		bh.Handle(m)
+		close(done)
+	}()
 
 	for i := 0; i < 2; i++ {
 		m.recvChan <- makeMessage(systemChain, []byte("Some bytes"))
@@ -169,86 +155,77 @@ func TestQueueOverflow(t *testing.T) {
 		}
 	}
 
+	mSysChain.rejectEnqueue = true
 	m.recvChan <- makeMessage(systemChain, []byte("Some bytes"))
 	reply := <-m.sendChan
 	if reply.Status != cb.Status_SERVICE_UNAVAILABLE {
 		t.Fatalf("Should not have successfully queued the message")
 	}
 
-}
-
-func TestMultiQueueOverflow(t *testing.T) {
-	mm := getMultichainManager()
-	defer mm.halt()
-	bh := NewHandlerImpl(mm, 2)
-	ms := []*mockB{newMockB(), newMockB(), newMockB()}
-
-	for _, m := range ms {
-		defer close(m.recvChan)
-		b := newBroadcaster(bh.(*handlerImpl))
-		go b.queueEnvelopes(m)
-	}
-
-	for _, m := range ms {
-		for i := 0; i < 2; i++ {
-			m.recvChan <- makeMessage(systemChain, []byte("Some bytes"))
-			reply := <-m.sendChan
-			if reply.Status != cb.Status_SUCCESS {
-				t.Fatalf("Should have successfully queued the message")
-			}
-		}
-	}
-
-	for _, m := range ms {
-		m.recvChan <- makeMessage(systemChain, []byte("Some bytes"))
-		reply := <-m.sendChan
-		if reply.Status != cb.Status_SERVICE_UNAVAILABLE {
-			t.Fatalf("Should not have successfully queued the message")
-		}
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatalf("Should have terminated the stream")
 	}
 }
 
 func TestEmptyEnvelope(t *testing.T) {
-	mm := getMultichainManager()
-	defer mm.halt()
-	bh := NewHandlerImpl(mm, 2)
+	mm, _ := getMockSupportManager()
+	bh := NewHandlerImpl(mm)
 	m := newMockB()
 	defer close(m.recvChan)
-	go bh.Handle(m)
+	done := make(chan struct{})
+	go func() {
+		bh.Handle(m)
+		close(done)
+	}()
 
 	m.recvChan <- &cb.Envelope{}
 	reply := <-m.sendChan
 	if reply.Status != cb.Status_BAD_REQUEST {
 		t.Fatalf("Should have rejected the null message")
 	}
 
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatalf("Should have terminated the stream")
+	}
 }
 
 func TestBadChainID(t *testing.T) {
-	mm := getMultichainManager()
-	defer mm.halt()
-	bh := NewHandlerImpl(mm, 2)
+	mm, _ := getMockSupportManager()
+	bh := NewHandlerImpl(mm)
 	m := newMockB()
 	defer close(m.recvChan)
-	go bh.Handle(m)
+	done := make(chan struct{})
+	go func() {
+		bh.Handle(m)
+		close(done)
+	}()
 
 	m.recvChan <- makeMessage("Wrong chain", []byte("Some bytes"))
 	reply := <-m.sendChan
 	if reply.Status != cb.Status_NOT_FOUND {
 		t.Fatalf("Should have rejected message to a chain which does not exist")
 	}
 
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatalf("Should have terminated the stream")
+	}
 }
 
 func TestNewChainID(t *testing.T) {
-	mm := getMultichainManager()
-	defer mm.halt()
-	bh := NewHandlerImpl(mm, 2)
+	mm, _ := getMockSupportManager()
+	bh := NewHandlerImpl(mm)
 	m := newMockB()
 	defer close(m.recvChan)
 	go bh.Handle(m)
+	newChainID := "New Chain"
 
-	m.recvChan <- makeConfigMessage("New chain")
+	m.recvChan <- makeConfigMessage(newChainID)
 	reply := <-m.sendChan
 	if reply.Status != cb.Status_SUCCESS {
 		t.Fatalf("Should have created a new chain, got %d", reply.Status)
@@ -258,9 +235,9 @@ func TestNewChainID(t *testing.T) {
 		t.Fatalf("Should have created a new chain")
 	}
 
-	m.recvChan <- makeMessage("New chain", []byte("Some bytes"))
+	m.recvChan <- makeMessage(newChainID, []byte("Some bytes"))
 	reply = <-m.sendChan
 	if reply.Status != cb.Status_SUCCESS {
-		t.Fatalf("Should have successfully sent message to new chain")
+		t.Fatalf("Should have successfully sent message to new chain, got %v", reply)
 	}
 }

orderer/server.go

@@ -49,8 +49,8 @@ func NewServer(ml multichain.Manager, queueSize, maxWindowSize int) ab.AtomicBro
 	logger.Infof("Starting orderer")
 
 	s := &server{
-		bh: broadcast.NewHandlerImpl(broadcastSupport{ml}, queueSize),
 		dh: deliver.NewHandlerImpl(deliverSupport{ml}),
+		bh: broadcast.NewHandlerImpl(broadcastSupport{ml}),
 	}
 	return s
 }