/* * * Copyright 2017 gRPC authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. * */ package test import ( "context" "fmt" "net" "sync" "testing" "time" "golang.org/x/net/http2" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/internal/stubserver" "google.golang.org/grpc/status" testgrpc "google.golang.org/grpc/interop/grpc_testing" testpb "google.golang.org/grpc/interop/grpc_testing" ) type delayListener struct { net.Listener closeCalled chan struct{} acceptCalled chan struct{} allowCloseCh chan struct{} dialed bool } func (d *delayListener) Accept() (net.Conn, error) { select { case <-d.acceptCalled: // On the second call, block until closed, then return an error. <-d.closeCalled <-d.allowCloseCh return nil, fmt.Errorf("listener is closed") default: close(d.acceptCalled) conn, err := d.Listener.Accept() if err != nil { return nil, err } // Allow closing of listener only after accept. // Note: Dial can return successfully, yet Accept // might now have finished. d.allowClose() return conn, nil } } func (d *delayListener) allowClose() { close(d.allowCloseCh) } func (d *delayListener) Close() error { close(d.closeCalled) go func() { <-d.allowCloseCh d.Listener.Close() }() return nil } func (d *delayListener) Dial(ctx context.Context) (net.Conn, error) { if d.dialed { // Only hand out one connection (net.Dial can return more even after the // listener is closed). This is not thread-safe, but Dial should never be // called concurrently in this environment. return nil, fmt.Errorf("no more conns") } d.dialed = true return (&net.Dialer{}).DialContext(ctx, "tcp", d.Listener.Addr().String()) } // TestGracefulStop ensures GracefulStop causes new connections to fail. // // Steps of this test: // 1. Start Server // 2. GracefulStop() Server after listener's Accept is called, but don't // allow Accept() to exit when Close() is called on it. // 3. Create a new connection to the server after listener.Close() is called. // Server should close this connection immediately, before handshaking. // 4. Send an RPC on the new connection. Should see Unavailable error // because the ClientConn is in transient failure. func (s) TestGracefulStop(t *testing.T) { lis, err := net.Listen("tcp", "localhost:0") if err != nil { t.Fatalf("Error listenening: %v", err) } dlis := &delayListener{ Listener: lis, acceptCalled: make(chan struct{}), closeCalled: make(chan struct{}), allowCloseCh: make(chan struct{}), } d := func(ctx context.Context, _ string) (net.Conn, error) { return dlis.Dial(ctx) } ss := &stubserver.StubServer{ FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error { _, err := stream.Recv() if err != nil { return err } return stream.Send(&testpb.StreamingOutputCallResponse{}) }, } s := grpc.NewServer() testgrpc.RegisterTestServiceServer(s, ss) // 1. Start Server wg := sync.WaitGroup{} wg.Add(1) go func() { s.Serve(dlis) wg.Done() }() // 2. GracefulStop() Server after listener's Accept is called, but don't // allow Accept() to exit when Close() is called on it. <-dlis.acceptCalled wg.Add(1) go func() { s.GracefulStop() wg.Done() }() // 3. Create a new connection to the server after listener.Close() is called. // Server should close this connection immediately, before handshaking. <-dlis.closeCalled // Block until GracefulStop calls dlis.Close() // Now dial. The listener's Accept method will return a valid connection, // even though GracefulStop has closed the listener. ctx, dialCancel := context.WithTimeout(context.Background(), defaultTestTimeout) defer dialCancel() cc, err := grpc.DialContext(ctx, "", grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithContextDialer(d)) if err != nil { t.Fatalf("grpc.DialContext(_, %q, _) = %v", lis.Addr().String(), err) } client := testgrpc.NewTestServiceClient(cc) defer cc.Close() // 4. Send an RPC on the new connection. // The server would send a GOAWAY first, but we are delaying the server's // writes for now until the client writes more than the preface. ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) if _, err = client.FullDuplexCall(ctx); err == nil || status.Code(err) != codes.Unavailable { t.Fatalf("FullDuplexCall= _, %v; want _, ", err) } cancel() wg.Wait() } // TestGracefulStopClosesConnAfterLastStream ensures that a server closes the // connections to its clients when the final stream has completed after // a GOAWAY. func (s) TestGracefulStopClosesConnAfterLastStream(t *testing.T) { handlerCalled := make(chan struct{}) gracefulStopCalled := make(chan struct{}) ts := &funcServer{streamingInputCall: func(stream testgrpc.TestService_StreamingInputCallServer) error { close(handlerCalled) // Initiate call to GracefulStop. <-gracefulStopCalled // Wait for GOAWAYs to be received by the client. return nil }} te := newTest(t, tcpClearEnv) te.startServer(ts) defer te.tearDown() te.withServerTester(func(st *serverTester) { st.writeHeadersGRPC(1, "/grpc.testing.TestService/StreamingInputCall", false) <-handlerCalled // Wait for the server to invoke its handler. // Gracefully stop the server. gracefulStopDone := make(chan struct{}) go func() { te.srv.GracefulStop() close(gracefulStopDone) }() st.wantGoAway(http2.ErrCodeNo) // Server sends a GOAWAY due to GracefulStop. pf := st.wantPing() // Server sends a ping to verify client receipt. st.writePing(true, pf.Data) // Send ping ack to confirm. st.wantGoAway(http2.ErrCodeNo) // Wait for subsequent GOAWAY to indicate no new stream processing. close(gracefulStopCalled) // Unblock server handler. fr := st.wantAnyFrame() // Wait for trailer. hdr, ok := fr.(*http2.MetaHeadersFrame) if !ok { t.Fatalf("Received unexpected frame of type (%T) from server: %v; want HEADERS", fr, fr) } if !hdr.StreamEnded() { t.Fatalf("Received unexpected HEADERS frame from server: %v; want END_STREAM set", fr) } st.wantRSTStream(http2.ErrCodeNo) // Server should send RST_STREAM because client did not half-close. <-gracefulStopDone // Wait for GracefulStop to return. }) } // TestGracefulStopBlocksUntilGRPCConnectionsTerminate ensures that // GracefulStop() blocks until all ongoing RPCs finished. func (s) TestGracefulStopBlocksUntilGRPCConnectionsTerminate(t *testing.T) { unblockGRPCCall := make(chan struct{}) grpcCallExecuting := make(chan struct{}) ss := &stubserver.StubServer{ UnaryCallF: func(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error) { close(grpcCallExecuting) <-unblockGRPCCall return &testpb.SimpleResponse{}, nil }, } err := ss.Start(nil) if err != nil { t.Fatalf("StubServer.start failed: %s", err) } t.Cleanup(ss.Stop) grpcClientCallReturned := make(chan struct{}) go func() { clt := ss.Client _, err := clt.UnaryCall(context.Background(), &testpb.SimpleRequest{}) if err != nil { t.Errorf("rpc failed with error: %s", err) } close(grpcClientCallReturned) }() gracefulStopReturned := make(chan struct{}) <-grpcCallExecuting go func() { ss.S.GracefulStop() close(gracefulStopReturned) }() select { case <-gracefulStopReturned: t.Error("GracefulStop returned before rpc method call ended") case <-time.After(defaultTestShortTimeout): } unblockGRPCCall <- struct{}{} <-grpcClientCallReturned <-gracefulStopReturned } // TestStopAbortsBlockingGRPCCall ensures that when Stop() is called while an ongoing RPC // is blocking that: // - Stop() returns // - and the RPC fails with an connection closed error on the client-side func (s) TestStopAbortsBlockingGRPCCall(t *testing.T) { unblockGRPCCall := make(chan struct{}) grpcCallExecuting := make(chan struct{}) ss := &stubserver.StubServer{ UnaryCallF: func(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error) { close(grpcCallExecuting) <-unblockGRPCCall return &testpb.SimpleResponse{}, nil }, } err := ss.Start(nil) if err != nil { t.Fatalf("StubServer.start failed: %s", err) } t.Cleanup(ss.Stop) grpcClientCallReturned := make(chan struct{}) go func() { clt := ss.Client _, err := clt.UnaryCall(context.Background(), &testpb.SimpleRequest{}) if err == nil || !isConnClosedErr(err) { t.Errorf("expected rpc to fail with connection closed error, got: %v", err) } close(grpcClientCallReturned) }() <-grpcCallExecuting ss.S.Stop() unblockGRPCCall <- struct{}{} <-grpcClientCallReturned }