Revert "Use bufconn in end2end tests." (#1381)

This reverts commit c7e2c00.

There is a race between Dial and Close that causes goroutine leaks in the
end2end tests.

test/end2end_test.go

@@ -53,7 +53,6 @@ import (
 	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/status"
 	"google.golang.org/grpc/tap"
-	"google.golang.org/grpc/test/bufconn"
 	testpb "google.golang.org/grpc/test/grpc_testing"
 )
 
@@ -355,7 +354,7 @@ const tlsDir = "testdata/"
 
 type env struct {
 	name        string
-	network     string // The type of network such as tcp, unix, bufconn, etc.
+	network     string // The type of network such as tcp, unix, etc.
 	security    string // The security protocol such as TLS, SSH, etc.
 	httpHandler bool   // whether to use the http.Handler ServerTransport; requires TLS
 	balancer    bool   // whether to use balancer
@@ -368,19 +367,21 @@ func (e env) runnable() bool {
 	return true
 }
 
-const bufconnNetwork = "bufconn"
+func (e env) dialer(addr string, timeout time.Duration) (net.Conn, error) {
+	return net.DialTimeout(e.network, addr, timeout)
+}
 
 var (
 	tcpClearEnv   = env{name: "tcp-clear", network: "tcp", balancer: true}
 	tcpTLSEnv     = env{name: "tcp-tls", network: "tcp", security: "tls", balancer: true}
 	unixClearEnv  = env{name: "unix-clear", network: "unix", balancer: true}
 	unixTLSEnv    = env{name: "unix-tls", network: "unix", security: "tls", balancer: true}
-	handlerEnv    = env{name: "handler-tls", network: bufconnNetwork, security: "tls", httpHandler: true, balancer: true}
-	noBalancerEnv = env{name: "no-balancer", network: bufconnNetwork, security: "tls", balancer: false}
+	handlerEnv    = env{name: "handler-tls", network: "tcp", security: "tls", httpHandler: true, balancer: true}
+	noBalancerEnv = env{name: "no-balancer", network: "tcp", security: "tls", balancer: false}
 	allEnv        = []env{tcpClearEnv, tcpTLSEnv, unixClearEnv, unixTLSEnv, handlerEnv, noBalancerEnv}
 )
 
-var onlyEnv = flag.String("only_env", "", "Restrict tests to the named environment. Empty means all.")
+var onlyEnv = flag.String("only_env", "", "If non-empty, one of 'tcp-clear', 'tcp-tls', 'unix-clear', 'unix-tls', or 'handler-tls' to only run the tests for that environment. Empty means all.")
 
 func listTestEnv() (envs []env) {
 	if *onlyEnv != "" {
@@ -441,7 +442,6 @@ type test struct {
 	// srv and srvAddr are set once startServer is called.
 	srv     *grpc.Server
 	srvAddr string
-	lis     net.Listener
 
 	cc          *grpc.ClientConn // nil until requested via clientConn
 	restoreLogs func()           // nil unless declareLogNoise is used
@@ -523,12 +523,7 @@ func (te *test) startServer(ts testpb.TestServiceServer) {
 		la = "/tmp/testsock" + fmt.Sprintf("%d", time.Now().UnixNano())
 		syscall.Unlink(la)
 	}
-	var err error
-	if te.e.network == bufconnNetwork {
-		te.lis = bufconn.Listen(1024 * 1024)
-	} else {
-		te.lis, err = net.Listen(te.e.network, la)
-	}
+	lis, err := net.Listen(te.e.network, la)
 	if err != nil {
 		te.t.Fatalf("Failed to listen: %v", err)
 	}
@@ -561,36 +556,24 @@ func (te *test) startServer(ts testpb.TestServiceServer) {
 	addr := la
 	switch te.e.network {
 	case "unix":
-	case "bufconn":
-		addr = te.lis.Addr().String()
 	default:
-		_, port, err := net.SplitHostPort(te.lis.Addr().String())
+		_, port, err := net.SplitHostPort(lis.Addr().String())
 		if err != nil {
 			te.t.Fatalf("Failed to parse listener address: %v", err)
 		}
 		addr = "localhost:" + port
 	}
 
-	go s.Serve(te.lis)
+	go s.Serve(lis)
 	te.srvAddr = addr
 }
 
-func (te *test) dialer(addr string, timeout time.Duration) (net.Conn, error) {
-	if te.lis == nil {
-		return nil, fmt.Errorf("no listener")
-	}
-	if te.e.network == bufconnNetwork {
-		return te.lis.(*bufconn.Listener).Dial()
-	}
-	return net.DialTimeout(te.e.network, addr, timeout)
-}
-
 func (te *test) clientConn() *grpc.ClientConn {
 	if te.cc != nil {
 		return te.cc
 	}
 	opts := []grpc.DialOption{
-		grpc.WithDialer(te.dialer),
+		grpc.WithDialer(te.e.dialer),
 		grpc.WithUserAgent(te.userAgent),
 	}
 
@@ -661,7 +644,7 @@ func (te *test) declareLogNoise(phrases ...string) {
 }
 
 func (te *test) withServerTester(fn func(st *serverTester)) {
-	c, err := te.dialer(te.srvAddr, 10*time.Second)
+	c, err := te.e.dialer(te.srvAddr, 10*time.Second)
 	if err != nil {
 		te.t.Fatal(err)
 	}
@@ -1106,7 +1089,7 @@ func testFailFast(t *testing.T, e env) {
 		if grpc.Code(err) == codes.Unavailable {
 			break
 		}
-		t.Logf("%v.EmptyCall(_, _) = _, %v", tc, err)
+		fmt.Printf("%v.EmptyCall(_, _) = _, %v", tc, err)
 		time.Sleep(10 * time.Millisecond)
 	}
 	// The client keeps reconnecting and ongoing fail-fast RPCs should fail with code.Unavailable.
@@ -2239,8 +2222,7 @@ func testPeerClientSide(t *testing.T, e env) {
 		t.Fatalf("TestService/EmptyCall(_, _) = _, %v, want _, <nil>", err)
 	}
 	pa := peer.Addr.String()
-	switch e.network {
-	case "unix", "bufconn":
+	if e.network == "unix" {
 		if pa != te.srvAddr {
 			t.Fatalf("peer.Addr = %v, want %v", pa, te.srvAddr)
 		}
@@ -3986,8 +3968,12 @@ func TestDialWithBlockErrorOnBadCertificates(t *testing.T) {
 	te.startServer(&testServer{security: te.e.security})
 	defer te.tearDown()
 
-	var err error
-	te.cc, err = grpc.Dial(te.srvAddr, grpc.WithTransportCredentials(clientAlwaysFailCred{}), grpc.WithBlock(), grpc.WithDialer(te.dialer))
+	var (
+		err  error
+		opts []grpc.DialOption
+	)
+	opts = append(opts, grpc.WithTransportCredentials(clientAlwaysFailCred{}), grpc.WithBlock())
+	te.cc, err = grpc.Dial(te.srvAddr, opts...)
 	if err != errClientAlwaysFailCred {
 		te.t.Fatalf("Dial(%q) = %v, want %v", te.srvAddr, err, errClientAlwaysFailCred)
 	}
@@ -4449,16 +4435,18 @@ func (ss *stubServer) FullDuplexCall(stream testpb.TestService_FullDuplexCallSer
 
 // Start starts the server and creates a client connected to it.
 func (ss *stubServer) Start() error {
-	lis := bufconn.Listen(1024 * 1024)
-	dialer := func(string, time.Duration) (net.Conn, error) { return lis.Dial() }
+	lis, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		return fmt.Errorf(`net.Listen("tcp", "localhost:0") = %v`, err)
+	}
 	ss.cleanups = append(ss.cleanups, func() { lis.Close() })
 
 	s := grpc.NewServer()
 	testpb.RegisterTestServiceServer(s, ss)
 	go s.Serve(lis)
 	ss.cleanups = append(ss.cleanups, s.Stop)
 
-	cc, err := grpc.Dial(lis.Addr().String(), grpc.WithInsecure(), grpc.WithBlock(), grpc.WithDialer(dialer))
+	cc, err := grpc.Dial(lis.Addr().String(), grpc.WithInsecure(), grpc.WithBlock())
 	if err != nil {
 		return fmt.Errorf("grpc.Dial(%q) = %v", lis.Addr().String(), err)
 	}