Reduce gRPC message sends in PostUploadPackjv-reduce-writes

On gitlab.com, PostUploadPack has to transfer a lot of data. Git
upload-pack does writes of 8KB and by default, each of these writes
would be forwarded as a gRPC message send. Although grpc-go may
coalesce sends into larger (32KB) socket writes, testing suggests it
often does not do this so you end up with an unnecessarily high number
of write syscalls.

1 files changed, 24 insertions, 4 deletions

diff --git a/internal/gitaly/service/smarthttp/upload_pack.go b/internal/gitaly/service/smarthttp/upload_pack.go
index 57b5ee501..d60dee70a 100644
--- a/internal/gitaly/service/smarthttp/upload_pack.go
+++ b/internal/gitaly/service/smarthttp/upload_pack.go
@@ -1,6 +1,7 @@
 package smarthttp
 
 import (
+	"bufio"
 	"context"
 	"crypto/sha1"
 	"fmt"
@@ -11,6 +12,7 @@ import (
 	"gitlab.com/gitlab-org/gitaly/internal/git"
 	"gitlab.com/gitlab-org/gitaly/internal/git/stats"
 	"gitlab.com/gitlab-org/gitaly/internal/gitaly/service/inspect"
+	"gitlab.com/gitlab-org/gitaly/internal/helper"
 	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
 	"gitlab.com/gitlab-org/gitaly/streamio"
 	"google.golang.org/grpc/codes"
@@ -42,10 +44,24 @@ func (s *server) PostUploadPack(stream gitalypb.SmartHTTPService_PostUploadPackS
 
 	var respBytes int64
 
-	stdoutWriter := streamio.NewWriter(func(p []byte) error {
-		respBytes += int64(len(p))
-		return stream.Send(&gitalypb.PostUploadPackResponse{Data: p})
-	})
+	stdoutWriter := helper.NewUnbufferedStartWriter(
+		bufio.NewWriterSize(
+			streamio.NewWriter(func(p []byte) error {
+				respBytes += int64(len(p))
+				return stream.Send(&gitalypb.PostUploadPackResponse{Data: p})
+			}),
+			streamio.WriteBufferSize,
+		),
+		// Git's progress messages "Enumerating objects" etc act as keepalives so
+		// they should not be delayed by buffering. This number of bytes should
+		// be large enough to hold the combined progress messages.
+		32*1024,
+	)
+	defer func() {
+		// In case of an early return, the output stream may contain messages for
+		// the user so we should still flush it.
+		_ = stdoutWriter.Flush()
+	}()
 
 	// TODO: it is first step of the https://gitlab.com/gitlab-org/gitaly/issues/1519
 	// needs to be removed after we get some statistics on this
@@ -95,6 +111,10 @@ func (s *server) PostUploadPack(stream gitalypb.SmartHTTPService_PostUploadPackS
 		return status.Errorf(codes.Unavailable, "PostUploadPack: %v", err)
 	}
 
+	if err := stdoutWriter.Flush(); err != nil {
+		return status.Errorf(codes.Unavailable, "PostUploadPack: %v", err)
+	}
+
 	ctxlogrus.Extract(ctx).WithField("request_sha", fmt.Sprintf("%x", h.Sum(nil))).WithField("response_bytes", respBytes).Info("request details")
 
 	return nil