SplitHTTP client: Add xmux (multiplex controller) for H3 & H2 (#3613)

#3613 (comment)

Closes #3560 (comment)

---------

Co-authored-by: mmmray <142015632+mmmray@users.noreply.github.com>

infra/conf/transport_internet.go

@@ -231,6 +231,14 @@ type SplitHTTPConfig struct {
 	ScMinPostsIntervalMs *Int32Range       `json:"scMinPostsIntervalMs"`
 	NoSSEHeader          bool              `json:"noSSEHeader"`
 	XPaddingBytes        *Int32Range       `json:"xPaddingBytes"`
+	Xmux                 Xmux              `json:"xmux"`
+}
+
+type Xmux struct {
+	maxConnections *Int32Range `json:"maxConnections"`
+	maxConcurrency *Int32Range `json:"maxConcurrency"`
+	cMaxReuseTimes *Int32Range `json:"cMaxReuseTimes"`
+	cMaxLifetimeMs *Int32Range `json:"cMaxLifetimeMs"`
 }
 
 func splithttpNewRandRangeConfig(input *Int32Range) *splithttp.RandRangeConfig {
@@ -254,6 +262,19 @@ func (c *SplitHTTPConfig) Build() (proto.Message, error) {
 	} else if c.Host == "" && c.Headers["Host"] != "" {
 		c.Host = c.Headers["Host"]
 	}
+
+	if c.Xmux.maxConnections != nil && c.Xmux.maxConcurrency != nil {
+		return nil, errors.New("maxConnections cannot be specified together with maxConcurrency")
+	}
+
+	// Multiplexing config
+	muxProtobuf := splithttp.Multiplexing{
+		MaxConnections: splithttpNewRandRangeConfig(c.Xmux.maxConnections),
+		MaxConcurrency: splithttpNewRandRangeConfig(c.Xmux.maxConcurrency),
+		CMaxReuseTimes: splithttpNewRandRangeConfig(c.Xmux.cMaxReuseTimes),
+		CMaxLifetimeMs: splithttpNewRandRangeConfig(c.Xmux.cMaxLifetimeMs),
+	}
+
 	config := &splithttp.Config{
 		Path:                 c.Path,
 		Host:                 c.Host,
@@ -263,6 +284,7 @@ func (c *SplitHTTPConfig) Build() (proto.Message, error) {
 		ScMinPostsIntervalMs: splithttpNewRandRangeConfig(c.ScMinPostsIntervalMs),
 		NoSSEHeader:          c.NoSSEHeader,
 		XPaddingBytes:        splithttpNewRandRangeConfig(c.XPaddingBytes),
+		Xmux:                 &muxProtobuf,
 	}
 	return config, nil
 }

transport/internet/splithttp/client.go

@@ -30,8 +30,7 @@ type DialerClient interface {
 // implements splithttp.DialerClient in terms of direct network connections
 type DefaultDialerClient struct {
 	transportConfig *Config
-	download        *http.Client
-	upload          *http.Client
+	client          *http.Client
 	isH2            bool
 	isH3            bool
 	// pool of net.Conn, created using dialUploadConn
@@ -80,7 +79,7 @@ func (c *DefaultDialerClient) OpenDownload(ctx context.Context, baseURL string)
 
 		req.Header = c.transportConfig.GetRequestHeader()
 
-		response, err := c.download.Do(req)
+		response, err := c.client.Do(req)
 		gotConn.Close()
 		if err != nil {
 			errors.LogInfoInner(ctx, err, "failed to send download http request")
@@ -138,7 +137,7 @@ func (c *DefaultDialerClient) SendUploadRequest(ctx context.Context, url string,
 	req.Header = c.transportConfig.GetRequestHeader()
 
 	if c.isH2 || c.isH3 {
-		resp, err := c.upload.Do(req)
+		resp, err := c.client.Do(req)
 		if err != nil {
 			return err
 		}

transport/internet/splithttp/config.go

@@ -105,6 +105,49 @@ func (c *Config) GetNormalizedXPaddingBytes() RandRangeConfig {
 	return *c.XPaddingBytes
 }
 
+func (m *Multiplexing) GetNormalizedCMaxReuseTimes() RandRangeConfig {
+	if m.CMaxReuseTimes == nil {
+		return RandRangeConfig{
+			From: 0,
+			To:   0,
+		}
+	}
+
+	return *m.CMaxReuseTimes
+}
+
+func (m *Multiplexing) GetNormalizedCMaxLifetimeMs() RandRangeConfig {
+	if m.CMaxLifetimeMs == nil || m.CMaxLifetimeMs.To == 0 {
+		return RandRangeConfig{
+			From: 0,
+			To:   0,
+		}
+	}
+	return *m.CMaxLifetimeMs
+}
+
+func (m *Multiplexing) GetNormalizedMaxConnections() RandRangeConfig {
+	if m.MaxConnections == nil {
+		return RandRangeConfig{
+			From: 0,
+			To:   0,
+		}
+	}
+
+	return *m.MaxConnections
+}
+
+func (m *Multiplexing) GetNormalizedMaxConcurrency() RandRangeConfig {
+	if m.MaxConcurrency == nil {
+		return RandRangeConfig{
+			From: 0,
+			To:   0,
+		}
+	}
+
+	return *m.MaxConcurrency
+}
+
 func init() {
 	common.Must(internet.RegisterProtocolConfigCreator(protocolName, func() interface{} {
 		return new(Config)

transport/internet/splithttp/config.proto

@@ -15,9 +15,17 @@ message Config {
   RandRangeConfig scMinPostsIntervalMs = 6;
   bool noSSEHeader = 7;
   RandRangeConfig xPaddingBytes = 8;
+  Multiplexing xmux = 9;
 }
 
 message RandRangeConfig {
     int32 from = 1;
     int32 to = 2;
 }
+
+message Multiplexing {
+  RandRangeConfig maxConnections = 1;
+  RandRangeConfig maxConcurrency = 2;
+  RandRangeConfig cMaxReuseTimes = 3;
+  RandRangeConfig cMaxLifetimeMs = 4;
+}