Make TransmitConcurrency a config var

smartcontractkit · Nov 18, 2024 · 89e51d7 · 89e51d7
1 parent c58b163
commit 89e51d7
Show file tree

Hide file tree

Showing 26 changed files with 60 additions and 10 deletions.
diff --git a/.changeset/mean-dots-move.md b/.changeset/mean-dots-move.md
@@ -0,0 +1,5 @@
+---
+"chainlink": patch
+---
+
+Add config var Mercury.Transmitter.TransmitConcurrency #added
diff --git a/core/config/docs/core.toml b/core/config/docs/core.toml
@@ -689,6 +689,10 @@ TransmitQueueMaxSize = 10_000 # Default
 # when sending a message to the mercury server, before aborting and considering
 # the transmission to be failed.
 TransmitTimeout = "5s" # Default
+# TransmitConcurrency is the max number of concurrent transmits to each server.
+#
+# Only has effect with LLO jobs.
+TransmitConcurrency = 100 # Default
 
 # Telemetry holds OTEL settings.
 # This data includes open telemetry metrics, traces, & logs.

diff --git a/core/config/mercury_config.go b/core/config/mercury_config.go
@@ -20,6 +20,7 @@ type MercuryTLS interface {
 type MercuryTransmitter interface {
 	TransmitQueueMaxSize() uint32
 	TransmitTimeout() commonconfig.Duration
+	TransmitConcurrency() uint32
 }
 
 type Mercury interface {

diff --git a/core/config/toml/types.go b/core/config/toml/types.go
@@ -1330,6 +1330,7 @@ func (m *MercuryTLS) ValidateConfig() (err error) {
 type MercuryTransmitter struct {
 	TransmitQueueMaxSize *uint32
 	TransmitTimeout      *commonconfig.Duration
+	TransmitConcurrency  *uint32
 }
 
 func (m *MercuryTransmitter) setFrom(f *MercuryTransmitter) {
@@ -1339,6 +1340,9 @@ func (m *MercuryTransmitter) setFrom(f *MercuryTransmitter) {
 	if v := f.TransmitTimeout; v != nil {
 		m.TransmitTimeout = v
 	}
+	if v := f.TransmitConcurrency; v != nil {
+		m.TransmitConcurrency = v
+	}
 }
 
 type Mercury struct {

diff --git a/core/services/chainlink/config_mercury.go b/core/services/chainlink/config_mercury.go
@@ -50,6 +50,10 @@ func (m *mercuryTransmitterConfig) TransmitTimeout() commonconfig.Duration {
 	return *m.c.TransmitTimeout
 }
 
+func (m *mercuryTransmitterConfig) TransmitConcurrency() uint32 {
+	return *m.c.TransmitConcurrency
+}
+
 type mercuryConfig struct {
 	c toml.Mercury
 	s toml.MercurySecrets

diff --git a/core/services/chainlink/config_test.go b/core/services/chainlink/config_test.go
@@ -838,6 +838,7 @@ func TestConfig_Marshal(t *testing.T) {
 		Transmitter: toml.MercuryTransmitter{
 			TransmitQueueMaxSize: ptr(uint32(123)),
 			TransmitTimeout:      commoncfg.MustNewDuration(234 * time.Second),
+			TransmitConcurrency:  ptr(uint32(456)),
 		},
 		VerboseLogging: ptr(true),
 	}
@@ -1348,6 +1349,7 @@ CertFile = '/path/to/cert.pem'
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 123
 TransmitTimeout = '3m54s'
+TransmitConcurrency = 456
 `},
 		{"full", full, fullTOML},
 		{"multi-chain", multiChain, multiChainTOML},

diff --git a/core/services/chainlink/testdata/config-empty-effective.toml b/core/services/chainlink/testdata/config-empty-effective.toml
@@ -237,6 +237,7 @@ CertFile = ''
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 10000
 TransmitTimeout = '5s'
+TransmitConcurrency = 100
 
 [Capabilities]
 [Capabilities.Peering]

diff --git a/core/services/chainlink/testdata/config-full.toml b/core/services/chainlink/testdata/config-full.toml
@@ -247,6 +247,7 @@ CertFile = '/path/to/cert.pem'
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 123
 TransmitTimeout = '3m54s'
+TransmitConcurrency = 456
 
 [Capabilities]
 [Capabilities.Peering]

diff --git a/core/services/chainlink/testdata/config-multi-chain-effective.toml b/core/services/chainlink/testdata/config-multi-chain-effective.toml
@@ -237,6 +237,7 @@ CertFile = ''
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 10000
 TransmitTimeout = '5s'
+TransmitConcurrency = 100
 
 [Capabilities]
 [Capabilities.Peering]

diff --git a/core/services/llo/mercurytransmitter/transmitter.go b/core/services/llo/mercurytransmitter/transmitter.go
@@ -29,14 +29,6 @@ import (
 const (
 	// Mercury server error codes
 	DuplicateReport = 2
-
-	// TransmitConcurrency is the number of concurrent transmits to each
-	// server. Number of goroutines per server will be roughly 2x this because
-	// each server has a delete queue and a transmit queue.
-	//
-	// This could be reduced by implementing transmit batching, see: https://smartcontract-it.atlassian.net/browse/MERC-6635
-	// TODO: Should this be a configuration variable (e.g. set to 1 or 2 for testing?)
-	TransmitConcurrency = 100
 )
 
 var (
@@ -105,6 +97,7 @@ var _ Transmitter = (*transmitter)(nil)
 type Config interface {
 	TransmitQueueMaxSize() uint32
 	TransmitTimeout() commonconfig.Duration
+	TransmitConcurrency() uint32
 }
 
 type transmitter struct {
@@ -172,10 +165,17 @@ func (mt *transmitter) Start(ctx context.Context) (err error) {
 					return err
 				}
 				s.q.Init(transmissions)
-				// starting pm after loading from it is fine because it simply spawns some garbage collection/prune goroutines
+				// starting pm after loading from it is fine because it simply
+				// spawns some garbage collection/prune goroutines
 				startClosers = append(startClosers, s.c, s.q, s.pm)
 
-				nThreads := int(TransmitConcurrency)
+				// Number of goroutines per server will be roughly
+				// 2*nServers*TransmitConcurrency because each server has a
+				// delete queue and a transmit queue.
+				//
+				// This could potentially be reduced by implementing transmit batching,
+				// see: https://smartcontract-it.atlassian.net/browse/MERC-6635
+				nThreads := int(mt.cfg.TransmitConcurrency())
 				mt.wg.Add(2 * nThreads)
 				for i := 0; i < nThreads; i++ {
 					go s.runDeleteQueueLoop(mt.stopCh, mt.wg)

diff --git a/core/services/llo/mercurytransmitter/transmitter_test.go b/core/services/llo/mercurytransmitter/transmitter_test.go
@@ -33,6 +33,10 @@ func (m mockCfg) TransmitTimeout() commonconfig.Duration {
 	return *commonconfig.MustNewDuration(1 * time.Hour)
 }
 
+func (m mockCfg) TransmitConcurrency() uint32 {
+	return 5
+}
+
 func Test_Transmitter_Transmit(t *testing.T) {
 	lggr := logger.TestLogger(t)
 	db := pgtest.NewSqlxDB(t)

diff --git a/core/services/ocr2/plugins/llo/helpers_test.go b/core/services/ocr2/plugins/llo/helpers_test.go
@@ -185,6 +185,7 @@ func setupNode(
 
 		// [Mercury]
 		c.Mercury.VerboseLogging = ptr(true)
+		c.Mercury.Transmitter.TransmitConcurrency = ptr(uint32(5)) // Avoid a ridiculous number of goroutines
 	})
 
 	lggr, observedLogs := logger.TestLoggerObserved(t, zapcore.DebugLevel)

diff --git a/core/web/resolver/testdata/config-empty-effective.toml b/core/web/resolver/testdata/config-empty-effective.toml
@@ -237,6 +237,7 @@ CertFile = ''
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 10000
 TransmitTimeout = '5s'
+TransmitConcurrency = 100
 
 [Capabilities]
 [Capabilities.Peering]

diff --git a/core/web/resolver/testdata/config-full.toml b/core/web/resolver/testdata/config-full.toml
@@ -247,6 +247,7 @@ CertFile = ''
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 123
 TransmitTimeout = '3m54s'
+TransmitConcurrency = 456
 
 [Capabilities]
 [Capabilities.Peering]

diff --git a/core/web/resolver/testdata/config-multi-chain-effective.toml b/core/web/resolver/testdata/config-multi-chain-effective.toml
@@ -237,6 +237,7 @@ CertFile = ''
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 10000
 TransmitTimeout = '5s'
+TransmitConcurrency = 100
 
 [Capabilities]
 [Capabilities.Peering]

diff --git a/docs/CONFIG.md b/docs/CONFIG.md
@@ -1875,6 +1875,7 @@ CertFile is the path to a PEM file of trusted root certificate authority certifi
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 10_000 # Default
 TransmitTimeout = "5s" # Default
+TransmitConcurrency = 100 # Default
 ```
 Mercury.Transmitter controls settings for the mercury transmitter
 
@@ -1897,6 +1898,14 @@ TransmitTimeout controls how long the transmitter will wait for a response
 when sending a message to the mercury server, before aborting and considering
 the transmission to be failed.
 
+### TransmitConcurrency
+```toml
+TransmitConcurrency = 100 # Default
+```
+TransmitConcurrency is the max number of concurrent transmits to each server.
+
+Only has effect with LLO jobs.
+
 ## Telemetry
 ```toml
 [Telemetry]

diff --git a/testdata/scripts/config/merge_raw_configs.txtar b/testdata/scripts/config/merge_raw_configs.txtar
@@ -384,6 +384,7 @@ CertFile = ''
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 10000
 TransmitTimeout = '5s'
+TransmitConcurrency = 100
 
 [Capabilities]
 [Capabilities.Peering]

diff --git a/testdata/scripts/node/validate/default.txtar b/testdata/scripts/node/validate/default.txtar
@@ -249,6 +249,7 @@ CertFile = ''
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 10000
 TransmitTimeout = '5s'
+TransmitConcurrency = 100
 
 [Capabilities]
 [Capabilities.Peering]

diff --git a/testdata/scripts/node/validate/defaults-override.txtar b/testdata/scripts/node/validate/defaults-override.txtar
@@ -310,6 +310,7 @@ CertFile = ''
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 10000
 TransmitTimeout = '5s'
+TransmitConcurrency = 100
 
 [Capabilities]
 [Capabilities.Peering]

diff --git a/testdata/scripts/node/validate/disk-based-logging-disabled.txtar b/testdata/scripts/node/validate/disk-based-logging-disabled.txtar
@@ -293,6 +293,7 @@ CertFile = ''
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 10000
 TransmitTimeout = '5s'
+TransmitConcurrency = 100
 
 [Capabilities]
 [Capabilities.Peering]

diff --git a/testdata/scripts/node/validate/disk-based-logging-no-dir.txtar b/testdata/scripts/node/validate/disk-based-logging-no-dir.txtar
@@ -293,6 +293,7 @@ CertFile = ''
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 10000
 TransmitTimeout = '5s'
+TransmitConcurrency = 100
 
 [Capabilities]
 [Capabilities.Peering]

diff --git a/testdata/scripts/node/validate/disk-based-logging.txtar b/testdata/scripts/node/validate/disk-based-logging.txtar
@@ -293,6 +293,7 @@ CertFile = ''
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 10000
 TransmitTimeout = '5s'
+TransmitConcurrency = 100
 
 [Capabilities]
 [Capabilities.Peering]

diff --git a/testdata/scripts/node/validate/invalid-ocr-p2p.txtar b/testdata/scripts/node/validate/invalid-ocr-p2p.txtar
@@ -278,6 +278,7 @@ CertFile = ''
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 10000
 TransmitTimeout = '5s'
+TransmitConcurrency = 100
 
 [Capabilities]
 [Capabilities.Peering]

diff --git a/testdata/scripts/node/validate/invalid.txtar b/testdata/scripts/node/validate/invalid.txtar
@@ -283,6 +283,7 @@ CertFile = ''
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 10000
 TransmitTimeout = '5s'
+TransmitConcurrency = 100
 
 [Capabilities]
 [Capabilities.Peering]

diff --git a/testdata/scripts/node/validate/valid.txtar b/testdata/scripts/node/validate/valid.txtar
@@ -290,6 +290,7 @@ CertFile = ''
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 10000
 TransmitTimeout = '5s'
+TransmitConcurrency = 100
 
 [Capabilities]
 [Capabilities.Peering]

diff --git a/testdata/scripts/node/validate/warnings.txtar b/testdata/scripts/node/validate/warnings.txtar
@@ -272,6 +272,7 @@ CertFile = ''
 [Mercury.Transmitter]
 TransmitQueueMaxSize = 10000
 TransmitTimeout = '5s'
+TransmitConcurrency = 100
 
 [Capabilities]
 [Capabilities.Peering]