[FAB-15897] Improve CLI perf w/ multiple endorsers

If you specify multiple peers to "peer chaincode invoke", currently
these endorsements are run one at a time, in order specified on the CLI.

This is slow, which might not be seen as a problem for the CLI, however
in the current version of chaincode lifecycle this command is used to
trigger the building of chaincode containers. This may be causing CI failures
when chaincode needs to be built in a single transaction across multiple
peers.

This change runs the endorsements concurrently, which should trigger
the chaincode to be built across all peers simultaneously. Also, this is
how the various SDKs behave today.

Signed-off-by: Simon Stone <sstone1@uk.ibm.com>
Change-Id: Ib9f69c7bbd7f4f1197f2959599e45d0a2545d366

Author: Simon Stone

internal/peer/chaincode/common.go

@@ -446,6 +446,36 @@ func InitCmdFactory(cmdName string, isEndorserRequired, isOrdererRequired bool)
 	}, nil
 }
 
+// processProposals sends a signed proposal to a set of peers, and gathers all the responses.
+func processProposals(endorserClients []pb.EndorserClient, signedProposal *pb.SignedProposal) ([]*pb.ProposalResponse, error) {
+	responsesCh := make(chan *pb.ProposalResponse, len(endorserClients))
+	errorCh := make(chan error, len(endorserClients))
+	wg := sync.WaitGroup{}
+	for _, endorser := range endorserClients {
+		wg.Add(1)
+		go func(endorser pb.EndorserClient) {
+			defer wg.Done()
+			proposalResp, err := endorser.ProcessProposal(context.Background(), signedProposal)
+			if err != nil {
+				errorCh <- err
+				return
+			}
+			responsesCh <- proposalResp
+		}(endorser)
+	}
+	wg.Wait()
+	close(responsesCh)
+	close(errorCh)
+	for err := range errorCh {
+		return nil, err
+	}
+	var responses []*pb.ProposalResponse
+	for response := range responsesCh {
+		responses = append(responses, response)
+	}
+	return responses, nil
+}
+
 // ChaincodeInvokeOrQuery invokes or queries the chaincode. If successful, the
 // INVOKE form prints the ProposalResponse to STDOUT, and the QUERY form prints
 // the query result on STDOUT. A command-line flag (-r, --raw) determines
@@ -496,13 +526,10 @@ func ChaincodeInvokeOrQuery(
 	if err != nil {
 		return nil, errors.WithMessagef(err, "error creating signed proposal for %s", funcName)
 	}
-	var responses []*pb.ProposalResponse
-	for _, endorser := range endorserClients {
-		proposalResp, err := endorser.ProcessProposal(context.Background(), signedProp)
-		if err != nil {
-			return nil, errors.WithMessagef(err, "error endorsing %s", funcName)
-		}
-		responses = append(responses, proposalResp)
+
+	responses, err := processProposals(endorserClients, signedProp)
+	if err != nil {
+		return nil, errors.WithMessagef(err, "error endorsing %s", funcName)
 	}
 
 	if len(responses) == 0 {

internal/peer/chaincode/common_test.go

@@ -13,6 +13,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"sort"
 	"testing"
 	"time"
 
@@ -695,3 +696,45 @@ func TestChaincodeInvokeOrQuery_waitForEvent(t *testing.T) {
 		close(delayChan)
 	})
 }
+
+func TestProcessProposals(t *testing.T) {
+	// Build clients that return a range of status codes (for verifying each client is called).
+	mockClients := []pb.EndorserClient{}
+	for i := 2; i <= 5; i++ {
+		response := &pb.ProposalResponse{
+			Response:    &pb.Response{Status: int32(i * 100)},
+			Endorsement: &pb.Endorsement{},
+		}
+		mockClients = append(mockClients, common.GetMockEndorserClient(response, nil))
+	}
+	mockErrorClient := common.GetMockEndorserClient(nil, errors.New("failed to call endorser"))
+	signedProposal := &pb.SignedProposal{}
+	t.Run("should process a proposal for a single peer", func(t *testing.T) {
+		responses, err := processProposals([]pb.EndorserClient{mockClients[0]}, signedProposal)
+		assert.NoError(t, err)
+		assert.Len(t, responses, 1)
+		assert.Equal(t, responses[0].Response.Status, int32(200))
+	})
+	t.Run("should process a proposal for multiple peers", func(t *testing.T) {
+		responses, err := processProposals(mockClients, signedProposal)
+		assert.NoError(t, err)
+		assert.Len(t, responses, 4)
+		// Sort the statuses (as they may turn up in different order) before comparing.
+		statuses := []int32{}
+		for _, response := range responses {
+			statuses = append(statuses, response.Response.Status)
+		}
+		sort.Slice(statuses, func(i, j int) bool { return statuses[i] < statuses[j] })
+		assert.EqualValues(t, []int32{200, 300, 400, 500}, statuses)
+	})
+	t.Run("should return an error from processing a proposal for a single peer", func(t *testing.T) {
+		responses, err := processProposals([]pb.EndorserClient{mockErrorClient}, signedProposal)
+		assert.EqualError(t, err, "failed to call endorser")
+		assert.Nil(t, responses)
+	})
+	t.Run("should return an error from processing a proposal for a single peer within multiple peers", func(t *testing.T) {
+		responses, err := processProposals([]pb.EndorserClient{mockClients[0], mockErrorClient, mockClients[1]}, signedProposal)
+		assert.EqualError(t, err, "failed to call endorser")
+		assert.Nil(t, responses)
+	})
+}