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

dialoptions.go

@@ -213,6 +213,7 @@
 func WithInitialWindowSize(s int32) DialOption {
 	return newFuncDialOption(func(o *dialOptions) {
 		o.copts.InitialWindowSize = s
+		o.copts.UseDynamicWindowSizing = false
 	})
 }
 
@@ -222,6 +223,26 @@
 func WithInitialConnWindowSize(s int32) DialOption {
 	return newFuncDialOption(func(o *dialOptions) {
 		o.copts.InitialConnWindowSize = s
+		o.copts.UseDynamicWindowSizing = false
+	})
+}
+
+// WithStaticStreamWindowSize returns a DialOption which sets the initial
+// stream window size to the value provided and disables dynamic flow control.
+func WithStaticStreamWindowSize(s int32) DialOption {
+	return newFuncDialOption(func(o *dialOptions) {
+		o.copts.InitialWindowSize = s
+		o.copts.UseDynamicWindowSizing = false
+	})
+}
+
+// WithStaticConnWindowSize returns a DialOption which sets the initial
+// connection window size to the value provided and disables dynamic flow
+// control.
+func WithStaticConnWindowSize(s int32) DialOption {
+	return newFuncDialOption(func(o *dialOptions) {
+		o.copts.InitialConnWindowSize = s
+		o.copts.UseDynamicWindowSizing = false
 	})
 }
 

internal/transport/http2_client.go

@@ -313,7 +313,7 @@ func NewHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
 	icwz := int32(initialWindowSize)
 	if opts.InitialConnWindowSize >= defaultWindowSize {
 		icwz = opts.InitialConnWindowSize
-		dynamicWindow = false
+		dynamicWindow = opts.UseDynamicWindowSizing
 	}
 	writeBufSize := opts.WriteBufferSize
 	readBufSize := opts.ReadBufferSize
@@ -381,7 +381,7 @@ func NewHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
 	t.controlBuf = newControlBuffer(t.ctxDone)
 	if opts.InitialWindowSize >= defaultWindowSize {
 		t.initialWindowSize = opts.InitialWindowSize
-		dynamicWindow = false
+		dynamicWindow = opts.UseDynamicWindowSizing
 	}
 	if dynamicWindow {
 		t.bdpEst = &bdpEstimator{

internal/transport/http2_server.go

@@ -180,12 +180,12 @@ func NewServerTransport(conn net.Conn, config *ServerConfig) (_ ServerTransport,
 	iwz := int32(initialWindowSize)
 	if config.InitialWindowSize >= defaultWindowSize {
 		iwz = config.InitialWindowSize
-		dynamicWindow = false
+		dynamicWindow = config.UseDynamicWindowSizing
 	}
 	icwz := int32(initialWindowSize)
 	if config.InitialConnWindowSize >= defaultWindowSize {
 		icwz = config.InitialConnWindowSize
-		dynamicWindow = false
+		dynamicWindow = config.UseDynamicWindowSizing
 	}
 	if iwz != defaultWindowSize {
 		isettings = append(isettings, http2.Setting{

internal/transport/transport.go

@@ -450,22 +450,23 @@ 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
+	UseDynamicWindowSizing bool
 }
 
 // ConnectOptions covers all relevant options for communicating with the server.
@@ -504,6 +505,8 @@ type ConnectOptions struct {
 	MaxHeaderListSize *uint32
 	// The mem.BufferPool to use when reading/writing to the wire.
 	BufferPool mem.BufferPool
+	// UseDynamicWindowSizing controls whether dynamic window sizing is enabled.
+	UseDynamicWindowSizing bool
 }
 
 // WriteOptions provides additional hints and information for message

server.go

@@ -151,34 +151,35 @@
 }
 
 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
+	useDynamicWindowSizing bool
 }
 
 var defaultServerOptions = serverOptions{
@@ -279,6 +280,7 @@
 func InitialWindowSize(s int32) ServerOption {
 	return newFuncServerOption(func(o *serverOptions) {
 		o.initialWindowSize = s
+		o.useDynamicWindowSizing = false
 	})
 }
 
@@ -287,6 +289,29 @@
 func InitialConnWindowSize(s int32) ServerOption {
 	return newFuncServerOption(func(o *serverOptions) {
 		o.initialConnWindowSize = s
+		o.useDynamicWindowSizing = false
+	})
+}
+
+// StaticStreamWindowSize returns a ServerOption to set the initial stream
+// window size to the value provided and disables dynamic flow control.
+// The lower bound for window size is 64K and any value smaller than that
+// will be ignored.
+func StaticStreamWindowSize(s int32) ServerOption {
+	return newFuncServerOption(func(o *serverOptions) {
+		o.initialWindowSize = s
+		o.useDynamicWindowSizing = false
+	})
+}
+
+// StaticConnWindowSize returns a ServerOption to set the initial connection
+// window size to the value provided and disables dynamic flow control.
+// The lower bound for window size is 64K and any value smaller than that
+// will be ignored.
+func StaticConnWindowSize(s int32) ServerOption {
+	return newFuncServerOption(func(o *serverOptions) {
+		o.initialConnWindowSize = s
+		o.useDynamicWindowSizing = false
 	})
 }
 
@@ -970,22 +995,23 @@
 // gRPC http2 server transport in transport/http2_server.go).
 func (s *Server) newHTTP2Transport(c net.Conn) transport.ServerTransport {
 	config := &transport.ServerConfig{
-		MaxStreams:            s.opts.maxConcurrentStreams,
-		ConnectionTimeout:     s.opts.connectionTimeout,
-		Credentials:           s.opts.creds,
-		InTapHandle:           s.opts.inTapHandle,
-		StatsHandlers:         s.opts.statsHandlers,
-		KeepaliveParams:       s.opts.keepaliveParams,
-		KeepalivePolicy:       s.opts.keepalivePolicy,
-		InitialWindowSize:     s.opts.initialWindowSize,
-		InitialConnWindowSize: s.opts.initialConnWindowSize,
-		WriteBufferSize:       s.opts.writeBufferSize,
-		ReadBufferSize:        s.opts.readBufferSize,
-		SharedWriteBuffer:     s.opts.sharedWriteBuffer,
-		ChannelzParent:        s.channelz,
-		MaxHeaderListSize:     s.opts.maxHeaderListSize,
-		HeaderTableSize:       s.opts.headerTableSize,
-		BufferPool:            s.opts.bufferPool,
+		MaxStreams:             s.opts.maxConcurrentStreams,
+		ConnectionTimeout:      s.opts.connectionTimeout,
+		Credentials:            s.opts.creds,
+		InTapHandle:            s.opts.inTapHandle,
+		StatsHandlers:          s.opts.statsHandlers,
+		KeepaliveParams:        s.opts.keepaliveParams,
+		KeepalivePolicy:        s.opts.keepalivePolicy,
+		InitialWindowSize:      s.opts.initialWindowSize,
+		InitialConnWindowSize:  s.opts.initialConnWindowSize,
+		WriteBufferSize:        s.opts.writeBufferSize,
+		ReadBufferSize:         s.opts.readBufferSize,
+		SharedWriteBuffer:      s.opts.sharedWriteBuffer,
+		ChannelzParent:         s.channelz,
+		MaxHeaderListSize:      s.opts.maxHeaderListSize,
+		HeaderTableSize:        s.opts.headerTableSize,
+		BufferPool:             s.opts.bufferPool,
+		UseDynamicWindowSizing: s.opts.useDynamicWindowSizing,
 	}
 	st, err := transport.NewServerTransport(c, config)
 	if err != nil {