Khalil/1899 async rpc sent tracker

cmd/access/node_builder/access_node_builder.go

@@ -1192,12 +1192,15 @@ func (builder *FlowAccessNodeBuilder) initPublicLibp2pNode(networkKey crypto.Pri
 	}
 
 	meshTracerCfg := &tracer.GossipSubMeshTracerConfig{
-		Logger:                       builder.Logger,
-		Metrics:                      networkMetrics,
-		IDProvider:                   builder.IdentityProvider,
-		LoggerInterval:               builder.FlowConfig.NetworkConfig.GossipSubConfig.LocalMeshLogInterval,
-		RpcSentTrackerCacheCollector: metrics.GossipSubRPCSentTrackerMetricFactory(builder.HeroCacheMetricsFactory(), network.PublicNetwork),
-		RpcSentTrackerCacheSize:      builder.FlowConfig.NetworkConfig.GossipSubConfig.RPCSentTrackerCacheSize,
+		Logger:                                  builder.Logger,
+		Metrics:                                 networkMetrics,
+		IDProvider:                              builder.IdentityProvider,
+		LoggerInterval:                          builder.FlowConfig.NetworkConfig.GossipSubConfig.LocalMeshLogInterval,
+		RpcSentTrackerCacheCollector:            metrics.GossipSubRPCSentTrackerMetricFactory(builder.HeroCacheMetricsFactory(), network.PublicNetwork),
+		RpcSentTrackerCacheSize:                 builder.FlowConfig.NetworkConfig.GossipSubConfig.RPCSentTrackerCacheSize,
+		RpcSentTrackerWorkerQueueCacheSize:      builder.FlowConfig.NetworkConfig.GossipSubConfig.RPCSentTrackerQueueCacheSize,
+		RpcSentTrackerNumOfWorkers:              builder.FlowConfig.NetworkConfig.GossipSubConfig.RpcSentTrackerNumOfWorkers,
+		RpcSentTrackerWorkerQueueCacheCollector: metrics.GossipSubRPCSentTrackerQueueMetricFactory(builder.HeroCacheMetricsFactory(), network.PublicNetwork),
 	}
 	meshTracer := tracer.NewGossipSubMeshTracer(meshTracerCfg)
 

cmd/observer/node_builder/observer_builder.go

@@ -703,12 +703,15 @@ func (builder *ObserverServiceBuilder) initPublicLibp2pNode(networkKey crypto.Pr
 	}
 
 	meshTracerCfg := &tracer.GossipSubMeshTracerConfig{
-		Logger:                       builder.Logger,
-		Metrics:                      builder.Metrics.Network,
-		IDProvider:                   builder.IdentityProvider,
-		LoggerInterval:               builder.FlowConfig.NetworkConfig.GossipSubConfig.LocalMeshLogInterval,
-		RpcSentTrackerCacheCollector: metrics.GossipSubRPCSentTrackerMetricFactory(builder.HeroCacheMetricsFactory(), network.PublicNetwork),
-		RpcSentTrackerCacheSize:      builder.FlowConfig.NetworkConfig.GossipSubConfig.RPCSentTrackerCacheSize,
+		Logger:                                  builder.Logger,
+		Metrics:                                 builder.Metrics.Network,
+		IDProvider:                              builder.IdentityProvider,
+		LoggerInterval:                          builder.FlowConfig.NetworkConfig.GossipSubConfig.LocalMeshLogInterval,
+		RpcSentTrackerCacheCollector:            metrics.GossipSubRPCSentTrackerMetricFactory(builder.HeroCacheMetricsFactory(), network.PublicNetwork),
+		RpcSentTrackerCacheSize:                 builder.FlowConfig.NetworkConfig.GossipSubConfig.RPCSentTrackerCacheSize,
+		RpcSentTrackerWorkerQueueCacheSize:      builder.FlowConfig.NetworkConfig.GossipSubConfig.RPCSentTrackerQueueCacheSize,
+		RpcSentTrackerNumOfWorkers:              builder.FlowConfig.NetworkConfig.GossipSubConfig.RpcSentTrackerNumOfWorkers,
+		RpcSentTrackerWorkerQueueCacheCollector: metrics.GossipSubRPCSentTrackerQueueMetricFactory(builder.HeroCacheMetricsFactory(), network.PublicNetwork),
 	}
 	meshTracer := tracer.NewGossipSubMeshTracer(meshTracerCfg)
 

config/default-config.yml

@@ -66,6 +66,10 @@ network-config:
   # The default RPC sent tracker cache size. The RPC sent tracker is used to track RPC control messages sent from the local node.
   # Note: this cache size must be large enough to keep a history of sent messages in a reasonable time window of past history.
   gossipsub-rpc-sent-tracker-cache-size: 1_000_000
+  # Cache size of the rpc sent tracker queue used for async tracking.
+  gossipsub-rpc-sent-tracker-queue-cache-size: 1000
+  # Number of workers for rpc sent tracker worker pool.
+  gossipsub-rpc-sent-tracker-workers: 5
   # Peer scoring is the default value for enabling peer scoring
   gossipsub-peer-scoring-enabled: true
   # Gossipsub rpc inspectors configs

follower/follower_builder.go

@@ -605,12 +605,15 @@ func (builder *FollowerServiceBuilder) initPublicLibp2pNode(networkKey crypto.Pr
 	}
 
 	meshTracerCfg := &tracer.GossipSubMeshTracerConfig{
-		Logger:                       builder.Logger,
-		Metrics:                      builder.Metrics.Network,
-		IDProvider:                   builder.IdentityProvider,
-		LoggerInterval:               builder.FlowConfig.NetworkConfig.GossipSubConfig.LocalMeshLogInterval,
-		RpcSentTrackerCacheCollector: metrics.GossipSubRPCSentTrackerMetricFactory(builder.HeroCacheMetricsFactory(), network.PublicNetwork),
-		RpcSentTrackerCacheSize:      builder.FlowConfig.NetworkConfig.GossipSubConfig.RPCSentTrackerCacheSize,
+		Logger:                                  builder.Logger,
+		Metrics:                                 builder.Metrics.Network,
+		IDProvider:                              builder.IdentityProvider,
+		LoggerInterval:                          builder.FlowConfig.NetworkConfig.GossipSubConfig.LocalMeshLogInterval,
+		RpcSentTrackerCacheCollector:            metrics.GossipSubRPCSentTrackerMetricFactory(builder.HeroCacheMetricsFactory(), network.PublicNetwork),
+		RpcSentTrackerCacheSize:                 builder.FlowConfig.NetworkConfig.GossipSubConfig.RPCSentTrackerCacheSize,
+		RpcSentTrackerWorkerQueueCacheSize:      builder.FlowConfig.NetworkConfig.GossipSubConfig.RPCSentTrackerQueueCacheSize,
+		RpcSentTrackerNumOfWorkers:              builder.FlowConfig.NetworkConfig.GossipSubConfig.RpcSentTrackerNumOfWorkers,
+		RpcSentTrackerWorkerQueueCacheCollector: metrics.GossipSubRPCSentTrackerQueueMetricFactory(builder.HeroCacheMetricsFactory(), network.PublicNetwork),
 	}
 	meshTracer := tracer.NewGossipSubMeshTracer(meshTracerCfg)
 

module/component/component.go

@@ -143,7 +143,8 @@ func NoopWorker(ctx irrecoverable.SignalerContext, ready ReadyFunc) {
 type ComponentManagerBuilder interface {
 	// AddWorker adds a worker routine for the ComponentManager
 	AddWorker(ComponentWorker) ComponentManagerBuilder
-
+	// AddWorkers adds n number of worker routines for the ComponentManager.
+	AddWorkers(int, ComponentWorker) ComponentManagerBuilder
 	// Build builds and returns a new ComponentManager instance
 	Build() *ComponentManager
 }
@@ -157,15 +158,23 @@ func NewComponentManagerBuilder() ComponentManagerBuilder {
 	return &componentManagerBuilderImpl{}
 }
 
-// AddWorker adds a ComponentWorker closure to the ComponentManagerBuilder
 // All worker functions will be run in parallel when the ComponentManager is started.
 // Note: AddWorker is not concurrency-safe, and should only be called on an individual builder
-// within a single goroutine.
+// within a single goroutine.// AddWorker adds a ComponentWorker closure to the ComponentManagerBuilder
+
 func (c *componentManagerBuilderImpl) AddWorker(worker ComponentWorker) ComponentManagerBuilder {
 	c.workers = append(c.workers, worker)
 	return c
 }
 
+// AddWorkers adds n number of workers for the ComponentManager.
+func (c *componentManagerBuilderImpl) AddWorkers(n int, worker ComponentWorker) ComponentManagerBuilder {
+	for i := 0; i < n; i++ {
+		c.workers = append(c.workers, worker)
+	}
+	return c
+}
+
 // Build returns a new ComponentManager instance with the configured workers
 // Build may be called multiple times to create multiple individual ComponentManagers. This will
 // result in the worker routines being called multiple times. If this is unsafe, do not call it

module/metrics/herocache.go

@@ -155,6 +155,15 @@ func GossipSubRPCSentTrackerMetricFactory(f HeroCacheMetricsFactory, networkType
 	return f(namespaceNetwork, r)
 }
 
+func GossipSubRPCSentTrackerQueueMetricFactory(f HeroCacheMetricsFactory, networkType network.NetworkingType) module.HeroCacheMetrics {
+	// we don't use the public prefix for the metrics here for sake of backward compatibility of metric name.
+	r := ResourceNetworkingRPCSentTrackerQueue
+	if networkType == network.PublicNetwork {
+		r = PrependPublicPrefix(r)
+	}
+	return f(namespaceNetwork, r)
+}
+
 func RpcInspectorNotificationQueueMetricFactory(f HeroCacheMetricsFactory, networkType network.NetworkingType) module.HeroCacheMetrics {
 	r := ResourceNetworkingRpcInspectorNotificationQueue
 	if networkType == network.PublicNetwork {

module/metrics/labels.go

@@ -93,6 +93,7 @@ const (
 	ResourceNetworkingRpcClusterPrefixReceivedCache    = "rpc_cluster_prefixed_received_cache"
 	ResourceNetworkingDisallowListCache                = "disallow_list_cache"
 	ResourceNetworkingRPCSentTrackerCache              = "gossipsub_rpc_sent_tracker_cache"
+	ResourceNetworkingRPCSentTrackerQueue              = "gossipsub_rpc_sent_tracker_queue"
 
 	ResourceFollowerPendingBlocksCache         = "follower_pending_block_cache"           // follower engine
 	ResourceFollowerLoopCertifiedBlocksChannel = "follower_loop_certified_blocks_channel" // follower loop, certified blocks buffered channel

network/internal/p2pfixtures/fixtures.go

@@ -104,12 +104,15 @@ func CreateNode(t *testing.T, networkKey crypto.PrivateKey, sporkID flow.Identif
 	require.NoError(t, err)
 
 	meshTracerCfg := &tracer.GossipSubMeshTracerConfig{
-		Logger:                       logger,
-		Metrics:                      metrics.NewNoopCollector(),
-		IDProvider:                   idProvider,
-		LoggerInterval:               defaultFlowConfig.NetworkConfig.GossipSubConfig.LocalMeshLogInterval,
-		RpcSentTrackerCacheCollector: metrics.NewNoopCollector(),
-		RpcSentTrackerCacheSize:      defaultFlowConfig.NetworkConfig.GossipSubConfig.RPCSentTrackerCacheSize,
+		Logger:                                  logger,
+		Metrics:                                 metrics.NewNoopCollector(),
+		IDProvider:                              idProvider,
+		LoggerInterval:                          defaultFlowConfig.NetworkConfig.GossipSubConfig.LocalMeshLogInterval,
+		RpcSentTrackerCacheCollector:            metrics.NewNoopCollector(),
+		RpcSentTrackerCacheSize:                 defaultFlowConfig.NetworkConfig.GossipSubConfig.RPCSentTrackerCacheSize,
+		RpcSentTrackerWorkerQueueCacheSize:      defaultFlowConfig.NetworkConfig.GossipSubConfig.RPCSentTrackerQueueCacheSize,
+		RpcSentTrackerNumOfWorkers:              defaultFlowConfig.NetworkConfig.GossipSubConfig.RpcSentTrackerNumOfWorkers,
+		RpcSentTrackerWorkerQueueCacheCollector: metrics.NewNoopCollector(),
 	}
 	meshTracer := tracer.NewGossipSubMeshTracer(meshTracerCfg)
 

network/netconf/flags.go

@@ -37,10 +37,12 @@ const (
 	gracePeriod   = "libp2p-grace-period"
 	silencePeriod = "libp2p-silence-period"
 	// gossipsub
-	peerScoring             = "gossipsub-peer-scoring-enabled"
-	localMeshLogInterval    = "gossipsub-local-mesh-logging-interval"
-	rpcSentTrackerCacheSize = "gossipsub-rpc-sent-tracker-cache-size"
-	scoreTracerInterval     = "gossipsub-score-tracer-interval"
+	peerScoring                  = "gossipsub-peer-scoring-enabled"
+	localMeshLogInterval         = "gossipsub-local-mesh-logging-interval"
+	rpcSentTrackerCacheSize      = "gossipsub-rpc-sent-tracker-cache-size"
+	rpcSentTrackerQueueCacheSize = "gossipsub-rpc-sent-tracker-queue-cache-size"
+	rpcSentTrackerNumOfWorkers   = "gossipsub-rpc-sent-tracker-workers"
+	scoreTracerInterval          = "gossipsub-score-tracer-interval"
 	// gossipsub validation inspector
 	gossipSubRPCInspectorNotificationCacheSize                 = "gossipsub-rpc-inspector-notification-cache-size"
 	validationInspectorNumberOfWorkers                         = "gossipsub-rpc-validation-inspector-workers"
@@ -67,8 +69,8 @@ func AllFlagNames() []string {
 	return []string{
 		networkingConnectionPruning, preferredUnicastsProtocols, receivedMessageCacheSize, peerUpdateInterval, unicastMessageTimeout, unicastCreateStreamRetryDelay,
 		dnsCacheTTL, disallowListNotificationCacheSize, dryRun, lockoutDuration, messageRateLimit, bandwidthRateLimit, bandwidthBurstLimit, memoryLimitRatio,
-		fileDescriptorsRatio, peerBaseLimitConnsInbound, highWatermark, lowWatermark, gracePeriod, silencePeriod, peerScoring, localMeshLogInterval, rpcSentTrackerCacheSize, scoreTracerInterval,
-		gossipSubRPCInspectorNotificationCacheSize, validationInspectorNumberOfWorkers, validationInspectorInspectMessageQueueCacheSize, validationInspectorClusterPrefixedTopicsReceivedCacheSize,
+		fileDescriptorsRatio, peerBaseLimitConnsInbound, highWatermark, lowWatermark, gracePeriod, silencePeriod, peerScoring, localMeshLogInterval, rpcSentTrackerCacheSize, rpcSentTrackerQueueCacheSize, rpcSentTrackerNumOfWorkers,
+		scoreTracerInterval, gossipSubRPCInspectorNotificationCacheSize, validationInspectorNumberOfWorkers, validationInspectorInspectMessageQueueCacheSize, validationInspectorClusterPrefixedTopicsReceivedCacheSize,
 		validationInspectorClusterPrefixedTopicsReceivedCacheDecay, validationInspectorClusterPrefixHardThreshold, ihaveSyncSampleSizePercentage, ihaveAsyncSampleSizePercentage,
 		ihaveMaxSampleSize, metricsInspectorNumberOfWorkers, metricsInspectorCacheSize, alspDisabled, alspSpamRecordCacheSize, alspSpamRecordQueueSize, alspHearBeatInterval,
 	}
@@ -109,6 +111,8 @@ func InitializeNetworkFlags(flags *pflag.FlagSet, config *Config) {
 	flags.Duration(localMeshLogInterval, config.GossipSubConfig.LocalMeshLogInterval, "logging interval for local mesh in gossipsub")
 	flags.Duration(scoreTracerInterval, config.GossipSubConfig.ScoreTracerInterval, "logging interval for peer score tracer in gossipsub, set to 0 to disable")
 	flags.Uint32(rpcSentTrackerCacheSize, config.GossipSubConfig.RPCSentTrackerCacheSize, "cache size of the rpc sent tracker used by the gossipsub mesh tracer.")
+	flags.Uint32(rpcSentTrackerQueueCacheSize, config.GossipSubConfig.RPCSentTrackerQueueCacheSize, "cache size of the rpc sent tracker worker queue.")
+	flags.Int(rpcSentTrackerNumOfWorkers, config.GossipSubConfig.RpcSentTrackerNumOfWorkers, "number of workers for the rpc sent tracker worker pool.")
 	// gossipsub RPC control message validation limits used for validation configuration and rate limiting
 	flags.Int(validationInspectorNumberOfWorkers, config.GossipSubConfig.GossipSubRPCInspectorsConfig.GossipSubRPCValidationInspectorConfigs.NumberOfWorkers, "number of gossupsub RPC control message validation inspector component workers")
 	flags.Uint32(validationInspectorInspectMessageQueueCacheSize, config.GossipSubConfig.GossipSubRPCInspectorsConfig.GossipSubRPCValidationInspectorConfigs.CacheSize, "cache size for gossipsub RPC validation inspector events worker pool queue.")

network/p2p/p2pbuilder/libp2pNodeBuilder.go

@@ -496,12 +496,15 @@ func DefaultNodeBuilder(
 	}
 
 	meshTracerCfg := &tracer.GossipSubMeshTracerConfig{
-		Logger:                       logger,
-		Metrics:                      metricsCfg.Metrics,
-		IDProvider:                   idProvider,
-		LoggerInterval:               gossipCfg.LocalMeshLogInterval,
-		RpcSentTrackerCacheCollector: metrics.GossipSubRPCSentTrackerMetricFactory(metricsCfg.HeroCacheFactory, flownet.PrivateNetwork),
-		RpcSentTrackerCacheSize:      gossipCfg.RPCSentTrackerCacheSize,
+		Logger:                                  logger,
+		Metrics:                                 metricsCfg.Metrics,
+		IDProvider:                              idProvider,
+		LoggerInterval:                          gossipCfg.LocalMeshLogInterval,
+		RpcSentTrackerCacheCollector:            metrics.GossipSubRPCSentTrackerMetricFactory(metricsCfg.HeroCacheFactory, flownet.PrivateNetwork),
+		RpcSentTrackerCacheSize:                 gossipCfg.RPCSentTrackerCacheSize,
+		RpcSentTrackerWorkerQueueCacheSize:      gossipCfg.RPCSentTrackerQueueCacheSize,
+		RpcSentTrackerNumOfWorkers:              gossipCfg.RpcSentTrackerNumOfWorkers,
+		RpcSentTrackerWorkerQueueCacheCollector: metrics.GossipSubRPCSentTrackerQueueMetricFactory(metricsCfg.HeroCacheFactory, flownet.PrivateNetwork),
 	}
 	meshTracer := tracer.NewGossipSubMeshTracer(meshTracerCfg)
 

network/p2p/p2pconf/gossipsub.go

@@ -23,6 +23,10 @@ type GossipSubConfig struct {
 	ScoreTracerInterval time.Duration `mapstructure:"gossipsub-score-tracer-interval"`
 	// RPCSentTrackerCacheSize cache size of the rpc sent tracker used by the gossipsub mesh tracer.
 	RPCSentTrackerCacheSize uint32 `mapstructure:"gossipsub-rpc-sent-tracker-cache-size"`
+	// RPCSentTrackerQueueCacheSize cache size of the rpc sent tracker queue used for async tracking.
+	RPCSentTrackerQueueCacheSize uint32 `mapstructure:"gossipsub-rpc-sent-tracker-queue-cache-size"`
+	// RpcSentTrackerNumOfWorkers number of workers for rpc sent tracker worker pool.
+	RpcSentTrackerNumOfWorkers int `mapstructure:"gossipsub-rpc-sent-tracker-workers"`
 	// PeerScoring is whether to enable GossipSub peer scoring.
 	PeerScoring bool `mapstructure:"gossipsub-peer-scoring-enabled"`
 }