transport: Pool read buffers used by the HTTP/2 framer (#9032)
## Problem
The HTTP/2 framer in gRPC uses a `bufio.Reader` with a 32KB buffer by
default. When there are a large number of transports, these buffers
consume significant memory, even when the transport is idle.
## Solution
#8964 added a `ReadyReader` interface that allows non-memory-pinning
reads. This PR replaces the standard `bufio.Reader` with a custom
`io.Reader` implementation that uses pooled buffers and releases the
buffer once all data is consumed.
To defer the re-allocation of the read buffer, the reader calls
`ReadOnReady` on the underlying `io.Reader`. For this to work, the
underlying `io.Reader` must implement either the `ReadyReader` interface
or `syscall.RawConn`. If neither condition is met, the framer gracefully
falls back to using the regular `bufio.Reader`.
Additional Changes:
* The ALTS connection has been refactored to implement the `ReadyReader`
interface.
* The write buffer pools used by the framer are updated to use the
`mem.BufferPool` interface, allowing the pools to be shared across both
read and write operations.
* Use `syscall.Read` instead of `unix.Read` to avoid triggering the race
detector, see comment for details.
* Add environment variable protection for the changes to allow fast
rollback.
## Benchmarks
In a [real-world
benchmark](https://github.com/arjan-bal/custom-go-client-benchmark/tree/retry-dp),
where a GCS directpath client downloads a file in a loop, the average
"in use" memory falls from 28.3MB to 21.3MB (-24%).
Local Benchmarks show no significant difference
```
❯ go run benchmark/benchresult/main.go streaming-before streaming-after
streaming-networkMode_Local-bufConn_false-keepalive_false-benchTime_2m0s-trace_false-latency_0s-kbps_0-MTU_0-maxConcurrentCa
lls_120-reqSize_1024B-respSize_1024B-compressor_off-channelz_false-preloader_false-clientReadBufferSize_-1-clientWriteBuffer
Size_-1-serverReadBufferSize_-1-serverWriteBufferSize_-1-sleepBetweenRPCs_0s-connections_1-recvBufferPool_simple-sharedWrite
Buffer_true
Title Before After Percentage
TotalOps 29981273 29966908 -0.05%
SendOps 0 0 NaN%
RecvOps 0 0 NaN%
Bytes/op 4971.06 4971.41 0.00%
Allocs/op 19.79 19.79 0.00%
ReqT/op 2046721570.13 2045740919.47 -0.05%
RespT/op 2046721570.13 2045740919.47 -0.05%
50th-Lat 461.523µs 460.906µs -0.13%
90th-Lat 654.435µs 655.327µs 0.14%
99th-Lat 1.225856ms 1.240984ms 1.23%
Avg-Lat 478.845µs 479.553µs 0.15%
GoVersion go1.25.0 go1.25.0
GrpcVersion 1.81.0-dev 1.81.0-dev
```
RELEASE NOTES:
* transport: Pool HTTP/2 framer read buffers to reduce idle memory
consumption. Currently limited to Linux for ALTS and non-encrypted
transports (TCP, Unix). To disable, set
`GRPC_GO_EXPERIMENTAL_HTTP_FRAMER_READ_BUFFER_POOLING=false` and report
any issues.diff --git a/credentials/alts/internal/conn/record.go b/credentials/alts/internal/conn/record.go
index 0ad5a1e..89bf226 100644
--- a/credentials/alts/internal/conn/record.go
+++ b/credentials/alts/internal/conn/record.go
@@ -27,8 +27,9 @@
"net"
core "google.golang.org/grpc/credentials/alts/internal"
- "google.golang.org/grpc/internal/mem"
+ imem "google.golang.org/grpc/internal/mem"
"google.golang.org/grpc/internal/transport/readyreader"
+ "google.golang.org/grpc/mem"
)
// ALTSRecordCrypto is the interface for gRPC ALTS record protocol.
@@ -75,16 +76,19 @@
var (
protocols = make(map[string]ALTSRecordFunc)
- writeBufPool *mem.BinaryTieredBufferPool
+ writeBufPool *imem.BinaryTieredBufferPool
// readBufPool pools buffers of at least `altsReadBufferInitialSize` size.
// Since the read buffer size is slightly larger than 32KB, using a regular
// BinaryTieredBufferPool results in allocating buffers of almost double the
// required length.
- readBufPool = mem.NewDirtySimplePool()
+ readBufPool = imem.NewDirtySimplePool()
+
+ // Compile-time check to ensure conn implements ReadyReader.
+ _ readyreader.Reader = &conn{}
)
func init() {
- pool, err := mem.NewDirtyBinaryTieredBufferPool(
+ pool, err := imem.NewDirtyBinaryTieredBufferPool(
8,
12, // Go page size, 4KB
14, // 16KB (max HTTP/2 frame size used by gRPC)
@@ -126,7 +130,8 @@
// nextFrame stores the next frame (in protected buffer) info.
nextFrame []byte
// overhead is the calculated overhead of each frame.
- overhead int
+ overhead int
+ constPool constBufferPool // stored as a field to avoid heap allocations.
}
// NewConn creates a new secure channel instance given the other party role and
@@ -163,11 +168,27 @@
return altsConn, nil
}
+type constBufferPool struct {
+ buffer []byte
+}
+
+func (p *constBufferPool) Get(int) *[]byte {
+ return &p.buffer
+}
+
+func (p *constBufferPool) Put(*[]byte) {}
+
// Read reads and decrypts a frame from the underlying connection, and copies the
// decrypted payload into b. If the size of the payload is greater than len(b),
// Read retains the remaining bytes in an internal buffer, and subsequent calls
// to Read will read from this buffer until it is exhausted.
func (p *conn) Read(b []byte) (n int, err error) {
+ p.constPool.buffer = b
+ _, n, err = p.ReadOnReady(len(b), &p.constPool)
+ return n, err
+}
+
+func (p *conn) ReadOnReady(bufSize int, pool mem.BufferPool) (*[]byte, int, error) {
if len(p.buf) == 0 {
var framedMsg []byte
var protected []byte
@@ -175,9 +196,10 @@
protected = *p.protectedHandle
protected = protected[:cap(protected)]
}
+ var err error
framedMsg, p.nextFrame, err = ParseFramedMsg(p.nextFrame, altsRecordLengthLimit)
if err != nil {
- return 0, err
+ return nil, 0, err
}
// Check whether the next frame to be decrypted has been
// completely received yet.
@@ -217,40 +239,42 @@
// Connection was idle, need to re-allocate the read buffer.
newBuf, nRead, err := p.reader.ReadOnReady(altsReadBufferInitialSize, readBufPool)
if err != nil {
- return 0, err
+ return nil, 0, err
}
p.protectedHandle = newBuf
protected = (*newBuf)[:nRead]
} else {
nRead, err := p.Conn.Read(protected[len(protected):cap(protected)])
if err != nil {
- return 0, err
+ return nil, 0, err
}
protected = protected[:len(protected)+nRead]
}
framedMsg, p.nextFrame, err = ParseFramedMsg(protected, altsRecordLengthLimit)
if err != nil {
- return 0, err
+ return nil, 0, err
}
}
// Now we have a complete frame, decrypted it.
msg := framedMsg[MsgLenFieldSize:]
msgType := binary.LittleEndian.Uint32(msg[:msgTypeFieldSize])
if msgType&0xff != altsRecordMsgType {
- return 0, fmt.Errorf("received frame with incorrect message type %v, expected lower byte %v",
+ return nil, 0, fmt.Errorf("received frame with incorrect message type %v, expected lower byte %v",
msgType, altsRecordMsgType)
}
ciphertext := msg[msgTypeFieldSize:]
// Decrypt directly into the buffer, avoiding a copy from p.buf if
// possible.
- if len(b) >= len(ciphertext) {
- dec, err := p.crypto.Decrypt(b[:0], ciphertext)
+ if bufSize >= len(ciphertext) {
+ allocatedBuf := pool.Get(bufSize)
+ dec, err := p.crypto.Decrypt((*allocatedBuf)[:0], ciphertext)
if err != nil {
- return 0, err
+ pool.Put(allocatedBuf)
+ return nil, 0, err
}
p.dropProtectedIfEmtpy()
- return len(dec), nil
+ return allocatedBuf, len(dec), nil
}
// Decrypt requires that if the dst and ciphertext alias, they
// must alias exactly. Code here used to use msg[:0], but msg
@@ -261,14 +285,15 @@
// check: https://golang.org/pkg/crypto/cipher/#AEAD.
p.buf, err = p.crypto.Decrypt(ciphertext[:0], ciphertext)
if err != nil {
- return 0, err
+ return nil, 0, err
}
}
- n = copy(b, p.buf)
+ allocatedBuf := pool.Get(bufSize)
+ n := copy(*allocatedBuf, p.buf)
p.buf = p.buf[n:]
p.dropProtectedIfEmtpy()
- return n, nil
+ return allocatedBuf, n, nil
}
func (p *conn) dropProtectedIfEmtpy() {
diff --git a/internal/envconfig/envconfig.go b/internal/envconfig/envconfig.go
index cf1b9a8..8b5b14b 100644
--- a/internal/envconfig/envconfig.go
+++ b/internal/envconfig/envconfig.go
@@ -142,6 +142,16 @@
//
// TODO: In release v1.82.0, env var will be enabled by default.
Enable8KBDefaultHeaderListSize = boolFromEnv("GRPC_GO_EXPERIMENTAL_ENABLE_8KB_DEFAULT_HEADER_LIST_SIZE", false)
+
+ // EnableHTTPFramerReadBufferPooling enables the use of the
+ // readyreader.Reader interface to perform non-memory-pinning reads,
+ // provided the underlying net.Conn supports it. This reduces memory usage
+ // when subchannels are idle.
+ //
+ // This environment variable serves as an escape hatch to disable the
+ // feature if unforeseen issues arise, and it will be removed in a future
+ // release.
+ EnableHTTPFramerReadBufferPooling = boolFromEnv("GRPC_GO_EXPERIMENTAL_HTTP_FRAMER_READ_BUFFER_POOLING", true)
)
func boolFromEnv(envVar string, def bool) bool {
diff --git a/internal/transport/http_util.go b/internal/transport/http_util.go
index 5bbb641..c34975f 100644
--- a/internal/transport/http_util.go
+++ b/internal/transport/http_util.go
@@ -36,6 +36,9 @@
"golang.org/x/net/http2"
"golang.org/x/net/http2/hpack"
"google.golang.org/grpc/codes"
+ "google.golang.org/grpc/internal/envconfig"
+ imem "google.golang.org/grpc/internal/mem"
+ "google.golang.org/grpc/internal/transport/readyreader"
"google.golang.org/grpc/mem"
)
@@ -296,7 +299,7 @@
}
type bufWriter struct {
- pool *sync.Pool
+ pool *imem.SimpleBufferPool
buf []byte
offset int
batchSize int
@@ -304,7 +307,7 @@
err error
}
-func newBufWriter(conn io.Writer, batchSize int, pool *sync.Pool) *bufWriter {
+func newBufWriter(conn io.Writer, batchSize int, pool *imem.SimpleBufferPool) *bufWriter {
w := &bufWriter{
batchSize: batchSize,
conn: conn,
@@ -326,7 +329,7 @@
return n, toIOError(err)
}
if w.buf == nil {
- b := w.pool.Get().(*[]byte)
+ b := w.pool.Get(w.batchSize)
w.buf = *b
}
written := 0
@@ -407,22 +410,32 @@
errDetail error
}
-var writeBufferPoolMap = make(map[int]*sync.Pool)
-var writeBufferMutex sync.Mutex
+var ioBufferPoolMap = make(map[int]*imem.SimpleBufferPool)
+var ioBufferMutex sync.Mutex
+
+func bufferedReader(r io.Reader, bufSize int) io.Reader {
+ if bufSize <= 0 {
+ return r
+ }
+ if envconfig.EnableHTTPFramerReadBufferPooling {
+ if rr := readyreader.NewNonBlocking(r); rr != nil {
+ readPool := ioBufferPool(bufSize)
+ return readyreader.NewBuffered(rr, bufSize, readPool)
+ }
+ }
+ return bufio.NewReaderSize(r, bufSize)
+}
func newFramer(conn io.ReadWriter, writeBufferSize, readBufferSize int, sharedWriteBuffer bool, maxHeaderListSize uint32, memPool mem.BufferPool) *framer {
if writeBufferSize < 0 {
writeBufferSize = 0
}
- var r io.Reader = conn
- if readBufferSize > 0 {
- r = bufio.NewReaderSize(r, readBufferSize)
- }
- var pool *sync.Pool
+ r := bufferedReader(conn, readBufferSize)
+ var writePool *imem.SimpleBufferPool
if sharedWriteBuffer {
- pool = getWriteBufferPool(writeBufferSize)
+ writePool = ioBufferPool(writeBufferSize)
}
- w := newBufWriter(conn, writeBufferSize, pool)
+ w := newBufWriter(conn, writeBufferSize, writePool)
f := &framer{
writer: w,
fr: http2.NewFramer(w, r),
@@ -578,20 +591,15 @@
return df.FrameHeader
}
-func getWriteBufferPool(size int) *sync.Pool {
- writeBufferMutex.Lock()
- defer writeBufferMutex.Unlock()
- pool, ok := writeBufferPoolMap[size]
+func ioBufferPool(size int) *imem.SimpleBufferPool {
+ ioBufferMutex.Lock()
+ defer ioBufferMutex.Unlock()
+ pool, ok := ioBufferPoolMap[size]
if ok {
return pool
}
- pool = &sync.Pool{
- New: func() any {
- b := make([]byte, size)
- return &b
- },
- }
- writeBufferPoolMap[size] = pool
+ pool = imem.NewDirtySimplePool()
+ ioBufferPoolMap[size] = pool
return pool
}
diff --git a/internal/transport/http_util_test.go b/internal/transport/http_util_test.go
index bda6d77..a16081e 100644
--- a/internal/transport/http_util_test.go
+++ b/internal/transport/http_util_test.go
@@ -19,6 +19,7 @@
package transport
import (
+ "bufio"
"bytes"
"errors"
"fmt"
@@ -31,6 +32,9 @@
"time"
"golang.org/x/net/http2"
+ "google.golang.org/grpc/internal/envconfig"
+ "google.golang.org/grpc/internal/testutils"
+ "google.golang.org/grpc/internal/transport/readyreader"
"google.golang.org/grpc/mem"
)
@@ -259,7 +263,7 @@
// Configure the bufWriter with a batchsize that results in data being flushed
// to the underlying conn, midway through Write().
writeBufferSize := (len(data) - 1) / 2
- writer := newBufWriter(&badNetworkConn{}, writeBufferSize, getWriteBufferPool(writeBufferSize))
+ writer := newBufWriter(&badNetworkConn{}, writeBufferSize, ioBufferPool(writeBufferSize))
errCh := make(chan error, 1)
go func() {
@@ -413,3 +417,74 @@
})
}
}
+
+type testReadyReader struct {
+ readyreader.Reader
+}
+
+func (t *testReadyReader) Read([]byte) (int, error) {
+ return 0, io.EOF
+}
+
+func (s) TestBufferedReader(t *testing.T) {
+ normalReader := bytes.NewReader(nil)
+
+ tests := []struct {
+ name string
+ reader io.Reader
+ bufSize int
+ enablePooling bool
+ wantTypeOf any
+ }{
+ {
+ name: "bufSize_0",
+ reader: normalReader,
+ bufSize: 0,
+ enablePooling: true,
+ wantTypeOf: (*bytes.Reader)(nil),
+ },
+ {
+ name: "env_var_disabled_normal_reader",
+ reader: normalReader,
+ bufSize: 10,
+ enablePooling: false,
+ wantTypeOf: (*bufio.Reader)(nil),
+ },
+ {
+ name: "env_var_disabled_ready_reader",
+ reader: &testReadyReader{},
+ bufSize: 10,
+ enablePooling: false,
+ wantTypeOf: (*bufio.Reader)(nil),
+ },
+ {
+ name: "env_var_enabled_normal_reader",
+ reader: normalReader,
+ bufSize: 10,
+ enablePooling: true,
+ wantTypeOf: (*bufio.Reader)(nil),
+ },
+ {
+ name: "env_var_enabled_ready_reader",
+ reader: &testReadyReader{},
+ bufSize: 10,
+ enablePooling: true,
+ wantTypeOf: readyreader.NewBuffered(nil, 10, mem.DefaultBufferPool()),
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnableHTTPFramerReadBufferPooling, tt.enablePooling)
+
+ got := bufferedReader(tt.reader, tt.bufSize)
+
+ gotType := reflect.TypeOf(got)
+ wantType := reflect.TypeOf(tt.wantTypeOf)
+
+ if gotType != wantType {
+ t.Errorf("bufferedReader() type = %v, want %v", gotType, wantType)
+ }
+ })
+ }
+}
diff --git a/internal/transport/readyreader/raw_conn_linux.go b/internal/transport/readyreader/raw_conn_linux.go
index a156e0a..56906c3 100644
--- a/internal/transport/readyreader/raw_conn_linux.go
+++ b/internal/transport/readyreader/raw_conn_linux.go
@@ -18,20 +18,22 @@
package readyreader
-import (
- "golang.org/x/sys/unix"
-)
+import "syscall"
func isRawConnSupported() bool {
return true
}
-// sysRead uses the modern unix package for Unix-like systems.
+// sysRead uses the standard syscall package rather than the modern unix package
+// to avoid triggering the race detector. Because both packages perform sync
+// operations on a local variable to satisfy the race detector, mixing them
+// for read and write syscalls causes data races. We use syscall here to remain
+// consistent with net.Conn implementations in standard library.
func sysRead(fd uintptr, p []byte) (int, error) {
- return unix.Read(int(fd), p)
+ return syscall.Read(int(fd), p)
}
// wouldBlock checks standard Unix non-blocking errors.
func wouldBlock(err error) bool {
- return err == unix.EAGAIN || err == unix.EWOULDBLOCK
+ return err == syscall.EAGAIN || err == syscall.EWOULDBLOCK
}
diff --git a/internal/transport/readyreader/ready_reader.go b/internal/transport/readyreader/ready_reader.go
index 05e63fa..250a300 100644
--- a/internal/transport/readyreader/ready_reader.go
+++ b/internal/transport/readyreader/ready_reader.go
@@ -63,9 +63,9 @@
buf *[]byte
}
-// newNonBlocking returns a ReadyReader if the passed reader supports
+// NewNonBlocking returns a ReadyReader if the passed reader supports
// non-memory-pinning reads, else nil.
-func newNonBlocking(r io.Reader) Reader {
+func NewNonBlocking(r io.Reader) Reader {
if rr, ok := r.(Reader); ok {
return rr
}
@@ -155,7 +155,7 @@
// If [syscall.RawConn] is unavailable, it falls back to using the simpler
// [io.Reader] interface for reads.
func New(r io.Reader) Reader {
- if r := newNonBlocking(r); r != nil {
+ if r := NewNonBlocking(r); r != nil {
return r
}
return &blockingReader{reader: r}
diff --git a/internal/transport/readyreader/ready_reader_ext_test.go b/internal/transport/readyreader/ready_reader_ext_test.go
index 310580c..cd686fd 100644
--- a/internal/transport/readyreader/ready_reader_ext_test.go
+++ b/internal/transport/readyreader/ready_reader_ext_test.go
@@ -123,7 +123,11 @@
}
}
-func (s) TestReadyReader_TCP_Blocking(t *testing.T) {
+// Tests the behavior of readers wrapping TCP connections. It ensures that read
+// operations do not hold onto an allocated buffer while the connection is idle.
+// Buffers should only be permanently allocated from the pool when data is
+// actually ready on the socket, optimizing memory usage for idle connections.
+func (s) TestReadyReader_TCP(t *testing.T) {
if runtime.GOOS != "linux" {
t.Skip("This test is only applicable for Linux, as RawConn functionality is not implemented for non-linux platforms.")
}
