transport: handler server transport continuously reads from streams without waiting for application
Author: wmxhwCreated May 23, 2024Updated Sep 17, 2026
LabelsStatus: Help WantedP2Type: BugfixitArea: Transport
What version of gRPC are you using?
grpc v1.51.0
What version of Go are you using (go version)?
go version go1.21.7 windows/amd64
What operating system (Linux, Windows, …) and version?
Windows 11
What did you do?
My client has a large amount of data to send, but the server's processing performance (such as writing to disk) may not be as good. Here is a test code, and it also exhibits the same issue.
test proto
service HelloService {
rpc SayHelloStream(stream SayHelloRequest) returns (google.protobuf.Empty){}
}
message SayHelloRequest {
string hello = 1;
}server code
func (s helloServer) SayHelloStream(stream hello.HelloService_SayHelloStreamServer) error {
for {
r, err := stream,Recv()
if err == io.EOF {
break
}
if err != nil {
return err
}
fmt.Println(r.Hello)
time.Sleep(10*time.Second) // Print once every 10 seconds.
}
}client code
func main() {
clt := hello.NewHelloServiceClient(cc)
body := bytes.NewBuffer(nil) // Simulate a large amount of data.
for i:=0; i< 1e6; i++ {
body.WriteString("n")
}
for {
err = stream.Send(&hello.SayHelloRequest{
Hello: body.String(),
})
if err != nil {
panic(err)
}
time.Sleep(10*time.Millisecond) // Send once every 10 milliseconds.
}
}What did you expect to see?
Even if the client sends a large amount of traffic, the server should not experience memory leaks. gRPC should have flow control and will not have memory leaks.
What did you see instead?
There is a memory leak occurring in the HandleStreams function at internal/transport/handler_server.go.
func (ht *serverHandlerTransport) HandleStreams(ctx context.Context, startStream func(*Stream)) {
....
go func() {
defer close(readerDone)
// TODO: minimize garbage, optimize recvBuffer code/ownership
const readSize = 8196
for buf := make([]byte, readSize); ; {
n, err := req.Body.Read(buf)
if n > 0 {
s.buf.put(recvMsg{buffer: bytes.NewBuffer(buf[:n:n])})
buf = buf[n:]
}
if err != nil {
s.buf.put(recvMsg{err: mapRecvMsgError(err)})
return
}
if len(buf) == 0 {
buf = make([]byte, readSize) // Does this piece of code have flow control?
}
}
}()
....
}Source: grpc/grpc-go