grpc: fix regression by freeing request bufferslice after processing unary

[coxley]

Summary

I was excited to see recycled buffers be released (#6619), and wanted to gauge performance before putting into production. It was surprising to see that the default codec in v1.65.0 performed better than the new one in v1.66.0.

There was a missed (mem.BufferSlice).Free() in the server during processUnaryRPC, leaving the incoming request bytes hanging around while still paying the added allocations that comes with managing the buffers.

I'm not super attached to the added BufferSlice docs, but it would have saved me a lot of time trying to figure out what was wrong with my interaction code. It wasn't clear to me that sometimes the provided slice kept a reference to the buffer, and would be released by SDK logic.

Testing

This is far from a pure benchmark (bufconn, client, etc contributing to allocs). However, it demonstrates the regression and fix.

Results

# Regression
> go test -bench=. -benchmem
goos: darwin
goarch: arm64
pkg: scratch
cpu: Apple M1 Pro
BenchmarkCodec/v1.66.0/sm-10               55629             20031 ns/op            9723 B/op        188 allocs/op
BenchmarkCodec/v1.66.0/md-10               52221             22292 ns/op           16907 B/op        187 allocs/op
BenchmarkCodec/v1.66.0/lg-10               29656             40643 ns/op           76818 B/op        198 allocs/op
PASS
ok      scratch 4.972s

# Baseline
> go test -bench=. -benchmem
goos: darwin
goarch: arm64
pkg: scratch
cpu: Apple M1 Pro
BenchmarkCodec/v1.65.0/sm-10               55014             20053 ns/op            9550 B/op        183 allocs/op
BenchmarkCodec/v1.65.0/md-10               52394             22663 ns/op           19042 B/op        183 allocs/op
BenchmarkCodec/v1.65.0/lg-10               28153             42928 ns/op          109144 B/op        189 allocs/op
PASS
ok      scratch 4.998s

# This PR
➜ go test -bench=. -benchmem
goos: darwin
goarch: arm64
pkg: scratch
cpu: Apple M1 Pro
BenchmarkCodec/fix/sm-10           55530             19899 ns/op            9724 B/op        188 allocs/op
BenchmarkCodec/fix/md-10           55821             21535 ns/op           12742 B/op        183 allocs/op
BenchmarkCodec/fix/lg-10           32937             38130 ns/op           43673 B/op        190 allocs/op
PASS
ok      scratch 5.058s

Code:

package main

import (
	"context"
	"net"
	"os"
	"regexp"
	"strings"
	"testing"

	"github.com/stretchr/testify/require"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	_ "google.golang.org/grpc/encoding/proto"
	health "google.golang.org/grpc/health/grpc_health_v1"
	"google.golang.org/grpc/test/bufconn"
)

var (
	small  = &health.HealthCheckRequest{Service: "Foo Bar"}
	medium = &health.HealthCheckRequest{Service: strings.Repeat(".", 3<<10)}
	large  = &health.HealthCheckRequest{Service: strings.Repeat(".", 31<<10)}
	msgs   = []*health.HealthCheckRequest{small, medium, large}
)

// hacky way to show the grpc version in the benchmark
func grpcVersion(t testing.TB) string {
	fb, err := os.ReadFile("go.mod")
	require.NoError(t, err)

	versionRE := regexp.MustCompile(`google.golang.org/grpc (\S*)`)
	matches := versionRE.FindStringSubmatch(string(fb))
	require.Len(t, matches, 2)

	replaceRE := regexp.MustCompile(`replace google.golang.org/grpc`)
	if replaceRE.Match(fb) {
		return "fix"
	}
	return matches[1]
}

func BenchmarkCodec(b *testing.B) {
	ctx := context.Background()
	client := testClient(b)

	names := []string{"sm", "md", "lg"}
	b.Run(grpcVersion(b), func(b *testing.B) {
		for i, msg := range msgs {
			b.Run(names[i], func(b *testing.B) {
				for range b.N {
					_, err := client.Check(ctx, msg)
					require.NoError(b, err)
				}
			})
		}
	})
}

func testClient(t testing.TB) health.HealthClient {
	t.Helper()

	srv := grpc.NewServer()
	handler := new(handler)
	health.RegisterHealthServer(srv, handler)

	lis := bufconn.Listen(1 << 20)
	go func() {
		require.NoError(t, srv.Serve(lis))
	}()

	t.Cleanup(func() {
		srv.GracefulStop()
	})

	conn, err := grpc.NewClient(
		"passthrough://bufnet",
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithContextDialer(func(ctx context.Context, s string) (net.Conn, error) {
			return lis.Dial()
		}),
	)
	require.NoError(t, err)

	healthClient := health.NewHealthClient(conn)
	return healthClient
}

type handler struct {
	health.UnimplementedHealthServer
}

func (handler) Check(context.Context, *health.HealthCheckRequest) (*health.HealthCheckResponse, error) {
	return new(health.HealthCheckResponse), nil
}

RELEASE NOTES:

• Fix minor server-side regression when pooling large (>1KiB) incoming request buffers

[linux-foundation-easycla]

The committers listed above are authorized under a signed CLA.

• ✅ login: coxley / name: Codey Oxley (0941401, 6deaf1e)

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 81.76%. Comparing base (55d820d) to head (6deaf1e).
Report is 3 commits behind head on master.

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #7571      +/-   ##
==========================================
- Coverage   81.92%   81.76%   -0.16%     
==========================================
  Files         361      361              
  Lines       27825    27826       +1     
==========================================
- Hits        22796    22753      -43     
- Misses       3840     3868      +28     
- Partials     1189     1205      +16     

Files with missing lines	Coverage Δ
mem/buffer_slice.go	96.20% <ø> (ø)
server.go	82.34% <100.00%> (+0.11%)	⬆️

... and 20 files with indirect coverage changes

[dfawley]

@PapaCharlie would you mind taking a quick look at this?

[PapaCharlie]

Yeah looking now. The tests are passing so I'm not too concerned there, but I'm surprised I missed this. This isn't restricted to unary RPCs, so it may need fixes elsewhere!

[PapaCharlie]

Nevermind, this is the equivalent for stream RPCs: https://github.com/grpc/grpc-go/blob/master/rpc_util.go#L923

Nice catch! I just have a tiny nit, for consistency with the other places defer Free is called

[PapaCharlie]

Would you mind moving this up near to where d is initialized, at line 1361?

	d, err := recvAndDecompress(&parser{r: stream, bufferPool: s.opts.bufferPool}, stream, dc, s.opts.maxReceiveMessageSize, payInfo, decomp, true)
	if err != nil {
		if e := t.WriteStatus(stream, status.Convert(err)); e != nil {
			channelz.Warningf(logger, s.channelz, "grpc: Server.processUnaryRPC failed to write status: %v", e)
		}
		return err
	}
	defer d.Free()

Just for consistency

[coxley]

Whoops, of course.

Done!

[dfawley]

Let's do a patch release of 1.66 with this fix when it's ready.

[PapaCharlie]

LGTM!

[dfawley]

Thanks for the fix!

[sluongng]

Will there be a v1.66.1 release with this fix included?

[easwars]

Yes, we are planning on doing that very soon.