feat(storagenode): add configurations for initial window size

This PR adds new gRPC initial flow control window size options to the storage node.

- `--server-initial-conn-window-size`
- `--server-initial-stream-window-size`

See:
- https://httpwg.org/specs/rfc7540.html#InitialWindowSize
- https://pkg.go.dev/google.golang.org/grpc#InitialConnWindowSize

bin/start_varlogsn.py

@@ -179,6 +179,12 @@ def start(args: argparse.Namespace) -> None:
             if args.server_write_buffer_size:
                 cmd.append(
                     f"--server-write-buffer-size={args.server_write_buffer_size}")
+            if args.server_initial_conn_window_size:
+                cmd.append(
+                        f"--server-initial-conn-window-size={args.server_initial_conn_window_size}")
+            if args.server_initial_stream_window_size:
+                cmd.append(
+                        f"--server-initial-stream-window-size={args.server_initial_stream_window_size}")
             if args.replication_client_read_buffer_size:
                 cmd.append(
                     f"--replication-client-read-buffer-size={args.replication_client_read_buffer_size}")
@@ -266,6 +272,8 @@ def main() -> None:
     # grpc options
     parser.add_argument("--server-read-buffer-size", type=str)
     parser.add_argument("--server-write-buffer-size", type=str)
+    parser.add_argument("--server-initial-conn-window-size", type=str)
+    parser.add_argument("--server-initial-stream-window-size", type=str)
     parser.add_argument("--replication-client-read-buffer-size", type=str)
     parser.add_argument("--replication-client-write-buffer-size", type=str)
 

cmd/varlogsn/cli.go

@@ -48,6 +48,8 @@ func newStartCommand() *cli.Command {
 			flagServerMaxRecvMsgSize.StringFlag(false, units.ToByteSizeString(storagenode.DefaultServerMaxRecvSize)),
 			flagReplicationClientReadBufferSize.StringFlag(false, units.ToByteSizeString(storagenode.DefaultReplicateClientReadBufferSize)),
 			flagReplicationClientWriteBufferSize.StringFlag(false, units.ToByteSizeString(storagenode.DefaultReplicateClientWriteBufferSize)),
+			flagServerInitialConnWindowSize,
+			flagServerInitialStreamWindowSize,
 
 			// lse options
 			flagLogStreamExecutorSequenceQueueCapacity.IntFlag(false, logstream.DefaultSequenceQueueCapacity),

cmd/varlogsn/flags.go

@@ -56,6 +56,16 @@ var (
 		Envs:    []string{"SERVER_MAX_MSG_SIZE", "SERVER_MAX_MESSAGE_SIZE"},
 		Usage:   "B, KiB, MiB, GiB",
 	}
+	flagServerInitialConnWindowSize = &cli.StringFlag{
+		Name:    "server-initial-conn-window-size",
+		EnvVars: []string{"SERVER_INITIAL_CONN_WINDOW_SIZE"},
+		Usage:   "Window size for a connection.",
+	}
+	flagServerInitialStreamWindowSize = &cli.StringFlag{
+		Name:    "server-initial-stream-window-size",
+		EnvVars: []string{"SERVER_INITIAL_STREAM_WINDOW_SIZE"},
+		Usage:   "Window size for stream.",
+	}
 	flagReplicationClientReadBufferSize = flags.FlagDesc{
 		Name:  "replication-client-read-buffer-size",
 		Envs:  []string{"REPLICATION_CLIENT_READ_BUFFER_SIZE"},

cmd/varlogsn/varlogsn.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math"
 	"os"
 	"os/signal"
 	"strings"
@@ -117,7 +118,7 @@ func start(c *cli.Context) error {
 		return err
 	}
 
-	sn, err := storagenode.NewStorageNode(
+	snOpts := []storagenode.Option{
 		storagenode.WithClusterID(clusterID),
 		storagenode.WithStorageNodeID(storageNodeID),
 		storagenode.WithListenAddress(c.String(flagListen.Name)),
@@ -138,7 +139,23 @@ func start(c *cli.Context) error {
 		storagenode.WithMaxLogStreamReplicasCount(int32(c.Int(flagMaxLogStreamReplicasCount.Name))),
 		storagenode.WithDefaultStorageOptions(storageOpts...),
 		storagenode.WithLogger(logger),
-	)
+	}
+	if initialConnWindowSize := c.String(flagServerInitialConnWindowSize.Name); initialConnWindowSize != "" {
+		size, err := units.FromByteSizeString(initialConnWindowSize, 0, math.MaxInt32)
+		if err != nil {
+			return err
+		}
+		snOpts = append(snOpts, storagenode.WithGRPCServerInitialConnWindowSize(int32(size)))
+	}
+	if initialStreamWindowSize := c.String(flagServerInitialStreamWindowSize.Name); initialStreamWindowSize != "" {
+		size, err := units.FromByteSizeString(initialStreamWindowSize, 0, math.MaxInt32)
+		if err != nil {
+			return err
+		}
+		snOpts = append(snOpts, storagenode.WithGRPCServerInitialWindowSize(int32(size)))
+	}
+
+	sn, err := storagenode.NewStorageNode(snOpts...)
 	if err != nil {
 		return err
 	}

internal/storagenode/config.go

@@ -33,6 +33,14 @@ type config struct {
 	grpcServerReadBufferSize        int64
 	grpcServerWriteBufferSize       int64
 	grpcServerMaxRecvMsgSize        int64
+	grpcServerInitialConnWindowSize struct {
+		value int32
+		set   bool
+	}
+	grpcServerInitialWindowSize struct {
+		value int32
+		set   bool
+	}
 	replicateClientReadBufferSize   int64
 	replicateClientWriteBufferSize  int64
 	maxLogStreamReplicasCount       int32
@@ -168,6 +176,20 @@ func WithGRPCServerMaxRecvMsgSize(grpcServerMaxRecvMsgSize int64) Option {
 	})
 }
 
+func WithGRPCServerInitialConnWindowSize(grpcServerInitialConnWindowSize int32) Option {
+	return newFuncOption(func(cfg *config) {
+		cfg.grpcServerInitialConnWindowSize.value = grpcServerInitialConnWindowSize
+		cfg.grpcServerInitialConnWindowSize.set = true
+	})
+}
+
+func WithGRPCServerInitialWindowSize(grpcServerInitialWindowSize int32) Option {
+	return newFuncOption(func(cfg *config) {
+		cfg.grpcServerInitialWindowSize.value = grpcServerInitialWindowSize
+		cfg.grpcServerInitialWindowSize.set = true
+	})
+}
+
 func WithReplicateClientReadBufferSize(replicateClientReadBufferSize int64) Option {
 	return newFuncOption(func(cfg *config) {
 		cfg.replicateClientReadBufferSize = replicateClientReadBufferSize

internal/storagenode/storagenode.go

@@ -98,7 +98,7 @@ func NewStorageNode(opts ...Option) (*StorageNode, error) {
 	}
 	dataDirs = filterValidDataDirectories(dataDirs, cfg.cid, cfg.snid, cfg.logger)
 
-	grpcServer := grpc.NewServer(
+	grpcServerOpts := []grpc.ServerOption{
 		grpc.ReadBufferSize(int(cfg.grpcServerReadBufferSize)),
 		grpc.WriteBufferSize(int(cfg.grpcServerWriteBufferSize)),
 		grpc.MaxRecvMsgSize(int(cfg.grpcServerMaxRecvMsgSize)),
@@ -118,12 +118,18 @@ func NewStorageNode(opts ...Option) (*StorageNode, error) {
 				return resp, err
 			},
 		),
-	)
+	}
+	if opt := cfg.grpcServerInitialConnWindowSize; opt.set {
+		grpcServerOpts = append(grpcServerOpts, grpc.InitialConnWindowSize(opt.value))
+	}
+	if opt := cfg.grpcServerInitialWindowSize; opt.set {
+		grpcServerOpts = append(grpcServerOpts, grpc.InitialWindowSize(opt.value))
+	}
 
 	sn := &StorageNode{
 		config:       cfg,
 		executors:    executorsmap.New(hintNumExecutors),
-		server:       grpcServer,
+		server:       grpc.NewServer(grpcServerOpts...),
 		healthServer: health.NewServer(),
 		closedC:      make(chan struct{}),
 		snPaths:      snPaths,
@@ -198,6 +204,8 @@ func (sn *StorageNode) Serve() error {
 		zap.Int64("grpcServerReadBufferSize", sn.grpcServerReadBufferSize),
 		zap.Int64("grpcServerWriteBufferSize", sn.grpcServerWriteBufferSize),
 		zap.Int64("grpcServerMaxRecvMsgSize", sn.grpcServerMaxRecvMsgSize),
+		zap.Int32("grpcServerInitialConnWindowSize", sn.grpcServerInitialConnWindowSize.value),
+		zap.Int32("grpcServerInitialStreamWindowSize", sn.grpcServerInitialWindowSize.value),
 		zap.Int64("grpcReplicateClientReadBufferSize", sn.replicateClientReadBufferSize),
 		zap.Int64("grpcReplicateClientWriteBufferSize", sn.replicateClientWriteBufferSize),
 	)