Established streams leak

[WQG6848]

What version of gRPC are you using?

761a6b364cb9e497a52e78286d686b1cd17d8ee3

What version of Go are you using (go version)?

1.9.2

What operating system (Linux, Windows, …) and version?

Ubuntu 16.04

What did you do?

There is a server that will return a large message body, the message body is always greater than 4M.
Then connect to the server, call the same remote function in a loop.

Server:

var message = strings.Repeat("hello", 1024*1024)

type server struct{}

func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
	return &pb.HelloReply{Message: message}, nil
}

Client:

function main() {
    ......

    for {
        _, err := c.SayHello(context.Background(), &pb.HelloRequest{})
        if err != nil {
            fmt.Println(err)
        }

        time.Sleep(time.Second)
    }
}

What did you expect to see?

Normal memory.

What did you see instead?

Client will print error like this :

"rpc error: code = ResourceExhausted desc = grpc: received message larger than max (5242885 vs. 4194304) ".

Worse, system memory grows quickly. Finally, the server will be killed.

[WQG6848]

When the client reads data error from Stream, why does estdStreams in the server's loopyWriter not delete this stream?
I tracked this behavior and did find a Stream leak.

[lyuxuan]

gRPC has a default 4MB recv message limit on client side. If you want to receive message that's larger than 4MB, then you need to set the call option MaxCallRecvMsgSize(s) to allow for message to be received up to size s.

(Note: Use the dial option WithDefaultCallOptions(MaxCallRecvMsgSize(s)) to apply the same limit to all RPCs using the connection. )

As for the stream leak, since you are sending unary RPC, so by the time client encounter the error, server side should have already finished the stream. I suspect it is something else that's causing the memory to grow. Could you please try the call option above to make client successfully receive the message, and see if the memory still grow quickly?

Thanks!

[WQG6848]

First thanks for your help @lyuxuan .

I can set the size of the data received by the client through the MaxCallRecvMsgSize(s) method. The problem is why the stream leaks when the size of the data returned by the server is larger than s MB.

Demo

hello.proto

syntax="proto3";

package hello;

service Greeter {
  // Sends a greeting
  rpc SayHello (HelloRequest) returns (HelloReply) {}
}

// The request message containing the user's name.
message HelloRequest {
  string name = 1;
}

// The response message containing the greetings
message HelloReply {
  string message = 1;
}

client.go

func main() {
	conn, err := grpc.Dial(":8888", grpc.WithInsecure())
	if err != nil {
		panic(err)
	}
	defer conn.Close()

	c := pb.NewGreeterClient(conn)

	for i := 0; i < 200; i++ {
		_, err := c.SayHello(context.Background(), &pb.HelloRequest{})
		if err != nil {
			fmt.Println(err)
		}
		time.Sleep(time.Second)
	}

	time.Sleep(time.Hour)
}

server.go

var message = strings.Repeat("hello", 1024*1024)

type server struct{}

func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
	return &pb.HelloReply{Message: message}, nil
}

func main() {
	fmt.Println("PID:", os.Getpid())

	lis, err := net.Listen("tcp", ":8888")
	if err != nil {
		panic(err)
	}

	go func() {
		// pprof
		if err := http.ListenAndServe(":8080", nil); err != nil {
			panic(err)
		}
	}()

	s := grpc.NewServer()
	pb.RegisterGreeterServer(s, &server{})

	s.Serve(lis)
}

After analyzing the code, I suspect that the stream is leaking, so I added some logs to the GPRC source code.

google.golang.org/grpc/internal/transport/controlbuf.go :515

func (l *loopyWriter) registerStreamHandler(h *registerStream) error {
	str := &outStream{
		id:    h.streamID,
		state: empty,
		itl:   &itemList{},
		wq:    h.wq,
	}
	l.estdStreams[h.streamID] = str
	fmt.Println("Register StreamID: ", h.streamID, "Now Streams Length:", len(l.estdStreams)) // log
	return nil
}

First

server.go

var message = strings.Repeat("hello", 1024*1024)

In this test, the client will report errors like this :

rpc error: code = ResourceExhausted desc = grpc: received message larger than max (5242885 vs. 4194304)
rpc error: code = ResourceExhausted desc = grpc: received message larger than max (5242885 vs. 4194304)
rpc error: code = ResourceExhausted desc = grpc: received message larger than max (5242885 vs. 4194304)
rpc error: code = ResourceExhausted desc = grpc: received message larger than max (5242885 vs. 4194304)
rpc error: code = ResourceExhausted desc = grpc: received message larger than max (5242885 vs. 4194304)
rpc error: code = ResourceExhausted desc = grpc: received message larger than max (5242885 vs. 4194304)
rpc error: code = ResourceExhausted desc = grpc: received message larger than max (5242885 vs. 4194304)
rpc error: code = ResourceExhausted desc = grpc: received message larger than max (5242885 vs. 4194304)
rpc error: code = ResourceExhausted desc = grpc: received message larger than max (5242885 vs. 4194304)

The server will print logs like this:

PID: 30168
Register StreamID:  1 Now Streams Length: 1
Register StreamID:  3 Now Streams Length: 2
Register StreamID:  5 Now Streams Length: 3
Register StreamID:  7 Now Streams Length: 4
Register StreamID:  9 Now Streams Length: 5
Register StreamID:  11 Now Streams Length: 6
Register StreamID:  13 Now Streams Length: 7
Register StreamID:  15 Now Streams Length: 8
Register StreamID:  17 Now Streams Length: 9
Register StreamID:  19 Now Streams Length: 10

..............

Register StreamID:  385 Now Streams Length: 193
Register StreamID:  387 Now Streams Length: 194
Register StreamID:  389 Now Streams Length: 195
Register StreamID:  391 Now Streams Length: 196
Register StreamID:  393 Now Streams Length: 197
Register StreamID:  395 Now Streams Length: 198
Register StreamID:  397 Now Streams Length: 199
Register StreamID:  399 Now Streams Length: 200

So, it seems that the stream leaked

Next, take a look at the memory.

> top -p 30168
PID USER      PR  NI    VIRT    RES    SHR S  %CPU %MEM     TIME+ COMMAND
30168 wqg+    20   0 1521120 1.027g   7808 S   0.0 13.2   0:03.25 server

Memroy leak did happen

pprof:

Second

server.go

var message = strings.Repeat("h", 1024*1024)

In this test, The server will print logs like this:

PID: 1662
Register StreamID:  1 Now Streams Length: 1
Register StreamID:  3 Now Streams Length: 1
Register StreamID:  5 Now Streams Length: 1
Register StreamID:  7 Now Streams Length: 1
Register StreamID:  9 Now Streams Length: 1
Register StreamID:  11 Now Streams Length: 1
Register StreamID:  13 Now Streams Length: 1
Register StreamID:  15 Now Streams Length: 1

.......................

Register StreamID:  381 Now Streams Length: 1
Register StreamID:  383 Now Streams Length: 1
Register StreamID:  385 Now Streams Length: 1
Register StreamID:  387 Now Streams Length: 1
Register StreamID:  389 Now Streams Length: 1
Register StreamID:  391 Now Streams Length: 1
Register StreamID:  393 Now Streams Length: 1
Register StreamID:  395 Now Streams Length: 1
Register StreamID:  397 Now Streams Length: 1
Register StreamID:  399 Now Streams Length: 1

It looks like everything is ok

Next, take a look at the memory.

> top -p 1662
PID USER      PR  NI    VIRT    RES    SHR S  %CPU %MEM     TIME+ COMMAND
1662 wqg+     20   0  665296  14496   7932 S   0.0  0.2   0:01.96 server

Everything is ok

[WQG6848]

I found something.

func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.Stream, srv *service, md *MethodDesc, trInfo *traceInfo) (err error) {
    ......
    // Step 1
    if err := s.sendResponse(t, stream, reply, cp, opts, comp); err != nil {
        ......
    }

    // Step 2
    err = t.WriteStatus(stream, status.New(codes.OK, ""))
    ....

    return err
}

After step 1, outStream.itl.head.it is *dataFrame. At this time, the server may already be returning data to the client.

After step 2, outStream.itl.head.it is *dataFrame. outStream.itl.head.next.it is *headerFrame.
The important thing is that the value of Stream.state changes from streamReadDone to streamDone. Then the program ignores the handleRSTStream frame sent by the client Stream.

In general, after all the data in the dataFrame is sent, the program will execute cleanupStreamHandler(headerFrame.cleanup) to delete the stream.

Unfortunately, if the client Stream reports an error after receiving the first data frame, for example:
rpc error: code = ResourceExhausted desc = grpc: received message larger than max.....
The client Stream sends a handleRSTStream frame to the server, after which the client Stream is closed.

However, because the value of Stream.states is streamDone, the server will ignore the handleRSTStream frame.

But the server's loopyWriter will still write data to the exception Stream until sendQuota is exhausted.

[lyuxuan]

@WQG6848 Thanks for your detailed explanation. It's very helpful. Yes, it's indeed a bug in grpc, we'll get it fixed soon. Let us know if you need any help when before it gets fixed. Thanks!

[canguler]

fixed by #2610

[jonathanbp]

I'm seeing something that looks like this issue with latest gRPC release (1.19.1), also with a 4MB max message limit set on clients, Math.Int32Max on server and messages are >4MB. If I kill all clients the memory is gc'ed, but as long as clients stay connected and request data which results in a resource exhausted message the memory increases until the process is OOM killed. If I increase the limit on clients to > message size then no leaks are apparent.

The code to reproduce looks very similar to the code posted here earlier by @WQG6848.

[canguler]

Hi @jonathanbp,

Can you share the code to reproduce it?