Merge branch 'jv-sidechannel-window' into 'master'

sidechannel: use default yamux max window size

Closes #4132

See merge request gitlab-org/gitaly!4439

4 files changed, 33 insertions, 6 deletions

diff --git a/internal/backchannel/backchannel.go b/internal/backchannel/backchannel.go
index a85ab437e..e32494f63 100644
--- a/internal/backchannel/backchannel.go
+++ b/internal/backchannel/backchannel.go
@@ -40,7 +40,7 @@ import (
 var magicBytes = []byte("backchannel")
 
 // muxConfig returns a new config to use with the multiplexing session.
-func muxConfig(logger io.Writer) *yamux.Config {
+func muxConfig(logger io.Writer, extra func(*yamux.Config)) *yamux.Config {
 	cfg := yamux.DefaultConfig()
 	cfg.LogOutput = logger
 	// The server only accepts a single stream from the client, which is the client's gRPC stream.
@@ -55,6 +55,10 @@ func muxConfig(logger io.Writer) *yamux.Config {
 	// receiver. This is can have a big impact on throughput as the latency increases, as the sender
 	// can't proceed sending without receiving an acknowledgement back.
 	cfg.MaxStreamWindowSize = 16 * 1024 * 1024
+
+	if extra != nil {
+		extra(cfg)
+	}
 	return cfg
 }
 
diff --git a/internal/backchannel/client.go b/internal/backchannel/client.go
index c77157407..031cca58b 100644
--- a/internal/backchannel/client.go
+++ b/internal/backchannel/client.go
@@ -28,25 +28,36 @@ type ServerFactory func() Server
 type ClientHandshaker struct {
 	logger        *logrus.Entry
 	serverFactory ServerFactory
+	yamuxConfig   func(*yamux.Config)
 }
 
 // NewClientHandshaker returns a new client side implementation of the backchannel. The provided
 // logger is used to log multiplexing errors.
 func NewClientHandshaker(logger *logrus.Entry, serverFactory ServerFactory) ClientHandshaker {
-	return ClientHandshaker{logger: logger, serverFactory: serverFactory}
+	return NewClientHandshakerWithYamuxConfig(logger, serverFactory, nil)
+}
+
+// NewClientHandshakerWithYamuxConfig returns a new client side implementation of the
+// backchannel. The provided logger is used to log multiplexing errors.
+// For each connection that we accept, the yamuxConfig callback is
+// invoked to allow yamux configuration overrides. If yamuxConfig is nil
+// it is ignored.
+func NewClientHandshakerWithYamuxConfig(logger *logrus.Entry, serverFactory ServerFactory, yamuxConfig func(*yamux.Config)) ClientHandshaker {
+	return ClientHandshaker{logger: logger, serverFactory: serverFactory, yamuxConfig: yamuxConfig}
 }
 
 // ClientHandshake returns TransportCredentials that perform the client side multiplexing handshake and
 // start the backchannel Server on the established connections. The transport credentials are used to intiliaze the
 // connection prior to the multiplexing.
 func (ch ClientHandshaker) ClientHandshake(tc credentials.TransportCredentials) credentials.TransportCredentials {
-	return clientHandshake{TransportCredentials: tc, serverFactory: ch.serverFactory, logger: ch.logger}
+	return clientHandshake{TransportCredentials: tc, serverFactory: ch.serverFactory, logger: ch.logger, yamuxConfig: ch.yamuxConfig}
 }
 
 type clientHandshake struct {
 	credentials.TransportCredentials
 	serverFactory ServerFactory
 	logger        *logrus.Entry
+	yamuxConfig   func(*yamux.Config)
 }
 
 func (ch clientHandshake) ClientHandshake(ctx context.Context, serverName string, conn net.Conn) (net.Conn, credentials.AuthInfo, error) {
@@ -92,7 +103,7 @@ func (ch clientHandshake) serve(ctx context.Context, conn net.Conn) (net.Conn, e
 	logger := ch.logger.WriterLevel(logrus.ErrorLevel)
 
 	// Initiate the multiplexing session.
-	muxSession, err := yamux.Client(conn, muxConfig(logger))
+	muxSession, err := yamux.Client(conn, muxConfig(logger, ch.yamuxConfig))
 	if err != nil {
 		logger.Close()
 		return nil, fmt.Errorf("open multiplexing session: %w", err)
diff --git a/internal/backchannel/server.go b/internal/backchannel/server.go
index a6502f26d..b277b727a 100644
--- a/internal/backchannel/server.go
+++ b/internal/backchannel/server.go
@@ -100,7 +100,7 @@ func (s *ServerHandshaker) Handshake(conn net.Conn, authInfo credentials.AuthInf
 	logger := s.logger.WriterLevel(logrus.ErrorLevel)
 
 	// Open the server side of the multiplexing session.
-	muxSession, err := yamux.Server(conn, muxConfig(logger))
+	muxSession, err := yamux.Server(conn, muxConfig(logger, nil))
 	if err != nil {
 		logger.Close()
 		return nil, nil, fmt.Errorf("create multiplexing session: %w", err)
diff --git a/internal/sidechannel/sidechannel.go b/internal/sidechannel/sidechannel.go
index 74c7d0ab8..5cffe9efc 100644
--- a/internal/sidechannel/sidechannel.go
+++ b/internal/sidechannel/sidechannel.go
@@ -9,6 +9,7 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/hashicorp/yamux"
 	"github.com/sirupsen/logrus"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/backchannel"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/gitaly/client"
@@ -133,12 +134,23 @@ func NewServerHandshaker(registry *Registry) *ServerHandshaker {
 // NewClientHandshaker is used to enable sidechannel support on outbound
 // gRPC connections.
 func NewClientHandshaker(logger *logrus.Entry, registry *Registry) client.Handshaker {
-	return backchannel.NewClientHandshaker(
+	return backchannel.NewClientHandshakerWithYamuxConfig(
 		logger,
 		func() backchannel.Server {
 			lm := listenmux.New(insecure.NewCredentials())
 			lm.Register(NewServerHandshaker(registry))
 			return grpc.NewServer(grpc.Creds(lm))
 		},
+		func(cfg *yamux.Config) {
+			// Backchannel sets a very large custom window size (16MB). This is not
+			// necessary for sidechannels because we use one stream per connection.
+			// Worse, this is wasteful, because a client that is serving many
+			// concurrent sidechannel calls may end up lazily creating a 16MB buffer
+			// for each ongoing call. See
+			// https://gitlab.com/gitlab-org/gitaly/-/issues/4132. To prevent this
+			// waste we change this value back to 256KB which is the default and
+			// minimum value.
+			cfg.MaxStreamWindowSize = 256 * 1024
+		},
 	)
 }