Skip to content

Commit

Permalink
Add leak goroutine checking to grpc/balancer tests (#1497)
Browse files Browse the repository at this point in the history
  • Loading branch information
@menghanl
menghanl authored Sep 7, 2017
1 parent 067cb1f commit d46a365
Show file tree
Hide file tree
Showing 6 changed files with 105 additions and 61 deletions.
103 changes: 55 additions & 48 deletions balancer_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ import (
"golang.org/x/net/context"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/naming"
"google.golang.org/grpc/test/leakcheck"
)

type testWatcher struct {
Expand Down Expand Up @@ -56,6 +57,7 @@ func (w *testWatcher) Next() (updates []*naming.Update, err error) {
}

func (w *testWatcher) Close() {
close(w.side)
}

// Inject naming resolution updates to the testWatcher.
Expand Down Expand Up @@ -89,7 +91,7 @@ func (r *testNameResolver) Resolve(target string) (naming.Watcher, error) {
return r.w, nil
}

func startServers(t *testing.T, numServers int, maxStreams uint32) ([]*server, *testNameResolver) {
func startServers(t *testing.T, numServers int, maxStreams uint32) ([]*server, *testNameResolver, func()) {
var servers []*server
for i := 0; i < numServers; i++ {
s := newTestServer()
Expand All @@ -100,18 +102,25 @@ func startServers(t *testing.T, numServers int, maxStreams uint32) ([]*server, *
// Point to server[0]
addr := "localhost:" + servers[0].port
return servers, &testNameResolver{
addr: addr,
}
addr: addr,
}, func() {
for i := 0; i < numServers; i++ {
servers[i].stop()
}
}
}

func TestNameDiscovery(t *testing.T) {
defer leakcheck.Check(t)
// Start 2 servers on 2 ports.
numServers := 2
servers, r := startServers(t, numServers, math.MaxUint32)
servers, r, cleanup := startServers(t, numServers, math.MaxUint32)
defer cleanup()
cc, err := Dial("foo.bar.com", WithBalancer(RoundRobin(r)), WithBlock(), WithInsecure(), WithCodec(testCodec{}))
if err != nil {
t.Fatalf("Failed to create ClientConn: %v", err)
}
defer cc.Close()
req := "port"
var reply string
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err == nil || ErrorDesc(err) != servers[0].port {
Expand All @@ -135,18 +144,17 @@ func TestNameDiscovery(t *testing.T) {
}
time.Sleep(10 * time.Millisecond)
}
cc.Close()
for i := 0; i < numServers; i++ {
servers[i].stop()
}
}

func TestEmptyAddrs(t *testing.T) {
servers, r := startServers(t, 1, math.MaxUint32)
defer leakcheck.Check(t)
servers, r, cleanup := startServers(t, 1, math.MaxUint32)
defer cleanup()
cc, err := Dial("foo.bar.com", WithBalancer(RoundRobin(r)), WithBlock(), WithInsecure(), WithCodec(testCodec{}))
if err != nil {
t.Fatalf("Failed to create ClientConn: %v", err)
}
defer cc.Close()
var reply string
if err := Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, cc); err != nil || reply != expectedResponse {
t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, reply = %q, want %q, <nil>", err, reply, expectedResponse)
Expand All @@ -168,18 +176,19 @@ func TestEmptyAddrs(t *testing.T) {
}
cancel()
}
cc.Close()
servers[0].stop()
}

func TestRoundRobin(t *testing.T) {
defer leakcheck.Check(t)
// Start 3 servers on 3 ports.
numServers := 3
servers, r := startServers(t, numServers, math.MaxUint32)
servers, r, cleanup := startServers(t, numServers, math.MaxUint32)
defer cleanup()
cc, err := Dial("foo.bar.com", WithBalancer(RoundRobin(r)), WithBlock(), WithInsecure(), WithCodec(testCodec{}))
if err != nil {
t.Fatalf("Failed to create ClientConn: %v", err)
}
defer cc.Close()
// Add servers[1] to the service discovery.
u := &naming.Update{
Op: naming.Add,
Expand Down Expand Up @@ -214,18 +223,17 @@ func TestRoundRobin(t *testing.T) {
t.Fatalf("Index %d: Invoke(_, _, _, _, _) = %v, want %s", i, err, servers[i%numServers].port)
}
}
cc.Close()
for i := 0; i < numServers; i++ {
servers[i].stop()
}
}

func TestCloseWithPendingRPC(t *testing.T) {
servers, r := startServers(t, 1, math.MaxUint32)
defer leakcheck.Check(t)
servers, r, cleanup := startServers(t, 1, math.MaxUint32)
defer cleanup()
cc, err := Dial("foo.bar.com", WithBalancer(RoundRobin(r)), WithBlock(), WithInsecure(), WithCodec(testCodec{}))
if err != nil {
t.Fatalf("Failed to create ClientConn: %v", err)
}
defer cc.Close()
var reply string
if err := Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, cc, FailFast(false)); err != nil {
t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, want %s", err, servers[0].port)
Expand Down Expand Up @@ -267,15 +275,17 @@ func TestCloseWithPendingRPC(t *testing.T) {
time.Sleep(5 * time.Millisecond)
cc.Close()
wg.Wait()
servers[0].stop()
}

func TestGetOnWaitChannel(t *testing.T) {
servers, r := startServers(t, 1, math.MaxUint32)
defer leakcheck.Check(t)
servers, r, cleanup := startServers(t, 1, math.MaxUint32)
defer cleanup()
cc, err := Dial("foo.bar.com", WithBalancer(RoundRobin(r)), WithBlock(), WithInsecure(), WithCodec(testCodec{}))
if err != nil {
t.Fatalf("Failed to create ClientConn: %v", err)
}
defer cc.Close()
// Remove all servers so that all upcoming RPCs will block on waitCh.
updates := []*naming.Update{{
Op: naming.Delete,
Expand Down Expand Up @@ -309,18 +319,19 @@ func TestGetOnWaitChannel(t *testing.T) {
r.w.inject(updates)
// Wait until the above RPC succeeds.
wg.Wait()
cc.Close()
servers[0].stop()
}

func TestOneServerDown(t *testing.T) {
defer leakcheck.Check(t)
// Start 2 servers.
numServers := 2
servers, r := startServers(t, numServers, math.MaxUint32)
servers, r, cleanup := startServers(t, numServers, math.MaxUint32)
defer cleanup()
cc, err := Dial("foo.bar.com", WithBalancer(RoundRobin(r)), WithBlock(), WithInsecure(), WithCodec(testCodec{}))
if err != nil {
t.Fatalf("Failed to create ClientConn: %v", err)
}
defer cc.Close()
// Add servers[1] to the service discovery.
var updates []*naming.Update
updates = append(updates, &naming.Update{
Expand Down Expand Up @@ -361,20 +372,19 @@ func TestOneServerDown(t *testing.T) {
}()
}
wg.Wait()
cc.Close()
for i := 0; i < numServers; i++ {
servers[i].stop()
}
}

func TestOneAddressRemoval(t *testing.T) {
defer leakcheck.Check(t)
// Start 2 servers.
numServers := 2
servers, r := startServers(t, numServers, math.MaxUint32)
servers, r, cleanup := startServers(t, numServers, math.MaxUint32)
defer cleanup()
cc, err := Dial("foo.bar.com", WithBalancer(RoundRobin(r)), WithBlock(), WithInsecure(), WithCodec(testCodec{}))
if err != nil {
t.Fatalf("Failed to create ClientConn: %v", err)
}
defer cc.Close()
// Add servers[1] to the service discovery.
var updates []*naming.Update
updates = append(updates, &naming.Update{
Expand Down Expand Up @@ -423,10 +433,6 @@ func TestOneAddressRemoval(t *testing.T) {
}()
}
wg.Wait()
cc.Close()
for i := 0; i < numServers; i++ {
servers[i].stop()
}
}

func checkServerUp(t *testing.T, currentServer *server) {
Expand All @@ -436,19 +442,20 @@ func checkServerUp(t *testing.T, currentServer *server) {
if err != nil {
t.Fatalf("Failed to create ClientConn: %v", err)
}
defer cc.Close()
var reply string
for {
if err := Invoke(context.Background(), "/foo/bar", &req, &reply, cc); err != nil && ErrorDesc(err) == port {
break
}
time.Sleep(10 * time.Millisecond)
}
cc.Close()
}

func TestPickFirstEmptyAddrs(t *testing.T) {
servers, r := startServers(t, 1, math.MaxUint32)
defer servers[0].stop()
defer leakcheck.Check(t)
servers, r, cleanup := startServers(t, 1, math.MaxUint32)
defer cleanup()
cc, err := Dial("foo.bar.com", WithBalancer(pickFirstBalancer(r)), WithBlock(), WithInsecure(), WithCodec(testCodec{}))
if err != nil {
t.Fatalf("Failed to create ClientConn: %v", err)
Expand Down Expand Up @@ -478,12 +485,14 @@ func TestPickFirstEmptyAddrs(t *testing.T) {
}

func TestPickFirstCloseWithPendingRPC(t *testing.T) {
servers, r := startServers(t, 1, math.MaxUint32)
defer servers[0].stop()
defer leakcheck.Check(t)
servers, r, cleanup := startServers(t, 1, math.MaxUint32)
defer cleanup()
cc, err := Dial("foo.bar.com", WithBalancer(pickFirstBalancer(r)), WithBlock(), WithInsecure(), WithCodec(testCodec{}))
if err != nil {
t.Fatalf("Failed to create ClientConn: %v", err)
}
defer cc.Close()
var reply string
if err := Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, cc, FailFast(false)); err != nil {
t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, want %s", err, servers[0].port)
Expand Down Expand Up @@ -528,12 +537,11 @@ func TestPickFirstCloseWithPendingRPC(t *testing.T) {
}

func TestPickFirstOrderAllServerUp(t *testing.T) {
defer leakcheck.Check(t)
// Start 3 servers on 3 ports.
numServers := 3
servers, r := startServers(t, numServers, math.MaxUint32)
for i := 0; i < numServers; i++ {
defer servers[i].stop()
}
servers, r, cleanup := startServers(t, numServers, math.MaxUint32)
defer cleanup()
cc, err := Dial("foo.bar.com", WithBalancer(pickFirstBalancer(r)), WithBlock(), WithInsecure(), WithCodec(testCodec{}))
if err != nil {
t.Fatalf("Failed to create ClientConn: %v", err)
Expand Down Expand Up @@ -642,12 +650,11 @@ func TestPickFirstOrderAllServerUp(t *testing.T) {
}

func TestPickFirstOrderOneServerDown(t *testing.T) {
defer leakcheck.Check(t)
// Start 3 servers on 3 ports.
numServers := 3
servers, r := startServers(t, numServers, math.MaxUint32)
for i := 0; i < numServers; i++ {
defer servers[i].stop()
}
servers, r, cleanup := startServers(t, numServers, math.MaxUint32)
defer cleanup()
cc, err := Dial("foo.bar.com", WithBalancer(pickFirstBalancer(r)), WithBlock(), WithInsecure(), WithCodec(testCodec{}))
if err != nil {
t.Fatalf("Failed to create ClientConn: %v", err)
Expand Down Expand Up @@ -702,6 +709,7 @@ func TestPickFirstOrderOneServerDown(t *testing.T) {
p, _ := strconv.Atoi(servers[0].port)
servers[0] = newTestServer()
go servers[0].start(t, p, math.MaxUint32)
defer servers[0].stop()
servers[0].wait(t, 2*time.Second)
checkServerUp(t, servers[0])

Expand Down Expand Up @@ -733,12 +741,11 @@ func TestPickFirstOrderOneServerDown(t *testing.T) {
}

func TestPickFirstOneAddressRemoval(t *testing.T) {
defer leakcheck.Check(t)
// Start 2 servers.
numServers := 2
servers, r := startServers(t, numServers, math.MaxUint32)
for i := 0; i < numServers; i++ {
defer servers[i].stop()
}
servers, r, cleanup := startServers(t, numServers, math.MaxUint32)
defer cleanup()
cc, err := Dial("localhost:"+servers[0].port, WithBalancer(pickFirstBalancer(r)), WithBlock(), WithInsecure(), WithCodec(testCodec{}))
if err != nil {
t.Fatalf("Failed to create ClientConn: %v", err)
Expand Down
6 changes: 6 additions & 0 deletions call_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ import (
"golang.org/x/net/context"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"google.golang.org/grpc/test/leakcheck"
"google.golang.org/grpc/transport"
)

Expand Down Expand Up @@ -211,6 +212,7 @@ func setUp(t *testing.T, port int, maxStreams uint32) (*server, *ClientConn) {
}

func TestInvoke(t *testing.T) {
defer leakcheck.Check(t)
server, cc := setUp(t, 0, math.MaxUint32)
var reply string
if err := Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply, cc); err != nil || reply != expectedResponse {
Expand All @@ -221,6 +223,7 @@ func TestInvoke(t *testing.T) {
}

func TestInvokeLargeErr(t *testing.T) {
defer leakcheck.Check(t)
server, cc := setUp(t, 0, math.MaxUint32)
var reply string
req := "hello"
Expand All @@ -237,6 +240,7 @@ func TestInvokeLargeErr(t *testing.T) {

// TestInvokeErrorSpecialChars checks that error messages don't get mangled.
func TestInvokeErrorSpecialChars(t *testing.T) {
defer leakcheck.Check(t)
server, cc := setUp(t, 0, math.MaxUint32)
var reply string
req := "weird error"
Expand All @@ -253,6 +257,7 @@ func TestInvokeErrorSpecialChars(t *testing.T) {

// TestInvokeCancel checks that an Invoke with a canceled context is not sent.
func TestInvokeCancel(t *testing.T) {
defer leakcheck.Check(t)
server, cc := setUp(t, 0, math.MaxUint32)
var reply string
req := "canceled"
Expand All @@ -271,6 +276,7 @@ func TestInvokeCancel(t *testing.T) {
// TestInvokeCancelClosedNonFail checks that a canceled non-failfast RPC
// on a closed client will terminate.
func TestInvokeCancelClosedNonFailFast(t *testing.T) {
defer leakcheck.Check(t)
server, cc := setUp(t, 0, math.MaxUint32)
var reply string
cc.Close()
Expand Down
Loading

0 comments on commit d46a365

Please sign in to comment.