Merge pull request #2351 from prnake/fix-max-conns

fix: try resolve the high concurrency issue to a single host

common/constants.go

@@ -121,6 +121,9 @@ var BatchUpdateInterval int
 
 
 var RelayTimeout int // unit is second
 var RelayTimeout int // unit is second
 
 
+var RelayMaxIdleConns int
+var RelayMaxIdleConnsPerHost int
+
 var GeminiSafetySetting string
 var GeminiSafetySetting string
 
 
 // https://docs.cohere.com/docs/safety-modes Type; NONE/CONTEXTUAL/STRICT
 // https://docs.cohere.com/docs/safety-modes Type; NONE/CONTEXTUAL/STRICT

common/init.go

@@ -90,6 +90,8 @@ func InitEnv() {
 	SyncFrequency = GetEnvOrDefault("SYNC_FREQUENCY", 60)
 	SyncFrequency = GetEnvOrDefault("SYNC_FREQUENCY", 60)
 	BatchUpdateInterval = GetEnvOrDefault("BATCH_UPDATE_INTERVAL", 5)
 	BatchUpdateInterval = GetEnvOrDefault("BATCH_UPDATE_INTERVAL", 5)
 	RelayTimeout = GetEnvOrDefault("RELAY_TIMEOUT", 0)
 	RelayTimeout = GetEnvOrDefault("RELAY_TIMEOUT", 0)
+	RelayMaxIdleConns = GetEnvOrDefault("RELAY_MAX_IDLE_CONNS", 500)
+	RelayMaxIdleConnsPerHost = GetEnvOrDefault("RELAY_MAX_IDLE_CONNS_PER_HOST", 100)
 
 
 	// Initialize string variables with GetEnvOrDefaultString
 	// Initialize string variables with GetEnvOrDefaultString
 	GeminiSafetySetting = GetEnvOrDefaultString("GEMINI_SAFETY_SETTING", "BLOCK_NONE")
 	GeminiSafetySetting = GetEnvOrDefaultString("GEMINI_SAFETY_SETTING", "BLOCK_NONE")

relay/helper/stream_scanner.go

@@ -72,6 +72,8 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon
 	if common.DebugEnabled {
 	if common.DebugEnabled {
 		// print timeout and ping interval for debugging
 		// print timeout and ping interval for debugging
 		println("relay timeout seconds:", common.RelayTimeout)
 		println("relay timeout seconds:", common.RelayTimeout)
+		println("relay max idle conns:", common.RelayMaxIdleConns)
+		println("relay max idle conns per host:", common.RelayMaxIdleConnsPerHost)
 		println("streaming timeout seconds:", int64(streamingTimeout.Seconds()))
 		println("streaming timeout seconds:", int64(streamingTimeout.Seconds()))
 		println("ping interval seconds:", int64(pingInterval.Seconds()))
 		println("ping interval seconds:", int64(pingInterval.Seconds()))
 	}
 	}

service/http_client.go

@@ -34,12 +34,20 @@ func checkRedirect(req *http.Request, via []*http.Request) error {
 }
 }
 
 
 func InitHttpClient() {
 func InitHttpClient() {
+	transport := &http.Transport{
+		MaxIdleConns:          common.RelayMaxIdleConns,
+		MaxIdleConnsPerHost:   common.RelayMaxIdleConnsPerHost,
+		ForceAttemptHTTP2:     true,
+	}
+
 	if common.RelayTimeout == 0 {
 	if common.RelayTimeout == 0 {
 		httpClient = &http.Client{
 		httpClient = &http.Client{
+			Transport:     transport,
 			CheckRedirect: checkRedirect,
 			CheckRedirect: checkRedirect,
 		}
 		}
 	} else {
 	} else {
 		httpClient = &http.Client{
 		httpClient = &http.Client{
+			Transport:     transport,
 			Timeout:       time.Duration(common.RelayTimeout) * time.Second,
 			Timeout:       time.Duration(common.RelayTimeout) * time.Second,
 			CheckRedirect: checkRedirect,
 			CheckRedirect: checkRedirect,
 		}
 		}
@@ -84,6 +92,9 @@ func NewProxyHttpClient(proxyURL string) (*http.Client, error) {
 	case "http", "https":
 	case "http", "https":
 		client := &http.Client{
 		client := &http.Client{
 			Transport: &http.Transport{
 			Transport: &http.Transport{
+				MaxIdleConns:          common.RelayMaxIdleConns,
+				MaxIdleConnsPerHost:   common.RelayMaxIdleConnsPerHost,
+				ForceAttemptHTTP2:     true,
 				Proxy: http.ProxyURL(parsedURL),
 				Proxy: http.ProxyURL(parsedURL),
 			},
 			},
 			CheckRedirect: checkRedirect,
 			CheckRedirect: checkRedirect,
@@ -116,6 +127,9 @@ func NewProxyHttpClient(proxyURL string) (*http.Client, error) {
 
 
 		client := &http.Client{
 		client := &http.Client{
 			Transport: &http.Transport{
 			Transport: &http.Transport{
+				MaxIdleConns:          common.RelayMaxIdleConns,
+				MaxIdleConnsPerHost:   common.RelayMaxIdleConnsPerHost,
+				ForceAttemptHTTP2:     true,
 				DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
 				DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
 					return dialer.Dial(network, addr)
 					return dialer.Dial(network, addr)
 				},
 				},