[FAB-9132] Fix data race in core/chaincode/shim

Fix data race in the shim

A data race existed between reading and setting the error value returned
from stream.Recv(). Modify the shim to avoid the race and add a bit more
alignment between the peer and shim stream processing functions.

Fix data races in MockCCComm

- Quit was causing a race between channel send and close. Resolved by
  wiring a channel to Run and closing send and receive after the done
  channel had closed.
- The keep alive go routine was racing on the keepAlive message during
  send. Wired the done channel into the keep-alive go routine.
- Some tests are setting responses while go routines are running. Work
  around this bad behavior with a lock to avoid restructuring tests.

Change-Id: I49bab82336bbbeb941b7db2d8bd279f254b4ec9f
Signed-off-by: Matthew Sykes <sykesmat@us.ibm.com>

common/mocks/peer/mockccstream.go

@@ -18,6 +18,7 @@ package peer
 
 import (
 	"fmt"
+	"sync"
 	"time"
 
 	pb "github.com/hyperledger/fabric/protos/peer"
@@ -58,8 +59,10 @@ type MockCCComm struct {
 	recvStream  chan *pb.ChaincodeMessage
 	sendStream  chan *pb.ChaincodeMessage
 	respIndex   int
+	respLock    sync.Mutex
 	respSet     *MockResponseSet
 	pong        bool
+	skipClose   bool
 }
 
 func (s *MockCCComm) SetName(newname string) {
@@ -68,9 +71,6 @@ func (s *MockCCComm) SetName(newname string) {
 
 //Send sends a message
 func (s *MockCCComm) Send(msg *pb.ChaincodeMessage) error {
-	defer func() {
-		recover()
-	}()
 	s.sendStream <- msg
 	return nil
 }
@@ -96,16 +96,11 @@ func (s *MockCCComm) GetSendStream() chan *pb.ChaincodeMessage {
 	return s.sendStream
 }
 
-//Quit closes the channels...this will also close chaincode side
+//Quit closes the channels...
 func (s *MockCCComm) Quit() {
-	if s.recvStream != nil {
+	if !s.skipClose {
 		close(s.recvStream)
-		s.recvStream = nil
-	}
-
-	if s.sendStream != nil {
 		close(s.sendStream)
-		s.sendStream = nil
 	}
 }
 
@@ -126,31 +121,32 @@ func (s *MockCCComm) SetKeepAlive(ka *pb.ChaincodeMessage) {
 
 //SetResponses sets responses for an Init or Invoke
 func (s *MockCCComm) SetResponses(respSet *MockResponseSet) {
+	s.respLock.Lock()
 	s.respSet = respSet
 	s.respIndex = 0
+	s.respLock.Unlock()
 }
 
 //keepAlive
-func (s *MockCCComm) ka() {
-	defer recover()
+func (s *MockCCComm) ka(done <-chan struct{}) {
 	for {
 		if s.keepAlive == nil {
 			return
 		}
 		s.Send(s.keepAlive)
-		time.Sleep(10 * time.Millisecond)
+		select {
+		case <-time.After(10 * time.Millisecond):
+		case <-done:
+			return
+		}
 	}
 }
 
 //Run receives and sends indefinitely
-func (s *MockCCComm) Run() error {
+func (s *MockCCComm) Run(done <-chan struct{}) error {
 	//start the keepalive
-	go s.ka()
-
-	//if we started keep alive this will kill it
-	defer func() {
-		s.keepAlive = nil
-	}()
+	go s.ka(done)
+	defer s.Quit()
 
 	for {
 		msg, err := s.Recv()
@@ -181,6 +177,9 @@ func (s *MockCCComm) respond(msg *pb.ChaincodeMessage) error {
 		return nil
 	}
 
+	s.respLock.Lock()
+	defer s.respLock.Unlock()
+
 	var err error
 	if s.respIndex < len(s.respSet.Responses) {
 		mockResp := s.respSet.Responses[s.respIndex]

common/mocks/peer/mockpeerccsupport.go

@@ -58,7 +58,7 @@ func (mp *MockPeerCCSupport) GetCCMirror(name string) *MockCCComm {
 		return nil
 	}
 
-	return &MockCCComm{name: name, recvStream: s.sendStream, sendStream: s.recvStream}
+	return &MockCCComm{name: name, recvStream: s.sendStream, sendStream: s.recvStream, skipClose: true}
 }
 
 //RemoveCC removes a cc

core/chaincode/chaincode_support_test.go

@@ -335,6 +335,9 @@ func startCC(t *testing.T, channelID string, ccname string, chaincodeSupport *Ch
 		done <- err
 	}
 
+	ccDone := make(chan struct{})
+	defer close(ccDone)
+
 	//start the mock peer
 	go func() {
 		respSet := &mockpeer.MockResponseSet{
@@ -346,7 +349,7 @@ func startCC(t *testing.T, channelID string, ccname string, chaincodeSupport *Ch
 			},
 		}
 		ccSide.SetResponses(respSet)
-		ccSide.Run()
+		ccSide.Run(ccDone)
 	}()
 
 	ccSide.Send(&pb.ChaincodeMessage{Type: pb.ChaincodeMessage_REGISTER, Payload: putils.MarshalOrPanic(&pb.ChaincodeID{Name: ccname + ":0"}), Txid: "0", ChannelId: channelID})

core/chaincode/handler.go

@@ -364,7 +364,7 @@ func (h *Handler) ProcessStream(stream ccintf.ChaincodeStream) error {
 		msg *pb.ChaincodeMessage
 		err error
 	}
-	msgAvail := make(chan *recvMsg)
+	msgAvail := make(chan *recvMsg, 1)
 
 	receiveMessage := func() {
 		in, err := h.chatStream.Recv()
@@ -374,32 +374,31 @@ func (h *Handler) ProcessStream(stream ccintf.ChaincodeStream) error {
 	go receiveMessage()
 	for {
 		select {
-		case rMsg := <-msgAvail:
+		case rmsg := <-msgAvail:
+			switch {
 			// Defer the deregistering of the this handler.
-			if rMsg.err == io.EOF {
-				chaincodeLogger.Debugf("received EOF, ending chaincode support stream: %s", rMsg.err)
-				return rMsg.err
-			} else if rMsg.err != nil {
-				err := errors.Wrap(rMsg.err, "receive failed")
+			case rmsg.err == io.EOF:
+				chaincodeLogger.Debugf("received EOF, ending chaincode support stream: %s", rmsg.err)
+				return rmsg.err
+			case rmsg.err != nil:
+				err := errors.Wrap(rmsg.err, "receive failed")
 				chaincodeLogger.Errorf("handling chaincode support stream: %+v", err)
 				return err
-			} else if rMsg.msg == nil {
+			case rmsg.msg == nil:
 				err := errors.New("received nil message, ending chaincode support stream")
 				chaincodeLogger.Debugf("%+v", err)
 				return err
+			default:
+				err := h.handleMessage(rmsg.msg)
+				if err != nil {
+					err = errors.WithMessage(err, "error handling message, ending stream")
+					chaincodeLogger.Errorf("[%s] %+v", shorttxid(rmsg.msg.Txid), err)
+					return err
+				}
+
+				go receiveMessage()
 			}
 
-			in := rMsg.msg
-
-			err := h.handleMessage(in)
-			if err != nil {
-				err = errors.WithMessage(err, "error handling message, ending stream")
-				chaincodeLogger.Errorf("[%s] %+v", shorttxid(in.Txid), err)
-				return err
-			}
-
-			go receiveMessage()
-
 		case sendErr := <-h.errChan:
 			err := errors.Wrapf(sendErr, "received error while sending message, ending chaincode support stream")
 			chaincodeLogger.Errorf("%s", err)

core/chaincode/shim/chaincode.go

@@ -262,83 +262,71 @@ func newPeerClientConnection() (*grpc.ClientConn, error) {
 }
 
 func chatWithPeer(chaincodename string, stream PeerChaincodeStream, cc Chaincode) error {
-
 	// Create the shim handler responsible for all control logic
 	handler := newChaincodeHandler(stream, cc)
-
 	defer stream.CloseSend()
+
 	// Send the ChaincodeID during register.
 	chaincodeID := &pb.ChaincodeID{Name: chaincodename}
 	payload, err := proto.Marshal(chaincodeID)
 	if err != nil {
 		return errors.Wrap(err, "error marshalling chaincodeID during chaincode registration")
 	}
+
 	// Register on the stream
 	chaincodeLogger.Debugf("Registering.. sending %s", pb.ChaincodeMessage_REGISTER)
 	if err = handler.serialSend(&pb.ChaincodeMessage{Type: pb.ChaincodeMessage_REGISTER, Payload: payload}); err != nil {
 		return errors.WithMessage(err, "error sending chaincode REGISTER")
 	}
-	waitc := make(chan struct{})
+
+	// holds return values from gRPC Recv below
+	type recvMsg struct {
+		msg *pb.ChaincodeMessage
+		err error
+	}
+	msgAvail := make(chan *recvMsg, 1)
 	errc := make(chan error)
-	go func() {
-		defer close(waitc)
-		msgAvail := make(chan *pb.ChaincodeMessage)
-		var in *pb.ChaincodeMessage
-		recv := true
-		for {
-			in = nil
-			err = nil
-			if recv {
-				recv = false
-				go func() {
-					var in2 *pb.ChaincodeMessage
-					in2, err = stream.Recv()
-					errc <- err
-					msgAvail <- in2
-				}()
-			}
-			select {
-			case sendErr := <-errc:
-				//serialSendAsync successful?
-				if sendErr == nil {
-					continue
-				}
-				//no, bail
-				err = errors.Wrap(sendErr, "error sending")
-				return
-			case in = <-msgAvail:
-				if err == io.EOF {
-					err = errors.Wrapf(err, "received EOF, ending chaincode stream")
-					chaincodeLogger.Debugf("%+v", err)
-					return
-				} else if err != nil {
-					chaincodeLogger.Errorf("Received error from server, ending chaincode stream: %+v", err)
-					return
-				} else if in == nil {
-					err = errors.New("received nil message, ending chaincode stream")
-					chaincodeLogger.Debugf("%+v", err)
-					return
+
+	receiveMessage := func() {
+		in, err := stream.Recv()
+		msgAvail <- &recvMsg{in, err}
+	}
+
+	go receiveMessage()
+	for {
+		select {
+		case rmsg := <-msgAvail:
+			switch {
+			case rmsg.err == io.EOF:
+				err = errors.Wrapf(rmsg.err, "received EOF, ending chaincode stream")
+				chaincodeLogger.Debugf("%+v", err)
+				return err
+			case rmsg.err != nil:
+				err := errors.Wrap(rmsg.err, "receive failed")
+				chaincodeLogger.Errorf("Received error from server, ending chaincode stream: %+v", err)
+				return err
+			case rmsg.msg == nil:
+				err := errors.New("received nil message, ending chaincode stream")
+				chaincodeLogger.Debugf("%+v", err)
+				return err
+			default:
+				chaincodeLogger.Debugf("[%s]Received message %s from peer", shorttxid(rmsg.msg.Txid), rmsg.msg.Type)
+				err := handler.handleMessage(rmsg.msg, errc)
+				if err != nil {
+					err = errors.WithMessage(err, "error handling message")
+					return err
 				}
-				chaincodeLogger.Debugf("[%s]Received message %s from peer", shorttxid(in.Txid), in.Type)
-				recv = true
-			}
 
-			err = handler.handleMessage(in, errc)
-			if err != nil {
-				err = errors.WithMessage(err, "error handling message")
-				return
+				go receiveMessage()
 			}
 
-			//keepalive messages are PONGs to the fabric's PINGs
-			if in.Type == pb.ChaincodeMessage_KEEPALIVE {
-				chaincodeLogger.Debug("Sending KEEPALIVE response")
-				//ignore any errors, maybe next KEEPALIVE will work
-				handler.serialSendAsync(in, nil)
+		case sendErr := <-errc:
+			if sendErr != nil {
+				err := errors.Wrap(sendErr, "error sending")
+				return err
 			}
 		}
-	}()
-	<-waitc
-	return err
+	}
 }
 
 // -- init stub ---

core/chaincode/shim/handler.go

@@ -728,7 +728,8 @@ func (handler *Handler) handleCreated(msg *pb.ChaincodeMessage, errc chan error)
 // handleMessage message handles loop for shim side of chaincode/peer stream.
 func (handler *Handler) handleMessage(msg *pb.ChaincodeMessage, errc chan error) error {
 	if msg.Type == pb.ChaincodeMessage_KEEPALIVE {
-		// Received a keep alive message, we don't do anything with it for now
+		chaincodeLogger.Debug("Sending KEEPALIVE response")
+		handler.serialSendAsync(msg, nil) // ignore errors, maybe next KEEPALIVE will work
 		return nil
 	}
 	chaincodeLogger.Debugf("[%s] Handling ChaincodeMessage of type: %s(state:%s)", shorttxid(msg.Txid), msg.Type, handler.state)