Method TestFlowControlLogicalRace

test/end2end_test.go:4962–5026

(t *testing.T)

}

func (s) TestFlowControlLogicalRace(t *testing.T) {
	// Test for a regression of https://github.com/grpc/grpc-go/issues/632,
	// and other flow control bugs.

	const (
		itemCount   = 100
		itemSize    = 1 << 10
		recvCount   = 2
		maxFailures = 3
	)

	requestCount := 3000
	if raceMode {
		requestCount = 1000
	}

	lis, err := net.Listen("tcp", "localhost:0")
	if err != nil {
		t.Fatalf("Failed to listen: %v", err)
	}
	defer lis.Close()

	s := grpc.NewServer()
	testgrpc.RegisterTestServiceServer(s, &flowControlLogicalRaceServer{
		itemCount: itemCount,
		itemSize:  itemSize,
	})
	defer s.Stop()

	go s.Serve(lis)

	cc, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatalf("grpc.NewClient(%q) = %v", lis.Addr().String(), err)
	}
	defer cc.Close()
	cl := testgrpc.NewTestServiceClient(cc)

	failures := 0
	for i := 0; i < requestCount; i++ {
		ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
		output, err := cl.StreamingOutputCall(ctx, &testpb.StreamingOutputCallRequest{})
		if err != nil {
			t.Fatalf("StreamingOutputCall; err = %q", err)
		}

		for j := 0; j < recvCount; j++ {
			if _, err := output.Recv(); err != nil {
				if err == io.EOF || status.Code(err) == codes.DeadlineExceeded {
					t.Errorf("got %d responses to request %d", j, i)
					failures++
					break
				}
				t.Fatalf("Recv; err = %q", err)
			}
		}
		cancel()