transport: Replace closures with interfaces to avoid heap allocations (#8630)

In Go, creating a closure results in a heap allocation if the compiler
determines the closure might outlive the function in which it was
created. This change removes two such closures, replacing them with
interfaces that are implemented by the `ClientStream` and `ServerStream`
structs.

While this pattern may slightly reduce readability, the performance
benefit is worthwhile, as this transport code is executed for every new
stream. This reduces allocs/unary RPC by 2.5%.

## Testing
```sh

# test command
 go run benchmark/benchmain/main.go -benchtime=60s -workloads=unary \
   -compression=off -maxConcurrentCalls=500 -trace=off \
   -reqSizeBytes=100 -respSizeBytes=100 -networkMode=Local -resultFile="${RUN_NAME}"   -recvBufferPool=simple

# results
go run benchmark/benchresult/main.go unary-before unary-after       
               Title       Before        After Percentage
            TotalOps      7593738      7708364     1.51%
             SendOps            0            0      NaN%
             RecvOps            0            0      NaN%
            Bytes/op     10218.45     10185.84    -0.32%
           Allocs/op       164.85       160.84    -2.43%
             ReqT/op 101249840.00 102778186.67     1.51%
            RespT/op 101249840.00 102778186.67     1.51%
            50th-Lat   3.617561ms   3.568623ms    -1.35%
            90th-Lat   5.218682ms   5.131828ms    -1.66%
            99th-Lat   6.052632ms   5.950261ms    -1.69%
             Avg-Lat   3.948414ms   3.889006ms    -1.50%
           GoVersion     go1.24.4     go1.24.4
         GrpcVersion   1.77.0-dev   1.77.0-dev

```

RELEASE NOTES: N/A

Author: arjan-bal

internal/transport/client_stream.go

@@ -144,3 +144,11 @@ func (s *ClientStream) TrailersOnly() bool {
 func (s *ClientStream) Status() *status.Status {
 	return s.status
 }
+
+func (s *ClientStream) requestRead(n int) {
+	s.ct.adjustWindow(s, uint32(n))
+}
+
+func (s *ClientStream) updateWindow(n int) {
+	s.ct.updateWindow(s, uint32(n))
+}

internal/transport/handler_server.go

@@ -387,6 +387,12 @@ func (ht *serverHandlerTransport) writeHeader(s *ServerStream, md metadata.MD) e
 	return err
 }
 
+func (ht *serverHandlerTransport) adjustWindow(*ServerStream, uint32) {
+}
+
+func (ht *serverHandlerTransport) updateWindow(*ServerStream, uint32) {
+}
+
 func (ht *serverHandlerTransport) HandleStreams(ctx context.Context, startStream func(*ServerStream)) {
 	// With this transport type there will be exactly 1 stream: this HTTP request.
 	var cancel context.CancelFunc
@@ -414,7 +420,6 @@ func (ht *serverHandlerTransport) HandleStreams(ctx context.Context, startStream
 		Stream: Stream{
 			id:             0, // irrelevant
 			ctx:            ctx,
-			requestRead:    func(int) {},
 			method:         req.URL.Path,
 			recvCompress:   req.Header.Get("grpc-encoding"),
 			contentSubtype: ht.contentSubtype,
@@ -424,9 +429,10 @@ func (ht *serverHandlerTransport) HandleStreams(ctx context.Context, startStream
 		headerWireLength: 0, // won't have access to header wire length until golang/go#18997.
 	}
 	s.Stream.buf.init()
+	s.readRequester = s
 	s.trReader = transportReader{
 		reader:        recvBufferReader{ctx: s.ctx, ctxDone: s.ctx.Done(), recv: &s.buf},
-		windowHandler: func(int) {},
+		windowHandler: s,
 	}
 
 	// readerDone is closed when the Body.Read-ing goroutine exits.

internal/transport/http2_client.go

@@ -493,9 +493,7 @@ func (t *http2Client) newStream(ctx context.Context, callHdr *CallHdr) *ClientSt
 	}
 	s.Stream.buf.init()
 	s.Stream.wq.init(defaultWriteQuota, s.done)
-	s.requestRead = func(n int) {
-		t.adjustWindow(s, uint32(n))
-	}
+	s.readRequester = s
 	// The client side stream context should have exactly the same life cycle with the user provided context.
 	// That means, s.ctx should be read-only. And s.ctx is done iff ctx is done.
 	// So we use the original context here instead of creating a copy.
@@ -509,9 +507,7 @@ func (t *http2Client) newStream(ctx context.Context, callHdr *CallHdr) *ClientSt
 				s.Close(err)
 			},
 		},
-		windowHandler: func(n int) {
-			t.updateWindow(s, uint32(n))
-		},
+		windowHandler: s,
 	}
 	return s
 }

internal/transport/http2_server.go

@@ -640,9 +640,7 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 		t.channelz.SocketMetrics.StreamsStarted.Add(1)
 		t.channelz.SocketMetrics.LastRemoteStreamCreatedTimestamp.Store(time.Now().UnixNano())
 	}
-	s.requestRead = func(n int) {
-		t.adjustWindow(s, uint32(n))
-	}
+	s.readRequester = s
 	s.ctxDone = s.ctx.Done()
 	s.Stream.wq.init(defaultWriteQuota, s.ctxDone)
 	s.trReader = transportReader{
@@ -651,9 +649,7 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 			ctxDone: s.ctxDone,
 			recv:    &s.buf,
 		},
-		windowHandler: func(n int) {
-			t.updateWindow(s, uint32(n))
-		},
+		windowHandler: s,
 	}
 	// Register the stream with loopy.
 	t.controlBuf.put(&registerStream{

internal/transport/server_stream.go

@@ -179,3 +179,11 @@ func (s *ServerStream) SetTrailer(md metadata.MD) error {
 	s.hdrMu.Unlock()
 	return nil
 }
+
+func (s *ServerStream) requestRead(n int) {
+	s.st.adjustWindow(s, uint32(n))
+}
+
+func (s *ServerStream) updateWindow(n int) {
+	s.st.updateWindow(s, uint32(n))
+}

internal/transport/transport.go

@@ -291,9 +291,7 @@ type Stream struct {
 	recvCompress string
 	sendCompress string
 
-	// Callback to state application's intentions to read data. This
-	// is used to adjust flow control, if needed.
-	requestRead func(int)
+	readRequester readRequester
 
 	// contentSubtype is the content-subtype for requests.
 	// this must be lowercase or the behavior is undefined.
@@ -310,6 +308,12 @@ type Stream struct {
 	wq       writeQuota
 }
 
+// readRequester is used to state application's intentions to read data. This
+// is used to adjust flow control, if needed.
+type readRequester interface {
+	requestRead(int)
+}
+
 func (s *Stream) swapState(st streamState) streamState {
 	return streamState(atomic.SwapUint32((*uint32)(&s.state), uint32(st)))
 }
@@ -357,7 +361,7 @@ func (s *Stream) ReadMessageHeader(header []byte) (err error) {
 	if er := s.trReader.er; er != nil {
 		return er
 	}
-	s.requestRead(len(header))
+	s.readRequester.requestRead(len(header))
 	for len(header) != 0 {
 		n, err := s.trReader.ReadMessageHeader(header)
 		header = header[n:]
@@ -380,7 +384,7 @@ func (s *Stream) read(n int) (data mem.BufferSlice, err error) {
 	if er := s.trReader.er; er != nil {
 		return nil, er
 	}
-	s.requestRead(n)
+	s.readRequester.requestRead(n)
 	for n != 0 {
 		buf, err := s.trReader.Read(n)
 		var bufLen int
@@ -422,18 +426,24 @@ type transportReader struct {
 	_ noCopy
 	// The handler to control the window update procedure for both this
 	// particular stream and the associated transport.
-	windowHandler func(int)
+	windowHandler windowHandler
 	er            error
 	reader        recvBufferReader
 }
 
+// The handler to control the window update procedure for both this
+// particular stream and the associated transport.
+type windowHandler interface {
+	updateWindow(int)
+}
+
 func (t *transportReader) ReadMessageHeader(header []byte) (int, error) {
 	n, err := t.reader.ReadMessageHeader(header)
 	if err != nil {
 		t.er = err
 		return 0, err
 	}
-	t.windowHandler(n)
+	t.windowHandler.updateWindow(n)
 	return n, nil
 }
 
@@ -443,7 +453,7 @@ func (t *transportReader) Read(n int) (mem.Buffer, error) {
 		t.er = err
 		return buf, err
 	}
-	t.windowHandler(buf.Len())
+	t.windowHandler.updateWindow(buf.Len())
 	return buf, nil
 }
 
@@ -629,6 +639,8 @@ type internalServerTransport interface {
 	write(s *ServerStream, hdr []byte, data mem.BufferSlice, opts *WriteOptions) error
 	writeStatus(s *ServerStream, st *status.Status) error
 	incrMsgRecv()
+	adjustWindow(s *ServerStream, n uint32)
+	updateWindow(s *ServerStream, n uint32)
 }
 
 // connectionErrorf creates an ConnectionError with the specified error description.

internal/transport/transport_test.go

@@ -1856,8 +1856,8 @@ func (s) TestReadGivesSameErrorAfterAnyErrorOccurs(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	s := &Stream{
-		ctx:         ctx,
-		requestRead: func(int) {},
+		ctx:           ctx,
+		readRequester: &fakeReadRequester{},
 	}
 	s.buf.init()
 	s.trReader = transportReader{
@@ -1866,7 +1866,9 @@ func (s) TestReadGivesSameErrorAfterAnyErrorOccurs(t *testing.T) {
 			ctxDone: s.ctx.Done(),
 			recv:    &s.buf,
 		},
-		windowHandler: func(int) {},
+		windowHandler: &mockWindowUpdater{
+			f: func(int) {},
+		},
 	}
 	testData := make([]byte, 1)
 	testData[0] = 5
@@ -3163,16 +3165,18 @@ func (s) TestReadMessageHeaderMultipleBuffers(t *testing.T) {
 	headerLen := 5
 	bytesRead := 0
 	s := Stream{
-		requestRead: func(int) {},
+		readRequester: &fakeReadRequester{},
 	}
 	s.buf.init()
 	recvBuffer := &s.buf
 	s.trReader = transportReader{
 		reader: recvBufferReader{
 			recv: recvBuffer,
 		},
-		windowHandler: func(i int) {
-			bytesRead += i
+		windowHandler: &mockWindowUpdater{
+			f: func(i int) {
+				bytesRead += i
+			},
 		},
 	}
 
@@ -3476,3 +3480,16 @@ func (s) TestDeleteStreamMetricsIncrementedOnlyOnce(t *testing.T) {
 		})
 	}
 }
+
+type fakeReadRequester struct {
+}
+
+func (f *fakeReadRequester) requestRead(int) {}
+
+type mockWindowUpdater struct {
+	f func(int)
+}
+
+func (m *mockWindowUpdater) updateWindow(n int) {
+	m.f(n)
+}