[chore/performance] Make sender multiplier configurable (#1750)

internal/config/config.go

@@ -141,6 +141,7 @@ type Configuration struct {
 	AdvancedRateLimitRequests    int           `name:"advanced-rate-limit-requests" usage:"Amount of HTTP requests to permit within a 5 minute window. 0 or less turns rate limiting off."`
 	AdvancedThrottlingMultiplier int           `name:"advanced-throttling-multiplier" usage:"Multiplier to use per cpu for http request throttling. 0 or less turns throttling off."`
 	AdvancedThrottlingRetryAfter time.Duration `name:"advanced-throttling-retry-after" usage:"Retry-After duration response to send for throttled requests."`
+	AdvancedSenderMultiplier     int           `name:"advanced-sender-multiplier" usage:"Multiplier to use per cpu for batching outgoing fedi messages. 0 or less turns batching off (not recommended)."`
 
 	// Cache configuration vars.
 	Cache CacheConfiguration `name:"cache"`

internal/config/defaults.go

@@ -116,6 +116,7 @@ var Defaults = Configuration{
 	AdvancedCookiesSamesite:      "lax",
 	AdvancedRateLimitRequests:    300, // 1 per second per 5 minutes
 	AdvancedThrottlingMultiplier: 8,   // 8 open requests per CPU
+	AdvancedSenderMultiplier:     2,   // 2 senders per CPU
 
 	Cache: CacheConfiguration{
 		GTS: GTSCacheConfiguration{

internal/config/flags.go

@@ -144,6 +144,7 @@ func (s *ConfigState) AddServerFlags(cmd *cobra.Command) {
 		cmd.Flags().Int(AdvancedRateLimitRequestsFlag(), cfg.AdvancedRateLimitRequests, fieldtag("AdvancedRateLimitRequests", "usage"))
 		cmd.Flags().Int(AdvancedThrottlingMultiplierFlag(), cfg.AdvancedThrottlingMultiplier, fieldtag("AdvancedThrottlingMultiplier", "usage"))
 		cmd.Flags().Duration(AdvancedThrottlingRetryAfterFlag(), cfg.AdvancedThrottlingRetryAfter, fieldtag("AdvancedThrottlingRetryAfter", "usage"))
+		cmd.Flags().Int(AdvancedSenderMultiplierFlag(), cfg.AdvancedSenderMultiplier, fieldtag("AdvancedSenderMultiplier", "usage"))
 
 		cmd.Flags().String(RequestIDHeaderFlag(), cfg.RequestIDHeader, fieldtag("RequestIDHeader", "usage"))
 	})

internal/config/helpers.gen.go

@@ -2124,6 +2124,31 @@ func GetAdvancedThrottlingRetryAfter() time.Duration { return global.GetAdvanced
 // SetAdvancedThrottlingRetryAfter safely sets the value for global configuration 'AdvancedThrottlingRetryAfter' field
 func SetAdvancedThrottlingRetryAfter(v time.Duration) { global.SetAdvancedThrottlingRetryAfter(v) }
 
+// GetAdvancedSenderMultiplier safely fetches the Configuration value for state's 'AdvancedSenderMultiplier' field
+func (st *ConfigState) GetAdvancedSenderMultiplier() (v int) {
+	st.mutex.Lock()
+	v = st.config.AdvancedSenderMultiplier
+	st.mutex.Unlock()
+	return
+}
+
+// SetAdvancedSenderMultiplier safely sets the Configuration value for state's 'AdvancedSenderMultiplier' field
+func (st *ConfigState) SetAdvancedSenderMultiplier(v int) {
+	st.mutex.Lock()
+	defer st.mutex.Unlock()
+	st.config.AdvancedSenderMultiplier = v
+	st.reloadToViper()
+}
+
+// AdvancedSenderMultiplierFlag returns the flag name for the 'AdvancedSenderMultiplier' field
+func AdvancedSenderMultiplierFlag() string { return "advanced-sender-multiplier" }
+
+// GetAdvancedSenderMultiplier safely fetches the value for global configuration 'AdvancedSenderMultiplier' field
+func GetAdvancedSenderMultiplier() int { return global.GetAdvancedSenderMultiplier() }
+
+// SetAdvancedSenderMultiplier safely sets the value for global configuration 'AdvancedSenderMultiplier' field
+func SetAdvancedSenderMultiplier(v int) { global.SetAdvancedSenderMultiplier(v) }
+
 // GetCacheGTSAccountMaxSize safely fetches the Configuration value for state's 'Cache.GTS.AccountMaxSize' field
 func (st *ConfigState) GetCacheGTSAccountMaxSize() (v int) {
 	st.mutex.Lock()

internal/transport/controller.go

@@ -57,10 +57,19 @@ type controller struct {
 
 // NewController returns an implementation of the Controller interface for creating new transports
 func NewController(state *state.State, federatingDB federatingdb.DB, clock pub.Clock, client httpclient.SigningClient) Controller {
-	applicationName := config.GetApplicationName()
-	host := config.GetHost()
-	proto := config.GetProtocol()
-	version := config.GetSoftwareVersion()
+	var (
+		applicationName  = config.GetApplicationName()
+		host             = config.GetHost()
+		proto            = config.GetProtocol()
+		version          = config.GetSoftwareVersion()
+		senderMultiplier = config.GetAdvancedSenderMultiplier()
+	)
+
+	senders := senderMultiplier * runtime.GOMAXPROCS(0)
+	if senders < 1 {
+		// Clamp senders to 1.
+		senders = 1
+	}
 
 	c := &controller{
 		state:     state,
@@ -69,7 +78,7 @@ func NewController(state *state.State, federatingDB federatingdb.DB, clock pub.C
 		client:    client,
 		trspCache: cache.New[string, *transport](0, 100, 0),
 		userAgent: fmt.Sprintf("%s (+%s://%s) gotosocial/%s", applicationName, proto, host, version),
-		senders:   2 * runtime.GOMAXPROCS(0), // on batch delivery, only ever send 2*GOMAXPROCS at a time.
+		senders:   senders,
 	}
 
 	return c