speech: API is leaking connections

[snktagarwal]

I am using Speech API v0.10.0 and running into open connections in my client. I am using the following stub code:
main() {
....
477 ctx := context.Background()
478 // Create a google speech API client
479 client, err = speech.NewClient(ctx, option.WithServiceAccountFile(os.Getenv("GOOGLE_APPLICATION_CREDENTIALS")))
480 if err != nil {
481 log.Println("Err: Cannot create speech client")
482 return
483 }
484 defer client.Close()
}

116 ctx := context.Background()
117 speechStream, err := client.StreamingRecognize(ctx)

I call the following when I finish reading audio like this:

264 err := speechStream.CloseSend()
265 if err != nil {
266 log.Printf("CloseSend failed due to %+v", err)
267 }

I am consistently seeing that CloseSend succeeds (err is nil) but I keep having a increase in number of connections. I tried to GRPC debug this before coming here: grpc/grpc-go#1402

Reading from above issue it looks like creating a new RPC should not always mean creating a new connection. Although, the way I am leaking it looks like each streaming RPC is leaking a dangling client connection. Other notes:

1. main() is called once and creates a RPC server
2. I stream audio every 55s (because of 1min limit) and that's when I create a new speechStream. I close the speechStream after the 55s deadline.

Is this the right way to go around the 55 sec deadline when doing transcription? Is there any behavior of closing streams down that I am not handling properly?

[snktagarwal]

Hi, I see you updated the bug. is this in triage stage or an issue you know about?

[snktagarwal]

`
import (
"io"

"fmt"                                                                                                               
_ "net"                                                                                                             
"net/http"                                                                                                          
_ "net/http/pprof"                                                                                                  
"testing"                                                                                                           
"cloud.google.com/go/speech/apiv1"                                                                                  
"golang.org/x/net/context"                                                                                          
speechpb "google.golang.org/genproto/googleapis/cloud/speech/v1"                                                    

)

func TestExampleClient_StreamingRecognize(t *testing.T) {
for i := 0; i < 10; i++ {
ctx := context.Background()
c, err := speech.NewClient(ctx)
if err != nil {
// TODO: Handle error.
}
stream, err := c.StreamingRecognize(ctx)
if err != nil {
// TODO: Handle error.
}
go func() {
reqs := []*speechpb.StreamingRecognizeRequest{
// TODO: Create requests.
}
for _, req := range reqs {
if err := stream.Send(req); err != nil {
// TODO: Handle error.
}
}
stream.CloseSend()
}()
for {
resp, err := stream.Recv()
if err == io.EOF {
break
}
if err != nil {
// TODO: handle error.
}
// TODO: Use resp.
_ = resp
}
}
fmt.Printf("Starting the server ...")
fmt.Println(http.ListenAndServe("localhost:6060", nil))
fmt.Printf("Ending the server ...")
}`

Use the above code to repro the issue. Go to localhost:6060/debug/pprof/ and if you will see 9 thread leaks like this:

10 @ 0x105aabd 0x106b8aa 0x106a4cc 0x1587e02 0x1089921
#	0x1587e01	google.golang.org/grpc/transport.(*http2Client).controller+0x9a1	/Users/sanket/go/src/google.golang.org/grpc/transport/http2_client.go:1227

10 @ 0x105aabd 0x106b8aa 0x106a4cc 0x15b228d 0x15cb747 0x1089921
#	0x15b228c	google.golang.org/grpc.(*addrConn).transportMonitor+0xc6c	/Users/sanket/go/src/google.golang.org/grpc/clientconn.go:1030
#	0x15cb746	google.golang.org/grpc.(*ClientConn).resetAddrConn.func1+0x266	/Users/sanket/go/src/google.golang.org/grpc/clientconn.go:759

[jba]

Thanks for the repro. It's not a known issue. I will look at it shortly.

[pongad]

@snktagarwal I ran your code and observed 37 goroutines at the end. Is this what you're seeing? If I increase the number of iterations to 20, I end up with 67 goroutines.

However, the Client object can be reused and safely used by multiple goroutines. If I call NewClient only once before the loop, I only end up with 7 goroutines. If I call c.Close before ListenAndServe, I see only 3. From the stack, it looks like all of them are spawned by the HTTP server itself, so I don't think there's a leak there.

From your issue description, your server is creating only one client, is this right?

[snktagarwal]

Ahan. Good catch. I never thought about reusing the "client" across goroutines. You're right, currently I create a client for each RPC call in my server.

I will use this approach to reduce the amount of leaks. But, I think its still a bug (maybe a lower priority now) that the Client has dangling RPC watchers (thread leaks that you saw when client was created multiple times).

[pongad]

@snktagarwal Potentially, yes. Though, I didn't see any leaks in my testing. As noted in my last comment, after closing the speech client, I only see 3 goroutines running, all of them seems to be spawned from ListenAndServe itself.

Since this is already tracked in grpc/grpc-go#1402 and the observed leak is from grpc not the client lib, I'll close this issue. Please reopen if you think it belongs here.

[snktagarwal]

(Not sure how to reopen the issue)

I think the purpose of the client is to be created once and used across. I have a (golang) server that is accepting audio from a NodeJS UI and pumping it into my server. I only want to create one NewClient() in main() and then use it throughout my RPCs. If I have to close the client for each RPC that defeats the purpose of the API and hence this is still a bad bug since it promotes bad coding practice.

On a second note, somehow even after doing CloseSend on the Client I am getting thread leaks in my server code. It may be the case that somewhere my server is holding onto objects and not letting them GC.

[pongad]

@snktagarwal I'm not sure what you're asking in the first paragraph. The client can be safely used concurrently by many goroutines. There shouldn't be anything stopping you from calling NewClient only once. Stream, on the other hand, are trickier, but those should be scoped to each request, right?

[snktagarwal]

What I am claiming is if I do something like:

c = NewClient()
while processingRequests {
rec = c.RecognizeRequest()
.....
rec.CloseSend() // <--!! This is creating a thread leak which does not go away until c.Close() is called
}

// wait for a long time to serve RPCs
c.Close()

during the time that there are 100s of RecognizeResults() are called thread leaks which is not acceptable for a server. Does this makes more sense? You can use the above code to repro this issue.

func TestExampleClient_StreamingRecognize(t *testing.T) {
	ctx := context.Background()
	c, err := speech.NewClient(ctx)
	for i := 0; i < 10; i++ {
		if err != nil {
			// TODO: Handle error.
		}
		stream, err := c.StreamingRecognize(ctx)
		if err != nil {
			// TODO: Handle error.
		}
		go func() {
			reqs := []*speechpb.StreamingRecognizeRequest{
			// TODO: Create requests.
			}
			for _, req := range reqs {
				if err := stream.Send(req); err != nil {
					// TODO: Handle error.
				}
			}
			stream.CloseSend()
		}()
		for {
			resp, err := stream.Recv()
			if err == io.EOF {
				break
			}
			if err != nil {
				// TODO: handle error.
			}
			// TODO: Use resp.
			_ = resp
		}
	}
	fmt.Printf("Starting the server ...")
	fmt.Println(http.ListenAndServe("localhost:6060", nil))
	fmt.Printf("Ending the server ...")
	c.Close() // call close after we have served all RPCs
}

[pongad]

I edited your code snippet for spacing. I'll take a look at this again.

[pongad]

@snktagarwal I ran the code above (importing "net/http/pprof"). After printing "starting the server", I went to http://localhost:6060/debug/pprof/goroutine?debug=1.

The dump shows only 7 goroutines. Both google.golang.org/grpc/transport.(*http2Client).controller and google.golang.org/grpc.(*addrConn).transportMonitor have 1 goroutine each.

Could it be you are using an outdated grpc package?

[snktagarwal]

Hmm. Can you tell me your grpc version?

…

On Aug 30, 2017 12:29 AM, "Michael Darakananda" ***@***.***> wrote: @snktagarwal <https://github.com/snktagarwal> I ran the code above (importing "net/http/pprof"). After printing "starting the server", I went to http://localhost:6060/debug/pprof/goroutine?debug=1. The dump shows only 7 goroutines. Both google.golang.org/grpc/ transport.(*http2Client).controller and google.golang.org/grpc.(* addrConn).transportMonitor have 1 goroutine each. Could it be you are using an outdated grpc package? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#711 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAG_HybOLbnvFF3OFzEMYhiF0Y4jCPd6ks5sdQ9xgaJpZM4OnOEY> .

[pongad]

@snktagarwal I used grpc from master earlier, but just tested with v1.5.2 and got the same results.

[pongad]

Since this issue has been inactive for a while, I'll close this. Please reopen if required.

[Bankq]

I'm seeing same issue with pubsub. Goroutine counts and mem usage keeps creeping up when I receive on a topic.

In my case there is only 1 goroutine calling receive.

Ok I found the issue: you have to explicitly call pubsub.Client.Close() to release all the connections. @snktagarwal this might apply to your case as well?

@jba do you think it's worth mentioning it explicitly in the doc? It caught me cuz doc says one don't have to call it at program exits but my service deal with pubsub topics from different projects thus I have to create pubsub clients on the fly. Not calling Close() leaks goroutines there.

[jba]

I tried to clarify the doc in https://code-review.googlesource.com/#/c/gocloud/+/20250.

[Bankq]

@jba tyvm! This is the moment I miss destructors :)