grpc: introduce new DialOption and ServerOption to configure initial window size without disabling BDP estimation.

default_dial_option_server_option_test.go

@@ -148,6 +148,33 @@ func (s) TestJoinDialOption(t *testing.T) {
 	}
 }
 
+// TestJoinDialOption_StaticConnAndStreamWindowSizes verifies that both static
+// stream and connection window sizes set via joined dial options are correctly
+// applied.
+func (s) TestJoinDialOption_StaticConnAndStreamWindowSizes(t *testing.T) {
+	const (
+		staticWindowSize = 123456
+		staticConnSize   = 654321
+	)
+	jdo := newJoinDialOption(
+		WithTransportCredentials(insecure.NewCredentials()),
+		WithStaticStreamWindowSize(staticWindowSize),
+		WithStaticConnWindowSize(staticConnSize),
+	)
+	cc, err := Dial("fake", jdo)
+	if err != nil {
+		t.Fatalf("Dialing with insecure credentials failed: %v", err)
+	}
+	defer cc.Close()
+	if cc.dopts.copts.StaticStreamWindowSize != staticWindowSize {
+		t.Fatalf("Unexpected cc.dopts.copts.StaticStreamWindowSize: %d != %d", cc.dopts.copts.StaticStreamWindowSize, staticWindowSize)
+	}
+	if cc.dopts.copts.StaticConnWindowSize != staticConnSize {
+		t.Fatalf("Expected StaticConnWindowSize = %d, got %d", staticConnSize, cc.dopts.copts.StaticConnWindowSize)
+	}
+
+}
+
 // TestJoinServerOption tests the join server option. It configures a joined
 // server option with three individual server options, and verifies that all
 // three are successfully applied.

dialoptions.go

@@ -225,6 +225,22 @@ func WithInitialConnWindowSize(s int32) DialOption {
 	})
 }
 
+// WithStaticStreamWindowSize returns a DialOption to set the static initial
+// stream window size.
+func WithStaticStreamWindowSize(s int32) DialOption {
+	return newFuncDialOption(func(o *dialOptions) {
+		o.copts.StaticStreamWindowSize = s
+	})
+}
+
+// WithStaticConnWindowSize returns a DialOption to set the static initial
+// connection window size.
+func WithStaticConnWindowSize(s int32) DialOption {
+	return newFuncDialOption(func(o *dialOptions) {
+		o.copts.StaticConnWindowSize = s
+	})
+}
+
 // WithMaxMsgSize returns a DialOption which sets the maximum message size the
 // client can receive.
 //

internal/transport/http2_client.go

@@ -311,7 +311,9 @@
 	}
 	dynamicWindow := true
 	icwz := int32(initialWindowSize)
-	if opts.InitialConnWindowSize >= defaultWindowSize {
+	if opts.StaticConnWindowSize > 0 {
+		icwz = opts.StaticConnWindowSize
+	} else if opts.InitialConnWindowSize >= defaultWindowSize {
 		icwz = opts.InitialConnWindowSize
 		dynamicWindow = false
 	}
@@ -379,7 +381,9 @@
 		t.md = md
 	}
 	t.controlBuf = newControlBuffer(t.ctxDone)
-	if opts.InitialWindowSize >= defaultWindowSize {
+	if opts.StaticStreamWindowSize > 0 {
+		t.initialWindowSize = opts.StaticStreamWindowSize
+	} else if opts.InitialWindowSize >= defaultWindowSize {
 		t.initialWindowSize = opts.InitialWindowSize
 		dynamicWindow = false
 	}

internal/transport/http2_server.go

@@ -177,12 +177,16 @@
 	}
 	dynamicWindow := true
 	iwz := int32(initialWindowSize)
-	if config.InitialWindowSize >= defaultWindowSize {
+	if config.StaticStreamWindowSize > 0 {
+		iwz = config.StaticStreamWindowSize
+	} else if config.InitialWindowSize >= defaultWindowSize {
 		iwz = config.InitialWindowSize
 		dynamicWindow = false
 	}
 	icwz := int32(initialWindowSize)
-	if config.InitialConnWindowSize >= defaultWindowSize {
+	if config.StaticConnWindowSize > 0 {
+		icwz = config.StaticConnWindowSize
+	} else if config.InitialConnWindowSize >= defaultWindowSize {
 		icwz = config.InitialConnWindowSize
 		dynamicWindow = false
 	}

internal/transport/transport.go

@@ -450,22 +450,24 @@ const (
 
 // ServerConfig consists of all the configurations to establish a server transport.
 type ServerConfig struct {
-	MaxStreams            uint32
-	ConnectionTimeout     time.Duration
-	Credentials           credentials.TransportCredentials
-	InTapHandle           tap.ServerInHandle
-	StatsHandlers         []stats.Handler
-	KeepaliveParams       keepalive.ServerParameters
-	KeepalivePolicy       keepalive.EnforcementPolicy
-	InitialWindowSize     int32
-	InitialConnWindowSize int32
-	WriteBufferSize       int
-	ReadBufferSize        int
-	SharedWriteBuffer     bool
-	ChannelzParent        *channelz.Server
-	MaxHeaderListSize     *uint32
-	HeaderTableSize       *uint32
-	BufferPool            mem.BufferPool
+	MaxStreams             uint32
+	ConnectionTimeout      time.Duration
+	Credentials            credentials.TransportCredentials
+	InTapHandle            tap.ServerInHandle
+	StatsHandlers          []stats.Handler
+	KeepaliveParams        keepalive.ServerParameters
+	KeepalivePolicy        keepalive.EnforcementPolicy
+	InitialWindowSize      int32
+	InitialConnWindowSize  int32
+	WriteBufferSize        int
+	ReadBufferSize         int
+	SharedWriteBuffer      bool
+	ChannelzParent         *channelz.Server
+	MaxHeaderListSize      *uint32
+	HeaderTableSize        *uint32
+	BufferPool             mem.BufferPool
+	StaticStreamWindowSize int32
+	StaticConnWindowSize   int32
 }
 
 // ConnectOptions covers all relevant options for communicating with the server.
@@ -503,7 +505,9 @@ type ConnectOptions struct {
 	// MaxHeaderListSize sets the max (uncompressed) size of header list that is prepared to be received.
 	MaxHeaderListSize *uint32
 	// The mem.BufferPool to use when reading/writing to the wire.
-	BufferPool mem.BufferPool
+	BufferPool             mem.BufferPool
+	StaticConnWindowSize   int32
+	StaticStreamWindowSize int32
 }
 
 // WriteOptions provides additional hints and information for message

server.go

@@ -151,34 +151,36 @@
 }
 
 type serverOptions struct {
-	creds                 credentials.TransportCredentials
-	codec                 baseCodec
-	cp                    Compressor
-	dc                    Decompressor
-	unaryInt              UnaryServerInterceptor
-	streamInt             StreamServerInterceptor
-	chainUnaryInts        []UnaryServerInterceptor
-	chainStreamInts       []StreamServerInterceptor
-	binaryLogger          binarylog.Logger
-	inTapHandle           tap.ServerInHandle
-	statsHandlers         []stats.Handler
-	maxConcurrentStreams  uint32
-	maxReceiveMessageSize int
-	maxSendMessageSize    int
-	unknownStreamDesc     *StreamDesc
-	keepaliveParams       keepalive.ServerParameters
-	keepalivePolicy       keepalive.EnforcementPolicy
-	initialWindowSize     int32
-	initialConnWindowSize int32
-	writeBufferSize       int
-	readBufferSize        int
-	sharedWriteBuffer     bool
-	connectionTimeout     time.Duration
-	maxHeaderListSize     *uint32
-	headerTableSize       *uint32
-	numServerWorkers      uint32
-	bufferPool            mem.BufferPool
-	waitForHandlers       bool
+	creds                  credentials.TransportCredentials
+	codec                  baseCodec
+	cp                     Compressor
+	dc                     Decompressor
+	unaryInt               UnaryServerInterceptor
+	streamInt              StreamServerInterceptor
+	chainUnaryInts         []UnaryServerInterceptor
+	chainStreamInts        []StreamServerInterceptor
+	binaryLogger           binarylog.Logger
+	inTapHandle            tap.ServerInHandle
+	statsHandlers          []stats.Handler
+	maxConcurrentStreams   uint32
+	maxReceiveMessageSize  int
+	maxSendMessageSize     int
+	unknownStreamDesc      *StreamDesc
+	keepaliveParams        keepalive.ServerParameters
+	keepalivePolicy        keepalive.EnforcementPolicy
+	initialWindowSize      int32
+	initialConnWindowSize  int32
+	writeBufferSize        int
+	readBufferSize         int
+	sharedWriteBuffer      bool
+	connectionTimeout      time.Duration
+	maxHeaderListSize      *uint32
+	headerTableSize        *uint32
+	numServerWorkers       uint32
+	bufferPool             mem.BufferPool
+	waitForHandlers        bool
+	staticStreamWindowSize int32
+	staticConnWindowSize   int32
 }
 
 var defaultServerOptions = serverOptions{
@@ -290,6 +292,22 @@
 	})
 }
 
+// StaticStreamWindowSize returns a ServerOption to set the static initial
+// stream window size.
+func StaticStreamWindowSize(s int32) ServerOption {
+	return newFuncServerOption(func(o *serverOptions) {
+		o.staticStreamWindowSize = s
+	})
+}
+
+// StaticConnWindowSize returns a ServerOption to set the static initial
+// connection window size.
+func StaticConnWindowSize(s int32) ServerOption {
+	return newFuncServerOption(func(o *serverOptions) {
+		o.staticConnWindowSize = s
+	})
+}
+
 // KeepaliveParams returns a ServerOption that sets keepalive and max-age parameters for the server.
 func KeepaliveParams(kp keepalive.ServerParameters) ServerOption {
 	if kp.Time > 0 && kp.Time < internal.KeepaliveMinServerPingTime {