Enable gateway concurrency limits

Add support for setting API invocation concurrency limits on the gateway service.  Uses the same mechanism as used by the endorser and deliver services.

The sample config limit has been set to 500 for now.  This can be adjusted in the future based on experience.

Signed-off-by: andrew-coleman <andrew_coleman@uk.ibm.com>
(cherry picked from commit b1b62bb)

core/peer/config.go

@@ -125,6 +125,10 @@ type Config struct {
 	// registered to deliver service for blocks and transaction events.
 	LimitsConcurrencyDeliverService int
 
+	// LimitsConcurrencyGatewayService sets the limits for concurrent requests to
+	// gateway service that handles the submission and evaluation of transactions.
+	LimitsConcurrencyGatewayService int
+
 	// ----- TLS -----
 	// Require server-side TLS.
 	// TODO: create separate sub-struct for PeerTLS config.
@@ -251,6 +255,7 @@ func (c *Config) load() error {
 	c.NetworkID = viper.GetString("peer.networkId")
 	c.LimitsConcurrencyEndorserService = viper.GetInt("peer.limits.concurrency.endorserService")
 	c.LimitsConcurrencyDeliverService = viper.GetInt("peer.limits.concurrency.deliverService")
+	c.LimitsConcurrencyGatewayService = viper.GetInt("peer.limits.concurrency.gatewayService")
 	c.DiscoveryEnabled = viper.GetBool("peer.discovery.enabled")
 	c.ProfileEnabled = viper.GetBool("peer.profile.enabled")
 	c.ProfileListenAddress = viper.GetString("peer.profile.listenAddress")

core/peer/config_test.go

@@ -274,6 +274,7 @@ func TestGlobalConfig(t *testing.T) {
 	viper.Set("peer.networkId", "testNetwork")
 	viper.Set("peer.limits.concurrency.endorserService", 2500)
 	viper.Set("peer.limits.concurrency.deliverService", 2500)
+	viper.Set("peer.limits.concurrency.gatewayService", 500)
 	viper.Set("peer.discovery.enabled", true)
 	viper.Set("peer.profile.enabled", false)
 	viper.Set("peer.profile.listenAddress", "peer.authentication.timewindow")
@@ -334,6 +335,7 @@ func TestGlobalConfig(t *testing.T) {
 		NetworkID:                             "testNetwork",
 		LimitsConcurrencyEndorserService:      2500,
 		LimitsConcurrencyDeliverService:       2500,
+		LimitsConcurrencyGatewayService:       500,
 		DiscoveryEnabled:                      true,
 		ProfileEnabled:                        false,
 		ProfileListenAddress:                  "peer.authentication.timewindow",

internal/peer/node/grpc_limiters.go

@@ -20,8 +20,9 @@ func initGrpcSemaphores(config *peer.Config) map[string]semaphore.Semaphore {
 	semaphores := make(map[string]semaphore.Semaphore)
 	endorserConcurrency := config.LimitsConcurrencyEndorserService
 	deliverConcurrency := config.LimitsConcurrencyDeliverService
+	gatewayConcurrency := config.LimitsConcurrencyGatewayService
 
-	// Currently concurrency limit is applied to endorser service and deliver service.
+	// Currently concurrency limit is applied to endorser service, deliver service and gateway service.
 	// These services are defined in fabric-protos and fabric-protos-go (generated from fabric-protos).
 	// Below service names must match their definitions.
 	if endorserConcurrency != 0 {
@@ -32,6 +33,10 @@ func initGrpcSemaphores(config *peer.Config) map[string]semaphore.Semaphore {
 		logger.Infof("concurrency limit for deliver service is %d", deliverConcurrency)
 		semaphores["/protos.Deliver"] = semaphore.New(deliverConcurrency)
 	}
+	if gatewayConcurrency != 0 {
+		logger.Infof("concurrency limit for gateway service is %d", gatewayConcurrency)
+		semaphores["/gateway.Gateway"] = semaphore.New(gatewayConcurrency)
+	}
 
 	return semaphores
 }

internal/peer/node/grpc_limiters_test.go

@@ -21,15 +21,17 @@ func TestInitGrpcSemaphores(t *testing.T) {
 	config := peer.Config{
 		LimitsConcurrencyEndorserService: 5,
 		LimitsConcurrencyDeliverService:  5,
+		LimitsConcurrencyGatewayService:  5,
 	}
 	semaphores := initGrpcSemaphores(&config)
-	require.Equal(t, 2, len(semaphores))
+	require.Equal(t, 3, len(semaphores))
 }
 
 func TestInitGrpcNoSemaphores(t *testing.T) {
 	config := peer.Config{
 		LimitsConcurrencyEndorserService: 0,
 		LimitsConcurrencyDeliverService:  0,
+		LimitsConcurrencyGatewayService:  0,
 	}
 	semaphores := initGrpcSemaphores(&config)
 	require.Equal(t, 0, len(semaphores))
@@ -75,4 +77,9 @@ func TestGetServiceName(t *testing.T) {
 	require.Equal(t, "/protos.Deliver", getServiceName("/protos.Deliver/Deliver"))
 	require.Equal(t, "/protos.Deliver", getServiceName("/protos.Deliver/DeliverFiltered"))
 	require.Equal(t, "/protos.Deliver", getServiceName("/protos.Deliver/DeliverWithPrivateData"))
+	require.Equal(t, "/gateway.Gateway", getServiceName("/gateway.Gateway/Evaluate"))
+	require.Equal(t, "/gateway.Gateway", getServiceName("/gateway.Gateway/Endorse"))
+	require.Equal(t, "/gateway.Gateway", getServiceName("/gateway.Gateway/Submit"))
+	require.Equal(t, "/gateway.Gateway", getServiceName("/gateway.Gateway/CommitStatus"))
+	require.Equal(t, "/gateway.Gateway", getServiceName("/gateway.Gateway/ChaincodeEvents"))
 }

sampleconfig/core.yaml

@@ -479,6 +479,8 @@ peer:
             endorserService: 2500
             # deliverService limits concurrent event listeners registered to deliver service for blocks and transaction events.
             deliverService: 2500
+            # gatewayService limits concurrent requests to gateway service that handles the submission and evaluation of transactions.
+            gatewayService: 500
 
     # Since all nodes should be consistent it is recommended to keep
     # the default value of 100MB for MaxRecvMsgSize & MaxSendMsgSize